diff --git a/backend/daily_data_loader.py b/backend/daily_data_loader.py index 2a7977c..d2bdc98 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 @@ -48,6 +49,86 @@ # ``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 + + +@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 @@ -103,6 +184,142 @@ 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, + *, + 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, 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: + # 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 + + +def _cache_covers_range( + first_date: date | None, + last_date: date | None, + requested_start: date, + requested_end: date, + *, + vendor_earliest: date | None = None, + checked_through: date | None = None, + allow_unpublished_tail: bool = False, +) -> bool: + """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. + + 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. + vendor_earliest: Earliest bar a qualifying vendor probe returned. + 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 + # Front and back are judged by separate evidence: how far the vendor's history + # goes (DATA-004) and which recent days it has actually published (DATA-003). + if not _cache_reaches_back_far_enough(first_date, requested_start, vendor_earliest): + return False + 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 + # 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 + + +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``). 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 + + def _date_bounds(candles: pd.DataFrame) -> tuple[date | None, date | None]: """Return the first/last valid candle dates in a cached frame. @@ -217,6 +434,7 @@ def __init__( max_consecutive_failures: int | None = None, sleep_func: Callable[[float], None] = time.sleep, fetch_workers: int | None = None, + today_func: Callable[[], date] = date.today, ): # The Dhan client is optional so cache-only callers (the legacy-file # cleanup step, the chart UI's `read_cached_history`) can build a loader @@ -242,6 +460,12 @@ def __init__( ) self.max_consecutive_failures = max(0, int(max_consecutive_failures or 0)) self.sleep_func = sleep_func + # Wall clock, injected like sleep_func so tests can pin it. Deliberately + # separate from the ``today`` argument callers pass to describe a DATA + # window: "the date I am asking about" and "the date it is now" are + # different questions, and conflating them let a marker's 30-day expiry be + # judged against a historical request boundary, so it never expired. + self.today_func = today_func # PERF-001: 1 (the default) keeps the long-standing sequential path # byte-identical. Values above 1 fetch with a thread pool while the # shared pacer holds the global inter-request delay. @@ -303,6 +527,150 @@ 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 + ) -> 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`` of **now** + (vendors do backfill occasionally, so the belief expires). Age is + measured against the injected wall clock, never against the requested + window: judging it by a request boundary meant a repeated historical + request always computed an age of zero and the marker never expired. + - 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. + + 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. + """ + evidence = _read_vendor_earliest_evidence(self.first_bar_path(symbol, security_id)) + if evidence is None: + return None + age_days = (self.today_func() - evidence.recorded_on).days + if age_days < 0 or age_days >= VENDOR_EARLIEST_RECHECK_DAYS: + return None + if evidence.requested_from > requested_start: + return None + return evidence.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, + ) -> None: + """Record the vendor's earliest bar when it fell short of what we asked for. + + 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 + + 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, + security_id, + requested_from=start, + earliest_available=first_date, + recorded_on=self.today_func(), + ) + 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. @@ -342,12 +710,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 @@ -364,7 +748,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. # @@ -381,11 +766,21 @@ 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 + vendor_earliest = self._vendor_earliest_for( + symbol, security_id, requested_start + ) + # 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, + vendor_earliest=vendor_earliest, + checked_through=checked_through, + allow_unpublished_tail=allow_unpublished_tail, ): if cached is None: # The footer is only an advisory index. The file can be @@ -393,11 +788,14 @@ def get_daily_history( # 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, + vendor_earliest=vendor_earliest, + checked_through=checked_through, + allow_unpublished_tail=allow_unpublished_tail, ): return self._slice_to_range(cached, start_date, end_date), True @@ -413,6 +811,9 @@ 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 + ) return self._slice_to_range(candles, start_date, end_date), False def fetch_window( @@ -494,6 +895,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 + ) return candles, "fresh_download" cached = pd.read_parquet(path) @@ -509,6 +913,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 + ) return candles, "fresh_download" first_date, last_date = _date_bounds(cached) @@ -525,9 +932,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 + ) 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) + 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. @@ -538,10 +954,15 @@ def ensure_daily_history( from_date=start, to_date=today, ) + self._record_vendor_earliest( + symbol, security_id, requested_from=start, candles=candles + ) 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" @@ -572,7 +993,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) ) @@ -886,7 +1309,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" @@ -902,6 +1331,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: @@ -941,6 +1371,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() @@ -963,6 +1398,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 @@ -1173,17 +1609,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/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/docs/architecture/components/data-acquisition.md b/docs/architecture/components/data-acquisition.md index 5eac3a8..bdeecfd 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). | @@ -67,14 +67,46 @@ 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 | |---|---|---| -| **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 +132,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 new file mode 100644 index 0000000..c9af500 --- /dev/null +++ b/tests/test_candle_cache_write_paths.py @@ -0,0 +1,266 @@ +"""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_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. + + 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..11b8232 100644 --- a/tests/test_daily_data_loader.py +++ b/tests/test_daily_data_loader.py @@ -1127,3 +1127,310 @@ 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 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( + { + "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, + allow_unpublished_tail=True, + ) + + 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), + allow_unpublished_tail=True, + ) + + 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), + allow_unpublished_tail=True, + ) + + 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), 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 + + +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), allow_unpublished_tail=True) 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), allow_unpublished_tail=True) 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), allow_unpublished_tail=True) 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), allow_unpublished_tail=True) 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), allow_unpublished_tail=True) 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), allow_unpublished_tail=True) 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), allow_unpublished_tail=True) 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), allow_unpublished_tail=True) 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), allow_unpublished_tail=True) 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), 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) 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..f296e17 --- /dev/null +++ b/tests/test_daily_data_loader_vendor_earliest.py @@ -0,0 +1,601 @@ +"""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 +import pytest + +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. + 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, + "open": 100.0, + "high": 105.0, + "low": 99.0, + "close": 104.0, + "volume": 1_000.0, + } + ) + + +def _loader( + tmp_path: Path, client: object, *, now: date = TODAY +) -> DailyDataLoader: + """Build a loader with a pinned wall clock. + + ``now`` is the injected clock, which is what the marker's expiry is measured + against. It is deliberately separate from the ``today`` argument these tests + pass to ``ensure_daily_history``: that one describes the *data* window. + """ + # 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, + today_func=lambda: now, + ) + + +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 + + +# --------------------------------------------------------------------------- +# The marker ages in real time, not against the requested window +# --------------------------------------------------------------------------- + + +def test_a_marker_from_a_historical_request_still_expires(tmp_path: Path): + """Codex review, PR #114. + + ``get_daily_history`` used to stamp ``recorded_on`` with the request's own + ``end_date``. Repeating a historical request then computed an age of zero every + time, so the 30-day expiry never fired and a vendor backfill could stay hidden + behind a partial cache indefinitely. + """ + historical_end = date(2020, 6, 30) + client = ListedLateClient(listed_on=LISTED_ON, through=historical_end) + # The clock is far ahead of the window being requested. + loader = _loader(tmp_path, client, now=TODAY) + + loader.get_daily_history(ROW, start_date=HISTORY_START, end_date=historical_end) + + payload = json.loads(_marker(loader).read_text(encoding="utf-8")) + # Stamped from the clock, not from the 2020 request boundary. + assert date.fromisoformat(payload["recorded_on"]) == TODAY + # And it is therefore already long expired relative to when it was "recorded". + stale_loader = _loader(tmp_path, client, now=TODAY + timedelta(days=31)) + assert stale_loader._vendor_earliest_for( + ROW["symbol"], ROW["security_id"], HISTORY_START + ) is None + + +def test_a_future_dated_request_does_not_extend_a_markers_life(tmp_path: Path): + """A request reaching into the future must not keep a stale belief alive.""" + client = ListedLateClient() + loader = _loader(tmp_path, client, now=TODAY) + loader._write_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=HISTORY_START, + earliest_available=LISTED_ON, recorded_on=TODAY - timedelta(days=45), + ) + + # Asking about a window that ends next year must not make a 45-day-old + # marker look fresh. + assert loader._vendor_earliest_for( + ROW["symbol"], ROW["security_id"], HISTORY_START + ) is None + + +def test_a_recent_marker_is_honoured_against_the_clock(tmp_path: Path): + """The positive case, so the expiry test above cannot pass vacuously.""" + loader = _loader(tmp_path, ListedLateClient(), now=TODAY) + loader._write_vendor_earliest( + ROW["symbol"], ROW["security_id"], requested_from=HISTORY_START, + earliest_available=LISTED_ON, recorded_on=TODAY - timedelta(days=5), + ) + + 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 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}