diff --git a/backend/app/plugins/fuyao/provider.py b/backend/app/plugins/fuyao/provider.py index 3e48e9e..576ff9f 100644 --- a/backend/app/plugins/fuyao/provider.py +++ b/backend/app/plugins/fuyao/provider.py @@ -451,13 +451,17 @@ class FuyaoProvider: - 兜底: 单标的 historical 接口(窗口早于 dump 覆盖 / dump 不可用; 10 年自动分片, 逐标的节流 + 进度回调)。 """ - chunks = list(self.iter_daily( - symbols, - start_time=start_time, - end_time=end_time, - asset_type=asset_type, - on_chunk_done=on_chunk_done, - )) + chunks = [ + df + for df in self.iter_daily( + symbols, + start_time=start_time, + end_time=end_time, + asset_type=asset_type, + on_chunk_done=on_chunk_done, + ) + if not df.is_empty() + ] return pl.concat(chunks, how="diagonal_relaxed") if chunks else pl.DataFrame() def iter_daily( @@ -467,7 +471,6 @@ class FuyaoProvider: end_time: datetime | None, asset_type: str = "stock", on_chunk_done: Callable[[int, int], None] | None = None, - strict: bool = False, ) -> Iterator[pl.DataFrame]: """分批产出日K,供历史同步逐批落盘,避免全市场结果累积在内存。""" if not symbols or asset_type != "stock": @@ -476,7 +479,6 @@ class FuyaoProvider: start_dt = start_time or (end_dt - timedelta(days=365)) start_d, end_d = start_dt.date(), end_dt.date() symset = set(symbols) - failed_symbols: list[str] = [] if (end_d - start_d).days <= _RECENT_DUMP_DAYS: try: @@ -501,7 +503,7 @@ class FuyaoProvider: if overlap_start <= overlap_end: sources.append(("dump", overlap_start, overlap_end)) tail_start = max(start_d, dump_max + timedelta(days=1)) - if tail_start <= end_d: + if tail_start <= end_d and not _tail_ok(end_d, dump_max): try: ten = self._ensure_dump(_DAILY10_DUMP_KIND, "daily_k_10d") ten_dates = pl.from_epoch( @@ -563,9 +565,7 @@ class FuyaoProvider: for symbol in batch: rows.extend(_kline_rows( symbol, - self._historical_bars( - symbol, source_start, source_end, failed_symbols=failed_symbols - ), + self._historical_bars(symbol, source_start, source_end), )) time.sleep(_HIST_INTERVAL_S) df = normalize_daily(rows, source=self.name) @@ -574,11 +574,6 @@ class FuyaoProvider: on_chunk_done(done, total) if not df.is_empty(): yield df - if strict and failed_symbols: - sample = ", ".join(failed_symbols[:10]) - raise FuyaoError( - f"扶摇日K同步部分标的失败: {len(failed_symbols)} 只 (样例: {sample})" - ) def _daily_dump_info(self) -> tuple[Path, date, date] | None: """返回多年 dump 的路径和覆盖范围,不把大文件读入进程内存。""" @@ -618,6 +613,7 @@ class FuyaoProvider: ): raw = pl.from_arrow(batch) if raw.is_empty() or "date_ms" not in raw.columns or "thscode" not in raw.columns: + yield pl.DataFrame() continue start_ms, end_ms = _ms_of_date(start_d), _ms_of_date(end_d) raw = raw.filter( @@ -626,6 +622,7 @@ class FuyaoProvider: & pl.col("thscode").is_in(sorted(symset)) ) if raw.is_empty(): + yield pl.DataFrame() continue raw = raw.with_columns( pl.from_epoch(pl.col("date_ms") + _SH_MS, time_unit="ms").dt.date().alias("date") @@ -662,84 +659,7 @@ class FuyaoProvider: cols = [c for c in DAILY_COLS if c in out.columns] return out.select(cols).sort(["symbol", "date"]) if not out.is_empty() else out.select(cols) - def _daily_from_big_dump( - self, symset: set[str], start_d: date, end_d: date - ) -> pl.DataFrame | None: - """深窗口主路径: 10 年全量 dump(lazy 按需筛) + 必要时 10d dump 补尾。 - - 覆盖不了(窗口早于 10 年 / dump 拉取失败)返回 None, 由调用方走单标的接口。 - """ - path = self._ensure_daily_big_dump(start_d) - if path is None: - return None - _, dmax = _dump_date_range(path) - big_hi = min(end_d, dmax) - # 窗口/标的过滤下推到 lazy 计划, 只物化需要的行(全量 10 年 ≈ 13.6M 行) - window = ( - pl.scan_parquet(path) - .with_columns( - pl.from_epoch(pl.col("date_ms") + _SH_MS, time_unit="ms").dt.date().alias("date") - ) - .filter( - (pl.col("date") >= start_d) - & (pl.col("date") <= big_hi) - & pl.col("thscode").is_in(sorted(symset)) - ) - .collect() - ) - parts = [self._map_daily_dump(window, symset, start_d, big_hi)] - if not _tail_ok(end_d, dmax): - # 末端缺口(如 10 年 dump 是旧 release, end 是最近交易日): 10d dump 补尾 - try: - ten = self._ensure_dump(_DAILY10_DUMP_KIND, "daily_k_10d") - ten_dates = pl.from_epoch( - ten["date_ms"].cast(pl.Int64) + _SH_MS, time_unit="ms" - ).dt.date() - ten_min, ten_max = ten_dates.min(), ten_dates.max() - if ( - ten_min is not None - and ten_min <= dmax + timedelta(days=1) - and _tail_ok(end_d, ten_max) - ): - tail_start = max(start_d, dmax + timedelta(days=1)) - parts.append(self._daily_from_dump(ten, symset, tail_start, end_d)) - else: - return None # 中段或尾部仍有缺口 → 单标的兜底, 不交缺口数据 - except FuyaoError as e: - logger.warning("扶摇 10d dump 补尾失败: %s", e) - return None - non_empty = [p for p in parts if not p.is_empty()] - if not non_empty: - return pl.DataFrame() - out = pl.concat(non_empty, how="vertical_relaxed") - return out.unique(subset=["symbol", "date"], keep="last").sort(["symbol", "date"]) - - def _daily_from_api( - self, - symbols: list[str], - start_d: date, - end_d: date, - on_chunk_done: Callable[[int, int], None] | None, - ) -> pl.DataFrame: - frames: list[pl.DataFrame] = [] - for i, sym in enumerate(symbols): - rows = self._historical_bars(sym, start_d, end_d) - time.sleep(_HIST_INTERVAL_S) - if rows: - df = normalize_daily(_kline_rows(sym, rows), default_symbol=sym, source=self.name) - if not df.is_empty(): - frames.append(df) - if on_chunk_done: - on_chunk_done(i + 1, len(symbols)) - return pl.concat(frames, how="diagonal_relaxed") if frames else pl.DataFrame() - - def _historical_bars( - self, - symbol: str, - start_d: date, - end_d: date, - failed_symbols: list[str] | None = None, - ) -> list[dict]: + def _historical_bars(self, symbol: str, start_d: date, end_d: date) -> list[dict]: """按 ≤10 年窗口分片拉取单标的原始日K。中途失败软返回已得行, 不抛出。""" out: list[dict] = [] s = _ms_of_date(start_d) @@ -750,8 +670,6 @@ class FuyaoProvider: out.extend(self._get_client().historical_kline(symbol, s, chunk_end, adjust="none")) except FuyaoError as err: logger.warning("扶摇日K拉取失败 %s [%s ~ %s]: %s", symbol, start_d, end_d, err) - if failed_symbols is not None: - failed_symbols.append(symbol) break if s + _HIST_MAX_SPAN_MS <= e: time.sleep(_HIST_INTERVAL_S) diff --git a/backend/app/services/kline_sync.py b/backend/app/services/kline_sync.py index d3bb589..acee83a 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -7,8 +7,10 @@ """ from __future__ import annotations +import contextlib import logging import shutil +import time import uuid from collections.abc import Callable from datetime import date, datetime, timedelta @@ -187,7 +189,6 @@ def sync_and_persist_daily_batch( start_time=start_time, end_time=end_time, on_chunk_done=on_chunk_done, - strict=True, ), repo, ) @@ -244,7 +245,9 @@ def sync_and_persist_daily_batch( def _persist_daily_chunks(chunks, repo: KlineRepository) -> int: """先把流式 provider 结果写入私有 staging,完整取数后再提交正式分区。""" - root = repo.store.data_dir / ".daily_sync_staging" / uuid.uuid4().hex + staging_base = repo.store.data_dir / ".daily_sync_staging" + _sweep_stale_daily_staging(staging_base) + root = staging_base / uuid.uuid4().hex written = 0 try: for index, df in enumerate(chunks): @@ -261,26 +264,28 @@ def _persist_daily_chunks(chunks, repo: KlineRepository) -> int: for date_dir in sorted(root.glob("date=*")): files = sorted(date_dir.glob("*.parquet")) if files: - repo.append_daily(pl.scan_parquet(files).collect(streaming=True)) + repo.append_daily(pl.scan_parquet(files).collect(engine="streaming")) finally: shutil.rmtree(root, ignore_errors=True) - if root.parent.exists() and not any(root.parent.iterdir()): + with contextlib.suppress(OSError): root.parent.rmdir() - if written: - try: - d = repo.store.data_dir.as_posix() - repo.db.execute( - f"""CREATE OR REPLACE VIEW kline_daily AS - SELECT * FROM read_parquet( - '{d}/kline_daily/**/*.parquet', union_by_name=true - )""" - ) - except Exception as e: # noqa: BLE001 - logger.warning("refresh view failed: %s", e) return written +def _sweep_stale_daily_staging(staging_base, max_age_s: int = 24 * 60 * 60) -> None: + """清理崩溃遗留的旧同步目录,不碰仍可能活跃的新目录。""" + if not staging_base.exists(): + return + cutoff = time.time() - max_age_s + for run_dir in staging_base.iterdir(): + try: + if run_dir.is_dir() and run_dir.stat().st_mtime < cutoff: + shutil.rmtree(run_dir) + except OSError: + logger.warning("failed to clean stale daily staging: %s", run_dir) + + def sync_daily_by_quotes(repo: KlineRepository) -> int: """用实时行情接口拉全市场当日数据,覆写 kline_daily 今天分区。 diff --git a/backend/tests/test_fuyao_daily_streaming_sync.py b/backend/tests/test_fuyao_daily_streaming_sync.py index 379c744..4e67d27 100644 --- a/backend/tests/test_fuyao_daily_streaming_sync.py +++ b/backend/tests/test_fuyao_daily_streaming_sync.py @@ -1,6 +1,7 @@ """扶摇流式日K同步: 大历史不得在内存中累积后再写入。""" from __future__ import annotations +import os from datetime import date, datetime import polars as pl @@ -84,3 +85,23 @@ def test_sync_discards_staging_when_streaming_fails(monkeypatch, repo): assert not list((repo.store.data_dir / "kline_daily").glob("date=*")) assert not (repo.store.data_dir / ".daily_sync_staging").exists() + + +def test_sync_sweeps_only_stale_staging(monkeypatch, repo): + staging = repo.store.data_dir / ".daily_sync_staging" + stale = staging / "stale" + fresh = staging / "fresh" + stale.mkdir(parents=True) + fresh.mkdir() + os.utime(stale, (1, 1)) + provider = _StreamingProvider([]) + _route_fuyao(monkeypatch, provider) + + written = kline_sync.sync_and_persist_daily_batch( + ["000001.SZ"], repo, object(), + start_date=datetime(2020, 1, 1), end_date=datetime(2020, 1, 2), + ) + + assert written == 0 + assert not stale.exists() + assert fresh.exists() diff --git a/backend/tests/test_fuyao_provider.py b/backend/tests/test_fuyao_provider.py index badb107..cd844e3 100644 --- a/backend/tests/test_fuyao_provider.py +++ b/backend/tests/test_fuyao_provider.py @@ -627,8 +627,9 @@ def _hist_provider(monkeypatch, fake: _FakeHistClient, allow_dumps: bool = False monkeypatch.setattr(fp, "get_api_key", lambda: "test-key") monkeypatch.setattr(fp, "_HIST_INTERVAL_S", 0.0) if not allow_dumps: - # 默认禁用 dump 档(单标的接口路径测试用); 大 dump 测试传 allow_dumps=True + # API 路径测试必须与开发机真实 dump 缓存隔离; dump 测试显式传 allow_dumps=True。 p._ensure_daily_big_dump = lambda start_d: None # type: ignore[assignment] + p._daily_dump_info = lambda: None # type: ignore[assignment] return p @@ -740,22 +741,6 @@ def test_daily_api_soft_fail_per_symbol(monkeypatch): assert df["symbol"].unique().to_list() == ["000001.SZ"] -def test_iter_daily_strict_raises_after_api_failure(monkeypatch): - provider = _hist_provider( - monkeypatch, - _FakeHistClient( - {"000001.SZ": [_bar(date(2018, 1, 2), 10.0)]}, - error_syms=("000002.SZ",), - ), - ) - with pytest.raises(fc.FuyaoError, match="部分标的失败"): - list(provider.iter_daily( - ["000001.SZ", "000002.SZ"], - datetime(2018, 1, 1), datetime(2018, 1, 3), - strict=True, - )) - - def test_iter_daily_api_is_bounded_by_symbol_batch(monkeypatch): """历史 API 每批只产出固定 symbol 集合,不积累全市场 frames。""" bars = { @@ -1008,6 +993,25 @@ def test_daily_big_dump_tail_filled_by_10d(monkeypatch, tmp_path): assert df["close"].to_list() == [10.9, 11.0, 11.2, 11.65] +def test_daily_big_dump_weekend_end_needs_no_tail(monkeypatch, tmp_path): + """多年 dump 覆盖到周五时,紧邻周末不应回退逐标的 API。""" + provider = _bigdump_provider( + monkeypatch, + tmp_path, + [ + _dump_bar("000001.SZ", date(2026, 8, 1), 10.9), + _dump_bar("000001.SZ", date(2026, 8, 28), 11.65), + ], + ) + + df = provider.get_daily( + ["000001.SZ"], datetime(2026, 8, 1), datetime(2026, 8, 30) + ) + + assert df["date"].to_list() == [date(2026, 8, 1), date(2026, 8, 28)] + assert provider._get_client().calls == [] + + def test_daily_big_dump_midgap_falls_back(monkeypatch, tmp_path): """大 dump 与 10d dump 之间有中段缺口 → 整体回退, 不拼缺口数据。""" big_rows = [ diff --git a/docs/plugin-development.md b/docs/plugin-development.md index 7bac229..a27383d 100644 --- a/docs/plugin-development.md +++ b/docs/plugin-development.md @@ -141,6 +141,10 @@ class MyProvider: on_chunk_done=None) -> pl.DataFrame: """日K: [symbol, date, open, high, low, close, volume, amount]; 不复权""" + def iter_daily(self, symbols, start_time, end_time, asset_type="stock", + on_chunk_done=None) -> Iterator[pl.DataFrame]: + """(可选)有界分批返回与 get_daily 同形的日K; 全市场历史同步优先消费。""" + def get_adj_factors(self, symbols, start_time, end_time, asset_type="stock", on_chunk_done=None) -> pl.DataFrame: """除权因子: [symbol, trade_date, ex_factor]""" @@ -241,6 +245,12 @@ provider 不应自行切换或回退到其他数据源。 | `get_depth_batch` | 单批异常由服务隔离并保留其他批次; 不跨数据源回退 | | `get_minute` | 抛异常时调用方自动回退 TickFlow 重试 | | `get_daily` / `get_adj_factors` / `get_financials` | 异常由上层同步流程捕获记录; 无数据返回空 DataFrame | +| `iter_daily` | 可选; 每批必须符合 `get_daily` 契约。流正常结束后才提交 staging; 未捕获异常会丢弃 staging。provider 内已定义的单标的软失败语义保持不变 | + +`iter_daily` 用于避免大范围日K同步在 provider 内累积完整 DataFrame。实现该方法后, +`kline_sync` 会优先消费它; 未实现的 provider 继续调用 `get_daily`,保持兼容。批次大小应有 +明确上界,不得先把全部结果放入列表再 `concat`。`on_chunk_done(cur, total)` 必须覆盖空批次, +确保最终 `cur == total`。 ### get_realtime 行字段