现在可以正常工作了

datasvc/app/main.py

@@ -5,8 +5,11 @@ import logging
 from contextlib import suppress
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta
+from time import time
 from pathlib import Path
-from typing import Dict, List, Optional, Tuple, Union
+from collections import defaultdict
+from threading import Event, RLock
+from typing import Dict, List, Optional, Set, Tuple, Union
 
 import ccxt
 import ccxt.async_support as ccxt_async
@@ -24,12 +27,7 @@ try:
 except ImportError:  # pragma: no cover - 环境缺失依赖时自动降级
     resample_to_interval = None  # type: ignore
 
-from .storage import (
-    ensure_storage,
-    read_candles,
-    upsert_candles,
-    get_last_timestamp,
-)
+from .storage import candle_path, ensure_storage, read_candles, write_candles_snapshot
 
 
 LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()
@@ -46,10 +44,8 @@ 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_MAX_CONCURRENCY = int(os.environ.get("REST_MAX_CONCURRENCY", "1"))
 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"],
@@ -62,6 +58,300 @@ AGGREGATION_PLAN: Dict[str, List[str]] = {
 CandleRow = List[Union[int, float]]
 
 
+CANDLE_COLUMNS = ["timestamp", "open", "high", "low", "close", "volume"]
+CANDLES_CACHE: Dict[Tuple[str, str], pd.DataFrame] = {}
+CACHE_LOCK = RLock()
+BASE_TIMEFRAMES: Set[str] = set()
+BASE_DIRTY_VERSION: Dict[Tuple[str, str], int] = {}
+BASE_FLUSH_INTERVAL_SECONDS = 600
+CACHE_FILE_MTIME: Dict[Tuple[str, str], float] = {}
+IS_ENGINE_PROCESS = os.environ.get("DATASVC_ENGINE") == "1"
+
+
+def _empty_frame() -> pd.DataFrame:
+    return pd.DataFrame(columns=CANDLE_COLUMNS)
+
+
+def _normalize_dataframe(df: pd.DataFrame) -> pd.DataFrame:
+    if df.empty:
+        return _empty_frame()
+    normalized = df.copy()
+    missing_columns = [col for col in CANDLE_COLUMNS if col not in normalized.columns]
+    for column in missing_columns:
+        normalized[column] = 0.0 if column != "timestamp" else 0
+    normalized = normalized[CANDLE_COLUMNS]
+    normalized["timestamp"] = normalized["timestamp"].astype("int64")
+    for column in CANDLE_COLUMNS[1:]:
+        normalized[column] = normalized[column].astype("float64")
+    normalized = normalized.drop_duplicates(subset=["timestamp"], keep="last").sort_values("timestamp").reset_index(drop=True)
+    return normalized
+
+
+def preload_candles_cache(symbols: List[str], timeframes: List[str]) -> None:
+    new_cache: Dict[Tuple[str, str], pd.DataFrame] = {}
+    new_mtime: Dict[Tuple[str, str], float] = {}
+    for symbol in symbols:
+        for timeframe in timeframes:
+            df = read_candles(DATA_DIR, symbol, timeframe, None, None)
+            normalized = _normalize_dataframe(df)
+            new_cache[(symbol, timeframe)] = normalized
+            try:
+                mtime = os.path.getmtime(candle_path(DATA_DIR, symbol, timeframe))
+            except OSError:
+                mtime = 0.0
+            new_mtime[(symbol, timeframe)] = mtime
+    with CACHE_LOCK:
+        CANDLES_CACHE.clear()
+        CANDLES_CACHE.update(new_cache)
+        BASE_DIRTY_VERSION.clear()
+        CACHE_FILE_MTIME.clear()
+        for key in new_cache:
+            if key[1] in BASE_TIMEFRAMES:
+                BASE_DIRTY_VERSION[key] = 0
+            CACHE_FILE_MTIME[key] = new_mtime.get(key, 0.0)
+
+
+def refresh_cache_from_disk(symbol: str, timeframe: str) -> None:
+    if timeframe not in BASE_TIMEFRAMES:
+        return
+    if IS_ENGINE_PROCESS:
+        return
+    key = (symbol, timeframe)
+    path = candle_path(DATA_DIR, symbol, timeframe)
+    try:
+        mtime = os.path.getmtime(path)
+    except FileNotFoundError:
+        with CACHE_LOCK:
+            if key not in CANDLES_CACHE:
+                CANDLES_CACHE[key] = _empty_frame()
+            CACHE_FILE_MTIME[key] = 0.0
+        return
+    except OSError:
+        return
+    with CACHE_LOCK:
+        cached_mtime = CACHE_FILE_MTIME.get(key, 0.0)
+    if mtime <= cached_mtime:
+        return
+    df = read_candles(DATA_DIR, symbol, timeframe, None, None)
+    normalized = _normalize_dataframe(df)
+    with CACHE_LOCK:
+        CANDLES_CACHE[key] = normalized
+        CACHE_FILE_MTIME[key] = mtime
+        if timeframe in BASE_TIMEFRAMES:
+            BASE_DIRTY_VERSION[key] = 0
+
+
+def update_cache_mtime(symbol: str, timeframe: str) -> None:
+    path = candle_path(DATA_DIR, symbol, timeframe)
+    try:
+        mtime = os.path.getmtime(path)
+    except OSError:
+        mtime = time()
+    with CACHE_LOCK:
+        CACHE_FILE_MTIME[(symbol, timeframe)] = mtime
+
+
+def rebuild_all_derived_timeframes(symbols: List[str]) -> None:
+    if not RESAMPLE_AVAILABLE:
+        return
+    for symbol in symbols:
+        for base_tf, targets in AGGREGATION_TARGETS.items():
+            if not targets:
+                continue
+            resample_and_store(symbol, base_tf, targets)
+
+
+def cache_get(symbol: str, timeframe: str, start: Optional[int] = None, end: Optional[int] = None) -> pd.DataFrame:
+    refresh_cache_from_disk(symbol, timeframe)
+    key = (symbol, timeframe)
+    with CACHE_LOCK:
+        df = CANDLES_CACHE.get(key)
+        if df is None:
+            df = _empty_frame()
+    result = df
+    if start is not None:
+        result = result[result["timestamp"] >= int(start)]
+    if end is not None:
+        result = result[result["timestamp"] <= int(end)]
+    return result.copy()
+
+
+def cache_get_last_timestamp(symbol: str, timeframe: str) -> Optional[int]:
+    if timeframe in BASE_TIMEFRAMES:
+        refresh_cache_from_disk(symbol, timeframe)
+    key = (symbol, timeframe)
+    with CACHE_LOCK:
+        df = CANDLES_CACHE.get(key)
+        if df is None:
+            CANDLES_CACHE[key] = _empty_frame()
+            return None
+        if df.empty:
+            return None
+        return int(df["timestamp"].iloc[-1])
+
+
+def cache_update(symbol: str, timeframe: str, candles: List[CandleRow]) -> Optional[pd.DataFrame]:
+    if not candles:
+        return None
+    new_df = _normalize_dataframe(pd.DataFrame(candles, columns=CANDLE_COLUMNS))
+    if new_df.empty:
+        return None
+    key = (symbol, timeframe)
+    with CACHE_LOCK:
+        existing = CANDLES_CACHE.get(key)
+    if existing is None or existing.empty:
+        merged = new_df
+    else:
+        merged = pd.concat([existing, new_df], ignore_index=True)
+        merged = _normalize_dataframe(merged)
+    with CACHE_LOCK:
+        CANDLES_CACHE[key] = merged
+        if timeframe in BASE_TIMEFRAMES:
+            BASE_DIRTY_VERSION[key] = BASE_DIRTY_VERSION.get(key, 0) + 1
+        snapshot = merged.copy()
+    return snapshot
+
+
+def collect_engine_status() -> dict:
+    updated_at = datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
+    tasks = [state.to_payload() for state in fetch_states.values()]
+    return {
+        "updated_at": updated_at,
+        "tasks": tasks,
+        "queues": {},
+    }
+
+
+def write_engine_status_snapshot() -> None:
+    try:
+        ENGINE_STATUS_PATH.parent.mkdir(parents=True, exist_ok=True)
+        snapshot = collect_engine_status()
+        ENGINE_STATUS_PATH.write_text(json.dumps(snapshot, ensure_ascii=False), encoding="utf-8")
+    except Exception:
+        logger.warning("写入引擎状态快照失败", exc_info=True)
+
+
+async def status_flush_worker(stop_event: Event, interval: float = 5.0) -> None:
+    await asyncio.to_thread(write_engine_status_snapshot)
+    try:
+        while not stop_event.is_set():
+            await asyncio.sleep(interval)
+            await asyncio.to_thread(write_engine_status_snapshot)
+    except asyncio.CancelledError:
+        raise
+
+
+def load_engine_status_snapshot() -> Optional[dict]:
+    try:
+        content = ENGINE_STATUS_PATH.read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return None
+    except Exception:
+        logger.warning("读取引擎状态快照失败", exc_info=True)
+        return None
+    try:
+        return json.loads(content)
+    except json.JSONDecodeError:
+        logger.warning("解析引擎状态快照失败")
+        return None
+
+
+async def flush_dirty_base_snapshots(force_all: bool = False) -> None:
+    with CACHE_LOCK:
+        if force_all:
+            target_entries = []
+            for key in CANDLES_CACHE.keys():
+                symbol, timeframe = key
+                if timeframe in BASE_TIMEFRAMES:
+                    version = BASE_DIRTY_VERSION.get(key, 0)
+                    target_entries.append((key, version))
+        else:
+            target_entries = [(key, version) for key, version in BASE_DIRTY_VERSION.items() if version > 0]
+        snapshots = {key: CANDLES_CACHE.get(key, _empty_frame()).copy() for key, _ in target_entries}
+    if not snapshots:
+        return
+    failed: Set[Tuple[str, str]] = set()
+    for key, snapshot in snapshots.items():
+        symbol, timeframe = key
+        try:
+            await asyncio.to_thread(write_candles_snapshot, DATA_DIR, symbol, timeframe, snapshot)
+        except Exception:
+            failed.add(key)
+            logger.exception(
+                "基础周期快照写入失败",
+                extra={"symbol": symbol, "timeframe": timeframe},
+            )
+        else:
+            update_cache_mtime(symbol, timeframe)
+    if not failed:
+        logger.debug(
+            "基础周期快照写入完成",
+            extra={"count": len(snapshots), "force_all": force_all},
+        )
+    with CACHE_LOCK:
+        for key, version in target_entries:
+            if key in failed:
+                continue
+            current_version = BASE_DIRTY_VERSION.get(key, 0)
+            if current_version == version:
+                BASE_DIRTY_VERSION[key] = 0
+
+
+async def base_flush_worker():
+    try:
+        logger.info(
+            "基础周期定时写盘任务已启动",
+            extra={"interval_seconds": BASE_FLUSH_INTERVAL_SECONDS},
+        )
+        while True:
+            await asyncio.sleep(BASE_FLUSH_INTERVAL_SECONDS)
+            await flush_dirty_base_snapshots()
+    except asyncio.CancelledError:
+        raise
+    finally:
+        with suppress(Exception):
+            await flush_dirty_base_snapshots(force_all=True)
+
+
+async def run_engine(stop_event: Optional[Event] = None):
+    if stop_event is None:
+        stop_event = Event()
+    logger.info("数据引擎启动")
+    fetch_tasks.clear()
+    try:
+        await asyncio.to_thread(rebuild_all_derived_timeframes, SYMBOLS)
+    except Exception:
+        logger.exception("初始化衍生周期失败，继续启动引擎")
+    try:
+        status_task = asyncio.create_task(status_flush_worker(stop_event), name="status::flush")
+        fetch_tasks.append(status_task)
+        flush_task = asyncio.create_task(base_flush_worker(), name="flush::base")
+        fetch_tasks.append(flush_task)
+        for s in SYMBOLS:
+            for tf in FETCH_TIMEFRAMES:
+                fetch_task = asyncio.create_task(fetch_loop(s, tf), name=f"fetch::{s}::{tf}")
+                fetch_tasks.append(fetch_task)
+        while not stop_event.is_set():
+            await asyncio.sleep(1.0)
+    finally:
+        stop_event.set()
+        if fetch_tasks:
+            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()
+        with suppress(Exception):
+            await flush_dirty_base_snapshots(force_all=True)
+        with suppress(Exception):
+            await asyncio.to_thread(write_engine_status_snapshot)
+    logger.info("数据引擎已停止")
+
+
 def _split_env_list(value: str) -> List[str]:
     return [item.strip() for item in value.split(",") if item.strip()]
 
@@ -158,6 +448,7 @@ for base_tf in FETCH_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}
+BASE_TIMEFRAMES = set(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
@@ -165,40 +456,18 @@ 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("/")
 
+BASE_FLUSH_INTERVAL_MINUTES = max(1, int(os.environ.get("BASE_FLUSH_INTERVAL_MINUTES", "10")))
+BASE_FLUSH_INTERVAL_SECONDS = BASE_FLUSH_INTERVAL_MINUTES * 60
+
+ENGINE_STATUS_PATH = Path(DATA_DIR) / "engine_status.json"
+
 VALID_SYMBOLS = set(SYMBOLS)
 VALID_TIMEFRAMES = set(AVAILABLE_TIMEFRAMES)
 
 ensure_storage(DATA_DIR)
+preload_candles_cache(SYMBOLS, AVAILABLE_TIMEFRAMES)
 
 
-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,
@@ -283,24 +552,16 @@ class Hub:
 hub = Hub()
 
 fetch_tasks: List[asyncio.Task] = []
-verification_queues: Dict[Tuple[str, str], asyncio.Queue["VerificationJob"]] = {}
+
+engine_runner_stop: Optional[Event] = None
+engine_runner_task: Optional[asyncio.Task] = None
 
 
 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:
+    if 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")
+
+    base_tf_ms = tf_to_ms(base_timeframe)
 
     updates: List[Tuple[str, List[List[float]]]] = []
     for target_tf in derived_timeframes:
@@ -308,8 +569,31 @@ def resample_and_store(symbol: str, base_timeframe: str, derived_timeframes: Lis
         if minutes is None:
             logger.warning("无法解析聚合周期", extra={"target_timeframe": target_tf})
             continue
+
+        last_ts = cache_get_last_timestamp(symbol, target_tf)
+        start_ts: Optional[int] = None
+        if last_ts is not None and base_tf_ms is not None and base_tf_ms > 0:
+            buffer_ms = minutes * 60_000 + base_tf_ms
+            start_ts = max(0, int(last_ts) - buffer_ms)
+
+        base_slice = cache_get(symbol, base_timeframe, start_ts, None)
+        if base_slice.empty:
+            continue
+        base_slice = base_slice.copy()
+        if "date" not in base_slice.columns:
+            base_slice["date"] = pd.to_datetime(base_slice["timestamp"], unit="ms", utc=True)
+        base_slice = (
+            base_slice.drop_duplicates(subset=["timestamp"], keep="last")
+            .sort_values("timestamp")
+            .reset_index(drop=True)
+        )
+        base_slice["timestamp"] = base_slice["timestamp"].astype("int64")
+
         try:
-            derived_df = resample_to_interval(base_df, minutes)  # type: ignore[misc]
+            if RESAMPLE_AVAILABLE:
+                derived_df = resample_to_interval(base_slice, minutes)  # type: ignore[misc]
+            else:
+                derived_df = _fallback_resample_to_interval(base_slice, minutes)
         except Exception:
             logger.exception(
                 "聚合周期计算失败",
@@ -341,7 +625,6 @@ def resample_and_store(symbol: str, base_timeframe: str, derived_timeframes: Lis
             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:
@@ -361,7 +644,7 @@ def resample_and_store(symbol: str, base_timeframe: str, derived_timeframes: Lis
             )
         if not records:
             continue
-        upsert_candles(DATA_DIR, symbol, target_tf, records)
+        cache_update(symbol, target_tf, records)
         updates.append((target_tf, records[-3:] if len(records) > 3 else records))
     return updates
 
@@ -412,6 +695,8 @@ def compute_live_derived_updates(
     if not candles or not derived_timeframes or base_tf_ms <= 0:
         return updates
 
+    pending_updates: Dict[str, List[CandleRow]] = defaultdict(list)
+
     derived_ms_map: Dict[str, int] = {}
     max_multiplier = 1
     for target_tf in derived_timeframes:
@@ -431,7 +716,7 @@ def compute_live_derived_updates(
     if base_start < 0:
         base_start = 0
 
-    base_df = read_candles(DATA_DIR, symbol, base_timeframe, base_start, newest_ts)
+    base_df = cache_get(symbol, base_timeframe, base_start, newest_ts)
     if base_df.empty:
         return updates
     base_df = base_df.sort_values("timestamp")
@@ -502,11 +787,13 @@ def compute_live_derived_updates(
                 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])
+            pending_updates[target_tf].append(row)
             rows_with_status.append((row, closed))
             current_start += derived_ms
         if rows_with_status:
             updates[target_tf] = rows_with_status
+    for target_tf, rows in pending_updates.items():
+        cache_update(symbol, target_tf, rows)
     return updates
 
 
@@ -519,7 +806,6 @@ class FetchState:
     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]:
@@ -535,68 +821,12 @@ class FetchState:
             "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,
@@ -604,7 +834,6 @@ async def process_candles(
     derived_timeframes: List[str],
     tf_ms: int,
     finalized: bool,
-    allow_verification: bool,
     closed_flags: Optional[List[bool]] = None,
 ) -> None:
     if not candles:
@@ -612,7 +841,7 @@ async def process_candles(
     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)
+    cache_update(symbol, timeframe, candles)
     base_records = list(zip(candles, closed_flags))
     last_closed_ts: Optional[int] = None
     for row, is_closed in base_records:
@@ -633,10 +862,9 @@ async def process_candles(
     )
     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:
+        if needs_resample:
             derived_updates = await asyncio.to_thread(
                 resample_and_store,
                 symbol,
@@ -650,7 +878,7 @@ async def process_candles(
                         "symbol": symbol,
                         "base_timeframe": timeframe,
                         "targets": [item[0] for item in derived_updates],
-                        "origin": "resample",
+                        "origin": "resample" if RESAMPLE_AVAILABLE else "fallback",
                     },
                 )
         live_derived_updates = await asyncio.to_thread(
@@ -699,10 +927,6 @@ async def process_candles(
                 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:
@@ -732,10 +956,6 @@ async def process_candles(
                 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:
@@ -765,16 +985,7 @@ async def process_candles(
         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(
@@ -836,7 +1047,6 @@ async def rest_catchup(
                 derived_timeframes,
                 tf_ms,
                 finalized=True,
-                allow_verification=False,
                 closed_flags=[True] * len(candles),
             )
             state.consecutive_errors = 0
@@ -950,7 +1160,6 @@ async def rest_poll_loop(
                         derived_timeframes,
                         tf_ms,
                         finalized=True,
-                        allow_verification=True,
                         closed_flags=closed_flags,
                     )
                 except asyncio.CancelledError:
@@ -1004,7 +1213,6 @@ async def stream_loop(symbol: str, timeframe: str, derived_timeframes: List[str]
                         derived_timeframes,
                         tf_ms,
                         finalized=is_closed,
-                        allow_verification=is_closed,
                         closed_flags=[is_closed],
                     )
         except asyncio.CancelledError:
@@ -1055,21 +1263,9 @@ async def fetch_loop(symbol: str, timeframe: str):
 
     tf_ms = tf_to_ms(timeframe)
     state_key = (symbol, timeframe)
-    last_ts = get_last_timestamp(DATA_DIR, symbol, timeframe)
+    last_ts = cache_get_last_timestamp(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},
-            )
+    initial_sync_flushed = False
 
     start_from = parse_start_from_ms(START_FROM)
     backoff = 1.0
@@ -1093,6 +1289,13 @@ async def fetch_loop(symbol: str, timeframe: str):
 
             try:
                 await rest_catchup(symbol, timeframe, derived_timeframes, tf_ms, initial_since)
+                if not initial_sync_flushed:
+                    await flush_dirty_base_snapshots(force_all=True)
+                    logger.info(
+                        "初次同步完成，基础周期数据已写盘",
+                        extra={"symbol": symbol, "timeframe": timeframe},
+                    )
+                    initial_sync_flushed = True
                 if WS_ENABLED:
                     await stream_loop(symbol, timeframe, derived_timeframes, tf_ms)
                 else:
@@ -1121,158 +1324,34 @@ async def fetch_loop(symbol: str, timeframe: str):
         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)
+    logger.info("API 服务启动完成")
+    if IS_ENGINE_PROCESS:
+        global engine_runner_stop, engine_runner_task
+        if engine_runner_task is None or engine_runner_task.done():
+            engine_runner_stop = Event()
+            engine_runner_task = asyncio.create_task(run_engine(engine_runner_stop))
 
 
 @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()
+    logger.info("API 服务准备退出")
+    if IS_ENGINE_PROCESS:
+        global engine_runner_stop, engine_runner_task
+        if engine_runner_stop is not None:
+            engine_runner_stop.set()
+        if engine_runner_task is not None:
+            with suppress(Exception):
+                await engine_runner_task
+            engine_runner_task = None
+            engine_runner_stop = None
 
 
 @app.get("/health")
 async def health():
     now = datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
+    engine_status = load_engine_status_snapshot() or {"updated_at": None, "tasks": [], "queues": {}}
     return {
         "status": "ok",
         "time": now,
@@ -1281,7 +1360,8 @@ async def health():
         "base_timeframes": FETCH_TIMEFRAMES,
         "derived_timeframes": DERIVED_TIMEFRAMES,
         "timeframes": AVAILABLE_TIMEFRAMES,
-        "tasks": [state.to_payload() for state in fetch_states.values()],
+        "engine": engine_status,
+        "tasks": engine_status.get("tasks", []),
     }
 
 
@@ -1294,7 +1374,7 @@ def api_candles(
 ):
     try:
         ensure_symbol_timeframe(symbol, tf)
-        df = read_candles(DATA_DIR, symbol, tf, start, end)
+        df = cache_get(symbol, tf, start, end)
         records = df.to_dict("records") if not df.empty else []
         return JSONResponse(records)
     except Exception as e:
@@ -1308,7 +1388,7 @@ async def ws_endpoint(websocket: WebSocket, symbol: str, tf: str, since: Optiona
         return
     await hub.subscribe(websocket, symbol, tf)
     try:
-        snap = read_candles(DATA_DIR, symbol, tf, since, None)
+        snap = cache_get(symbol, tf, since, None)
         await websocket.send_text(
             json.dumps(
                 {
@@ -1344,3 +1424,34 @@ def root():
     }
 
 
+def _fallback_resample_to_interval(df: pd.DataFrame, minutes: int) -> pd.DataFrame:
+    if df.empty or minutes <= 0:
+        return pd.DataFrame(columns=CANDLE_COLUMNS)
+    working = df.copy()
+    if "timestamp" not in working.columns:
+        return pd.DataFrame(columns=CANDLE_COLUMNS)
+    working["date"] = pd.to_datetime(working["timestamp"], unit="ms", utc=True)
+    working = working.set_index("date", drop=True)
+    columns = ["open", "high", "low", "close", "volume"]
+    for column in columns:
+        if column not in working.columns:
+            working[column] = 0.0
+    working = working[columns]
+    rule = f"{minutes}T"
+    aggregated = working.resample(rule, label="left", closed="left").agg(
+        {
+            "open": "first",
+            "high": "max",
+            "low": "min",
+            "close": "last",
+            "volume": "sum",
+        }
+    )
+    aggregated = aggregated.dropna(subset=["open", "high", "low", "close"]).reset_index()
+    aggregated["timestamp"] = (aggregated["date"].astype("int64") // 1_000_000)
+    aggregated = aggregated.drop(columns=["date"], errors="ignore")
+    aggregated = aggregated.dropna(subset=["timestamp"]).reset_index(drop=True)
+    aggregated["timestamp"] = aggregated["timestamp"].astype("int64")
+    return aggregated[CANDLE_COLUMNS]
+
+

datasvc/app/storage.py

@@ -85,6 +85,31 @@ def upsert_candles(base_dir: str, symbol: str, timeframe: str, candles: List[Lis
                     pass
 
 
+def write_candles_snapshot(base_dir: str, symbol: str, timeframe: str, df: pd.DataFrame):
+    columns = ["timestamp", "open", "high", "low", "close", "volume"]
+    if df.empty:
+        safe_df = pd.DataFrame(columns=columns)
+    else:
+        safe_df = df[columns].copy()
+        safe_df = safe_df.drop_duplicates(subset=["timestamp"], keep="last").sort_values("timestamp").reset_index(drop=True)
+    p = _path(base_dir, symbol, timeframe)
+    with _lock:
+        temp_path = f"{p}.tmp"
+        try:
+            safe_df.to_parquet(temp_path, index=False)
+            os.replace(temp_path, p)
+        finally:
+            if os.path.exists(temp_path):
+                try:
+                    os.remove(temp_path)
+                except OSError:
+                    pass
+
+
+def candle_path(base_dir: str, symbol: str, timeframe: str) -> str:
+    return _path(base_dir, symbol, timeframe)
+
+
 def get_last_timestamp(base_dir: str, symbol: str, timeframe: str) -> Optional[int]:
     p = _path(base_dir, symbol, timeframe)
     if not os.path.exists(p):

datasvc/app/worker.py

@@ -0,0 +1,27 @@
+import asyncio
+import signal
+from threading import Event
+
+from .main import logger, run_engine
+
+
+async def _async_main():
+    stop_event = Event()
+    loop = asyncio.get_running_loop()
+    for sig in (signal.SIGINT, signal.SIGTERM):
+        try:
+            loop.add_signal_handler(sig, stop_event.set)
+        except NotImplementedError:
+            # 信号处理在某些平台（如 Windows）不可用，忽略即可
+            pass
+    await run_engine(stop_event)
+
+
+def main():
+    logger.info("worker 进程启动")
+    asyncio.run(_async_main())
+
+
+if __name__ == "__main__":
+    main()
+

datasvc/docker-compose.yml

@@ -10,7 +10,8 @@ services:
       - POLL_FACTOR=0.5
       - DATA_DIR=/data
       - TZ=Asia/Shanghai
-      - VERIFY_INTERVALS=1m=5,5m=2,10m=2
+      - WS_ENABLED=false
+      - DATASVC_ENGINE=1
     ports:
       - "9000:9000"
     volumes: