1347 lines
50 KiB
Python
1347 lines
50 KiB
Python
import os
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
from contextlib import suppress
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Tuple, Union
|
||
|
||
import ccxt
|
||
import ccxt.async_support as ccxt_async
|
||
import pandas as pd
|
||
import websockets
|
||
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query, HTTPException, status
|
||
from fastapi.responses import JSONResponse
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
|
||
# docker compose down && docker compose build --no-cache && docker compose up -d
|
||
# docker compose down && docker compose build && docker compose up -d
|
||
|
||
try:
|
||
from technical.util import resample_to_interval # type: ignore
|
||
except ImportError: # pragma: no cover - 环境缺失依赖时自动降级
|
||
resample_to_interval = None # type: ignore
|
||
|
||
from .storage import (
|
||
ensure_storage,
|
||
read_candles,
|
||
upsert_candles,
|
||
get_last_timestamp,
|
||
)
|
||
|
||
|
||
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()
|
||
logging.basicConfig(
|
||
level=LOG_LEVEL,
|
||
format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
|
||
)
|
||
logger = logging.getLogger("datasvc")
|
||
BASE_DIR = Path(__file__).resolve().parent.parent
|
||
DEFAULT_PAIRS_FILE = BASE_DIR / "pairs.json"
|
||
RESAMPLE_AVAILABLE = resample_to_interval is not None
|
||
RESAMPLE_WARNING_EMITTED = False
|
||
WS_ENABLED = os.environ.get("WS_ENABLED", "false").lower() in {"1", "true", "yes"}
|
||
REST_POLL_INTERVAL = max(1.0, float(os.environ.get("REST_POLL_INTERVAL", "5")))
|
||
REST_POLL_WINDOW = max(1, int(os.environ.get("REST_POLL_WINDOW", "10")))
|
||
|
||
REST_MAX_CONCURRENCY = int(os.environ.get("REST_MAX_CONCURRENCY", "4"))
|
||
VERIFY_MAX_CONCURRENCY = int(os.environ.get("VERIFY_MAX_CONCURRENCY", "2"))
|
||
REST_FETCH_SEMAPHORE = asyncio.Semaphore(max(1, REST_MAX_CONCURRENCY))
|
||
VERIFY_FETCH_SEMAPHORE = asyncio.Semaphore(max(1, VERIFY_MAX_CONCURRENCY))
|
||
|
||
AGGREGATION_PLAN: Dict[str, List[str]] = {
|
||
"1m": ["2m", "3m", "4m", "5m", "10m", "15m", "20m", "25m", "30m"],
|
||
"1h": ["2h", "3h", "4h", "6h", "8h", "12h", "16h"],
|
||
"1d": ["2d", "3d", "4d", "5d", "6d"],
|
||
"1w": ["2w"],
|
||
"1M": ["2M", "3M", "6M"],
|
||
}
|
||
|
||
CandleRow = List[Union[int, float]]
|
||
|
||
|
||
def _split_env_list(value: str) -> List[str]:
|
||
return [item.strip() for item in value.split(",") if item.strip()]
|
||
|
||
|
||
def _unique_preserve(values: List[str]) -> List[str]:
|
||
seen = set()
|
||
ordered: List[str] = []
|
||
for item in values:
|
||
if item not in seen:
|
||
ordered.append(item)
|
||
seen.add(item)
|
||
return ordered
|
||
|
||
|
||
def _load_symbols() -> List[str]:
|
||
path = DEFAULT_PAIRS_FILE
|
||
if not path.is_file():
|
||
default_symbols = ["BTC/USDT:USDT"]
|
||
logger.warning("交易对配置文件不存在，使用默认值", extra={"file": str(path), "symbols": default_symbols})
|
||
return default_symbols
|
||
try:
|
||
content = path.read_text(encoding="utf-8")
|
||
data = json.loads(content)
|
||
except Exception:
|
||
default_symbols = ["BTC/USDT:USDT"]
|
||
logger.exception("读取交易对配置文件失败，使用默认值", extra={"file": str(path), "symbols": default_symbols})
|
||
return default_symbols
|
||
|
||
raw_symbols: List[str] = []
|
||
if isinstance(data, list):
|
||
raw_symbols = [str(item).strip() for item in data if isinstance(item, str) and item.strip()]
|
||
elif isinstance(data, dict):
|
||
candidates = data.get("symbols") or data.get("pairs")
|
||
if isinstance(candidates, list):
|
||
raw_symbols = [str(item).strip() for item in candidates if isinstance(item, str) and item.strip()]
|
||
if not raw_symbols:
|
||
default_symbols = ["BTC/USDT:USDT"]
|
||
logger.warning("交易对配置文件未提供有效列表，使用默认值", extra={"file": str(path), "symbols": default_symbols})
|
||
return default_symbols
|
||
|
||
symbols = _unique_preserve(raw_symbols)
|
||
logger.info("已从配置文件载入交易对", extra={"file": str(path), "symbols": symbols})
|
||
return symbols
|
||
|
||
|
||
def timeframe_to_minutes(tf: str) -> Optional[int]:
|
||
if not tf:
|
||
return None
|
||
unit = tf[-1]
|
||
try:
|
||
value = int(tf[:-1])
|
||
except ValueError:
|
||
return None
|
||
multiplier = {
|
||
"m": 1,
|
||
"h": 60,
|
||
"d": 1440,
|
||
"w": 10080,
|
||
"M": 43200, # 30 天近似
|
||
}.get(unit)
|
||
if multiplier is None:
|
||
return None
|
||
return value * multiplier
|
||
|
||
|
||
def binance_stream_symbol(symbol: str) -> str:
|
||
try:
|
||
base, rest = symbol.split("/", 1)
|
||
except ValueError:
|
||
cleaned = symbol.replace("/", "").split(":")[0]
|
||
return cleaned.lower()
|
||
quote = rest.split(":")[0]
|
||
return f"{base}{quote}".lower()
|
||
|
||
|
||
def build_stream_url(symbol: str, timeframe: str) -> str:
|
||
stream_symbol = binance_stream_symbol(symbol)
|
||
return f"{BINANCE_WS_BASE}/{stream_symbol}@kline_{timeframe}"
|
||
|
||
DATA_DIR = os.environ.get("DATA_DIR", "/data")
|
||
EXCHANGE = os.environ.get("EXCHANGE", "binance")
|
||
SYMBOLS = _load_symbols()
|
||
|
||
_default_timeframes = ["1m", "1h", "1d", "1w", "1M"]
|
||
requested_timeframes = _split_env_list(os.environ.get("TIMEFRAMES", ",".join(_default_timeframes)))
|
||
if not requested_timeframes:
|
||
requested_timeframes = _default_timeframes
|
||
|
||
FETCH_TIMEFRAMES = _unique_preserve(requested_timeframes)
|
||
AVAILABLE_TIMEFRAMES = list(FETCH_TIMEFRAMES)
|
||
for base_tf in FETCH_TIMEFRAMES:
|
||
for derived_tf in AGGREGATION_PLAN.get(base_tf, []):
|
||
if derived_tf not in AVAILABLE_TIMEFRAMES:
|
||
AVAILABLE_TIMEFRAMES.append(derived_tf)
|
||
DERIVED_TIMEFRAMES = [tf for tf in AVAILABLE_TIMEFRAMES if tf not in FETCH_TIMEFRAMES]
|
||
AGGREGATION_TARGETS = {tf: AGGREGATION_PLAN.get(tf, []) for tf in FETCH_TIMEFRAMES}
|
||
|
||
START_FROM = os.environ.get("START_FROM", "2025-09-01") # 首次启动拉取起始日期（UTC）
|
||
POLL_FACTOR = float(os.environ.get("POLL_FACTOR", "0.5")) # 轮询间隔 = tf_ms * factor
|
||
BACKOFF_BASE = float(os.environ.get("BACKOFF_BASE", "2.0"))
|
||
BACKOFF_MAX = float(os.environ.get("BACKOFF_MAX", "30.0"))
|
||
BINANCE_WS_BASE = os.environ.get("BINANCE_WS_BASE", "wss://fstream.binance.com/ws").rstrip("/")
|
||
|
||
VALID_SYMBOLS = set(SYMBOLS)
|
||
VALID_TIMEFRAMES = set(AVAILABLE_TIMEFRAMES)
|
||
|
||
ensure_storage(DATA_DIR)
|
||
|
||
|
||
def _load_verify_intervals() -> Dict[str, int]:
|
||
mapping: Dict[str, int] = {}
|
||
raw = os.environ.get("VERIFY_INTERVALS", "")
|
||
if not raw:
|
||
return mapping
|
||
parts = [item.strip() for item in raw.split(",") if item.strip()]
|
||
for part in parts:
|
||
if "=" not in part:
|
||
continue
|
||
key, value = part.split("=", 1)
|
||
key = key.strip()
|
||
value = value.strip()
|
||
if not key or not value:
|
||
continue
|
||
try:
|
||
parsed = int(value)
|
||
except ValueError:
|
||
logger.warning("解析 VERIFY_INTERVALS 失败，已忽略条目", extra={"entry": part})
|
||
continue
|
||
if parsed <= 0:
|
||
continue
|
||
mapping[key] = parsed
|
||
return mapping
|
||
|
||
|
||
DEFAULT_VERIFY_INTERVAL_MULTIPLIER = max(1, int(os.environ.get("VERIFY_DEFAULT_INTERVAL", "1")))
|
||
VERIFY_INTERVAL_MULTIPLIERS = _load_verify_intervals()
|
||
|
||
app = FastAPI(title="Local Data Service", version="0.1.0")
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"],
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
|
||
def tf_to_ms(tf: str) -> int:
|
||
minutes = timeframe_to_minutes(tf)
|
||
if minutes is None:
|
||
logger.warning("无法解析时间周期，默认使用 60 秒", extra={"timeframe": tf})
|
||
return 60_000
|
||
return minutes * 60_000
|
||
|
||
|
||
def ensure_symbol_timeframe(symbol: str, timeframe: str) -> None:
|
||
if symbol not in VALID_SYMBOLS:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail=f"symbol 必须为 {sorted(VALID_SYMBOLS)} 之一。",
|
||
)
|
||
if timeframe not in VALID_TIMEFRAMES:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail=f"tf 必须为 {sorted(VALID_TIMEFRAMES)} 之一。",
|
||
)
|
||
|
||
|
||
def parse_start_from_ms(val: str) -> int:
|
||
"""将 START_FROM 解析成毫秒级时间戳。
|
||
支持两种格式：
|
||
- YYYY-MM-DD（UTC 00:00:00）
|
||
- 整型毫秒时间戳字符串
|
||
"""
|
||
try:
|
||
return int(val)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
dt = datetime.fromisoformat(val) # 允许 '2022-01-01' 或 '2022-01-01T00:00:00'
|
||
except Exception:
|
||
# 回退到固定日期
|
||
dt = datetime(2025, 9, 1)
|
||
return int(dt.timestamp() * 1000)
|
||
|
||
|
||
class Hub:
|
||
def __init__(self) -> None:
|
||
self.subscribers: Dict[str, List[WebSocket]] = {}
|
||
|
||
def topic(self, symbol: str, timeframe: str) -> str:
|
||
return f"candles::{symbol}::{timeframe}"
|
||
|
||
async def subscribe(self, ws: WebSocket, symbol: str, timeframe: str):
|
||
topic = self.topic(symbol, timeframe)
|
||
await ws.accept()
|
||
self.subscribers.setdefault(topic, []).append(ws)
|
||
|
||
def _clean(self, topic: str):
|
||
conns = self.subscribers.get(topic, [])
|
||
self.subscribers[topic] = [w for w in conns if not w.client_state.name == "DISCONNECTED"]
|
||
|
||
async def publish(self, symbol: str, timeframe: str, payload: dict):
|
||
topic = self.topic(symbol, timeframe)
|
||
conns = self.subscribers.get(topic, [])
|
||
if not conns:
|
||
return
|
||
message = json.dumps(payload, ensure_ascii=False)
|
||
dead: List[WebSocket] = []
|
||
for ws in conns:
|
||
try:
|
||
await ws.send_text(message)
|
||
except Exception:
|
||
dead.append(ws)
|
||
if dead:
|
||
self.subscribers[topic] = [w for w in conns if w not in dead]
|
||
|
||
|
||
hub = Hub()
|
||
|
||
fetch_tasks: List[asyncio.Task] = []
|
||
verification_queues: Dict[Tuple[str, str], asyncio.Queue["VerificationJob"]] = {}
|
||
|
||
|
||
def resample_and_store(symbol: str, base_timeframe: str, derived_timeframes: List[str]) -> List[Tuple[str, List[CandleRow]]]:
|
||
if not RESAMPLE_AVAILABLE or not derived_timeframes:
|
||
return []
|
||
base_df = read_candles(DATA_DIR, symbol, base_timeframe, None, None)
|
||
if base_df.empty:
|
||
return []
|
||
base_df = base_df.copy()
|
||
if "date" not in base_df.columns:
|
||
base_df["date"] = pd.to_datetime(base_df["timestamp"], unit="ms", utc=True)
|
||
base_df = (
|
||
base_df.drop_duplicates(subset=["timestamp"], keep="last")
|
||
.sort_values("timestamp")
|
||
.reset_index(drop=True)
|
||
)
|
||
base_df["timestamp"] = base_df["timestamp"].astype("int64")
|
||
|
||
updates: List[Tuple[str, List[List[float]]]] = []
|
||
for target_tf in derived_timeframes:
|
||
minutes = timeframe_to_minutes(target_tf)
|
||
if minutes is None:
|
||
logger.warning("无法解析聚合周期", extra={"target_timeframe": target_tf})
|
||
continue
|
||
try:
|
||
derived_df = resample_to_interval(base_df, minutes) # type: ignore[misc]
|
||
except Exception:
|
||
logger.exception(
|
||
"聚合周期计算失败",
|
||
extra={"symbol": symbol, "base_timeframe": base_timeframe, "target_timeframe": target_tf},
|
||
)
|
||
continue
|
||
if derived_df is None or derived_df.empty:
|
||
continue
|
||
derived_df = derived_df.copy()
|
||
if "timestamp" not in derived_df.columns:
|
||
if "date" in derived_df.columns:
|
||
dates = pd.to_datetime(derived_df["date"], utc=True, errors="coerce")
|
||
derived_df["timestamp"] = (dates.astype("int64") // 1_000_000)
|
||
elif isinstance(derived_df.index, pd.DatetimeIndex):
|
||
idx = derived_df.index
|
||
if idx.tz is None:
|
||
idx = idx.tz_localize("UTC")
|
||
else:
|
||
idx = idx.tz_convert("UTC")
|
||
derived_df["timestamp"] = (idx.astype("int64") // 1_000_000)
|
||
if "timestamp" not in derived_df.columns:
|
||
logger.warning(
|
||
"聚合结果缺少 timestamp 列，已跳过",
|
||
extra={"target_timeframe": target_tf},
|
||
)
|
||
continue
|
||
derived_df = derived_df.dropna(subset=["timestamp", "open", "high", "low", "close", "volume"])
|
||
if derived_df.empty:
|
||
continue
|
||
derived_df["timestamp"] = derived_df["timestamp"].astype("int64")
|
||
derived_df = derived_df.sort_values("timestamp")
|
||
last_ts = get_last_timestamp(DATA_DIR, symbol, target_tf)
|
||
if last_ts is not None:
|
||
derived_df = derived_df[derived_df["timestamp"] > last_ts]
|
||
if derived_df.empty:
|
||
continue
|
||
numpy_rows = derived_df[["timestamp", "open", "high", "low", "close", "volume"]].to_numpy()
|
||
records: List[CandleRow] = []
|
||
for ts, o, h, l, c, v in numpy_rows:
|
||
records.append(
|
||
[
|
||
int(ts),
|
||
float(o),
|
||
float(h),
|
||
float(l),
|
||
float(c),
|
||
float(v),
|
||
]
|
||
)
|
||
if not records:
|
||
continue
|
||
upsert_candles(DATA_DIR, symbol, target_tf, records)
|
||
updates.append((target_tf, records[-3:] if len(records) > 3 else records))
|
||
return updates
|
||
|
||
|
||
def normalize_candles_for_timeframe(candles: List[CandleRow], tf_ms: int) -> Tuple[List[CandleRow], List[int]]:
|
||
if not candles:
|
||
return [], []
|
||
normalized_map: Dict[int, CandleRow] = {}
|
||
for row in candles:
|
||
if not row:
|
||
continue
|
||
try:
|
||
ts = int(row[0])
|
||
o = float(row[1])
|
||
h = float(row[2])
|
||
l = float(row[3])
|
||
c = float(row[4])
|
||
v = float(row[5])
|
||
except (TypeError, ValueError, IndexError):
|
||
continue
|
||
normalized_map[ts] = [ts, o, h, l, c, v]
|
||
ordered_ts = sorted(normalized_map.keys())
|
||
normalized: List[CandleRow] = []
|
||
missing: List[int] = []
|
||
last_ts: Optional[int] = None
|
||
for ts in ordered_ts:
|
||
normalized.append(normalized_map[ts])
|
||
if last_ts is not None and tf_ms > 0:
|
||
delta = ts - last_ts
|
||
if delta > tf_ms:
|
||
gap_ts = last_ts + tf_ms
|
||
while gap_ts < ts:
|
||
missing.append(gap_ts)
|
||
gap_ts += tf_ms
|
||
last_ts = ts
|
||
return normalized, missing
|
||
|
||
|
||
def compute_live_derived_updates(
|
||
symbol: str,
|
||
base_timeframe: str,
|
||
derived_timeframes: List[str],
|
||
base_tf_ms: int,
|
||
candles: List[CandleRow],
|
||
last_closed_ts: Optional[int],
|
||
) -> Dict[str, List[Tuple[CandleRow, bool]]]:
|
||
updates: Dict[str, List[Tuple[CandleRow, bool]]] = {}
|
||
if not candles or not derived_timeframes or base_tf_ms <= 0:
|
||
return updates
|
||
|
||
derived_ms_map: Dict[str, int] = {}
|
||
max_multiplier = 1
|
||
for target_tf in derived_timeframes:
|
||
derived_ms = tf_to_ms(target_tf)
|
||
if derived_ms is None or derived_ms <= 0 or derived_ms % base_tf_ms != 0:
|
||
continue
|
||
multiplier = derived_ms // base_tf_ms
|
||
derived_ms_map[target_tf] = derived_ms
|
||
if multiplier > max_multiplier:
|
||
max_multiplier = multiplier
|
||
if not derived_ms_map:
|
||
return updates
|
||
|
||
window_ms = max_multiplier * base_tf_ms
|
||
newest_ts = max(int(row[0]) for row in candles if row)
|
||
base_start = newest_ts - window_ms + base_tf_ms
|
||
if base_start < 0:
|
||
base_start = 0
|
||
|
||
base_df = read_candles(DATA_DIR, symbol, base_timeframe, base_start, newest_ts)
|
||
if base_df.empty:
|
||
return updates
|
||
base_df = base_df.sort_values("timestamp")
|
||
|
||
base_rows: List[Tuple[int, float, float, float, float, float]] = []
|
||
for record in candles:
|
||
try:
|
||
ts = int(record[0])
|
||
if ts < base_start:
|
||
continue
|
||
base_rows.append(
|
||
(
|
||
ts,
|
||
float(record[1]),
|
||
float(record[2]),
|
||
float(record[3]),
|
||
float(record[4]),
|
||
float(record[5]),
|
||
)
|
||
)
|
||
except (TypeError, ValueError, IndexError):
|
||
continue
|
||
|
||
if base_rows:
|
||
temp_df = pd.DataFrame(
|
||
base_rows,
|
||
columns=["timestamp", "open", "high", "low", "close", "volume"],
|
||
)
|
||
base_df = pd.concat([base_df, temp_df], ignore_index=True)
|
||
|
||
if base_df.empty:
|
||
return updates
|
||
|
||
base_df = (
|
||
base_df.drop_duplicates(subset=["timestamp"], keep="last")
|
||
.sort_values("timestamp")
|
||
.reset_index(drop=True)
|
||
)
|
||
|
||
base_df_indexed = base_df.set_index("timestamp", drop=False)
|
||
if base_df_indexed.empty:
|
||
return updates
|
||
|
||
for target_tf, derived_ms in derived_ms_map.items():
|
||
multiplier = derived_ms // base_tf_ms
|
||
rows_with_status: List[Tuple[CandleRow, bool]] = []
|
||
latest_available_ts = int(base_df_indexed.index.max())
|
||
candidate_start = max(base_start, int(base_df_indexed.index.min()))
|
||
first_bucket = (candidate_start // derived_ms) * derived_ms
|
||
if first_bucket < candidate_start:
|
||
first_bucket += derived_ms
|
||
last_possible_start = latest_available_ts - (multiplier - 1) * base_tf_ms
|
||
current_start = first_bucket
|
||
while current_start <= last_possible_start:
|
||
expected_ts = [current_start + i * base_tf_ms for i in range(multiplier)]
|
||
subset = base_df_indexed.reindex(expected_ts)
|
||
if subset.isna().any().any():
|
||
current_start += derived_ms
|
||
continue
|
||
start_ts = current_start
|
||
end_ts = start_ts + derived_ms - base_tf_ms
|
||
row: CandleRow = [
|
||
start_ts,
|
||
float(subset.iloc[0]["open"]),
|
||
float(subset["high"].max()),
|
||
float(subset["low"].min()),
|
||
float(subset.iloc[-1]["close"]),
|
||
float(subset["volume"].sum()),
|
||
]
|
||
closed = last_closed_ts is not None and last_closed_ts >= end_ts
|
||
upsert_candles(DATA_DIR, symbol, target_tf, [row])
|
||
rows_with_status.append((row, closed))
|
||
current_start += derived_ms
|
||
if rows_with_status:
|
||
updates[target_tf] = rows_with_status
|
||
return updates
|
||
|
||
|
||
@dataclass
|
||
class FetchState:
|
||
symbol: str
|
||
timeframe: str
|
||
started_at: datetime = field(default_factory=datetime.utcnow)
|
||
last_fetch_at: Optional[datetime] = None
|
||
last_candle_ts: Optional[int] = None
|
||
consecutive_errors: int = 0
|
||
last_error: Optional[str] = None
|
||
last_verified_ts: Optional[int] = None
|
||
|
||
def to_payload(self) -> dict:
|
||
def serialize_dt(dt: Optional[datetime]) -> Optional[str]:
|
||
if not dt:
|
||
return None
|
||
return dt.replace(microsecond=0).isoformat() + "Z"
|
||
|
||
return {
|
||
"symbol": self.symbol,
|
||
"timeframe": self.timeframe,
|
||
"started_at": serialize_dt(self.started_at),
|
||
"last_fetch_at": serialize_dt(self.last_fetch_at),
|
||
"last_candle_ts": self.last_candle_ts,
|
||
"consecutive_errors": self.consecutive_errors,
|
||
"last_error": self.last_error,
|
||
"last_verified_ts": self.last_verified_ts,
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class VerificationJob:
|
||
timestamp: int
|
||
count: int = 1
|
||
|
||
|
||
fetch_states: Dict[Tuple[str, str], FetchState] = {}
|
||
|
||
|
||
def enqueue_verification_job(symbol: str, timeframe: str, timestamp: int, count: int) -> None:
|
||
state_key = (symbol, timeframe)
|
||
queue = verification_queues.get(state_key)
|
||
if queue is None:
|
||
return
|
||
state = fetch_states.get(state_key)
|
||
if state is None:
|
||
state = FetchState(symbol=symbol, timeframe=timeframe)
|
||
fetch_states[state_key] = state
|
||
tf_ms = tf_to_ms(timeframe)
|
||
interval_multiplier = VERIFY_INTERVAL_MULTIPLIERS.get(timeframe, DEFAULT_VERIFY_INTERVAL_MULTIPLIER)
|
||
min_gap_ms: Optional[int] = None
|
||
if tf_ms and tf_ms > 0:
|
||
min_gap_ms = tf_ms * max(1, interval_multiplier)
|
||
if state.last_verified_ts is not None and timestamp <= state.last_verified_ts:
|
||
logger.debug(
|
||
"跳过校验任务，已验证更晚时间",
|
||
extra={"symbol": symbol, "timeframe": timeframe, "timestamp": timestamp, "last_verified": state.last_verified_ts},
|
||
)
|
||
return
|
||
if min_gap_ms is not None and state.last_verified_ts is not None:
|
||
gap = timestamp - state.last_verified_ts
|
||
if gap < min_gap_ms:
|
||
logger.debug(
|
||
"跳过校验任务，间隔不足",
|
||
extra={
|
||
"symbol": symbol,
|
||
"timeframe": timeframe,
|
||
"timestamp": timestamp,
|
||
"last_verified": state.last_verified_ts,
|
||
"required_gap_ms": min_gap_ms,
|
||
"actual_gap_ms": gap,
|
||
},
|
||
)
|
||
return
|
||
job = VerificationJob(timestamp=int(timestamp), count=max(1, int(count)))
|
||
try:
|
||
queue.put_nowait(job)
|
||
logger.info(
|
||
"排入校验任务",
|
||
extra={"symbol": symbol, "timeframe": timeframe, "timestamp": timestamp, "count": count},
|
||
)
|
||
except asyncio.QueueFull:
|
||
logger.warning(
|
||
"验证队列已满，丢弃校验任务",
|
||
extra={"symbol": symbol, "timeframe": timeframe, "timestamp": timestamp, "count": count},
|
||
)
|
||
|
||
|
||
async def process_candles(
|
||
symbol: str,
|
||
timeframe: str,
|
||
candles: List[CandleRow],
|
||
derived_timeframes: List[str],
|
||
tf_ms: int,
|
||
finalized: bool,
|
||
allow_verification: bool,
|
||
closed_flags: Optional[List[bool]] = None,
|
||
) -> None:
|
||
if not candles:
|
||
return
|
||
if closed_flags is None or len(closed_flags) != len(candles):
|
||
closed_flags = [finalized] * len(candles)
|
||
state_key = (symbol, timeframe)
|
||
upsert_candles(DATA_DIR, symbol, timeframe, candles)
|
||
base_records = list(zip(candles, closed_flags))
|
||
last_closed_ts: Optional[int] = None
|
||
for row, is_closed in base_records:
|
||
if is_closed:
|
||
if last_closed_ts is None or row[0] > last_closed_ts:
|
||
last_closed_ts = row[0]
|
||
if last_closed_ts is None:
|
||
last_closed_ts = candles[-1][0] - tf_ms
|
||
logger.info(
|
||
"基础周期 K 线更新完成",
|
||
extra={
|
||
"symbol": symbol,
|
||
"timeframe": timeframe,
|
||
"count": len(candles),
|
||
"finalized": finalized,
|
||
"last_closed_ts": last_closed_ts,
|
||
},
|
||
)
|
||
derived_updates: List[Tuple[str, List[CandleRow]]] = []
|
||
live_derived_updates: Dict[str, List[Tuple[CandleRow, bool]]] = {}
|
||
derived_closed_ts: Dict[str, int] = {}
|
||
if derived_timeframes:
|
||
needs_resample = finalized or len(candles) > 1
|
||
if needs_resample and RESAMPLE_AVAILABLE:
|
||
derived_updates = await asyncio.to_thread(
|
||
resample_and_store,
|
||
symbol,
|
||
timeframe,
|
||
derived_timeframes,
|
||
)
|
||
if derived_updates:
|
||
logger.info(
|
||
"衍生周期批量聚合完成",
|
||
extra={
|
||
"symbol": symbol,
|
||
"base_timeframe": timeframe,
|
||
"targets": [item[0] for item in derived_updates],
|
||
"origin": "resample",
|
||
},
|
||
)
|
||
live_derived_updates = await asyncio.to_thread(
|
||
compute_live_derived_updates,
|
||
symbol,
|
||
timeframe,
|
||
derived_timeframes,
|
||
tf_ms,
|
||
candles,
|
||
last_closed_ts,
|
||
)
|
||
if live_derived_updates:
|
||
logger.info(
|
||
"衍生周期实时聚合完成",
|
||
extra={
|
||
"symbol": symbol,
|
||
"base_timeframe": timeframe,
|
||
"targets": list(live_derived_updates.keys()),
|
||
"origin": "live",
|
||
},
|
||
)
|
||
for row, is_closed in base_records[-3:]:
|
||
payload = {
|
||
"topic": f"candles.{symbol}.{timeframe}",
|
||
"type": "upsert",
|
||
"data": {
|
||
"t": row[0],
|
||
"o": row[1],
|
||
"h": row[2],
|
||
"l": row[3],
|
||
"c": row[4],
|
||
"v": row[5],
|
||
"closed": bool(is_closed),
|
||
},
|
||
}
|
||
await hub.publish(symbol, timeframe, payload)
|
||
last_closed_ts_for_derived = last_closed_ts
|
||
for target_tf, rows in derived_updates:
|
||
if not rows:
|
||
continue
|
||
target_tf_ms = tf_to_ms(target_tf)
|
||
for row in rows:
|
||
ts = int(row[0])
|
||
o, h, l, c, v = map(float, row[1:])
|
||
if target_tf_ms and target_tf_ms > 0:
|
||
derived_closed = last_closed_ts_for_derived is not None and last_closed_ts_for_derived >= ts + target_tf_ms - tf_ms
|
||
else:
|
||
derived_closed = last_closed_ts_for_derived is not None and last_closed_ts_for_derived >= ts
|
||
if derived_closed:
|
||
previous = derived_closed_ts.get(target_tf)
|
||
if previous is None or ts > previous:
|
||
derived_closed_ts[target_tf] = ts
|
||
derived_state_key = (symbol, target_tf)
|
||
derived_state = fetch_states.get(derived_state_key)
|
||
if derived_state is None:
|
||
derived_state = FetchState(symbol=symbol, timeframe=target_tf)
|
||
fetch_states[derived_state_key] = derived_state
|
||
derived_state.last_fetch_at = datetime.utcnow()
|
||
derived_state.last_candle_ts = ts
|
||
derived_state.consecutive_errors = 0
|
||
derived_state.last_error = None
|
||
payload = {
|
||
"topic": f"candles.{symbol}.{target_tf}",
|
||
"type": "upsert",
|
||
"data": {
|
||
"t": ts,
|
||
"o": o,
|
||
"h": h,
|
||
"l": l,
|
||
"c": c,
|
||
"v": v,
|
||
"closed": bool(derived_closed),
|
||
},
|
||
}
|
||
await hub.publish(symbol, target_tf, payload)
|
||
if live_derived_updates:
|
||
for target_tf, items in live_derived_updates.items():
|
||
if not items:
|
||
continue
|
||
for row, derived_closed in items:
|
||
ts = int(row[0])
|
||
if derived_closed:
|
||
previous = derived_closed_ts.get(target_tf)
|
||
if previous is None or ts > previous:
|
||
derived_closed_ts[target_tf] = ts
|
||
derived_state_key = (symbol, target_tf)
|
||
derived_state = fetch_states.get(derived_state_key)
|
||
if derived_state is None:
|
||
derived_state = FetchState(symbol=symbol, timeframe=target_tf)
|
||
fetch_states[derived_state_key] = derived_state
|
||
derived_state.last_fetch_at = datetime.utcnow()
|
||
derived_state.last_candle_ts = ts
|
||
derived_state.consecutive_errors = 0
|
||
derived_state.last_error = None
|
||
payload = {
|
||
"topic": f"candles.{symbol}.{target_tf}",
|
||
"type": "upsert",
|
||
"data": {
|
||
"t": ts,
|
||
"o": float(row[1]),
|
||
"h": float(row[2]),
|
||
"l": float(row[3]),
|
||
"c": float(row[4]),
|
||
"v": float(row[5]),
|
||
"closed": bool(derived_closed),
|
||
},
|
||
}
|
||
await hub.publish(symbol, target_tf, payload)
|
||
state = fetch_states.get(state_key)
|
||
if state:
|
||
state.last_fetch_at = datetime.utcnow()
|
||
state.last_candle_ts = candles[-1][0]
|
||
state.consecutive_errors = 0
|
||
state.last_error = None
|
||
last_closed_flag = closed_flags[-1] if closed_flags else finalized
|
||
if allow_verification and last_closed_flag:
|
||
latest_ts = candles[-1][0]
|
||
now_ms = int(datetime.utcnow().timestamp() * 1000)
|
||
if latest_ts > 0 and now_ms - latest_ts <= 2 * tf_ms:
|
||
range_count = 10 if timeframe == "1m" else 1
|
||
enqueue_verification_job(symbol, timeframe, latest_ts, range_count)
|
||
if allow_verification and derived_closed_ts:
|
||
for target_tf, closed_ts in derived_closed_ts.items():
|
||
enqueue_verification_job(symbol, target_tf, closed_ts, 10)
|
||
|
||
|
||
async def rest_catchup(
|
||
symbol: str,
|
||
timeframe: str,
|
||
derived_timeframes: List[str],
|
||
tf_ms: int,
|
||
start_since: int,
|
||
) -> None:
|
||
state_key = (symbol, timeframe)
|
||
state = fetch_states[state_key]
|
||
exchange = build_exchange()
|
||
since = start_since
|
||
backoff = 1.0
|
||
gap_retry: Dict[int, int] = {}
|
||
logger.info("开始 REST 补齐历史", extra={"symbol": symbol, "timeframe": timeframe, "since": since})
|
||
try:
|
||
while True:
|
||
now_ms = int(datetime.utcnow().timestamp() * 1000)
|
||
if since >= now_ms - tf_ms:
|
||
break
|
||
try:
|
||
async with REST_FETCH_SEMAPHORE:
|
||
candles = await exchange.fetch_ohlcv(symbol, timeframe, since=since, limit=1000)
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except (ccxt.NetworkError, ccxt.ExchangeNotAvailable, ccxt.RequestTimeout) as exc:
|
||
logger.warning(
|
||
"历史补齐网络异常，准备重试",
|
||
extra={"symbol": symbol, "timeframe": timeframe, "error": str(exc)},
|
||
)
|
||
state.last_error = str(exc)
|
||
state.consecutive_errors += 1
|
||
backoff = min(backoff * BACKOFF_BASE, BACKOFF_MAX)
|
||
await asyncio.sleep(backoff)
|
||
continue
|
||
except Exception as exc:
|
||
logger.exception(
|
||
"历史补齐发生异常，准备重试",
|
||
extra={"symbol": symbol, "timeframe": timeframe},
|
||
)
|
||
state.last_error = str(exc)
|
||
state.consecutive_errors += 1
|
||
backoff = min(backoff * BACKOFF_BASE, BACKOFF_MAX)
|
||
await asyncio.sleep(backoff)
|
||
continue
|
||
if not candles:
|
||
break
|
||
candles, missing_ts = normalize_candles_for_timeframe(candles, tf_ms)
|
||
if not candles:
|
||
since += tf_ms
|
||
backoff = 1.0
|
||
await asyncio.sleep(0.2)
|
||
continue
|
||
await process_candles(
|
||
symbol,
|
||
timeframe,
|
||
candles,
|
||
derived_timeframes,
|
||
tf_ms,
|
||
finalized=True,
|
||
allow_verification=False,
|
||
closed_flags=[True] * len(candles),
|
||
)
|
||
state.consecutive_errors = 0
|
||
state.last_error = None
|
||
backoff = 1.0
|
||
if missing_ts:
|
||
gap_start = missing_ts[0]
|
||
attempts = gap_retry.get(gap_start, 0) + 1
|
||
gap_retry[gap_start] = attempts
|
||
if attempts <= 3:
|
||
logger.warning(
|
||
"检测到缺失 K 线，准备回补",
|
||
extra={
|
||
"symbol": symbol,
|
||
"timeframe": timeframe,
|
||
"missing_from": gap_start,
|
||
"missing_to": missing_ts[-1],
|
||
"attempt": attempts,
|
||
},
|
||
)
|
||
since = gap_start
|
||
await asyncio.sleep(0.2)
|
||
continue
|
||
logger.error(
|
||
"缺失 K 线多次回补失败，已跳过",
|
||
extra={
|
||
"symbol": symbol,
|
||
"timeframe": timeframe,
|
||
"missing_from": gap_start,
|
||
"missing_to": missing_ts[-1],
|
||
},
|
||
)
|
||
gap_retry.pop(gap_start, None)
|
||
else:
|
||
gap_retry.clear()
|
||
|
||
since = candles[-1][0] + tf_ms
|
||
|
||
now_ms = int(datetime.utcnow().timestamp() * 1000)
|
||
lag = now_ms - since
|
||
if lag > tf_ms * 10:
|
||
await asyncio.sleep(0.2)
|
||
else:
|
||
await asyncio.sleep(max(1.0, tf_ms * POLL_FACTOR / 1000.0))
|
||
finally:
|
||
with suppress(Exception):
|
||
await exchange.close()
|
||
logger.info("REST 补齐完成", extra={"symbol": symbol, "timeframe": timeframe, "latest": state.last_candle_ts})
|
||
|
||
|
||
async def rest_poll_loop(
|
||
symbol: str,
|
||
timeframe: str,
|
||
derived_timeframes: List[str],
|
||
tf_ms: int,
|
||
) -> None:
|
||
state_key = (symbol, timeframe)
|
||
window = max(REST_POLL_WINDOW, 1)
|
||
interval = max(REST_POLL_INTERVAL, 1.0)
|
||
exchange = build_exchange()
|
||
try:
|
||
while True:
|
||
state = fetch_states.get(state_key)
|
||
latest_ts = state.last_candle_ts if state else None
|
||
if latest_ts is None or latest_ts <= 0:
|
||
since = parse_start_from_ms(START_FROM)
|
||
else:
|
||
since = max(0, latest_ts - (window - 1) * tf_ms)
|
||
try:
|
||
async with REST_FETCH_SEMAPHORE:
|
||
candles = await exchange.fetch_ohlcv(
|
||
symbol,
|
||
timeframe,
|
||
since=since,
|
||
limit=max(window + 2, window),
|
||
)
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except (ccxt.NetworkError, ccxt.ExchangeNotAvailable, ccxt.RequestTimeout) as exc:
|
||
logger.warning(
|
||
"实时轮询网络异常，准备重试",
|
||
extra={"symbol": symbol, "timeframe": timeframe, "error": str(exc)},
|
||
)
|
||
await asyncio.sleep(interval)
|
||
continue
|
||
except Exception as exc:
|
||
logger.exception(
|
||
"实时轮询发生异常",
|
||
extra={"symbol": symbol, "timeframe": timeframe},
|
||
)
|
||
await asyncio.sleep(interval)
|
||
continue
|
||
candles, _ = normalize_candles_for_timeframe(candles, tf_ms)
|
||
if candles:
|
||
closed_flags = [True] * len(candles)
|
||
logger.info(
|
||
"轮询拉取基础周期完成",
|
||
extra={
|
||
"symbol": symbol,
|
||
"timeframe": timeframe,
|
||
"count": len(candles),
|
||
"since": since,
|
||
"mode": "rest_poll",
|
||
},
|
||
)
|
||
try:
|
||
await process_candles(
|
||
symbol,
|
||
timeframe,
|
||
candles,
|
||
derived_timeframes,
|
||
tf_ms,
|
||
finalized=True,
|
||
allow_verification=True,
|
||
closed_flags=closed_flags,
|
||
)
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except Exception as exc:
|
||
logger.exception(
|
||
"处理基础周期 K 线失败 [%s %s]",
|
||
symbol,
|
||
timeframe,
|
||
)
|
||
state = fetch_states.get(state_key)
|
||
if state:
|
||
state.last_error = str(exc)
|
||
state.consecutive_errors += 1
|
||
await asyncio.sleep(interval)
|
||
continue
|
||
await asyncio.sleep(interval)
|
||
except asyncio.CancelledError:
|
||
raise
|
||
finally:
|
||
with suppress(Exception):
|
||
await exchange.close()
|
||
logger.info("轮询任务退出", extra={"symbol": symbol, "timeframe": timeframe})
|
||
|
||
|
||
async def stream_loop(symbol: str, timeframe: str, derived_timeframes: List[str], tf_ms: int):
|
||
state_key = (symbol, timeframe)
|
||
url = build_stream_url(symbol, timeframe)
|
||
while True:
|
||
try:
|
||
async with websockets.connect(url, ping_interval=20, ping_timeout=20) as ws:
|
||
logger.info("WebSocket 已连接", extra={"symbol": symbol, "timeframe": timeframe, "url": url})
|
||
async for message in ws:
|
||
data = json.loads(message)
|
||
kline = data.get("k")
|
||
if not kline:
|
||
continue
|
||
is_closed = bool(kline.get("x"))
|
||
row: CandleRow = [
|
||
int(kline["t"]),
|
||
float(kline["o"]),
|
||
float(kline["h"]),
|
||
float(kline["l"]),
|
||
float(kline["c"]),
|
||
float(kline["v"]),
|
||
]
|
||
await process_candles(
|
||
symbol,
|
||
timeframe,
|
||
[row],
|
||
derived_timeframes,
|
||
tf_ms,
|
||
finalized=is_closed,
|
||
allow_verification=is_closed,
|
||
closed_flags=[is_closed],
|
||
)
|
||
except asyncio.CancelledError:
|
||
logger.info("取消 WebSocket 任务", extra={"symbol": symbol, "timeframe": timeframe})
|
||
raise
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"WebSocket 连接异常，准备重连",
|
||
extra={"symbol": symbol, "timeframe": timeframe, "error": str(exc)},
|
||
)
|
||
state = fetch_states.get(state_key)
|
||
start_since = None
|
||
if state and state.last_candle_ts:
|
||
start_since = state.last_candle_ts + tf_ms
|
||
if start_since:
|
||
await rest_catchup(symbol, timeframe, derived_timeframes, tf_ms, start_since)
|
||
await asyncio.sleep(5.0)
|
||
|
||
|
||
def build_exchange():
|
||
if EXCHANGE.lower() == "binance":
|
||
return ccxt_async.binance(
|
||
{
|
||
"enableRateLimit": True,
|
||
"timeout": 20_000,
|
||
"options": {
|
||
"adjustForTimeDifference": True,
|
||
"defaultType": "future",
|
||
"defaultSubType": "linear",
|
||
"defaultMarket": "future",
|
||
"defaultSettle": "USDT",
|
||
},
|
||
}
|
||
)
|
||
raise RuntimeError(f"Unsupported EXCHANGE: {EXCHANGE}")
|
||
|
||
|
||
async def fetch_loop(symbol: str, timeframe: str):
|
||
"""初次通过 REST 补齐历史，随后持续轮询/流式拉取增量。"""
|
||
derived_timeframes = AGGREGATION_TARGETS.get(timeframe, [])
|
||
global RESAMPLE_WARNING_EMITTED
|
||
if derived_timeframes and not RESAMPLE_AVAILABLE and not RESAMPLE_WARNING_EMITTED:
|
||
logger.warning(
|
||
"缺少 technical.util.resample_to_interval 模块，聚合时间周期生成已跳过",
|
||
extra={"timeframe": timeframe},
|
||
)
|
||
RESAMPLE_WARNING_EMITTED = True
|
||
|
||
tf_ms = tf_to_ms(timeframe)
|
||
state_key = (symbol, timeframe)
|
||
last_ts = get_last_timestamp(DATA_DIR, symbol, timeframe)
|
||
fetch_states[state_key] = FetchState(symbol=symbol, timeframe=timeframe, last_candle_ts=last_ts)
|
||
queue = verification_queues.get(state_key)
|
||
if queue is None:
|
||
queue = asyncio.Queue(maxsize=500)
|
||
verification_queues[state_key] = queue
|
||
if last_ts is not None and last_ts > 0:
|
||
initial_count = 10 if timeframe == "1m" else 1
|
||
try:
|
||
queue.put_nowait(VerificationJob(timestamp=last_ts, count=initial_count))
|
||
except asyncio.QueueFull:
|
||
logger.warning(
|
||
"重启后无法排入校验任务，队列已满",
|
||
extra={"symbol": symbol, "timeframe": timeframe, "timestamp": last_ts, "count": initial_count},
|
||
)
|
||
|
||
start_from = parse_start_from_ms(START_FROM)
|
||
backoff = 1.0
|
||
|
||
try:
|
||
while True:
|
||
state = fetch_states[state_key]
|
||
state.started_at = datetime.utcnow()
|
||
if state.last_candle_ts is not None:
|
||
rewind_since = max(0, state.last_candle_ts - tf_ms)
|
||
initial_since = max(start_from, rewind_since)
|
||
else:
|
||
initial_since = start_from
|
||
|
||
logger.info(
|
||
"启动拉取任务 [%s %s] since=%s",
|
||
symbol,
|
||
timeframe,
|
||
initial_since,
|
||
)
|
||
|
||
try:
|
||
await rest_catchup(symbol, timeframe, derived_timeframes, tf_ms, initial_since)
|
||
if WS_ENABLED:
|
||
await stream_loop(symbol, timeframe, derived_timeframes, tf_ms)
|
||
else:
|
||
await rest_poll_loop(symbol, timeframe, derived_timeframes, tf_ms)
|
||
except asyncio.CancelledError:
|
||
logger.info("取消拉取任务 [%s %s]", symbol, timeframe)
|
||
state.last_error = "cancelled"
|
||
raise
|
||
except Exception as exc:
|
||
state.last_error = str(exc)
|
||
state.consecutive_errors += 1
|
||
logger.exception(
|
||
"拉取任务异常 [%s %s]，%.1f 秒后重启",
|
||
symbol,
|
||
timeframe,
|
||
backoff,
|
||
)
|
||
await asyncio.sleep(backoff)
|
||
backoff = min(backoff * BACKOFF_BASE, BACKOFF_MAX)
|
||
continue
|
||
else:
|
||
backoff = 1.0
|
||
logger.warning("拉取循环提前结束 [%s %s]，1 秒后重启", symbol, timeframe)
|
||
await asyncio.sleep(1.0)
|
||
finally:
|
||
logger.info("拉取任务退出", extra={"symbol": symbol, "timeframe": timeframe})
|
||
|
||
|
||
async def verification_worker(symbol: str, timeframe: str):
|
||
state_key = (symbol, timeframe)
|
||
queue = verification_queues.get(state_key)
|
||
if queue is None:
|
||
return
|
||
interval_ms = tf_to_ms(timeframe)
|
||
exchange = build_exchange()
|
||
try:
|
||
while True:
|
||
job = await queue.get()
|
||
try:
|
||
state = fetch_states.get(state_key)
|
||
if state and state.last_verified_ts is not None and job.timestamp <= state.last_verified_ts:
|
||
logger.debug(
|
||
"跳过校验任务，时间戳已验证",
|
||
extra={"symbol": symbol, "timeframe": timeframe, "timestamp": job.timestamp},
|
||
)
|
||
continue
|
||
verify_count = max(1, job.count)
|
||
if interval_ms <= 0:
|
||
interval_ms = tf_to_ms(timeframe)
|
||
start_ts = job.timestamp - (verify_count - 1) * interval_ms
|
||
if start_ts < 0:
|
||
start_ts = 0
|
||
limit = max(verify_count + 2, 2)
|
||
async with VERIFY_FETCH_SEMAPHORE:
|
||
fetched = await exchange.fetch_ohlcv(symbol, timeframe, since=start_ts, limit=limit)
|
||
normalized, _ = normalize_candles_for_timeframe(fetched, interval_ms)
|
||
if not normalized:
|
||
logger.warning(
|
||
"验证未获取到任何数据",
|
||
extra={
|
||
"symbol": symbol,
|
||
"timeframe": timeframe,
|
||
"timestamp": job.timestamp,
|
||
"count": verify_count,
|
||
},
|
||
)
|
||
continue
|
||
logger.info(
|
||
"开始校验 K 线",
|
||
extra={
|
||
"symbol": symbol,
|
||
"timeframe": timeframe,
|
||
"timestamp": job.timestamp,
|
||
"count": verify_count,
|
||
},
|
||
)
|
||
remote_map = {int(row[0]): row for row in normalized}
|
||
target_ts_list = [start_ts + i * interval_ms for i in range(verify_count)]
|
||
local_df = read_candles(DATA_DIR, symbol, timeframe, start_ts, job.timestamp)
|
||
updated = False
|
||
for ts in target_ts_list:
|
||
candidate = remote_map.get(ts)
|
||
if candidate is None:
|
||
logger.warning(
|
||
"验证缺失远端数据",
|
||
extra={"symbol": symbol, "timeframe": timeframe, "timestamp": ts},
|
||
)
|
||
continue
|
||
stored_rows = local_df[local_df["timestamp"] == ts]
|
||
needs_upsert = stored_rows.empty
|
||
reason = "missing"
|
||
if not needs_upsert:
|
||
stored_row = stored_rows.iloc[0]
|
||
diffs = (
|
||
abs(float(stored_row["open"]) - float(candidate[1])),
|
||
abs(float(stored_row["high"]) - float(candidate[2])),
|
||
abs(float(stored_row["low"]) - float(candidate[3])),
|
||
abs(float(stored_row["close"]) - float(candidate[4])),
|
||
abs(float(stored_row["volume"]) - float(candidate[5])),
|
||
)
|
||
if any(diff > 1e-9 for diff in diffs):
|
||
needs_upsert = True
|
||
reason = "mismatch"
|
||
if needs_upsert:
|
||
upsert_candles(DATA_DIR, symbol, timeframe, [candidate])
|
||
updated = True
|
||
logger.info(
|
||
"验证回补完成",
|
||
extra={"symbol": symbol, "timeframe": timeframe, "timestamp": ts, "reason": reason},
|
||
)
|
||
state = fetch_states.get(state_key)
|
||
if state:
|
||
if state.last_verified_ts is None or job.timestamp > state.last_verified_ts:
|
||
state.last_verified_ts = job.timestamp
|
||
if updated:
|
||
state.last_error = None
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"验证请求失败",
|
||
extra={
|
||
"symbol": symbol,
|
||
"timeframe": timeframe,
|
||
"timestamp": job.timestamp,
|
||
"count": job.count,
|
||
"error": str(exc),
|
||
},
|
||
)
|
||
await asyncio.sleep(1.0)
|
||
finally:
|
||
queue.task_done()
|
||
except asyncio.CancelledError:
|
||
raise
|
||
finally:
|
||
with suppress(Exception):
|
||
await exchange.close()
|
||
logger.info("验证任务退出", extra={"symbol": symbol, "timeframe": timeframe})
|
||
|
||
|
||
@app.on_event("startup")
|
||
async def on_start():
|
||
ensure_storage(DATA_DIR)
|
||
fetch_tasks.clear()
|
||
verification_queues.clear()
|
||
for s in SYMBOLS:
|
||
for tf in FETCH_TIMEFRAMES:
|
||
state_key = (s, tf)
|
||
verification_queues[state_key] = asyncio.Queue(maxsize=500)
|
||
task = asyncio.create_task(fetch_loop(s, tf), name=f"fetch::{s}::{tf}")
|
||
fetch_tasks.append(task)
|
||
verify_task = asyncio.create_task(verification_worker(s, tf), name=f"verify::{s}::{tf}")
|
||
fetch_tasks.append(verify_task)
|
||
for tf in DERIVED_TIMEFRAMES:
|
||
state_key = (s, tf)
|
||
if state_key not in verification_queues:
|
||
verification_queues[state_key] = asyncio.Queue(maxsize=500)
|
||
verify_task = asyncio.create_task(verification_worker(s, tf), name=f"verify::{s}::{tf}")
|
||
fetch_tasks.append(verify_task)
|
||
|
||
|
||
@app.on_event("shutdown")
|
||
async def on_shutdown():
|
||
if not fetch_tasks:
|
||
return
|
||
logger.info("正在停止拉取任务")
|
||
tasks = list(fetch_tasks)
|
||
for task in tasks:
|
||
task.cancel()
|
||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||
for result in results:
|
||
if isinstance(result, Exception) and not isinstance(result, asyncio.CancelledError):
|
||
logger.warning("任务停止时出现异常：%s", result)
|
||
fetch_tasks.clear()
|
||
verification_queues.clear()
|
||
|
||
|
||
@app.get("/health")
|
||
async def health():
|
||
now = datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
|
||
return {
|
||
"status": "ok",
|
||
"time": now,
|
||
"exchange": EXCHANGE,
|
||
"symbols": SYMBOLS,
|
||
"base_timeframes": FETCH_TIMEFRAMES,
|
||
"derived_timeframes": DERIVED_TIMEFRAMES,
|
||
"timeframes": AVAILABLE_TIMEFRAMES,
|
||
"tasks": [state.to_payload() for state in fetch_states.values()],
|
||
}
|
||
|
||
|
||
@app.get("/api/candles")
|
||
def api_candles(
|
||
symbol: str = Query(..., description="如 BTC/USDT:USDT"),
|
||
tf: str = Query("1m", description="时间周期"),
|
||
start: Optional[int] = Query(None, description="开始时间戳(ms)"),
|
||
end: Optional[int] = Query(None, description="结束时间戳(ms)"),
|
||
):
|
||
try:
|
||
ensure_symbol_timeframe(symbol, tf)
|
||
df = read_candles(DATA_DIR, symbol, tf, start, end)
|
||
records = df.to_dict("records") if not df.empty else []
|
||
return JSONResponse(records)
|
||
except Exception as e:
|
||
return JSONResponse({"error": str(e)}, status_code=500)
|
||
|
||
|
||
@app.websocket("/ws")
|
||
async def ws_endpoint(websocket: WebSocket, symbol: str, tf: str, since: Optional[int] = None):
|
||
if symbol not in VALID_SYMBOLS or tf not in VALID_TIMEFRAMES:
|
||
await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="invalid symbol/timeframe")
|
||
return
|
||
await hub.subscribe(websocket, symbol, tf)
|
||
try:
|
||
snap = read_candles(DATA_DIR, symbol, tf, since, None)
|
||
await websocket.send_text(
|
||
json.dumps(
|
||
{
|
||
"topic": f"candles.{symbol}.{tf}",
|
||
"type": "snapshot",
|
||
"data": [
|
||
{"t": int(r["timestamp"]), "o": r["open"], "h": r["high"], "l": r["low"], "c": r["close"], "v": r["volume"]}
|
||
for _, r in snap.iterrows()
|
||
],
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
while True:
|
||
await asyncio.sleep(30)
|
||
await websocket.send_text(json.dumps({"type": "ping", "ts": int(datetime.utcnow().timestamp() * 1000)}))
|
||
except WebSocketDisconnect:
|
||
return
|
||
|
||
|
||
@app.get("/")
|
||
def root():
|
||
return {
|
||
"service": "Local Data Service",
|
||
"exchange": EXCHANGE,
|
||
"symbols": SYMBOLS,
|
||
"base_timeframes": FETCH_TIMEFRAMES,
|
||
"derived_timeframes": DERIVED_TIMEFRAMES,
|
||
"timeframes": AVAILABLE_TIMEFRAMES,
|
||
}
|
||
|
||
|