fithia2/apps/paper_trader/engine.py

"""PaperTradingEngine: daily processing loop for paper trading.

Strategy decisions (WHAT to buy/sell) are made locally using backtest logic.
Order execution (HOW to execute) is done via Alpaca Paper Trading API.
"""
from __future__ import annotations

import datetime as dt
import json
import math
import time
from dataclasses import dataclass, field
from typing import Any

from libs.backtest.allocator import build_planned_order
from libs.backtest.domain import (
    BacktestConfig,
    Candidate,
    DailyPortfolioState,
    ExecutionConfig,
    OpenPosition,
    PlannedOrder,
    PositionStatus,
)
from libs.backtest.execution import simulate_exit, simulate_scheduled_open_exit, update_trailing_stop
from libs.backtest.manifests import load_manifest, resolve_config
from libs.backtest.selector import select_candidates
from libs.common.logging import get_logger

from apps.paper_trader.alpaca_broker import AlpacaBroker, AccountInfo, Order, Position
from apps.paper_trader.event_detector import EventDetector
from apps.paper_trader.state import (
    DailySnapshotRow,
    SessionRow,
    StateManager,
    StrategyStateRow,
)

logger = get_logger(__name__)

# Kill-switch threshold (matches backtest)
_KILL_SWITCH_DRAWDOWN_PCT = 25.0


@dataclass
class ReconciliationReport:
    """Result of comparing Alpaca positions vs local strategy states."""
    orphaned_alpaca: list[str] = field(default_factory=list)  # on Alpaca but not local
    ghost_local: list[str] = field(default_factory=list)      # in local but not on Alpaca
    reconciled_exits: list[str] = field(default_factory=list)  # ghost positions auto-closed
    stale_orders_cancelled: list[str] = field(default_factory=list)

    @property
    def has_issues(self) -> bool:
        return bool(self.orphaned_alpaca or self.ghost_local)


def _is_reaction_close_entry(candidate_json: str) -> bool:
    """Return True if the position was entered at the reaction-day CLOSE (MOC order).

    Same-day events (timing_class == "same_day") enter via MOC; after-close events
    enter at the next open. This distinction matters for exit checking: MOC entries
    must not have their stop checked against the entry bar's intraday low/high,
    because the position did not exist during that intraday period.
    """
    try:
        import json
        cand = json.loads(candidate_json)
        return cand.get("timing_class") == "same_day"
    except Exception:
        return False


