"""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 from zoneinfo import ZoneInfo 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 def _daily_bars_cover_requested_range(bars: list[dict]) -> bool: if not bars: return False if _bulk_daily_bars_look_truncated(bars): return False try: start_dt = date.fromisoformat(start_date) end_dt = date.fromisoformat(end_date) row_dates = [ date.fromisoformat(str(row.get("date", ""))[:10]) for row in bars if row.get("date") ] except Exception: return True if not row_dates: return False first_row = min(row_dates) last_row = max(row_dates) if (end_dt - last_row).days > 10: return False # Late starts can be legitimate IPO/listing history. A very small # payload after a wide requested warmup is almost always a local # intraday-cache fragment and must be repaired from Oracle instead. if (first_row - start_dt).days > 10 and len(row_dates) < 20: return False return True 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) if cache is not None and bool(getattr(cache, "strict_misses", False)): for ticker, (cached_bars, _) in partial_hits.items(): if _daily_bars_cover_requested_range(cached_bars): results[ticker] = cached_bars completed = len(tickers) _emit_progress() return _ordered_results() # 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 if _daily_bars_cover_requested_range(merged): 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 if the requested range is still covered. for ticker in chunk: cached_bars, _ = partial_hits[ticker] if _daily_bars_cover_requested_range(cached_bars): async with lock: results[ticker] = cached_bars completed += 1 else: async with lock: 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 and _daily_bars_cover_requested_range(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 ] if _daily_bars_cover_requested_range(bars): async with lock: results[ticker] = bars if cache: await asyncio.to_thread(cache.put, ticker, start_date, end_date, bars) else: rebuilt = await _rebuild_daily_from_intraday_cache(ticker) if rebuilt and _daily_bars_cover_requested_range(rebuilt): async with lock: results[ticker] = rebuilt if cache: await asyncio.to_thread( cache.put, ticker, start_date, end_date, rebuilt ) except Exception: rebuilt = await _rebuild_daily_from_intraday_cache(ticker) if rebuilt and _daily_bars_cover_requested_range(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 not _daily_bars_cover_requested_range(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 and _daily_bars_cover_requested_range(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 allowed_event_types = _momentum_allowed_event_types(strategy) 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, event_score = _momentum_effective_event_state(info, allowed_event_types) 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_allowed_event_types(strategy) -> set[str]: if strategy is None: return set() return { str(value).strip().lower() for value in getattr(strategy, "candidate_allowed_event_types", []) if str(value).strip() } def _momentum_event_types_pass( info: dict, allowed_event_types: set[str], ) -> bool: if not allowed_event_types: return True raw_event_types = info.get("event_types") or [] event_types = { str(value).strip().lower() for value in raw_event_types if str(value).strip() } if not event_types: return False return any(event_type in allowed_event_types for event_type in event_types) def _momentum_effective_event_state( info: dict, allowed_event_types: set[str], ) -> tuple[bool, float]: event_flag = bool(info.get("event_flag")) event_score = float(info.get("event_score") or 0.0) if not event_flag: return False, 0.0 if allowed_event_types and not _momentum_event_types_pass(info, allowed_event_types): return False, 0.0 return True, event_score 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) allowed_event_types = _momentum_allowed_event_types(strategy) 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, event_score = _momentum_effective_event_state(info, allowed_event_types) 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.""" from libs.intraday.simulator import _same_day_support_score 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 ) support_score = _same_day_support_score(info) 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_support_score", 0.0) or 0.0) * support_score score += float(getattr(strategy, "candidate_intraday_weight_liquid_largecap", 0.0) or 0.0) * ( 1.0 if info.get("is_liquid_largecap") else 0.0 ) score += float(getattr(strategy, "candidate_intraday_weight_moderate_gap_liquid", 0.0) or 0.0) * ( 1.0 if info.get("is_moderate_gap_liquid") else 0.0 ) 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_sector_thrust", 0.0) or 0.0) * _clip_unit( float(info.get("sector_thrust_score") or 0.0), 1.0, ) 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_liquid_continuation_score( info: dict, ) -> tuple[float, float, float, float, float, float, float, float, float]: from libs.intraday.simulator import _same_day_support_score entropy_20d = info.get("entropy_20d") return ( 1.0 if info.get("is_moderate_gap_liquid") else 0.0, 1.0 if info.get("is_liquid_largecap") else 0.0, 1.0 if info.get("is_sector_thrust") else 0.0, _same_day_support_score(info), float(info.get("confirmation_return_pct") or 0.0), float(info.get("entry_dollar_volume") or 0.0), float(info.get("avg_dollar_vol_30d") or 0.0), float(info.get("gain_pct") or 0.0), -(float(entropy_20d) if entropy_20d is not None else 1.0), ) def _momentum_intraday_event_reserve_eligible( daily_info: dict, strategy, ) -> bool: if not bool(daily_info.get("event_flag")): return False allowed_event_types = _momentum_allowed_event_types(strategy) if not _momentum_event_types_pass(daily_info, allowed_event_types): 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, ticker_sectors: dict[str, str] | 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 ( _annotate_sector_thrust_features, _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, ) gains = _annotate_sector_thrust_features( gains, shortlist_strategy, ticker_sectors, ) 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 if rank_mode == "liquid_continuation": rankable_liquid_gains = { ticker: info for ticker, info in rankable_gains.items() if ( info.get("is_moderate_gap_liquid") or info.get("is_liquid_largecap") or info.get("is_sector_thrust") ) } if not rankable_liquid_gains: continue ranked = sorted( rankable_liquid_gains.items(), key=lambda item: _momentum_intraday_liquid_continuation_score(item[1]), 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=ticker_sectors, ) 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 now_et = datetime.now(ZoneInfo("America/New_York")) today_et = now_et.date().isoformat() # 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]: # Same-day bars are mutable. A live or delayed historical response can # contain only a partial session, so never let today's cache satisfy a # backtest fetch. Cache writes are still allowed for tomorrow's reuse. if cache and day != today_et 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 use_live_today_first = time(4, 0) <= now_et.time() < time(16, 15) def _normalize_intraday_response(raw: object, chunk: list[str]) -> dict[str, list[dict]]: fetched: dict[str, list[dict]] = {} bars_by_ticker = raw.get("bars", {}) if isinstance(raw, dict) else {} 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 def _intraday_bars_complete_for_use( day: str, bars: list[dict], *, live_today: bool, ) -> bool: if not IntradayCache.is_complete_enough(bars): return False if day != today_et: return True # During the live session partial same-day bars are expected. After the # close, though, treating a noon-truncated response as a full backtest day # creates false exits and false positives. if now_et.time() < time(16, 15): return True last_et: datetime | None = None for bar in bars: ts_raw = str(bar.get("timestamp", "")) if not ts_raw: continue try: ts = ( datetime.fromisoformat(ts_raw[:-1] + "+00:00") if ts_raw.endswith("Z") else datetime.fromisoformat(ts_raw) ) ts_et = ts.astimezone(ZoneInfo("America/New_York")) except Exception: continue if ts_et.date().isoformat() == day and (last_et is None or ts_et > last_et): last_et = ts_et if last_et is None: return False return last_et.time() >= time(15, 45) async def _request_intraday_chunk( day: str, chunk: list[str], *, live_today: bool, ) -> dict[str, list[dict]]: path = "/api/v1/alpaca/intraday/today" if live_today else "/api/v1/alpaca/intraday" params = {"tickers": ",".join(chunk), "interval": oracle_interval} if not live_today: params.update({"start_date": day, "end_date": day}) async with semaphore: raw = await client.get(path, params=params) return _normalize_intraday_response(raw, chunk) async def _fetch_intraday_chunk_from_api(day: str, chunk: list[str]) -> tuple[dict[str, list[dict]], bool]: is_today = day == today_et attempts = ( [True] # /today only; historical endpoint rejects today's date during live session if is_today and use_live_today_first else [False, True] if is_today else [False] ) last_error: Exception | None = None for live_today in attempts: try: fetched = await _request_intraday_chunk(day, chunk, live_today=live_today) except Exception as exc: last_error = exc continue if any( _intraday_bars_complete_for_use(day, bars, live_today=live_today) for bars in fetched.values() ): return fetched, live_today # After-hours /intraday/today can legitimately be empty; try the # historical same-day endpoint before declaring this chunk empty. if not is_today or live_today != attempts[-1]: continue return fetched, live_today if last_error is not None: raise last_error return {ticker: [] for ticker in chunk}, bool(attempts[-1]) async def fetch_day_chunk(day: str, chunk: list[str]) -> None: nonlocal api_completed, api_calls fetched: dict[str, list[dict]] = {} live_today_response = False request_succeeded = False try: fetched, live_today_response = 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, []) complete_for_use = _intraday_bars_complete_for_use( day, bars, live_today=live_today_response, ) if cache: if complete_for_use and not live_today_response: cache.put(ticker, day, bars) elif request_succeeded and day != today_et: reason = "sparse" if bars else "empty" if not live_today_response: cache.put_negative(ticker, day, reason=reason) if complete_for_use: 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)