diff --git a/libs/intraday/orb_simulator.py b/libs/intraday/orb_simulator.py index e5f3af2..71b0a1c 100644 --- a/libs/intraday/orb_simulator.py +++ b/libs/intraday/orb_simulator.py @@ -13,6 +13,8 @@ from __future__ import annotations import datetime as dt from collections import deque +from dataclasses import dataclass, field +from typing import Callable from zoneinfo import ZoneInfo from libs.intraday.domain import DayResult, IntradayTrade, ORBStrategyParams @@ -27,6 +29,7 @@ from libs.intraday.simulator import ( _ET = ZoneInfo("America/New_York") +_PREMARKET_OPEN = dt.time(4, 0) _MARKET_OPEN = dt.time(9, 30) _MARKET_CLOSE = dt.time(16, 0) _MIN_BARS = 5 # minimum market-hours bars required @@ -35,6 +38,78 @@ _MIN_BARS = 5 # minimum market-hours bars required _DOJI_THRESHOLD = 0.001 +def _compute_running_vwap(bars: list[dict], up_to_ts: dt.datetime) -> float | None: + """Compute running VWAP from market open up to (and including) the given timestamp. + + Uses typical price = (high + low + close) / 3 for each bar. + Returns None if no bars with volume are found. + """ + cum_pv = 0.0 + cum_vol = 0.0 + for b in bars: + ts = _parse_ts(b["timestamp"]) + if ts > up_to_ts: + break + vol = float(b.get("volume", 0) or 0) + if vol <= 0: + continue + typical = (float(b["high"]) + float(b["low"]) + float(b["close"])) / 3.0 + cum_pv += typical * vol + cum_vol += vol + if cum_vol <= 0: + return None + return cum_pv / cum_vol + + +def _linear_scaler( + value: float | None, + low: float | None, + high: float | None, + floor: float = 1.0, + *, + invert: bool = False, +) -> float: + """Linear interpolation scaler (same logic as simulator._linear_scaler).""" + if value is None or low is None or high is None or high <= low: + return 1.0 + floor = max(0.0, min(1.0, floor)) + if invert: + if value <= low: + return floor + if value >= high: + return 1.0 + frac = (value - low) / (high - low) + return floor + frac * (1.0 - floor) + if value <= low: + return 1.0 + if value >= high: + return floor + frac = (value - low) / (high - low) + return 1.0 - frac * (1.0 - floor) + + +def _orb_vix_size_scaler(vix_value: float | None, params: ORBStrategyParams) -> float: + """Position-size scaler based on VIX level for ORB strategy.""" + return _linear_scaler( + vix_value, + params.vix_size_scale_low, + params.vix_size_scale_high, + params.vix_size_scale_min, + ) + + +@dataclass +class ORBSimulationState: + """Rolling ORB simulation state for chunked backtests.""" + + equity: float + ticker_last_traded: dict[str, str] = field(default_factory=dict) + settled_cash: float | None = None + pending_settlements: list[tuple[str, float]] = field(default_factory=list) + recent_daily_pnl: list[float] = field(default_factory=list) + """Recent daily PnL history for cross-chunk rolling loss filter continuity.""" + + # ── Bar Aggregation ─────────────────────────────────────────────────────── @@ -124,6 +199,36 @@ def _compute_composite_score( ) +def _effective_orb_engine_family(params: ORBStrategyParams) -> str: + family = getattr(params, "engine_family", "quality_breakout") or "quality_breakout" + if family not in { + "classic_breakout", + "quality_breakout", + "compression_breakout", + "gainers_leader", + "leader_followthrough", + "stocks_in_play_dual_regime", + }: + return "quality_breakout" + return family + + +def _compute_premarket_dollar_vol(all_bars: list[dict], date_str: str) -> float: + """Premarket dollar volume proxy from 04:00-09:30 ET bars on the trade date.""" + total = 0.0 + for bar in all_bars: + ts = _parse_ts(bar["timestamp"]).astimezone(_ET) + if ts.date().isoformat() != date_str: + continue + if not (_PREMARKET_OPEN <= ts.time() < _MARKET_OPEN): + continue + price = bar.get("close") or bar.get("open") or 0.0 + volume = bar.get("volume", 0) or 0.0 + if price > 0 and volume > 0: + total += price * volume + return total + + # ── ORB Candidate Selection ──────────────────────────────────────────────── @@ -134,6 +239,8 @@ def compute_orb_candidates( enrichment: dict[str, dict[str, dict]], blacklisted_tickers: set[str] | None = None, spy_bars: list[dict] | None = None, + ticker_sectors: dict[str, str] | None = None, + _stats_out: dict | None = None, ) -> list[dict]: """Identify and rank ORB candidates for a given trading day. @@ -160,6 +267,7 @@ def compute_orb_candidates( Each dict: {ticker, orb_bar, direction, rvol, gap_pct, atr, first_bar_dollar_vol, score, mkt_bars} """ market_open = _market_open_ts(date_str) + engine_family = _effective_orb_engine_family(params) raw_candidates: list[dict] = [] @@ -177,7 +285,8 @@ def compute_orb_candidates( # Verify first bar is near market open (allow data irregularities up to 10 min) first_bar_ts = _parse_ts(mkt_bars[0]["timestamp"]) - if abs((first_bar_ts - market_open).total_seconds() / 60) > 10: + _late_diff = abs((first_bar_ts - market_open).total_seconds() / 60) + if _late_diff > 10: _f_late += 1 continue @@ -199,6 +308,18 @@ def compute_orb_candidates( "close": orb_bars_raw[-1]["close"], "volume": sum(b.get("volume", 0) or 0 for b in orb_bars_raw), } + orb_vwap = None + orb_vwap_num = 0.0 + orb_vwap_den = 0.0 + for bar in orb_bars_raw: + bar_vwap = bar.get("vwap") + bar_volume = float(bar.get("volume", 0) or 0) + if bar_vwap is None or bar_volume <= 0: + continue + orb_vwap_num += float(bar_vwap) * bar_volume + orb_vwap_den += bar_volume + if orb_vwap_den > 0: + orb_vwap = orb_vwap_num / orb_vwap_den # Price filter (use ORB candle open as current price) open_price = orb_bar.get("open", 0) @@ -208,10 +329,31 @@ def compute_orb_candidates( # Direction filter (uses aggregated ORB candle open/close) direction = classify_orb_candle(orb_bar) - if params.entry_direction == "long_only" and direction != "bullish": - _f_dir += 1 - continue - if direction == "doji": + followthrough_engine = engine_family in { + "gainers_leader", + "leader_followthrough", + "stocks_in_play_dual_regime", + } + allow_doji_breakout = ( + followthrough_engine + and bool(getattr(params, "allow_doji_breakout", False)) + ) + allow_red_to_green_breakout = ( + followthrough_engine + and bool(getattr(params, "allow_red_to_green_breakout", False)) + ) + if params.entry_direction == "long_only": + if direction == "bearish" and not allow_red_to_green_breakout: + _f_dir += 1 + continue + if direction == "bearish" and allow_red_to_green_breakout: + direction = "bullish" + if direction == "doji" and not allow_doji_breakout: + _f_dir += 1 + continue + if direction == "doji" and allow_doji_breakout: + direction = "bullish" + elif direction == "doji": _f_dir += 1 continue @@ -221,11 +363,20 @@ def compute_orb_candidates( avg_dollar_vol = ticker_enrich.get("avg_dollar_vol_30d") avg_daily_vol = ticker_enrich.get("avg_daily_vol_14d") prev_close = ticker_enrich.get("prev_close") + premarket_dollar_vol = _compute_premarket_dollar_vol(all_bars, date_str) # ATR filter if atr is None or atr < params.min_atr_14: _f_atr += 1 continue + if prev_close and prev_close > 0: + atr_ratio = atr / prev_close + if params.min_atr_pct is not None and atr_ratio < params.min_atr_pct: + _f_atr += 1 + continue + if params.max_atr_pct is not None and atr_ratio > params.max_atr_pct: + _f_atr += 1 + continue # Dollar volume filter if avg_dollar_vol is None or avg_dollar_vol < params.min_avg_dollar_volume: @@ -243,13 +394,39 @@ def compute_orb_candidates( gap_pct = 0.0 if prev_close and prev_close > 0: gap_pct = (open_price - prev_close) / prev_close + abs_gap_pct = abs(gap_pct) + + used_small_gap_attention_override = False + if params.min_abs_gap_pct is not None and abs_gap_pct < params.min_abs_gap_pct: + small_gap_attention_override = ( + followthrough_engine + and getattr(params, "small_gap_attention_override_premarket_dollar_vol", None) is not None + and premarket_dollar_vol + >= float(getattr(params, "small_gap_attention_override_premarket_dollar_vol")) + ) + if small_gap_attention_override: + small_gap_rvol_min = getattr(params, "small_gap_attention_override_rvol", None) + if small_gap_rvol_min is not None and (rvol is None or rvol < float(small_gap_rvol_min)): + _f_gap += 1 + continue + used_small_gap_attention_override = True + else: + _f_gap += 1 + continue - # Max gap filter: exclude over-extended stocks (gap-up > threshold) - # Stocks that open >10% above prev_close are prone to mean-reversion, not continuation. + # Optional max gap filter. Useful for classical ORB continuation, but typically + # disabled in gainers/leader style engines that explicitly seek outsized movers. if params.max_gap_pct is not None and gap_pct > params.max_gap_pct: _f_gap += 1 continue + if ( + params.min_premarket_dollar_vol is not None + and premarket_dollar_vol < params.min_premarket_dollar_vol + ): + _f_dolvol += 1 + continue + # First-bar dollar volume (ORB window total) first_bar_dollar_vol = orb_vol * open_price @@ -264,8 +441,27 @@ def compute_orb_candidates( body_ratio = max((orb_close - orb_open_price) / orb_range, 0.0) else: body_ratio = max((orb_open_price - orb_close) / orb_range, 0.0) + close_location = (orb_close - orb_bar["low"]) / orb_range else: body_ratio = 0.0 + close_location = 0.5 + + # ORB range quality filter: skip if range is too narrow or too wide relative to ATR + if atr > 0 and orb_range > 0: + orb_range_atr_ratio = orb_range / atr + if params.orb_range_atr_min is not None and orb_range_atr_ratio < params.orb_range_atr_min: + _f_dir += 1 + continue + if params.orb_range_atr_max is not None and orb_range_atr_ratio > params.orb_range_atr_max: + _f_dir += 1 + continue + + if engine_family != "classic_breakout" and body_ratio < getattr(params, "min_body_ratio", 0.0): + _f_dir += 1 + continue + if engine_family == "leader_followthrough" and close_location < getattr(params, "min_close_location", 0.0): + _f_dir += 1 + continue # 5-day prior momentum in the direction of the breakout. # For longs: positive ret_5d = stock already trending up (momentum alignment). @@ -276,19 +472,153 @@ def compute_orb_candidates( else: momentum = 0.0 + entropy_20d = ticker_enrich.get("entropy_20d") + atr_ratio_10_60 = ticker_enrich.get("atr_ratio_10_60") + range_compression_10_60 = ticker_enrich.get("range_compression_10_60") + gap_zscore_20d = ticker_enrich.get("gap_zscore_20d") + event_flag = bool(ticker_enrich.get("event_flag")) + raw_event_types = ticker_enrich.get("event_types") or [] + event_types = [str(v) for v in raw_event_types if str(v)] + event_score = float(ticker_enrich.get("event_score") or 0.0) + attention_wiki_spike_10d = ticker_enrich.get("attention_wiki_spike_10d") + attention_wiki_zscore_20d = ticker_enrich.get("attention_wiki_zscore_20d") + attention_article_count_3d = int(ticker_enrich.get("attention_article_count_3d") or 0) + attention_us_article_count_3d = int(ticker_enrich.get("attention_us_article_count_3d") or 0) + attention_resolver_confidence = float(ticker_enrich.get("attention_resolver_confidence") or 0.0) + + allowed_event_types = {str(v).lower() for v in getattr(params, "allowed_event_types", []) if str(v)} + if allowed_event_types and event_flag: + if not any(str(event_type).lower() in allowed_event_types for event_type in event_types): + event_flag = False + event_score = 0.0 + event_types = [] + + if engine_family == "compression_breakout": + min_entropy = getattr(params, "min_entropy", None) + max_entropy = getattr(params, "max_entropy", None) + compression_ratio_max = getattr(params, "compression_ratio_max", None) + if min_entropy is not None and (entropy_20d is None or entropy_20d < min_entropy): + _f_rvol += 1 + continue + if max_entropy is not None and (entropy_20d is None or entropy_20d > max_entropy): + _f_rvol += 1 + continue + if compression_ratio_max is not None and ( + range_compression_10_60 is None or range_compression_10_60 > compression_ratio_max + ): + _f_rvol += 1 + continue + + if engine_family == "stocks_in_play_dual_regime": + if getattr(params, "require_event_flag", False) and not event_flag: + _f_gap += 1 + continue + if ( + getattr(params, "attention_min_wiki_spike_10d", None) is not None + and ( + attention_wiki_spike_10d is None + or attention_wiki_spike_10d < float(getattr(params, "attention_min_wiki_spike_10d")) + ) + ): + _f_gap += 1 + continue + if ( + getattr(params, "attention_min_wiki_zscore_20d", None) is not None + and ( + attention_wiki_zscore_20d is None + or attention_wiki_zscore_20d < float(getattr(params, "attention_min_wiki_zscore_20d")) + ) + ): + _f_gap += 1 + continue + if ( + getattr(params, "attention_min_article_count_3d", None) is not None + and attention_article_count_3d < int(getattr(params, "attention_min_article_count_3d")) + ): + _f_gap += 1 + continue + if ( + getattr(params, "attention_min_us_article_count_3d", None) is not None + and attention_us_article_count_3d < int(getattr(params, "attention_min_us_article_count_3d")) + ): + _f_gap += 1 + continue + if ( + getattr(params, "attention_min_resolver_confidence", None) is not None + and attention_resolver_confidence < float(getattr(params, "attention_min_resolver_confidence")) + ): + _f_gap += 1 + continue + if direction == "bullish": + if close_location < getattr(params, "min_close_location", 0.0): + _f_dir += 1 + continue + if ( + getattr(params, "require_vwap_confirmation", False) + and orb_vwap is not None + and orb_close < orb_vwap + ): + _f_dir += 1 + continue + elif direction == "bearish": + if not getattr(params, "allow_failed_orb_short", False): + _f_dir += 1 + continue + if gap_pct <= 0: + _f_gap += 1 + continue + if close_location > getattr(params, "max_close_location_short", 1.0): + _f_dir += 1 + continue + if ( + getattr(params, "require_vwap_confirmation", False) + and orb_vwap is not None + and orb_close > orb_vwap + ): + _f_dir += 1 + continue + else: + _f_dir += 1 + continue + raw_candidates.append({ "ticker": ticker, + "sector": (ticker_sectors or {}).get(ticker, "UNKNOWN"), "orb_bar": orb_bar, "direction": direction, "rvol": rvol, "gap_pct": gap_pct, + "abs_gap_pct": abs_gap_pct, "atr": atr, "first_bar_dollar_vol": first_bar_dollar_vol, + "premarket_dollar_vol": premarket_dollar_vol, "body_ratio": body_ratio, + "close_location": close_location, "momentum": momentum, + "entropy_20d": entropy_20d or 0.0, + "atr_ratio_10_60": atr_ratio_10_60 or 0.0, + "range_compression_10_60": range_compression_10_60, + "gap_zscore_20d": gap_zscore_20d or 0.0, + "event_flag": event_flag, + "event_types": event_types, + "event_score": event_score, + "attention_wiki_spike_10d": attention_wiki_spike_10d or 0.0, + "attention_article_count_3d": attention_article_count_3d, + "attention_us_article_count_3d": attention_us_article_count_3d, + "attention_resolver_confidence": attention_resolver_confidence, + "orb_vwap": orb_vwap, + "used_small_gap_attention_override": used_small_gap_attention_override, + "orb_return": ((orb_close - orb_open_price) / orb_open_price) if orb_open_price > 0 else 0.0, "mkt_bars": mkt_bars, }) + _filter_stats = { + "gap": _f_gap, "rvol": _f_rvol, "atr": _f_atr, "dolvol": _f_dolvol, + "dir": _f_dir, "no_bars": _f_no_bars, "late": _f_late, "price": _f_price, + } + if _stats_out is not None: + _stats_out.update(_filter_stats) + if not raw_candidates: total = len(bars_by_ticker) import sys @@ -301,30 +631,135 @@ def compute_orb_candidates( ) return [] + if engine_family == "stocks_in_play_dual_regime": + sector_returns: dict[tuple[str, str], list[float]] = {} + for cand in raw_candidates: + key = (str(cand.get("sector") or "UNKNOWN"), str(cand["direction"])) + sector_returns.setdefault(key, []).append(float(cand.get("orb_return") or 0.0)) + filtered_candidates: list[dict] = [] + for cand in raw_candidates: + key = (str(cand.get("sector") or "UNKNOWN"), str(cand["direction"])) + sector_avg = ( + sum(sector_returns.get(key, [0.0])) / len(sector_returns.get(key, [0.0])) + if sector_returns.get(key) + else 0.0 + ) + sector_relative_strength = float(cand.get("orb_return") or 0.0) - sector_avg + cand["sector_relative_strength"] = sector_relative_strength + if ( + cand["direction"] == "bullish" + and getattr(params, "min_sector_relative_strength", None) is not None + and sector_relative_strength < float(getattr(params, "min_sector_relative_strength")) + ): + continue + filtered_candidates.append(cand) + raw_candidates = filtered_candidates + + if not raw_candidates: + return [] + # Normalize and score rvol_vals = [c["rvol"] for c in raw_candidates] - gap_vals = [max(c["gap_pct"], 0.0) for c in raw_candidates] # clip negative gaps + if engine_family in {"gainers_leader", "leader_followthrough"}: + gap_vals = [c["abs_gap_pct"] for c in raw_candidates] + else: + gap_vals = [max(c["gap_pct"], 0.0) for c in raw_candidates] # clip negative gaps dolvol_vals = [c["first_bar_dollar_vol"] for c in raw_candidates] + premarket_dolvol_vals = [c["premarket_dollar_vol"] for c in raw_candidates] body_vals = [c["body_ratio"] for c in raw_candidates] + close_location_vals = [c["close_location"] for c in raw_candidates] momentum_vals = [max(c["momentum"], 0.0) for c in raw_candidates] # only reward aligned momentum + event_vals = [c["event_score"] for c in raw_candidates] + attention_wiki_vals = [c["attention_wiki_spike_10d"] for c in raw_candidates] + attention_news_vals = [ + max(c["attention_article_count_3d"], c["attention_us_article_count_3d"]) + for c in raw_candidates + ] + entropy_vals = [c["entropy_20d"] for c in raw_candidates] + atr_ratio_vals = [c["atr_ratio_10_60"] for c in raw_candidates] + gap_zscore_vals = [c["gap_zscore_20d"] for c in raw_candidates] + structure_vals = [ + c["close_location"] if c["direction"] == "bullish" else 1.0 - c["close_location"] + for c in raw_candidates + ] norm_rvol = _normalize_scores(rvol_vals) norm_gap = _normalize_scores(gap_vals) norm_dolvol = _normalize_scores(dolvol_vals) + norm_premarket_dolvol = _normalize_scores(premarket_dolvol_vals) norm_body = _normalize_scores(body_vals) + norm_close_location = _normalize_scores(close_location_vals) + norm_structure = _normalize_scores(structure_vals) norm_momentum = _normalize_scores(momentum_vals) + norm_event = _normalize_scores(event_vals) + norm_attention_wiki = _normalize_scores(attention_wiki_vals) + norm_attention_news = _normalize_scores(attention_news_vals) + norm_entropy = _normalize_scores(entropy_vals) + norm_atr_ratio = _normalize_scores(atr_ratio_vals) + norm_gap_zscore = _normalize_scores(gap_zscore_vals) for i, cand in enumerate(raw_candidates): - cand["score"] = ( + score = ( norm_rvol[i] * params.weight_rvol + norm_gap[i] * params.weight_gap + norm_dolvol[i] * params.weight_dollar_vol - + norm_body[i] * params.weight_body_ratio - + norm_momentum[i] * params.weight_momentum + + norm_premarket_dolvol[i] * params.weight_premarket_dollar_vol ) + if engine_family != "classic_breakout": + score += norm_body[i] * params.weight_body_ratio + score += norm_momentum[i] * params.weight_momentum + if engine_family in { + "gainers_leader", "leader_followthrough", "stocks_in_play_dual_regime" + }: + score += norm_structure[i] * params.weight_close_location + score += norm_gap_zscore[i] * params.weight_gap_zscore + if engine_family == "stocks_in_play_dual_regime": + score += norm_event[i] * params.weight_event_catalyst + score += norm_attention_wiki[i] * params.weight_attention_wiki + score += norm_attention_news[i] * params.weight_attention_news + if engine_family in { + "compression_breakout", "gainers_leader", "leader_followthrough", + "stocks_in_play_dual_regime", + }: + score += norm_entropy[i] * params.weight_entropy + score += norm_atr_ratio[i] * params.weight_atr_ratio + if engine_family == "compression_breakout": + # gap_zscore only added here for compression_breakout; + # gainers_leader/leader_followthrough already add it above + score += norm_gap_zscore[i] * params.weight_gap_zscore + cand["score"] = score # Sort by score descending, take top N raw_candidates.sort(key=lambda c: c["score"], reverse=True) + max_per_sector = getattr(params, "max_candidates_per_sector", None) + max_small_gap_attention = getattr(params, "max_small_gap_attention_candidates", None) + if ( + (max_per_sector is not None and max_per_sector > 0) + or (max_small_gap_attention is not None and max_small_gap_attention >= 0) + ): + selected: list[dict] = [] + sector_counts: dict[str, int] = {} + small_gap_attention_count = 0 + for cand in raw_candidates: + sector = str(cand.get("sector") or "UNKNOWN") + if max_per_sector is not None and max_per_sector > 0: + if sector_counts.get(sector, 0) >= max_per_sector: + continue + if ( + max_small_gap_attention is not None + and max_small_gap_attention >= 0 + and cand.get("used_small_gap_attention_override") + ): + if small_gap_attention_count >= max_small_gap_attention: + continue + selected.append(cand) + if max_per_sector is not None and max_per_sector > 0: + sector_counts[sector] = sector_counts.get(sector, 0) + 1 + if cand.get("used_small_gap_attention_override"): + small_gap_attention_count += 1 + if len(selected) >= params.max_candidates: + break + return selected return raw_candidates[: params.max_candidates] @@ -384,6 +819,11 @@ def simulate_orb_trade( ticker: str, available_cash: float | None = None, sizing_capital: float | None = None, + score_rank_pct: float = 0.0, + prev_close: float | None = None, + entry_after_ts: dt.datetime | None = None, + spy_bars: list[dict] | None = None, + is_soft_day: bool = False, ) -> IntradayTrade | None: """Simulate a single ORB trade with ATR-based stops. @@ -414,7 +854,17 @@ def simulate_orb_trade( if atr <= 0: return None - stop_distance = atr * params.atr_stop_multiplier + effective_atr_stop_mult = ( + params.atr_stop_multiplier_weak + if (is_soft_day and params.atr_stop_multiplier_weak is not None) + else params.atr_stop_multiplier + ) + effective_breakeven_at_r = ( + params.breakeven_at_r_weak + if (is_soft_day and params.breakeven_at_r_weak is not None) + else params.breakeven_at_r + ) + stop_distance = atr * effective_atr_stop_mult if stop_distance <= 0: return None @@ -457,12 +907,29 @@ def simulate_orb_trade( if ts <= orb_ts: continue # skip ORB bar and anything before it - # Check timeout - if ts > timeout_ts: + # Re-entry mode: skip bars before the previous exit + if entry_after_ts is not None and ts <= entry_after_ts: + continue + + # Check timeout (disabled for re-entries — they happen later in the day) + if entry_after_ts is None and ts > timeout_ts: return None # no fill before timeout # Check breakout - if direction == "long" and b["high"] >= breakout_level: + # entry_on_bar_close: require bar CLOSE above/below level (filters wick-only touches) + use_bar_close_entry = params.entry_on_bar_close + if direction == "long": + bar_triggered = ( + b["close"] >= breakout_level if use_bar_close_entry + else b["high"] >= breakout_level + ) + else: + bar_triggered = ( + b["close"] <= breakout_level if use_bar_close_entry + else b["low"] <= breakout_level + ) + + if bar_triggered and direction == "long": if group_size > 1: # Signal is only known at the END of the aggregated bar. # Fill at the first 5-min bar's open after the signal bar ends — @@ -475,11 +942,15 @@ def simulate_orb_trade( return None # near close — no next bar available to fill entry_price_raw = max(breakout_level, fill_raw["open"]) entry_bar = fill_raw # entry_ts and entry_time use the fill bar + elif use_bar_close_entry: + # Enter at bar close — trader waits for bar to complete + entry_price_raw = b["close"] + entry_bar = b else: entry_price_raw = max(breakout_level, b["open"]) entry_bar = b break - elif direction == "short" and b["low"] <= breakout_level: + elif bar_triggered and direction == "short": if group_size > 1: agg_ts = _parse_ts(b["timestamp"]) fill_raw = next( @@ -489,6 +960,9 @@ def simulate_orb_trade( return None entry_price_raw = min(breakout_level, fill_raw["open"]) entry_bar = fill_raw + elif use_bar_close_entry: + entry_price_raw = b["close"] + entry_bar = b else: entry_price_raw = min(breakout_level, b["open"]) entry_bar = b @@ -497,16 +971,142 @@ def simulate_orb_trade( if entry_bar is None: return None # no breakout fill + # --- Pullback continuation entry --- + # Instead of entering on the breakout, wait for a pullback and continuation. + # 1. Record the breakout, then look for a bar that retraces from the post-breakout peak + # 2. After the pullback, look for continuation (new bar making progress) + # 3. Enter at the continuation bar close with stop at pullback extreme + if params.pullback_entry: + initial_breakout_bar = entry_bar + initial_breakout_ts = _parse_ts(initial_breakout_bar["timestamp"]) + + # Reset entry — we'll find a better one after pullback + entry_bar = None + entry_price_raw = 0.0 + + post_breakout_peak = breakout_level + pullback_extreme = breakout_level # lowest point during pullback (long) + pullback_found = False + bars_after_breakout = 0 + + for b in mkt_bars: + ts = _parse_ts(b["timestamp"]) + if ts <= initial_breakout_ts: + continue + if ts >= exit_target: + break # too late in the day + + bars_after_breakout += 1 + if bars_after_breakout > params.pullback_max_bars: + break + + if direction == "long": + post_breakout_peak = max(post_breakout_peak, b["high"]) + move_from_breakout = post_breakout_peak - breakout_level + + if not pullback_found: + # Look for pullback: price retraces from peak + if move_from_breakout > 0: + retracement = (post_breakout_peak - b["low"]) / move_from_breakout + if retracement >= params.pullback_min_retracement_pct: + pullback_found = True + pullback_extreme = b["low"] + continue + + # Pullback found — track the low and look for continuation + pullback_extreme = min(pullback_extreme, b["low"]) + + # Continuation: bar closes green and above prior bar's high + if b["close"] > b["open"] and b["close"] > pullback_extreme: + entry_price_raw = b["close"] + entry_bar = b + if params.pullback_stop_at_low and pullback_extreme < entry_price_raw: + stop_distance = entry_price_raw - pullback_extreme + break + + else: # short + post_breakout_peak = min(post_breakout_peak, b["low"]) # trough + move_from_breakout = breakout_level - post_breakout_peak + + if not pullback_found: + if move_from_breakout > 0: + retracement = (b["high"] - post_breakout_peak) / move_from_breakout + if retracement >= params.pullback_min_retracement_pct: + pullback_found = True + pullback_extreme = b["high"] + continue + + pullback_extreme = max(pullback_extreme, b["high"]) + + if b["close"] < b["open"] and b["close"] < pullback_extreme: + entry_price_raw = b["close"] + entry_bar = b + if params.pullback_stop_at_low and pullback_extreme > entry_price_raw: + stop_distance = pullback_extreme - entry_price_raw + break + + if entry_bar is None: + return None # no pullback-continuation pattern found + + # --- Breakout volume confirmation --- + # Reject breakouts on thin volume (low conviction, likely to fail). + if params.min_breakout_rel_vol is not None: + entry_vol = entry_bar.get("volume", 0) or 0 + # Average volume of all post-ORB bars (excluding ORB bar itself) + post_orb_vols = [ + b.get("volume", 0) or 0 + for b in mkt_bars + if _parse_ts(b["timestamp"]) > orb_ts + ] + avg_bar_vol = sum(post_orb_vols) / len(post_orb_vols) if post_orb_vols else 0 + if avg_bar_vol > 0 and entry_vol < avg_bar_vol * params.min_breakout_rel_vol: + return None # breakout bar volume too low + # --- Position sizing (must happen before stop check so shares are known) --- initial_stop = ( entry_price_raw - stop_distance if direction == "long" else entry_price_raw + stop_distance ) + # --- Confirmation bar requirement (lookahead-free) --- + # After breakout, wait one bar. If confirmation bar closes in the right direction, + # enter at the confirmation bar's close (the price available AFTER seeing confirmation). + # This avoids retroactive cancellation bias — unconfirmed trades simply don't enter. + if params.require_confirmation_bar: + confirm_bar = None + for b in mkt_bars: + ts = _parse_ts(b["timestamp"]) + if ts <= _parse_ts(entry_bar["timestamp"]): + continue + confirm_bar = b + break + if confirm_bar is None: + return None # no bar after entry (near close) + if direction == "long" and confirm_bar["close"] < entry_price_raw: + return None # confirmation failed — don't enter + elif direction == "short" and confirm_bar["close"] > entry_price_raw: + return None # confirmation failed — don't enter + # Confirmation passed — shift entry to confirmation bar's close + # (the price available to the trader AFTER observing the confirmation) + entry_price_raw = confirm_bar["close"] + entry_bar = confirm_bar + entry_ts = _parse_ts(confirm_bar["timestamp"]) + # Recalculate stop with new entry price + initial_stop = ( + entry_price_raw - stop_distance if direction == "long" + else entry_price_raw + stop_distance + ) + # Use sizing_capital for position sizing (simple/compound mode). # sizing_capital = initial_capital when compound_returns=False, else current equity. cap = sizing_capital if sizing_capital is not None else equity - risk_dollars = cap * params.risk_per_trade_pct + # Score-based position sizing: top-ranked candidates get larger positions + if params.score_sizing_multiplier is not None and params.score_sizing_multiplier > 1.0: + # score_rank_pct: 1.0 = top rank, 0.0 = bottom rank + sizing_mult = 1.0 + score_rank_pct * (params.score_sizing_multiplier - 1.0) + else: + sizing_mult = 1.0 + risk_dollars = cap * params.risk_per_trade_pct * sizing_mult shares_from_risk = risk_dollars / stop_distance max_shares_by_capital = (cap * params.max_position_pct) / entry_price_raw @@ -533,7 +1133,8 @@ def simulate_orb_trade( # Only apply for 5-min bars (group_size == 1). For 30-min (or larger) bars, # we skip same-bar stop detection — the user only checks every N minutes, # so the stop is evaluated at the NEXT bar's open, not within the entry bar. - if group_size == 1 and direction == "long" and entry_bar["low"] <= initial_stop: + # Also skip when entry_on_bar_close — trader enters at bar close, not exposed to intra-bar action. + if group_size == 1 and not params.entry_on_bar_close and direction == "long" and entry_bar["low"] <= initial_stop: exit_price_raw = initial_stop exit_price = _apply_slippage_exit(exit_price_raw, slippage) pnl_pct = (exit_price - entry_price_filled) / entry_price_filled @@ -559,8 +1160,9 @@ def simulate_orb_trade( rvol=round(rvol, 3), atr_at_entry=round(atr, 4), r_multiple_at_exit=-1.0, + stop_level_at_exit="initial", ) - if group_size == 1 and direction == "short" and entry_bar["high"] >= initial_stop: + if group_size == 1 and not params.entry_on_bar_close and direction == "short" and entry_bar["high"] >= initial_stop: exit_price_raw = initial_stop exit_price = _apply_slippage_entry(exit_price_raw, slippage) pnl_pct = (entry_price_filled - exit_price) / entry_price_filled @@ -586,6 +1188,7 @@ def simulate_orb_trade( rvol=round(rvol, 3), atr_at_entry=round(atr, 4), r_multiple_at_exit=-1.0, + stop_level_at_exit="initial", ) # --- Phase 2: Manage position --- @@ -598,9 +1201,61 @@ def simulate_orb_trade( exit_time_str = entry_bar["timestamp"] exit_reason = "close" final_r = 0.0 + stop_level = "initial" # 'initial' | 'breakeven' | 'trailing' — for diagnostics use_atr_trail = params.trailing_stop_atr_multiplier > 0 + # VWAP exit setup + use_vwap_exit = params.vwap_exit_mode in ("exit", "floor") + vwap_exit_buffer = atr * params.vwap_exit_buffer_atr + + # Max hold time exit + max_hold_exit_ts: dt.datetime | None = None + if params.max_hold_minutes is not None: + max_hold_exit_ts = entry_ts + dt.timedelta(minutes=params.max_hold_minutes) + + # Gap fill: track previous close for emergency exit + use_gap_fill_exit = params.exit_on_gap_fill and prev_close is not None and prev_close > 0 + + # Track peak R for VWAP activation threshold + peak_r = 0.0 + + # Time-decay trailing: precompute decay schedule + use_time_decay = ( + params.time_decay_start_minutes is not None + and use_atr_trail + ) + if use_time_decay: + decay_start_ts = market_open + dt.timedelta(minutes=params.time_decay_start_minutes) + decay_end_ts = exit_target # decay completes at exit time + decay_span = (decay_end_ts - decay_start_ts).total_seconds() + else: + decay_start_ts = decay_end_ts = None + decay_span = 0.0 + + # SPY intraday guard: precompute SPY open price for intraday comparison + spy_guard_active = ( + params.spy_intraday_guard_pct is not None + and spy_bars is not None + and len(spy_bars) > 0 + ) + spy_open = 0.0 + if spy_guard_active: + spy_open = spy_bars[0].get("open", 0.0) if spy_bars else 0.0 + + # Partial exit state (Option B: blended single trade result) + original_shares = shares + remaining_shares = shares + partial_exited = False + partial_pnl = 0.0 + partial_exit_r_val: float | None = None + partial_exit_slippage = 0.0 + + # Pyramid state: track add-on legs separately for PnL + pyramid_count = 0 + pyramid_legs: list[tuple[int, float, float]] = [] # (shares, entry_filled, entry_raw) + pyramid_total_shares = 0 + for b in mkt_bars: ts = _parse_ts(b["timestamp"]) if ts <= entry_ts: @@ -624,24 +1279,130 @@ def simulate_orb_trade( final_r = (exit_price_raw - entry_price_raw) / stop_distance break + # ── Step 1b: Gap fill protection ── + if use_gap_fill_exit and bar_close < prev_close: + exit_price_raw = bar_close + exit_time_str = b["timestamp"] + exit_reason = "gap_fill" + final_r = (exit_price_raw - entry_price_raw) / stop_distance + break + + # ── Step 1c: Max hold time exit ── + if max_hold_exit_ts is not None and ts >= max_hold_exit_ts: + exit_price_raw = bar_close + exit_time_str = b["timestamp"] + exit_reason = "max_hold" + final_r = (exit_price_raw - entry_price_raw) / stop_distance + break + # ── Step 2: Update peak using actual bar high ── peak_price = max(peak_price, bar_high) # ── Step 3: R-multiple from close (trader sees close to decide adjustments) ── current_r = (bar_close - entry_price_raw) / stop_distance + peak_r = max(peak_r, current_r) + + # ── Step 3b: VWAP exit check ── + if use_vwap_exit and peak_r >= params.vwap_exit_after_r: + running_vwap = _compute_running_vwap(mkt_bars, ts) + if running_vwap is not None: + vwap_level = running_vwap - vwap_exit_buffer + if params.vwap_exit_mode == "exit" and bar_close < vwap_level: + exit_price_raw = bar_close + exit_time_str = b["timestamp"] + exit_reason = "vwap_exit" + final_r = current_r + break + elif params.vwap_exit_mode == "floor" and trailing_active: + # VWAP as trailing stop floor + if vwap_level > current_stop: + current_stop = vwap_level + + # ── Step 3c: Profit target exit ── + if params.profit_target_r is not None and current_r >= params.profit_target_r: + exit_price_raw = bar_close + exit_time_str = b["timestamp"] + exit_reason = "profit_target" + final_r = current_r + break + + # Partial exit: lock in profits at configured R-multiple + if ( + params.partial_exit_at_r is not None + and not partial_exited + and current_r >= params.partial_exit_at_r + ): + p_shares = int(original_shares * params.partial_exit_pct) + if p_shares > 0 and p_shares < remaining_shares: + p_exit_raw = bar_close + p_exit = _apply_slippage_exit(p_exit_raw, slippage) + partial_pnl = (p_exit - entry_price_filled) * p_shares + partial_exit_slippage = abs(p_exit - p_exit_raw) * p_shares + remaining_shares -= p_shares + partial_exited = True + partial_exit_r_val = current_r + # Protect remainder: move stop to breakeven if not already + if current_stop < entry_price_raw: + current_stop = entry_price_raw + stop_level = "breakeven" + + # Pyramiding: add to winning position at configured R-multiple + if ( + params.pyramid_at_r is not None + and pyramid_count < params.pyramid_max_adds + and current_r >= params.pyramid_at_r * (1 + pyramid_count) + ): + add_shares = int(original_shares * params.pyramid_add_pct) + if add_shares > 0: + p_entry_raw = bar_close + p_entry_filled = _apply_slippage_entry(p_entry_raw, slippage) + pyramid_legs.append((add_shares, p_entry_filled, p_entry_raw)) + pyramid_total_shares += add_shares + pyramid_count += 1 # Move stop to breakeven at configured R-multiple - if current_r >= params.breakeven_at_r and current_stop < entry_price_raw: + if current_r >= effective_breakeven_at_r and current_stop < entry_price_raw: current_stop = entry_price_raw + stop_level = "breakeven" # Activate trailing stop at configured R-multiple if current_r >= params.trailing_at_r: trailing_active = True + stop_level = "trailing" # ── Step 4: Update trailing stop for NEXT bar ── if trailing_active: if use_atr_trail: - candidate_stop = peak_price - atr * params.trailing_stop_atr_multiplier + # Two-stage trailing: wider trail initially, tightens at a higher R + atr_mult = params.trailing_stop_atr_multiplier + # Gap-adaptive trailing: override base multiplier based on gap size + if params.gap_trail_wide_threshold is not None: + if abs(gap_pct) > params.gap_trail_wide_threshold: + atr_mult = params.gap_trail_wide_atr_multiplier + elif params.gap_trail_tight_atr_multiplier is not None: + atr_mult = params.gap_trail_tight_atr_multiplier + if ( + params.trailing_tighten_at_r is not None + and current_r >= params.trailing_tighten_at_r + and params.trailing_stop_atr_multiplier_tight > 0 + ): + atr_mult = params.trailing_stop_atr_multiplier_tight + # Time-decay: linearly shrink trail width toward close + if use_time_decay and ts >= decay_start_ts and decay_span > 0: + elapsed = min((ts - decay_start_ts).total_seconds(), decay_span) + decay_pct = elapsed / decay_span # 0 → 1 + atr_mult *= 1.0 - decay_pct * (1.0 - params.time_decay_factor) + # SPY intraday guard: tighten trail when SPY drops from open + if spy_guard_active and spy_bars: + spy_bar = next( + (sb for sb in spy_bars if sb.get("timestamp") == b.get("timestamp")), + None, + ) + if spy_bar is not None and spy_open > 0: + spy_change = (spy_bar["close"] - spy_open) / spy_open + if spy_change < params.spy_intraday_guard_pct: + atr_mult *= params.spy_intraday_guard_tighten + candidate_stop = peak_price - atr * atr_mult else: swing_low_window.append(bar_close) candidate_stop = max(swing_low_window) @@ -657,22 +1418,125 @@ def simulate_orb_trade( final_r = (entry_price_raw - exit_price_raw) / stop_distance break + # ── Step 1b: Gap fill protection (short: price rises above prev_close) ── + if use_gap_fill_exit and bar_close > prev_close: + exit_price_raw = bar_close + exit_time_str = b["timestamp"] + exit_reason = "gap_fill" + final_r = (entry_price_raw - exit_price_raw) / stop_distance + break + + # ── Step 1c: Max hold time exit ── + if max_hold_exit_ts is not None and ts >= max_hold_exit_ts: + exit_price_raw = bar_close + exit_time_str = b["timestamp"] + exit_reason = "max_hold" + final_r = (entry_price_raw - exit_price_raw) / stop_distance + break + # ── Step 2: Update trough using actual bar low ── peak_price = min(peak_price, bar_low) # ── Step 3: R-multiple from close ── current_r = (entry_price_raw - bar_close) / stop_distance + peak_r = max(peak_r, current_r) + + # ── Step 3b: VWAP exit check (short) ── + if use_vwap_exit and peak_r >= params.vwap_exit_after_r: + running_vwap = _compute_running_vwap(mkt_bars, ts) + if running_vwap is not None: + vwap_level = running_vwap + vwap_exit_buffer + if params.vwap_exit_mode == "exit" and bar_close > vwap_level: + exit_price_raw = bar_close + exit_time_str = b["timestamp"] + exit_reason = "vwap_exit" + final_r = current_r + break + elif params.vwap_exit_mode == "floor" and trailing_active: + if vwap_level < current_stop: + current_stop = vwap_level + + # ── Step 3c: Profit target exit (short) ── + if params.profit_target_r is not None and current_r >= params.profit_target_r: + exit_price_raw = bar_close + exit_time_str = b["timestamp"] + exit_reason = "profit_target" + final_r = current_r + break - if current_r >= params.breakeven_at_r and current_stop > entry_price_raw: + # Partial exit (short) + if ( + params.partial_exit_at_r is not None + and not partial_exited + and current_r >= params.partial_exit_at_r + ): + p_shares = int(original_shares * params.partial_exit_pct) + if p_shares > 0 and p_shares < remaining_shares: + p_exit_raw = bar_close + p_exit = _apply_slippage_entry(p_exit_raw, slippage) + partial_pnl = (entry_price_filled - p_exit) * p_shares + partial_exit_slippage = abs(p_exit - p_exit_raw) * p_shares + remaining_shares -= p_shares + partial_exited = True + partial_exit_r_val = current_r + if current_stop > entry_price_raw: + current_stop = entry_price_raw + stop_level = "breakeven" + + # Pyramiding (short): add to winning position + if ( + params.pyramid_at_r is not None + and pyramid_count < params.pyramid_max_adds + and current_r >= params.pyramid_at_r * (1 + pyramid_count) + ): + add_shares = int(original_shares * params.pyramid_add_pct) + if add_shares > 0: + p_entry_raw = bar_close + p_entry_filled = _apply_slippage_exit(p_entry_raw, slippage) + pyramid_legs.append((add_shares, p_entry_filled, p_entry_raw)) + pyramid_total_shares += add_shares + pyramid_count += 1 + + if current_r >= effective_breakeven_at_r and current_stop > entry_price_raw: current_stop = entry_price_raw + stop_level = "breakeven" if current_r >= params.trailing_at_r: trailing_active = True + stop_level = "trailing" # ── Step 4: Update trailing stop for NEXT bar ── if trailing_active: if use_atr_trail: - candidate_stop = peak_price + atr * params.trailing_stop_atr_multiplier + atr_mult = params.trailing_stop_atr_multiplier + # Gap-adaptive trailing: override base multiplier based on gap size + if params.gap_trail_wide_threshold is not None: + if abs(gap_pct) > params.gap_trail_wide_threshold: + atr_mult = params.gap_trail_wide_atr_multiplier + elif params.gap_trail_tight_atr_multiplier is not None: + atr_mult = params.gap_trail_tight_atr_multiplier + if ( + params.trailing_tighten_at_r is not None + and current_r >= params.trailing_tighten_at_r + and params.trailing_stop_atr_multiplier_tight > 0 + ): + atr_mult = params.trailing_stop_atr_multiplier_tight + # Time-decay: linearly shrink trail width toward close + if use_time_decay and ts >= decay_start_ts and decay_span > 0: + elapsed = min((ts - decay_start_ts).total_seconds(), decay_span) + decay_pct = elapsed / decay_span + atr_mult *= 1.0 - decay_pct * (1.0 - params.time_decay_factor) + # SPY intraday guard (short): tighten trail when SPY rallies from open + if spy_guard_active and spy_bars: + spy_bar = next( + (sb for sb in spy_bars if sb.get("timestamp") == b.get("timestamp")), + None, + ) + if spy_bar is not None and spy_open > 0: + spy_change = (spy_bar["close"] - spy_open) / spy_open + if spy_change > abs(params.spy_intraday_guard_pct): + atr_mult *= params.spy_intraday_guard_tighten + candidate_stop = peak_price + atr * atr_mult else: swing_low_window.append(bar_close) candidate_stop = min(swing_low_window) @@ -698,22 +1562,42 @@ def simulate_orb_trade( else: final_r = (entry_price_raw - exit_price_raw) / stop_distance - # Apply slippage to exit + # Apply slippage to exit (on remaining original shares + pyramid shares) exit_price = ( _apply_slippage_exit(exit_price_raw, slippage) if direction == "long" else _apply_slippage_entry(exit_price_raw, slippage) ) if direction == "long": - pnl_pct = (exit_price - entry_price_filled) / entry_price_filled + remainder_pnl_pct = (exit_price - entry_price_filled) / entry_price_filled else: - pnl_pct = (entry_price_filled - exit_price) / entry_price_filled + remainder_pnl_pct = (entry_price_filled - exit_price) / entry_price_filled - pnl = pnl_pct * (shares * entry_price_filled) + # Blended PnL: partial exit + final exit on remaining original shares + remainder_pnl = remainder_pnl_pct * (remaining_shares * entry_price_filled) + pnl = remainder_pnl + partial_pnl - entry_slippage = abs(entry_price_filled - entry_price_raw) * shares - exit_slippage = abs(exit_price - exit_price_raw) * shares - slippage_cost = entry_slippage + exit_slippage + # Pyramid PnL: add-on legs exit at the same price as the main position + pyr_pnl = 0.0 + pyr_entry_slippage = 0.0 + if pyramid_legs: + for p_shares, p_entry_filled, p_entry_raw in pyramid_legs: + if direction == "long": + pyr_pnl += (exit_price - p_entry_filled) * p_shares + else: + pyr_pnl += (p_entry_filled - exit_price) * p_shares + pyr_entry_slippage += abs(p_entry_filled - p_entry_raw) * p_shares + pnl += pyr_pnl + + # pnl_pct as return on total deployed capital (original + pyramid) + total_deployed_cost = original_shares * entry_price_filled + sum( + s * e for s, e, _ in pyramid_legs + ) + pnl_pct = pnl / total_deployed_cost if total_deployed_cost > 0 else 0.0 + + entry_slippage = abs(entry_price_filled - entry_price_raw) * original_shares + exit_slippage = abs(exit_price - exit_price_raw) * (remaining_shares + pyramid_total_shares) + slippage_cost = entry_slippage + exit_slippage + partial_exit_slippage + pyr_entry_slippage return IntradayTrade( date=date_str, @@ -722,7 +1606,7 @@ def simulate_orb_trade( exit_price=round(exit_price, 4), entry_time=entry_bar["timestamp"], exit_time=exit_time_str, - shares=round(shares, 4), + shares=round(original_shares, 4), pnl=round(pnl, 4), pnl_pct=round(pnl_pct, 6), exit_reason=exit_reason, @@ -732,6 +1616,11 @@ def simulate_orb_trade( rvol=round(rvol, 3), atr_at_entry=round(atr, 4), r_multiple_at_exit=round(final_r, 3), + stop_level_at_exit=stop_level, + partial_exit_r=round(partial_exit_r_val, 3) if partial_exit_r_val is not None else None, + pyramid_adds=pyramid_count, + pyramid_pnl=round(pyr_pnl, 4), + total_capital_deployed=round(total_deployed_cost, 4), ) @@ -746,16 +1635,19 @@ def simulate_orb_day( equity: float, blacklisted_tickers: set[str] | None = None, spy_bars: list[dict] | None = None, + ticker_sectors: dict[str, str] | None = None, available_cash: float | None = None, sizing_capital: float | None = None, + vix_value: float | None = None, ) -> DayResult: """Simulate one full trading day using the ORB strategy. - 1. SPY regime check (daily gap from enrichment) — skip bad market days - 2. compute_orb_candidates — filter and rank candidates - 3. If fewer than min_candidates_to_trade → skip day - 4. For each candidate: simulate_orb_trade - 5. Apply daily loss limit and max-stops kill switch + 1. VIX regime check — skip high-VIX days (max_vix) + 2. SPY regime check (daily gap from enrichment) — skip bad market days + 3. compute_orb_candidates — filter and rank candidates + 4. If fewer than min_candidates_to_trade → skip day + 5. For each candidate: simulate_orb_trade (with VIX size scaling) + 6. Apply daily loss limit and max-stops kill switch Args: bars_by_ticker: {ticker: [bars]} for this day. @@ -765,29 +1657,55 @@ def simulate_orb_day( equity: Current portfolio equity (for risk-based sizing). blacklisted_tickers: Tickers in cooldown. spy_bars: SPY bars (unused — regime check now uses enrichment). + vix_value: Previous close VIX for this trading day (None if unavailable). Returns: DayResult compatible with compute_metrics(). """ result = DayResult(date=date_str) + # VIX regime check: skip the entire day if VIX is too high. + # vix_value is the prior close VIX (lookahead-free). + if params.max_vix is not None and vix_value is not None: + if vix_value > params.max_vix: + result.skip_reason = "vix_gate" + return result # skip high-VIX days + + # VIX position size scaler (applied to sizing_capital later) + vix_scaler = _orb_vix_size_scaler(vix_value, params) + # Market regime check: index ETF daily gap (lookahead-free via enrichment) - # Uses prev_close and today_open from enrichment, which are computed from - # prior daily bars only — no intraday data needed. - if params.market_regime_spy_threshold is not None: + regime_scaler = 1.0 + if params.market_regime_spy_threshold is not None or params.regime_size_scale_low is not None: regime_ticker = getattr(params, "market_regime_ticker", None) or "SPY" regime_enrich = 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 < params.market_regime_spy_threshold: - return result # skip bearish-open days + # Hard skip floor (V20 param or V19 legacy threshold) + if params.regime_skip_below is not None and regime_gap < params.regime_skip_below: + result.skip_reason = "market_regime" + return result + if (params.regime_size_scale_low is None + and params.market_regime_spy_threshold is not None + and regime_gap < params.market_regime_spy_threshold): + result.skip_reason = "market_regime" + return result + # Soft scaler (V20 path) + if params.regime_size_scale_low is not None and params.regime_size_scale_high is not None: + regime_scaler = _linear_scaler( + regime_gap, + params.regime_size_scale_low, + params.regime_size_scale_high, + params.regime_size_scale_min, + invert=True, + ) - # Candidate breadth filter: skip day if too few intraday tickers gapped up. - # More robust than single-ETF regime check — measures actual candidate pool sentiment. + # Candidate breadth filter + breadth_scaler = 1.0 min_breadth = getattr(params, "min_candidate_breadth", None) - if min_breadth is not None: + if min_breadth is not None or params.breadth_size_scale_low is not None: pos_gap_count = 0 total_with_data = 0 for ticker in bars_by_ticker: @@ -798,9 +1716,32 @@ def simulate_orb_day( total_with_data += 1 if today_o > prev_c: pos_gap_count += 1 - if total_with_data > 0 and (pos_gap_count / total_with_data) < min_breadth: - return result # skip low-breadth day + if total_with_data > 0: + breadth_ratio = pos_gap_count / total_with_data + if params.breadth_skip_below is not None and breadth_ratio < params.breadth_skip_below: + result.skip_reason = "breadth" + return result + if (params.breadth_size_scale_low is None + and min_breadth is not None + and breadth_ratio < min_breadth): + result.skip_reason = "breadth" + return result + if params.breadth_size_scale_low is not None and params.breadth_size_scale_high is not None: + breadth_scaler = _linear_scaler( + breadth_ratio, + params.breadth_size_scale_low, + params.breadth_size_scale_high, + params.breadth_size_scale_min, + invert=True, + ) + + combined_scaler = regime_scaler * breadth_scaler + is_soft_day = combined_scaler < params.soft_day_scaler_threshold + result.regime_scaler = regime_scaler + result.breadth_scaler = breadth_scaler + result.is_soft_day = is_soft_day + _cand_stats: dict = {} candidates = compute_orb_candidates( bars_by_ticker, date_str, @@ -808,10 +1749,14 @@ def simulate_orb_day( enrichment, blacklisted_tickers=blacklisted_tickers, spy_bars=None, # handled above via enrichment + ticker_sectors=ticker_sectors, + _stats_out=_cand_stats, ) result.candidates_found = len(candidates) + result.candidate_filter_stats = _cand_stats if _cand_stats else None if len(candidates) < params.min_candidates_to_trade: + result.skip_reason = "no_candidates" if len(candidates) == 0 else "below_min_candidates" return result # ── Pass 1: Find breakout times for all candidates ── @@ -835,11 +1780,21 @@ def simulate_orb_day( # ── Pass 2: Simulate in chronological order with capital constraints ── sizing_cap = sizing_capital if sizing_capital is not None else equity + # Apply VIX + regime/breadth scalers to sizing capital + combined_size_mult = vix_scaler * combined_scaler + adjusted_sizing = sizing_cap * combined_size_mult if combined_size_mult < 1.0 else sizing_cap daily_loss_limit = sizing_cap * params.daily_max_loss_pct remaining_cash = available_cash # None → no constraint (settlement_days=0) result.available_cash_start = available_cash if available_cash is not None else equity skipped_cash = 0 + entries_at_ts: dict[str, int] = {} # timestamp_str → entries taken at that bar + total_deployed = 0.0 # cumulative deployed capital for deployment cap + max_deploy = ( + sizing_cap * params.max_total_deployment_pct + if params.max_total_deployment_pct is not None + else None + ) for idx, (breakout_ts, cand, direction_str) in enumerate(timed_candidates): # Kill switch: only count losses from trades that have ALREADY EXITED @@ -864,11 +1819,42 @@ def simulate_orb_day( if realized_stops >= params.max_stops_per_day: break + # Simultaneous-entry cap: limit correlated risk when all candidates break out + # on the same bar (typically 09:35). Top-ranked candidates are taken first + # because timed_candidates is sorted by breakout_ts (ties preserve ranking order). + if params.max_simultaneous_entries is not None: + ts_key = breakout_ts.isoformat() + if entries_at_ts.get(ts_key, 0) >= params.max_simultaneous_entries: + continue + # Cash exhaustion if remaining_cash is not None and remaining_cash <= 0: skipped_cash += len(timed_candidates) - idx break + # Portfolio deployment cap + if max_deploy is not None and total_deployed >= max_deploy: + continue + + # Score rank percentage: 1.0 = top ranked, 0.0 = bottom ranked + n_cands = len(candidates) + cand_rank = next( + (i for i, c in enumerate(candidates) if c["ticker"] == cand["ticker"]), + n_cands - 1, + ) + score_rank_pct = 1.0 - (cand_rank / max(n_cands - 1, 1)) + + # Soft-day selection gates + if is_soft_day: + if params.soft_day_max_trades is not None and len(result.trades) >= params.soft_day_max_trades: + continue + if params.soft_day_min_score_pct is not None and score_rank_pct < params.soft_day_min_score_pct: + continue + + # Previous close for gap fill protection + ticker_enrich = enrichment.get(cand["ticker"], {}).get(date_str, {}) + cand_prev_close = ticker_enrich.get("prev_close") + trade = simulate_orb_trade( mkt_bars=cand["mkt_bars"], orb_bar=cand["orb_bar"], @@ -881,7 +1867,11 @@ def simulate_orb_day( date_str=date_str, ticker=cand["ticker"], available_cash=remaining_cash, - sizing_capital=sizing_capital, + sizing_capital=adjusted_sizing if combined_size_mult < 1.0 else sizing_capital, + score_rank_pct=score_rank_pct, + prev_close=cand_prev_close, + spy_bars=spy_bars, + is_soft_day=is_soft_day, ) if trade is None: @@ -890,12 +1880,92 @@ def simulate_orb_day( result.trades.append(trade) result.daily_pnl += trade.pnl + # Track simultaneous entries count + ts_key = breakout_ts.isoformat() + entries_at_ts[ts_key] = entries_at_ts.get(ts_key, 0) + 1 + + # Track total deployed capital (original + pyramid) + trade_deployed = trade.total_capital_deployed or (trade.shares * trade.entry_price) + total_deployed += trade_deployed + # Deduct deployed capital from remaining settled cash if remaining_cash is not None: - remaining_cash -= trade.shares * trade.entry_price + remaining_cash -= trade_deployed + + # ── Pass 3 (optional): Re-entry after stop-out ── + if params.reentry_after_stop: + stopped_trades = [ + t for t in result.trades + if t.exit_reason == "stop_loss" and not t.is_reentry + ] + for stopped_trade in stopped_trades: + # Check re-entry count for this ticker + reentries_done = sum( + 1 for t in result.trades if t.ticker == stopped_trade.ticker and t.is_reentry + ) + if reentries_done >= params.reentry_max_per_ticker: + continue + + # Deployment cap check + if max_deploy is not None and total_deployed >= max_deploy: + break + + # Find original candidate data + cand_match = next( + ((c, d) for _, c, d in timed_candidates if c["ticker"] == stopped_trade.ticker), + None, + ) + if cand_match is None: + continue + cand, direction_str = cand_match + + # Score rank for re-entry (same as original) + n_cands = len(candidates) + cand_rank = next( + (i for i, c in enumerate(candidates) if c["ticker"] == cand["ticker"]), + n_cands - 1, + ) + score_rank_pct = 1.0 - (cand_rank / max(n_cands - 1, 1)) + + ticker_enrich = enrichment.get(cand["ticker"], {}).get(date_str, {}) + cand_prev_close = ticker_enrich.get("prev_close") + + exit_ts = _parse_ts(stopped_trade.exit_time) + reentry_trade = simulate_orb_trade( + mkt_bars=cand["mkt_bars"], + orb_bar=cand["orb_bar"], + direction=direction_str, + atr=cand["atr"], + rvol=cand["rvol"], + gap_pct=cand["gap_pct"], + params=params, + equity=equity, + date_str=date_str, + ticker=cand["ticker"], + available_cash=remaining_cash, + sizing_capital=adjusted_sizing if combined_size_mult < 1.0 else sizing_capital, + score_rank_pct=score_rank_pct, + prev_close=cand_prev_close, + entry_after_ts=exit_ts, + spy_bars=spy_bars, + is_soft_day=is_soft_day, + ) + + if reentry_trade is not None: + reentry_trade.is_reentry = True + result.trades.append(reentry_trade) + result.daily_pnl += reentry_trade.pnl + re_deployed = reentry_trade.total_capital_deployed or ( + reentry_trade.shares * reentry_trade.entry_price + ) + total_deployed += re_deployed + if remaining_cash is not None: + remaining_cash -= re_deployed result.skipped_insufficient_cash = skipped_cash - result.capital_deployed = sum(t.shares * t.entry_price for t in result.trades) + result.capital_deployed = sum( + t.total_capital_deployed or (t.shares * t.entry_price) for t in result.trades + ) # Note: daily_return_pct is set by run_orb_simulation (portfolio-level: PnL/equity). # Default 0.0 is correct for no-trade days. @@ -905,12 +1975,16 @@ def simulate_orb_day( # ── Full Backtest Simulation ─────────────────────────────────────────────── -def run_orb_simulation( +def run_orb_simulation_with_state( all_intraday: dict[str, dict[str, list[dict]]], trading_days: list[str], params: ORBStrategyParams, enrichment: dict[str, dict[str, dict]], -) -> list[DayResult]: + ticker_sectors: dict[str, str] | None = None, + state: ORBSimulationState | None = None, + progress_callback: Callable[[int, int], None] | None = None, + vix_by_day: dict[str, float] | None = None, +) -> tuple[list[DayResult], ORBSimulationState]: """Run the full ORB backtest simulation across all trading days. Key differences from run_simulation() (momentum): @@ -929,27 +2003,49 @@ def run_orb_simulation( enrichment: {ticker: {date: features}} from enrich_daily_bars(). Returns: - List of DayResult objects (one per day that had intraday data). - Compatible with compute_metrics() and all report formatters. + (day_results, next_state) where next_state can be fed into the next chunk. """ results: list[DayResult] = [] - equity = params.initial_capital + equity = state.equity if state is not None else params.initial_capital # Ticker cooldown tracker - ticker_last_traded: dict[str, dt.date] = {} + ticker_last_traded: dict[str, dt.date] = ( + {ticker: dt.date.fromisoformat(last_date) for ticker, last_date in state.ticker_last_traded.items()} + if state is not None else {} + ) # GFV / settlement tracking (only active when settlement_days > 0) # settled_cash: funds available for new day-trade positions (GFV-safe) # pending_settlements: (settlement_date_str, amount) — proceeds awaiting settlement settlement_enabled = params.settlement_days > 0 - settled_cash = params.initial_capital - pending_settlements: list[tuple[str, float]] = [] + settled_cash = ( + state.settled_cash + if state is not None and state.settled_cash is not None + else params.initial_capital + ) + pending_settlements: list[tuple[str, float]] = ( + list(state.pending_settlements) if state is not None else [] + ) + # Rolling PnL window: persists cross-chunk daily PnL history for rolling loss filter. + # Trimmed to rolling_loss_days length so memory stays bounded. + rolling_pnl_window: list[float] = list(state.recent_daily_pnl) if state is not None else [] + + # Drawdown governor: track peak equity to detect drawdowns + peak_equity = equity + + # Streak sizing: track recent trade outcomes for streak-based sizing + streak_outcomes: list[bool] = [] # True = win, False = loss (from recent trades) + + total_days = len(trading_days) for day_idx, date_str in enumerate(trading_days): + if progress_callback: + progress_callback(day_idx + 1, total_days) bars_by_ticker = all_intraday.get(date_str) if not bars_by_ticker: # No intraday data for this day — still record it (0% return, no trades) results.append(DayResult(date=date_str)) + rolling_pnl_window.append(0.0) continue # Step 1: Move proceeds that have reached their settlement date into settled_cash @@ -977,10 +2073,138 @@ def run_orb_simulation( else None ) + # Rolling strategy loss filter: pause trading after sustained self-drawdown. + # Uses rolling_pnl_window which persists across chunk boundaries (unlike results[]). + if ( + params.rolling_loss_days is not None + and params.rolling_loss_threshold is not None + and len(rolling_pnl_window) >= params.rolling_loss_days + ): + n_roll = params.rolling_loss_days + rolling_pnl = sum(rolling_pnl_window[-n_roll:]) + if params.daily_budget_reset or not params.compound_returns: + sizing_capital_for_check = params.initial_capital + else: + sizing_capital_for_check = equity + if sizing_capital_for_check > 0: + rolling_return = rolling_pnl / sizing_capital_for_check + if rolling_return < params.rolling_loss_threshold: + results.append(DayResult(date=date_str, skip_reason="rolling_loss")) + rolling_pnl_window.append(0.0) + if settlement_enabled: + still_pending = [] + for settle_date, amount in pending_settlements: + if settle_date <= date_str: + settled_cash += amount + else: + still_pending.append((settle_date, amount)) + pending_settlements = still_pending + continue + + # Multi-day SPY trend filter: skip if SPY is in a sustained downtrend + # Uses enrichment[spy_ticker][date]["prev_close"] for N-day cumulative return. + # enrichment[D]["prev_close"] = close of trading day before D. + # N-day return = (close_yesterday - close_N_days_ago) / close_N_days_ago + # = (enrich[today]["prev_close"] - enrich[trading_days[day_idx-N+1]]["prev_close"]) + # / enrich[trading_days[day_idx-N+1]]["prev_close"] + if ( + params.market_regime_spy_trend_days is not None + and params.market_regime_spy_trend_threshold is not None + and day_idx >= params.market_regime_spy_trend_days + ): + spy_trend_ticker = getattr(params, "market_regime_ticker", None) or "SPY" + spy_enrich = enrichment.get(spy_trend_ticker, {}) + close_yesterday = spy_enrich.get(date_str, {}).get("prev_close") + n_days_back = params.market_regime_spy_trend_days + look_back_date = trading_days[day_idx - n_days_back + 1] + close_n_ago = spy_enrich.get(look_back_date, {}).get("prev_close") + if close_yesterday and close_n_ago and close_n_ago > 0: + spy_trend_return = (close_yesterday - close_n_ago) / close_n_ago + if spy_trend_return < params.market_regime_spy_trend_threshold: + results.append(DayResult(date=date_str, skip_reason="spy_trend")) + rolling_pnl_window.append(0.0) + if settlement_enabled: + still_pending = [] + for settle_date, amount in pending_settlements: + if settle_date <= date_str: + settled_cash += amount + else: + still_pending.append((settle_date, amount)) + pending_settlements = still_pending + continue + available_cash = settled_cash if settlement_enabled else None - # Simple vs compound sizing: fixed initial_capital vs growing equity - sizing_capital = None if params.compound_returns else params.initial_capital + # Sizing mode resolution: + # - daily_budget_reset: research mode, always initial_capital (ignores path) + # - compound_returns: sizing_capital=None → simulate_orb_trade uses equity + # - simple: fixed initial_capital + if params.daily_budget_reset: + sizing_capital = params.initial_capital + elif params.compound_returns: + sizing_capital = None + else: + sizing_capital = params.initial_capital + + # Drawdown governor: scale down sizing when equity drops below peak + if params.drawdown_governor_threshold is not None and peak_equity > 0: + dd_pct = (peak_equity - equity) / peak_equity # 0.0 = at peak, 0.05 = 5% DD + if dd_pct > params.drawdown_governor_threshold: + # Linear ramp from 1.0 at threshold to min_scale at 2× threshold + dd_range = params.drawdown_governor_threshold # same width for the ramp + dd_excess = dd_pct - params.drawdown_governor_threshold + governor_scale = max( + params.drawdown_governor_min_scale, + 1.0 - (1.0 - params.drawdown_governor_min_scale) * min(dd_excess / dd_range, 1.0), + ) + if sizing_capital is not None: + sizing_capital = sizing_capital * governor_scale + else: + # compound mode: scale equity for sizing + sizing_capital = equity * governor_scale + + # Streak sizing: apply win/loss streak multiplier after governor + if (params.streak_sizing_win_bonus is not None or params.streak_sizing_loss_penalty is not None) and streak_outcomes: + # Count consecutive wins or losses from the END of the list + streak_len = 0 + is_winning = streak_outcomes[-1] + for outcome in reversed(streak_outcomes): + if outcome == is_winning: + streak_len += 1 + else: + break + + streak_mult = 1.0 + if is_winning and params.streak_sizing_win_bonus is not None: + streak_mult = 1.0 + streak_len * params.streak_sizing_win_bonus + elif not is_winning and params.streak_sizing_loss_penalty is not None: + streak_mult = 1.0 - streak_len * params.streak_sizing_loss_penalty + + streak_mult = max(params.streak_sizing_min, min(params.streak_sizing_max, streak_mult)) + + if sizing_capital is not None: + sizing_capital = sizing_capital * streak_mult + else: + sizing_capital = equity * streak_mult + + # Rolling WR sizing: apply bonus/penalty based on recent win rate + if params.rolling_wr_sizing_window is not None and len(streak_outcomes) >= params.rolling_wr_sizing_window: + recent = streak_outcomes[-params.rolling_wr_sizing_window:] + rolling_wr = sum(recent) / len(recent) + + wr_mult = 1.0 + if rolling_wr > params.rolling_wr_sizing_threshold: + wr_mult = 1.0 + params.rolling_wr_sizing_bonus + elif params.rolling_wr_sizing_penalty_threshold is not None and rolling_wr < params.rolling_wr_sizing_penalty_threshold: + wr_mult = 1.0 - params.rolling_wr_sizing_penalty + + if wr_mult != 1.0: + if sizing_capital is not None: + sizing_capital = sizing_capital * wr_mult + else: + sizing_capital = equity * wr_mult + + day_vix = vix_by_day.get(date_str) if vix_by_day else None day_result = simulate_orb_day( bars_by_ticker, @@ -990,8 +2214,10 @@ def run_orb_simulation( equity=equity, blacklisted_tickers=blacklisted if blacklisted else None, spy_bars=spy_bars, + ticker_sectors=ticker_sectors, available_cash=available_cash, sizing_capital=sizing_capital, + vix_value=day_vix, ) # Override daily_return_pct with portfolio-level return (PnL / equity at start of day). # simulate_orb_day uses deployed capital as denominator — that inflates returns. @@ -1000,10 +2226,19 @@ def run_orb_simulation( day_result.daily_return_pct = day_result.daily_pnl / equity results.append(day_result) + rolling_pnl_window.append(day_result.daily_pnl) + + # Update streak outcomes from today's trades + for trade in day_result.trades: + streak_outcomes.append(trade.pnl > 0) + # Keep only last 20 outcomes to bound memory + if len(streak_outcomes) > 20: + streak_outcomes = streak_outcomes[-20:] # Update equity equity += day_result.daily_pnl equity = max(equity, 1.0) # prevent zero/negative equity from crashing + peak_equity = max(peak_equity, equity) # Step 2: After the day, deduct deployed capital and schedule proceeds for settlement if settlement_enabled: @@ -1025,4 +2260,34 @@ def run_orb_simulation( for trade in day_result.trades: ticker_last_traded[trade.ticker] = current_date + max_roll = params.rolling_loss_days or 0 + next_state = ORBSimulationState( + equity=equity, + ticker_last_traded={ + ticker: last_dt.isoformat() for ticker, last_dt in ticker_last_traded.items() + }, + settled_cash=settled_cash if settlement_enabled else None, + pending_settlements=list(pending_settlements), + recent_daily_pnl=rolling_pnl_window[-max_roll:] if max_roll > 0 else [], + ) + return results, next_state + + +def run_orb_simulation( + all_intraday: dict[str, dict[str, list[dict]]], + trading_days: list[str], + params: ORBStrategyParams, + enrichment: dict[str, dict[str, dict]], + ticker_sectors: dict[str, str] | None = None, + vix_by_day: dict[str, float] | None = None, +) -> list[DayResult]: + """Run the full ORB backtest simulation across all trading days.""" + results, _ = run_orb_simulation_with_state( + all_intraday, + trading_days, + params, + enrichment, + ticker_sectors=ticker_sectors, + vix_by_day=vix_by_day, + ) return results diff --git a/libs/intraday/screener.py b/libs/intraday/screener.py index f68b0f2..c3b53b8 100644 --- a/libs/intraday/screener.py +++ b/libs/intraday/screener.py @@ -8,13 +8,19 @@ This approach reduces API calls by ~97% vs brute-force (fetching intraday for al from __future__ import annotations import asyncio +import hashlib +import json +import math import sys from collections import defaultdict +from datetime import date, datetime, time, timezone +from pathlib import Path from typing import Callable import yaml -from libs.intraday.cache import IntradayCache +from libs.common.config import get_settings +from libs.intraday.cache import DailyBarCache, IntradayCache from libs.intraday.domain import UniverseParams from libs.oracle_client import OracleClient from libs.oracle_client.price import PriceService @@ -27,9 +33,90 @@ _UNIVERSE_YAML_MAP = { "midlarge": "configs/symbols_midlarge_snapshot_exact.yaml", "largecap": "configs/symbols.yaml", "midcap": "configs/symbols_midcap.yaml", + "smallmid": "configs/symbols_smallmid.yaml", } +class _ScreenerUniverseSnapshotStore: + """Persist successful live screener universes for recent sanity fallback. + + This is intentionally only a best-effort cache. It is used when the live + Oracle/Yahoo screener is temporarily unavailable, so recent-day sanity + backtests can still resolve a broad tradable universe instead of failing. + """ + + def __init__(self) -> None: + settings = get_settings() + self._root = Path(settings.data_root) / "cache" / "screener_universe" + + @staticmethod + def _query_payload( + *, + market_cap_min: float | None, + min_avg_volume: int | None, + price_min: float | None, + ) -> dict[str, float | int | None]: + return { + "market_cap_min": market_cap_min, + "min_avg_volume": min_avg_volume, + "price_min": price_min, + } + + def _path(self, payload: dict[str, float | int | None]) -> Path: + key = hashlib.sha1( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + return self._root / f"{key}.json" + + def load( + self, + *, + market_cap_min: float | None, + min_avg_volume: int | None, + price_min: float | None, + ) -> list[str] | None: + payload = self._query_payload( + market_cap_min=market_cap_min, + min_avg_volume=min_avg_volume, + price_min=price_min, + ) + path = self._path(payload) + if not path.exists(): + return None + try: + raw = json.loads(path.read_text()) + except Exception: + return None + symbols = raw.get("symbols") + if not isinstance(symbols, list): + return None + return sorted({str(symbol).upper() for symbol in symbols if symbol}) + + def save( + self, + *, + market_cap_min: float | None, + min_avg_volume: int | None, + price_min: float | None, + symbols: list[str], + ) -> None: + payload = self._query_payload( + market_cap_min=market_cap_min, + min_avg_volume=min_avg_volume, + price_min=price_min, + ) + path = self._path(payload) + path.parent.mkdir(parents=True, exist_ok=True) + body = { + "created_at_utc": datetime.now(timezone.utc).isoformat(), + "query": payload, + "symbols": sorted({str(symbol).upper() for symbol in symbols if symbol}), + } + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps(body, indent=2, sort_keys=True)) + tmp.replace(path) + + async def resolve_universe(params: UniverseParams, client: OracleClient) -> list[str]: """Resolve the list of ticker symbols based on universe config. @@ -39,6 +126,7 @@ async def resolve_universe(params: UniverseParams, client: OracleClient) -> list - 'midlarge': 971-ticker YAML fallback file - 'largecap': 741-ticker YAML fallback file - 'midcap': 802-ticker YAML file + - 'smallmid': small+midcap YAML file - 'yaml': custom YAML file (requires symbols_file param) - 'screener': live screener query @@ -66,15 +154,37 @@ async def resolve_universe(params: UniverseParams, client: OracleClient) -> list # Live screener query if source == "screener": screener = ScreenerService(client) - stocks = await screener.search_all_stocks( - market_cap_min=params.market_cap_min, - min_avg_volume=params.avg_volume_min, - price_min=params.min_price if params.min_price > 0 else None, - ) - return sorted({s.symbol.upper() for s in stocks if s.symbol}) + snapshot_store = _ScreenerUniverseSnapshotStore() + market_cap_min = params.market_cap_min + min_avg_volume = params.avg_volume_min + price_min = params.min_price if params.min_price > 0 else None + try: + stocks = await screener.search_all_stocks( + market_cap_min=market_cap_min, + min_avg_volume=min_avg_volume, + price_min=price_min, + ) + symbols = sorted({s.symbol.upper() for s in stocks if s.symbol}) + if symbols: + snapshot_store.save( + market_cap_min=market_cap_min, + min_avg_volume=min_avg_volume, + price_min=price_min, + symbols=symbols, + ) + return symbols + except Exception: + snapshot = snapshot_store.load( + market_cap_min=market_cap_min, + min_avg_volume=min_avg_volume, + price_min=price_min, + ) + if snapshot: + return snapshot + raise raise ValueError(f"Unknown universe source: {source!r}. " - "Use: sp500, nasdaq100, midlarge, largecap, midcap, yaml, screener") + "Use: sp500, nasdaq100, midlarge, largecap, midcap, smallmid, yaml, screener") def _load_yaml_symbols(path: str) -> list[str]: @@ -105,10 +215,14 @@ async def fetch_daily_bars_bulk( start_date: str, end_date: str, client: OracleClient, - concurrency: int = 20, + cache: DailyBarCache | None = None, + intraday_cache_fallback: IntradayCache | None = None, + prefer_intraday_fallback: bool = False, + skip_oracle_when_unhealthy: bool = False, + concurrency: int = 3, progress_callback: Callable[[int, int], None] | None = None, ) -> dict[str, list[dict]]: - """Fetch daily OHLCV bars for all tickers in parallel. + """Fetch daily OHLCV bars using bulk chunks, with per-ticker fallback. Returns {ticker: [bar_dict, ...]}. Tickers with no data or errors are silently omitted. @@ -118,6 +232,206 @@ async def fetch_daily_bars_bulk( results: dict[str, list[dict]] = {} lock = asyncio.Lock() completed = 0 + # Keep daily bulk payloads moderate to avoid destabilizing Oracle during + # multi-year research runs over large universes. + chunk_size = 25 + + def _ordered_results() -> dict[str, list[dict]]: + return {ticker: results[ticker] for ticker in tickers if ticker in results} + + def _emit_progress() -> None: + if progress_callback: + progress_callback(completed, len(tickers)) + + def _normalize_bulk_bars(raw_bars: list[dict]) -> list[dict]: + bars: list[dict] = [] + for b in raw_bars: + bars.append( + { + "date": str(b.get("date", "")), + "open": float(b.get("open", 0) or 0), + "high": float(b.get("high", 0) or 0), + "low": float(b.get("low", 0) or 0), + "close": float(b.get("close", 0) or 0), + "volume": float(b.get("volume", 0) or 0), + } + ) + return bars + + def _bulk_daily_bars_look_truncated(bars: list[dict]) -> bool: + if not bars: + return False + try: + span_days = (date.fromisoformat(end_date) - date.fromisoformat(start_date)).days + except Exception: + return False + return span_days >= 45 and len(bars) < 20 + + async def _rebuild_daily_from_intraday_cache(ticker: str) -> list[dict] | None: + if intraday_cache_fallback is None: + return None + dates = await asyncio.to_thread( + intraday_cache_fallback.available_dates, + ticker, + start_date, + end_date, + ) + if not dates: + return None + + from datetime import datetime as _dt, time as _time + from zoneinfo import ZoneInfo + _et = ZoneInfo("America/New_York") + _mkt_open = _time(9, 30) + _mkt_close = _time(16, 0) + + def _to_et(ts_str: str) -> _dt: + if ts_str.endswith("Z"): + return _dt.fromisoformat(ts_str[:-1] + "+00:00").astimezone(_et) + return _dt.fromisoformat(ts_str).astimezone(_et) + + rows: list[dict] = [] + for day in dates: + bars = await asyncio.to_thread(intraday_cache_fallback.get, ticker, day) + if not bars: + continue + # Use only regular market hours (9:30–16:00 ET) for OHLCV reconstruction. + # After-hours moves distort prev_close, corrupting next-day gap calculations. + mkt_bars = [] + for b in bars: + try: + ts = _to_et(b.get("timestamp", "")) + if _mkt_open <= ts.time() < _mkt_close: + mkt_bars.append(b) + except Exception: + continue + if not mkt_bars: + continue + rows.append( + { + "date": day, + "open": float(mkt_bars[0].get("open", 0.0) or 0.0), + "high": max(float(b.get("high", 0.0) or 0.0) for b in mkt_bars), + "low": min(float(b.get("low", 0.0) or 0.0) for b in mkt_bars), + "close": float(mkt_bars[-1].get("close", 0.0) or 0.0), + "volume": sum(float(b.get("volume", 0.0) or 0.0) for b in mkt_bars), + } + ) + return rows or None + + # Phase 1a: read cache — full hits, partial hits (end_date advanced), and true misses. + # Partial hits occur when the clock crosses a day boundary between runs (e.g. 16:00 ET), + # shifting end_date by one trading day. We reuse cached bars up to coverage_end and + # only fetch the small tail from Oracle instead of refetching all 500+ tickers. + full_misses: list[str] = [] + # ticker -> (cached_bars, tail_start_date) + partial_hits: dict[str, tuple[list[dict], str]] = {} + + if cache: + read_semaphore = asyncio.Semaphore(32) + + async def read_one(ticker: str) -> None: + async with read_semaphore: + bars, tail_start = await asyncio.to_thread( + cache.get_with_tail, ticker, start_date, end_date + ) + async with lock: + if bars is None: + full_misses.append(ticker) + elif tail_start is None: + results[ticker] = bars # full cache hit + else: + partial_hits[ticker] = (bars, tail_start) + + await asyncio.gather(*[asyncio.create_task(read_one(ticker)) for ticker in tickers]) + completed = len(results) + _emit_progress() + else: + full_misses = list(tickers) + + # Phase 1b: tail-only fetches for partial hits. + # Group by tail_start so tickers cached from the same run share one bulk API call. + if partial_hits: + tail_groups: dict[str, list[str]] = {} + for ticker, (_, ts) in partial_hits.items(): + tail_groups.setdefault(ts, []).append(ticker) + + tail_semaphore = asyncio.Semaphore(concurrency) + + async def fetch_tail_chunk(chunk: list[str], tail_from: str) -> None: + nonlocal completed + try: + async with tail_semaphore: + raw = await client.get( + "/api/v1/price/data", + params={ + "tickers": ",".join(chunk), + "start_date": tail_from, + "end_date": end_date, + }, + ) + bars_by_ticker = raw.get("bars", {}) + for ticker in chunk: + tail_bars = _normalize_bulk_bars(bars_by_ticker.get(ticker, [])) + cached_bars, _ = partial_hits[ticker] + merged = cached_bars + tail_bars + async with lock: + results[ticker] = merged + if cache and tail_bars: + await asyncio.to_thread( + cache.put, ticker, start_date, end_date, merged + ) + async with lock: + completed += len(chunk) + _emit_progress() + except Exception: + # Fall back: use the cached portion without the tail so the run + # can still complete. The missing day just has no candidates. + for ticker in chunk: + cached_bars, _ = partial_hits[ticker] + async with lock: + results[ticker] = cached_bars + completed += 1 + _emit_progress() + + tail_tasks = [ + asyncio.create_task(fetch_tail_chunk(tail_tickers[i : i + chunk_size], tail_from)) + for tail_from, tail_tickers in tail_groups.items() + for i in range(0, len(tail_tickers), chunk_size) + ] + await asyncio.gather(*tail_tasks) + + misses = full_misses + + if prefer_intraday_fallback and misses and intraday_cache_fallback is not None: + rebuild_semaphore = asyncio.Semaphore(max(1, min(concurrency * 2, 16))) + remaining: list[str] = [] + + async def rebuild_one(ticker: str) -> None: + nonlocal completed + async with rebuild_semaphore: + rebuilt = await _rebuild_daily_from_intraday_cache(ticker) + if rebuilt: + async with lock: + results[ticker] = rebuilt + completed += 1 + if cache: + await asyncio.to_thread(cache.put, ticker, start_date, end_date, rebuilt) + _emit_progress() + return + async with lock: + remaining.append(ticker) + + await asyncio.gather(*[asyncio.create_task(rebuild_one(ticker)) for ticker in misses]) + misses = remaining + + oracle_available: bool | None = None + if skip_oracle_when_unhealthy and misses: + oracle_available = await client.health_check_fast() + if not oracle_available: + completed += len(misses) + _emit_progress() + return _ordered_results() async def fetch_one(ticker: str) -> None: nonlocal completed @@ -138,17 +452,75 @@ async def fetch_daily_bars_bulk( ] async with lock: results[ticker] = bars + if cache: + await asyncio.to_thread(cache.put, ticker, start_date, end_date, bars) except Exception: - pass + rebuilt = await _rebuild_daily_from_intraday_cache(ticker) + if rebuilt: + async with lock: + results[ticker] = rebuilt + if cache: + await asyncio.to_thread(cache.put, ticker, start_date, end_date, rebuilt) finally: async with lock: completed += 1 - if progress_callback: - progress_callback(completed, len(tickers)) + _emit_progress() + + async def fetch_chunk(chunk: list[str]) -> None: + nonlocal completed + try: + async with semaphore: + raw = await client.get( + "/api/v1/price/data", + params={ + "tickers": ",".join(chunk), + "start_date": start_date, + "end_date": end_date, + }, + ) - tasks = [asyncio.create_task(fetch_one(t)) for t in tickers] + bars_by_ticker = raw.get("bars", {}) + suspicious: list[str] = [] + for ticker in chunk: + raw_bars = bars_by_ticker.get(ticker, []) + if raw_bars: + bars = _normalize_bulk_bars(raw_bars) + if _bulk_daily_bars_look_truncated(bars): + suspicious.append(ticker) + continue + async with lock: + results[ticker] = bars + if cache: + await asyncio.to_thread(cache.put, ticker, start_date, end_date, bars) + for ticker in suspicious: + await fetch_one(ticker) + async with lock: + completed += len(chunk) - len(suspicious) + _emit_progress() + except Exception: + # Fall back to single-ticker requests so one bad chunk does not + # drop the whole backtest run. If Oracle remains unavailable but + # 5-minute bars are already cached locally, rebuild the daily tape + # from those cached intraday files instead of hanging on repeated + # network retries. + for ticker in chunk: + rebuilt = await _rebuild_daily_from_intraday_cache(ticker) + if rebuilt: + async with lock: + results[ticker] = rebuilt + completed += 1 + if cache: + await asyncio.to_thread(cache.put, ticker, start_date, end_date, rebuilt) + _emit_progress() + continue + await fetch_one(ticker) + + tasks = [ + asyncio.create_task(fetch_chunk(misses[i : i + chunk_size])) + for i in range(0, len(misses), chunk_size) + ] await asyncio.gather(*tasks) - return results + return _ordered_results() # ── Phase 2: Pre-Screening ───────────────────────────────────────────────── @@ -159,45 +531,53 @@ def pre_screen_candidates( trading_days: list[str], threshold: float = 0.015, max_per_day: int = 30, + enrichment: dict[str, dict[str, dict]] | None = None, ) -> dict[str, list[str]]: - """Identify candidate ticker-days using daily bar heuristics. + """Identify candidate ticker-days using lookahead-free opening-gap heuristics. + + A ticker-day is a candidate if the stock opens at least ``threshold`` above + the prior close: - A ticker-day is a candidate if (high - open) / open >= threshold. - This indicates the stock rose significantly above its opening price at some - point during the day — a necessary condition for being a morning gainer. + gap_pct = (today_open - prev_close) / prev_close + + This uses only information known at the open of the session plus prior-day + data. It intentionally avoids using the same day's high/close because those + values are not available when selecting the morning candidate universe. Returns: - {date: [ticker1, ticker2, ...]} — ranked by (high-open)/open descending, + {date: [ticker1, ticker2, ...]} — ranked by opening gap descending, capped at max_per_day per date. - - Note: This is a conservative filter. If threshold is too high, some actual - morning gainers may be missed. 1.5% default catches most meaningful movers. """ - # Build per-ticker lookup: {ticker -> {date -> bar}} - ticker_date_bar: dict[str, dict[str, dict]] = {} - for ticker, bars in daily_bars.items(): - date_map: dict[str, dict] = {} - for b in bars: - d = b["date"][:10] - date_map[d] = b - ticker_date_bar[ticker] = date_map + trading_day_set = set(trading_days) candidates: dict[str, list[str]] = defaultdict(list) day_scores: dict[str, dict[str, float]] = defaultdict(dict) - for ticker, date_map in ticker_date_bar.items(): - for day in trading_days: - b = date_map.get(day) - if not b: - continue - open_p = b.get("open", 0) - high_p = b.get("high", 0) - if open_p <= 0: - continue - score = (high_p - open_p) / open_p - if score >= threshold: + for ticker, bars in daily_bars.items(): + sorted_bars = sorted(bars, key=lambda bar: bar["date"]) + prev_close: float | None = None + for bar in sorted_bars: + day = bar["date"][:10] + open_p = bar.get("open") + gap_pct: float | None = None + + ticker_enrich = enrichment.get(ticker, {}).get(day, {}) if enrichment else {} + if ticker_enrich: + gap_pct = ticker_enrich.get("gap_pct") + elif ( + day in trading_day_set + and prev_close is not None and prev_close > 0 + and open_p is not None and open_p > 0 + ): + gap_pct = (open_p - prev_close) / prev_close + + if day in trading_day_set and gap_pct is not None and gap_pct >= threshold: candidates[day].append(ticker) - day_scores[day][ticker] = score + day_scores[day][ticker] = gap_pct + + close_p = bar.get("close") + if close_p is not None and close_p > 0: + prev_close = close_p # Rank and cap per day result: dict[str, list[str]] = {} @@ -208,6 +588,327 @@ def pre_screen_candidates( return result +def momentum_pre_screen_candidates( + daily_bars: dict[str, list[dict]], + trading_days: list[str], + enrichment: dict[str, dict[str, dict]], + threshold: float = 0.0, + max_per_day: int | None = 30, + *, + strategy=None, +) -> dict[str, list[str]]: + """Identify momentum candidates with only open-time and prior-day information. + + Inclusion: + - opening gap >= threshold + + Ranking (all lookahead-free): + 1. larger opening gap + 2. stronger prior 5-day return + 3. lower entropy_20d + 4. higher avg_dollar_vol_30d + 5. higher ATR/open + """ + from collections import defaultdict + + trading_day_set = set(trading_days) + candidates: dict[str, list[str]] = defaultdict(list) + day_scores: dict[str, dict[str, tuple[float, float, float, float, float, float]]] = defaultdict(dict) + + require_event_flag = bool(getattr(strategy, "candidate_require_event_flag", False)) if strategy else False + min_event_score = getattr(strategy, "candidate_min_event_score", None) if strategy else None + min_wiki_spike = getattr(strategy, "candidate_min_attention_wiki_spike_10d", None) if strategy else None + min_article_count = ( + getattr(strategy, "candidate_min_attention_article_count_3d", None) + if strategy else None + ) + min_us_article_count = ( + getattr(strategy, "candidate_min_attention_us_article_count_3d", None) + if strategy else None + ) + min_resolver_conf = ( + getattr(strategy, "candidate_min_attention_resolver_confidence", None) + if strategy else None + ) + weight_event = float(getattr(strategy, "candidate_weight_event_score", 0.0) or 0.0) if strategy else 0.0 + weight_wiki = float(getattr(strategy, "candidate_weight_attention_wiki", 0.0) or 0.0) if strategy else 0.0 + weight_news = float(getattr(strategy, "candidate_weight_attention_news", 0.0) or 0.0) if strategy else 0.0 + + for ticker, bars in daily_bars.items(): + date_map = {bar["date"][:10]: bar for bar in bars} + for day in trading_days: + bar = date_map.get(day) + if not bar: + continue + open_p = bar.get("open") + if open_p is None or open_p <= 0: + continue + + info = enrichment.get(ticker, {}).get(day, {}) + gap_pct = info.get("gap_pct") + if gap_pct is None or gap_pct < threshold: + continue + + event_flag = bool(info.get("event_flag")) + event_score = float(info.get("event_score") or 0.0) + wiki_spike = float(info.get("attention_wiki_spike_10d") or 0.0) + article_count = int(info.get("attention_article_count_3d") or 0) + us_article_count = int(info.get("attention_us_article_count_3d") or 0) + resolver_confidence = float(info.get("attention_resolver_confidence") or 0.0) + + if require_event_flag and not event_flag: + continue + if min_event_score is not None and event_score < min_event_score: + continue + if min_wiki_spike is not None and wiki_spike < min_wiki_spike: + continue + if min_article_count is not None and article_count < min_article_count: + continue + if min_us_article_count is not None and us_article_count < min_us_article_count: + continue + if min_resolver_conf is not None and resolver_confidence < min_resolver_conf: + continue + + ret_5d = float(info.get("ret_5d") or 0.0) + entropy = info.get("entropy_20d") + entropy_rank = -(float(entropy) if entropy is not None else 1.0) + avg_dollar_vol = float(info.get("avg_dollar_vol_30d") or 0.0) + atr_14 = float(info.get("atr_14") or 0.0) + atr_pct = atr_14 / float(open_p) if open_p and atr_14 > 0 else 0.0 + wiki_rank = min(max(wiki_spike, 0.0), 10.0) / 10.0 + news_rank = min(max(float(max(article_count, us_article_count)), 0.0), 20.0) / 20.0 + signal_rank = ( + event_score * weight_event + + wiki_rank * weight_wiki + + news_rank * weight_news + ) + + candidates[day].append(ticker) + day_scores[day][ticker] = ( + signal_rank, + float(gap_pct), + ret_5d, + entropy_rank, + avg_dollar_vol, + atr_pct, + ) + + result: dict[str, list[str]] = {} + for day, tickers in candidates.items(): + ranked = sorted( + tickers, + key=lambda ticker: day_scores[day].get(ticker, (0.0, 0.0, 0.0, -1.0, 0.0, 0.0)), + reverse=True, + ) + result[day] = ranked if max_per_day is None else ranked[:max_per_day] + + return result + + +def _momentum_candidate_signal_passes( + info: dict, + strategy, +) -> bool: + """Return True when same-day catalyst/attention gates pass.""" + if strategy is None: + return True + require_event_flag = bool(getattr(strategy, "candidate_require_event_flag", False)) + min_event_score = getattr(strategy, "candidate_min_event_score", None) + min_wiki_spike = getattr(strategy, "candidate_min_attention_wiki_spike_10d", None) + min_article_count = getattr(strategy, "candidate_min_attention_article_count_3d", None) + min_us_article_count = getattr(strategy, "candidate_min_attention_us_article_count_3d", None) + min_resolver_conf = getattr(strategy, "candidate_min_attention_resolver_confidence", None) + + event_flag = bool(info.get("event_flag")) + event_score = float(info.get("event_score") or 0.0) + wiki_spike = float(info.get("attention_wiki_spike_10d") or 0.0) + article_count = int(info.get("attention_article_count_3d") or 0) + us_article_count = int(info.get("attention_us_article_count_3d") or 0) + resolver_confidence = float(info.get("attention_resolver_confidence") or 0.0) + + if require_event_flag and not event_flag: + return False + if min_event_score is not None and event_score < min_event_score: + return False + if min_wiki_spike is not None and wiki_spike < min_wiki_spike: + return False + if min_article_count is not None and article_count < min_article_count: + return False + if min_us_article_count is not None and us_article_count < min_us_article_count: + return False + if min_resolver_conf is not None and resolver_confidence < min_resolver_conf: + return False + return True + + +def _momentum_intraday_weighted_score( + info: dict, + daily_info: dict, + strategy, +) -> float: + """Weighted same-day candidate score for intraday-first ranking.""" + + def _clip_unit(value: float | None, cap: float) -> float: + if value is None or cap <= 0: + return 0.0 + return min(max(float(value), 0.0), cap) / cap + + def _dollar_vol_score(value: float | None) -> float: + if value is None or value <= 0: + return 0.0 + # 100k -> 0, 100M -> 1 on a log scale; enough to distinguish noisy + # small names from genuinely liquid intraday leaders. + scaled = (math.log10(float(value)) - 5.0) / 3.0 + return min(max(scaled, 0.0), 1.0) + + def _prior_dollar_vol_score(value: float | None) -> float: + if value is None or value <= 0: + return 0.0 + # 100M -> 0, 10B -> 1 on a log scale. This is intentionally narrower + # than entry-time dollar volume so only genuinely liquid large-cap + # leaders receive a meaningful ranking boost. + scaled = (math.log10(float(value)) - 8.0) / 2.0 + return min(max(scaled, 0.0), 1.0) + + gain_pct = float(info.get("gain_pct") or 0.0) + confirmation_return_pct = float(info.get("confirmation_return_pct") or 0.0) + volume_ratio_14d = float(info.get("volume_ratio_14d") or 0.0) + entry_dollar_volume = float(info.get("entry_dollar_volume") or 0.0) + avg_dollar_vol_30d = float(info.get("avg_dollar_vol_30d") or 0.0) + gap_pct = float(info.get("gap_pct") or 0.0) + ret_5d = float(info.get("ret_5d") or 0.0) + entropy_20d = info.get("entropy_20d") + low_entropy = ( + 1.0 - min(max(float(entropy_20d), 0.0), 1.0) + if entropy_20d is not None + else 0.0 + ) + + event_score = float(daily_info.get("event_score") or 0.0) + wiki_spike = float(daily_info.get("attention_wiki_spike_10d") or 0.0) + article_count = float( + max( + int(daily_info.get("attention_article_count_3d") or 0), + int(daily_info.get("attention_us_article_count_3d") or 0), + ) + ) + + score = 0.0 + score += float(getattr(strategy, "candidate_intraday_weight_gain", 0.0) or 0.0) * _clip_unit( + gain_pct, 0.10 + ) + score += float(getattr(strategy, "candidate_intraday_weight_confirmation", 0.0) or 0.0) * _clip_unit( + confirmation_return_pct, 0.02 + ) + score += float(getattr(strategy, "candidate_intraday_weight_volume_ratio", 0.0) or 0.0) * _clip_unit( + volume_ratio_14d, 0.20 + ) + score += float(getattr(strategy, "candidate_intraday_weight_entry_dollar_volume", 0.0) or 0.0) * _dollar_vol_score( + entry_dollar_volume + ) + score += float(getattr(strategy, "candidate_intraday_weight_avg_dollar_vol_30d", 0.0) or 0.0) * _prior_dollar_vol_score( + avg_dollar_vol_30d + ) + score += float(getattr(strategy, "candidate_intraday_weight_gap", 0.0) or 0.0) * _clip_unit( + gap_pct, 0.10 + ) + score += float(getattr(strategy, "candidate_intraday_weight_ret_5d", 0.0) or 0.0) * _clip_unit( + ret_5d, 0.20 + ) + score += float(getattr(strategy, "candidate_intraday_weight_low_entropy", 0.0) or 0.0) * low_entropy + score += float(getattr(strategy, "candidate_intraday_weight_event_score", 0.0) or 0.0) * _clip_unit( + event_score, 3.0 + ) + score += float(getattr(strategy, "candidate_intraday_weight_attention_wiki", 0.0) or 0.0) * _clip_unit( + wiki_spike, 10.0 + ) + score += float(getattr(strategy, "candidate_intraday_weight_attention_news", 0.0) or 0.0) * _clip_unit( + article_count, 20.0 + ) + return score + + +def momentum_intraday_first_candidates( + all_intraday: dict[str, dict[str, list[dict]]], + trading_days: list[str], + strategy, + *, + daily_enrichment: dict[str, dict[str, dict]] | None = None, + max_per_day: int | None = None, +) -> dict[str, list[str]]: + """Build the final momentum shortlist from entry-time intraday information. + + This is used after a broader, still lookahead-free daily seed shortlist has + already bounded the intraday fetch set. The final ranking uses only + information known by the entry / confirmation bar of the same day. + """ + from libs.intraday.simulator import _select_momentum_sleeves, compute_morning_gains + + if max_per_day is None: + max_per_day = max(1, int(getattr(strategy, "candidate_final_max_per_day", 30) or 30)) + + shortlist_strategy = strategy.model_copy( + update={ + "top_n": max_per_day, + "ticker_cooldown_days": 0, + "max_positions_per_sector": None, + "market_regime_spy_threshold": None, + "max_vix": None, + } + ) + + result: dict[str, list[str]] = {} + for day in trading_days: + day_bars = all_intraday.get(day, {}) + if not day_bars: + continue + gains = compute_morning_gains( + day_bars, + shortlist_strategy, + day, + daily_features_by_ticker={ + ticker: (daily_enrichment or {}).get(ticker, {}).get(day, {}) + for ticker in day_bars.keys() + }, + ) + if not gains: + continue + filtered_gains = { + ticker: info + for ticker, info in gains.items() + if _momentum_candidate_signal_passes( + (daily_enrichment or {}).get(ticker, {}).get(day, {}), + strategy, + ) + } + if not filtered_gains: + continue + rank_mode = str(getattr(strategy, "candidate_intraday_rank_mode", "sleeves") or "sleeves").lower() + if rank_mode == "weighted": + ranked = sorted( + filtered_gains.items(), + key=lambda item: ( + _momentum_intraday_weighted_score( + item[1], + (daily_enrichment or {}).get(item[0], {}).get(day, {}), + strategy, + ), + item[1].get("confirmation_return_pct") or 0.0, + item[1].get("gain_pct") or 0.0, + item[1].get("entry_dollar_volume") or 0.0, + item[1].get("volume_ratio_14d") or 0.0, + ), + reverse=True, + ) + if ranked: + result[day] = [ticker for ticker, _info in ranked[:max_per_day]] + continue + picks = _select_momentum_sleeves(filtered_gains, shortlist_strategy, ticker_sectors=None) + if picks: + result[day] = [ticker for ticker, _sleeve in picks[:max_per_day]] + return result + + # ── ORB Pre-Screening ───────────────────────────────────────────────────── @@ -292,7 +993,8 @@ async def fetch_intraday_bulk( client: OracleClient, cache: IntradayCache | None, interval: str = "5min", - concurrency: int = 8, + skip_oracle_when_unhealthy: bool = False, + concurrency: int = 3, progress_callback: Callable[[int, int, int, int], None] | None = None, ) -> dict[str, dict[str, list[dict]]]: """Fetch 5-min intraday bars for candidate ticker-days (cache-first). @@ -320,65 +1022,122 @@ async def fetch_intraday_bulk( result: dict[str, dict[str, list[dict]]] = defaultdict(dict) misses: dict[str, list[str]] = defaultdict(list) # {day: [ticker, ...]} + hits: list[tuple[str, str]] = [] # (day, ticker) to read later - # Phase 1: resolve cache hits synchronously + # Phase 1a: fast existence check — only calls p.exists() + metadata header, + # defers full Parquet read until after API fetches to show early progress. + PROGRESS_INTERVAL = 2000 for day in sorted(candidates.keys()): for ticker in candidates[day]: - cached = cache.get(ticker, day) if cache else None - if cached is not None: - result[day][ticker] = cached - completed += 1 + if cache and cache.has(ticker, day): + hits.append((day, ticker)) cache_hits += 1 else: misses[day].append(ticker) + completed += 1 + if progress_callback and completed % PROGRESS_INTERVAL == 0: + progress_callback(completed, total, cache_hits, api_calls) if progress_callback: progress_callback(completed, total, cache_hits, api_calls) + # Phase 1b: read cached Parquet files in parallel (avoid blocking event loop) + if hits: + READ_CONCURRENCY = 32 + read_semaphore = asyncio.Semaphore(READ_CONCURRENCY) + + async def read_one(day: str, ticker: str) -> None: + async with read_semaphore: + cached = await asyncio.to_thread(cache.get, ticker, day) # type: ignore[union-attr] + if cached: + async with lock: + result[day][ticker] = cached + + await asyncio.gather(*[ + asyncio.create_task(read_one(day, ticker)) + for day, ticker in hits + ]) + # Phase 2: fetch cache misses via multi-ticker endpoint (batched by date, chunk ≤ 75) - CHUNK = 75 + miss_total = sum(len(v) for v in misses.values()) + api_completed = 0 # Separate counter: starts at 0 for Phase 2 - async def fetch_day_chunk(day: str, chunk: list[str]) -> None: - nonlocal completed, api_calls + # Signal Phase 2 start so the caller can reset its progress bar + if progress_callback and miss_total > 0: + progress_callback(0, miss_total, cache_hits, 0) + + if skip_oracle_when_unhealthy and miss_total > 0: + oracle_available = await client.health_check_fast() + if not oracle_available: + if progress_callback: + progress_callback(miss_total, miss_total, cache_hits, 0) + return dict(result) + + CHUNK = 30 + + async def _fetch_intraday_chunk_from_api(day: str, chunk: list[str]) -> dict[str, list[dict]]: fetched: dict[str, list[dict]] = {} async with semaphore: - try: - raw = await client.get( - "/api/v1/alpaca/intraday", - params={ - "tickers": ",".join(chunk), - "interval": oracle_interval, - "start_date": day, - "end_date": day, - }, - ) - bars_by_ticker = raw.get("bars", {}) - for ticker in chunk: - fetched[ticker] = [ - { - "timestamp": b.get("timestamp", ""), - "open": float(b.get("open", 0)), - "high": float(b.get("high", 0)), - "low": float(b.get("low", 0)), - "close": float(b.get("close", 0)), - "volume": float(b.get("volume", 0)), - "vwap": float(b.get("vwap", 0) or 0), - } - for b in bars_by_ticker.get(ticker, []) - ] - except Exception: - pass + raw = await client.get( + "/api/v1/alpaca/intraday", + params={ + "tickers": ",".join(chunk), + "interval": oracle_interval, + "start_date": day, + "end_date": day, + }, + ) + bars_by_ticker = raw.get("bars", {}) + for ticker in chunk: + fetched[ticker] = [ + { + "timestamp": b.get("timestamp", ""), + "open": float(b.get("open", 0)), + "high": float(b.get("high", 0)), + "low": float(b.get("low", 0)), + "close": float(b.get("close", 0)), + "volume": float(b.get("volume", 0)), + "vwap": float(b.get("vwap", 0) or 0), + } + for b in bars_by_ticker.get(ticker, []) + ] + return fetched + + async def fetch_day_chunk(day: str, chunk: list[str]) -> None: + nonlocal api_completed, api_calls + fetched: dict[str, list[dict]] = {} + request_succeeded = False + try: + fetched = await _fetch_intraday_chunk_from_api(day, chunk) + request_succeeded = True + except Exception: + # Split failed chunks recursively down to singleton requests. + # This prevents one transient bulk failure from leaving permanent + # holes that get re-fetched on every later backtest run. + if len(chunk) == 1: + fetched = {} + else: + mid = len(chunk) // 2 + left = chunk[:mid] + right = chunk[mid:] + await fetch_day_chunk(day, left) + await fetch_day_chunk(day, right) + return async with lock: for ticker in chunk: bars = fetched.get(ticker, []) - if cache and bars: - cache.put(ticker, day, bars) - if bars: + if cache: + if IntradayCache.is_complete_enough(bars): + cache.put(ticker, day, bars) + elif request_succeeded: + reason = "sparse" if bars else "empty" + cache.put_negative(ticker, day, reason=reason) + if IntradayCache.is_complete_enough(bars): result[day][ticker] = bars - completed += len(chunk) + api_completed += len(chunk) api_calls += 1 - _c, _t, _ch, _ac = completed, total, cache_hits, api_calls + _c, _t, _ch, _ac = api_completed, miss_total, cache_hits, api_calls if progress_callback: progress_callback(_c, _t, _ch, _ac)