diff --git a/backend/daily_data_loader.py b/backend/daily_data_loader.py index d09ab24..2dc643c 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,7 +387,19 @@ def get_daily_history( and first_date <= requested_start and last_date >= requested_end ): - return self._slice_to_range(cached, start_date, end_date), True + if cached is None: + # The footer is only an advisory index. The file can be + # replaced after the metadata read, and a valid footer + # does not prove every data page is readable. + cached = pd.read_parquet(path) + actual_first, actual_last = _date_bounds(cached) + if ( + actual_first is not None + and actual_last is not None + and actual_first <= requested_start + and actual_last >= requested_end + ): + 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. @@ -999,6 +1022,9 @@ def _ensure_one_row( """Run ``ensure_daily_history`` for one row, capturing a safe outcome.""" symbol = str(row.get("symbol", "?")).strip() or "?" try: + # 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 new file mode 100644 index 0000000..4b730f6 --- /dev/null +++ b/backend/parquet_stats.py @@ -0,0 +1,103 @@ +"""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 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 +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". +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..f309afc --- /dev/null +++ b/tests/test_daily_data_loader_footer_stats.py @@ -0,0 +1,281 @@ +"""Loader integration tests for the PERF-002 footer-stats fast paths. + +Two behaviors changed and two must NOT have changed: + +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 + "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_validates_frame_before_fresh_verdict(monkeypatch, tmp_path): + """A fresh prefetch verdict must validate the actual Parquet data pages. + + 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) + 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( + [_instrument()], years_back=YEARS_BACK, today=TODAY + ) + ) + + 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): + """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_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) + + 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) 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)