oandapy

오안다피

oandapy는 OANDA의 REST API를 위한 Python 래퍼입니다.

설치하다

핍 사용:

 $ pip install git+https://github.com/oanda/oandapy.git

oandapy는 자동으로 설치되는 python-requests에 의존합니다.

용법

oandapy 모듈을 포함하고 계정 자격 증명으로 oandapy 인스턴스를 생성합니다. FxGame 및 FxTrade의 경우 액세스 토큰을 제공해야 합니다.

 import oandapy

oanda = oandapy.API(environment="practice", access_token="abcdefghijk...")

함수에 대한 키워드 인수는 Oanda API 문서의 각 엔드포인트에 사용 가능한 함수에 매핑되므로 API에 대한 변경 사항은 이 라이브러리에서 이를 사용하여 보류되지 않습니다. 각 API 호출에 대해 oandapy는 JSON에서 변환된 기본 Python 개체를 반환하므로 그럴 필요가 없습니다.

oandapy.py의 EndpointsMixin 클래스는 모든 Oanda API 엔드포인트의 혼합을 보유합니다.

예

악기 가격 확인

 response = oanda.get_prices(instruments="EUR_USD")
prices = response.get("prices")
asking_price = prices[0].get("ask")

지정가 주문 열기

 # required datetime functions
from datetime import datetime, timedelta

# sample account_id
account_id = 1813880

# set the trade to expire after one day
trade_expire = datetime.utcnow() + timedelta(days=1)
trade_expire = trade_expire.isoformat("T") + "Z"

response = oanda.create_order(account_id,
    instrument="USD_CAD",
    units=1000,
    side='sell',
    type='limit',
    price=1.15,
    expiry=trade_expire
)

요금 스트리밍

데이터 처리 방법을 설정하려면 사용자 정의 스트리머 클래스를 만듭니다. 각 틱은 on_success 및 on_error 함수를 통해 전송됩니다. 이러한 메서드는 추상 메서드이므로 스트리밍 데이터를 처리하려면 이러한 메서드를 재정의해야 합니다.

다음 예에서는 스트림의 틱 수 를 인쇄한 다음 연결을 끊습니다.

 class MyStreamer(oandapy.Streamer):
    def __init__(self, count=10, *args, **kwargs):
        super(MyStreamer, self).__init__(*args, **kwargs)
        self.count = count
        self.reccnt = 0

    def on_success(self, data):
        print data, "n"
        self.reccnt += 1
        if self.reccnt == self.count:
            self.disconnect()

    def on_error(self, data):
        self.disconnect()

사용자 정의 스트리머의 인스턴스를 초기화하고 스트림 연결을 시작하세요. 추가 문서는 http://developer.oanda.com/rest-live/streaming/을 참조하세요.

 account = "12345"
stream = MyStreamer(environment="practice", access_token="abcdefghijk...")
stream.rates(account, instruments="EUR_USD,EUR_JPY,US30_USD,DE30_EUR")

스트리밍 이벤트에도 동일한 절차를 사용할 수 있습니다.

 stream = MyStreamer(environment="practice", access_token="abcdefghijk...")
stream.events(ignore_heartbeat=False)