88 lines
3.0 KiB
Python
88 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import json
|
|
from typing import List, Callable, Awaitable
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from fastapi.responses import StreamingResponse
|
|
|
|
from app.streaming.binance_hub import hub as binance_hub
|
|
from app.streaming.alpha_hub import alpha_hub
|
|
|
|
router = APIRouter(prefix="/stream", tags=["Stream"])
|
|
|
|
|
|
@router.get("/sse")
|
|
async def stream_sse(symbols: str = "BTCUSDT,XAUUSD", timeframe: str = "1m"):
|
|
"""
|
|
Server-Sent Events (SSE) multiplexer for multiple symbols over a single connection.
|
|
- Supports 1m timeframe (server streams 1m updates; clients can resample locally).
|
|
- symbols: comma-separated list (e.g., BTCUSDT,ETHUSDT,XAUUSD)
|
|
"""
|
|
if not symbols:
|
|
raise HTTPException(status_code=400, detail="symbols must not be empty")
|
|
if timeframe != "1m":
|
|
raise HTTPException(status_code=400, detail="Only timeframe=1m is supported")
|
|
|
|
syms: List[str] = [s.strip().upper().replace("/", "") for s in symbols.split(",") if s.strip()]
|
|
if not syms:
|
|
raise HTTPException(status_code=400, detail="No valid symbols provided")
|
|
|
|
out_queue: asyncio.Queue = asyncio.Queue(maxsize=1000)
|
|
tasks: List[asyncio.Task] = []
|
|
unsubscribers: List[Callable[[], Awaitable[None]]] = []
|
|
|
|
async def add_subscription(sym: str):
|
|
if sym.startswith("XAU"):
|
|
q, unsubscribe = await alpha_hub.subscribe(sym, timeframe="1m")
|
|
else:
|
|
q, unsubscribe = await binance_hub.subscribe(sym, timeframe="1m")
|
|
unsubscribers.append(unsubscribe)
|
|
|
|
async def worker():
|
|
try:
|
|
while True:
|
|
evt = await q.get()
|
|
if evt is None:
|
|
break
|
|
try:
|
|
await out_queue.put(evt)
|
|
except Exception:
|
|
await asyncio.sleep(0)
|
|
except asyncio.CancelledError:
|
|
pass
|
|
tasks.append(asyncio.create_task(worker()))
|
|
|
|
for s in syms:
|
|
await add_subscription(s)
|
|
|
|
async def event_generator():
|
|
try:
|
|
while True:
|
|
try:
|
|
evt = await asyncio.wait_for(out_queue.get(), timeout=15.0)
|
|
data = json.dumps(evt, separators=(",", ":"))
|
|
yield f"event: kline\n".encode("utf-8")
|
|
yield f"data: {data}\n\n".encode("utf-8")
|
|
except asyncio.TimeoutError:
|
|
# Keep-alive comment
|
|
yield b": ping\n\n"
|
|
finally:
|
|
for t in tasks:
|
|
t.cancel()
|
|
for t in tasks:
|
|
with contextlib.suppress(Exception):
|
|
await t
|
|
for u in unsubscribers:
|
|
with contextlib.suppress(Exception):
|
|
await u()
|
|
|
|
headers = {
|
|
"Cache-Control": "no-cache",
|
|
"Connection": "keep-alive",
|
|
"X-Accel-Buffering": "no",
|
|
}
|
|
return StreamingResponse(event_generator(), media_type="text/event-stream", headers=headers)
|