oandapy es un contenedor de Python para la API REST de OANDA.
Usando pipa:
$ pip install git+https://github.com/oanda/oandapy.git
oandapy depende de las solicitudes de Python, que se instalarán automáticamente.
Incluya el módulo oandapy y cree una instancia de oandapy con las credenciales de su cuenta. Para FxGame y FxTrade, se debe proporcionar un token de acceso.
import oandapy
oanda = oandapy.API(environment="practice", access_token="abcdefghijk...")
Los argumentos de palabras clave para funciones se asignan a las funciones disponibles para cada punto final en los documentos de la API de Oanda, por lo que esta biblioteca no impide que usted use los cambios en la API. Para cada llamada a la API, oandapy devuelve un objeto Python nativo, convertido de JSON para que usted no tenga que hacerlo.
La clase EndpointsMixin en oandapy.py contiene una combinación de todos los puntos finales de la API de Oanda.
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
)
Cree una clase de transmisor personalizada para configurar cómo desea manejar los datos. Cada tick se envía a través de las funciones on_success y on_error . Dado que estos métodos son métodos abstractos, debe anularlos para manejar la transmisión de datos.
El siguiente ejemplo imprime los ticks de recuento de la transmisión y luego se desconecta.
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()
Inicialice una instancia de su transmisor personalizado y comience a conectarse a la transmisión. Consulte http://developer.oanda.com/rest-live/streaming/ para obtener más documentación.
account = "12345"
stream = MyStreamer(environment="practice", access_token="abcdefghijk...")
stream.rates(account, instruments="EUR_USD,EUR_JPY,US30_USD,DE30_EUR")
El mismo procedimiento se puede utilizar para eventos en streaming.
stream = MyStreamer(environment="practice", access_token="abcdefghijk...")
stream.events(ignore_heartbeat=False)
