From 86d55e01f9dead1037b9337ef80b714b84238e3b Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Wed, 1 Apr 2026 01:26:55 -0700 Subject: [PATCH] Fix PIT snapshot store regressions for backtests --- libs/backtest/selector.py | 82 +- libs/backtest/snapshot_store.py | 975 ++++++++++++++++++++- tests/unit/backtest/test_selector.py | 117 +++ tests/unit/backtest/test_snapshot_store.py | 348 +++++++- 4 files changed, 1485 insertions(+), 37 deletions(-) diff --git a/libs/backtest/selector.py b/libs/backtest/selector.py index 1ac756c..f3e077d 100644 --- a/libs/backtest/selector.py +++ b/libs/backtest/selector.py @@ -87,9 +87,58 @@ def build_candidate( ): return None + source_symbol = str(row.get("source_symbol") or row.get("symbol", row.get("ticker", ""))).upper() + trade_symbol_mode = ( + strategy_engine.trade_symbol_mode + if strategy_engine is not None + else "event" + ) + trade_symbol = source_symbol + execution_entry_price = row.get("entry_price") or row.get("entry_price_est") + execution_event_close = row.get("event_close") + execution_avg_dollar_volume = row.get( + "avg_dollar_volume_20d", + row.get("avg_dollar_volume", 0.0), + ) + execution_atr_14_raw = row.get("atr_14") + execution_features = {k: v for k, v in row.items() if k not in _RESERVED_KEYS} + + if strategy_engine and trade_symbol_mode == "sector_etf": + proxy_symbol = str(row.get("sector_etf_proxy") or "").upper() + if not proxy_symbol: + logger.debug( + "skip_candidate_no_sector_etf_proxy", + event_id=event_id, + source_symbol=source_symbol, + engine_id=strategy_engine.engine_id, + ) + return None + trade_symbol = proxy_symbol + execution_entry_price = row.get("sector_etf_entry_price") or execution_entry_price + execution_event_close = row.get("sector_etf_event_close") or execution_event_close + execution_avg_dollar_volume = ( + row.get("sector_etf_avg_dollar_volume") + or execution_avg_dollar_volume + ) + execution_atr_14_raw = row.get("sector_etf_atr_14") + execution_features.update( + { + "source_symbol": source_symbol, + "trade_symbol_mode": trade_symbol_mode, + "proxy_trade_symbol": trade_symbol, + "proxy_reference_sector": row.get("sector"), + "source_event_close": row.get("event_close"), + "source_reaction_day_low": row.get("reaction_day_low"), + "source_reaction_day_high": row.get("reaction_day_high"), + "event_close": row.get("sector_etf_event_close"), + "reaction_day_low": row.get("sector_etf_reaction_day_low"), + "reaction_day_high": row.get("sector_etf_reaction_day_high"), + } + ) + if strategy_engine and strategy_engine.entry_timing_policy == "reaction_close": execution_date = reaction_date - entry_price_est = row.get("event_close") or row.get("entry_price_est") + entry_price_est = execution_event_close or row.get("entry_price_est") if not entry_price_est: logger.debug( "skip_candidate_no_event_close", @@ -98,7 +147,7 @@ def build_candidate( ) return None else: - entry_price_est = row.get("entry_price") or row.get("entry_price_est") + entry_price_est = execution_entry_price if not entry_price_est: logger.warning("skip_candidate_no_entry_price", event_id=event_id) @@ -109,8 +158,8 @@ def build_candidate( return None score = float(row.get("score", 0.0)) - avg_dollar_volume = float(row.get("avg_dollar_volume", 0.0)) - atr_14_raw = row.get("atr_14") + avg_dollar_volume = float(execution_avg_dollar_volume or 0.0) + atr_14_raw = execution_atr_14_raw atr_14 = float(atr_14_raw) if atr_14_raw is not None else None event_direction = str(row.get("event_direction", "")).lower() guidance_status = str(row.get("guidance_status", "")).lower() @@ -229,7 +278,8 @@ def build_candidate( return Candidate( event_id=event_id, - symbol=str(row.get("symbol", row.get("ticker", ""))), + symbol=trade_symbol, + source_symbol=source_symbol, issuer_id=row.get("issuer_id"), score=score, sector=str(row.get("sector") or "UNKNOWN"), @@ -356,6 +406,7 @@ def build_candidate( if strategy_engine else None ), + trade_symbol_mode=trade_symbol_mode, trade_direction=trade_direction, parent_position_id=row.get("parent_position_id"), is_add_on=bool(row.get("is_add_on", False)), @@ -364,7 +415,7 @@ def build_candidate( if row.get("forced_shares") not in (None, "") else None ), - features={k: v for k, v in row.items() if k not in _RESERVED_KEYS}, + features=execution_features, ) @@ -1056,11 +1107,8 @@ def select_candidates( excluded_symbols = {symbol.upper() for symbol in (excluded_symbols or set())} for row in raw_rows: event_id = str(row.get("event_id", "")) - symbol = str(row.get("symbol", row.get("ticker", ""))).upper() if event_id and event_id in excluded_event_ids: continue - if symbol and symbol in excluded_symbols: - continue prepared_row = _prepare_row_for_strategy_engine( row, signal_config=signal_config, @@ -1072,6 +1120,8 @@ def select_candidates( engine_lookup=engine_lookup, ) if c is not None: + if c.symbol and c.symbol.upper() in excluded_symbols: + continue candidates.append(c) candidates = filter_by_universe(candidates, universe_config) @@ -1086,6 +1136,8 @@ def select_candidates( if event_type_profiles: candidates = filter_by_event_type(candidates, event_type_profiles) candidates = rank_candidates(candidates, signal_config.ranking_fields) + if strategy_engine and strategy_engine.trade_symbol_mode != "event": + candidates = _dedupe_candidates_by_symbol(candidates) candidates = truncate_candidates( candidates, truncate_to if truncate_to is not None else signal_config.max_candidates_per_day, @@ -1093,6 +1145,18 @@ def select_candidates( return candidates +def _dedupe_candidates_by_symbol(candidates: list[Candidate]) -> list[Candidate]: + seen_symbols: set[str] = set() + deduped: list[Candidate] = [] + for candidate in candidates: + symbol = candidate.symbol.upper() + if symbol in seen_symbols: + continue + seen_symbols.add(symbol) + deduped.append(candidate) + return deduped + + def _resolve_score_threshold( signal_config: SignalConfig, strategy_engine: StrategyEngineConfig | None, diff --git a/libs/backtest/snapshot_store.py b/libs/backtest/snapshot_store.py index a0a7f8f..098c5c5 100644 --- a/libs/backtest/snapshot_store.py +++ b/libs/backtest/snapshot_store.py @@ -13,18 +13,46 @@ from __future__ import annotations import asyncio import datetime as dt +import hashlib import json +import os +from functools import partial +import pickle from pathlib import Path +import time from typing import Any from zoneinfo import ZoneInfo +import pyarrow as pa import pyarrow.parquet as pq +from libs.backtest.proxies import sector_etf_for_sector +from libs.features.market_features import compute_market_features from libs.common.config import get_settings from libs.common.logging import get_logger logger = get_logger(__name__) _UTC = ZoneInfo("UTC") +_PRICE_DERIVED_FEATURE_COLUMNS = ( + "pre_event_volatility_20d", + "pre_event_rsi_14", + "pre_event_bb_position", + "pre_event_obv_slope_20d", + "pre_event_hurst_60d", + "pre_event_entropy_60d", + "pre_event_ou_theta_60d", + "pre_event_gravitational_pull", + "pre_event_market_temperature", +) +_MACRO_BACKFILL_FEATURE_COLUMNS = ( + "macro_vix", + "macro_hy_spread", + "macro_t10y2y", +) +_PRICE_FEATURE_WARMUP_DAYS = 120 +_RUNTIME_CACHE_VERSION = 3 +_RUNTIME_CACHE_WAIT_TIMEOUT_SECONDS = 1800.0 +_RUNTIME_CACHE_WAIT_LOG_INTERVAL_SECONDS = 5.0 class SnapshotStore: @@ -183,10 +211,78 @@ class SnapshotStore: "Use SnapshotStore._async_load() directly in async contexts." ) - data = asyncio.run( - cls._async_load(Path(snapshot_dir), split_name, oracle_url, db_dsn, scoring_fn) + snapshot_path = Path(snapshot_dir) + return cls._load_with_runtime_cache( + snapshot_path=snapshot_path, + split_names=[split_name], + scoring_fn=scoring_fn, + builder=lambda: asyncio.run( + cls._async_load(snapshot_path, split_name, oracle_url, db_dsn, scoring_fn) + ), + ) + + @classmethod + def load_merged( + cls, + snapshot_dir: str | Path, + split_names: list[str], + oracle_url: str, + db_dsn: str, + scoring_fn: Any | None = None, + ) -> "SnapshotStore": + """Load multiple splits in one pass to avoid duplicate DB/Oracle fetches.""" + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = None + + if loop is not None and loop.is_running(): + raise RuntimeError( + "SnapshotStore.load_merged() cannot be called from a running event loop. " + "Use SnapshotStore._async_load_merged() directly in async contexts." + ) + + snapshot_path = Path(snapshot_dir) + return cls._load_with_runtime_cache( + snapshot_path=snapshot_path, + split_names=split_names, + scoring_fn=scoring_fn, + builder=lambda: asyncio.run( + cls._async_load_merged(snapshot_path, split_names, oracle_url, db_dsn, scoring_fn) + ), + ) + + @classmethod + def materialize_snapshot_dir( + cls, + snapshot_dir: str | Path, + oracle_url: str, + db_dsn: str, + *, + split_names: list[str] | None = None, + output_dir: str | Path | None = None, + ) -> list[str]: + """Persist runtime-backfilled feature columns into snapshot parquet files.""" + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = None + + if loop is not None and loop.is_running(): + raise RuntimeError( + "SnapshotStore.materialize_snapshot_dir() cannot be called from a running event loop. " + "Use SnapshotStore._async_materialize_snapshot_dir() directly in async contexts." + ) + + return asyncio.run( + cls._async_materialize_snapshot_dir( + Path(snapshot_dir), + oracle_url=oracle_url, + db_dsn=db_dsn, + split_names=split_names, + output_dir=Path(output_dir) if output_dir is not None else None, + ) ) - return cls(**data) @classmethod async def _async_load( @@ -202,23 +298,248 @@ class SnapshotStore: if not parquet_path.exists(): raise FileNotFoundError(f"Parquet file not found: {parquet_path}") - # Step 1: Read Parquet + row_list = cls._read_parquet_rows(parquet_path) + logger.info("snapshot_store_rows_loaded", split=split_name, count=len(row_list)) + + return await cls._build_init_kwargs_from_rows( + row_list=row_list, + oracle_url=oracle_url, + db_dsn=db_dsn, + scoring_fn=scoring_fn, + ) + + @classmethod + async def _async_load_merged( + cls, + snapshot_dir: Path, + split_names: list[str], + oracle_url: str, + db_dsn: str, + scoring_fn: Any | None = None, + ) -> dict[str, Any]: + row_list: list[dict[str, Any]] = [] + loaded_splits: list[str] = [] + for split_name in split_names: + parquet_path = snapshot_dir / f"{split_name}.parquet" + if not parquet_path.exists(): + continue + loaded_splits.append(split_name) + row_list.extend(cls._read_parquet_rows(parquet_path)) + + if not loaded_splits: + raise FileNotFoundError("No snapshot splits found.") + + logger.info( + "snapshot_store_rows_loaded", + split="merged", + loaded_splits=loaded_splits, + count=len(row_list), + ) + + return await cls._build_init_kwargs_from_rows( + row_list=row_list, + oracle_url=oracle_url, + db_dsn=db_dsn, + scoring_fn=scoring_fn, + ) + + @classmethod + async def _async_materialize_snapshot_dir( + cls, + snapshot_dir: Path, + *, + oracle_url: str, + db_dsn: str, + split_names: list[str] | None = None, + output_dir: Path | None = None, + ) -> list[str]: + split_names = list(split_names or ["train", "valid", "test"]) + output_dir = output_dir or snapshot_dir + output_dir.mkdir(parents=True, exist_ok=True) + + tables_by_split: dict[str, pa.Table] = {} + rows_by_split: dict[str, list[dict[str, Any]]] = {} + row_list: list[dict[str, Any]] = [] + needed_columns: set[str] = set() + + for split_name in split_names: + parquet_path = snapshot_dir / f"{split_name}.parquet" + if not parquet_path.exists(): + continue + table = pq.read_table(str(parquet_path)) + rows = table.to_pylist() + tables_by_split[split_name] = table + rows_by_split[split_name] = rows + row_list.extend(rows) + for column in (*_PRICE_DERIVED_FEATURE_COLUMNS, *_MACRO_BACKFILL_FEATURE_COLUMNS): + if column not in table.column_names or table.column(column).null_count > 0: + needed_columns.add(column) + + if not tables_by_split: + raise FileNotFoundError(f"No snapshot splits found under {snapshot_dir}") + + if not needed_columns: + if output_dir != snapshot_dir: + for split_name in tables_by_split: + pq.write_table(tables_by_split[split_name], output_dir / f"{split_name}.parquet") + manifest_path = snapshot_dir / "manifest.json" + if manifest_path.exists(): + (output_dir / "manifest.json").write_text(manifest_path.read_text()) + return [] + + date_range = cls._compute_date_range(row_list) + unique_symbols = sorted({ + str(symbol).upper() + for symbol in ( + row.get("ticker") or row.get("symbol") + for row in row_list + ) + if symbol + }) + + bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]] = {} + macro_by_date: dict[dt.date, dict[str, Any]] = {} + if any(column in needed_columns for column in _PRICE_DERIVED_FEATURE_COLUMNS): + bars_by_symbol, _ = await cls._fetch_price_data(unique_symbols, date_range, oracle_url) + if any(column in needed_columns for column in _MACRO_BACKFILL_FEATURE_COLUMNS): + macro_by_date = await cls._fetch_macro(date_range, db_dsn) + + _vix_pairs = sorted( + (d, v["VIXCLS"]) + for d, v in macro_by_date.items() + if "VIXCLS" in v and v["VIXCLS"] is not None + ) + _hy_series_key = next( + (k for k in next(iter(macro_by_date.values()), {}) if "BAMLH0" in k), + None, + ) if macro_by_date else None + _hy_pairs = sorted( + (d, v[_hy_series_key]) + for d, v in macro_by_date.items() + if _hy_series_key and _hy_series_key in v and v[_hy_series_key] is not None + ) if _hy_series_key else [] + _t10y2y_pairs = sorted( + (d, v["T10Y2Y"]) + for d, v in macro_by_date.items() + if "T10Y2Y" in v and v["T10Y2Y"] is not None + ) + + price_bar_cache: dict[str, list[Any]] = {} + derived_feature_cache: dict[tuple[str, dt.date], dict[str, Any]] = {} + materialized_columns: set[str] = set() + + for split_name, table in tables_by_split.items(): + rows = rows_by_split[split_name] + for row in rows: + symbol = (row.get("ticker") or row.get("symbol") or "") + look_date = ( + cls._normalize_date(row.get("event_date")) + or cls._normalize_date(row.get("execution_date")) + or cls._normalize_date(row.get("entry_date")) + ) + if symbol and look_date: + if any(column in needed_columns for column in _PRICE_DERIVED_FEATURE_COLUMNS): + cls._backfill_price_derived_features( + row, + symbol=str(symbol).upper(), + event_date=look_date, + bars_by_symbol=bars_by_symbol, + price_bar_cache=price_bar_cache, + derived_feature_cache=derived_feature_cache, + ) + if "macro_vix" in needed_columns and row.get("macro_vix") is None: + value = cls._lookup_as_of(_vix_pairs, look_date) + if value is not None: + row["macro_vix"] = value + if "macro_hy_spread" in needed_columns and row.get("macro_hy_spread") is None: + value = cls._lookup_as_of(_hy_pairs, look_date) + if value is not None: + row["macro_hy_spread"] = value + if "macro_t10y2y" in needed_columns and row.get("macro_t10y2y") is None: + value = cls._lookup_as_of(_t10y2y_pairs, look_date) + if value is not None: + row["macro_t10y2y"] = value + + updated_table = table + for column in sorted(needed_columns): + values = [row.get(column) for row in rows] + if not any(value is not None for value in values): + continue + array = pa.array(values, type=pa.float64()) + if column in updated_table.column_names: + idx = updated_table.column_names.index(column) + updated_table = updated_table.set_column(idx, column, array) + else: + updated_table = updated_table.append_column(column, array) + materialized_columns.add(column) + + pq.write_table(updated_table, output_dir / f"{split_name}.parquet") + + manifest_path = snapshot_dir / "manifest.json" + if manifest_path.exists(): + manifest = json.loads(manifest_path.read_text()) + manifest["output_dir"] = str(output_dir.resolve()) + manifest["materialized_feature_columns"] = sorted(materialized_columns) + manifest["materialized_feature_last_refresh_utc"] = dt.datetime.now(dt.UTC).isoformat() + (output_dir / "manifest.json").write_text(json.dumps(manifest, indent=2)) + + return sorted(materialized_columns) + + @staticmethod + def _read_parquet_rows(parquet_path: Path) -> list[dict[str, Any]]: logger.info("snapshot_store_reading_parquet", path=str(parquet_path)) table = pq.read_table(str(parquet_path)) - rows: list[dict[str, Any]] = table.to_pydict() - # Convert column-oriented dict to list of row dicts + cols = table.to_pydict() num_rows = table.num_rows - col_names = list(rows.keys()) - row_list: list[dict[str, Any]] = [ - {col: rows[col][i] for col in col_names} for i in range(num_rows) - ] - logger.info("snapshot_store_rows_loaded", count=num_rows) + col_names = list(cols.keys()) + return [{col: cols[col][i] for col in col_names} for i in range(num_rows)] + + @classmethod + async def _build_init_kwargs_from_rows( + cls, + *, + row_list: list[dict[str, Any]], + oracle_url: str, + db_dsn: str, + scoring_fn: Any | None = None, + ) -> dict[str, Any]: + raw_row_count = len(row_list) + + # De-duplicate merged-boundary overlaps before any expensive fetches. + deduped_rows: list[dict[str, Any]] = [] + seen_keys: set[tuple[Any, ...]] = set() + for row in row_list: + key = ( + row.get("event_id"), + row.get("ticker") or row.get("symbol"), + row.get("entry_date") or row.get("execution_date"), + row.get("reaction_date"), + ) + if key in seen_keys: + continue + seen_keys.add(key) + deduped_rows.append(row) + row_list = deduped_rows + logger.info( + "snapshot_store_prepare_rows", + raw_rows=raw_row_count, + deduped_rows=len(row_list), + ) # Collect event_ids for DB lookup event_ids = [str(r.get("event_id", "")) for r in row_list] + date_range = cls._compute_date_range(row_list) # Step 2–6: DB + Oracle enrichment + logger.info( + "snapshot_store_event_metadata_fetch_start", + event_count=len(event_ids), + ) event_meta = await cls._fetch_event_metadata(event_ids, db_dsn) + logger.info( + "snapshot_store_event_metadata_fetch_done", + resolved_events=len(event_meta), + ) unique_symbols = sorted({ str(symbol).upper() for symbol in ( @@ -228,18 +549,61 @@ class SnapshotStore: ) if symbol }) - - date_range = cls._compute_date_range(row_list) - bars_by_symbol, avg_dvol = await cls._fetch_price_data( - unique_symbols, date_range, oracle_url + logger.info( + "snapshot_store_sector_fetch_start", + symbol_count=len(unique_symbols), ) sectors = await cls._fetch_sectors(unique_symbols, oracle_url) + logger.info( + "snapshot_store_sector_fetch_done", + symbol_count=len(sectors), + ) + proxy_symbols = sorted( + { + proxy_symbol + for proxy_symbol in ( + sector_etf_for_sector(sectors.get(symbol)) + for symbol in unique_symbols + ) + if proxy_symbol + } + ) + price_symbols = sorted(set(unique_symbols) | set(proxy_symbols)) + logger.info( + "snapshot_store_enrichment_plan", + symbol_count=len(unique_symbols), + trade_symbol_count=len(price_symbols), + start_date=date_range[0].isoformat() if date_range else None, + end_date=date_range[1].isoformat() if date_range else None, + ) + logger.info( + "snapshot_store_price_fetch_start", + symbol_count=len(price_symbols), + ) + bars_by_symbol, avg_dvol = await cls._fetch_price_data( + price_symbols, date_range, oracle_url + ) + logger.info( + "snapshot_store_price_fetch_done", + symbol_count=len(bars_by_symbol), + avg_dvol_symbols=sum(1 for value in avg_dvol.values() if value > 0), + ) + logger.info("snapshot_store_macro_fetch_start") macro_by_date = await cls._fetch_macro(date_range, db_dsn) + logger.info( + "snapshot_store_macro_fetch_done", + macro_dates=len(macro_by_date), + ) # Fetch SPY bars for macro regime filter (SMA computation) + logger.info("snapshot_store_spy_macro_fetch_start") spy_macro = await cls._fetch_spy_macro(date_range, oracle_url) for d, spy_data in spy_macro.items(): macro_by_date.setdefault(d, {}).update(spy_data) + logger.info( + "snapshot_store_spy_macro_fetch_done", + spy_dates=len(spy_macro), + ) # Compute VIX change rates for composite parking gate sorted_macro_dates = sorted(macro_by_date.keys()) @@ -270,6 +634,8 @@ class SnapshotStore: for d, v in macro_by_date.items() if _hy_series_key and _hy_series_key in v and v[_hy_series_key] is not None ) if _hy_series_key else [] + price_bar_cache: dict[str, list[Any]] = {} + derived_feature_cache: dict[tuple[str, dt.date], dict[str, Any]] = {} def _lookup_fred_as_of(pairs: list, as_of: dt.date) -> float | None: """Return most recent FRED value on or before as_of date.""" @@ -285,7 +651,9 @@ class SnapshotStore: return result candidates_by_exec_date: dict[dt.date, list[dict[str, Any]]] = {} - for row in row_list: + total_rows = len(row_list) + progress_step = max(1000, total_rows // 5) if total_rows else 1000 + for idx, row in enumerate(row_list, start=1): eid = str(row.get("event_id", "")) meta = event_meta.get(eid, {}) ticker = meta.get("ticker") or row.get("ticker") or row.get("symbol") @@ -315,7 +683,6 @@ class SnapshotStore: meta.get("event_timestamp") or cls._normalize_timestamp(row.get("event_timestamp"), fallback_event_date) ) - enriched["avg_dollar_volume"] = avg_dvol.get(ticker, 0.0) enriched["sector"] = sectors.get(ticker, "UNKNOWN") # Backfill macro features from FRED data when absent in Parquet @@ -329,6 +696,43 @@ class SnapshotStore: if enriched.get("macro_hy_spread") is None and _hy_pairs: enriched["macro_hy_spread"] = _lookup_fred_as_of(_hy_pairs, look_date) + if look_date is not None: + cls._backfill_price_derived_features( + enriched, + symbol=ticker, + event_date=look_date, + bars_by_symbol=bars_by_symbol, + price_bar_cache=price_bar_cache, + derived_feature_cache=derived_feature_cache, + ) + cls._backfill_avg_dollar_volume_features( + enriched, + symbol=ticker, + event_date=look_date, + bars_by_symbol=bars_by_symbol, + price_bar_cache=price_bar_cache, + fallback_avg_dvol=avg_dvol.get(ticker, 0.0), + ) + else: + cls._backfill_avg_dollar_volume_features( + enriched, + symbol=ticker, + event_date=None, + bars_by_symbol=bars_by_symbol, + price_bar_cache=price_bar_cache, + fallback_avg_dvol=avg_dvol.get(ticker, 0.0), + ) + reaction_date = cls._normalize_date(enriched.get("reaction_date")) or exec_date + cls._attach_sector_etf_proxy_features( + enriched, + sector=enriched.get("sector"), + reaction_date=reaction_date, + execution_date=exec_date, + bars_by_symbol=bars_by_symbol, + avg_dvol=avg_dvol, + price_bar_cache=price_bar_cache, + ) + # Map Parquet-specific columns to canonical backtest names # event_close (reaction-day close) → entry_price_est baseline if "entry_price_est" not in enriched and "event_close" in enriched: @@ -343,6 +747,12 @@ class SnapshotStore: enriched["score"] = compute_entry_score(enriched) candidates_by_exec_date.setdefault(exec_date, []).append(enriched) + if idx % progress_step == 0 or idx == total_rows: + logger.info( + "snapshot_store_candidate_build_progress", + processed=idx, + total=total_rows, + ) # Build bars_by_symbol_date: symbol -> date -> bar dict bars_by_symbol_date: dict[str, dict[dt.date, dict[str, Any]]] = {} @@ -370,6 +780,7 @@ class SnapshotStore: "snapshot_store_built", exec_dates=len(candidates_by_exec_date), symbols=len(bars_by_symbol_date), + candidates=sum(len(rows) for rows in candidates_by_exec_date.values()), ) return { "candidates_by_exec_date": candidates_by_exec_date, @@ -381,6 +792,315 @@ class SnapshotStore: # Internal async helpers # ------------------------------------------------------------------ + @staticmethod + def _runtime_cache_dir(snapshot_dir: Path) -> Path: + return snapshot_dir / ".runtime_cache" + + @staticmethod + def _scoring_cache_token(scoring_fn: Any | None) -> str: + if scoring_fn is None: + return "none" + if isinstance(scoring_fn, partial): + func = scoring_fn.func + return ( + f"partial:{getattr(func, '__module__', '')}.{getattr(func, '__qualname__', getattr(func, '__name__', type(func).__name__))}:" + f"args={repr(scoring_fn.args)}:keywords={repr(scoring_fn.keywords)}" + ) + return ( + f"callable:{getattr(scoring_fn, '__module__', '')}." + f"{getattr(scoring_fn, '__qualname__', getattr(scoring_fn, '__name__', type(scoring_fn).__name__))}" + ) + + @classmethod + def _runtime_cache_file( + cls, + snapshot_dir: Path, + split_names: list[str], + scoring_fn: Any | None, + ) -> Path: + split_part = "__".join(sorted(split_names)) + score_token = cls._scoring_cache_token(scoring_fn) + token_hash = hashlib.sha1(score_token.encode("utf-8")).hexdigest()[:16] + return cls._runtime_cache_dir(snapshot_dir) / f"{split_part}__{token_hash}.pkl" + + @classmethod + def _runtime_cache_lock_file( + cls, + snapshot_dir: Path, + split_names: list[str], + scoring_fn: Any | None, + ) -> Path: + cache_file = cls._runtime_cache_file(snapshot_dir, split_names, scoring_fn) + return cache_file.with_suffix(f"{cache_file.suffix}.lock") + + @staticmethod + def _runtime_cache_fingerprint(snapshot_dir: Path, split_names: list[str]) -> str: + paths: list[Path] = [snapshot_dir / "manifest.json"] + paths.extend(snapshot_dir / f"{split_name}.parquet" for split_name in sorted(split_names)) + parts: list[str] = [f"v={_RUNTIME_CACHE_VERSION}"] + for path in paths: + if not path.exists(): + parts.append(f"{path.name}:missing") + continue + stat = path.stat() + parts.append(f"{path.name}:{stat.st_size}:{stat.st_mtime_ns}") + return hashlib.sha1("|".join(parts).encode("utf-8")).hexdigest() + + @classmethod + def _try_load_runtime_cache( + cls, + snapshot_dir: Path, + split_names: list[str], + *, + scoring_fn: Any | None, + ) -> dict[str, Any] | None: + cache_file = cls._runtime_cache_file(snapshot_dir, split_names, scoring_fn) + if not cache_file.exists(): + logger.info( + "snapshot_store_runtime_cache_miss", + reason="file_missing", + cache_file=str(cache_file), + ) + return None + try: + payload = pickle.loads(cache_file.read_bytes()) + expected_fingerprint = cls._runtime_cache_fingerprint(snapshot_dir, split_names) + if payload.get("fingerprint") != expected_fingerprint: + logger.info( + "snapshot_store_runtime_cache_miss", + reason="fingerprint_mismatch", + cache_file=str(cache_file), + ) + return None + data = payload.get("data") + if not isinstance(data, dict): + logger.info( + "snapshot_store_runtime_cache_miss", + reason="invalid_payload", + cache_file=str(cache_file), + ) + return None + logger.info( + "snapshot_store_runtime_cache_hit", + cache_file=str(cache_file), + split="merged" if len(split_names) > 1 else split_names[0], + ) + return data + except Exception as exc: + logger.warning( + "snapshot_store_runtime_cache_read_failed", + cache_file=str(cache_file), + error=str(exc), + ) + return None + + @classmethod + def _load_with_runtime_cache( + cls, + *, + snapshot_path: Path, + split_names: list[str], + scoring_fn: Any | None, + builder: Any, + ) -> "SnapshotStore": + cache_data = cls._try_load_runtime_cache( + snapshot_path, + split_names, + scoring_fn=scoring_fn, + ) + if cache_data is not None: + return cls(**cache_data) + + lock_file = cls._runtime_cache_lock_file(snapshot_path, split_names, scoring_fn) + if cls._acquire_runtime_cache_lock(lock_file): + try: + cache_data = cls._try_load_runtime_cache( + snapshot_path, + split_names, + scoring_fn=scoring_fn, + ) + if cache_data is not None: + return cls(**cache_data) + data = builder() + cls._write_runtime_cache( + snapshot_path, + split_names, + scoring_fn=scoring_fn, + data=data, + ) + return cls(**data) + finally: + cls._release_runtime_cache_lock(lock_file) + + logger.info( + "snapshot_store_runtime_cache_wait_start", + cache_file=str(cls._runtime_cache_file(snapshot_path, split_names, scoring_fn)), + split="merged" if len(split_names) > 1 else split_names[0], + ) + cache_data = cls._wait_for_runtime_cache( + snapshot_path, + split_names, + scoring_fn=scoring_fn, + lock_file=lock_file, + ) + if cache_data is not None: + return cls(**cache_data) + + if cls._acquire_runtime_cache_lock(lock_file): + try: + cache_data = cls._try_load_runtime_cache( + snapshot_path, + split_names, + scoring_fn=scoring_fn, + ) + if cache_data is not None: + return cls(**cache_data) + data = builder() + cls._write_runtime_cache( + snapshot_path, + split_names, + scoring_fn=scoring_fn, + data=data, + ) + return cls(**data) + finally: + cls._release_runtime_cache_lock(lock_file) + + logger.warning( + "snapshot_store_runtime_cache_wait_failed", + cache_file=str(cls._runtime_cache_file(snapshot_path, split_names, scoring_fn)), + ) + data = builder() + cls._write_runtime_cache( + snapshot_path, + split_names, + scoring_fn=scoring_fn, + data=data, + ) + return cls(**data) + + @staticmethod + def _runtime_cache_lock_payload() -> dict[str, Any]: + return { + "pid": os.getpid(), + "created_at": time.time(), + } + + @staticmethod + def _process_is_running(pid: int | None) -> bool: + if not isinstance(pid, int) or pid <= 0: + return False + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + return True + + @classmethod + def _lock_is_stale(cls, lock_file: Path) -> bool: + try: + payload = json.loads(lock_file.read_text()) + except Exception: + return True + pid = payload.get("pid") + created_at = payload.get("created_at") + if not cls._process_is_running(pid): + return True + if not isinstance(created_at, (int, float)): + return True + return (time.time() - float(created_at)) > _RUNTIME_CACHE_WAIT_TIMEOUT_SECONDS + + @classmethod + def _acquire_runtime_cache_lock(cls, lock_file: Path) -> bool: + lock_file.parent.mkdir(parents=True, exist_ok=True) + payload = json.dumps(cls._runtime_cache_lock_payload()) + for _ in range(2): + try: + fd = os.open(str(lock_file), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(payload) + return True + except FileExistsError: + if cls._lock_is_stale(lock_file): + try: + lock_file.unlink() + except FileNotFoundError: + pass + continue + return False + return False + + @staticmethod + def _release_runtime_cache_lock(lock_file: Path) -> None: + try: + lock_file.unlink() + except FileNotFoundError: + pass + + @classmethod + def _wait_for_runtime_cache( + cls, + snapshot_dir: Path, + split_names: list[str], + *, + scoring_fn: Any | None, + lock_file: Path, + ) -> dict[str, Any] | None: + deadline = time.monotonic() + _RUNTIME_CACHE_WAIT_TIMEOUT_SECONDS + next_log = time.monotonic() + while time.monotonic() < deadline: + cache_file = cls._runtime_cache_file(snapshot_dir, split_names, scoring_fn) + if cache_file.exists(): + cache_data = cls._try_load_runtime_cache( + snapshot_dir, + split_names, + scoring_fn=scoring_fn, + ) + if cache_data is not None: + return cache_data + if not lock_file.exists(): + return None + if cls._lock_is_stale(lock_file): + cls._release_runtime_cache_lock(lock_file) + return None + now = time.monotonic() + if now >= next_log: + logger.info( + "snapshot_store_runtime_cache_waiting", + cache_file=str(cache_file), + split="merged" if len(split_names) > 1 else split_names[0], + ) + next_log = now + _RUNTIME_CACHE_WAIT_LOG_INTERVAL_SECONDS + time.sleep(1.0) + return None + + @classmethod + def _write_runtime_cache( + cls, + snapshot_dir: Path, + split_names: list[str], + *, + scoring_fn: Any | None, + data: dict[str, Any], + ) -> None: + cache_file = cls._runtime_cache_file(snapshot_dir, split_names, scoring_fn) + cache_dir = cache_file.parent + cache_dir.mkdir(parents=True, exist_ok=True) + payload = { + "fingerprint": cls._runtime_cache_fingerprint(snapshot_dir, split_names), + "data": data, + } + tmp_file = cache_file.with_suffix(f"{cache_file.suffix}.tmp.{os.getpid()}") + tmp_file.write_bytes(pickle.dumps(payload, protocol=pickle.HIGHEST_PROTOCOL)) + os.replace(tmp_file, cache_file) + logger.info( + "snapshot_store_runtime_cache_written", + cache_file=str(cache_file), + split="merged" if len(split_names) > 1 else split_names[0], + ) + @staticmethod async def _fetch_event_metadata( event_ids: list[str], @@ -434,7 +1154,7 @@ class SnapshotStore: symbols: list[str], date_range: tuple[dt.date, dt.date] | None, oracle_url: str, - concurrency: int = 4, + concurrency: int = 12, ) -> tuple[dict[str, dict[dt.date, dict[str, Any]]], dict[str, float]]: """Fetch daily OHLCV bars and compute avg_dollar_volume per symbol.""" if not symbols or date_range is None: @@ -442,12 +1162,15 @@ class SnapshotStore: try: from libs.oracle_client import OracleClient, PriceService - start_str = date_range[0].isoformat() + fetch_start = date_range[0] - dt.timedelta(days=_PRICE_FEATURE_WARMUP_DAYS) + start_str = fetch_start.isoformat() end_str = date_range[1].isoformat() bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]] = {} avg_dvol: dict[str, float] = {} semaphore = asyncio.Semaphore(concurrency) + total_symbols = len(symbols) + progress_step = max(25, total_symbols // 10) if total_symbols else 25 async with OracleClient(base_url=oracle_url) as client: svc = PriceService(client) @@ -483,26 +1206,203 @@ class SnapshotStore: ) return sym, {}, 0.0 - results = await asyncio.gather(*(_fetch_symbol(sym) for sym in symbols)) - for sym, date_bars, mean_dvol in results: + tasks = [asyncio.create_task(_fetch_symbol(sym)) for sym in symbols] + completed = 0 + for result in asyncio.as_completed(tasks): + sym, date_bars, mean_dvol = await result bars_by_symbol[sym] = date_bars avg_dvol[sym] = mean_dvol + completed += 1 + if completed % progress_step == 0 or completed == total_symbols: + logger.info( + "snapshot_store_price_fetch_progress", + completed=completed, + total=total_symbols, + ) return bars_by_symbol, avg_dvol except Exception as exc: logger.warning("snapshot_store_oracle_failed", error=str(exc)) return {}, {} + @staticmethod + def _backfill_price_derived_features( + row: dict[str, Any], + *, + symbol: str, + event_date: dt.date, + bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]], + price_bar_cache: dict[str, list[Any]], + derived_feature_cache: dict[tuple[str, dt.date], dict[str, Any]], + ) -> None: + if all(row.get(col) is not None for col in _PRICE_DERIVED_FEATURE_COLUMNS): + return + + cache_key = (symbol, event_date) + if cache_key not in derived_feature_cache: + price_bars = SnapshotStore._get_price_bars_for_symbol( + symbol, + bars_by_symbol=bars_by_symbol, + price_bar_cache=price_bar_cache, + ) + if not price_bars: + derived_feature_cache[cache_key] = {} + return + features = compute_market_features(price_bars, event_date.isoformat()) + derived_feature_cache[cache_key] = { + col: features.get(col) + for col in _PRICE_DERIVED_FEATURE_COLUMNS + } + + for col, value in derived_feature_cache[cache_key].items(): + if row.get(col) is None and value is not None: + row[col] = value + + @classmethod + def _backfill_avg_dollar_volume_features( + cls, + row: dict[str, Any], + *, + symbol: str, + event_date: dt.date | None, + bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]], + price_bar_cache: dict[str, list[Any]], + fallback_avg_dvol: float, + ) -> None: + existing_adv_20d = row.get("avg_dollar_volume_20d") + existing_adv = row.get("avg_dollar_volume") + if existing_adv_20d is not None and existing_adv is not None: + return + + derived_adv = None + if event_date is not None: + price_bars = cls._get_price_bars_for_symbol( + symbol, + bars_by_symbol=bars_by_symbol, + price_bar_cache=price_bar_cache, + ) + if price_bars: + features = compute_market_features(price_bars, event_date.isoformat()) + derived_adv = features.get("avg_dollar_volume_20d") + + resolved_adv = ( + existing_adv_20d + if existing_adv_20d is not None + else derived_adv + if derived_adv is not None + else existing_adv + if existing_adv is not None + else fallback_avg_dvol + ) + if row.get("avg_dollar_volume_20d") is None: + row["avg_dollar_volume_20d"] = resolved_adv + if row.get("avg_dollar_volume") is None: + row["avg_dollar_volume"] = resolved_adv + + @staticmethod + def _get_price_bars_for_symbol( + symbol: str, + *, + bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]], + price_bar_cache: dict[str, list[Any]], + ) -> list[Any]: + if symbol in price_bar_cache: + return price_bar_cache[symbol] + + symbol_bars = bars_by_symbol.get(symbol) or {} + if not symbol_bars: + price_bar_cache[symbol] = [] + return price_bar_cache[symbol] + + from libs.oracle_client.models import PriceBar + + price_bar_cache[symbol] = [ + PriceBar( + date=bar_date.isoformat(), + open=float(bar.get("open", 0.0) or 0.0), + high=float(bar.get("high", 0.0) or 0.0), + low=float(bar.get("low", 0.0) or 0.0), + close=float(bar.get("close", 0.0) or 0.0), + volume=int(bar.get("volume", 0) or 0), + ) + for bar_date, bar in sorted(symbol_bars.items()) + ] + return price_bar_cache[symbol] + + @classmethod + def _attach_sector_etf_proxy_features( + cls, + row: dict[str, Any], + *, + sector: str | None, + reaction_date: dt.date, + execution_date: dt.date, + bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]], + avg_dvol: dict[str, float], + price_bar_cache: dict[str, list[Any]], + ) -> None: + proxy_symbol = sector_etf_for_sector(sector) + if not proxy_symbol: + return + + row["sector_etf_proxy"] = proxy_symbol + proxy_bars = cls._get_price_bars_for_symbol( + proxy_symbol, + bars_by_symbol=bars_by_symbol, + price_bar_cache=price_bar_cache, + ) + if not proxy_bars: + return + + proxy_features = compute_market_features(proxy_bars, reaction_date.isoformat()) + row["sector_etf_event_close"] = proxy_features.get("event_close") + row["sector_etf_reaction_day_low"] = proxy_features.get("reaction_day_low") + row["sector_etf_reaction_day_high"] = proxy_features.get("reaction_day_high") + row["sector_etf_reaction_day_return"] = proxy_features.get("reaction_day_return") + row["sector_etf_volume_ratio_20d"] = proxy_features.get("volume_ratio_20d") + row["sector_etf_gap_size"] = proxy_features.get("gap_size") + row["sector_etf_close_location"] = proxy_features.get("close_location") + row["sector_etf_avg_dollar_volume"] = ( + proxy_features.get("avg_dollar_volume_20d") + or avg_dvol.get(proxy_symbol, 0.0) + ) + row["sector_etf_atr_14"] = proxy_features.get("atr_14") + + execution_bar = (bars_by_symbol.get(proxy_symbol) or {}).get(execution_date) + if execution_bar is not None: + row["sector_etf_entry_price"] = ( + execution_bar.get("open") + or execution_bar.get("close") + ) + else: + row["sector_etf_entry_price"] = row.get("sector_etf_event_close") + + @staticmethod + def _lookup_as_of(pairs: list[tuple[dt.date, float]], as_of: dt.date) -> float | None: + lo, hi = 0, len(pairs) - 1 + result = None + while lo <= hi: + mid = (lo + hi) // 2 + if pairs[mid][0] <= as_of: + result = pairs[mid][1] + lo = mid + 1 + else: + hi = mid - 1 + return result + @staticmethod async def _fetch_sectors( symbols: list[str], oracle_url: str, - concurrency: int = 4, + concurrency: int = 12, ) -> dict[str, str]: """Fetch company sector for each symbol. Default 'UNKNOWN' if unavailable.""" if not symbols: return {} + cache = SnapshotStore._load_sector_cache() result: dict[str, str] = {} semaphore = asyncio.Semaphore(concurrency) + total_symbols = len(symbols) + progress_step = max(25, total_symbols // 10) if total_symbols else 25 try: from libs.oracle_client import CompanyService, OracleClient @@ -510,21 +1410,41 @@ class SnapshotStore: company_svc = CompanyService(client) async def _fetch_sector(sym: str) -> tuple[str, str]: async with semaphore: + cached = cache.get(sym) + if isinstance(cached, str) and cached: + return sym, cached try: info = await company_svc.get_company(sym) - return sym, info.sector or "UNKNOWN" + sector = info.sector or "UNKNOWN" + if SnapshotStore._sector_info_is_placeholder(info): + yf_sector = SnapshotStore._fetch_sector_from_yfinance(sym) + if isinstance(yf_sector, str) and yf_sector and yf_sector != "UNKNOWN": + sector = yf_sector + return sym, sector except Exception: - return sym, "UNKNOWN" + return sym, cached or "UNKNOWN" - sector_results = await asyncio.gather(*(_fetch_sector(sym) for sym in symbols)) - for sym, sector in sector_results: + tasks = [asyncio.create_task(_fetch_sector(sym)) for sym in symbols] + completed = 0 + for sector_result in asyncio.as_completed(tasks): + sym, sector = await sector_result result[sym] = sector + completed += 1 + if completed % progress_step == 0 or completed == total_symbols: + logger.info( + "snapshot_store_sector_fetch_progress", + completed=completed, + total=total_symbols, + ) except Exception as exc: logger.warning("snapshot_store_sector_fetch_failed", error=str(exc)) # Default all remaining to UNKNOWN for sym in symbols: - result.setdefault(sym, "UNKNOWN") + result.setdefault(sym, cache.get(sym, "UNKNOWN")) + if result[sym] != "UNKNOWN": + cache[sym] = result[sym] + SnapshotStore._write_sector_cache(cache) return result @staticmethod @@ -987,6 +1907,7 @@ class SnapshotStore: dates: list[dt.date] = [] for r in rows: for raw in ( + r.get("event_date"), r.get("entry_date"), r.get("execution_date"), r.get("reaction_date"), diff --git a/tests/unit/backtest/test_selector.py b/tests/unit/backtest/test_selector.py index d9666ae..a00ffb7 100644 --- a/tests/unit/backtest/test_selector.py +++ b/tests/unit/backtest/test_selector.py @@ -85,6 +85,17 @@ class TestBuildCandidate: assert c is not None assert c.sector == "UNKNOWN" + def test_prefers_avg_dollar_volume_20d_when_present(self): + from libs.backtest.selector import build_candidate + + row = _make_raw_row( + avg_dollar_volume=999_000_000.0, + avg_dollar_volume_20d=12_345_678.0, + ) + c = build_candidate(row) + assert c is not None + assert c.avg_dollar_volume == 12_345_678.0 + def test_score_bucket_classification(self): from libs.backtest.selector import build_candidate @@ -119,6 +130,39 @@ class TestBuildCandidate: assert c.timing_class == "same_day" assert c.trade_direction == "short" + def test_sector_etf_proxy_uses_proxy_trade_fields(self): + from libs.backtest.selector import build_candidate + + engine = StrategyEngineConfig( + engine_id="sector_etf_proxy", + event_types=["earnings"], + trade_symbol_mode="sector_etf", + ) + row = _make_raw_row( + score=0.9, + event_close=150.0, + reaction_day_low=145.0, + reaction_day_high=153.0, + sector_etf_proxy="XLK", + sector_etf_event_close=210.0, + sector_etf_entry_price=211.5, + sector_etf_reaction_day_low=206.0, + sector_etf_reaction_day_high=212.0, + sector_etf_avg_dollar_volume=250_000_000.0, + sector_etf_atr_14=4.2, + ) + c = build_candidate(row, strategy_engine=engine) + assert c is not None + assert c.symbol == "XLK" + assert c.source_symbol == "AAPL" + assert c.trade_symbol_mode == "sector_etf" + assert c.entry_price_est == 211.5 + assert c.avg_dollar_volume == 250_000_000.0 + assert c.atr_14 == 4.2 + assert c.features["event_close"] == 210.0 + assert c.features["reaction_day_low"] == 206.0 + assert c.features["source_event_close"] == 150.0 + def test_engine_can_force_long_direction_for_negative_reaction(self): from libs.backtest.selector import build_candidate @@ -1741,6 +1785,79 @@ class TestSelectCandidates: assert "C" not in symbols # low ADV assert "D" not in symbols # below min_price + def test_sector_etf_proxy_dedupes_by_trade_symbol(self): + from libs.backtest.selector import select_candidates + + rows = [ + _make_raw_row( + event_id="EVT::TEST::001", + symbol="AAPL", + score=0.9, + sector="Technology", + sector_etf_proxy="XLK", + sector_etf_event_close=210.0, + sector_etf_entry_price=211.0, + sector_etf_avg_dollar_volume=250_000_000.0, + sector_etf_atr_14=4.0, + ), + _make_raw_row( + event_id="EVT::TEST::002", + symbol="MSFT", + score=0.8, + sector="Technology", + sector_etf_proxy="XLK", + sector_etf_event_close=210.0, + sector_etf_entry_price=211.0, + sector_etf_avg_dollar_volume=250_000_000.0, + sector_etf_atr_14=4.0, + ), + ] + u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=1_000_000) + s = SignalConfig(score_threshold=0.1, max_candidates_per_day=5) + engine = StrategyEngineConfig( + engine_id="sector_etf_proxy", + event_types=["earnings"], + trade_symbol_mode="sector_etf", + ) + + result = select_candidates(rows, u, s, strategy_engine=engine) + assert len(result) == 1 + assert result[0].symbol == "XLK" + assert result[0].source_symbol == "AAPL" + + def test_sector_etf_proxy_respects_excluded_trade_symbol(self): + from libs.backtest.selector import select_candidates + + rows = [ + _make_raw_row( + event_id="EVT::TEST::001", + symbol="AAPL", + score=0.9, + sector="Technology", + sector_etf_proxy="XLK", + sector_etf_event_close=210.0, + sector_etf_entry_price=211.0, + sector_etf_avg_dollar_volume=250_000_000.0, + sector_etf_atr_14=4.0, + ), + ] + u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=1_000_000) + s = SignalConfig(score_threshold=0.1, max_candidates_per_day=5) + engine = StrategyEngineConfig( + engine_id="sector_etf_proxy", + event_types=["earnings"], + trade_symbol_mode="sector_etf", + ) + + result = select_candidates( + rows, + u, + s, + strategy_engine=engine, + excluded_symbols={"XLK"}, + ) + assert result == [] + def test_pipeline_with_event_type_profiles(self): from libs.backtest.selector import select_candidates diff --git a/tests/unit/backtest/test_snapshot_store.py b/tests/unit/backtest/test_snapshot_store.py index 9cace9f..1947b79 100644 --- a/tests/unit/backtest/test_snapshot_store.py +++ b/tests/unit/backtest/test_snapshot_store.py @@ -177,24 +177,136 @@ class TestSnapshotStoreLoadGuard: asyncio.run(_test()) +class TestSnapshotStoreRuntimeCache: + def test_load_merged_reuses_runtime_cache(self, tmp_path, monkeypatch): + from libs.backtest.snapshot_store import SnapshotStore + + (tmp_path / "manifest.json").write_text(json.dumps({"snapshot_id": "test"})) + for split_name in ("train", "valid", "test"): + pq.write_table( + pa.table({"event_id": [f"EVT::{split_name}"], "ticker": ["AAPL"]}), + tmp_path / f"{split_name}.parquet", + ) + + calls = {"count": 0} + data = { + "candidates_by_exec_date": { + dt.date(2026, 1, 6): [{"event_id": "EVT::001", "symbol": "AAPL"}], + }, + "bars_by_symbol_date": { + "AAPL": { + dt.date(2026, 1, 6): {"date": dt.date(2026, 1, 6), "close": 100.0}, + } + }, + "macro_by_date": {}, + } + + async def _fake_async_load_merged(*args, **kwargs): + calls["count"] += 1 + return data + + monkeypatch.setattr(SnapshotStore, "_async_load_merged", _fake_async_load_merged) + + first = SnapshotStore.load_merged( + snapshot_dir=tmp_path, + split_names=["train", "valid", "test"], + oracle_url="http://localhost", + db_dsn="postgres://localhost/test", + scoring_fn=None, + ) + assert calls["count"] == 1 + assert first.get_candidates_for_date(dt.date(2026, 1, 6))[0]["symbol"] == "AAPL" + + second = SnapshotStore.load_merged( + snapshot_dir=tmp_path, + split_names=["train", "valid", "test"], + oracle_url="http://localhost", + db_dsn="postgres://localhost/test", + scoring_fn=None, + ) + assert calls["count"] == 1 + assert second.get_candidates_for_date(dt.date(2026, 1, 6))[0]["symbol"] == "AAPL" + + def test_load_merged_waits_for_inflight_runtime_cache(self, tmp_path, monkeypatch): + from libs.backtest.snapshot_store import SnapshotStore + + (tmp_path / "manifest.json").write_text(json.dumps({"snapshot_id": "test"})) + for split_name in ("train", "valid", "test"): + pq.write_table( + pa.table({"event_id": [f"EVT::{split_name}"], "ticker": ["AAPL"]}), + tmp_path / f"{split_name}.parquet", + ) + + data = { + "candidates_by_exec_date": { + dt.date(2026, 1, 6): [{"event_id": "EVT::001", "symbol": "AAPL"}], + }, + "bars_by_symbol_date": { + "AAPL": { + dt.date(2026, 1, 6): {"date": dt.date(2026, 1, 6), "close": 100.0}, + } + }, + "macro_by_date": {}, + } + calls = {"builder": 0} + + async def _fake_async_load_merged(*args, **kwargs): + calls["builder"] += 1 + return data + + monkeypatch.setattr(SnapshotStore, "_async_load_merged", _fake_async_load_merged) + monkeypatch.setattr(SnapshotStore, "_try_load_runtime_cache", lambda *args, **kwargs: None) + monkeypatch.setattr(SnapshotStore, "_acquire_runtime_cache_lock", lambda *args, **kwargs: False) + monkeypatch.setattr(SnapshotStore, "_wait_for_runtime_cache", lambda *args, **kwargs: data) + + result = SnapshotStore.load_merged( + snapshot_dir=tmp_path, + split_names=["train", "valid", "test"], + oracle_url="http://localhost", + db_dsn="postgres://localhost/test", + scoring_fn=None, + ) + assert calls["builder"] == 0 + assert result.get_candidates_for_date(dt.date(2026, 1, 6))[0]["symbol"] == "AAPL" + + class TestSnapshotStoreFromParquet: def test_compute_date_range(self, tmp_path): from libs.backtest.snapshot_store import SnapshotStore rows = [ + {"event_date": "2026-01-03"}, {"entry_date": "2026-01-05"}, {"entry_date": "2026-01-10"}, {"entry_date": "2026-01-07"}, {"reaction_date": "2026-01-04"}, ] result = SnapshotStore._compute_date_range(rows) - assert result == (dt.date(2026, 1, 4), dt.date(2026, 1, 10)) + assert result == (dt.date(2026, 1, 3), dt.date(2026, 1, 10)) def test_compute_date_range_empty(self, tmp_path): from libs.backtest.snapshot_store import SnapshotStore assert SnapshotStore._compute_date_range([]) is None + def test_backfill_avg_dollar_volume_prefers_row_level_20d_value(self, tmp_path): + from libs.backtest.snapshot_store import SnapshotStore + + row = { + "avg_dollar_volume_20d": 12_345_678.0, + "avg_dollar_volume": None, + } + SnapshotStore._backfill_avg_dollar_volume_features( + row, + symbol="AAPL", + event_date=dt.date(2026, 1, 6), + bars_by_symbol={}, + price_bar_cache={}, + fallback_avg_dvol=999_000_000.0, + ) + assert row["avg_dollar_volume_20d"] == 12_345_678.0 + assert row["avg_dollar_volume"] == 12_345_678.0 + def test_async_load_falls_back_to_parquet_metadata_when_db_unavailable(self, tmp_path, monkeypatch): from libs.backtest.snapshot_store import SnapshotStore @@ -271,6 +383,240 @@ class TestSnapshotStoreFromParquet: assert rows[0]["score"] == pytest.approx(0.77) assert rows[0]["event_timestamp"] is not None + def test_async_load_backfills_missing_price_derived_features(self, tmp_path, monkeypatch): + from libs.backtest.snapshot_store import SnapshotStore + + parquet_path = tmp_path / "train.parquet" + event_date = dt.date(2026, 1, 5) + table = pa.table({ + "event_id": ["EVT::ROW::002"], + "ticker": ["AAPL"], + "event_date": [event_date.isoformat()], + "event_type": ["other_material_event"], + "reaction_date": [event_date.isoformat()], + "entry_date": ["2026-01-06"], + "event_close": [150.0], + "entry_price": [151.0], + "atr_14": [3.0], + }) + pq.write_table(table, parquet_path) + + async def _fake_event_meta(*args, **kwargs): + return {} + + async def _fake_price_data(*args, **kwargs): + bars = {} + start = event_date - dt.timedelta(days=120) + series = {} + for i in range(121): + d = start + dt.timedelta(days=i) + close = 100.0 + i * 0.5 + ((i % 5) - 2) * 0.1 + series[d] = { + "date": d, + "open": close - 0.4, + "high": close + 0.8, + "low": close - 0.9, + "close": close, + "volume": 1_000_000 + i * 1000, + } + bars["AAPL"] = series + return bars, {"AAPL": 125_000_000.0} + + async def _fake_sectors(*args, **kwargs): + return {"AAPL": "Technology"} + + async def _fake_macro(*args, **kwargs): + return {} + + monkeypatch.setattr(SnapshotStore, "_fetch_event_metadata", staticmethod(_fake_event_meta)) + monkeypatch.setattr(SnapshotStore, "_fetch_price_data", staticmethod(_fake_price_data)) + monkeypatch.setattr(SnapshotStore, "_fetch_sectors", staticmethod(_fake_sectors)) + monkeypatch.setattr(SnapshotStore, "_fetch_macro", staticmethod(_fake_macro)) + monkeypatch.setattr(SnapshotStore, "_fetch_spy_macro", staticmethod(_fake_macro)) + + data = asyncio.run( + SnapshotStore._async_load( + tmp_path, + "train", + oracle_url="http://localhost:18001", + db_dsn="postgresql+asyncpg://unused", + scoring_fn=lambda row: 0.77, + ) + ) + store = SnapshotStore(**data) + rows = store.get_candidates_for_date(dt.date(2026, 1, 6)) + + assert len(rows) == 1 + row = rows[0] + assert row["pre_event_hurst_60d"] is not None + assert row["pre_event_entropy_60d"] is not None + assert row["pre_event_bb_position"] is not None + assert row["pre_event_gravitational_pull"] is not None + assert row["pre_event_market_temperature"] is not None + + def test_async_load_merged_dedupes_rows_and_fetches_once(self, tmp_path, monkeypatch): + from libs.backtest.snapshot_store import SnapshotStore + + event_date = dt.date(2026, 1, 5) + train_table = pa.table({ + "event_id": ["EVT::ROW::003"], + "ticker": ["AAPL"], + "event_date": [event_date.isoformat()], + "event_type": ["earnings_release"], + "reaction_date": [event_date.isoformat()], + "entry_date": ["2026-01-06"], + "event_close": [150.0], + "entry_price": [151.0], + "atr_14": [3.0], + }) + valid_table = pa.table({ + "event_id": ["EVT::ROW::003", "EVT::ROW::004"], + "ticker": ["AAPL", "MSFT"], + "event_date": [event_date.isoformat(), "2026-01-07"], + "event_type": ["earnings_release", "guidance"], + "reaction_date": [event_date.isoformat(), "2026-01-07"], + "entry_date": ["2026-01-06", "2026-01-08"], + "event_close": [150.0, 300.0], + "entry_price": [151.0, 301.0], + "atr_14": [3.0, 5.0], + }) + pq.write_table(train_table, tmp_path / "train.parquet") + pq.write_table(valid_table, tmp_path / "valid.parquet") + + call_state: dict[str, object] = {"price_calls": 0, "date_range": None} + + async def _fake_event_meta(*args, **kwargs): + return {} + + async def _fake_price_data(symbols, date_range, *args, **kwargs): + call_state["price_calls"] = int(call_state["price_calls"]) + 1 + call_state["date_range"] = date_range + bars = {} + for sym in symbols: + bars[sym] = { + event_date: { + "date": event_date, + "open": 100.0, + "high": 101.0, + "low": 99.0, + "close": 100.5, + "volume": 1_000_000, + }, + dt.date(2026, 1, 7): { + "date": dt.date(2026, 1, 7), + "open": 101.0, + "high": 102.0, + "low": 100.0, + "close": 101.5, + "volume": 1_000_000, + }, + dt.date(2026, 1, 8): { + "date": dt.date(2026, 1, 8), + "open": 102.0, + "high": 103.0, + "low": 101.0, + "close": 102.5, + "volume": 1_000_000, + }, + } + return bars, {sym: 100_000_000.0 for sym in symbols} + + async def _fake_sectors(symbols, *args, **kwargs): + return {sym: "Technology" for sym in symbols} + + async def _fake_macro(*args, **kwargs): + return {} + + monkeypatch.setattr(SnapshotStore, "_fetch_event_metadata", staticmethod(_fake_event_meta)) + monkeypatch.setattr(SnapshotStore, "_fetch_price_data", staticmethod(_fake_price_data)) + monkeypatch.setattr(SnapshotStore, "_fetch_sectors", staticmethod(_fake_sectors)) + monkeypatch.setattr(SnapshotStore, "_fetch_macro", staticmethod(_fake_macro)) + monkeypatch.setattr(SnapshotStore, "_fetch_spy_macro", staticmethod(_fake_macro)) + + data = asyncio.run( + SnapshotStore._async_load_merged( + tmp_path, + ["train", "valid", "test"], + oracle_url="http://localhost:18001", + db_dsn="postgresql+asyncpg://unused", + scoring_fn=lambda row: 0.55, + ) + ) + store = SnapshotStore(**data) + + assert call_state["price_calls"] == 1 + assert call_state["date_range"] == (dt.date(2026, 1, 5), dt.date(2026, 1, 8)) + assert len(store.get_candidates_for_date(dt.date(2026, 1, 6))) == 1 + assert len(store.get_candidates_for_date(dt.date(2026, 1, 8))) == 1 + + def test_materialize_snapshot_dir_persists_runtime_backfilled_columns(self, tmp_path, monkeypatch): + from libs.backtest.snapshot_store import SnapshotStore + + event_date = dt.date(2026, 1, 5) + table = pa.table({ + "event_id": ["EVT::ROW::005"], + "ticker": ["AAPL"], + "event_date": [event_date.isoformat()], + "reaction_date": [event_date.isoformat()], + "entry_date": ["2026-01-06"], + "event_close": [150.0], + "entry_price": [151.0], + "macro_vix": [25.0], + "macro_hy_spread": [4.5], + }) + pq.write_table(table, tmp_path / "train.parquet") + (tmp_path / "manifest.json").write_text(json.dumps({"snapshot_id": "unit_test_snapshot"})) + + async def _fake_price_data(symbols, date_range, *args, **kwargs): + start = event_date - dt.timedelta(days=120) + bars = { + "AAPL": { + (start + dt.timedelta(days=i)): { + "date": start + dt.timedelta(days=i), + "open": 100.0 + i, + "high": 100.5 + i, + "low": 99.5 + i, + "close": 100.0 + i, + "volume": 1_000_000 + i, + } + for i in range(121) + } + } + return bars, {"AAPL": 100_000_000.0} + + async def _fake_macro(*args, **kwargs): + return { + event_date - dt.timedelta(days=1): {"T10Y2Y": 0.55}, + event_date: {"T10Y2Y": 0.60}, + } + + monkeypatch.setattr(SnapshotStore, "_fetch_price_data", staticmethod(_fake_price_data)) + monkeypatch.setattr(SnapshotStore, "_fetch_macro", staticmethod(_fake_macro)) + + materialized = SnapshotStore.materialize_snapshot_dir( + tmp_path, + oracle_url="http://localhost:18001", + db_dsn="postgresql+asyncpg://unused", + ) + + updated = pq.read_table(tmp_path / "train.parquet") + row = updated.to_pylist()[0] + + assert "pre_event_hurst_60d" in updated.column_names + assert "pre_event_market_temperature" in updated.column_names + assert "macro_t10y2y" in updated.column_names + assert row["pre_event_hurst_60d"] is not None + assert row["pre_event_entropy_60d"] is not None + assert row["pre_event_bb_position"] is not None + assert row["pre_event_market_temperature"] is not None + assert row["macro_t10y2y"] == pytest.approx(0.60) + assert "pre_event_hurst_60d" in materialized + assert "macro_t10y2y" in materialized + + manifest = json.loads((tmp_path / "manifest.json").read_text()) + assert "pre_event_hurst_60d" in manifest["materialized_feature_columns"] + assert "macro_t10y2y" in manifest["materialized_feature_columns"] + class TestSnapshotStoreSectorFetch: def test_fetch_sectors_falls_back_from_placeholder_oracle(self, tmp_path, monkeypatch):