class PaperTradingEngine:
    """Daily processing loop. Mirrors BacktestRunner._simulate_day() for live use."""

    def __init__(
        self,
        session: SessionRow,
        broker: AlpacaBroker,
        state: StateManager,
        event_detector: EventDetector,
        snapshot_store: "SnapshotStore | None" = None,
    ) -> None:
        self._session = session
        self._broker = broker
        self._state = state
        self._detector = event_detector
        self._snapshot_store = snapshot_store

        manifest = load_manifest(session.config_path)
        self._config: BacktestConfig = resolve_config(manifest)

        # Apply parking preset override stored at session creation time
        if session.parking_preset:
            self._config.risk.cash_parking_preset = session.parking_preset
            self._config.risk.apply_parking_preset()
        if session.idle_alpha_preset:
            self._config.idle_alpha_sleeve_preset = session.idle_alpha_preset
            self._config.apply_idle_alpha_sleeve_preset()
        if session.form4_sleeve_preset:
            self._config.form4_capture_sleeve_preset = session.form4_sleeve_preset
            self._config.apply_form4_capture_sleeve_preset()
        if session.ownership_sleeve_preset:
            self._config.ownership_capture_sleeve_preset = session.ownership_sleeve_preset
            self._config.apply_ownership_capture_sleeve_preset()
        if getattr(session, "risk_off_alpha_sleeve_preset", None):
            self._config.risk_off_alpha_sleeve_preset = session.risk_off_alpha_sleeve_preset
            self._config.apply_risk_off_alpha_sleeve_preset()

        # Shared attention filtering service (matches BacktestRunner)
        from libs.backtest.attention import AttentionFilterService
        oracle_url = event_detector._oracle_url if hasattr(event_detector, '_oracle_url') else ""
        self._attention_service = AttentionFilterService(
            oracle_url=oracle_url,
            scoring_model=self._config.signal.scoring_model,
        )
        self._capital_bucket_specs: dict[str, float] = {}
        get_strategy_engines = getattr(self._config, "get_strategy_engines", None)
        strategy_engines = get_strategy_engines() if callable(get_strategy_engines) else []
        for engine in strategy_engines or []:
            bucket_id = getattr(engine, "capital_bucket_id", None)
            allocation = getattr(engine, "capital_bucket_allocation_pct", None)
            if not bucket_id or allocation is None or allocation <= 0:
                continue
            self._capital_bucket_specs[bucket_id] = max(
                self._capital_bucket_specs.get(bucket_id, 0.0),
                float(allocation),
            )
        # Lookback entry: fired once per daemon session on the first run_next_open call
        self._lookback_injected: bool = False
        # Overlay shock brake cooldown (in-memory, session-scoped)
        self._parking_brake_cooldown_remaining: int = 0
        self._parking_brake_skip_buy_today: bool = False  # skip same-day re-buy after brake fires

    def _get_candidate_capital_bucket_id(self, candidate: Candidate) -> str | None:
        return candidate.engine_capital_bucket_id

    def _get_strategy_state_capital_bucket_id(self, state: StrategyStateRow) -> str | None:
        try:
            payload = json.loads(state.candidate_json)
        except Exception:
            return None
        bucket_id = payload.get("engine_capital_bucket_id") or payload.get("capital_bucket_id")
        if not bucket_id:
            return None
        return str(bucket_id)

    def _active_capital_bucket_ids_for_candidates(
        self,
        candidates: list[Candidate],
        strategy_states: dict[str, StrategyStateRow],
    ) -> set[str]:
        active_bucket_ids = {
            bucket_id
            for bucket_id in (
                self._get_candidate_capital_bucket_id(candidate)
                for candidate in candidates
            )
            if bucket_id
        }
        for state in strategy_states.values():
            bucket_id = self._get_strategy_state_capital_bucket_id(state)
            if bucket_id:
                active_bucket_ids.add(bucket_id)
        return active_bucket_ids

    def _capital_bucket_notional(
        self,
        bucket_id: str,
        alpaca_positions: list[Position],
        strategy_states: dict[str, StrategyStateRow],
    ) -> float:
        notional = 0.0
        for position in alpaca_positions:
            state = strategy_states.get(position.symbol)
            if state is None or self._get_strategy_state_capital_bucket_id(state) != bucket_id:
                continue
            notional += abs(float(position.market_value))
        return notional

    def _capital_bucket_entry_cost(
        self,
        bucket_id: str,
        alpaca_positions: list[Position],
        strategy_states: dict[str, StrategyStateRow],
    ) -> float:
        entry_cost = 0.0
        for position in alpaca_positions:
            state = strategy_states.get(position.symbol)
            if state is None or self._get_strategy_state_capital_bucket_id(state) != bucket_id:
                continue
            entry_cost += abs(float(position.avg_entry_price) * float(position.qty))
        return entry_cost

    def _capital_bucket_realized_pnl(self, session_id: str, bucket_id: str) -> float:
        realized = 0.0
        for trade in self._state.list_trades(session_id):
            if trade.get("capital_bucket_id") != bucket_id:
                continue
            realized += float(trade.get("net_pnl") or 0.0)
        return realized

    def _capital_bucket_equity(
        self,
        bucket_id: str,
        session_id: str,
        alpaca_positions: list[Position],
        strategy_states: dict[str, StrategyStateRow],
    ) -> float:
        allocation = self._capital_bucket_specs.get(bucket_id)
        if allocation is None:
            return 0.0
        initial_bucket_equity = self._session.initial_equity * allocation
        market_value = self._capital_bucket_notional(bucket_id, alpaca_positions, strategy_states)
        entry_cost = self._capital_bucket_entry_cost(bucket_id, alpaca_positions, strategy_states)
        unrealized = market_value - entry_cost
        return max(
            0.0,
            initial_bucket_equity
            + self._capital_bucket_realized_pnl(session_id, bucket_id)
            + unrealized,
        )

    def _capital_bucket_cash_available(
        self,
        bucket_id: str,
        session_id: str,
        alpaca_positions: list[Position],
        strategy_states: dict[str, StrategyStateRow],
    ) -> float:
        market_value = self._capital_bucket_notional(bucket_id, alpaca_positions, strategy_states)
        return max(
            0.0,
            self._capital_bucket_equity(bucket_id, session_id, alpaca_positions, strategy_states)
            - market_value,
        )

    def _adjust_portfolio_state_for_candidate(
        self,
        *,
        session_id: str,
        candidate: Candidate,
        portfolio_state: DailyPortfolioState,
        active_bucket_ids: set[str],
        alpaca_positions: list[Position],
        strategy_states: dict[str, StrategyStateRow],
    ) -> DailyPortfolioState:
        if not self._capital_bucket_specs or portfolio_state.cash_available <= 0:
            return portfolio_state

        configured_bucket_ids = set(self._capital_bucket_specs)
        candidate_bucket = self._get_candidate_capital_bucket_id(candidate)
        relevant_bucket_ids = configured_bucket_ids & active_bucket_ids
        if candidate_bucket and candidate_bucket in configured_bucket_ids:
            relevant_bucket_ids.add(candidate_bucket)
        if not relevant_bucket_ids:
            return portfolio_state

        bucket_cash_available = {
            bucket_id: self._capital_bucket_cash_available(
                bucket_id, session_id, alpaca_positions, strategy_states
            )
            for bucket_id in relevant_bucket_ids
        }
        bucket_equity = {
            bucket_id: self._capital_bucket_equity(
                bucket_id, session_id, alpaca_positions, strategy_states
            )
            for bucket_id in relevant_bucket_ids
        }
        sizing_equity = portfolio_state.sizing_equity or portfolio_state.equity
        if candidate_bucket and candidate_bucket in relevant_bucket_ids:
            adjusted_cash = min(
                portfolio_state.cash_available,
                bucket_cash_available[candidate_bucket],
            )
            adjusted_sizing_equity = bucket_equity[candidate_bucket]
        else:
            adjusted_cash = max(
                0.0,
                portfolio_state.cash_available - sum(bucket_cash_available.values()),
            )
            adjusted_sizing_equity = max(
                0.0,
                sizing_equity - sum(bucket_equity.values()),
            )

        if (
            math.isclose(adjusted_cash, portfolio_state.cash_available, rel_tol=0.0, abs_tol=1e-9)
            and math.isclose(adjusted_sizing_equity, sizing_equity, rel_tol=0.0, abs_tol=1e-9)
        ):
            return portfolio_state

        return portfolio_state.model_copy(
            update={
                "cash_available": adjusted_cash,
                "sizing_equity": adjusted_sizing_equity,
            }
        )

    # ------------------------------------------------------------------ #
    # Reconciliation & safety
    # ------------------------------------------------------------------ #

    def _cancel_stale_orders(self) -> list[str]:
        """Cancel all open orders. Daily system — any leftover is stale."""
        cancelled: list[str] = []
        try:
            open_orders = self._broker.list_orders("open")
            for order in open_orders:
                try:
                    self._broker.cancel_order(order.id)
                    cancelled.append(f"{order.symbol}:{order.id}")
                    logger.warning(
                        "paper_engine_stale_order_cancelled",
                        symbol=order.symbol, order_id=order.id,
                    )
                except Exception as exc:
                    logger.error(
                        "paper_engine_cancel_failed",
                        order_id=order.id, error=f"{type(exc).__name__}: {exc}",
                    )
        except Exception as exc:
            logger.error("paper_engine_list_orders_failed", error=f"{type(exc).__name__}: {exc}")
        return cancelled

    def _reconcile_positions(
        self,
        alpaca_positions: list[Position],
        strategy_states: dict[str, StrategyStateRow],
        today: dt.date,
    ) -> ReconciliationReport:
        """Compare Alpaca positions vs local state. Fix ghost positions."""
        session_id = self._session.session_id
        report = ReconciliationReport()

        alpaca_symbols = {p.symbol for p in alpaca_positions}
        local_symbols = set(strategy_states.keys())

        # Orphaned: on Alpaca but no local state (e.g. manual buy, or state save failed)
        for sym in sorted(alpaca_symbols - local_symbols):
            report.orphaned_alpaca.append(sym)
            logger.warning(
                "paper_engine_orphaned_position",
                symbol=sym,
                msg="Position on Alpaca but no local strategy state — skipping (manual intervention needed)",
            )

        # Ghost: local state but no Alpaca position (e.g. manually closed, or order never filled)
        for sym in sorted(local_symbols - alpaca_symbols):
            ss = strategy_states[sym]
            report.ghost_local.append(sym)
            logger.warning(
                "paper_engine_ghost_position",
                symbol=sym,
                entry_date=ss.entry_date,
                msg="Local state exists but no Alpaca position — auto-closing",
            )
            # Close local state and record as reconciled
            self._state.close_strategy_state(session_id, sym)
            self._state.close_trade(
                session_id=session_id,
                symbol=sym,
                engine_id=ss.engine_id,
                capital_bucket_id=self._get_strategy_state_capital_bucket_id(ss),
                entry_date=ss.entry_date,
                exit_date=today.isoformat(),
                entry_price=None,
                exit_price=0.0,
                exit_reason="RECONCILED",
                shares=0,
                net_pnl=0.0,
                r_multiple=0.0,
                holding_days=ss.days_held,
            )
            report.reconciled_exits.append(sym)

        if report.has_issues:
            logger.info(
                "paper_engine_reconciliation_summary",
                orphaned=len(report.orphaned_alpaca),
                ghost=len(report.ghost_local),
                reconciled=len(report.reconciled_exits),
            )
        return report

    def _verify_order_fill(self, order_id: str, symbol: str, timeout_sec: float = 15.0) -> tuple[Order | None, str]:
        """Poll broker to verify order fill.

        Returns (order, "") on success.
        Returns (None, "alpaca_rejected:{status}:{order_id}") if Alpaca explicitly rejects.
        Returns (None, "order_timeout:{order_id}") if fill not confirmed within timeout_sec.

        Callers must distinguish the two failure modes:
          - alpaca_rejected → record permanently in processed_events (real problem)
          - order_timeout   → do NOT record (allow retry on next run_next_open)
        """
        deadline = time.monotonic() + timeout_sec
        while time.monotonic() < deadline:
            try:
                order = self._broker.get_order(order_id)
                if order.status == "filled" and order.filled_qty > 0:
                    return order, ""
                if order.status in ("canceled", "expired", "rejected", "cancelled"):
                    reason = f"alpaca_rejected:{order.status}:{order_id}"
                    logger.warning(
                        "paper_engine_order_rejected",
                        symbol=symbol, order_id=order_id, alpaca_status=order.status,
                        hint="Check Alpaca dashboard for rejection details",
                    )
                    return None, reason
            except Exception:
                pass
            time.sleep(0.5)
        # Timeout — likely submitted outside market hours or Alpaca latency
        reason = f"order_timeout:{order_id}"
        logger.warning(
            "paper_engine_order_fill_timeout",
            symbol=symbol, order_id=order_id, timeout_sec=timeout_sec,
            hint="Order may have been submitted outside market hours — will retry on next run",
        )
        return None, reason

    @staticmethod
    def _is_market_open() -> bool:
        """Return True if US equity market is currently open (9:30–16:00 ET, weekdays)."""
        import zoneinfo
        now_et = dt.datetime.now(tz=zoneinfo.ZoneInfo("America/New_York"))
        if now_et.weekday() >= 5:  # Saturday=5, Sunday=6
            return False
        market_open = now_et.replace(hour=9, minute=30, second=0, microsecond=0)
        market_close = now_et.replace(hour=16, minute=0, second=0, microsecond=0)
        return market_open <= now_et < market_close

    def _check_kill_switch(self, drawdown_pct: float, session_st: Any) -> bool:
        """Activate kill switch if drawdown exceeds threshold. Returns True if triggered."""
        if drawdown_pct >= _KILL_SWITCH_DRAWDOWN_PCT and not session_st.kill_switch_triggered:
            session_st.kill_switch_triggered = True
            self._state.update_session_state(session_st)
            logger.critical(
                "paper_engine_kill_switch_triggered",
                drawdown_pct=round(drawdown_pct, 2),
                threshold=_KILL_SWITCH_DRAWDOWN_PCT,
            )
            return True
        return session_st.kill_switch_triggered

    # ------------------------------------------------------------------ #
    # Main entry point
    # ------------------------------------------------------------------ #

    async def run_daily(self, target_date: dt.date | None = None, force: bool = False) -> dict[str, Any]:
        """Process one trading day. Returns a summary dict for the CLI to display."""
        today = target_date or dt.date.today()
        session_id = self._session.session_id

        # 1. Idempotency: skip if already processed (unless forced)
        if not force and self._state.is_date_processed(session_id, today):
            logger.info("paper_engine_already_processed", date=today.isoformat())
            return {"date": today, "status": "already_processed"}

        # 2. Check trading day
        from libs.common.time_utils import is_trading_day
        if not is_trading_day(today):
            logger.info("paper_engine_non_trading_day", date=today.isoformat())
            return {"date": today, "status": "non_trading_day"}

        # 3. Fetch Alpaca state
        account = self._broker.get_account()
        alpaca_positions = self._broker.list_positions()
        held_symbols = [p.symbol for p in alpaca_positions]

        # 4. Load local strategy states
        strategy_states = {
            ss.symbol: ss
            for ss in self._state.get_open_strategy_states(session_id)
        }

        # 4a. Cancel stale open orders
        stale_cancelled = self._cancel_stale_orders()

        # 4b. Reconcile Alpaca vs local state
        recon = self._reconcile_positions(alpaca_positions, strategy_states, today)
        recon.stale_orders_cancelled = stale_cancelled
        # Remove ghost symbols so exit phase doesn't process them
        for sym in recon.ghost_local:
            strategy_states.pop(sym, None)

        # 4c. Kill switch check
        session_st = self._state.get_session_state(session_id)
        if session_st.kill_switch_triggered:
            logger.warning("paper_engine_kill_switch_active", date=today.isoformat())
            self._state.mark_date_processed(session_id, today)
            return {
                "date": today, "status": "kill_switch_active",
                "reconciliation": recon,
            }

        # 5. Fetch price bars for held positions (last 30 days)
        bars_by_symbol: dict[str, dict[dt.date, dict]] = {}
        if held_symbols:
            bar_start = today - dt.timedelta(days=30)
            bars_by_symbol = self._broker.get_bars_as_dict(held_symbols, bar_start, today)

        # ============================================================
        # EXIT PHASE
        # ============================================================
        exits: list[dict[str, Any]] = []
        session_st = self._state.get_session_state(session_id)
        net_pnl_today = 0.0

        for alpaca_pos in alpaca_positions:
            sym = alpaca_pos.symbol
            ss = strategy_states.get(sym)
            if ss is None:
                logger.debug("paper_engine_no_local_state", symbol=sym)
                continue

            ss.days_held += 1

            sym_bars = bars_by_symbol.get(sym, {})
            bar = sym_bars.get(today)
            if bar is None:
                logger.warning("paper_engine_no_bar", symbol=sym, date=today.isoformat())
                self._state.update_strategy_state(
                    session_id, sym, days_held=ss.days_held
                )
                continue

            # Convert to OpenPosition for backtest logic
            open_pos = self._to_open_position(alpaca_pos, ss)

            # Update trailing stop
            effective_exec = self._resolve_execution_config(ss)
            if effective_exec.trailing_model:
                update_trailing_stop(
                    open_pos,
                    bar,
                    trailing_model=effective_exec.trailing_model,
                    warmup_days=effective_exec.trailing_warmup_days,
                )
                ss.current_stop = open_pos.current_stop
                ss.peak_price = open_pos.peak_price

            # Check exit
            filled_trade = simulate_exit(open_pos, bar, effective_exec, today)
            if filled_trade is not None:
                try:
                    close_qty = None
                    if filled_trade.shares < alpaca_pos.qty:
                        close_qty = filled_trade.shares
                    self._broker.close_position(sym, qty=close_qty, fill_price=filled_trade.exit_price)
                    logger.info(
                        "paper_engine_exit",
                        symbol=sym,
                        reason=filled_trade.exit_reason.value,
                        pnl=filled_trade.net_pnl,
                    )
                except Exception as exc:
                    logger.error("paper_engine_close_failed", symbol=sym, error=str(exc))
                    continue

                self._state.close_strategy_state(session_id, sym)
                self._state.close_trade(
                    session_id=session_id,
                    symbol=sym,
                    engine_id=ss.engine_id,
                    capital_bucket_id=self._get_strategy_state_capital_bucket_id(ss),
                    entry_date=ss.entry_date,
                    exit_date=today.isoformat(),
                    entry_price=alpaca_pos.avg_entry_price,
                    exit_price=filled_trade.exit_price,
                    exit_reason=filled_trade.exit_reason.value,
                    shares=filled_trade.shares,
                    net_pnl=filled_trade.net_pnl,
                    r_multiple=filled_trade.r_multiple,
                    holding_days=ss.days_held,
                )
                net_pnl_today += filled_trade.net_pnl

                # Update consecutive losses / cooldown
                if filled_trade.net_pnl < 0:
                    session_st.consecutive_losses += 1
                    streak = self._config.risk.cooldown_after_loss_streak
                    if streak > 0 and session_st.consecutive_losses >= streak:
                        session_st.cooldown_remaining = self._config.risk.cooldown_days
                        session_st.consecutive_losses = 0
                else:
                    session_st.consecutive_losses = 0

                exits.append({
                    "symbol": sym,
                    "reason": filled_trade.exit_reason.value,
                    "pnl": filled_trade.net_pnl,
                    "r_multiple": filled_trade.r_multiple,
                    "shares": filled_trade.shares,
                    "exit_price": filled_trade.exit_price,
                })
            else:
                # No exit — persist updated trailing state
                self._state.update_strategy_state(
                    session_id,
                    sym,
                    days_held=ss.days_held,
                    current_stop=ss.current_stop,
                    peak_price=ss.peak_price,
                )

        # Decrement cooldown
        if session_st.cooldown_remaining > 0:
            session_st.cooldown_remaining -= 1

        # Reset daily risk usage
        session_st.daily_new_risk_used = 0.0

        # ============================================================
        # CASH PARKING: accrue interest, update peak, check gate (before entries)
        # ============================================================
        parking_sold_today = False
        if self._config.risk.cash_parking_enabled:
            parking_st = self._state.get_parking_state(session_id)
            # Reset sold_today flag from yesterday
            if parking_st and parking_st.get("sold_today", 0):
                self._state.update_parking_gate_state(session_id, sold_today=0)
            # Update peak price for trailing stop / top-up
            if parking_st and parking_st["symbol"] != "SGOV":
                sym = parking_st["symbol"]
                bars = self._broker.get_latest_bars([sym])
                if sym in bars:
                    cur_price = bars[sym].close
                    peak = parking_st.get("peak_price", 0) or parking_st["avg_price"]
                    if cur_price > peak:
                        self._state.update_parking_peak(session_id, cur_price)
            # Gate check and sell if signal changed
            parking_sold_today = self._parking_check_and_sell(session_id, today)

        # ============================================================
        # ENTRY PHASE
        # ============================================================
        entries: list[dict[str, Any]] = []
        rejected: list[dict[str, Any]] = []

        # Refresh account/positions after exits
        account = self._broker.get_account()
        alpaca_positions_after_exits = self._broker.list_positions()
        strategy_states_after_exits = {
            ss.symbol: ss
            for ss in self._state.get_open_strategy_states(session_id)
        }

        # Get event candidates for today — reaction_close convention only
        candidate_rows = await self._detector.get_candidates_for_date(
            today, self._config, convention="reaction_close"
        )

        # Filter already-processed events
        n_before = len(candidate_rows)
        candidate_rows = [
            r for r in candidate_rows
            if not self._state.has_processed_event(session_id, str(r.get("event_id", "")))
        ]
        if n_before != len(candidate_rows):
            logger.debug(
                "paper_engine_candidates_after_dedup",
                before=n_before,
                after=len(candidate_rows),
            )

        # Run selection pipeline for each enabled engine
        open_positions = self._to_open_positions(alpaca_positions_after_exits, strategy_states_after_exits)
        portfolio_state = self._build_portfolio_state(account, alpaca_positions_after_exits, today)

        # Fetch macro data
        macro_data = await self._fetch_macro(today)

        engine_daily_risk_used: dict[str, float] = {}
        engines = self._config.get_active_strategy_engines()
        logger.debug(
            "paper_engine_selection_input",
            date=today.isoformat(),
            candidate_rows=len(candidate_rows),
            engines=len(engines),
            symbols=[r.get("symbol") for r in candidate_rows],
        )
        if candidate_rows:
            sample = candidate_rows[0]
            logger.debug(
                "paper_engine_sample_row",
                symbol=sample.get("symbol"),
                event_type=sample.get("event_type"),
                event_direction=sample.get("event_direction"),
                filing_time_bucket=sample.get("filing_time_bucket"),
                entry_price_est=sample.get("entry_price_est"),
                avg_dollar_volume=sample.get("avg_dollar_volume"),
                avg_dollar_volume_20d=sample.get("avg_dollar_volume_20d"),
                event_close=sample.get("event_close"),
                close_location=sample.get("close_location"),
                gap_size=sample.get("gap_size"),
                reaction_day_return=sample.get("reaction_day_return"),
                market_cap_proxy=sample.get("market_cap_proxy"),
                execution_date=str(sample.get("execution_date")),
                event_timestamp=str(sample.get("event_timestamp")),
            )

        if engines:
            # Residual reserve: engines that set residual_reserve_selected=True
            # prevent later engines from picking the same event_id/symbol.
            # Matches BacktestRunner._select_candidates_for_date().
            reserved_event_ids: set[str] = {
                ss.event_id for ss in strategy_states_after_exits.values()
            }
            reserved_symbols: set[str] = {
                p.symbol for p in alpaca_positions_after_exits
                if p.symbol in strategy_states_after_exits
            }
            engine_batches: list[tuple[Any, list[Candidate]]] = []
            for engine_cfg in engines:
                prelimit = self._config.signal.max_candidates_per_day
                if self._attention_service.engine_requires_attention(engine_cfg):
                    prelimit = max(prelimit * 5, prelimit)
                engine_candidates = select_candidates(
                    raw_rows=candidate_rows,
                    universe_config=self._config.universe,
                    signal_config=self._config.signal,
                    event_type_profiles=self._config.event_type_profiles or {},
                    strategy_engine=engine_cfg,
                    truncate_to=prelimit,
                    excluded_event_ids=reserved_event_ids,
                    excluded_symbols=reserved_symbols,
                )
                # Attention filtering (matches BacktestRunner)
                engine_candidates = self._attention_service.apply_filters(
                    engine_candidates, engine_cfg, self._config.signal,
                )
                # Residual reserve for next engine
                if engine_cfg.residual_reserve_selected and engine_candidates:
                    reserved_event_ids.update(c.event_id for c in engine_candidates)
                    reserved_symbols.update(c.symbol.upper() for c in engine_candidates)
                engine_batches.append((engine_cfg, engine_candidates))

            active_bucket_ids = self._active_capital_bucket_ids_for_candidates(
                [
                    candidate
                    for _, batch_candidates in engine_batches
                    for candidate in batch_candidates
                ],
                strategy_states_after_exits,
            )
            for engine_cfg, engine_candidates in engine_batches:
                for candidate in engine_candidates:
                    engine_risk_used = engine_daily_risk_used.get(engine_cfg.engine_id, 0.0)
                    candidate_portfolio_state = self._adjust_portfolio_state_for_candidate(
                        session_id=session_id,
                        candidate=candidate,
                        portfolio_state=portfolio_state,
                        active_bucket_ids=active_bucket_ids,
                        alpaca_positions=alpaca_positions_after_exits,
                        strategy_states=strategy_states_after_exits,
                    )
                    plan = build_planned_order(
                        candidate=candidate,
                        portfolio_state=candidate_portfolio_state,
                        open_positions=open_positions,
                        config=self._config,
                        cooldown_remaining=session_st.cooldown_remaining,
                        macro_data=macro_data,
                        engine_daily_new_risk_used=engine_risk_used,
                    )
                    if plan.skip_reason == "insufficient_cash":
                        # Attempt to free parking cash before giving up
                        needed = plan.shares * float(candidate.entry_price_est) if plan.shares else float(
                            candidate.entry_price_est * 1
                        )
                        if self._parking_liquidate_for_event(session_id, today, needed):
                            account = self._broker.get_account()
                            _ap2 = self._broker.list_positions()
                            alpaca_positions_after_exits = _ap2
                            portfolio_state = self._build_portfolio_state(account, _ap2, today)
                            candidate_portfolio_state = self._adjust_portfolio_state_for_candidate(
                                session_id=session_id,
                                candidate=candidate,
                                portfolio_state=portfolio_state,
                                active_bucket_ids=active_bucket_ids,
                                alpaca_positions=_ap2,
                                strategy_states=strategy_states_after_exits,
                            )
                            plan = build_planned_order(
                                candidate=candidate,
                                portfolio_state=candidate_portfolio_state,
                                open_positions=open_positions,
                                config=self._config,
                                cooldown_remaining=session_st.cooldown_remaining,
                                macro_data=macro_data,
                                engine_daily_new_risk_used=engine_risk_used,
                            )

                    if plan.skip_reason:
                        self._state.record_processed_event(
                            session_id, candidate.event_id, today.isoformat(),
                            "rejected", skip_reason=plan.skip_reason,
                        )
                        rejected.append({
                            "symbol": candidate.symbol,
                            "event_type": candidate.event_type,
                            "score": candidate.score,
                            "reason": plan.skip_reason,
                        })
                        continue

                    # Submit market buy via Alpaca
                    if not self._is_market_open():
                        logger.warning(
                            "paper_engine_market_closed_skip_entry",
                            symbol=candidate.symbol,
                            hint="Market is closed — skipping without recording so retry fires next run",
                        )
                        rejected.append({
                            "symbol": candidate.symbol,
                            "event_type": candidate.event_type,
                            "score": candidate.score,
                            "reason": "market_closed",
                        })
                        continue
                    try:
                        order = self._broker.submit_market_buy(candidate.symbol, plan.shares)
                        logger.info(
                            "paper_engine_buy_submitted",
                            symbol=candidate.symbol,
                            qty=plan.shares,
                            order_id=order.id,
                        )
                    except Exception as exc:
                        logger.error(
                            "paper_engine_buy_failed",
                            symbol=candidate.symbol,
                            error=str(exc),
                        )
                        self._state.record_processed_event(
                            session_id, candidate.event_id, today.isoformat(),
                            "rejected", skip_reason=f"order_failed:{exc}",
                        )
                        rejected.append({
                            "symbol": candidate.symbol,
                            "event_type": candidate.event_type,
                            "score": candidate.score,
                            "reason": f"order_failed:{exc}",
                        })
                        continue

                    # Verify fill
                    verified, fill_fail_reason = self._verify_order_fill(order.id, candidate.symbol)
                    if verified is None:
                        if fill_fail_reason.startswith("alpaca_rejected:"):
                            # Alpaca explicitly rejected — record permanently
                            self._state.record_processed_event(
                                session_id, candidate.event_id, today.isoformat(),
                                "rejected", skip_reason=fill_fail_reason,
                            )
                        else:
                            # Timeout — likely market closed or transient; do NOT record so retry works
                            logger.warning(
                                "paper_engine_fill_timeout_not_recorded",
                                symbol=candidate.symbol, order_id=order.id,
                            )
                        rejected.append({
                            "symbol": candidate.symbol,
                            "event_type": candidate.event_type,
                            "score": candidate.score,
                            "reason": fill_fail_reason,
                        })
                        continue
                    fill_price = verified.filled_avg_price or plan.entry_price_limit

                    # Save local strategy state
                    self._state.save_strategy_state(
                        session_id,
                        StrategyStateRow(
                            session_id=session_id,
                            symbol=candidate.symbol,
                            event_id=candidate.event_id,
                            engine_id=candidate.engine_id,
                            order_id=order.id,
                            entry_date=today.isoformat(),
                            stop_price=plan.stop_price,
                            target_price=plan.target_price,
                            current_stop=plan.stop_price,
                            peak_price=fill_price,
                            days_held=0,
                            trade_direction=candidate.trade_direction,
                            candidate_json=candidate.model_dump_json(),
                            plan_json=plan.model_dump_json(),
                            status="open",
                        ),
                    )
                    self._state.open_trade(
                        session_id=session_id,
                        symbol=candidate.symbol,
                        engine_id=candidate.engine_id,
                        capital_bucket_id=candidate.engine_capital_bucket_id,
                        entry_date=today.isoformat(),
                        entry_price=fill_price,
                        shares=plan.shares,
                    )
                    self._state.record_processed_event(
                        session_id, candidate.event_id, today.isoformat(), "entered",
                    )

                    trade_risk_state = candidate_portfolio_state.sizing_equity or candidate_portfolio_state.equity
                    trade_risk = trade_risk_state * (
                        candidate.engine_per_trade_risk_pct
                        or self._config.risk.per_trade_risk_pct
                    )
                    engine_daily_risk_used[engine_cfg.engine_id] = engine_risk_used + trade_risk
                    session_st.daily_new_risk_used += trade_risk

                    # Refresh portfolio state after each entry
                    alpaca_positions_after_exits = self._broker.list_positions()
                    strategy_states_after_exits = {
                        ss.symbol: ss
                        for ss in self._state.get_open_strategy_states(session_id)
                    }
                    open_positions = self._to_open_positions(
                        alpaca_positions_after_exits, strategy_states_after_exits
                    )
                    # Add the new virtual position to open_positions for gate checks
                    new_open = self._virtual_open_position(candidate, plan, today)
                    open_positions.append(new_open)
                    portfolio_state = DailyPortfolioState(
                        date=portfolio_state.date,
                        equity=portfolio_state.equity,
                        sizing_equity=portfolio_state.sizing_equity,
                        cash_available=max(
                            0.0,
                            portfolio_state.cash_available - plan.entry_price_limit * plan.shares,
                        ),
                        gross_exposure=portfolio_state.gross_exposure + plan.entry_price_limit * plan.shares,
                        net_exposure=portfolio_state.net_exposure + plan.entry_price_limit * plan.shares,
                        reserved_risk_budget=portfolio_state.reserved_risk_budget,
                        unrealized_pnl=portfolio_state.unrealized_pnl,
                        realized_pnl=portfolio_state.realized_pnl,
                        open_positions=[p.position_id for p in open_positions],
                        daily_new_risk_used=session_st.daily_new_risk_used,
                        peak_equity=portfolio_state.peak_equity,
                        current_drawdown_pct=portfolio_state.current_drawdown_pct,
                    )

                    entries.append({
                        "symbol": candidate.symbol,
                        "event_type": candidate.event_type,
                        "score": candidate.score,
                        "shares": plan.shares,
                        "entry_price": plan.entry_price_limit,
                        "stop": plan.stop_price,
                        "target": plan.target_price,
                        "order_id": order.id,
                    })

        else:
            # No engines defined — use flat candidate selection
            all_candidates = select_candidates(
                raw_rows=candidate_rows,
                universe_config=self._config.universe,
                signal_config=self._config.signal,
                event_type_profiles=self._config.event_type_profiles or {},
                excluded_event_ids={ss.event_id for ss in strategy_states_after_exits.values()},
                excluded_symbols={p.symbol for p in alpaca_positions_after_exits if p.symbol in strategy_states_after_exits},
            )
            active_bucket_ids = self._active_capital_bucket_ids_for_candidates(
                list(all_candidates),
                strategy_states_after_exits,
            )
            for candidate in all_candidates:
                candidate_portfolio_state = self._adjust_portfolio_state_for_candidate(
                    session_id=session_id,
                    candidate=candidate,
                    portfolio_state=portfolio_state,
                    active_bucket_ids=active_bucket_ids,
                    alpaca_positions=alpaca_positions_after_exits,
                    strategy_states=strategy_states_after_exits,
                )
                plan = build_planned_order(
                    candidate=candidate,
                    portfolio_state=candidate_portfolio_state,
                    open_positions=open_positions,
                    config=self._config,
                    cooldown_remaining=session_st.cooldown_remaining,
                    macro_data=macro_data,
                )
                if plan.skip_reason:
                    self._state.record_processed_event(
                        session_id, candidate.event_id, today.isoformat(),
                        "rejected", skip_reason=plan.skip_reason,
                    )
                    rejected.append({
                        "symbol": candidate.symbol,
                        "event_type": candidate.event_type,
                        "score": candidate.score,
                        "reason": plan.skip_reason,
                    })
                    continue

                if not self._is_market_open():
                    logger.warning(
                        "paper_engine_market_closed_skip_entry",
                        symbol=candidate.symbol,
                        hint="Market is closed — skipping without recording so retry fires next run",
                    )
                    rejected.append({
                        "symbol": candidate.symbol,
                        "event_type": candidate.event_type,
                        "score": candidate.score,
                        "reason": "market_closed",
                    })
                    continue
                try:
                    order = self._broker.submit_market_buy(candidate.symbol, plan.shares)
                except Exception as exc:
                    self._state.record_processed_event(
                        session_id, candidate.event_id, today.isoformat(),
                        "rejected", skip_reason=f"order_failed:{exc}",
                    )
                    rejected.append({
                        "symbol": candidate.symbol,
                        "event_type": candidate.event_type,
                        "score": candidate.score,
                        "reason": f"order_failed:{exc}",
                    })
                    continue

                verified, fill_fail_reason = self._verify_order_fill(order.id, candidate.symbol)
                if verified is None:
                    if fill_fail_reason.startswith("alpaca_rejected:"):
                        self._state.record_processed_event(
                            session_id, candidate.event_id, today.isoformat(),
                            "rejected", skip_reason=fill_fail_reason,
                        )
                    else:
                        logger.warning(
                            "paper_engine_fill_timeout_not_recorded",
                            symbol=candidate.symbol, order_id=order.id,
                        )
                    rejected.append({
                        "symbol": candidate.symbol,
                        "event_type": candidate.event_type,
                        "score": candidate.score,
                        "reason": fill_fail_reason,
                    })
                    continue
                fill_price = verified.filled_avg_price or plan.entry_price_limit

                self._state.save_strategy_state(
                    session_id,
                    StrategyStateRow(
                        session_id=session_id,
                        symbol=candidate.symbol,
                        event_id=candidate.event_id,
                        engine_id=candidate.engine_id,
                        order_id=order.id,
                        entry_date=today.isoformat(),
                        stop_price=plan.stop_price,
                        target_price=plan.target_price,
                        current_stop=plan.stop_price,
                        peak_price=fill_price,
                        days_held=0,
                        trade_direction=candidate.trade_direction,
                        candidate_json=candidate.model_dump_json(),
                        plan_json=plan.model_dump_json(),
                        status="open",
                    ),
                )
                self._state.open_trade(
                    session_id=session_id,
                    symbol=candidate.symbol,
                    engine_id=candidate.engine_id,
                    capital_bucket_id=candidate.engine_capital_bucket_id,
                    entry_date=today.isoformat(),
                    entry_price=fill_price,
                    shares=plan.shares,
                )
                self._state.record_processed_event(
                    session_id, candidate.event_id, today.isoformat(), "entered",
                )
                entries.append({
                    "symbol": candidate.symbol,
                    "event_type": candidate.event_type,
                    "score": candidate.score,
                    "shares": plan.shares,
                    "entry_price": plan.entry_price_limit,
                    "stop": plan.stop_price,
                    "target": plan.target_price,
                    "order_id": order.id,
                })

        # ============================================================
        # CASH PARKING: buy with idle cash (after all entries)
        # ============================================================
        if self._config.risk.cash_parking_enabled and not parking_sold_today:
            self._parking_buy(session_id, today)

        summary = self._finalize_day(today, session_st, exits, entries, rejected, len(candidate_rows))
        summary["reconciliation"] = recon
        self._state.mark_date_processed(session_id, today)
        return summary

    # ------------------------------------------------------------------ #
    # Cash Parking
    # ------------------------------------------------------------------ #

    def _parking_evaluate_gate(self, today: dt.date) -> str:
        """Evaluate parking gate and return target symbol or 'sgov'.

        Uses broker price data to compute all gate signals, matching
        backtester's _evaluate_parking_target() logic including temperature,
        entropy, VRP, Hurst, autocorrelation, and kurtosis.
        """
        import math

        risk = self._config.risk
        gate_mode = risk.cash_parking_gate_mode
        # Actual parking symbol (qqqm, qqq, spy, etc.)
        actual_sym = risk.cash_parking_symbol
        # Gate signal source (always qqq or spy for indicators)
        gate_sym = actual_sym if actual_sym in ("spy", "qqq") else "qqq"

        # Fetch historical bars for gate calculation
        lookback = max(risk.cash_parking_gate_vol_lookback + 5, 80)
        start = today - dt.timedelta(days=lookback * 2)
        bars_dict = self._broker.get_bars_as_dict([gate_sym.upper()], start, today)
        bars = bars_dict.get(gate_sym.upper(), {})
        if len(bars) < 20:
            return "sgov"  # insufficient data → safe default

        sorted_dates = sorted(bars.keys())
        closes = [bars[d]["close"] for d in sorted_dates]
        n = len(closes)

        # --- Compute indicators ---
        # Volatility (multiple lookbacks)
        def _compute_vol(lb: int) -> float | None:
            if n < lb + 1:
                return None
            log_rets = []
            for i in range(n - lb, n):
                if closes[i - 1] > 0:
                    log_rets.append(math.log(closes[i] / closes[i - 1]))
            if not log_rets:
                return None
            mean_r = sum(log_rets) / len(log_rets)
            var_r = sum((r - mean_r) ** 2 for r in log_rets) / len(log_rets)
            return math.sqrt(var_r * 252)

        vol = _compute_vol(risk.cash_parking_gate_vol_lookback)
        vol_15 = _compute_vol(15)
        vol_50 = _compute_vol(50)

        # Momentum
        def _compute_mom(period: int) -> float | None:
            if n <= period or closes[-1 - period] <= 0:
                return None
            return (closes[-1] - closes[-1 - period]) / closes[-1 - period]

        # Entropy
        def _compute_entropy(lb: int) -> float | None:
            if n < lb + 1:
                return None
            daily_rets = []
            for i in range(n - lb, n):
                if closes[i - 1] > 0:
                    daily_rets.append(closes[i] / closes[i - 1] - 1)
            if len(daily_rets) < lb - 1:
                return None
            n_pos = sum(1 for r in daily_rets if r > 0.001)
            n_neg = sum(1 for r in daily_rets if r < -0.001)
            n_flat = len(daily_rets) - n_pos - n_neg
            n_total = len(daily_rets)
            entropy = 0.0
            for count in (n_pos, n_neg, n_flat):
                if count > 0:
                    p = count / n_total
                    entropy -= p * math.log2(p)
            return entropy

        # Autocorrelation (lag-1)
        def _compute_autocorr(lb: int) -> float | None:
            if n < lb + 2:
                return None
            rets = []
            for i in range(n - lb - 1, n):
                if closes[i - 1] > 0:
                    rets.append(closes[i] / closes[i - 1] - 1)
            if len(rets) < lb:
                return None
            x, y = rets[:-1], rets[1:]
            n_ac = len(x)
            mx = sum(x) / n_ac
            my = sum(y) / n_ac
            cov = sum((x[k] - mx) * (y[k] - my) for k in range(n_ac)) / n_ac
            sx = (sum((x[k] - mx) ** 2 for k in range(n_ac)) / n_ac) ** 0.5
            sy = (sum((y[k] - my) ** 2 for k in range(n_ac)) / n_ac) ** 0.5
            if sx < 1e-12 or sy < 1e-12:
                return None
            return cov / (sx * sy)

        # --- Determine SGOV state from parking_state ---
        parking_st = self._state.get_parking_state(self._session.session_id)
        in_sgov = parking_st is not None and parking_st["symbol"] == "SGOV"

        # --- Gate evaluation (volatility mode) ---
        if gate_mode == "volatility":
            # Vol gate
            if vol is not None and vol >= risk.cash_parking_gate_vol_threshold:
                return "sgov"

            # Entropy check
            ent_thr = risk.cash_parking_entropy_threshold
            if ent_thr > 0:
                ent_lb = risk.cash_parking_entropy_lookback
                entropy = _compute_entropy(ent_lb)
                if entropy is not None and entropy > ent_thr and not in_sgov:
                    return "sgov"
                if in_sgov and not risk.cash_parking_require_trend:
                    if entropy is not None and entropy <= ent_thr * 0.8:
                        pass  # allow recovery
                    elif entropy is not None:
                        return "sgov"

            # VRP check
            vrp_thr = risk.cash_parking_vrp_threshold
            if vrp_thr > 0 and vol is not None:
                # Try to get VIX from broker or skip
                try:
                    vix_bars = self._broker.get_bars_as_dict(["VIXY"], today - dt.timedelta(days=5), today)
                    # Fallback: estimate VRP from vol ratio if VIX unavailable
                except Exception:
                    pass  # VRP check skipped in live (VIX not easily available)

            # Temperature check
            temp_thr = risk.cash_parking_temperature_threshold
            if temp_thr > 0 and vol_15 is not None and vol_50 is not None and vol_50 > 0:
                temp = vol_15 / vol_50
                if temp > temp_thr and not in_sgov:
                    return "sgov"
                if in_sgov and not risk.cash_parking_require_trend:
                    if temp <= temp_thr * 0.7:
                        pass  # allow recovery
                    else:
                        return "sgov"

            # Autocorrelation check
            ac_thr = risk.cash_parking_autocorr_threshold
            if ac_thr > -99:
                ac = _compute_autocorr(20)
                if ac is not None and ac < ac_thr and not in_sgov:
                    return "sgov"
                if in_sgov and not risk.cash_parking_require_trend:
                    if ac is not None and ac >= ac_thr + 0.1:
                        pass
                    elif ac is not None:
                        return "sgov"

            # Momentum trend check (asymmetric re-entry)
            if risk.cash_parking_require_trend and risk.cash_parking_trend_mode == "momentum":
                period = risk.cash_parking_trend_sma_period
                mom = _compute_mom(period)
                reentry_pct = risk.cash_parking_trend_reentry_pct

                if in_sgov:
                    mom_ok = mom is not None and mom > reentry_pct
                    if mom_ok:
                        pass  # allow recovery
                    else:
                        return "sgov"
                else:
                    if mom is not None and mom <= 0:
                        return "sgov"
                    # Entropy check within momentum (for vme presets)
                    if ent_thr > 0:
                        ent_lb = risk.cash_parking_entropy_lookback
                        entropy = _compute_entropy(ent_lb)
                        if entropy is not None and entropy > ent_thr:
                            return "sgov"

        return actual_sym

    def _parking_check_shock_brake(self, today: dt.date) -> bool:
        """Check QQQ/SMH acceleration signals for fast overlay exit.

        Returns True if any brake signal fires.
        """
        import math

        risk = self._config.risk
        rv_ratio = risk.cash_parking_overlay_shock_brake_rv_ratio
        dd5_pct = risk.cash_parking_overlay_shock_brake_dd5_pct
        sma_cross = risk.cash_parking_overlay_shock_brake_sma_cross

        lookback = 55  # enough for vol_20 + buffer
        start = today - dt.timedelta(days=lookback * 2)
        qqq_bars_dict = self._broker.get_bars_as_dict(["QQQ"], start, today)
        qqq_bars = qqq_bars_dict.get("QQQ", {})
        if len(qqq_bars) < 22:
            return False

        sorted_dates = sorted(qqq_bars.keys())
        qqq_closes = [qqq_bars[d]["close"] for d in sorted_dates]
        n = len(qqq_closes)

        def _vol(lb: int) -> float | None:
            if n < lb + 1:
                return None
            log_rets = []
            for i in range(n - lb, n):
                if qqq_closes[i - 1] > 0:
                    log_rets.append(math.log(qqq_closes[i] / qqq_closes[i - 1]))
            if not log_rets:
                return None
            mean_r = sum(log_rets) / len(log_rets)
            var_r = sum((r - mean_r) ** 2 for r in log_rets) / len(log_rets)
            return math.sqrt(var_r * 252)

        # Signal 1: vol acceleration
        vol5 = _vol(5)
        vol20 = _vol(20)
        if vol5 is not None and vol20 is not None and vol20 > 0:
            if vol5 / vol20 > rv_ratio:
                return True

        # Signal 2: trend break + sector weakness
        if sma_cross and n >= 11:
            qqq_close_now = qqq_closes[-1]
            qqq_sma10 = sum(qqq_closes[-10:]) / 10
            if qqq_close_now < qqq_sma10:
                # Check SMH mom_5
                smh_bars_dict = self._broker.get_bars_as_dict(["SMH"], start, today)
                smh_bars = smh_bars_dict.get("SMH", {})
                if len(smh_bars) >= 7:
                    smh_sorted = sorted(smh_bars.keys())
                    smh_closes = [smh_bars[d]["close"] for d in smh_sorted]
                    if len(smh_closes) >= 6 and smh_closes[-6] > 0:
                        smh_mom5 = (smh_closes[-1] - smh_closes[-6]) / smh_closes[-6]
                        if smh_mom5 < 0:
                            return True

        # Signal 3: sharp 5-day drawdown
        if n >= 6:
            qqq_high5 = max(qqq_closes[-5:])
            qqq_close_now = qqq_closes[-1]
            if qqq_high5 > 0 and (qqq_high5 - qqq_close_now) / qqq_high5 > dd5_pct:
                return True

        # Signal 4: near-SMA buffer — exit when QQQ within sma_buffer% ABOVE SMA10 (pre-emptive)
        # Requires vol5/vol20 ∈ (0.25, rv_ratio_upper) — "barely elevated" pre-crash signature.
        # Upper bound filters out high-vol days (regular gate handles those) and false alarms.
        sma_buffer = getattr(risk, "cash_parking_overlay_shock_brake_sma_buffer", 0.0)
        if sma_buffer > 0 and n >= 11:
            qqq_close_now = qqq_closes[-1]
            qqq_sma10_pt = sum(qqq_closes[-10:]) / 10
            vol5_pt = _vol(5)
            vol20_pt = _vol(20)
            if (
                qqq_sma10_pt > 0
                and vol5_pt is not None and vol20_pt is not None and vol20_pt > 0
            ):
                rv = vol5_pt / vol20_pt
                rv_upper = getattr(risk, "cash_parking_overlay_shock_brake_rv_ratio_upper", 0.0)
                vol_ok = rv > 0.25 and (rv_upper <= 0 or rv < rv_upper)
                if vol_ok:
                    sma_gap = (qqq_close_now - qqq_sma10_pt) / qqq_sma10_pt
                    if 0 < sma_gap < sma_buffer:
                        return True

        return False

    def _parking_check_and_sell(self, session_id: str, today: dt.date) -> bool:
        """Check gate signal, apply confirmation, trailing stop. Returns True if sold."""
        import math

        parking_st = self._state.get_parking_state(session_id)
        if parking_st is None:
            return False

        current_sym = parking_st["symbol"].lower()
        risk = self._config.risk
        target = self._parking_evaluate_gate(today)

        # --- Decrement overlay brake cooldown ---
        brake_cooldown = int(parking_st.get("overlay_brake_cooldown", 0))
        if brake_cooldown > 0:
            brake_cooldown -= 1
            self._state.update_parking_gate_state(session_id, overlay_brake_cooldown=brake_cooldown)

        # --- Shock brake check ---
        overlay_sym = (risk.cash_parking_low_vol_overlay_symbol or "").lower()
        if (
            overlay_sym
            and risk.cash_parking_overlay_shock_brake_enabled
            and current_sym == overlay_sym
        ):
            brake_fired = self._parking_check_shock_brake(today)
            if brake_fired:
                logger.info("parking_shock_brake", symbol=current_sym, date=str(today))
                # Execute sell immediately (bypass confirmation)
                sym = parking_st["symbol"]
                qty = parking_st["qty"]
                try:
                    if qty > 0:
                        self._broker.close_position(sym, qty=qty)
                        import time
                        time.sleep(1)
                    self._state.close_parking_state(session_id)
                    bars_now = self._broker.get_latest_bars([sym])
                    exit_price = bars_now[sym].close if sym in bars_now else parking_st["avg_price"]
                    self._state.record_trade(
                        session_id=session_id,
                        symbol=sym,
                        engine_id=None,
                        capital_bucket_id=None,
                        entry_date=parking_st["entry_date"],
                        exit_date=today.isoformat(),
                        entry_price=parking_st["avg_price"],
                        exit_price=exit_price,
                        exit_reason="PARKING_SHOCK_BRAKE",
                        shares=qty,
                        net_pnl=(exit_price - parking_st["avg_price"]) * qty,
                        r_multiple=0.0,
                        holding_days=(today - dt.date.fromisoformat(parking_st["entry_date"])).days,
                    )
                except Exception as e:
                    logger.warning("parking_shock_brake_sell_failed", error=str(e))
                # Set cooldown for next active parking state (will be created on re-buy)
                # We store cooldown on a session-level attribute for now
                self._parking_brake_cooldown_remaining = risk.cash_parking_overlay_shock_brake_cooldown_days
                self._parking_brake_skip_buy_today = True  # skip same-day re-buy
                return True

        # --- Dwell cap check ---
        overlay_hold_days = int(parking_st.get("overlay_hold_days", 0))
        if overlay_sym and current_sym == overlay_sym:
            overlay_hold_days += 1
            self._state.update_parking_gate_state(session_id, overlay_hold_days=overlay_hold_days)

            max_hold = risk.cash_parking_overlay_max_hold_days
            if max_hold > 0 and overlay_hold_days >= max_hold:
                logger.info("parking_dwell_cap", symbol=current_sym, hold_days=overlay_hold_days)
                sym = parking_st["symbol"]
                qty = parking_st["qty"]
                try:
                    if qty > 0:
                        self._broker.close_position(sym, qty=qty)
                        import time
                        time.sleep(1)
                    self._state.close_parking_state(session_id)
                    bars_now = self._broker.get_latest_bars([sym])
                    exit_price = bars_now[sym].close if sym in bars_now else parking_st["avg_price"]
                    self._state.record_trade(
                        session_id=session_id,
                        symbol=sym,
                        engine_id=None,
                        capital_bucket_id=None,
                        entry_date=parking_st["entry_date"],
                        exit_date=today.isoformat(),
                        entry_price=parking_st["avg_price"],
                        exit_price=exit_price,
                        exit_reason="PARKING_DWELL_CAP",
                        shares=qty,
                        net_pnl=(exit_price - parking_st["avg_price"]) * qty,
                        r_multiple=0.0,
                        holding_days=(today - dt.date.fromisoformat(parking_st["entry_date"])).days,
                    )
                except Exception as e:
                    logger.warning("parking_dwell_cap_sell_failed", error=str(e))
                return True

        # --- Trailing stop check ---
        stop_pct = risk.cash_parking_stop_pct
        if stop_pct > 0 and current_sym != "sgov":
            peak = parking_st.get("peak_price", 0) or parking_st["avg_price"]
            bars = self._broker.get_latest_bars([parking_st["symbol"]])
            if parking_st["symbol"] in bars:
                cur_price = bars[parking_st["symbol"]].close
                if peak > 0 and cur_price < peak * (1 - stop_pct):
                    logger.info("parking_trailing_stop", symbol=parking_st["symbol"],
                                peak=round(peak, 2), current=round(cur_price, 2))
                    target = "sgov"  # force exit

        # --- Target confirmation (2-day) to prevent whipsaw ---
        committed = parking_st.get("committed_target", current_sym)
        pending = parking_st.get("pending_target", "")
        pending_days = parking_st.get("pending_days", 0)

        if target == committed:
            # Signal agrees with committed → reset pending
            if pending:
                self._state.update_parking_gate_state(
                    session_id, pending_target="", pending_days=0)
            # No change needed
            if target == current_sym:
                return False
        else:
            # Signal disagrees with committed → accumulate pending
            if target == pending:
                pending_days += 1
            else:
                pending = target
                pending_days = 1
            self._state.update_parking_gate_state(
                session_id, pending_target=pending, pending_days=pending_days)

            if pending_days < 2:
                return False  # not confirmed yet, hold current
            # Confirmed after 2 days — commit and execute
            committed = target
            self._state.update_parking_gate_state(
                session_id, committed_target=committed, pending_target="", pending_days=0)
            if committed == current_sym:
                return False  # same symbol after confirmation

        # --- Execute sell ---
        sym = parking_st["symbol"]
        qty = parking_st["qty"]
        logger.info("parking_sell", symbol=sym, qty=qty, reason=f"gate→{target}")
        try:
            if qty > 0:
                self._broker.close_position(sym, qty=qty)
                time.sleep(1)
            self._state.close_parking_state(session_id)

            # Record trade
            bars = self._broker.get_latest_bars([sym])
            exit_price = bars[sym].close if sym in bars else parking_st["avg_price"]
            self._state.record_trade(
                session_id=session_id,
                symbol=sym,
                engine_id=None,
                capital_bucket_id=None,
                entry_date=parking_st["entry_date"],
                exit_date=today.isoformat(),
                entry_price=parking_st["avg_price"],
                exit_price=exit_price,
                exit_reason="PARKING",
                shares=qty,
                net_pnl=(exit_price - parking_st["avg_price"]) * qty,
                r_multiple=0.0,
                holding_days=(today - dt.date.fromisoformat(parking_st["entry_date"])).days,
            )
            # Mark sold today
            # Note: parking_state is now closed, so we track sold_today via instance var
        except Exception as e:
            logger.warning("parking_sell_failed", error=str(e))
        return True

    def _parking_liquidate_for_event(
        self, session_id: str, today: dt.date, needed: float
    ) -> bool:
        """Release parking cash to fund an event entry that has insufficient cash.

        Mirrors BacktestRunner._liquidate_parking_for_cash().
        Sells shares via broker; waits 1 s for fill.
        Returns True if any cash was freed.
        """
        parking_st = self._state.get_parking_state(session_id)
        if parking_st is None:
            return False

        sym = parking_st["symbol"]

        # Broker position (SGOV / QQQM / QQQ / SPY)
        qty = parking_st.get("qty", 0)
        if qty <= 0:
            return False
        bars = self._broker.get_latest_bars([sym])
        cur_price = bars[sym].close if sym in bars else float(parking_st.get("avg_price", 0))
        if cur_price <= 0:
            return False
        shares_to_sell = min(qty, max(1, math.ceil(needed / cur_price)))
        try:
            self._broker.close_position(sym, qty=shares_to_sell)
            time.sleep(3)
            new_qty = qty - shares_to_sell
            if new_qty <= 0:
                self._state.close_parking_state(session_id)
            else:
                avg = float(parking_st.get("avg_price", cur_price))
                self._state.save_parking_state(
                    session_id, sym,
                    parking_st.get("entry_date", today),
                    new_qty, avg, new_qty * avg,
                    peak_price=float(parking_st.get("peak_price", 0) or avg),
                    gate_in_sgov=int(parking_st.get("gate_in_sgov", 0)),
                    sgov_entry_value=0.0,
                )
            logger.info(
                "parking_partial_sell_for_event",
                symbol=sym, shares_sold=shares_to_sell, remaining_qty=max(0, new_qty),
            )
            return True
        except Exception as exc:
            logger.warning("parking_sell_for_event_failed", error=str(exc))
            return False

    def _parking_position_value(
        self, session_id: str, alpaca_positions: "list[Any]"
    ) -> "tuple[float, float]":
        """Return (parking_mv, parking_unrealized_pnl) for this session's parking.

        Uses the already-fetched alpaca_positions to avoid an extra API call.
        Each session tracks its own qty, so two sessions parking in the same
        symbol (e.g. QQQ) get correctly separated market values.
        """
        parking_st = self._state.get_parking_state(session_id)
        if parking_st is None:
            return 0.0, 0.0
        sym = parking_st["symbol"]
        # Broker position (SGOV / QQQ / SPY / TQQQ / QQQM …)
        qty = parking_st.get("qty", 0)
        avg = float(parking_st.get("avg_price", 0))
        if qty <= 0:
            return 0.0, 0.0
        cur_price = avg  # fallback to avg if not found in positions
        for p in alpaca_positions:
            if p.symbol == sym:
                cur_price = float(p.current_price)
                break
        return cur_price * qty, (cur_price - avg) * qty

    def _session_cash(self, session_id: str) -> tuple[float, float]:
        """Return (session_equity, session_cash) derived from local state.

        Avoids using the shared Alpaca account directly so multiple sessions
        on the same broker account each see their own isolated balance.
        Includes parking positions so available cash is correctly reduced.
        """
        from apps.paper_trader.mock_broker import MockBroker
        if isinstance(self._broker, MockBroker):
            acct = self._broker.get_account()
            return float(acct.equity), float(acct.cash)

        # Derive from session snapshots + open positions
        session_symbols = {
            ss.symbol for ss in self._state.get_open_strategy_states(session_id)
        }
        alpaca_positions = self._broker.list_positions()
        session_mv = sum(p.market_value for p in alpaca_positions if p.symbol in session_symbols)
        session_unreal = sum(p.unrealized_pl for p in alpaca_positions if p.symbol in session_symbols)

        # Include parking position so cash isn't over-stated
        parking_mv, _ = self._parking_position_value(session_id, alpaca_positions)
        session_mv += parking_mv

        snapshots = self._state.list_snapshots(session_id)
        if snapshots:
            session_equity = snapshots[-1]["equity"]
        else:
            total_realized = sum(
                t.get("net_pnl", 0.0) for t in self._state.list_trades(session_id)
            )
            session_equity = self._session.initial_equity + total_realized + session_unreal
        session_cash = max(0.0, session_equity - session_mv)
        return session_equity, session_cash

    def _parking_buy(self, session_id: str, today: dt.date) -> None:
        """Buy or top-up parking with idle cash after all entries are done."""
        risk = self._config.risk
        parking_st = self._state.get_parking_state(session_id)

        # --- Top-up existing parking ---
        if parking_st is not None:
            sym = parking_st["symbol"]
            # Check top-up conditions
            topup_dd = getattr(risk, "cash_parking_topup_max_peak_drawdown_pct", 0)
            if topup_dd > 0:
                peak = parking_st.get("peak_price", 0) or parking_st["avg_price"]
                bars = self._broker.get_latest_bars([sym])
                if sym in bars and peak > 0:
                    cur_price = bars[sym].close
                    dd_from_peak = (peak - cur_price) / peak
                    if dd_from_peak > topup_dd:
                        logger.info("parking_topup_blocked",
                                    symbol=sym, dd=f"{dd_from_peak:.2%}", threshold=f"{topup_dd:.2%}")
                        # Send blocked cash to SGOV instead
                        session_equity, session_cash = self._session_cash(session_id)
                        reserve = session_equity * risk.cash_parking_reserve_pct
                        investable = max(0.0, session_cash - reserve)
                        if investable > 100:
                            self._state.save_parking_state(
                                session_id, "SGOV", today, 1, investable, investable,
                                gate_in_sgov=0, sgov_entry_value=investable,
                            )
                        return
            # Allow top-up (no blocking condition met)
            return  # For now, no actual top-up execution (matches backtester behavior of holding)

        # --- New parking position ---
        # Skip same-day re-buy if brake fired today; next day gate re-evaluates fresh.
        if self._parking_brake_skip_buy_today:
            self._parking_brake_skip_buy_today = False
            return

        target = self._parking_evaluate_gate(today)

        # Overlay brake cooldown: suppress TQQQ overlay re-entry during cooldown
        overlay_sym_cfg = (risk.cash_parking_low_vol_overlay_symbol or "").lower()
        if (
            overlay_sym_cfg
            and target.lower() == overlay_sym_cfg
            and self._parking_brake_cooldown_remaining > 0
        ):
            self._parking_brake_cooldown_remaining -= 1
            # Fall back to base park symbol
            target = risk.cash_parking_symbol
            logger.info("parking_overlay_cooldown_active", remaining=self._parking_brake_cooldown_remaining)

        # SGOV/QQQ/SPY/QQQM/TQQQ: buy through broker
        sym = target.upper()
        session_equity, session_cash = self._session_cash(session_id)
        reserve = session_equity * risk.cash_parking_reserve_pct
        investable = max(0.0, session_cash - reserve)

        bars = self._broker.get_latest_bars([sym])
        if sym not in bars:
            return
        price = bars[sym].close
        if price <= 0 or investable < price:
            return

        qty = int(investable / price)
        if qty <= 0:
            return

        logger.info("parking_buy", symbol=sym, qty=qty, price=round(price, 2))
        try:
            order = self._broker.submit_market_buy(sym, qty)
            for _ in range(5):
                time.sleep(1)
                filled = self._broker.get_order(order.id)
                if filled and filled.filled_avg_price:
                    avg_price = filled.filled_avg_price
                    self._state.save_parking_state(
                        session_id, sym, today, qty, avg_price, avg_price * qty,
                        peak_price=avg_price, gate_in_sgov=1 if target == "sgov" else 0,
                        committed_target=target,
                    )
                    logger.info("parking_filled", symbol=sym, qty=qty, price=round(avg_price, 2))
                    return
            logger.warning("parking_buy_timeout", symbol=sym, order_id=order.id)
        except Exception as e:
            logger.warning("parking_buy_failed", error=str(e))

    # ------------------------------------------------------------------ #
    # Phased execution: reaction_close / next_open / monitor
    # ------------------------------------------------------------------ #

    async def run_reaction_close(
        self, target_date: dt.date | None = None, force: bool = False
    ) -> dict[str, Any]:
        """장 마감 직전 (~3:40 PM ET): same-day 이벤트 후보 → MOC 매수 주문.

        파이프라인 없이도 호출 가능. 당일 DB에 이미 적재된 이벤트를 사용.
        """
        today = target_date or dt.date.today()
        session_id = self._session.session_id
        phase = "reaction_close"

        if not force and self._state.is_phase_processed(session_id, today, phase):
            logger.info("paper_engine_already_processed", date=today.isoformat(), phase=phase)
            return {"date": today, "status": "already_processed", "phase": phase}

        from libs.common.time_utils import is_trading_day
        if not is_trading_day(today):
            return {"date": today, "status": "non_trading_day", "phase": phase}

        account = self._broker.get_account()
        alpaca_positions = self._broker.list_positions()
        session_st = self._state.get_session_state(session_id)
        strategy_states = {
            ss.symbol: ss
            for ss in self._state.get_open_strategy_states(session_id)
        }

        all_rows = await self._detector.get_candidates_for_date(
            today, self._config, convention="reaction_close"
        )
        macro_data = await self._fetch_macro(today)

        entries, rejected = await self._process_entries(
            today, all_rows, account, alpaca_positions, strategy_states,
            session_st, macro_data, self._broker.submit_moc_buy,
            entry_timing="reaction_close",
        )

        session_st.last_processed_date = today.isoformat()
        self._state.update_session_state(session_st)
        self._state.mark_phase_processed(session_id, today, phase)

        session_eq, session_ca = self._session_cash(session_id)
        logger.info(
            "paper_engine_reaction_close_done",
            date=today.isoformat(),
            candidates=len(all_rows),
            entries=len(entries),
            rejected=len(rejected),
        )
        return {
            "date": today,
            "status": "processed",
            "phase": phase,
            "entries": entries,
            "rejected": rejected,
            "candidates_detected": len(all_rows),
            "account": {"equity": session_eq, "cash": session_ca,
                        "market_value": account.long_market_value},
        }

    async def run_next_open(
        self, target_date: dt.date | None = None, force: bool = False
    ) -> dict[str, Any]:
        """장 시작 직후 (~9:30 AM ET): 전날 바로 exit 판단 + after-close 이벤트 → 시장가 매수.

        파이프라인이 전날 저녁 실행됐다고 가정.
        """
        today = target_date or dt.date.today()
        session_id = self._session.session_id
        phase = "next_open"

        if not force and self._state.is_phase_processed(session_id, today, phase):
            logger.info("paper_engine_already_processed", date=today.isoformat(), phase=phase)
            return {"date": today, "status": "already_processed", "phase": phase}

        from libs.common.time_utils import is_trading_day
        if not is_trading_day(today):
            return {"date": today, "status": "non_trading_day", "phase": phase}

        account = self._broker.get_account()
        alpaca_positions = self._broker.list_positions()
        session_st = self._state.get_session_state(session_id)
        strategy_states = {
            ss.symbol: ss
            for ss in self._state.get_open_strategy_states(session_id)
        }

        # Cancel stale orders + reconcile
        self._cancel_stale_orders()
        recon = self._reconcile_positions(alpaca_positions, strategy_states, today)
        for sym in recon.ghost_local:
            strategy_states.pop(sym, None)

        # Exit: 전날 (bar_date = today - 1) 종가 기준으로 exit 판단
        prev_date = today - dt.timedelta(days=1)
        exits = await self._process_exits(
            today, prev_date, alpaca_positions, strategy_states, session_st
        )

        # Refresh Alpaca state after exits
        account = self._broker.get_account()
        alpaca_positions = self._broker.list_positions()
        strategy_states = {
            ss.symbol: ss
            for ss in self._state.get_open_strategy_states(session_id)
        }

        # ============================================================
        # CASH PARKING: sell gate check first to free cash for entries
        # ============================================================
        parking_sold_today = False
        if self._config.risk.cash_parking_enabled:
            parking_sold_today = self._parking_check_and_sell(session_id, today)

        # Lookback entry: on first run_next_open, pick up events from previous days
        # that are still within their holding window (fires once per daemon session).
        lookback_entries: list[dict[str, Any]] = []
        lookback_rejected: list[dict[str, Any]] = []
        if self._config.execution.lookback_entry_enabled and not self._lookback_injected:
            self._lookback_injected = True
            if self._snapshot_store is not None:
                from libs.backtest.calendar import get_trading_days
                start_lb = self._lookback_start_date(today)
                lookback_rows = []
                for lb_date in get_trading_days(start_lb, today):
                    if lb_date >= today:
                        continue
                    for row in self._snapshot_store.get_candidates_for_date(lb_date):
                        row = dict(row)
                        row["is_lookback_entry"] = True
                        tdays = get_trading_days(lb_date, today)
                        row["lookback_days_elapsed"] = max(0, len(tdays) - 1)
                        lookback_rows.append(row)
            else:
                lookback_rows = await self._detector.get_candidates_for_lookback(
                    today, self._lookback_start_date(today), self._config
                )
            # Override entry_price_est with current market price for lookback rows.
            # Sizing was designed for the historical reaction-day price; filling at
            # today's market price without updating entry_price_est causes share counts
            # to be based on a stale price, which can result in the account going into
            # negative cash (cost = shares × current_price > shares × hist_price).
            if lookback_rows:
                lb_symbols = list({str(r.get("symbol", "")).upper() for r in lookback_rows if r.get("symbol")})
                try:
                    latest_bars = self._broker.get_latest_bars(lb_symbols)
                    for row in lookback_rows:
                        sym = str(row.get("symbol", "")).upper()
                        bar = latest_bars.get(sym)
                        if bar is not None:
                            row["lookback_original_entry_price_est"] = row.get("entry_price_est")
                            row["entry_price_est"] = bar.close
                except Exception as e:
                    logger.warning("lookback_price_override_failed", error=str(e))

            if lookback_rows:
                macro_data_lb = await self._fetch_macro(today)
                lookback_entries, lookback_rejected = await self._process_entries(
                    today, lookback_rows, account, alpaca_positions, strategy_states,
                    session_st, macro_data_lb, self._broker.submit_market_buy,
                    entry_timing="next_open",
                )
                # Refresh state after lookback entries
                alpaca_positions = self._broker.list_positions()
                strategy_states = {
                    ss.symbol: ss
                    for ss in self._state.get_open_strategy_states(session_id)
                }

        # Entry: all events with entry_date == today, across both conventions.
        # Mirrors BacktestRunner: next_open engines call get_candidates_for_date(date)
        # which returns ALL rows with execution_date==date regardless of entry_convention.
        #
        # Exclusion: same-day events with reaction_close convention are excluded here
        # because in the Parquet their execution_date is reaction_date+1 (next_open_after
        # _reaction_close entry), not reaction_date. Their next_open entry will appear
        # tomorrow with entry_convention='next_open_after_reaction_close'.
        # ENB (after-close with reaction_close convention, entry_date=reaction_date) is
        # correctly included because event_date != reaction_date (not same_day).
        if self._snapshot_store is not None:
            all_rows = self._snapshot_store.get_candidates_for_date(today)
        else:
            all_rows = await self._detector.get_candidates_for_date(
                today, self._config, convention=None
            )
        next_open_rows = [
            r for r in all_rows
            if not (self._is_same_day_event(r) and r.get("entry_convention") == "reaction_close")
        ]
        macro_data = await self._fetch_macro(today)

        # ============================================================
        # ROTATION: proactively close weak positions when good
        # candidates exist today. Must run before _process_entries
        # so freed cash is available for new entries.
        # Mirrors BacktestRunner._attempt_rotation_exits() (run.py:994).
        # ============================================================
        rotation_candidates = select_candidates(
            raw_rows=next_open_rows,
            universe_config=self._config.universe,
            signal_config=self._config.signal,
            event_type_profiles=self._config.event_type_profiles or {},
        )
        n_rotated = self._attempt_rotation_exits_paper(
            today, rotation_candidates, strategy_states, alpaca_positions
        )
        if n_rotated > 0:
            alpaca_positions = self._broker.list_positions()
            strategy_states = {
                ss.symbol: ss
                for ss in self._state.get_open_strategy_states(session_id)
            }
            account = self._broker.get_account()

        # ============================================================
        # MACRO RISK-ON: generate synthetic ETF candidates when broad
        # market breadth trigger fires. Mirrors BacktestRunner
        # _schedule_macro_long_candidates() (run.py:5509).
        # ============================================================
        open_symbols_now = {ss.symbol.upper() for ss in strategy_states.values()}
        macro_long_add_ons = await self._generate_macro_long_candidates(
            today, macro_data, open_symbols_now,
            event_breadth_count=len(rotation_candidates),
        )
        add_on_candidates: list[Candidate] = []
        if macro_long_add_ons:
            for row in macro_long_add_ons:
                try:
                    add_on_candidates.append(Candidate.model_validate(row))
                except Exception as exc:
                    logger.warning("paper_engine_macro_long_candidate_invalid", error=str(exc))

        entries, rejected = await self._process_entries(
            today, next_open_rows, account, alpaca_positions, strategy_states,
            session_st, macro_data, self._broker.submit_market_buy,
            entry_timing="next_open",
            add_on_candidates=add_on_candidates or None,
        )
        entries = lookback_entries + entries
        rejected = lookback_rejected + rejected

        # CASH PARKING: buy with remaining idle cash after entries
        if self._config.risk.cash_parking_enabled and not parking_sold_today:
            self._parking_buy(session_id, today)

        summary = self._finalize_day(today, session_st, exits, entries, rejected, len(next_open_rows))
        summary["phase"] = phase
        self._state.mark_phase_processed(session_id, today, phase)
        return summary

    async def run_monitor(self, interval_sec: int = 60) -> None:
        """장중 실시간 모니터링: stop/target 조건 충족 시 즉시 청산.

        Ctrl+C 로 종료. 별도 터미널에서 실행 권장.
        """
        import asyncio as _asyncio
        session_id = self._session.session_id
        logger.info("paper_engine_monitor_start", session=session_id, interval_sec=interval_sec)

        while True:
            try:
                alpaca_positions = self._broker.list_positions()
                strategy_states = {
                    ss.symbol: ss
                    for ss in self._state.get_open_strategy_states(session_id)
                }

                for pos in alpaca_positions:
                    ss = strategy_states.get(pos.symbol)
                    if ss is None:
                        continue

                    price = pos.current_price

                    # Trailing peak 업데이트
                    if price > ss.peak_price:
                        ss.peak_price = price
                        self._state.update_strategy_state(
                            session_id, pos.symbol, peak_price=price
                        )

                    # Stop 조건
                    if price < ss.current_stop:
                        logger.info(
                            "paper_engine_monitor_stop_hit",
                            symbol=pos.symbol, price=price, stop=ss.current_stop,
                        )
                        self._monitor_close(pos, ss, "STOP_INTRADAY", price)

                    # Target 조건
                    elif ss.target_price and price >= ss.target_price:
                        logger.info(
                            "paper_engine_monitor_target_hit",
                            symbol=pos.symbol, price=price, target=ss.target_price,
                        )
                        self._monitor_close(pos, ss, "TARGET_INTRADAY", price)

            except Exception as exc:
                logger.error("paper_engine_monitor_error", error=str(exc))

            await _asyncio.sleep(interval_sec)

    def _monitor_close(self, pos: Any, ss: Any, reason: str, price: float) -> None:
        """모니터링 루프에서 포지션 청산 처리."""
        session_id = self._session.session_id
        try:
            self._broker.close_position(pos.symbol, fill_price=price)
            self._state.close_strategy_state(session_id, pos.symbol)
            self._state.close_trade(
                session_id=session_id,
                symbol=pos.symbol,
                engine_id=ss.engine_id,
                capital_bucket_id=self._get_strategy_state_capital_bucket_id(ss),
                entry_date=ss.entry_date,
                exit_date=dt.date.today().isoformat(),
                entry_price=pos.avg_entry_price,
                exit_price=price,
                exit_reason=reason,
                shares=pos.qty,
                net_pnl=(price - pos.avg_entry_price) * pos.qty,
                r_multiple=0.0,
                holding_days=ss.days_held,
            )
        except Exception as exc:
            logger.error("paper_engine_monitor_close_failed", symbol=pos.symbol, error=str(exc))

    # ------------------------------------------------------------------ #
    # Shared helpers for phased execution
    # ------------------------------------------------------------------ #

    def _lookback_start_date(self, today: dt.date) -> dt.date:
        """Return the earliest date to search for lookback events (calendar-day buffer)."""
        from apps.backtester.run import _compute_max_effective_mhd
        max_mhd = _compute_max_effective_mhd(self._config)
        return today - dt.timedelta(days=max_mhd * 2)

    @staticmethod
    def _is_same_day_event(row: dict[str, Any]) -> bool:
        """event_date == reaction_date 이면 same-day (종가 진입) 이벤트."""
        def _pd(v: Any) -> dt.date | None:
            if isinstance(v, dt.datetime):
                return v.date()
            if isinstance(v, dt.date):
                return v
            if isinstance(v, str):
                try:
                    return dt.date.fromisoformat(v[:10])
                except ValueError:
                    return None
            return None
        ed = _pd(row.get("event_date"))
        rd = _pd(row.get("reaction_date"))
        return ed is not None and rd is not None and ed == rd

    async def _process_exits(
        self,
        today: dt.date,
        bar_date: dt.date,
        alpaca_positions: list[Any],
        strategy_states: dict[str, Any],
        session_st: Any,
    ) -> list[dict[str, Any]]:
        """보유 포지션에 대해 bar_date 기준 exit 로직 실행."""
        session_id = self._session.session_id
        exits: list[dict[str, Any]] = []
        held_symbols = [p.symbol for p in alpaca_positions]
        if not held_symbols:
            session_st.daily_new_risk_used = 0.0
            return exits

        bar_start = bar_date - dt.timedelta(days=30)
        # Fetch up to today so NO_PROGRESS / EARLY_FAILURE can execute at today's open.
        bars_by_symbol = self._broker.get_bars_as_dict(held_symbols, bar_start, today)

        for alpaca_pos in alpaca_positions:
            sym = alpaca_pos.symbol
            ss = strategy_states.get(sym)
            if ss is None:
                logger.debug("paper_engine_no_local_state", symbol=sym)
                continue

            ss.days_held += 1

            # Skip exit check for reaction_close (same-day) positions on their entry bar.
            # These positions were opened at the CLOSE of bar_date, so the intraday
            # bar data (low/high) precedes the actual entry and must not trigger stops.
            # next_open positions are NOT skipped — they entered at the OPEN so the
            # full day bar is valid for exit checking.
            if ss.entry_date == bar_date.isoformat() and _is_reaction_close_entry(ss.candidate_json):
                self._state.update_strategy_state(
                    session_id, sym, days_held=ss.days_held,
                    current_stop=ss.current_stop, peak_price=ss.peak_price,
                )
                continue

            sym_bars = bars_by_symbol.get(sym, {})
            available = [d for d in sym_bars if d <= bar_date]
            bar = sym_bars[max(available)] if available else None

            if bar is None:
                logger.warning("paper_engine_no_bar", symbol=sym, date=bar_date.isoformat())
                self._state.update_strategy_state(session_id, sym, days_held=ss.days_held)
                continue

            open_pos = self._to_open_position(alpaca_pos, ss)
            effective_exec = self._resolve_execution_config(ss)

            if effective_exec.trailing_model:
                update_trailing_stop(
                    open_pos, bar,
                    trailing_model=effective_exec.trailing_model,
                    warmup_days=effective_exec.trailing_warmup_days,
                )
                ss.current_stop = open_pos.current_stop
                ss.peak_price = open_pos.peak_price

            filled_trade = simulate_exit(open_pos, bar, effective_exec, bar_date)

            # NO_PROGRESS / EARLY_FAILURE: check yesterday's close, execute at today's open.
            # Mirrors BacktestRunner._evaluate_pending_open_exit + _process_pending_open_exits.
            if filled_trade is None:
                close_val = bar.get("close")
                if close_val is not None:
                    close_val = float(close_val)
                    np_days = effective_exec.early_failure_no_progress_days
                    np_r = effective_exec.early_failure_no_progress_r
                    scheduled_reason: str | None = None
                    # NO_PROGRESS: not enough progress by day N
                    if (
                        np_days is not None and np_r is not None
                        and ss.days_held == np_days
                        and open_pos.status.value != "partial"
                    ):
                        initial_r = abs(alpaca_pos.avg_entry_price - ss.current_stop)
                        progress_price = alpaca_pos.avg_entry_price + initial_r * np_r
                        if close_val < progress_price:
                            scheduled_reason = "NO_PROGRESS"
                    # EARLY_FAILURE: day-1 close below both entry and reaction close
                    if scheduled_reason is None and (
                        effective_exec.early_failure_close_below_entry_and_reaction_close
                        and ss.days_held == 1
                        and close_val < alpaca_pos.avg_entry_price
                    ):
                        reaction_close = float(open_pos.plan.candidate.features.get("event_close") or close_val)
                        if close_val < reaction_close:
                            scheduled_reason = "EARLY_FAILURE"
                    if scheduled_reason is not None:
                        today_bar = bars_by_symbol.get(sym, {}).get(today)
                        if today_bar is not None:
                            filled_trade = simulate_scheduled_open_exit(
                                position=open_pos,
                                bar=today_bar,
                                config=effective_exec,
                                current_date=today,
                                reason=scheduled_reason,
                                fraction=float(effective_exec.early_failure_no_progress_fraction or 1.0),
                            )

            if filled_trade is not None:
                is_partial = filled_trade.shares < alpaca_pos.qty
                try:
                    self._broker.close_position(sym, qty=filled_trade.shares if is_partial else None, fill_price=filled_trade.exit_price)
                    logger.info(
                        "paper_engine_exit",
                        symbol=sym, reason=filled_trade.exit_reason.value, pnl=filled_trade.net_pnl,
                        partial=is_partial,
                    )
                except Exception as exc:
                    logger.error("paper_engine_close_failed", symbol=sym, error=str(exc))
                    continue

                if is_partial:
                    # T1 partial exit: keep position tracked with breakeven stop
                    self._state.update_strategy_state(
                        session_id, sym,
                        days_held=ss.days_held,
                        current_stop=open_pos.current_stop,  # set to entry_price by simulate_exit
                        peak_price=ss.peak_price,
                        status="partial",
                    )
                else:
                    self._state.close_strategy_state(session_id, sym)
                self._state.close_trade(
                    session_id=session_id, symbol=sym,
                    engine_id=ss.engine_id,
                    capital_bucket_id=self._get_strategy_state_capital_bucket_id(ss),
                    entry_date=ss.entry_date, exit_date=today.isoformat(),
                    entry_price=alpaca_pos.avg_entry_price, exit_price=filled_trade.exit_price,
                    exit_reason=filled_trade.exit_reason.value, shares=filled_trade.shares,
                    net_pnl=filled_trade.net_pnl, r_multiple=filled_trade.r_multiple,
                    holding_days=ss.days_held,
                )
                if filled_trade.net_pnl < 0:
                    session_st.consecutive_losses += 1
                    streak = self._config.risk.cooldown_after_loss_streak
                    if streak > 0 and session_st.consecutive_losses >= streak:
                        session_st.cooldown_remaining = self._config.risk.cooldown_days
                        session_st.consecutive_losses = 0
                else:
                    session_st.consecutive_losses = 0
                exits.append({
                    "symbol": sym, "reason": filled_trade.exit_reason.value,
                    "pnl": filled_trade.net_pnl, "r_multiple": filled_trade.r_multiple,
                    "shares": filled_trade.shares, "exit_price": filled_trade.exit_price,
                })
            else:
                self._state.update_strategy_state(
                    session_id, sym, days_held=ss.days_held,
                    current_stop=ss.current_stop, peak_price=ss.peak_price,
                )

        if session_st.cooldown_remaining > 0:
            session_st.cooldown_remaining -= 1
        session_st.daily_new_risk_used = 0.0
        return exits

    async def _process_entries(
        self,
        today: dt.date,
        candidate_rows: list[dict[str, Any]],
        account: Any,
        alpaca_positions: list[Any],
        strategy_states: dict[str, Any],
        session_st: Any,
        macro_data: dict[str, Any],
        order_fn: Any,
        entry_timing: str | None = None,
        add_on_candidates: list[Candidate] | None = None,
    ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
        """후보군에 대해 진입 판단 + 주문 제출. order_fn = submit_market_buy | submit_moc_buy.

        entry_timing: 'reaction_close' or 'next_open'. When set, only engines with
        matching entry_timing_policy are used. Mirrors BacktestRunner's per-engine
        get_candidates_for_date vs get_candidates_for_reaction_date split.
        add_on_candidates: pre-built Candidate objects (e.g. macro_long synthetics)
            that bypass select_candidates() and are appended directly to candidate_batches.
        """
        session_id = self._session.session_id
        entries: list[dict[str, Any]] = []
        rejected: list[dict[str, Any]] = []

        # Kill switch gate
        if session_st.kill_switch_triggered:
            logger.warning("paper_engine_kill_switch_blocks_entries")
            return entries, rejected

        candidate_rows = [
            r for r in candidate_rows
            if not self._state.has_processed_event(session_id, str(r.get("event_id", "")))
        ]

        # Inject macro values from _fetch_macro() into candidate rows.
        # EventDetector (PostgreSQL) rows lack macro_vix/macro_hy_spread; the
        # Parquet snapshot pre-embeds them. Without this injection, any engine
        # with macro_vix_max set will reject all candidates (None fails the check).
        _macro_vix = macro_data.get("VIXCLS")
        _macro_hy = macro_data.get("BAMLH0A0HYM2")
        if _macro_vix is not None or _macro_hy is not None:
            for row in candidate_rows:
                if _macro_vix is not None and row.get("macro_vix") is None:
                    row["macro_vix"] = _macro_vix
                if _macro_hy is not None and row.get("macro_hy_spread") is None:
                    row["macro_hy_spread"] = _macro_hy

        open_positions = self._to_open_positions(alpaca_positions, strategy_states)
        portfolio_state = self._build_portfolio_state(account, alpaca_positions, today)
        engines = self._config.get_active_strategy_engines()
        engine_daily_risk_used: dict[str, float] = {}

        logger.debug(
            "paper_engine_selection_input",
            date=today.isoformat(),
            candidate_rows=len(candidate_rows),
            engines=len(engines),
            symbols=[r.get("symbol") for r in candidate_rows],
        )
        if candidate_rows:
            s = candidate_rows[0]
            logger.debug(
                "paper_engine_sample_row",
                symbol=s.get("symbol"), event_type=s.get("event_type"),
                event_direction=s.get("event_direction"), filing_time_bucket=s.get("filing_time_bucket"),
                entry_price_est=s.get("entry_price_est"), avg_dollar_volume=s.get("avg_dollar_volume"),
                avg_dollar_volume_20d=s.get("avg_dollar_volume_20d"), event_close=s.get("event_close"),
                close_location=s.get("close_location"), gap_size=s.get("gap_size"),
                reaction_day_return=s.get("reaction_day_return"), market_cap_proxy=s.get("market_cap_proxy"),
                execution_date=str(s.get("execution_date")), event_timestamp=str(s.get("event_timestamp")),
            )

        engine_list = engines if engines else [None]
        reserved_event_ids: set[str] = {ss.event_id for ss in strategy_states.values()}
        reserved_symbols: set[str] = {p.symbol for p in alpaca_positions if p.symbol in strategy_states}
        candidate_batches: list[tuple[Any | None, list[Candidate]]] = []
        for engine_cfg in engine_list:
            if engine_cfg is not None:
                # Skip engines that don't match the requested entry timing policy.
                # Mirrors BacktestRunner: reaction_close engines use get_candidates_for_reaction_date,
                # next_open engines use get_candidates_for_date.
                if entry_timing is not None and engine_cfg.entry_timing_policy != entry_timing:
                    continue
                prelimit = self._config.signal.max_candidates_per_day
                if self._attention_service.engine_requires_attention(engine_cfg):
                    prelimit = max(prelimit * 5, prelimit)
                engine_candidates = select_candidates(
                    raw_rows=candidate_rows,
                    universe_config=self._config.universe,
                    signal_config=self._config.signal,
                    event_type_profiles=self._config.event_type_profiles or {},
                    strategy_engine=engine_cfg,
                    truncate_to=prelimit,
                    excluded_event_ids=reserved_event_ids,
                    excluded_symbols=reserved_symbols,
                )
                # Attention filtering (matches BacktestRunner)
                engine_candidates = self._attention_service.apply_filters(
                    engine_candidates, engine_cfg, self._config.signal,
                )
                if engine_cfg.residual_reserve_selected and engine_candidates:
                    reserved_event_ids.update(c.event_id for c in engine_candidates)
                    reserved_symbols.update(c.symbol.upper() for c in engine_candidates)
            else:
                engine_candidates = select_candidates(
                    raw_rows=candidate_rows,
                    universe_config=self._config.universe,
                    signal_config=self._config.signal,
                    event_type_profiles=self._config.event_type_profiles or {},
                    excluded_event_ids=reserved_event_ids,
                    excluded_symbols=reserved_symbols,
                )
            candidate_batches.append((engine_cfg, engine_candidates))

        # Add pre-built add-on candidates (e.g. macro_long synthetics) directly.
        # Mirrors BacktestRunner scheduled_add_ons injection (run.py:1918-1925).
        if add_on_candidates:
            from collections import defaultdict as _defaultdict
            grouped_add_ons: dict[str, list[Candidate]] = _defaultdict(list)
            for c in add_on_candidates:
                if not self._state.has_processed_event(session_id, str(c.event_id)):
                    grouped_add_ons[c.engine_id].append(c)
            for eid, acs in grouped_add_ons.items():
                ecfg = next((e for e in engines if e.engine_id == eid), None)
                candidate_batches.append((ecfg, acs))

        # ============================================================
        # Lookback MHD expiration filter (mirrors BacktestRunner run.py:1832-1843).
        # Reject lookback candidates whose holding window has expired or whose
        # remaining holding period is too short to be worthwhile.
        # ============================================================
        min_remaining = self._config.execution.lookback_min_remaining_days
        if min_remaining is not None or True:  # always run expiry check
            from libs.backtest.execution import build_effective_execution_config
            filtered_batches: list[tuple[Any, list[Any]]] = []
            for ecfg, batch in candidate_batches:
                filtered: list[Any] = []
                for candidate in batch:
                    if not candidate.features.get("is_lookback_entry", False):
                        filtered.append(candidate)
                        continue
                    elapsed = int(candidate.features.get("lookback_days_elapsed", 0))
                    eff_exec = build_effective_execution_config(candidate, self._config)
                    candidate_mhd = eff_exec.max_holding_days
                    if elapsed >= candidate_mhd:
                        logger.info(
                            "lookback_skip_expired",
                            symbol=candidate.symbol, elapsed=elapsed, mhd=candidate_mhd,
                        )
                        continue
                    remaining = candidate_mhd - elapsed
                    if min_remaining is not None and remaining < min_remaining:
                        logger.info(
                            "lookback_skip_insufficient_remaining",
                            symbol=candidate.symbol, elapsed=elapsed, mhd=candidate_mhd,
                            remaining=remaining, min_required=min_remaining,
                        )
                        continue
                    filtered.append(candidate)
                filtered_batches.append((ecfg, filtered))
            candidate_batches = filtered_batches

        # ============================================================
        # Phase 2: Cross-Engine Global Score Ranking
        # Apply strategy_engine_selection_mode before allocation.
        # Mirrors BacktestRunner._select_candidates_for_date() mode logic
        # (run.py:1931-1941).
        # ============================================================
        selection_mode = getattr(self._config, "strategy_engine_selection_mode", "interleave")
        if selection_mode in ("global_score", "interleave_head_score"):
            from libs.backtest.selector import rank_candidates
            engine_queues: dict[str, list[Any]] = {}
            for ecfg, batch in candidate_batches:
                eid = ecfg.engine_id if ecfg is not None else "__default__"
                engine_queues[eid] = batch

            if selection_mode == "global_score":
                merged: list[Any] = []
                for batch in engine_queues.values():
                    merged.extend(batch)
                merged = rank_candidates(merged, self._config.signal.ranking_fields)
                candidate_batches = [(None, merged)]
            else:  # interleave_head_score
                engine_order = {
                    e.engine_id: idx
                    for idx, e in enumerate(engines)
                }
                working = {eid: list(v) for eid, v in engine_queues.items()}
                ordered: list[Any] = []
                while True:
                    head_pool: list[tuple[float, int, Any]] = []
                    for eid, queue in working.items():
                        if not queue:
                            continue
                        head_pool.append((queue[0].score, -engine_order.get(eid, 0), queue[0]))
                    if not head_pool:
                        break
                    _, _, winner = max(head_pool, key=lambda item: (item[0], item[1]))
                    ordered.append(working[winner.engine_id].pop(0))
                max_cands = self._config.signal.max_candidates_per_day
                candidate_batches = [(None, ordered[:max_cands])]

        active_bucket_ids = self._active_capital_bucket_ids_for_candidates(
            [
                candidate
                for _, batch_candidates in candidate_batches
                for candidate in batch_candidates
            ],
            strategy_states,
        )
        for engine_cfg, engine_candidates in candidate_batches:
            for candidate in engine_candidates:
                engine_risk_used = (
                    engine_daily_risk_used.get(engine_cfg.engine_id, 0.0)
                    if engine_cfg is not None
                    else 0.0
                )
                candidate_portfolio_state = self._adjust_portfolio_state_for_candidate(
                    session_id=session_id,
                    candidate=candidate,
                    portfolio_state=portfolio_state,
                    active_bucket_ids=active_bucket_ids,
                    alpaca_positions=alpaca_positions,
                    strategy_states=strategy_states,
                )
                plan = build_planned_order(
                    candidate=candidate, portfolio_state=candidate_portfolio_state,
                    open_positions=open_positions, config=self._config,
                    cooldown_remaining=session_st.cooldown_remaining,
                    macro_data=macro_data,
                    engine_daily_new_risk_used=engine_risk_used if engine_cfg else 0.0,
                )
                freed = False
                if plan.skip_reason == "insufficient_cash":
                    # 1) Attempt to free parking cash before giving up
                    needed = plan.shares * float(candidate.entry_price_est) if plan.shares else float(candidate.entry_price_est)
                    freed = self._parking_liquidate_for_event(session_id, today, needed)
                    # 2) If parking didn't help, try recycle (sell weak position)
                    if not freed and engine_cfg is not None and engine_cfg.recycle_on_cash_block:
                        freed = self._attempt_recycle_paper(
                            today, candidate, strategy_states, alpaca_positions, engine_cfg
                        )
                    if freed:
                        account = self._broker.get_account()
                        _ap2 = self._broker.list_positions()
                        alpaca_positions = _ap2
                        strategy_states = {
                            ss.symbol: ss
                            for ss in self._state.get_open_strategy_states(session_id)
                        }
                        portfolio_state = self._build_portfolio_state(account, _ap2, today)
                        candidate_portfolio_state = self._adjust_portfolio_state_for_candidate(
                            session_id=session_id,
                            candidate=candidate,
                            portfolio_state=portfolio_state,
                            active_bucket_ids=active_bucket_ids,
                            alpaca_positions=_ap2,
                            strategy_states=strategy_states,
                        )
                        plan = build_planned_order(
                            candidate=candidate, portfolio_state=candidate_portfolio_state,
                            open_positions=open_positions, config=self._config,
                            cooldown_remaining=session_st.cooldown_remaining,
                            macro_data=macro_data,
                            engine_daily_new_risk_used=engine_risk_used if engine_cfg else 0.0,
                        )

                if plan.skip_reason:
                    # If we just freed parking cash but plan still shows insufficient_cash,
                    # Alpaca may not have settled the SGOV sell yet → skip without recording
                    # so the event retries on the next run_next_open.
                    if freed and plan.skip_reason == "insufficient_cash":
                        logger.warning(
                            "paper_engine_parking_freed_cash_not_settled",
                            symbol=candidate.symbol,
                            hint="SGOV sold but cash not yet reflected in account — will retry on next run",
                        )
                        continue
                    self._state.record_processed_event(
                        session_id, candidate.event_id, today.isoformat(),
                        "rejected", skip_reason=plan.skip_reason,
                    )
                    rejected.append({
                        "symbol": candidate.symbol, "event_type": candidate.event_type,
                        "score": candidate.score, "reason": plan.skip_reason,
                    })
                    continue
                # Gap cap check for next_open entries (matches BacktestRunner).
                # Skipped for lookback entries: multi-day price drift vs reaction-day
                # close is not comparable to an overnight gap.
                is_lookback = bool(candidate.features.get("is_lookback_entry", False))
                if not is_lookback:
                    from libs.backtest.execution import check_next_open_gap_cap
                    today_bar = self._broker.get_bar(candidate.symbol) if hasattr(self._broker, 'get_bar') else None
                    gap_reason = check_next_open_gap_cap(candidate, today_bar)
                    if gap_reason:
                        self._state.record_processed_event(
                            session_id, candidate.event_id, today.isoformat(),
                            "rejected", skip_reason=gap_reason,
                        )
                        rejected.append({
                            "symbol": candidate.symbol, "event_type": candidate.event_type,
                            "score": candidate.score, "reason": gap_reason,
                        })
                        continue
                # Market-hours guard: skip (without recording) if market is closed.
                # Allows retry on next run_next_open when market is open.
                is_moc = (order_fn != self._broker.submit_market_buy)
                if not is_moc and not self._is_market_open():
                    logger.warning(
                        "paper_engine_market_closed_skip_entry",
                        symbol=candidate.symbol,
                        hint="Market is closed — skipping without recording so retry fires next run",
                    )
                    rejected.append({"symbol": candidate.symbol, "event_type": candidate.event_type, "score": candidate.score, "reason": "market_closed"})
                    continue
                try:
                    order = order_fn(candidate.symbol, plan.shares)
                    logger.info("paper_engine_buy_submitted", symbol=candidate.symbol, qty=plan.shares, order_id=order.id)
                except Exception as exc:
                    logger.error("paper_engine_buy_failed", symbol=candidate.symbol, error=str(exc))
                    self._state.record_processed_event(
                        session_id, candidate.event_id, today.isoformat(),
                        "rejected", skip_reason=f"order_failed:{exc}",
                    )
                    rejected.append({"symbol": candidate.symbol, "event_type": candidate.event_type, "score": candidate.score, "reason": f"order_failed:{exc}"})
                    continue

                # Verify fill (skip for MOC orders — they fill at close)
                if not is_moc:
                    verified, fill_fail_reason = self._verify_order_fill(order.id, candidate.symbol)
                    if verified is None:
                        if fill_fail_reason.startswith("alpaca_rejected:"):
                            # Alpaca explicitly rejected — record permanently
                            self._state.record_processed_event(
                                session_id, candidate.event_id, today.isoformat(),
                                "rejected", skip_reason=fill_fail_reason,
                            )
                        else:
                            # Timeout — do NOT record so retry works next run
                            logger.warning(
                                "paper_engine_fill_timeout_not_recorded",
                                symbol=candidate.symbol, order_id=order.id,
                            )
                        rejected.append({"symbol": candidate.symbol, "event_type": candidate.event_type, "score": candidate.score, "reason": fill_fail_reason})
                        continue
                    fill_price = verified.filled_avg_price or plan.entry_price_limit
                else:
                    fill_price = plan.entry_price_limit  # MOC: actual price unknown until close

                initial_days_held = int(candidate.features.get("lookback_days_elapsed", 0))
                self._state.save_strategy_state(
                    session_id,
                    StrategyStateRow(
                        session_id=session_id, symbol=candidate.symbol, event_id=candidate.event_id,
                        engine_id=candidate.engine_id, order_id=order.id, entry_date=today.isoformat(),
                        stop_price=plan.stop_price, target_price=plan.target_price,
                        current_stop=plan.stop_price, peak_price=fill_price,
                        days_held=initial_days_held, trade_direction=candidate.trade_direction,
                        candidate_json=candidate.model_dump_json(), plan_json=plan.model_dump_json(),
                        status="open",
                    ),
                )
                self._state.open_trade(
                    session_id=session_id,
                    symbol=candidate.symbol,
                    engine_id=candidate.engine_id,
                    capital_bucket_id=candidate.engine_capital_bucket_id,
                    entry_date=today.isoformat(),
                    entry_price=fill_price,
                    shares=plan.shares,
                )
                self._state.record_processed_event(
                    session_id, candidate.event_id, today.isoformat(), "entered",
                )
                trade_risk_state = candidate_portfolio_state.sizing_equity or candidate_portfolio_state.equity
                trade_risk = trade_risk_state * (
                    candidate.engine_per_trade_risk_pct or self._config.risk.per_trade_risk_pct
                )
                if engine_cfg:
                    engine_daily_risk_used[engine_cfg.engine_id] = engine_risk_used + trade_risk
                session_st.daily_new_risk_used += trade_risk

                alpaca_positions = self._broker.list_positions()
                strategy_states = {
                    ss.symbol: ss
                    for ss in self._state.get_open_strategy_states(session_id)
                }
                open_positions = self._to_open_positions(alpaca_positions, strategy_states)
                open_positions.append(self._virtual_open_position(candidate, plan, today))
                portfolio_state = DailyPortfolioState(
                    date=portfolio_state.date, equity=portfolio_state.equity,
                    sizing_equity=portfolio_state.sizing_equity,
                    cash_available=max(0.0, portfolio_state.cash_available - plan.entry_price_limit * plan.shares),
                    gross_exposure=portfolio_state.gross_exposure + plan.entry_price_limit * plan.shares,
                    net_exposure=portfolio_state.net_exposure + plan.entry_price_limit * plan.shares,
                    reserved_risk_budget=portfolio_state.reserved_risk_budget,
                    unrealized_pnl=portfolio_state.unrealized_pnl, realized_pnl=portfolio_state.realized_pnl,
                    open_positions=[p.position_id for p in open_positions],
                    daily_new_risk_used=session_st.daily_new_risk_used,
                    peak_equity=portfolio_state.peak_equity, current_drawdown_pct=portfolio_state.current_drawdown_pct,
                )
                entries.append({
                    "symbol": candidate.symbol, "event_type": candidate.event_type,
                    "score": candidate.score, "shares": plan.shares,
                    "entry_price": plan.entry_price_limit,
                    "stop": plan.stop_price, "target": plan.target_price, "order_id": order.id,
                })

        return entries, rejected

    def _finalize_day(
        self,
        today: dt.date,
        session_st: Any,
        exits: list[dict[str, Any]],
        entries: list[dict[str, Any]],
        rejected: list[dict[str, Any]],
        candidates_detected: int,
    ) -> dict[str, Any]:
        """일일 스냅샷 저장 + summary dict 반환."""
        session_id = self._session.session_id
        # 세션 소유 포지션만 집계 (Alpaca 전체 계좌가 아닌 세션 기준)
        alpaca_positions_final = self._broker.list_positions()
        session_symbols_final = {
            ss.symbol for ss in self._state.get_open_strategy_states(session_id)
        }
        session_positions_final = [p for p in alpaca_positions_final if p.symbol in session_symbols_final]
        session_market_value_final = sum(p.market_value for p in session_positions_final)
        session_unrealized_pl_final = sum(p.unrealized_pl for p in session_positions_final)

        # Include parking position in MV and unrealized P&L
        parking_mv_final, parking_unreal_final = self._parking_position_value(
            session_id, alpaca_positions_final
        )
        session_market_value_final += parking_mv_final
        session_unrealized_pl_final += parking_unreal_final

        # 세션 equity = initial_equity + 전체 실현 P&L + 현재 미실현 P&L (parking 포함)
        total_realized_pnl = sum(
            t.get("net_pnl", 0.0) for t in self._state.list_trades(session_id)
        )
        session_equity_final = self._session.initial_equity + total_realized_pnl + session_unrealized_pl_final
        session_cash_final = max(0.0, session_equity_final - session_market_value_final)

        prev_snapshots = self._state.list_snapshots(session_id)
        prev_equity = prev_snapshots[-1]["equity"] if prev_snapshots else self._session.initial_equity

        peak_equity = self._state.get_peak_equity(session_id, self._session.initial_equity)
        peak_equity = max(peak_equity, session_equity_final)
        total_pnl = session_equity_final - self._session.initial_equity
        drawdown_pct = (
            max(0.0, (peak_equity - session_equity_final) / peak_equity * 100)
            if peak_equity > 0 else 0.0
        )
        self._state.save_daily_snapshot(
            DailySnapshotRow(
                session_id=session_id, date=today.isoformat(), equity=session_equity_final,
                cash=session_cash_final, market_value=session_market_value_final,
                daily_pnl=session_equity_final - prev_equity, total_pnl=total_pnl,
                drawdown_pct=drawdown_pct, open_position_count=len(session_positions_final),
            )
        )
        # Kill switch check after drawdown computation
        self._check_kill_switch(drawdown_pct, session_st)

        session_st.last_processed_date = today.isoformat()
        self._state.update_session_state(session_st)
        logger.info(
            "paper_engine_day_done",
            date=today.isoformat(), exits=len(exits), entries=len(entries), rejected=len(rejected),
        )
        return {
            "date": today, "status": "processed",
            "exits": exits, "entries": entries, "rejected": rejected,
            "candidates_detected": candidates_detected,
            "account": {
                "equity": session_equity_final, "cash": session_cash_final,
                "market_value": session_market_value_final,
                "total_pnl": total_pnl, "drawdown_pct": drawdown_pct,
            },
        }

    # ------------------------------------------------------------------ #
    # Conversion helpers
    # ------------------------------------------------------------------ #

    def _to_open_position(
        self, alpaca_pos: Position, ss: StrategyStateRow
    ) -> OpenPosition:
        """Convert Alpaca position + local state to backtest OpenPosition."""
        candidate = Candidate.model_validate_json(ss.candidate_json)
        plan = PlannedOrder.model_validate_json(ss.plan_json)
        pos_status = (
            PositionStatus.PARTIALLY_EXITED if ss.status == "partial"
            else PositionStatus.ENTERED
        )
        return OpenPosition(
            position_id=ss.order_id or ss.symbol,
            plan=plan,
            entry_date=dt.date.fromisoformat(ss.entry_date),
            entry_price=alpaca_pos.avg_entry_price,
            entry_fill_slippage_bps=0.0,
            current_stop=ss.current_stop,
            target_price=ss.target_price,
            peak_price=ss.peak_price,
            shares_open=alpaca_pos.qty,
            shares_total=alpaca_pos.qty,
            days_held=ss.days_held,
            status=pos_status,
        )

    def _to_open_positions(
        self,
        alpaca_positions: list[Position],
        strategy_states: dict[str, StrategyStateRow],
    ) -> list[OpenPosition]:
        result: list[OpenPosition] = []
        for alpaca_pos in alpaca_positions:
            ss = strategy_states.get(alpaca_pos.symbol)
            if ss is None:
                continue
            result.append(self._to_open_position(alpaca_pos, ss))
        return result

    def _virtual_open_position(
        self, candidate: Candidate, plan: PlannedOrder, entry_date: dt.date
    ) -> OpenPosition:
        """Create a virtual OpenPosition for gate-checking after a new entry."""
        return OpenPosition(
            position_id=f"virtual_{candidate.symbol}",
            plan=plan,
            entry_date=entry_date,
            entry_price=plan.entry_price_limit,
            entry_fill_slippage_bps=0.0,
            current_stop=plan.stop_price,
            target_price=plan.target_price,
            peak_price=plan.entry_price_limit,
            shares_open=plan.shares,
            shares_total=plan.shares,
            days_held=0,
            status=PositionStatus.ENTERED,
        )

    def _build_portfolio_state(
        self,
        account: AccountInfo,
        alpaca_positions: list[Position],
        date: dt.date,
    ) -> DailyPortfolioState:
        """세션별 독립 equity/cash 기준으로 포트폴리오 상태 계산.

        Alpaca 계좌는 여러 세션이 공유하므로 account.equity/cash를 직접 쓰면
        안 됨. 대신 이 세션 고유의 equity(SQLite 스냅샷 기준)와 이 세션이
        보유한 포지션만 사용한다.
        """
        session_id = self._session.session_id

        # 이 세션 소유 포지션만 (SQLite strategy_states 기준)
        session_symbols = {
            ss.symbol for ss in self._state.get_open_strategy_states(session_id)
        }
        session_positions = [p for p in alpaca_positions if p.symbol in session_symbols]
        session_market_value = sum(p.market_value for p in session_positions)
        session_unrealized_pl = sum(p.unrealized_pl for p in session_positions)

        # Include parking position so available cash is not over-stated
        parking_mv, parking_unreal = self._parking_position_value(session_id, alpaca_positions)
        session_market_value += parking_mv
        session_unrealized_pl += parking_unreal

        # MockBroker (backtest): broker IS the session, use actual cash directly.
        # AlpacaBroker (live): multiple sessions may share account, derive from snapshot.
        from apps.paper_trader.mock_broker import MockBroker
        if isinstance(self._broker, MockBroker):
            session_cash = max(0.0, account.cash)
            session_equity = session_cash + session_market_value
        else:
            snapshots = self._state.list_snapshots(session_id)
            session_equity = (
                snapshots[-1]["equity"] if snapshots else self._session.initial_equity
            )
            session_cash = max(0.0, session_equity - session_market_value)

        peak_equity = self._state.get_peak_equity(session_id, self._session.initial_equity)
        peak_equity = max(peak_equity, session_equity)
        drawdown_pct = (
            max(0.0, (peak_equity - session_equity) / peak_equity * 100)
            if peak_equity > 0 else 0.0
        )
        session_st = self._state.get_session_state(session_id)
        return DailyPortfolioState(
            date=date,
            equity=session_equity,
            sizing_equity=session_equity,
            cash_available=session_cash,
            gross_exposure=session_market_value,
            net_exposure=session_market_value,
            reserved_risk_budget=0.0,
            unrealized_pnl=session_unrealized_pl,
            realized_pnl=0.0,
            open_positions=[p.symbol for p in session_positions],
            daily_new_risk_used=session_st.daily_new_risk_used,
            peak_equity=peak_equity,
            current_drawdown_pct=drawdown_pct,
        )

    def _resolve_execution_config(self, ss: StrategyStateRow) -> ExecutionConfig:
        """Get effective ExecutionConfig using shared function.

        Delegates to libs.backtest.execution.build_effective_execution_config()
        for consistency with BacktestRunner.
        """
        from libs.backtest.execution import build_effective_execution_config
        plan = PlannedOrder.model_validate_json(ss.plan_json)
        return build_effective_execution_config(plan.candidate, self._config)

    # ------------------------------------------------------------------ #
    # Macro data
    # ------------------------------------------------------------------ #

    async def _fetch_macro(self, date: dt.date) -> dict[str, Any]:
        """Fetch SPY/QQQ macro data for regime filtering."""
        try:
            sma_period = self._config.risk.macro_sma_period
            start = date - dt.timedelta(days=sma_period * 2 + 10)
            symbols = ["SPY", "QQQ"]

            # Fast path: use bars_cache from EventDetector (backtest mode)
            bars_cache = getattr(self._detector, "_bars_cache", None)
            if bars_cache is not None:
                macro: dict[str, Any] = {}
                for sym in symbols:
                    all_bars = bars_cache.get(sym, {})
                    closes = [
                        float(all_bars[d]["close"])
                        for d in sorted(all_bars.keys())
                        if start <= d <= date
                    ]
                    if closes:
                        key_prefix = sym.lower()
                        macro[f"{key_prefix}_close"] = closes[-1]
                        if len(closes) >= sma_period:
                            macro[f"{key_prefix}_sma_{sma_period}"] = sum(closes[-sma_period:]) / sma_period
                return macro

            from libs.oracle_client import OracleClient, PriceService

            async with OracleClient(base_url=self._detector._oracle_url) as client:
                svc = PriceService(client)

                async def _fetch_macro_sym(sym: str) -> tuple[str, list[Any]]:
                    try:
                        resp = await svc.get_daily_bars(sym, start=start.isoformat(), end=date.isoformat())
                        return sym, resp.bars
                    except Exception:
                        return sym, []

                results = await __import__("asyncio").gather(
                    *(_fetch_macro_sym(sym) for sym in symbols)
                )

            macro = {}
            for sym, bars in results:
                if not bars:
                    continue
                key_prefix = sym.lower()
                closes = [float(b.close) for b in bars]
                if closes:
                    macro[f"{key_prefix}_close"] = closes[-1]
                    if len(closes) >= sma_period:
                        macro[f"{key_prefix}_sma_{sma_period}"] = sum(closes[-sma_period:]) / sma_period

            # Fetch FRED macro data (VIX, HY spread) for regime sizing
            # Matches SnapshotStore._fetch_macro() which loads MacroObservation from DB
            try:
                from libs.oracle_client import FredService, OracleClient as _OC
                async with _OC(base_url=self._detector._oracle_url) as fred_client:
                    fred_svc = FredService(fred_client)
                    for series_id in ("VIXCLS", "BAMLH0A0HYM2"):
                        try:
                            resp = await fred_svc.get_observations(series_id, start=start.isoformat(), end=date.isoformat())
                            if resp.observations:
                                latest = [o for o in resp.observations if o.value is not None]
                                if latest:
                                    macro[series_id] = latest[-1].value
                        except Exception:
                            pass
            except Exception:
                pass

            return macro

        except Exception as exc:
            logger.warning("paper_engine_macro_fetch_failed", error=str(exc))
            return {}

    # ------------------------------------------------------------------ #
    # Macro Risk-On Sleeve (Phase 3)
    # ------------------------------------------------------------------ #

    async def _fetch_etf_bars(
        self,
        symbols: list[str],
        as_of_date: dt.date,
        lookback_days: int = 30,
    ) -> dict[str, dict[dt.date, dict[str, Any]]]:
        """Fetch daily bar data for ETF symbols from Oracle/Alpaca.

        Returns {symbol -> {date -> bar_dict}}.
        """
        start = as_of_date - dt.timedelta(days=lookback_days)
        result: dict[str, dict[dt.date, dict[str, Any]]] = {}

        # Fast path: use EventDetector bars_cache (backtest mode)
        bars_cache = getattr(self._detector, "_bars_cache", None)
        if bars_cache is not None:
            for sym in symbols:
                sym_bars = bars_cache.get(sym.upper(), {})
                result[sym.upper()] = {
                    d: bar for d, bar in sym_bars.items()
                    if start <= d <= as_of_date
                }
            return result

        # Live path: Oracle PriceService
        try:
            from libs.oracle_client import OracleClient, PriceService

            async def _fetch_sym(sym: str) -> tuple[str, list[Any]]:
                try:
                    async with OracleClient(base_url=self._detector._oracle_url) as client:
                        svc = PriceService(client)
                        resp = await svc.get_daily_bars(sym, start=start.isoformat(), end=as_of_date.isoformat())
                        return sym, resp.bars
                except Exception:
                    return sym, []

            import asyncio as _aio
            fetched = await _aio.gather(*(_fetch_sym(s) for s in symbols))
            for sym, bars in fetched:
                if not bars:
                    continue
                bar_dict: dict[dt.date, dict[str, Any]] = {}
                for b in bars:
                    bar_date = dt.date.fromisoformat(str(b.date)[:10]) if hasattr(b, "date") else None
                    if bar_date is None:
                        continue
                    bar_dict[bar_date] = {
                        "open": float(b.open), "high": float(b.high),
                        "low": float(b.low), "close": float(b.close),
                        "volume": float(b.volume),
                    }
                result[sym.upper()] = bar_dict
        except Exception as exc:
            logger.warning("paper_engine_etf_bars_fetch_failed", error=str(exc))

        return result

    async def _generate_macro_long_candidates(
        self,
        today: dt.date,
        macro_data: dict[str, Any],
        open_symbols: set[str],
        event_breadth_count: int = 0,
    ) -> list[dict[str, Any]]:
        """Generate synthetic macro_long row dicts when breadth trigger fires.

        Uses MacroLongScreener to evaluate each engine with macro_long_symbol set.
        Returns list of row dicts that can be passed to _process_entries as
        synthetic event candidates (already as Candidate via the screener).
        """
        from libs.backtest.macro_screener import MacroLongScreener, compute_market_features_from_bars
        from libs.common.time_utils import next_trading_day

        macro_engines = [
            e for e in self._config.get_active_strategy_engines()
            if e.macro_long_symbol
        ]
        if not macro_engines:
            return []

        # Collect all unique symbols we need bar data for
        all_symbols: set[str] = set()
        for engine in macro_engines:
            all_symbols.add(str(engine.macro_long_symbol).upper())
            for bs in (engine.macro_long_breadth_symbols or []):
                all_symbols.add(str(bs).upper())
        if any(e.macro_long_leadership_vs_spy_min is not None for e in macro_engines):
            all_symbols.add("SPY")

        # The trigger is evaluated on yesterday (signal_date = today - 1 trading day)
        # and the candidate is entered today (execution_date = today).
        prev_dates = [d for d in [today - dt.timedelta(days=i) for i in range(1, 8)]]
        from libs.common.time_utils import is_trading_day
        signal_date = next((d for d in prev_dates if is_trading_day(d)), None)
        if signal_date is None:
            return []
        execution_date = today

        bars_by_symbol = await self._fetch_etf_bars(list(all_symbols), signal_date, lookback_days=35)

        macro_vix = macro_data.get("VIXCLS")
        screener = MacroLongScreener()
        candidates: list[dict[str, Any]] = []

        for engine in macro_engines:
            trigger_sym = str(engine.macro_long_symbol).upper()
            trigger_bars = bars_by_symbol.get(trigger_sym, {})
            market_features = compute_market_features_from_bars(trigger_bars, signal_date)
            if not market_features:
                continue

            breadth_features: dict[str, dict[str, Any]] = {}
            for bs in (engine.macro_long_breadth_symbols or []):
                bsym = str(bs).upper()
                bbars = bars_by_symbol.get(bsym, {})
                bfeat = compute_market_features_from_bars(bbars, signal_date)
                if bfeat:
                    breadth_features[bsym] = bfeat
            if "SPY" not in breadth_features:
                spy_bars = bars_by_symbol.get("SPY", {})
                spy_feat = compute_market_features_from_bars(spy_bars, signal_date)
                if spy_feat:
                    breadth_features["SPY"] = spy_feat

            # Execution bar for price reference
            exec_bars = bars_by_symbol.get(trigger_sym, {})
            exec_bar = exec_bars.get(execution_date)

            candidate = screener.screen(
                signal_date=signal_date,
                execution_date=execution_date,
                engine=engine,
                macro_vix=macro_vix,
                market_features=market_features,
                breadth_features=breadth_features,
                open_symbols=open_symbols,
                event_breadth_count=event_breadth_count,
                execution_bar=exec_bar,
            )
            if candidate is not None:
                # Convert Candidate to row dict compatible with _process_entries
                row = candidate.model_dump()
                # Mark as synthetic so _process_entries doesn't double-check Oracle
                row["is_macro_long_synthetic"] = True
                candidates.append(row)
                logger.info(
                    "paper_engine_macro_long_triggered",
                    date=today.isoformat(),
                    engine=engine.engine_id,
                    symbol=candidate.symbol,
                    score=round(candidate.score, 3),
                    reaction_return=market_features.get("reaction_day_return"),
                )

        return candidates

    # ------------------------------------------------------------------ #
    # Momentum Breakout Sleeve (Phase 4)
    # ------------------------------------------------------------------ #

    # ------------------------------------------------------------------ #
    # Rotation & Recycle (Phase 1)
    # ------------------------------------------------------------------ #

    def _compute_hold_fitness_paper(
        self,
        ss: StrategyStateRow,
        current_price: float,
        execution_config: ExecutionConfig,
    ) -> float:
        """fitness = 0.4*progress + 0.3*(1-time_used) + 0.3*(1-peak_dd).

        Mirrors BacktestRunner._compute_hold_fitness() (run.py:7775).
        Uses current Alpaca price instead of bar["close"].
        """
        max_days = execution_config.max_holding_days or 25
        time_used = min(1.0, ss.days_held / max_days)

        stop_dist = abs(ss.stop_price - ss.current_stop)  # entry_price - stop approximation
        # Better: reconstruct from plan_json
        try:
            plan = PlannedOrder.model_validate_json(ss.plan_json)
            entry_price = plan.entry_price_limit
            stop_price = plan.stop_price
            target_r = execution_config.target_1_r or execution_config.a_tier_target_1_r or 2.0
        except Exception:
            entry_price = ss.stop_price  # fallback
            stop_price = ss.current_stop
            target_r = 2.0

        stop_dist = abs(entry_price - stop_price)
        unrealized_r = (current_price - entry_price) / stop_dist if stop_dist > 0 else 0.0
        progress = max(-1.0, unrealized_r / target_r) if target_r > 0 else 0.0

        peak = ss.peak_price if ss.peak_price > 0 else entry_price
        peak_dd = (peak - current_price) / peak if peak > 0 else 0.0

        return 0.4 * progress + 0.3 * (1.0 - time_used) + 0.3 * (1.0 - peak_dd)

    def _attempt_rotation_exits_paper(
        self,
        today: dt.date,
        candidates: list[Any],
        strategy_states: dict[str, StrategyStateRow],
        alpaca_positions: list[Any],
    ) -> int:
        """Proactively close stale positions when good opportunities exist.

        Mirrors BacktestRunner._attempt_rotation_exits() (run.py:7797).
        Returns number of positions rotated out.
        """
        session_id = self._session.session_id

        rotation_engines = {
            e.engine_id: e
            for e in self._config.get_active_strategy_engines()
            if e.rotation_enabled
        }
        if not rotation_engines:
            return 0

        min_score = min(e.rotation_min_candidate_score for e in rotation_engines.values())
        has_opportunity = any(c.score >= min_score for c in candidates)
        if not has_opportunity:
            return 0

        rotated = 0
        for alpaca_pos in list(alpaca_positions):
            sym = alpaca_pos.symbol
            ss = strategy_states.get(sym)
            if ss is None:
                continue
            engine_cfg = rotation_engines.get(ss.engine_id)
            if engine_cfg is None:
                continue
            if ss.days_held < engine_cfg.rotation_min_days_held:
                continue
            if ss.status != "open":
                continue

            current_price = float(alpaca_pos.current_price)
            effective_exec = self._resolve_execution_config(ss)
            fitness = self._compute_hold_fitness_paper(ss, current_price, effective_exec)

            plan = PlannedOrder.model_validate_json(ss.plan_json)
            stop_dist = abs(plan.entry_price_limit - plan.stop_price)
            unrealized_r = (
                (current_price - plan.entry_price_limit) / stop_dist
                if stop_dist > 0 else 0.0
            )

            if fitness >= engine_cfg.rotation_fitness_threshold:
                continue
            if (
                engine_cfg.rotation_max_unrealized_r is not None
                and unrealized_r > engine_cfg.rotation_max_unrealized_r
            ):
                continue
            if (
                engine_cfg.rotation_min_unrealized_r is not None
                and unrealized_r >= engine_cfg.rotation_min_unrealized_r
            ):
                continue

            try:
                self._broker.close_position(sym)
            except Exception as exc:
                logger.error("paper_engine_rotation_close_failed", symbol=sym, error=str(exc))
                continue

            pnl = (current_price - plan.entry_price_limit) * alpaca_pos.qty
            r_multiple = unrealized_r

            self._state.close_strategy_state(session_id, sym)
            self._state.close_trade(
                session_id=session_id,
                symbol=sym,
                engine_id=ss.engine_id,
                capital_bucket_id=self._get_strategy_state_capital_bucket_id(ss),
                entry_date=ss.entry_date,
                exit_date=today.isoformat(),
                entry_price=plan.entry_price_limit,
                exit_price=current_price,
                exit_reason="ROTATION",
                shares=alpaca_pos.qty,
                net_pnl=pnl,
                r_multiple=r_multiple,
                holding_days=ss.days_held,
            )
            logger.info(
                "paper_engine_rotation_exit",
                date=today.isoformat(), symbol=sym,
                fitness=round(fitness, 3), days_held=ss.days_held,
                unrealized_r=round(unrealized_r, 2), pnl=round(pnl, 2),
            )
            rotated += 1

        return rotated

    def _attempt_recycle_paper(
        self,
        today: dt.date,
        candidate: Any,
        strategy_states: dict[str, StrategyStateRow],
        alpaca_positions: list[Any],
        engine_cfg: Any,
    ) -> bool:
        """Sell a weak existing position to free cash for a better candidate.

        Mirrors BacktestRunner._attempt_same_day_cash_recycle() (run.py:7600).
        Returns True if a victim was found and sold.
        """
        if engine_cfg is None or not engine_cfg.recycle_on_cash_block:
            return False

        session_id = self._session.session_id
        allow_any_engine = bool(getattr(engine_cfg, "recycle_allow_any_victim_engine", False))
        allowed_victims = None if allow_any_engine else set(
            engine_cfg.recycle_allowed_victim_engine_ids or [candidate.engine_id]
        )
        min_days = engine_cfg.recycle_min_days_held or 0
        min_delta = engine_cfg.recycle_min_score_delta or 0.0

        eligible: list[tuple[float, float, float, Any, Any]] = []
        for alpaca_pos in alpaca_positions:
            sym = alpaca_pos.symbol
            ss = strategy_states.get(sym)
            if ss is None:
                continue
            if allowed_victims is not None and ss.engine_id not in allowed_victims:
                continue
            if ss.days_held < min_days:
                continue
            current_price = float(alpaca_pos.current_price)
            if current_price <= 0:
                continue

            plan = PlannedOrder.model_validate_json(ss.plan_json)
            victim_score = 0.0
            try:
                victim_cand = Candidate.model_validate_json(ss.candidate_json)
                victim_score = victim_cand.score
            except Exception:
                pass

            if candidate.score < (victim_score + min_delta):
                continue
            if engine_cfg.recycle_positive_pnl_only and current_price < plan.entry_price_limit:
                continue

            proceeds = current_price * alpaca_pos.qty
            effective_exec = self._resolve_execution_config(ss)
            fitness = self._compute_hold_fitness_paper(ss, current_price, effective_exec)
            stop_dist = abs(plan.entry_price_limit - plan.stop_price)
            unrealized_r = (
                (current_price - plan.entry_price_limit) / stop_dist
                if stop_dist > 0 else 0.0
            )

            max_victim_fitness = getattr(engine_cfg, "recycle_max_victim_fitness", None)
            max_victim_r = getattr(engine_cfg, "recycle_max_victim_unrealized_r", None)
            if max_victim_fitness is not None and fitness > max_victim_fitness:
                continue
            if max_victim_r is not None and unrealized_r > max_victim_r:
                continue

            eligible.append((fitness, unrealized_r, victim_score, alpaca_pos, ss))

        if not eligible:
            return False

        # Sort by fitness ascending (weakest first)
        eligible.sort(key=lambda x: (x[0], x[1]))
        victim_pos, victim_ss = eligible[0][3], eligible[0][4]
        victim_price = float(victim_pos.current_price)
        victim_plan = PlannedOrder.model_validate_json(victim_ss.plan_json)

        try:
            self._broker.close_position(victim_pos.symbol)
        except Exception as exc:
            logger.error("paper_engine_recycle_close_failed", symbol=victim_pos.symbol, error=str(exc))
            return False

        pnl = (victim_price - victim_plan.entry_price_limit) * victim_pos.qty
        stop_dist = abs(victim_plan.entry_price_limit - victim_plan.stop_price)
        r_mult = (victim_price - victim_plan.entry_price_limit) / stop_dist if stop_dist > 0 else 0.0

        self._state.close_strategy_state(session_id, victim_pos.symbol)
        self._state.close_trade(
            session_id=session_id,
            symbol=victim_pos.symbol,
            engine_id=victim_ss.engine_id,
            capital_bucket_id=self._get_strategy_state_capital_bucket_id(victim_ss),
            entry_date=victim_ss.entry_date,
            exit_date=today.isoformat(),
            entry_price=victim_plan.entry_price_limit,
            exit_price=victim_price,
            exit_reason="RECYCLE",
            shares=victim_pos.qty,
            net_pnl=pnl,
            r_multiple=r_mult,
            holding_days=victim_ss.days_held,
        )
        logger.info(
            "paper_engine_recycle_exit",
            date=today.isoformat(),
            victim_symbol=victim_pos.symbol,
            replacement_symbol=candidate.symbol,
            pnl=round(pnl, 2),
        )
        return True