"""ORB single-ticker advisor. Wraps the same data fetch + pre-screen + run_orb_simulation pipeline the intraday backtester uses, scoped to one (ticker, asof). Verdicts match the historical backtest entries on that date for that ticker. """ from __future__ import annotations import datetime as dt import os from datetime import timedelta from pathlib import Path from typing import Any, Literal from zoneinfo import ZoneInfo import yaml from pydantic import BaseModel, Field from libs.common.logging import get_logger from libs.intraday.cache import DailyBarCache, IntradayCache from libs.intraday.domain import ORBStrategyParams from libs.intraday.features import enrich_daily_bars from libs.intraday.orb_simulator import run_orb_simulation from libs.intraday.screener import ( fetch_daily_bars_bulk, fetch_intraday_bulk, orb_pre_screen_candidates, ) from libs.oracle_client.client import OracleClient logger = get_logger(__name__) _ET = ZoneInfo("America/New_York") _MARKET_OPEN = dt.time(9, 30) _MARKET_CLOSE = dt.time(16, 0) Signal = Literal["BUY", "SELL", "HOLD", "NO_SIGNAL", "ERROR"] class OrbVerdict(BaseModel): signal: Signal score: float | None = None reason: str details: dict[str, Any] = Field(default_factory=dict) warning: str | None = None def _oracle_url() -> str: return os.environ.get("ORACLE_URL") or os.environ.get( "STOCK_ORACLE_URL", "http://localhost:8000" ) def _load_orb_params(yaml_path: str | Path) -> ORBStrategyParams: with open(yaml_path) as f: raw = yaml.safe_load(f) or {} return ORBStrategyParams(**raw.get("orb_strategy", {})) def _support_tickers(params: ORBStrategyParams) -> list[str]: """Auxiliary tickers needed for regime / market-quality gates.""" candidates: set[str] = {"SPY", "QQQ"} for attr in ( "market_regime_ticker", "market_orb_quality_ticker", "market_orb_quality_secondary_ticker", ): value = getattr(params, attr, None) if value: candidates.add(str(value)) return sorted(candidates) def _now_et() -> dt.datetime: return dt.datetime.now(_ET) def _orb_window_end(params: ORBStrategyParams, asof: dt.date) -> dt.datetime: orb_min = int(getattr(params, "orb_minutes", 5) or 5) return dt.datetime.combine(asof, _MARKET_OPEN, tzinfo=_ET) + timedelta(minutes=orb_min) async def evaluate_orb_buy( ticker: str, asof: dt.date, yaml_path: str | Path, ) -> OrbVerdict: params = _load_orb_params(yaml_path) upper = ticker.upper() now = _now_et() today_et = now.date() warning: str | None = None if asof > today_et: return OrbVerdict( signal="ERROR", reason=f"asof={asof.isoformat()} is in the future", ) if asof == today_et: if now.time() < _MARKET_OPEN: return OrbVerdict( signal="NO_SIGNAL", reason="market not yet open in ET; ORB requires the opening 5-min range", ) orb_end_dt = _orb_window_end(params, asof) if now < orb_end_dt: return OrbVerdict( signal="NO_SIGNAL", reason=( f"ORB window not yet closed; check back at " f"{orb_end_dt.strftime('%H:%M ET')} " f"(orb_minutes={params.orb_minutes})" ), ) if now.time() >= _MARKET_CLOSE: warning = "ORB session is over for today; this is a historical signal" else: warning = "ORB session in progress; intraday data may be partial" else: warning = f"evaluating historical date {asof.isoformat()}; signal is post-hoc" asof_str = asof.isoformat() daily_start = (asof - timedelta(days=120)).isoformat() tickers_needed = sorted({upper, *_support_tickers(params)}) daily_cache = DailyBarCache() intraday_cache = IntradayCache() async with OracleClient(base_url=_oracle_url()) as client: daily_bars = await fetch_daily_bars_bulk( tickers_needed, daily_start, asof_str, client, cache=daily_cache, intraday_cache_fallback=intraday_cache, prefer_intraday_fallback=True, skip_oracle_when_unhealthy=True, concurrency=8, ) if upper not in daily_bars or not daily_bars[upper]: return OrbVerdict( signal="ERROR", reason=f"no daily bars available for {upper}; ticker invalid or data unavailable", warning=warning, ) if not any(str(b.get("date", ""))[:10] == asof_str for b in daily_bars[upper]): return OrbVerdict( signal="NO_SIGNAL", reason=( f"no daily bar for {upper} on {asof_str}; not a trading day or data missing" ), warning=warning, ) trading_days = [asof_str] enrichment = enrich_daily_bars(daily_bars, trading_days) candidate_dict = orb_pre_screen_candidates( daily_bars, trading_days, enrichment, min_price=params.min_price, min_atr=params.min_atr_14, min_avg_dollar_vol=params.min_avg_dollar_volume, max_per_day=None, ) if upper not in candidate_dict.get(asof_str, []): ticker_enrich = enrichment.get(upper, {}).get(asof_str, {}) atr = ticker_enrich.get("atr_14") adv = ticker_enrich.get("avg_dollar_vol_30d") today_open = ticker_enrich.get("today_open") reasons: list[str] = [] if today_open is not None and today_open < params.min_price: reasons.append(f"open ${today_open:.2f} < min_price ${params.min_price}") if atr is None or atr < params.min_atr_14: reasons.append(f"atr_14={atr} < {params.min_atr_14}") if adv is None or adv < params.min_avg_dollar_volume: reasons.append( f"avg_dollar_vol_30d={adv} < {params.min_avg_dollar_volume:.0f}" ) return OrbVerdict( signal="NO_SIGNAL", reason=( f"pre-screen rejected {upper}: " f"{', '.join(reasons) or 'liquidity/volatility gate'}" ), details={ "atr_14": atr, "avg_dollar_vol_30d": adv, "today_open": today_open, }, warning=warning, ) intraday_candidates: dict[str, list[str]] = {asof_str: tickers_needed} all_intraday = await fetch_intraday_bulk( intraday_candidates, client, intraday_cache, concurrency=4, skip_oracle_when_unhealthy=True, ) if upper not in all_intraday.get(asof_str, {}): return OrbVerdict( signal="NO_SIGNAL", reason=f"no intraday bars available for {upper} on {asof_str}", warning=warning, ) ticker_sectors = {t: "UNKNOWN" for t in tickers_needed} day_results = run_orb_simulation( all_intraday=all_intraday, trading_days=[asof_str], params=params, enrichment=enrichment, ticker_sectors=ticker_sectors, ) if not day_results: return OrbVerdict( signal="NO_SIGNAL", reason="ORB simulator returned no day result", warning=warning, ) day = day_results[0] ticker_trades = [t for t in day.trades if t.ticker.upper() == upper] if not ticker_trades: skip_msg = day.skip_reason or "no entry generated (gap/RVOL/breakout filter)" return OrbVerdict( signal="NO_SIGNAL", reason=f"ORB simulation produced no trade for {upper}: {skip_msg}", details={ "day_skip_reason": day.skip_reason, "candidates_found": day.candidates_found, "regime_scaler": day.regime_scaler, "breadth_scaler": day.breadth_scaler, "candidate_filter_stats": day.candidate_filter_stats, }, warning=warning, ) trade = ticker_trades[0] return OrbVerdict( signal="BUY", reason=f"ORB breakout entry @ {trade.entry_time}", details={ "entry_price": trade.entry_price, "entry_time": trade.entry_time, "shares": trade.shares, "atr_at_entry": trade.atr_at_entry, "rvol": trade.rvol, "gap_pct": trade.gap_pct, "orb_direction": trade.orb_direction, "simulated_exit_price": trade.exit_price, "simulated_exit_time": trade.exit_time, "simulated_exit_reason": trade.exit_reason, }, warning=warning, ) async def evaluate_orb_sell( ticker: str, entry_date: dt.date, asof: dt.date, yaml_path: str | Path, ) -> OrbVerdict: """ORB exits intraday on the entry day at 15:55 ET. If the user holds past the entry day, the position is outside ORB's risk model — return HOLD with a manual-exit warning. If entry_date == asof, re-run the simulation and report whether the modeled exit has fired. """ upper = ticker.upper() if entry_date != asof: return OrbVerdict( signal="HOLD", reason=( f"ORB exits intraday at 15:55 ET on the entry day; entry_date=" f"{entry_date.isoformat()} != asof={asof.isoformat()} means this position " "is outside ORB's lifecycle. Consider manual exit." ), warning="position outside ORB lifecycle — strategy provides no further signal", ) buy = await evaluate_orb_buy(upper, asof, yaml_path) if buy.signal != "BUY": return OrbVerdict( signal="HOLD", reason=( f"no ORB entry simulated for {upper} on {asof.isoformat()}; cannot derive " "an engine-consistent exit" ), details=buy.details, warning=buy.warning or "ORB intraday in progress; check back after 16:00 ET", ) sim_exit_time = buy.details.get("simulated_exit_time") sim_exit_reason = buy.details.get("simulated_exit_reason") sim_exit_price = buy.details.get("simulated_exit_price") if not sim_exit_time: return OrbVerdict( signal="HOLD", reason="ORB simulation has no exit yet; intraday session still in progress", warning="ORB exits at 15:55 ET; check back after market close", ) now = _now_et() exit_dt: dt.datetime | None = None try: exit_dt = dt.datetime.fromisoformat(str(sim_exit_time).replace("Z", "+00:00")) if exit_dt.tzinfo is None: exit_dt = exit_dt.replace(tzinfo=_ET) except Exception: exit_dt = None base = { "exit_time": sim_exit_time, "exit_reason": sim_exit_reason, "exit_price": sim_exit_price, } if exit_dt is None or exit_dt <= now: return OrbVerdict( signal="SELL", reason=f"ORB exit fired in simulation: {sim_exit_reason}", details=base, warning=buy.warning, ) return OrbVerdict( signal="HOLD", reason=f"simulated exit at {sim_exit_time} has not yet occurred", details=base, warning="ORB intraday in progress", )