fithia2/apps/orb_trader/engine.py

"""ORB Paper Trading Engine.

One engine instance is created per session per trading day.
The scheduler calls phase methods in order:
  1. run_pre_screen()    — 09:20 ET (daily bars + enrichment + quality filter)
  2. run_orb_detection() — 09:40 ET (intraday bars → candidates)
     └─ run_pre_screen 미실행 시 full fallback (daily bars도 자체 fetch)
  3. run_breakout_check() — every sim_bar_minutes from orb_end until order_timeout
  4. run_stop_check()   — every sim_bar_minutes from orb_end until 15:55 ET
  5. run_eod_exit()     — 15:55 ET
  6. run_post_close()   — 16:00 ET

Stop management logic mirrors orb_simulator.py:477-580 exactly.
"""
from __future__ import annotations

import copy
import datetime as dt
import json
import logging
import time
import uuid
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo

from apps.orb_trader.models import (
    ORBCandidateRow,
    ORBDailySnapshotRow,
    ORBPositionRow,
    ORBTradeRow,
)
from apps.orb_trader.screener import (
    bars_to_enrichment_format,
    intraday_bars_to_format,
    live_pre_screen,
    load_universe,
)
from apps.orb_trader.state import ORBStateManager
from libs.intraday.features import enrich_daily_bars
from libs.intraday.orb_simulator import (
    _aggregate_bars,
    _bar_close_location,
    _bar_return_pct,
    _find_late_breakout_time,
    _first_regular_bar,
    _find_vwap_reclaim_time,
    _linear_range_scaler,
    _linear_scaler,
    _opening_breadth_stats,
    _orb_soft_day_sector_confirmation_override_allows,
    _orb_soft_day_setup_profile_allows,
    _orb_soft_day_vwap_min_score_pct,
    _orb_soft_day_vwap_reason_allowed,
    _orb_soft_day_vwap_reason_param,
    _orb_soft_day_vwap_size_scale,
    compute_orb_candidates,
)
from libs.intraday.simulator import _market_open_ts, _parse_ts, filter_market_hours
from libs.oracle_client.alpaca import get_multi_intraday_bars_today, get_snapshots

log = logging.getLogger(__name__)

# Structured logger — events flow to journal/events.db via libs.common.logging
# sink processor (configured by the daemon at startup). Used at trade/error
# sites so the Logs/Health UI gets ticker, fill_price, qty, etc. as fields,
# not embedded in a free-text message.
try:
    from libs.common.logging import get_logger as _get_struct_logger
    structured_log = _get_struct_logger("apps.orb_trader.engine")
except Exception:  # pragma: no cover — defensive; structlog should always import
    structured_log = None  # type: ignore[assignment]


def _emit(event: str, **fields: Any) -> None:
    """Emit a structured event if structlog is available; no-op otherwise."""
    if structured_log is None:
        return
    level = fields.pop("_level", "info")
    try:
        getattr(structured_log, level)(event, **fields)
    except Exception:
        pass

_ET = ZoneInfo("America/New_York")
_ACCOUNT_CIRCUIT_BREAKER_PCT = 25.0  # halt if equity drops >25% from peak
_AUXILIARY_TRIGGER_TYPES = {
    "vwap_reclaim",
    "soft_day_vwap_reclaim",
    "late_breakout",
    "intraday_continuation_reclaim",
}
_MARKET_THRUST_LIQUID_TRIGGER_TYPES = {
    "market_thrust_liquid_continuation",
    "market_thrust_opening_burst",
    "market_thrust_opening_followthrough",
}
_MARKET_THRUST_IMPULSE_TRIGGER_TYPES = {
    "market_thrust_opening_impulse_reclaim",
}
# Mirrors orb_simulator.py:9947-9954. Aux + market-thrust triggers bypass the
# soft-day score/profile/ret5/rvol gates because they have their own gating.
_SOFT_DAY_PRIMARY_FILTER_EXEMPT_TRIGGER_TYPES = {
    "soft_day_vwap_reclaim",
    "broad_gapup_continuation",
    "market_thrust_liquid_continuation",
    "market_thrust_opening_burst",
    "market_thrust_opening_followthrough",
    "market_thrust_opening_impulse_reclaim",
    "intraday_continuation_reclaim",
}
# Mirrors orb_simulator.py:9935-9941. soft_day_max_trades caps total day trades
# but exempts these trigger types from being subject to the cap themselves.
_SOFT_DAY_MAX_TRADES_EXEMPT_TRIGGER_TYPES = {
    "broad_gapup_continuation",
    "market_thrust_opening_burst",
    "market_thrust_opening_followthrough",
    "market_thrust_opening_impulse_reclaim",
    "intraday_continuation_reclaim",
}


