diff --git a/apps/orb_trader/engine.py b/apps/orb_trader/engine.py new file mode 100644 index 0000000..fa7a02c --- /dev/null +++ b/apps/orb_trader/engine.py @@ -0,0 +1,790 @@ +"""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) diff --git a/apps/web/orb_trading_service.py b/apps/web/orb_trading_service.py new file mode 100644 index 0000000..92ad6e5 --- /dev/null +++ b/apps/web/orb_trading_service.py @@ -0,0 +1,580 @@ +"""ORB paper trading in-process service layer. + +Provides: + - ORBAutoScheduler: asyncio-based intraday scheduler + - orb_auto_scheduler: module-level singleton + - State persistence to .orb_auto_state.json + +Schedule is computed dynamically from strategy params (orb_minutes, +order_timeout_minutes, sim_bar_minutes) rather than being hardcoded. + +Daily event sequence: + 9:30 + orb_minutes → orb_detect (daily bars + 5-min ORB bars → candidates) + orb_end + 1 min → breakout (Oracle snapshot price check, every 1 min) + ...repeat until order_timeout_minutes elapses + orb_end + N×sim_bar → stop_check (fetch 5-min bars, aggregate, check stops) + ...repeat until 15:50 + 15:55 → eod_exit + 16:00 → post_close +""" +from __future__ import annotations + +import asyncio +import datetime as dt +import json +import logging +import traceback +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from zoneinfo import ZoneInfo + +log = logging.getLogger(__name__) + +_TZ_ET = ZoneInfo("America/New_York") +_TZ_PHOENIX = ZoneInfo("America/Phoenix") # UTC-7 always (no DST) +_DEFAULT_DB = "data/paper/orb.db" + +_MARKET_OPEN = dt.time(9, 30) # ET +_EOD_EXIT = dt.time(15, 55) # ET +_POST_CLOSE = dt.time(16, 0) # ET + + +# ── Dynamic schedule builder ────────────────────────────────────────────────── + +def build_schedule( + date: dt.date, + orb_minutes: int = 10, + order_timeout_minutes: int = 45, + sim_bar_minutes: int = 90, +) -> list[dict[str, Any]]: + """Build today's event list from strategy parameters. + + Args: + date: The trading date. + orb_minutes: Length of the ORB window (e.g. 10 → window is 9:30–9:40). + order_timeout_minutes: How many minutes after ORB end to keep looking for + breakouts (one check per minute). + sim_bar_minutes: Interval between stop-management checks after breakout + window closes (matches the sim_bar_minutes strategy param). + + Returns: + List of event dicts sorted chronologically. Each dict has: + name, label, kind, et_dt (aware datetime in ET) + + Note: breakout checks run every sim_bar_minutes (matching the backtest bar + aggregation interval), up to order_timeout_minutes after *market open* + (not after ORB end) — consistent with orb_simulator.py timeout semantics. + """ + mkt_open = dt.datetime(date.year, date.month, date.day, 9, 30, tzinfo=_TZ_ET) + orb_end = mkt_open + dt.timedelta(minutes=orb_minutes) + eod = dt.datetime(date.year, date.month, date.day, 15, 55, tzinfo=_TZ_ET) + post = dt.datetime(date.year, date.month, date.day, 16, 0, tzinfo=_TZ_ET) + + events: list[dict[str, Any]] = [] + + # ── ORB window monitoring: 9:30 → orb_end-1, every minute (no-op) ───────── + for i in range(orb_minutes): + t = mkt_open + dt.timedelta(minutes=i) + events.append({ + "name": f"orb_monitor_{i + 1}", + "label": f"ORB 윈도우 ({t.strftime('%H:%M')} ET, {i + 1}/{orb_minutes}분)", + "kind": "orb_monitor", + "et_dt": t, + }) + + # ── ORB detection at window close (fetch bars + rank candidates) ────────── + events.append({ + "name": "orb_detect", + "label": f"ORB 감지 ({orb_end.strftime('%H:%M')} ET)", + "kind": "orb_detect", + "et_dt": orb_end, + }) + + # ── Breakout checks: every sim_bar_minutes from first bar close to timeout ── + # Timeout is measured from market open (matches orb_simulator.py semantics). + timeout_dt = mkt_open + dt.timedelta(minutes=order_timeout_minutes) + i = 1 + t = orb_end + dt.timedelta(minutes=sim_bar_minutes) + while t <= timeout_dt: + if t >= eod: + break + events.append({ + "name": f"breakout_{i}", + "label": f"브레이크아웃 ({t.strftime('%H:%M')} ET, +{i * sim_bar_minutes}분)", + "kind": "breakout", + "et_dt": t, + }) + t += dt.timedelta(minutes=sim_bar_minutes) + i += 1 + + # ── Stop checks: every sim_bar_minutes from orb_end ─────────────────────── + t = orb_end + dt.timedelta(minutes=sim_bar_minutes) + idx = 1 + while t < eod: + events.append({ + "name": f"stop_{idx}", + "label": f"스톱 체크 ({t.strftime('%H:%M')} ET, +{sim_bar_minutes}분)", + "kind": "stop_check", + "et_dt": t, + }) + t += dt.timedelta(minutes=sim_bar_minutes) + idx += 1 + + # ── EOD + post-close ────────────────────────────────────────────────────── + events.append({"name": "eod_exit", "label": "EOD 청산 (15:55 ET)", + "kind": "eod_exit", "et_dt": eod}) + events.append({"name": "post_close", "label": "마감 후 스냅샷 (16:00 ET)", + "kind": "post_close", "et_dt": post}) + + return sorted(events, key=lambda e: e["et_dt"]) + + +def _load_session_params(db_path: str, session_name: str) -> dict[str, Any]: + """Load strategy params (orb_minutes, sim_bar_minutes, order_timeout_minutes) + from the session's config YAML. Returns defaults on any error. + """ + defaults = {"orb_minutes": 10, "sim_bar_minutes": 90, "order_timeout_minutes": 45} + try: + import yaml + from apps.orb_trader.state import ORBStateManager + session = ORBStateManager(db_path).get_session(session_name) + if session is None: + return defaults + raw = yaml.safe_load(Path(session.config_path).read_text()) or {} + orb = raw.get("orb_strategy", {}) + return { + "orb_minutes": orb.get("orb_minutes", defaults["orb_minutes"]), + "sim_bar_minutes": orb.get("sim_bar_minutes", defaults["sim_bar_minutes"]), + "order_timeout_minutes": orb.get("order_timeout_minutes", defaults["order_timeout_minutes"]), + } + except Exception: + return defaults + + + +# ── State file helpers ──────────────────────────────────────────────────────── + +def _state_file_path(db_path: str) -> Path: + return Path(db_path).parent / ".orb_auto_state.json" + + +def load_orb_saved_state(db_path: str) -> dict[str, Any] | None: + sf = _state_file_path(db_path) + if sf.exists(): + try: + return json.loads(sf.read_text()) + except Exception: + pass + return None + + +# ── ORBAutoScheduler ────────────────────────────────────────────────────────── + +class ORBAutoScheduler: + """In-process ORB intraday auto-trader. + + Schedule is built dynamically each trading day from the sessions' strategy + params (orb_minutes, order_timeout_minutes, sim_bar_minutes). + """ + + def __init__(self) -> None: + self._task: asyncio.Task | None = None # type: ignore[type-arg] + self._sessions: list[str] = [] + self._db_path: str = _DEFAULT_DB + self._dry_run: bool = False + self._log_lines: list[str] = [] + self._completed: set[str] = set() + self._today_schedule: list[dict[str, Any]] = [] # built each trading day + self._engines: dict[str, Any] = {} + self._engine_date: str = "" + + @property + def running(self) -> bool: + return self._task is not None and not self._task.done() + + def _log_file_path(self) -> Path: + return Path(self._db_path).parent / "orb_scheduler.log" + + def _load_persisted_log(self, max_lines: int = 3000) -> list[str]: + """Load existing log lines from file (survives server restarts). + Trims to last max_lines if the file has grown too large. + """ + try: + lf = self._log_file_path() + if lf.exists(): + lines = lf.read_text(encoding="utf-8").splitlines() + if len(lines) > max_lines: + # Keep last max_lines; rewrite trimmed file + lines = lines[-max_lines:] + try: + lf.write_text("\n".join(lines) + "\n", encoding="utf-8") + except Exception: + pass + return lines + except Exception: + pass + return [] + + def start(self, sessions: list[str], db_path: str, dry_run: bool = False) -> None: + if self.running: + raise RuntimeError("ORBAutoScheduler already running") + self._sessions = sessions + self._db_path = db_path + self._dry_run = dry_run + self._log_lines = self._load_persisted_log() # restore previous logs + self._completed = set() + self._today_schedule = [] + self._engines = {} + self._engine_date = "" + self._save_state() + self._task = asyncio.create_task(self._run_loop()) + + def stop(self) -> None: + if self._task and not self._task.done(): + self._task.cancel() + self._clear_state() + + def shutdown(self) -> None: + """Server shutdown — cancel task but keep state for auto-restart.""" + if self._task and not self._task.done(): + self._task.cancel() + + def get_log(self, lines: int = 200) -> str: + return "\n".join(self._log_lines[-lines:]) + + def get_log_tail(self, lines: int = 80) -> list[str]: + return self._log_lines[-lines:] + + @property + def log_line_count(self) -> int: + return len(self._log_lines) + + def get_status(self) -> dict[str, Any]: + now_et = self._now_et() + today = now_et.date() + schedule_view = [] + for ev in self._today_schedule: + ev_dt = ev["et_dt"] + past = ev_dt <= now_et + wait = (ev_dt - now_et).total_seconds() + schedule_view.append({ + "name": ev["name"], + "kind": ev["kind"], + "session": ev.get("session", ""), + "label": ev["label"], + "et_time": ev_dt.strftime("%H:%M ET"), + "et_iso": ev_dt.isoformat(), + "past": past, + "done": ev["name"] in self._completed, + "wait_secs": max(0, wait), + }) + return { + "running": self.running, + "sessions": self._sessions, + "dry_run": self._dry_run, + "schedule": schedule_view, + "log_tail": list(self._log_lines), # full log, not truncated + "log_line_count": self.log_line_count, + } + + # ── State persistence ────────────────────────────────────────────────────── + + def _save_state(self) -> None: + try: + _state_file_path(self._db_path).write_text(json.dumps({ + "running": True, + "sessions": self._sessions, + "dry_run": self._dry_run, + "db_path": self._db_path, + })) + except Exception: + pass + + def _clear_state(self) -> None: + try: + sf = _state_file_path(self._db_path) + if sf.exists(): + sf.unlink() + except Exception: + pass + + # ── Helpers ──────────────────────────────────────────────────────────────── + + def _log(self, msg: str) -> None: + ts = datetime.now(tz=_TZ_PHOENIX).strftime("%H:%M MST") + line = f"{ts} {msg}" + self._log_lines.append(line) + log.info("[ORBScheduler] %s", msg) + # Persist to file so logs survive server restarts + try: + with self._log_file_path().open("a", encoding="utf-8") as f: + f.write(line + "\n") + except Exception: + pass + + def _now_et(self) -> dt.datetime: + return dt.datetime.now(tz=_TZ_ET) + + def _is_trading_day(self, date: dt.date) -> bool: + try: + from libs.common.time_utils import is_trading_day + return is_trading_day(date) + except Exception: + return date.weekday() < 5 + + def _next_trading_day(self, from_date: dt.date) -> dt.date: + check = from_date + dt.timedelta(days=1) + for _ in range(14): + if self._is_trading_day(check): + return check + check += dt.timedelta(days=1) + raise RuntimeError("No trading day found in next 14 days") + + @staticmethod + def _fmt_countdown(seconds: float) -> str: + if seconds <= 0: + return "now" + h = int(seconds // 3600) + m = int((seconds % 3600) // 60) + s = int(seconds % 60) + if h > 0: + return f"{h}h {m:02d}m" + if m > 0: + return f"{m}m {s:02d}s" + return f"{s}s" + + def _get_active_sessions(self) -> list[str]: + try: + from apps.orb_trader.state import ORBStateManager + return [ + s.session_name + for s in ORBStateManager(self._db_path).list_sessions() + if s.status == "active" + ] + except Exception: + return [] + + def _build_today_schedule(self, date: dt.date, sessions: list[str]) -> list[dict[str, Any]]: + """Build per-session schedules and merge into one sorted timeline. + + Each event is tagged with its session so the loop can dispatch + the right engine for the right session at the right time. + Sessions that no longer exist in the DB are skipped with a warning. + """ + from apps.orb_trader.state import ORBStateManager + state_mgr = ORBStateManager(self._db_path) + + combined: list[dict[str, Any]] = [] + valid_sessions: list[str] = [] + for session_name in sessions: + # Verify session still exists in DB + session_obj = state_mgr.get_session(session_name) + if session_obj is None: + self._log(f" WARNING: session '{session_name}' not found in DB — skipping") + continue + valid_sessions.append(session_name) + params = _load_session_params(self._db_path, session_name) + events = build_schedule(date, **params) + for ev in events: + ev["session"] = session_name + ev["name"] = f"{session_name}:{ev['name']}" + combined.extend(events) + self._log( + f" {session_name}: orb={params['orb_minutes']}min, " + f"timeout={params['order_timeout_minutes']}min, " + f"bar={params['sim_bar_minutes']}min → {len(events)} events" + ) + + if not valid_sessions and sessions: + # All specified sessions are gone — fall back to all active sessions + active = self._get_active_sessions() + self._log( + f" All specified sessions missing; falling back to active sessions: " + f"{', '.join(active) or 'none'}" + ) + return self._build_today_schedule(date, active) + + combined.sort(key=lambda e: (e["et_dt"], e["session"])) + self._log(f"Total schedule: {len(combined)} events across {len(valid_sessions)} session(s)") + return combined + + def _get_or_create_engine(self, session_name: str, date_str: str) -> Any: + if self._engine_date != date_str: + self._engines = {} + self._engine_date = date_str + + if session_name not in self._engines: + try: + from apps.orb_trader.state import ORBStateManager + from apps.orb_trader.engine import make_orb_engine + state_mgr = ORBStateManager(self._db_path) + session = state_mgr.get_session(session_name) + if session is None: + return None + self._engines[session_name] = make_orb_engine(session, self._db_path) + except Exception as exc: + self._log(f" ERROR creating engine for {session_name}: {exc}") + return None + + return self._engines[session_name] + + # ── Engine operation runners ─────────────────────────────────────────────── + + async def _run_trading(self, kind: str, sessions: list[str], date_str: str) -> None: + # ORB 윈도우 모니터링은 no-op (범위 형성 중, 장중 데이터는 orb_detect에서 일괄 fetch) + if kind == "orb_monitor": + self._log(f" ORB 윈도우 모니터링 중...") + return + + from apps.orb_trader.state import ORBStateManager + state_mgr = ORBStateManager(self._db_path) + + for session_name in sessions: + session = state_mgr.get_session(session_name) + if session is None or session.status != "active": + self._log(f" {session_name}: skipped (not active)") + continue + if self._dry_run: + self._log(f" [DRY] {kind} — {session_name}") + continue + + engine = self._get_or_create_engine(session_name, date_str) + if engine is None: + continue + + self._log(f" ▶ {kind} — {session_name}") + try: + def _run_sync(e=engine, k=kind, d=date_str) -> dict[str, Any]: + import asyncio as _asyncio + loop = _asyncio.new_event_loop() + _asyncio.set_event_loop(loop) + try: + if k == "orb_detect": + return e.run_orb_detection(d) + elif k == "breakout": + return e.run_breakout_check(d) + elif k == "stop_check": + return e.run_stop_check(d) + elif k == "eod_exit": + return e.run_eod_exit(d) + elif k == "post_close": + return e.run_post_close(d) + return {} + finally: + loop.close() + + summary = await asyncio.to_thread(_run_sync) + self._log(f" ✓ {session_name}: {summary}") + except Exception as exc: + tb = traceback.format_exc() + self._log(f" ERROR {session_name}: {exc}") + log.error("ORB engine error: %s\n%s", exc, tb) + + # ── Main scheduler loop ──────────────────────────────────────────────────── + + async def _run_loop(self) -> None: + resolved = self._sessions or self._get_active_sessions() + if not resolved: + self._log("No active ORB sessions. Stopping.") + return + + self._log(f"ORB auto-scheduler started — sessions: {', '.join(resolved)}") + if self._dry_run: + self._log("DRY RUN — orders will not be placed") + + last_schedule_date: dt.date | None = None + + try: + while True: + now_et = self._now_et() + today = now_et.date() + date_str = today.isoformat() + + # ── New trading day: rebuild schedule ───────────────────────── + if last_schedule_date != today: + self._completed.clear() + last_schedule_date = today + + if not self._sessions: + fresh = self._get_active_sessions() + if set(fresh) != set(resolved): + self._log(f"Sessions refreshed: {', '.join(fresh) or 'none'}") + resolved = fresh + + self._log( + f"━━━ {today.strftime('%a %Y-%m-%d')} " + f"— sessions: {', '.join(resolved) or 'none'} ━━━" + ) + + if not self._is_trading_day(today): + self._today_schedule = [] + next_td = self._next_trading_day(today) + self._log(f"Non-trading day. Next: {next_td}") + else: + self._today_schedule = self._build_today_schedule(today, resolved) + # Skip events already past + for ev in self._today_schedule: + if ev["et_dt"] <= now_et: + self._completed.add(ev["name"]) + self._log(f"Past (skipped): {ev['label']}") + + if not self._is_trading_day(today): + await asyncio.sleep(1800) + continue + + if not resolved: + await asyncio.sleep(300) + continue + + pending = [ev for ev in self._today_schedule if ev["name"] not in self._completed] + + if not pending: + next_td = self._next_trading_day(today) + # Wake up just before the ORB detect event of next trading day + # (schedule isn't built yet, assume market open 9:30 ET) + wake_et = dt.datetime( + next_td.year, next_td.month, next_td.day, 9, 25, tzinfo=_TZ_ET + ) + wait = (wake_et - now_et).total_seconds() + self._log( + f"All done today. Sleeping until " + f"{wake_et.strftime('%I:%M %p ET')} on {next_td} " + f"({self._fmt_countdown(wait)})" + ) + await asyncio.sleep(min(wait, 3600)) + continue + + next_ev = pending[0] + next_et = next_ev["et_dt"] + wait = (next_et - now_et).total_seconds() + + if wait > 90: + label = next_ev.get("label", next_ev["name"]) + sess = next_ev.get("session", "") + self._log(f"Next: [{sess}] {label} — {self._fmt_countdown(wait)}") + await asyncio.sleep(min(wait - 60, 600)) + continue + + if wait > 0: + await asyncio.sleep(wait) + + # ── Execute all events at this time slot (per session) ───────── + batch = [ev for ev in pending if ev["et_dt"] == next_et] + for ev in batch: + session = ev.get("session", "") + self._log(f"▶ [{session}] {ev['label']}") + try: + await self._run_trading(ev["kind"], [session] if session else resolved, date_str) + except Exception as exc: + self._log(f" ERROR [{session}]: {exc}") + self._completed.add(ev["name"]) + + except asyncio.CancelledError: + self._log("ORB auto-scheduler stopped.") + raise + + +# ── Module-level singleton ──────────────────────────────────────────────────── + +orb_auto_scheduler = ORBAutoScheduler()