fix intraday evidence pipeline

backend/app/providers/replay_data_provider.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from datetime import datetime, timezone
+from datetime import datetime, time as clock_time, timedelta, timezone
 from typing import Protocol
 from urllib.parse import quote
 
@@ -13,6 +13,11 @@ from app.providers.yfinance_provider import NasdaqHistoricalProvider, StooqProvi
 
 
 REPLAY_TIMEFRAMES = frozenset({"1d", "15m", "5m", "1m"})
+INTRADAY_RETENTION_DAYS = {
+    "1m": 7,
+    "5m": 59,
+    "15m": 59,
+}
 
 
 @dataclass(frozen=True)
@@ -41,6 +46,88 @@ class ReplayDataProvider(Protocol):
     def fetch(self, request: ReplayDataRequest) -> ProviderBars: ...
 
 
+class NasdaqChartReplayDataProvider:
+    """Public US intraday fallback using Nasdaq's observed last-sale chart path."""
+
+    name = "nasdaq_chart"
+    supported_timeframes = frozenset({"1m", "5m", "15m"})
+    source_metadata = {
+        "source": "Nasdaq public quote chart API",
+        "license": "Public endpoint; downstream use remains subject to provider terms.",
+        "bar_construction": "observed_last_sale_path",
+        "ohlcv_completeness": "Price path only; volume unavailable and OHLC is aggregated from observed minute values.",
+        "data_quality_score": 55.0,
+    }
+
+    def __init__(self) -> None:
+        self._observed_cache: dict[tuple[str, str], tuple[pd.DataFrame, str | None]] = {}
+
+    def fetch(self, request: ReplayDataRequest) -> ProviderBars:
+        if request.timeframe not in self.supported_timeframes:
+            return ProviderBars(self.name, source_metadata=self.source_metadata, blockers=["UNSUPPORTED_TIMEFRAME"])
+        if str(request.market or "").strip().upper() not in {"USA", "US", "UNITED STATES", "NASDAQ", "NYSE"}:
+            return ProviderBars(self.name, source_metadata=self.source_metadata, blockers=["UNSUPPORTED_MARKET"])
+        observed, blocker = self._observed_path(request)
+        if blocker:
+            return ProviderBars(self.name, source_metadata=self.source_metadata, blockers=[blocker])
+        start = _naive_utc(request.start)
+        end = _naive_utc(request.end)
+        observed = observed.loc[(observed.index >= start) & (observed.index <= end)]
+        if observed.empty:
+            return ProviderBars(self.name, source_metadata=self.source_metadata, blockers=["NO_INTRADAY_HISTORY"])
+        if request.timeframe == "1m":
+            frame = observed.assign(Open=observed["Close"], High=observed["Close"], Low=observed["Close"], Volume=None)
+        else:
+            frequency = {"5m": "5min", "15m": "15min"}[request.timeframe]
+            frame = observed["Close"].resample(frequency).agg(Open="first", High="max", Low="min", Close="last")
+            frame["Volume"] = None
+        return ProviderBars(self.name, normalize_replay_frame(frame), self.source_metadata, [])
+
+    def _observed_path(self, request: ReplayDataRequest) -> tuple[pd.DataFrame, str | None]:
+        cache_key = (request.source_symbol.upper(), str(request.market or "").upper())
+        cached = self._observed_cache.get(cache_key)
+        if cached is not None:
+            return cached[0].copy(), cached[1]
+        url = f"https://api.nasdaq.com/api/quote/{quote(request.source_symbol)}/chart"
+        try:
+            response = requests.get(
+                url,
+                params={"assetclass": "stocks"},
+                headers=provider_headers(),
+                timeout=12,
+            )
+            response.raise_for_status()
+            payload = response.json()
+        except Exception:
+            result = (pd.DataFrame(), "PROVIDER_UNAVAILABLE")
+            self._observed_cache[cache_key] = result
+            return result
+        chart = (payload.get("data") or {}).get("chart") or []
+        points = [row for row in chart if row.get("x") is not None and row.get("y") is not None]
+        if not points:
+            result = (pd.DataFrame(), "NO_INTRADAY_HISTORY")
+            self._observed_cache[cache_key] = result
+            return result
+        index = pd.to_datetime([row["x"] for row in points], unit="ms", utc=True)
+        values = pd.to_numeric([row["y"] for row in points], errors="coerce")
+        observed = pd.DataFrame({"Close": values}, index=index).dropna(subset=["Close"])
+        if observed.empty:
+            result = (pd.DataFrame(), "NO_INTRADAY_HISTORY")
+            self._observed_cache[cache_key] = result
+            return result
+        eastern = observed.index.tz_convert("America/New_York")
+        regular_session = (eastern.time >= clock_time(9, 30)) & (eastern.time <= clock_time(16, 0))
+        observed = observed.loc[regular_session]
+        if observed.empty:
+            result = (pd.DataFrame(), "MARKET_SESSION_UNAVAILABLE")
+            self._observed_cache[cache_key] = result
+            return result
+        observed.index = observed.index.tz_convert(None)
+        result = (observed, None)
+        self._observed_cache[cache_key] = result
+        return observed.copy(), None
+
+
 class YahooReplayDataProvider:
     name = "yahoo_chart"
     supported_timeframes = REPLAY_TIMEFRAMES
@@ -53,10 +140,11 @@ class YahooReplayDataProvider:
     def fetch(self, request: ReplayDataRequest) -> ProviderBars:
         if request.timeframe not in self.supported_timeframes:
             return ProviderBars(self.name, source_metadata=self.source_metadata, blockers=["UNSUPPORTED_TIMEFRAME"])
+        start, end = provider_request_window(request)
         url = f"https://query1.finance.yahoo.com/v8/finance/chart/{quote(request.source_symbol)}"
         params = {
-            "period1": int(_as_utc(request.start).timestamp()),
-            "period2": int(_as_utc(request.end).timestamp()) + 1,
+            "period1": int(_as_utc(start).timestamp()),
+            "period2": int(_as_utc(end).timestamp()) + 1,
             "interval": request.timeframe,
             "includePrePost": "false",
             "events": "div,splits",
@@ -100,11 +188,12 @@ class YFinanceReplayDataProvider:
     }
 
     def fetch(self, request: ReplayDataRequest) -> ProviderBars:
+        start, end = provider_request_window(request)
         try:
             frame = yf.download(
                 request.source_symbol,
-                start=request.start,
-                end=request.end,
+                start=start,
+                end=end,
                 interval=request.timeframe,
                 auto_adjust=True,
                 progress=False,
@@ -148,7 +237,7 @@ class DailyProviderReplayAdapter:
 
 
 def default_replay_providers(*, include_yfinance: bool = True) -> list[ReplayDataProvider]:
-    providers: list[ReplayDataProvider] = [YahooReplayDataProvider()]
+    providers: list[ReplayDataProvider] = [NasdaqChartReplayDataProvider(), YahooReplayDataProvider()]
     if include_yfinance:
         providers.append(YFinanceReplayDataProvider())
     providers.extend([DailyProviderReplayAdapter(StooqProvider()), DailyProviderReplayAdapter(NasdaqHistoricalProvider())])
@@ -175,6 +264,15 @@ def normalize_replay_frame(frame: pd.DataFrame) -> pd.DataFrame:
     return output.loc[~invalid]
 
 
+def provider_request_window(request: ReplayDataRequest) -> tuple[datetime, datetime]:
+    """Clamp intraday requests to the retention windows exposed by Yahoo sources."""
+
+    retention_days = INTRADAY_RETENTION_DAYS.get(request.timeframe)
+    if retention_days is None:
+        return request.start, request.end
+    return max(request.start, request.end - timedelta(days=retention_days)), request.end
+
+
 def _as_utc(value: datetime) -> datetime:
     if value.tzinfo is None:
         return value.replace(tzinfo=timezone.utc)

backend/app/services/intraday_opportunity.py

@@ -134,7 +134,10 @@ class BlumIntradayOpportunityEngine:
             return IntradayDecision(INTRADAY_BLOCKED, "SPREAD_TOO_WIDE", "Estimated spread is too large relative to the target.", costs=costs, **base)
 
         confidence = min(95.0, strategy.walk_forward_score)
-        edge_score = min(100.0, 50.0 + float(strategy.metrics.get("expectancy_r") or 0.0) * 100.0)
+        expectancy_r = strategy.metrics.get("expectancy_r")
+        if expectancy_r is None:
+            expectancy_r = strategy.metrics.get("net_expectancy_r")
+        edge_score = min(100.0, 50.0 + float(expectancy_r or 0.0) * 100.0)
         if confidence < strategy.minimum_confidence or edge_score < strategy.minimum_edge_score:
             return IntradayDecision(INTRADAY_BLOCKED, "QUANT_EDGE_BELOW_THRESHOLD", "Promoted strategy does not meet current confidence or Quant Edge threshold.", confidence=confidence, edge_score=edge_score, regime=regime, session=session, costs=costs, **base)
         size = self.sizer.size(

backend/app/services/intraday_paper_engine.py

@@ -88,6 +88,7 @@ class BlumIntradayPaperEngine:
         self.learning = IntradayPaperLearningService()
         self.no_trade_learning = IntradayNoTradeLearningService(evaluation_minutes=settings.intraday_no_trade_evaluation_minutes)
         self._desk_context: dict[int, tuple[str, str]] = {}
+        self._discovery_metadata: dict = {}
 
     def run_once(self, db: Session, *, trigger: str = "manual", assets: Iterable[Asset] | None = None) -> dict:
         if not settings.intraday_paper_enabled:
@@ -165,6 +166,7 @@ class BlumIntradayPaperEngine:
             run.summary_json = {
                 "decisions": decisions[:50],
                 "lifecycle": lifecycle,
+                "asset_discovery": self._discovery_metadata,
                 "policy": "Strict 1d/15m/5m/1m paper-forward evidence; no timeframe fallback or broker execution.",
             }
             self.paper.refresh_live_game_counts(db, game)
@@ -227,9 +229,32 @@ class BlumIntradayPaperEngine:
                 seen.add(asset.id)
                 self._desk_context[asset.id] = (agent.agent_name, agent.benchmark)
                 ranked.append(asset)
-                if len(ranked) >= self.max_assets:
-                    return ranked
-        return ranked
+
+        ranked.sort(key=lambda row: (row.ticker.upper(), row.id))
+        previous = db.scalar(
+            select(IntradayPaperRun)
+            .where(IntradayPaperRun.status != "RUNNING")
+            .order_by(desc(IntradayPaperRun.id))
+            .limit(1)
+        )
+        previous_discovery = (previous.summary_json or {}).get("asset_discovery") if previous else {}
+        cursor_before = (previous_discovery or {}).get("cursor_after")
+        start_index = 0
+        if cursor_before is not None:
+            for index, asset in enumerate(ranked):
+                if asset.id == cursor_before:
+                    start_index = (index + 1) % max(1, len(ranked))
+                    break
+        rotated = ranked[start_index:] + ranked[:start_index]
+        selected = rotated[: self.max_assets]
+        self._discovery_metadata = {
+            "policy": "round_robin_stored_market_universe",
+            "universe_size": len(ranked),
+            "cursor_before": cursor_before,
+            "cursor_after": selected[-1].id if selected else cursor_before,
+            "selected_tickers": [asset.ticker for asset in selected],
+        }
+        return selected
 
     def _portfolio_state(self, db: Session, game: LiveForwardPaperGame) -> IntradayPortfolioState:
         rows = db.scalars(
@@ -822,6 +847,7 @@ def intraday_snapshot_summary(db: Session, *, now: datetime | None = None) -> di
     no_trade_pending = int(
         db.scalar(select(func.count(IntradayNoTradeDecision.id)).where(IntradayNoTradeDecision.status == "PENDING")) or 0
     )
+    strategy_registry = BlumPromotedStrategyRegistry().status(db)
     if not latest_run:
         inactivity_reason = "No intraday run has completed."
         next_action = "Wait for the bounded backend worker or invoke the explicit manual POST."
@@ -869,6 +895,7 @@ def intraday_snapshot_summary(db: Session, *, now: datetime | None = None) -> di
         "desks": dict(Counter(row.desk or "UNKNOWN" for row in rows)),
         "setups": dict(Counter(row.setup_type for row in rows)),
         "latest_blockers": latest_run.data_blockers if latest_run else ["No intraday run has completed."],
+        "strategy_registry": strategy_registry,
         "evidence_type": PAPER_FORWARD_INTRADAY,
         "policy": "Read-only forward paper evidence. No trade or recalculation is triggered by this snapshot.",
     }

backend/app/services/live_forward_paper_trading.py

@@ -547,11 +547,24 @@ class LiveForwardPaperTradingService(_TradingLabLiveForwardService):
         waiting: list[dict] = []
         data_blocked: list[dict] = []
         skipped: list[dict] = []
+        open_tickers = {
+            str(ticker).upper()
+            for ticker in db.scalars(
+                select(LiveForwardPaperTrade.ticker).where(
+                    LiveForwardPaperTrade.game_id == live_game.id,
+                    LiveForwardPaperTrade.status.in_(["ORDER_SUBMITTED", "PARTIALLY_FILLED", "OPEN"]),
+                )
+            ).all()
+            if ticker
+        }
 
         for trade in rows:
             if int(live_game.open_positions or 0) >= settings.live_trading_game_max_open_positions:
                 skipped.append({"trade_id": trade.id, "ticker": trade.ticker, "reason": "max_open_positions_reached"})
                 break
+            if trade.ticker.upper() in open_tickers:
+                skipped.append({"trade_id": trade.id, "ticker": trade.ticker, "reason": "ticker_position_already_open"})
+                continue
 
             if classification_from_trade(trade) != TRADE_CANDIDATE:
                 waiting.append({
@@ -581,6 +594,7 @@ class LiveForwardPaperTradingService(_TradingLabLiveForwardService):
                 continue
 
             opened.append(self.open_candidate_trade(db, live_game, trade, latest_date, latest_price, condition))
+            open_tickers.add(trade.ticker.upper())
 
         return {"opened": opened, "waiting": waiting, "data_blocked": data_blocked, "skipped": skipped}
 

backend/app/services/promoted_strategy_registry.py

@@ -57,8 +57,13 @@ class BlumPromotedStrategyRegistry:
             return None
         sample_size = int(row.sample_size or 0)
         stability = number(metrics.get("stability_score"), 0.0)
-        expectancy = number(metrics.get("expectancy_r"), 0.0)
+        expectancy = number(
+            metrics.get("expectancy_r"),
+            number(metrics.get("net_expectancy_r"), 0.0),
+        )
         benchmark_excess = number(metrics.get("benchmark_excess"), 0.0)
+        deflated_sharpe = number(metrics.get("deflated_sharpe_probability"), 0.0) * 100.0
+        derived_walk_forward_score = min(stability, deflated_sharpe) if deflated_sharpe > 0 else stability
         timeframe_stack = tuple(metrics.get("timeframe_stack") or REQUIRED_INTRADAY_TIMEFRAMES)
         if sample_size < settings.replay_min_promotion_samples:
             return None
@@ -83,7 +88,7 @@ class BlumPromotedStrategyRegistry:
             minimum_confidence=number(metrics.get("minimum_confidence"), 60.0),
             minimum_edge_score=number(metrics.get("minimum_edge_score"), settings.blum_quant_edge_min_score),
             validated_trade_count=sample_size,
-            walk_forward_score=number(metrics.get("walk_forward_score"), 0.0),
+            walk_forward_score=number(metrics.get("walk_forward_score"), derived_walk_forward_score),
             expected_costs=dict(metrics.get("expected_costs") or {}),
             max_allowed_drawdown=abs(number(metrics.get("max_drawdown"), 0.0)),
             promotion_timestamp=row.created_at,

backend/app/services/replay_data.py

@@ -79,7 +79,6 @@ class MultiProviderReplayDataService:
         for provider in self.providers:
             if timeframe not in provider.supported_timeframes:
                 attempts.append({"provider": provider.name, "status": "blocked", "blockers": ["UNSUPPORTED_TIMEFRAME"]})
-                blockers.append("UNSUPPORTED_TIMEFRAME")
                 continue
             request = ReplayDataRequest(
                 source_symbol=asset.ticker,
@@ -112,7 +111,7 @@ class MultiProviderReplayDataService:
                     "blockers": list(result.blockers),
                 }
             )
-            blockers.extend(result.blockers)
+            blockers.extend(code for code in result.blockers if not code.startswith("UNSUPPORTED_"))
             if normalized.empty:
                 continue
             self._persist_bars(db, asset, timeframe, provider.name, normalized, result.source_metadata)
@@ -197,7 +196,9 @@ class MultiProviderReplayDataService:
                 )
             ).all()
         )
-        quality = _frame_quality(frame)
+        provider_quality = metadata.get("data_quality_score") if isinstance(metadata, dict) else None
+        quality = _number(provider_quality) if provider_quality is not None else _frame_quality(frame)
+        quality = max(0.0, min(100.0, float(quality or 0.0)))
         acquired_at = datetime.utcnow()
         for timestamp, (_, row) in zip(timestamps, frame.iterrows()):
             if timestamp in existing:

backend/tests/test_live_forward_paper_trading.py

@@ -681,6 +681,36 @@ def test_paper_forward_lifecycle_opens_candidate_when_trigger_is_met():
         assert any(event.event_type == "POSITION_OPENED" for event in events)
 
 
+def test_paper_forward_lifecycle_allows_only_one_open_position_per_ticker():
+    with setup_db() as db:
+        asset = seed_asset(db)
+        service = LiveForwardPaperTradingService()
+        first_payload = candidate(target_1=130.0, target_2=150.0)
+        second_payload = candidate(target_1=128.0, target_2=145.0)
+        second_payload["setup"] = {"setup_type": "pullback"}
+        second_payload["trade_plan"] = {
+            **second_payload["trade_plan"],
+            "entry_trigger": "pullback reclaim above support",
+        }
+        service.create_candidate(db, first_payload)
+        service.create_candidate(db, second_payload)
+        add_price(db, asset, 1, 101.0)
+
+        report = run_lifecycle(service, db)
+        open_rows = db.scalars(
+            select(PaperForwardTrade).where(
+                PaperForwardTrade.ticker == "NVDA",
+                PaperForwardTrade.status == "OPEN",
+            )
+        ).all()
+
+    assert len(open_rows) == 1
+    assert any(
+        row.get("reason") == "ticker_position_already_open"
+        for row in report["phases"]["open_eligible_trades"]["skipped"]
+    )
+
+
 def test_paper_forward_lifecycle_keeps_candidate_when_trigger_not_met():
     with setup_db() as db:
         asset = seed_asset(db)

backend/tests/test_live_intraday_paper_engine.py

@@ -19,6 +19,7 @@ from app.models import (
     LiveForwardPaperTradeEvent,
     PaperExecutionFill,
     PaperExecutionOrder,
+    PriceHistory,
     ReplayMarketBar,
     ReplayStrategyValidation,
     StrategyCandidateVariant,
@@ -198,6 +199,30 @@ def test_registry_rejects_unstable_negative_or_overfit_strategy():
     assert rows == []
 
 
+def test_registry_accepts_metrics_emitted_by_alpha_strategy_factory():
+    with setup_db() as db:
+        promoted = seed_validation(
+            db,
+            metrics={
+                "expectancy_r": None,
+                "walk_forward_score": None,
+                "net_expectancy_r": 0.25,
+                "deflated_sharpe_probability": 0.97,
+                "stability_score": 74.0,
+            },
+        )
+
+        rows = BlumPromotedStrategyRegistry().list_eligible(
+            db,
+            market="USA",
+            asset_class="Stock",
+        )
+
+    assert [row.validation_id for row in rows] == [promoted.id]
+    assert rows[0].metrics["net_expectancy_r"] == 0.25
+    assert rows[0].walk_forward_score >= 60.0
+
+
 def test_data_gateway_requires_all_four_timeframes_without_fallback():
     with setup_db() as db:
         asset = seed_asset(db)
@@ -285,6 +310,47 @@ def test_intraday_trade_evidence_constant_is_forward_only():
     assert PAPER_FORWARD_INTRADAY != "REPLAY_EVIDENCE"
 
 
+def test_intraday_discovery_rotates_across_the_stored_asset_universe():
+    with setup_db() as db:
+        for index in range(25):
+            asset = seed_asset(db, ticker=f"T{index:02d}", market="USA")
+            db.add(
+                PriceHistory(
+                    asset_id=asset.id,
+                    date=NOW.date(),
+                    open=100.0,
+                    high=101.0,
+                    low=99.0,
+                    close=100.0,
+                    volume=1_000_000,
+                    provider="fixture",
+                )
+            )
+        db.commit()
+        engine = BlumIntradayPaperEngine(
+            now_provider=lambda: NOW,
+            refresh_missing=False,
+            max_assets=10,
+        )
+
+        first = engine.run_once(db, trigger="test")
+        second = engine.run_once(db, trigger="test")
+
+    first_tickers = {
+        row["ticker"]
+        for row in first["blockers"]
+        if row.get("reason") == "NO_PROMOTED_INTRADAY_STRATEGY"
+    }
+    second_tickers = {
+        row["ticker"]
+        for row in second["blockers"]
+        if row.get("reason") == "NO_PROMOTED_INTRADAY_STRATEGY"
+    }
+    assert len(first_tickers) == 10
+    assert len(second_tickers) == 10
+    assert first_tickers.isdisjoint(second_tickers)
+
+
 def test_intraday_trade_persists_strategy_cost_and_lifecycle_metadata():
     with setup_db() as db:
         game = LiveForwardPaperGame(game_id="intraday-test", starting_capital=10_000, current_capital=10_000, cash=10_000)
@@ -518,6 +584,7 @@ def test_intraday_snapshot_is_read_only_and_reports_no_activity_truthfully():
     assert before == after == 0
     assert snapshot["status"] == "NO_INTRADAY_RUNS"
     assert snapshot["trades_opened_today"] == 0
+    assert snapshot["strategy_registry"]["status"] == "NO_PROMOTED_STRATEGIES"
 
 
 def test_intraday_run_reports_data_blocked_when_no_eligible_market_data_exists():

backend/tests/test_replay_data_engine.py

@@ -1,4 +1,4 @@
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 import pandas as pd
 import pytest
@@ -8,7 +8,13 @@ from sqlalchemy.orm import Session
 
 from app.core.database import Base
 from app.models import Asset, PriceHistory, ReplayMarketBar
-from app.providers.replay_data_provider import ProviderBars, ReplayDataRequest
+from app.providers.replay_data_provider import (
+    NasdaqChartReplayDataProvider,
+    ProviderBars,
+    ReplayDataRequest,
+    YahooReplayDataProvider,
+    YFinanceReplayDataProvider,
+)
 from app.services.replay_data import MultiProviderReplayDataService
 
 
@@ -119,6 +125,177 @@ def frame(start: datetime, periods: int, frequency: str) -> pd.DataFrame:
     )
 
 
+class FakeYahooResponse:
+    def raise_for_status(self) -> None:
+        return None
+
+    def json(self) -> dict:
+        return {
+            "chart": {
+                "result": [
+                    {
+                        "timestamp": [1_752_657_600],
+                        "indicators": {
+                            "quote": [
+                                {
+                                    "open": [100.0],
+                                    "high": [101.0],
+                                    "low": [99.0],
+                                    "close": [100.5],
+                                    "volume": [1_000_000],
+                                }
+                            ]
+                        },
+                    }
+                ],
+                "error": None,
+            }
+        }
+
+
+class FakeNasdaqResponse:
+    def raise_for_status(self) -> None:
+        return None
+
+    def json(self) -> dict:
+        start = datetime(2026, 7, 15, 13, 30, tzinfo=timezone.utc)
+        return {
+            "data": {
+                "symbol": "NVDA",
+                "chart": [
+                    {
+                        "x": int((start + timedelta(minutes=index)).timestamp() * 1000),
+                        "y": 170.0 + index / 10,
+                    }
+                    for index in range(31)
+                ],
+            }
+        }
+
+
+def test_yahoo_replay_clamps_one_minute_request_to_provider_retention(monkeypatch):
+    captured: dict = {}
+
+    def fake_get(url, *, params, headers, timeout):
+        captured.update(params)
+        return FakeYahooResponse()
+
+    monkeypatch.setattr("app.providers.replay_data_provider.requests.get", fake_get)
+    end = datetime(2026, 7, 16, 12, 0)
+    request = ReplayDataRequest("NVDA", "NVDA", "USA", "1m", end - timedelta(days=365), end)
+
+    YahooReplayDataProvider().fetch(request)
+
+    requested_start = datetime.utcfromtimestamp(captured["period1"])
+    assert requested_start == end - timedelta(days=7)
+
+
+def test_yfinance_replay_clamps_five_minute_request_to_provider_retention(monkeypatch):
+    captured: dict = {}
+
+    def fake_download(symbol, **kwargs):
+        captured.update(kwargs)
+        return pd.DataFrame()
+
+    monkeypatch.setattr("app.providers.replay_data_provider.yf.download", fake_download)
+    end = datetime(2026, 7, 16, 12, 0)
+    request = ReplayDataRequest("NVDA", "NVDA", "USA", "5m", end - timedelta(days=365), end)
+
+    YFinanceReplayDataProvider().fetch(request)
+
+    assert captured["start"] == end - timedelta(days=59)
+
+
+def test_nasdaq_chart_provider_resamples_observed_price_path(monkeypatch):
+    monkeypatch.setattr(
+        "app.providers.replay_data_provider.requests.get",
+        lambda *args, **kwargs: FakeNasdaqResponse(),
+    )
+    end = datetime(2026, 7, 16, 12, 0)
+
+    result = NasdaqChartReplayDataProvider().fetch(
+        ReplayDataRequest("NVDA", "NVDA", "USA", "15m", end - timedelta(days=10), end)
+    )
+
+    assert result.blockers == []
+    assert len(result.frame) == 3
+    assert result.frame.iloc[0]["Open"] == 170.0
+    assert result.frame.iloc[0]["Close"] == 171.4
+    assert result.source_metadata["bar_construction"] == "observed_last_sale_path"
+    assert result.source_metadata["data_quality_score"] == 55.0
+
+
+def test_nasdaq_chart_provider_never_returns_bars_after_requested_end(monkeypatch):
+    monkeypatch.setattr(
+        "app.providers.replay_data_provider.requests.get",
+        lambda *args, **kwargs: FakeNasdaqResponse(),
+    )
+    end = datetime(2026, 7, 14, 20, 0)
+
+    result = NasdaqChartReplayDataProvider().fetch(
+        ReplayDataRequest("NVDA", "NVDA", "USA", "1m", end - timedelta(days=2), end)
+    )
+
+    assert result.frame.empty
+    assert result.blockers == ["NO_INTRADAY_HISTORY"]
+
+
+def test_nasdaq_chart_provider_reuses_one_price_path_for_all_timeframes(monkeypatch):
+    request_count = 0
+
+    def fake_get(*args, **kwargs):
+        nonlocal request_count
+        request_count += 1
+        return FakeNasdaqResponse()
+
+    monkeypatch.setattr("app.providers.replay_data_provider.requests.get", fake_get)
+    provider = NasdaqChartReplayDataProvider()
+    end = datetime(2026, 7, 16, 12, 0)
+    for timeframe in ("1m", "5m", "15m"):
+        provider.fetch(ReplayDataRequest("NVDA", "NVDA", "USA", timeframe, end - timedelta(days=10), end))
+
+    assert request_count == 1
+
+
+def test_replay_persistence_honors_provider_quality_override():
+    start = datetime(2026, 7, 10, 8, 0)
+    provider = RecordingProvider(frame(start, periods=5, frequency="15min"))
+    provider.source_metadata = {"license": "test", "data_quality_score": 55.0}
+    with setup_db() as db:
+        asset = seed_asset(db)
+        MultiProviderReplayDataService([provider]).ensure_coverage(
+            db,
+            asset=asset,
+            timeframe="15m",
+            start=start,
+            end=start + timedelta(hours=1),
+        )
+        rows = db.scalars(select(ReplayMarketBar).where(ReplayMarketBar.asset_id == asset.id)).all()
+
+    assert rows
+    assert {row.data_quality_score for row in rows} == {55.0}
+
+
+def test_unsupported_provider_does_not_pollute_successful_fallback_blockers():
+    start = datetime(2026, 7, 10, 8, 0)
+    unsupported = RecordingProvider()
+    unsupported.supported_timeframes = frozenset({"1d"})
+    fallback = RecordingProvider(frame(start, periods=5, frequency="15min"))
+    fallback.name = "fallback"
+    with setup_db() as db:
+        asset = seed_asset(db)
+        result = MultiProviderReplayDataService([unsupported, fallback]).ensure_coverage(
+            db,
+            asset=asset,
+            timeframe="15m",
+            start=start,
+            end=start + timedelta(hours=1),
+        )
+
+    assert result.provider == "fallback"
+    assert "UNSUPPORTED_TIMEFRAME" not in result.blockers
+
+
 def seed_replay_bars(db: Session, asset: Asset, start: datetime, count: int, timeframe: str = "5m") -> None:
     for index in range(count):
         timestamp = start + timedelta(minutes=5 * index)