From d1caac86b5ef70a41b6236b837a13583263845ff Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Tue, 4 Aug 2026 07:39:40 +0200 Subject: [PATCH] fix: preserve OHLCV stale detection --- app/scheduler.py | 9 +++++++- app/services/ingestion_service.py | 25 +++++++++++++++++++-- frontend/src/pages/TickerDetailPage.tsx | 5 +++-- tests/unit/test_ingestion_service.py | 29 +++++++++++++++++++++++-- 4 files changed, 61 insertions(+), 7 deletions(-) diff --git a/app/scheduler.py b/app/scheduler.py index e0e703f..7bc4092 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -512,6 +512,7 @@ async def collect_ohlcv( job_name: str = "data_collector", *, refetch_days: int = 0, + refresh_sr: bool = True, ) -> None: """Fetch latest daily OHLCV for all tracked tickers. @@ -580,6 +581,7 @@ async def collect_ohlcv( try: result = await ingestion_service.fetch_and_ingest( db, provider, symbol, start_date=backfill_start, end_date=end_date, + refresh_sr=refresh_sr, ) _last_successful[job_name] = symbol processed += 1 @@ -619,6 +621,11 @@ async def collect_ohlcv( _runtime_finish(job_name, "error", processed=processed, total=total, message=str(exc)) +async def collect_ohlcv_for_scan() -> None: + """Near-close fetch; the scanner immediately rebuilds S/R per ticker.""" + await collect_ohlcv(refresh_sr=False) + + async def backfill_ohlcv() -> None: """Deep historical backfill: re-fetch the full ``settings.ohlcv_history_days`` window for every ticker, ignoring incremental resume. @@ -1484,7 +1491,7 @@ async def sync_ticker_universe() -> None: _FINAL_REFETCH_DAYS = 5 _DAILY_PIPELINE_STEPS = [ - ("data_collector", "collect_ohlcv"), + ("data_collector", "collect_ohlcv_for_scan"), ("benchmark_collector", "collect_benchmark"), ("sentiment_collector", "collect_sentiment"), ("market_regime", "compute_market_regime"), diff --git a/app/services/ingestion_service.py b/app/services/ingestion_service.py index 07628d5..3610cf1 100644 --- a/app/services/ingestion_service.py +++ b/app/services/ingestion_service.py @@ -100,6 +100,8 @@ async def fetch_and_ingest( symbol: str, start_date: date | None = None, end_date: date | None = None, + *, + refresh_sr: bool = True, ) -> IngestionResult: """Fetch OHLCV data from provider and upsert into Price Store. @@ -244,7 +246,7 @@ async def fetch_and_ingest( ticker.symbol, ingested_count, ) - if ingested_count > 0: + if ingested_count > 0 and refresh_sr: await _refresh_structural_sr(db, ticker.symbol) return IngestionResult( symbol=ticker.symbol, @@ -254,9 +256,28 @@ async def fetch_and_ingest( message=f"Rate limited. Ingested {ingested_count} records. Resume available.", ) - if ingested_count > 0: + if ingested_count > 0 and refresh_sr: await _refresh_structural_sr(db, ticker.symbol) + # Incremental fetches deliberately overlap the latest stored session so an + # in-progress bar can be updated. A halted/delisted symbol can therefore + # return one old bar forever; non-empty no longer means fresh. Judge stale + # state from the newest stored session after the upserts instead. + latest = await _get_latest_ohlcv_date(db, ticker.id) + gap_days = (end_date - latest).days if latest is not None else None + if gap_days is not None and gap_days > _STALE_OHLCV_GAP_DAYS: + return IngestionResult( + symbol=ticker.symbol, + records_ingested=ingested_count, + last_date=latest, + status="stale", + message=( + f"No new bars since {latest.isoformat()} ({gap_days}d gap). " + "The symbol may be halted, delisted, or renamed under a new ticker — " + "check the listing and add/fetch the current symbol if it changed." + ), + ) + return IngestionResult( symbol=ticker.symbol, records_ingested=ingested_count, diff --git a/frontend/src/pages/TickerDetailPage.tsx b/frontend/src/pages/TickerDetailPage.tsx index 68c90ce..1b799e0 100644 --- a/frontend/src/pages/TickerDetailPage.tsx +++ b/frontend/src/pages/TickerDetailPage.tsx @@ -77,7 +77,8 @@ function marketDate(date = new Date()): string { } function formatSessionDate(isoDate: string): string { - if (isoDate === marketDate()) return 'Today'; + const currentMarketDate = marketDate(); + if (isoDate === currentMarketDate) return 'Today'; // Parse date-only market sessions explicitly. Parsing YYYY-MM-DD directly as // a Date means midnight UTC and makes today's bar look many hours old. @@ -86,7 +87,7 @@ function formatSessionDate(isoDate: string): string { return new Intl.DateTimeFormat(undefined, { month: 'short', day: 'numeric', - year: year === new Date().getFullYear() ? undefined : 'numeric', + year: year === Number(currentMarketDate.slice(0, 4)) ? undefined : 'numeric', timeZone: 'UTC', }).format(new Date(Date.UTC(year, month - 1, day))); } diff --git a/tests/unit/test_ingestion_service.py b/tests/unit/test_ingestion_service.py index 58026a9..8dd478f 100644 --- a/tests/unit/test_ingestion_service.py +++ b/tests/unit/test_ingestion_service.py @@ -130,9 +130,34 @@ async def test_empty_fetch_with_stale_history_reports_stale(session): ] await svc.fetch_and_ingest(session, MockMarketDataProvider(ohlcv_data=old), "SATS") - result = await svc.fetch_and_ingest(session, MockMarketDataProvider(ohlcv_data=[]), "SATS") + # Incremental overlap means Alpaca can keep returning the final historical + # bar. That is still stale: the latest session did not advance. + result = await svc.fetch_and_ingest( + session, + MockMarketDataProvider(ohlcv_data=[old[-1]]), + "SATS", + ) assert result.status == "stale" - assert result.records_ingested == 0 + assert result.records_ingested == 1 assert result.last_date is not None assert "renamed" in (result.message or "").lower() or "halted" in (result.message or "").lower() + + +async def test_ingest_can_skip_sr_refresh_when_scanner_follows(session, monkeypatch): + await _add_ticker(session, "SCAN") + calls: list[str] = [] + + async def fake_refresh(db, symbol): + calls.append(symbol) + + monkeypatch.setattr(svc, "_refresh_structural_sr", fake_refresh) + result = await svc.fetch_and_ingest( + session, + MockMarketDataProvider(ohlcv_data=_bars("SCAN", 3)), + "SCAN", + refresh_sr=False, + ) + + assert result.status == "complete" + assert calls == []