class ORBTradingEngine:
    """Intraday paper trading engine for the ORB strategy.

    Holds per-day state (enrichment, candidates, date_str) as instance variables.
    On server restart mid-day, state is reconstructed from the DB.
    """

    def __init__(
        self,
        session: Any,
        broker: Any,
        state: ORBStateManager,
        params: Any,
        log_callback: Any = None,
    ) -> None:
        self._session = session
        self._broker = broker
        self._state = state
        # Keep live execution overrides local to this engine instance. Strategy
        # filters must remain identical to the backtest config; otherwise a
        # session named "v49.86" can make different candidate decisions live.
        self._params = params.model_copy(deep=True) if hasattr(params, "model_copy") else copy.copy(params)
        self._log_callback = log_callback  # optional scheduler._log for UI visibility

        # Live execution details. These affect paper/live order handling, not
        # strategy candidate selection.
        self._params.settlement_days = 0  # paper trading; no real T+1 settlement
        self._params.slippage_bps = 0.0   # real fills, no simulated slippage

        # Per-day in-memory state (reset each day)
        self._date_str: str = ""
        self._enrichment: dict[str, dict[str, dict]] = {}
        self._daily_bars: dict[str, list[dict]] = {}
        self._candidates: list[dict] = []      # computed by run_orb_detection
        self._pending_cands: list[dict] = []   # not yet filled (for breakout checks)
        # Pre-screened tickers: None = pre_screen not yet run, [] = ran but nothing passed
        self._pre_screened_tickers: list[str] | None = None
        self._day_size_scale: float = 1.0
        self._soft_day_reason: str | None = None
        self._market_orb_quality_max_trades: int | None = None
        self._market_orb_quality_reason: str | None = None
        self._market_thrust_breadth_override_active: bool = False
        self._market_thrust_index_breadth_override_active: bool = False
        self._market_thrust_opening_breadth_override_active: bool = False

    def _get_equity(self) -> float:
        """Current equity = last snapshot equity, or initial if no snapshots."""
        eq = self._state.get_equity(self._session.session_id)
        return eq if eq is not None else self._session.initial_equity

    @staticmethod
    def _last_trading_day(ref: dt.date) -> dt.date:
        """Return the most recent weekday strictly before ref.

        Used for daily-bar end_date: today's bar is incomplete during market hours,
        and weekend dates cause Alpaca to return 502 Bad Gateway.
        """
        d = ref - dt.timedelta(days=1)
        while d.weekday() >= 5:  # 5=Sat, 6=Sun
            d -= dt.timedelta(days=1)
        return d

    def _log(self, msg: str) -> None:
        log.info("[ORB:%s] %s", self._session.session_name, msg)
        if self._log_callback is not None:
            try:
                self._log_callback(f"  [engine] {msg}")
            except Exception:
                pass

    def _market_context_tickers(self) -> list[str]:
        """Tickers needed for live regime/quality gates even if not trade candidates."""
        tickers: list[str] = []
        regime_ticker = getattr(self._params, "market_regime_ticker", None) or "QQQ"
        tickers.append(regime_ticker)
        for attr in (
            "market_regime_gap_ticker",
            "market_orb_quality_ticker",
            "market_orb_quality_secondary_ticker",
        ):
            ticker = getattr(self._params, attr, None)
            if ticker:
                tickers.append(str(ticker))
        return list(dict.fromkeys(t.upper() for t in tickers if t))

    def _with_market_context_tickers(self, tickers: list[str]) -> list[str]:
        """Fetch market context bars without expanding the tradable universe."""
        return list(dict.fromkeys([*self._market_context_tickers(), *tickers]))

    def _reset_day_context(self) -> None:
        self._day_size_scale = 1.0
        self._soft_day_reason = None
        self._market_orb_quality_max_trades = None
        self._market_orb_quality_reason = None
        self._market_thrust_breadth_override_active = False
        self._market_thrust_index_breadth_override_active = False
        self._market_thrust_opening_breadth_override_active = False

    @staticmethod
    def _metadata_value(value: Any) -> Any:
        if value is None or isinstance(value, (str, int, float, bool)):
            return value
        if isinstance(value, (dt.date, dt.datetime)):
            return value.isoformat()
        try:
            json.dumps(value)
            return value
        except TypeError:
            return str(value)

    def _candidate_metadata_json(self, cand: dict[str, Any]) -> str:
        keys = (
            "body_ratio",
            "broad_gapup_continuation",
            "close_location",
            "filter_first_bar_dollar_vol",
            "filter_premarket_dollar_vol",
            "filter_rvol",
            "first_bar_dollar_vol",
            "gap_up_fill_exit_active",
            "market_thrust_liquid_continuation",
            "market_thrust_opening_impulse_reclaim",
            "orb_return",
            "premarket_dollar_vol",
            "ret_5d",
            "rvol_rank_pct",
            "score_rank_pct",
            "sector_confirmation_active",
            "trigger_type",
        )
        metadata: dict[str, Any] = {}
        for key in keys:
            if key in cand:
                metadata[key] = self._metadata_value(cand.get(key))

        orb_bar = cand.get("orb_bar")
        if isinstance(orb_bar, dict):
            metadata["orb_bar"] = {
                key: self._metadata_value(orb_bar.get(key))
                for key in ("timestamp", "open", "high", "low", "close", "volume")
                if key in orb_bar
            }

        metadata["live_day_context"] = {
            "day_size_scale": self._metadata_value(self._day_size_scale),
            "soft_day_reason": self._metadata_value(self._soft_day_reason),
            "market_orb_quality_max_trades": self._metadata_value(
                self._market_orb_quality_max_trades
            ),
            "market_orb_quality_reason": self._metadata_value(
                self._market_orb_quality_reason
            ),
            "market_thrust_breadth_override_active": self._metadata_value(
                self._market_thrust_breadth_override_active
            ),
            "market_thrust_index_breadth_override_active": self._metadata_value(
                self._market_thrust_index_breadth_override_active
            ),
            "market_thrust_opening_breadth_override_active": self._metadata_value(
                self._market_thrust_opening_breadth_override_active
            ),
        }
        return json.dumps(metadata, sort_keys=True, separators=(",", ":"))

    @staticmethod
    def _candidate_metadata_from_row(row: dict[str, Any]) -> dict[str, Any]:
        raw = row.get("metadata_json")
        if not raw:
            return {}
        try:
            parsed = json.loads(str(raw))
        except (TypeError, ValueError):
            return {}
        return parsed if isinstance(parsed, dict) else {}

    def _restore_day_context_from_metadata(self, metadata: dict[str, Any]) -> None:
        context = metadata.get("live_day_context")
        if not isinstance(context, dict):
            return
        if context.get("day_size_scale") is not None:
            try:
                self._day_size_scale = float(context["day_size_scale"])
            except (TypeError, ValueError):
                pass
        if context.get("soft_day_reason") is not None:
            self._soft_day_reason = str(context["soft_day_reason"])
        if context.get("market_orb_quality_max_trades") is not None:
            try:
                self._market_orb_quality_max_trades = int(
                    context["market_orb_quality_max_trades"]
                )
            except (TypeError, ValueError):
                pass
        if context.get("market_orb_quality_reason") is not None:
            self._market_orb_quality_reason = str(context["market_orb_quality_reason"])
        self._market_thrust_breadth_override_active = bool(
            context.get("market_thrust_breadth_override_active", False)
        )
        self._market_thrust_index_breadth_override_active = bool(
            context.get("market_thrust_index_breadth_override_active", False)
        )
        self._market_thrust_opening_breadth_override_active = bool(
            context.get("market_thrust_opening_breadth_override_active", False)
        )

    @staticmethod
    def _copy_params_with_updates(params: Any, updates: dict[str, Any]) -> Any:
        """Return a params copy with updates for parity scans."""
        if not updates:
            return params
        if hasattr(params, "model_copy"):
            return params.model_copy(update=updates)
        copied = copy.copy(params)
        for key, value in updates.items():
            setattr(copied, key, value)
        return copied

    @staticmethod
    def _meets_min(value: float | None, threshold: Any) -> bool:
        return threshold is None or (value is not None and value >= float(threshold))

    def _iex_live_intraday_threshold(self, threshold: Any) -> float | None:
        """Convert a SIP-calibrated intraday volume threshold to raw IEX units."""
        if threshold is None:
            return None
        multiplier = max(
            1.0,
            float(getattr(self._params, "iex_live_intraday_volume_multiplier", 1.0) or 1.0),
        )
        return float(threshold) / multiplier

    def _apply_market_thrust_breadth_override(
        self,
        bars_by_ticker: dict[str, list[dict]],
        date_str: str,
        *,
        soft_day_reason_parts: list[str],
        regime_gap_pct: float | None,
        breadth_ratio: float | None,
        breadth_scaler: float,
    ) -> tuple[float, list[str]]:
        """Mirror the backtest gate that enables market-thrust auxiliary sleeves."""
        index_override_enabled = bool(
            getattr(self._params, "market_thrust_breadth_override_enabled", False)
        )
        opening_breadth_override_enabled = bool(
            getattr(self._params, "market_thrust_opening_breadth_override_enabled", False)
        )
        if not (index_override_enabled or opening_breadth_override_enabled):
            self._market_thrust_breadth_override_active = False
            self._market_thrust_index_breadth_override_active = False
            self._market_thrust_opening_breadth_override_active = False
            return breadth_scaler, soft_day_reason_parts

        market_open = dt.datetime.fromisoformat(f"{date_str}T09:30:00").replace(tzinfo=_ET)
        primary_ticker = (
            getattr(self._params, "market_orb_quality_ticker", None)
            or getattr(self._params, "market_regime_ticker", None)
            or "SPY"
        )
        secondary_ticker = getattr(self._params, "market_orb_quality_secondary_ticker", None)
        primary_bar = _first_regular_bar(bars_by_ticker.get(str(primary_ticker), []), market_open)
        secondary_bar = (
            _first_regular_bar(bars_by_ticker.get(str(secondary_ticker), []), market_open)
            if secondary_ticker
            else None
        )
        primary_close_loc = _bar_close_location(primary_bar)
        secondary_close_loc = _bar_close_location(secondary_bar)
        primary_ret = _bar_return_pct(primary_bar)
        secondary_ret = _bar_return_pct(secondary_bar)

        quality_blocks_thrust = self._market_quality_blocks_thrust(
            primary_close_loc,
            secondary_close_loc,
        )
        active = (
            index_override_enabled
            and "breadth" in soft_day_reason_parts
            and "hard_breadth" not in soft_day_reason_parts
            and not quality_blocks_thrust
            and self._meets_min(
                primary_close_loc,
                getattr(
                    self._params,
                    "market_thrust_breadth_override_min_primary_close_location",
                    None,
                ),
            )
            and self._meets_min(
                secondary_close_loc,
                getattr(
                    self._params,
                    "market_thrust_breadth_override_min_secondary_close_location",
                    None,
                ),
            )
            and self._meets_min(
                primary_ret,
                getattr(
                    self._params,
                    "market_thrust_breadth_override_min_primary_return_pct",
                    None,
                ),
            )
            and self._meets_min(
                secondary_ret,
                getattr(
                    self._params,
                    "market_thrust_breadth_override_min_secondary_return_pct",
                    None,
                ),
            )
            and self._meets_min(
                regime_gap_pct,
                getattr(self._params, "market_thrust_breadth_override_min_regime_gap_pct", None),
            )
            and self._meets_min(
                breadth_ratio,
                getattr(self._params, "market_thrust_breadth_override_min_breadth_ratio", None),
            )
        )
        opening_breadth_active = False
        if opening_breadth_override_enabled:
            opening_reason_allowed = bool(soft_day_reason_parts) and (
                (
                    "market_regime" in soft_day_reason_parts
                    and bool(
                        getattr(
                            self._params,
                            "market_thrust_opening_breadth_override_allow_regime_soft_day",
                            True,
                        )
                    )
                )
                or (
                    "breadth" in soft_day_reason_parts
                    and bool(
                        getattr(
                            self._params,
                            "market_thrust_opening_breadth_override_allow_breadth_soft_day",
                            True,
                        )
                    )
                )
                or (
                    "hard_breadth" in soft_day_reason_parts
                    and bool(
                        getattr(
                            self._params,
                            "market_thrust_opening_breadth_override_allow_hard_breadth",
                            False,
                        )
                    )
                )
            )
            opening_stats = _opening_breadth_stats(
                bars_by_ticker,
                date_str,
                min_first_bar_dollar_vol=self._iex_live_intraday_threshold(
                    getattr(
                        self._params,
                        "market_thrust_opening_breadth_override_min_first_bar_dollar_vol",
                        None,
                    )
                ),
                strong_close_location=float(
                    getattr(
                        self._params,
                        "market_thrust_opening_breadth_override_strong_close_location",
                        0.65,
                    )
                    or 0.65
                ),
            )
            opening_total_count = int(opening_stats.get("total_count") or 0)
            opening_positive_ratio = opening_stats.get("positive_ratio")
            opening_avg_return = opening_stats.get("avg_return_pct")
            opening_strong_ratio = opening_stats.get("strong_close_location_ratio")
            opening_breadth_active = (
                opening_reason_allowed
                and opening_total_count
                >= int(
                    getattr(
                        self._params,
                        "market_thrust_opening_breadth_override_min_total_count",
                        100,
                    )
                    or 0
                )
                and self._meets_min(
                    opening_positive_ratio,
                    getattr(
                        self._params,
                        "market_thrust_opening_breadth_override_min_positive_ratio",
                        None,
                    ),
                )
                and self._meets_min(
                    opening_avg_return,
                    getattr(
                        self._params,
                        "market_thrust_opening_breadth_override_min_avg_return_pct",
                        None,
                    ),
                )
                and self._meets_min(
                    opening_strong_ratio,
                    getattr(
                        self._params,
                        "market_thrust_opening_breadth_override_min_strong_close_location_ratio",
                        None,
                    ),
                )
            )
            self._log(
                "Opening-breadth thrust override: "
                f"active={opening_breadth_active}, count={opening_total_count}, "
                f"positive={opening_positive_ratio if opening_positive_ratio is not None else 'NA'}, "
                f"avg_ret={opening_avg_return if opening_avg_return is not None else 'NA'}, "
                f"strong_close={opening_strong_ratio if opening_strong_ratio is not None else 'NA'}"
            )

        self._market_thrust_index_breadth_override_active = active
        self._market_thrust_opening_breadth_override_active = opening_breadth_active
        self._market_thrust_breadth_override_active = active or opening_breadth_active
        if active:
            floor = max(
                0.0,
                float(
                    getattr(self._params, "market_thrust_breadth_override_size_scale_floor", 1.0)
                    or 0.0
                ),
            )
            breadth_scaler = max(breadth_scaler, floor)
            if (
                bool(getattr(self._params, "market_thrust_breadth_override_clear_soft_day", False))
                and breadth_scaler >= getattr(self._params, "soft_day_scaler_threshold", 1.0)
            ):
                soft_day_reason_parts = [
                    reason for reason in soft_day_reason_parts if reason != "breadth"
                ]
        if opening_breadth_active:
            regime_floor = max(
                0.0,
                float(
                    getattr(
                        self._params,
                        "market_thrust_opening_breadth_override_regime_size_scale_floor",
                        1.0,
                    )
                    or 0.0
                ),
            )
            breadth_floor = max(
                0.0,
                float(
                    getattr(
                        self._params,
                        "market_thrust_opening_breadth_override_breadth_size_scale_floor",
                        1.0,
                    )
                    or 0.0
                ),
            )
            # The caller has already applied regime_scaler, so live can only lift
            # breadth sizing here. Backtest parity for regime-floor lifting is
            # handled before combined sizing in the simulator.
            if regime_floor > 0:
                breadth_scaler = max(breadth_scaler, min(regime_floor, breadth_floor))
            else:
                breadth_scaler = max(breadth_scaler, breadth_floor)
            if getattr(self._params, "market_thrust_opening_breadth_override_clear_soft_day", False):
                soft_day_reason_parts = [
                    reason
                    for reason in soft_day_reason_parts
                    if not (
                        (
                            reason == "market_regime"
                            and getattr(
                                self._params,
                                "market_thrust_opening_breadth_override_allow_regime_soft_day",
                                True,
                            )
                        )
                        or (
                            reason == "breadth"
                            and getattr(
                                self._params,
                                "market_thrust_opening_breadth_override_allow_breadth_soft_day",
                                True,
                            )
                        )
                        or (
                            reason == "hard_breadth"
                            and getattr(
                                self._params,
                                "market_thrust_opening_breadth_override_allow_hard_breadth",
                                False,
                            )
                        )
                    )
                ]
        self._log(
            "Market thrust breadth override: "
            f"active={active}, {primary_ticker} close_loc={primary_close_loc if primary_close_loc is not None else 'NA'} "
            f"ret={primary_ret if primary_ret is not None else 'NA'}, "
            f"{secondary_ticker or '-'} close_loc={secondary_close_loc if secondary_close_loc is not None else 'NA'} "
            f"ret={secondary_ret if secondary_ret is not None else 'NA'}, "
            f"breadth={breadth_ratio if breadth_ratio is not None else 'NA'}, "
            f"regime_gap={regime_gap_pct if regime_gap_pct is not None else 'NA'}"
        )
        return breadth_scaler, soft_day_reason_parts

    def _market_quality_blocks_thrust(
        self,
        primary_close_loc: float | None,
        secondary_close_loc: float | None,
    ) -> bool:
        """Backtest market-thrust override is blocked by defensive quality states."""
        primary_strong_above = getattr(self._params, "market_orb_quality_primary_strong_above", None)
        secondary_weak_below = getattr(self._params, "market_orb_quality_secondary_weak_below", None)
        secondary_weak_above = getattr(self._params, "market_orb_quality_secondary_weak_above", None)
        divergence = (
            primary_strong_above is not None
            and secondary_weak_below is not None
            and primary_close_loc is not None
            and secondary_close_loc is not None
            and primary_close_loc >= primary_strong_above
            and secondary_close_loc <= secondary_weak_below
            and (secondary_weak_above is None or secondary_close_loc >= secondary_weak_above)
        )

        primary_weak_below = getattr(self._params, "market_orb_quality_primary_weak_below", None)
        primary_weak_above = getattr(self._params, "market_orb_quality_primary_weak_above", None)
        secondary_strong_above = getattr(self._params, "market_orb_quality_secondary_strong_above", None)
        primary_weak_secondary_strong = (
            primary_weak_below is not None
            and secondary_strong_above is not None
            and primary_close_loc is not None
            and secondary_close_loc is not None
            and primary_close_loc <= primary_weak_below
            and (primary_weak_above is None or primary_close_loc >= primary_weak_above)
            and secondary_close_loc >= secondary_strong_above
        )

        primary_lag_above = getattr(self._params, "market_orb_quality_primary_lag_above", None)
        primary_lag_below = getattr(self._params, "market_orb_quality_primary_lag_below", None)
        secondary_lead_above = getattr(self._params, "market_orb_quality_secondary_lead_above", None)
        secondary_lead_below = getattr(self._params, "market_orb_quality_secondary_lead_below", None)
        primary_lag_secondary_lead = (
            primary_lag_below is not None
            and secondary_lead_above is not None
            and primary_close_loc is not None
            and secondary_close_loc is not None
            and primary_close_loc <= primary_lag_below
            and (primary_lag_above is None or primary_close_loc >= primary_lag_above)
            and secondary_close_loc >= secondary_lead_above
            and (secondary_lead_below is None or secondary_close_loc <= secondary_lead_below)
        )

        joint_weak_primary_below = getattr(self._params, "market_orb_quality_joint_weak_primary_below", None)
        joint_weak_primary_above = getattr(self._params, "market_orb_quality_joint_weak_primary_above", None)
        joint_weak_secondary_below = getattr(self._params, "market_orb_quality_joint_weak_secondary_below", None)
        joint_weak_secondary_above = getattr(self._params, "market_orb_quality_joint_weak_secondary_above", None)
        joint_weak = (
            joint_weak_primary_below is not None
            and joint_weak_secondary_below is not None
            and primary_close_loc is not None
            and secondary_close_loc is not None
            and primary_close_loc <= joint_weak_primary_below
            and (joint_weak_primary_above is None or primary_close_loc >= joint_weak_primary_above)
            and secondary_close_loc <= joint_weak_secondary_below
            and (joint_weak_secondary_above is None or secondary_close_loc >= joint_weak_secondary_above)
        )

        joint_panic_primary_below = getattr(self._params, "market_orb_quality_joint_panic_primary_below", None)
        joint_panic_secondary_below = getattr(self._params, "market_orb_quality_joint_panic_secondary_below", None)
        joint_panic = (
            joint_panic_primary_below is not None
            and joint_panic_secondary_below is not None
            and primary_close_loc is not None
            and secondary_close_loc is not None
            and primary_close_loc <= joint_panic_primary_below
            and secondary_close_loc <= joint_panic_secondary_below
        )

        return bool(
            divergence
            or primary_weak_secondary_strong
            or primary_lag_secondary_lead
            or joint_weak
            or joint_panic
        )

    def _apply_market_orb_quality(
        self,
        bars_by_ticker: dict[str, list[dict]],
        date_str: str,
    ) -> None:
        """Mirror the backtest market-ORB-quality day scaler in live trading."""
        use_quality = any(
            getattr(self._params, attr, None) is not None
            for attr 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 not use_quality:
            return

        market_open = dt.datetime.fromisoformat(f"{date_str}T09:30:00").replace(tzinfo=_ET)
        primary_ticker = (
            getattr(self._params, "market_orb_quality_ticker", None)
            or getattr(self._params, "market_regime_ticker", None)
            or "SPY"
        )
        secondary_ticker = getattr(self._params, "market_orb_quality_secondary_ticker", None)
        primary_bar = _first_regular_bar(bars_by_ticker.get(str(primary_ticker), []), market_open)
        secondary_bar = (
            _first_regular_bar(bars_by_ticker.get(str(secondary_ticker), []), market_open)
            if secondary_ticker
            else None
        )
        primary_close_loc = _bar_close_location(primary_bar)
        secondary_close_loc = _bar_close_location(secondary_bar)
        primary_ret = _bar_return_pct(primary_bar)
        secondary_ret = _bar_return_pct(secondary_bar)

        scaler = 1.0
        reasons: list[str] = []
        max_trades: int | None = None

        low = getattr(self._params, "market_orb_quality_size_scale_low", None)
        high = getattr(self._params, "market_orb_quality_size_scale_high", None)
        if low is not None and high is not None:
            scaler *= _linear_range_scaler(
                primary_close_loc,
                low,
                high,
                getattr(self._params, "market_orb_quality_size_scale_min", 1.0),
                getattr(self._params, "market_orb_quality_size_scale_max", 1.0),
            )

        def _cap_trades(raw: Any) -> None:
            nonlocal max_trades
            if raw is None:
                return
            cap = max(0, int(raw))
            max_trades = cap if max_trades is None else min(max_trades, cap)

        primary_strong_above = getattr(self._params, "market_orb_quality_primary_strong_above", None)
        secondary_weak_above = getattr(self._params, "market_orb_quality_secondary_weak_above", None)
        secondary_weak_below = getattr(self._params, "market_orb_quality_secondary_weak_below", None)
        divergence = (
            primary_strong_above is not None
            and secondary_weak_above is not None
            and primary_close_loc is not None
            and secondary_close_loc is not None
            and primary_close_loc >= primary_strong_above
            and secondary_close_loc >= secondary_weak_above
            and (secondary_weak_below is None or secondary_close_loc <= secondary_weak_below)
        )
        if divergence:
            scale = getattr(self._params, "market_orb_quality_divergence_scale", None)
            if scale is not None:
                scaler *= max(0.0, float(scale))
            _cap_trades(getattr(self._params, "market_orb_quality_divergence_max_trades", None))
            reasons.append("divergence")

        primary_weak_below = getattr(self._params, "market_orb_quality_primary_weak_below", None)
        primary_weak_above = getattr(self._params, "market_orb_quality_primary_weak_above", None)
        secondary_strong_above = getattr(self._params, "market_orb_quality_secondary_strong_above", None)
        primary_weak_secondary_strong = (
            primary_weak_below is not None
            and secondary_strong_above is not None
            and primary_close_loc is not None
            and secondary_close_loc is not None
            and primary_close_loc <= primary_weak_below
            and (primary_weak_above is None or primary_close_loc >= primary_weak_above)
            and secondary_close_loc >= secondary_strong_above
        )
        if primary_weak_secondary_strong:
            scale = getattr(self._params, "market_orb_quality_primary_weak_secondary_strong_scale", None)
            if scale is not None:
                scaler *= max(0.0, float(scale))
            _cap_trades(getattr(self._params, "market_orb_quality_primary_weak_secondary_strong_max_trades", None))
            reasons.append("primary_weak_secondary_strong")

        primary_lag_above = getattr(self._params, "market_orb_quality_primary_lag_above", None)
        primary_lag_below = getattr(self._params, "market_orb_quality_primary_lag_below", None)
        secondary_lead_above = getattr(self._params, "market_orb_quality_secondary_lead_above", None)
        secondary_lead_below = getattr(self._params, "market_orb_quality_secondary_lead_below", None)
        primary_lag_secondary_lead = (
            primary_lag_below is not None
            and secondary_lead_above is not None
            and primary_close_loc is not None
            and secondary_close_loc is not None
            and primary_close_loc <= primary_lag_below
            and (primary_lag_above is None or primary_close_loc >= primary_lag_above)
            and secondary_close_loc >= secondary_lead_above
            and (secondary_lead_below is None or secondary_close_loc <= secondary_lead_below)
        )
        if primary_lag_secondary_lead:
            scale = getattr(self._params, "market_orb_quality_primary_lag_secondary_lead_scale", None)
            if scale is not None:
                scaler *= max(0.0, float(scale))
            _cap_trades(getattr(self._params, "market_orb_quality_primary_lag_secondary_lead_max_trades", None))
            reasons.append("primary_lag_secondary_lead")

        joint_weak_primary_below = getattr(self._params, "market_orb_quality_joint_weak_primary_below", None)
        joint_weak_primary_above = getattr(self._params, "market_orb_quality_joint_weak_primary_above", None)
        joint_weak_secondary_below = getattr(self._params, "market_orb_quality_joint_weak_secondary_below", None)
        joint_weak_secondary_above = getattr(self._params, "market_orb_quality_joint_weak_secondary_above", None)
        joint_weak = (
            joint_weak_primary_below is not None
            and joint_weak_secondary_below is not None
            and primary_close_loc is not None
            and secondary_close_loc is not None
            and primary_close_loc <= joint_weak_primary_below
            and (joint_weak_primary_above is None or primary_close_loc >= joint_weak_primary_above)
            and secondary_close_loc <= joint_weak_secondary_below
            and (joint_weak_secondary_above is None or secondary_close_loc >= joint_weak_secondary_above)
        )
        if joint_weak:
            scale = getattr(self._params, "market_orb_quality_joint_weak_scale", None)
            if scale is not None:
                scaler *= max(0.0, float(scale))
            _cap_trades(getattr(self._params, "market_orb_quality_joint_weak_max_trades", None))
            reasons.append("joint_weak")

        joint_panic_primary_below = getattr(self._params, "market_orb_quality_joint_panic_primary_below", None)
        joint_panic_secondary_below = getattr(self._params, "market_orb_quality_joint_panic_secondary_below", None)
        joint_panic = (
            joint_panic_primary_below is not None
            and joint_panic_secondary_below is not None
            and primary_close_loc is not None
            and secondary_close_loc is not None
            and primary_close_loc <= joint_panic_primary_below
            and secondary_close_loc <= joint_panic_secondary_below
        )
        if joint_panic:
            scale = getattr(self._params, "market_orb_quality_joint_panic_scale", None)
            if scale is not None:
                scaler *= max(0.0, float(scale))
            _cap_trades(getattr(self._params, "market_orb_quality_joint_panic_max_trades", None))
            reasons.append("joint_panic")

        self._day_size_scale *= max(0.0, scaler)
        self._market_orb_quality_max_trades = max_trades
        self._market_orb_quality_reason = "+".join(reasons) if reasons else None
        self._log(
            "Market ORB quality: "
            f"{primary_ticker} close_loc={primary_close_loc if primary_close_loc is not None else 'NA'} "
            f"ret={primary_ret if primary_ret is not None else 'NA'}, "
            f"{secondary_ticker or '-'} close_loc={secondary_close_loc if secondary_close_loc is not None else 'NA'} "
            f"ret={secondary_ret if secondary_ret is not None else 'NA'}, "
            f"scale={scaler:.3f}, max_trades={max_trades}, "
            f"reason={self._market_orb_quality_reason or '-'}"
        )

    # ── Phase 1: Pre-market Screening (daily bars + enrichment + quality filter) ─

    def run_pre_screen(self, date_str: str) -> dict[str, Any]:
        """Pre-market screening: fetch daily bars, compute enrichment, filter universe.

        Called at ~9:20 ET (before market open). Narrows the universe from ~971
        to ~250-350 tickers using quality filters (price, ATR, dollar volume).
        The expensive intraday bar fetch in run_orb_detection() then only
        fetches data for pre-screened tickers.

        If this method is never called (late start, failure), run_orb_detection()
        falls back to the full pipeline automatically.

        Returns summary dict.
        """
        self._date_str = date_str
        self._state.update_daily_state(
            self._session.session_id, date_str, phase="pre_screen"
        )

        # Cheap Oracle health probe — warns early so the operator can restart Oracle
        # before the bar-fetch loop (5 chunks × 15s timeout) burns 75s silently.
        try:
            import httpx as _httpx
            from libs.oracle_client.alpaca import _base_url as _oracle_base_url
            _oracle_health_url = _oracle_base_url() + "/api/v1/health"
            _r = _httpx.get(_oracle_health_url, timeout=3.0)
            _ok = _r.status_code < 400
        except Exception:
            _ok = False
        if not _ok:
            self._log(
                "CRITICAL: Oracle is unreachable — bar fetches will likely fail. "
                "Start Oracle before 09:20 ET on live trading days."
            )
            _emit(
                "orb_engine_oracle_unreachable",
                _level="error",
                session_id=self._session.session_id,
                session_name=self._session.session_name,
            )

        universe_source = getattr(self._params, "_universe_source", "midlarge")
        universe_symbols_file = getattr(self._params, "_universe_symbols_file", None)
        tickers = load_universe(universe_source, universe_symbols_file)

        # Always include the regime ticker so regime/breadth filters have data
        regime_ticker = getattr(self._params, "market_regime_ticker", None) or "QQQ"
        fetch_tickers = list(dict.fromkeys([regime_ticker] + tickers))  # deduplicate, regime first
        self._log(f"Pre-screen: {len(tickers)} tickers (+{regime_ticker}) — fetching daily bars")

        today = dt.date.fromisoformat(date_str)
        bars_end = self._last_trading_day(today)
        start = bars_end - dt.timedelta(days=65)

        raw_bars: dict[str, list] = {}
        chunk_size = 200
        for i in range(0, len(fetch_tickers), chunk_size):
            chunk = fetch_tickers[i : i + chunk_size]
            try:
                raw_bars.update(self._broker.get_bars(chunk, start, bars_end))
            except Exception as e:
                self._log(f"  WARNING: daily bars chunk {i//chunk_size+1} failed ({e}) — skipping")
                _emit(
                    "orb_engine_bars_chunk_failed",
                    _level="warning",
                    session_id=self._session.session_id,
                    chunk_kind="daily",
                    chunk_idx=i // chunk_size + 1,
                    error=str(e),
                )

        daily_bars_dict = bars_to_enrichment_format(raw_bars)

        # Add synthetic today row so enrich_daily_bars() produces entries for date_str
        for sym, bars in daily_bars_dict.items():
            if bars:
                last = bars[-1]
                if last["date"] < date_str:
                    daily_bars_dict[sym] = bars + [{
                        "date": date_str,
                        "open": last["close"], "high": last["close"],
                        "low": last["close"], "close": last["close"],
                        "volume": 0,
                    }]

        self._enrichment = enrich_daily_bars(daily_bars_dict, [date_str])
        self._daily_bars = daily_bars_dict

        qualified = live_pre_screen(self._enrichment, date_str, self._params)
        self._pre_screened_tickers = qualified

        daily_bars_count = len([s for s, b in raw_bars.items() if b])
        self._log(
            f"Pre-screen 완료: {daily_bars_count} daily bars → "
            f"{len(qualified)} qualified (전체 {len(tickers)}개 중)"
        )
        return {
            "universe_size": len(tickers),
            "daily_bars": daily_bars_count,
            "pre_screened": len(qualified),
        }

    # ── Phase 2: ORB Detection (intraday bars → candidates) ──────────────────

    def run_orb_detection(self, date_str: str) -> dict[str, Any]:
        """Fetch 5-min ORB bars for (pre-screened or full) universe, then compute candidates.

        Called once at 9:30 + orb_minutes (e.g., 9:40 for a 10-min ORB).
        If run_pre_screen() was called earlier, uses cached enrichment and
        pre-screened ticker list (skips daily bars fetch). Otherwise runs
        the full pipeline as a fallback.

        Returns summary dict.
        """
        self._date_str = date_str
        self._reset_day_context()
        self._state.update_daily_state(
            self._session.session_id, date_str, phase="orb_detection"
        )

        # Rolling loss filter: skip day if recent N-day equity return is below threshold
        roll_days = getattr(self._params, "rolling_loss_days", None)
        roll_thresh = getattr(self._params, "rolling_loss_threshold", None)
        if roll_days is not None and roll_thresh is not None:
            snapshots = self._state.list_snapshots(self._session.session_id)
            past = [s for s in snapshots if s["date"] < date_str]
            if len(past) >= roll_days:
                window = past[-roll_days:]
                rolling_pnl = sum(s["daily_pnl"] for s in window)
                sizing_base = self._session.initial_equity  # daily_budget_reset mode
                if sizing_base > 0 and rolling_pnl / sizing_base < roll_thresh:
                    self._log(
                        f"Rolling loss filter triggered ({rolling_pnl/sizing_base:.2%} "
                        f"< {roll_thresh:.2%}) — skipping today"
                    )
                    self._state.update_daily_state(
                        self._session.session_id, date_str, phase="done"
                    )
                    return {
                        "universe_size": 0,
                        "daily_bars": 0,
                        "intraday_bars": 0,
                        "orb_candidates": 0,
                        "long": 0,
                        "short": 0,
                        "skip_reason": "rolling_loss",
                    }

        # Account-level circuit breaker: halt if equity has dropped >25% from peak
        equity_now = self._get_equity()
        peak_eq = self._state.get_peak_equity(
            self._session.session_id, self._session.initial_equity
        )
        if peak_eq > 0:
            account_dd_pct = (peak_eq - equity_now) / peak_eq * 100
            if account_dd_pct >= _ACCOUNT_CIRCUIT_BREAKER_PCT:
                self._log(
                    f"CIRCUIT BREAKER: account drawdown {account_dd_pct:.1f}% "
                    f">= {_ACCOUNT_CIRCUIT_BREAKER_PCT}% — session halted"
                )
                _emit(
                    "orb_engine_circuit_breaker",
                    _level="error",
                    session_id=self._session.session_id,
                    drawdown_pct=round(float(account_dd_pct), 2),
                    threshold_pct=_ACCOUNT_CIRCUIT_BREAKER_PCT,
                    peak_equity=round(float(peak_eq), 2),
                    current_equity=round(float(equity_now), 2),
                )
                self._state.set_session_status(self._session.session_id, "paused")
                self._state.update_daily_state(
                    self._session.session_id, date_str, phase="done"
                )
                return {
                    "universe_size": 0, "daily_bars": 0, "intraday_bars": 0,
                    "orb_candidates": 0, "long": 0, "short": 0,
                    "skip_reason": "circuit_breaker",
                }

        # ── Determine intraday_tickers: use pre-screen cache or fetch daily bars ─
        if self._enrichment and self._pre_screened_tickers is not None:
            # Pre-screen already ran — skip daily bars fetch
            candidate_tickers = list(self._pre_screened_tickers)
            candidate_ticker_set = set(candidate_tickers)
            intraday_tickers = self._with_market_context_tickers(candidate_tickers)
            daily_bars_count = len([s for s, b in self._daily_bars.items() if b])
            self._log(
                f"Pre-screened universe 사용: {len(candidate_tickers)} tickers "
                f"(daily bars 캐시됨)"
            )
        else:
            # Fallback: full pipeline (pre_screen missed or failed)
            universe_source = getattr(self._params, "_universe_source", "midlarge")
            universe_symbols_file = getattr(self._params, "_universe_symbols_file", None)
            tickers = load_universe(universe_source, universe_symbols_file)
            regime_ticker = getattr(self._params, "market_regime_ticker", None) or "QQQ"
            fetch_tickers = list(dict.fromkeys([regime_ticker] + tickers))
            self._log(f"Universe: {len(tickers)} tickers (+{regime_ticker}) — fetching daily bars")

            today = dt.date.fromisoformat(date_str)
            bars_end = self._last_trading_day(today)
            start = bars_end - dt.timedelta(days=65)

            raw_bars: dict[str, list] = {}
            chunk_size = 200
            for i in range(0, len(fetch_tickers), chunk_size):
                chunk = fetch_tickers[i : i + chunk_size]
                try:
                    raw_bars.update(self._broker.get_bars(chunk, start, bars_end))
                except Exception as e:
                    self._log(f"  WARNING: daily bars chunk {i//chunk_size+1} failed ({e}) — skipping")

            daily_bars_dict = bars_to_enrichment_format(raw_bars)

            # Add synthetic today row (yesterday's close as placeholder) so
            # enrich_daily_bars() produces a keyed entry for date_str
            for sym, bars in daily_bars_dict.items():
                if bars:
                    last = bars[-1]
                    if last["date"] < date_str:
                        daily_bars_dict[sym] = bars + [{
                            "date": date_str,
                            "open": last["close"], "high": last["close"],
                            "low": last["close"], "close": last["close"],
                            "volume": 0,
                        }]

            self._enrichment = enrich_daily_bars(daily_bars_dict, [date_str])
            self._daily_bars = daily_bars_dict
            daily_bars_count = len([s for s, b in raw_bars.items() if b])
            candidate_tickers = list(tickers)
            candidate_ticker_set = set(candidate_tickers)
            intraday_tickers = self._with_market_context_tickers(candidate_tickers)

        context_added = len(intraday_tickers) - len(candidate_ticker_set)
        if context_added > 0:
            self._log(
                f"Market context bars 포함: {context_added} tickers "
                f"({', '.join(self._market_context_tickers())})"
            )

        # ── Fetch 5-min intraday bars for (pre-screened or full) universe ──────
        today = dt.date.fromisoformat(date_str)
        market_open = dt.datetime(today.year, today.month, today.day, 9, 30, tzinfo=_ET)
        orb_end = market_open + dt.timedelta(minutes=self._params.orb_minutes + 5)
        fetch_end = dt.datetime.now(_ET).replace(second=0, microsecond=0)
        if fetch_end < orb_end:
            fetch_end = orb_end

        intraday_raw: dict[str, list[dict]] = {}
        chunk_size = 100
        for i in range(0, len(intraday_tickers), chunk_size):
            chunk = intraday_tickers[i : i + chunk_size]
            try:
                chunk_bars = self._broker.get_intraday_bars(
                    chunk,
                    start=market_open,
                    end=fetch_end,
                    timeframe_minutes=5,
                )
                intraday_raw.update(chunk_bars)
            except Exception as e:
                self._log(f"  WARNING: intraday bars chunk {i//chunk_size+1} failed ({e}) — skipping")
                _emit(
                    "orb_engine_bars_chunk_failed",
                    _level="warning",
                    session_id=self._session.session_id,
                    chunk_kind="intraday",
                    chunk_idx=i // chunk_size + 1,
                    error=str(e),
                )

        bars_by_ticker = intraday_bars_to_format(intraday_raw)
        market_bars_by_ticker = {
            ticker: mkt_bars
            for ticker, bars in bars_by_ticker.items()
            if (mkt_bars := filter_market_hours(bars))
        }
        intraday_count = len(market_bars_by_ticker)
        self._log(
            f"Daily bars: {daily_bars_count} tickers | "
            f"Intraday bars: {intraday_count} tickers"
        )
        if daily_bars_count == 0:
            self._log("WARNING: no daily bars fetched — enrichment will be empty; check Oracle/Alpaca connection")
            _emit(
                "orb_engine_no_daily_bars",
                _level="error",
                session_id=self._session.session_id,
            )
        if intraday_count == 0:
            self._log("WARNING: no intraday bars fetched — zero candidates will be produced")
            _emit(
                "orb_engine_no_intraday_bars",
                _level="error",
                session_id=self._session.session_id,
            )

        # Patch today_open in enrichment with actual first-bar open from intraday data.
        # The pre_screen synthetic row uses prev_close as today_open (gap=0), which breaks
        # market_regime_spy_threshold and breadth filters. Overwrite with real opening price.
        for ticker, ticker_bars in market_bars_by_ticker.items():
            if not ticker_bars:
                continue
            first_bar = ticker_bars[0]
            real_open = first_bar.get("open")
            if real_open and ticker in self._enrichment:
                if date_str in self._enrichment[ticker]:
                    self._enrichment[ticker][date_str]["today_open"] = real_open
                else:
                    # Fallback: find the entry that was created for this date
                    for d in sorted(self._enrichment[ticker].keys(), reverse=True):
                        if d <= date_str:
                            # Create a date_str entry inheriting from latest
                            self._enrichment[ticker][date_str] = copy.copy(
                                self._enrichment[ticker][d]
                            )
                            self._enrichment[ticker][date_str]["today_open"] = real_open
                            break

        # Market regime / breadth checks mirror simulate_orb_day's soft-day fallback.
        regime_scaler = 1.0
        breadth_scaler = 1.0
        regime_gap_pct: float | None = None
        breadth_ratio: float | None = None
        soft_day_reason_parts: list[str] = []

        regime_thresh = getattr(self._params, "market_regime_spy_threshold", None)
        if regime_thresh is not None or getattr(self._params, "regime_size_scale_low", None) is not None:
            regime_ticker = getattr(self._params, "market_regime_ticker", None) or "QQQ"
            regime_enrich = self._enrichment.get(regime_ticker, {}).get(date_str, {})
            regime_prev_close = regime_enrich.get("prev_close")
            regime_today_open = regime_enrich.get("today_open")
            if regime_prev_close and regime_today_open and regime_prev_close > 0:
                regime_gap = (regime_today_open - regime_prev_close) / regime_prev_close
                regime_gap_pct = regime_gap
                regime_skip_below = getattr(self._params, "regime_skip_below", None)
                if regime_skip_below is not None and regime_gap < regime_skip_below:
                    self._log(
                        f"Regime filter: {regime_ticker} gap {regime_gap:.3%} "
                        f"< hard floor {regime_skip_below:.3%} — skipping today"
                    )
                    self._state.update_daily_state(
                        self._session.session_id, date_str, phase="done"
                    )
                    return {
                        "universe_size": intraday_count, "daily_bars": daily_bars_count,
                        "intraday_bars": intraday_count, "orb_candidates": 0,
                        "long": 0, "short": 0, "skip_reason": "market_regime",
                    }
                if (
                    getattr(self._params, "regime_size_scale_low", None) is None
                    and regime_thresh is not None
                    and regime_gap < regime_thresh
                ):
                    if getattr(self._params, "soft_day_fallback_on_regime_skip", False):
                        regime_scaler = max(
                            0.0,
                            float(
                                getattr(
                                    self._params,
                                    "soft_day_regime_skip_size_scale",
                                    1.0,
                                )
                                or 0.0
                            ),
                        )
                        soft_day_reason_parts.append("market_regime")
                        self._log(
                            f"Regime soft fallback: {regime_ticker} gap {regime_gap:.3%} "
                            f"< {regime_thresh:.3%}; size_scale={regime_scaler:.3f}"
                        )
                    else:
                        self._log(
                            f"Regime filter: {regime_ticker} gap {regime_gap:.3%} "
                            f"< {regime_thresh:.3%} — skipping today"
                        )
                        self._state.update_daily_state(
                            self._session.session_id, date_str, phase="done"
                        )
                        return {
                            "universe_size": intraday_count, "daily_bars": daily_bars_count,
                            "intraday_bars": intraday_count, "orb_candidates": 0,
                            "long": 0, "short": 0, "skip_reason": "market_regime",
                        }
                regime_low = getattr(self._params, "regime_size_scale_low", None)
                regime_high = getattr(self._params, "regime_size_scale_high", None)
                if regime_low is not None and regime_high is not None:
                    regime_scaler = _linear_scaler(
                        regime_gap,
                        regime_low,
                        regime_high,
                        getattr(self._params, "regime_size_scale_min", 0.0),
                        invert=True,
                    )

        # Breadth filter
        min_breadth = getattr(self._params, "min_candidate_breadth", None)
        if min_breadth is not None or getattr(self._params, "breadth_size_scale_low", None) is not None:
            pos_gap_count = 0
            total_with_data = 0
            for ticker in bars_by_ticker:
                t_enrich = self._enrichment.get(ticker, {}).get(date_str, {})
                prev_c = t_enrich.get("prev_close")
                today_o = t_enrich.get("today_open")
                if prev_c and today_o and prev_c > 0:
                    total_with_data += 1
                    if today_o > prev_c:
                        pos_gap_count += 1
            if total_with_data > 0:
                breadth_ratio = pos_gap_count / total_with_data
                hard_breadth_fallback_active = False
                breadth_skip_below = getattr(self._params, "breadth_skip_below", None)
                if breadth_skip_below is not None and breadth_ratio < breadth_skip_below:
                    fallback_min = getattr(
                        self._params,
                        "hard_breadth_soft_fallback_min_breadth",
                        None,
                    )
                    if (
                        getattr(self._params, "hard_breadth_soft_fallback_enabled", False)
                        and (fallback_min is None or breadth_ratio >= float(fallback_min))
                    ):
                        breadth_scaler = max(
                            0.0,
                            float(
                                getattr(
                                    self._params,
                                    "hard_breadth_soft_fallback_size_scale",
                                    0.03,
                                )
                                or 0.0
                            ),
                        )
                        soft_day_reason_parts.append("hard_breadth")
                        hard_breadth_fallback_active = True
                        self._log(
                            f"Hard breadth soft fallback: {breadth_ratio:.1%} "
                            f"< {breadth_skip_below:.1%}; size_scale={breadth_scaler:.3f}"
                        )
                    else:
                        self._log(
                            f"Breadth filter: {breadth_ratio:.1%} positive gaps "
                            f"< hard floor {breadth_skip_below:.1%} — skipping today"
                        )
                        self._state.update_daily_state(
                            self._session.session_id, date_str, phase="done"
                        )
                        return {
                            "universe_size": intraday_count, "daily_bars": daily_bars_count,
                            "intraday_bars": intraday_count, "orb_candidates": 0,
                            "long": 0, "short": 0, "skip_reason": "breadth",
                        }
                if (
                    getattr(self._params, "breadth_size_scale_low", None) is None
                    and min_breadth is not None
                    and breadth_ratio < min_breadth
                    and not hard_breadth_fallback_active
                ):
                    if getattr(self._params, "soft_day_fallback_on_breadth_skip", False):
                        breadth_scaler = max(
                            0.0,
                            float(
                                getattr(
                                    self._params,
                                    "soft_day_breadth_skip_size_scale",
                                    1.0,
                                )
                                or 0.0
                            ),
                        )
                        soft_day_reason_parts.append("breadth")
                        self._log(
                            f"Breadth soft fallback: {breadth_ratio:.1%} "
                            f"< {min_breadth:.1%}; size_scale={breadth_scaler:.3f}"
                        )
                    else:
                        self._log(
                            f"Breadth filter: {breadth_ratio:.1%} positive gaps "
                            f"< {min_breadth:.1%} — skipping today"
                        )
                        self._state.update_daily_state(
                            self._session.session_id, date_str, phase="done"
                        )
                        return {
                            "universe_size": intraday_count, "daily_bars": daily_bars_count,
                            "intraday_bars": intraday_count, "orb_candidates": 0,
                            "long": 0, "short": 0, "skip_reason": "breadth",
                        }
                breadth_low = getattr(self._params, "breadth_size_scale_low", None)
                breadth_high = getattr(self._params, "breadth_size_scale_high", None)
                if breadth_low is not None and breadth_high is not None:
                    breadth_scaler = _linear_scaler(
                        breadth_ratio,
                        breadth_low,
                        breadth_high,
                        getattr(self._params, "breadth_size_scale_min", 0.0),
                        invert=True,
                    )

        breadth_scaler, soft_day_reason_parts = self._apply_market_thrust_breadth_override(
            market_bars_by_ticker,
            date_str,
            soft_day_reason_parts=soft_day_reason_parts,
            regime_gap_pct=regime_gap_pct,
            breadth_ratio=breadth_ratio,
            breadth_scaler=breadth_scaler,
        )
        self._day_size_scale = max(0.0, regime_scaler * breadth_scaler)
        if (
            soft_day_reason_parts
            and getattr(self._params, "soft_day_combined_size_scale_floor", None) is not None
        ):
            self._day_size_scale = max(
                self._day_size_scale,
                max(
                    0.0,
                    float(getattr(self._params, "soft_day_combined_size_scale_floor")),
                ),
            )
        if soft_day_reason_parts:
            self._soft_day_reason = "+".join(soft_day_reason_parts)
            self._log(
                f"Soft-day active: reason={self._soft_day_reason}, "
                f"day_size_scale={self._day_size_scale:.3f}"
            )

        self._apply_market_orb_quality(market_bars_by_ticker, date_str)
        if self._day_size_scale <= 0:
            self._log("Day sizing disabled by market ORB quality — candidates may be recorded, but no entries will be placed")

        broad_gapup_fallback_mode = bool(
            getattr(self._params, "broad_gapup_continuation_enabled", False)
        )
        market_thrust_liquid_mode = bool(
            getattr(self._params, "market_thrust_liquid_continuation_enabled", False)
        )
        market_thrust_opening_impulse_mode = bool(
            getattr(self._params, "market_thrust_opening_impulse_reclaim_enabled", False)
        )
        base_candidate_updates: dict[str, Any] = {}
        if broad_gapup_fallback_mode:
            base_candidate_updates["broad_gapup_continuation_enabled"] = False
        if market_thrust_liquid_mode:
            base_candidate_updates["market_thrust_liquid_continuation_enabled"] = False
        if market_thrust_opening_impulse_mode:
            base_candidate_updates["market_thrust_opening_impulse_reclaim_enabled"] = False
        candidate_params = self._copy_params_with_updates(
            self._params,
            base_candidate_updates,
        )
        computed_candidates = compute_orb_candidates(
            bars_by_ticker=bars_by_ticker,
            date_str=date_str,
            params=candidate_params,
            enrichment=self._enrichment,
            iex_live_mode=True,
        )
        normal_candidates = [
            cand for cand in computed_candidates
            if cand["ticker"] in candidate_ticker_set
        ]
        broad_candidates: list[dict[str, Any]] = []
        allow_broad_candidates = (
            broad_gapup_fallback_mode
            and (
                normal_candidates == []
                or not getattr(
                    self._params,
                    "broad_gapup_continuation_only_when_no_primary_entries",
                    False,
                )
            )
        )
        if allow_broad_candidates:
            broad_scan_params = self._copy_params_with_updates(
                self._params,
                {
                    "max_candidates": max(
                        int(getattr(self._params, "max_candidates", 0) or 0),
                        len(candidate_ticker_set),
                    ),
                    "max_candidates_per_sector": None,
                    "market_thrust_liquid_continuation_enabled": False,
                    "market_thrust_opening_impulse_reclaim_enabled": False,
                },
            )
            broad_candidates = [
                cand for cand in compute_orb_candidates(
                    bars_by_ticker=bars_by_ticker,
                    date_str=date_str,
                    params=broad_scan_params,
                    enrichment=self._enrichment,
                    iex_live_mode=True,
                )
                if cand["ticker"] in candidate_ticker_set
                and cand.get("broad_gapup_continuation")
            ]

        used_tickers = {cand["ticker"] for cand in normal_candidates + broad_candidates}
        market_thrust_liquid_candidates: list[dict[str, Any]] = []
        opening_breadth_only_thrust = (
            self._market_thrust_opening_breadth_override_active
            and not self._market_thrust_index_breadth_override_active
        )
        liquid_allowed_on_opening_breadth = bool(
            getattr(
                self._params,
                "market_thrust_opening_breadth_override_activate_liquid_continuation",
                True,
            )
        )
        market_thrust_liquid_scan_active = (
            market_thrust_liquid_mode
            and self._market_thrust_breadth_override_active
            and (
                not opening_breadth_only_thrust
                or liquid_allowed_on_opening_breadth
            )
        )
        if market_thrust_liquid_scan_active:
            market_thrust_scan_params = self._copy_params_with_updates(
                self._params,
                {
                    "max_candidates": max(
                        int(getattr(self._params, "max_candidates", 0) or 0),
                        len(candidate_ticker_set),
                    ),
                    "max_candidates_per_sector": None,
                    "broad_gapup_continuation_enabled": False,
                    "market_thrust_opening_impulse_reclaim_enabled": False,
                },
            )
            market_thrust_liquid_candidates = [
                cand for cand in compute_orb_candidates(
                    bars_by_ticker=bars_by_ticker,
                    date_str=date_str,
                    params=market_thrust_scan_params,
                    enrichment=self._enrichment,
                    iex_live_mode=True,
                )
                if cand["ticker"] in candidate_ticker_set
                and cand.get("market_thrust_liquid_continuation")
                and cand["ticker"] not in used_tickers
            ]
            max_market_thrust_candidates = getattr(
                self._params,
                "market_thrust_liquid_continuation_max_candidates",
                None,
            )
            if max_market_thrust_candidates is not None:
                market_thrust_liquid_candidates = market_thrust_liquid_candidates[
                    : max(0, int(max_market_thrust_candidates))
                ]
            used_tickers.update(cand["ticker"] for cand in market_thrust_liquid_candidates)

        market_thrust_opening_impulse_candidates: list[dict[str, Any]] = []
        if market_thrust_opening_impulse_mode and self._market_thrust_breadth_override_active:
            market_thrust_impulse_params = self._copy_params_with_updates(
                self._params,
                {
                    "max_candidates": max(
                        int(getattr(self._params, "max_candidates", 0) or 0),
                        len(candidate_ticker_set),
                    ),
                    "max_candidates_per_sector": None,
                    "broad_gapup_continuation_enabled": False,
                    "market_thrust_liquid_continuation_enabled": False,
                },
            )
            market_thrust_opening_impulse_candidates = [
                cand for cand in compute_orb_candidates(
                    bars_by_ticker=bars_by_ticker,
                    date_str=date_str,
                    params=market_thrust_impulse_params,
                    enrichment=self._enrichment,
                    iex_live_mode=True,
                )
                if cand["ticker"] in candidate_ticker_set
                and cand.get("market_thrust_opening_impulse_reclaim")
                and cand["ticker"] not in used_tickers
            ]
            max_market_thrust_impulse_candidates = getattr(
                self._params,
                "market_thrust_opening_impulse_reclaim_max_candidates",
                None,
            )
            if max_market_thrust_impulse_candidates is not None:
                market_thrust_opening_impulse_candidates = market_thrust_opening_impulse_candidates[
                    : max(0, int(max_market_thrust_impulse_candidates))
                ]

        self._candidates = (
            normal_candidates
            + broad_candidates
            + market_thrust_liquid_candidates
            + market_thrust_opening_impulse_candidates
        )

        # Save candidates to DB
        for cand in self._candidates:
            orb_bar = cand["orb_bar"]
            direction = cand["direction"]
            breakout_level = orb_bar["high"] if direction == "bullish" else orb_bar["low"]
            cand["trigger_type"] = self._candidate_trigger_type(cand)
            size_scale = self._candidate_size_scale(cand)
            cand["size_scale"] = size_scale
            row = ORBCandidateRow(
                session_id=self._session.session_id,
                date=date_str,
                ticker=cand["ticker"],
                direction=direction,
                orb_high=orb_bar["high"],
                orb_low=orb_bar["low"],
                breakout_level=breakout_level,
                atr=cand["atr"],
                rvol=cand["rvol"],
                gap_pct=cand["gap_pct"],
                composite_score=cand["score"],
                size_scale=size_scale,
                metadata_json=self._candidate_metadata_json(cand),
            )
            self._state.save_candidate(row)

        # Keep as pending (not yet filled)
        self._pending_cands = list(self._candidates)

        n_long = sum(1 for c in self._candidates if c["direction"] == "bullish")
        n_short = sum(1 for c in self._candidates if c["direction"] == "bearish")
        self._log(
            f"ORB candidates: {len(self._candidates)} "
            f"(long={n_long}, short={n_short})"
        )

        self._state.update_daily_state(
            self._session.session_id, date_str, phase="breakout"
        )

        return {
            "universe_size": len(intraday_tickers),
            "daily_bars": daily_bars_count,
            "intraday_bars": intraday_count,
            "orb_candidates": len(self._candidates),
            "long": n_long,
            "short": n_short,
        }

    # ── Phase 3: Breakout Check ───────────────────────────────────────────────

    def _post_orb_bar_high_breakout(
        self,
        bars: list[dict],
        direction: str,
        breakout_level: float,
        date_str: str,
    ) -> bool:
        """Return True if any post-ORB bar inside the order timeout crossed the breakout level.

        Mirrors libs.intraday.orb_simulator._find_breakout_time semantics: iterate
        bars after the ORB candle (ts > market_open, since 5-min bar timestamps
        denote bar start and the ORB bar starts at 09:30:00 ET), stop at
        order_timeout_minutes from market open, and use bar high (low for short)
        as the trigger. Independent of the scheduler's timing — the timeout is
        enforced internally so a manually triggered late check cannot enter a
        candidate whose only crossing was post-timeout. Used only when
        params.live_breakout_use_bar_high is True.
        """
        timeout_minutes = int(getattr(self._params, "order_timeout_minutes", 45) or 45)
        market_open = _market_open_ts(date_str)
        timeout_ts = market_open + dt.timedelta(minutes=timeout_minutes)
        for b in bars:
            try:
                ts = _parse_ts(b["timestamp"])
            except Exception:
                continue
            if ts <= market_open:
                continue
            if ts > timeout_ts:
                break
            if direction == "bullish":
                bar_high = float(b.get("high") or 0.0)
                if bar_high >= breakout_level:
                    return True
            else:
                bar_low = float(b.get("low") or 0.0)
                if bar_low > 0 and bar_low <= breakout_level:
                    return True
        return False

    def run_breakout_check(self, date_str: str) -> dict[str, Any]:
        """Check for breakouts and place orders for unfilled candidates.

        Called every sim_bar_minutes from orb_end until order_timeout elapses.
        """
        self._date_str = date_str

        # Reload state if engine was recreated (e.g., server restart)
        if not self._pending_cands and not self._candidates:
            rebuilt = self._rebuild_pending_candidates(date_str)
            self._pending_cands = rebuilt
            self._candidates = list(rebuilt)

        if not self._pending_cands:
            return {"checked": 0, "filled": 0, "remaining": 0}

        daily_state = self._state.get_daily_state(
            self._session.session_id, date_str
        )
        if daily_state.kill_switch:
            self._log("Kill switch active — skipping breakout check")
            _emit(
                "orb_engine_kill_switch_active",
                _level="warning",
                session_id=self._session.session_id,
                phase="breakout",
            )
            return {"checked": 0, "filled": 0, "remaining": 0, "kill_switch": True}

        equity = self._get_equity()

        # Fetch real-time snapshots for pending candidates via Oracle API
        tickers = [c["ticker"] for c in self._pending_cands]
        snapshots = get_snapshots(tickers)

        # Optional: also fetch today's 5-min bars to detect spike-and-retrace
        # breakouts that snapshot polling missed (live_breakout_use_bar_high).
        use_bar_high = bool(getattr(self._params, "live_breakout_use_bar_high", False))
        intraday_bars_today: dict[str, list[dict]] = {}
        if use_bar_high:
            try:
                intraday_bars_today = get_multi_intraday_bars_today(
                    tickers, interval="5min"
                )
            except Exception as e:
                self._log(f"  bar-high intraday fetch failed: {e}")
                intraday_bars_today = {}

        filled_count = 0
        still_pending = []

        # Build ticker→rank map from the master candidate list so soft-day
        # score_rank_pct matches the simulator (which ranks against the full
        # day's candidates, not the shrinking pending list).
        master_candidates = self._candidates or self._pending_cands
        n_master = len(master_candidates)
        ticker_rank_map: dict[str, int] = {
            str(c.get("ticker") or ""): i for i, c in enumerate(master_candidates)
        }

        for cand in self._pending_cands:
            ticker = cand["ticker"]
            direction = cand["direction"]
            orb_bar = cand["orb_bar"]
            atr = cand["atr"]
            score = cand["score"]
            rvol = cand["rvol"]
            trigger_type = self._candidate_trigger_type(cand)

            breakout_level = orb_bar["high"] if direction == "bullish" else orb_bar["low"]

            # Check if already traded today or at max simultaneous positions
            open_positions = self._state.get_open_positions(
                self._session.session_id, date_str
            )
            if any(p.ticker == ticker for p in open_positions):
                self._state.update_candidate_status(
                    self._session.session_id, date_str, ticker, "filled"
                )
                continue
            quality_max_trades = getattr(self, "_market_orb_quality_max_trades", None)
            if quality_max_trades is not None and len(open_positions) >= quality_max_trades:
                self._log(
                    f"  {ticker}: market ORB quality max_trades={quality_max_trades} "
                    f"reached ({self._market_orb_quality_reason or 'quality_gate'}) — cancelling"
                )
                self._state.update_candidate_status(
                    self._session.session_id, date_str, ticker, "cancelled"
                )
                continue
            max_sim = getattr(self._params, "max_simultaneous_entries", None)
            if max_sim is not None and len(open_positions) >= max_sim:
                still_pending.append(cand)
                continue

            # Check breakout using real-time snapshot price
            snap = snapshots.get(ticker)
            if snap is None or snap.price is None:
                still_pending.append(cand)
                continue

            current_price = snap.price
            opening_burst_entry = self._candidate_uses_opening_burst_entry(cand)
            if opening_burst_entry and not self._opening_burst_live_entry_allowed(date_str):
                self._state.update_candidate_status(
                    self._session.session_id, date_str, ticker, "timeout"
                )
                continue
            broke_out = opening_burst_entry or (
                (direction == "bullish" and current_price >= breakout_level)
                or (direction == "bearish" and current_price <= breakout_level)
            )
            triggered_by_bar_high = False
            if not broke_out and use_bar_high:
                bars = intraday_bars_today.get(ticker) or []
                if bars and self._post_orb_bar_high_breakout(
                    bars, direction, breakout_level, date_str
                ):
                    broke_out = True
                    triggered_by_bar_high = True
            if not broke_out:
                still_pending.append(cand)
                continue

            # Soft-day primary filter (mirrors orb_simulator.py:9932-9983).
            # Without this gate, live would trade primary ORB candidates on
            # regime/breadth soft days that the backtest skips by design,
            # producing systematic divergence from backtest expectations.
            if getattr(self, "_soft_day_reason", None):
                cand_rank = ticker_rank_map.get(ticker, n_master - 1)
                score_rank_pct = self._score_rank_pct(cand_rank, n_master)
                ticker_enrich = self._enrichment.get(ticker, {}).get(date_str, {})
                todays_trades = [
                    t for t in self._state.list_trades(self._session.session_id)
                    if str(t.get("date")) == date_str
                ]
                total_trades_today = len(open_positions) + len(todays_trades)
                reject_reason = self._evaluate_soft_day_primary_filter(
                    cand,
                    trigger_type=trigger_type,
                    score_rank_pct=score_rank_pct,
                    ticker_enrich=ticker_enrich,
                    total_trades_today=total_trades_today,
                )
                if reject_reason is not None:
                    self._log(
                        f"  {ticker}: {trigger_type} rejected by {reject_reason} "
                        f"(soft_day={self._soft_day_reason}, "
                        f"score_rank={score_rank_pct:.2f})"
                    )
                    _emit(
                        "orb_engine_soft_day_reject",
                        session_id=self._session.session_id,
                        ticker=ticker,
                        trigger_type=trigger_type,
                        reject_reason=reject_reason,
                        soft_day_reason=self._soft_day_reason,
                        score_rank_pct=round(float(score_rank_pct), 4),
                    )
                    self._state.update_candidate_status(
                        self._session.session_id, date_str, ticker, "cancelled"
                    )
                    continue

            # Breakout detected — compute position size and place order
            stop_distance = atr * self._params.atr_stop_multiplier
            if stop_distance <= 0:
                still_pending.append(cand)
                continue

            day_size_scale = max(0.0, float(getattr(self, "_day_size_scale", 1.0) or 0.0))
            candidate_size_scale = self._candidate_size_scale(cand)
            sizing_capital = (
                self._compute_sizing_capital(equity)
                * day_size_scale
                * candidate_size_scale
            )
            if sizing_capital <= 0:
                self._log(
                    f"  {ticker}: day_size_scale={day_size_scale:.3f}, "
                    f"candidate_size_scale={candidate_size_scale:.3f}; sizing disabled"
                )
                self._state.update_candidate_status(
                    self._session.session_id, date_str, ticker, "cancelled"
                )
                continue
            risk_dollars = sizing_capital * self._params.risk_per_trade_pct
            shares_from_risk = risk_dollars / stop_distance

            entry_price_est = (
                float(current_price)
                if opening_burst_entry
                else max(breakout_level, current_price)
            )
            max_shares_by_capital = (sizing_capital * self._params.max_position_pct) / entry_price_est
            shares = int(min(shares_from_risk, max_shares_by_capital))
            if shares <= 0:
                self._log(f"  {ticker}: shares=0 after sizing — skipping")
                self._state.update_candidate_status(
                    self._session.session_id, date_str, ticker, "cancelled"
                )
                continue

            # Check buying power
            try:
                acct = self._broker.get_account()
                if acct.buying_power < shares * entry_price_est:
                    self._log(f"  {ticker}: insufficient buying power — skipping")
                    self._state.update_candidate_status(
                        self._session.session_id, date_str, ticker, "cancelled"
                    )
                    continue
            except Exception as e:
                self._log(f"  {ticker}: account check error: {e}")

            # Place order
            try:
                if direction == "bullish":
                    order = self._broker.submit_market_buy(ticker, shares)
                else:
                    order = self._broker.submit_market_sell(ticker, shares)
                self._log(
                    f"  {ticker}: {trigger_type} → {shares} shares (order {order.id})"
                )
                _emit(
                    "orb_engine_buy_submitted",
                    session_id=self._session.session_id,
                    ticker=ticker, direction=direction, qty=shares,
                    trigger_type=trigger_type,
                    entry_price_est=round(entry_price_est, 4),
                    order_id=order.id,
                    category="order",
                )
            except Exception as e:
                self._log(f"  {ticker}: order failed: {e}")
                _emit(
                    "orb_engine_buy_rejected",
                    _level="error",
                    session_id=self._session.session_id,
                    ticker=ticker, direction=direction, qty=shares,
                    error=str(e),
                    category="order",
                )
                still_pending.append(cand)
                continue

            # Wait for fill (poll up to 30s)
            fill_price = entry_price_est
            order_rejected = False
            reject_status: str | None = None
            for _ in range(6):
                time.sleep(5)
                try:
                    filled_order = self._broker.get_order(order.id)
                    if filled_order.status == "filled" and filled_order.filled_avg_price:
                        fill_price = filled_order.filled_avg_price
                        break
                    if filled_order.status in ("cancelled", "rejected", "expired"):
                        self._log(f"  {ticker}: order {filled_order.status} — no position created")
                        order_rejected = True
                        reject_status = filled_order.status
                        break
                except Exception:
                    pass

            if order_rejected:
                _emit(
                    "orb_engine_buy_rejected",
                    _level="error",
                    session_id=self._session.session_id,
                    ticker=ticker, qty=shares,
                    order_id=order.id,
                    status=reject_status or "unknown",
                    category="order",
                )
                self._state.update_candidate_status(
                    self._session.session_id, date_str, ticker, "cancelled"
                )
                continue
            if triggered_by_bar_high:
                gap = float(breakout_level) - float(fill_price)
                self._log(
                    f"  {ticker}: bar-high trigger (snap retrace) — "
                    f"breakout_level={breakout_level:.4f}, fill={float(fill_price):.4f}, "
                    f"gap={gap:.4f}"
                )
                _emit(
                    "orb_engine_bar_high_trigger",
                    session_id=self._session.session_id,
                    ticker=ticker, direction=direction,
                    breakout_level=round(float(breakout_level), 4),
                    fill_price=round(float(fill_price), 4),
                    gap=round(gap, 4),
                    category="order",
                )
            _emit(
                "orb_engine_buy_filled",
                session_id=self._session.session_id,
                ticker=ticker, direction=direction, qty=shares,
                fill_price=round(float(fill_price), 4),
                trigger_type=trigger_type,
                order_id=order.id,
                category="order",
            )

            # Record position
            initial_stop = (
                fill_price - stop_distance
                if direction == "bullish"
                else fill_price + stop_distance
            )
            pos = ORBPositionRow(
                session_id=self._session.session_id,
                date=date_str,
                ticker=ticker,
                direction="long" if direction == "bullish" else "short",
                entry_price=fill_price,
                entry_time=dt.datetime.now(_ET).isoformat(),
                shares=shares,
                orb_high=orb_bar["high"],
                orb_low=orb_bar["low"],
                atr_at_entry=atr,
                stop_distance=stop_distance,
                current_stop=initial_stop,
                peak_price=fill_price,
                trigger_type=trigger_type,
                rvol=rvol,
                composite_score=score,
                order_id=order.id,
            )
            self._state.save_position(pos)
            self._state.update_candidate_status(
                self._session.session_id, date_str, ticker, "filled"
            )
            filled_count += 1

            # Check kill switches
            daily_state = self._state.get_daily_state(
                self._session.session_id, date_str
            )
            if daily_state.kill_switch:
                self._log("Kill switch triggered — stopping breakout monitoring")
                _emit(
                    "orb_engine_kill_switch_triggered",
                    _level="error",
                    session_id=self._session.session_id,
                    phase="breakout",
                )
                break

        self._pending_cands = still_pending
        self._log(
            f"Breakout check: filled={filled_count}, remaining={len(still_pending)}"
        )

        return {
            "checked": len(tickers),
            "filled": filled_count,
            "remaining": len(still_pending),
        }

    def _now_et(self) -> dt.datetime:
        return dt.datetime.now(_ET)

    @staticmethod
    def _score_rank_pct(rank: int, count: int) -> float:
        if count <= 1:
            return 1.0
        return 1.0 - (rank / max(count - 1, 1))

    @staticmethod
    def _candidate_direction(cand: dict[str, Any]) -> str | None:
        direction = cand.get("direction")
        if direction == "bullish":
            return "long"
        if direction == "bearish":
            return "short"
        if direction in {"long", "short"}:
            return str(direction)
        return None

    def _candidate_trigger_type(self, cand: dict[str, Any]) -> str:
        """Return the backtest trigger family that this live candidate represents."""
        if cand.get("broad_gapup_continuation"):
            return "broad_gapup_continuation"
        if cand.get("market_thrust_liquid_continuation"):
            entry_mode = str(
                getattr(
                    self._params,
                    "market_thrust_liquid_continuation_entry_mode",
                    "breakout",
                )
                or "breakout"
            ).lower()
            if entry_mode == "opening_burst":
                return "market_thrust_opening_burst"
            if entry_mode in {"opening_followthrough", "followthrough_burst"}:
                return "market_thrust_opening_followthrough"
            return "market_thrust_liquid_continuation"
        if cand.get("market_thrust_opening_impulse_reclaim"):
            return "market_thrust_opening_impulse_reclaim"
        return str(cand.get("trigger_type") or "orb")

    def _candidate_uses_opening_burst_entry(self, cand: dict[str, Any]) -> bool:
        if cand.get("broad_gapup_continuation"):
            entry_mode = str(
                getattr(self._params, "broad_gapup_continuation_entry_mode", "breakout")
                or "breakout"
            ).lower()
            return entry_mode == "opening_burst"
        if cand.get("market_thrust_liquid_continuation"):
            entry_mode = str(
                getattr(
                    self._params,
                    "market_thrust_liquid_continuation_entry_mode",
                    "breakout",
                )
                or "breakout"
            ).lower()
            return entry_mode == "opening_burst"
        if cand.get("market_thrust_opening_impulse_reclaim"):
            entry_mode = str(
                getattr(
                    self._params,
                    "market_thrust_opening_impulse_reclaim_entry_mode",
                    "vwap_reclaim",
                )
                or "vwap_reclaim"
            ).lower()
            return entry_mode == "opening_burst"
        return False

    def _opening_burst_live_entry_allowed(self, date_str: str) -> bool:
        """Allow live opening-burst entries only near the intended opening window."""
        now_et = self._now_et()
        market_open = _market_open_ts(date_str)
        max_minutes = getattr(
            self._params,
            "opening_burst_liquid_max_entry_minutes_after_open",
            None,
        )
        if max_minutes is None:
            max_minutes = max(
                int(getattr(self._params, "orb_minutes", 5) or 5)
                + int(getattr(self._params, "sim_bar_minutes", 5) or 5),
                int(getattr(self._params, "order_timeout_minutes", 25) or 25),
            )
        return now_et <= market_open + dt.timedelta(minutes=float(max_minutes))

    @staticmethod
    def _trade_trigger_type(row: Any) -> str:
        if isinstance(row, dict):
            return str(row.get("trigger_type") or "orb")
        return str(getattr(row, "trigger_type", None) or "orb")

    def _live_primary_trade_exists(
        self,
        trades: list[Any],
        *,
        allowed_primary_families: set[str] | None = None,
    ) -> bool:
        allowed = set(_AUXILIARY_TRIGGER_TYPES)
        if allowed_primary_families:
            allowed.update(allowed_primary_families)
        return any(self._trade_trigger_type(trade) not in allowed for trade in trades)

    def _live_trigger_count_today(self, trigger_type: str) -> int:
        if not getattr(self, "_date_str", None):
            return 0
        open_positions = self._state.get_open_positions(
            self._session.session_id,
            self._date_str,
        )
        open_count = sum(
            1 for pos in open_positions
            if self._trade_trigger_type(pos) == trigger_type
        )
        todays_trades = [
            t for t in self._state.list_trades(self._session.session_id)
            if str(t.get("date")) == self._date_str
        ]
        closed_count = sum(
            1 for trade in todays_trades
            if self._trade_trigger_type(trade) == trigger_type
        )
        return open_count + closed_count

    def _nofill_vwap_live_allowed(self, score_rank_pct: float) -> bool:
        min_score = getattr(self._params, "nofill_vwap_reclaim_min_score_pct", None)
        return min_score is None or score_rank_pct >= float(min_score)

    def _candidate_decision_premarket_dollar_vol(self, cand: dict[str, Any]) -> float:
        raw = cand.get("filter_premarket_dollar_vol", cand.get("premarket_dollar_vol"))
        try:
            return float(raw or 0.0)
        except (TypeError, ValueError):
            return 0.0

    def _soft_day_vwap_live_allowed(
        self,
        cand: dict[str, Any],
        score_rank_pct: float,
        todays_trades: list[dict[str, Any]],
    ) -> bool:
        if not bool(getattr(self._params, "soft_day_vwap_reclaim_enabled", False)):
            return False
        soft_day_reason = getattr(self, "_soft_day_reason", None)
        if not soft_day_reason:
            return False
        if not _orb_soft_day_vwap_reason_allowed(self._params, soft_day_reason):
            return False
        if getattr(self._params, "soft_day_vwap_reclaim_only_when_no_existing_trades", False):
            if self._state.get_open_positions(self._session.session_id, self._date_str):
                return False
            todays_trades = [
                t for t in self._state.list_trades(self._session.session_id)
                if str(t.get("date")) == self._date_str
            ]
            if todays_trades:
                return False
        if (
            getattr(self._params, "soft_day_vwap_reclaim_only_when_no_primary_trades", False)
            and self._live_primary_trade_exists(todays_trades)
        ):
            return False
        min_score = _orb_soft_day_vwap_min_score_pct(self._params, soft_day_reason)
        if min_score is not None and score_rank_pct < float(min_score):
            return False
        min_pm = _orb_soft_day_vwap_reason_param(
            self._params,
            soft_day_reason,
            "soft_day_vwap_reclaim_min_premarket_dollar_vol",
        )
        if min_pm is not None and self._candidate_decision_premarket_dollar_vol(cand) < min_pm:
            return False
        min_ret5 = _orb_soft_day_vwap_reason_param(
            self._params,
            soft_day_reason,
            "soft_day_vwap_reclaim_min_ret_5d",
        )
        ret_5d = cand.get("ret_5d")
        try:
            ret_5d_float = None if ret_5d is None else float(ret_5d)
        except (TypeError, ValueError):
            ret_5d_float = None
        if min_ret5 is not None and (ret_5d_float is None or ret_5d_float < min_ret5):
            return False
        gap_down_max_ret5 = _orb_soft_day_vwap_reason_param(
            self._params,
            soft_day_reason,
            "soft_day_vwap_reclaim_gap_down_max_ret_5d",
        )
        gap_pct = cand.get("gap_pct")
        try:
            gap_pct_float = None if gap_pct is None else float(gap_pct)
        except (TypeError, ValueError):
            gap_pct_float = None
        if (
            gap_down_max_ret5 is not None
            and gap_pct_float is not None
            and gap_pct_float < 0
            and ret_5d_float is not None
            and ret_5d_float > gap_down_max_ret5
        ):
            return False
        weak_rank = _orb_soft_day_vwap_reason_param(
            self._params,
            soft_day_reason,
            "soft_day_vwap_reclaim_weak_participation_max_rvol_rank_pct",
        )
        weak_body = _orb_soft_day_vwap_reason_param(
            self._params,
            soft_day_reason,
            "soft_day_vwap_reclaim_weak_participation_max_body_ratio",
        )
        if weak_rank is not None and weak_body is not None:
            try:
                rvol_rank = float(cand.get("rvol_rank_pct"))
                body_ratio = float(cand.get("body_ratio"))
            except (TypeError, ValueError):
                rvol_rank = None
                body_ratio = None
            if (
                rvol_rank is not None
                and body_ratio is not None
                and rvol_rank <= weak_rank
                and body_ratio <= weak_body
            ):
                return False
        min_rvol = _orb_soft_day_vwap_reason_param(
            self._params,
            soft_day_reason,
            "soft_day_vwap_reclaim_min_rvol",
        )
        max_rvol = _orb_soft_day_vwap_reason_param(
            self._params,
            soft_day_reason,
            "soft_day_vwap_reclaim_max_rvol",
        )
        try:
            rvol = float(cand.get("filter_rvol", cand.get("rvol")))
        except (TypeError, ValueError):
            rvol = None
        if min_rvol is not None and (rvol is None or rvol < min_rvol):
            return False
        if max_rvol is not None and (rvol is None or rvol > max_rvol):
            return False
        min_body = _orb_soft_day_vwap_reason_param(
            self._params,
            soft_day_reason,
            "soft_day_vwap_reclaim_min_body_ratio",
        )
        if min_body is not None:
            try:
                body_ratio = float(cand.get("body_ratio"))
            except (TypeError, ValueError):
                body_ratio = None
            if body_ratio is None or body_ratio < min_body:
                return False
        min_close = _orb_soft_day_vwap_reason_param(
            self._params,
            soft_day_reason,
            "soft_day_vwap_reclaim_min_close_location",
        )
        if min_close is not None:
            try:
                close_location = float(cand.get("close_location"))
            except (TypeError, ValueError):
                close_location = None
            if close_location is None or close_location < min_close:
                return False
        return True

    def _evaluate_soft_day_primary_filter(
        self,
        cand: dict[str, Any],
        trigger_type: str,
        score_rank_pct: float,
        ticker_enrich: dict[str, Any],
        total_trades_today: int,
    ) -> str | None:
        """Mirror orb_simulator.py:9932-9983 soft-day gates for primary triggers.

        Returns a rejection reason string or None if the candidate passes.
        Only applies when soft_day_reason is set; aux/market-thrust triggers are
        exempt because they have their own gating earlier in the pipeline.
        """
        soft_day_reason = getattr(self, "_soft_day_reason", None)
        if not soft_day_reason:
            return None

        # soft_day_max_trades caps total day trade count for non-exempt triggers.
        max_trades = getattr(self._params, "soft_day_max_trades", None)
        if (
            trigger_type not in _SOFT_DAY_MAX_TRADES_EXEMPT_TRIGGER_TYPES
            and max_trades is not None
            and total_trades_today >= int(max_trades)
        ):
            return "soft_day_max_trades"

        # Score / profile / ret5 / rvol gates only apply to primary breakout
        # triggers; aux + market-thrust families are exempt.
        if trigger_type in _SOFT_DAY_PRIMARY_FILTER_EXEMPT_TRIGGER_TYPES:
            return None

        # Sector confirmation override can bypass the score/profile gates.
        sector_override_active = _orb_soft_day_sector_confirmation_override_allows(
            self._params,
            soft_day_reason,
            cand,
            score_rank_pct=score_rank_pct,
            trigger_type=trigger_type,
        )
        max_sector_override = getattr(
            self._params,
            "soft_day_sector_confirmation_override_max_trades",
            1,
        )
        if (
            sector_override_active
            and max_sector_override is not None
            and self._live_trigger_count_today("sector_confirmation_override")
            >= max(0, int(max_sector_override))
        ):
            sector_override_active = False
        if sector_override_active:
            return None

        min_score_pct = getattr(self._params, "soft_day_min_score_pct", None)
        if min_score_pct is not None and score_rank_pct < float(min_score_pct):
            return "soft_day_score"

        min_cand_score = getattr(self._params, "soft_day_min_candidate_score", None)
        if min_cand_score is not None:
            try:
                cand_score = float(cand.get("score") or 0.0)
            except (TypeError, ValueError):
                cand_score = 0.0
            if cand_score < float(min_cand_score):
                return "soft_day_candidate_score"

        min_ret_5d = getattr(self._params, "soft_day_min_ret_5d", None)
        if min_ret_5d is not None:
            ret_5d_raw = cand.get("ret_5d", ticker_enrich.get("ret_5d"))
            try:
                ret_5d_val = None if ret_5d_raw is None else float(ret_5d_raw)
            except (TypeError, ValueError):
                ret_5d_val = None
            if ret_5d_val is None or ret_5d_val < float(min_ret_5d):
                return "soft_day_ret5"

        max_rvol = getattr(self._params, "soft_day_max_rvol", None)
        if max_rvol is not None:
            cand_rvol_raw = cand.get("rvol")
            try:
                cand_rvol = None if cand_rvol_raw is None else float(cand_rvol_raw)
            except (TypeError, ValueError):
                cand_rvol = None
            if cand_rvol is None or cand_rvol > float(max_rvol):
                return "soft_day_rvol"

        if not _orb_soft_day_setup_profile_allows(
            self._params,
            soft_day_reason,
            cand,
            ticker_enrich,
        ):
            return "soft_day_profile"

        return None

    def _reclaim_trade_slot_available(self, trigger_type: str) -> bool:
        counts = getattr(self, "_live_reclaim_trade_counts", {})
        if not isinstance(counts, dict):
            counts = {}
        if trigger_type == "vwap_reclaim":
            max_trades = getattr(self._params, "nofill_vwap_reclaim_max_trades", 1)
        elif trigger_type == "market_thrust_opening_impulse_reclaim":
            max_trades = getattr(
                self._params,
                "market_thrust_opening_impulse_reclaim_max_trades",
                1,
            )
            if not getattr(self, "_market_thrust_breadth_override_active", False):
                no_thrust_max = getattr(
                    self._params,
                    "market_thrust_opening_impulse_reclaim_no_thrust_max_trades",
                    None,
                )
                if no_thrust_max is not None:
                    max_trades = no_thrust_max
        else:
            max_trades = getattr(self._params, "soft_day_vwap_reclaim_max_trades", 1)
        live_count = max(
            int(counts.get(trigger_type, 0)),
            self._live_trigger_count_today(trigger_type),
        )
        return max_trades is None or live_count < int(max_trades)

    def _increment_reclaim_trade_count(self, trigger_type: str) -> None:
        counts = getattr(self, "_live_reclaim_trade_counts", {})
        if not isinstance(counts, dict):
            counts = {}
        counts[trigger_type] = int(counts.get(trigger_type, 0)) + 1
        self._live_reclaim_trade_counts = counts

    def _reclaim_live_size_scale(self, trigger_type: str) -> float:
        if trigger_type == "vwap_reclaim":
            raw = getattr(self._params, "nofill_vwap_reclaim_size_scale", 1.0)
        elif trigger_type == "market_thrust_opening_impulse_reclaim":
            raw = getattr(
                self._params,
                "market_thrust_opening_impulse_reclaim_size_scale",
                1.0,
            )
            if not getattr(self, "_market_thrust_breadth_override_active", False):
                no_thrust_raw = getattr(
                    self._params,
                    "market_thrust_opening_impulse_reclaim_no_thrust_size_scale",
                    None,
                )
                if no_thrust_raw is not None:
                    raw = no_thrust_raw
        else:
            raw = _orb_soft_day_vwap_size_scale(
                self._params,
                getattr(self, "_soft_day_reason", None),
            )
        return max(0.0, min(1.0, float(raw or 0.0)))

    def _market_thrust_impulse_late_live_enabled(self) -> bool:
        if not bool(
            getattr(
                self._params,
                "market_thrust_opening_impulse_reclaim_enabled",
                False,
            )
        ):
            return False
        entry_mode = str(
            getattr(
                self._params,
                "market_thrust_opening_impulse_reclaim_entry_mode",
                "vwap_reclaim",
            )
            or "vwap_reclaim"
        ).lower()
        return entry_mode == "late_breakout"

    def _market_thrust_impulse_late_live_allowed(
        self,
        cand: dict[str, Any],
        score_rank_pct: float,
        todays_trades: list[dict[str, Any]],
    ) -> bool:
        if not self._market_thrust_impulse_late_live_enabled():
            return False
        if not bool(cand.get("market_thrust_opening_impulse_reclaim")):
            return False
        if (
            getattr(
                self._params,
                "market_thrust_opening_impulse_reclaim_only_when_no_primary_trades",
                False,
            )
            and not getattr(self, "_market_thrust_breadth_override_active", False)
            and self._live_primary_trade_exists(
                todays_trades,
                allowed_primary_families=(
                    _MARKET_THRUST_LIQUID_TRIGGER_TYPES
                    | _MARKET_THRUST_IMPULSE_TRIGGER_TYPES
                ),
            )
        ):
            return False
        min_score = getattr(
            self._params,
            "market_thrust_opening_impulse_reclaim_min_score_pct",
            None,
        )
        return min_score is None or score_rank_pct >= float(min_score)

    def _submit_live_entry(
        self,
        cand: dict[str, Any],
        date_str: str,
        *,
        current_price: float,
        entry_price_est: float,
        trigger_label: str,
    ) -> bool:
        ticker = cand["ticker"]
        direction = cand["direction"]
        orb_bar = cand["orb_bar"]
        atr = cand["atr"]
        score = cand["score"]
        rvol = cand.get("rvol")

        open_positions = self._state.get_open_positions(self._session.session_id, date_str)
        if any(p.ticker == ticker for p in open_positions):
            return False
        max_sim = getattr(self._params, "max_simultaneous_entries", None)
        if max_sim is not None and len(open_positions) >= int(max_sim):
            return False

        stop_distance = atr * self._params.atr_stop_multiplier
        if stop_distance <= 0:
            return False

        equity = self._get_equity()
        day_size_scale = max(0.0, float(getattr(self, "_day_size_scale", 1.0) or 0.0))
        candidate_size_scale = self._candidate_size_scale(cand)
        sizing_capital = (
            self._compute_sizing_capital(equity)
            * day_size_scale
            * candidate_size_scale
        )
        if sizing_capital <= 0:
            self._log(
                f"  {ticker}: {trigger_label} sizing disabled "
                f"(day={day_size_scale:.3f}, candidate={candidate_size_scale:.3f})"
            )
            return False

        risk_dollars = sizing_capital * self._params.risk_per_trade_pct
        shares_from_risk = risk_dollars / stop_distance
        max_shares_by_capital = (sizing_capital * self._params.max_position_pct) / entry_price_est
        shares = int(min(shares_from_risk, max_shares_by_capital))
        if shares <= 0:
            return False

        try:
            acct = self._broker.get_account()
            if acct.buying_power < shares * entry_price_est:
                self._log(f"  {ticker}: insufficient buying power for {trigger_label}")
                return False
        except Exception as exc:
            self._log(f"  {ticker}: account check error during {trigger_label}: {exc}")

        try:
            if direction == "bullish":
                order = self._broker.submit_market_buy(ticker, shares)
            else:
                order = self._broker.submit_market_sell(ticker, shares)
            self._log(
                f"  {ticker}: {trigger_label} → {shares} shares "
                f"(order {order.id})"
            )
            _emit(
                "orb_engine_reclaim_buy_submitted",
                session_id=self._session.session_id,
                ticker=ticker,
                direction=direction,
                qty=shares,
                trigger_type=trigger_label,
                entry_price_est=round(entry_price_est, 4),
                current_price=round(current_price, 4),
                order_id=order.id,
                category="order",
            )
        except Exception as exc:
            self._log(f"  {ticker}: {trigger_label} order failed: {exc}")
            return False

        fill_price = entry_price_est
        for _ in range(6):
            time.sleep(5)
            try:
                filled_order = self._broker.get_order(order.id)
                if filled_order.status == "filled" and filled_order.filled_avg_price:
                    fill_price = filled_order.filled_avg_price
                    break
                if filled_order.status in ("cancelled", "rejected", "expired"):
                    return False
            except Exception:
                pass

        initial_stop = (
            fill_price - stop_distance
            if direction == "bullish"
            else fill_price + stop_distance
        )
        pos = ORBPositionRow(
            session_id=self._session.session_id,
            date=date_str,
            ticker=ticker,
            direction="long" if direction == "bullish" else "short",
            entry_price=fill_price,
            entry_time=self._now_et().isoformat(),
            shares=shares,
            orb_high=orb_bar["high"],
            orb_low=orb_bar["low"],
            atr_at_entry=atr,
            stop_distance=stop_distance,
            current_stop=initial_stop,
            peak_price=fill_price,
            trigger_type=trigger_label,
            rvol=rvol,
            composite_score=score,
            order_id=order.id,
        )
        self._state.save_position(pos)
        self._state.update_candidate_status(
            self._session.session_id,
            date_str,
            ticker,
            "filled",
        )
        _emit(
            "orb_engine_reclaim_buy_filled",
            session_id=self._session.session_id,
            ticker=ticker,
            direction=direction,
            qty=shares,
            fill_price=round(float(fill_price), 4),
            trigger_type=trigger_label,
            order_id=order.id,
            category="order",
        )
        return True

    def run_reclaim_check(self, date_str: str) -> dict[str, Any]:
        """Live timed-entry check for fallback sleeves.

        Backtests generate VWAP reclaim and market-thrust late-breakout entries
        as timed pass-2 entries. Live needs an explicit polling path because
        normal breakout checks end at `order_timeout_minutes`, while these
        windows usually start later.
        """
        self._date_str = date_str
        vwap_reclaim_enabled = (
            bool(getattr(self._params, "nofill_vwap_reclaim_enabled", False))
            or bool(getattr(self._params, "soft_day_vwap_reclaim_enabled", False))
        )
        impulse_late_enabled = self._market_thrust_impulse_late_live_enabled()
        if not (vwap_reclaim_enabled or impulse_late_enabled):
            return {"reclaim_checked": 0, "reclaim_filled": 0, "skip_reason": "disabled"}
        daily_state = self._state.get_daily_state(self._session.session_id, date_str)
        if bool(getattr(daily_state, "kill_switch", False)):
            return {"reclaim_checked": 0, "reclaim_filled": 0, "skip_reason": "kill_switch"}

        now_et = self._now_et()
        market_open = _market_open_ts(date_str)
        windows: list[tuple[int, int]] = []
        if vwap_reclaim_enabled:
            windows.append(
                (
                    int(getattr(self._params, "vwap_reclaim_window_start_min", 30) or 30),
                    int(getattr(self._params, "vwap_reclaim_window_end_min", 150) or 150),
                )
            )
        if impulse_late_enabled:
            windows.append(
                (
                    int(getattr(self._params, "late_breakout_window_start_min", 30) or 30),
                    int(getattr(self._params, "late_breakout_window_end_min", 150) or 150),
                )
            )
        window_start = market_open + dt.timedelta(minutes=min(start for start, _ in windows))
        window_end = market_open + dt.timedelta(minutes=max(end for _, end in windows))
        if now_et < window_start:
            return {"reclaim_checked": 0, "reclaim_filled": 0, "skip_reason": "not_started"}
        grace_minutes = max(5, int(getattr(self._params, "sim_bar_minutes", 5) or 5))
        if now_et > window_end + dt.timedelta(minutes=grace_minutes):
            return {"reclaim_checked": 0, "reclaim_filled": 0, "skip_reason": "ended"}

        candidates = list(getattr(self, "_candidates", []) or [])
        if not candidates:
            rebuilt = self._rebuild_pending_candidates(date_str)
            self._pending_cands = rebuilt
            self._candidates = list(rebuilt)
            candidates = list(rebuilt)
        if not candidates:
            return {"reclaim_checked": 0, "reclaim_filled": 0, "skip_reason": "no_live_candidates"}

        open_positions = self._state.get_open_positions(self._session.session_id, date_str)
        open_tickers = {p.ticker for p in open_positions}
        todays_trades = [
            t for t in self._state.list_trades(self._session.session_id)
            if str(t.get("date")) == date_str
        ]
        traded_tickers = {str(t.get("ticker")) for t in todays_trades}
        unavailable_tickers = open_tickers | traded_tickers

        max_trades = getattr(self._params, "max_trades_per_day", None)
        if max_trades is not None and len(open_positions) + len(todays_trades) >= int(max_trades):
            return {"reclaim_checked": 0, "reclaim_filled": 0, "skip_reason": "max_trades"}

        tickers = [
            str(c.get("ticker"))
            for c in candidates
            if c.get("ticker") and str(c.get("ticker")) not in unavailable_tickers
        ]
        if not tickers:
            return {"reclaim_checked": 0, "reclaim_filled": 0, "skip_reason": "no_available_candidates"}

        try:
            bars_raw = self._broker.get_intraday_bars(
                tickers,
                start=market_open,
                end=now_et.replace(second=0, microsecond=0),
                timeframe_minutes=5,
            )
        except Exception as exc:
            self._log(f"  reclaim intraday fetch failed: {exc}")
            return {"reclaim_checked": len(tickers), "reclaim_filled": 0, "skip_reason": "bars_failed"}

        bars_by_ticker = intraday_bars_to_format(bars_raw)
        snapshots = get_snapshots(tickers)
        seen_signals: set[tuple[str, str, str]] = getattr(
            self,
            "_live_reclaim_seen_signals",
            set(),
        )
        self._live_reclaim_seen_signals = seen_signals

        signals: list[tuple[dt.datetime, float, dict[str, Any], str, float]] = []
        for rank, base_cand in enumerate(candidates):
            ticker = str(base_cand.get("ticker") or "")
            if not ticker or ticker in unavailable_tickers:
                continue
            mkt_bars = filter_market_hours(bars_by_ticker.get(ticker, []))
            if not mkt_bars:
                continue
            cand = dict(base_cand)
            cand["mkt_bars"] = mkt_bars
            score_rank_pct = self._score_rank_pct(rank, len(candidates))
            direction = self._candidate_direction(cand)
            if direction is None:
                continue

            if bool(getattr(self._params, "nofill_vwap_reclaim_enabled", False)):
                nofill = _find_vwap_reclaim_time(
                    mkt_bars,
                    cand["orb_bar"],
                    direction,
                    self._params,
                    date_str,
                )
                if nofill is not None and self._nofill_vwap_live_allowed(score_rank_pct):
                    key = (ticker, "vwap_reclaim", nofill[0].isoformat())
                    if key not in seen_signals:
                        signals.append((nofill[0], nofill[1], cand, "vwap_reclaim", score_rank_pct))

            if self._soft_day_vwap_live_allowed(cand, score_rank_pct, todays_trades):
                soft = _find_vwap_reclaim_time(
                    mkt_bars,
                    cand["orb_bar"],
                    direction,
                    self._params,
                    date_str,
                )
                if soft is not None:
                    key = (ticker, "soft_day_vwap_reclaim", soft[0].isoformat())
                    if key not in seen_signals:
                        signals.append((soft[0], soft[1], cand, "soft_day_vwap_reclaim", score_rank_pct))

            if self._market_thrust_impulse_late_live_allowed(
                cand,
                score_rank_pct,
                todays_trades,
            ):
                late = _find_late_breakout_time(
                    mkt_bars,
                    cand["orb_bar"],
                    direction,
                    self._params,
                    date_str,
                )
                if late is not None:
                    key = (
                        ticker,
                        "market_thrust_opening_impulse_reclaim",
                        late[0].isoformat(),
                    )
                    if key not in seen_signals:
                        signals.append(
                            (
                                late[0],
                                late[1],
                                cand,
                                "market_thrust_opening_impulse_reclaim",
                                score_rank_pct,
                            )
                        )

        if not signals:
            return {"reclaim_checked": len(tickers), "reclaim_filled": 0}

        signals.sort(key=lambda item: (item[0], -float(item[2].get("score") or 0.0)))
        filled = 0
        for signal_ts, signal_price, cand, trigger_type, _score_rank_pct in signals:
            if signal_ts > now_et:
                continue
            key = (cand["ticker"], trigger_type, signal_ts.isoformat())
            if key in seen_signals:
                continue
            if not self._reclaim_trade_slot_available(trigger_type):
                continue
            snap = snapshots.get(cand["ticker"])
            if snap is None or snap.price is None:
                continue
            sleeve_scale = self._reclaim_live_size_scale(trigger_type)
            base_scale = self._candidate_size_scale(cand)
            cand["size_scale"] = max(0.0, min(1.0, base_scale * sleeve_scale))
            if self._submit_live_entry(
                cand,
                date_str,
                current_price=float(snap.price),
                entry_price_est=max(float(snap.price), float(signal_price)),
                trigger_label=trigger_type,
            ):
                seen_signals.add(key)
                self._increment_reclaim_trade_count(trigger_type)
                filled += 1
                if max_trades is not None and len(open_positions) + len(todays_trades) + filled >= int(max_trades):
                    break

        if filled:
            self._log(f"Reclaim check: filled={filled}, checked={len(tickers)}")
        return {"reclaim_checked": len(tickers), "reclaim_filled": filled}

    # ── Phase 4: Stop Check (sim_bar_minutes checkpoints) ────────────────────

    def run_stop_check(self, date_str: str) -> dict[str, Any]:
        """Evaluate stops for all open positions using aggregated bars.

        Called every sim_bar_minutes from orb_end until 15:55 ET.
        Stop logic mirrors orb_simulator.py:477-580 exactly.
        """
        self._date_str = date_str
        reclaim_summary = self.run_reclaim_check(date_str)

        positions = self._state.get_open_positions(self._session.session_id, date_str)
        if not positions:
            return {"positions_checked": 0, "stops_hit": 0, **reclaim_summary}

        daily_state = self._state.get_daily_state(
            self._session.session_id, date_str
        )
        if daily_state.kill_switch:
            return {"positions_checked": 0, "stops_hit": 0, "kill_switch": True, **reclaim_summary}

        today = dt.date.fromisoformat(date_str)
        market_open = dt.datetime(today.year, today.month, today.day, 9, 30, tzinfo=_ET)
        now_et = dt.datetime.now(_ET)

        tickers = [p.ticker for p in positions]
        bars_raw = self._broker.get_intraday_bars(
            tickers,
            start=market_open,
            end=now_et,
            timeframe_minutes=5,
        )


        group_size = self._params.sim_bar_minutes // 5

        stops_hit = 0
        equity = self._get_equity()

        for pos in positions:
            ticker = pos.ticker
            all_bars = bars_raw.get(ticker, [])
            mkt_bars = filter_market_hours(all_bars)

            if not mkt_bars:
                continue

            # Filter bars after entry time
            entry_ts = _parse_ts(pos.entry_time)
            post_entry = [b for b in mkt_bars if _parse_ts(b["timestamp"]) > entry_ts]

            if not post_entry:
                continue

            # Aggregate to sim_bar_minutes (e.g., 90-min)
            agg_bars = _aggregate_bars(post_entry, group_size)

            # Run stop management on each aggregated bar
            current_stop = pos.current_stop
            peak_price = pos.peak_price
            trailing_active = pos.trailing_active
            stop_distance = pos.stop_distance
            atr = pos.atr_at_entry
            exit_bar = None
            exit_reason = "close"

            use_atr_trail = self._params.trailing_stop_atr_multiplier > 0

            for bar in agg_bars:
                bar_high = bar["high"]
                bar_low = bar["low"]

                if pos.direction == "long":
                    peak_price = max(peak_price, bar_high)
                    current_r = (bar_high - pos.entry_price) / stop_distance if stop_distance > 0 else 0

                    if current_r >= self._params.breakeven_at_r and current_stop < pos.entry_price:
                        current_stop = pos.entry_price

                    if current_r >= self._params.trailing_at_r:
                        trailing_active = True

                    # Check stop hit BEFORE updating trailing
                    if bar_low <= current_stop:
                        exit_bar = bar
                        exit_reason = "trailing_stop" if trailing_active else "stop_loss"
                        break

                    # Update trailing AFTER stop check
                    if trailing_active:
                        if use_atr_trail:
                            tighten_r = getattr(self._params, "trailing_tighten_at_r", None)
                            tight_mult = getattr(self._params, "trailing_stop_atr_multiplier_tight", 0.0)
                            if (tighten_r is not None and current_r >= tighten_r and tight_mult > 0):
                                atr_mult = tight_mult
                            else:
                                atr_mult = self._params.trailing_stop_atr_multiplier
                            candidate = peak_price - atr * atr_mult
                        else:
                            candidate = max(bar_low, current_stop)
                        if candidate > current_stop:
                            current_stop = candidate

                else:  # short
                    peak_price = min(peak_price, bar_low)
                    current_r = (pos.entry_price - bar_low) / stop_distance if stop_distance > 0 else 0

                    if current_r >= self._params.breakeven_at_r and current_stop > pos.entry_price:
                        current_stop = pos.entry_price

                    if current_r >= self._params.trailing_at_r:
                        trailing_active = True

                    if bar_high >= current_stop:
                        exit_bar = bar
                        exit_reason = "trailing_stop" if trailing_active else "stop_loss"
                        break

                    if trailing_active:
                        if use_atr_trail:
                            tighten_r = getattr(self._params, "trailing_tighten_at_r", None)
                            tight_mult = getattr(self._params, "trailing_stop_atr_multiplier_tight", 0.0)
                            if (tighten_r is not None and current_r >= tighten_r and tight_mult > 0):
                                atr_mult = tight_mult
                            else:
                                atr_mult = self._params.trailing_stop_atr_multiplier
                            candidate = peak_price + atr * atr_mult
                        else:
                            candidate = min(bar_high, current_stop)
                        if candidate < current_stop:
                            current_stop = candidate

            # Update DB stop levels
            self._state.update_position_stop(
                self._session.session_id, date_str, ticker,
                current_stop, peak_price, trailing_active,
            )

            if exit_bar:
                # Close position — use qty so only this session's shares are closed
                # (other sessions may hold the same ticker in the same Alpaca account).
                exit_price = current_stop  # fallback if fill poll fails
                try:
                    close_order = self._broker.close_position(ticker, qty=int(pos.shares))
                    # Poll for actual broker fill price (captures gap-through losses)
                    for _ in range(4):
                        time.sleep(3)
                        try:
                            o = self._broker.get_order(close_order.id)
                            if o.filled_avg_price:
                                exit_price = o.filled_avg_price
                                break
                        except Exception:
                            pass
                    self._log(
                        f"  {ticker}: stop hit ({exit_reason}) @ {exit_price:.2f}"
                    )
                    _emit(
                        "orb_engine_close_filled",
                        session_id=self._session.session_id,
                        ticker=ticker, qty=int(pos.shares),
                        exit_price=round(float(exit_price), 4),
                        entry_price=round(float(pos.entry_price), 4),
                        reason="stop",
                        exit_subreason=exit_reason,
                        order_id=close_order.id,
                        category="order",
                    )
                except Exception as e:
                    self._log(f"  {ticker}: close error: {e}")
                    _emit(
                        "orb_engine_close_failed",
                        _level="error",
                        session_id=self._session.session_id,
                        ticker=ticker, qty=int(pos.shares),
                        reason="stop",
                        exit_subreason=exit_reason,
                        error=str(e),
                        category="order",
                    )

                self._record_trade(pos, exit_price, exit_bar["timestamp"], exit_reason, equity)
                stops_hit += 1

                # Update daily kill switches
                loss = (exit_price - pos.entry_price) * pos.shares
                if pos.direction == "short":
                    loss = (pos.entry_price - exit_price) * pos.shares
                if loss < 0:
                    new_cum_loss = daily_state.cumulative_loss + abs(loss)
                    new_stops = daily_state.stops_hit + 1
                    kill = (
                        new_cum_loss >= equity * self._params.daily_max_loss_pct
                        or new_stops >= self._params.max_stops_per_day
                    )
                    self._state.update_daily_state(
                        self._session.session_id, date_str,
                        cumulative_loss=new_cum_loss,
                        stops_hit=new_stops,
                        kill_switch=kill,
                    )
                    daily_state = self._state.get_daily_state(
                        self._session.session_id, date_str
                    )
                    if kill:
                        self._log("Kill switch triggered!")
                        _emit(
                            "orb_engine_kill_switch_triggered",
                            _level="error",
                            session_id=self._session.session_id,
                            phase="stop_check",
                            cumulative_loss=round(float(new_cum_loss), 2),
                            stops_hit=int(new_stops),
                        )
                        break

        self._log(f"Stop check: {len(positions)} positions, {stops_hit} stops hit")
        return {"positions_checked": len(positions), "stops_hit": stops_hit, **reclaim_summary}

    # ── Phase 5: EOD Exit ─────────────────────────────────────────────────────

    def run_eod_exit(self, date_str: str) -> dict[str, Any]:
        """Close all remaining open positions at 15:55 ET."""
        self._date_str = date_str
        self._state.update_daily_state(
            self._session.session_id, date_str, phase="eod_exit"
        )

        # Cancel any unfilled breakout candidates — must run unconditionally so
        # stale pending records are cleaned up even when there are no open positions
        # (e.g., server restarted after ORB detection but before any breakout).
        for cand in self._pending_cands:
            self._state.update_candidate_status(
                self._session.session_id, date_str, cand["ticker"], "timeout"
            )
        self._pending_cands = []

        db_cands = self._state.list_candidates(self._session.session_id, date_str)
        for c in db_cands:
            if c["status"] == "pending":
                self._state.update_candidate_status(
                    self._session.session_id, date_str, c["ticker"], "timeout"
                )

        positions = self._state.get_open_positions(self._session.session_id, date_str)
        if not positions:
            self._log("EOD: no open positions")
            return {"closed": 0}

        equity = self._get_equity()
        closed = 0
        now_str = dt.datetime.now(_ET).isoformat()

        for pos in positions:
            try:
                # Use qty so only this session's shares are closed
                close_order = self._broker.close_position(pos.ticker, qty=int(pos.shares))
                exit_price = pos.entry_price
                for _ in range(4):
                    time.sleep(3)
                    try:
                        o = self._broker.get_order(close_order.id)
                        if o.filled_avg_price:
                            exit_price = o.filled_avg_price
                            break
                    except Exception:
                        pass

                self._record_trade(pos, exit_price, now_str, "close", equity)
                self._log(f"  EOD close: {pos.ticker} @ {exit_price:.2f}")
                _emit(
                    "orb_engine_close_filled",
                    session_id=self._session.session_id,
                    ticker=pos.ticker, qty=int(pos.shares),
                    exit_price=round(float(exit_price), 4),
                    entry_price=round(float(pos.entry_price), 4),
                    reason="eod",
                    order_id=close_order.id,
                    category="order",
                )
                closed += 1
            except Exception as e:
                self._log(f"  EOD close error {pos.ticker}: {e}")
                _emit(
                    "orb_engine_close_failed",
                    _level="error",
                    session_id=self._session.session_id,
                    ticker=pos.ticker, qty=int(pos.shares),
                    reason="eod",
                    error=str(e),
                    category="order",
                )
                # Mark as closed in DB anyway to prevent zombie positions
                self._state.close_position_record(
                    self._session.session_id, date_str, pos.ticker
                )

        return {"closed": closed}

    # ── Phase 6: Post-close ───────────────────────────────────────────────────

    def run_post_close(self, date_str: str) -> dict[str, Any]:
        """Record daily equity snapshot and finalize day."""
        self._date_str = date_str
        self._state.update_daily_state(
            self._session.session_id, date_str, phase="done"
        )

        trades_today = self._state.list_trades(self._session.session_id)
        today_trades = [t for t in trades_today if t["date"] == date_str]

        daily_pnl = sum(t["pnl"] for t in today_trades)
        stops_hit = sum(1 for t in today_trades if t["exit_reason"] in ("stop_loss", "trailing_stop"))

        prev_equity = self._get_equity()
        if prev_equity is None:
            prev_equity = self._session.initial_equity
        new_equity = max(prev_equity + daily_pnl, 0.01)

        # Drawdown
        peak_equity = self._state.get_peak_equity(
            self._session.session_id, self._session.initial_equity
        )
        drawdown_pct = ((new_equity - peak_equity) / peak_equity * 100) if peak_equity > 0 else 0.0

        snap = ORBDailySnapshotRow(
            session_id=self._session.session_id,
            date=date_str,
            equity=new_equity,
            daily_pnl=daily_pnl,
            total_pnl=new_equity - self._session.initial_equity,
            trades_taken=len(today_trades),
            stops_hit=stops_hit,
            drawdown_pct=drawdown_pct,
        )
        self._state.save_daily_snapshot(snap)

        self._log(
            f"Post-close: equity={new_equity:.2f}, pnl={daily_pnl:+.2f}, "
            f"trades={len(today_trades)}, stops={stops_hit}"
        )

        return {
            "equity": new_equity,
            "daily_pnl": daily_pnl,
            "trades": len(today_trades),
            "stops_hit": stops_hit,
            "drawdown_pct": drawdown_pct,
        }

    # ── Helpers ───────────────────────────────────────────────────────────────

    def _record_trade(
        self,
        pos: ORBPositionRow,
        exit_price: float,
        exit_time: str,
        exit_reason: str,
        equity: float,
    ) -> None:
        """Record a completed trade in the DB and close the position record."""
        if pos.direction == "long":
            pnl = (exit_price - pos.entry_price) * pos.shares
        else:
            pnl = (pos.entry_price - exit_price) * pos.shares

        r_multiple = (
            pnl / (pos.stop_distance * pos.shares)
            if pos.stop_distance > 0 and pos.shares > 0
            else 0.0
        )

        trade = ORBTradeRow(
            trade_id=str(uuid.uuid4())[:12],
            session_id=self._session.session_id,
            date=pos.date,
            ticker=pos.ticker,
            direction=pos.direction,
            entry_price=pos.entry_price,
            exit_price=exit_price,
            entry_time=pos.entry_time,
            exit_time=exit_time,
            shares=pos.shares,
            pnl=round(pnl, 4),
            r_multiple=round(r_multiple, 3),
            exit_reason=exit_reason,
            trigger_type=pos.trigger_type,
            atr_at_entry=pos.atr_at_entry,
            rvol=pos.rvol,
            composite_score=pos.composite_score,
        )
        self._state.save_trade(trade)
        self._state.close_position_record(
            self._session.session_id, pos.date, pos.ticker
        )

    def _compute_sizing_capital(self, equity: float) -> float:
        """Replicate backtest sizing_capital formula: governor + streak multiplier.

        Mirrors libs/intraday/orb_simulator.py:2141-2187.
        V23 uses daily_budget_reset=True: base sizing = initial_equity (not equity).
        This matches the backtest 단리 mode where each day starts from $10k.
        """
        # daily_budget_reset: fixed daily budget matches V23 backtest 단리 mode
        daily_reset = getattr(self._params, "daily_budget_reset", False)
        sizing = self._session.initial_equity if daily_reset else equity

        # Drawdown governor: scale down when equity drops below peak
        gov_thresh = getattr(self._params, "drawdown_governor_threshold", None)
        gov_min = getattr(self._params, "drawdown_governor_min_scale", 0.30)
        if gov_thresh is not None:
            peak_equity = self._state.get_peak_equity(
                self._session.session_id, self._session.initial_equity
            )
            if peak_equity > 0:
                dd_pct = (peak_equity - equity) / peak_equity
                if dd_pct > gov_thresh:
                    dd_excess = dd_pct - gov_thresh
                    governor_scale = max(
                        gov_min,
                        1.0 - (1.0 - gov_min) * min(dd_excess / gov_thresh, 1.0),
                    )
                    sizing = sizing * governor_scale

        # Streak sizing: amplify after consecutive wins, reduce after consecutive losses.
        # list_trades returns DESC (newest first) — outcomes[0] = most recent trade.
        win_bonus = getattr(self._params, "streak_sizing_win_bonus", None)
        loss_penalty = getattr(self._params, "streak_sizing_loss_penalty", None)
        streak_max = getattr(self._params, "streak_sizing_max", 2.5)
        streak_min = getattr(self._params, "streak_sizing_min", 0.5)
        if win_bonus is not None or loss_penalty is not None:
            trades = self._state.list_trades(self._session.session_id)
            if trades:
                outcomes = [t["pnl"] > 0 for t in trades]  # newest first
                is_winning = outcomes[0]  # most recent outcome
                streak_len = 0
                for o in outcomes:  # count from newest
                    if o == is_winning:
                        streak_len += 1
                    else:
                        break
                streak_mult = 1.0
                if is_winning and win_bonus is not None:
                    streak_mult = 1.0 + streak_len * win_bonus
                elif not is_winning and loss_penalty is not None:
                    streak_mult = 1.0 - streak_len * loss_penalty
                streak_mult = max(streak_min, min(streak_max, streak_mult))
                sizing = sizing * streak_mult

        return sizing

    def _candidate_size_scale(self, cand: dict[str, Any]) -> float:
        """Per-candidate live sizing scale for sleeves that survive in memory/DB."""
        stored_scale = cand.get("size_scale")
        if stored_scale is not None:
            return max(0.0, min(1.0, float(stored_scale)))

        def _float(value: Any) -> float | None:
            try:
                return None if value is None else float(value)
            except (TypeError, ValueError):
                return None

        def _trigger_allowed(allowed: Any, trigger: str) -> bool:
            return not allowed or trigger in {str(item) for item in allowed}

        def _ticker_enrichment() -> dict[str, Any]:
            enrichment = getattr(self, "_enrichment", {}) or {}
            if not isinstance(enrichment, dict):
                return {}
            raw = enrichment.get(str(cand.get("ticker") or ""))
            return raw if isinstance(raw, dict) else {}

        scale = 1.0
        trigger_type = self._candidate_trigger_type(cand)
        gap_pct = _float(cand.get("gap_pct"))
        enrich = _ticker_enrichment()
        ret_5d = _float(cand.get("ret_5d", enrich.get("ret_5d")))
        raw_premarket_dollar_vol = _float(
            cand.get("premarket_dollar_vol", enrich.get("premarket_dollar_vol"))
        )
        filter_premarket_dollar_vol = _float(cand.get("filter_premarket_dollar_vol"))
        decision_premarket_dollar_vol = (
            filter_premarket_dollar_vol
            if filter_premarket_dollar_vol is not None
            else raw_premarket_dollar_vol
        )
        premarket_dollar_vol = decision_premarket_dollar_vol or 0.0
        rvol = _float(cand.get("filter_rvol"))
        if rvol is None:
            rvol = _float(cand.get("rvol"))
        first_bar_dollar_vol = _float(cand.get("filter_first_bar_dollar_vol"))
        if first_bar_dollar_vol is None:
            first_bar_dollar_vol = _float(cand.get("first_bar_dollar_vol"))
        body_ratio = _float(cand.get("body_ratio"))
        close_location = _float(cand.get("close_location"))
        orb_return = _float(cand.get("orb_return"))

        if bool(getattr(self._params, "broad_gapup_continuation_enabled", False)):
            is_broad_gapup = bool(cand.get("broad_gapup_continuation"))
            if not is_broad_gapup:
                max_gap = getattr(self._params, "max_gap_pct", None)
                is_broad_gapup = (
                    max_gap is not None
                    and gap_pct is not None
                    and gap_pct > float(max_gap)
                )
            if is_broad_gapup:
                raw_scale = getattr(
                    self._params,
                    "broad_gapup_continuation_size_scale",
                    1.0,
                )
                scale *= max(0.0, min(1.0, float(raw_scale or 0.0)))

        is_bullish = cand.get("direction") in {"bullish", "long"}
        sector_confirmed = bool(cand.get("sector_confirmation_active"))
        overextended_min_gap = getattr(
            self._params,
            "overextended_downside_reclaim_min_abs_gap_pct",
            None,
        )
        overextended_min_ret = getattr(
            self._params,
            "overextended_downside_reclaim_min_ret_5d",
            None,
        )
        if (
            is_bullish
            and gap_pct is not None
            and overextended_min_gap is not None
            and overextended_min_ret is not None
            and gap_pct <= -float(overextended_min_gap)
            and ret_5d is not None
            and ret_5d >= float(overextended_min_ret)
            and _trigger_allowed(
                getattr(
                    self._params,
                    "overextended_downside_reclaim_allowed_trigger_types",
                    None,
                ),
                trigger_type,
            )
            and (
                getattr(
                    self._params,
                    "overextended_downside_reclaim_min_premarket_dollar_vol",
                    None,
                )
                is None
                or premarket_dollar_vol >= float(
                    getattr(
                        self._params,
                        "overextended_downside_reclaim_min_premarket_dollar_vol",
                    )
                )
            )
        ):
            raw_scale = getattr(
                self._params,
                "overextended_downside_reclaim_size_scale",
                1.0,
            )
            scale *= max(0.0, min(1.0, float(raw_scale or 0.0)))

        mid_attention_scale = getattr(
            self._params,
            "mid_attention_exhaustion_size_scale",
            1.0,
        )
        if (
            is_bullish
            and mid_attention_scale is not None
            and float(mid_attention_scale) < 1.0
            and rvol is not None
            and getattr(self._params, "mid_attention_exhaustion_min_rvol", None) is not None
            and getattr(self._params, "mid_attention_exhaustion_max_rvol", None) is not None
            and getattr(
                self._params,
                "mid_attention_exhaustion_min_premarket_dollar_vol",
                None,
            )
            is not None
            and getattr(
                self._params,
                "mid_attention_exhaustion_max_premarket_dollar_vol",
                None,
            )
            is not None
            and _trigger_allowed(
                getattr(
                    self._params,
                    "mid_attention_exhaustion_allowed_trigger_types",
                    None,
                ),
                trigger_type,
            )
            and float(getattr(self._params, "mid_attention_exhaustion_min_rvol"))
            <= rvol
            <= float(getattr(self._params, "mid_attention_exhaustion_max_rvol"))
            and float(
                getattr(
                    self._params,
                    "mid_attention_exhaustion_min_premarket_dollar_vol",
                )
            )
            <= premarket_dollar_vol
            <= float(
                getattr(
                    self._params,
                    "mid_attention_exhaustion_max_premarket_dollar_vol",
                )
            )
        ):
            scale *= max(0.0, min(1.0, float(mid_attention_scale)))

        mid_liquidity_scale = getattr(
            self._params,
            "mid_liquidity_fragility_size_scale",
            1.0,
        )
        if (
            is_bullish
            and mid_liquidity_scale is not None
            and float(mid_liquidity_scale) < 1.0
            and decision_premarket_dollar_vol is not None
            and _trigger_allowed(
                getattr(
                    self._params,
                    "mid_liquidity_fragility_allowed_trigger_types",
                    None,
                ),
                trigger_type,
            )
        ):
            thin_min_pm = getattr(
                self._params,
                "mid_liquidity_fragility_thin_min_premarket_dollar_vol",
                None,
            )
            thin_max_pm = getattr(
                self._params,
                "mid_liquidity_fragility_thin_max_premarket_dollar_vol",
                None,
            )
            thin_active = False
            if thin_min_pm is not None and thin_max_pm is not None:
                thin_profile = (
                    float(thin_min_pm)
                    <= decision_premarket_dollar_vol
                    <= float(thin_max_pm)
                )
                if thin_profile:
                    thin_clauses = []
                    thin_max_rvol = getattr(
                        self._params,
                        "mid_liquidity_fragility_thin_max_rvol",
                        None,
                    )
                    if thin_max_rvol is not None:
                        thin_clauses.append(
                            rvol is not None and rvol <= float(thin_max_rvol)
                        )
                    thin_min_ret = getattr(
                        self._params,
                        "mid_liquidity_fragility_thin_min_ret_5d",
                        None,
                    )
                    thin_max_ret = getattr(
                        self._params,
                        "mid_liquidity_fragility_thin_max_ret_5d",
                        None,
                    )
                    if thin_min_ret is not None and thin_max_ret is not None:
                        thin_clauses.append(
                            ret_5d is not None
                            and float(thin_min_ret)
                            <= ret_5d
                            <= float(thin_max_ret)
                        )
                    thin_min_body = getattr(
                        self._params,
                        "mid_liquidity_fragility_thin_min_body_ratio",
                        None,
                    )
                    if thin_min_body is not None:
                        thin_clauses.append(
                            body_ratio is not None
                            and body_ratio >= float(thin_min_body)
                        )
                    thin_active = any(thin_clauses)

            mid_min_pm = getattr(
                self._params,
                "mid_liquidity_fragility_mid_min_premarket_dollar_vol",
                None,
            )
            mid_max_pm = getattr(
                self._params,
                "mid_liquidity_fragility_mid_max_premarket_dollar_vol",
                None,
            )
            mid_min_body = getattr(
                self._params,
                "mid_liquidity_fragility_mid_min_body_ratio",
                None,
            )
            mid_max_body = getattr(
                self._params,
                "mid_liquidity_fragility_mid_max_body_ratio",
                None,
            )
            mid_active = (
                mid_min_pm is not None
                and mid_max_pm is not None
                and mid_min_body is not None
                and mid_max_body is not None
                and body_ratio is not None
                and float(mid_min_pm)
                <= decision_premarket_dollar_vol
                <= float(mid_max_pm)
                and float(mid_min_body) <= body_ratio <= float(mid_max_body)
            )
            if thin_active or mid_active:
                scale *= max(0.0, min(1.0, float(mid_liquidity_scale)))

        orphan_thin_scale = getattr(
            self._params,
            "orphan_thin_attention_size_scale",
            1.0,
        )
        orphan_thin_max_pm = getattr(
            self._params,
            "orphan_thin_attention_max_premarket_dollar_vol",
            None,
        )
        if (
            is_bullish
            and orphan_thin_scale is not None
            and float(orphan_thin_scale) < 1.0
            and orphan_thin_max_pm is not None
            and decision_premarket_dollar_vol is not None
            and decision_premarket_dollar_vol <= float(orphan_thin_max_pm)
            and not sector_confirmed
            and _trigger_allowed(
                getattr(
                    self._params,
                    "orphan_thin_attention_allowed_trigger_types",
                    None,
                ),
                trigger_type,
            )
        ):
            scale *= max(0.0, min(1.0, float(orphan_thin_scale)))

        gap_fill_trap_scale = getattr(
            self._params,
            "gap_up_fill_trap_size_scale",
            1.0,
        )
        gap_fill_trap_max_orb_return = getattr(
            self._params,
            "gap_up_fill_trap_max_orb_return",
            None,
        )
        if (
            is_bullish
            and gap_fill_trap_scale is not None
            and float(gap_fill_trap_scale) < 1.0
            and gap_fill_trap_max_orb_return is not None
            and orb_return is not None
            and orb_return <= float(gap_fill_trap_max_orb_return)
            and _trigger_allowed(
                getattr(
                    self._params,
                    "gap_up_fill_trap_allowed_trigger_types",
                    None,
                ),
                trigger_type,
            )
        ):
            gap_fill_exit_active = bool(cand.get("gap_up_fill_exit_active"))
            gap_fill_min_gap = getattr(
                self._params,
                "gap_up_fill_exit_min_gap_pct",
                None,
            )
            if not gap_fill_exit_active and gap_fill_min_gap is not None:
                gap_fill_exit_active = (
                    gap_pct is not None
                    and gap_pct >= float(gap_fill_min_gap)
                    and (
                        getattr(self._params, "gap_up_fill_exit_max_gap_pct", None)
                        is None
                        or gap_pct
                        <= float(getattr(self._params, "gap_up_fill_exit_max_gap_pct"))
                    )
                    and (
                        getattr(self._params, "gap_up_fill_exit_min_ret_5d", None)
                        is None
                        or (
                            ret_5d is not None
                            and ret_5d
                            >= float(getattr(self._params, "gap_up_fill_exit_min_ret_5d"))
                        )
                    )
                    and (
                        getattr(self._params, "gap_up_fill_exit_max_ret_5d", None)
                        is None
                        or (
                            ret_5d is not None
                            and ret_5d
                            <= float(getattr(self._params, "gap_up_fill_exit_max_ret_5d"))
                        )
                    )
                    and (
                        getattr(
                            self._params,
                            "gap_up_fill_exit_max_premarket_dollar_vol",
                            None,
                        )
                        is None
                        or premarket_dollar_vol
                        <= float(
                            getattr(
                                self._params,
                                "gap_up_fill_exit_max_premarket_dollar_vol",
                            )
                        )
                    )
                    and (
                        getattr(self._params, "gap_up_fill_exit_min_body_ratio", None)
                        is None
                        or (
                            body_ratio is not None
                            and body_ratio
                            >= float(getattr(self._params, "gap_up_fill_exit_min_body_ratio"))
                        )
                    )
                    and (
                        getattr(self._params, "gap_up_fill_exit_max_body_ratio", None)
                        is None
                        or (
                            body_ratio is not None
                            and body_ratio
                            <= float(getattr(self._params, "gap_up_fill_exit_max_body_ratio"))
                        )
                    )
                    and (
                        getattr(
                            self._params,
                            "gap_up_fill_exit_max_close_location",
                            None,
                        )
                        is None
                        or (
                            close_location is not None
                            and close_location
                            <= float(
                                getattr(
                                    self._params,
                                    "gap_up_fill_exit_max_close_location",
                                )
                            )
                        )
                    )
                )
            if gap_fill_exit_active:
                scale *= max(0.0, min(1.0, float(gap_fill_trap_scale)))

        low_quality_scale = getattr(
            self._params,
            "low_candidate_quality_size_scale",
            1.0,
        )
        low_quality_max_score = getattr(
            self._params,
            "low_candidate_quality_max_score",
            None,
        )
        candidate_score = _float(cand.get("score", cand.get("candidate_score")))
        if (
            is_bullish
            and low_quality_scale is not None
            and float(low_quality_scale) < 1.0
            and low_quality_max_score is not None
            and candidate_score is not None
            and candidate_score <= float(low_quality_max_score)
            and _trigger_allowed(
                getattr(
                    self._params,
                    "low_candidate_quality_allowed_trigger_types",
                    None,
                ),
                trigger_type,
            )
        ):
            scale *= max(0.0, min(1.0, float(low_quality_scale)))

        unsupported_scale = getattr(
            self._params,
            "unsupported_attention_size_scale",
            1.0,
        )
        if (
            is_bullish
            and unsupported_scale is not None
            and float(unsupported_scale) < 1.0
            and _trigger_allowed(
                getattr(
                    self._params,
                    "unsupported_attention_allowed_trigger_types",
                    None,
                ),
                trigger_type,
            )
        ):
            high_conviction_scales = [
                _float(cand.get(name)) or 1.0
                for name in (
                    "red_to_green_acceleration_size_scale",
                    "liquid_leader_conviction_size_scale",
                    "opening_burst_liquid_size_scale",
                    "ownership_initial_size_scale",
                    "form4_size_scale",
                )
            ]
            liquid_min_pm = getattr(
                self._params,
                "unsupported_attention_liquid_min_premarket_dollar_vol",
                None,
            )
            liquid_max_score = getattr(
                self._params,
                "unsupported_attention_liquid_max_candidate_score",
                None,
            )
            liquid_active = (
                liquid_min_pm is not None
                and liquid_max_score is not None
                and decision_premarket_dollar_vol is not None
                and decision_premarket_dollar_vol >= float(liquid_min_pm)
                and candidate_score is not None
                and candidate_score <= float(liquid_max_score)
                and (
                    not bool(
                        getattr(
                            self._params,
                            "unsupported_attention_liquid_require_no_sector_confirmation",
                            True,
                        )
                    )
                    or not sector_confirmed
                )
            )

            thin_max_pm = getattr(
                self._params,
                "unsupported_attention_thin_max_premarket_dollar_vol",
                None,
            )
            thin_min_gap = getattr(
                self._params,
                "unsupported_attention_thin_min_gap_pct",
                None,
            )
            thin_min_rvol = getattr(
                self._params,
                "unsupported_attention_thin_min_rvol",
                None,
            )
            thin_max_first_bar = getattr(
                self._params,
                "unsupported_attention_thin_max_first_bar_dollar_vol",
                None,
            )
            thin_active = (
                thin_max_pm is not None
                and thin_min_gap is not None
                and thin_min_rvol is not None
                and decision_premarket_dollar_vol is not None
                and decision_premarket_dollar_vol <= float(thin_max_pm)
                and gap_pct is not None
                and gap_pct >= float(thin_min_gap)
                and rvol is not None
                and rvol >= float(thin_min_rvol)
                and (
                    thin_max_first_bar is None
                    or (
                        first_bar_dollar_vol is not None
                        and first_bar_dollar_vol <= float(thin_max_first_bar)
                    )
                )
            )
            thin_max_rvol = getattr(
                self._params,
                "unsupported_attention_thin_max_rvol",
                None,
            )
            if thin_active and thin_max_rvol is not None:
                thin_active = rvol is not None and rvol <= float(thin_max_rvol)

            ignore_high_conviction = bool(
                getattr(
                    self._params,
                    "unsupported_attention_ignore_high_conviction",
                    True,
                )
            )
            if (
                (liquid_active or thin_active)
                and (
                    not ignore_high_conviction
                    or not any(item > 1.0 for item in high_conviction_scales)
                )
            ):
                scale *= max(0.0, min(1.0, float(unsupported_scale)))

        if is_bullish and gap_pct is not None and not sector_confirmed:
            body_ratio = _float(cand.get("body_ratio")) or 0.0
            close_location = _float(cand.get("close_location")) or 0.0
            orb_return = _float(cand.get("orb_return")) or 0.0
            score_rank_pct = _float(cand.get("score_rank_pct"))

            isolated_min_gap = getattr(
                self._params,
                "isolated_downside_loss_cap_min_abs_gap_pct",
                None,
            )
            if (
                isolated_min_gap is not None
                and gap_pct <= -float(isolated_min_gap)
                and _trigger_allowed(
                    getattr(
                        self._params,
                        "isolated_downside_loss_cap_allowed_trigger_types",
                        None,
                    ),
                    trigger_type,
                )
                and (
                    getattr(self._params, "isolated_downside_loss_cap_max_ret_5d", None)
                    is None
                    or (ret_5d is not None and ret_5d <= float(getattr(self._params, "isolated_downside_loss_cap_max_ret_5d")))
                )
                and (
                    getattr(
                        self._params,
                        "isolated_downside_loss_cap_min_premarket_dollar_vol",
                        None,
                    )
                    is None
                    or premarket_dollar_vol >= float(getattr(self._params, "isolated_downside_loss_cap_min_premarket_dollar_vol"))
                )
                and (
                    getattr(
                        self._params,
                        "isolated_downside_loss_cap_max_premarket_dollar_vol",
                        None,
                    )
                    is None
                    or premarket_dollar_vol <= float(getattr(self._params, "isolated_downside_loss_cap_max_premarket_dollar_vol"))
                )
                and (
                    getattr(self._params, "isolated_downside_loss_cap_max_body_ratio", None)
                    is None
                    or body_ratio <= float(getattr(self._params, "isolated_downside_loss_cap_max_body_ratio"))
                )
                and (
                    getattr(self._params, "isolated_downside_loss_cap_min_body_ratio", None)
                    is None
                    or body_ratio >= float(getattr(self._params, "isolated_downside_loss_cap_min_body_ratio"))
                )
                and (
                    getattr(self._params, "isolated_downside_loss_cap_max_close_location", None)
                    is None
                    or close_location <= float(getattr(self._params, "isolated_downside_loss_cap_max_close_location"))
                )
                and (
                    getattr(self._params, "isolated_downside_loss_cap_min_close_location", None)
                    is None
                    or close_location >= float(getattr(self._params, "isolated_downside_loss_cap_min_close_location"))
                )
                and (
                    getattr(self._params, "isolated_downside_loss_cap_min_orb_return", None)
                    is None
                    or orb_return >= float(getattr(self._params, "isolated_downside_loss_cap_min_orb_return"))
                )
                and (
                    getattr(self._params, "isolated_downside_loss_cap_max_score_rank_pct", None)
                    is None
                    or (
                        score_rank_pct is not None
                        and score_rank_pct <= float(getattr(self._params, "isolated_downside_loss_cap_max_score_rank_pct"))
                    )
                )
            ):
                raw_scale = getattr(self._params, "isolated_downside_size_scale", 1.0)
                scale *= max(0.0, min(1.0, float(raw_scale or 0.0)))

            pressure_min_gap = getattr(
                self._params,
                "isolated_downside_pressure_min_abs_gap_pct",
                None,
            )
            if (
                pressure_min_gap is not None
                and gap_pct <= -float(pressure_min_gap)
                and _trigger_allowed(
                    getattr(
                        self._params,
                        "isolated_downside_pressure_allowed_trigger_types",
                        None,
                    ),
                    trigger_type,
                )
                and (
                    getattr(self._params, "isolated_downside_pressure_max_ret_5d", None)
                    is None
                    or (ret_5d is not None and ret_5d <= float(getattr(self._params, "isolated_downside_pressure_max_ret_5d")))
                )
                and (
                    getattr(
                        self._params,
                        "isolated_downside_pressure_min_premarket_dollar_vol",
                        None,
                    )
                    is None
                    or premarket_dollar_vol >= float(getattr(self._params, "isolated_downside_pressure_min_premarket_dollar_vol"))
                )
                and (
                    getattr(self._params, "isolated_downside_pressure_max_body_ratio", None)
                    is None
                    or body_ratio <= float(getattr(self._params, "isolated_downside_pressure_max_body_ratio"))
                )
                and (
                    getattr(self._params, "isolated_downside_pressure_max_close_location", None)
                    is None
                    or close_location <= float(getattr(self._params, "isolated_downside_pressure_max_close_location"))
                )
            ):
                raw_scale = getattr(
                    self._params,
                    "isolated_downside_pressure_size_scale",
                    1.0,
                )
                scale *= max(0.0, min(1.0, float(raw_scale or 0.0)))

        return max(0.0, min(1.0, scale))

    def _rebuild_pending_candidates(self, date_str: str) -> list[dict]:
        """Reconstruct pending candidates from DB (after server restart)."""
        db_cands = self._state.list_candidates(self._session.session_id, date_str)
        open_positions = self._state.get_open_positions(
            self._session.session_id, date_str
        )
        filled_tickers = {p.ticker for p in open_positions}

        result = []
        for c in db_cands:
            if c["status"] != "pending":
                continue
            if c["ticker"] in filled_tickers:
                continue
            metadata = self._candidate_metadata_from_row(c)
            self._restore_day_context_from_metadata(metadata)
            metadata_orb_bar = metadata.get("orb_bar")
            orb_bar = metadata_orb_bar if isinstance(metadata_orb_bar, dict) else {}
            if not orb_bar:
                orb_bar = {
                    "timestamp": _market_open_ts(date_str).isoformat(),
                    "open": c["orb_low"],
                    "high": c["orb_high"],
                    "low": c["orb_low"],
                    "close": c["orb_high"],
                    "volume": 0,
                }
            else:
                orb_bar = {
                    "timestamp": orb_bar.get("timestamp")
                    or _market_open_ts(date_str).isoformat(),
                    "open": orb_bar.get("open", c["orb_low"]),
                    "high": orb_bar.get("high", c["orb_high"]),
                    "low": orb_bar.get("low", c["orb_low"]),
                    "close": orb_bar.get("close", c["orb_high"]),
                    "volume": orb_bar.get("volume", 0),
                }

            # Reconstruct candidate dict for breakout and timed-entry checks.
            candidate = {
                "ticker": c["ticker"],
                "direction": c["direction"],
                "orb_bar": orb_bar,
                "atr": c["atr"],
                "rvol": c["rvol"],
                "gap_pct": c["gap_pct"],
                "score": c["composite_score"],
                "size_scale": c.get("size_scale", 1.0),
            }
            for key, value in metadata.items():
                if key not in {"orb_bar", "live_day_context"}:
                    candidate[key] = value
            result.append(candidate)
        return result


# ── Engine factory ────────────────────────────────────────────────────────────

def make_orb_engine(
    session: Any,
    db_path: str | None = None,
    broker_override: Any = None,
    log_callback: Any = None,
) -> ORBTradingEngine:
    """Create an ORBTradingEngine for the given session.

    broker_override: pass a MockORBBroker (or any duck-typed broker) to avoid
    real Alpaca API calls during testing.
    """
    from apps.orb_trader.state import ORBStateManager
    from libs.intraday.domain import IntradayConfig
    from apps.intraday_bt.run import _load_config_yaml

    if broker_override is not None:
        broker = broker_override
    else:
        from apps.paper_trader.alpaca_broker import AlpacaBroker
        broker = AlpacaBroker.from_env()

    state = ORBStateManager(db_path)

    # Load strategy params from a standalone YAML config.
    raw = _load_config_yaml(Path(session.config_path)) or {}
    # Strip _meta and other non-model keys
    config_data = {k: v for k, v in raw.items() if not k.startswith("_")}
    config = IntradayConfig(**config_data)
    params = config.orb_strategy
    if params is None:
        from libs.intraday.domain import ORBStrategyParams
        params = ORBStrategyParams()

    # Store universe source on params for runtime use
    params._universe_source = config.universe.source
    params._universe_symbols_file = config.universe.symbols_file

    return ORBTradingEngine(
        session=session,
        broker=broker,
        state=state,
        params=params,
        log_callback=log_callback,
    )