Python WebSocket 相場購読と自動再接続の実装

pip install websockets

1import asyncio, json, os, random, uuid
2import websockets
3
4WS_URL = "wss://quote.alltick.co/quote-stock-b-ws-api?token={token}"
5CMD_SUBSCRIBE_TRADE = 22004
6CMD_HEARTBEAT = 22000
7HEARTBEAT_SECONDS = 10
8CODES = ["700.HK", "AAPL.US"]
9
10def message(cmd_id: int, data: dict) -> str:
11    return json.dumps({"cmd_id": cmd_id, "seq_id": 1, "trace": str(uuid.uuid4()), "data": data})
12
13async def heartbeat(ws):
14    while True:
15        await asyncio.sleep(HEARTBEAT_SECONDS)
16        await ws.send(message(CMD_HEARTBEAT, {}))
17
18async def run_once(url: str):
19    async with websockets.connect(url, ping_interval=None) as ws:
20        await ws.send(message(CMD_SUBSCRIBE_TRADE, {"symbol_list": [{"code": c} for c in CODES]}))
21        beat = asyncio.create_task(heartbeat(ws))
22        try:
23            async for raw in ws:
24                handle(json.loads(raw))
25        finally:
26            beat.cancel()
27
28def handle(msg: dict):
29    if msg.get("cmd_id") == CMD_HEARTBEAT + 1:
30        return
31    print(msg)
32
33async def main():
34    token = os.getenv("ALLTICK_API_TOKEN", "YOUR_API_TOKEN")
35    url = WS_URL.format(token=token)
36    delay = 1
37    while True:
38        try:
39            await run_once(url)
40            delay = 1
41        except (websockets.WebSocketException, OSError) as exc:
42            wait = min(delay, 60) + random.uniform(0, 1)
43            print(f"切断検出: {exc!r}、{wait:.1f} 秒後に再接続します")
44            await asyncio.sleep(wait)
45            delay *= 2
46
47if __name__ == "__main__":
48    asyncio.run(main())