feat: exit codes for scheduler retry + README scheduling/diagram updates

src/cotdata/providers/cftc.py

@@ -99,12 +99,16 @@ def _parse_zip(zip_path: Path) -> pd.DataFrame:
     return df
 
 
-def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None:
+def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> dict:
     """Download + parse CFTC Legacy futures COT; write full per-code history.
 
     codes: iterable of CFTC codes; default = all registry codes.
     Rebuilds the complete per-code table each run (parse is cheap; downloads are
     cached and skip when unchanged). Incremental append is a future optimization.
+
+    Returns a result dict {"kind", "ok", "wrote"}. ``ok`` is False only on a hard
+    failure to fetch the *current* year (the file a new release lands in) — so a
+    scheduler can retry a real outage without treating "no new data yet" as failure.
     """
     code_to_sym = {}
     for s in all_symbols():
@@ -120,20 +124,26 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None:
         want = set(code_to_sym.keys())
 
     frames = []
+    latest_ok = True
     for year in range(first_year, last_year + 1):
         zp = _download_year(year)
         if zp is None:
+            if year == last_year:
+                latest_ok = False  # couldn't fetch the current year — may have missed a release
             continue
         try:
             frames.append(_parse_zip(zp))
         except Exception as e:  # noqa: BLE001
             print(f"  {year}: parse failed — {e}")
+            if year == last_year:
+                latest_ok = False
 
     if not frames:
         print("cftc: no data parsed")
-        return
+        return {"kind": "cot_legacy", "ok": False, "wrote": 0}
 
     allrows = pd.concat(frames, ignore_index=True)
+    wrote = 0
     for code in sorted(want):
         sub = allrows[allrows[CONTRACT_CODE] == code].copy()
         if sub.empty:
@@ -143,4 +153,6 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None:
         sym_name = code_to_sym.get(code)
         file_name = f"{sym_name}_{code}" if sym_name else code
         store.write_cot_legacy(file_name, sub, source="cftc")
+        wrote += 1
         print(f"{file_name}: {len(sub):5d} weeks (legacy) -> store")
+    return {"kind": "cot_legacy", "ok": latest_ok, "wrote": wrote}

src/cotdata/providers/cftc_disagg.py

@@ -84,11 +84,12 @@ def _parse_zip(zip_path: Path) -> pd.DataFrame:
     return df
 
 
-def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None:
+def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> dict:
     """Download + parse CFTC Disaggregated futures COT; write full per-code history.
 
     codes: iterable of CFTC codes; default = all registry codes.
-    Rebuilds the complete per-code table each run.
+    Rebuilds the complete per-code table each run. Returns {"kind", "ok", "wrote"};
+    ``ok`` is False only on a hard failure to fetch the current year.
     """
     code_to_sym = {}
     for s in all_symbols():
@@ -120,34 +121,42 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None:
                 print(f"  hist_2006_2016 (disagg): parse failed — {e}")
 
     # Fetch individual years for 2017+ (or first_year if it's > 2016)
+    latest_ok = True
     for year in range(max(2017, first_year), last_year + 1):
         zp = _download_url(f"{URL_PREFIX}{year}.zip", f"fut_disagg_txt_{year}.zip")
         if not zp:
+            if year == last_year:
+                latest_ok = False  # couldn't fetch current year — may have missed a release
             continue
         try:
             frames.append(_parse_zip(zp))
         except Exception as e:  # noqa: BLE001
             print(f"  {year} (disagg): parse failed — {e}")
+            if year == last_year:
+                latest_ok = False
 
     if not frames:
         print("cftc_disagg: no data parsed")
-        return
+        return {"kind": "cot_disagg", "ok": False, "wrote": 0}
 
     allrows = pd.concat(frames, ignore_index=True)
-    
+
     # Parquet cannot serialize mixed-type object columns after concat (due to NaNs)
     for col in allrows.select_dtypes(include=['object']).columns:
         if col not in [CONTRACT_CODE, REPORT_DATE]:
             allrows[col] = allrows[col].astype(str)
 
+    wrote = 0
     for code in sorted(want):
         sub = allrows[allrows[CONTRACT_CODE] == code].copy()
         if sub.empty:
             continue
-        
+
         # Index by report date (DatetimeIndex → manifest last_date)
         sub = sub.sort_values(REPORT_DATE).set_index(REPORT_DATE)
         sym_name = code_to_sym.get(code)
         file_name = f"{sym_name}_{code}" if sym_name else code
         store.write_cot_disagg(file_name, sub, source="cftc_disagg")
