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")