"""Morning Momentum Intraday Backtester — Main Entry Point. Usage: # Default run (sp500, 40 trading days, default params) python -m apps.intraday_bt.run # Override params inline python -m apps.intraday_bt.run --days 200 --universe midlarge --top-n 5 --stop-loss -0.03 # Custom config file python -m apps.intraday_bt.run --config configs/intraday/default.yaml # Parameter sweep python -m apps.intraday_bt.run --sweep configs/intraday/sweep_basic.yaml # Cache management python -m apps.intraday_bt.run --no-cache python -m apps.intraday_bt.run --refresh-cache # Verbose daily output python -m apps.intraday_bt.run --verbose """ from __future__ import annotations import argparse import asyncio import json import math import pickle import re import sys import uuid from datetime import date, timedelta from pathlib import Path import yaml from apps.intraday_bt.oracle import make_intraday_oracle_client from libs.common.config import get_settings from libs.common.time_utils import is_trading_day, to_eastern, trading_days_between, utc_now from libs.intraday.cache import ( DailyBarCache, IntradayCache, LayeredDailyBarCache, ReadOnlyDailyBarCache, ) from libs.intraday.catalyst import ( AttentionEventCache, FilingEventCache, PriorEventFeatureSnapshotCache, fetch_attention_features_bulk, fetch_filing_event_features_bulk, ) from libs.intraday.domain import ( BacktestParams, CacheParams, IntradayConfig, ORBStrategyParams, OutputParams, StrategyParams, UniverseParams, ) from libs.intraday.features import compute_gap_pct, enrich_daily_bars from libs.intraday.metrics import ( compute_metrics, format_daily_breakdown, format_summary, format_sweep_comparison, format_top_trades, write_results, ) from libs.intraday.screener import ( fetch_daily_bars_bulk, fetch_intraday_bulk, momentum_intraday_first_candidates, momentum_pre_screen_candidates, orb_pre_screen_candidates, pre_screen_candidates, resolve_universe, ) from libs.intraday.simulator import ( SECTOR_PROXY_TICKERS, _bar_at_offset, _dollar_volume_up_to_bar, _market_open_ts, _volume_up_to_bar, filter_market_hours, run_simulation, ) from libs.oracle_client import CompanyService from libs.oracle_client.fred import FredService # ── Config Loading ───────────────────────────────────────────────────────── def _load_config_yaml(path: Path) -> dict: """Load a standalone YAML config. `extends` inheritance is intentionally unsupported. Strategy YAML files must be materialized in full so a single file is always the source of truth. """ resolved = path.resolve() with open(resolved) as f: raw = yaml.safe_load(f) or {} if "extends" in raw: raise ValueError( f"Config inheritance via 'extends' is no longer supported: {resolved}. " "Use a fully materialized standalone YAML file." ) return raw def load_config(path: str | None) -> IntradayConfig: """Load config from YAML file or return defaults.""" if path is None: default_path = Path("configs/intraday/default.yaml") if default_path.exists(): path = str(default_path) else: return IntradayConfig() raw = _load_config_yaml(Path(path)) strategy_mode = raw.get("strategy_mode", "momentum") strategy = StrategyParams(**raw.get("strategy", {})) universe = UniverseParams(**raw.get("universe", {})) backtest = BacktestParams(**raw.get("backtest", {})) cache = CacheParams(**raw.get("cache", {})) output = OutputParams(**raw.get("output", {})) orb_strategy = None if strategy_mode == "orb": orb_strategy = ORBStrategyParams(**raw.get("orb_strategy", {})) return IntradayConfig( strategy_mode=strategy_mode, strategy=strategy, orb_strategy=orb_strategy, universe=universe, backtest=backtest, cache=cache, output=output, ) def apply_cli_overrides(config: IntradayConfig, args: argparse.Namespace) -> IntradayConfig: """Apply CLI argument overrides to config.""" strategy_dict = config.strategy.model_dump() universe_dict = config.universe.model_dump() backtest_dict = config.backtest.model_dump() cache_dict = config.cache.model_dump() output_dict = config.output.model_dump() if args.days is not None: backtest_dict["lookback_trading_days"] = args.days if getattr(args, "start", None) is not None: backtest_dict["start_date"] = args.start if getattr(args, "end", None) is not None: backtest_dict["end_date"] = args.end if args.universe is not None: universe_dict["source"] = args.universe if args.top_n is not None: strategy_dict["top_n"] = args.top_n if args.stop_loss is not None: strategy_dict["stop_loss_pct"] = None if args.stop_loss.lower() == "none" else float(args.stop_loss) if args.entry_min is not None: strategy_dict["entry_minutes_after_open"] = args.entry_min if args.exit_min is not None: strategy_dict["exit_minutes_before_close"] = args.exit_min if args.min_gain is not None: strategy_dict["min_morning_gain_pct"] = args.min_gain if args.no_cache: cache_dict["enabled"] = False if args.verbose: output_dict["verbose"] = True if args.output_dir is not None: output_dict["dir"] = args.output_dir strategy_mode = config.strategy_mode if hasattr(args, "strategy") and args.strategy is not None: strategy_mode = args.strategy orb_strategy = config.orb_strategy if strategy_mode == "orb" and orb_strategy is None: orb_strategy = ORBStrategyParams() # Override compound_returns / initial_capital / daily_budget_reset via CLI. # "compound" is a sizing mode, so make the CLI flag disable reset unless # the user explicitly asks for reset in the same command. compound_override = getattr(args, "compound_returns", None) capital_override = getattr(args, "initial_capital", None) reset_override = getattr(args, "daily_budget_reset", None) if compound_override is True and reset_override is None: reset_override = False elif reset_override is True and compound_override is None: compound_override = False if strategy_mode == "orb" and orb_strategy is not None: if compound_override is not None or capital_override is not None or reset_override is not None: orb_dict = orb_strategy.model_dump() if compound_override is not None: orb_dict["compound_returns"] = compound_override if capital_override is not None: orb_dict["initial_capital"] = capital_override if reset_override is not None: orb_dict["daily_budget_reset"] = reset_override orb_strategy = ORBStrategyParams(**orb_dict) # Keep the generic strategy block in saved result metadata aligned with # the ORB block. The simulator uses orb_strategy, but humans and UI # diagnostics often inspect config.strategy first. strategy_dict["compound_returns"] = orb_strategy.compound_returns strategy_dict["daily_budget_reset"] = orb_strategy.daily_budget_reset strategy_dict["initial_capital"] = orb_strategy.initial_capital elif strategy_mode != "orb": if compound_override is not None or capital_override is not None or reset_override is not None: if compound_override is not None: strategy_dict["compound_returns"] = compound_override if capital_override is not None: strategy_dict["initial_capital"] = capital_override if reset_override is not None: strategy_dict["daily_budget_reset"] = reset_override return IntradayConfig( strategy_mode=strategy_mode, strategy=StrategyParams(**strategy_dict), orb_strategy=orb_strategy, universe=UniverseParams(**universe_dict), backtest=BacktestParams(**backtest_dict), cache=CacheParams(**cache_dict), output=OutputParams(**output_dict), ) # ── Trading Day Resolution ───────────────────────────────────────────────── async def get_trading_days( client: OracleClient, start_date: str | None, end_date: str | None, lookback: int, ) -> list[str]: """Resolve the list of trading days for the backtest period.""" # Use the local NYSE calendar for deterministic date resolution. # This avoids long-range Oracle timeouts when resolving multi-year periods. today = _latest_backtest_date() if end_date: end = date.fromisoformat(end_date) else: end = today if start_date: start = date.fromisoformat(start_date) else: # Fetch extra calendar days to account for weekends/holidays. start = end - timedelta(days=lookback * 2) days = [d.isoformat() for d in trading_days_between(start, end)] # If an explicit start_date was pinned, return all days in range (no lookback cap) if start_date: return days # Otherwise return last N trading days return days[-lookback:] def _latest_backtest_date(now_et=None) -> date: """Return the latest completed date safe for historical backtests. - Trading day after 16:00 ET: include today - Trading day before 16:00 ET: stop at yesterday - Non-trading day: use today as upper bound; the NYSE calendar trims to the last completed trading session automatically. """ if now_et is None: now_et = to_eastern(utc_now()) today = now_et.date() if not is_trading_day(today) or now_et.hour >= 16: return today return today - timedelta(days=1) def _latest_completed_trading_day(now_et=None) -> date: """Return the most recent completed trading session date.""" latest = _latest_backtest_date(now_et) while not is_trading_day(latest): latest -= timedelta(days=1) return latest def _load_ticker_sectors(tickers: list[str]) -> dict[str, str]: """Load cached sector labels for intraday basket diversification filters.""" settings = get_settings() path = Path(settings.data_root) / "cache" / "sector_cache.json" if not path.exists(): return {ticker: "UNKNOWN" for ticker in tickers} try: payload = json.loads(path.read_text()) except Exception: return {ticker: "UNKNOWN" for ticker in tickers} return {ticker: str(payload.get(ticker) or "UNKNOWN") for ticker in tickers} def _write_ticker_sector_cache(payload: dict[str, str]) -> None: settings = get_settings() path = Path(settings.data_root) / "cache" / "sector_cache.json" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(dict(sorted(payload.items())), indent=2)) def _sector_info_is_placeholder(info: object) -> bool: sector = getattr(info, "sector", None) industry = getattr(info, "industry", None) exchange = getattr(info, "exchange", None) market_cap = getattr(info, "market_cap", None) return ( sector == "Technology" and industry == "Software" and exchange is None and market_cap is None ) async def _load_ticker_sectors_with_oracle( tickers: list[str], client, *, concurrency: int = 12, ) -> dict[str, str]: """Load sector labels from cache, backfilling missing values from Oracle.""" if not tickers: return {} result = _load_ticker_sectors(tickers) missing = sorted( ticker for ticker, sector in result.items() if not sector or str(sector).upper() == "UNKNOWN" ) if not missing or client is None: return result settings = get_settings() path = Path(settings.data_root) / "cache" / "sector_cache.json" try: cache_payload = json.loads(path.read_text()) if path.exists() else {} except Exception: cache_payload = {} semaphore = asyncio.Semaphore(max(1, concurrency)) company_svc = CompanyService(client) async def _fetch_sector(ticker: str) -> tuple[str, str | None]: async with semaphore: try: info = await company_svc.get_company(ticker) except Exception: return ticker, None if _sector_info_is_placeholder(info): return ticker, None sector = str(getattr(info, "sector", None) or "").strip() if not sector or sector.upper() == "UNKNOWN": return ticker, None return ticker, sector updated = False tasks = [asyncio.create_task(_fetch_sector(ticker)) for ticker in missing] for task in asyncio.as_completed(tasks): ticker, sector = await task if not sector: continue result[ticker] = sector cache_payload[ticker] = sector updated = True if updated: _write_ticker_sector_cache(cache_payload) return result def _load_orb_ticker_sectors(tickers: list[str]) -> dict[str, str]: """Backward-compatible alias for ORB paths.""" return _load_ticker_sectors(tickers) def _orb_strategy_uses_catalyst(params: ORBStrategyParams) -> bool: return ( params.engine_family == "stocks_in_play_dual_regime" or params.require_event_flag or params.weight_event_catalyst > 0 ) def _orb_strategy_uses_attention(params: ORBStrategyParams) -> bool: return ( params.engine_family == "stocks_in_play_dual_regime" or params.weight_attention_wiki > 0 or params.weight_attention_news > 0 or params.attention_min_wiki_spike_10d is not None or params.attention_min_wiki_zscore_20d is not None or params.attention_min_article_count_3d is not None or params.attention_min_us_article_count_3d is not None or params.attention_min_resolver_confidence is not None ) def _orb_strategy_uses_ownership_13dg(params: ORBStrategyParams) -> bool: return ( int(getattr(params, "ownership_13dg_lookback_days", 0) or 0) > 0 and ( float(getattr(params, "weight_ownership_13dg", 0.0) or 0.0) > 0.0 or float(getattr(params, "weight_ownership_initial_13dg", 0.0) or 0.0) > 0.0 or float(getattr(params, "ownership_initial_size_scale", 1.0) or 1.0) > 1.0 or _orb_strategy_uses_ownership_seed_overlay(params) or _orb_strategy_uses_idle_sleeve(params) ) ) def _orb_strategy_uses_ownership_seed_overlay(params: ORBStrategyParams) -> bool: return ( int(getattr(params, "candidate_seed_ownership_overlay_slots", 0) or 0) > 0 and int(getattr(params, "ownership_13dg_lookback_days", 0) or 0) > 0 ) def _orb_strategy_uses_form4_seed_overlay(params: ORBStrategyParams) -> bool: return ( int(getattr(params, "candidate_seed_form4_overlay_slots", 0) or 0) > 0 and int(getattr(params, "form4_lookback_days", 0) or 0) > 0 ) def _orb_strategy_uses_form4(params: ORBStrategyParams) -> bool: return ( int(getattr(params, "form4_lookback_days", 0) or 0) > 0 and ( float(getattr(params, "form4_size_scale", 1.0) or 1.0) > 1.0 or _orb_strategy_uses_form4_seed_overlay(params) or _orb_strategy_uses_idle_sleeve(params) ) ) def _orb_strategy_uses_vix(params: ORBStrategyParams) -> bool: return any( value is not None and value != default for value, default in [ (params.max_vix, None), (params.vix_size_scale_low, None), (params.vix_size_scale_high, None), ] ) or params.vix_size_scale_min != 1.0 def _orb_strategy_requires_regime_ticker_daily(params: ORBStrategyParams) -> bool: """Whether ORB simulation needs daily bars for the configured regime ticker.""" return any( value is not None and value != default for value, default in [ (params.market_regime_spy_threshold, None), (params.regime_size_scale_low, None), (params.regime_size_scale_high, None), (params.regime_skip_below, None), ] ) def _build_intraday_data_provenance(config: IntradayConfig) -> dict[str, object]: provenance: dict[str, object] = {} if config.strategy_mode != "orb": return provenance orb_params = config.orb_strategy or ORBStrategyParams() daily_snapshot_id = getattr(orb_params, "daily_bar_snapshot_id", None) if daily_snapshot_id: provenance["daily_bars"] = { "enabled": True, "snapshot_id": str(daily_snapshot_id), "overlay_enabled": bool( getattr(orb_params, "daily_bar_snapshot_overlay_enabled", False) ), } if ( int(getattr(orb_params, "prior_event_lookback_days", 0) or 0) > 0 and float(getattr(orb_params, "weight_event_catalyst", 0.0) or 0.0) > 0.0 ): cache = PriorEventFeatureSnapshotCache( snapshot_id=getattr(orb_params, "prior_event_snapshot_id", None), ) provenance["prior_event"] = { "enabled": True, "snapshot_id": cache.snapshot_id, "lookback_calendar_days": int(orb_params.prior_event_lookback_days), "event_types": list(getattr(orb_params, "prior_event_types", None) or []), } if getattr(orb_params, "prior_event_decay_half_life_days", None) is not None: provenance["prior_event"]["decay_half_life_days"] = float( orb_params.prior_event_decay_half_life_days ) provenance["prior_event"]["decay_min_score"] = getattr( orb_params, "prior_event_decay_min_score", None ) provenance["prior_event"]["guidance_score_scale"] = float( getattr(orb_params, "prior_event_guidance_score_scale", 1.0) or 1.0 ) if _orb_strategy_uses_ownership_13dg(orb_params): provenance["ownership_13dg"] = { "reference_path": getattr(orb_params, "ownership_13dg_reference_path", None), "lookback_calendar_days": int( getattr(orb_params, "ownership_13dg_lookback_days", 0) or 0 ), "pit_rule": "filing_date < trade_date", "candidate_seed_overlay_slots": int( getattr(orb_params, "candidate_seed_ownership_overlay_slots", 0) or 0 ), } if _orb_strategy_uses_form4(orb_params): provenance["form4"] = { "reference_path": getattr(orb_params, "form4_reference_path", None), "lookback_calendar_days": int( getattr(orb_params, "form4_lookback_days", 0) or 0 ), "pit_rule": "filing_date < trade_date", } return provenance def _resolve_daily_cache_dir(config: IntradayConfig) -> str: default_dir = str(Path(config.cache.dir).with_name("daily")) if config.strategy_mode != "orb" or config.orb_strategy is None: return default_dir snapshot_id = getattr(config.orb_strategy, "daily_bar_snapshot_id", None) if not snapshot_id: return default_dir return str(Path(config.cache.dir).with_name("daily_snapshots") / str(snapshot_id)) def _build_daily_cache( config: IntradayConfig, ) -> DailyBarCache | LayeredDailyBarCache | ReadOnlyDailyBarCache: snapshot_dir = _resolve_daily_cache_dir(config) if config.strategy_mode != "orb" or config.orb_strategy is None: return DailyBarCache(snapshot_dir) snapshot_id = getattr(config.orb_strategy, "daily_bar_snapshot_id", None) if not snapshot_id: return DailyBarCache(snapshot_dir) if not bool(getattr(config.orb_strategy, "daily_bar_snapshot_overlay_enabled", False)): return ReadOnlyDailyBarCache(DailyBarCache(snapshot_dir)) overlay_dir = Path(config.cache.dir).with_name("daily_overlays") / str(snapshot_id) return LayeredDailyBarCache( DailyBarCache(snapshot_dir), DailyBarCache(str(overlay_dir)), ) def _orb_prior_event_decay_enabled(params: ORBStrategyParams) -> bool: return getattr(params, "prior_event_decay_half_life_days", None) is not None def _prior_event_features_have_detail(features: dict[str, dict[str, dict]]) -> bool: """Return True when a prior-event feature map has enough detail for decay.""" for day_map in features.values(): for payload in day_map.values(): if not payload.get("event_flag"): continue if payload.get("prior_event_candidates"): return True return False return True def _prior_event_type_score(event_type: str, params: ORBStrategyParams) -> float: event_type_l = str(event_type or "").lower() if "guidance" in event_type_l: return max( 0.0, float(getattr(params, "prior_event_guidance_score_scale", 1.0) or 1.0), ) if "earnings" in event_type_l or "results" in event_type_l: return 1.0 return 0.75 def _apply_orb_prior_event_decay_features( event_features: dict[str, dict[str, dict]], params: ORBStrategyParams, ) -> dict[str, dict[str, dict]]: """Apply PIT-safe recency/type decay to prior-event features. D-1 keeps full event strength; older events decay by configured half-life. The raw prior-event candidates remain attached for diagnostics and cache reuse. """ half_life = getattr(params, "prior_event_decay_half_life_days", None) if half_life is None: return event_features half_life = float(half_life) if half_life <= 0: return event_features min_score = getattr(params, "prior_event_decay_min_score", None) min_score_f = float(min_score) if min_score is not None else None for day_map in event_features.values(): for payload in day_map.values(): if not payload.get("event_flag"): continue candidates = payload.get("prior_event_candidates") or [] best_score = 0.0 best_candidate: dict | None = None for candidate in candidates: try: days_ago = int(candidate.get("days_ago")) except (TypeError, ValueError): continue if days_ago < 1: continue type_score = _prior_event_type_score( str(candidate.get("event_type") or ""), params, ) age_after_first_day = max(days_ago - 1, 0) decay = math.pow(0.5, age_after_first_day / half_life) score = type_score * decay if score > best_score: best_score = score best_candidate = candidate if best_candidate is None: continue payload["prior_event_binary_score"] = float(payload.get("event_score") or 0.0) payload["prior_event_decay_score"] = round(best_score, 6) payload["prior_event_best_days_ago"] = int(best_candidate["days_ago"]) payload["prior_event_best_event_type"] = str(best_candidate.get("event_type") or "") if min_score_f is not None and best_score < min_score_f: payload["event_flag"] = False payload["event_score"] = 0.0 else: payload["event_score"] = round(best_score, 6) return event_features def _coerce_ownership_date(value: object) -> date | None: if value is None: return None if isinstance(value, date): return value text = str(value).strip() if not text: return None try: return date.fromisoformat(text[:10]) except ValueError: return None def _load_orb_ownership_13dg_features( tickers: list[str], trading_days: list[str], params: ORBStrategyParams, ) -> dict[str, dict[str, dict]]: """Load PIT-safe 13D/13G ownership flags from local parquet. The parquet is local reference data, so this path is deterministic and does not depend on the live Oracle API. Filing dates are strictly before the trade date to avoid same-day filing lookahead. """ lookback_days = int(getattr(params, "ownership_13dg_lookback_days", 0) or 0) if lookback_days <= 0 or not tickers or not trading_days: return {} path = Path( getattr( params, "ownership_13dg_reference_path", "data/reference/ownership_13d13g_events_pit.parquet", ) or "data/reference/ownership_13d13g_events_pit.parquet" ) if not path.exists(): return {} import pyarrow.parquet as pq ticker_set = {str(ticker).strip().upper() for ticker in tickers if str(ticker).strip()} if not ticker_set: return {} rows = pq.read_table(str(path)).to_pylist() events_by_ticker: dict[str, list[dict[str, object]]] = {} for row in rows: symbol = str(row.get("symbol") or "").strip().upper() if symbol not in ticker_set: continue filing_date = _coerce_ownership_date(row.get("filing_date")) if filing_date is None: continue events_by_ticker.setdefault(symbol, []).append( { "filing_date": filing_date, "form_type": str(row.get("form_type") or "").strip().upper(), "is_initial_for_owner": bool(row.get("is_initial_for_owner") or False), "is_13g_to_13d_transition": bool( row.get("is_13g_to_13d_transition") or False ), "activist_flag": bool(row.get("activist_flag") or False), "ownership_strength_score": row.get("ownership_strength_score"), } ) trading_dates = [(day, date.fromisoformat(day)) for day in trading_days] result: dict[str, dict[str, dict]] = {} for ticker, ticker_events in events_by_ticker.items(): ticker_events.sort(key=lambda item: item["filing_date"]) for day, trade_date in trading_dates: cutoff = trade_date - timedelta(days=lookback_days) eligible = [ event for event in ticker_events if cutoff <= event["filing_date"] < trade_date ] if not eligible: continue days_since = min( (trade_date - event["filing_date"]).days for event in eligible ) strength_scores: list[float] = [] for event in eligible: try: strength_scores.append(float(event.get("ownership_strength_score") or 0.0)) except (TypeError, ValueError): continue max_strength = max(strength_scores) if strength_scores else 0.0 has_initial = any(bool(event.get("is_initial_for_owner")) for event in eligible) has_active_13d = any( str(event.get("form_type") or "") in {"SC 13D", "SCHEDULE 13D"} for event in eligible ) result.setdefault(ticker, {})[day] = { "ownership_13dg_flag": True, "ownership_13dg_initial_flag": has_initial, "ownership_13dg_active_13d_flag": has_active_13d, "ownership_13dg_transition_flag": any( bool(event.get("is_13g_to_13d_transition")) for event in eligible ), "ownership_13dg_activist_flag": any( bool(event.get("activist_flag")) for event in eligible ), "ownership_13dg_event_count": len(eligible), "ownership_13dg_days_since": days_since, "ownership_13dg_strength_score": max_strength, "ownership_13dg_score": 1.0, "ownership_13dg_initial_score": 1.0 if has_initial else 0.0, } return result def _merge_orb_ownership_13dg_features( enrichment: dict[str, dict[str, dict]], ownership_features: dict[str, dict[str, dict]], ) -> None: for ticker, day_map in ownership_features.items(): ticker_enrichment = enrichment.setdefault(ticker, {}) for day, payload in day_map.items(): ticker_enrichment.setdefault(day, {}).update(payload) def _load_orb_form4_features( tickers: list[str], trading_days: list[str], params: ORBStrategyParams, ) -> dict[str, dict[str, dict]]: """Load PIT-safe Form 4 purchase-cluster features from local parquet.""" lookback_days = int(getattr(params, "form4_lookback_days", 0) or 0) if lookback_days <= 0 or not tickers or not trading_days: return {} path = Path( getattr( params, "form4_reference_path", "data/reference/form4_daily_events_pit.parquet", ) or "data/reference/form4_daily_events_pit.parquet" ) if not path.exists(): return {} import pyarrow.parquet as pq ticker_set = {str(ticker).strip().upper() for ticker in tickers if str(ticker).strip()} if not ticker_set: return {} rows = pq.read_table(str(path)).to_pylist() events_by_ticker: dict[str, list[dict[str, object]]] = {} for row in rows: symbol = str(row.get("symbol") or "").strip().upper() if symbol not in ticker_set: continue filing_date = _coerce_ownership_date(row.get("filing_date")) if filing_date is None: continue events_by_ticker.setdefault(symbol, []).append( { "filing_date": filing_date, "total_value": row.get("total_value"), "owner_count": row.get("owner_count"), "transaction_count": row.get("transaction_count"), "c_suite_count": row.get("c_suite_count"), "role_weight_score": row.get("role_weight_score"), "has_officer_or_director": row.get("has_officer_or_director"), } ) trading_dates = [(day, date.fromisoformat(day)) for day in trading_days] result: dict[str, dict[str, dict]] = {} for ticker, ticker_events in events_by_ticker.items(): ticker_events.sort(key=lambda item: item["filing_date"]) for day, trade_date in trading_dates: cutoff = trade_date - timedelta(days=lookback_days) eligible = [ event for event in ticker_events if cutoff <= event["filing_date"] < trade_date ] if not eligible: continue def _num(event: dict[str, object], key: str) -> float: try: return float(event.get(key) or 0.0) except (TypeError, ValueError): return 0.0 def _int_num(event: dict[str, object], key: str) -> int: try: return int(event.get(key) or 0) except (TypeError, ValueError): return 0 days_since = min( (trade_date - event["filing_date"]).days for event in eligible ) result.setdefault(ticker, {})[day] = { "form4_flag": True, "form4_days_since": days_since, "form4_event_count": len(eligible), "form4_total_value": sum(_num(event, "total_value") for event in eligible), "form4_owner_count": max(_int_num(event, "owner_count") for event in eligible), "form4_transaction_count": max( _int_num(event, "transaction_count") for event in eligible ), "form4_c_suite_count": max( _int_num(event, "c_suite_count") for event in eligible ), "form4_role_weight_score": max( _num(event, "role_weight_score") for event in eligible ), "form4_has_officer_or_director": any( bool(event.get("has_officer_or_director")) for event in eligible ), } return result def _merge_orb_form4_features( enrichment: dict[str, dict[str, dict]], form4_features: dict[str, dict[str, dict]], ) -> None: for ticker, day_map in form4_features.items(): ticker_enrichment = enrichment.setdefault(ticker, {}) for day, payload in day_map.items(): ticker_enrichment.setdefault(day, {}).update(payload) async def _fetch_prior_event_features_db_live( tickers: list[str], trading_days: list[str], lookback_calendar_days: int = 7, event_types: tuple[str, ...] = ("earnings_release", "guidance_update"), ) -> dict[str, dict[str, dict]]: """Bulk-fetch prior earnings/guidance events from DB and build event feature map. For each ticker, finds events of the specified types in the DB events table within [start - lookback_calendar_days, end], then marks each trading day within `lookback_calendar_days` after an event as event_flag=True, event_score=1.0. Trading days with no recent event retain event_flag=False, event_score=0.0. """ import asyncpg from datetime import datetime, timedelta if not tickers or not trading_days: return {} start_date = trading_days[0] end_date = trading_days[-1] # Expand lookback window to capture events before the first trading day start_dt = datetime.strptime(start_date, "%Y-%m-%d").date() end_dt = datetime.strptime(end_date, "%Y-%m-%d").date() # symbol_id format: SYM::{TICKER}::US symbol_ids = [f"SYM::{t}::US" for t in tickers] ticker_from_sid = {f"SYM::{t}::US": t for t in tickers} dsn = get_settings().postgres_dsn.replace("+asyncpg", "") conn = await asyncpg.connect(dsn=dsn) try: rows = await conn.fetch( """ SELECT symbol_id, event_date::text AS event_date, event_type, event_direction, filed_at_utc::text AS filed_at_utc FROM events WHERE symbol_id = ANY($1) AND event_type = ANY($2) AND event_date BETWEEN $3 AND $4 ORDER BY symbol_id, event_date """, symbol_ids, list(event_types), (start_dt - timedelta(days=lookback_calendar_days)), end_dt, ) finally: await conn.close() # Group events per ticker. Event timestamps are kept for diagnostics, but # prior-event scoring only admits events from earlier calendar days. from collections import defaultdict events_by_ticker: dict[str, list[dict]] = defaultdict(list) for row in rows: ticker = ticker_from_sid.get(row["symbol_id"]) if ticker: events_by_ticker[ticker].append({ "event_date": row["event_date"], "event_type": row["event_type"], "event_direction": row["event_direction"], "filed_at_utc": row["filed_at_utc"], }) # Build feature map: for each trading day, mark True if within lookback window after an event result: dict[str, dict[str, dict]] = {} for ticker in tickers: ticker_events = events_by_ticker.get(ticker, []) if not ticker_events: continue ticker_map: dict[str, dict] = {} for td_str in trading_days: td = datetime.strptime(td_str, "%Y-%m-%d").date() eligible_events: list[dict] = [] for event in ticker_events: ev = datetime.strptime(str(event.get("event_date")), "%Y-%m-%d").date() days_ago = (td - ev).days if 1 <= days_ago <= lookback_calendar_days: eligible_events.append({ "event_date": str(event.get("event_date")), "event_type": str(event.get("event_type") or ""), "event_direction": event.get("event_direction"), "filed_at_utc": event.get("filed_at_utc"), "days_ago": days_ago, }) if eligible_events: event_types_out = sorted({ str(event.get("event_type") or "") for event in eligible_events if str(event.get("event_type") or "") }) ticker_map[td_str] = { "event_flag": True, "event_score": 1.0, "event_count": len(eligible_events), "event_types": event_types_out, "prior_event_days_ago": min(int(e["days_ago"]) for e in eligible_events), "prior_event_candidates": sorted( eligible_events, key=lambda e: (int(e["days_ago"]), str(e.get("event_type") or "")), ), } if ticker_map: result[ticker] = ticker_map return result async def _prefetch_prior_event_features_db( tickers: list[str], trading_days: list[str], lookback_calendar_days: int = 7, event_types: tuple[str, ...] = ("earnings_release", "guidance_update"), snapshot_cache: PriorEventFeatureSnapshotCache | None = None, require_event_details: bool = False, ) -> dict[str, dict[str, dict]]: """Return prior-event enrichment, freezing each request in a local snapshot.""" if not tickers or not trading_days: return {} normalized_types = tuple( sorted({ str(event_type).strip() for event_type in event_types if str(event_type).strip() }) ) if snapshot_cache is not None: cached = await asyncio.to_thread( snapshot_cache.get, tickers, trading_days, lookback_calendar_days, normalized_types, ) if cached is not None and ( not require_event_details or _prior_event_features_have_detail(cached) ): return cached result = await _fetch_prior_event_features_db_live( tickers, trading_days, lookback_calendar_days=lookback_calendar_days, event_types=normalized_types, ) if snapshot_cache is not None: await asyncio.to_thread( snapshot_cache.put, tickers, trading_days, lookback_calendar_days, normalized_types, result, ) return result def _merge_orb_event_features( enrichment: dict[str, dict[str, dict]], event_features: dict[str, dict[str, dict]], ) -> None: for ticker, day_map in event_features.items(): ticker_enrichment = enrichment.setdefault(ticker, {}) for day, features in day_map.items(): ticker_day = ticker_enrichment.setdefault(day, {}) ticker_day.update(features) def _merge_orb_attention_features( enrichment: dict[str, dict[str, dict]], attention_features: dict[str, dict[str, dict]], ) -> None: for ticker, day_map in attention_features.items(): ticker_enrichment = enrichment.setdefault(ticker, {}) for day, features in day_map.items(): ticker_day = ticker_enrichment.setdefault(day, {}) ticker_day.update(features) def _orb_candidate_event_tickers( candidates: dict[str, list[str]], enrichment: dict[str, dict[str, dict]], params: ORBStrategyParams, ) -> list[str]: """Reduce catalyst fetches to the most relevant daily stocks-in-play names.""" selected: set[str] = set() min_abs_gap = getattr(params, "min_abs_gap_pct", None) per_day_limit = max(int(getattr(params, "max_candidates", 20) * 2), 12) for day, day_tickers in candidates.items(): ranked: list[tuple[float, str]] = [] for ticker in day_tickers: ticker_day = enrichment.get(ticker, {}).get(day, {}) prev_close = ticker_day.get("prev_close") today_open = ticker_day.get("today_open") if prev_close and today_open and prev_close > 0: abs_gap = abs((today_open - prev_close) / prev_close) if min_abs_gap is not None and abs_gap < min_abs_gap: continue ranked.append((abs_gap, ticker)) ranked.sort(reverse=True) for _, ticker in ranked[:per_day_limit]: selected.add(ticker) return sorted(selected) def _orb_candidate_event_pairs( candidates: dict[str, list[str]], enrichment: dict[str, dict[str, dict]], params: ORBStrategyParams, ) -> list[tuple[str, str]]: selected: list[tuple[str, str]] = [] min_abs_gap = getattr(params, "min_abs_gap_pct", None) per_day_limit = max(int(getattr(params, "max_candidates", 20) * 2), 12) for day, day_tickers in candidates.items(): ranked: list[tuple[float, str]] = [] for ticker in day_tickers: ticker_day = enrichment.get(ticker, {}).get(day, {}) prev_close = ticker_day.get("prev_close") today_open = ticker_day.get("today_open") if prev_close and today_open and prev_close > 0: abs_gap = abs((today_open - prev_close) / prev_close) if min_abs_gap is not None and abs_gap < min_abs_gap: continue ranked.append((abs_gap, ticker)) ranked.sort(reverse=True) for _, ticker in ranked[:per_day_limit]: selected.append((ticker, day)) return selected def _momentum_strategy_uses_daily_enrichment(params: StrategyParams) -> bool: return any( value is not None and value != default for value, default in [ (params.min_gap_pct, None), (params.max_gap_pct, None), (params.min_volume_ratio_14d, None), (params.min_ret_5d, None), (params.min_entropy_20d, None), (params.max_entropy_20d, None), (params.entropy_size_scale_low, None), (params.entropy_size_scale_high, None), (params.use_moderate_gap_liquid_sleeve, False), (params.use_liquid_cluster_engine, False), (params.use_sector_etf_sleeve, False), (params.liquid_cluster_require_special_liquidity_gate, False), (params.liquid_cluster_min_avg_dollar_vol_30d, None), (params.liquid_cluster_max_avg_dollar_vol_30d, None), (params.liquid_cluster_min_volume_ratio_14d, None), (params.liquid_cluster_max_entropy_20d, None), (params.candidate_seed_moderate_liquid_overlay_slots, 0), (params.candidate_intraday_moderate_liquid_reserve_slots, 0), (params.market_regime_gap_threshold, None), (params.regime_size_scale_low, None), (params.regime_size_scale_high, None), (params.regime_skip_below, None), (params.min_candidate_breadth, None), (params.breadth_size_scale_low, None), (params.breadth_size_scale_high, None), (params.breadth_skip_below, None), ] ) or params.use_five_sleeves def _momentum_strategy_uses_sector_labels(params: StrategyParams) -> bool: return bool( params.max_positions_per_sector or params.use_sector_thrust_sleeve or params.use_liquid_cluster_engine or params.use_sector_etf_sleeve ) def _momentum_strategy_uses_sector_proxies(params: StrategyParams) -> bool: return bool( params.use_sector_etf_sleeve and params.sector_etf_capital_fraction > 0 and params.sector_etf_max_positions > 0 ) def _momentum_strategy_requires_regime_ticker_daily(params: StrategyParams) -> bool: return any( value is not None and value != default for value, default in [ (params.market_regime_gap_threshold, None), (params.regime_size_scale_low, None), (params.regime_size_scale_high, None), (params.regime_skip_below, None), ] ) def _momentum_strategy_uses_catalyst(params: StrategyParams) -> bool: return ( params.candidate_require_event_flag or params.candidate_min_event_score is not None or bool(params.candidate_allowed_event_types) or params.candidate_seed_event_overlay_slots > 0 or params.candidate_seed_event_min_score is not None or params.candidate_weight_event_score > 0 or params.candidate_intraday_weight_event_score > 0 or params.candidate_intraday_event_reserve_slots > 0 or params.candidate_intraday_event_reserve_min_score is not None or params.use_event_sleeve or params.event_weight > 0 or params.event_min_score is not None or params.use_event_day_liquid_sleeve or params.event_day_liquid_min_event_score is not None or params.event_day_liquid_min_event_support_score is not None or params.event_day_liquid_min_total_event_entry_dollar_volume is not None ) def _momentum_strategy_uses_candidate_stage_catalyst(params: StrategyParams) -> bool: return ( params.candidate_require_event_flag or params.candidate_min_event_score is not None or bool(params.candidate_allowed_event_types) or params.candidate_seed_event_overlay_slots > 0 or params.candidate_seed_event_min_score is not None ) def _momentum_candidate_allowed_event_types(strategy: StrategyParams) -> set[str]: return { str(value).strip().lower() for value in strategy.candidate_allowed_event_types if str(value).strip() } def _momentum_candidate_event_types_pass(info: dict, strategy: StrategyParams) -> bool: allowed_event_types = _momentum_candidate_allowed_event_types(strategy) if not allowed_event_types: return True raw_event_types = info.get("event_types") or [] event_types = { str(value).strip().lower() for value in raw_event_types if str(value).strip() } if not event_types: return False return any(event_type in allowed_event_types for event_type in event_types) def _momentum_strategy_uses_attention(params: StrategyParams) -> bool: return ( params.candidate_weight_attention_wiki > 0 or params.candidate_weight_attention_news > 0 or params.candidate_intraday_weight_attention_wiki > 0 or params.candidate_intraday_weight_attention_news > 0 or params.candidate_min_attention_wiki_spike_10d is not None or params.candidate_min_attention_article_count_3d is not None or params.candidate_min_attention_us_article_count_3d is not None or params.candidate_min_attention_resolver_confidence is not None ) def _momentum_strategy_uses_vix(params: StrategyParams) -> bool: return any( value is not None and value != default for value, default in [ (params.max_vix, None), (params.vix_size_scale_low, None), (params.vix_size_scale_high, None), ] ) or params.vix_size_scale_min != 1.0 def _momentum_uses_historical_intraday_first(params: StrategyParams) -> bool: return str(getattr(params, "candidate_source_mode", "daily_gap")).lower() == "intraday_first" def _should_use_recent_live_scan( strategy: StrategyParams, trading_days: list[str], ) -> bool: if strategy.recent_live_scan_days <= 0 or not trading_days: return False if len(trading_days) > strategy.recent_live_scan_days: return False latest_safe = _latest_backtest_date() end_day = date.fromisoformat(trading_days[-1]) delta_days = (latest_safe - end_day).days return 0 <= delta_days <= strategy.recent_live_scan_days def _recent_live_scan_universe(strategy: StrategyParams) -> UniverseParams: return UniverseParams( source="screener", market_cap_min=strategy.recent_live_scan_market_cap_min, avg_volume_min=strategy.recent_live_scan_avg_volume_min, min_price=strategy.recent_live_scan_min_price, ) def _strategy_for_recent_live_scan( strategy: StrategyParams, *, recent_live_scan: bool, ) -> StrategyParams: """Apply recent-window-only overrides without affecting research/Q1 configs.""" if not recent_live_scan: return strategy updates: dict[str, object] = {} override_fields = [ ("recent_live_scan_top_n", "top_n"), ("recent_live_scan_min_morning_gain_pct", "min_morning_gain_pct"), ("recent_live_scan_max_morning_gain_pct", "max_morning_gain_pct"), ("recent_live_scan_min_confirmation_return_pct", "min_confirmation_return_pct"), ("recent_live_scan_min_entry_dollar_volume", "min_entry_dollar_volume"), ("recent_live_scan_max_gap_pct", "max_gap_pct"), ("recent_live_scan_max_entropy_20d", "max_entropy_20d"), ("recent_live_scan_use_slow_ignite_sleeve", "use_slow_ignite_sleeve"), ("recent_live_scan_slow_ignite_weight", "slow_ignite_weight"), ("recent_live_scan_slow_ignite_min_gain_pct", "slow_ignite_min_gain_pct"), ("recent_live_scan_slow_ignite_max_gain_pct", "slow_ignite_max_gain_pct"), ("recent_live_scan_slow_ignite_min_entry_dollar_volume", "slow_ignite_min_entry_dollar_volume"), ("recent_live_scan_slow_ignite_max_entropy_20d", "slow_ignite_max_entropy_20d"), ("recent_live_scan_use_liquid_largecap_sleeve", "use_liquid_largecap_sleeve"), ("recent_live_scan_liquid_largecap_weight", "liquid_largecap_weight"), ("recent_live_scan_liquid_largecap_min_gain_pct", "liquid_largecap_min_gain_pct"), ("recent_live_scan_liquid_largecap_max_gain_pct", "liquid_largecap_max_gain_pct"), ("recent_live_scan_liquid_largecap_min_confirmation_return_pct", "liquid_largecap_min_confirmation_return_pct"), ("recent_live_scan_liquid_largecap_min_entry_dollar_volume", "liquid_largecap_min_entry_dollar_volume"), ("recent_live_scan_liquid_largecap_min_avg_dollar_vol_30d", "liquid_largecap_min_avg_dollar_vol_30d"), ("recent_live_scan_liquid_largecap_max_entropy_20d", "liquid_largecap_max_entropy_20d"), ] for source_field, target_field in override_fields: value = getattr(strategy, source_field, None) if value is not None: updates[target_field] = value if not updates: return strategy return strategy.model_copy(update=updates) def _should_use_recent_intraday_first_scan(trading_days: list[str]) -> bool: """Recent live-scan windows should use intraday-first candidate generation. Using daily-first on 1-5 day sanity windows misses obvious leaders such as recent Yahoo gainers that are absent from the static research universe or do not pass a daily pre-screen despite strong same-day intraday momentum. """ return bool(trading_days) def _merge_momentum_event_features( enrichment: dict[str, dict[str, dict]], event_features: dict[str, dict[str, dict]], ) -> None: for ticker, day_map in event_features.items(): ticker_enrichment = enrichment.setdefault(ticker, {}) for day, features in day_map.items(): ticker_day = ticker_enrichment.setdefault(day, {}) ticker_day.update(features) def _merge_momentum_attention_features( enrichment: dict[str, dict[str, dict]], attention_features: dict[str, dict[str, dict]], ) -> None: for ticker, day_map in attention_features.items(): ticker_enrichment = enrichment.setdefault(ticker, {}) for day, features in day_map.items(): ticker_day = ticker_enrichment.setdefault(day, {}) ticker_day.update(features) def _momentum_preliminary_candidates( daily_bars: dict[str, list[dict]], trading_days: list[str], enrichment: dict[str, dict[str, dict]], threshold: float, strategy: StrategyParams, ) -> dict[str, list[str]]: return momentum_pre_screen_candidates( daily_bars, trading_days, enrichment, threshold=threshold, max_per_day=None, strategy=None, ) def _momentum_intraday_seed_candidates( daily_bars: dict[str, list[dict]], trading_days: list[str], enrichment: dict[str, dict[str, dict]], strategy: StrategyParams, *, default_threshold: float, use_signal_features: bool = False, ) -> dict[str, list[str]]: seed_threshold = strategy.candidate_seed_threshold seed_max_per_day = strategy.candidate_seed_max_per_day if not _momentum_uses_historical_intraday_first(strategy): seed_threshold = default_threshold seed_max_per_day = max(strategy.top_n * 20, 120) return momentum_pre_screen_candidates( daily_bars, trading_days, enrichment, threshold=seed_threshold, max_per_day=seed_max_per_day, strategy=strategy if use_signal_features else None, ) def _augment_momentum_seed_candidates_with_liquid_overlay( candidates: dict[str, list[str]], daily_bars: dict[str, list[dict]], trading_days: list[str], enrichment: dict[str, dict[str, dict]], strategy: StrategyParams, ) -> tuple[dict[str, list[str]], dict[str, set[str]]]: """Returns (augmented_candidates, overlay_tickers_per_day). overlay_tickers_per_day maps each date to the set of tickers added by the overlay (leader + liquid + event). Used by the ORB path to bypass gapper-specific filters for these tickers. """ slots = int(getattr(strategy, "candidate_seed_liquid_overlay_slots", 0) or 0) leader_slots = int(getattr(strategy, "candidate_seed_leader_overlay_slots", 0) or 0) moderate_slots = int(getattr(strategy, "candidate_seed_moderate_liquid_overlay_slots", 0) or 0) event_slots = int(getattr(strategy, "candidate_seed_event_overlay_slots", 0) or 0) ownership_slots = int(getattr(strategy, "candidate_seed_ownership_overlay_slots", 0) or 0) form4_slots = int(getattr(strategy, "candidate_seed_form4_overlay_slots", 0) or 0) if ( slots <= 0 and leader_slots <= 0 and moderate_slots <= 0 and event_slots <= 0 and ownership_slots <= 0 and form4_slots <= 0 ): return {day: list(day_tickers) for day, day_tickers in candidates.items() if day_tickers}, {} ticker_day_bar: dict[str, dict[str, dict]] = {} for ticker, bars in daily_bars.items(): day_map: dict[str, dict] = {} for bar in bars: day_map[str(bar["date"])[:10]] = bar ticker_day_bar[ticker] = day_map result: dict[str, list[str]] = {} overlay_by_day: dict[str, set[str]] = {} min_gap = getattr(strategy, "candidate_seed_liquid_min_gap_pct", None) max_gap = getattr(strategy, "candidate_seed_liquid_max_gap_pct", None) min_avg_dollar_vol = getattr(strategy, "candidate_seed_liquid_min_avg_dollar_vol_30d", None) min_ret_5d = getattr(strategy, "candidate_seed_liquid_min_ret_5d", None) max_entropy = getattr(strategy, "candidate_seed_liquid_max_entropy_20d", None) leader_min_gap = getattr(strategy, "candidate_seed_leader_min_gap_pct", None) leader_max_gap = getattr(strategy, "candidate_seed_leader_max_gap_pct", None) leader_min_avg_dollar_vol = getattr(strategy, "candidate_seed_leader_min_avg_dollar_vol_30d", None) leader_min_ret_5d = getattr(strategy, "candidate_seed_leader_min_ret_5d", None) leader_min_atr_pct = getattr(strategy, "candidate_seed_leader_min_atr_pct", None) leader_max_entropy = getattr(strategy, "candidate_seed_leader_max_entropy_20d", None) moderate_min_gap = getattr(strategy, "candidate_seed_moderate_liquid_min_gap_pct", None) moderate_max_gap = getattr(strategy, "candidate_seed_moderate_liquid_max_gap_pct", None) moderate_min_avg_dollar_vol = getattr( strategy, "candidate_seed_moderate_liquid_min_avg_dollar_vol_30d", None, ) moderate_max_avg_dollar_vol = getattr( strategy, "candidate_seed_moderate_liquid_max_avg_dollar_vol_30d", None, ) moderate_min_ret_5d = getattr(strategy, "candidate_seed_moderate_liquid_min_ret_5d", None) moderate_max_entropy = getattr(strategy, "candidate_seed_moderate_liquid_max_entropy_20d", None) event_min_score = getattr(strategy, "candidate_seed_event_min_score", None) event_min_gap = getattr(strategy, "candidate_seed_event_min_gap_pct", None) event_max_gap = getattr(strategy, "candidate_seed_event_max_gap_pct", None) event_min_avg_dollar_vol = getattr(strategy, "candidate_seed_event_min_avg_dollar_vol_30d", None) event_min_ret_5d = getattr(strategy, "candidate_seed_event_min_ret_5d", None) event_max_entropy = getattr(strategy, "candidate_seed_event_max_entropy_20d", None) ownership_initial_only = bool(getattr(strategy, "candidate_seed_ownership_initial_only", True)) ownership_min_strength = getattr(strategy, "candidate_seed_ownership_min_strength_score", None) ownership_min_gap = getattr(strategy, "candidate_seed_ownership_min_gap_pct", None) ownership_max_gap = getattr(strategy, "candidate_seed_ownership_max_gap_pct", None) ownership_min_avg_dollar_vol = getattr( strategy, "candidate_seed_ownership_min_avg_dollar_vol_30d", None, ) ownership_min_ret_5d = getattr(strategy, "candidate_seed_ownership_min_ret_5d", None) ownership_max_entropy = getattr(strategy, "candidate_seed_ownership_max_entropy_20d", None) form4_min_total_value = getattr(strategy, "candidate_seed_form4_min_total_value", None) form4_min_owner_count = getattr(strategy, "candidate_seed_form4_min_owner_count", None) form4_min_c_suite_count = getattr(strategy, "candidate_seed_form4_min_c_suite_count", None) form4_require_cluster_or_csuite = bool( getattr(strategy, "candidate_seed_form4_require_cluster_or_csuite", False) ) form4_min_gap = getattr(strategy, "candidate_seed_form4_min_gap_pct", None) form4_max_gap = getattr(strategy, "candidate_seed_form4_max_gap_pct", None) form4_min_avg_dollar_vol = getattr( strategy, "candidate_seed_form4_min_avg_dollar_vol_30d", None, ) form4_min_ret_5d = getattr(strategy, "candidate_seed_form4_min_ret_5d", None) form4_max_entropy = getattr(strategy, "candidate_seed_form4_max_entropy_20d", None) def overlay_gap_pct(info: dict, bar: dict) -> float | None: gap_pct = info.get("gap_pct") if gap_pct is not None: return float(gap_pct) prev_close = info.get("prev_close") open_price = bar.get("open") if prev_close in (None, 0) or open_price in (None, 0): return None return (float(open_price) - float(prev_close)) / float(prev_close) for day in trading_days: day_candidates = list(candidates.get(day, [])) chosen = set(day_candidates) day_overlay_tickers: set[str] = set() overlay_ranked: list[tuple[float, float, float, str]] = [] for ticker, date_map in ticker_day_bar.items(): bar = date_map.get(day) if not bar: continue info = enrichment.get(ticker, {}).get(day, {}) gap_pct = overlay_gap_pct(info, bar) if gap_pct is None: continue if min_gap is not None and gap_pct < min_gap: continue if max_gap is not None and gap_pct > max_gap: continue avg_dollar_vol = info.get("avg_dollar_vol_30d") if min_avg_dollar_vol is not None and ( avg_dollar_vol is None or avg_dollar_vol < min_avg_dollar_vol ): continue ret_5d = info.get("ret_5d") if min_ret_5d is not None and (ret_5d is None or ret_5d < min_ret_5d): continue entropy_20d = info.get("entropy_20d") if max_entropy is not None and ( entropy_20d is None or entropy_20d > max_entropy ): continue overlay_ranked.append( ( float(avg_dollar_vol or 0.0), float(gap_pct or 0.0), float(ret_5d or 0.0), ticker, ) ) if overlay_ranked: ranked_overlay = [ticker for *_rest, ticker in sorted(overlay_ranked, reverse=True)[:slots]] for ticker in ranked_overlay: if ticker not in chosen: day_candidates.append(ticker) chosen.add(ticker) day_overlay_tickers.add(ticker) moderate_ranked: list[tuple[float, float, float, float, str]] = [] for ticker, date_map in ticker_day_bar.items(): bar = date_map.get(day) if not bar: continue info = enrichment.get(ticker, {}).get(day, {}) gap_pct = overlay_gap_pct(info, bar) if moderate_min_gap is not None and (gap_pct is None or gap_pct < moderate_min_gap): continue if moderate_max_gap is not None and (gap_pct is None or gap_pct > moderate_max_gap): continue avg_dollar_vol = info.get("avg_dollar_vol_30d") if moderate_min_avg_dollar_vol is not None and ( avg_dollar_vol is None or avg_dollar_vol < moderate_min_avg_dollar_vol ): continue if moderate_max_avg_dollar_vol is not None and ( avg_dollar_vol is None or avg_dollar_vol > moderate_max_avg_dollar_vol ): continue ret_5d = info.get("ret_5d") if moderate_min_ret_5d is not None and (ret_5d is None or ret_5d < moderate_min_ret_5d): continue entropy_20d = info.get("entropy_20d") if moderate_max_entropy is not None and ( entropy_20d is None or entropy_20d > moderate_max_entropy ): continue moderate_ranked.append( ( float(gap_pct or 0.0), float(avg_dollar_vol or 0.0), float(ret_5d or 0.0), -float(entropy_20d if entropy_20d is not None else 1.0), ticker, ) ) if moderate_ranked: ranked_moderate = [ ticker for *_rest, ticker in sorted(moderate_ranked, reverse=True)[:moderate_slots] ] for ticker in ranked_moderate: if ticker not in chosen: day_candidates.append(ticker) chosen.add(ticker) day_overlay_tickers.add(ticker) enrichment.setdefault(ticker, {}).setdefault(day, {})[ "candidate_seed_moderate_liquid_overlay" ] = True event_ranked: list[tuple[float, float, float, float, float, str]] = [] for ticker, date_map in ticker_day_bar.items(): bar = date_map.get(day) if not bar: continue info = enrichment.get(ticker, {}).get(day, {}) if not bool(info.get("event_flag")): continue if not _momentum_candidate_event_types_pass(info, strategy): continue event_score = float(info.get("event_score") or 0.0) if event_min_score is not None and event_score < float(event_min_score): continue gap_pct = overlay_gap_pct(info, bar) if event_min_gap is not None and (gap_pct is None or gap_pct < event_min_gap): continue if event_max_gap is not None and (gap_pct is None or gap_pct > event_max_gap): continue avg_dollar_vol = info.get("avg_dollar_vol_30d") if event_min_avg_dollar_vol is not None and ( avg_dollar_vol is None or avg_dollar_vol < event_min_avg_dollar_vol ): continue ret_5d = info.get("ret_5d") if event_min_ret_5d is not None and (ret_5d is None or ret_5d < event_min_ret_5d): continue entropy_20d = info.get("entropy_20d") if event_max_entropy is not None and ( entropy_20d is None or entropy_20d > event_max_entropy ): continue event_ranked.append( ( event_score, float(avg_dollar_vol or 0.0), float(gap_pct or 0.0), float(ret_5d or 0.0), -float(entropy_20d if entropy_20d is not None else 1.0), ticker, ) ) if event_ranked: ranked_events = [ ticker for *_rest, ticker in sorted(event_ranked, reverse=True)[:event_slots] ] for ticker in ranked_events: if ticker not in chosen: day_candidates.append(ticker) chosen.add(ticker) day_overlay_tickers.add(ticker) ownership_ranked: list[tuple[float, float, float, int, float, str]] = [] for ticker, date_map in ticker_day_bar.items(): bar = date_map.get(day) if not bar: continue info = enrichment.get(ticker, {}).get(day, {}) if not bool(info.get("ownership_13dg_flag")): continue if ownership_initial_only and not bool(info.get("ownership_13dg_initial_flag")): continue strength = float(info.get("ownership_13dg_strength_score") or 0.0) if ownership_min_strength is not None and strength < float(ownership_min_strength): continue gap_pct = overlay_gap_pct(info, bar) if ownership_min_gap is not None and (gap_pct is None or gap_pct < ownership_min_gap): continue if ownership_max_gap is not None and (gap_pct is None or gap_pct > ownership_max_gap): continue avg_dollar_vol = info.get("avg_dollar_vol_30d") if ownership_min_avg_dollar_vol is not None and ( avg_dollar_vol is None or avg_dollar_vol < ownership_min_avg_dollar_vol ): continue ret_5d = info.get("ret_5d") if ownership_min_ret_5d is not None and (ret_5d is None or ret_5d < ownership_min_ret_5d): continue entropy_20d = info.get("entropy_20d") if ownership_max_entropy is not None and ( entropy_20d is None or entropy_20d > ownership_max_entropy ): continue days_since = int(info.get("ownership_13dg_days_since") or 9999) ownership_ranked.append( ( strength, float(avg_dollar_vol or 0.0), float(ret_5d or 0.0), -days_since, -float(entropy_20d if entropy_20d is not None else 1.0), ticker, ) ) if ownership_ranked: ranked_ownership = [ ticker for *_rest, ticker in sorted(ownership_ranked, reverse=True)[:ownership_slots] ] for ticker in ranked_ownership: if ticker not in chosen: day_candidates.append(ticker) chosen.add(ticker) day_overlay_tickers.add(ticker) enrichment.setdefault(ticker, {}).setdefault(day, {})[ "candidate_seed_ownership_overlay" ] = True form4_ranked: list[tuple[float, int, int, float, float, str]] = [] for ticker, date_map in ticker_day_bar.items(): bar = date_map.get(day) if not bar: continue info = enrichment.get(ticker, {}).get(day, {}) if not bool(info.get("form4_flag")): continue total_value = float(info.get("form4_total_value") or 0.0) if form4_min_total_value is not None and total_value < float(form4_min_total_value): continue owner_count = int(info.get("form4_owner_count") or 0) c_suite_count = int(info.get("form4_c_suite_count") or 0) if form4_require_cluster_or_csuite: cluster_ok = ( form4_min_owner_count is not None and owner_count >= int(form4_min_owner_count) ) csuite_ok = ( form4_min_c_suite_count is not None and c_suite_count >= int(form4_min_c_suite_count) ) if not (cluster_ok or csuite_ok): continue else: if form4_min_owner_count is not None and owner_count < int(form4_min_owner_count): continue if ( form4_min_c_suite_count is not None and c_suite_count < int(form4_min_c_suite_count) ): continue gap_pct = overlay_gap_pct(info, bar) if form4_min_gap is not None and (gap_pct is None or gap_pct < form4_min_gap): continue if form4_max_gap is not None and (gap_pct is None or gap_pct > form4_max_gap): continue avg_dollar_vol = info.get("avg_dollar_vol_30d") if form4_min_avg_dollar_vol is not None and ( avg_dollar_vol is None or avg_dollar_vol < form4_min_avg_dollar_vol ): continue ret_5d = info.get("ret_5d") if form4_min_ret_5d is not None and (ret_5d is None or ret_5d < form4_min_ret_5d): continue entropy_20d = info.get("entropy_20d") if form4_max_entropy is not None and ( entropy_20d is None or entropy_20d > form4_max_entropy ): continue days_since = int(info.get("form4_days_since") or 9999) role_weight = float(info.get("form4_role_weight_score") or 0.0) form4_ranked.append( ( role_weight, owner_count, c_suite_count, total_value, -days_since, ticker, ) ) if form4_ranked: ranked_form4 = [ ticker for *_rest, ticker in sorted(form4_ranked, reverse=True)[:form4_slots] ] for ticker in ranked_form4: if ticker not in chosen: day_candidates.append(ticker) chosen.add(ticker) day_overlay_tickers.add(ticker) enrichment.setdefault(ticker, {}).setdefault(day, {})[ "candidate_seed_form4_overlay" ] = True leader_ranked: list[tuple[float, float, float, float, str]] = [] for ticker, date_map in ticker_day_bar.items(): bar = date_map.get(day) if not bar: continue info = enrichment.get(ticker, {}).get(day, {}) gap_pct = overlay_gap_pct(info, bar) if leader_min_gap is not None and (gap_pct is None or gap_pct < leader_min_gap): continue if leader_max_gap is not None and (gap_pct is None or gap_pct > leader_max_gap): continue avg_dollar_vol = info.get("avg_dollar_vol_30d") if leader_min_avg_dollar_vol is not None and ( avg_dollar_vol is None or avg_dollar_vol < leader_min_avg_dollar_vol ): continue ret_5d = info.get("ret_5d") if leader_min_ret_5d is not None and (ret_5d is None or ret_5d < leader_min_ret_5d): continue entropy_20d = info.get("entropy_20d") if leader_max_entropy is not None and ( entropy_20d is None or entropy_20d > leader_max_entropy ): continue atr_14 = info.get("atr_14") open_price = bar.get("open") atr_pct = ( float(atr_14) / float(open_price) if atr_14 is not None and open_price not in (None, 0) else None ) if leader_min_atr_pct is not None and (atr_pct is None or atr_pct < leader_min_atr_pct): continue leader_ranked.append( ( float(ret_5d or 0.0), float(avg_dollar_vol or 0.0), float(atr_pct or 0.0), -float(entropy_20d or 1.0), ticker, ) ) if leader_ranked: ranked_leaders = [ ticker for *_rest, ticker in sorted(leader_ranked, reverse=True)[:leader_slots] ] for ticker in ranked_leaders: if ticker not in chosen: day_candidates.append(ticker) chosen.add(ticker) day_overlay_tickers.add(ticker) if day_candidates: result[day] = day_candidates if day_overlay_tickers: overlay_by_day[day] = day_overlay_tickers return result, overlay_by_day def _normalize_candidate_map( candidates: dict[str, list[str]] | tuple[dict[str, list[str]], object], ) -> dict[str, list[str]]: """Normalize helper output to a plain {day: tickers} map. Some call sites work with helpers that return `(candidate_map, metadata)`. Normalizing here keeps run/sweep code robust and avoids shape drift across paths. """ if isinstance(candidates, tuple): candidates = candidates[0] return { day: list(day_tickers) for day, day_tickers in candidates.items() if day_tickers } def _orb_intraday_support_tickers(params: ORBStrategyParams) -> list[str]: """Extra intraday-only tickers needed by the ORB simulator beyond trade candidates.""" support: set[str] = set() uses_market_orb_quality = any( getattr(params, key, None) is not None for key in ( "market_orb_quality_size_scale_low", "market_orb_quality_size_scale_high", "market_orb_quality_primary_strong_above", "market_orb_quality_secondary_weak_above", "market_orb_quality_secondary_weak_below", "market_orb_quality_divergence_scale", "market_orb_quality_primary_weak_below", "market_orb_quality_primary_weak_above", "market_orb_quality_secondary_strong_above", "market_orb_quality_primary_weak_secondary_strong_scale", "market_orb_quality_primary_lag_above", "market_orb_quality_primary_lag_below", "market_orb_quality_secondary_lead_above", "market_orb_quality_secondary_lead_below", "market_orb_quality_primary_lag_secondary_lead_scale", "market_orb_quality_joint_weak_primary_below", "market_orb_quality_joint_weak_primary_above", "market_orb_quality_joint_weak_secondary_below", "market_orb_quality_joint_weak_secondary_above", "market_orb_quality_joint_weak_scale", "market_orb_quality_joint_panic_primary_below", "market_orb_quality_joint_panic_secondary_below", "market_orb_quality_joint_panic_scale", ) ) if ( uses_market_orb_quality or bool(getattr(params, "market_orb_quality_joint_weak_require_confirmation", False)) or bool(getattr(params, "market_orb_quality_joint_panic_require_confirmation", False)) ): quality_ticker = ( getattr(params, "market_orb_quality_ticker", None) or getattr(params, "market_regime_ticker", None) or "SPY" ) if quality_ticker: support.add(str(quality_ticker).upper()) secondary_quality_ticker = getattr(params, "market_orb_quality_secondary_ticker", None) if secondary_quality_ticker: support.add(str(secondary_quality_ticker).upper()) conditional_confirmation_ticker = getattr(params, "conditional_confirmation_ticker", None) if conditional_confirmation_ticker: support.add(str(conditional_confirmation_ticker).upper()) if _orb_strategy_uses_idle_sleeve(params): for ticker in getattr(params, "orb_idle_sleeve_parking_symbols", []) or []: if ticker: support.add(str(ticker).upper()) for attr in ( "orb_idle_sleeve_parking_base_symbol", "orb_idle_sleeve_parking_overlay_symbol", "orb_idle_sleeve_parking_defensive_symbol", ): ticker = getattr(params, attr, None) if ticker: support.add(str(ticker).upper()) for ticker in getattr(params, "orb_idle_sleeve_risk_off_symbols", []) or []: if ticker: support.add(str(ticker).upper()) for ticker in getattr(params, "orb_idle_sleeve_sector_rotation_symbols", []) or []: if ticker: support.add(str(ticker).upper()) market_ticker = getattr(params, "orb_idle_sleeve_market_ticker", None) if market_ticker: support.add(str(market_ticker).upper()) return sorted(support) def _orb_strategy_uses_idle_sleeve(params: ORBStrategyParams) -> bool: return bool(getattr(params, "orb_idle_sleeve_enabled", False)) def _augment_candidate_map_with_support_tickers( candidates: dict[str, list[str]], support_tickers: list[str], trading_days: list[str] | None = None, *, include_empty_days: bool = False, ) -> dict[str, list[str]]: """Append support tickers to each existing candidate day without duplicating names.""" if not support_tickers: return candidates result: dict[str, list[str]] = {} days = trading_days if include_empty_days and trading_days is not None else list(candidates.keys()) for day in days: day_tickers = candidates.get(day, []) merged = list(day_tickers) seen = set(merged) for ticker in support_tickers: if ticker not in seen: merged.append(ticker) seen.add(ticker) result[day] = merged return result def _merge_candidate_maps(candidate_maps: list[dict[str, list[str]]]) -> dict[str, list[str]]: """Union multiple candidate maps day-by-day without duplicating tickers.""" merged_by_day: dict[str, list[str]] = {} for candidate_map in candidate_maps: for day, tickers in candidate_map.items(): day_list = merged_by_day.setdefault(day, []) seen = set(day_list) for ticker in tickers: if ticker not in seen: day_list.append(ticker) seen.add(ticker) return merged_by_day def _momentum_candidate_event_tickers( candidates: dict[str, list[str]], ) -> list[str]: selected: set[str] = set() for day_tickers in candidates.values(): selected.update(day_tickers) return sorted(selected) def _momentum_candidate_event_pairs( candidates: dict[str, list[str]], enrichment: dict[str, dict[str, dict]], strategy: StrategyParams, ) -> list[tuple[str, str]]: per_day_limit = max(strategy.top_n * 15, 60) selected: list[tuple[str, str]] = [] for day, day_tickers in candidates.items(): ranked = sorted( day_tickers, key=lambda ticker: ( float(enrichment.get(ticker, {}).get(day, {}).get("gap_pct") or 0.0), float(enrichment.get(ticker, {}).get(day, {}).get("ret_5d") or 0.0), -float(enrichment.get(ticker, {}).get(day, {}).get("entropy_20d") or 1.0), float(enrichment.get(ticker, {}).get(day, {}).get("avg_dollar_vol_30d") or 0.0), ), reverse=True, ) for ticker in ranked[:per_day_limit]: selected.append((ticker, day)) return selected def _recent_intraday_first_candidates( all_intraday: dict[str, dict[str, list[dict]]], trading_days: list[str], strategy: StrategyParams, ) -> dict[str, list[str]]: """Build a recent-day candidate shortlist directly from intraday action. This path is only used for very recent sanity windows where a static universe and daily-first pre-screen can miss obvious day leaders (for example names that are absent from the research universe but present in today's Yahoo gainers list). """ result: dict[str, list[str]] = {} shortlist_size = max(strategy.top_n * 25, strategy.recent_live_scan_max_candidates_per_day) volume_gate = max(100_000, int((strategy.min_entry_volume or 0) * 0.5)) liquid_dollar_gate = max( 50_000_000.0, float(strategy.min_entry_dollar_volume or 0.0) * 10.0, ) largecap_dollar_gate = max(liquid_dollar_gate, 100_000_000.0) for day in trading_days: bars_by_ticker = all_intraday.get(day, {}) if not bars_by_ticker: continue market_open = _market_open_ts(day) fast_scored: list[tuple[float, float, str]] = [] largecap_scored: list[tuple[float, float, float, str]] = [] liquid_scored: list[tuple[float, float, float, str]] = [] slow_scored: list[tuple[float, float, float, str]] = [] for ticker, all_bars in bars_by_ticker.items(): mkt_bars = filter_market_hours(all_bars) if len(mkt_bars) < 5: continue open_price = float(mkt_bars[0].get("open", 0.0) or 0.0) if open_price <= 0: continue entry_bar = _bar_at_offset(mkt_bars, market_open, strategy.entry_minutes_after_open) if entry_bar is None: continue entry_price = float(entry_bar.get("close", 0.0) or 0.0) if entry_price <= 0: continue gain_pct = (entry_price - open_price) / open_price confirmation_bar = entry_bar if strategy.confirmation_minutes_after_entry > 0: confirmation_bar = _bar_at_offset( mkt_bars, market_open, strategy.entry_minutes_after_open + strategy.confirmation_minutes_after_entry, ) if confirmation_bar is None: continue confirmation_price = float(confirmation_bar.get("close", 0.0) or 0.0) if confirmation_price <= 0: continue confirmation_gain_pct = (confirmation_price - open_price) / open_price confirmation_return = (confirmation_price - entry_price) / entry_price if entry_price > 0 else 0.0 entry_ts = _parse_et_bar_timestamp(confirmation_bar["timestamp"]) entry_volume = _volume_up_to_bar(mkt_bars, entry_ts) if entry_volume < volume_gate: continue entry_dollar_volume = _dollar_volume_up_to_bar(mkt_bars, entry_ts) if gain_pct > 0: fast_scored.append((gain_pct, entry_volume, ticker)) if ( confirmation_gain_pct >= 0.002 and confirmation_return >= -0.001 and entry_dollar_volume >= largecap_dollar_gate ): largecap_scored.append((entry_dollar_volume, confirmation_return, confirmation_gain_pct, ticker)) if ( confirmation_gain_pct >= 0.003 and confirmation_return >= 0.0 and entry_dollar_volume >= liquid_dollar_gate ): liquid_scored.append((entry_dollar_volume, confirmation_gain_pct, confirmation_return, ticker)) if ( confirmation_gain_pct >= 0.003 and confirmation_gain_pct < max(strategy.min_morning_gain_pct, 0.015) and confirmation_return >= 0.0 and entry_dollar_volume >= liquid_dollar_gate ): slow_scored.append((confirmation_return, entry_dollar_volume, confirmation_gain_pct, ticker)) if not fast_scored and not largecap_scored and not liquid_scored and not slow_scored: continue fast_slots = max(strategy.top_n * 8, int(shortlist_size * 0.45)) largecap_slots = max(strategy.top_n * 4, int(shortlist_size * 0.20)) liquid_slots = max(strategy.top_n * 4, int(shortlist_size * 0.20)) slow_slots = max(strategy.top_n * 2, shortlist_size - fast_slots - largecap_slots - liquid_slots) ranked_fast = [ticker for _gain, _vol, ticker in sorted(fast_scored, reverse=True)[:fast_slots]] ranked_largecap = [ ticker for _dvol, _conf_ret, _gain, ticker in sorted(largecap_scored, reverse=True)[:largecap_slots] ] ranked_liquid = [ ticker for _dvol, _gain, _conf, ticker in sorted(liquid_scored, reverse=True)[:liquid_slots] ] ranked_slow = [ ticker for _conf, _dvol, _gain, ticker in sorted(slow_scored, reverse=True)[:slow_slots] ] merged: list[str] = [] for source in (ranked_fast, ranked_largecap, ranked_liquid, ranked_slow): for ticker in source: if ticker not in merged: merged.append(ticker) if len(merged) >= shortlist_size: break if len(merged) >= shortlist_size: break if merged: result[day] = merged return result def _retain_recent_intraday_shortlist( candidates: dict[str, list[str]], daily_bars: dict[str, list[dict]], *, require_daily_features: bool, ) -> dict[str, list[str]]: """Keep the intraday-first shortlist instead of overwriting it with daily pre-screen. When momentum strategies need daily enrichment, recent same-day leaders still need a cached daily history row for gap/entropy/trend features. Otherwise retain the intraday-first shortlist as-is. """ if not require_daily_features: return {day: list(day_tickers) for day, day_tickers in candidates.items() if day_tickers} retained: dict[str, list[str]] = {} for day, day_tickers in candidates.items(): kept = [ticker for ticker in day_tickers if ticker in daily_bars] if kept: retained[day] = kept return retained def _parse_et_bar_timestamp(timestamp: str): from libs.intraday.simulator import _parse_ts return _parse_ts(timestamp) async def _fetch_vix_by_day( client: OracleClient, trading_days: list[str], ) -> dict[str, float]: if not trading_days: return {} try: if not await client.health_check_fast(): return _load_vix_from_local_macro_snapshots(trading_days) except Exception: return _load_vix_from_local_macro_snapshots(trading_days) try: fred = FredService(client) response = await fred.get_observations("VIXCLS", start=trading_days[0], end=trading_days[-1]) result: dict[str, float] = {} for obs in response.observations: if obs.value is None: continue result[obs.date] = float(obs.value) if result: return result except Exception: pass return _load_vix_from_local_macro_snapshots(trading_days) _MACRO_WINDOW_RE = re.compile(r"macro_window_(\d{4}-\d{2}-\d{2})_(\d{4}-\d{2}-\d{2})\.pkl$") def _load_vix_from_local_macro_snapshots(trading_days: list[str]) -> dict[str, float]: """Best-effort local fallback when Oracle/FRED is unavailable. Several backtester workflows persist macro_window_*.pkl files with point-in-time VIX values. Reusing them keeps intraday research and official backtests from hard-failing when the FRED proxy is temporarily down. """ if not trading_days: return {} settings = get_settings() root = Path(settings.data_root) / "parquet" if not root.exists(): return {} start = trading_days[0] end = trading_days[-1] best_path: Path | None = None best_span: int | None = None for path in root.rglob("macro_window_*.pkl"): match = _MACRO_WINDOW_RE.search(path.name) if not match: continue window_start, window_end = match.groups() if window_start > start or window_end < end: continue span = (date.fromisoformat(window_end) - date.fromisoformat(window_start)).days if best_span is None or span < best_span: best_span = span best_path = path if best_path is None: return {} try: with best_path.open("rb") as fh: payload = pickle.load(fh) except Exception: return {} if not isinstance(payload, dict): return {} result: dict[str, float] = {} for key, values in payload.items(): if not isinstance(values, dict): continue vix_value = values.get("VIXCLS") if vix_value is None: continue if hasattr(key, "isoformat"): key_str = key.isoformat() else: key_str = str(key) if start <= key_str <= end: result[key_str] = float(vix_value) return result def _momentum_enrichment_for_days( daily_bars: dict[str, list[dict]], trading_days: list[str], ) -> dict[str, dict[str, dict]]: enrichment = enrich_daily_bars(daily_bars, trading_days) for ticker, day_map in enrichment.items(): ticker_bars = sorted(daily_bars.get(ticker, []), key=lambda bar: bar["date"]) by_day = {bar["date"][:10]: bar for bar in ticker_bars} for day, features in day_map.items(): prev_close = features.get("prev_close") today_open = features.get("today_open") features["gap_pct"] = ( compute_gap_pct(prev_close, today_open) if prev_close is not None and today_open is not None else None ) today_bar = by_day.get(day) if today_bar and today_bar.get("volume") and features.get("avg_daily_vol_14d"): avg_daily_vol = features["avg_daily_vol_14d"] features["daily_volume_ratio_14d"] = ( today_bar["volume"] / avg_daily_vol if avg_daily_vol and avg_daily_vol > 0 else None ) else: features["daily_volume_ratio_14d"] = None return enrichment def _append_synthetic_today_daily_rows( daily_bars: dict[str, list[dict]], trading_days: list[str], ) -> int: """Add a placeholder row for the current ET trading day when daily data is stale. Same-day daily endpoints often lag intraday/today. The placeholder only makes prior-bar enrichment keys exist; actual entry open/gap still comes from intraday bars. """ if not trading_days: return 0 today_et = to_eastern(utc_now()).date().isoformat() if today_et not in set(trading_days): return 0 added = 0 for ticker, bars in list(daily_bars.items()): if not bars: continue sorted_bars = sorted(bars, key=lambda bar: str(bar.get("date", ""))[:10]) if any(str(bar.get("date", ""))[:10] == today_et for bar in sorted_bars): daily_bars[ticker] = sorted_bars continue prior_bars = [ bar for bar in sorted_bars if str(bar.get("date", ""))[:10] < today_et ] if not prior_bars: daily_bars[ticker] = sorted_bars continue last = prior_bars[-1] prev_close = float(last.get("close", 0.0) or 0.0) if prev_close <= 0: daily_bars[ticker] = sorted_bars continue daily_bars[ticker] = sorted_bars + [ { "date": today_et, "open": prev_close, "high": prev_close, "low": prev_close, "close": prev_close, "volume": 0.0, "synthetic_today_daily": True, } ] added += 1 return added # ── Progress Reporting ───────────────────────────────────────────────────── def _make_progress_bar(completed: int, total: int, width: int = 30) -> str: pct = completed / total if total > 0 else 0 filled = int(width * pct) bar = "█" * filled + "░" * (width - filled) return f"[{bar}] {completed}/{total} ({pct*100:.0f}%)" def _chunk_trading_days_by_pairs( trading_days: list[str], candidates: dict[str, list[str]], max_pairs_per_chunk: int = 5_000, ) -> list[list[str]]: """Split trading days into contiguous chunks capped by candidate pair count. ORB single-run backtests previously loaded every candidate day's 5-minute bars into memory before simulation. Large windows can exceed multiple GB in Python objects, so we stream a few days at a time instead. """ chunks: list[list[str]] = [] current: list[str] = [] current_pairs = 0 for day in trading_days: day_pairs = len(candidates.get(day, [])) if current and current_pairs + day_pairs > max_pairs_per_chunk: chunks.append(current) current = [] current_pairs = 0 current.append(day) current_pairs += day_pairs if current: chunks.append(current) return chunks # ── Main Orchestrator ────────────────────────────────────────────────────── async def run(config: IntradayConfig, refresh_cache: bool = False) -> tuple: """Full pipeline: universe → daily bars → intraday bars → simulate → metrics. Returns (day_results, metrics, all_intraday, trading_days). """ settings = get_settings() is_orb = config.strategy_mode == "orb" if is_orb: p = config.orb_strategy or ORBStrategyParams() timeout_minutes = 9 * 60 + 30 + p.order_timeout_minutes timeout_hour, timeout_minute = divmod(timeout_minutes, 60) print( f"\nORB | {config.universe.source} | ${p.initial_capital:,.0f} | " f"last {config.backtest.lookback_trading_days}d | " f"orb:{p.orb_minutes}min stop:{p.atr_stop_multiplier}xATR " f"be:{p.breakeven_at_r}R tr:{p.trailing_at_r}R timeout:{timeout_hour:02d}:{timeout_minute:02d} " f"rvol:{p.min_rvol} risk:{p.risk_per_trade_pct*100:.2f}%" ) else: s = config.strategy print( f"\nMoMo | {config.universe.source} | ${s.initial_capital:,.0f} | " f"last {config.backtest.lookback_trading_days}d | " f"entry:+{s.entry_minutes_after_open}min exit:-{s.exit_minutes_before_close}min " f"stop:{s.stop_loss_pct or 'off'} gain:{s.min_morning_gain_pct*100:.1f}% top:{s.top_n}" ) cache = IntradayCache(config.cache.dir) if config.cache.enabled else None daily_cache = _build_daily_cache(config) if config.cache.enabled else None orb_snapshot_id = ( getattr(config.orb_strategy, "prior_event_snapshot_id", None) if is_orb and config.orb_strategy is not None else None ) event_cache = ( FilingEventCache(str(Path(config.cache.dir).with_name("orb_catalyst"))) if config.cache.enabled else None ) prior_event_snapshot_cache = PriorEventFeatureSnapshotCache( str(Path(config.cache.dir).with_name("orb_prior_event")), snapshot_id=orb_snapshot_id, ) attention_cache = ( AttentionEventCache(str(Path(config.cache.dir).with_name("orb_attention"))) if config.cache.enabled else None ) attention_cache = ( AttentionEventCache(str(Path(config.cache.dir).with_name("orb_attention"))) if config.cache.enabled else None ) if refresh_cache and cache: print("\n Refreshing cache (evicting all entries)...") removed = cache.evict() print(f" Removed {removed} cached files.") if cache: stats = cache.stats() print(f"\n Cache: {stats['total_files']} files, {stats['total_mb']} MB") async with make_intraday_oracle_client(settings) as client: # Step 1: Get trading days print("[1/4] Resolving trading calendar...") trading_days = await get_trading_days( client, config.backtest.start_date, config.backtest.end_date, config.backtest.lookback_trading_days, ) print(f" {trading_days[0]} → {trading_days[-1]} ({len(trading_days)} days)") # Step 2: Resolve universe recent_live_scan = (not is_orb) and _should_use_recent_live_scan(config.strategy, trading_days) if not is_orb: config = config.model_copy( update={"strategy": _strategy_for_recent_live_scan(config.strategy, recent_live_scan=recent_live_scan)} ) universe_params = _recent_live_scan_universe(config.strategy) if recent_live_scan else config.universe universe_label = ( f"{config.universe.source} + recent-live-scan" if recent_live_scan else config.universe.source ) recent_intraday_first_scan = recent_live_scan and _should_use_recent_intraday_first_scan(trading_days) print(f"[2/4] Resolving universe ({universe_label})...") tickers = await resolve_universe(universe_params, client) print(f" {len(tickers)} tickers") if recent_intraday_first_scan: print( f"[3/4] Phase 1: Fetching intraday bars for recent live scan " f"({len(tickers)} tickers across {len(trading_days)} days)..." ) intraday_seed = {day: tickers for day in trading_days} _live_last_pct = [-1] def live_intraday_progress(completed: int, total: int, hits: int, calls: int) -> None: if completed == 0 and calls == 0 and total > 0: sys.stdout.write("\n") sys.stdout.flush() _live_last_pct[0] = -1 pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _live_last_pct[0] or completed == total: _live_last_pct[0] = pct sys.stdout.write( f"\r {_make_progress_bar(completed, total)} cache:{hits} api:{calls}" ) sys.stdout.flush() all_intraday = await fetch_intraday_bulk( intraday_seed, client, cache, concurrency=8, progress_callback=live_intraday_progress, ) print() candidates = _recent_intraday_first_candidates(all_intraday, trading_days, config.strategy) total_pairs = sum(len(v) for v in candidates.values()) print( f" Intraday-first shortlisted: {total_pairs} ticker-day pairs " f"across {len(candidates)} days" ) shortlisted_tickers = sorted({ticker for day in candidates.values() for ticker in day}) daily_bars: dict[str, list[dict]] = {} if shortlisted_tickers and _momentum_strategy_uses_daily_enrichment(config.strategy): print(f" Fetching daily enrichment bars for {len(shortlisted_tickers)} shortlisted tickers...") n_done = [0] _daily_last_pct = [-1] def daily_progress(completed: int, total: int) -> None: n_done[0] = completed pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _daily_last_pct[0] or completed == total: _daily_last_pct[0] = pct sys.stdout.write(f"\r {_make_progress_bar(completed, total)}") sys.stdout.flush() daily_fetch_start = ( date.fromisoformat(trading_days[0]) - timedelta(days=90) ).isoformat() daily_bars = await fetch_daily_bars_bulk( shortlisted_tickers, daily_fetch_start, trading_days[-1], client, cache=daily_cache, intraday_cache_fallback=cache, prefer_intraday_fallback=True, concurrency=20, progress_callback=daily_progress, ) print(f"\n {len(daily_bars)}/{len(shortlisted_tickers)} shortlisted tickers with daily data") else: # Step 3: Phase 1 — Fetch daily bars + pre-screen print(f"[3/4] Phase 1: Fetching daily bars for {len(tickers)} tickers...") ticker_sectors = ( _load_orb_ticker_sectors(tickers) if is_orb else ( await _load_ticker_sectors_with_oracle(tickers, client) if _momentum_strategy_uses_sector_labels(config.strategy) else {} ) ) n_done = [0] _daily_last_pct = [-1] def daily_progress(completed: int, total: int) -> None: n_done[0] = completed pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _daily_last_pct[0] or completed == total: _daily_last_pct[0] = pct sys.stdout.write(f"\r {_make_progress_bar(completed, total)}") sys.stdout.flush() # For ORB mode, fetch extra prior history for enrichment (ATR/volume warmup). # enrich_daily_bars() uses only bars BEFORE each trading day, so extra bars # before trading_days[0] act as warmup and never appear in simulation results. # Without this, a short (e.g. single-day) backtest has no prior bars and all # candidates get filtered out (ATR/dollar-vol = None → zero trades). if not recent_intraday_first_scan: if is_orb or _momentum_strategy_uses_daily_enrichment(config.strategy): warmup_start = ( date.fromisoformat(trading_days[0]) - timedelta(days=90) ).isoformat() daily_fetch_start = warmup_start else: daily_fetch_start = trading_days[0] daily_bars = await fetch_daily_bars_bulk( tickers, daily_fetch_start, trading_days[-1], client, cache=daily_cache, intraday_cache_fallback=cache, prefer_intraday_fallback=True, skip_oracle_when_unhealthy=True, concurrency=20, progress_callback=daily_progress, ) print(f"\n {len(daily_bars)}/{len(tickers)} tickers with data") momentum_enrichment: dict[str, dict[str, dict]] | None = None momentum_vix_by_day: dict[str, float] | None = None momentum_seed_candidates: dict[str, list[str]] | None = None if is_orb: # ORB: enrich daily bars, then filter by quality metrics orb_params = config.orb_strategy or ORBStrategyParams() # Ensure regime ticker is in daily_bars for market regime filter regime_ticker = getattr(orb_params, "market_regime_ticker", "SPY") or "SPY" if _orb_strategy_requires_regime_ticker_daily(orb_params) and regime_ticker not in daily_bars: extra_bars = await fetch_daily_bars_bulk( [regime_ticker], daily_fetch_start, trading_days[-1], client, cache=daily_cache, intraday_cache_fallback=cache, concurrency=1 ) daily_bars.update(extra_bars) print(" Computing ATR/volume enrichment...") synthetic_today_rows = _append_synthetic_today_daily_rows(daily_bars, trading_days) if synthetic_today_rows: print( " Added synthetic same-day daily rows " f"for {synthetic_today_rows} tickers" ) enrichment = enrich_daily_bars(daily_bars, trading_days) candidates = orb_pre_screen_candidates( daily_bars, trading_days, enrichment, min_price=orb_params.min_price, min_atr=orb_params.min_atr_14, min_avg_dollar_vol=orb_params.min_avg_dollar_volume, max_per_day=None, ) orb_overlay_tickers_per_day: dict[str, set[str]] | None = None orb_ownership_features: dict[str, dict[str, dict]] | None = None orb_form4_features: dict[str, dict[str, dict]] | None = None if _orb_strategy_uses_ownership_seed_overlay(orb_params): all_tickers = sorted(str(ticker).upper() for ticker in daily_bars.keys()) print( " Loading PIT 13D/13G ownership seed features " f"(D-{int(getattr(orb_params, 'ownership_13dg_lookback_days', 0) or 0)}) " f"for {len(all_tickers)} tickers..." ) orb_ownership_features = _load_orb_ownership_13dg_features( all_tickers, trading_days, orb_params, ) _merge_orb_ownership_13dg_features(enrichment, orb_ownership_features) print(f" 13D/13G seed coverage: {len(orb_ownership_features)} tickers with events") if _orb_strategy_uses_form4_seed_overlay(orb_params): all_tickers = sorted(str(ticker).upper() for ticker in daily_bars.keys()) print( " Loading PIT Form 4 seed features " f"(D-{int(getattr(orb_params, 'form4_lookback_days', 0) or 0)}) " f"for {len(all_tickers)} tickers..." ) orb_form4_features = _load_orb_form4_features( all_tickers, trading_days, orb_params, ) _merge_orb_form4_features(enrichment, orb_form4_features) print(f" Form 4 seed coverage: {len(orb_form4_features)} tickers with events") if ( int(getattr(orb_params, "candidate_seed_leader_overlay_slots", 0) or 0) > 0 or int(getattr(orb_params, "candidate_seed_liquid_overlay_slots", 0) or 0) > 0 or int(getattr(orb_params, "candidate_seed_moderate_liquid_overlay_slots", 0) or 0) > 0 or int(getattr(orb_params, "candidate_seed_event_overlay_slots", 0) or 0) > 0 or _orb_strategy_uses_ownership_seed_overlay(orb_params) or _orb_strategy_uses_form4_seed_overlay(orb_params) ): candidates, orb_overlay_tickers_per_day = _augment_momentum_seed_candidates_with_liquid_overlay( # type: ignore[arg-type] candidates, daily_bars, trading_days, enrichment, orb_params ) if _orb_strategy_uses_catalyst(orb_params): _prior_lookback = int(getattr(orb_params, "prior_event_lookback_days", 0) or 0) if _prior_lookback > 0: all_tickers = list({t for day_tickers in candidates.values() for t in day_tickers}) _prior_event_types = tuple(getattr(orb_params, "prior_event_types", None) or ["earnings_release", "guidance_update"]) print( " Prefetching prior-event features " f"(snapshot={prior_event_snapshot_cache.snapshot_id}, " f"D-{_prior_lookback}, types={_prior_event_types}) " f"for {len(all_tickers)} tickers..." ) event_features = await _prefetch_prior_event_features_db( all_tickers, trading_days, lookback_calendar_days=_prior_lookback, event_types=_prior_event_types, snapshot_cache=prior_event_snapshot_cache, require_event_details=_orb_prior_event_decay_enabled(orb_params), ) event_features = _apply_orb_prior_event_decay_features( event_features, orb_params, ) print(f" Prior-event coverage: {len(event_features)} tickers with events") else: event_tickers = _orb_candidate_event_tickers(candidates, enrichment, orb_params) print(f" Fetching filing catalyst events for {len(event_tickers)} tickers...") _evt_last_pct = [-1] def event_progress(completed: int, total: int) -> None: pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _evt_last_pct[0] or completed == total: _evt_last_pct[0] = pct sys.stdout.write(f"\r {_make_progress_bar(completed, total)}") sys.stdout.flush() event_features = await fetch_filing_event_features_bulk( event_tickers, trading_days[0], trading_days[-1], client, cache=event_cache, concurrency=16, progress_callback=event_progress, ) print() _merge_orb_event_features(enrichment, event_features) if _orb_strategy_uses_attention(orb_params): attention_pairs = _orb_candidate_event_pairs(candidates, enrichment, orb_params) print(f" Fetching event attention for {len(attention_pairs)} ticker-days...") _attn_last_pct = [-1] def attention_progress(completed: int, total: int) -> None: pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _attn_last_pct[0] or completed == total: _attn_last_pct[0] = pct sys.stdout.write(f"\r {_make_progress_bar(completed, total)}") sys.stdout.flush() attention_features = await fetch_attention_features_bulk( attention_pairs, client, cache=attention_cache, concurrency=16, progress_callback=attention_progress, ) print() _merge_orb_attention_features(enrichment, attention_features) if _orb_strategy_uses_ownership_13dg(orb_params) and orb_ownership_features is None: all_tickers = list({t for day_tickers in candidates.values() for t in day_tickers}) print( " Loading PIT 13D/13G ownership features " f"(D-{int(getattr(orb_params, 'ownership_13dg_lookback_days', 0) or 0)}) " f"for {len(all_tickers)} tickers..." ) ownership_features = _load_orb_ownership_13dg_features( all_tickers, trading_days, orb_params, ) _merge_orb_ownership_13dg_features(enrichment, ownership_features) print(f" 13D/13G coverage: {len(ownership_features)} tickers with events") if _orb_strategy_uses_form4(orb_params) and orb_form4_features is None: all_tickers = list({t for day_tickers in candidates.values() for t in day_tickers}) print( " Loading PIT Form 4 features " f"(D-{int(getattr(orb_params, 'form4_lookback_days', 0) or 0)}) " f"for {len(all_tickers)} tickers..." ) form4_features = _load_orb_form4_features( all_tickers, trading_days, orb_params, ) _merge_orb_form4_features(enrichment, form4_features) print(f" Form 4 coverage: {len(form4_features)} tickers with events") if _orb_strategy_uses_vix(orb_params): print(" Fetching VIX regime series for ORB...") momentum_vix_by_day = await _fetch_vix_by_day(client, trading_days) candidates = _augment_candidate_map_with_support_tickers( candidates, _orb_intraday_support_tickers(orb_params), trading_days, include_empty_days=_orb_strategy_uses_idle_sleeve(orb_params), ) else: enrichment = {} if recent_intraday_first_scan: candidates = _retain_recent_intraday_shortlist( candidates, daily_bars, require_daily_features=_momentum_strategy_uses_daily_enrichment(config.strategy), ) if _momentum_strategy_requires_regime_ticker_daily(config.strategy): regime_ticker = config.strategy.market_regime_gap_ticker or "SPY" if regime_ticker not in daily_bars: extra_bars = await fetch_daily_bars_bulk( [regime_ticker], daily_fetch_start, trading_days[-1], client, cache=daily_cache, intraday_cache_fallback=cache, prefer_intraday_fallback=True, skip_oracle_when_unhealthy=True, concurrency=1, ) daily_bars.update(extra_bars) if _momentum_strategy_uses_daily_enrichment(config.strategy): print(" Computing momentum daily enrichment...") momentum_enrichment = _momentum_enrichment_for_days(daily_bars, trading_days) if _momentum_strategy_uses_vix(config.strategy): print(" Fetching VIX regime series...") momentum_vix_by_day = await _fetch_vix_by_day(client, trading_days) if ( not recent_intraday_first_scan and momentum_enrichment is not None and ( _momentum_strategy_uses_catalyst(config.strategy) or _momentum_strategy_uses_attention(config.strategy) ) ): preliminary_candidates = _normalize_candidate_map( _momentum_intraday_seed_candidates( daily_bars, trading_days, momentum_enrichment, config.strategy, default_threshold=config.backtest.pre_screen_threshold, use_signal_features=False, ) ) if _momentum_strategy_uses_catalyst(config.strategy): if _momentum_strategy_uses_candidate_stage_catalyst(config.strategy): event_tickers = sorted(daily_bars.keys()) else: event_tickers = _momentum_candidate_event_tickers(preliminary_candidates) print(f" Fetching momentum filing catalysts for {len(event_tickers)} tickers...") _evt_last_pct = [-1] def event_progress(completed: int, total: int) -> None: pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _evt_last_pct[0] or completed == total: _evt_last_pct[0] = pct sys.stdout.write(f"\r {_make_progress_bar(completed, total)}") sys.stdout.flush() event_features = await fetch_filing_event_features_bulk( event_tickers, trading_days[0], trading_days[-1], client, cache=event_cache, concurrency=16, progress_callback=event_progress, ) print() _merge_momentum_event_features(momentum_enrichment, event_features) if _momentum_strategy_uses_attention(config.strategy): attention_pairs = _momentum_candidate_event_pairs( preliminary_candidates, momentum_enrichment, config.strategy, ) print(f" Fetching momentum attention for {len(attention_pairs)} ticker-days...") _attn_last_pct = [-1] def attention_progress(completed: int, total: int) -> None: pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _attn_last_pct[0] or completed == total: _attn_last_pct[0] = pct sys.stdout.write(f"\r {_make_progress_bar(completed, total)}") sys.stdout.flush() attention_features = await fetch_attention_features_bulk( attention_pairs, client, cache=attention_cache, concurrency=16, progress_callback=attention_progress, ) print() _merge_momentum_attention_features(momentum_enrichment, attention_features) if not recent_intraday_first_scan: momentum_seed_candidates = _momentum_intraday_seed_candidates( daily_bars, trading_days, momentum_enrichment or {}, config.strategy, default_threshold=config.backtest.pre_screen_threshold, use_signal_features=True, ) momentum_seed_candidates, _ = _augment_momentum_seed_candidates_with_liquid_overlay( momentum_seed_candidates, daily_bars, trading_days, momentum_enrichment or {}, config.strategy, ) if _momentum_uses_historical_intraday_first(config.strategy): candidates = _normalize_candidate_map(momentum_seed_candidates) else: candidates = momentum_pre_screen_candidates( daily_bars, trading_days, momentum_enrichment or {}, threshold=config.backtest.pre_screen_threshold, max_per_day=config.strategy.candidate_final_max_per_day, strategy=config.strategy, ) total_pairs = sum(len(v) for v in candidates.values()) print(f" Pre-screened: {total_pairs} ticker-day pairs across {len(candidates)} days") if is_orb: from libs.intraday.orb_simulator import run_orb_simulation_with_state # Step 4: ORB fetch + simulate in bounded day batches. # Large windows (e.g. 90k+ ticker-days) can exceed several GB if every # 5-minute bar is materialized in one giant dict before simulation. day_chunks = _chunk_trading_days_by_pairs(trading_days, candidates) print( f"[4/4] Phase 2: Fetching intraday bars ({total_pairs} pairs) " f"in {len(day_chunks)} batches..." ) print(" Streaming ORB simulation to keep memory bounded...") orb_params = config.orb_strategy or ORBStrategyParams() day_results = [] sim_state = None all_intraday: dict[str, dict[str, list[dict]]] = {} simulated_days = 0 for chunk_idx, day_chunk in enumerate(day_chunks, start=1): chunk_candidates = { day: candidates[day] for day in day_chunk if candidates.get(day) } chunk_pairs = sum(len(v) for v in chunk_candidates.values()) print( f"\n Batch {chunk_idx}/{len(day_chunks)}: " f"{day_chunk[0]} → {day_chunk[-1]} ({chunk_pairs} pairs)" ) chunk_last_pct = [-1] def chunk_intraday_progress(completed: int, total: int, hits: int, calls: int) -> None: if completed == 0 and calls == 0 and total > 0: sys.stdout.write("\n") sys.stdout.flush() chunk_last_pct[0] = -1 pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > chunk_last_pct[0] or completed == total: chunk_last_pct[0] = pct sys.stdout.write( f"\r {_make_progress_bar(completed, total)} " f"cache:{hits} api:{calls}" ) sys.stdout.flush() chunk_intraday = await fetch_intraday_bulk( chunk_candidates, client, cache, skip_oracle_when_unhealthy=True, concurrency=8, progress_callback=chunk_intraday_progress, ) if chunk_pairs > 0: print() next_day_after_chunk_idx = simulated_days + len(day_chunk) next_day_after_chunk = ( trading_days[next_day_after_chunk_idx] if next_day_after_chunk_idx < len(trading_days) else None ) chunk_results, sim_state = run_orb_simulation_with_state( chunk_intraday, day_chunk, orb_params, enrichment, ticker_sectors=ticker_sectors, state=sim_state, vix_by_day=momentum_vix_by_day, overlay_tickers_per_day=orb_overlay_tickers_per_day, next_trading_day_after_window=next_day_after_chunk, ) day_results.extend(chunk_results) simulated_days += len(day_chunk) print(f" Simulated {simulated_days}/{len(trading_days)} days") del chunk_intraday print(f"\n Done. Simulated {len(day_results)} days") else: # Step 4: Phase 2 — Fetch intraday bars (cache-first) if recent_intraday_first_scan: print(f"[4/4] Phase 2: Reusing recent live-scan intraday bars ({total_pairs} pairs)...") shortlisted_intraday: dict[str, dict[str, list[dict]]] = {} for day, day_candidates in candidates.items(): day_bars = all_intraday.get(day, {}) subset = { ticker: day_bars[ticker] for ticker in day_candidates if ticker in day_bars } if subset: shortlisted_intraday[day] = subset all_intraday = shortlisted_intraday print(f" Done. {len(all_intraday)} days with shortlisted intraday data") else: print(f"[4/4] Phase 2: Fetching intraday bars ({total_pairs} pairs)...") n_api = [0] _intra_last_pct = [-1] def intraday_progress(completed: int, total: int, hits: int, calls: int) -> None: n_api[0] = calls if completed == 0 and calls == 0 and total > 0: # Phase 2 reset signal: new total = miss_total, restart bar from 0 sys.stdout.write("\n") sys.stdout.flush() _intra_last_pct[0] = -1 pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _intra_last_pct[0] or completed == total: _intra_last_pct[0] = pct sys.stdout.write( f"\r {_make_progress_bar(completed, total)} " f"cache:{hits} api:{calls}" ) sys.stdout.flush() all_intraday = await fetch_intraday_bulk( candidates, client, cache, skip_oracle_when_unhealthy=True, concurrency=8, progress_callback=intraday_progress, ) print(f"\n Done. {len(all_intraday)} days with intraday data") if _momentum_uses_historical_intraday_first(config.strategy): candidates = momentum_intraday_first_candidates( all_intraday, trading_days, config.strategy, daily_enrichment=momentum_enrichment, ticker_sectors=ticker_sectors, max_per_day=config.strategy.candidate_final_max_per_day, ) total_pairs = sum(len(v) for v in candidates.values()) shortlisted_intraday: dict[str, dict[str, list[dict]]] = {} for day, day_candidates in candidates.items(): day_bars = all_intraday.get(day, {}) subset = { ticker: day_bars[ticker] for ticker in day_candidates if ticker in day_bars } if subset: shortlisted_intraday[day] = subset all_intraday = shortlisted_intraday print( " Intraday-first reranked: " f"{total_pairs} ticker-day pairs across {len(candidates)} days" ) sector_proxy_intraday: dict[str, dict[str, list[dict]]] | None = None if (not is_orb) and _momentum_strategy_uses_sector_proxies(config.strategy): print(" Fetching sector ETF proxy bars...") proxy_candidates = {day: list(SECTOR_PROXY_TICKERS) for day in trading_days} _proxy_last_pct = [-1] def proxy_progress(completed: int, total: int, hits: int, calls: int) -> None: if completed == 0 and calls == 0 and total > 0: sys.stdout.write("\n") sys.stdout.flush() _proxy_last_pct[0] = -1 pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _proxy_last_pct[0] or completed == total: _proxy_last_pct[0] = pct sys.stdout.write( f"\r {_make_progress_bar(completed, total)} " f"cache:{hits} api:{calls}" ) sys.stdout.flush() sector_proxy_intraday = await fetch_intraday_bulk( proxy_candidates, client, cache, skip_oracle_when_unhealthy=True, concurrency=4, progress_callback=proxy_progress, ) print(f"\n Done. {len(sector_proxy_intraday)} days with sector ETF proxy data") if not is_orb: # Step 5: Simulate (momentum mode still runs after full preload) print("\nSimulating trades...") _sim_last_pct = [-1] def sim_progress(done: int, total: int) -> None: pct = int(done / total * 10) * 10 if total > 0 else 0 if pct > _sim_last_pct[0] or done == total: _sim_last_pct[0] = pct sys.stdout.write(f"\r {_make_progress_bar(done, total)}") sys.stdout.flush() day_results = run_simulation( all_intraday, trading_days, config.strategy, daily_enrichment=momentum_enrichment, vix_by_day=momentum_vix_by_day, ticker_sectors=ticker_sectors, sector_proxy_intraday_by_day=sector_proxy_intraday, ) print() # Step 6: Compute metrics run_id = str(uuid.uuid4())[:8] metrics = compute_metrics(day_results, config, run_id=run_id) return day_results, metrics, all_intraday, trading_days async def run_with_sweep(config: IntradayConfig, sweep_path: str) -> None: """Run data pipeline once, then sweep over parameter combinations.""" from apps.intraday_bt.sweep import ( apply_overrides, generate_combinations, load_sweep_config, run_sweep, ) import copy is_orb = config.strategy_mode == "orb" sweep = load_sweep_config(sweep_path, config) mode_label = "ORB" if is_orb else "MoMo" print(f"\nSweep [{mode_label}] {sweep.total_combinations} combos | " + " | ".join(f"{k}:{v}" for k, v in sweep.sweep_params.items())) settings = get_settings() cache = IntradayCache(config.cache.dir) if config.cache.enabled else None daily_cache = _build_daily_cache(config) if config.cache.enabled else None orb_snapshot_id = ( getattr(config.orb_strategy, "prior_event_snapshot_id", None) if is_orb and config.orb_strategy is not None else None ) event_cache = ( FilingEventCache(str(Path(config.cache.dir).with_name("orb_catalyst"))) if config.cache.enabled else None ) prior_event_snapshot_cache = PriorEventFeatureSnapshotCache( str(Path(config.cache.dir).with_name("orb_prior_event")), snapshot_id=orb_snapshot_id, ) attention_cache = ( AttentionEventCache(str(Path(config.cache.dir).with_name("orb_attention"))) if config.cache.enabled else None ) async with make_intraday_oracle_client(settings) as client: print(f"\n[1/4] Resolving universe ({config.universe.source})...") tickers = await resolve_universe(config.universe, client) print(f" {len(tickers)} tickers") print("[2/4] Resolving trading calendar...") trading_days = await get_trading_days( client, config.backtest.start_date, config.backtest.end_date, config.backtest.lookback_trading_days, ) print(f" {trading_days[0]} → {trading_days[-1]} ({len(trading_days)} days)") ticker_sectors = ( _load_orb_ticker_sectors(tickers) if is_orb else ( await _load_ticker_sectors_with_oracle(tickers, client) if _momentum_strategy_uses_sector_labels(config.strategy) else {} ) ) print(f"[3/4] Phase 1: Fetching daily bars for {len(tickers)} tickers...") _daily_prog_last = [-1] def daily_prog(completed: int, total: int) -> None: pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _daily_prog_last[0] or completed == total: _daily_prog_last[0] = pct sys.stdout.write(f"\r {_make_progress_bar(completed, total)}") sys.stdout.flush() daily_fetch_start_sw = ( (date.fromisoformat(trading_days[0]) - timedelta(days=90)).isoformat() if (is_orb or _momentum_strategy_uses_daily_enrichment(config.strategy)) else trading_days[0] ) daily_bars = await fetch_daily_bars_bulk( tickers, daily_fetch_start_sw, trading_days[-1], client, cache=daily_cache, intraday_cache_fallback=cache, prefer_intraday_fallback=True, skip_oracle_when_unhealthy=True, concurrency=20, progress_callback=daily_prog, ) print(f"\n {len(daily_bars)}/{len(tickers)} with data") momentum_enrichment: dict[str, dict[str, dict]] | None = None momentum_vix_by_day: dict[str, float] | None = None orb_sweep_overlay_tickers_per_day: dict[str, set[str]] | None = None orb_sweep_context_resolver = None if is_orb: # Ensure regime ticker is in daily_bars for market regime filter (sweep mode) orb_params_sweep_check = config.orb_strategy or ORBStrategyParams() orb_candidate_variant_param_keys = ( "min_price", "min_atr_14", "min_avg_dollar_volume", "candidate_seed_liquid_overlay_slots", "candidate_seed_liquid_min_gap_pct", "candidate_seed_liquid_max_gap_pct", "candidate_seed_liquid_min_avg_dollar_vol_30d", "candidate_seed_liquid_min_ret_5d", "candidate_seed_liquid_max_entropy_20d", "candidate_seed_leader_overlay_slots", "candidate_seed_leader_min_gap_pct", "candidate_seed_leader_max_gap_pct", "candidate_seed_leader_min_avg_dollar_vol_30d", "candidate_seed_leader_min_ret_5d", "candidate_seed_leader_min_atr_pct", "candidate_seed_leader_max_entropy_20d", "candidate_seed_moderate_liquid_overlay_slots", "candidate_seed_moderate_liquid_min_gap_pct", "candidate_seed_moderate_liquid_max_gap_pct", "candidate_seed_moderate_liquid_min_avg_dollar_vol_30d", "candidate_seed_moderate_liquid_max_avg_dollar_vol_30d", "candidate_seed_moderate_liquid_min_ret_5d", "candidate_seed_moderate_liquid_max_entropy_20d", "candidate_seed_event_overlay_slots", "candidate_seed_event_min_score", "candidate_seed_event_min_gap_pct", "candidate_seed_event_max_gap_pct", "candidate_seed_event_min_avg_dollar_vol_30d", "candidate_seed_event_min_ret_5d", "candidate_seed_event_max_entropy_20d", "candidate_allowed_event_types", "candidate_seed_ownership_overlay_slots", "candidate_seed_ownership_initial_only", "candidate_seed_ownership_min_strength_score", "candidate_seed_ownership_min_gap_pct", "candidate_seed_ownership_max_gap_pct", "candidate_seed_ownership_min_avg_dollar_vol_30d", "candidate_seed_ownership_min_ret_5d", "candidate_seed_ownership_max_entropy_20d", "candidate_seed_form4_overlay_slots", "candidate_seed_form4_min_total_value", "candidate_seed_form4_min_owner_count", "candidate_seed_form4_min_c_suite_count", "candidate_seed_form4_require_cluster_or_csuite", "candidate_seed_form4_min_gap_pct", "candidate_seed_form4_max_gap_pct", "candidate_seed_form4_min_avg_dollar_vol_30d", "candidate_seed_form4_min_ret_5d", "candidate_seed_form4_max_entropy_20d", ) orb_sweep_dynamic_candidate_params = any( key in sweep.sweep_params for key in orb_candidate_variant_param_keys ) orb_sweep_dynamic_quality_params = any( key in sweep.sweep_params for key in ( "market_orb_quality_ticker", "market_orb_quality_secondary_ticker", "market_orb_quality_size_scale_low", "market_orb_quality_size_scale_high", "market_orb_quality_size_scale_min", "market_orb_quality_size_scale_max", "conditional_confirmation_ticker", "market_orb_quality_joint_weak_require_confirmation", "market_orb_quality_joint_panic_require_confirmation", ) ) orb_ownership_param_keys = ( "ownership_13dg_lookback_days", "ownership_13dg_reference_path", "weight_ownership_13dg", "weight_ownership_initial_13dg", "ownership_initial_size_scale", "ownership_initial_min_score_rank_pct", "ownership_initial_allowed_trigger_types", "ownership_initial_ignore_scaled_risk_overlays", "candidate_seed_ownership_overlay_slots", "candidate_seed_ownership_initial_only", "candidate_seed_ownership_min_strength_score", "candidate_seed_ownership_min_gap_pct", "candidate_seed_ownership_max_gap_pct", "candidate_seed_ownership_min_avg_dollar_vol_30d", "candidate_seed_ownership_min_ret_5d", "candidate_seed_ownership_max_entropy_20d", ) orb_ownership_seed_param_keys = ( "candidate_seed_ownership_overlay_slots", "candidate_seed_ownership_initial_only", "candidate_seed_ownership_min_strength_score", "candidate_seed_ownership_min_gap_pct", "candidate_seed_ownership_max_gap_pct", "candidate_seed_ownership_min_avg_dollar_vol_30d", "candidate_seed_ownership_min_ret_5d", "candidate_seed_ownership_max_entropy_20d", ) orb_form4_seed_param_keys = ( "candidate_seed_form4_overlay_slots", "candidate_seed_form4_min_total_value", "candidate_seed_form4_min_owner_count", "candidate_seed_form4_min_c_suite_count", "candidate_seed_form4_require_cluster_or_csuite", "candidate_seed_form4_min_gap_pct", "candidate_seed_form4_max_gap_pct", "candidate_seed_form4_min_avg_dollar_vol_30d", "candidate_seed_form4_min_ret_5d", "candidate_seed_form4_max_entropy_20d", ) orb_sweep_uses_ownership_13dg = ( _orb_strategy_uses_ownership_13dg(orb_params_sweep_check) or any(key in sweep.sweep_params for key in orb_ownership_param_keys) ) orb_sweep_uses_ownership_seed_overlay = ( _orb_strategy_uses_ownership_seed_overlay(orb_params_sweep_check) or any(key in sweep.sweep_params for key in orb_ownership_seed_param_keys) ) orb_form4_param_keys = ( "form4_lookback_days", "form4_reference_path", "form4_size_scale", "form4_min_total_value", "form4_min_owner_count", "form4_min_c_suite_count", "form4_require_cluster_or_csuite", "form4_allowed_trigger_types", "form4_ignore_scaled_risk_overlays", *orb_form4_seed_param_keys, ) orb_sweep_uses_form4_seed_overlay = ( _orb_strategy_uses_form4_seed_overlay(orb_params_sweep_check) or any(key in sweep.sweep_params for key in orb_form4_seed_param_keys) ) orb_sweep_uses_form4 = ( _orb_strategy_uses_form4(orb_params_sweep_check) or any(key in sweep.sweep_params for key in orb_form4_param_keys) ) orb_sweep_dynamic_event_params = any( key in sweep.sweep_params for key in ( "prior_event_lookback_days", "prior_event_types", "prior_event_decay_half_life_days", "prior_event_decay_min_score", "prior_event_guidance_score_scale", ) ) orb_sweep_requires_event_details = any( key in sweep.sweep_params for key in ( "prior_event_decay_half_life_days", "prior_event_decay_min_score", "prior_event_guidance_score_scale", ) ) or _orb_prior_event_decay_enabled(orb_params_sweep_check) def _orb_candidate_variant_key(params: ORBStrategyParams) -> tuple[object, ...]: return tuple( tuple(value) if isinstance(value, list) else value for value in ( getattr(params, key, None) for key in orb_candidate_variant_param_keys ) ) regime_ticker_sw = getattr(orb_params_sweep_check, "market_regime_ticker", "SPY") or "SPY" if _orb_strategy_requires_regime_ticker_daily(orb_params_sweep_check) and regime_ticker_sw not in daily_bars: extra_bars_sw = await fetch_daily_bars_bulk( [regime_ticker_sw], daily_fetch_start_sw, trading_days[-1], client, cache=daily_cache, intraday_cache_fallback=cache, concurrency=1 ) daily_bars.update(extra_bars_sw) orb_params = config.orb_strategy or ORBStrategyParams() print(" Computing ATR/volume enrichment...") synthetic_today_rows_sw = _append_synthetic_today_daily_rows(daily_bars, trading_days) if synthetic_today_rows_sw: print( " Added synthetic same-day daily rows " f"for {synthetic_today_rows_sw} tickers" ) enrichment = enrich_daily_bars(daily_bars, trading_days) orb_sweep_ownership_features: dict[str, dict[str, dict]] | None = None orb_sweep_form4_features: dict[str, dict[str, dict]] | None = None if orb_sweep_uses_ownership_seed_overlay: ownership_params = orb_params if "ownership_13dg_lookback_days" in sweep.sweep_params: lookbacks = [ int(value or 0) for value in sweep.sweep_params.get("ownership_13dg_lookback_days", []) ] if lookbacks: ownership_params = ownership_params.model_copy( update={"ownership_13dg_lookback_days": max(lookbacks)} ) if int(getattr(ownership_params, "ownership_13dg_lookback_days", 0) or 0) > 0: all_daily_tickers = sorted(str(ticker).upper() for ticker in daily_bars.keys()) print( " Loading PIT 13D/13G ownership seed features " f"(D-{int(getattr(ownership_params, 'ownership_13dg_lookback_days', 0) or 0)}) " f"for {len(all_daily_tickers)} tickers..." ) orb_sweep_ownership_features = _load_orb_ownership_13dg_features( all_daily_tickers, trading_days, ownership_params, ) _merge_orb_ownership_13dg_features( enrichment, orb_sweep_ownership_features, ) print( f" 13D/13G seed coverage: {len(orb_sweep_ownership_features)} " "tickers with events" ) if orb_sweep_uses_form4_seed_overlay: form4_params = orb_params if "form4_lookback_days" in sweep.sweep_params: lookbacks = [ int(value or 0) for value in sweep.sweep_params.get("form4_lookback_days", []) ] if lookbacks: form4_params = form4_params.model_copy( update={"form4_lookback_days": max(lookbacks)} ) if int(getattr(form4_params, "form4_lookback_days", 0) or 0) > 0: all_daily_tickers = sorted(str(ticker).upper() for ticker in daily_bars.keys()) print( " Loading PIT Form 4 seed features " f"(D-{int(getattr(form4_params, 'form4_lookback_days', 0) or 0)}) " f"for {len(all_daily_tickers)} tickers..." ) orb_sweep_form4_features = _load_orb_form4_features( all_daily_tickers, trading_days, form4_params, ) _merge_orb_form4_features(enrichment, orb_sweep_form4_features) print( f" Form 4 seed coverage: {len(orb_sweep_form4_features)} " "tickers with events" ) orb_base_enrichment = ( copy.deepcopy(enrichment) if orb_sweep_dynamic_event_params else None ) candidates = orb_pre_screen_candidates( daily_bars, trading_days, enrichment, min_price=orb_params.min_price, min_atr=orb_params.min_atr_14, min_avg_dollar_vol=orb_params.min_avg_dollar_volume, max_per_day=None, ) if ( int(getattr(orb_params, "candidate_seed_leader_overlay_slots", 0) or 0) > 0 or int(getattr(orb_params, "candidate_seed_liquid_overlay_slots", 0) or 0) > 0 or int(getattr(orb_params, "candidate_seed_moderate_liquid_overlay_slots", 0) or 0) > 0 or int(getattr(orb_params, "candidate_seed_event_overlay_slots", 0) or 0) > 0 or _orb_strategy_uses_ownership_seed_overlay(orb_params) or _orb_strategy_uses_form4_seed_overlay(orb_params) ): candidates, orb_sweep_overlay_tickers_per_day = _augment_momentum_seed_candidates_with_liquid_overlay( # type: ignore[arg-type] candidates, daily_bars, trading_days, enrichment, orb_params ) orb_candidate_variant_maps: dict[tuple[float, float, float], dict[str, list[str]]] = {} orb_overlay_variant_maps: dict[tuple[float, float, float], dict[str, set[str]] | None] = {} if orb_sweep_dynamic_candidate_params: for overrides in generate_combinations(sweep): combo_config = apply_overrides(config, overrides) combo_params = combo_config.orb_strategy or ORBStrategyParams() variant_key = _orb_candidate_variant_key(combo_params) if variant_key in orb_candidate_variant_maps: continue combo_candidates = orb_pre_screen_candidates( daily_bars, trading_days, enrichment, min_price=combo_params.min_price, min_atr=combo_params.min_atr_14, min_avg_dollar_vol=combo_params.min_avg_dollar_volume, max_per_day=None, ) combo_overlay_tickers: dict[str, set[str]] | None = None if ( int(getattr(combo_params, "candidate_seed_leader_overlay_slots", 0) or 0) > 0 or int(getattr(combo_params, "candidate_seed_liquid_overlay_slots", 0) or 0) > 0 or int(getattr(combo_params, "candidate_seed_moderate_liquid_overlay_slots", 0) or 0) > 0 or int(getattr(combo_params, "candidate_seed_event_overlay_slots", 0) or 0) > 0 or _orb_strategy_uses_ownership_seed_overlay(combo_params) or _orb_strategy_uses_form4_seed_overlay(combo_params) ): combo_candidates, combo_overlay_tickers = _augment_momentum_seed_candidates_with_liquid_overlay( # type: ignore[arg-type] combo_candidates, daily_bars, trading_days, enrichment, combo_params, ) orb_candidate_variant_maps[variant_key] = _normalize_candidate_map(combo_candidates) orb_overlay_variant_maps[variant_key] = combo_overlay_tickers if orb_sweep_dynamic_candidate_params and orb_candidate_variant_maps: candidates = _merge_candidate_maps(list(orb_candidate_variant_maps.values())) all_tickers_sw = list({t for day_tickers in candidates.values() for t in day_tickers}) if _orb_strategy_uses_catalyst(orb_params_sweep_check): _prior_lookback_sw = int(getattr(orb_params, "prior_event_lookback_days", 0) or 0) if _prior_lookback_sw > 0: _prior_event_types_sw = tuple(getattr(orb_params, "prior_event_types", None) or ["earnings_release", "guidance_update"]) print( " Prefetching prior-event features " f"(snapshot={prior_event_snapshot_cache.snapshot_id}, " f"D-{_prior_lookback_sw}, types={_prior_event_types_sw}) " f"for {len(all_tickers_sw)} tickers..." ) event_features = await _prefetch_prior_event_features_db( all_tickers_sw, trading_days, lookback_calendar_days=_prior_lookback_sw, event_types=_prior_event_types_sw, snapshot_cache=prior_event_snapshot_cache, require_event_details=orb_sweep_requires_event_details, ) event_features = _apply_orb_prior_event_decay_features( event_features, orb_params, ) print(f" Prior-event coverage: {len(event_features)} tickers with events") else: event_tickers = _orb_candidate_event_tickers(candidates, enrichment, orb_params) _evt_prog_last = [-1] def event_prog(completed: int, total: int) -> None: pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _evt_prog_last[0] or completed == total: _evt_prog_last[0] = pct sys.stdout.write(f"\r {_make_progress_bar(completed, total)}") sys.stdout.flush() event_features = await fetch_filing_event_features_bulk( event_tickers, trading_days[0], trading_days[-1], client, cache=event_cache, concurrency=16, progress_callback=event_prog, ) print() _merge_orb_event_features(enrichment, event_features) if orb_sweep_dynamic_event_params: event_enrichment_variants: dict[tuple[object, ...], dict[str, dict[str, dict]]] = {} unique_event_variants: dict[tuple[object, ...], ORBStrategyParams] = {} def _orb_event_variant_key(params: ORBStrategyParams) -> tuple[object, ...]: return ( int(getattr(params, "prior_event_lookback_days", 0) or 0), tuple(getattr(params, "prior_event_types", None) or ["earnings_release", "guidance_update"]), getattr(params, "prior_event_decay_half_life_days", None), getattr(params, "prior_event_decay_min_score", None), float(getattr(params, "prior_event_guidance_score_scale", 1.0) or 1.0), ) for overrides in generate_combinations(sweep): combo_config = apply_overrides(config, overrides) combo_params = combo_config.orb_strategy or ORBStrategyParams() if not _orb_strategy_uses_catalyst(combo_params): continue combo_lookback = int(getattr(combo_params, "prior_event_lookback_days", 0) or 0) if combo_lookback <= 0: continue unique_event_variants[_orb_event_variant_key(combo_params)] = combo_params if unique_event_variants: print( " Prefetching sweep prior-event enrichment variants " f"for {len(unique_event_variants)} setups..." ) base_variant_key = _orb_event_variant_key(orb_params) for variant_key, combo_params in sorted( unique_event_variants.items(), key=lambda item: repr(item[0]), ): combo_lookback = int(variant_key[0] or 0) combo_types = tuple(variant_key[1] or ()) if combo_lookback <= 0: continue if variant_key == base_variant_key: event_enrichment_variants[variant_key] = enrichment continue variant_features = await _prefetch_prior_event_features_db( all_tickers_sw, trading_days, lookback_calendar_days=combo_lookback, event_types=combo_types, snapshot_cache=prior_event_snapshot_cache, require_event_details=_orb_prior_event_decay_enabled(combo_params), ) variant_features = _apply_orb_prior_event_decay_features( variant_features, combo_params, ) variant_enrichment = copy.deepcopy(orb_base_enrichment if orb_base_enrichment is not None else enrichment) _merge_orb_event_features(variant_enrichment, variant_features) event_enrichment_variants[variant_key] = variant_enrichment if _orb_strategy_uses_attention(orb_params_sweep_check): attention_pairs = _orb_candidate_event_pairs(candidates, enrichment, orb_params) print(f" Fetching event attention for {len(attention_pairs)} ticker-days...") _attn_prog_last = [-1] def attention_prog(completed: int, total: int) -> None: pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _attn_prog_last[0] or completed == total: _attn_prog_last[0] = pct sys.stdout.write(f"\r {_make_progress_bar(completed, total)}") sys.stdout.flush() attention_features = await fetch_attention_features_bulk( attention_pairs, client, cache=attention_cache, concurrency=16, progress_callback=attention_prog, ) print() _merge_orb_attention_features(enrichment, attention_features) if orb_base_enrichment is not None: _merge_orb_attention_features(orb_base_enrichment, attention_features) if orb_sweep_dynamic_event_params: for variant_enrichment in event_enrichment_variants.values(): if variant_enrichment is not enrichment: _merge_orb_attention_features( variant_enrichment, attention_features, ) if orb_sweep_uses_ownership_13dg and orb_sweep_ownership_features is None: ownership_params = orb_params if "ownership_13dg_lookback_days" in sweep.sweep_params: lookbacks = [ int(value or 0) for value in sweep.sweep_params.get("ownership_13dg_lookback_days", []) ] if lookbacks: ownership_params = ownership_params.model_copy( update={"ownership_13dg_lookback_days": max(lookbacks)} ) if int(getattr(ownership_params, "ownership_13dg_lookback_days", 0) or 0) > 0: print( " Loading PIT 13D/13G ownership features " f"(D-{int(getattr(ownership_params, 'ownership_13dg_lookback_days', 0) or 0)}) " f"for {len(all_tickers_sw)} tickers..." ) ownership_features = _load_orb_ownership_13dg_features( all_tickers_sw, trading_days, ownership_params, ) _merge_orb_ownership_13dg_features(enrichment, ownership_features) if orb_base_enrichment is not None: _merge_orb_ownership_13dg_features( orb_base_enrichment, ownership_features, ) if orb_sweep_dynamic_event_params: for variant_enrichment in event_enrichment_variants.values(): if variant_enrichment is not enrichment: _merge_orb_ownership_13dg_features( variant_enrichment, ownership_features, ) print(f" 13D/13G coverage: {len(ownership_features)} tickers with events") if orb_sweep_uses_form4 and orb_sweep_form4_features is None: form4_params = orb_params if "form4_lookback_days" in sweep.sweep_params: lookbacks = [ int(value or 0) for value in sweep.sweep_params.get("form4_lookback_days", []) ] if lookbacks: form4_params = form4_params.model_copy( update={"form4_lookback_days": max(lookbacks)} ) if int(getattr(form4_params, "form4_lookback_days", 0) or 0) > 0: print( " Loading PIT Form 4 features " f"(D-{int(getattr(form4_params, 'form4_lookback_days', 0) or 0)}) " f"for {len(all_tickers_sw)} tickers..." ) form4_features = _load_orb_form4_features( all_tickers_sw, trading_days, form4_params, ) _merge_orb_form4_features(enrichment, form4_features) if orb_base_enrichment is not None: _merge_orb_form4_features( orb_base_enrichment, form4_features, ) if orb_sweep_dynamic_event_params: for variant_enrichment in event_enrichment_variants.values(): if variant_enrichment is not enrichment: _merge_orb_form4_features( variant_enrichment, form4_features, ) print(f" Form 4 coverage: {len(form4_features)} tickers with events") if _orb_strategy_uses_vix(orb_params_sweep_check): print(" Fetching VIX regime series for ORB sweep...") momentum_vix_by_day = await _fetch_vix_by_day(client, trading_days) orb_support_tickers = set(_orb_intraday_support_tickers(orb_params)) if orb_sweep_dynamic_quality_params: for overrides in generate_combinations(sweep): combo_config = apply_overrides(config, overrides) combo_params = combo_config.orb_strategy or ORBStrategyParams() orb_support_tickers.update(_orb_intraday_support_tickers(combo_params)) if orb_sweep_dynamic_candidate_params and orb_candidate_variant_maps: for variant_key, candidate_map in list(orb_candidate_variant_maps.items()): orb_candidate_variant_maps[variant_key] = _augment_candidate_map_with_support_tickers( candidate_map, sorted(orb_support_tickers), trading_days, include_empty_days=_orb_strategy_uses_idle_sleeve(orb_params), ) candidates = _merge_candidate_maps(list(orb_candidate_variant_maps.values())) else: candidates = _augment_candidate_map_with_support_tickers( candidates, sorted(orb_support_tickers), trading_days, include_empty_days=_orb_strategy_uses_idle_sleeve(orb_params), ) if orb_sweep_dynamic_event_params or orb_sweep_dynamic_candidate_params: def orb_sweep_context_resolver( combo_config: IntradayConfig, ) -> tuple[ dict[str, dict[str, dict]], dict[str, float] | None, dict[str, set[str]] | None, dict[str, list[str]] | None, ]: combo_params = combo_config.orb_strategy or ORBStrategyParams() if orb_sweep_dynamic_event_params: if not _orb_strategy_uses_catalyst(combo_params): combo_enrichment = orb_base_enrichment if orb_base_enrichment is not None else enrichment else: combo_lookback = int(getattr(combo_params, "prior_event_lookback_days", 0) or 0) combo_key = _orb_event_variant_key(combo_params) combo_enrichment = ( event_enrichment_variants.get(combo_key, enrichment) if combo_lookback > 0 else enrichment ) else: combo_enrichment = enrichment combo_candidate_map = None combo_overlay_tickers = orb_sweep_overlay_tickers_per_day if orb_sweep_dynamic_candidate_params: variant_key = _orb_candidate_variant_key(combo_params) combo_candidate_map = orb_candidate_variant_maps.get(variant_key) combo_overlay_tickers = orb_overlay_variant_maps.get(variant_key) return combo_enrichment, momentum_vix_by_day, combo_overlay_tickers, combo_candidate_map else: enrichment = {} if _momentum_strategy_requires_regime_ticker_daily(config.strategy): regime_ticker_sw = config.strategy.market_regime_gap_ticker or "SPY" if regime_ticker_sw not in daily_bars: extra_bars_sw = await fetch_daily_bars_bulk( [regime_ticker_sw], daily_fetch_start_sw, trading_days[-1], client, cache=daily_cache, intraday_cache_fallback=cache, prefer_intraday_fallback=True, skip_oracle_when_unhealthy=True, concurrency=1, ) daily_bars.update(extra_bars_sw) if _momentum_strategy_uses_daily_enrichment(config.strategy): print(" Computing momentum daily enrichment...") momentum_enrichment = _momentum_enrichment_for_days(daily_bars, trading_days) if _momentum_strategy_uses_vix(config.strategy): print(" Fetching VIX regime series...") momentum_vix_by_day = await _fetch_vix_by_day(client, trading_days) if ( momentum_enrichment is not None and ( _momentum_strategy_uses_catalyst(config.strategy) or _momentum_strategy_uses_attention(config.strategy) ) ): preliminary_candidates = _normalize_candidate_map( _momentum_intraday_seed_candidates( daily_bars, trading_days, momentum_enrichment, config.strategy, default_threshold=config.backtest.pre_screen_threshold, use_signal_features=False, ) ) if _momentum_strategy_uses_catalyst(config.strategy): if _momentum_strategy_uses_candidate_stage_catalyst(config.strategy): event_tickers = sorted(daily_bars.keys()) else: event_tickers = _momentum_candidate_event_tickers(preliminary_candidates) print(f" Fetching momentum filing catalysts for {len(event_tickers)} tickers...") _evt_prog_last = [-1] def event_prog(completed: int, total: int) -> None: pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _evt_prog_last[0] or completed == total: _evt_prog_last[0] = pct sys.stdout.write(f"\r {_make_progress_bar(completed, total)}") sys.stdout.flush() event_features = await fetch_filing_event_features_bulk( event_tickers, trading_days[0], trading_days[-1], client, cache=event_cache, concurrency=16, progress_callback=event_prog, ) print() _merge_momentum_event_features(momentum_enrichment, event_features) if _momentum_strategy_uses_attention(config.strategy): attention_pairs = _momentum_candidate_event_pairs( preliminary_candidates, momentum_enrichment, config.strategy, ) print(f" Fetching momentum attention for {len(attention_pairs)} ticker-days...") _attn_prog_last = [-1] def attention_prog(completed: int, total: int) -> None: pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _attn_prog_last[0] or completed == total: _attn_prog_last[0] = pct sys.stdout.write(f"\r {_make_progress_bar(completed, total)}") sys.stdout.flush() attention_features = await fetch_attention_features_bulk( attention_pairs, client, cache=attention_cache, concurrency=16, progress_callback=attention_prog, ) print() _merge_momentum_attention_features(momentum_enrichment, attention_features) candidates = _momentum_intraday_seed_candidates( daily_bars, trading_days, momentum_enrichment or {}, config.strategy, default_threshold=config.backtest.pre_screen_threshold, use_signal_features=True, ) candidates, _ = _augment_momentum_seed_candidates_with_liquid_overlay( candidates, daily_bars, trading_days, momentum_enrichment or {}, config.strategy, ) candidates = _normalize_candidate_map(candidates) total_pairs = sum(len(v) for v in candidates.values()) print(f" {total_pairs} candidate pairs") print(f"[4/4] Phase 2: Fetching intraday bars...") _intra_prog_last = [-1] def intra_prog(completed: int, total: int, hits: int, calls: int) -> None: if completed == 0 and calls == 0 and total > 0: sys.stdout.write("\n") sys.stdout.flush() _intra_prog_last[0] = -1 pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _intra_prog_last[0] or completed == total: _intra_prog_last[0] = pct sys.stdout.write( f"\r {_make_progress_bar(completed, total)} cache:{hits} api:{calls}" ) sys.stdout.flush() all_intraday = await fetch_intraday_bulk( candidates, client, cache, concurrency=8, progress_callback=intra_prog, ) print(f"\n Done. {len(all_intraday)} days with data") if (not is_orb) and _momentum_uses_historical_intraday_first(config.strategy): candidates = momentum_intraday_first_candidates( all_intraday, trading_days, config.strategy, daily_enrichment=momentum_enrichment, ticker_sectors=ticker_sectors, max_per_day=config.strategy.candidate_final_max_per_day, ) total_pairs = sum(len(v) for v in candidates.values()) shortlisted_intraday: dict[str, dict[str, list[dict]]] = {} for day, day_candidates in candidates.items(): day_bars = all_intraday.get(day, {}) subset = { ticker: day_bars[ticker] for ticker in day_candidates if ticker in day_bars } if subset: shortlisted_intraday[day] = subset all_intraday = shortlisted_intraday print( " Intraday-first reranked: " f"{total_pairs} ticker-day pairs across {len(candidates)} days" ) sector_proxy_intraday: dict[str, dict[str, list[dict]]] | None = None if (not is_orb) and _momentum_strategy_uses_sector_proxies(config.strategy): print(" Fetching sector ETF proxy bars for sweep...") proxy_candidates = {day: list(SECTOR_PROXY_TICKERS) for day in trading_days} _proxy_prog_last = [-1] def proxy_prog(completed: int, total: int, hits: int, calls: int) -> None: if completed == 0 and calls == 0 and total > 0: sys.stdout.write("\n") sys.stdout.flush() _proxy_prog_last[0] = -1 pct = int(completed / total * 10) * 10 if total > 0 else 0 if pct > _proxy_prog_last[0] or completed == total: _proxy_prog_last[0] = pct sys.stdout.write( f"\r {_make_progress_bar(completed, total)} cache:{hits} api:{calls}" ) sys.stdout.flush() sector_proxy_intraday = await fetch_intraday_bulk( proxy_candidates, client, cache, concurrency=4, progress_callback=proxy_prog, ) print(f"\n Done. {len(sector_proxy_intraday)} days with sector ETF proxy data") print(f"\nRunning {sweep.total_combinations} sweep combinations...") completed_sw = [0] _sweep_last_pct = [-1] def sweep_prog(done: int, total: int) -> None: completed_sw[0] = done pct = int(done / total * 10) * 10 if total > 0 else 0 if pct > _sweep_last_pct[0] or done == total: _sweep_last_pct[0] = pct sys.stdout.write(f"\r {_make_progress_bar(done, total)}") sys.stdout.flush() sweep_results = run_sweep( sweep, all_intraday, trading_days, progress_callback=sweep_prog, enrichment=enrichment, momentum_enrichment=momentum_enrichment, vix_by_day=momentum_vix_by_day, ticker_sectors=ticker_sectors, sector_proxy_intraday_by_day=sector_proxy_intraday, overlay_tickers_per_day=orb_sweep_overlay_tickers_per_day if is_orb else None, orb_context_resolver=orb_sweep_context_resolver if is_orb else None, ) print() # Display top results print(format_sweep_comparison(sweep_results, top_n=20)) # Also show full details for the #1 configuration if sweep_results: best = sweep_results[0] best_config = apply_overrides(config, best.params) if is_orb: best_enrichment = enrichment or {} best_vix = momentum_vix_by_day best_overlay = orb_sweep_overlay_tickers_per_day best_candidate_map = None if orb_sweep_context_resolver is not None: resolved = orb_sweep_context_resolver(best_config) if len(resolved) == 4: best_enrichment, best_vix, best_overlay, best_candidate_map = resolved else: best_enrichment, best_vix, best_overlay = resolved from libs.intraday.orb_simulator import run_orb_simulation best_all_intraday = all_intraday if best_candidate_map: best_all_intraday = { day: { ticker: day_intraday[ticker] for ticker in tickers if ticker in day_intraday } for day, tickers in best_candidate_map.items() if (day_intraday := all_intraday.get(day)) } best_day_results = run_orb_simulation( best_all_intraday, trading_days, best_config.orb_strategy, best_enrichment, ticker_sectors=ticker_sectors, vix_by_day=best_vix, overlay_tickers_per_day=best_overlay, ) else: best_day_results = run_simulation( all_intraday, trading_days, best_config.strategy, daily_enrichment=momentum_enrichment, vix_by_day=momentum_vix_by_day, ticker_sectors=ticker_sectors, sector_proxy_intraday_by_day=sector_proxy_intraday, ) print("\n=== Best Configuration Detail ===") print(format_summary(best.metrics, best_config)) print(format_top_trades(best_day_results, n=5)) # Save sweep results out = Path(config.output.dir) out.mkdir(parents=True, exist_ok=True) from datetime import datetime import json ts = datetime.now().strftime("%Y%m%d_%H%M%S") sweep_file = out / f"sweep_{ts}.json" sweep_data = [ { "params": sr.params, "metrics": sr.metrics.model_dump(), "objective_score": sr.objective_score, } for sr in sweep_results ] sweep_file.write_text(json.dumps(sweep_data, indent=2, default=str)) print(f"\nSweep results saved to: {sweep_file}") # ── CLI ──────────────────────────────────────────────────────────────────── def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Morning Momentum Intraday Backtester", formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument("--config", help="Strategy config YAML file path") parser.add_argument("--strategy", choices=["momentum", "orb"], default=None, help="Strategy mode: 'momentum' (default) or 'orb'") parser.add_argument("--sweep", help="Parameter sweep YAML file path") parser.add_argument("--days", type=int, help="Number of trading days to backtest") parser.add_argument("--start", help="Start date YYYY-MM-DD (overrides --days)") parser.add_argument("--end", help="End date YYYY-MM-DD (default: today)") parser.add_argument("--universe", choices=["sp500", "nasdaq100", "broad", "midlarge", "largecap", "midcap", "smallmid", "screener"], help="Universe source") parser.add_argument("--top-n", type=int, dest="top_n", help="Number of stocks to buy per day") parser.add_argument("--stop-loss", dest="stop_loss", help="Stop loss %% (e.g. -0.02) or 'none' to disable") parser.add_argument("--entry-min", type=int, dest="entry_min", help="Minutes after open to enter (default 30)") parser.add_argument("--exit-min", type=int, dest="exit_min", help="Minutes before close to exit (default 30)") parser.add_argument("--min-gain", type=float, dest="min_gain", help="Minimum morning gain to qualify (default 0.01 = 1%%)") parser.add_argument("--no-cache", action="store_true", dest="no_cache", help="Disable cache use for this run") parser.add_argument("--refresh-cache", action="store_true", dest="refresh_cache", help="Force re-download all intraday data") parser.add_argument("--verbose", action="store_true", help="Show detailed per-day output") parser.add_argument("--output-dir", dest="output_dir", help="Override output directory") parser.add_argument("--initial-capital", type=float, dest="initial_capital", default=None, help="Override initial capital (e.g. 50000)") parser.add_argument("--compound-returns", dest="compound_returns", action="store_true", default=None, help="Override: use compound returns (복리) and disable daily budget reset unless explicitly set") parser.add_argument("--no-compound-returns", dest="compound_returns", action="store_false", help="Override: use simple returns (단리) regardless of config") parser.add_argument("--daily-budget-reset", dest="daily_budget_reset", action="store_true", default=None, help="Research mode: reset sizing capital and buying power to initial_capital every day " "(does not compound prior PnL; takes precedence over compound_returns)") parser.add_argument("--no-daily-budget-reset", dest="daily_budget_reset", action="store_false", help="Disable daily budget reset mode") parser.add_argument("--cache-stats", action="store_true", dest="cache_stats", help="Show cache statistics and exit") return parser.parse_args() async def main_async() -> None: args = parse_args() # Cache stats shortcut if args.cache_stats: config = load_config(args.config) cache = IntradayCache(config.cache.dir) stats = cache.stats() print(f"Cache directory: {config.cache.dir}") print(f" Files: {stats['total_files']}") print(f" Size: {stats['total_mb']} MB") print(f" Tickers: {stats['tickers']}") print(f" Date range: {stats['date_min']} → {stats['date_max']}") return # Load + apply config config = load_config(args.config) config = apply_cli_overrides(config, args) # Sweep mode if args.sweep: await run_with_sweep(config, args.sweep) return # Single run mode day_results, metrics, all_intraday, trading_days = await run( config, refresh_cache=args.refresh_cache ) # Display results print(format_summary(metrics, config)) if config.output.verbose or args.verbose: print(format_daily_breakdown(day_results)) print(format_top_trades(day_results, n=5)) # Always show daily breakdown (compact version) if not (config.output.verbose or args.verbose): print(format_daily_breakdown(day_results)) # Save results if day_results: out_file = write_results( metrics, day_results, config, config.output.dir, data_provenance=_build_intraday_data_provenance(config), ) print(f"\nResults saved to: {out_file}") def main() -> None: asyncio.run(main_async()) if __name__ == "__main__": main()