fix: --check measures lag on write time, not on data age

src/cotdata/status.py

@@ -19,8 +19,8 @@ def status_path():
 
 # Domains shown by --check, in report order.
 _DOMAINS = ["prices", "metadata", "cot", "cot_legacy", "cot_disagg", "cot_tff"]
-# An entry lagging more than this many days behind its domain's newest data
-# probably failed to update while its peers succeeded (a partial run).
+# An entry whose producer last touched it more than this many days behind its
+# domain's newest write probably failed while its peers succeeded (a partial run).
 _LAG_DAYS = 3
 
 
@@ -31,6 +31,16 @@ def _parse_date(s: Optional[str]) -> Optional[dt.date]:
         return None
 
 
+def _parse_write(s: Optional[str]) -> Optional[dt.datetime]:
+    """Parse a manifest ``updated_at`` (ISO-8601, trailing Z) to a naive UTC datetime."""
+    if not s:
+        return None
+    try:
+        return dt.datetime.fromisoformat(str(s).replace("Z", "+00:00")).replace(tzinfo=None)
+    except (ValueError, TypeError):
+        return None
+
+
 def hist_code_names() -> set:
     """Entry names for symbols' predecessor (hist_code) COT contracts, e.g.
     ``RTY_23977A``. These are retired codes, legitimately frozen at an old date,
@@ -60,15 +70,36 @@ def summarize(manifest: dict, today: Optional[dt.date] = None,
         dated = {n: _parse_date(e.get("last_date")) for n, e in entries.items()}
         dates = [d for d in dated.values() if d]
         newest = max(dates) if dates else None
-        lagging = []
-        if newest:
-            for name, d in dated.items():
-                if name in ignore_lag:
-                    continue
-                behind = (newest - d).days if d else None
-                if behind is not None and behind > _LAG_DAYS:
-                    lagging.append((name, entries[name].get("last_date"), behind))
         writes = [e.get("updated_at") for e in entries.values() if e.get("updated_at")]
+
+        # Lag is measured on WHEN THE PRODUCER LAST TOUCHED THE ENTRY, not on how
+        # recent its data is. The check exists to catch a partial run (one entry
+        # failed while its peers succeeded), and only updated_at distinguishes that
+        # from a source that simply has nothing newer.
+        #
+        # Comparing last_date instead produced permanent false positives for the two
+        # legitimate cases of an old-but-correct entry: thin markets that drop out of
+        # the CFTC report below the reporting threshold and reappear months later
+        # (NKD and ZO each have ~20 gaps over 10 days across 19 years, the longest
+        # 168d and 294d), and retired hist_code contracts frozen by design. Both are
+        # written on every pass, so both carry a current updated_at.
+        parsed = {n: _parse_write(e.get("updated_at")) for n, e in entries.items()}
+        usable = [w for w in parsed.values() if w]
+        newest_write = max(usable) if usable else None
+        lagging = []
+        for name, e in entries.items():
+            if name in ignore_lag:
+                continue
+            w = parsed[name]
+            if w is None:
+                # No usable write timestamp: report it rather than assume it is fine.
+                # Silently passing here would turn "cannot check" into "all current",
+                # which is the failure mode this whole check exists to avoid.
+                lagging.append((name, e.get("updated_at") or "never", 9999))
+                continue
+            behind = (newest_write - w).days
+            if behind > _LAG_DAYS:
+                lagging.append((name, e.get("updated_at"), behind))
         out["domains"][domain] = {
             "entries": len(entries),
             "rows": sum(int(e.get("n_rows") or 0) for e in entries.values()),
@@ -106,14 +137,15 @@ def format_report(manifest: dict, root: str = "", today: Optional[dt.date] = Non
         if d["lagging"]:
             L.append("")
             L.append(f"⚠ {domain}: {len(d['lagging'])} entr{'y' if len(d['lagging'])==1 else 'ies'} "
-                     f"lag >{_LAG_DAYS}d behind newest ({d['newest']}):")
-            for name, last, behind in d["lagging"][:15]:
-                L.append(f"    {name:<18}{last}  ({behind}d behind)")
+                     f"NOT WRITTEN in the last run (>{_LAG_DAYS}d behind the domain's "
+                     f"newest write, {d['last_write']}):")
+            for name, wrote, behind in d["lagging"][:15]:
+                L.append(f"    {name:<18}last written {wrote}  ({behind}d behind)")
             if len(d["lagging"]) > 15:
                 L.append(f"    … and {len(d['lagging']) - 15} more")
     if not any(d["lagging"] for d in s["domains"].values()):
         L.append("")
-        L.append("✓ all entries current (none lag behind their domain's newest).")
+        L.append("✓ every entry was written by the latest producer pass.")
     return "\n".join(L)
 
 

tests/test_status.py

@@ -31,28 +31,64 @@ def test_summarize_counts_and_newest():
 def test_summarize_flags_lagging_entry():
     s = status.summarize(_manifest(), today=dt.date(2026, 7, 15))
     lagging = s["domains"]["prices"]["lagging"]
-    # NQ_backadj (2026-07-05) is 9 days behind the domain newest (2026-07-14).
+    # NQ_backadj was last WRITTEN 2026-07-06, 9 days behind the domain's newest
+    # write (2026-07-15) — i.e. the producer skipped it while its peers succeeded.
     assert [name for name, _, _ in lagging] == ["NQ_backadj"]
     assert lagging[0][2] == 9
 
 
-def test_ignore_lag_suppresses_hist_codes():
-    # RTY's predecessor code is frozen at 2018 but must NOT be flagged as lagging.
+def test_old_data_is_not_lagging_if_it_was_written_this_pass():
+    """Lag is about the PRODUCER skipping an entry, not about the data being old.
+
+    Two legitimate old-but-correct cases, both seen in the real store: thin markets
+    that drop out of the CFTC report for months and reappear (NKD, ZO), and retired
+    hist_code contracts frozen by design. Both are rewritten every pass, so both
+    carry a current updated_at and neither is a problem.
+    """
+    m = {
+        "schema_version": 2,
+        "cot_legacy": {
+            "RTY_current": {"last_date": "2026-07-07", "n_rows": 100,
+                            "updated_at": "2026-07-15T04:00:00Z"},
+            "RTY_23977A":  {"last_date": "2018-06-05", "n_rows": 500,   # retired code
+                            "updated_at": "2026-07-15T04:00:00Z"},
+            "ZO_004603":   {"last_date": "2026-06-02", "n_rows": 900,   # intermittent
+                            "updated_at": "2026-07-15T04:00:00Z"},
+            "NQ_skipped":  {"last_date": "2026-07-07", "n_rows": 100,   # genuinely missed
+                            "updated_at": "2026-06-20T04:00:00Z"},
+        },
+    }
+    lagging = status.summarize(m, dt.date(2026, 7, 15))["domains"]["cot_legacy"]["lagging"]
+    # Only the entry the producer actually skipped, even though two others carry
+    # far older data.
+    assert {n for n, _, _ in lagging} == {"NQ_skipped"}
+
+
+def test_ignore_lag_still_suppresses_by_name():
     m = {
         "schema_version": 2,
         "cot_legacy": {
-            "RTY_current": {"last_date": "2026-07-07", "n_rows": 100, "updated_at": "x"},
-            "RTY_23977A":  {"last_date": "2018-06-05", "n_rows": 500, "updated_at": "x"},
-            "NQ_current":  {"last_date": "2020-01-01", "n_rows": 100, "updated_at": "x"},
+            "RTY_current": {"last_date": "2026-07-07", "n_rows": 100,
+                            "updated_at": "2026-07-15T04:00:00Z"},
+            "RTY_23977A":  {"last_date": "2018-06-05", "n_rows": 500,
+                            "updated_at": "2026-06-01T04:00:00Z"},
         },
     }
     today = dt.date(2026, 7, 15)
-    # Without the ignore set, both stale entries are flagged.
-    plain = status.summarize(m, today)["domains"]["cot_legacy"]["lagging"]
-    assert {n for n, _, _ in plain} == {"RTY_23977A", "NQ_current"}
-    # With the hist_code suppressed, only the genuinely-stale current code remains.
+    assert {n for n, _, _ in status.summarize(m, today)["domains"]["cot_legacy"]["lagging"]} \
+        == {"RTY_23977A"}
     filt = status.summarize(m, today, ignore_lag={"RTY_23977A"})["domains"]["cot_legacy"]["lagging"]
-    assert {n for n, _, _ in filt} == {"NQ_current"}
+    assert filt == []
+
+
+def test_unusable_write_timestamp_is_reported_not_assumed_fine():
+    """"Cannot check" must never render as "all current" — that is the exact failure
+    this check exists to catch."""
+    m = {"schema_version": 2,
+         "cot_legacy": {"A": {"last_date": "2026-07-07", "n_rows": 1, "updated_at": "x"},
+                        "B": {"last_date": "2026-07-07", "n_rows": 1}}}
+    lagging = status.summarize(m, dt.date(2026, 7, 15))["domains"]["cot_legacy"]["lagging"]
+    assert {n for n, _, _ in lagging} == {"A", "B"}
 
 
 def test_empty_store_report():