fix(DATA-003): stop the scan re-dirtying the candle cache it just repaired

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)
 
 

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 `<symbol>_<security-id>.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
 

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}