+        wrote += 1
         print(f"{file_name}: {len(sub):5d} weeks (disagg) -> store")
+    return {"kind": "cot_disagg", "ok": latest_ok, "wrote": wrote}

src/cotdata/providers/cftc_tff.py

@@ -84,11 +84,12 @@ def _parse_zip(zip_path: Path) -> pd.DataFrame:
     return df
 
 
-def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None:
+def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> dict:
     """Download + parse CFTC TFF futures COT; write full per-code history.
 
     codes: iterable of CFTC codes; default = all registry codes.
-    Rebuilds the complete per-code table each run.
+    Rebuilds the complete per-code table each run. Returns {"kind", "ok", "wrote"};
+    ``ok`` is False only on a hard failure to fetch the current year.
     """
     code_to_sym = {}
     for s in all_symbols():
@@ -120,34 +121,42 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> None:
                 print(f"  hist_2006_2016 (tff): parse failed — {e}")
 
     # Fetch individual years for 2017+ (or first_year if it's > 2016)
+    latest_ok = True
     for year in range(max(2017, first_year), last_year + 1):
         zp = _download_url(f"{URL_PREFIX}{year}.zip", f"fut_fin_txt_{year}.zip")
         if not zp:
+            if year == last_year:
+                latest_ok = False  # couldn't fetch current year — may have missed a release
             continue
         try:
             frames.append(_parse_zip(zp))
         except Exception as e:  # noqa: BLE001
             print(f"  {year} (tff): parse failed — {e}")
+            if year == last_year:
+                latest_ok = False
 
     if not frames:
         print("cftc_tff: no data parsed")
-        return
+        return {"kind": "cot_tff", "ok": False, "wrote": 0}
 
     allrows = pd.concat(frames, ignore_index=True)
-    
+
     # Parquet cannot serialize mixed-type object columns after concat (due to NaNs)
     for col in allrows.select_dtypes(include=['object']).columns:
         if col not in [CONTRACT_CODE, REPORT_DATE]:
             allrows[col] = allrows[col].astype(str)
 
+    wrote = 0
     for code in sorted(want):
         sub = allrows[allrows[CONTRACT_CODE] == code].copy()
         if sub.empty:
             continue
-        
+
         # Index by report date (DatetimeIndex → manifest last_date)
         sub = sub.sort_values(REPORT_DATE).set_index(REPORT_DATE)
         sym_name = code_to_sym.get(code)
         file_name = f"{sym_name}_{code}" if sym_name else code
         store.write_cot_tff(file_name, sub, source="cftc_tff")
+        wrote += 1
         print(f"{file_name}: {len(sub):5d} weeks (tff) -> store")
+    return {"kind": "cot_tff", "ok": latest_ok, "wrote": wrote}

src/cotdata/providers/norgate.py

