From 7d703ea5248eb8cedd937a06828610074aeb8d13 Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Mon, 3 Aug 2026 23:13:17 +0200 Subject: [PATCH] fix: refresh same-day OHLCV bars --- app/scheduler.py | 6 +-- app/services/ingestion_service.py | 7 +++- frontend/src/pages/TickerDetailPage.tsx | 45 +++++++++++++++++++-- tests/unit/test_ingestion_service.py | 53 ++++++++++++++++++++++++- 4 files changed, 102 insertions(+), 9 deletions(-) diff --git a/app/scheduler.py b/app/scheduler.py index 65ec390..e0e703f 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -518,7 +518,7 @@ async def collect_ohlcv( Uses AlpacaOHLCVProvider. Processes each ticker independently. On rate limit, records last successful ticker for resume. Start date is resolved by ingestion progress: - - existing ticker: resume from last_ingested_date + 1 + - existing ticker: overlap last_ingested_date so partial bars refresh - new ticker: backfill the configured history window ``full_backfill`` forces every ticker to re-fetch the full @@ -1496,8 +1496,8 @@ _DAILY_PIPELINE_STEPS = [ ("alerts", "dispatch_alerts_job"), ] -# Near-close (~15:30 ET Mon–Fri): refresh in-progress day-t bars (already how -# the intraday pipeline keeps the dashboard live), then the only daily +# Near-close (~15:30 ET Mon–Fri): refresh in-progress day-t bars (incremental +# ingestion overlaps the latest stored session), then the only daily # qualifying R:R scan, then Telegram immediately so manual fills can still hit # MOC cutoffs (~15:50/15:55). Under a 15-minute delayed SIP feed a 15:30 scan # may see ~15:15 prices — immaterial for a 12-1 momentum signal. diff --git a/app/services/ingestion_service.py b/app/services/ingestion_service.py index 444aa6b..07628d5 100644 --- a/app/services/ingestion_service.py +++ b/app/services/ingestion_service.py @@ -129,7 +129,12 @@ async def fetch_and_ingest( if bar_count < minimum_backfill_bars: start_date = backfill_start elif progress is not None: - start_date = progress.last_ingested_date + timedelta(days=1) + # Re-fetch the latest stored session so an in-progress daily bar can + # be overwritten as the market moves. Starting one day later makes + # every subsequent intraday, near-close, and manual refresh skip + # today's bar once the first partial snapshot has been stored. + # The price-store upsert keeps this one-session overlap idempotent. + start_date = progress.last_ingested_date else: start_date = backfill_start diff --git a/frontend/src/pages/TickerDetailPage.tsx b/frontend/src/pages/TickerDetailPage.tsx index 1bea549..68c90ce 100644 --- a/frontend/src/pages/TickerDetailPage.tsx +++ b/frontend/src/pages/TickerDetailPage.tsx @@ -64,10 +64,43 @@ function timeAgo(iso: string): string { return `${days}d ago`; } +function marketDate(date = new Date()): string { + const parts = new Intl.DateTimeFormat('en-US', { + timeZone: 'America/New_York', + year: 'numeric', + month: '2-digit', + day: '2-digit', + }).formatToParts(date); + const value = (type: Intl.DateTimeFormatPartTypes) => + parts.find((part) => part.type === type)?.value ?? ''; + return value('year') + '-' + value('month') + '-' + value('day'); +} + +function formatSessionDate(isoDate: string): string { + if (isoDate === marketDate()) 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. + const [year, month, day] = isoDate.split('-').map(Number); + if (!year || !month || !day) return isoDate; + return new Intl.DateTimeFormat(undefined, { + month: 'short', + day: 'numeric', + year: year === new Date().getFullYear() ? undefined : 'numeric', + timeZone: 'UTC', + }).format(new Date(Date.UTC(year, month - 1, day))); +} + +function formatOHLCVFreshness(sessionDate: string, updatedAt?: string | null): string { + const session = formatSessionDate(sessionDate); + return updatedAt ? session + ' · updated ' + timeAgo(updatedAt) : session; +} + interface DataStatusItem { label: string; available: boolean; timestamp?: string | null; + timestampLabel?: string | null; selector: FetchSelector; // what a refresh of this row fetches paid?: boolean; // provider call that may cost money/quota } @@ -100,7 +133,7 @@ function DataFreshnessBar({ }`} /> {item.label} {item.available && item.timestamp ? ( - {timeAgo(item.timestamp)} + {item.timestampLabel ?? timeAgo(item.timestamp)} ) : !item.available ? ( no data ) : null} @@ -171,10 +204,16 @@ export default function TickerDetailPage() { const dataStatus: DataStatusItem[] = useMemo(() => [ { label: 'OHLCV', - // Market age of the latest bar (session date), not DB insert time — - // created_at stays frozen when the provider returns no new sessions. + // Keep the market session date distinct from the last successful bar + // write; treating YYYY-MM-DD as an instant makes today's session look old. available: !!ohlcv.data && ohlcv.data.length > 0, timestamp: ohlcv.data?.[ohlcv.data.length - 1]?.date, + timestampLabel: ohlcv.data?.length + ? formatOHLCVFreshness( + ohlcv.data[ohlcv.data.length - 1].date, + ohlcv.data[ohlcv.data.length - 1].created_at, + ) + : null, selector: ['ohlcv'] as FetchSelector, paid: true, }, diff --git a/tests/unit/test_ingestion_service.py b/tests/unit/test_ingestion_service.py index 1efa1d6..58026a9 100644 --- a/tests/unit/test_ingestion_service.py +++ b/tests/unit/test_ingestion_service.py @@ -6,6 +6,8 @@ from datetime import date, timedelta import pytest +from app.models.ohlcv import OHLCVRecord +from app.models.settings import IngestionProgress from app.models.ticker import Ticker from app.providers.protocol import OHLCVData from app.services import ingestion_service as svc @@ -18,9 +20,12 @@ async def session(): yield s -async def _add_ticker(session, symbol: str) -> None: - session.add(Ticker(symbol=symbol)) +async def _add_ticker(session, symbol: str) -> Ticker: + ticker = Ticker(symbol=symbol) + session.add(ticker) await session.commit() + await session.refresh(ticker) + return ticker def _bars(symbol: str, n: int) -> list[OHLCVData]: @@ -51,6 +56,50 @@ async def test_happy_path_ingests_bars(session): assert result.records_ingested == 3 +async def test_incremental_fetch_overlaps_latest_session_and_updates_partial_bar(session): + """Once today exists, a live refresh must fetch and overwrite it again.""" + ticker = await _add_ticker(session, "LIVE") + today = date.today() + session.add_all([ + OHLCVRecord( + ticker_id=ticker.id, + date=today - timedelta(days=i), + open=100.0, + high=101.0, + low=99.0, + close=100.0, + volume=1000, + ) + for i in range(200) + ]) + session.add(IngestionProgress(ticker_id=ticker.id, last_ingested_date=today)) + await session.commit() + + provider = MockMarketDataProvider(ohlcv_data=[ + OHLCVData( + ticker="LIVE", + date=today, + open=100.0, + high=124.0, + low=99.0, + close=123.0, + volume=2000, + ) + ]) + result = await svc.fetch_and_ingest(session, provider, "LIVE") + + assert provider.calls == [{ + "ticker": "LIVE", + "start_date": today, + "end_date": today, + }] + assert result.status == "complete" + assert result.records_ingested == 1 + records = await svc.price_service.query_ohlcv(session, "LIVE", today, today) + assert records[0].close == 123.0 + assert records[0].volume == 2000 + + async def test_empty_fetch_with_existing_history_is_up_to_date(session): # Covered ticker, just no new bars in the window → complete, not no_data. await _add_ticker(session, "BBB")