fix: preserve OHLCV stale detection
This commit is contained in:
+8
-1
@@ -512,6 +512,7 @@ async def collect_ohlcv(
|
|||||||
job_name: str = "data_collector",
|
job_name: str = "data_collector",
|
||||||
*,
|
*,
|
||||||
refetch_days: int = 0,
|
refetch_days: int = 0,
|
||||||
|
refresh_sr: bool = True,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Fetch latest daily OHLCV for all tracked tickers.
|
"""Fetch latest daily OHLCV for all tracked tickers.
|
||||||
|
|
||||||
@@ -580,6 +581,7 @@ async def collect_ohlcv(
|
|||||||
try:
|
try:
|
||||||
result = await ingestion_service.fetch_and_ingest(
|
result = await ingestion_service.fetch_and_ingest(
|
||||||
db, provider, symbol, start_date=backfill_start, end_date=end_date,
|
db, provider, symbol, start_date=backfill_start, end_date=end_date,
|
||||||
|
refresh_sr=refresh_sr,
|
||||||
)
|
)
|
||||||
_last_successful[job_name] = symbol
|
_last_successful[job_name] = symbol
|
||||||
processed += 1
|
processed += 1
|
||||||
@@ -619,6 +621,11 @@ async def collect_ohlcv(
|
|||||||
_runtime_finish(job_name, "error", processed=processed, total=total, message=str(exc))
|
_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:
|
async def backfill_ohlcv() -> None:
|
||||||
"""Deep historical backfill: re-fetch the full ``settings.ohlcv_history_days``
|
"""Deep historical backfill: re-fetch the full ``settings.ohlcv_history_days``
|
||||||
window for every ticker, ignoring incremental resume.
|
window for every ticker, ignoring incremental resume.
|
||||||
@@ -1484,7 +1491,7 @@ async def sync_ticker_universe() -> None:
|
|||||||
_FINAL_REFETCH_DAYS = 5
|
_FINAL_REFETCH_DAYS = 5
|
||||||
|
|
||||||
_DAILY_PIPELINE_STEPS = [
|
_DAILY_PIPELINE_STEPS = [
|
||||||
("data_collector", "collect_ohlcv"),
|
("data_collector", "collect_ohlcv_for_scan"),
|
||||||
("benchmark_collector", "collect_benchmark"),
|
("benchmark_collector", "collect_benchmark"),
|
||||||
("sentiment_collector", "collect_sentiment"),
|
("sentiment_collector", "collect_sentiment"),
|
||||||
("market_regime", "compute_market_regime"),
|
("market_regime", "compute_market_regime"),
|
||||||
|
|||||||
@@ -100,6 +100,8 @@ async def fetch_and_ingest(
|
|||||||
symbol: str,
|
symbol: str,
|
||||||
start_date: date | None = None,
|
start_date: date | None = None,
|
||||||
end_date: date | None = None,
|
end_date: date | None = None,
|
||||||
|
*,
|
||||||
|
refresh_sr: bool = True,
|
||||||
) -> IngestionResult:
|
) -> IngestionResult:
|
||||||
"""Fetch OHLCV data from provider and upsert into Price Store.
|
"""Fetch OHLCV data from provider and upsert into Price Store.
|
||||||
|
|
||||||
@@ -244,7 +246,7 @@ async def fetch_and_ingest(
|
|||||||
ticker.symbol,
|
ticker.symbol,
|
||||||
ingested_count,
|
ingested_count,
|
||||||
)
|
)
|
||||||
if ingested_count > 0:
|
if ingested_count > 0 and refresh_sr:
|
||||||
await _refresh_structural_sr(db, ticker.symbol)
|
await _refresh_structural_sr(db, ticker.symbol)
|
||||||
return IngestionResult(
|
return IngestionResult(
|
||||||
symbol=ticker.symbol,
|
symbol=ticker.symbol,
|
||||||
@@ -254,9 +256,28 @@ async def fetch_and_ingest(
|
|||||||
message=f"Rate limited. Ingested {ingested_count} records. Resume available.",
|
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)
|
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(
|
return IngestionResult(
|
||||||
symbol=ticker.symbol,
|
symbol=ticker.symbol,
|
||||||
records_ingested=ingested_count,
|
records_ingested=ingested_count,
|
||||||
|
|||||||
@@ -77,7 +77,8 @@ function marketDate(date = new Date()): string {
|
|||||||
}
|
}
|
||||||
|
|
||||||
function formatSessionDate(isoDate: string): 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
|
// 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.
|
// 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, {
|
return new Intl.DateTimeFormat(undefined, {
|
||||||
month: 'short',
|
month: 'short',
|
||||||
day: 'numeric',
|
day: 'numeric',
|
||||||
year: year === new Date().getFullYear() ? undefined : 'numeric',
|
year: year === Number(currentMarketDate.slice(0, 4)) ? undefined : 'numeric',
|
||||||
timeZone: 'UTC',
|
timeZone: 'UTC',
|
||||||
}).format(new Date(Date.UTC(year, month - 1, day)));
|
}).format(new Date(Date.UTC(year, month - 1, day)));
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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")
|
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.status == "stale"
|
||||||
assert result.records_ingested == 0
|
assert result.records_ingested == 1
|
||||||
assert result.last_date is not None
|
assert result.last_date is not None
|
||||||
assert "renamed" in (result.message or "").lower() or "halted" in (result.message or "").lower()
|
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 == []
|
||||||
|
|||||||
Reference in New Issue
Block a user