@@ -10,6 +10,7 @@
 """
 from __future__ import annotations  # PEP 604 unions (dict | None) on Python 3.9
 
+import datetime as dt
 import pandas as pd
 import numpy as np
 import re
@@ -19,6 +20,13 @@
 from .. import store
 
 CCB_SUFFIX = "_CCB"  # Norgate "Continuous Contract Back-adjusted"
+
+# Norgate database names (from norgatedata.databases()) that cotdata reads: the
+# continuous series and the individual contracts used for volume reconstruction.
+FINAL_DATABASES = ("Futures", "Continuous Futures")
+# Default local-time cutoff after which Norgate's "Final" futures prices are in
+# (≈ Continuous Futures Final at ~8:55pm ET per Norgate's update schedule).
+DEFAULT_FINAL_CUTOFF = "20:55"
 # If roll-day overnight moves exceed this multiple of the normal-day median, the
 # series looks UNADJUSTED (calendar-spread gaps not stitched). Self-calibrating
 # per symbol, so it works across products with different spread magnitudes.
@@ -226,6 +234,47 @@ def _check_roll_gaps(internal_symbol: str, df: pd.DataFrame) -> bool:
     return False
 
 
+def _to_naive_local(t):
+    """norgatedata returns tz-AWARE local datetimes (e.g. ...-04:00); normalize to
+    naive local so we can compare against a naive local cutoff. Naive inputs (older
+    norgatedata) are assumed already local and passed through."""
+    if t is None:
+        return None
+    if t.tzinfo is not None:
+        t = t.astimezone().replace(tzinfo=None)  # → local wall-clock, drop tzinfo
+    return t
+
+
+def _finals_ready(db_times: dict, cutoff: str = DEFAULT_FINAL_CUTOFF, now=None):
+    """Pure core of :func:`finals_ready` — testable without norgatedata.
+
+    db_times maps database name → its last-update datetime (tz-aware or naive local,
+    or None). Ready when every database was refreshed at/after today's `cutoff`
+    (local HH:MM). Returns (ready: bool, detail: dict)."""
+    now = now or dt.datetime.now()
+    h, m = (int(x) for x in cutoff.split(":"))
+    cut = now.replace(hour=h, minute=m, second=0, microsecond=0)
+    detail, ready = {}, True
+    for db, t in db_times.items():
+        detail[db] = t.isoformat() if t else None
+        tt = _to_naive_local(t)
+        if tt is None or tt < cut:
+            ready = False
+    detail["cutoff"] = cut.isoformat()
+    return ready, detail
+
+
+def finals_ready(cutoff: str = DEFAULT_FINAL_CUTOFF, now=None):
+    """True once Norgate has this day's FINAL futures prices — i.e. it has refreshed
+    both the 'Futures' and 'Continuous Futures' databases at/after today's local
+    `cutoff`. Uses norgatedata.last_database_update_time (the local PC time of the
+    last DB refresh). Lets a scheduled run avoid capturing interim (non-final) bars.
+    Returns (ready: bool, detail: dict)."""
+    import norgatedata  # Windows producer only
+    times = {db: norgatedata.last_database_update_time(db) for db in FINAL_DATABASES}
+    return _finals_ready(times, cutoff, now)
+
+
 def update(symbols=None, full: bool = False) -> None:
     """Fetch + write to the store for the given internal symbols (backadj and unadj).
 

src/cotdata/update.py

@@ -22,6 +22,14 @@ def main() -> None:
     p.add_argument("--full", action="store_true",
                    help="Full rebuild of reconstructed volume (ignore the incremental "
                         "60-day window). Use after a reconstruction-logic change.")
+    p.add_argument("--require-final", action="store_true",
+                   help="For --prices: only fetch once Norgate's FINAL futures prices are "
+                        "in (last_database_update_time for 'Futures' and 'Continuous Futures' "
+                        ">= --final-cutoff today). Otherwise defer with a non-zero exit so a "
+                        "scheduler retries. Avoids capturing interim (non-final) bars.")
+    p.add_argument("--final-cutoff", default="20:55", metavar="HH:MM",
+                   help="Local time after which Norgate's futures Finals are expected "
+                        "(default 20:55, ≈ Continuous Futures Final in ET).")
     p.add_argument("--check", action="store_true",
                    help="Print store status (row counts, newest data, staleness) from "
                         "the manifest and exit. Read-only, cross-platform, no network.")
@@ -61,37 +69,65 @@ def main() -> None:
 
     kinds = []
     last_run = None
+    failed_kinds = []  # domains that hard-failed → non-zero exit so a scheduler retries
+    deferred = []      # work skipped because inputs aren't ready yet (also non-zero exit)
     if args.prices:
-        last_run = norgate.update(symbols=args.symbols, full=args.full)
-        kinds.append("prices")
+        ready = True
+        if args.require_final:
+            ready, detail = norgate.finals_ready(args.final_cutoff)
+            if not ready:
+                print(f"prices: Norgate Finals not in yet (--require-final, cutoff "
+                      f"{args.final_cutoff}) — deferring. {detail}")
+                deferred.append("prices")
+        if ready:
+            last_run = norgate.update(symbols=args.symbols, full=args.full)
+            kinds.append("prices")
+            if last_run and last_run.get("symbols_failed"):
+                failed_kinds.append("prices")
 
     if args.metadata:
         norgate.update_metadata(symbols=args.symbols)
         kinds.append("metadata")
 
     if args.cot_legacy or args.cot_all:
         from .providers import cftc
-        cftc.update()
+        r = cftc.update()
         kinds.append("cot_legacy")
+        if not (r or {}).get("ok", True):
+            failed_kinds.append("cot_legacy")
 
     if args.cot_disagg or args.cot_all:
         from .providers import cftc_disagg
-        cftc_disagg.update()
+        r = cftc_disagg.update()
         kinds.append("cot_disagg")
