diff --git a/.DS_Store b/.DS_Store index 7514a52..fd04eab 100644 Binary files a/.DS_Store and b/.DS_Store differ diff --git a/.gitignore b/.gitignore index 15160ab..8e52078 100644 --- a/.gitignore +++ b/.gitignore @@ -29,3 +29,4 @@ feature_meta .DS_Store 交易记录/~$交易规则.docx /datasvc/data +.DS_Store diff --git a/datasvc/app/main.py b/datasvc/app/main.py index 61728f5..d165777 100644 --- a/datasvc/app/main.py +++ b/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] + + diff --git a/datasvc/app/storage.py b/datasvc/app/storage.py index 7e4d7fc..f00a2ef 100644 --- a/datasvc/app/storage.py +++ b/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): diff --git a/datasvc/app/worker.py b/datasvc/app/worker.py new file mode 100644 index 0000000..7a5863a --- /dev/null +++ b/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() + diff --git a/datasvc/docker-compose.yml b/datasvc/docker-compose.yml index 31cf276..3319b58 100644 --- a/datasvc/docker-compose.yml +++ b/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: