dekun
118 lines
3.7 KiB
Python
118 lines
3.7 KiB
Python
"""多周期 K 线：优先 SQLite，不足或过期再请求币安。"""
|
|
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
from .binance import binance_client
|
|
from .chart_intervals import (
|
|
CHART_INTERVALS,
|
|
cache_minutes_for_interval,
|
|
limit_for_interval,
|
|
validate_interval,
|
|
)
|
|
from .db import get_kline_meta, get_klines_from_db, save_klines
|
|
from .exceptions import BinanceRateLimitedError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _is_db_fresh(symbol: str, interval: str, min_bars: int) -> bool:
|
|
meta = get_kline_meta(symbol, interval)
|
|
if not meta or meta.get("bar_count", 0) < min_bars:
|
|
return False
|
|
try:
|
|
last_fetch = datetime.fromisoformat(meta["last_fetch_at"])
|
|
except ValueError:
|
|
return False
|
|
age = (datetime.now() - last_fetch).total_seconds()
|
|
return age < cache_minutes_for_interval(interval) * 60
|
|
|
|
|
|
async def sync_klines(
|
|
symbol: str, interval: str, limit: int | None = None
|
|
) -> list[dict]:
|
|
"""从币安拉取指定周期并写入本地库。"""
|
|
sym = symbol.upper()
|
|
iv = validate_interval(interval)
|
|
n = min(limit or limit_for_interval(iv), 1500)
|
|
candles = await binance_client.get_klines_limit(sym, iv, n)
|
|
save_klines(sym, iv, candles)
|
|
logger.info("Saved %d %s klines for %s to DB", len(candles), iv, sym)
|
|
return candles
|
|
|
|
|
|
async def sync_daily_klines(symbol: str, limit: int | None = None) -> list[dict]:
|
|
return await sync_klines(symbol, "1d", limit)
|
|
|
|
|
|
async def get_candles(
|
|
symbol: str,
|
|
interval: str = "1d",
|
|
limit: int | None = None,
|
|
force_refresh: bool = False,
|
|
) -> tuple[list[dict], str]:
|
|
"""
|
|
返回 (candles, source)。
|
|
source: db | db_stale | binance
|
|
"""
|
|
sym = symbol.upper().strip()
|
|
iv = validate_interval(interval)
|
|
n = min(limit or limit_for_interval(iv), 1500)
|
|
min_bars = min(n, 50)
|
|
|
|
if not force_refresh and _is_db_fresh(sym, iv, min_bars):
|
|
candles = get_klines_from_db(sym, iv, n)
|
|
if len(candles) >= min_bars:
|
|
return candles, "db"
|
|
|
|
stored = get_klines_from_db(sym, iv, n)
|
|
if binance_client.is_rate_limited():
|
|
if stored:
|
|
logger.warning("Rate limited, serve stale DB klines for %s %s", sym, iv)
|
|
return stored, "db_stale"
|
|
raise BinanceRateLimitedError(binance_client.rate_limit_remaining_sec(), sym)
|
|
|
|
try:
|
|
candles = await sync_klines(sym, iv, n)
|
|
return candles, "binance"
|
|
except BinanceRateLimitedError:
|
|
if stored:
|
|
return stored, "db_stale"
|
|
raise
|
|
except Exception:
|
|
if stored:
|
|
return stored, "db_stale"
|
|
raise
|
|
|
|
|
|
async def get_daily_candles(
|
|
symbol: str,
|
|
limit: int | None = None,
|
|
force_refresh: bool = False,
|
|
) -> tuple[list[dict], str]:
|
|
return await get_candles(symbol, "1d", limit, force_refresh)
|
|
|
|
|
|
async def prefetch_symbols(symbols: list[str]) -> None:
|
|
"""后台预拉 Top 币种全周期 K 线入库（串行，避免 418）。"""
|
|
seen: set[str] = set()
|
|
for raw in symbols:
|
|
sym = raw.upper().strip()
|
|
if not sym or sym in seen or not sym.endswith("USDT"):
|
|
continue
|
|
seen.add(sym)
|
|
for interval in CHART_INTERVALS:
|
|
n = limit_for_interval(interval)
|
|
if _is_db_fresh(sym, interval, min(50, n)):
|
|
continue
|
|
if binance_client.is_rate_limited():
|
|
logger.warning("Prefetch stopped — rate limited")
|
|
return
|
|
try:
|
|
await sync_klines(sym, interval, n)
|
|
except BinanceRateLimitedError:
|
|
logger.warning("Prefetch rate limited at %s %s", sym, interval)
|
|
return
|
|
except Exception as e:
|
|
logger.warning("Prefetch %s %s failed: %s", sym, interval, e)
|