+        if not (r or {}).get("ok", True):
+            failed_kinds.append("cot_disagg")
 
     if args.cot_tff or args.cot_all:
         from .providers import cftc_tff
-        cftc_tff.update()
+        r = cftc_tff.update()
         kinds.append("cot_tff")
+        if not (r or {}).get("ok", True):
+            failed_kinds.append("cot_tff")
 
     # Structured heartbeat for downstream tools: rebuild status.json from the now-
     # updated manifest. Pollers detect new data via newest_data[<domain>].
     from . import status
     run = dict(last_run or {})
     run["kinds"] = kinds
+    run["failed"] = failed_kinds
+    run["deferred"] = deferred
     run["at"] = _dt.datetime.utcnow().isoformat(timespec="seconds") + "Z"
     path = status.write_status_file(last_run=run)
     print(f"status written -> {path}")
 
+    # Exit non-zero so Task Scheduler / cron retries, either on a hard failure
+    # (source unreachable) or when --require-final deferred because the Finals
+    # aren't in yet. Ordinary "no new data yet" is NOT a failure.
+    if failed_kinds or deferred:
+        parts = ([f"failed: {', '.join(failed_kinds)}"] if failed_kinds else []) + \
+                ([f"deferred: {', '.join(deferred)}"] if deferred else [])
+        raise SystemExit("cotdata-update: " + " | ".join(parts))
+
 if __name__ == "__main__":
     main()

tests/test_cli_exit.py

@@ -0,0 +1,60 @@
+"""cotdata-update exit codes: non-zero on a hard fetch failure so a scheduler
+(Windows Task Scheduler / cron) can retry, zero on success or 'no new data'."""
+import sys
+from unittest import mock
+
+import pytest
+
+
+def _argv(monkeypatch, tmp_path, *args):
+    monkeypatch.setenv("COTDATA_STORE", str(tmp_path))
+    monkeypatch.setattr(sys, "argv", ["cotdata-update", *args])
+
+
+def test_exits_nonzero_when_cot_hard_fails(tmp_path, monkeypatch):
+    _argv(monkeypatch, tmp_path, "--cot-legacy")
+    from cotdata import update
+    with mock.patch("cotdata.providers.cftc.update",
+                    return_value={"kind": "cot_legacy", "ok": False, "wrote": 0}):
+        with pytest.raises(SystemExit) as ei:
+            update.main()
+    assert ei.value.code not in (0, None)  # non-zero exit
+
+
+def test_exits_zero_on_cot_success(tmp_path, monkeypatch):
+    _argv(monkeypatch, tmp_path, "--cot-legacy")
+    from cotdata import update
+    with mock.patch("cotdata.providers.cftc.update",
+                    return_value={"kind": "cot_legacy", "ok": True, "wrote": 5}):
+        update.main()  # must not raise SystemExit
+
+
+def test_exits_nonzero_when_prices_have_failures(tmp_path, monkeypatch):
+    _argv(monkeypatch, tmp_path, "--prices")
+    from cotdata import update
+    with mock.patch("cotdata.providers.norgate.update",
+                    return_value={"kind": "prices", "symbols_failed": ["GC"], "ok": []}):
+        with pytest.raises(SystemExit) as ei:
+            update.main()
+    assert ei.value.code not in (0, None)
+
+
+def test_require_final_defers_when_not_ready(tmp_path, monkeypatch):
+    _argv(monkeypatch, tmp_path, "--prices", "--require-final")
+    from cotdata import update
+    with mock.patch("cotdata.providers.norgate.finals_ready", return_value=(False, {"Futures": None})), \
+         mock.patch("cotdata.providers.norgate.update") as m_update:
+        with pytest.raises(SystemExit) as ei:
+            update.main()
+    assert ei.value.code not in (0, None)   # non-zero -> scheduler retries
+    m_update.assert_not_called()            # did NOT capture interim prices
+
+
+def test_require_final_runs_when_ready(tmp_path, monkeypatch):
+    _argv(monkeypatch, tmp_path, "--prices", "--require-final")
+    from cotdata import update
+    with mock.patch("cotdata.providers.norgate.finals_ready", return_value=(True, {})), \
+         mock.patch("cotdata.providers.norgate.update",
+                    return_value={"kind": "prices", "symbols_failed": [], "ok": ["ES"]}) as m_update:
+        update.main()                       # must not raise
+    m_update.assert_called_once()