diff --git a/TF_DF.py b/TF_DF.py index e99a62a..1dc09e6 100644 --- a/TF_DF.py +++ b/TF_DF.py @@ -123,12 +123,12 @@ class TF_DF(): return klu_state_list def check_fx(self, klc): if klc.pre and klc.next: - if klc.high > klc.pre.high and klc.high > klc.next.high and klc.low > klc.pre.low and klc.low > klc.next.low and klc.close > klc.next.close: + if klc.high > klc.pre.high and klc.high > klc.next.high and klc.low > klc.pre.low and klc.low > klc.next.low: #if (klc.close > klc.ema52 or klc.next.close > klc.next.ema52) and klc.macd > 0 and klc.macd > klc.macdhist: klc.set_fx(Chan_FX_TYPE.TOP) #print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "TOP") return Chan_FX_TYPE.TOP - elif klc.low < klc.pre.low and klc.low < klc.next.low and klc.high < klc.pre.high and klc.high < klc.next.high and klc.close < klc.next.close: + elif klc.low < klc.pre.low and klc.low < klc.next.low and klc.high < klc.pre.high and klc.high < klc.next.high: #if (klc.close < klc.ema52 or klc.next.close < klc.next.ema52) and klc.macd < 0 and klc.macd < klc.macdhist: klc.set_fx(Chan_FX_TYPE.BOTTOM) #print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "BOTTOM") diff --git a/datasvc/Dockerfile b/datasvc/Dockerfile index 6c10c63..7307616 100644 --- a/datasvc/Dockerfile +++ b/datasvc/Dockerfile @@ -14,7 +14,7 @@ COPY pairs.json /app/pairs.json ENV DATA_DIR=/data \ EXCHANGE=binance \ TIMEFRAMES=1m,1h,1d,1w,1M \ - START_FROM=2022-01-01 \ + START_FROM=2025-09-01 \ POLL_FACTOR=0.5 VOLUME ["/data"] diff --git a/datasvc/README.md b/datasvc/README.md index 7656ec9..0625074 100644 --- a/datasvc/README.md +++ b/datasvc/README.md @@ -22,6 +22,9 @@ | `POLL_FACTOR` | 拉取间隔因子，实际间隔 = 周期毫秒 × factor | `0.5` | | `REST_MAX_CONCURRENCY` | REST 历史拉取并发数 | `4` | | `VERIFY_MAX_CONCURRENCY` | 校验请求并发数 | `2` | +| `WS_ENABLED` | 是否启用 Binance WebSocket 增量（`true`/`false`） | `false` | +| `REST_POLL_INTERVAL` | 实时轮询 REST 的间隔秒数 | `5` | +| `REST_POLL_WINDOW` | 实时轮询时拉取的最新 K 线数量 | `10` | | `BACKOFF_BASE / BACKOFF_MAX` | 异常重试的指数退避参数 | `2.0 / 30.0` | > 衍生周期列表由程序自动推导，无需手动写入 `TIMEFRAMES`。 diff --git a/datasvc/app/main.py b/datasvc/app/main.py index 3626008..000d8b3 100644 --- a/datasvc/app/main.py +++ b/datasvc/app/main.py @@ -16,6 +16,9 @@ from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query, HTTPExceptio 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 - 环境缺失依赖时自动降级 @@ -26,7 +29,6 @@ from .storage import ( read_candles, upsert_candles, get_last_timestamp, - read_candle_exact, ) @@ -40,6 +42,9 @@ 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")) @@ -154,7 +159,7 @@ for base_tf in FETCH_TIMEFRAMES: 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", "2022-01-01") # 首次启动拉取起始日期（UTC） +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")) @@ -165,6 +170,35 @@ 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, @@ -210,7 +244,7 @@ def parse_start_from_ms(val: str) -> int: dt = datetime.fromisoformat(val) # 允许 '2022-01-01' 或 '2022-01-01T00:00:00' except Exception: # 回退到固定日期 - dt = datetime(2022, 1, 1) + dt = datetime(2025, 9, 1) return int(dt.timestamp() * 1000) @@ -249,7 +283,7 @@ class Hub: hub = Hub() fetch_tasks: List[asyncio.Task] = [] -verification_queues: Dict[Tuple[str, str], asyncio.Queue[int]] = {} +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]]]: @@ -261,7 +295,12 @@ def resample_and_store(symbol: str, base_timeframe: str, derived_timeframes: Lis 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.sort_values("timestamp") + 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: @@ -327,6 +366,150 @@ def resample_and_store(symbol: str, base_timeframe: str, derived_timeframes: Lis 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 @@ -356,30 +539,140 @@ class FetchState: } +@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, - schedule_verification: bool, + 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]]] = [] - if derived_timeframes and RESAMPLE_AVAILABLE: - derived_updates = await asyncio.to_thread( - resample_and_store, + 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, ) - for row in candles[-3:]: + 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", @@ -390,15 +683,35 @@ async def process_candles( "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", @@ -409,30 +722,59 @@ async def process_candles( "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 - if schedule_verification: - queue = verification_queues.get(state_key) - if queue: - now_ms = int(datetime.utcnow().timestamp() * 1000) - latest_ts = candles[-1][0] - if now_ms - latest_ts <= 2 * tf_ms: - verify_ts = latest_ts - tf_ms - if verify_ts > 0 and (state.last_verified_ts is None or verify_ts > state.last_verified_ts): - try: - queue.put_nowait(verify_ts) - except asyncio.QueueFull: - logger.warning( - "验证队列已满，丢弃此次校验请求", - extra={"symbol": symbol, "timeframe": timeframe, "timestamp": verify_ts}, - ) + 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( @@ -447,6 +789,7 @@ async def rest_catchup( 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: @@ -480,11 +823,57 @@ async def rest_catchup( continue if not candles: break - await process_candles(symbol, timeframe, candles, derived_timeframes, tf_ms, schedule_verification=False) - since = candles[-1][0] + tf_ms + 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 @@ -498,6 +887,80 @@ async def rest_catchup( 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", + }, + ) + await process_candles( + symbol, + timeframe, + candles, + derived_timeframes, + tf_ms, + finalized=True, + allow_verification=True, + closed_flags=closed_flags, + ) + 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) @@ -508,8 +971,9 @@ async def stream_loop(symbol: str, timeframe: str, derived_timeframes: List[str] async for message in ws: data = json.loads(message) kline = data.get("k") - if not kline or not kline.get("x"): + if not kline: continue + is_closed = bool(kline.get("x")) row: CandleRow = [ int(kline["t"]), float(kline["o"]), @@ -518,7 +982,16 @@ async def stream_loop(symbol: str, timeframe: str, derived_timeframes: List[str] float(kline["c"]), float(kline["v"]), ] - await process_candles(symbol, timeframe, [row], derived_timeframes, tf_ms, schedule_verification=True) + 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 @@ -573,13 +1046,14 @@ async def fetch_loop(symbol: str, timeframe: str): if queue is None: queue = asyncio.Queue(maxsize=500) verification_queues[state_key] = queue - if last_ts is not None: + if last_ts is not None and last_ts > 0: + initial_count = 10 if timeframe == "1m" else 1 try: - queue.put_nowait(last_ts) + queue.put_nowait(VerificationJob(timestamp=last_ts, count=initial_count)) except asyncio.QueueFull: logger.warning( "重启后无法排入校验任务，队列已满", - extra={"symbol": symbol, "timeframe": timeframe, "timestamp": last_ts}, + extra={"symbol": symbol, "timeframe": timeframe, "timestamp": last_ts, "count": initial_count}, ) start_since = parse_start_from_ms(START_FROM) @@ -593,7 +1067,10 @@ async def fetch_loop(symbol: str, timeframe: str): try: await rest_catchup(symbol, timeframe, derived_timeframes, tf_ms, initial_since) - await stream_loop(symbol, timeframe, derived_timeframes, tf_ms) + 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("取消拉取任务", extra={"symbol": symbol, "timeframe": timeframe}) state = fetch_states.get(state_key) @@ -609,59 +1086,105 @@ async def verification_worker(symbol: str, timeframe: str): queue = verification_queues.get(state_key) if queue is None: return + interval_ms = tf_to_ms(timeframe) exchange = build_exchange() try: while True: - verify_ts = await queue.get() + job = await queue.get() try: state = fetch_states.get(state_key) - if state and state.last_verified_ts is not None and verify_ts <= state.last_verified_ts: - queue.task_done() + 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: - verification = await exchange.fetch_ohlcv(symbol, timeframe, since=verify_ts, limit=2) - target_rows = [row for row in verification if row and row[0] == verify_ts] - if not target_rows: + 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": verify_ts}, + "验证未获取到任何数据", + extra={ + "symbol": symbol, + "timeframe": timeframe, + "timestamp": job.timestamp, + "count": verify_count, + }, ) - queue.task_done() continue - candidate = target_rows[-1] - stored = read_candle_exact(DATA_DIR, symbol, timeframe, verify_ts) - needs_upsert = stored.empty - reason = "missing" - if not needs_upsert: - stored_row = stored.iloc[0] - open_diff = abs(float(stored_row["open"]) - float(candidate[1])) - high_diff = abs(float(stored_row["high"]) - float(candidate[2])) - low_diff = abs(float(stored_row["low"]) - float(candidate[3])) - close_diff = abs(float(stored_row["close"]) - float(candidate[4])) - volume_diff = abs(float(stored_row["volume"]) - float(candidate[5])) - if any(diff > 1e-9 for diff in (open_diff, high_diff, low_diff, close_diff, volume_diff)): - needs_upsert = True - reason = "mismatch" - if needs_upsert: - upsert_candles(DATA_DIR, symbol, timeframe, [candidate]) - logger.info( - "验证回补完成", - extra={"symbol": symbol, "timeframe": timeframe, "timestamp": verify_ts, "reason": reason}, - ) + 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: - state.last_verified_ts = verify_ts - queue.task_done() + 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: - queue.task_done() raise except Exception as exc: logger.warning( "验证请求失败", - extra={"symbol": symbol, "timeframe": timeframe, "timestamp": verify_ts, "error": str(exc)}, + extra={ + "symbol": symbol, + "timeframe": timeframe, + "timestamp": job.timestamp, + "count": job.count, + "error": str(exc), + }, ) - queue.task_done() await asyncio.sleep(1.0) + finally: + queue.task_done() except asyncio.CancelledError: raise finally: @@ -683,6 +1206,12 @@ async def on_start(): 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") diff --git a/datasvc/docker-compose.yml b/datasvc/docker-compose.yml index ed58b2b..31cf276 100644 --- a/datasvc/docker-compose.yml +++ b/datasvc/docker-compose.yml @@ -6,10 +6,11 @@ services: environment: - EXCHANGE=binance - TIMEFRAMES=1m,1h,1d,1w,1M - - START_FROM=2022-01-01 + - START_FROM=2025-09-01 - POLL_FACTOR=0.5 - DATA_DIR=/data - TZ=Asia/Shanghai + - VERIFY_INTERVALS=1m=5,5m=2,10m=2 ports: - "9000:9000" volumes: diff --git a/datasvc/pairs.json b/datasvc/pairs.json index ebb2365..ffe4f03 100644 --- a/datasvc/pairs.json +++ b/datasvc/pairs.json @@ -1,22 +1,5 @@ [ "BTC/USDT:USDT", "ETH/USDT:USDT", - "SOL/USDT:USDT", - "WIF/USDT:USDT", - "1000PEPE/USDT:USDT", - "DOGS/USDT:USDT", - "ORDI/USDT:USDT", - "AAVE/USDT:USDT", - "REEF/USDT:USDT", - "1000SATS/USDT:USDT", - "SUI/USDT:USDT", - "1INCH/USDT:USDT", - "DOGE/USDT:USDT", - "TON/USDT:USDT", - "UNI/USDT:USDT", - "XRP/USDT:USDT", - "SUN/USDT:USDT", - "NOT/USDT:USDT", - "RARE/USDT:USDT", - "RDNT/USDT:USDT" + "SOL/USDT:USDT" ]