"""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 asyncio 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 build_effective_execution_config, 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()) # Exclude parking position from orphaned check (it's tracked in parking_state, not strategy_states) parking_st = self._state.get_parking_state(self._session.session_id) parking_sym = parking_st["symbol"].upper() if parking_st else None # Orphaned: on Alpaca but no local state (e.g. manual buy, or state save failed) for sym in sorted(alpaca_symbols - local_symbols): if parking_sym and sym == parking_sym: continue # parking position — tracked separately, not truly orphaned 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) } # Lookback entry: pick up prior-day candidates still within holding window. # Fires once per engine instance (same gate as run_next_open). 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) if lookback_rows: lb_symbols = list({str(r.get("symbol", "")).upper() for r in lookback_rows if r.get("symbol") and "-" not in str(r.get("symbol", ""))}) try: latest_bars = self._broker.get_latest_bars(lb_symbols) except Exception as e: logger.warning("lookback_price_override_failed", error=str(e)) latest_bars = {} 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 # Also override event_close so reaction_close engines use today's # price (not the historical reaction-day close) for entry/sizing. row["event_close"] = bar.close 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_after_exits, strategy_states_after_exits, session_st, macro_data_lb, self._broker.submit_market_buy, entry_timing=None, ) 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) } entries.extend(lookback_entries) rejected.extend(lookback_rejected) # 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, execution_config=build_effective_execution_config(candidate, 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, execution_config=build_effective_execution_config(candidate, 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, execution_config=build_effective_execution_config(candidate, 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: await 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.close_trade( session_id=session_id, symbol=sym, engine_id="parking", 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.close_trade( session_id=session_id, symbol=sym, engine_id="parking", 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.close_trade( session_id=session_id, symbol=sym, engine_id="parking", 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 avg = float(parking_st.get("avg_price", cur_price)) raw_entry_date = parking_st.get("entry_date", today) if isinstance(raw_entry_date, str): raw_entry_date = dt.date.fromisoformat(raw_entry_date[:10]) if new_qty <= 0: self._state.close_parking_state(session_id) else: self._state.save_parking_state( session_id, sym, raw_entry_date, 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, ) self._state.record_trade( session_id=session_id, symbol=sym, engine_id="parking", capital_bucket_id=None, entry_date=raw_entry_date.isoformat(), exit_date=today.isoformat(), entry_price=avg, exit_price=cur_price, exit_reason="PARKING_LIQUIDATE_FOR_EVENT", shares=shares_to_sell, net_pnl=(cur_price - avg) * shares_to_sell, r_multiple=0.0, holding_days=(today - raw_entry_date).days, ) 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) or repr(exc), exc_type=type(exc).__name__, symbol=sym, qty=shares_to_sell) 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. Equity is computed from first principles (initial + realized + unrealized) rather than the stale daily snapshot so that intraday partial exits and position gains are properly reflected when sizing parking purchases. """ 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 live positions (same session symbols only) 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, parking_unreal = self._parking_position_value(session_id, alpaca_positions) session_mv += parking_mv # Compute equity from first principles (same as _finalize_day) so intraday # realized P&L (e.g. partial exits) is reflected. Using the stale daily # snapshot underestimates cash when positions gain value within the day. total_realized = sum( (t.get("net_pnl") or 0.0) for t in self._state.list_trades(session_id) ) session_equity = self._session.initial_equity + total_realized + session_unreal + parking_unreal session_cash = max(0.0, session_equity - session_mv) return session_equity, session_cash async 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 MAX_ATTEMPTS = 3 CONFIRM_SECS = 30 for attempt in range(1, MAX_ATTEMPTS + 1): if attempt > 1: 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), attempt=attempt) try: order = self._broker.submit_market_buy(sym, qty) except Exception as e: logger.warning("parking_buy_submit_failed", error=str(e), attempt=attempt) if attempt < MAX_ATTEMPTS: await asyncio.sleep(5) continue filled_price: float | None = None for _ in range(CONFIRM_SECS): await asyncio.sleep(1) o = self._broker.get_order(order.id) if o and o.filled_avg_price: filled_price = o.filled_avg_price break if filled_price is None: o = self._broker.get_order(order.id) if o and o.filled_avg_price: filled_price = o.filled_avg_price if filled_price is not None: self._state.save_parking_state( session_id, sym, today, qty, filled_price, filled_price * qty, peak_price=filled_price, gate_in_sgov=1 if target == "sgov" else 0, committed_target=target, ) self._state.open_trade( session_id, sym, "parking", None, today.isoformat(), filled_price, qty, ) logger.info("parking_filled", symbol=sym, qty=qty, price=round(filled_price, 2)) return logger.warning("parking_buy_timeout", symbol=sym, order_id=order.id, attempt=attempt) try: self._broker.cancel_order(order.id) except Exception: # cancel may fail if order filled between final check and cancel call o = self._broker.get_order(order.id) if o and o.filled_avg_price: self._state.save_parking_state( session_id, sym, today, qty, o.filled_avg_price, o.filled_avg_price * qty, peak_price=o.filled_avg_price, gate_in_sgov=1 if target == "sgov" else 0, committed_target=target, ) self._state.open_trade( session_id, sym, "parking", None, today.isoformat(), o.filled_avg_price, qty, ) logger.info("parking_filled_after_cancel_fail", symbol=sym) return if attempt < MAX_ATTEMPTS: await asyncio.sleep(5) logger.error("parking_buy_all_attempts_failed", symbol=sym) # ------------------------------------------------------------------ # # 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: try: parking_sold_today = self._parking_check_and_sell(session_id, today) except Exception as _pcs_exc: logger.warning( "paper_engine_parking_check_failed_skipped", error=str(_pcs_exc) or repr(_pcs_exc), exc_type=type(_pcs_exc).__name__, ) # 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") and "-" not in str(r.get("symbol", ""))}) try: latest_bars = self._broker.get_latest_bars(lb_symbols) except Exception as e: logger.warning("lookback_price_override_failed", error=str(e)) latest_bars = {} 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 # Also override event_close so reaction_close engines use today's # price (not the historical reaction-day close) for entry/sizing. row["event_close"] = bar.close 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=None, # lookback: allow all engines regardless of original convention ) # 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: await 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, execution_config=build_effective_execution_config(candidate, 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 parking_had_qty = False if plan.skip_reason == "insufficient_cash": # Check if parking has shares before attempting liquidation _pst = self._state.get_parking_state(session_id) parking_had_qty = _pst is not None and (_pst.get("qty") or 0) > 0 # 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, execution_config=build_effective_execution_config(candidate, 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: # Case 1: parking freed but cash not yet reflected in account (Alpaca latency) 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 # Case 2: parking existed but sell failed (transient Alpaca error) # Don't permanently reject — retry next run when sell may succeed. if not freed and parking_had_qty and plan.skip_reason == "insufficient_cash": logger.warning( "paper_engine_parking_sell_failed_skip", symbol=candidate.symbol, hint="Parking sell failed (transient?) — skipping without recording to allow retry", ) 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 # Lookback entries start at days_held=0 (actual hold time from today), # not lookback_days_elapsed. The signal validity was already checked # (elapsed < mhd, min_remaining_days) before entry was allowed. initial_days_held = 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") or 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 import asyncio as _asyncio from libs.oracle_client import OracleClient, PriceService async def _fetch_price_bars() -> dict[str, Any]: _macro: dict[str, Any] = {} async with OracleClient(base_url=self._detector._oracle_url, timeout=5.0) 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 _asyncio.gather( *(_fetch_macro_sym(sym) for sym in symbols) ) 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 try: from libs.oracle_client import FredService, OracleClient as _OC async with _OC(base_url=self._detector._oracle_url, timeout=5.0) 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 # Hard cap: don't let Oracle hangs block trading for more than 8s try: macro = await _asyncio.wait_for(_fetch_price_bars(), timeout=8.0) except _asyncio.TimeoutError: logger.warning("paper_engine_macro_fetch_timeout") macro = {} 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