From 1a616a50614d8d62d3dd9ab365b632ee315ec506 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Wed, 15 Jul 2026 07:19:45 -0700 Subject: [PATCH] feat(data): all-A historical backfill (chunked, resumable, failure-tolerant) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add `data-backfill`, a separate manual command that warms the same tushare read-through caches as the nightly `data-update`, but over a WIDE [backfill.start, today] window (and full 1min history) instead of the incremental lookback tail. The nightly incremental job is unchanged. - qt/data_backfill.py: run_data_backfill + BackfillResult + format_summary. Symbols are chunked into batches of backfill.chunk_size; each batch is warmed over the wide window. A persistent per-batch fetch failure is logged (secret-free: batch index + exception type only) and skipped, and the run continues — the failed batch's gaps stay uncovered and are retried on re-run (opposite of the fail-fast incremental job, by design). Resumability is inherent to the coverage ledgers (fixed window => stable re-runs). Minute history is warmed over the full window only when include_minute is true. A future backfill.start (config typo) raises readably instead of a silent no-op (an inverted window subtracts to zero gaps). The result/summary surface the concurrency mode (serial vs N workers + global rate limit). - qt/data_updater.py: extract behavior-preserving _build_scheduler / _build_caches helpers out of run_data_update (which now calls them) so backfill reuses the exact same cache + scheduler construction. Incremental behavior is byte-identical (existing data_update suites pass unchanged). - qt/config.py: BackfillCfg (start / chunk_size / include_minute, validated) as an all-defaults DataUpdateCfg.backfill sub-config, so every existing config still validates unchanged. - qt/cli.py: data-backfill subcommand (mirrors data-update; runs no factor/alpha/portfolio/backtest/PanelStore). - config/data_update_all_a.yaml: backfill block with an honest cost note (a cold all-A minute backfill is a long, many-hour, resumable operation). - docs/ops/data_update_schedule.md: Historical backfill section (run before enabling the nightly timer). - tests/test_data_backfill.py: 22 network-free tests (chunking, per-batch failure tolerance, minute full-window control, all-A resolution, config validation, future-start guard, real-cache resumability, and a joint failure+resume test proving a re-run re-fetches only the previously-failed symbol). Caches stay raw-only; no token is stored or logged; coverage semantics unchanged. phase0 anchor unchanged (ic_mean=0.9600, annual_return=0.8408). --- config/data_update_all_a.yaml | 23 ++ docs/ops/data_update_schedule.md | 47 +++ qt/cli.py | 25 ++ qt/config.py | 52 +++ qt/data_backfill.py | 253 +++++++++++++++ qt/data_updater.py | 80 +++-- tests/test_data_backfill.py | 525 +++++++++++++++++++++++++++++++ 7 files changed, 977 insertions(+), 28 deletions(-) create mode 100644 qt/data_backfill.py create mode 100644 tests/test_data_backfill.py diff --git a/config/data_update_all_a.yaml b/config/data_update_all_a.yaml index 0b87241..4a7844e 100644 --- a/config/data_update_all_a.yaml +++ b/config/data_update_all_a.yaml @@ -116,6 +116,29 @@ data_update: # per thread; store + ledger writes stay deterministic. concurrency: max_workers: 4 + # PR-2 historical backfill knobs — consumed ONLY by the SEPARATE, manual + # python -m qt.cli data-backfill --config config/data_update_all_a.yaml + # command. The nightly `data-update` job IGNORES this block entirely. + # + # Backfill warms the caches over the WIDE window [backfill.start, today] (the + # full history), NOT the incremental `today - lookback_days` tail. It processes + # the universe in per-symbol batches of `chunk_size`, tolerates a per-batch + # failure (logged + skipped + retryable on re-run), and is resumable via the + # coverage ledgers (a re-run fetches only still-uncovered gaps — safe to Ctrl-C + # and re-run). Run it BEFORE enabling the nightly timer: backfill history first, + # then keep it current nightly. + # + # ⚠️ HONEST COST — a COLD all-A MINUTE backfill is a LONG, MANY-HOUR (potentially + # multi-DAY) operation. At all-A scale (~5500 symbols × ~5 years of 1min bars, + # paged ≤23 trading days / ≤8000 rows per call) `include_minute: true` is by far + # the most expensive path here — tens of thousands of stk_mins calls, run against + # the rate limit. It is resumable, so run it in stages and re-run to finish; if + # you only need daily/universe/financial history first, set include_minute: false + # for a much shorter run and backfill minutes separately. + backfill: + start: "2020-01-01" # WIDE backfill window start (window = [start, today]) + chunk_size: 300 # per-batch symbol count (durable + failure-isolated) + include_minute: true # also backfill 1min history (see HONEST COST above) # --- unused by data-update (schema-only stubs) ------------------------------- # factors: diff --git a/docs/ops/data_update_schedule.md b/docs/ops/data_update_schedule.md index 1a0530c..1e63578 100644 --- a/docs/ops/data_update_schedule.md +++ b/docs/ops/data_update_schedule.md @@ -136,6 +136,53 @@ symbols this becomes statistically likely on any given night. Important properti - **Per-symbol failure isolation is a planned PR-2 hardening** (changing the fail-fast behavior is a warm-path semantic change, deliberately out of scope for PR-1). +## Historical backfill (PR-2) — run FIRST, then keep current + +The nightly `data-update` job only tops up the moving front edge (`lookback_days` +relative to *today*) and, for minutes, a 7-day tail. It does **not** load deep +history. To populate the caches with the full historical window, use the separate, +manual **`data-backfill`** command: + +```bash +cd /home/shaofl/Projects/financial_projects/stocks_market/Quantitative_Trading +/home/shaofl/Development/env_tools/envs/quant_mf/bin/python -m qt.cli \ + data-backfill --config config/data_update_all_a.yaml +``` + +It reuses the SAME caches, feeds, universe resolution, and rate limiter as +`data-update`, but: + +- **Wide window.** It warms `[backfill.start, today]` (the full history), NOT the + incremental `today - lookback_days` tail. +- **Chunked.** Symbols are processed in per-symbol batches of + `data_update.backfill.chunk_size` (default 300), so progress is durable + batch-by-batch and memory stays bounded. +- **Full minute history (optional).** With `data_update.backfill.include_minute: + true` it warms 1min bars over the whole `[backfill.start, today]` window — NOT + the nightly 7-day tail. Set it to `false` to skip minute history for a much + shorter run. +- **Per-symbol / per-batch failure-tolerant.** Unlike the nightly job (which stays + fail-fast on purpose), a batch whose fetch fails persistently (after the feeds' + retries) is **logged (secret-free: batch index + exception type only) and + skipped**; the run CONTINUES to the next batch. The failed batch's gaps stay + uncovered (records no coverage), so they are retried on the next run. The + `OK data-backfill: …` summary tallies `failed_batches` and lists failed symbols + (bounded). +- **Resumable.** Resumability is inherent to the coverage ledgers — a re-run over + the same window fetches only still-uncovered gaps. It is **safe to Ctrl-C and + re-run**; durable successes before an interruption are already persisted. + +⚠️ **This is a LONG, MANY-HOUR (potentially multi-day), manual run** — especially +with `include_minute: true` at all-A scale (~5500 symbols × years of 1min bars). +Run it in stages and re-run to finish. It never runs factors / alpha / portfolio / +backtest / PanelStore, and (like `data-update`) stores RAW endpoint facts only — +no token, no qfq, no derived flag. + +**Order of operations:** backfill the history FIRST (this command), confirm the +summary looks sane, and only THEN enable the nightly incremental timer above so it +keeps the caches current. The backfill window `[start, today]` is fixed, so a +re-run after enabling the timer stays warm except for still-uncovered gaps. + ## Notes - Scope is set by `data_update.universe_scope: all_a` in the config; the default diff --git a/qt/cli.py b/qt/cli.py index 2d7a2f6..02fff19 100644 --- a/qt/cli.py +++ b/qt/cli.py @@ -229,6 +229,24 @@ def _cmd_data_update(args: argparse.Namespace) -> int: return 0 +def _cmd_data_backfill(args: argparse.Namespace) -> int: + """Backfill the tushare caches over the WIDE historical window (PR-2); no backtest.""" + from qt.data_backfill import format_summary, run_data_backfill + + try: + result = run_data_backfill(args.config) + except (ConfigError, ValueError, FileNotFoundError) as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 1 + print( + f"OK data-backfill: {result.universe_size} symbols in " + f"{result.n_batches} batch(es), failed_batches={result.failed_batches} " + f"({result.elapsed_seconds:.1f}s)\n" + f"{format_summary(result)}" + ) + return 0 + + def _cmd_fetch_data(args: argparse.Namespace) -> int: """Stage helper: run the spine and report the data-fetch stage.""" return _run_pipeline_cmd(args.config, "fetch-data") @@ -295,6 +313,13 @@ def build_parser() -> argparse.ArgumentParser: p_du.add_argument("--config", required=True, help="Path to the YAML config.") p_du.set_defaults(func=_cmd_data_update) + p_bf = sub.add_parser( + "data-backfill", + help="Backfill the tushare raw caches over the WIDE history (PR-2); no backtest.", + ) + p_bf.add_argument("--config", required=True, help="Path to the YAML config.") + p_bf.set_defaults(func=_cmd_data_backfill) + p_i5a = sub.add_parser( "run-phase-i5a-intraday", help="Run the I5a intraday tail-rebalance architecture smoke (minute cache).", diff --git a/qt/config.py b/qt/config.py index f16b601..454908c 100644 --- a/qt/config.py +++ b/qt/config.py @@ -718,6 +718,55 @@ def _check_report_name(cls, v: str) -> str: return name +class BackfillCfg(_Strict): + """PR-2 historical backfill knobs for ``data-backfill`` (all defaults => a + no-op addition: every existing config still validates unchanged). + + ``data-backfill`` is a SEPARATE, manual, long-running command from the nightly + incremental ``data-update``; it warms the SAME caches over a WIDE window. + + * ``start`` — the WIDE backfill window start. The window is ``[start, today]``, + NOT the incremental ``today - lookback_days`` tail (validated as a date). + * ``chunk_size`` — symbols are processed in batches of this size so progress is + durable batch-by-batch, memory stays bounded, and a per-batch failure is + isolated (must be >= 1). + * ``include_minute`` — also backfill 1min bars over the FULL window (a LONG, + resumable operation at all-A scale). ``false`` skips minute history. + """ + + start: str = "2020-01-01" + chunk_size: int = 300 + include_minute: bool = True + + @field_validator("start", mode="before") + @classmethod + def _coerce_date_to_str(cls, v: Any) -> Any: + if isinstance(v, (_date, datetime)): + return v.strftime("%Y-%m-%d") + return v + + @field_validator("start") + @classmethod + def _check_start(cls, v: str) -> str: + try: + datetime.strptime(v, "%Y-%m-%d") + except ValueError as exc: + raise ValueError( + f"data_update.backfill.start must be a 'YYYY-MM-DD' date; " + f"got {v!r} ({exc})." + ) from exc + return v + + @field_validator("chunk_size") + @classmethod + def _check_chunk_size(cls, v: int) -> int: + if v < 1: + raise ValueError( + f"data_update.backfill.chunk_size must be >= 1; got {v}." + ) + return v + + class DataUpdateConcurrencyCfg(_Strict): """D5 opt-in bounded concurrency for ``data-update`` cache warms (default serial). @@ -775,6 +824,9 @@ class DataUpdateCfg(_Strict): concurrency: DataUpdateConcurrencyCfg = Field( default_factory=DataUpdateConcurrencyCfg ) + # PR-2 historical backfill knobs (consumed ONLY by the separate data-backfill + # command; the nightly data-update ignores them). All-defaults => unchanged. + backfill: BackfillCfg = Field(default_factory=BackfillCfg) @field_validator("endpoints", "force_refresh") @classmethod diff --git a/qt/data_backfill.py b/qt/data_backfill.py new file mode 100644 index 0000000..ce862f9 --- /dev/null +++ b/qt/data_backfill.py @@ -0,0 +1,253 @@ +"""Historical backfill of the Tushare read-through caches (PR-2). + +A SEPARATE, manual entry point from :func:`qt.data_updater.run_data_update` (the +21:00 incremental warm). Backfill warms the SAME caches over a WIDE window +``[backfill.start, today]`` — the full history, not the incremental +``today - lookback_days`` tail — and (optionally) the full 1min-bar history rather +than the 7-day tail the nightly job tops up. + +Three properties make it safe to run for hours on the whole A-share market: + +* **Chunked**: symbols are processed in batches of ``backfill.chunk_size``, so + progress is durable batch-by-batch and memory stays bounded. +* **Per-batch failure-tolerant**: a persistent per-batch fetch error (after the + feeds' own retries) is LOGGED (secret-free: batch index + exception TYPE only) + and the run CONTINUES to the next batch. The failed batch's gaps stay + uncovered, so a re-run retries them — consistent with the cache's "a failed + fetch records no coverage" semantic. This DIFFERS on purpose from the nightly + ``run_data_update``, which stays fail-fast (unchanged). +* **Resumable**: resumability is INHERENT to the coverage ledgers — a re-run over + the same window fetches only still-uncovered gaps. Nothing extra is + implemented here; the fixed ``[start, today]`` window keeps re-runs stable. + +Like the updater it NEVER computes factors / builds an alpha or portfolio / runs +a backtest / writes a ``PanelStore``, and the caches store RAW endpoint facts +only (no token, no qfq, no derived flag). +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass, field +from datetime import datetime + +import pandas as pd + +from qt.config import load_config +from qt.data_updater import ( + _build_caches, + _build_feeds, + _build_scheduler, + _resolve_symbols, + _resolve_today, + update_endpoints, +) + +_LOGGER = logging.getLogger("qt.data_backfill") + +# Cap the failed-symbol list carried in the result / summary so a wide failure +# never produces an unbounded (or memory-heavy) DTO or log line. +_MAX_FAILED_SYMBOLS = 200 + + +@dataclass(frozen=True) +class BackfillResult: + """Outcome of a historical backfill run (immutable).""" + + window_start: pd.Timestamp + window_end: pd.Timestamp + universe_size: int + chunk_size: int + n_batches: int + include_minute: bool + endpoints: list[str] + summary: dict[str, dict[str, int]] + failed_batches: int + failed_symbols: list[str] + elapsed_seconds: float = 0.0 + # Concurrency mode of the warm (max_workers=1 == serial; rate_limit is the + # global per-minute cap). Surfaced because a manual multi-hour backfill's + # throughput profile is operationally relevant (mirrors UpdateResult). + max_workers: int = 1 + rate_limit_per_min: int = 0 + notes: list[str] = field(default_factory=list) + + +def _chunk(symbols: list[str], size: int) -> list[list[str]]: + """Split ``symbols`` into contiguous batches of at most ``size`` (order kept).""" + if size < 1: + raise ValueError(f"chunk_size must be >= 1; got {size}.") + return [list(symbols[i : i + size]) for i in range(0, len(symbols), size)] + + +def run_data_backfill(config_path: str, *, today=None) -> BackfillResult: + """Warm the tushare caches over the WIDE historical window from ``config_path``. + + Mirrors ``run_data_update``'s guards (tushare source, external secret, cache + enabled, a ``data_update`` section), reuses the SAME cache + feed construction, + resolves the same universe (all-A / static / index), then warms in per-symbol + batches over ``[backfill.start, today]`` with per-batch failure tolerance. + Returns an immutable :class:`BackfillResult`. + """ + t0 = time.perf_counter() + cfg = load_config(config_path) + du = cfg.data_update + if du is None: + raise ValueError("data-backfill requires a 'data_update' config section.") + if cfg.data.source != "tushare": + raise ValueError( + "data-backfill requires data.source='tushare' (it pulls real data)." + ) + if not cfg.data.external_secret_file: + raise ValueError("data-backfill requires data.external_secret_file (token).") + if not cfg.data.cache.enabled: + raise ValueError("data-backfill requires data.cache.enabled=true.") + + bf = du.backfill + today_ts = _resolve_today(cfg, today) + end = today_ts.strftime("%Y-%m-%d") + start = bf.start # WIDE window start, NOT today - lookback_days + + # Guard the inverted-window footgun: an interval [start, end] with start AFTER + # end subtracts to ZERO gaps everywhere, so a future ``backfill.start`` (a + # config typo) would iterate every batch, fetch nothing, record no coverage, + # and exit 0 "OK" — silent no-op. Fail readably instead (mirrors + # ``DataCfg._check_date_order``). ``start`` is already date-validated by + # ``BackfillCfg``; ``today_ts`` is normalized, so compare against it directly. + if datetime.strptime(start, "%Y-%m-%d") > today_ts.to_pydatetime(): + raise ValueError( + f"data_update.backfill.start ({start}) is after today ({end}); the " + "backfill window is empty (check the date — a future start would " + "fetch nothing and exit OK)." + ) + + scheduler = _build_scheduler(du) + cache, intraday_cache = _build_caches(cfg, du, today_ts) + feeds = _build_feeds( + cfg, cache, intraday_cache, du.rate_limit_per_min, scheduler=scheduler + ) + + symbols = _resolve_symbols(cfg, feeds, start, end) + + # include_minute is the SOLE control of minute warming for backfill: drop any + # configured stk_mins_1min from the dense set, then re-add it (with the FULL + # window + the intraday cache) only when include_minute is on. Minutes are + # warmed over [start, today] — NOT the nightly 7-day tail. + dense_endpoints = [e for e in du.endpoints if e != "stk_mins_1min"] + warm_endpoints = list(dense_endpoints) + if bf.include_minute: + warm_endpoints.append("stk_mins_1min") + intraday_window = ( + f"{start} 00:00:00", + today_ts.strftime("%Y-%m-%d 23:59:59"), + ) + + batches = _chunk(symbols, bf.chunk_size) + n_batches = len(batches) + failed_batches = 0 + failed_symbols: list[str] = [] + + for i, batch in enumerate(batches, start=1): + try: + update_endpoints( + cache, + feeds, + batch, + start=start, + end=end, + endpoints=warm_endpoints, + index_codes=du.index_codes, + fina_fields=du.fina_fields, + sw_level=cfg.processing.neutralize.industry_level, + intraday_cache=intraday_cache if bf.include_minute else None, + intraday_window=intraday_window if bf.include_minute else None, + ) + except Exception as exc: # noqa: BLE001 - tolerate ANY per-batch failure + failed_batches += 1 + if len(failed_symbols) < _MAX_FAILED_SYMBOLS: + failed_symbols.extend( + batch[: _MAX_FAILED_SYMBOLS - len(failed_symbols)] + ) + # Secret-free: batch index + size + exception TYPE only (never the + # token, the fetch args, or the exception message). + _LOGGER.warning( + "backfill batch %d/%d FAILED (%d symbols): %s — skipped; gaps stay " + "uncovered (retryable on re-run)", + i, + n_batches, + len(batch), + type(exc).__name__, + ) + # Progress line, secret-free (cumulative gap-fetch counts across endpoints). + cum_requests = sum(cache.stats().values()) + if bf.include_minute: + cum_requests += sum(intraday_cache.stats().values()) + _LOGGER.info( + "backfill batch %d/%d: %d symbols, cumulative requests=%d, elapsed=%.1fs", + i, + n_batches, + len(batch), + cum_requests, + time.perf_counter() - t0, + ) + + # Build the per-endpoint summary once, from the shared caches, so it is robust + # to a per-batch failure (a failed last batch never drops the tallies). + summary = cache.update_summary() + if bf.include_minute: + st = intraday_cache.stats() + summary["stk_mins_1min"] = { + "requests": int(st.get("stk_mins_1min", 0)), + "rows_written": 0, + "not_ready": 0, + } + + return BackfillResult( + window_start=pd.Timestamp(start), + window_end=pd.Timestamp(end), + universe_size=len(symbols), + chunk_size=bf.chunk_size, + n_batches=n_batches, + include_minute=bf.include_minute, + endpoints=list(warm_endpoints), + summary=summary, + failed_batches=failed_batches, + failed_symbols=failed_symbols, + elapsed_seconds=time.perf_counter() - t0, + max_workers=du.concurrency.max_workers, + rate_limit_per_min=du.rate_limit_per_min, + ) + + +def format_summary(result: BackfillResult) -> str: + """One human line per endpoint + a batch-failure tally (secret-free).""" + mode = "serial" if result.max_workers <= 1 else f"{result.max_workers} workers" + lines = [ + f"data-backfill window [{result.window_start.date()} .. " + f"{result.window_end.date()}], {result.universe_size} symbols in " + f"{result.n_batches} batch(es) of {result.chunk_size} " + f"(include_minute={result.include_minute}; {mode}, global " + f"rate_limit_per_min={result.rate_limit_per_min})" + ] + for ep in result.endpoints: + s = result.summary.get(ep, {}) + lines.append( + f" {ep}: requests={s.get('requests', 0)} " + f"rows_written={s.get('rows_written', 0)} " + f"not_ready={s.get('not_ready', 0)}" + ) + if result.failed_batches: + shown = ", ".join(result.failed_symbols[:10]) + more = ( + "" + if len(result.failed_symbols) <= 10 + else f" (+{len(result.failed_symbols) - 10} more)" + ) + lines.append( + f" FAILED batches: {result.failed_batches}/{result.n_batches}; " + f"failed symbols: {shown}{more} — gaps uncovered, retry by re-running" + ) + else: + lines.append(f" all {result.n_batches} batch(es) OK") + return "\n".join(lines) diff --git a/qt/data_updater.py b/qt/data_updater.py index b5bf1cd..ef55507 100644 --- a/qt/data_updater.py +++ b/qt/data_updater.py @@ -164,6 +164,56 @@ def _resolve_symbols(cfg: RootConfig, feeds: UpdateFeeds, start: str, end: str) return sorted(syms) +def _build_scheduler(du) -> GlobalRateLimiter | None: + """D5: build the ONE shared global rate limiter when bounded concurrency is on. + + ``max_workers == 1`` (serial, the default) returns ``None`` — every feed keeps + its per-call throttle, byte-identical to before; ``> 1`` funnels all workers + through a single per-minute budget (reusing ``rate_limit_per_min``) so the + Tushare quota is never multiplied per thread. Shared by the incremental warm + and the historical backfill so both throttle identically. + """ + if du.concurrency.max_workers > 1: + return GlobalRateLimiter(du.rate_limit_per_min) + return None + + +def _build_caches(cfg: RootConfig, du, today_ts): + """Construct the shared daily + intraday read-through caches (raw-only). + + Both the incremental warm and the historical backfill build the SAME caches + from ``data.cache.root_dir`` — same not-ready pending window, same fina + late-disclosure tail, same coverage semantics, same ``max_workers``. The + caches store RAW endpoint facts only; no token / qfq / factor result is ever + cached. Returns ``(daily_cache, intraday_cache)``. + """ + from data.cache import ( + CacheParquetStore, + CoverageLedger, + IntradayCoverageLedger, + IntradayParquetStore, + TushareCache, + TushareIntradayCache, + ) + + root = cfg.data.cache.root_dir + cache = TushareCache( + CacheParquetStore(root), + CoverageLedger(root), + refresh_recent_days=du.tail_refresh_days, + refresh_dimension_days=cfg.data.cache.refresh_dimension_days, + force_refresh=tuple(du.force_refresh), + today=today_ts, + not_ready_days=du.not_ready_days, + recent_tail_overrides={"fina_indicator": du.fina_tail_days}, + max_workers=du.concurrency.max_workers, + ) + intraday_cache = TushareIntradayCache( + IntradayParquetStore(root), IntradayCoverageLedger(root) + ) + return cache, intraday_cache + + def _build_feeds( cfg: RootConfig, cache, intraday_cache, rate_limit: int, scheduler=None ) -> UpdateFeeds: @@ -234,39 +284,13 @@ def run_data_update(config_path: str, *, today=None) -> UpdateResult: end = today_ts.strftime("%Y-%m-%d") start = (today_ts - pd.Timedelta(days=du.lookback_days)).strftime("%Y-%m-%d") - from data.cache import ( - CacheParquetStore, - CoverageLedger, - IntradayCoverageLedger, - IntradayParquetStore, - TushareCache, - TushareIntradayCache, - ) - # D5: bounded concurrency is opt-in. max_workers==1 (default) builds NO # scheduler — every feed keeps its per-call throttle and the cache stays serial, # byte-identical to before. >1 builds ONE global rate limiter shared by all # feeds (so the quota is global, not per-thread) and a multi-worker cache. max_workers = du.concurrency.max_workers - scheduler = ( - GlobalRateLimiter(du.rate_limit_per_min) if max_workers > 1 else None - ) - - root = cfg.data.cache.root_dir - cache = TushareCache( - CacheParquetStore(root), - CoverageLedger(root), - refresh_recent_days=du.tail_refresh_days, - refresh_dimension_days=cfg.data.cache.refresh_dimension_days, - force_refresh=tuple(du.force_refresh), - today=today_ts, - not_ready_days=du.not_ready_days, - recent_tail_overrides={"fina_indicator": du.fina_tail_days}, - max_workers=max_workers, - ) - intraday_cache = TushareIntradayCache( - IntradayParquetStore(root), IntradayCoverageLedger(root) - ) + scheduler = _build_scheduler(du) + cache, intraday_cache = _build_caches(cfg, du, today_ts) feeds = _build_feeds( cfg, cache, intraday_cache, du.rate_limit_per_min, scheduler=scheduler ) diff --git a/tests/test_data_backfill.py b/tests/test_data_backfill.py new file mode 100644 index 0000000..7faebe2 --- /dev/null +++ b/tests/test_data_backfill.py @@ -0,0 +1,525 @@ +"""PR-2 historical backfill: chunking / failure-tolerance / minute-window / +resumability / all-A universe / config (network-free, FAKE feeds). + +Everything here is a fake — no test hits tushare or reads the real token. The +recorder-style tests fake ``_build_feeds`` + ``update_endpoints`` to capture the +per-batch symbol partitions and windows; the resumability test drives a REAL +read-through cache with a fake tushare client so the coverage ledger genuinely +skips already-covered gaps on the second run. + +The incremental ``data-update`` path is proven UNCHANGED by the existing +``test_data_updater`` / ``test_data_update_all_a`` / ``test_data_update_concurrency`` +suites passing unaltered (they exercise ``run_data_update`` / +``update_endpoints`` / ``_build_feeds`` / ``_resolve_symbols``). +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pandas as pd +import pytest +import yaml +from pydantic import ValidationError + +from data.feed.tushare_feed import TushareFeed +from qt.config import BackfillCfg, ConfigError, DataUpdateCfg, load_config +from qt.data_backfill import BackfillResult, _chunk, format_summary, run_data_backfill +from qt.data_updater import UpdateFeeds + +_CONFIG_DIR = Path(__file__).resolve().parents[1] / "config" + + +# --------------------------------------------------------------------------- # +# config +# --------------------------------------------------------------------------- # +def test_backfill_cfg_defaults(): + bf = BackfillCfg() + assert bf.start == "2020-01-01" + assert bf.chunk_size == 300 + assert bf.include_minute is True + + +def test_data_update_cfg_has_backfill_default(): + # A DataUpdateCfg built with no backfill block gets the all-defaults sub-config. + du = DataUpdateCfg() + assert du.backfill.chunk_size == 300 + assert du.backfill.start == "2020-01-01" + assert du.backfill.include_minute is True + + +def test_existing_data_update_yaml_gets_default_backfill(): + # config/data_update.yaml has NO backfill block -> defaults, still validates. + cfg = load_config(str(_CONFIG_DIR / "data_update.yaml")) + assert cfg.data_update is not None + assert cfg.data_update.backfill.chunk_size == 300 + + +def test_all_a_config_backfill_validates(): + cfg = load_config(str(_CONFIG_DIR / "data_update_all_a.yaml")) + bf = cfg.data_update.backfill + assert bf.start == "2020-01-01" + assert bf.chunk_size == 300 + assert bf.include_minute is True + + +@pytest.mark.parametrize("bad", [0, -1, -50]) +def test_bad_chunk_size_rejected(bad): + with pytest.raises(ValidationError): + BackfillCfg(chunk_size=bad) + + +@pytest.mark.parametrize("bad", ["not-a-date", "2024-13-01", "20240101"]) +def test_bad_start_rejected(bad): + with pytest.raises(ValidationError): + BackfillCfg(start=bad) + + +def test_bad_chunk_size_rejected_via_load_config(tmp_path): + raw = yaml.safe_load((_CONFIG_DIR / "data_update_all_a.yaml").read_text(encoding="utf-8")) + raw["data_update"]["backfill"] = {"start": "2020-01-01", "chunk_size": 0} + p = tmp_path / "bad.yaml" + p.write_text(yaml.safe_dump(raw), encoding="utf-8") + with pytest.raises(ConfigError, match="chunk_size"): + load_config(str(p)) + + +def test_chunk_helper_partitions_in_order(): + assert _chunk(["a", "b", "c", "d", "e"], 2) == [["a", "b"], ["c", "d"], ["e"]] + assert _chunk([], 3) == [] + assert _chunk(["a"], 5) == [["a"]] + with pytest.raises(ValueError, match="chunk_size"): + _chunk(["a"], 0) + + +# --------------------------------------------------------------------------- # +# fakes for the recorder-style run tests +# --------------------------------------------------------------------------- # +class _RecCovariates: + """Fake covariates feed: all_a_symbols() returns the whole (fake) market.""" + + def __init__(self, symbols): + self.symbols = list(symbols) + self.calls = 0 + + def all_a_symbols(self): + self.calls += 1 + return list(self.symbols) + + +class _EndpointRecorder: + """Stand-in for update_endpoints: records each per-batch warm; can raise.""" + + def __init__(self, *, fail_if_contains=None, exc=ConnectionError): + self.calls: list[dict] = [] + self.fail_if_contains = fail_if_contains + self.exc = exc + + def __call__( + self, + cache, + feeds, + symbols, + *, + start, + end, + endpoints, + index_codes, + fina_fields, + sw_level="L1", + intraday_cache=None, + intraday_window=None, + capture=None, + ): + self.calls.append({ + "symbols": list(symbols), + "start": start, + "end": end, + "endpoints": list(endpoints), + "intraday_cache": intraday_cache, + "intraday_window": intraday_window, + }) + if self.fail_if_contains is not None and self.fail_if_contains in symbols: + raise self.exc("transient boom") + return {} + + +def _write_cfg( + tmp_path, + *, + base="data_update_all_a.yaml", + backfill=None, + endpoints=None, + universe=None, + universe_scope=None, + index_codes=None, + tail_refresh_days=None, + not_ready_days=None, +): + raw = yaml.safe_load((_CONFIG_DIR / base).read_text(encoding="utf-8")) + raw["data"]["cache"]["root_dir"] = str(tmp_path / "cache") + if backfill is not None: + raw["data_update"]["backfill"] = backfill + if endpoints is not None: + raw["data_update"]["endpoints"] = endpoints + if universe_scope is not None: + raw["data_update"]["universe_scope"] = universe_scope + if index_codes is not None: + raw["data_update"]["index_codes"] = index_codes + if tail_refresh_days is not None: + raw["data_update"]["tail_refresh_days"] = tail_refresh_days + if not_ready_days is not None: + raw["data_update"]["not_ready_days"] = not_ready_days + if universe is not None: + raw["universe"] = universe + p = tmp_path / "cfg.yaml" + p.write_text(yaml.safe_dump(raw), encoding="utf-8") + return str(p) + + +def _syms(n): + return [f"{i:06d}.SZ" for i in range(1, n + 1)] + + +# --------------------------------------------------------------------------- # +# chunking +# --------------------------------------------------------------------------- # +def test_chunking_partitions_symbols_over_wide_window(tmp_path, monkeypatch): + all_a = _syms(7) + cfg_path = _write_cfg( + tmp_path, + backfill={"start": "2024-01-01", "chunk_size": 3, "include_minute": False}, + endpoints=["market_daily", "adj_factor"], + ) + feeds = UpdateFeeds(covariates=_RecCovariates(all_a)) + monkeypatch.setattr("qt.data_backfill._build_feeds", lambda *a, **k: feeds) + rec = _EndpointRecorder() + monkeypatch.setattr("qt.data_backfill.update_endpoints", rec) + + result = run_data_backfill(cfg_path, today="2024-06-30") + + assert isinstance(result, BackfillResult) + assert result.n_batches == 3 # ceil(7 / 3) + assert [len(c["symbols"]) for c in rec.calls] == [3, 3, 1] + # in-order, exact partition of the resolved universe + assert [s for c in rec.calls for s in c["symbols"]] == all_a + # EVERY batch warmed over the WIDE window [start, today] (not a lookback tail) + for c in rec.calls: + assert c["start"] == "2024-01-01" + assert c["end"] == "2024-06-30" + assert result.window_start == pd.Timestamp("2024-01-01") + assert result.window_end == pd.Timestamp("2024-06-30") + assert result.universe_size == 7 + assert result.failed_batches == 0 + assert result.failed_symbols == [] + + +# --------------------------------------------------------------------------- # +# per-batch failure tolerance (the key PR-2 requirement) +# --------------------------------------------------------------------------- # +def test_batch_failure_is_isolated_and_tallied(tmp_path, monkeypatch): + all_a = _syms(6) # chunk 2 -> 3 batches; batch 2 = 000003 + 000004 + cfg_path = _write_cfg( + tmp_path, + backfill={"start": "2024-01-01", "chunk_size": 2, "include_minute": False}, + endpoints=["market_daily", "adj_factor"], + ) + feeds = UpdateFeeds(covariates=_RecCovariates(all_a)) + monkeypatch.setattr("qt.data_backfill._build_feeds", lambda *a, **k: feeds) + rec = _EndpointRecorder(fail_if_contains="000003.SZ") + monkeypatch.setattr("qt.data_backfill.update_endpoints", rec) + + # must NOT raise + result = run_data_backfill(cfg_path, today="2024-06-30") + + assert result.n_batches == 3 + assert len(rec.calls) == 3 # every batch attempted, INCLUDING those after the fail + assert result.failed_batches == 1 + assert result.failed_symbols == ["000003.SZ", "000004.SZ"] + # the other two batches were still warmed (failure isolated to its batch) + warmed = [c["symbols"] for c in rec.calls] + assert ["000001.SZ", "000002.SZ"] in warmed + assert ["000005.SZ", "000006.SZ"] in warmed + + +def test_failure_summary_is_secret_free_and_tallied(tmp_path, monkeypatch): + cfg_path = _write_cfg( + tmp_path, + backfill={"start": "2024-01-01", "chunk_size": 1, "include_minute": False}, + endpoints=["market_daily", "adj_factor"], + ) + feeds = UpdateFeeds(covariates=_RecCovariates(["000001.SZ", "000002.SZ"])) + monkeypatch.setattr("qt.data_backfill._build_feeds", lambda *a, **k: feeds) + rec = _EndpointRecorder(fail_if_contains="000002.SZ") + monkeypatch.setattr("qt.data_backfill.update_endpoints", rec) + + result = run_data_backfill(cfg_path, today="2024-06-30") + text = format_summary(result) + + assert "FAILED batches: 1/2" in text + assert "000002.SZ" in text # the failed symbol is reported (benign) + # secret-free: no token / config path leaks into the summary + assert "token" not in text.lower() + assert ".config.json" not in text + + +# --------------------------------------------------------------------------- # +# minute full-window control +# --------------------------------------------------------------------------- # +def test_include_minute_warms_full_window(tmp_path, monkeypatch): + cfg_path = _write_cfg( + tmp_path, + backfill={"start": "2022-01-01", "chunk_size": 10, "include_minute": True}, + endpoints=["market_daily", "adj_factor"], + ) + feeds = UpdateFeeds(covariates=_RecCovariates(["000001.SZ", "000002.SZ"])) + monkeypatch.setattr("qt.data_backfill._build_feeds", lambda *a, **k: feeds) + rec = _EndpointRecorder() + monkeypatch.setattr("qt.data_backfill.update_endpoints", rec) + + result = run_data_backfill(cfg_path, today="2024-06-30") + + assert result.n_batches == 1 + c = rec.calls[0] + assert "stk_mins_1min" in c["endpoints"] + assert c["intraday_cache"] is not None + # the FULL window, NOT a 7-day tail + assert c["intraday_window"] == ("2022-01-01 00:00:00", "2024-06-30 23:59:59") + assert "stk_mins_1min" in result.endpoints + assert result.include_minute is True + assert "stk_mins_1min" in result.summary + + +def test_include_minute_false_skips_minutes_even_if_configured(tmp_path, monkeypatch): + # include_minute=False is the SOLE control: a configured stk_mins_1min endpoint + # must NOT trigger minute warming when include_minute is off. + cfg_path = _write_cfg( + tmp_path, + backfill={"start": "2022-01-01", "chunk_size": 10, "include_minute": False}, + endpoints=["market_daily", "adj_factor", "stk_mins_1min"], + ) + feeds = UpdateFeeds(covariates=_RecCovariates(["000001.SZ", "000002.SZ"])) + monkeypatch.setattr("qt.data_backfill._build_feeds", lambda *a, **k: feeds) + rec = _EndpointRecorder() + monkeypatch.setattr("qt.data_backfill.update_endpoints", rec) + + result = run_data_backfill(cfg_path, today="2024-06-30") + + c = rec.calls[0] + assert "stk_mins_1min" not in c["endpoints"] + assert c["intraday_cache"] is None + assert c["intraday_window"] is None + assert "stk_mins_1min" not in result.endpoints + assert result.include_minute is False + + +# --------------------------------------------------------------------------- # +# all-A universe resolution +# --------------------------------------------------------------------------- # +def test_all_a_universe_resolved_and_chunked(tmp_path, monkeypatch): + all_a = _syms(10) + cfg_path = _write_cfg( + tmp_path, + universe_scope="all_a", + backfill={"start": "2024-01-01", "chunk_size": 4, "include_minute": False}, + endpoints=["market_daily", "adj_factor"], + ) + cov = _RecCovariates(all_a) + feeds = UpdateFeeds(covariates=cov) + monkeypatch.setattr("qt.data_backfill._build_feeds", lambda *a, **k: feeds) + rec = _EndpointRecorder() + monkeypatch.setattr("qt.data_backfill.update_endpoints", rec) + + result = run_data_backfill(cfg_path, today="2024-06-30") + + assert cov.calls == 1 # resolved once via all_a_symbols() + assert result.universe_size == 10 + assert result.n_batches == 3 # ceil(10 / 4) + assert [s for c in rec.calls for s in c["symbols"]] == all_a + + +# --------------------------------------------------------------------------- # +# guards +# --------------------------------------------------------------------------- # +def test_backfill_requires_cache_enabled(tmp_path, monkeypatch): + raw = yaml.safe_load((_CONFIG_DIR / "data_update_all_a.yaml").read_text(encoding="utf-8")) + raw["data"]["cache"]["enabled"] = False + raw["data"]["cache"]["root_dir"] = str(tmp_path / "cache") + p = tmp_path / "nocache.yaml" + p.write_text(yaml.safe_dump(raw), encoding="utf-8") + with pytest.raises(ValueError, match="cache.enabled"): + run_data_backfill(str(p), today="2024-06-30") + + +def test_future_start_raises_not_silent_noop(tmp_path): + # A future backfill.start (config typo) would subtract to ZERO gaps everywhere + # -> silent no-op exit-0. The runtime guard must raise readably instead. + cfg_path = _write_cfg( + tmp_path, + backfill={"start": "2099-01-01", "chunk_size": 10, "include_minute": False}, + ) + with pytest.raises(ValueError, match="after today"): + run_data_backfill(cfg_path, today="2024-06-30") + + +# --------------------------------------------------------------------------- # +# resumability (REAL read-through cache + fake tushare client) +# --------------------------------------------------------------------------- # +class _FakeMarketPro: + """Fake tushare client: deterministic daily/adj rows + counters. + + ``fail_symbols`` (mutable) makes ``daily``/``adj_factor`` RAISE for those + ts_codes (a transient error); ``daily_fetched`` records every ts_code a fetch + was attempted for, so a test can assert WHICH symbols hit the API. + """ + + def __init__(self, fail_symbols=()): + self.daily_calls = 0 + self.adj_calls = 0 + self.fail_symbols = set(fail_symbols) + self.daily_fetched: list[str] = [] + self.adj_fetched: list[str] = [] + + def _days(self, start_date, end_date): + return pd.bdate_range(pd.Timestamp(start_date), pd.Timestamp(end_date)) + + def daily(self, ts_code, start_date, end_date, **_): + self.daily_calls += 1 + self.daily_fetched.append(ts_code) + if ts_code in self.fail_symbols: + raise ConnectionError("transient boom") + days = self._days(start_date, end_date) + if len(days) == 0: + return pd.DataFrame( + columns=["ts_code", "trade_date", "open", "high", "low", + "close", "vol", "amount"] + ) + rows = [] + for i, d in enumerate(days): + px = 10.0 + i + rows.append({ + "ts_code": ts_code, "trade_date": d.strftime("%Y%m%d"), + "open": px - 0.2, "high": px + 0.3, "low": px - 0.4, + "close": px, "vol": 1000.0 + i, "amount": 1.0e6 + i, + }) + return pd.DataFrame(rows) + + def adj_factor(self, ts_code, start_date, end_date, **_): + self.adj_calls += 1 + self.adj_fetched.append(ts_code) + if ts_code in self.fail_symbols: + raise ConnectionError("transient boom") + days = self._days(start_date, end_date) + if len(days) == 0: + return pd.DataFrame(columns=["ts_code", "trade_date", "adj_factor"]) + return pd.DataFrame({ + "ts_code": [ts_code] * len(days), + "trade_date": [d.strftime("%Y%m%d") for d in days], + "adj_factor": [1.0] * len(days), + }) + + +def _static_market_cfg(tmp_path, **overrides): + """A config/data_update.yaml-based, config-scope, static-universe, market-only + backfill config (tail/not_ready = 0 so an identical re-run is fully warm).""" + base = dict( + base="data_update.yaml", + universe={ + "type": "static", + "symbols": ["000001.SZ", "000002.SZ"], + "min_listing_days": 60, + "filters": {"missing_close": True, "suspended": False, + "st": False, "limit_up_down": False}, + }, + universe_scope="config", + endpoints=["market_daily", "adj_factor"], + index_codes=[], + tail_refresh_days=0, + not_ready_days=0, + backfill={"start": "2024-01-02", "chunk_size": 1, "include_minute": False}, + ) + base.update(overrides) + return _write_cfg(tmp_path, **base) + + +def test_backfill_second_run_is_warm_zero_fetches(tmp_path, monkeypatch): + # Static universe + config scope so _resolve_symbols needs no feed; endpoints + # only market -> a single fake market feed suffices. tail/not_ready = 0 so the + # identical 2nd run over the SAME fixed [start, today] window is fully warm + # (no moving-tail / pending-day refetch) — isolating the coverage-skip. + import tushare as ts + + fake_pro = _FakeMarketPro() + monkeypatch.setattr(ts, "pro_api", lambda token=None: fake_pro) + + cfg_path = _static_market_cfg(tmp_path) + + fake_secret = tmp_path / "secret.json" + fake_secret.write_text(json.dumps({"tushare": {"token": "FAKE_TOKEN"}}), encoding="utf-8") + + def _fake_build_feeds(cfg, cache, intraday_cache, rate_limit, scheduler=None): + market = TushareFeed(secret_file=str(fake_secret), cache=cache) + return UpdateFeeds(market=market) + + monkeypatch.setattr("qt.data_backfill._build_feeds", _fake_build_feeds) + + r1 = run_data_backfill(cfg_path, today="2024-01-31") + n1 = fake_pro.daily_calls + assert n1 > 0 # cold run: real gap fetches happened + assert r1.failed_batches == 0 + assert r1.summary["market_daily"]["requests"] > 0 + + r2 = run_data_backfill(cfg_path, today="2024-01-31") + assert fake_pro.daily_calls == n1 # warm run: ZERO new daily fetches (all covered) + assert fake_pro.adj_calls # (sanity: it was called at least once overall) + assert r2.failed_batches == 0 + assert r2.summary["market_daily"]["requests"] == 0 # coverage ledger skipped it + assert r2.summary["adj_factor"]["requests"] == 0 + + +def test_failed_batch_reretried_covered_skipped_real_cache(tmp_path, monkeypatch): + # The operator workflow, end to end, against a REAL TushareCache/CoverageLedger: + # run 1 fails for ONE symbol (its batch stays uncovered) while the other symbol + # is durable; run 2 (healthy client) re-fetches ONLY the previously-failed + # symbol and skips the already-covered one. chunk_size=1 isolates each symbol + # to its own batch. The feed's real retry runs, sped up by no-op'ing its sleep. + import tushare as ts + + import data.feed.throttle as throttle + + monkeypatch.setattr(throttle.time, "sleep", lambda *a, **k: None) # instant retries + + fake_pro = _FakeMarketPro(fail_symbols={"000002.SZ"}) + monkeypatch.setattr(ts, "pro_api", lambda token=None: fake_pro) + + cfg_path = _static_market_cfg(tmp_path) + fake_secret = tmp_path / "secret.json" + fake_secret.write_text(json.dumps({"tushare": {"token": "FAKE_TOKEN"}}), encoding="utf-8") + + def _fake_build_feeds(cfg, cache, intraday_cache, rate_limit, scheduler=None): + return UpdateFeeds(market=TushareFeed(secret_file=str(fake_secret), cache=cache)) + + monkeypatch.setattr("qt.data_backfill._build_feeds", _fake_build_feeds) + + # RUN 1: 000002 fails (uncovered, retryable); 000001 succeeds (durable). + r1 = run_data_backfill(cfg_path, today="2024-01-31") + assert r1.failed_batches == 1 + assert r1.failed_symbols == ["000002.SZ"] + assert "000001.SZ" in fake_pro.daily_fetched # the good one was fetched+stored + + # RUN 2: healthy client. Only the previously-failed symbol's gap is refetched. + fake_pro.fail_symbols = set() + fake_pro.daily_fetched.clear() + fake_pro.adj_fetched.clear() + r2 = run_data_backfill(cfg_path, today="2024-01-31") + + assert r2.failed_batches == 0 + assert "000002.SZ" in fake_pro.daily_fetched # previously-failed -> refetched + assert "000001.SZ" not in fake_pro.daily_fetched # already covered -> skipped + assert "000001.SZ" not in fake_pro.adj_fetched + # after run 2 the whole universe is durable: a 3rd run is fully warm. + fake_pro.daily_fetched.clear() + r3 = run_data_backfill(cfg_path, today="2024-01-31") + assert fake_pro.daily_fetched == [] + assert r3.failed_batches == 0