"""ORB Paper Trading Engine. One engine instance is created per session per trading day. The scheduler calls phase methods in order: 1. run_pre_market() — 08:00 ET 2. run_orb_detection() — 09:40 ET (ORB window closes) 3. run_breakout_check() — 09:45...10:15 ET (called 7 times, every 5 min) 4. run_stop_check() — 11:10, 12:40, 14:10, 15:40 ET (90-min checkpoints) 5. run_eod_exit() — 15:55 ET 6. run_post_close() — 16:00 ET Stop management logic mirrors orb_simulator.py:477-580 exactly. """ from __future__ import annotations import datetime as dt import logging import uuid from typing import Any from zoneinfo import ZoneInfo from apps.orb_trader.models import ( ORBCandidateRow, ORBDailySnapshotRow, ORBPositionRow, ORBTradeRow, ) from apps.orb_trader.screener import ( bars_to_enrichment_format, intraday_bars_to_format, load_universe, ) from apps.orb_trader.state import ORBStateManager log = logging.getLogger(__name__) _ET = ZoneInfo("America/New_York") class ORBTradingEngine: """Intraday paper trading engine for the ORB strategy. Holds per-day state (enrichment, candidates, date_str) as instance variables. On server restart mid-day, state is reconstructed from the DB. """ def __init__( self, session: Any, broker: Any, state: ORBStateManager, params: Any, ) -> None: self._session = session self._broker = broker self._state = state self._params = params # Force live trading overrides self._params.compound_returns = True self._params.settlement_days = 0 self._params.slippage_bps = 0.0 # Per-day in-memory state (reset each day) self._date_str: str = "" self._enrichment: dict[str, dict[str, dict]] = {} self._daily_bars: dict[str, list[dict]] = {} self._candidates: list[dict] = [] # computed by run_orb_detection self._pending_cands: list[dict] = [] # not yet filled (for breakout checks) def _get_equity(self) -> float: """Current equity = last snapshot equity, or initial if no snapshots.""" eq = self._state.get_equity(self._session.session_id) return eq if eq is not None else self._session.initial_equity def _log(self, msg: str) -> None: log.info("[ORB:%s] %s", self._session.session_name, msg) # ── Phase 1: ORB Detection (includes daily-bar fetch for full universe) ───── def run_orb_detection(self, date_str: str) -> dict[str, Any]: """Fetch daily bars for the full universe + 5-min ORB bars, then compute candidates. Called once at 9:30 + orb_minutes (e.g., 9:40 for a 10-min ORB). No separate pre-market step: daily enrichment and ORB bars are fetched together so all universe tickers are evaluated without a prior filter pass. Returns summary dict. """ self._date_str = date_str self._state.update_daily_state( self._session.session_id, date_str, phase="orb_detection" ) # ── Step 1: Fetch 65 days of daily bars for full universe ───────────── universe_source = getattr(self._params, "_universe_source", "midlarge") universe_symbols_file = getattr(self._params, "_universe_symbols_file", None) tickers = load_universe(universe_source, universe_symbols_file) self._log(f"Universe: {len(tickers)} tickers — fetching daily bars") today = dt.date.fromisoformat(date_str) start = today - dt.timedelta(days=65) raw_bars: dict[str, list] = {} chunk_size = 200 for i in range(0, len(tickers), chunk_size): chunk = tickers[i : i + chunk_size] raw_bars.update(self._broker.get_bars(chunk, start, today)) daily_bars_dict = bars_to_enrichment_format(raw_bars) # Add synthetic today row (yesterday's close as placeholder) so # enrich_daily_bars() produces a keyed entry for date_str for sym, bars in daily_bars_dict.items(): if bars: last = bars[-1] if last["date"] < date_str: daily_bars_dict[sym] = bars + [{ "date": date_str, "open": last["close"], "high": last["close"], "low": last["close"], "close": last["close"], "volume": 0, }] from libs.intraday.features import enrich_daily_bars self._enrichment = enrich_daily_bars(daily_bars_dict, [date_str]) self._daily_bars = daily_bars_dict # ── Step 2: Fetch 5-min intraday bars for full universe (ORB window) ── market_open = dt.datetime(today.year, today.month, today.day, 9, 30, tzinfo=_ET) orb_end = market_open + dt.timedelta(minutes=self._params.orb_minutes + 5) fetch_end = dt.datetime.now(_ET).replace(second=0, microsecond=0) if fetch_end < orb_end: fetch_end = orb_end intraday_raw: dict[str, list[dict]] = {} chunk_size = 100 for i in range(0, len(tickers), chunk_size): chunk = tickers[i : i + chunk_size] chunk_bars = self._broker.get_intraday_bars( chunk, start=market_open, end=fetch_end, timeframe_minutes=5, ) intraday_raw.update(chunk_bars) bars_by_ticker = intraday_bars_to_format(intraday_raw) self._log( f"Daily bars: {len([s for s,b in raw_bars.items() if b])} tickers | " f"Intraday bars: {len(bars_by_ticker)} tickers" ) from libs.intraday.orb_simulator import compute_orb_candidates self._candidates = compute_orb_candidates( bars_by_ticker=bars_by_ticker, date_str=date_str, params=self._params, enrichment=self._enrichment, ) # Save candidates to DB for cand in self._candidates: orb_bar = cand["orb_bar"] direction = cand["direction"] breakout_level = orb_bar["high"] if direction == "bullish" else orb_bar["low"] row = ORBCandidateRow( session_id=self._session.session_id, date=date_str, ticker=cand["ticker"], direction=direction, orb_high=orb_bar["high"], orb_low=orb_bar["low"], breakout_level=breakout_level, atr=cand["atr"], rvol=cand["rvol"], gap_pct=cand["gap_pct"], composite_score=cand["score"], ) self._state.save_candidate(row) # Keep as pending (not yet filled) self._pending_cands = list(self._candidates) n_long = sum(1 for c in self._candidates if c["direction"] == "bullish") n_short = sum(1 for c in self._candidates if c["direction"] == "bearish") self._log( f"ORB candidates: {len(self._candidates)} " f"(long={n_long}, short={n_short})" ) self._state.update_daily_state( self._session.session_id, date_str, phase="breakout" ) return { "universe_size": len(tickers), "orb_candidates": len(self._candidates), "long": n_long, "short": n_short, } # ── Phase 3: Breakout Check ─────────────────────────────────────────────── def run_breakout_check(self, date_str: str) -> dict[str, Any]: """Check for breakouts and place orders for unfilled candidates. Called every 5 minutes from 9:45 to 10:15 ET (7 checks total). """ self._date_str = date_str # Reload state if engine was recreated (e.g., server restart) if not self._pending_cands and not self._candidates: self._pending_cands = self._rebuild_pending_candidates(date_str) if not self._pending_cands: return {"checked": 0, "filled": 0, "remaining": 0} daily_state = self._state.get_daily_state( self._session.session_id, date_str ) if daily_state.kill_switch: self._log("Kill switch active — skipping breakout check") return {"checked": 0, "filled": 0, "remaining": 0, "kill_switch": True} equity = self._get_equity() # Fetch real-time snapshots for pending candidates via Oracle API from libs.oracle_client.alpaca import get_snapshots tickers = [c["ticker"] for c in self._pending_cands] snapshots = get_snapshots(tickers) filled_count = 0 still_pending = [] for cand in self._pending_cands: ticker = cand["ticker"] direction = cand["direction"] orb_bar = cand["orb_bar"] atr = cand["atr"] score = cand["score"] rvol = cand["rvol"] breakout_level = orb_bar["high"] if direction == "bullish" else orb_bar["low"] # Check if already traded today open_positions = self._state.get_open_positions( self._session.session_id, date_str ) if any(p.ticker == ticker for p in open_positions): self._state.update_candidate_status( self._session.session_id, date_str, ticker, "filled" ) continue # Check breakout using real-time snapshot price snap = snapshots.get(ticker) if snap is None or snap.price is None: still_pending.append(cand) continue current_price = snap.price broke_out = ( (direction == "bullish" and current_price >= breakout_level) or (direction == "bearish" and current_price <= breakout_level) ) if not broke_out: still_pending.append(cand) continue # Breakout detected — compute position size and place order stop_distance = atr * self._params.atr_stop_multiplier if stop_distance <= 0: still_pending.append(cand) continue risk_dollars = equity * self._params.risk_per_trade_pct shares_from_risk = risk_dollars / stop_distance entry_price_est = max(breakout_level, current_price) max_shares_by_capital = (equity * self._params.max_position_pct) / entry_price_est shares = int(min(shares_from_risk, max_shares_by_capital)) if shares <= 0: self._log(f" {ticker}: shares=0 after sizing — skipping") self._state.update_candidate_status( self._session.session_id, date_str, ticker, "cancelled" ) continue # Check buying power try: acct = self._broker.get_account() if acct.buying_power < shares * entry_price_est: self._log(f" {ticker}: insufficient buying power — skipping") self._state.update_candidate_status( self._session.session_id, date_str, ticker, "cancelled" ) continue except Exception as e: self._log(f" {ticker}: account check error: {e}") # Place order try: if direction == "bullish": order = self._broker.submit_market_buy(ticker, shares) else: order = self._broker.submit_market_sell(ticker, shares) self._log( f" {ticker}: {direction} breakout → {shares} shares " f"(order {order.id})" ) except Exception as e: self._log(f" {ticker}: order failed: {e}") still_pending.append(cand) continue # Wait for fill (poll up to 30s) fill_price = entry_price_est order_rejected = False import time for _ in range(6): time.sleep(5) try: filled_order = self._broker.get_order(order.id) if filled_order.status == "filled" and filled_order.filled_avg_price: fill_price = filled_order.filled_avg_price break if filled_order.status in ("cancelled", "rejected", "expired"): self._log(f" {ticker}: order {filled_order.status} — no position created") order_rejected = True break except Exception: pass if order_rejected: self._state.update_candidate_status( self._session.session_id, date_str, ticker, "cancelled" ) continue # Record position initial_stop = ( fill_price - stop_distance if direction == "bullish" else fill_price + stop_distance ) pos = ORBPositionRow( session_id=self._session.session_id, date=date_str, ticker=ticker, direction="long" if direction == "bullish" else "short", entry_price=fill_price, entry_time=dt.datetime.now(_ET).isoformat(), shares=shares, orb_high=orb_bar["high"], orb_low=orb_bar["low"], atr_at_entry=atr, stop_distance=stop_distance, current_stop=initial_stop, peak_price=fill_price, rvol=rvol, composite_score=score, order_id=order.id, ) self._state.save_position(pos) self._state.update_candidate_status( self._session.session_id, date_str, ticker, "filled" ) filled_count += 1 # Check kill switches daily_state = self._state.get_daily_state( self._session.session_id, date_str ) if daily_state.kill_switch: self._log("Kill switch triggered — stopping breakout monitoring") break self._pending_cands = still_pending self._log( f"Breakout check: filled={filled_count}, remaining={len(still_pending)}" ) return { "checked": len(tickers), "filled": filled_count, "remaining": len(still_pending), } # ── Phase 4: Stop Check (90-min checkpoints) ────────────────────────────── def run_stop_check(self, date_str: str) -> dict[str, Any]: """Evaluate stops for all open positions using 90-min aggregated bars. Called at 11:10, 12:40, 14:10, 15:40 ET. Stop logic mirrors orb_simulator.py:477-580 exactly. """ self._date_str = date_str positions = self._state.get_open_positions(self._session.session_id, date_str) if not positions: return {"positions_checked": 0, "stops_hit": 0} daily_state = self._state.get_daily_state( self._session.session_id, date_str ) if daily_state.kill_switch: return {"positions_checked": 0, "stops_hit": 0, "kill_switch": True} today = dt.date.fromisoformat(date_str) market_open = dt.datetime(today.year, today.month, today.day, 9, 30, tzinfo=_ET) now_et = dt.datetime.now(_ET) tickers = [p.ticker for p in positions] bars_raw = self._broker.get_intraday_bars( tickers, start=market_open, end=now_et, timeframe_minutes=5, ) from libs.intraday.orb_simulator import _aggregate_bars from libs.intraday.simulator import _parse_ts, filter_market_hours group_size = self._params.sim_bar_minutes // 5 stops_hit = 0 equity = self._get_equity() for pos in positions: ticker = pos.ticker all_bars = bars_raw.get(ticker, []) mkt_bars = filter_market_hours(all_bars) if not mkt_bars: continue # Filter bars after entry time entry_ts = _parse_ts(pos.entry_time) post_entry = [b for b in mkt_bars if _parse_ts(b["timestamp"]) > entry_ts] if not post_entry: continue # Aggregate to sim_bar_minutes (e.g., 90-min) agg_bars = _aggregate_bars(post_entry, group_size) # Run stop management on each aggregated bar current_stop = pos.current_stop peak_price = pos.peak_price trailing_active = pos.trailing_active stop_distance = pos.stop_distance atr = pos.atr_at_entry exit_bar = None exit_reason = "close" use_atr_trail = self._params.trailing_stop_atr_multiplier > 0 for bar in agg_bars: bar_high = bar["high"] bar_low = bar["low"] if pos.direction == "long": peak_price = max(peak_price, bar_high) current_r = (bar_high - pos.entry_price) / stop_distance if stop_distance > 0 else 0 if current_r >= self._params.breakeven_at_r and current_stop < pos.entry_price: current_stop = pos.entry_price if current_r >= self._params.trailing_at_r: trailing_active = True # Check stop hit BEFORE updating trailing if bar_low <= current_stop: exit_bar = bar exit_reason = "trailing_stop" if trailing_active else "stop_loss" break # Update trailing AFTER stop check if trailing_active: if use_atr_trail: candidate = peak_price - atr * self._params.trailing_stop_atr_multiplier else: candidate = max(bar_low, current_stop) if candidate > current_stop: current_stop = candidate else: # short peak_price = min(peak_price, bar_low) current_r = (pos.entry_price - bar_low) / stop_distance if stop_distance > 0 else 0 if current_r >= self._params.breakeven_at_r and current_stop > pos.entry_price: current_stop = pos.entry_price if current_r >= self._params.trailing_at_r: trailing_active = True if bar_high >= current_stop: exit_bar = bar exit_reason = "trailing_stop" if trailing_active else "stop_loss" break if trailing_active: if use_atr_trail: candidate = peak_price + atr * self._params.trailing_stop_atr_multiplier else: candidate = min(bar_high, current_stop) if candidate < current_stop: current_stop = candidate # Update DB stop levels self._state.update_position_stop( self._session.session_id, date_str, ticker, current_stop, peak_price, trailing_active, ) if exit_bar: # Close position — use qty so only this session's shares are closed # (other sessions may hold the same ticker in the same Alpaca account). exit_price = current_stop # fallback if fill poll fails try: import time close_order = self._broker.close_position(ticker, qty=int(pos.shares)) # Poll for actual broker fill price (captures gap-through losses) for _ in range(4): time.sleep(3) try: o = self._broker.get_order(close_order.id) if o.filled_avg_price: exit_price = o.filled_avg_price break except Exception: pass self._log( f" {ticker}: stop hit ({exit_reason}) @ {exit_price:.2f}" ) except Exception as e: self._log(f" {ticker}: close error: {e}") self._record_trade(pos, exit_price, exit_bar["timestamp"], exit_reason, equity) stops_hit += 1 # Update daily kill switches loss = (exit_price - pos.entry_price) * pos.shares if pos.direction == "short": loss = (pos.entry_price - exit_price) * pos.shares if loss < 0: new_cum_loss = daily_state.cumulative_loss + abs(loss) new_stops = daily_state.stops_hit + 1 kill = ( new_cum_loss >= equity * self._params.daily_max_loss_pct or new_stops >= self._params.max_stops_per_day ) self._state.update_daily_state( self._session.session_id, date_str, cumulative_loss=new_cum_loss, stops_hit=new_stops, kill_switch=kill, ) daily_state = self._state.get_daily_state( self._session.session_id, date_str ) if kill: self._log("Kill switch triggered!") break self._log(f"Stop check: {len(positions)} positions, {stops_hit} stops hit") return {"positions_checked": len(positions), "stops_hit": stops_hit} # ── Phase 5: EOD Exit ───────────────────────────────────────────────────── def run_eod_exit(self, date_str: str) -> dict[str, Any]: """Close all remaining open positions at 15:55 ET.""" self._date_str = date_str self._state.update_daily_state( self._session.session_id, date_str, phase="eod_exit" ) positions = self._state.get_open_positions(self._session.session_id, date_str) if not positions: self._log("EOD: no open positions") return {"closed": 0} equity = self._get_equity() closed = 0 now_str = dt.datetime.now(_ET).isoformat() for pos in positions: try: # Use qty so only this session's shares are closed close_order = self._broker.close_position(pos.ticker, qty=int(pos.shares)) import time exit_price = pos.entry_price for _ in range(4): time.sleep(3) try: o = self._broker.get_order(close_order.id) if o.filled_avg_price: exit_price = o.filled_avg_price break except Exception: pass self._record_trade(pos, exit_price, now_str, "close", equity) self._log(f" EOD close: {pos.ticker} @ {exit_price:.2f}") closed += 1 except Exception as e: self._log(f" EOD close error {pos.ticker}: {e}") # Mark as closed in DB anyway to prevent zombie positions self._state.close_position_record( self._session.session_id, date_str, pos.ticker ) # Cancel any unfilled breakout orders for cand in self._pending_cands: self._state.update_candidate_status( self._session.session_id, date_str, cand["ticker"], "timeout" ) return {"closed": closed} # ── Phase 6: Post-close ─────────────────────────────────────────────────── def run_post_close(self, date_str: str) -> dict[str, Any]: """Record daily equity snapshot and finalize day.""" self._date_str = date_str self._state.update_daily_state( self._session.session_id, date_str, phase="done" ) trades_today = self._state.list_trades(self._session.session_id) today_trades = [t for t in trades_today if t["date"] == date_str] daily_pnl = sum(t["pnl"] for t in today_trades) stops_hit = sum(1 for t in today_trades if t["exit_reason"] in ("stop_loss", "trailing_stop")) prev_equity = self._get_equity() if prev_equity is None: prev_equity = self._session.initial_equity new_equity = max(prev_equity + daily_pnl, 0.01) # Drawdown peak_equity = self._state.get_peak_equity( self._session.session_id, self._session.initial_equity ) drawdown_pct = ((new_equity - peak_equity) / peak_equity * 100) if peak_equity > 0 else 0.0 snap = ORBDailySnapshotRow( session_id=self._session.session_id, date=date_str, equity=new_equity, daily_pnl=daily_pnl, total_pnl=new_equity - self._session.initial_equity, trades_taken=len(today_trades), stops_hit=stops_hit, drawdown_pct=drawdown_pct, ) self._state.save_daily_snapshot(snap) self._log( f"Post-close: equity={new_equity:.2f}, pnl={daily_pnl:+.2f}, " f"trades={len(today_trades)}, stops={stops_hit}" ) return { "equity": new_equity, "daily_pnl": daily_pnl, "trades": len(today_trades), "stops_hit": stops_hit, "drawdown_pct": drawdown_pct, } # ── Helpers ─────────────────────────────────────────────────────────────── def _record_trade( self, pos: ORBPositionRow, exit_price: float, exit_time: str, exit_reason: str, equity: float, ) -> None: """Record a completed trade in the DB and close the position record.""" if pos.direction == "long": pnl = (exit_price - pos.entry_price) * pos.shares else: pnl = (pos.entry_price - exit_price) * pos.shares r_multiple = ( pnl / (pos.stop_distance * pos.shares) if pos.stop_distance > 0 and pos.shares > 0 else 0.0 ) trade = ORBTradeRow( trade_id=str(uuid.uuid4())[:12], session_id=self._session.session_id, date=pos.date, ticker=pos.ticker, direction=pos.direction, entry_price=pos.entry_price, exit_price=exit_price, entry_time=pos.entry_time, exit_time=exit_time, shares=pos.shares, pnl=round(pnl, 4), r_multiple=round(r_multiple, 3), exit_reason=exit_reason, atr_at_entry=pos.atr_at_entry, rvol=pos.rvol, composite_score=pos.composite_score, ) self._state.save_trade(trade) self._state.close_position_record( self._session.session_id, pos.date, pos.ticker ) def _rebuild_pending_candidates(self, date_str: str) -> list[dict]: """Reconstruct pending candidates from DB (after server restart).""" db_cands = self._state.list_candidates(self._session.session_id, date_str) open_positions = self._state.get_open_positions( self._session.session_id, date_str ) filled_tickers = {p.ticker for p in open_positions} result = [] for c in db_cands: if c["status"] != "pending": continue if c["ticker"] in filled_tickers: continue # Reconstruct minimal candidate dict for breakout check result.append({ "ticker": c["ticker"], "direction": c["direction"], "orb_bar": { "high": c["orb_high"], "low": c["orb_low"], "timestamp": "", "open": 0, "close": 0, "volume": 0, }, "atr": c["atr"], "rvol": c["rvol"], "gap_pct": c["gap_pct"], "score": c["composite_score"], }) return result # ── Engine factory ──────────────────────────────────────────────────────────── def make_orb_engine( session: Any, db_path: str | None = None, broker_override: Any = None, ) -> ORBTradingEngine: """Create an ORBTradingEngine for the given session. broker_override: pass a MockORBBroker (or any duck-typed broker) to avoid real Alpaca API calls during testing. """ import yaml from apps.orb_trader.state import ORBStateManager from libs.intraday.domain import IntradayConfig if broker_override is not None: broker = broker_override else: from apps.paper_trader.alpaca_broker import AlpacaBroker broker = AlpacaBroker.from_env() state = ORBStateManager(db_path) # Load strategy params from YAML config with open(session.config_path) as f: raw = yaml.safe_load(f) # Strip _meta and other non-model keys config_data = {k: v for k, v in raw.items() if not k.startswith("_")} config = IntradayConfig(**config_data) params = config.orb_strategy if params is None: from libs.intraday.domain import ORBStrategyParams params = ORBStrategyParams() # Store universe source on params for runtime use params._universe_source = config.universe.source params._universe_symbols_file = config.universe.symbols_file return ORBTradingEngine(session=session, broker=broker, state=state, params=params)