fithia2/apps/intraday_bt/run.py

"""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()