oandapy

โออันดาปี

oandapy เป็น wrapper หลามสำหรับ REST API ของ OANDA

ติดตั้ง

ใช้ pip:

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

oandapy ขึ้นอยู่กับคำขอของ python ซึ่งจะถูกติดตั้งโดยอัตโนมัติ

การใช้งาน

รวมโมดูล oandapy และสร้างอินสแตนซ์ oandapy ด้วยข้อมูลประจำตัวบัญชีของคุณ สำหรับ FxGame และ FxTrade จะต้องระบุโทเค็นการเข้าถึง

 import oandapy

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

อาร์กิวเมนต์ของคีย์เวิร์ดสำหรับฟังก์ชันต่างๆ จะถูกแมปกับฟังก์ชันที่มีให้ใช้งานสำหรับแต่ละตำแหน่งข้อมูลในเอกสาร Oanda API ดังนั้นไลบรารีนี้จะไม่มีการระงับการเปลี่ยนแปลงใน API ให้คุณใช้งาน สำหรับการเรียก API แต่ละครั้ง oandapy จะส่งคืนออบเจ็กต์ Python ดั้งเดิม ซึ่งแปลงจาก JSON ดังนั้นคุณไม่จำเป็นต้องทำ

คลาส EndpointsMixin ใน oandapy.py มีมิกซ์อินของตำแหน่งข้อมูล 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)