From 3c65c72a729ae12fb061bfd55bfc2affd2f4d722 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Tue, 24 Mar 2026 08:25:08 -0700 Subject: [PATCH] Auto-refresh snapshot when paper backtest end_date exceeds snapshot coverage When `fithia2 paper backtest --end ` requests a date beyond the snapshot's latest event, automatically runs the pipeline: 1. Filing poller (discover new 8-Ks) 2. Filing fetcher (download exhibits) 3. Event parser (parse events) 4. Feature builder (compute features) 5. Label generator (compute labels) 6. Dataset export (re-generate Parquet snapshot) Staleness check: snapshot is stale if its latest event_date is >14 days before the requested end_date, or if the manifest is >7 days old. If refresh fails, falls back to existing snapshot data gracefully. Co-Authored-By: Claude Opus 4.6 (1M context) --- apps/paper_trader/backtest_sim.py | 149 +++++++++++++++++++++++++++++- 1 file changed, 148 insertions(+), 1 deletion(-) diff --git a/apps/paper_trader/backtest_sim.py b/apps/paper_trader/backtest_sim.py index 9c74536..393ca12 100644 --- a/apps/paper_trader/backtest_sim.py +++ b/apps/paper_trader/backtest_sim.py @@ -181,6 +181,126 @@ def _convert_from_runner( } +def _snapshot_needs_refresh( + snapshot_id: str, + end_date: dt.date, + snapshot_dir: str = "data/datasets/snapshots", +) -> bool: + """Check if the Parquet snapshot is stale (doesn't cover end_date).""" + import json + + manifest_path = Path(snapshot_dir) / snapshot_id / "manifest.json" + if not manifest_path.exists(): + return True + + try: + manifest = json.loads(manifest_path.read_text()) + created = manifest.get("created_at_utc", "")[:10] + if created and dt.date.fromisoformat(created) < end_date - dt.timedelta(days=7): + return True + except Exception: + return True + + # Check if the latest event_date in the data covers end_date + train_path = Path(snapshot_dir) / snapshot_id / "train.parquet" + test_path = Path(snapshot_dir) / snapshot_id / "test.parquet" + latest_path = test_path if test_path.exists() else train_path + if not latest_path.exists(): + return True + + try: + import pyarrow.parquet as pq + table = pq.read_table(str(latest_path), columns=["event_date"]) + dates = table.column("event_date").to_pylist() + max_date = max(dates) if dates else "" + if isinstance(max_date, str): + max_date = dt.date.fromisoformat(max_date[:10]) + # Stale if snapshot's latest event is more than 14 days before end_date + return max_date < end_date - dt.timedelta(days=14) + except Exception: + return True + + +async def _refresh_snapshot( + snapshot_id: str, + universe_profile: str | None, + console=None, +) -> None: + """Re-run pipeline steps and re-export the snapshot.""" + if console: + console.print("\n[bold yellow]Snapshot stale — refreshing pipeline...[/]") + + # Step 1: Run pending pipeline steps + if console: + console.print(" [dim]1/4 Polling new filings...[/]") + try: + from apps.pipeline.filing_poller.main import poll_filings + from libs.common.ids import new_job_run_id + await poll_filings(new_job_run_id()) + except Exception as exc: + if console: + console.print(f" [yellow]Filing poller skipped: {exc}[/]") + + if console: + console.print(" [dim]2/4 Fetching exhibits...[/]") + try: + from apps.pipeline.filing_fetcher.main import fetch_exhibits + from libs.common.ids import new_job_run_id + await fetch_exhibits(new_job_run_id()) + except Exception as exc: + if console: + console.print(f" [yellow]Fetcher skipped: {exc}[/]") + + if console: + console.print(" [dim]3/4 Parsing events & building features...[/]") + try: + from apps.pipeline.event_parser.main import run_event_parser + from libs.common.ids import new_job_run_id + await run_event_parser(new_job_run_id()) + except Exception as exc: + if console: + console.print(f" [yellow]Parser skipped: {exc}[/]") + + try: + from apps.pipeline.feature_builder.main import run_feature_builder + from libs.common.ids import new_job_run_id + await run_feature_builder(new_job_run_id()) + except Exception as exc: + if console: + console.print(f" [yellow]Feature builder skipped: {exc}[/]") + + try: + from apps.pipeline.label_generator.main import run_label_generator + from libs.common.ids import new_job_run_id + await run_label_generator(new_job_run_id()) + except Exception as exc: + if console: + console.print(f" [yellow]Label generator skipped: {exc}[/]") + + # Step 2: Re-export snapshot + if console: + console.print(" [dim]4/4 Exporting snapshot...[/]") + try: + from libs.db.session import get_session + from libs.export.snapshot_export import export_dataset_snapshot + + async with get_session() as session: + await export_dataset_snapshot( + session=session, + snapshot_id=snapshot_id, + split_policy="temporal_70_15_15", + output_dir="data/datasets/snapshots", + feature_versions=["market_v1", "event_v1"], + universe_profile=universe_profile, + ) + if console: + console.print(" [green]Snapshot refreshed.[/]") + except Exception as exc: + if console: + console.print(f" [red]Snapshot export failed: {exc}[/]") + raise + + async def run_backtest( configs: list[str], capital: float, @@ -190,7 +310,11 @@ async def run_backtest( oracle_url: str, console=None, ) -> list[dict[str, Any]]: - """Run multiple strategies sequentially using BacktestRunner.""" + """Run multiple strategies sequentially using BacktestRunner. + + Automatically refreshes the Parquet snapshot if it doesn't cover + the requested end_date (runs pipeline + re-export). + """ from libs.common.time_utils import is_trading_day from libs.common.logging import configure_logging @@ -207,6 +331,29 @@ async def run_backtest( console.print(f"[bold]Trading days:[/] {trading_days[0]} → {trading_days[-1]} ({len(trading_days)} days)") console.print("[bold]Engine:[/] BacktestRunner (identical to research backtester)") + # Check if snapshots need refresh for each config + for config_path in configs: + from apps.backtester.run import load_manifest, resolve_config + manifest = load_manifest(config_path) + config = resolve_config(manifest) + snapshot_id = config.dataset_snapshot_id + + if _snapshot_needs_refresh(snapshot_id, end_date): + # Determine universe_profile from snapshot export config + universe_profile = None + if "midlarge" in snapshot_id: + universe_profile = "midlarge-liquid-long-v1" + elif "midwide" in snapshot_id: + universe_profile = "midwide-liquid-long-v1" + elif "smallcap" in snapshot_id: + universe_profile = "smallcap-liquid-long-v1" + + try: + await _refresh_snapshot(snapshot_id, universe_profile, console=console) + except Exception as exc: + if console: + console.print(f"[yellow]Snapshot refresh failed, using existing data: {exc}[/]") + configure_logging("WARNING") results = []