"""Two-phase intraday data loading pipeline. Phase 1: Fetch daily bars for the full universe → identify candidate ticker-days. Phase 2: Fetch 5-min intraday bars for candidates (cache-first). This approach reduces API calls by ~97% vs brute-force (fetching intraday for all stocks every day). """ 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.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 from libs.oracle_client.screener import ScreenerService # ── Universe Shortcuts ───────────────────────────────────────────────────── _UNIVERSE_YAML_MAP = { "broad": "configs/symbols_broad_snapshot_3408.yaml", "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. Supported sources: - 'sp500': S&P 500 constituents from Oracle index endpoint - 'nasdaq100': Nasdaq 100 constituents from Oracle index endpoint - 'broad': broad 3408-ticker YAML snapshot - '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 Returns sorted, deduplicated list of uppercase ticker symbols. """ source = params.source.lower() # Index-based sources if source in ("sp500", "nasdaq100"): data = await client.get(f"/api/v1/stocks/index/{source}") tickers = [c["symbol"].upper() for c in data.get("constituents", []) if c.get("symbol")] return sorted(set(tickers)) # YAML file shortcuts if source in _UNIVERSE_YAML_MAP: yaml_path = _UNIVERSE_YAML_MAP[source] return _load_yaml_symbols(yaml_path) # Custom YAML file if source == "yaml": if not params.symbols_file: raise ValueError("source='yaml' requires symbols_file to be set") return _load_yaml_symbols(params.symbols_file) # Live screener query if source == "screener": screener = ScreenerService(client) 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, broad, midlarge, largecap, midcap, smallmid, yaml, screener") def _load_yaml_symbols(path: str) -> list[str]: """Load ticker list from a YAML symbols file.""" with open(path) as f: data = yaml.safe_load(f) if isinstance(data, list): return sorted({str(s).upper() for s in data if s}) if isinstance(data, dict): symbols: list[str] = [] for key in ("symbols", "existing_only", "screener", "existing-only"): if key in data: val = data[key] if isinstance(val, list): symbols.extend(str(s).upper() for s in val if s) if symbols: return sorted(set(symbols)) # Flat dict of symbol -> metadata return sorted({str(k).upper() for k in data.keys() if not k.startswith("_")}) raise ValueError(f"Unexpected YAML format in {path}") # ── Phase 1: Daily Bar Bulk Fetch ────────────────────────────────────────── async def fetch_daily_bars_bulk( tickers: list[str], start_date: str, end_date: str, client: OracleClient, 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 using bulk chunks, with per-ticker fallback. Returns {ticker: [bar_dict, ...]}. Tickers with no data or errors are silently omitted. """ svc = PriceService(client) semaphore = asyncio.Semaphore(concurrency) 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 async with semaphore: try: resp = await svc.get_daily_bars(ticker, start=start_date, end=end_date) if resp.bars: bars = [ { "date": b.date, "open": b.open, "high": b.high, "low": b.low, "close": b.close, "volume": b.volume, } for b in resp.bars ] async with lock: results[ticker] = bars if cache: await asyncio.to_thread(cache.put, ticker, start_date, end_date, bars) except Exception: 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 _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, }, ) 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 _ordered_results() # ── Phase 2: Pre-Screening ───────────────────────────────────────────────── def pre_screen_candidates( daily_bars: dict[str, list[dict]], 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 lookahead-free opening-gap heuristics. A ticker-day is a candidate if the stock opens at least ``threshold`` above the prior close: 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 opening gap descending, capped at max_per_day per date. """ trading_day_set = set(trading_days) candidates: dict[str, list[str]] = defaultdict(list) day_scores: dict[str, dict[str, float]] = defaultdict(dict) 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] = 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]] = {} for day, tickers in candidates.items(): ranked = sorted(tickers, key=lambda t: day_scores[day].get(t, 0), reverse=True) result[day] = ranked[:max_per_day] 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_event_reserve_eligible( daily_info: dict, strategy, ) -> bool: if not bool(daily_info.get("event_flag")): return False min_score = getattr(strategy, "candidate_intraday_event_reserve_min_score", None) if min_score is None: return True return float(daily_info.get("event_score") or 0.0) >= float(min_score) def _momentum_intraday_event_reserve_score( info: dict, daily_info: dict, ) -> tuple[float, float, float, float, float]: entropy_20d = info.get("entropy_20d") return ( float(daily_info.get("event_score") or 0.0), float(info.get("confirmation_return_pct") or 0.0), float(info.get("gain_pct") or 0.0), float(info.get("entry_dollar_volume") or 0.0), -(float(entropy_20d) if entropy_20d is not None else 1.0), ) def _momentum_intraday_moderate_liquid_score( info: dict, ) -> tuple[float, float, float, float, float]: entropy_20d = info.get("entropy_20d") return ( float(info.get("confirmation_return_pct") or 0.0), float(info.get("entry_dollar_volume") or 0.0), float(info.get("gain_pct") or 0.0), float(info.get("avg_dollar_vol_30d") or 0.0), -(float(entropy_20d) if entropy_20d is not None else 1.0), ) def _apply_momentum_intraday_moderate_liquid_reserve( ranked_tickers: list[str], filtered_gains: dict[str, dict], daily_info_by_ticker: dict[str, dict], strategy, *, max_per_day: int, ) -> list[str]: reserve_slots = max( 0, int(getattr(strategy, "candidate_intraday_moderate_liquid_reserve_slots", 0) or 0), ) if reserve_slots <= 0 or max_per_day <= 0: return ranked_tickers[:max_per_day] base = list(ranked_tickers[:max_per_day]) trigger_below = max( 0, int(getattr(strategy, "candidate_intraday_moderate_liquid_reserve_trigger_below", 0) or 0), ) if trigger_below > 0 and base: from libs.intraday.simulator import _select_momentum_sleeves base_gains = { ticker: filtered_gains[ticker] for ticker in base if ticker in filtered_gains } selection_strategy = strategy.model_copy( update={ "ticker_cooldown_days": 0, "max_positions_per_sector": None, "market_regime_spy_threshold": None, "max_vix": None, } ) base_picks = _select_momentum_sleeves(base_gains, selection_strategy, ticker_sectors=None) if len(base_picks) >= trigger_below: return base reserve_ranked = [ ticker for ticker, _info in sorted( ( (ticker, filtered_gains[ticker]) for ticker in filtered_gains if filtered_gains[ticker].get("is_moderate_gap_liquid") ), key=lambda item: _momentum_intraday_moderate_liquid_score(item[1]), reverse=True, ) ] if not reserve_ranked: return base existing = sum( 1 for ticker in base if filtered_gains.get(ticker, {}).get("is_moderate_gap_liquid") ) needed = max(0, min(reserve_slots, len(reserve_ranked)) - existing) if needed <= 0: return base replacement_indices = [ idx for idx in range(len(base) - 1, -1, -1) if ( not filtered_gains.get(base[idx], {}).get("is_moderate_gap_liquid") and not _momentum_intraday_event_reserve_eligible( daily_info_by_ticker.get(base[idx], {}), strategy, ) ) ] result = list(base) used = set(result) for ticker in reserve_ranked: if needed <= 0: break if ticker in used: continue if len(result) < max_per_day: result.append(ticker) used.add(ticker) needed -= 1 continue if not replacement_indices: break replace_idx = replacement_indices.pop(0) used.discard(result[replace_idx]) result[replace_idx] = ticker used.add(ticker) needed -= 1 return result[:max_per_day] def _momentum_intraday_soft_day( day_bars: dict[str, list[dict]], daily_info_by_ticker: dict[str, dict], strategy, ) -> bool: from libs.intraday.simulator import _day_breadth_scaler, _day_regime_scaler regime_scaler, regime_skip = _day_regime_scaler(strategy, daily_info_by_ticker) if regime_skip is not None: return True breadth_scaler, breadth_skip = _day_breadth_scaler(strategy, day_bars, daily_info_by_ticker) if breadth_skip is not None: return True return (regime_scaler * breadth_scaler) < float( getattr(strategy, "soft_day_scaler_threshold", 1.0) or 1.0 ) def _apply_momentum_intraday_event_reserve( ranked_tickers: list[str], filtered_gains: dict[str, dict], daily_info_by_ticker: dict[str, dict], strategy, *, day_bars: dict[str, list[dict]], max_per_day: int, ) -> list[str]: reserve_slots = max(0, int(getattr(strategy, "candidate_intraday_event_reserve_slots", 0) or 0)) if reserve_slots <= 0 or max_per_day <= 0: return ranked_tickers[:max_per_day] if ( bool(getattr(strategy, "candidate_intraday_event_reserve_soft_day_only", False)) and not _momentum_intraday_soft_day(day_bars, daily_info_by_ticker, strategy) ): return ranked_tickers[:max_per_day] base = list(ranked_tickers[:max_per_day]) reserve_ranked = [ ticker for ticker, _info in sorted( ( (ticker, filtered_gains[ticker]) for ticker in filtered_gains if _momentum_intraday_event_reserve_eligible( daily_info_by_ticker.get(ticker, {}), strategy, ) ), key=lambda item: _momentum_intraday_event_reserve_score( item[1], daily_info_by_ticker.get(item[0], {}), ), reverse=True, ) ] if not reserve_ranked: return base existing_reserve = sum( 1 for ticker in base if _momentum_intraday_event_reserve_eligible( daily_info_by_ticker.get(ticker, {}), strategy, ) ) needed = max(0, min(reserve_slots, len(reserve_ranked)) - existing_reserve) if needed <= 0: return base replacement_indices = [ idx for idx in range(len(base) - 1, -1, -1) if not _momentum_intraday_event_reserve_eligible( daily_info_by_ticker.get(base[idx], {}), strategy, ) ] result = list(base) used = set(result) for ticker in reserve_ranked: if needed <= 0: break if ticker in used: continue if len(result) < max_per_day: result.append(ticker) used.add(ticker) needed -= 1 continue if not replacement_indices: break replace_idx = replacement_indices.pop(0) used.discard(result[replace_idx]) result[replace_idx] = ticker used.add(ticker) needed -= 1 return result 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 daily_info_by_ticker = { ticker: (daily_enrichment or {}).get(ticker, {}).get(day, {}) for ticker in day_bars.keys() } gains = compute_morning_gains( day_bars, shortlist_strategy, day, daily_features_by_ticker=daily_info_by_ticker, ) 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 rankable_gains = filtered_gains if int(getattr(strategy, "candidate_intraday_moderate_liquid_reserve_trigger_below", 0) or 0) > 0: rankable_gains = { ticker: info for ticker, info in filtered_gains.items() if not daily_info_by_ticker.get(ticker, {}).get( "candidate_seed_moderate_liquid_overlay" ) } if not rankable_gains: rankable_gains = filtered_gains rank_mode = str(getattr(strategy, "candidate_intraday_rank_mode", "sleeves") or "sleeves").lower() if rank_mode == "weighted": ranked = sorted( rankable_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: ranked_tickers = _apply_momentum_intraday_event_reserve( [ticker for ticker, _info in ranked], filtered_gains, daily_info_by_ticker, strategy, day_bars=day_bars, max_per_day=max_per_day, ) result[day] = _apply_momentum_intraday_moderate_liquid_reserve( ranked_tickers, filtered_gains, daily_info_by_ticker, strategy, max_per_day=max_per_day, ) continue picks = _select_momentum_sleeves(rankable_gains, shortlist_strategy, ticker_sectors=None) if picks: ranked_tickers = _apply_momentum_intraday_event_reserve( [ticker for ticker, _sleeve in picks], filtered_gains, daily_info_by_ticker, strategy, day_bars=day_bars, max_per_day=max_per_day, ) result[day] = _apply_momentum_intraday_moderate_liquid_reserve( ranked_tickers, filtered_gains, daily_info_by_ticker, strategy, max_per_day=max_per_day, ) return result # ── ORB Pre-Screening ───────────────────────────────────────────────────── def orb_pre_screen_candidates( daily_bars: dict[str, list[dict]], trading_days: list[str], enrichment: dict[str, dict[str, dict]], min_price: float = 10.0, min_atr: float = 0.50, min_avg_dollar_vol: float = 25_000_000.0, max_per_day: int | None = 50, ) -> dict[str, list[str]]: """Identify ORB candidate ticker-days using liquidity and volatility filters. All filters use PRIOR-day information only (no lookahead): - price >= min_price (from today's daily bar open — known at market open) - ATR(14) >= min_atr (from prior bars via enrichment) - avg_dollar_vol_30d >= min_avg_dollar_vol (from prior bars via enrichment) Ranking uses ATR/price (expected daily range as % of price) — a lookahead-free proxy for intraday volatility. Higher ATR/price = more likely to produce meaningful ORB breakouts. Returns: {date: [ticker1, ticker2, ...]} — ranked by ATR/price descending, capped at max_per_day per date when max_per_day is set. """ from collections import defaultdict # Build {ticker -> {date -> bar}} for quick lookup 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 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) if open_p < min_price: continue # All filters from enrichment (computed from PRIOR bars — no lookahead) ticker_enrich = enrichment.get(ticker, {}).get(day, {}) atr = ticker_enrich.get("atr_14") avg_dollar_vol = ticker_enrich.get("avg_dollar_vol_30d") if atr is None or atr < min_atr: continue if avg_dollar_vol is None or avg_dollar_vol < min_avg_dollar_vol: continue # Score: ATR / price = expected daily range % (lookahead-free) atr_pct = atr / open_p candidates[day].append(ticker) day_scores[day][ticker] = atr_pct # Rank by ATR/price descending and optionally cap result: dict[str, list[str]] = {} for day, tickers in candidates.items(): ranked = sorted(tickers, key=lambda t: day_scores[day].get(t, 0), reverse=True) result[day] = ranked if max_per_day is None else ranked[:max_per_day] return result # ── Phase 2: Intraday Bar Bulk Fetch ────────────────────────────────────── async def fetch_intraday_bulk( candidates: dict[str, list[str]], client: OracleClient, cache: IntradayCache | None, interval: str = "5min", 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). Returns {date: {ticker: [bar_dict, ...]}}. Args: candidates: {date -> [ticker, ...]} from pre_screen_candidates(). client: Oracle API client. cache: Intraday Parquet cache. interval: Candle interval (default '5min'). concurrency: Max concurrent API calls (Semaphore). Keep <= 8 to respect rate limits. progress_callback: Called with (completed, total, cache_hits, api_calls). """ # Oracle endpoint uses "5m" format; config uses "5min" format oracle_interval = interval.replace("min", "m") # Build flat list and separate cache hits from misses total = sum(len(tickers) for tickers in candidates.values()) completed = 0 cache_hits = 0 api_calls = 0 lock = asyncio.Lock() semaphore = asyncio.Semaphore(concurrency) 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 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]: 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) miss_total = sum(len(v) for v in misses.values()) api_completed = 0 # Separate counter: starts at 0 for Phase 2 # 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: 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: 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 api_completed += len(chunk) api_calls += 1 _c, _t, _ch, _ac = api_completed, miss_total, cache_hits, api_calls if progress_callback: progress_callback(_c, _t, _ch, _ac) tasks = [ asyncio.create_task(fetch_day_chunk(day, chunk)) for day, tickers in sorted(misses.items()) for chunk in ( tickers[i : i + CHUNK] for i in range(0, len(tickers), CHUNK) ) ] await asyncio.gather(*tasks) return dict(result)