116 lines
4.3 KiB
Python
116 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Query, HTTPException
|
|
from typing import List, Dict, Any
|
|
|
|
from app.services.crypto.binance_rest import fetch_klines as binance_klines
|
|
from app.services.metals.alpha_fx import fetch_fx_intraday, fetch_fx_daily
|
|
from app.utils.cache import TTLCache
|
|
from app.streaming.live_store import live_store
|
|
|
|
router = APIRouter(prefix="/ohlcv", tags=["OHLCV"])
|
|
|
|
_cache = TTLCache(default_ttl=60, maxsize=128)
|
|
|
|
|
|
def _resample(data: List[Dict[str, Any]], timeframe: str) -> List[Dict[str, Any]]:
|
|
# data is ascending, 1m or 5m depending on source
|
|
import math
|
|
seconds_map = {"1m": 60, "5m": 300, "1h": 3600, "4h": 14400, "1d": 86400}
|
|
tf_sec = seconds_map.get(timeframe, 60)
|
|
buckets: Dict[int, Dict[str, Any]] = {}
|
|
for d in data:
|
|
b = (d["time"] // tf_sec) * tf_sec
|
|
cur = buckets.get(b)
|
|
if cur is None:
|
|
buckets[b] = {
|
|
"time": b,
|
|
"open": d["open"],
|
|
"high": d["high"],
|
|
"low": d["low"],
|
|
"close": d["close"],
|
|
"volume": d.get("volume", 0.0),
|
|
}
|
|
else:
|
|
cur["high"] = max(cur["high"], d["high"])
|
|
cur["low"] = min(cur["low"], d["low"])
|
|
cur["close"] = d["close"]
|
|
cur["volume"] = cur.get("volume", 0.0) + d.get("volume", 0.0)
|
|
out = list(buckets.values())
|
|
out.sort(key=lambda x: x["time"])
|
|
return out
|
|
|
|
|
|
def _ttl_for(sym: str, timeframe: str) -> int:
|
|
# Tune TTL based on timeframe and provider characteristics
|
|
if sym.startswith("XAU"):
|
|
# Alpha Vantage free tier ~ 1/min practical cadence
|
|
if timeframe in ("1m", "5m"): return 60
|
|
if timeframe in ("1h", "4h"): return 300
|
|
return 3600
|
|
else:
|
|
# Binance updates are frequent; cache briefly
|
|
if timeframe == "1m": return 10
|
|
if timeframe in ("5m",): return 20
|
|
if timeframe in ("1h", "4h"): return 120
|
|
return 900
|
|
|
|
|
|
@router.get("")
|
|
async def get_ohlcv(
|
|
symbol: str = Query(..., description="e.g., BTCUSDT, ETHUSDT, XAUUSD"),
|
|
timeframe: str = Query("1m", description="1m,5m,1h,4h,1d"),
|
|
limit: int = Query(500, ge=10, le=1000),
|
|
) -> List[Dict[str, Any]]:
|
|
try:
|
|
sym = symbol.upper().replace("/", "")
|
|
key = (sym, timeframe)
|
|
cached = _cache.get(key)
|
|
if cached is not None:
|
|
return cached[-limit:]
|
|
|
|
if sym.startswith("XAU"):
|
|
# Prefer live store 1m if available (ingested by alpha_hub)
|
|
live_1m = live_store.get_history(sym, "1m")
|
|
if live_1m:
|
|
if timeframe == "1m":
|
|
return live_1m[-limit:]
|
|
data = _resample(live_1m, timeframe)
|
|
return data[-limit:]
|
|
# Fallback to Alpha Vantage REST
|
|
if timeframe in ("1m", "5m"):
|
|
base_tf = timeframe
|
|
data = await fetch_fx_intraday(sym, interval="1min" if timeframe == "1m" else "5min")
|
|
elif timeframe in ("1h", "4h"):
|
|
base_tf = "5m"
|
|
data = await fetch_fx_intraday(sym, interval="5min")
|
|
else: # daily
|
|
base_tf = "1d"
|
|
data = await fetch_fx_daily(sym)
|
|
if timeframe != base_tf:
|
|
data = _resample(data, timeframe)
|
|
ttl = _ttl_for(sym, timeframe)
|
|
_cache.set(key, data, ttl=ttl)
|
|
return data[-limit:]
|
|
else:
|
|
# Binance
|
|
if timeframe not in ("1m", "5m", "1h", "4h", "1d"):
|
|
raise HTTPException(status_code=400, detail="Unsupported timeframe")
|
|
# Prefer live store for 1m data if available
|
|
live_1m = live_store.get_history(sym, "1m")
|
|
if live_1m:
|
|
if timeframe == "1m":
|
|
return live_1m[-limit:]
|
|
# Resample from 1m to requested timeframe
|
|
data = _resample(live_1m, timeframe)
|
|
return data[-limit:]
|
|
# Fallback to REST
|
|
data = await binance_klines(sym, interval=timeframe, limit=1000)
|
|
ttl = _ttl_for(sym, timeframe)
|
|
_cache.set(key, data, ttl=ttl)
|
|
return data[-limit:]
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|