216 lines
8.8 KiB
Python
216 lines
8.8 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from typing import Dict, Set, Tuple, Any
|
|
|
|
import httpx
|
|
import asyncio
|
|
import time
|
|
import random
|
|
|
|
from app.config import settings
|
|
from app.streaming.live_store import live_store
|
|
|
|
ALPHA_BASE = "https://www.alphavantage.co/query"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AVKey:
|
|
symbol: str # e.g., XAUUSD
|
|
timeframe: str # '1m' only for hub
|
|
|
|
|
|
class AlphaVantageHub:
|
|
"""
|
|
Polls Alpha Vantage FX_INTRADAY for the latest bar per (symbol, 1m),
|
|
fans out to subscribers via asyncio.Queue, and ingests into live_store.
|
|
Ensures a single poller per (symbol,timeframe).
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._subs: Dict[AVKey, Set[asyncio.Queue]] = {}
|
|
self._tasks: Dict[AVKey, asyncio.Task] = {}
|
|
self._lock = asyncio.Lock()
|
|
# Conditional request caches
|
|
self._etags: Dict[AVKey, str] = {}
|
|
self._last_mod: Dict[AVKey, str] = {}
|
|
# Global token-bucket (Alpha free tier ~5 req/min)
|
|
self._rate_lock = asyncio.Lock()
|
|
self._tokens: float = 5.0
|
|
self._max_tokens: float = 5.0
|
|
self._refill_rate_per_sec: float = 5.0 / 60.0
|
|
self._last_refill_ts: float = time.time()
|
|
|
|
async def _acquire_token(self) -> None:
|
|
# Simple async token bucket
|
|
while True:
|
|
async with self._rate_lock:
|
|
now = time.time()
|
|
elapsed = now - self._last_refill_ts
|
|
if elapsed > 0:
|
|
self._tokens = min(self._max_tokens, self._tokens + elapsed * self._refill_rate_per_sec)
|
|
self._last_refill_ts = now
|
|
if self._tokens >= 1.0:
|
|
self._tokens -= 1.0
|
|
return
|
|
# Not enough tokens, compute wait time for next token
|
|
need = 1.0 - self._tokens
|
|
wait = max(0.1, need / self._refill_rate_per_sec)
|
|
await asyncio.sleep(min(wait, 5.0))
|
|
|
|
def get_status(self) -> list[dict]:
|
|
out: list[dict] = []
|
|
for key, subs in self._subs.items():
|
|
hist = live_store.get_history(key.symbol, key.timeframe)
|
|
last_ts = hist[-1]["time"] if hist else None
|
|
last_iso = None
|
|
if isinstance(last_ts, (int, float)):
|
|
try:
|
|
last_iso = datetime.utcfromtimestamp(int(last_ts)).isoformat() + "Z"
|
|
except Exception:
|
|
last_iso = None
|
|
out.append({
|
|
"symbol": key.symbol,
|
|
"timeframe": key.timeframe,
|
|
"subscribers": len(subs),
|
|
"last_event_time": last_iso,
|
|
})
|
|
return out
|
|
|
|
async def subscribe(self, symbol: str, timeframe: str = "1m") -> Tuple[asyncio.Queue, Any]:
|
|
if timeframe != "1m":
|
|
raise ValueError("AlphaVantageHub currently supports timeframe '1m' only")
|
|
key = AVKey(symbol=symbol.upper().replace("/", ""), timeframe=timeframe)
|
|
q: asyncio.Queue = asyncio.Queue(maxsize=100)
|
|
async with self._lock:
|
|
subs = self._subs.get(key)
|
|
if not subs:
|
|
subs = set()
|
|
self._subs[key] = subs
|
|
subs.add(q)
|
|
if key not in self._tasks:
|
|
self._tasks[key] = asyncio.create_task(self._run_poller(key))
|
|
|
|
async def _unsubscribe() -> None:
|
|
async with self._lock:
|
|
s = self._subs.get(key)
|
|
if s and q in s:
|
|
s.remove(q)
|
|
try:
|
|
q.put_nowait(None)
|
|
except Exception:
|
|
pass
|
|
if s is not None and len(s) == 0:
|
|
t = self._tasks.pop(key, None)
|
|
if t:
|
|
t.cancel()
|
|
self._subs.pop(key, None)
|
|
return q, _unsubscribe
|
|
|
|
async def _run_poller(self, key: AVKey) -> None:
|
|
symbol = key.symbol
|
|
from_symbol = symbol[:3]
|
|
to_symbol = symbol[3:]
|
|
apikey = settings.ALPHA_VANTAGE_API_KEY or "demo"
|
|
last_ts: int | None = None
|
|
poll_interval = 60 # seconds
|
|
backoff_cap = 300 # max 5 min
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
while True:
|
|
try:
|
|
await self._acquire_token()
|
|
params = {
|
|
"function": "FX_INTRADAY",
|
|
"from_symbol": from_symbol,
|
|
"to_symbol": to_symbol,
|
|
"interval": "1min",
|
|
"outputsize": "compact",
|
|
"apikey": apikey,
|
|
}
|
|
headers = {}
|
|
et = self._etags.get(key)
|
|
lm = self._last_mod.get(key)
|
|
if et:
|
|
headers["If-None-Match"] = et
|
|
if lm:
|
|
headers["If-Modified-Since"] = lm
|
|
r = await client.get(ALPHA_BASE, params=params, headers=headers)
|
|
if r.status_code == 304:
|
|
# Not modified, keep interval
|
|
delay = poll_interval + random.uniform(0, 2)
|
|
await asyncio.sleep(delay)
|
|
continue
|
|
# Raise for other non-2xx
|
|
r.raise_for_status()
|
|
# Store caching headers for next time
|
|
etag = r.headers.get("ETag")
|
|
if etag:
|
|
self._etags[key] = etag
|
|
last_mod = r.headers.get("Last-Modified")
|
|
if last_mod:
|
|
self._last_mod[key] = last_mod
|
|
js = r.json()
|
|
series = js.get("Time Series FX (1min)") or {}
|
|
if series:
|
|
latest_ts_str = max(series.keys())
|
|
dt = datetime.fromisoformat(latest_ts_str)
|
|
tsec = int(dt.timestamp())
|
|
if last_ts is None or tsec > last_ts:
|
|
row = series[latest_ts_str]
|
|
evt = {
|
|
"symbol": symbol,
|
|
"timeframe": key.timeframe,
|
|
"open_time": dt.isoformat(),
|
|
"close_time": dt.isoformat(),
|
|
"open": float(row["1. open"]),
|
|
"high": float(row["2. high"]),
|
|
"low": float(row["3. low"]),
|
|
"close": float(row["4. close"]),
|
|
"volume": float(row.get("5. volume", 0.0)),
|
|
"is_closed": True,
|
|
"source": "alpha_vantage",
|
|
}
|
|
try:
|
|
live_store.ingest_bar(symbol=symbol, timeframe="1m", bar={
|
|
"time": tsec,
|
|
"open": evt["open"],
|
|
"high": evt["high"],
|
|
"low": evt["low"],
|
|
"close": evt["close"],
|
|
"volume": evt["volume"],
|
|
})
|
|
except Exception:
|
|
pass
|
|
subs = self._subs.get(key) or set()
|
|
for q in list(subs):
|
|
try:
|
|
if q.full():
|
|
q.get_nowait()
|
|
q.put_nowait(evt)
|
|
except Exception:
|
|
try:
|
|
subs.remove(q)
|
|
except Exception:
|
|
pass
|
|
last_ts = tsec
|
|
# success -> reset interval
|
|
poll_interval = 60
|
|
except httpx.HTTPStatusError as e:
|
|
status = e.response.status_code if e.response else None
|
|
# 429 or 5xx -> exponential backoff
|
|
if status == 429 or (status and 500 <= status < 600):
|
|
poll_interval = min(backoff_cap, max(60, int(poll_interval * 2)))
|
|
# else, keep interval
|
|
except Exception:
|
|
# network or parse error
|
|
poll_interval = min(backoff_cap, max(60, int(poll_interval * 2)))
|
|
# sleep with small jitter
|
|
delay = poll_interval + random.uniform(0, 2)
|
|
await asyncio.sleep(delay)
|
|
|
|
|
|
# Singleton hub
|
|
alpha_hub = AlphaVantageHub()
|