From 03c5148c2914507415c0ac6c1ef9c33247d400f1 Mon Sep 17 00:00:00 2001 From: Porter Date: Sun, 16 Nov 2025 11:59:52 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E6=96=AD=E7=BA=BF=E9=87=8D?= =?UTF-8?q?=E8=BF=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .DS_Store | Bin 10244 -> 10244 bytes data_provider/config.json | 14 +--- data_provider/main.py | 148 +++++++++++++++++++++++++++++++++++--- 3 files changed, 138 insertions(+), 24 deletions(-) diff --git a/.DS_Store b/.DS_Store index ca5e05936b9e7b439cf7d37a6058fd0fe95210df..e997eb39c2deb35e8a2bab09e3d0bbc0afa49ae2 100644 GIT binary patch delta 122 zcmZn(XbG6$RCU^hRb-ew+wi_APp<;4X_Ir&Kp3=BIbD~K0rNLE*yo15q;7#SGX z>L^qj8XB7EC|DR;)Yfuxh$`z_2gPUS(BbnO)%*%Vr@_W@Z4@uq1B) delta 159 zcmZn(XbG6$pKU^hRb{$?J5i_GdN#mPBI`T04F3=9nH3@HqW3?)EZ%uvRVoK#+1 zkd%|31eDo5nP0q6OS-z+(9FU}N5R<8pjJnr+R(_@SVzIoz|5kymXkwNS>HM+K07Bj wFTZ#4N^yC{zR71q timeframe -> since_ms） + self._resume_since: Dict[str, Dict[str, int]] = {} + # 恢复点持久化文件 + self._resume_file: Path = self.data_dir / "resume_since.json" + # 尝试加载历史恢复点 + self._load_resume_since() def _load_config(self) -> Dict[str, object]: if not self.config_path.exists(): @@ -330,6 +336,95 @@ class DataProvider: def resample_df(self, df: pd.DataFrame, interval: int) -> pd.DataFrame: return resample_to_interval(df, interval) + def _save_resume_since(self) -> None: + path = self._resume_file + path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = path.with_suffix(path.suffix + ".tmp") + with self._lock: + snapshot = { + symbol: {tf: int(since) for tf, since in tf_map.items()} + for symbol, tf_map in self._resume_since.items() + } + try: + with tmp_path.open("w", encoding="utf-8") as fp: + json.dump(snapshot, fp, ensure_ascii=False, separators=(",", ":")) + os.replace(tmp_path, path) + finally: + if tmp_path.exists(): + try: + tmp_path.unlink() + except OSError: + pass + logger.debug("恢复点已保存到磁盘: %s", path) + + def _load_resume_since(self) -> None: + path = self._resume_file + if not path.exists(): + return + try: + with path.open("r", encoding="utf-8") as fp: + raw = json.load(fp) + except Exception as exc: + logger.warning("恢复点文件读取失败，忽略: %s (%s)", path, exc) + return + if not isinstance(raw, dict): + logger.warning("恢复点文件格式错误，忽略: %s", path) + return + loaded: Dict[str, Dict[str, int]] = {} + for symbol, tf_map in raw.items(): + if not isinstance(tf_map, dict): + continue + per_symbol: Dict[str, int] = {} + for timeframe, since in tf_map.items(): + try: + per_symbol[str(timeframe)] = int(since) + except Exception: + continue + if per_symbol: + loaded[str(symbol)] = per_symbol + if not loaded: + return + with self._lock: + # 合并为更早的 since，避免遗漏 + for symbol, tf_map in loaded.items(): + cur = self._resume_since.setdefault(symbol, {}) + for timeframe, since in tf_map.items(): + prev = cur.get(timeframe) + if prev is None or since < prev: + cur[timeframe] = since + logger.info("已加载恢复点: %s", path) + + def _get_resume_since(self, symbol: str, timeframe: str) -> Optional[int]: + with self._lock: + return self._resume_since.get(symbol, {}).get(timeframe) + + def _set_resume_since(self, symbol: str, timeframe: str, since_ms: int) -> None: + with self._lock: + per_symbol = self._resume_since.setdefault(symbol, {}) + prev = per_symbol.get(timeframe) + # 取更早的 since，避免跳过数据 + if prev is None or since_ms < prev: + per_symbol[timeframe] = since_ms + logger.warning( + "记录断线恢复点: %s %s since=%s (%s)", + symbol, + timeframe, + since_ms, + to_utc_iso(since_ms), + ) + # 同步写盘 + self._save_resume_since() + + def _clear_resume_since(self, symbol: str, timeframe: str) -> None: + with self._lock: + if symbol in self._resume_since and timeframe in self._resume_since[symbol]: + del self._resume_since[symbol][timeframe] + if not self._resume_since[symbol]: + del self._resume_since[symbol] + logger.info("清除断线恢复点: %s %s", symbol, timeframe) + # 同步写盘 + self._save_resume_since() + def start_background_workers(self) -> None: if self._fetch_thread and self._fetch_thread.is_alive(): return @@ -353,21 +448,50 @@ class DataProvider: for symbol in self.symbols: for timeframe in self.timeframes: try: - candles = self.exchange.fetch_ohlcv( - symbol, - timeframe=timeframe, - limit=RECENT_CANDLE_LIMIT, - ) + # 若存在断线恢复点，则优先从该 since 补齐历史数据 + resume_since = self._get_resume_since(symbol, timeframe) + if resume_since is not None: + logger.info( + "开始断线后补数: %s %s since=%s (%s)", + symbol, + timeframe, + resume_since, + to_utc_iso(resume_since), + ) + history = self._fetch_history(symbol, timeframe, resume_since) + with self._lock: + current = self.data.setdefault(symbol, {}).get(timeframe, []) + merged = self._merge_candles(timeframe, current, history) + self.data[symbol][timeframe] = merged + self._clear_resume_since(symbol, timeframe) + else: + # 正常增量获取最近若干根K线 + candles = self.exchange.fetch_ohlcv( + symbol, + timeframe=timeframe, + limit=RECENT_CANDLE_LIMIT, + ) + if not candles: + continue + with self._lock: + current = self.data.setdefault(symbol, {}).get(timeframe, []) + merged = self._merge_candles(timeframe, current, candles) + self.data[symbol][timeframe] = merged except ccxt.BaseError as exc: logger.error("更新最新K线失败 (%s %s): %s", symbol, timeframe, exc) + # 记录应当从何时恢复拉取，避免重连后从当前时间开始导致丢K + with self._lock: + current = self.data.get(symbol, {}).get(timeframe, []) + if current: + last_ts = int(current[-1]["timestamp"]) + else: + last_ts = self.start_time_ms + tf_ms = TIMEFRAME_TO_MS[timeframe] + # 回退一个周期，确保包含可能未完全收盘的K线，去重由 _merge_candles 处理 + since_ms = max(self.start_time_ms, last_ts - tf_ms) + self._set_resume_since(symbol, timeframe, since_ms) time.sleep(2) continue - if not candles: - continue - with self._lock: - current = self.data.setdefault(symbol, {}).get(timeframe, []) - merged = self._merge_candles(timeframe, current, candles) - self.data[symbol][timeframe] = merged if self._stop_event.wait(RECENT_FETCH_INTERVAL): break @@ -386,6 +510,8 @@ class DataProvider: for symbol, tf_map in snapshot.items(): for timeframe, data in tf_map.items(): self._write_to_disk(symbol, timeframe, data) + # 周期性也保存一次恢复点，保证一致性 + self._save_resume_since() def is_ready(self) -> bool: return self._ready.is_set()