From eab5bc552715a15f63de68f49ca25a58cb8a3f4e Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Mon, 24 Aug 2026 20:27:50 +0530 Subject: [PATCH 1/6] fix(DATA-003): stop the scan re-dirtying the cache it just repaired Nine symbols were quarantined with DUPLICATE_DATE on scan run_id=4, 46 seconds after the DATA-002 repair had cleaned them. The repair was not at fault: it ran, worked, and was overwritten by the scan itself. Aligning the DB (UTC) with the app log (IST) shows the parquet files being rewritten at 19:11:57-19:15:24, between the repair finishing at 19:11:11 and the failures at 19:17:50. Two defects, one causing the other. Defect A - vendor duplicates were persisted verbatim. normalize_daily_payload sorted but never de-duplicated, and of the six cache-write sites in the loader only the incremental merge deduped first. DhanHQ repeats bars: AEGISLOG carries two byte-identical rows for 2024-06-05. Fixed at the vendor boundary with drop_duplicates() over all six columns, so no write path can persist a redundant bar and every consumer benefits, including frames handed straight to screeners. Deliberately narrow: only rows identical in EVERY column are dropped. Bars sharing a date but differing in any value - including volume alone, which is a partial-vs-final bar - survive to be reported, because choosing between them would fabricate a price series that never existed. A guard test asserts a conflicting bar still reaches disk so DATA-001 quarantines it. Defect B - the cache-hit test could not be satisfied. get_daily_history required last_date >= requested_end, and scans request "through today" while the vendor's newest published bar is Friday's. Zero of 577 symbols qualified, so every scan re-downloaded the whole universe, which is what fired Defect A across the cache and made the scan take six minutes. The end comparison now tolerates STALE_LATEST_TOLERANCE_DAYS, the constant DATA-001 already defines as how far the newest candle may trail today before that is suspicious; a frame inside it still raises STALE_LATEST_CANDLE, so nothing is hidden. The start comparison stays strict, preserving the original guard against running a long-lookback screener on an interrupted prefetch's partial file. Verified against a copy of the real 577-file cache: fatal symbols 18 -> 5, 1,264 rows removed, and a scan-shaped read of the repaired symbols is now a cache hit that leaves the files untouched (previously it re-downloaded and re-dirtied them). The remaining five carry price-level conflicts from the vendor and stay quarantined by design. Co-Authored-By: Claude Opus 5 --- backend/daily_data_loader.py | 44 +++++-- backend/dhan_client.py | 14 +++ tests/test_candle_cache_write_paths.py | 158 +++++++++++++++++++++++++ tests/test_daily_data_loader.py | 76 ++++++++++++ tests/test_dhan_client.py | 77 ++++++++++++ 5 files changed, 358 insertions(+), 11 deletions(-) create mode 100644 tests/test_candle_cache_write_paths.py diff --git a/backend/daily_data_loader.py b/backend/daily_data_loader.py index 2a7977c..1b44a97 100644 --- a/backend/daily_data_loader.py +++ b/backend/daily_data_loader.py @@ -27,6 +27,7 @@ dhan_request_delay_seconds, ) from backend.data_quality import CandleQualityReport, validate_candles +from backend.data_quality.candles import STALE_LATEST_TOLERANCE_DAYS from backend.dhan_client import DhanDataClient, DhanRateLimitError from backend.observability import ( EVENT_CANDLE_DATA_QUALITY_FAILED, @@ -103,6 +104,35 @@ def safe_file_stem(value: object) -> str: return cleaned +def _cache_covers_range( + first_date: date | None, + last_date: date | None, + requested_start: date, + requested_end: date, +) -> bool: + """Return True when a cached range is good enough to answer a request. + + The **start** is compared strictly: a parquet missing early history (common + after an interrupted prefetch) must never silently run a long-lookback + screener on too little data. + + The **end** is compared with ``STALE_LATEST_TOLERANCE_DAYS`` of slack, because + callers routinely ask for data "through today" while the newest bar the vendor + has published is Friday's. Demanding an exact match made every symbol a cache + miss outside market hours, so each scan re-downloaded the entire universe and + overwrote the cache with raw vendor data (DATA-003). + + Reusing DATA-001's constant is deliberate: it is already the app's definition + of "how far the newest candle may trail today before that is suspicious", and + a frame inside it still raises ``STALE_LATEST_CANDLE`` as a warning, so nothing + is hidden by treating it as current enough to serve. + """ + if first_date is None or last_date is None: + return False + tolerated_end = requested_end - timedelta(days=STALE_LATEST_TOLERANCE_DAYS) + return first_date <= requested_start and last_date >= tolerated_end + + def _date_bounds(candles: pd.DataFrame) -> tuple[date | None, date | None]: """Return the first/last valid candle dates in a cached frame. @@ -381,23 +411,15 @@ def get_daily_history( first_date, last_date = _date_bounds(cached) requested_start = _coerce_date(start_date) requested_end = _coerce_date(end_date) - if ( - first_date is not None - and last_date is not None - and first_date <= requested_start - and last_date >= requested_end - ): + if _cache_covers_range(first_date, last_date, requested_start, requested_end): if cached is None: # The footer is only an advisory index. The file can be # replaced after the metadata read, and a valid footer # does not prove every data page is readable. cached = pd.read_parquet(path) actual_first, actual_last = _date_bounds(cached) - if ( - actual_first is not None - and actual_last is not None - and actual_first <= requested_start - and actual_last >= requested_end + if _cache_covers_range( + actual_first, actual_last, requested_start, requested_end ): return self._slice_to_range(cached, start_date, end_date), True diff --git a/backend/dhan_client.py b/backend/dhan_client.py index f780ce4..a53c5f4 100644 --- a/backend/dhan_client.py +++ b/backend/dhan_client.py @@ -120,6 +120,20 @@ def normalize_daily_payload(data: Any) -> pd.DataFrame: out["timestamp"] = timestamps.dt.tz_convert("Asia/Kolkata").dt.tz_localize(None) out = out.drop(columns=["timestamp_raw"]).dropna(subset=["timestamp", "open", "high", "low", "close"]) out = out[["timestamp", "open", "high", "low", "close", "volume"]] + # DATA-003: drop bars the vendor repeated verbatim. DhanHQ occasionally returns + # the same candle twice (observed live for AEGISLOG on 2024-06-05, two + # byte-identical rows), and every cache-write path used to persist that copy — + # which then failed DATA-001's DUPLICATE_DATE check and dropped the symbol from + # every scan. + # + # ``drop_duplicates()`` here compares ALL six columns, so it only removes rows + # that are identical in every respect. Two identical bars for one day cannot + # both be real observations, so removing one loses nothing and costs no trading + # day. Bars that share a date but differ in ANY value — including volume alone, + # which is a partial-vs-final bar — deliberately survive: choosing between them + # would fabricate a price series that never existed. Those still surface as + # DUPLICATE_DATE for the DATA-002 repair to resolve against the vendor. + out = out.drop_duplicates() return out.sort_values("timestamp").reset_index(drop=True) diff --git a/tests/test_candle_cache_write_paths.py b/tests/test_candle_cache_write_paths.py new file mode 100644 index 0000000..4daf02a --- /dev/null +++ b/tests/test_candle_cache_write_paths.py @@ -0,0 +1,158 @@ +"""Guard: no loader write path may persist a vendor's duplicate bar (DATA-003). + +Beginner note: +DhanHQ sometimes repeats a candle verbatim in its response. Before this guard, the +loader had six places that wrote a frame to the cache and only *one* of them +de-duplicated first, so a redundant bar reached disk, failed DATA-001's +DUPLICATE_DATE check, and silently dropped the symbol from every scan. + +That is exactly what happened live: the DATA-002 repair cleaned the cache during +the prefetch, and the very next scan's cache-miss re-download put the duplicates +straight back. This file locks the invariant at every entry point rather than +trusting each call site to remember. +""" + +from __future__ import annotations + +from datetime import date, timedelta +from pathlib import Path +from typing import cast + +import pandas as pd + +from backend.daily_data_loader import DailyDataLoader +from backend.data_quality.candles import validate_candles +from backend.dhan_client import DhanDataClient, normalize_daily_response + +TODAY = date(2026, 8, 24) +ROW = {"symbol": "DEMO", "security_id": "1"} + + +class DuplicatingClient: + """A client whose vendor response repeats one bar verbatim, as Dhan does.""" + + def __init__(self) -> None: + self.calls = 0 + + def fetch_daily_candles(self, **_kwargs) -> pd.DataFrame: + self.calls += 1 + # Routed through the real normalizer so the test exercises the same + # boundary production uses, not a hand-built clean frame. + return normalize_daily_response( + { + "status": "success", + "data": [ + { + "timestamp": (TODAY - timedelta(days=offset)).isoformat(), + "open": 100.0, + "high": 105.0, + "low": 99.0, + "close": 104.0, + "volume": 1_000.0, + } + # `4` appears twice: the repeated bar. + for offset in (8, 7, 6, 5, 4, 4, 3, 2, 1, 0) + ], + } + ) + + +def _assert_cache_is_clean(loader: DailyDataLoader) -> pd.DataFrame: + """The cached parquet must not carry a DUPLICATE_DATE finding.""" + stored = pd.read_parquet(loader.cache_path(ROW["symbol"], ROW["security_id"])) + report = validate_candles(stored, symbol="DEMO", expected_latest_date=TODAY) + assert "DUPLICATE_DATE" not in {finding.code for finding in report.findings} + return stored + + +def test_get_daily_history_cache_miss_writes_no_duplicates(tmp_path: Path): + """The path that re-dirtied the cache after every repair.""" + loader = DailyDataLoader( + cast(DhanDataClient, DuplicatingClient()), + cache_dir=tmp_path, + request_delay_seconds=0.0, + ) + + loader.get_daily_history(ROW, start_date=TODAY - timedelta(days=8), end_date=TODAY) + + _assert_cache_is_clean(loader) + + +def test_ensure_daily_history_fresh_download_writes_no_duplicates(tmp_path: Path): + """First-ever download for a symbol.""" + loader = DailyDataLoader( + cast(DhanDataClient, DuplicatingClient()), + cache_dir=tmp_path, + request_delay_seconds=0.0, + ) + + _frame, status = loader.ensure_daily_history(ROW, years_back=1, today=TODAY) + + assert status == "fresh_download" + _assert_cache_is_clean(loader) + + +def test_ensure_daily_history_backfill_writes_no_duplicates(tmp_path: Path): + """A cache current at the back but missing early history is refetched whole.""" + loader = DailyDataLoader( + cast(DhanDataClient, DuplicatingClient()), + cache_dir=tmp_path, + request_delay_seconds=0.0, + ) + # Recent-only cache, so the backfill branch runs. + pd.DataFrame( + { + "timestamp": pd.to_datetime([TODAY - timedelta(days=1), TODAY]), + "open": [100.0, 100.0], + "high": [105.0, 105.0], + "low": [99.0, 99.0], + "close": [104.0, 104.0], + "volume": [1_000.0, 1_000.0], + } + ).to_parquet(loader.cache_path(ROW["symbol"], ROW["security_id"]), index=False) + + _frame, status = loader.ensure_daily_history(ROW, years_back=5, today=TODAY) + + assert status == "backfilled" + _assert_cache_is_clean(loader) + + +def test_a_conflicting_bar_still_reaches_the_cache_to_be_reported(tmp_path: Path): + """The guard must not become a silent price-picker. + + Two bars sharing a date but disagreeing on value are a real vendor conflict. + They must survive to disk so DATA-001 quarantines the symbol and the DATA-002 + repair resolves them, rather than one being quietly discarded here. + """ + + class ConflictingClient: + def fetch_daily_candles(self, **_kwargs) -> pd.DataFrame: + return normalize_daily_response( + { + "status": "success", + "data": [ + { + "timestamp": TODAY.isoformat(), + "open": 100.0, "high": 105.0, "low": 99.0, + "close": 104.0, "volume": 1_000.0, + }, + { + "timestamp": TODAY.isoformat(), + "open": 12.0, "high": 13.0, "low": 11.0, + "close": 12.5, "volume": 2_000.0, + }, + ], + } + ) + + loader = DailyDataLoader( + cast(DhanDataClient, ConflictingClient()), + cache_dir=tmp_path, + request_delay_seconds=0.0, + ) + + loader.get_daily_history(ROW, start_date=TODAY - timedelta(days=1), end_date=TODAY) + + stored = pd.read_parquet(loader.cache_path(ROW["symbol"], ROW["security_id"])) + report = validate_candles(stored, symbol="DEMO", expected_latest_date=TODAY) + assert "DUPLICATE_DATE" in {finding.code for finding in report.findings} diff --git a/tests/test_daily_data_loader.py b/tests/test_daily_data_loader.py index 3b8e127..d26bba5 100644 --- a/tests/test_daily_data_loader.py +++ b/tests/test_daily_data_loader.py @@ -1127,3 +1127,79 @@ def test_fetch_workers_setting_clamps_and_defaults(monkeypatch, tmp_path): # The loader clamps explicit constructor values the same way. loader = DailyDataLoader(None, cache_dir=tmp_path, fetch_workers=99) assert loader.fetch_workers == 8 + + +# --------------------------------------------------------------------------- +# DATA-003: a cache the vendor cannot improve on is still a hit +# --------------------------------------------------------------------------- + + +def _covering_cache(loader, *, last_date, first_date): + """Write a cache spanning [first_date, last_date] for the standard instrument.""" + frame = pd.DataFrame( + { + "timestamp": pd.to_datetime([first_date, last_date]), + "open": [100.0, 101.0], + "high": [105.0, 106.0], + "low": [99.0, 100.0], + "close": [104.0, 105.0], + "volume": [1_000.0, 1_100.0], + } + ) + frame.to_parquet(loader.cache_path("DEMO", "1"), index=False) + return frame + + +def test_cache_short_of_today_by_a_weekend_is_still_a_hit(tmp_path): + """The live failure: a scan asks through today, the vendor's newest bar is Friday. + + Requiring last_date >= today made every symbol a miss, so each scan + re-downloaded the whole universe and overwrote the cache with raw vendor data. + """ + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + today = date(2026, 8, 24) # Monday; newest published bar is Friday the 21st + _covering_cache(loader, first_date=date(2016, 8, 1), last_date=date(2026, 8, 21)) + + _frame, from_cache = loader.get_daily_history( + {"symbol": "DEMO", "security_id": "1"}, + start_date=date(2016, 8, 24), + end_date=today, + ) + + assert from_cache is True + assert client.calls == 0 # nothing re-downloaded, nothing overwritten + + +def test_cache_far_behind_the_requested_end_is_still_a_miss(tmp_path): + """Tolerance is for unpublished bars, not for genuinely abandoned history.""" + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _covering_cache(loader, first_date=date(2016, 8, 1), last_date=date(2026, 7, 1)) + + _frame, from_cache = loader.get_daily_history( + {"symbol": "DEMO", "security_id": "1"}, + start_date=date(2016, 8, 24), + end_date=date(2026, 8, 24), + ) + + assert from_cache is False + + +def test_cache_missing_early_history_is_a_miss_however_current_it_is(tmp_path): + """The START comparison stays strict. + + A partial parquet from an interrupted prefetch must never silently run a + long-lookback screener on too little history. + """ + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _covering_cache(loader, first_date=date(2025, 1, 2), last_date=date(2026, 8, 24)) + + _frame, from_cache = loader.get_daily_history( + {"symbol": "DEMO", "security_id": "1"}, + start_date=date(2016, 8, 24), + end_date=date(2026, 8, 24), + ) + + assert from_cache is False diff --git a/tests/test_dhan_client.py b/tests/test_dhan_client.py index 5b23f59..b0f5968 100644 --- a/tests/test_dhan_client.py +++ b/tests/test_dhan_client.py @@ -3,6 +3,7 @@ from __future__ import annotations import concurrent.futures +import datetime as dt import sys import threading import time @@ -206,3 +207,79 @@ def historical_daily_data(self, **kwargs): assert client.dhan is raw_client assert all(not frame.empty for frame in frames) assert raw_client.max_active_calls == 1 + + +# --------------------------------------------------------------------------- +# DATA-003: vendor duplicates must not reach the cache +# --------------------------------------------------------------------------- + + +def test_exact_duplicate_bars_are_collapsed(): + """DhanHQ sometimes repeats a bar verbatim; one copy carries all the info. + + Observed live for AEGISLOG on 2024-06-05 — two byte-identical rows. Persisting + both makes the symbol fail DATA-001's DUPLICATE_DATE check and drops it from + every scan, so the redundant copy is removed at the vendor boundary. + """ + payload = [ + {"timestamp": "2024-06-04", "open": 1.0, "high": 2.0, "low": 0.5, "close": 1.5, "volume": 10}, + {"timestamp": "2024-06-05", "open": 700.0, "high": 730.1, "low": 664.75, "close": 705.45, "volume": 1122766}, + {"timestamp": "2024-06-05", "open": 700.0, "high": 730.1, "low": 664.75, "close": 705.45, "volume": 1122766}, + ] + + frame = normalize_daily_payload(payload) + + assert len(frame.index) == 2 + # No trading day may be lost to the dedupe. + assert list(pd.to_datetime(frame["timestamp"]).dt.date) == [ + dt.date(2024, 6, 4), + dt.date(2024, 6, 5), + ] + + +def test_conflicting_bars_for_one_date_are_preserved(): + """Two *different* bars for one day must survive to be reported, not guessed at. + + Collapsing these would stitch together a price series that never existed. They + stay so DATA-001 quarantines the symbol and the DATA-002 repair resolves them + against the vendor instead. + """ + payload = [ + {"timestamp": "2024-06-05", "open": 700.0, "high": 730.0, "low": 664.0, "close": 705.0, "volume": 1_000}, + {"timestamp": "2024-06-05", "open": 12.0, "high": 13.0, "low": 11.0, "close": 12.5, "volume": 2_000}, + ] + + frame = normalize_daily_payload(payload) + + assert len(frame.index) == 2 + + +def test_duplicate_bars_differing_only_in_volume_are_preserved(): + """Same OHLC, different volume is a partial-vs-final bar, not a redundant copy. + + Only the DATA-002 repair may resolve that (highest volume wins); the vendor + boundary must not silently pick one. + """ + payload = [ + {"timestamp": "2024-06-05", "open": 700.0, "high": 730.0, "low": 664.0, "close": 705.0, "volume": 1_122_766}, + {"timestamp": "2024-06-05", "open": 700.0, "high": 730.0, "low": 664.0, "close": 705.0, "volume": 2_245}, + ] + + frame = normalize_daily_payload(payload) + + assert len(frame.index) == 2 + + +def test_deduped_payload_passes_the_data_quality_gate(): + """The end-to-end point: an exact-duplicate payload no longer quarantines.""" + from backend.data_quality.candles import validate_candles + + payload = [ + {"timestamp": "2024-06-04", "open": 100.0, "high": 105.0, "low": 99.0, "close": 104.0, "volume": 10}, + {"timestamp": "2024-06-05", "open": 100.0, "high": 105.0, "low": 99.0, "close": 104.0, "volume": 20}, + {"timestamp": "2024-06-05", "open": 100.0, "high": 105.0, "low": 99.0, "close": 104.0, "volume": 20}, + ] + + report = validate_candles(normalize_daily_payload(payload), symbol="AEGISLOG") + + assert "DUPLICATE_DATE" not in {finding.code for finding in report.findings} From 5a9e5eff698235f92f58b99384edbf64a5a4ceb1 Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Tue, 25 Aug 2026 00:28:24 +0530 Subject: [PATCH 2/6] fix(DATA-004): stop re-downloading stocks that listed after the window opened Both cache-coverage checks require first_date <= requested_start, where requested_start is today minus ten years. A stock that listed after that date can never satisfy it, because DhanHQ has nothing earlier to give. 200 of 577 cached symbols were in that state (DMART listed 2017-03-21, RBLBANK 2016-08-31, LTTS, COHANCE, ~196 more): every prefetch and every scan re-downloaded their full history, wrote the same short frame back, and did it again next time. The two cases look identical from the cached file alone and must not be conflated: an interrupted prefetch leaves a partial file while the vendor DOES have the missing years and must be refetched; a later listing means the vendor has nothing earlier and refetching is waste forever. Fixed by recording what the vendor actually served, following the sidecar precedent DATA-002 established with .checked and .repaired. A new .firstbar marker stores the probe's requested_from, the earliest_available bar that came back, and recorded_on. A cache is then treated as reaching back far enough when it either literally covers the requested start or already begins at the vendor's earliest known bar. Three properties keep the marker from ever hiding real missing history: - it only counts when the recorded probe reached at least as far back as the current request, so a five-year probe cannot suppress a ten-year refetch; - it expires after VENDOR_EARLIEST_RECHECK_DAYS (30), because vendors do occasionally backfill history; - any read or parse failure returns None and the strict rule applies, so a corrupt marker can only cost an extra request. Suppressing the backfill deliberately falls through to the normal freshness and incremental logic rather than returning early: a later listing still needs its daily top-up, and the point is to stop the pointless ten-year refetch, not to freeze the symbol. Measured on a copy of the real 577-file cache, nifty_500 (500 rows): prefetch pass 1 500 requests, 176 full-window backfills (learns) prefetch pass 2 176 requests, 0 full-window backfills prefetch pass 3 0 requests, 0 full-window backfills scan path before: 324 hits / 176 misses (176 full re-downloads) after: 500 hits / 0 misses, 0 files rewritten Stacked on DATA-003 (PR #112), which introduced _cache_covers_range and is not yet merged; this branch targets that one. Co-Authored-By: Claude Opus 5 --- backend/daily_data_loader.py | 192 ++++++++++- .../test_daily_data_loader_vendor_earliest.py | 311 ++++++++++++++++++ 2 files changed, 494 insertions(+), 9 deletions(-) create mode 100644 tests/test_daily_data_loader_vendor_earliest.py diff --git a/backend/daily_data_loader.py b/backend/daily_data_loader.py index 1b44a97..1e3475c 100644 --- a/backend/daily_data_loader.py +++ b/backend/daily_data_loader.py @@ -8,6 +8,7 @@ from __future__ import annotations import concurrent.futures +import json import logging import re import threading @@ -49,6 +50,13 @@ # ``history_start_date``) in one place stops the three callers from drifting apart. DEFAULT_HISTORY_YEARS_BACK = 10 +# How long we trust a recorded "the vendor has nothing earlier than this" answer +# before probing again. Vendors do occasionally backfill history, so the belief +# expires rather than becoming permanent; a month keeps the cost negligible +# (one request per affected symbol per month) while never hiding real data for +# long. Mirrors the DATA-002 ``REPAIR_RETRY_AFTER_DAYS`` cooldown precedent. +VENDOR_EARLIEST_RECHECK_DAYS = 30 + def history_start_date( years_back: int = DEFAULT_HISTORY_YEARS_BACK, today: date | None = None @@ -109,6 +117,7 @@ def _cache_covers_range( last_date: date | None, requested_start: date, requested_end: date, + vendor_earliest: date | None = None, ) -> bool: """Return True when a cached range is good enough to answer a request. @@ -130,7 +139,35 @@ def _cache_covers_range( if first_date is None or last_date is None: return False tolerated_end = requested_end - timedelta(days=STALE_LATEST_TOLERANCE_DAYS) - return first_date <= requested_start and last_date >= tolerated_end + if last_date < tolerated_end: + return False + return _cache_reaches_back_far_enough(first_date, requested_start, vendor_earliest) + + +def _cache_reaches_back_far_enough( + first_date: date, + requested_start: date, + vendor_earliest: date | None, +) -> bool: + """Return True when nothing earlier is missing that could still be fetched. + + Normally that means the cache literally reaches ``requested_start``. But a + stock that listed *after* that date can never satisfy it — DhanHQ has nothing + earlier to give — and demanding it made 200 of 577 symbols a permanent cache + miss, re-downloading their full history on every prefetch and every scan + (DATA-004). + + ``vendor_earliest`` is the earliest bar the vendor actually served for a probe + that reached at least as far back as this request (see + ``DailyDataLoader._vendor_earliest_for``). When the cache already starts at or + before that bar, it is as complete as it can ever be, so asking again is pure + waste. ``None`` means we have no such evidence and the strict rule applies — + which keeps an interrupted prefetch's partial file being refetched, since + there the vendor genuinely does have the missing years. + """ + if first_date <= requested_start: + return True + return vendor_earliest is not None and first_date <= vendor_earliest def _date_bounds(candles: pd.DataFrame) -> tuple[date | None, date | None]: @@ -333,6 +370,106 @@ def _write_checked_through( except OSError: logger.warning("Could not write daily-cache checked marker for %s", symbol) + def first_bar_path(self, symbol: str, security_id: str | int) -> Path: + """Return the sidecar recording how far back the vendor's history goes. + + A third marker alongside ``.checked`` (an empty tail) and ``.repaired`` + (a repair cooldown), following the same pattern: remember an answer the + vendor already gave so we do not pay for the identical request forever. + """ + return self.cache_path(symbol, security_id).with_suffix(".firstbar") + + def _vendor_earliest_for( + self, symbol: str, security_id: str | int, requested_start: date, today: date + ) -> date | None: + """The vendor's earliest bar, when we have evidence that answers this request. + + Returns ``None`` — meaning "no evidence, apply the strict rule" — unless + all three hold: + + - a marker exists and parses; + - it was recorded within ``VENDOR_EARLIEST_RECHECK_DAYS`` (vendors do + backfill occasionally, so the belief expires); + - the recorded probe reached **at least as far back** as this request. + Learning that nothing exists before 2021 when you only asked from 2021 + says nothing about 2016, so a shallower probe must not suppress a + deeper refetch. + + Any read or parse problem returns ``None``, so a corrupt marker can only + ever cost an extra request — never hide history that really is missing. + """ + path = self.first_bar_path(symbol, security_id) + if not path.exists(): + return None + try: + payload = json.loads(path.read_text(encoding="utf-8")) + requested_from = _coerce_date(str(payload["requested_from"])) + earliest_available = _coerce_date(str(payload["earliest_available"])) + recorded_on = _coerce_date(str(payload["recorded_on"])) + except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError): + return None + if (today - recorded_on).days >= VENDOR_EARLIEST_RECHECK_DAYS: + return None + if requested_from > requested_start: + return None + return earliest_available + + def _write_vendor_earliest( + self, + symbol: str, + security_id: str | int, + *, + requested_from: date, + earliest_available: date, + recorded_on: date, + ) -> None: + """Persist what the vendor served, so the next pass need not ask again.""" + try: + self.first_bar_path(symbol, security_id).write_text( + json.dumps( + { + "requested_from": requested_from.isoformat(), + "earliest_available": earliest_available.isoformat(), + "recorded_on": recorded_on.isoformat(), + } + ), + encoding="utf-8", + ) + except OSError: + # The marker is an optimisation, never a correctness requirement. + logger.warning("Could not write daily-cache first-bar marker for %s", symbol) + + def _record_vendor_earliest( + self, + symbol: str, + security_id: str | int, + *, + requested_from: date | datetime | str, + candles: pd.DataFrame, + today: date, + ) -> None: + """Record the vendor's earliest bar when it fell short of what we asked for. + + Called after any full-window download. A response that *does* reach the + requested start teaches us nothing worth storing, and an empty response is + no evidence at all, so both are skipped. + """ + if candles.empty: + return + first_date, _last_date = _date_bounds(candles) + if first_date is None: + return + start = _coerce_date(requested_from) + if first_date <= start: + return + self._write_vendor_earliest( + symbol, + security_id, + requested_from=start, + earliest_available=first_date, + recorded_on=today, + ) + def read_cached_history(self, symbol: str, security_id: str | int) -> pd.DataFrame: """Return the cached daily candles for one stock; empty DataFrame if missing. @@ -411,7 +548,12 @@ def get_daily_history( first_date, last_date = _date_bounds(cached) requested_start = _coerce_date(start_date) requested_end = _coerce_date(end_date) - if _cache_covers_range(first_date, last_date, requested_start, requested_end): + vendor_earliest = self._vendor_earliest_for( + symbol, security_id, requested_start, requested_end + ) + if _cache_covers_range( + first_date, last_date, requested_start, requested_end, vendor_earliest + ): if cached is None: # The footer is only an advisory index. The file can be # replaced after the metadata read, and a valid footer @@ -419,7 +561,11 @@ def get_daily_history( cached = pd.read_parquet(path) actual_first, actual_last = _date_bounds(cached) if _cache_covers_range( - actual_first, actual_last, requested_start, requested_end + actual_first, + actual_last, + requested_start, + requested_end, + vendor_earliest, ): return self._slice_to_range(cached, start_date, end_date), True @@ -435,6 +581,13 @@ def get_daily_history( if not candles.empty: path.parent.mkdir(parents=True, exist_ok=True) candles.to_parquet(path, index=False) + self._record_vendor_earliest( + symbol, + security_id, + requested_from=start_date, + candles=candles, + today=_coerce_date(end_date), + ) return self._slice_to_range(candles, start_date, end_date), False def fetch_window( @@ -516,6 +669,9 @@ def ensure_daily_history( if not candles.empty: path.parent.mkdir(parents=True, exist_ok=True) candles.to_parquet(path, index=False) + self._record_vendor_earliest( + symbol, security_id, requested_from=start, candles=candles, today=today + ) return candles, "fresh_download" cached = pd.read_parquet(path) @@ -531,6 +687,9 @@ def ensure_daily_history( ) if not candles.empty: candles.to_parquet(path, index=False) + self._record_vendor_earliest( + symbol, security_id, requested_from=start, candles=candles, today=today + ) return candles, "fresh_download" first_date, last_date = _date_bounds(cached) @@ -547,9 +706,18 @@ def ensure_daily_history( ) if not candles.empty: candles.to_parquet(path, index=False) + self._record_vendor_earliest( + symbol, security_id, requested_from=start, candles=candles, today=today + ) return candles, "fresh_download" - if first_date > start: + # A cache that starts late is either an interrupted prefetch (the vendor + # HAS the missing years, so refetch) or a stock that listed after the + # window opened (the vendor has nothing earlier, so refetching is waste + # forever). ``_vendor_earliest_for`` is what tells the two apart; without + # evidence it returns None and the original always-refetch rule applies. + vendor_earliest = self._vendor_earliest_for(symbol, security_id, start, today) + if not _cache_reaches_back_far_enough(first_date, start, vendor_earliest): # The cache may be current at the back but missing years at the # front, usually after an old interrupted prefetch. Refetch the # intended full window so long-lookback screeners see real history. @@ -560,10 +728,15 @@ def ensure_daily_history( from_date=start, to_date=today, ) + self._record_vendor_earliest( + symbol, security_id, requested_from=start, candles=candles, today=today + ) if not candles.empty: candles.to_parquet(path, index=False) return candles, "backfilled" return cached, "fresh" + # Falling through on purpose: a later listing still needs its daily + # top-up. Suppressing the backfill must not freeze the symbol's tail. if last_date >= today: return cached, "fresh" @@ -1195,17 +1368,18 @@ def cleanup_stale_cache_files( choose the age threshold, and only daily parquet files plus their sidecar markers are touched. - Two sidecar kinds travel with a parquet: `.checked` (the empty-increment - marker written here) and `.repaired` (the DATA-002 repair cooldown). Both - are meaningless without their parquet, so an orphan of either is removed - regardless of age. + Three sidecar kinds travel with a parquet: `.checked` (the empty-increment + marker written here), `.repaired` (the DATA-002 repair cooldown), and + `.firstbar` (the DATA-004 record of how far back the vendor's history + goes). All are meaningless without their parquet, so an orphan of any of + them is removed regardless of age. """ if not self.cache_dir.exists(): return 0 now = now or datetime.now() cutoff = now - timedelta(days=max(1, int(max_age_days))) targets: set[Path] = set() - sidecar_suffixes = (".checked", ".repaired") + sidecar_suffixes = (".checked", ".repaired", ".firstbar") for parquet in self.cache_dir.glob("*.parquet"): modified = datetime.fromtimestamp(parquet.stat().st_mtime) diff --git a/tests/test_daily_data_loader_vendor_earliest.py b/tests/test_daily_data_loader_vendor_earliest.py new file mode 100644 index 0000000..5d8968f --- /dev/null +++ b/tests/test_daily_data_loader_vendor_earliest.py @@ -0,0 +1,311 @@ +"""Tests for remembering how far back the vendor's history actually goes (DATA-004). + +Beginner note: +The cache-coverage test asks "does this file reach back to the start of the +requested window?". For a stock that listed *after* that start — DMART listed in +2017, well inside a ten-year window — the answer is permanently no, because DhanHQ +has nothing earlier to give. Before this change every prefetch and every scan +re-downloaded those symbols' whole history, wrote the same short frame back, and +did it again next time. 200 of 577 cached symbols were in that state. + +The fix records what the vendor actually served, following the ``.checked`` and +``.repaired`` sidecar precedent already in this codebase: once we have asked from +a given date and learned the earliest bar that exists, a cache reaching that bar is +as complete as it can ever be. + +The distinction these tests protect is the whole point: + +- an **interrupted prefetch** left a partial file and the vendor *does* have + earlier data → must still refetch; +- a **later listing** means the vendor has nothing earlier → refetching is waste. +""" + +from __future__ import annotations + +import json +from datetime import date, timedelta +from pathlib import Path +from typing import cast + +import pandas as pd + +from backend.daily_data_loader import DailyDataLoader +from backend.dhan_client import DhanDataClient + +TODAY = date(2026, 8, 24) +HISTORY_START = date(2016, 8, 24) # TODAY minus ten years +LISTED_ON = date(2017, 3, 21) # a DMART-style later listing +ROW = {"symbol": "DMART", "security_id": "1"} + + +def _month_series(first: date, last: date) -> pd.DatetimeIndex: + """Monthly bars that begin exactly on ``first`` and end on or before ``last``. + + ``pd.date_range(..., freq="MS")`` snaps to month starts, which would silently + move a mid-month listing date *and* leave the newest bar weeks short of the + requested end. Both endpoints are therefore inserted explicitly: these tests + turn on the first bar being exactly the listing date, and a snapped last bar + would make the frame fail the freshness half of the coverage test for reasons + that have nothing to do with what is being tested. + """ + months = pd.date_range(first, last, freq="MS") + return ( + pd.DatetimeIndex([pd.Timestamp(first), *months, pd.Timestamp(last)]) + .unique() + .sort_values() + ) + + +class ListedLateClient: + """A vendor that has no data before ``listed_on``, whatever you ask for. + + This is the real behaviour that made the coverage test unsatisfiable: the + request reaches back ten years, the response starts at the listing date. + """ + + def __init__(self, listed_on: date = LISTED_ON, *, through: date = TODAY) -> None: + self.listed_on = listed_on + self.through = through + # The loader passes real dates on every path these tests exercise. + self.calls: list[tuple[date, date]] = [] + + def fetch_daily_candles(self, *, from_date, to_date, **_kwargs) -> pd.DataFrame: + self.calls.append((from_date, to_date)) + # Monthly bars keep the fixture small; only the date bounds matter. The + # listing date itself is forced in because "MS" snaps to month starts, + # and the first bar being exactly the listing date is the whole point. + dates = _month_series(self.listed_on, self.through) + return pd.DataFrame( + { + "timestamp": dates, + "open": 100.0, + "high": 105.0, + "low": 99.0, + "close": 104.0, + "volume": 1_000.0, + } + ) + + +def _loader(tmp_path: Path, client: object) -> DailyDataLoader: + # The loader only calls fetch_daily_candles, so the duck-typed fake stands in. + return DailyDataLoader( + cast(DhanDataClient, client), cache_dir=tmp_path, request_delay_seconds=0.0 + ) + + +def _write_cache(loader: DailyDataLoader, first: date, last: date) -> Path: + path = loader.cache_path(ROW["symbol"], ROW["security_id"]) + pd.DataFrame( + { + "timestamp": _month_series(first, last), + "open": 100.0, + "high": 105.0, + "low": 99.0, + "close": 104.0, + "volume": 1_000.0, + } + ).to_parquet(path, index=False) + return path + + +def _marker(loader: DailyDataLoader) -> Path: + return loader.first_bar_path(ROW["symbol"], ROW["security_id"]) + + +# --------------------------------------------------------------------------- +# Learning the vendor's earliest bar +# --------------------------------------------------------------------------- + + +def test_a_full_download_records_the_vendors_earliest_bar(tmp_path: Path): + client = ListedLateClient() + loader = _loader(tmp_path, client) + + _frame, status = loader.ensure_daily_history(ROW, years_back=10, today=TODAY) + + assert status == "fresh_download" + payload = json.loads(_marker(loader).read_text(encoding="utf-8")) + # We asked from the ten-year start and learned the vendor begins at listing. + assert date.fromisoformat(payload["requested_from"]) == HISTORY_START + assert date.fromisoformat(payload["earliest_available"]) == LISTED_ON + + +def test_no_marker_is_written_when_the_vendor_covers_the_whole_window(tmp_path: Path): + """Nothing to remember when the request was fully satisfied.""" + client = ListedLateClient(listed_on=date(2016, 1, 1)) + loader = _loader(tmp_path, client) + + loader.ensure_daily_history(ROW, years_back=10, today=TODAY) + + assert not _marker(loader).exists() + + +def test_an_empty_response_records_nothing(tmp_path: Path): + """An empty answer is no evidence about how far back history goes.""" + + class EmptyClient: + def fetch_daily_candles(self, **_kwargs) -> pd.DataFrame: + return pd.DataFrame() + + loader = _loader(tmp_path, EmptyClient()) + + loader.ensure_daily_history(ROW, years_back=10, today=TODAY) + + assert not _marker(loader).exists() + + +# --------------------------------------------------------------------------- +# The bug: re-downloading a later listing forever +# --------------------------------------------------------------------------- + + +def test_a_later_listing_is_not_backfilled_again_on_the_next_pass(tmp_path: Path): + """The core defect. Second prefetch must not re-download the same history.""" + client = ListedLateClient() + loader = _loader(tmp_path, client) + + _frame, first_status = loader.ensure_daily_history(ROW, years_back=10, today=TODAY) + calls_after_first = len(client.calls) + _frame, second_status = loader.ensure_daily_history(ROW, years_back=10, today=TODAY) + + assert first_status == "fresh_download" + assert second_status != "backfilled" + # The tail top-up may still ask; the ten-year backfill must not. + backfills = [c for c in client.calls[calls_after_first:] if c[0] == HISTORY_START] + assert backfills == [] + + +def test_a_later_listing_still_gets_its_daily_top_up(tmp_path: Path): + """Skipping the pointless backfill must not freeze the symbol's tail. + + The whole value of the cache is that it keeps advancing; suppressing the + backfill must only suppress the backfill. + """ + client = ListedLateClient() + loader = _loader(tmp_path, client) + _write_cache(loader, first=LISTED_ON, last=TODAY - timedelta(days=40)) + loader._write_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=HISTORY_START, + earliest_available=LISTED_ON, recorded_on=TODAY, + ) + + _frame, status = loader.ensure_daily_history(ROW, years_back=10, today=TODAY) + + assert status == "incremental" + # The request started after the cached tail, not at the ten-year start. + assert client.calls[-1][0] > TODAY - timedelta(days=41) + + +def test_a_later_listing_is_a_cache_hit_for_scans(tmp_path: Path): + """The scan path benefits too: no fetch, no rewrite.""" + client = ListedLateClient() + loader = _loader(tmp_path, client) + _write_cache(loader, first=LISTED_ON, last=TODAY) + loader._write_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=HISTORY_START, + earliest_available=LISTED_ON, recorded_on=TODAY, + ) + + _frame, from_cache = loader.get_daily_history( + ROW, start_date=HISTORY_START, end_date=TODAY + ) + + assert from_cache is True + assert client.calls == [] + + +# --------------------------------------------------------------------------- +# The distinction that must not be lost +# --------------------------------------------------------------------------- + + +def test_a_genuinely_partial_cache_is_still_backfilled(tmp_path: Path): + """An interrupted prefetch, where the vendor DOES have earlier data. + + The marker says history begins in 2016, but the cache starts in 2020 — so + there are real bars missing and the backfill must run. + """ + client = ListedLateClient(listed_on=date(2016, 1, 1)) + loader = _loader(tmp_path, client) + _write_cache(loader, first=date(2020, 1, 1), last=TODAY) + loader._write_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=HISTORY_START, + earliest_available=date(2016, 1, 1), recorded_on=TODAY, + ) + + _frame, status = loader.ensure_daily_history(ROW, years_back=10, today=TODAY) + + assert status == "backfilled" + + +def test_a_shallower_probe_does_not_prove_anything_about_earlier_history(tmp_path: Path): + """A marker from a 5-year probe cannot answer a 10-year request. + + Learning that nothing exists before 2021 when you only asked from 2021 says + nothing about 2016, so the deeper request must still go to the vendor. + """ + client = ListedLateClient(listed_on=date(2021, 6, 1)) + loader = _loader(tmp_path, client) + _write_cache(loader, first=date(2021, 6, 1), last=TODAY) + loader._write_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=date(2021, 1, 1), + earliest_available=date(2021, 6, 1), recorded_on=TODAY, + ) + + _frame, status = loader.ensure_daily_history(ROW, years_back=10, today=TODAY) + + assert status == "backfilled" + + +def test_a_stale_marker_is_re_probed(tmp_path: Path): + """Vendors do occasionally backfill history, so the belief expires.""" + client = ListedLateClient() + loader = _loader(tmp_path, client) + _write_cache(loader, first=LISTED_ON, last=TODAY) + loader._write_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=HISTORY_START, + earliest_available=LISTED_ON, recorded_on=date(2026, 1, 1), + ) + + _frame, status = loader.ensure_daily_history(ROW, years_back=10, today=TODAY) + + assert status == "backfilled" + + +def test_an_unreadable_marker_fails_open_to_a_refetch(tmp_path: Path): + """A corrupt marker may cost a request; it must never hide missing history.""" + client = ListedLateClient() + loader = _loader(tmp_path, client) + _write_cache(loader, first=LISTED_ON, last=TODAY) + _marker(loader).write_text("not json", encoding="utf-8") + + _frame, status = loader.ensure_daily_history(ROW, years_back=10, today=TODAY) + + assert status == "backfilled" + + +# --------------------------------------------------------------------------- +# Housekeeping +# --------------------------------------------------------------------------- + + +def test_orphan_first_bar_markers_are_cleaned_up(tmp_path: Path): + """`.firstbar` travels with its parquet, like `.checked` and `.repaired`.""" + loader = _loader(tmp_path, ListedLateClient()) + orphan = tmp_path / "GONE_9.firstbar" + orphan.write_text("{}", encoding="utf-8") + + removed = loader.cleanup_stale_cache_files(max_age_days=30) + + assert removed >= 1 + assert not orphan.exists() + + +def test_the_marker_suffix_is_distinct_from_the_other_sidecars(tmp_path: Path): + loader = _loader(tmp_path, ListedLateClient()) + path = _marker(loader) + + assert path.suffix == ".firstbar" + assert path.with_suffix(".checked") != path + assert path.with_suffix(".repaired") != path From 551b3af50c24a37f2760f751d09719faf3760f99 Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Tue, 25 Aug 2026 14:32:48 +0530 Subject: [PATCH 3/6] fix(DATA-003): serve a stale cache only on evidence, not a time window Addresses Codex P1 on PR #112. The blanket "tolerate STALE_LATEST_TOLERANCE_DAYS" rule served any cache within four calendar days of the requested end, so a cache ending Thursday was handed to a Monday scan even though Friday's bar had been published. Screeners then ran on prices demonstrably behind the market with no refresh attempted. Worse, the docstring justified this by claiming DATA-001 would still raise STALE_LATEST_CANDLE as a backstop. It does not: the warning fires only when the gap is GREATER than the same constant, so the two rules cover disjoint ranges and everything the coverage test tolerated passed silently. That claim was simply wrong and is now corrected in place. The end test is now satisfied by evidence rather than by elapsed time: - _only_unpublished_days_missing: nothing is absent except weekends and possibly the requested end itself. Tolerating the request end is load-bearing, because the current session's EOD bar is routinely unpublished when a scan runs; without it every weekday scan becomes a miss again, which is the problem DATA-003 set out to fix. - the loader's existing .checked marker: written by the prefetch precisely when it asked Dhan for this tail and got nothing back. That covers market holidays, which are weekdays and so fail the arithmetic above despite there being no bar to fetch. Codex suggested limiting the relaxation to known non-trading days via a trading calendar. That dependency was explicitly rejected in the DATA-001 design as heavier than the problem warrants, so this gets the same result from weekday arithmetic plus a receipt the loader already writes. Behaviour, verified by test: cache Fri, scan Mon hit (weekend only - the case DATA-003 exists for) cache Thu, scan Mon MISS (Friday's bar exists - Codex's case) cache Fri, scan Tue MISS (Monday's bar exists) cache Mon, scan Tue hit (today's bar not published yet) cache 3 weeks old MISS holiday gap + .checked marker hit same gap without the marker MISS marker older than the request end MISS STALE_LATEST_TOLERANCE_DAYS is no longer imported by the loader; the constant keeps its DATA-001 meaning and is no longer overloaded to mean two things. Co-Authored-By: Claude Opus 5 --- backend/daily_data_loader.py | 84 ++++++++++++++++++----- tests/test_daily_data_loader.py | 115 ++++++++++++++++++++++++++++++++ 2 files changed, 184 insertions(+), 15 deletions(-) diff --git a/backend/daily_data_loader.py b/backend/daily_data_loader.py index 1b44a97..3ab6e66 100644 --- a/backend/daily_data_loader.py +++ b/backend/daily_data_loader.py @@ -27,7 +27,6 @@ dhan_request_delay_seconds, ) from backend.data_quality import CandleQualityReport, validate_candles -from backend.data_quality.candles import STALE_LATEST_TOLERANCE_DAYS from backend.dhan_client import DhanDataClient, DhanRateLimitError from backend.observability import ( EVENT_CANDLE_DATA_QUALITY_FAILED, @@ -104,11 +103,47 @@ def safe_file_stem(value: object) -> str: return cleaned +# The longest gap the weekday walk below can ever accept is a long weekend, so a +# wider gap is rejected without walking it. Purely a performance guard: a decade-old +# cache should not iterate 3,650 days to reach the same "no" a subtraction gives. +_MAX_TOLERABLE_GAP_DAYS = 7 + + +def _only_unpublished_days_missing(last_date: date, requested_end: date) -> bool: + """True when nothing the market has actually published is missing from the cache. + + Two kinds of day may be absent without the cache being out of date: + + - **weekends**, when the exchange did not trade at all; and + - **``requested_end`` itself**, because the current session's end-of-day bar is + routinely not published yet when a scan runs. + + That second carve-out is load-bearing. Without it every weekday scan would be a + miss again, which is the whole problem DATA-003 set out to fix. + + Any *other* missing weekday means a bar the market really did publish is absent, + so the cache is genuinely behind and must be refreshed. This replaces an earlier + blanket "tolerate four calendar days" rule, which silently served a cache ending + on Thursday to a Monday scan even though Friday's bar existed. + """ + if last_date >= requested_end: + return True + if (requested_end - last_date).days > _MAX_TOLERABLE_GAP_DAYS: + return False + day = last_date + timedelta(days=1) + while day < requested_end: + if day.weekday() < 5: # Monday-Friday + return False + day += timedelta(days=1) + return True + + def _cache_covers_range( first_date: date | None, last_date: date | None, requested_start: date, requested_end: date, + checked_through: date | None = None, ) -> bool: """Return True when a cached range is good enough to answer a request. @@ -116,21 +151,30 @@ def _cache_covers_range( after an interrupted prefetch) must never silently run a long-lookback screener on too little data. - The **end** is compared with ``STALE_LATEST_TOLERANCE_DAYS`` of slack, because - callers routinely ask for data "through today" while the newest bar the vendor - has published is Friday's. Demanding an exact match made every symbol a cache - miss outside market hours, so each scan re-downloaded the entire universe and - overwrote the cache with raw vendor data (DATA-003). - - Reusing DATA-001's constant is deliberate: it is already the app's definition - of "how far the newest candle may trail today before that is suspicious", and - a frame inside it still raises ``STALE_LATEST_CANDLE`` as a warning, so nothing - is hidden by treating it as current enough to serve. + The **end** is satisfied by evidence rather than by a time window, in one of + two ways: + + - ``_only_unpublished_days_missing`` — nothing absent but weekends and possibly + the current session's own unpublished bar; or + - ``checked_through`` — the loader's existing ``.checked`` marker, written by + the prefetch precisely when it asked Dhan for this tail and got nothing back. + That covers market holidays, which are weekdays and so fail the arithmetic + above despite there being no bar to fetch. + + Note this rule and DATA-001's ``STALE_LATEST_CANDLE`` warning cover **disjoint** + ranges: the warning fires only when the newest bar trails by *more* than + ``STALE_LATEST_TOLERANCE_DAYS``, so anything served here is below its threshold + and passes silently. An earlier version of this docstring claimed the warning + still fired as a backstop — it does not, which is exactly why the end test has + to stand on its own evidence rather than on a tolerance window. """ if first_date is None or last_date is None: return False - tolerated_end = requested_end - timedelta(days=STALE_LATEST_TOLERANCE_DAYS) - return first_date <= requested_start and last_date >= tolerated_end + if first_date > requested_start: + return False + if _only_unpublished_days_missing(last_date, requested_end): + return True + return checked_through is not None and checked_through >= requested_end def _date_bounds(candles: pd.DataFrame) -> tuple[date | None, date | None]: @@ -411,7 +455,13 @@ def get_daily_history( first_date, last_date = _date_bounds(cached) requested_start = _coerce_date(start_date) requested_end = _coerce_date(end_date) - if _cache_covers_range(first_date, last_date, requested_start, requested_end): + # The prefetch's marker is the only evidence that distinguishes a + # market holiday (a weekday with no bar to fetch) from a weekday whose + # bar we simply have not collected yet. + checked_through = self._read_checked_through(symbol, security_id) + if _cache_covers_range( + first_date, last_date, requested_start, requested_end, checked_through + ): if cached is None: # The footer is only an advisory index. The file can be # replaced after the metadata read, and a valid footer @@ -419,7 +469,11 @@ def get_daily_history( cached = pd.read_parquet(path) actual_first, actual_last = _date_bounds(cached) if _cache_covers_range( - actual_first, actual_last, requested_start, requested_end + actual_first, + actual_last, + requested_start, + requested_end, + checked_through, ): return self._slice_to_range(cached, start_date, end_date), True diff --git a/tests/test_daily_data_loader.py b/tests/test_daily_data_loader.py index d26bba5..f6c7fc6 100644 --- a/tests/test_daily_data_loader.py +++ b/tests/test_daily_data_loader.py @@ -1203,3 +1203,118 @@ def test_cache_missing_early_history_is_a_miss_however_current_it_is(tmp_path): ) assert from_cache is False + + +# --------------------------------------------------------------------------- +# DATA-003 follow-up: only genuinely unpublishable days may be tolerated +# --------------------------------------------------------------------------- + + +def _cache_ending(loader, last_date, *, first_date=date(2016, 8, 1)): + pd.DataFrame( + { + "timestamp": pd.to_datetime([first_date, last_date]), + "open": [100.0, 101.0], + "high": [105.0, 106.0], + "low": [99.0, 100.0], + "close": [104.0, 105.0], + "volume": [1_000.0, 1_100.0], + } + ).to_parquet(loader.cache_path("DEMO", "1"), index=False) + + +def _is_hit(loader, *, end_date, start_date=date(2016, 8, 24)): + _frame, from_cache = loader.get_daily_history( + {"symbol": "DEMO", "security_id": "1"}, + start_date=start_date, + end_date=end_date, + ) + return from_cache + + +def test_a_weekend_only_gap_is_a_hit(tmp_path): + """Friday's bar, scanned on Monday: only Sat/Sun are missing, so nothing is lost. + + This is the case DATA-003 exists to serve — without it every scan re-downloads + the whole universe. + """ + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _cache_ending(loader, date(2026, 8, 21)) # Friday + + assert _is_hit(loader, end_date=date(2026, 8, 24)) is True # Monday + assert client.calls == 0 + + +def test_a_missing_published_weekday_is_a_miss(tmp_path): + """Thursday's bar, scanned on Monday: Friday's bar exists and is absent. + + Serving this silently would run screeners on prices that are demonstrably + behind the market. + """ + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _cache_ending(loader, date(2026, 8, 20)) # Thursday + + assert _is_hit(loader, end_date=date(2026, 8, 24)) is False # Monday + + +def test_todays_own_bar_may_be_unpublished(tmp_path): + """The request end itself is always tolerated — EOD data lands late.""" + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _cache_ending(loader, date(2026, 8, 24)) # Monday + + assert _is_hit(loader, end_date=date(2026, 8, 25)) is True # Tuesday + assert client.calls == 0 + + +def test_a_skipped_midweek_day_is_a_miss(tmp_path): + """Friday's bar scanned on Tuesday: Monday's bar exists and is absent.""" + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _cache_ending(loader, date(2026, 8, 21)) # Friday + + assert _is_hit(loader, end_date=date(2026, 8, 25)) is False # Tuesday + + +def test_a_long_stale_cache_is_a_miss(tmp_path): + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _cache_ending(loader, date(2026, 8, 3)) + + assert _is_hit(loader, end_date=date(2026, 8, 24)) is False + + +def test_a_market_holiday_is_a_hit_when_the_prefetch_already_asked(tmp_path): + """A holiday Monday is a weekday, so arithmetic alone would force a refetch. + + The prefetch's existing `.checked` marker records that Dhan was asked for this + tail and had nothing, which is direct evidence rather than a guess. + """ + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _cache_ending(loader, date(2026, 8, 21)) # Friday + loader.checked_path("DEMO", "1").write_text("2026-08-25", encoding="utf-8") + + assert _is_hit(loader, end_date=date(2026, 8, 25)) is True # holiday Monday behind us + assert client.calls == 0 + + +def test_the_same_holiday_gap_is_a_miss_without_that_evidence(tmp_path): + """No marker means we genuinely do not know, so ask.""" + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _cache_ending(loader, date(2026, 8, 21)) + + assert _is_hit(loader, end_date=date(2026, 8, 25)) is False + + +def test_a_stale_checked_marker_does_not_vouch_for_a_later_request(tmp_path): + """Evidence from last week says nothing about today's missing bars.""" + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _cache_ending(loader, date(2026, 8, 21)) + loader.checked_path("DEMO", "1").write_text("2026-08-22", encoding="utf-8") + + assert _is_hit(loader, end_date=date(2026, 8, 25)) is False From 6ab43f67db9978aa7f1bb28ba98b9dbd40613d43 Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Tue, 25 Aug 2026 21:05:52 +0530 Subject: [PATCH 4/6] fix(DATA-003): bound the .checked fallback to a short gap Found by a self-review pass over the previous commit. The .checked branch of _cache_covers_range had no staleness bound, so the marker could certify a cache that was arbitrarily far behind as fully covering the request. Concretely: normalize_daily_response turns a "no data" / "no records" vendor response into an empty frame rather than raising, so a DhanHQ data outage has ensure_daily_history stamp .checked = today on every prefetch. Since the fallback only tested checked_through >= requested_end and never how far last_date trailed, a scan weeks later still got from_cache=True and computed signals on stale prices while the cache-miss counters reported a clean hit. The weekday walk it bypasses is bounded by _MAX_TOLERABLE_GAP_DAYS; the fallback skipped that guard entirely. The marker exists to rescue market holidays, which are short gaps by definition, so it now carries the same bound. A genuinely halted or delisted symbol therefore becomes a cache miss again rather than being served silently - the honest outcome, and DATA-001 already raises STALE_LATEST_CANDLE for it. Co-Authored-By: Claude Opus 5 --- backend/daily_data_loader.py | 10 +++++++++- tests/test_daily_data_loader.py | 27 +++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/backend/daily_data_loader.py b/backend/daily_data_loader.py index 3ab6e66..dce9d7a 100644 --- a/backend/daily_data_loader.py +++ b/backend/daily_data_loader.py @@ -159,7 +159,8 @@ def _cache_covers_range( - ``checked_through`` — the loader's existing ``.checked`` marker, written by the prefetch precisely when it asked Dhan for this tail and got nothing back. That covers market holidays, which are weekdays and so fail the arithmetic - above despite there being no bar to fetch. + above despite there being no bar to fetch. It is bounded by + ``_MAX_TOLERABLE_GAP_DAYS`` too, so it can only ever rescue a short gap. Note this rule and DATA-001's ``STALE_LATEST_CANDLE`` warning cover **disjoint** ranges: the warning fires only when the newest bar trails by *more* than @@ -174,6 +175,13 @@ def _cache_covers_range( return False if _only_unpublished_days_missing(last_date, requested_end): return True + # The marker exists to rescue market holidays, which are by definition short + # gaps, so it is bounded by the same window as the weekday walk above. Without + # that bound a vendor outage answering "no data" instead of erroring would have + # the prefetch stamp `.checked` daily, certifying an arbitrarily stale cache as + # complete while scans reported a clean hit. + if (requested_end - last_date).days > _MAX_TOLERABLE_GAP_DAYS: + return False return checked_through is not None and checked_through >= requested_end diff --git a/tests/test_daily_data_loader.py b/tests/test_daily_data_loader.py index f6c7fc6..bf1d14e 100644 --- a/tests/test_daily_data_loader.py +++ b/tests/test_daily_data_loader.py @@ -1318,3 +1318,30 @@ def test_a_stale_checked_marker_does_not_vouch_for_a_later_request(tmp_path): loader.checked_path("DEMO", "1").write_text("2026-08-22", encoding="utf-8") assert _is_hit(loader, end_date=date(2026, 8, 25)) is False + + +def test_a_checked_marker_cannot_vouch_for_an_unboundedly_stale_cache(tmp_path): + """Self-review finding: the holiday rescue must stay a holiday rescue. + + A vendor outage that answers "no data" rather than erroring makes the prefetch + stamp `.checked` every day. Without a staleness bound that marker would certify + a month-old cache as complete, and scans would compute signals on stale prices + while reporting a clean cache hit. + """ + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _cache_ending(loader, date(2026, 7, 20)) # a month behind + loader.checked_path("DEMO", "1").write_text("2026-08-25", encoding="utf-8") + + assert _is_hit(loader, end_date=date(2026, 8, 25)) is False + + +def test_a_checked_marker_still_rescues_a_short_holiday_gap(tmp_path): + """The bound must not break the case the fallback exists for.""" + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _cache_ending(loader, date(2026, 8, 21)) # Friday + loader.checked_path("DEMO", "1").write_text("2026-08-25", encoding="utf-8") + + assert _is_hit(loader, end_date=date(2026, 8, 25)) is True + assert client.calls == 0 From 8f4e31e702ad7cb5f33a81dc5d6b208e35d8c509 Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Tue, 1 Sep 2026 15:42:22 +0530 Subject: [PATCH 5/6] fix(DATA-003): scope unpublished tail to scanner callers Co-authored-by: Codex --- backend/daily_data_loader.py | 145 +++++++++++++----- .../components/data-acquisition.md | 11 +- tests/test_candle_cache_write_paths.py | 108 +++++++++++++ tests/test_daily_data_loader.py | 111 ++++++++++++-- 4 files changed, 317 insertions(+), 58 deletions(-) diff --git a/backend/daily_data_loader.py b/backend/daily_data_loader.py index dce9d7a..728d32f 100644 --- a/backend/daily_data_loader.py +++ b/backend/daily_data_loader.py @@ -109,30 +109,42 @@ def safe_file_stem(value: object) -> str: _MAX_TOLERABLE_GAP_DAYS = 7 -def _only_unpublished_days_missing(last_date: date, requested_end: date) -> bool: - """True when nothing the market has actually published is missing from the cache. +def _only_unpublished_days_missing( + last_date: date, + requested_end: date, + *, + allow_requested_end: bool = False, +) -> bool: + """Return whether the cache misses only days with no published candle. Two kinds of day may be absent without the cache being out of date: - **weekends**, when the exchange did not trade at all; and - - **``requested_end`` itself**, because the current session's end-of-day bar is - routinely not published yet when a scan runs. - - That second carve-out is load-bearing. Without it every weekday scan would be a - miss again, which is the whole problem DATA-003 set out to fix. - - Any *other* missing weekday means a bar the market really did publish is absent, - so the cache is genuinely behind and must be refreshed. This replaces an earlier - blanket "tolerate four calendar days" rule, which silently served a cache ending - on Thursday to a Monday scan even though Friday's bar existed. + - ``requested_end`` itself, but only when the caller explicitly says it is + running a current scanner session whose end-of-day bar may not exist yet. + + Args: + last_date: The newest valid candle date in the cached Parquet frame. + requested_end: The inclusive end date requested by the caller. + allow_requested_end: Whether this caller may treat a missing weekday at + exactly ``requested_end`` as an unpublished current-session candle. + + Beginner note: + A direct caller can ask for a historical date where every weekday candle + is already expected to exist. The scanner alone knows it is asking for + the still-publishing current session, so this helper requires that + caller to grant ``allow_requested_end`` deliberately. Any other missing + weekday means the cache is behind and must be refreshed. """ if last_date >= requested_end: return True if (requested_end - last_date).days > _MAX_TOLERABLE_GAP_DAYS: return False day = last_date + timedelta(days=1) - while day < requested_end: - if day.weekday() < 5: # Monday-Friday + while day <= requested_end: + # Weekends never have daily exchange candles. The requested weekday may + # be absent only for the scanner's explicitly authorised live tail. + if day.weekday() < 5 and (day != requested_end or not allow_requested_end): return False day += timedelta(days=1) return True @@ -144,37 +156,48 @@ def _cache_covers_range( requested_start: date, requested_end: date, checked_through: date | None = None, + *, + allow_unpublished_tail: bool = False, ) -> bool: - """Return True when a cached range is good enough to answer a request. + """Return whether a cached range has the authority to answer a request. The **start** is compared strictly: a parquet missing early history (common after an interrupted prefetch) must never silently run a long-lookback screener on too little data. - The **end** is satisfied by evidence rather than by a time window, in one of - two ways: - - - ``_only_unpublished_days_missing`` — nothing absent but weekends and possibly - the current session's own unpublished bar; or - - ``checked_through`` — the loader's existing ``.checked`` marker, written by - the prefetch precisely when it asked Dhan for this tail and got nothing back. - That covers market holidays, which are weekdays and so fail the arithmetic - above despite there being no bar to fetch. It is bounded by - ``_MAX_TOLERABLE_GAP_DAYS`` too, so it can only ever rescue a short gap. - - Note this rule and DATA-001's ``STALE_LATEST_CANDLE`` warning cover **disjoint** - ranges: the warning fires only when the newest bar trails by *more* than - ``STALE_LATEST_TOLERANCE_DAYS``, so anything served here is below its threshold - and passes silently. An earlier version of this docstring claimed the warning - still fired as a backstop — it does not, which is exactly why the end test has - to stand on its own evidence rather than on a tolerance window. + By default, a weekday ``requested_end`` must be present and ``.checked`` + evidence is ignored. A caller that is actively loading the scanner universe + may opt into a bounded unpublished tail. That opt-in permits the requested + end itself to be unpublished and accepts a recent ``.checked`` marker for a + short weekday-holiday gap. + + Args: + first_date: Oldest valid candle date in the candidate cache. + last_date: Newest valid candle date in the candidate cache. + requested_start: Inclusive requested start date. + requested_end: Inclusive requested end date. + checked_through: Optional prefetch sidecar date recording an empty tail. + allow_unpublished_tail: Scanner-only authority to use current-session + and sidecar-marker tail relaxation. + + Beginner note: + A `.checked` marker means a previous prefetch received no newer rows; it + does not prove a historical request is complete. Keeping this authority + at the scanner call site prevents forward-return and other historical + calculations from silently using an incomplete weekday range. """ if first_date is None or last_date is None: return False if first_date > requested_start: return False - if _only_unpublished_days_missing(last_date, requested_end): + if _only_unpublished_days_missing( + last_date, + requested_end, + allow_requested_end=allow_unpublished_tail, + ): return True + if not allow_unpublished_tail: + return False # The marker exists to rescue market holidays, which are by definition short # gaps, so it is bounded by the same window as the weekday walk above. Without # that bound a vendor outage answering "no data" instead of erroring would have @@ -424,12 +447,28 @@ def get_daily_history( start_date: date | datetime | str, end_date: date | datetime | str, force_refresh: bool = False, + *, + allow_unpublished_tail: bool = False, ) -> tuple[pd.DataFrame, bool]: - """ - Return daily candles for one instrument, sliced to the requested range. - - The boolean indicates whether the result was answered without hitting - Dhan (i.e., served entirely from the local Parquet cache). + """Return daily candles for one instrument, sliced to the requested range. + + Args: + instrument: Universe row containing symbol, security ID, and Dhan + instrument metadata. + start_date: Inclusive first date required by the caller. + end_date: Inclusive final date required by the caller. + force_refresh: Bypass a usable cache and fetch the requested range. + allow_unpublished_tail: Scanner-only authority to accept a bounded + current-session/weekend/marker cache tail. + + Returns: + The requested candle frame and whether it was served from cache. + + Beginner note: + This public method also serves historical consumers such as + forward-return validation. They must receive complete weekday data, + so the current-session relaxation is opt-in rather than inferred + from dates or the machine clock. """ row = dict(instrument) # Universe CSV rows are the source of truth for how to ask Dhan for a @@ -446,7 +485,8 @@ def get_daily_history( path = self.cache_path(symbol, security_id) if path.exists() and not force_refresh: - # Cache hit only when the file covers the entire requested range. + # A cache hit requires complete coverage unless this scanner caller + # explicitly grants the bounded unpublished-tail exception below. # A partial parquet is common after interrupted prefetches; slicing # it would silently run long-lookback screeners on too little data. # @@ -468,7 +508,12 @@ def get_daily_history( # bar we simply have not collected yet. checked_through = self._read_checked_through(symbol, security_id) if _cache_covers_range( - first_date, last_date, requested_start, requested_end, checked_through + first_date, + last_date, + requested_start, + requested_end, + checked_through, + allow_unpublished_tail=allow_unpublished_tail, ): if cached is None: # The footer is only an advisory index. The file can be @@ -482,6 +527,7 @@ def get_daily_history( requested_start, requested_end, checked_through, + allow_unpublished_tail=allow_unpublished_tail, ): return self._slice_to_range(cached, start_date, end_date), True @@ -656,7 +702,9 @@ def ensure_daily_history( merged = ( pd.concat([cached, new_rows], ignore_index=True) - .drop_duplicates(subset=["timestamp"], keep="last") + # Keep same-date disagreements for DATA-001/DATA-002 to investigate; + # only a row identical across all six canonical columns is redundant. + .drop_duplicates() .sort_values("timestamp") .reset_index(drop=True) ) @@ -970,7 +1018,13 @@ def _iter_history_sequential( force_refresh: bool, progress_callback: ProgressCallback | None, ): - """The long-standing one-symbol-at-a-time path (fetch_workers == 1).""" + """The long-standing one-symbol-at-a-time path (fetch_workers == 1). + + Beginner note: + This is a scanner-owned caller, so it explicitly permits a short + current-session or checked-marker tail. Direct historical callers + use ``get_daily_history`` without that authority and remain strict. + """ consecutive_failures = 0 for index, row in enumerate(rows, start=1): symbol = str(row.get("symbol", "")).strip().upper() or "UNKNOWN" @@ -986,6 +1040,7 @@ def _iter_history_sequential( start_date=start_date, end_date=end_date, force_refresh=force_refresh, + allow_unpublished_tail=True, ) consecutive_failures = 0 if from_cache: @@ -1025,6 +1080,11 @@ def _iter_history_parallel( consumed normally (the request already happened; discarding the data would help nobody). Rows never submitted yield breaker items, exactly like the sequential path. + + Beginner note: + Parallel workers receive the same explicit scanner-only tail + authority as the sequential path. Keeping the flag here, rather + than making it the public default, protects historical consumers. """ window = self.fetch_workers * 2 pending: deque[tuple[int, dict, str, concurrent.futures.Future]] = deque() @@ -1047,6 +1107,7 @@ def submit_next() -> bool: start_date=start_date, end_date=end_date, force_refresh=force_refresh, + allow_unpublished_tail=True, ) pending.append((index, row, symbol, future)) return True diff --git a/docs/architecture/components/data-acquisition.md b/docs/architecture/components/data-acquisition.md index 5eac3a8..20e39a1 100644 --- a/docs/architecture/components/data-acquisition.md +++ b/docs/architecture/components/data-acquisition.md @@ -58,7 +58,7 @@ flowchart TD |---|---| | `DailyDataLoader(client, cache_dir, request_delay_seconds, rate_limit_retry_delays, fetch_timeout_seconds, max_consecutive_failures, fetch_workers, sleep_func)` | `client=None` ⇒ cache-only mode (fetches fail loudly). `fetch_workers` (1–8, default 1 via `SCANNER_DHAN_FETCH_WORKERS`) opts into parallel fetch behind a shared `_RequestPacer` (PERF-001). | | `.read_cached_history(symbol, security_id)` | Disk-only read (chart UI path); empty frame if missing/corrupt. | -| `.get_daily_history(instrument, start, end, force_refresh=False)` | `(frame, served_from_cache)`; **cache hit only when the file covers the entire requested range.** | +| `.get_daily_history(instrument, start, end, force_refresh=False, *, allow_unpublished_tail=False)` | `(frame, served_from_cache)`; direct and historical callers require complete weekday coverage. Only scanner universe loading explicitly opts into a bounded current-session/weekend/marker tail. | | `.ensure_daily_history(instrument, years_back=10, today=None)` | `(frame, status)` where status ∈ `fresh`/`incremental`/`fresh_download`/`backfilled`. The prefetch engine. | | `.fetch_window(instrument, start, end)` | Network-only fetch (same pacing, DH-904 backoff, and optional timeout) that **does not write the cache**. Added for the DATA-002 repair, which merges a bounded window *over* existing history — writing it directly would truncate a ten-year file to that window. An empty frame means the vendor has no rows there, not an error. | | `.iter_universe_history(...)` | Yields `HistoryLoadItem` per symbol (streaming — compute as you load). | @@ -71,10 +71,10 @@ flowchart TD | Decision | Rationale | Alternative rejected | |---|---|---| -| **Normalize at the boundary** | Screeners get one stable 6-col frame; SDK wire-shape changes are absorbed here. | Per-screener parsing — duplicated, fragile. | +| **Normalize exact rows at the boundary** | Screeners get one stable `timestamp, open, high, low, close, volume` frame; exact duplicates across all six canonical columns are removed before every cache write. Same-date rows that differ in any column remain for DATA-001/DATA-002 rather than silently picking a price. | Per-screener parsing or date-only de-duplication — duplicated, fragile, or able to hide a vendor conflict. | | **One file per `(symbol, security_id)`, no date in name** | Different scan windows reuse one growing cache; incremental top-up only fetches missing tail. | Date-range filenames (legacy) — duplicate files, re-fetches. `cleanup_legacy_cache_files` removes those. | -| **Cache hit requires full-range coverage** | A partial parquet (interrupted prefetch) would silently run a long-lookback screener on too little data. | Slice whatever exists — silent wrong results. | -| **`.checked` sidecar marker** | Remembers a no-new-rows tail (weekend/holiday) so the next launch doesn't re-pay for the same empty request. | Re-request every launch — wasted quota. | +| **Conservative direct cache coverage** | Historical and direct `get_daily_history` callers require every requested weekday candle; they ignore `.checked` evidence. Weekend-only gaps remain valid because no daily bar exists on those dates. | Infer that every request is a live scanner session — can silently feed incomplete history to forward-return calculations. | +| **Explicit scanner unpublished-tail opt-in** | Only the sequential and parallel universe-loading paths pass `allow_unpublished_tail=True`, allowing the current session's requested end and a `.checked`-verified weekday-holiday gap. The marker may rescue at most seven calendar days (`_MAX_TOLERABLE_GAP_DAYS = 7`), preventing a stale cache from being certified indefinitely. | Depend on a trading calendar or apply the relaxation to all callers — extra dependency or unsafe historical behavior. | | **Deterministic DH-904 backoff `[2,5,10]s`** | Predictable, testable retry without random jitter; raises after the list is exhausted. | Infinite/exponential random retry — unbounded, flaky tests. | | **Optional wall-clock timeout via worker thread** | The SDK exposes no timeout; a thread + `future.result(timeout)` lets a stuck call not freeze the Streamlit run (Python can't kill it, but `shutdown(wait=False)` moves on). | Block forever — frozen UI. | | **`client=None` cache-only mode fails loudly on fetch** | Chart UI / cleanup can build a loader without creds, but a real fetch attempt raises a clear error not `AttributeError`. | Silent no-op — confusing empty results. | @@ -100,7 +100,8 @@ flowchart TD ## 7. Testing - [`tests/test_dhan_client.py`](../../../tests/test_dhan_client.py) — payload normalization, epoch inference, rate-limit detection, "no data". -- [`tests/test_daily_data_loader.py`](../../../tests/test_daily_data_loader.py) — cache hit/miss, incremental/backfill statuses, `.checked` marker, retries, circuit breaker, cleanup, streaming. +- [`tests/test_daily_data_loader.py`](../../../tests/test_daily_data_loader.py) — cache hit/miss, conservative direct historical coverage, explicit sequential/parallel scanner tail opt-in, seven-day `.checked` marker bound, incremental/backfill statuses, retries, circuit breaker, cleanup, streaming. +- [`tests/test_candle_cache_write_paths.py`](../../../tests/test_candle_cache_write_paths.py) — all six cache-write paths, including malformed-cache recovery and incremental exact-row normalization while preserving conflicting same-date rows for data-quality reporting. ## 8. Extension points diff --git a/tests/test_candle_cache_write_paths.py b/tests/test_candle_cache_write_paths.py index 4daf02a..c9af500 100644 --- a/tests/test_candle_cache_write_paths.py +++ b/tests/test_candle_cache_write_paths.py @@ -117,6 +117,114 @@ def test_ensure_daily_history_backfill_writes_no_duplicates(tmp_path: Path): _assert_cache_is_clean(loader) +def test_ensure_daily_history_missing_timestamp_cache_recovery_writes_no_duplicates(tmp_path: Path): + """A malformed cache is replaced through the full-download write path.""" + loader = DailyDataLoader( + cast(DhanDataClient, DuplicatingClient()), + cache_dir=tmp_path, + request_delay_seconds=0.0, + ) + pd.DataFrame( + { + "open": [100.0], + "high": [105.0], + "low": [99.0], + "close": [104.0], + "volume": [1_000.0], + } + ).to_parquet(loader.cache_path(ROW["symbol"], ROW["security_id"]), index=False) + + _frame, status = loader.ensure_daily_history(ROW, years_back=1, today=TODAY) + + assert status == "fresh_download" + assert len(_assert_cache_is_clean(loader)) == 9 + + +def test_ensure_daily_history_all_nat_cache_recovery_writes_no_duplicates(tmp_path: Path): + """An all-NaT timestamp column is replaced through the same guarded write path.""" + loader = DailyDataLoader( + cast(DhanDataClient, DuplicatingClient()), + cache_dir=tmp_path, + request_delay_seconds=0.0, + ) + pd.DataFrame( + { + # Explicit dtype keeps the malformed fixture all-NaT while making + # the intended datetime64 column clear to static type checking. + "timestamp": pd.Series([None], dtype="datetime64[ns]"), + "open": [100.0], + "high": [105.0], + "low": [99.0], + "close": [104.0], + "volume": [1_000.0], + } + ).to_parquet(loader.cache_path(ROW["symbol"], ROW["security_id"]), index=False) + + _frame, status = loader.ensure_daily_history(ROW, years_back=1, today=TODAY) + + assert status == "fresh_download" + assert len(_assert_cache_is_clean(loader)) == 9 + + +def test_incremental_merge_drops_exact_rows_but_preserves_conflicting_rows(tmp_path: Path): + """Incremental storage removes only exact rows across all six canonical columns. + + The new response includes an exact repeated current candle and a different + correction for the previous cached date. The write must remove the repeated + row but retain both values for the conflicting date so DATA-001 can report it. + """ + + class CanonicalButDuplicatingClient: + def fetch_daily_candles(self, **_kwargs) -> pd.DataFrame: + repeated_current = { + "timestamp": pd.Timestamp(TODAY), + "open": 100.0, + "high": 105.0, + "low": 99.0, + "close": 104.0, + "volume": 1_000.0, + } + return pd.DataFrame( + [ + { + "timestamp": pd.Timestamp(TODAY - timedelta(days=1)), + "open": 120.0, + "high": 125.0, + "low": 119.0, + "close": 124.0, + "volume": 2_000.0, + }, + repeated_current, + repeated_current, + ] + ) + + loader = DailyDataLoader( + cast(DhanDataClient, CanonicalButDuplicatingClient()), + cache_dir=tmp_path, + request_delay_seconds=0.0, + ) + pd.DataFrame( + { + "timestamp": pd.to_datetime([date(2025, 8, 24), TODAY - timedelta(days=1)]), + "open": [100.0, 100.0], + "high": [105.0, 105.0], + "low": [99.0, 99.0], + "close": [104.0, 104.0], + "volume": [1_000.0, 1_000.0], + } + ).to_parquet(loader.cache_path(ROW["symbol"], ROW["security_id"]), index=False) + + _frame, status = loader.ensure_daily_history(ROW, years_back=1, today=TODAY) + + assert status == "incremental" + stored = pd.read_parquet(loader.cache_path(ROW["symbol"], ROW["security_id"])) + report = validate_candles(stored, symbol="DEMO", expected_latest_date=TODAY) + assert "DUPLICATE_DATE" in {finding.code for finding in report.findings} + assert len(stored.loc[stored["timestamp"].eq(pd.Timestamp(TODAY))]) == 1 + assert len(stored.loc[stored["timestamp"].eq(pd.Timestamp(TODAY - timedelta(days=1)))]) == 2 + + def test_a_conflicting_bar_still_reaches_the_cache_to_be_reported(tmp_path: Path): """The guard must not become a silent price-picker. diff --git a/tests/test_daily_data_loader.py b/tests/test_daily_data_loader.py index bf1d14e..11b8232 100644 --- a/tests/test_daily_data_loader.py +++ b/tests/test_daily_data_loader.py @@ -1134,6 +1134,49 @@ def test_fetch_workers_setting_clamps_and_defaults(monkeypatch, tmp_path): # --------------------------------------------------------------------------- +def test_direct_history_missing_weekday_requested_end_fetches(tmp_path): + """Historical callers must not inherit the scanner's current-session tail rule. + + The cache reaches Monday, but a direct historical request includes Tuesday. + Tuesday is a weekday, so only the caller that is explicitly running the + current scanner session may accept its absence as unpublished. + """ + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _covering_cache(loader, first_date=date(2016, 8, 1), last_date=date(2026, 8, 24)) + + _frame, from_cache = loader.get_daily_history( + {"symbol": "DEMO", "security_id": "1"}, + start_date=date(2016, 8, 24), + end_date=date(2026, 8, 25), + ) + + assert from_cache is False + assert client.calls == 1 + + +def test_direct_history_ignores_checked_marker_for_missing_weekday_requested_end(tmp_path): + """A historical request needs the candle even when a later marker exists. + + A `.checked` sidecar proves only that the scanner's prefetch saw no row for + the current-session tail. It cannot authorise a different direct caller to + treat a missing weekday as historical truth. + """ + client = FakeDhanClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _covering_cache(loader, first_date=date(2016, 8, 1), last_date=date(2026, 8, 24)) + loader.checked_path("DEMO", "1").write_text("2026-08-26", encoding="utf-8") + + _frame, from_cache = loader.get_daily_history( + {"symbol": "DEMO", "security_id": "1"}, + start_date=date(2016, 8, 24), + end_date=date(2026, 8, 25), + ) + + assert from_cache is False + assert client.calls == 1 + + def _covering_cache(loader, *, last_date, first_date): """Write a cache spanning [first_date, last_date] for the standard instrument.""" frame = pd.DataFrame( @@ -1165,6 +1208,7 @@ def test_cache_short_of_today_by_a_weekend_is_still_a_hit(tmp_path): {"symbol": "DEMO", "security_id": "1"}, start_date=date(2016, 8, 24), end_date=today, + allow_unpublished_tail=True, ) assert from_cache is True @@ -1181,6 +1225,7 @@ def test_cache_far_behind_the_requested_end_is_still_a_miss(tmp_path): {"symbol": "DEMO", "security_id": "1"}, start_date=date(2016, 8, 24), end_date=date(2026, 8, 24), + allow_unpublished_tail=True, ) assert from_cache is False @@ -1200,6 +1245,7 @@ def test_cache_missing_early_history_is_a_miss_however_current_it_is(tmp_path): {"symbol": "DEMO", "security_id": "1"}, start_date=date(2016, 8, 24), end_date=date(2026, 8, 24), + allow_unpublished_tail=True, ) assert from_cache is False @@ -1223,11 +1269,12 @@ def _cache_ending(loader, last_date, *, first_date=date(2016, 8, 1)): ).to_parquet(loader.cache_path("DEMO", "1"), index=False) -def _is_hit(loader, *, end_date, start_date=date(2016, 8, 24)): +def _is_hit(loader, *, end_date, start_date=date(2016, 8, 24), allow_unpublished_tail=False): _frame, from_cache = loader.get_daily_history( {"symbol": "DEMO", "security_id": "1"}, start_date=start_date, end_date=end_date, + allow_unpublished_tail=allow_unpublished_tail, ) return from_cache @@ -1242,7 +1289,7 @@ def test_a_weekend_only_gap_is_a_hit(tmp_path): loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) _cache_ending(loader, date(2026, 8, 21)) # Friday - assert _is_hit(loader, end_date=date(2026, 8, 24)) is True # Monday + assert _is_hit(loader, end_date=date(2026, 8, 24), allow_unpublished_tail=True) is True # Monday assert client.calls == 0 @@ -1256,7 +1303,7 @@ def test_a_missing_published_weekday_is_a_miss(tmp_path): loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) _cache_ending(loader, date(2026, 8, 20)) # Thursday - assert _is_hit(loader, end_date=date(2026, 8, 24)) is False # Monday + assert _is_hit(loader, end_date=date(2026, 8, 24), allow_unpublished_tail=True) is False # Monday def test_todays_own_bar_may_be_unpublished(tmp_path): @@ -1265,7 +1312,7 @@ def test_todays_own_bar_may_be_unpublished(tmp_path): loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) _cache_ending(loader, date(2026, 8, 24)) # Monday - assert _is_hit(loader, end_date=date(2026, 8, 25)) is True # Tuesday + assert _is_hit(loader, end_date=date(2026, 8, 25), allow_unpublished_tail=True) is True # Tuesday assert client.calls == 0 @@ -1275,7 +1322,7 @@ def test_a_skipped_midweek_day_is_a_miss(tmp_path): loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) _cache_ending(loader, date(2026, 8, 21)) # Friday - assert _is_hit(loader, end_date=date(2026, 8, 25)) is False # Tuesday + assert _is_hit(loader, end_date=date(2026, 8, 25), allow_unpublished_tail=True) is False # Tuesday def test_a_long_stale_cache_is_a_miss(tmp_path): @@ -1283,7 +1330,7 @@ def test_a_long_stale_cache_is_a_miss(tmp_path): loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) _cache_ending(loader, date(2026, 8, 3)) - assert _is_hit(loader, end_date=date(2026, 8, 24)) is False + assert _is_hit(loader, end_date=date(2026, 8, 24), allow_unpublished_tail=True) is False def test_a_market_holiday_is_a_hit_when_the_prefetch_already_asked(tmp_path): @@ -1297,7 +1344,7 @@ def test_a_market_holiday_is_a_hit_when_the_prefetch_already_asked(tmp_path): _cache_ending(loader, date(2026, 8, 21)) # Friday loader.checked_path("DEMO", "1").write_text("2026-08-25", encoding="utf-8") - assert _is_hit(loader, end_date=date(2026, 8, 25)) is True # holiday Monday behind us + assert _is_hit(loader, end_date=date(2026, 8, 25), allow_unpublished_tail=True) is True # holiday Monday behind us assert client.calls == 0 @@ -1307,7 +1354,7 @@ def test_the_same_holiday_gap_is_a_miss_without_that_evidence(tmp_path): loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) _cache_ending(loader, date(2026, 8, 21)) - assert _is_hit(loader, end_date=date(2026, 8, 25)) is False + assert _is_hit(loader, end_date=date(2026, 8, 25), allow_unpublished_tail=True) is False def test_a_stale_checked_marker_does_not_vouch_for_a_later_request(tmp_path): @@ -1317,7 +1364,7 @@ def test_a_stale_checked_marker_does_not_vouch_for_a_later_request(tmp_path): _cache_ending(loader, date(2026, 8, 21)) loader.checked_path("DEMO", "1").write_text("2026-08-22", encoding="utf-8") - assert _is_hit(loader, end_date=date(2026, 8, 25)) is False + assert _is_hit(loader, end_date=date(2026, 8, 25), allow_unpublished_tail=True) is False def test_a_checked_marker_cannot_vouch_for_an_unboundedly_stale_cache(tmp_path): @@ -1333,7 +1380,7 @@ def test_a_checked_marker_cannot_vouch_for_an_unboundedly_stale_cache(tmp_path): _cache_ending(loader, date(2026, 7, 20)) # a month behind loader.checked_path("DEMO", "1").write_text("2026-08-25", encoding="utf-8") - assert _is_hit(loader, end_date=date(2026, 8, 25)) is False + assert _is_hit(loader, end_date=date(2026, 8, 25), allow_unpublished_tail=True) is False def test_a_checked_marker_still_rescues_a_short_holiday_gap(tmp_path): @@ -1343,5 +1390,47 @@ def test_a_checked_marker_still_rescues_a_short_holiday_gap(tmp_path): _cache_ending(loader, date(2026, 8, 21)) # Friday loader.checked_path("DEMO", "1").write_text("2026-08-25", encoding="utf-8") - assert _is_hit(loader, end_date=date(2026, 8, 25)) is True + assert _is_hit(loader, end_date=date(2026, 8, 25), allow_unpublished_tail=True) is True assert client.calls == 0 + + +def _assert_universe_current_session_tail_is_a_cache_hit(tmp_path, *, fetch_workers): + """Run the real scanner iterator with a Friday cache through Monday.""" + client = FakeDhanClient() + loader = DailyDataLoader( + client, + cache_dir=tmp_path, + request_delay_seconds=0.0, + fetch_workers=fetch_workers, + ) + pd.DataFrame( + { + "timestamp": pd.to_datetime([date(2016, 8, 1), date(2026, 8, 21)]), + "open": [100.0, 101.0], + "high": [105.0, 106.0], + "low": [99.0, 100.0], + "close": [104.0, 105.0], + "volume": [1_000.0, 1_100.0], + } + ).to_parquet(loader.cache_path("RELIANCE", "2885"), index=False) + + items = list( + loader.iter_universe_history( + mapped_universe(), + start_date=date(2016, 8, 24), + end_date=date(2026, 8, 24), + ) + ) + + assert [(item.symbol, item.from_cache) for item in items] == [("RELIANCE", True)] + assert client.calls == 0 + + +def test_sequential_universe_loading_allows_the_current_session_tail(tmp_path): + """The sequential scanner path explicitly authorises Friday-to-Monday tail reuse.""" + _assert_universe_current_session_tail_is_a_cache_hit(tmp_path, fetch_workers=1) + + +def test_parallel_universe_loading_allows_the_current_session_tail(tmp_path): + """The parallel scanner path grants the same narrow tail authority to workers.""" + _assert_universe_current_session_tail_is_a_cache_hit(tmp_path, fetch_workers=2) From 3e0104f69eeb7020f2d15c5138b7470413dc8f95 Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Fri, 4 Sep 2026 18:28:27 +0530 Subject: [PATCH 6/6] fix(DATA-004): harden vendor earliest evidence Co-authored-by: Codex --- backend/daily_data_loader.py | 156 ++++++++++-- .../components/data-acquisition.md | 32 +++ .../test_daily_data_loader_vendor_earliest.py | 222 +++++++++++++++++- 3 files changed, 387 insertions(+), 23 deletions(-) diff --git a/backend/daily_data_loader.py b/backend/daily_data_loader.py index 92a8545..d2bdc98 100644 --- a/backend/daily_data_loader.py +++ b/backend/daily_data_loader.py @@ -57,6 +57,79 @@ VENDOR_EARLIEST_RECHECK_DAYS = 30 +@dataclass(frozen=True) +class _VendorEarliestEvidence: + """Validated `.firstbar` evidence that bounds a vendor's history. + + Beginner note: + This small immutable object separates untrusted JSON on disk from the dates + the cache-coverage decision is allowed to trust. Once constructed, no caller + can accidentally change one date and leave the chronology inconsistent. + """ + + requested_from: date + earliest_available: date + recorded_on: date + + +def _read_vendor_earliest_evidence(path: Path) -> _VendorEarliestEvidence | None: + """Read one coherent `.firstbar` sidecar, or return no evidence. + + The sidecar is deliberately fail-open: malformed content cannot certify an + incomplete cache, so callers treat it as absent and refetch if necessary. + Its three dates must use canonical ``YYYY-MM-DD`` form and establish the + exact chronology ``requested_from < earliest_available <= recorded_on``. + Unknown JSON object fields are ignored to allow future metadata additions. + + Args: + path: Sidecar path next to the daily-cache parquet file. + + Returns: + Immutable evidence when all required fields are valid; otherwise ``None``. + + Beginner note: + ``date.fromisoformat`` also accepts compact and ISO-week strings. Checking + ``isoformat()`` after parsing prevents those alternate spellings from becoming + an undocumented on-disk format that future readers might interpret differently. + """ + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError): + return None + if not isinstance(payload, dict): + return None + + requested_from = payload.get("requested_from") + earliest_available = payload.get("earliest_available") + recorded_on = payload.get("recorded_on") + if ( + not isinstance(requested_from, str) + or not isinstance(earliest_available, str) + or not isinstance(recorded_on, str) + ): + return None + + try: + parsed_requested_from = date.fromisoformat(requested_from) + parsed_earliest_available = date.fromisoformat(earliest_available) + parsed_recorded_on = date.fromisoformat(recorded_on) + except ValueError: + return None + if ( + parsed_requested_from.isoformat() != requested_from + or parsed_earliest_available.isoformat() != earliest_available + or parsed_recorded_on.isoformat() != recorded_on + ): + return None + if not parsed_requested_from < parsed_earliest_available <= parsed_recorded_on: + return None + return _VendorEarliestEvidence( + requested_from=parsed_requested_from, + earliest_available=parsed_earliest_available, + recorded_on=parsed_recorded_on, + ) + + def history_start_date( years_back: int = DEFAULT_HISTORY_YEARS_BACK, today: date | None = None ) -> date: @@ -235,15 +308,16 @@ def _cache_reaches_back_far_enough( ``vendor_earliest`` is the earliest bar the vendor actually served for a probe that reached at least as far back as this request (see - ``DailyDataLoader._vendor_earliest_for``). When the cache already starts at or - before that bar, it is as complete as it can ever be, so asking again is pure - waste. ``None`` means we have no such evidence and the strict rule applies — - which keeps an interrupted prefetch's partial file being refetched, since - there the vendor genuinely does have the missing years. + ``DailyDataLoader._vendor_earliest_for``). It must match the cache's literal + first date: accepting an earlier or later cache start would let contradictory + evidence certify an incomplete or corrupted cache. ``None`` means we have no + such evidence and the strict rule applies — which keeps an interrupted + prefetch's partial file being refetched, since there the vendor genuinely has + the missing years. """ if first_date <= requested_start: return True - return vendor_earliest is not None and first_date <= vendor_earliest + return vendor_earliest is not None and first_date == vendor_earliest def _date_bounds(candles: pd.DataFrame) -> tuple[date | None, date | None]: @@ -483,22 +557,29 @@ def _vendor_earliest_for( Any read or parse problem returns ``None``, so a corrupt marker can only ever cost an extra request — never hide history that really is missing. + + Args: + symbol: Instrument symbol used to locate the sidecar. + security_id: Vendor identifier paired with the symbol in cache paths. + requested_start: Earliest date the current caller needs covered. + + Returns: + The qualifying earliest available date, or ``None`` for no authority. + + Beginner note: + A marker that is too new in the future is no more trustworthy than a stale + one: both have a clock relationship that cannot describe a completed + vendor request, so the cache takes the safe (refetch) path. """ - path = self.first_bar_path(symbol, security_id) - if not path.exists(): - return None - try: - payload = json.loads(path.read_text(encoding="utf-8")) - requested_from = _coerce_date(str(payload["requested_from"])) - earliest_available = _coerce_date(str(payload["earliest_available"])) - recorded_on = _coerce_date(str(payload["recorded_on"])) - except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError): + evidence = _read_vendor_earliest_evidence(self.first_bar_path(symbol, security_id)) + if evidence is None: return None - if (self.today_func() - recorded_on).days >= VENDOR_EARLIEST_RECHECK_DAYS: + age_days = (self.today_func() - evidence.recorded_on).days + if age_days < 0 or age_days >= VENDOR_EARLIEST_RECHECK_DAYS: return None - if requested_from > requested_start: + if evidence.requested_from > requested_start: return None - return earliest_available + return evidence.earliest_available def _write_vendor_earliest( self, @@ -535,21 +616,52 @@ def _record_vendor_earliest( ) -> None: """Record the vendor's earliest bar when it fell short of what we asked for. - Called after any full-window download. A response that *does* reach the - requested start teaches us nothing worth storing, and an empty response is - no evidence at all, so both are skipped. + Called after any full-window download. Empty or invalid frames are + inconclusive and leave an existing marker untouched. A shallower probe + cannot replace or renew fresh deeper evidence. Expired or future-dated + evidence is not authoritative and may be replaced. An equally + deep/deeper response that reaches the requested start invalidates its + old marker; otherwise a later first bar becomes new evidence. ``recorded_on`` is stamped from the injected wall clock rather than from the caller's requested window, so the marker ages in real time whatever range was asked for. + + Args: + symbol: Instrument symbol used to locate the sidecar. + security_id: Vendor identifier paired with the symbol in cache paths. + requested_from: Inclusive beginning of the completed vendor probe. + candles: Raw non-empty response used to learn its first bar. + + Beginner note: + The depth comparison is between what the vendor was asked, not what it + returned. A late-listed stock can return the same first bar for many + windows, but only the request that began furthest back proves the stronger + "nothing exists earlier" statement. """ + start = _coerce_date(requested_from) if candles.empty: return first_date, _last_date = _date_bounds(candles) if first_date is None: return - start = _coerce_date(requested_from) + + path = self.first_bar_path(symbol, security_id) + existing = _read_vendor_earliest_evidence(path) + # A probe beginning later asks less of the vendor. Preserve evidence + # collected by a deeper request only while its wall-clock TTL is valid; + # expired/future-dated evidence is not authoritative and must not block + # this fresh answer from replacing it. + if existing is not None and existing.requested_from < start: + age_days = (self.today_func() - existing.recorded_on).days + if 0 <= age_days < VENDOR_EARLIEST_RECHECK_DAYS: + return if first_date <= start: + try: + path.unlink(missing_ok=True) + except OSError: + # The marker is optional, so a locked sidecar must not fail a fetch. + logger.warning("Could not remove obsolete daily-cache first-bar marker for %s", symbol) return self._write_vendor_earliest( symbol, diff --git a/docs/architecture/components/data-acquisition.md b/docs/architecture/components/data-acquisition.md index 20e39a1..bdeecfd 100644 --- a/docs/architecture/components/data-acquisition.md +++ b/docs/architecture/components/data-acquisition.md @@ -67,6 +67,38 @@ flowchart TD | `history_start_date(years_back, today)` | Leap-safe "subtract whole years" (Feb 29 → Feb 28). | | `safe_file_stem(value)` | Path-traversal-safe filename fragment. | +### DATA-004 `.firstbar` earliest-history evidence + +When a vendor request begins before a stock listed, the returned frame begins at +the stock's earliest available candle. `DailyDataLoader` stores that answer next +to the parquet as `_.firstbar`, a JSON object with canonical +`requested_from`, `earliest_available`, and `recorded_on` dates. The public cache +contract remains strict: a request is front-complete only when the parquet first +date literally reaches `requested_start`, or a fresh qualifying `.firstbar` +exists **and its `earliest_available` exactly equals the parquet first date**. +The exact binding prevents contradictory cache/sidecar dates from certifying +history that cannot be proved complete. + +The marker is internal, optional evidence—not a user input or a database record. +Its JSON reader accepts only a JSON object with string `YYYY-MM-DD` fields and +the chronology `requested_from < earliest_available <= recorded_on`; extra fields +are ignored for forwards compatibility. Its 30-day TTL is measured against the +injected wall clock, not the requested data window. Future, stale (age 30 days or +more), malformed, noncanonical, or shallower-than-request evidence is ignored and +therefore causes a safe refetch. A shallow probe preserves a deeper marker only +while that marker is fresh; expired or future-dated evidence is replaced by the +new probe, without renewing a fresh marker's timestamp. An equally deep/deeper +non-empty response that still starts late replaces the marker; one that reaches +the requested start removes the now-obsolete marker best-effort. Empty/invalid +frames leave prior evidence unchanged. + +`.firstbar` shares the cache lifecycle: it travels with its parquet and +`cleanup_stale_cache_files()` removes it when orphaned or when the associated +cache ages out. `tests/test_daily_data_loader_vendor_earliest.py` covers marker +creation, strict parsing/chronology, exact cache binding, wall-clock TTL, +shallower/equally-deep update rules, cleanup, and the request-bounded late-listing +vendor fixture. + ## 4. Key design decisions & trade-offs | Decision | Rationale | Alternative rejected | diff --git a/tests/test_daily_data_loader_vendor_earliest.py b/tests/test_daily_data_loader_vendor_earliest.py index 9ec10f1..f296e17 100644 --- a/tests/test_daily_data_loader_vendor_earliest.py +++ b/tests/test_daily_data_loader_vendor_earliest.py @@ -28,6 +28,7 @@ from typing import cast import pandas as pd +import pytest from backend.daily_data_loader import DailyDataLoader from backend.dhan_client import DhanDataClient @@ -74,7 +75,10 @@ def fetch_daily_candles(self, *, from_date, to_date, **_kwargs) -> pd.DataFrame: # Monthly bars keep the fixture small; only the date bounds matter. The # listing date itself is forced in because "MS" snaps to month starts, # and the first bar being exactly the listing date is the whole point. - dates = _month_series(self.listed_on, self.through) + first_returned_date = max(self.listed_on, from_date) + if first_returned_date > self.through: + return pd.DataFrame() + dates = _month_series(first_returned_date, self.through) return pd.DataFrame( { "timestamp": dates, @@ -379,3 +383,219 @@ def test_a_recent_marker_is_honoured_against_the_clock(tmp_path: Path): assert loader._vendor_earliest_for( ROW["symbol"], ROW["security_id"], HISTORY_START ) == LISTED_ON + + +# --------------------------------------------------------------------------- +# Marker hardening: evidence is useful only when it is self-consistent +# --------------------------------------------------------------------------- + + +def test_a_future_dated_marker_fails_open_to_a_backfill(tmp_path: Path): + """Future evidence must not certify a cache before that day has arrived. + + Beginner note: + The marker is only an optimisation. Treating a future stamp as fresh would + let a clock error or manually edited sidecar hide missing history, whereas + rejecting it merely asks the vendor again. + """ + client = ListedLateClient() + loader = _loader(tmp_path, client) + _write_cache(loader, first=LISTED_ON, last=TODAY) + loader._write_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=HISTORY_START, + earliest_available=LISTED_ON, recorded_on=TODAY + timedelta(days=1), + ) + + _frame, from_cache = loader.get_daily_history( + ROW, start_date=HISTORY_START, end_date=TODAY + ) + + assert from_cache is False + assert client.calls == [(HISTORY_START, TODAY)] + + +def test_invalid_utf8_marker_fails_open_to_a_backfill(tmp_path: Path): + """Undecodable marker bytes must trigger a safe refetch, not escape parsing.""" + client = ListedLateClient() + loader = _loader(tmp_path, client) + _write_cache(loader, first=LISTED_ON, last=TODAY) + _marker(loader).write_bytes(b"\xff") + + _frame, from_cache = loader.get_daily_history( + ROW, start_date=HISTORY_START, end_date=TODAY + ) + + assert from_cache is False + assert client.calls == [(HISTORY_START, TODAY)] + + +def test_a_marker_must_match_the_cached_first_bar_to_avoid_a_backfill(tmp_path: Path): + """Contradictory cache and marker starts must trigger a refetch.""" + client = ListedLateClient() + loader = _loader(tmp_path, client) + _write_cache(loader, first=LISTED_ON - timedelta(days=1), last=TODAY) + loader._write_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=HISTORY_START, + earliest_available=LISTED_ON, recorded_on=TODAY, + ) + + _frame, from_cache = loader.get_daily_history( + ROW, start_date=HISTORY_START, end_date=TODAY + ) + + assert from_cache is False + assert client.calls == [(HISTORY_START, TODAY)] + + +def test_noncanonical_non_string_or_impossible_marker_fields_are_rejected(tmp_path: Path): + """The parser accepts only a coherent object with canonical ISO dates.""" + loader = _loader(tmp_path, ListedLateClient()) + invalid_payloads = [ + { + "requested_from": 20160824, + "earliest_available": LISTED_ON.isoformat(), + "recorded_on": TODAY.isoformat(), + }, + { + "requested_from": "20160824", + "earliest_available": LISTED_ON.isoformat(), + "recorded_on": TODAY.isoformat(), + }, + { + "requested_from": HISTORY_START.isoformat(), + "earliest_available": HISTORY_START.isoformat(), + "recorded_on": TODAY.isoformat(), + }, + { + "requested_from": HISTORY_START.isoformat(), + "earliest_available": (TODAY + timedelta(days=1)).isoformat(), + "recorded_on": TODAY.isoformat(), + }, + ] + + for payload in invalid_payloads: + _marker(loader).write_text(json.dumps(payload), encoding="utf-8") + + assert loader._vendor_earliest_for( + ROW["symbol"], ROW["security_id"], HISTORY_START + ) is None + + +def test_a_fresh_deeper_marker_survives_a_shallower_qualifying_probe(tmp_path: Path): + """A shallower response cannot replace or renew fresh stronger evidence.""" + loader = _loader(tmp_path, ListedLateClient()) + loader._write_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=HISTORY_START, + earliest_available=LISTED_ON, recorded_on=TODAY - timedelta(days=2), + ) + original = _marker(loader).read_text(encoding="utf-8") + shallow_start = HISTORY_START + timedelta(days=100) + shallow_candles = pd.DataFrame( + {"timestamp": _month_series(date(2020, 1, 1), TODAY)} + ) + + loader._record_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=shallow_start, + candles=shallow_candles, + ) + + assert _marker(loader).read_text(encoding="utf-8") == original + + +@pytest.mark.parametrize( + "recorded_on", + [TODAY - timedelta(days=30), TODAY + timedelta(days=1)], + ids=["expired", "future-dated"], +) +def test_expired_or_future_deeper_marker_does_not_block_a_shallower_probe( + tmp_path: Path, recorded_on: date +): + """Non-fresh deeper evidence must not suppress a new shallower answer. + + Beginner note: + A deeper marker is stronger only while its 30-day wall-clock TTL is valid. + Once it is expired or dated in the future, keeping it would let old or + clock-skewed evidence block a new probe forever. The new probe therefore + replaces it, just as it would when no marker existed. + """ + loader = _loader(tmp_path, ListedLateClient()) + loader._write_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=HISTORY_START, + earliest_available=LISTED_ON, recorded_on=recorded_on, + ) + shallow_start = HISTORY_START + timedelta(days=100) + shallow_candles = pd.DataFrame( + {"timestamp": _month_series(date(2020, 1, 1), TODAY)} + ) + + loader._record_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=shallow_start, + candles=shallow_candles, + ) + + payload = json.loads(_marker(loader).read_text(encoding="utf-8")) + assert payload["requested_from"] == shallow_start.isoformat() + assert payload["earliest_available"] == date(2020, 1, 1).isoformat() + assert payload["recorded_on"] == TODAY.isoformat() + + +def test_a_full_equally_deep_probe_removes_obsolete_marker(tmp_path: Path): + """A response reaching the requested start makes its old marker obsolete.""" + loader = _loader(tmp_path, ListedLateClient()) + loader._write_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=HISTORY_START, + earliest_available=LISTED_ON, recorded_on=TODAY, + ) + complete_candles = pd.DataFrame( + {"timestamp": _month_series(HISTORY_START, TODAY)} + ) + + loader._record_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=HISTORY_START, + candles=complete_candles, + ) + + assert not _marker(loader).exists() + + +def test_marker_age_29_days_is_fresh_but_age_30_days_is_rejected(tmp_path: Path): + """The 30-day wall-clock TTL is inclusive at its expiration boundary.""" + loader = _loader(tmp_path, ListedLateClient()) + loader._write_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=HISTORY_START, + earliest_available=LISTED_ON, recorded_on=TODAY - timedelta(days=29), + ) + + assert loader._vendor_earliest_for( + ROW["symbol"], ROW["security_id"], HISTORY_START + ) == LISTED_ON + + loader._write_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=HISTORY_START, + earliest_available=LISTED_ON, recorded_on=TODAY - timedelta(days=30), + ) + + assert loader._vendor_earliest_for( + ROW["symbol"], ROW["security_id"], HISTORY_START + ) is None + + +def test_listed_late_client_respects_the_requested_start_and_end_dates() -> None: + """The fixture must model a bounded vendor response, not a full cache read.""" + client = ListedLateClient() + requested_start = TODAY - timedelta(days=10) + frame = client.fetch_daily_candles(from_date=requested_start, to_date=TODAY) + + assert frame["timestamp"].min().date() == requested_start + assert frame["timestamp"].max().date() == TODAY + + +def test_listed_late_client_returns_no_rows_after_its_configured_through_date() -> None: + """A request beginning after the fixture's vendor horizon is empty evidence.""" + client = ListedLateClient() + + frame = client.fetch_daily_candles( + from_date=TODAY + timedelta(days=1), to_date=TODAY + timedelta(days=2) + ) + + assert frame.empty