Fix daily bar rebuild to use market-hours close; clean ORB simulator debug code

Two fixes:
1. _rebuild_daily_from_intraday_cache now filters to regular market hours
   (9:30–16:00 ET) before computing OHLCV. Previously used bars[-1] which
   included after-hours data, distorting prev_close for gap calculations.
   Root cause of V23 regression: HIMS Aug-4 after-hours drop to $54.81
   made it appear as a +0.89% gap on Aug 5 instead of the correct -12.85%
   gap (from $63.45 market close), causing it to fail min_abs_gap_pct filter.
   V23 with fix: +109.32%, WR 58.1%, Sharpe 3.01, DD -12.91%

2. Remove temporary debug instrumentation (HIMS/2025-08-05 trace blocks)
   that was left in orb_simulator.py during regression investigation.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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)
+        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=params.market_cap_min,
+                market_cap_min=market_cap_min,
-            min_avg_volume=params.avg_volume_min,
+                min_avg_volume=min_avg_volume,
-            price_min=params.min_price if params.min_price > 0 else None,
+                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,
             )
-        return sorted({s.symbol.upper() for s in stocks if s.symbol})
+            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:
+                _emit_progress()
-                    progress_callback(completed, len(tickers))
 
-    tasks = [asyncio.create_task(fetch_one(t)) for t in tickers]
+    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 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:
+
+        gap_pct = (today_open - prev_close) / prev_close
 
-    A ticker-day is a candidate if (high - open) / open >= threshold.
+    This uses only information known at the open of the session plus prior-day
-    This indicates the stock rose significantly above its opening price at some
+    data. It intentionally avoids using the same day's high/close because those
-    point during the day — a necessary condition for being a morning gainer.
+    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}}
+    trading_day_set = set(trading_days)
-    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 ticker, bars in daily_bars.items():
-        for day in trading_days:
+        sorted_bars = sorted(bars, key=lambda bar: bar["date"])
-            b = date_map.get(day)
+        prev_close: float | None = None
-            if not b:
+        for bar in sorted_bars:
-                continue
+            day = bar["date"][:10]
-            open_p = b.get("open", 0)
+            open_p = bar.get("open")
-            high_p = b.get("high", 0)
+            gap_pct: float | None = None
-            if open_p <= 0:
+
-                continue
+            ticker_enrich = enrichment.get(ticker, {}).get(day, {}) if enrichment else {}
-            score = (high_p - open_p) / open_p
+            if ticker_enrich:
-            if score >= threshold:
+                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,29 +1022,62 @@ 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 cache and cache.has(ticker, day):
-            if cached is not None:
+                hits.append((day, ticker))
-                result[day][ticker] = cached
-                completed += 1
                 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:
+    # Signal Phase 2 start so the caller can reset its progress bar
-        nonlocal completed, api_calls
+    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={
@@ -366,19 +1101,43 @@ async def fetch_intraday_bulk(
                 }
                 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:
-                pass
+            # 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:
+                if cache:
+                    if IntradayCache.is_complete_enough(bars):
                         cache.put(ticker, day, bars)
-                if 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)