"""ORB Scanner service — v49.100 manual entry/exit checker. Provides three operations: check() — single-ticker entry filter breakdown + verdict exit_check() — pure-math stop/trailing state machine (no Oracle calls) gainers_scan() — batched pipeline over Oracle gainers universe """ from __future__ import annotations import datetime as dt import os from datetime import timedelta from pathlib import Path from zoneinfo import ZoneInfo import httpx from pydantic import BaseModel from apps.intraday_bt.run import _load_config_yaml from libs.common.logging import get_logger from libs.intraday.cache import IntradayCache from libs.intraday.domain import ORBStrategyParams from libs.intraday.features import compute_rvol_approx, 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 from libs.oracle_client.price import PriceService logger = get_logger(__name__) _ET = ZoneInfo("America/New_York") _MARKET_OPEN = dt.time(9, 30) _ORB_END = dt.time(9, 35) _ORDER_TIMEOUT = dt.time(9, 55) _FORCE_EXIT_TIME = dt.time(15, 55) _ORB_DEFAULT_STRATEGY = "orb_gainers_v49_100_mid_hot_rtg_reserve" _STRATEGIES_DIR = Path("configs/intraday/strategies") _params_cache: dict[str, ORBStrategyParams] = {} def _get_params(strategy_id: str | None = None) -> ORBStrategyParams: key = strategy_id or _ORB_DEFAULT_STRATEGY if key not in _params_cache: yaml_path = _STRATEGIES_DIR / f"{key}.yaml" raw = _load_config_yaml(yaml_path) _params_cache[key] = ORBStrategyParams(**raw.get("orb_strategy", {})) return _params_cache[key] def list_strategies() -> list[dict]: import yaml as _yaml result = [] for p in sorted(_STRATEGIES_DIR.glob("orb_gainers_*.yaml")): try: with open(p) as f: raw = _yaml.safe_load(f) or {} meta = raw.get("_meta") or {} name = meta.get("name") or raw.get("name") or p.stem result.append({"id": p.stem, "name": name}) except Exception: pass return result def _oracle_url() -> str: return os.environ.get("STOCK_ORACLE_URL", "http://localhost:18001") def _now_et() -> dt.datetime: return dt.datetime.now(_ET) def _market_status_str() -> str: t = _now_et().time() if t < _MARKET_OPEN: return "pre_market" elif t < _ORB_END: return "orb_forming" elif t < _ORDER_TIMEOUT: return "entry_window" elif t < _FORCE_EXIT_TIME: return "active" else: return "closed" def _premarket_dollar_vol(bars: list[dict], date_str: str) -> float: """Sum close*volume for premarket bars (04:00-09:30 ET) on date_str.""" total = 0.0 for bar in bars: ts_raw = bar.get("timestamp", "") if not ts_raw: continue try: ts = dt.datetime.fromisoformat(str(ts_raw).replace("Z", "+00:00")) if ts.tzinfo is None: ts = ts.replace(tzinfo=_ET) ts_et = ts.astimezone(_ET) except Exception: continue if ts_et.date().isoformat() != date_str: continue if not (dt.time(4, 0) <= ts_et.time() < _MARKET_OPEN): continue price = bar.get("close") or bar.get("open") or 0.0 volume = bar.get("volume") or 0.0 if price > 0 and volume > 0: total += price * volume return total def _orb_bar(bars: list[dict], date_str: str) -> dict | None: """Find the first regular 5-min bar at 09:30 ET on date_str.""" candidates: list[tuple[dt.datetime, dict]] = [] for bar in bars: ts_raw = bar.get("timestamp", "") if not ts_raw: continue try: ts = dt.datetime.fromisoformat(str(ts_raw).replace("Z", "+00:00")) if ts.tzinfo is None: ts = ts.replace(tzinfo=_ET) ts_et = ts.astimezone(_ET) except Exception: continue if ts_et.date().isoformat() != date_str: continue if ts_et.time() == _MARKET_OPEN: candidates.append((ts_et, bar)) if not candidates: # Fallback: first bar on that date within regular hours for bar in bars: ts_raw = bar.get("timestamp", "") if not ts_raw: continue try: ts = dt.datetime.fromisoformat(str(ts_raw).replace("Z", "+00:00")) if ts.tzinfo is None: ts = ts.replace(tzinfo=_ET) ts_et = ts.astimezone(_ET) except Exception: continue if ts_et.date().isoformat() == date_str and _MARKET_OPEN <= ts_et.time() < dt.time(10, 0): candidates.append((ts_et, bar)) if not candidates: return None candidates.sort(key=lambda x: x[0]) return candidates[0][1] def _latest_bar(bars: list[dict], date_str: str) -> dict | None: """Most recent regular-hours bar on date_str.""" matches: list[tuple[dt.datetime, dict]] = [] for bar in bars: ts_raw = bar.get("timestamp", "") if not ts_raw: continue try: ts = dt.datetime.fromisoformat(str(ts_raw).replace("Z", "+00:00")) if ts.tzinfo is None: ts = ts.replace(tzinfo=_ET) ts_et = ts.astimezone(_ET) except Exception: continue if ts_et.date().isoformat() != date_str: continue if _MARKET_OPEN <= ts_et.time(): matches.append((ts_et, bar)) if not matches: return None matches.sort(key=lambda x: x[0]) return matches[-1][1] def _parse_gainers(data: object) -> list[dict]: """Extract ticker + price/change info from unknown gainers response shape.""" if isinstance(data, list): return data if isinstance(data, dict): for key in ("gainers", "stocks", "tickers", "data", "results"): val = data.get(key) if isinstance(val, list): return val return [] def _gainer_ticker(item: object) -> str | None: if isinstance(item, str): return item.upper() if isinstance(item, dict): for key in ("symbol", "ticker", "Symbol", "Ticker"): v = item.get(key) if v and isinstance(v, str): return v.upper() return None # ── Pydantic models ──────────────────────────────────────────────────────── class FilterResult(BaseModel): name: str passed: bool value: str | None = None threshold: str | None = None note: str | None = None class StrategyInfo(BaseModel): id: str name: str class StrategiesResponse(BaseModel): strategies: list[StrategyInfo] default: str class OrbCheckRequest(BaseModel): ticker: str asof: str | None = None strategy: str | None = None class OrbCheckResponse(BaseModel): ticker: str asof: str evaluated_at: str market_status: str overall_signal: str filters: list[FilterResult] entry_details: dict | None = None warning: str | None = None class OrbExitCheckRequest(BaseModel): ticker: str entry_price: float | None = None # auto-fetched from ORB bar if omitted atr_at_entry: float | None = None # auto-fetched from enrichment if omitted current_price: float | None = None # auto-fetched from intraday if omitted peak_price: float | None = None # auto-fetched from intraday if omitted entry_time: str | None = None # "HH:MM" ET; defaults to 09:35 strategy: str | None = None class OrbExitCheckResponse(BaseModel): ticker: str current_stop: float stop_phase: str should_exit: bool reason: str r_multiple: float profit_r: float time_status: str current_price: float peak_price: float entry_time_used: str | None = None entry_price_used: float | None = None atr_at_entry_used: float | None = None class GainerResult(BaseModel): ticker: str price: float | None = None change_pct: float | None = None signal: str scale_factor: float | None = None filter_summary: str failure_reason: str | None = None class GainersScanResponse(BaseModel): scan_time: str market_status: str count_fetched: int count_passed: int results: list[GainerResult] # ── Service functions ────────────────────────────────────────────────────── async def check(req: OrbCheckRequest) -> OrbCheckResponse: params = _get_params(req.strategy) now = _now_et() asof = dt.date.fromisoformat(req.asof) if req.asof else now.date() asof_str = asof.isoformat() mstat = _market_status_str() evaluated_at = now.isoformat() if asof == now.date() and mstat == "pre_market": return OrbCheckResponse( ticker=req.ticker.upper(), asof=asof_str, evaluated_at=evaluated_at, market_status=mstat, overall_signal="NOT_YET", filters=[], warning="Market not yet open. Check after 09:30 ET.", ) if asof == now.date() and mstat == "orb_forming": return OrbCheckResponse( ticker=req.ticker.upper(), asof=asof_str, evaluated_at=evaluated_at, market_status=mstat, overall_signal="NOT_YET", filters=[], warning="ORB window still forming. Check back after 09:35 ET.", ) ticker = req.ticker.strip().upper() daily_start = (asof - timedelta(days=120)).isoformat() support_tickers = ["SPY", "QQQ"] all_tickers = sorted({ticker, *support_tickers}) warning: str | None = None if mstat == "closed": warning = "Market session over. Signal is based on today's completed session." elif mstat == "active": warning = "Entry window closed (09:55 ET). Shown for reference only." try: async with OracleClient(base_url=_oracle_url()) as client: daily_bars = await fetch_daily_bars_bulk( all_tickers, daily_start, asof_str, client, skip_oracle_when_unhealthy=True, concurrency=4, ) if ticker not in daily_bars or not daily_bars[ticker]: # Per-ticker fallback: try individual price endpoint directly try: svc_fallback = PriceService(client) pd_resp = await svc_fallback.get_daily_bars(ticker, start=daily_start, end=asof_str) if pd_resp.bars: daily_bars[ticker] = [ {"date": b.date, "open": b.open, "high": b.high, "low": b.low, "close": b.close, "volume": b.volume} for b in pd_resp.bars ] except Exception: pass if ticker not in daily_bars or not daily_bars[ticker]: return OrbCheckResponse( ticker=ticker, asof=asof_str, evaluated_at=evaluated_at, market_status=mstat, overall_signal="ERROR", filters=[], warning=f"No price data for {ticker}. Oracle may be loading — try again in a moment.", ) # Inject synthetic today row (Oracle only has yesterday's daily bars when market is open) for sym, bars in daily_bars.items(): if bars: last_bar = max(bars, key=lambda b: b["date"]) if last_bar["date"] < asof_str: daily_bars[sym] = bars + [{ "date": asof_str, "open": last_bar["close"], "high": last_bar["close"], "low": last_bar["close"], "close": last_bar["close"], "volume": 0, "synthetic_today_daily": True, }] enrichment = enrich_daily_bars(daily_bars, [asof_str]) # Fetch intraday for ticker + support (for regime gates in simulation) intraday_candidates = {asof_str: all_tickers} all_intraday = await fetch_intraday_bulk( intraday_candidates, client, cache=None, skip_oracle_when_unhealthy=True, concurrency=4, ) except Exception as exc: logger.exception("orb_scanner_check_failed", ticker=ticker) return OrbCheckResponse( ticker=ticker, asof=asof_str, evaluated_at=evaluated_at, market_status=mstat, overall_signal="ERROR", filters=[], warning=f"Data fetch failed: {exc}", ) ticker_enrich = enrichment.get(ticker, {}).get(asof_str, {}) ticker_bars = all_intraday.get(asof_str, {}).get(ticker, []) # ── Build filter breakdown ───────────────────────────────────────────── filters: list[FilterResult] = [] # 1. Price today_open = ticker_enrich.get("today_open") filters.append(FilterResult( name="Price", passed=today_open is not None and today_open >= params.min_price, value=f"${today_open:.2f}" if today_open else "N/A", threshold=f"≥ ${params.min_price:.2f}", )) # 2. Avg Daily Dollar Volume (30d) avg_dvol = ticker_enrich.get("avg_dollar_vol_30d") min_dvol = params.min_avg_dollar_volume filters.append(FilterResult( name="Avg Daily $Vol (30d)", passed=avg_dvol is not None and avg_dvol >= min_dvol, value=f"${avg_dvol / 1e6:.1f}M" if avg_dvol else "N/A", threshold=f"≥ ${min_dvol / 1e6:.0f}M", )) # 3. ATR(14) atr = ticker_enrich.get("atr_14") filters.append(FilterResult( name="ATR(14)", passed=atr is not None and atr >= params.min_atr_14, value=f"{atr:.3f}" if atr else "N/A", threshold=f"≥ {params.min_atr_14:.2f}", )) # 4. Gap% prev_close = ticker_enrich.get("prev_close") gap_pct: float | None = None if today_open and prev_close and prev_close > 0: gap_pct = (today_open - prev_close) / prev_close min_gap = params.min_abs_gap_pct or 0.0 max_gap = getattr(params, "max_gap_pct", None) or 1.0 gap_ok = gap_pct is not None and min_gap <= abs(gap_pct) <= max_gap filters.append(FilterResult( name="Gap%", passed=gap_ok, value=f"{gap_pct * 100:+.2f}%" if gap_pct is not None else "N/A", threshold=f"{min_gap * 100:.0f}%–{max_gap * 100:.0f}% (abs)", )) # ORB-specific filters require intraday bars orb = _orb_bar(ticker_bars, asof_str) latest = _latest_bar(ticker_bars, asof_str) premarket_dvol = _premarket_dollar_vol(ticker_bars, asof_str) avg_vol_14d = ticker_enrich.get("avg_daily_vol_14d") # Refine gap% using real ORB bar open (synthetic today row uses prev_close as open) if orb and ticker_enrich.get("synthetic_today_daily"): real_open = orb.get("open") if real_open and real_open > 0: today_open = real_open if prev_close and prev_close > 0: gap_pct = (today_open - prev_close) / prev_close gap_ok = min_gap <= abs(gap_pct) <= max_gap # Update the already-appended Gap% filter for f in filters: if f.name == "Gap%": f.value = f"{gap_pct * 100:+.2f}%" f.passed = gap_ok # 5. RVOL (informational — v49 uses volume_attention scoring, no simple cutoff) rvol: float | None = None if orb and avg_vol_14d and avg_vol_14d > 0: rvol = compute_rvol_approx(orb.get("volume", 0), avg_vol_14d) min_rvol = params.min_rvol filters.append(FilterResult( name="RVOL", passed=True if min_rvol is None else (rvol is not None and rvol >= min_rvol), value=f"{rvol:.1f}x" if rvol is not None else ("No intraday bars" if not orb else "N/A"), threshold=f"≥ {min_rvol:.1f}x" if min_rvol is not None else "volume_attention scored", note="informational" if min_rvol is None else None, )) # 6. Premarket $Vol min_premarket = params.min_premarket_dollar_vol premarket_unavailable = premarket_dvol == 0.0 and orb is not None filters.append(FilterResult( name="Premarket $Vol", passed=True if (min_premarket is None or premarket_unavailable) else premarket_dvol >= min_premarket, value=f"${premarket_dvol / 1e6:.2f}M" if not premarket_unavailable else "0 (no premarket bars)", threshold=f"≥ ${min_premarket / 1e6:.1f}M" if min_premarket else "N/A (no global floor)", note="IEX feed may not include extended hours" if premarket_unavailable else None, )) # 7. ORB Bullish orb_bullish = False if orb: orb_open_p = orb.get("open", 0) or 0.0 orb_close_p = orb.get("close", 0) or 0.0 orb_bullish = orb_close_p > orb_open_p filters.append(FilterResult( name="ORB Bullish", passed=orb_bullish, value=(f"close {orb.get('close', 0):.2f} > open {orb.get('open', 0):.2f}" if orb and orb_bullish else (f"close {orb.get('close', 0):.2f} ≤ open {orb.get('open', 0):.2f}" if orb else "N/A")), threshold="close > open", )) # 8. Breakout (current price ≥ ORB high) orb_high = orb.get("high", 0) if orb else 0.0 current_price = latest.get("close", 0) if latest else 0.0 breakout = current_price >= orb_high if (orb_high > 0 and current_price > 0) else False filters.append(FilterResult( name="Breakout", passed=breakout, value=f"${current_price:.2f}", threshold=f"≥ ORB high ${orb_high:.2f}" if orb_high > 0 else "ORB high N/A", )) # 9. Hot Reclaim Guard ret_5d = ticker_enrich.get("ret_5d") hot_min_ret5d = params.hot_reclaim_min_ret_5d hot_max_premarket = params.hot_reclaim_max_premarket_dollar_vol hot_triggered = ( hot_min_ret5d is not None and hot_max_premarket is not None and ret_5d is not None and ret_5d >= hot_min_ret5d and premarket_dvol <= hot_max_premarket ) if hot_min_ret5d is not None: scale = params.hot_reclaim_size_scale if hot_triggered else None filters.append(FilterResult( name="Hot Reclaim Guard", passed=True, value=f"5d ret {ret_5d * 100:.1f}%" if ret_5d is not None else "N/A", threshold=f"5d ≥ {hot_min_ret5d * 100:.0f}% + premarket ≤ ${(hot_max_premarket or 0) / 1e6:.1f}M → scale", note=f"triggered → {scale:.1f}x size" if hot_triggered and scale else None, )) # 10. Stale OBV Gate obv_slope = ticker_enrich.get("obv_slope_20") stale_max_obv = params.stale_obv_reversal_max_obv_slope_20d if stale_max_obv is not None and obv_slope is not None: stale_triggered = obv_slope <= stale_max_obv stale_scale = params.stale_obv_reversal_size_scale if stale_triggered else None filters.append(FilterResult( name="Stale OBV Gate", passed=True, value=f"OBV slope {obv_slope:.3f}", threshold=f"slope ≤ {stale_max_obv:.2f} → scale", note=f"triggered → {stale_scale:.2f}x size" if stale_triggered and stale_scale else None, )) # ── Run simulation for final verdict ────────────────────────────────── ticker_sectors = {t: "UNKNOWN" for t in all_tickers} try: day_results = run_orb_simulation( all_intraday, [asof_str], params, enrichment, ticker_sectors=ticker_sectors, ) except Exception as exc: logger.warning("orb_simulation_failed", ticker=ticker, error=str(exc)) day_results = [] day = day_results[0] if day_results else None trade = next((t for t in (day.trades if day else []) if t.ticker == ticker), None) if trade: overall_signal = "ENTRY" entry_details = { "entry_price": trade.entry_price, "atr_at_entry": trade.atr_at_entry, "rvol": trade.rvol, "gap_pct": trade.gap_pct, "orb_direction": trade.orb_direction, "body_ratio": trade.body_ratio, "close_location": trade.close_location, "premarket_dollar_vol": trade.premarket_dollar_vol, "hot_reclaim_size_scale": trade.hot_reclaim_size_scale, "entry_time": trade.entry_time, } # Refine filter values from simulation if available if trade.rvol is not None and rvol is None: for f in filters: if f.name == "RVOL": f.value = f"{trade.rvol:.1f}x" f.passed = True if trade.premarket_dollar_vol is not None: for f in filters: if f.name == "Premarket $Vol": f.value = f"${trade.premarket_dollar_vol / 1e6:.2f}M" f.passed = trade.premarket_dollar_vol >= (params.min_premarket_dollar_vol or 0) f.note = None # Update Hot Reclaim Guard scale from actual trade if trade.hot_reclaim_size_scale is not None and trade.hot_reclaim_size_scale < 1.0: for f in filters: if f.name == "Hot Reclaim Guard": f.note = f"triggered → {trade.hot_reclaim_size_scale:.1f}x size" else: overall_signal = "NO_ENTRY" entry_details = None return OrbCheckResponse( ticker=ticker, asof=asof_str, evaluated_at=evaluated_at, market_status=mstat, overall_signal=overall_signal, filters=filters, entry_details=entry_details, warning=warning, ) async def exit_check(req: OrbExitCheckRequest) -> OrbExitCheckResponse: """Stop state machine. Auto-fetches prices/ATR from Oracle when not provided.""" ticker = req.ticker.upper() now_et = _now_et() asof_str = now_et.date().isoformat() # Parse entry time (ET) entry_time_used: str | None = None entry_dt: dt.datetime | None = None if req.entry_time: try: parts = req.entry_time.replace(" ", "").split(":") h, m = int(parts[0]), int(parts[1]) entry_dt = now_et.replace(hour=h, minute=m, second=0, microsecond=0) entry_time_used = f"{h:02d}:{m:02d}" except Exception: pass if entry_dt is None: entry_dt = now_et.replace(hour=9, minute=35, second=0, microsecond=0) entry_time_used = "09:35" need_quote = req.current_price is None or req.entry_price is None need_intraday = req.peak_price is None # intraday needed for peak tracking need_daily = req.atr_at_entry is None bars_raw: list[dict] = [] atr = req.atr_at_entry live_price: float | None = None try: async with OracleClient(base_url=_oracle_url()) as client: # Real-time quote for current_price / entry_price if need_quote: try: svc = PriceService(client) quote = await svc.get_quote(ticker) live_price = quote.price except Exception as qexc: logger.warning("exit_check_quote_failed", ticker=ticker, error=str(qexc)) # Intraday bars for peak tracking if need_intraday or need_quote: intraday = await fetch_intraday_bulk( {asof_str: [ticker]}, client, cache=None, skip_oracle_when_unhealthy=True, concurrency=2, ) bars_raw = intraday.get(asof_str, {}).get(ticker, []) if need_daily: daily_start = (now_et.date() - timedelta(days=60)).isoformat() daily_bars_atr = await fetch_daily_bars_bulk( [ticker], daily_start, asof_str, client, skip_oracle_when_unhealthy=True, concurrency=2, ) bars = daily_bars_atr.get(ticker, []) if bars: last_bar = max(bars, key=lambda b: b["date"]) if last_bar["date"] < asof_str: daily_bars_atr[ticker] = bars + [{ "date": asof_str, "open": last_bar["close"], "high": last_bar["close"], "low": last_bar["close"], "close": last_bar["close"], "volume": 0, }] enrichment = enrich_daily_bars(daily_bars_atr, [asof_str]) atr = enrichment.get(ticker, {}).get(asof_str, {}).get("atr_14") except Exception as exc: logger.warning("exit_check_fetch_failed", ticker=ticker, error=str(exc)) # Parse intraday timestamps for peak tracking timed: list[tuple[dt.datetime, dict]] = [] for bar in bars_raw: ts_raw = bar.get("timestamp", "") if not ts_raw: continue try: ts = dt.datetime.fromisoformat(str(ts_raw).replace("Z", "+00:00")) if ts.tzinfo is None: ts = ts.replace(tzinfo=_ET) ts_et = ts.astimezone(_ET) except Exception: continue if ts_et.date().isoformat() == asof_str and ts_et.time() >= _MARKET_OPEN: timed.append((ts_et, bar)) timed.sort(key=lambda x: x[0]) # Resolve entry_price: real-time quote when in Current mode entry_price = req.entry_price if entry_price is None: entry_price = live_price or (float(timed[-1][1].get("close") or 0) if timed else 0.0) # current_price: always use real-time quote when available; fallback to latest bar current_price = req.current_price if current_price is None: current_price = live_price or (float(timed[-1][1].get("close") or entry_price) if timed else entry_price) # peak_price: max high of intraday bars since entry_time since_entry = [(ts, b) for ts, b in timed if ts >= entry_dt] peak_price = req.peak_price if peak_price is None: if since_entry: peak_price = max((float(b.get("high") or entry_price) for _, b in since_entry), default=entry_price) peak_price = max(peak_price, entry_price) else: peak_price = entry_price if current_price is None: current_price = entry_price if peak_price is None: peak_price = entry_price if not atr or atr <= 0: atr = 0.001 exit_params = _get_params(req.strategy) atr_mult = exit_params.atr_stop_multiplier or 0.75 trailing_atr = exit_params.trailing_stop_atr_multiplier or 0.6 tight_atr = exit_params.trailing_stop_atr_multiplier_tight or 0.2 trailing_r = exit_params.trailing_at_r or 1.0 tighten_r = exit_params.trailing_tighten_at_r # may be None breakeven_r = exit_params.breakeven_at_r or 1.0 stop_distance = atr_mult * atr initial_stop = entry_price - stop_distance peak_r = (peak_price - entry_price) / stop_distance current_r = (current_price - entry_price) / stop_distance if tighten_r is not None and peak_r >= tighten_r: current_stop = peak_price - tight_atr * atr phase = "trailing_tight" elif peak_r >= trailing_r: current_stop = peak_price - trailing_atr * atr phase = "trailing" else: current_stop = initial_stop phase = "initial" # Enforce breakeven floor if current_r >= breakeven_r and current_stop < entry_price: current_stop = entry_price if phase == "initial": phase = "breakeven" # 15:55 ET force exit force_exit = now_et.time() >= _FORCE_EXIT_TIME if force_exit: phase = "force_exit" time_status = "force_exit_due" elif now_et.time() >= _MARKET_OPEN: time_status = "in_window" else: time_status = "post_close" hit_stop = current_price <= current_stop if force_exit: should_exit = True reason = "Force exit at 15:55 ET" elif hit_stop: should_exit = True reason = f"Stop hit: ${current_price:.2f} ≤ stop ${current_stop:.2f} ({phase})" else: should_exit = False reason = f"Hold: ${current_price:.2f} > stop ${current_stop:.2f} ({phase})" return OrbExitCheckResponse( ticker=ticker, current_stop=round(current_stop, 4), stop_phase=phase, should_exit=should_exit, reason=reason, r_multiple=round(current_r, 3), profit_r=round(current_r, 3), time_status=time_status, current_price=round(current_price, 4), peak_price=round(peak_price, 4), entry_time_used=entry_time_used, entry_price_used=round(entry_price, 4), atr_at_entry_used=round(atr, 4), ) async def gainers_scan(count: int = 200, strategy: str | None = None) -> GainersScanResponse: params = _get_params(strategy) now = _now_et() asof = now.date() asof_str = asof.isoformat() mstat = _market_status_str() scan_time = now.isoformat() # 1. Fetch gainers from Oracle gainers_url = f"{_oracle_url()}/api/v1/stocks/gainers" try: resp = httpx.get(gainers_url, params={"count": count}, timeout=10.0) resp.raise_for_status() raw_gainers = resp.json() except Exception as exc: logger.error("gainers_fetch_failed", error=str(exc)) return GainersScanResponse( scan_time=scan_time, market_status=mstat, count_fetched=0, count_passed=0, results=[], ) gainer_items = _parse_gainers(raw_gainers) if not gainer_items: logger.warning("gainers_response_unrecognized", raw=str(raw_gainers)[:200]) return GainersScanResponse( scan_time=scan_time, market_status=mstat, count_fetched=0, count_passed=0, results=[], ) # Build gainer metadata map gainer_meta: dict[str, dict] = {} for item in gainer_items: sym = _gainer_ticker(item) if not sym: continue meta: dict = {} if isinstance(item, dict): meta["price"] = item.get("price") or item.get("regularMarketPrice") pct_raw = ( item.get("change_percent") or item.get("pct_change") or item.get("regularMarketChangePercent") or item.get("change_pct") ) # Oracle returns change_percent as a whole number (e.g. 12.5 = +12.5%) meta["change_pct"] = pct_raw / 100.0 if pct_raw is not None else None gainer_meta[sym] = meta all_gainer_tickers = sorted(gainer_meta.keys()) logger.info("gainers_fetched", count=len(all_gainer_tickers)) if not all_gainer_tickers: return GainersScanResponse( scan_time=scan_time, market_status=mstat, count_fetched=0, count_passed=0, results=[], ) support_tickers = ["SPY", "QQQ"] all_tickers = sorted({*all_gainer_tickers, *support_tickers}) daily_start = (asof - timedelta(days=120)).isoformat() try: async with OracleClient(base_url=_oracle_url()) as client: # 2. Fetch daily bars for all gainers daily_bars = await fetch_daily_bars_bulk( all_tickers, daily_start, asof_str, client, skip_oracle_when_unhealthy=True, concurrency=8, ) # 3. Inject synthetic today row (Oracle only has yesterday's daily bars when market is open) for sym, bars in daily_bars.items(): if bars: last_bar = max(bars, key=lambda b: b["date"]) if last_bar["date"] < asof_str: daily_bars[sym] = bars + [{ "date": asof_str, "open": last_bar["close"], "high": last_bar["close"], "low": last_bar["close"], "close": last_bar["close"], "volume": 0, "synthetic_today_daily": True, }] # 4. Enrich enrichment = enrich_daily_bars(daily_bars, [asof_str]) # 5. Pre-screen prescreened = orb_pre_screen_candidates( daily_bars, [asof_str], 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, ) candidate_tickers = prescreened.get(asof_str, []) logger.info("gainers_prescreened", total=len(all_gainer_tickers), passed=len(candidate_tickers)) # 6. Fetch intraday only for candidates + support intraday_candidates = {asof_str: sorted({*candidate_tickers, *support_tickers})} all_intraday: dict = {} if candidate_tickers: all_intraday = await fetch_intraday_bulk( intraday_candidates, client, cache=None, skip_oracle_when_unhealthy=True, concurrency=4, ) except Exception as exc: logger.exception("gainers_scan_failed") return GainersScanResponse( scan_time=scan_time, market_status=mstat, count_fetched=len(all_gainer_tickers), count_passed=0, results=[], ) # 6. Run simulation ticker_sectors = {t: "UNKNOWN" for t in all_tickers} day_results: list = [] if candidate_tickers and all_intraday: try: day_results = run_orb_simulation( all_intraday, [asof_str], params, enrichment, ticker_sectors=ticker_sectors, ) except Exception as exc: logger.warning("gainers_simulation_failed", error=str(exc)) # 7. Build results day = day_results[0] if day_results else None traded_tickers = {t.ticker: t for t in (day.trades if day else [])} pre_screened_set = set(candidate_tickers) results: list[GainerResult] = [] for sym in all_gainer_tickers: meta = gainer_meta.get(sym, {}) price = meta.get("price") change_pct = meta.get("change_pct") if sym in traded_tickers: trade = traded_tickers[sym] size_scale = trade.hot_reclaim_size_scale if size_scale is not None and size_scale < 1.0: signal = "SCALE_ENTRY" else: signal = "ENTRY" parts = [] if trade.gap_pct is not None: parts.append(f"gap {trade.gap_pct * 100:+.1f}%") if trade.rvol is not None: parts.append(f"rvol {trade.rvol:.1f}x") if trade.atr_at_entry is not None: parts.append(f"atr {trade.atr_at_entry:.2f}") filter_summary = " | ".join(parts) if parts else "entry confirmed" results.append(GainerResult( ticker=sym, price=price, change_pct=change_pct, signal=signal, scale_factor=size_scale, filter_summary=filter_summary, failure_reason=None, )) elif sym in pre_screened_set: # Passed pre-screen but failed ORB simulation filters e = enrichment.get(sym, {}).get(asof_str, {}) atr = e.get("atr_14") dvol = e.get("avg_dollar_vol_30d") summary = f"atr {atr:.2f} | dvol ${(dvol or 0) / 1e6:.0f}M" if atr else "prescreened" reason = "ORB filters: gap/RVOL/breakout/direction" if day and day.skip_reason: reason = f"day skipped: {day.skip_reason}" results.append(GainerResult( ticker=sym, price=price, change_pct=change_pct, signal="NO_ENTRY", scale_factor=None, filter_summary=summary, failure_reason=reason, )) else: # Failed pre-screen e = enrichment.get(sym, {}).get(asof_str, {}) atr = e.get("atr_14") dvol = e.get("avg_dollar_vol_30d") today_open = e.get("today_open") reasons = [] if today_open is not None and today_open < params.min_price: reasons.append(f"price ${today_open:.2f} < ${params.min_price:.0f}") elif atr is None or atr < params.min_atr_14: reasons.append(f"atr {atr or 'N/A'} < {params.min_atr_14}") elif dvol is None or dvol < params.min_avg_dollar_volume: reasons.append(f"dvol ${(dvol or 0) / 1e6:.0f}M < ${params.min_avg_dollar_volume / 1e6:.0f}M") else: reasons.append("no daily bar on date") results.append(GainerResult( ticker=sym, price=price, change_pct=change_pct, signal="NO_ENTRY", scale_factor=None, filter_summary="pre-screen fail", failure_reason=", ".join(reasons), )) # Sort: ENTRY → SCALE_ENTRY → NO_ENTRY _order = {"ENTRY": 0, "SCALE_ENTRY": 1, "NO_ENTRY": 2} results.sort(key=lambda r: _order.get(r.signal, 3)) count_passed = sum(1 for r in results if r.signal in ("ENTRY", "SCALE_ENTRY")) return GainersScanResponse( scan_time=scan_time, market_status=mstat, count_fetched=len(all_gainer_tickers), count_passed=count_passed, results=results, )