You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1131 lines
48 KiB
Python
1131 lines
48 KiB
Python
"""ORB Paper Trading Engine.
|
|
|
|
One engine instance is created per session per trading day.
|
|
The scheduler calls phase methods in order:
|
|
1. run_pre_screen() — 09:20 ET (daily bars + enrichment + quality filter)
|
|
2. run_orb_detection() — 09:40 ET (intraday bars → candidates)
|
|
└─ run_pre_screen 미실행 시 full fallback (daily bars도 자체 fetch)
|
|
3. run_breakout_check() — 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 copy
|
|
import datetime as dt
|
|
import logging
|
|
import time
|
|
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,
|
|
live_pre_screen,
|
|
load_universe,
|
|
)
|
|
from apps.orb_trader.state import ORBStateManager
|
|
from libs.intraday.features import enrich_daily_bars
|
|
from libs.intraday.orb_simulator import _aggregate_bars, compute_orb_candidates
|
|
from libs.intraday.simulator import _parse_ts, filter_market_hours
|
|
from libs.oracle_client.alpaca import get_snapshots
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
_ET = ZoneInfo("America/New_York")
|
|
_ACCOUNT_CIRCUIT_BREAKER_PCT = 25.0 # halt if equity drops >25% from peak
|
|
|
|
|
|
class ORBTradingEngine:
|
|
"""Intraday paper trading engine for the ORB strategy.
|
|
|
|
Holds per-day state (enrichment, candidates, date_str) as instance variables.
|
|
On server restart mid-day, state is reconstructed from the DB.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
session: Any,
|
|
broker: Any,
|
|
state: ORBStateManager,
|
|
params: Any,
|
|
log_callback: Any = None,
|
|
) -> None:
|
|
self._session = session
|
|
self._broker = broker
|
|
self._state = state
|
|
self._params = params
|
|
self._log_callback = log_callback # optional scheduler._log for UI visibility
|
|
|
|
# Live trading overrides
|
|
self._params.settlement_days = 0 # paper trading; no real T+1 settlement
|
|
self._params.slippage_bps = 0.0 # real fills, no simulated slippage
|
|
# IEX feed has ~1-3% market share vs SIP; RVOL computed from IEX ORB volume
|
|
# relative to SIP avg_daily_vol would be ~0.01-0.03 (min_rvol=1.0 would filter
|
|
# everything). Disable the threshold filter; RVOL is still used for ranking.
|
|
self._params.min_rvol = 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)
|
|
# Pre-screened tickers: None = pre_screen not yet run, [] = ran but nothing passed
|
|
self._pre_screened_tickers: list[str] | None = None
|
|
|
|
def _get_equity(self) -> float:
|
|
"""Current equity = last snapshot equity, or initial if no snapshots."""
|
|
eq = self._state.get_equity(self._session.session_id)
|
|
return eq if eq is not None else self._session.initial_equity
|
|
|
|
@staticmethod
|
|
def _last_trading_day(ref: dt.date) -> dt.date:
|
|
"""Return the most recent weekday strictly before ref.
|
|
|
|
Used for daily-bar end_date: today's bar is incomplete during market hours,
|
|
and weekend dates cause Alpaca to return 502 Bad Gateway.
|
|
"""
|
|
d = ref - dt.timedelta(days=1)
|
|
while d.weekday() >= 5: # 5=Sat, 6=Sun
|
|
d -= dt.timedelta(days=1)
|
|
return d
|
|
|
|
def _log(self, msg: str) -> None:
|
|
log.info("[ORB:%s] %s", self._session.session_name, msg)
|
|
if self._log_callback is not None:
|
|
try:
|
|
self._log_callback(f" [engine] {msg}")
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Phase 1: Pre-market Screening (daily bars + enrichment + quality filter) ─
|
|
|
|
def run_pre_screen(self, date_str: str) -> dict[str, Any]:
|
|
"""Pre-market screening: fetch daily bars, compute enrichment, filter universe.
|
|
|
|
Called at ~9:20 ET (before market open). Narrows the universe from ~971
|
|
to ~250-350 tickers using quality filters (price, ATR, dollar volume).
|
|
The expensive intraday bar fetch in run_orb_detection() then only
|
|
fetches data for pre-screened tickers.
|
|
|
|
If this method is never called (late start, failure), run_orb_detection()
|
|
falls back to the full pipeline automatically.
|
|
|
|
Returns summary dict.
|
|
"""
|
|
self._date_str = date_str
|
|
self._state.update_daily_state(
|
|
self._session.session_id, date_str, phase="pre_screen"
|
|
)
|
|
|
|
universe_source = getattr(self._params, "_universe_source", "midlarge")
|
|
universe_symbols_file = getattr(self._params, "_universe_symbols_file", None)
|
|
tickers = load_universe(universe_source, universe_symbols_file)
|
|
|
|
# Always include the regime ticker so regime/breadth filters have data
|
|
regime_ticker = getattr(self._params, "market_regime_ticker", None) or "QQQ"
|
|
fetch_tickers = list(dict.fromkeys([regime_ticker] + tickers)) # deduplicate, regime first
|
|
self._log(f"Pre-screen: {len(tickers)} tickers (+{regime_ticker}) — fetching daily bars")
|
|
|
|
today = dt.date.fromisoformat(date_str)
|
|
bars_end = self._last_trading_day(today)
|
|
start = bars_end - dt.timedelta(days=65)
|
|
|
|
raw_bars: dict[str, list] = {}
|
|
chunk_size = 200
|
|
for i in range(0, len(fetch_tickers), chunk_size):
|
|
chunk = fetch_tickers[i : i + chunk_size]
|
|
try:
|
|
raw_bars.update(self._broker.get_bars(chunk, start, bars_end))
|
|
except Exception as e:
|
|
self._log(f" WARNING: daily bars chunk {i//chunk_size+1} failed ({e}) — skipping")
|
|
|
|
daily_bars_dict = bars_to_enrichment_format(raw_bars)
|
|
|
|
# Add synthetic today row so enrich_daily_bars() produces entries for date_str
|
|
for sym, bars in daily_bars_dict.items():
|
|
if bars:
|
|
last = bars[-1]
|
|
if last["date"] < date_str:
|
|
daily_bars_dict[sym] = bars + [{
|
|
"date": date_str,
|
|
"open": last["close"], "high": last["close"],
|
|
"low": last["close"], "close": last["close"],
|
|
"volume": 0,
|
|
}]
|
|
|
|
self._enrichment = enrich_daily_bars(daily_bars_dict, [date_str])
|
|
self._daily_bars = daily_bars_dict
|
|
|
|
qualified = live_pre_screen(self._enrichment, date_str, self._params)
|
|
self._pre_screened_tickers = qualified
|
|
|
|
daily_bars_count = len([s for s, b in raw_bars.items() if b])
|
|
self._log(
|
|
f"Pre-screen 완료: {daily_bars_count} daily bars → "
|
|
f"{len(qualified)} qualified (전체 {len(tickers)}개 중)"
|
|
)
|
|
return {
|
|
"universe_size": len(tickers),
|
|
"daily_bars": daily_bars_count,
|
|
"pre_screened": len(qualified),
|
|
}
|
|
|
|
# ── Phase 2: ORB Detection (intraday bars → candidates) ──────────────────
|
|
|
|
def run_orb_detection(self, date_str: str) -> dict[str, Any]:
|
|
"""Fetch 5-min ORB bars for (pre-screened or full) universe, then compute candidates.
|
|
|
|
Called once at 9:30 + orb_minutes (e.g., 9:40 for a 10-min ORB).
|
|
If run_pre_screen() was called earlier, uses cached enrichment and
|
|
pre-screened ticker list (skips daily bars fetch). Otherwise runs
|
|
the full pipeline as a fallback.
|
|
|
|
Returns summary dict.
|
|
"""
|
|
self._date_str = date_str
|
|
self._state.update_daily_state(
|
|
self._session.session_id, date_str, phase="orb_detection"
|
|
)
|
|
|
|
# Rolling loss filter: skip day if recent N-day equity return is below threshold
|
|
roll_days = getattr(self._params, "rolling_loss_days", None)
|
|
roll_thresh = getattr(self._params, "rolling_loss_threshold", None)
|
|
if roll_days is not None and roll_thresh is not None:
|
|
snapshots = self._state.list_snapshots(self._session.session_id)
|
|
past = [s for s in snapshots if s["date"] < date_str]
|
|
if len(past) >= roll_days:
|
|
window = past[-roll_days:]
|
|
rolling_pnl = sum(s["daily_pnl"] for s in window)
|
|
sizing_base = self._session.initial_equity # daily_budget_reset mode
|
|
if sizing_base > 0 and rolling_pnl / sizing_base < roll_thresh:
|
|
self._log(
|
|
f"Rolling loss filter triggered ({rolling_pnl/sizing_base:.2%} "
|
|
f"< {roll_thresh:.2%}) — skipping today"
|
|
)
|
|
self._state.update_daily_state(
|
|
self._session.session_id, date_str, phase="done"
|
|
)
|
|
return {
|
|
"universe_size": 0,
|
|
"daily_bars": 0,
|
|
"intraday_bars": 0,
|
|
"orb_candidates": 0,
|
|
"long": 0,
|
|
"short": 0,
|
|
"skip_reason": "rolling_loss",
|
|
}
|
|
|
|
# Account-level circuit breaker: halt if equity has dropped >25% from peak
|
|
equity_now = self._get_equity()
|
|
peak_eq = self._state.get_peak_equity(
|
|
self._session.session_id, self._session.initial_equity
|
|
)
|
|
if peak_eq > 0:
|
|
account_dd_pct = (peak_eq - equity_now) / peak_eq * 100
|
|
if account_dd_pct >= _ACCOUNT_CIRCUIT_BREAKER_PCT:
|
|
self._log(
|
|
f"CIRCUIT BREAKER: account drawdown {account_dd_pct:.1f}% "
|
|
f">= {_ACCOUNT_CIRCUIT_BREAKER_PCT}% — session halted"
|
|
)
|
|
self._state.set_session_status(self._session.session_id, "paused")
|
|
self._state.update_daily_state(
|
|
self._session.session_id, date_str, phase="done"
|
|
)
|
|
return {
|
|
"universe_size": 0, "daily_bars": 0, "intraday_bars": 0,
|
|
"orb_candidates": 0, "long": 0, "short": 0,
|
|
"skip_reason": "circuit_breaker",
|
|
}
|
|
|
|
# ── Determine intraday_tickers: use pre-screen cache or fetch daily bars ─
|
|
if self._enrichment and self._pre_screened_tickers is not None:
|
|
# Pre-screen already ran — skip daily bars fetch
|
|
intraday_tickers = self._pre_screened_tickers
|
|
daily_bars_count = len([s for s, b in self._daily_bars.items() if b])
|
|
self._log(
|
|
f"Pre-screened universe 사용: {len(intraday_tickers)} tickers "
|
|
f"(daily bars 캐시됨)"
|
|
)
|
|
else:
|
|
# Fallback: full pipeline (pre_screen missed or failed)
|
|
universe_source = getattr(self._params, "_universe_source", "midlarge")
|
|
universe_symbols_file = getattr(self._params, "_universe_symbols_file", None)
|
|
tickers = load_universe(universe_source, universe_symbols_file)
|
|
regime_ticker = getattr(self._params, "market_regime_ticker", None) or "QQQ"
|
|
fetch_tickers = list(dict.fromkeys([regime_ticker] + tickers))
|
|
self._log(f"Universe: {len(tickers)} tickers (+{regime_ticker}) — fetching daily bars")
|
|
|
|
today = dt.date.fromisoformat(date_str)
|
|
bars_end = self._last_trading_day(today)
|
|
start = bars_end - dt.timedelta(days=65)
|
|
|
|
raw_bars: dict[str, list] = {}
|
|
chunk_size = 200
|
|
for i in range(0, len(fetch_tickers), chunk_size):
|
|
chunk = fetch_tickers[i : i + chunk_size]
|
|
try:
|
|
raw_bars.update(self._broker.get_bars(chunk, start, bars_end))
|
|
except Exception as e:
|
|
self._log(f" WARNING: daily bars chunk {i//chunk_size+1} failed ({e}) — skipping")
|
|
|
|
daily_bars_dict = bars_to_enrichment_format(raw_bars)
|
|
|
|
# Add synthetic today row (yesterday's close as placeholder) so
|
|
# enrich_daily_bars() produces a keyed entry for date_str
|
|
for sym, bars in daily_bars_dict.items():
|
|
if bars:
|
|
last = bars[-1]
|
|
if last["date"] < date_str:
|
|
daily_bars_dict[sym] = bars + [{
|
|
"date": date_str,
|
|
"open": last["close"], "high": last["close"],
|
|
"low": last["close"], "close": last["close"],
|
|
"volume": 0,
|
|
}]
|
|
|
|
self._enrichment = enrich_daily_bars(daily_bars_dict, [date_str])
|
|
self._daily_bars = daily_bars_dict
|
|
daily_bars_count = len([s for s, b in raw_bars.items() if b])
|
|
intraday_tickers = tickers # full universe (no pre-screen)
|
|
|
|
# ── Fetch 5-min intraday bars for (pre-screened or full) universe ──────
|
|
today = dt.date.fromisoformat(date_str)
|
|
market_open = dt.datetime(today.year, today.month, today.day, 9, 30, tzinfo=_ET)
|
|
orb_end = market_open + dt.timedelta(minutes=self._params.orb_minutes + 5)
|
|
fetch_end = dt.datetime.now(_ET).replace(second=0, microsecond=0)
|
|
if fetch_end < orb_end:
|
|
fetch_end = orb_end
|
|
|
|
intraday_raw: dict[str, list[dict]] = {}
|
|
chunk_size = 100
|
|
for i in range(0, len(intraday_tickers), chunk_size):
|
|
chunk = intraday_tickers[i : i + chunk_size]
|
|
try:
|
|
chunk_bars = self._broker.get_intraday_bars(
|
|
chunk,
|
|
start=market_open,
|
|
end=fetch_end,
|
|
timeframe_minutes=5,
|
|
)
|
|
intraday_raw.update(chunk_bars)
|
|
except Exception as e:
|
|
self._log(f" WARNING: intraday bars chunk {i//chunk_size+1} failed ({e}) — skipping")
|
|
|
|
bars_by_ticker = intraday_bars_to_format(intraday_raw)
|
|
intraday_count = len(bars_by_ticker)
|
|
self._log(
|
|
f"Daily bars: {daily_bars_count} tickers | "
|
|
f"Intraday bars: {intraday_count} tickers"
|
|
)
|
|
if daily_bars_count == 0:
|
|
self._log("WARNING: no daily bars fetched — enrichment will be empty; check Oracle/Alpaca connection")
|
|
if intraday_count == 0:
|
|
self._log("WARNING: no intraday bars fetched — zero candidates will be produced")
|
|
|
|
# Patch today_open in enrichment with actual first-bar open from intraday data.
|
|
# The pre_screen synthetic row uses prev_close as today_open (gap=0), which breaks
|
|
# market_regime_spy_threshold and breadth filters. Overwrite with real opening price.
|
|
for ticker, ticker_bars in bars_by_ticker.items():
|
|
if not ticker_bars:
|
|
continue
|
|
first_bar = ticker_bars[0]
|
|
real_open = first_bar.get("open")
|
|
if real_open and ticker in self._enrichment:
|
|
if date_str in self._enrichment[ticker]:
|
|
self._enrichment[ticker][date_str]["today_open"] = real_open
|
|
else:
|
|
# Fallback: find the entry that was created for this date
|
|
for d in sorted(self._enrichment[ticker].keys(), reverse=True):
|
|
if d <= date_str:
|
|
# Create a date_str entry inheriting from latest
|
|
self._enrichment[ticker][date_str] = copy.copy(
|
|
self._enrichment[ticker][d]
|
|
)
|
|
self._enrichment[ticker][date_str]["today_open"] = real_open
|
|
break
|
|
|
|
# Market regime check (mirrors simulate_day:1678-1693)
|
|
regime_thresh = getattr(self._params, "market_regime_spy_threshold", None)
|
|
if regime_thresh is not None:
|
|
regime_ticker = getattr(self._params, "market_regime_ticker", None) or "QQQ"
|
|
regime_enrich = self._enrichment.get(regime_ticker, {}).get(date_str, {})
|
|
regime_prev_close = regime_enrich.get("prev_close")
|
|
regime_today_open = regime_enrich.get("today_open")
|
|
if regime_prev_close and regime_today_open and regime_prev_close > 0:
|
|
regime_gap = (regime_today_open - regime_prev_close) / regime_prev_close
|
|
if regime_gap < regime_thresh:
|
|
self._log(
|
|
f"Regime filter: {regime_ticker} gap {regime_gap:.3%} "
|
|
f"< {regime_thresh:.3%} — skipping today"
|
|
)
|
|
self._state.update_daily_state(
|
|
self._session.session_id, date_str, phase="done"
|
|
)
|
|
return {
|
|
"universe_size": intraday_count, "daily_bars": daily_bars_count,
|
|
"intraday_bars": intraday_count, "orb_candidates": 0,
|
|
"long": 0, "short": 0, "skip_reason": "market_regime",
|
|
}
|
|
|
|
# Breadth filter (mirrors simulate_day:1706-1727)
|
|
min_breadth = getattr(self._params, "min_candidate_breadth", None)
|
|
if min_breadth is not None:
|
|
pos_gap_count = 0
|
|
total_with_data = 0
|
|
for ticker in bars_by_ticker:
|
|
t_enrich = self._enrichment.get(ticker, {}).get(date_str, {})
|
|
prev_c = t_enrich.get("prev_close")
|
|
today_o = t_enrich.get("today_open")
|
|
if prev_c and today_o and prev_c > 0:
|
|
total_with_data += 1
|
|
if today_o > prev_c:
|
|
pos_gap_count += 1
|
|
if total_with_data > 0:
|
|
breadth_ratio = pos_gap_count / total_with_data
|
|
if breadth_ratio < min_breadth:
|
|
self._log(
|
|
f"Breadth filter: {breadth_ratio:.1%} positive gaps "
|
|
f"< {min_breadth:.1%} — skipping today"
|
|
)
|
|
self._state.update_daily_state(
|
|
self._session.session_id, date_str, phase="done"
|
|
)
|
|
return {
|
|
"universe_size": intraday_count, "daily_bars": daily_bars_count,
|
|
"intraday_bars": intraday_count, "orb_candidates": 0,
|
|
"long": 0, "short": 0, "skip_reason": "breadth",
|
|
}
|
|
|
|
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(intraday_tickers),
|
|
"daily_bars": daily_bars_count,
|
|
"intraday_bars": intraday_count,
|
|
"orb_candidates": len(self._candidates),
|
|
"long": n_long,
|
|
"short": n_short,
|
|
}
|
|
|
|
# ── Phase 3: Breakout Check ───────────────────────────────────────────────
|
|
|
|
def 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
|
|
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 or at max simultaneous positions
|
|
open_positions = self._state.get_open_positions(
|
|
self._session.session_id, date_str
|
|
)
|
|
if any(p.ticker == ticker for p in open_positions):
|
|
self._state.update_candidate_status(
|
|
self._session.session_id, date_str, ticker, "filled"
|
|
)
|
|
continue
|
|
max_sim = getattr(self._params, "max_simultaneous_entries", None)
|
|
if max_sim is not None and len(open_positions) >= max_sim:
|
|
still_pending.append(cand)
|
|
continue
|
|
|
|
# Check breakout using real-time snapshot price
|
|
snap = snapshots.get(ticker)
|
|
if snap is None or snap.price is None:
|
|
still_pending.append(cand)
|
|
continue
|
|
|
|
current_price = snap.price
|
|
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
|
|
|
|
sizing_capital = self._compute_sizing_capital(equity)
|
|
risk_dollars = sizing_capital * 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 = (sizing_capital * self._params.max_position_pct) / entry_price_est
|
|
shares = int(min(shares_from_risk, max_shares_by_capital))
|
|
if shares <= 0:
|
|
self._log(f" {ticker}: shares=0 after sizing — skipping")
|
|
self._state.update_candidate_status(
|
|
self._session.session_id, date_str, ticker, "cancelled"
|
|
)
|
|
continue
|
|
|
|
# Check buying power
|
|
try:
|
|
acct = self._broker.get_account()
|
|
if acct.buying_power < shares * entry_price_est:
|
|
self._log(f" {ticker}: insufficient buying power — skipping")
|
|
self._state.update_candidate_status(
|
|
self._session.session_id, date_str, ticker, "cancelled"
|
|
)
|
|
continue
|
|
except Exception as e:
|
|
self._log(f" {ticker}: account check error: {e}")
|
|
|
|
# Place order
|
|
try:
|
|
if direction == "bullish":
|
|
order = self._broker.submit_market_buy(ticker, shares)
|
|
else:
|
|
order = self._broker.submit_market_sell(ticker, shares)
|
|
self._log(
|
|
f" {ticker}: {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
|
|
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 (sim_bar_minutes checkpoints) ────────────────────
|
|
|
|
def run_stop_check(self, date_str: str) -> dict[str, Any]:
|
|
"""Evaluate stops for all open positions using aggregated bars.
|
|
|
|
Called every sim_bar_minutes from orb_end until 15:55 ET.
|
|
Stop logic mirrors orb_simulator.py:477-580 exactly.
|
|
"""
|
|
self._date_str = date_str
|
|
|
|
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,
|
|
)
|
|
|
|
|
|
group_size = self._params.sim_bar_minutes // 5
|
|
|
|
stops_hit = 0
|
|
equity = self._get_equity()
|
|
|
|
for pos in positions:
|
|
ticker = pos.ticker
|
|
all_bars = bars_raw.get(ticker, [])
|
|
mkt_bars = filter_market_hours(all_bars)
|
|
|
|
if not mkt_bars:
|
|
continue
|
|
|
|
# Filter bars after entry time
|
|
entry_ts = _parse_ts(pos.entry_time)
|
|
post_entry = [b for b in mkt_bars if _parse_ts(b["timestamp"]) > entry_ts]
|
|
|
|
if not post_entry:
|
|
continue
|
|
|
|
# Aggregate to sim_bar_minutes (e.g., 90-min)
|
|
agg_bars = _aggregate_bars(post_entry, group_size)
|
|
|
|
# Run stop management on each aggregated bar
|
|
current_stop = pos.current_stop
|
|
peak_price = pos.peak_price
|
|
trailing_active = pos.trailing_active
|
|
stop_distance = pos.stop_distance
|
|
atr = pos.atr_at_entry
|
|
exit_bar = None
|
|
exit_reason = "close"
|
|
|
|
use_atr_trail = self._params.trailing_stop_atr_multiplier > 0
|
|
|
|
for bar in agg_bars:
|
|
bar_high = bar["high"]
|
|
bar_low = bar["low"]
|
|
|
|
if pos.direction == "long":
|
|
peak_price = max(peak_price, bar_high)
|
|
current_r = (bar_high - pos.entry_price) / stop_distance if stop_distance > 0 else 0
|
|
|
|
if current_r >= self._params.breakeven_at_r and current_stop < pos.entry_price:
|
|
current_stop = pos.entry_price
|
|
|
|
if current_r >= self._params.trailing_at_r:
|
|
trailing_active = True
|
|
|
|
# Check stop hit BEFORE updating trailing
|
|
if bar_low <= current_stop:
|
|
exit_bar = bar
|
|
exit_reason = "trailing_stop" if trailing_active else "stop_loss"
|
|
break
|
|
|
|
# Update trailing AFTER stop check
|
|
if trailing_active:
|
|
if use_atr_trail:
|
|
tighten_r = getattr(self._params, "trailing_tighten_at_r", None)
|
|
tight_mult = getattr(self._params, "trailing_stop_atr_multiplier_tight", 0.0)
|
|
if (tighten_r is not None and current_r >= tighten_r and tight_mult > 0):
|
|
atr_mult = tight_mult
|
|
else:
|
|
atr_mult = self._params.trailing_stop_atr_multiplier
|
|
candidate = peak_price - atr * atr_mult
|
|
else:
|
|
candidate = max(bar_low, current_stop)
|
|
if candidate > current_stop:
|
|
current_stop = candidate
|
|
|
|
else: # short
|
|
peak_price = min(peak_price, bar_low)
|
|
current_r = (pos.entry_price - bar_low) / stop_distance if stop_distance > 0 else 0
|
|
|
|
if current_r >= self._params.breakeven_at_r and current_stop > pos.entry_price:
|
|
current_stop = pos.entry_price
|
|
|
|
if current_r >= self._params.trailing_at_r:
|
|
trailing_active = True
|
|
|
|
if bar_high >= current_stop:
|
|
exit_bar = bar
|
|
exit_reason = "trailing_stop" if trailing_active else "stop_loss"
|
|
break
|
|
|
|
if trailing_active:
|
|
if use_atr_trail:
|
|
tighten_r = getattr(self._params, "trailing_tighten_at_r", None)
|
|
tight_mult = getattr(self._params, "trailing_stop_atr_multiplier_tight", 0.0)
|
|
if (tighten_r is not None and current_r >= tighten_r and tight_mult > 0):
|
|
atr_mult = tight_mult
|
|
else:
|
|
atr_mult = self._params.trailing_stop_atr_multiplier
|
|
candidate = peak_price + atr * atr_mult
|
|
else:
|
|
candidate = min(bar_high, current_stop)
|
|
if candidate < current_stop:
|
|
current_stop = candidate
|
|
|
|
# Update DB stop levels
|
|
self._state.update_position_stop(
|
|
self._session.session_id, date_str, ticker,
|
|
current_stop, peak_price, trailing_active,
|
|
)
|
|
|
|
if exit_bar:
|
|
# Close position — use qty so only this session's shares are closed
|
|
# (other sessions may hold the same ticker in the same Alpaca account).
|
|
exit_price = current_stop # fallback if fill poll fails
|
|
try:
|
|
close_order = self._broker.close_position(ticker, qty=int(pos.shares))
|
|
# Poll for actual broker fill price (captures gap-through losses)
|
|
for _ in range(4):
|
|
time.sleep(3)
|
|
try:
|
|
o = self._broker.get_order(close_order.id)
|
|
if o.filled_avg_price:
|
|
exit_price = o.filled_avg_price
|
|
break
|
|
except Exception:
|
|
pass
|
|
self._log(
|
|
f" {ticker}: stop hit ({exit_reason}) @ {exit_price:.2f}"
|
|
)
|
|
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))
|
|
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 candidates (in-memory and DB)
|
|
for cand in self._pending_cands:
|
|
self._state.update_candidate_status(
|
|
self._session.session_id, date_str, cand["ticker"], "timeout"
|
|
)
|
|
self._pending_cands = []
|
|
|
|
# Also sweep DB for any pending records not in in-memory list (e.g. after restart)
|
|
db_cands = self._state.list_candidates(self._session.session_id, date_str)
|
|
for c in db_cands:
|
|
if c["status"] == "pending":
|
|
self._state.update_candidate_status(
|
|
self._session.session_id, date_str, c["ticker"], "timeout"
|
|
)
|
|
|
|
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 _compute_sizing_capital(self, equity: float) -> float:
|
|
"""Replicate backtest sizing_capital formula: governor + streak multiplier.
|
|
|
|
Mirrors libs/intraday/orb_simulator.py:2141-2187.
|
|
V23 uses daily_budget_reset=True: base sizing = initial_equity (not equity).
|
|
This matches the backtest 단리 mode where each day starts from $10k.
|
|
"""
|
|
# daily_budget_reset: fixed daily budget matches V23 backtest 단리 mode
|
|
daily_reset = getattr(self._params, "daily_budget_reset", False)
|
|
sizing = self._session.initial_equity if daily_reset else equity
|
|
|
|
# Drawdown governor: scale down when equity drops below peak
|
|
gov_thresh = getattr(self._params, "drawdown_governor_threshold", None)
|
|
gov_min = getattr(self._params, "drawdown_governor_min_scale", 0.30)
|
|
if gov_thresh is not None:
|
|
peak_equity = self._state.get_peak_equity(
|
|
self._session.session_id, self._session.initial_equity
|
|
)
|
|
if peak_equity > 0:
|
|
dd_pct = (peak_equity - equity) / peak_equity
|
|
if dd_pct > gov_thresh:
|
|
dd_excess = dd_pct - gov_thresh
|
|
governor_scale = max(
|
|
gov_min,
|
|
1.0 - (1.0 - gov_min) * min(dd_excess / gov_thresh, 1.0),
|
|
)
|
|
sizing = sizing * governor_scale
|
|
|
|
# Streak sizing: amplify after consecutive wins, reduce after consecutive losses.
|
|
# list_trades returns DESC (newest first) — outcomes[0] = most recent trade.
|
|
win_bonus = getattr(self._params, "streak_sizing_win_bonus", None)
|
|
loss_penalty = getattr(self._params, "streak_sizing_loss_penalty", None)
|
|
streak_max = getattr(self._params, "streak_sizing_max", 2.5)
|
|
streak_min = getattr(self._params, "streak_sizing_min", 0.5)
|
|
if win_bonus is not None or loss_penalty is not None:
|
|
trades = self._state.list_trades(self._session.session_id)
|
|
if trades:
|
|
outcomes = [t["pnl"] > 0 for t in trades] # newest first
|
|
is_winning = outcomes[0] # most recent outcome
|
|
streak_len = 0
|
|
for o in outcomes: # count from newest
|
|
if o == is_winning:
|
|
streak_len += 1
|
|
else:
|
|
break
|
|
streak_mult = 1.0
|
|
if is_winning and win_bonus is not None:
|
|
streak_mult = 1.0 + streak_len * win_bonus
|
|
elif not is_winning and loss_penalty is not None:
|
|
streak_mult = 1.0 - streak_len * loss_penalty
|
|
streak_mult = max(streak_min, min(streak_max, streak_mult))
|
|
sizing = sizing * streak_mult
|
|
|
|
return sizing
|
|
|
|
def _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,
|
|
log_callback: Any = None,
|
|
) -> ORBTradingEngine:
|
|
"""Create an ORBTradingEngine for the given session.
|
|
|
|
broker_override: pass a MockORBBroker (or any duck-typed broker) to avoid
|
|
real Alpaca API calls during testing.
|
|
"""
|
|
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,
|
|
log_callback=log_callback,
|
|
)
|