Oanda WebSocket Sample

test_oanda.py

import asyncio
import json

import aiohttp
from rich.pretty import pprint


async def main():
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(
            "wss://widget.oanda.jp/ws/app/oanda_websocket?protocol=7", heartbeat=10.0
        ) as ws:
            await ws.send_str(
                r'{"event":"pusher:subscribe","data":{"auth":"","channel":"live-rate-fx"}}'
            )
            async for msg in ws:
                data = msg.json()
                if data["event"] == "live-rate":
                    data = json.loads(data["data"])
                    try:
                        usdjpy = next(
                            filter(lambda x: x["key"] == "USD_JPY", data["rates"])
                        )
                        pprint(usdjpy, expand_all=True)
                    except StopIteration:
                        pass


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

Requirements

pip install -U aiohttp rich

Reference

https://www.oanda.jp/lab-education/instruments/usdjpy/