From 1c022a1c7ca402e07aa8984327cc911d43c97591 Mon Sep 17 00:00:00 2001 From: 0112020179 Date: Sun, 6 Sep 2026 12:11:05 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix(fuyao):=20=E6=B5=81=E5=BC=8F=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E5=8E=86=E5=8F=B2=E6=97=A5K=E9=81=BF=E5=85=8D?= =?UTF-8?q?=E5=86=85=E5=AD=98=E6=BA=A2=E5=87=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/plugins/fuyao/provider.py | 193 ++++++++++++++++-- backend/app/services/kline_sync.py | 53 +++++ .../tests/test_fuyao_daily_streaming_sync.py | 86 ++++++++ backend/tests/test_fuyao_provider.py | 64 ++++++ 4 files changed, 379 insertions(+), 17 deletions(-) create mode 100644 backend/tests/test_fuyao_daily_streaming_sync.py diff --git a/backend/app/plugins/fuyao/provider.py b/backend/app/plugins/fuyao/provider.py index a457904..3e48e9e 100644 --- a/backend/app/plugins/fuyao/provider.py +++ b/backend/app/plugins/fuyao/provider.py @@ -32,12 +32,13 @@ import logging import math import re import time -from collections.abc import Callable +from collections.abc import Callable, Iterator from dataclasses import dataclass, field from datetime import UTC, date, datetime, timedelta from pathlib import Path import polars as pl +import pyarrow.parquet as pq from app.data_providers.normalizer import DAILY_COLS, normalize_daily from app.indicators.pipeline import filter_halt_days @@ -69,6 +70,12 @@ _DAILY10_DUMP_KIND = "daily-k-10d" _DAILY_DUMP_KIND = "daily-k" # 10 年全量日K dump(约 172MB), 深窗口一次下载覆盖全市场 _RECENT_DUMP_DAYS = 12 # 窗口跨度 ≤ 此天数时优先走 10d dump(覆盖 ≈10 个交易日) _PREV_CLOSE_BACKDAYS = 30 # 推导因子时向前找"除权日前收盘"的回看天数(容忍长期停牌) +_DAILY_DUMP_BATCH_ROWS = 100_000 +_HIST_SYMBOL_BATCH = 50 +_DAILY_DUMP_COLUMNS = [ + "thscode", "adjusted", "date_ms", "open_price", "high_price", "low_price", + "close_price", "volume", "turnover", +] def get_api_key() -> str: @@ -444,34 +451,152 @@ 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, + )) + return pl.concat(chunks, how="diagonal_relaxed") if chunks else pl.DataFrame() + + def iter_daily( + self, + symbols: list[str], + start_time: datetime | None, + 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": - return pl.DataFrame() + return end_dt = end_time or datetime.now() 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: dump = self._ensure_dump(_DAILY10_DUMP_KIND, "daily_k_10d") if _dump_covers(dump, start_d, end_d): - df = self._daily_from_dump(dump, set(symbols), start_d, end_d) + df = self._daily_from_dump(dump, symset, start_d, end_d) if on_chunk_done: on_chunk_done(1, 1) - logger.info("扶摇日K(10d dump)完成: %d 行 [%s ~ %s]", df.height, start_d, end_d) - return df - logger.info("扶摇 10d dump 未覆盖窗口 [%s ~ %s], 尝试 10 年 dump", start_d, end_d) + if not df.is_empty(): + yield df + return except FuyaoError as e: - logger.warning("扶摇日K 10d dump 不可用, 尝试 10 年 dump: %s", e) + logger.warning("扶摇 10d dump 不可用: %s", e) - df = self._daily_from_big_dump(set(symbols), start_d, end_d) - if df is not None: - if on_chunk_done: - on_chunk_done(1, 1) - logger.info("扶摇日K(10 年 dump)完成: %d 行 [%s ~ %s]", df.height, start_d, end_d) - return df - df = self._daily_from_api(symbols, start_d, end_d, on_chunk_done) - logger.info("扶摇日K(单标的接口)完成: %d 行 [%s ~ %s]", df.height, start_d, end_d) - return df + dump_info = self._daily_dump_info() + sources: list[tuple[str, date, date]] = [] + if dump_info: + _, dump_min, dump_max = dump_info + if start_d < dump_min: + sources.append(("api", start_d, min(end_d, dump_min - timedelta(days=1)))) + overlap_start, overlap_end = max(start_d, dump_min), min(end_d, dump_max) + 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: + 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() + except (FuyaoError, KeyError): + ten_min = ten_max = None + if ( + ten_min is not None + and ten_min <= dump_max + timedelta(days=1) + and _tail_ok(end_d, ten_max) + ): + sources.append(("10d", tail_start, end_d)) + else: + # 多年 dump 与请求终点之间存在不可验证的缺口,不能返回半段数据。 + sources = [("api", start_d, end_d)] + else: + sources.append(("api", start_d, end_d)) + + dump_batch_count = 0 + if dump_info: + dump_rows = pq.ParquetFile(dump_info[0]).metadata.num_rows + dump_batch_count = max( + 1, (dump_rows + _DAILY_DUMP_BATCH_ROWS - 1) // _DAILY_DUMP_BATCH_ROWS + ) + api_batch_count = (len(symbols) + _HIST_SYMBOL_BATCH - 1) // _HIST_SYMBOL_BATCH + total = sum( + dump_batch_count if kind == "dump" else 1 if kind == "10d" else api_batch_count + for kind, _, _ in sources + ) + done = 0 + for kind, source_start, source_end in sources: + if source_start > source_end: + continue + if kind == "dump": + path = dump_info[0] # type: ignore[index] + for df in self._iter_big_dump(path, symset, source_start, source_end): + done += 1 + if on_chunk_done: + on_chunk_done(done, total) + if not df.is_empty(): + yield df + elif kind == "10d": + ten = self._ensure_dump(_DAILY10_DUMP_KIND, "daily_k_10d") + df = self._daily_from_dump(ten, symset, source_start, source_end) + done += 1 + if on_chunk_done: + on_chunk_done(done, total) + if not df.is_empty(): + yield df + else: + batches = [ + symbols[i:i + _HIST_SYMBOL_BATCH] + for i in range(0, len(symbols), _HIST_SYMBOL_BATCH) + ] + for batch in batches: + rows: list[dict] = [] + for symbol in batch: + rows.extend(_kline_rows( + symbol, + self._historical_bars( + symbol, source_start, source_end, failed_symbols=failed_symbols + ), + )) + time.sleep(_HIST_INTERVAL_S) + df = normalize_daily(rows, source=self.name) + done += 1 + if on_chunk_done: + 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 的路径和覆盖范围,不把大文件读入进程内存。""" + path = None + for candidate in sorted(_cache_dir().glob("daily_k__*.parquet"), reverse=True): + try: + dmin, dmax = _dump_date_range(candidate) + except Exception: + continue + if dmin is not None and dmax is not None: + return candidate, dmin, dmax + try: + path = self._ensure_dump_path(_DAILY_DUMP_KIND, "daily_k") + dmin, dmax = _dump_date_range(path) + except FuyaoError as e: + logger.warning("扶摇 10 年 dump 不可用, 回退单标的接口: %s", e) + return None + return (path, dmin, dmax) if dmin is not None and dmax is not None else None def _daily_from_dump( self, dump: pl.DataFrame, symset: set[str], start_d: date, end_d: date @@ -481,6 +606,32 @@ class FuyaoProvider: ) return self._map_daily_dump(df, symset, start_d, end_d) + def _iter_big_dump( + self, path: Path, symset: set[str], start_d: date, end_d: date + ) -> Iterator[pl.DataFrame]: + """按固定 record batch 读取多年 dump,不做单次全量 collect。""" + parquet = pq.ParquetFile(path) + columns = [name for name in _DAILY_DUMP_COLUMNS if name in parquet.schema.names] + for batch in parquet.iter_batches( + batch_size=_DAILY_DUMP_BATCH_ROWS, + columns=columns, + ): + raw = pl.from_arrow(batch) + if raw.is_empty() or "date_ms" not in raw.columns or "thscode" not in raw.columns: + continue + start_ms, end_ms = _ms_of_date(start_d), _ms_of_date(end_d) + raw = raw.filter( + (pl.col("date_ms") >= start_ms) + & (pl.col("date_ms") <= end_ms) + & pl.col("thscode").is_in(sorted(symset)) + ) + if raw.is_empty(): + continue + raw = raw.with_columns( + pl.from_epoch(pl.col("date_ms") + _SH_MS, time_unit="ms").dt.date().alias("date") + ) + yield self._map_daily_dump(raw, symset, start_d, end_d) + def _map_daily_dump( self, df: pl.DataFrame, symset: set[str], start_d: date, end_d: date ) -> pl.DataFrame: @@ -582,7 +733,13 @@ class FuyaoProvider: 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) -> list[dict]: + def _historical_bars( + self, + symbol: str, + start_d: date, + end_d: date, + failed_symbols: list[str] | None = None, + ) -> list[dict]: """按 ≤10 年窗口分片拉取单标的原始日K。中途失败软返回已得行, 不抛出。""" out: list[dict] = [] s = _ms_of_date(start_d) @@ -593,6 +750,8 @@ 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 d634479..d3bb589 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -8,6 +8,8 @@ from __future__ import annotations import logging +import shutil +import uuid from collections.abc import Callable from datetime import date, datetime, timedelta @@ -177,6 +179,18 @@ def sync_and_persist_daily_batch( end_time = end_date or datetime.now() days = count or 365 start_time = start_date or (end_time - timedelta(days=days)) + iter_daily = getattr(provider, "iter_daily", None) + if callable(iter_daily): + return _persist_daily_chunks( + iter_daily( + symbols, + start_time=start_time, + end_time=end_time, + on_chunk_done=on_chunk_done, + strict=True, + ), + repo, + ) df = provider.get_daily( symbols, start_time=start_time, @@ -228,6 +242,45 @@ def sync_and_persist_daily_batch( return df.height +def _persist_daily_chunks(chunks, repo: KlineRepository) -> int: + """先把流式 provider 结果写入私有 staging,完整取数后再提交正式分区。""" + root = repo.store.data_dir / ".daily_sync_staging" / uuid.uuid4().hex + written = 0 + try: + for index, df in enumerate(chunks): + if df.is_empty(): + continue + for date_df in df.partition_by("date"): + dt = date_df["date"][0] + ds = dt.isoformat() if hasattr(dt, "isoformat") else str(dt) + out = root / f"date={ds}" / f"part-{index}.parquet" + out.parent.mkdir(parents=True, exist_ok=True) + date_df.write_parquet(out) + written += date_df.height + + 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)) + finally: + shutil.rmtree(root, ignore_errors=True) + if root.parent.exists() and not any(root.parent.iterdir()): + 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 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 new file mode 100644 index 0000000..379c744 --- /dev/null +++ b/backend/tests/test_fuyao_daily_streaming_sync.py @@ -0,0 +1,86 @@ +"""扶摇流式日K同步: 大历史不得在内存中累积后再写入。""" +from __future__ import annotations + +from datetime import date, datetime + +import polars as pl +import pytest + +from app.services import kline_sync +from app.tickflow.repository import DataStore, KlineRepository + + +def _daily(symbol: str, day: date) -> pl.DataFrame: + return pl.DataFrame({ + "symbol": [symbol], "date": [day], "open": [10.0], "high": [11.0], + "low": [9.0], "close": [10.5], "volume": [100.0], "amount": [1050.0], + }) + + +class _StreamingProvider: + def __init__(self, chunks, error: Exception | None = None): + self.chunks = chunks + self.error = error + self.get_daily_called = False + + def iter_daily(self, *args, **kwargs): + yield from self.chunks + if self.error: + raise self.error + + def get_daily(self, *args, **kwargs): + self.get_daily_called = True + raise AssertionError("streaming provider must not collect a full daily DataFrame") + + +@pytest.fixture +def repo(tmp_path): + return KlineRepository(DataStore(tmp_path)) + + +def _route_fuyao(monkeypatch, provider): + monkeypatch.setattr(kline_sync.preferences, "get_daily_data_provider", lambda: "fuyao") + from app.data_providers import custom + monkeypatch.setattr( + custom, + "provider_has_dataset", + lambda name, dataset: name == "fuyao" and dataset == "daily", + ) + monkeypatch.setattr(custom, "get_provider", lambda name: provider) + + +def test_sync_persists_streamed_fuyao_chunks_only_after_fetch(monkeypatch, repo): + provider = _StreamingProvider([ + _daily("000001.SZ", date(2020, 1, 2)), + _daily("000002.SZ", date(2020, 1, 2)), + ]) + _route_fuyao(monkeypatch, provider) + + written = kline_sync.sync_and_persist_daily_batch( + ["000001.SZ", "000002.SZ"], repo, object(), + start_date=datetime(2020, 1, 1), end_date=datetime(2020, 1, 2), + ) + + assert written == 2 + assert provider.get_daily_called is False + stored = pl.read_parquet( + repo.store.data_dir / "kline_daily" / "date=2020-01-02" / "part.parquet" + ) + assert set(stored["symbol"].to_list()) == {"000001.SZ", "000002.SZ"} + assert not (repo.store.data_dir / ".daily_sync_staging").exists() + + +def test_sync_discards_staging_when_streaming_fails(monkeypatch, repo): + provider = _StreamingProvider( + [_daily("000001.SZ", date(2020, 1, 2))], error=RuntimeError("network lost") + ) + _route_fuyao(monkeypatch, provider) + + with pytest.raises(RuntimeError, match="network lost"): + kline_sync.sync_and_persist_daily_batch( + ["000001.SZ"], repo, object(), + start_date=datetime(2020, 1, 1), end_date=datetime(2020, 1, 2), + ) + + assert not list((repo.store.data_dir / "kline_daily").glob("date=*")) + assert not (repo.store.data_dir / ".daily_sync_staging").exists() diff --git a/backend/tests/test_fuyao_provider.py b/backend/tests/test_fuyao_provider.py index aaddf77..badb107 100644 --- a/backend/tests/test_fuyao_provider.py +++ b/backend/tests/test_fuyao_provider.py @@ -740,6 +740,39 @@ 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 = { + "000001.SZ": [_bar(date(2018, 1, 2), 10.0)], + "000002.SZ": [_bar(date(2018, 1, 2), 11.0)], + "000003.SZ": [_bar(date(2018, 1, 2), 12.0)], + } + provider = _hist_provider(monkeypatch, _FakeHistClient(bars)) + monkeypatch.setattr(fp, "_HIST_SYMBOL_BATCH", 2) + chunks = list(provider.iter_daily( + list(bars), datetime(2018, 1, 1), datetime(2018, 1, 3) + )) + assert [set(df["symbol"].to_list()) for df in chunks] == [ + {"000001.SZ", "000002.SZ"}, {"000003.SZ"} + ] + + def test_daily_empty_symbols_or_non_stock_returns_empty(monkeypatch): provider = _hist_provider(monkeypatch, _FakeHistClient({})) assert provider.get_daily([], None, None).is_empty() @@ -899,6 +932,37 @@ def test_daily_deep_window_uses_big_dump(monkeypatch, tmp_path): assert df["volume"].to_list() == [975_701.0, 12_345.0, 12_345.0] +def test_iter_daily_big_dump_reads_bounded_record_batches(monkeypatch, tmp_path): + """多年 dump 分批读取,不经 LazyFrame collect() 物化整份文件。""" + rows = [_dump_bar("000001.SZ", date(2026, 1, 1) + timedelta(days=i), 10.0 + i) + for i in range(5)] + provider = _bigdump_provider(monkeypatch, tmp_path, rows) + monkeypatch.setattr(fp, "_DAILY_DUMP_BATCH_ROWS", 2) + chunks = list(provider.iter_daily( + ["000001.SZ"], datetime(2026, 1, 1), datetime(2026, 1, 5) + )) + assert [chunk.height for chunk in chunks] == [2, 2, 1] + assert [d for chunk in chunks for d in chunk["date"].to_list()] == [ + date(2026, 1, 1), date(2026, 1, 2), date(2026, 1, 3), + date(2026, 1, 4), date(2026, 1, 5), + ] + + +def test_iter_daily_splits_older_history_from_big_dump(monkeypatch, tmp_path): + """请求跨 dump 起点时,只把 dump 外那段回退到历史 API。""" + provider = _bigdump_provider( + monkeypatch, + tmp_path, + [_dump_bar("000001.SZ", date(2020, 1, 2), 10.0)], + ) + provider._client = _FakeHistClient({"000001.SZ": [_bar(date(2019, 12, 31), 9.0)]}) + rows = pl.concat(list(provider.iter_daily( + ["000001.SZ"], datetime(2019, 12, 30), datetime(2020, 1, 2) + ))) + assert rows["date"].to_list() == [date(2019, 12, 31), date(2020, 1, 2)] + assert len(provider._get_client().calls) == 1 + + def test_daily_big_dump_tail_filled_by_10d(monkeypatch, tmp_path): """大 dump 末端缺口(dmax 旧)由 10d dump 补尾, 两段拼接无缝。""" big_rows = [ From 3d6beb3c35c6ee8dcc06c0ad309d86a4f8ffd484 Mon Sep 17 00:00:00 2001 From: 0112020179 Date: Sun, 6 Sep 2026 17:39:43 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(fuyao):=20=E6=A0=B9=E6=8D=AE=E5=AE=A1?= =?UTF-8?q?=E6=9F=A5=E5=AE=8C=E5=96=84=E6=B5=81=E5=BC=8F=E5=90=8C=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/plugins/fuyao/provider.py | 114 +++--------------- backend/app/services/kline_sync.py | 35 +++--- .../tests/test_fuyao_daily_streaming_sync.py | 21 ++++ backend/tests/test_fuyao_provider.py | 38 +++--- docs/plugin-development.md | 10 ++ 5 files changed, 88 insertions(+), 130 deletions(-) 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 行字段