From 058e1f3f1ee9ee5c9cfa2c36f4cb1db89188ec77 Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Sat, 11 Jul 2026 01:28:33 +0530 Subject: [PATCH 1/2] perf(PERF-002): answer cache-coverage decisions from parquet footer stats The prefetch discards the candle frame (it only needs the status), yet the "is this cache fresh?" verdict decompressed the whole multi-year parquet just to learn its first/last dates. Parquet footers already carry per-row- group min/max statistics, and backend/health.py has read them for its cache snapshot since OBS-002. This generalizes that trick: - backend/parquet_stats.py: timestamp_bounds(path) -> (first, last) dates from footer statistics ONLY; (None, None) whenever the footer cannot answer authoritatively (missing stats/column, all-null, corrupt) - never raises, so the caller's full-read fallback is the error handler. - daily_data_loader._ensure_one_row: footer-provably-fresh caches yield "fresh" without any pandas I/O; anything else takes the unchanged ensure_daily_history slow path (public contract untouched). - daily_data_loader.get_daily_history: the coverage check reads the footer first; an insufficient cache now goes straight to the Dhan fetch without decompressing the frame it was about to discard. The covered path still reads the frame (it returns the candles). Statless files fall back to the original full read, so no former cache hit can become a miss. Benchmark (100 synthetic 10-year caches, pandas-written, warm, monotonic): OLD full-read + _date_bounds : 861.0 ms (8.61 ms/file) NEW footer timestamp_bounds : 72.0 ms (0.72 ms/file) ~12x On a ~500-symbol universe that removes seconds of pure decision overhead from every app relaunch. health.py deliberately NOT rewired: its exception handling feeds the unreadable-file counter, which this helper's never-raise contract would silently change. Tests: 9 unit tests for timestamp_bounds (multi-row-group, statless, all-NaT, mistyped column, corrupt file) + 6 loader integration tests that prove the fast paths do no pandas I/O (read_parquet monkeypatched to raise) and that statless files keep their old behavior. The pre-existing loader suite (33 tests) passes unmodified. Gates: 1,401 passed, coverage 88.19% (floor 87); pre-commit validate, compileall, ruff, mypy (120 files), bandit, pip-audit all clean. Co-Authored-By: Claude Fable 5 --- backend/daily_data_loader.py | 51 ++++- backend/parquet_stats.py | 98 +++++++++ tests/test_daily_data_loader_footer_stats.py | 200 +++++++++++++++++++ tests/test_parquet_stats.py | 97 +++++++++ 4 files changed, 444 insertions(+), 2 deletions(-) create mode 100644 backend/parquet_stats.py create mode 100644 tests/test_daily_data_loader_footer_stats.py create mode 100644 tests/test_parquet_stats.py diff --git a/backend/daily_data_loader.py b/backend/daily_data_loader.py index d09ab24..aeafbdb 100644 --- a/backend/daily_data_loader.py +++ b/backend/daily_data_loader.py @@ -34,6 +34,7 @@ EVENT_EXTERNAL_API_FAILED, log_event, ) +from backend.parquet_stats import timestamp_bounds from backend.security import redact_text # Module-level logger. Streamlit captures stderr, so logger output appears in the @@ -366,8 +367,18 @@ def get_daily_history( # Cache hit only when the file covers the entire requested range. # A partial parquet is common after interrupted prefetches; slicing # it would silently run long-lookback screeners on too little data. - cached = pd.read_parquet(path) - first_date, last_date = _date_bounds(cached) + # + # PERF-002: ask the Parquet footer for the bounds first. When the + # file does NOT cover the range, this skips decompressing a + # multi-year frame that would be thrown away for a Dhan refetch. + # A footer that cannot answer (missing statistics, odd writer) + # falls back to the original full read, so no file that used to + # count as a cache hit can become a miss. + cached: pd.DataFrame | None = None + first_date, last_date = timestamp_bounds(path) + if first_date is None or last_date is None: + cached = pd.read_parquet(path) + first_date, last_date = _date_bounds(cached) requested_start = _coerce_date(start_date) requested_end = _coerce_date(end_date) if ( @@ -376,6 +387,9 @@ def get_daily_history( and first_date <= requested_start and last_date >= requested_end ): + if cached is None: + # The covered path still needs the actual candles. + cached = pd.read_parquet(path) return self._slice_to_range(cached, start_date, end_date), True # Cache miss (or force_refresh): fetch the requested window from Dhan @@ -993,12 +1007,45 @@ def submit_next() -> bool: submit_next() yield outcome + def _covers_full_window(self, row: dict, years_back: int, today: date | None) -> bool: + """Answer "is this symbol's cache already fresh?" from footer stats only. + + PERF-002: the prefetch discards the candle frame — it only needs the + status — yet ``ensure_daily_history``'s "fresh" verdict used to load + the whole multi-year parquet just to learn its first/last dates. The + Parquet footer answers the same two dates in a few kilobytes. This + mirrors ``ensure_daily_history``'s exact fresh condition + (``first_date <= start and last_date >= today``); any file the footer + cannot vouch for returns False so the unchanged slow path decides. + """ + symbol = str(row.get("symbol", "")).strip().upper() + security_id = str(row.get("security_id", "")).strip() + if not symbol or not security_id: + # Let ensure_daily_history raise its documented ValueError. + return False + resolved_today = today or date.today() + start = history_start_date(int(years_back), resolved_today) + path = self.cache_path(symbol, security_id) + if not path.exists(): + return False + first_date, last_date = timestamp_bounds(path) + return ( + first_date is not None + and last_date is not None + and first_date <= start + and last_date >= resolved_today + ) + def _ensure_one_row( self, row: dict, years_back: int, today: date | None ) -> PrefetchOutcome: """Run ``ensure_daily_history`` for one row, capturing a safe outcome.""" symbol = str(row.get("symbol", "?")).strip() or "?" try: + # PERF-002 fast path: a cache the footer proves fresh skips the + # full-frame read entirely. Same status the slow path would return. + if self._covers_full_window(row, years_back, today): + return PrefetchOutcome(symbol=symbol, status="fresh") _, status = self.ensure_daily_history(row, years_back=years_back, today=today) return PrefetchOutcome(symbol=symbol, status=status) except Exception as exc: diff --git a/backend/parquet_stats.py b/backend/parquet_stats.py new file mode 100644 index 0000000..a23d9bc --- /dev/null +++ b/backend/parquet_stats.py @@ -0,0 +1,98 @@ +"""Cheap candle-date bounds from Parquet footer statistics (PERF-002). + +Beginner note: +Parquet files end with a footer that records optional per-row-group +minimum/maximum statistics for every column. pandas' ``to_parquet`` (via +pyarrow) writes those statistics by default, so for the daily candle cache +the first and last candle dates can usually be answered by reading a few +kilobytes of footer instead of decompressing a whole multi-year frame. +``backend/health.py`` has used this trick for its cache snapshot since +OBS-002; this module generalizes it for the data loader's cache-coverage +decisions ("is this file already fresh?", "does it cover the requested +range?"), which previously loaded the entire frame just to learn two dates. + +Callers MUST treat ``(None, None)`` as "the footer cannot answer cheaply" and +fall back to their existing full-read logic — never as "the file is empty". +Statistics can be legitimately absent (a writer passed +``write_statistics=False``, an all-null column, a truncated file), and the +loader's behavior for those files has to stay exactly what it was before +PERF-002. +""" + +from __future__ import annotations + +import datetime as dt +from pathlib import Path +from typing import Any + +import pyarrow.parquet as pq + + +def timestamp_bounds(path: Path) -> tuple[dt.date | None, dt.date | None]: + """Return ``(first, last)`` candle dates using footer statistics only. + + Reads the Parquet footer (schema + row-group statistics) and never the + data pages. Returns ``(None, None)`` whenever the footer cannot answer + authoritatively: missing file, no ``timestamp`` column, zero row groups, + any row group without min/max statistics, a non-date-like statistic, or + any read/parse error. Deliberately never raises — the caller's full-read + fallback is the error handler. + """ + try: + parquet_file = pq.ParquetFile(path) + timestamp_index = parquet_file.schema_arrow.get_field_index("timestamp") + if timestamp_index < 0: + return (None, None) + metadata = parquet_file.metadata + if metadata.num_row_groups == 0: + return (None, None) + + earliest: dt.date | None = None + latest: dt.date | None = None + for row_group_index in range(metadata.num_row_groups): + column = metadata.row_group(row_group_index).column(timestamp_index) + statistics = column.statistics + # PyArrow exposes one flag for the min/max pair. Older versions do + # not provide a separate ``has_max`` attribute, so use the stable + # ``has_min_max`` API before reading either value. One statless + # row group makes the whole answer untrustworthy: its rows could + # extend past every other group's bounds. + if statistics is None or not statistics.has_min_max: + return (None, None) + first = _as_date(statistics.min) + last = _as_date(statistics.max) + if first is None or last is None: + return (None, None) + if earliest is None or first < earliest: + earliest = first + if latest is None or last > latest: + latest = last + return (earliest, latest) + except Exception: + # PyArrow raises several format-specific exception classes for corrupt + # or non-Parquet files. All of them mean the same thing here: the + # footer cannot answer, so the caller must use its full-read fallback. + return (None, None) + + +def _as_date(value: Any) -> dt.date | None: + """Normalize common PyArrow timestamp-statistic values to a date. + + Mirrors the coercion ``backend/health.py`` applies to the same statistics + (datetime/date objects, pandas Timestamps via ``to_pydatetime``, ISO + strings). Anything else — e.g. integer or bytes statistics from a column + that is not really a timestamp — returns ``None`` so the caller falls + back rather than trusting a mistyped column. + """ + if value is None: + return None + if isinstance(value, dt.datetime): + return value.date() + if isinstance(value, dt.date): + return value + if hasattr(value, "to_pydatetime"): + return value.to_pydatetime().date() + try: + return dt.datetime.fromisoformat(str(value)).date() + except (TypeError, ValueError): + return None diff --git a/tests/test_daily_data_loader_footer_stats.py b/tests/test_daily_data_loader_footer_stats.py new file mode 100644 index 0000000..3fd1fc5 --- /dev/null +++ b/tests/test_daily_data_loader_footer_stats.py @@ -0,0 +1,200 @@ +"""Loader integration tests for the PERF-002 footer-stats fast paths. + +Two behaviors changed and two must NOT have changed: + +1. NEW: the prefetch fresh check and ``get_daily_history``'s coverage check + answer from the Parquet footer, so those paths no longer decompress a + multi-year frame just to learn two dates (proven here by making + ``pd.read_parquet`` explode). +2. UNCHANGED: any file the footer cannot vouch for (``write_statistics=False`` + is the canonical case) behaves exactly as before via the full-read + fallback — a covered range is still a cache hit, a fresh cache is still + "fresh". The pre-existing suite (test_daily_data_loader.py) runs + unmodified as the broader behavior lock. +""" + +from __future__ import annotations + +from datetime import date, timedelta + +import pandas as pd +import pyarrow as pa +import pyarrow.parquet as pq +import pytest + +from backend import daily_data_loader +from backend.daily_data_loader import DailyDataLoader, history_start_date + +TODAY = date(2026, 7, 10) +YEARS_BACK = 10 + + +def _instrument() -> dict: + return {"symbol": "DEMO", "security_id": "1"} + + +def _covering_frame() -> pd.DataFrame: + """Business-day candles spanning the full prefetch window ending today.""" + start = history_start_date(YEARS_BACK, TODAY) - timedelta(days=5) + stamps = pd.date_range(start, TODAY, freq="B") + if stamps[-1].date() != TODAY: + stamps = stamps.append(pd.DatetimeIndex([pd.Timestamp(TODAY)])) + return pd.DataFrame( + { + "timestamp": stamps, + "open": 100.0, + "high": 110.0, + "low": 95.0, + "close": 105.0, + "volume": 1000.0, + } + ) + + +def _write_cache(loader: DailyDataLoader, frame: pd.DataFrame, *, statistics: bool) -> None: + """Write DEMO's cache file with or without footer statistics.""" + path = loader.cache_path("DEMO", "1") + path.parent.mkdir(parents=True, exist_ok=True) + if statistics: + frame.to_parquet(path, index=False) + else: + pq.write_table(pa.Table.from_pandas(frame), path, write_statistics=False) + + +def _forbid_full_reads(monkeypatch) -> None: + """Make any pandas parquet read fail loudly inside the loader module.""" + + def _explode(*_args, **_kwargs): + raise AssertionError("this path must answer from footer statistics alone") + + monkeypatch.setattr(daily_data_loader.pd, "read_parquet", _explode) + + +def test_prefetch_fresh_verdict_reads_no_frame(monkeypatch, tmp_path): + """A footer-provably-fresh cache yields 'fresh' without pandas I/O. + + client=None doubles as a second proof: cache-only loaders raise on any + fetch attempt, so reaching Dhan would fail the test too. + """ + loader = DailyDataLoader(client=None, cache_dir=tmp_path, request_delay_seconds=0.0) + _write_cache(loader, _covering_frame(), statistics=True) + _forbid_full_reads(monkeypatch) + + outcomes = list( + loader.iter_ensure_universe_history( + [_instrument()], years_back=YEARS_BACK, today=TODAY + ) + ) + + assert [(outcome.symbol, outcome.status) for outcome in outcomes] == [("DEMO", "fresh")] + + +def test_prefetch_falls_back_to_full_read_without_footer_statistics(tmp_path): + """A statless covering cache must still be 'fresh' via the slow path.""" + loader = DailyDataLoader(client=None, cache_dir=tmp_path, request_delay_seconds=0.0) + _write_cache(loader, _covering_frame(), statistics=False) + + outcomes = list( + loader.iter_ensure_universe_history( + [_instrument()], years_back=YEARS_BACK, today=TODAY + ) + ) + + assert [(outcome.symbol, outcome.status) for outcome in outcomes] == [("DEMO", "fresh")] + + +def test_stale_cache_still_reaches_the_slow_path(tmp_path): + """The footer shortcut must not swallow the incremental top-up. + + With a cache ending before today, the fast path declines and + ensure_daily_history runs; a cache-only loader then raises its documented + RuntimeError, which the prefetch surfaces as a redacted failure. + """ + loader = DailyDataLoader(client=None, cache_dir=tmp_path, request_delay_seconds=0.0) + stale = _covering_frame() + stale = stale.loc[stale["timestamp"] < pd.Timestamp(TODAY) - pd.Timedelta(days=30)] + _write_cache(loader, stale, statistics=True) + + outcomes = list( + loader.iter_ensure_universe_history( + [_instrument()], years_back=YEARS_BACK, today=TODAY + ) + ) + + assert outcomes[0].status == "failed" + assert "cache-only mode" in (outcomes[0].message or "") + + +def test_get_daily_history_miss_decision_reads_no_frame(monkeypatch, tmp_path): + """An insufficient cache goes straight to the fetch, frame unread. + + Before PERF-002 the loader decompressed the whole cached frame, computed + its bounds, discarded it, and fetched from Dhan. Now the footer answers + the coverage question, so the only pandas work is the fetched result. + """ + fetched = pd.DataFrame( + { + "timestamp": pd.date_range(date(2026, 6, 1), date(2026, 7, 10), freq="B"), + "open": 100.0, + "high": 110.0, + "low": 95.0, + "close": 105.0, + "volume": 1000.0, + } + ) + + class OneShotClient: + """Serve one canned frame and count how often Dhan is called.""" + + def __init__(self) -> None: + self.calls = 0 + + def fetch_daily_candles(self, **_kwargs) -> pd.DataFrame: + self.calls += 1 + return fetched.copy(deep=True) + + client = OneShotClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + short = fetched.loc[fetched["timestamp"] >= pd.Timestamp(date(2026, 7, 1))] + _write_cache(loader, short, statistics=True) + # Forbid reads AFTER writing the cache; to_parquet is unaffected. + _forbid_full_reads(monkeypatch) + + frame, from_cache = loader.get_daily_history( + _instrument(), date(2026, 6, 1), date(2026, 7, 10) + ) + + assert from_cache is False + assert client.calls == 1 + assert not frame.empty + + +def test_get_daily_history_covered_range_without_statistics_is_still_a_hit(tmp_path): + """PERF-002 must not turn any old cache hit into a Dhan fetch.""" + + class ForbiddenClient: + """Fail the test if the loader asks Dhan for anything.""" + + def fetch_daily_candles(self, **_kwargs) -> pd.DataFrame: + raise AssertionError("a covered cache must not reach Dhan") + + loader = DailyDataLoader(ForbiddenClient(), cache_dir=tmp_path, request_delay_seconds=0.0) + _write_cache(loader, _covering_frame(), statistics=False) + + frame, from_cache = loader.get_daily_history( + _instrument(), TODAY - timedelta(days=30), TODAY + ) + + assert from_cache is True + assert not frame.empty + assert frame["timestamp"].min() >= pd.Timestamp(TODAY - timedelta(days=30)) + + +def test_covers_full_window_requires_symbol_and_security_id(tmp_path): + """Malformed rows decline the fast path so the slow path raises as before.""" + loader = DailyDataLoader(client=None, cache_dir=tmp_path, request_delay_seconds=0.0) + + assert loader._covers_full_window({"symbol": "", "security_id": "1"}, YEARS_BACK, TODAY) is False + assert loader._covers_full_window({"symbol": "DEMO", "security_id": ""}, YEARS_BACK, TODAY) is False + with pytest.raises(ValueError, match="missing symbol"): + loader.ensure_daily_history({"symbol": "", "security_id": "1"}, today=TODAY) diff --git a/tests/test_parquet_stats.py b/tests/test_parquet_stats.py new file mode 100644 index 0000000..3b000cc --- /dev/null +++ b/tests/test_parquet_stats.py @@ -0,0 +1,97 @@ +"""Unit tests for the footer-statistics date bounds helper (PERF-002). + +The contract under test: ``timestamp_bounds`` answers from the Parquet footer +alone, and returns ``(None, None)`` — never a guess, never an exception — +whenever the footer cannot answer authoritatively. The loader's full-read +fallback depends on that fail-safe shape. +""" + +from __future__ import annotations + +import datetime as dt + +import pandas as pd +import pyarrow as pa +import pyarrow.parquet as pq + +from backend.parquet_stats import timestamp_bounds + + +def _candles(dates: list[dt.date]) -> pd.DataFrame: + return pd.DataFrame( + { + "timestamp": [pd.Timestamp(value) for value in dates], + "open": [100.0] * len(dates), + "close": [101.0] * len(dates), + } + ) + + +def test_bounds_from_a_normal_pandas_written_cache_file(tmp_path): + path = tmp_path / "DEMO_1.parquet" + _candles( + [dt.date(2016, 7, 11), dt.date(2020, 1, 2), dt.date(2026, 7, 10)] + ).to_parquet(path, index=False) + + assert timestamp_bounds(path) == (dt.date(2016, 7, 11), dt.date(2026, 7, 10)) + + +def test_bounds_span_multiple_row_groups(tmp_path): + """Min/max must be folded across ALL row groups, not read from the first.""" + path = tmp_path / "DEMO_1.parquet" + frame = _candles( + [dt.date(2020, 1, 2), dt.date(2021, 6, 1), dt.date(2024, 3, 3), dt.date(2026, 7, 10)] + ) + # row_group_size=2 forces two groups; the true bounds straddle them. + pq.write_table(pa.Table.from_pandas(frame), path, row_group_size=2) + assert pq.ParquetFile(path).metadata.num_row_groups == 2 + + assert timestamp_bounds(path) == (dt.date(2020, 1, 2), dt.date(2026, 7, 10)) + + +def test_missing_file_returns_none_pair(tmp_path): + assert timestamp_bounds(tmp_path / "absent.parquet") == (None, None) + + +def test_non_parquet_file_returns_none_pair(tmp_path): + path = tmp_path / "corrupt.parquet" + path.write_bytes(b"this is not a parquet footer") + assert timestamp_bounds(path) == (None, None) + + +def test_missing_timestamp_column_returns_none_pair(tmp_path): + path = tmp_path / "DEMO_1.parquet" + pd.DataFrame({"close": [1.0, 2.0]}).to_parquet(path, index=False) + assert timestamp_bounds(path) == (None, None) + + +def test_empty_frame_returns_none_pair(tmp_path): + path = tmp_path / "DEMO_1.parquet" + _candles([]).to_parquet(path, index=False) + assert timestamp_bounds(path) == (None, None) + + +def test_all_null_timestamps_return_none_pair(tmp_path): + """A column of NaT has no min/max statistics — must fall back, not guess.""" + path = tmp_path / "DEMO_1.parquet" + pd.DataFrame( + {"timestamp": pd.to_datetime([None, None]), "close": [1.0, 2.0]} + ).to_parquet(path, index=False) + assert timestamp_bounds(path) == (None, None) + + +def test_writer_without_statistics_returns_none_pair(tmp_path): + """``write_statistics=False`` is the canonical 'footer cannot answer' case.""" + path = tmp_path / "DEMO_1.parquet" + frame = _candles([dt.date(2020, 1, 2), dt.date(2026, 7, 10)]) + pq.write_table(pa.Table.from_pandas(frame), path, write_statistics=False) + + assert timestamp_bounds(path) == (None, None) + + +def test_non_datelike_timestamp_column_returns_none_pair(tmp_path): + """Integer statistics in a mistyped 'timestamp' column must not be trusted.""" + path = tmp_path / "DEMO_1.parquet" + pd.DataFrame({"timestamp": [5, 9], "close": [1.0, 2.0]}).to_parquet(path, index=False) + + assert timestamp_bounds(path) == (None, None) From 4224ac1f23e22e6934ba63dcf221855c7fe9978c Mon Sep 17 00:00:00 2001 From: DoRmAmMu1997 Date: Sat, 11 Jul 2026 18:39:19 +0530 Subject: [PATCH 2/2] fix: validate parquet caches before freshness Co-authored-by: Hemant Co-authored-by: Codex --- backend/daily_data_loader.py | 49 +++------ backend/parquet_stats.py | 11 +- tests/test_daily_data_loader_footer_stats.py | 107 ++++++++++++++++--- 3 files changed, 116 insertions(+), 51 deletions(-) diff --git a/backend/daily_data_loader.py b/backend/daily_data_loader.py index aeafbdb..2dc643c 100644 --- a/backend/daily_data_loader.py +++ b/backend/daily_data_loader.py @@ -388,9 +388,18 @@ def get_daily_history( and last_date >= requested_end ): if cached is None: - # The covered path still needs the actual candles. + # 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) - return self._slice_to_range(cached, start_date, end_date), True + 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 + ): + return self._slice_to_range(cached, start_date, end_date), True # Cache miss (or force_refresh): fetch the requested window from Dhan # and save under the stable filename for future calls. @@ -1007,45 +1016,15 @@ def submit_next() -> bool: submit_next() yield outcome - def _covers_full_window(self, row: dict, years_back: int, today: date | None) -> bool: - """Answer "is this symbol's cache already fresh?" from footer stats only. - - PERF-002: the prefetch discards the candle frame — it only needs the - status — yet ``ensure_daily_history``'s "fresh" verdict used to load - the whole multi-year parquet just to learn its first/last dates. The - Parquet footer answers the same two dates in a few kilobytes. This - mirrors ``ensure_daily_history``'s exact fresh condition - (``first_date <= start and last_date >= today``); any file the footer - cannot vouch for returns False so the unchanged slow path decides. - """ - symbol = str(row.get("symbol", "")).strip().upper() - security_id = str(row.get("security_id", "")).strip() - if not symbol or not security_id: - # Let ensure_daily_history raise its documented ValueError. - return False - resolved_today = today or date.today() - start = history_start_date(int(years_back), resolved_today) - path = self.cache_path(symbol, security_id) - if not path.exists(): - return False - first_date, last_date = timestamp_bounds(path) - return ( - first_date is not None - and last_date is not None - and first_date <= start - and last_date >= resolved_today - ) - def _ensure_one_row( self, row: dict, years_back: int, today: date | None ) -> PrefetchOutcome: """Run ``ensure_daily_history`` for one row, capturing a safe outcome.""" symbol = str(row.get("symbol", "?")).strip() or "?" try: - # PERF-002 fast path: a cache the footer proves fresh skips the - # full-frame read entirely. Same status the slow path would return. - if self._covers_full_window(row, years_back, today): - return PrefetchOutcome(symbol=symbol, status="fresh") + # A prefetch freshness verdict must come from the frame itself. + # Footer statistics can survive damaged data pages, so using them + # here would let an unreadable cache masquerade as healthy. _, status = self.ensure_daily_history(row, years_back=years_back, today=today) return PrefetchOutcome(symbol=symbol, status=status) except Exception as exc: diff --git a/backend/parquet_stats.py b/backend/parquet_stats.py index a23d9bc..4b730f6 100644 --- a/backend/parquet_stats.py +++ b/backend/parquet_stats.py @@ -4,12 +4,17 @@ Parquet files end with a footer that records optional per-row-group minimum/maximum statistics for every column. pandas' ``to_parquet`` (via pyarrow) writes those statistics by default, so for the daily candle cache -the first and last candle dates can usually be answered by reading a few +the first and last candle dates can usually be estimated by reading a few kilobytes of footer instead of decompressing a whole multi-year frame. ``backend/health.py`` has used this trick for its cache snapshot since OBS-002; this module generalizes it for the data loader's cache-coverage -decisions ("is this file already fresh?", "does it cover the requested -range?"), which previously loaded the entire frame just to learn two dates. +miss decisions ("does this old file definitely fail to cover the requested +range?"), which previously loaded an entire frame that was then discarded. + +Footer bounds are advisory. A valid footer can coexist with corrupt data pages +or describe a file that a concurrent writer replaces before the caller reads +it. Callers must validate the frame they actually use before returning a cache +hit or reporting a cache as fresh. Callers MUST treat ``(None, None)`` as "the footer cannot answer cheaply" and fall back to their existing full-read logic — never as "the file is empty". diff --git a/tests/test_daily_data_loader_footer_stats.py b/tests/test_daily_data_loader_footer_stats.py index 3fd1fc5..f309afc 100644 --- a/tests/test_daily_data_loader_footer_stats.py +++ b/tests/test_daily_data_loader_footer_stats.py @@ -2,10 +2,10 @@ Two behaviors changed and two must NOT have changed: -1. NEW: the prefetch fresh check and ``get_daily_history``'s coverage check - answer from the Parquet footer, so those paths no longer decompress a - multi-year frame just to learn two dates (proven here by making - ``pd.read_parquet`` explode). +1. NEW: ``get_daily_history`` can use the footer to skip a frame that would be + discarded on a definite cache miss. Prefetch still reads the frame before + declaring it fresh because footer metadata cannot prove data pages remain + readable. 2. UNCHANGED: any file the footer cannot vouch for (``write_statistics=False`` is the canonical case) behaves exactly as before via the full-read fallback — a covered range is still a cache hit, a fresh cache is still @@ -70,15 +70,23 @@ def _explode(*_args, **_kwargs): monkeypatch.setattr(daily_data_loader.pd, "read_parquet", _explode) -def test_prefetch_fresh_verdict_reads_no_frame(monkeypatch, tmp_path): - """A footer-provably-fresh cache yields 'fresh' without pandas I/O. +def test_prefetch_validates_frame_before_fresh_verdict(monkeypatch, tmp_path): + """A fresh prefetch verdict must validate the actual Parquet data pages. - client=None doubles as a second proof: cache-only loaders raise on any - fetch attempt, so reaching Dhan would fail the test too. + Beginner note: footer statistics are a quick index, not an integrity + check. A file may keep a readable footer even when a data page is damaged, + so this path deliberately pays for one full read before saying "fresh". """ loader = DailyDataLoader(client=None, cache_dir=tmp_path, request_delay_seconds=0.0) _write_cache(loader, _covering_frame(), statistics=True) - _forbid_full_reads(monkeypatch) + real_read_parquet = pd.read_parquet + read_paths = [] + + def _record_read(path, *args, **kwargs): + read_paths.append(path) + return real_read_parquet(path, *args, **kwargs) + + monkeypatch.setattr(daily_data_loader.pd, "read_parquet", _record_read) outcomes = list( loader.iter_ensure_universe_history( @@ -87,6 +95,32 @@ def test_prefetch_fresh_verdict_reads_no_frame(monkeypatch, tmp_path): ) assert [(outcome.symbol, outcome.status) for outcome in outcomes] == [("DEMO", "fresh")] + assert read_paths == [loader.cache_path("DEMO", "1")] + + +def test_prefetch_does_not_call_unreadable_data_pages_fresh(monkeypatch, tmp_path): + """A footer-only success must not hide a corrupt Parquet data page.""" + loader = DailyDataLoader(client=None, cache_dir=tmp_path, request_delay_seconds=0.0) + _write_cache(loader, _covering_frame(), statistics=True) + monkeypatch.setattr( + daily_data_loader, + "timestamp_bounds", + lambda _path: (history_start_date(YEARS_BACK, TODAY), TODAY), + ) + + def _unreadable(*_args, **_kwargs): + raise OSError("parquet data page is corrupt") + + monkeypatch.setattr(daily_data_loader.pd, "read_parquet", _unreadable) + + outcomes = list( + loader.iter_ensure_universe_history( + [_instrument()], years_back=YEARS_BACK, today=TODAY + ) + ) + + assert outcomes[0].status == "failed" + assert "parquet data page is corrupt" in (outcomes[0].message or "") def test_prefetch_falls_back_to_full_read_without_footer_statistics(tmp_path): @@ -190,11 +224,58 @@ def fetch_daily_candles(self, **_kwargs) -> pd.DataFrame: assert frame["timestamp"].min() >= pd.Timestamp(TODAY - timedelta(days=30)) -def test_covers_full_window_requires_symbol_and_security_id(tmp_path): - """Malformed rows decline the fast path so the slow path raises as before.""" +def test_get_daily_history_rechecks_frame_after_footer_claim(monkeypatch, tmp_path): + """A file replaced after its footer is read must not become a false hit. + + This simulates a concurrent writer replacing the cache between the cheap + footer check and the full read. The frame actually returned by pandas is + authoritative, so an insufficient replacement must trigger a refetch. + """ + requested_start = date(2026, 6, 1) + requested_end = TODAY + replacement = _covering_frame().loc[ + lambda frame: frame["timestamp"] >= pd.Timestamp(date(2026, 7, 1)) + ] + fetched = _covering_frame().loc[ + lambda frame: frame["timestamp"] >= pd.Timestamp(requested_start) + ] + + class OneShotClient: + def __init__(self) -> None: + self.calls = 0 + + def fetch_daily_candles(self, **_kwargs) -> pd.DataFrame: + self.calls += 1 + return fetched.copy(deep=True) + + client = OneShotClient() + loader = DailyDataLoader(client, cache_dir=tmp_path, request_delay_seconds=0.0) + _write_cache(loader, _covering_frame(), statistics=True) + monkeypatch.setattr( + daily_data_loader, + "timestamp_bounds", + lambda _path: (requested_start, requested_end), + ) + monkeypatch.setattr( + daily_data_loader.pd, + "read_parquet", + lambda *_args, **_kwargs: replacement.copy(deep=True), + ) + + frame, from_cache = loader.get_daily_history( + _instrument(), requested_start, requested_end + ) + + assert from_cache is False + assert client.calls == 1 + assert frame["timestamp"].min() <= pd.Timestamp(requested_start) + + +def test_prefetch_requires_symbol_and_security_id(tmp_path): + """Malformed rows still fail through the documented validation path.""" loader = DailyDataLoader(client=None, cache_dir=tmp_path, request_delay_seconds=0.0) - assert loader._covers_full_window({"symbol": "", "security_id": "1"}, YEARS_BACK, TODAY) is False - assert loader._covers_full_window({"symbol": "DEMO", "security_id": ""}, YEARS_BACK, TODAY) is False with pytest.raises(ValueError, match="missing symbol"): loader.ensure_daily_history({"symbol": "", "security_id": "1"}, today=TODAY) + with pytest.raises(ValueError, match="missing security_id"): + loader.ensure_daily_history({"symbol": "DEMO", "security_id": ""}, today=TODAY)