Databento two-stage price producer + per-symbol provider selection (ADR-0006)

README.md

@@ -22,6 +22,8 @@ cotdata separates *fetching* data (a "producer" that talks to vendors) from *usi
 |------|--------|------|---------|
 | CFTC COT — legacy / disaggregated / TFF | [cftc.gov](https://www.cftc.gov/) | **Free** | any OS |
 | Futures prices + contract specs | [Norgate Data](https://norgatedata.com/) | Paid subscription | **Windows** (producer only) |
+| Futures prices, back-adjusted | [Databento](https://databento.com/) GLBX.MDP3 | Paid per query | any OS |
+| Futures prices, research-grade fallback | Yahoo Finance | **Free** | any OS |
 | *Reading the store* (any of the above) | — | Free | any OS |
 
 ## Contents
@@ -116,7 +118,7 @@ Date
 
 ## Producing data (producer)
 
-Run on the machine that can reach the source. Norgate prices require Windows; CFTC COT runs anywhere.
+Run on the machine that can reach the source. Norgate prices require Windows, CFTC COT runs anywhere, and a server without Norgate can build prices from Databento instead (see [Cross-platform prices without Norgate](#cross-platform-prices-without-norgate-databento)).
 
 ```bash
 COTDATA_STORE=/store  cotdata-update --prices                    # Norgate prices, ALL registry symbols (Windows)
@@ -138,6 +140,35 @@ pip install "cotdata[norgate]"     # adds the norgatedata dependency (Windows)
 
 The `norgatedata` package talks locally to the Norgate Data Updater application — there are no API keys. You just need the Updater installed, authenticated, and running.
 
+### Cross-platform prices without Norgate (databento)
+
+A server that cannot run Norgate (for example the public dashboard host) can build the price store from Databento instead. One provider owns each symbol end to end, so this is a full replacement, not a blend. Install the extras and set the environment:
+
+```bash
+pip install "cotdata[databento,yahoo]"      # databento producer + the Yahoo fallback
+
+export COTDATA_STORE=/path/to/store         # the store the dashboard reads
+export COTDATA_PRICE_SOURCE=databento       # deployment default, so softs/MSCI fall to Yahoo
+export DATABENTO_API_KEY=db-...             # for the paid ingest step only
+# optional: export COTDATA_DATABENTO_RAW=/path/to/raw   # defaults to $COTDATA_STORE/_raw/databento
+```
+
+Then build the store in order (ingest before build):
+
+```bash
+cotdata-update --ingest-databento     # Stage 1 (PAID): raw .n.0/.n.1 ohlcv-1d + statistics -> raw store
+cotdata-update --build-databento      # Stage 2 (FREE): additive back-adjustment -> $COTDATA_STORE/prices
+cotdata-update --prices-yahoo         # softs, lumber, MSCI proxies (resolve to Yahoo on this deployment)
+cotdata-update --cot-all              # CFTC COT, the dashboard needs it too
+cotdata-update --check                # coverage, newest dates, staleness
+```
+
+- **Two stages, one paid.** Stage 1 is the only step that hits the API. It writes an append-only raw store and resumes from the last fetched date, so re-runs pull only new days. Stage 2 reads that raw store with no API cost, so the back-adjustment can be iterated offline. The raw store is producer-internal, so keep it out of any sync to consumers.
+- **History starts 2010-06-06** (the GLBX floor), shallower than Norgate. Markets not on CME Globex (ICE softs, lumber, MSCI intl) fall back to Yahoo.
+- **First-run check.** A healthy symbol prints `built unadj+backadj (N bars, K rolls)`. If it prints `no rolls detected`, back-adjustment is a no-op for that symbol, so investigate before trusting it.
+- **Validate against Norgate** (optional gate) with `scripts/validate_databento_vs_norgate.py` if you have both stores.
+- **Schedule** the two price commands nightly and `--cot-all` weekly (cron on Linux, the same idea as the Windows section below).
+
 ### Scheduling on Windows (Task Scheduler)
 
 The goal: **prices daily**, and **COT caught within minutes of its Friday ~3:30pm ET release** while surviving holiday delays. Two properties make this simple:
@@ -206,6 +237,55 @@ foreach ($t in "cotdata prices","cotdata COT (Fri release)","cotdata COT (catch-
 
 **View / manage the jobs** any time in the Windows **Task Scheduler** GUI — press `Win + R` and run `taskschd.msc`, or open *Task Scheduler* from the Start menu — then look under **Task Scheduler Library** for the `cotdata …` tasks.
 
+### Scheduling on Linux (cron)
+
+A databento server schedules the same way: **prices nightly**, **COT caught soon after its Friday ~3:30pm ET release**, with a daily catch-up for holiday delays. The same two properties hold:
+
+- **Idempotent.** `--cot-all` HEAD-checks each CFTC year zip and skips it if unchanged. `--ingest-databento` resumes from the last fetched date, so a re-run pulls only new days. Running before new data lands is a harmless no-op.
+- **Fails loudly.** A run exits non-zero only on a hard fetch error (source unreachable), not when there is simply nothing new. Because ingest is resumable and COT is idempotent, a failed or missed run is picked up by the next one, so no explicit retry logic is needed.
+
+Cron runs with a bare environment, so put the config and the venv path in a wrapper script (one per command, mirroring the Windows pair). Replace the `<...>` placeholders: `<STORE>` = your store, `<VENV>` = your virtualenv, `<KEY>` = your Databento key, `<DIR>` = the folder holding these scripts.
+
+`run-prices.sh` — the two-stage databento build plus the Yahoo fallback:
+
+```bash
+#!/usr/bin/env bash
+set -euo pipefail
+export COTDATA_STORE=<STORE>
+export COTDATA_PRICE_SOURCE=databento
+export DATABENTO_API_KEY=<KEY>
+BIN=<VENV>/bin/cotdata-update
+"$BIN" --ingest-databento     # Stage 1 (paid): raw .n.0/.n.1 -> raw store
+"$BIN" --build-databento      # Stage 2 (free): back-adjusted prices
+"$BIN" --prices-yahoo         # softs / lumber / MSCI fallback
+```
+
+`run-cot.sh` — COT (note the different command):
+
+```bash
+#!/usr/bin/env bash
+set -euo pipefail
+export COTDATA_STORE=<STORE>
+<VENV>/bin/cotdata-update --cot-all
+```
+
+Make them executable (`chmod +x run-prices.sh run-cot.sh`), then add the jobs with `crontab -e`. Cron uses the **server's local** timezone, so convert the ET times below if it is not on Eastern (or set the server to a known zone). `flock` stops a slow run from overlapping the next, and the redirect keeps a log:
+
+```cron
+# Prices — nightly (Mon-Sat). GLBX settlements are disseminated the morning after the
+# session, so an early-morning run captures the prior session's finalized settlement.
+30 6 * * 1-6     flock -n /tmp/cotdata-prices.lock <DIR>/run-prices.sh >> <DIR>/prices.log 2>&1
+
+# COT — daily morning catch-up (holiday-delayed releases and a safety net).
+10 8 * * *       flock -n /tmp/cotdata-cot.lock <DIR>/run-cot.sh >> <DIR>/cot.log 2>&1
+
+# COT — Friday release window: every 2 min across the ~3:30pm ET release (times in ET,
+# each run a cheap no-op until the zip changes). Convert to the server's local time.
+*/2 15-16 * * 5  flock -n /tmp/cotdata-cot.lock <DIR>/run-cot.sh >> <DIR>/cot.log 2>&1
+```
+
+Set `MAILTO=you@example.com` at the top of the crontab to have cron email the output of any run that writes to stderr or exits non-zero. For tighter alerting, point your monitoring at the log files or the store's `status.json` (see [Operations](#operations)), or run the wrappers from a systemd timer with `OnFailure=`. Check coverage and freshness any time with `cotdata-update --check`.
+
 **Monitoring:** after any run, `status.json` reflects `newest_data.<domain>` and `last_run.symbols_failed` — poll it to confirm the Friday COT actually advanced, or to alert on failures (see [Operations](#operations)).
 
 > The Friday window intentionally over-polls (every 2 minutes across a 45-minute window); idempotency makes every run after the release lands a no-op. If you'd rather actively wait out *late* releases, a wrapper can loop until `status.json`'s `newest_data.cot_legacy` reaches the expected Tuesday — but daily catch-up already covers holiday slips with far less machinery.
@@ -272,8 +352,11 @@ Norgate publishes continuous futures in only two forms: unadjusted and **additiv
 
 ### Providers & authentication
 
-- **Norgate Data (primary prices).** No Python API keys — the `norgatedata` package talks locally to the Norgate Data Updater app, which must be installed, authenticated, and running on Windows.
-- **Databento (dormant / intraday).** Kept as a dormant provider for potential intraday use. If enabled, provide `DATABENTO_API_KEY` via the environment.
+Which vendor prices a symbol is a deployment choice, not a fixed fact (the same ES is Norgate for local research and databento on a public-dash server). It is resolved when a producer runs, from three inputs: the deployment default `COTDATA_PRICE_SOURCE` (`norgate` if unset), per-symbol capability (the `norgate` / `databento` / `yahoo` mappings in the registry, `null` where a vendor has no series), and an optional per-symbol `price_source` override. A symbol uses its override if set, otherwise the default when that vendor can serve it, otherwise a Yahoo fallback where a ticker exists. Each producer writes only the symbols that resolve to it, so a symbol is never blended across vendors. One provider owns each symbol end to end.
+
+- **Norgate Data (paid, Windows).** No Python API key. The `norgatedata` package talks locally to the Norgate Data Updater app, which must be installed, authenticated, and running on Windows. This is the default for local research (`cotdata-update --prices`).
+- **Databento (paid, cross-platform).** A two-stage producer for a server that cannot run Norgate. Stage 1 (`cotdata-update --ingest-databento`) pulls raw `.n.0` / `.n.1` `ohlcv-1d` and `statistics` into an append-only raw store (`$COTDATA_DATABENTO_RAW`, else `_raw/databento` under the store). This is the only paid step, and it is resumable, so re-runs fetch only new dates. Stage 2 (`cotdata-update --build-databento`) derives the back-adjusted prices from the raw store with no API cost, so the build logic can be iterated offline. Set `DATABENTO_API_KEY` and `COTDATA_PRICE_SOURCE=databento`. History starts 2010-06-06, and markets not on CME Globex (ICE softs, lumber, MSCI intl) fall back to Yahoo. The raw store is producer-internal, so exclude it from any consumer sync.
+- **Yahoo Finance (free, research-grade).** `cotdata-update --prices-yahoo` prices the markets that resolve to yfinance on this deployment: the MSCI ETF proxies always, plus the softs and lumber on a databento server. Expect gaps and silent revisions. This is not a production replacement for the paid feeds. Requires the `[yahoo]` extra.
 
 ### The symbol registry
 

pyproject.toml

@@ -34,7 +34,8 @@ Issues = "https://github.com/mspinola/cotdata/issues"
 [project.optional-dependencies]
 # Producer-only, Windows: the Norgate Data Updater must be installed & running.
 norgate = ["norgatedata"]
-# Dormant provider, retained for the intraday news-failure work + settlement cross-check.
+# Cross-platform price producer: two-stage ingest/build back-adjustment (see the
+# Providers section of the README and ADR-0006). Also serves the intraday work.
 databento = ["databento>=0.30"]
 
 yahoo = ["yfinance"]

scripts/validate_databento_vs_norgate.py

@@ -0,0 +1,187 @@
+#!/usr/bin/env python
+"""ADR-0006 item 6: validate databento back-adjustment against Norgate.
+
+The databento and Norgate `backadj` series are built independently (single provider
+per symbol, no stitching), so their absolute LEVELS may differ: different roll
+calendars pick different roll dates, and the additive back-adjust anchor floats to
+each provider's most-recent price. What must agree is the SHAPE.
+
+Additive (Panama) back-adjustment preserves absolute daily price CHANGES, not percent
+returns — a floating anchor changes the price level and hence the return denominator,
+but `Close.diff()` on any segment equals the true settlement change either way. So the
+core check is: daily changes match between the two series, except near roll dates the
+two providers place differently. This harness quantifies that.
+
+It reads `backadj` for a few liquid symbols from two stores — one built by Norgate,
+one built by databento (`cotdata-update --build-databento`) — and reports, per symbol:
+  * overlap (date spans, common days),
+  * change correlation, scale ratio, and normalized worst-day change diff (the shape
+    check, unitless so one tolerance works across symbols),
+  * level difference stats (expected non-zero; informational),
+  * roll-date agreement (from Delivery Month changes).
+
+It exits non-zero if any symbol falls outside tolerance, so it can gate a promotion.
+It needs real data, so it is NOT run in CI — the comparison logic is unit tested in
+tests/test_validate_databento.py against synthetic frames.
+
+Usage:
+    python scripts/validate_databento_vs_norgate.py \
+        --norgate-store /path/to/norgate_store \
+        --databento-store /path/to/databento_store \
+        --symbols ES CL GC
+"""
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+import pandas as pd
+
+_DEFAULT_SYMBOLS = ["ES", "CL", "GC"]
+# Defaults are deliberately lenient — tighten once you have seen a clean run. A few
+# days a year land on a mismatched roll and legitimately differ, so judge on the
+# correlation and the bulk of days, not a single worst day.
+_DEFAULT_CHANGE_CORR_MIN = 0.999
+_DEFAULT_SCALE_BAND = 0.02          # |scale_ratio - 1| allowed (catches unit/settlement scale bugs)
+_DEFAULT_REL_TOL = 0.10             # worst |Δchange| as a fraction of a typical daily change
+_DEFAULT_MAX_OUTLIER_DAYS = 8       # days/yr allowed to exceed REL_TOL (roll-date mismatches)
+
+
+def read_backadj(store_path: str, symbol: str) -> pd.DataFrame:
+    """Read one symbol's backadj OHLC frame straight from a store's parquet, normalized
+    the way get_prices would (tz-naive daily Date index). Empty if absent."""
+    p = Path(store_path) / "prices" / f"{symbol}_backadj.parquet"
+    if not p.exists():
+        return pd.DataFrame()
+    df = pd.read_parquet(p)
+    df.index = pd.to_datetime(df.index).tz_localize(None).normalize()
+    df.index.name = "Date"
+    return df.sort_index()
+
+
+def _roll_dates(df: pd.DataFrame) -> pd.DatetimeIndex:
+    """Roll dates = days the Delivery Month changes. Empty if the column is absent.
+    Values differ between providers (Norgate '202606' vs databento 'ESM6'), but the
+    DATES they change on are comparable."""
+    if "Delivery Month" not in df.columns:
+        return pd.DatetimeIndex([])
+    dm = df["Delivery Month"]
+    changed = dm.ne(dm.shift()) & dm.shift().notna()
+    return df.index[changed.fillna(False)]
+
+
+def compare(norgate: pd.DataFrame, databento: pd.DataFrame, symbol: str = "?") -> dict:
+    """Compare two backadj frames over their date overlap. Returns a metrics dict."""
+    idx = norgate.index.intersection(databento.index)
+    m: dict = {"symbol": symbol, "n_common": int(len(idx))}
+    if len(idx) < 3:
+        m["error"] = "insufficient overlap"
+        return m
+
+    n = norgate.loc[idx, "Close"].astype(float)
+    d = databento.loc[idx, "Close"].astype(float)
+
+    # Shape check on daily CHANGES (additive back-adjust preserves changes, not returns).
+    nc = n.diff().dropna()
+    dc = d.diff().reindex(nc.index)
+    scale = float(nc.abs().median()) or 1.0        # a typical daily change, for unitless norm
+    rel = (nc - dc).abs() / scale
+    m["change_corr"] = float(nc.corr(dc))
+    m["scale_ratio"] = float(dc.std() / nc.std()) if float(nc.std()) else float("nan")
+    m["change_max_rel_diff"] = float(rel.max())
+    m["change_rmse_points"] = float(((nc - dc) ** 2).mean() ** 0.5)
+
+    # Level difference: informational (a piecewise-constant offset is expected).
+    lvl = (n - d).abs()
+    m["level_mean_abs"] = float(lvl.mean())
+    m["level_max_abs"] = float(lvl.max())
+
+    # Coverage + roll agreement.
+    m["norgate_span"] = f"{n.index.min().date()}..{n.index.max().date()}"
+    m["databento_span"] = f"{d.index.min().date()}..{d.index.max().date()}"
+    nr_rolls = _roll_dates(norgate).intersection(idx)
+    db_rolls = _roll_dates(databento).intersection(idx)
+    m["rolls_norgate"] = int(len(nr_rolls))
+    m["rolls_databento"] = int(len(db_rolls))
+    m["rolls_common"] = int(len(nr_rolls.intersection(db_rolls)))
+
+    years = max((idx.max() - idx.min()).days / 365.25, 1e-9)
+    m["outlier_days"] = int((rel > _DEFAULT_REL_TOL).sum())
+    m["outlier_days_per_yr"] = m["outlier_days"] / years
+    return m
+
+
+def evaluate(m: dict, corr_min: float, scale_band: float, rel_tol: float,
+             max_outliers_per_yr: float) -> list:
+    """Return a list of failure reasons for one symbol's metrics (empty = pass)."""
+    if "error" in m:
+        return [m["error"]]
+    fails = []
+    if m["change_corr"] < corr_min:
+        fails.append(f"change_corr {m['change_corr']:.5f} < {corr_min}")
+    if abs(m["scale_ratio"] - 1.0) > scale_band:
+        fails.append(f"scale_ratio {m['scale_ratio']:.4f} off 1.0 by > {scale_band} "
+                     f"(unit/settlement scale mismatch?)")
+    # A big worst-day diff is only a failure if it happens on too many days (roll noise
+    # on a handful of days a year is expected and fine).
+    n_over = round(m["outlier_days_per_yr"], 1)
+    if m["change_max_rel_diff"] > rel_tol and m["outlier_days_per_yr"] > max_outliers_per_yr:
+        fails.append(f"{n_over} days/yr exceed rel tol {rel_tol} "
+                     f"(worst {m['change_max_rel_diff']:.2f}x a typical day)")
+    return fails
+
+
+def format_report(m: dict, fails: list) -> str:
+    if "error" in m:
+        return f"  {m['symbol']:<5} SKIP  ({m['error']}, n_common={m['n_common']})"
+    status = "PASS" if not fails else "FAIL"
+    lines = [
+        f"  {m['symbol']:<5} {status}  common={m['n_common']}  "
+        f"norgate={m['norgate_span']}  databento={m['databento_span']}",
+        f"        change_corr={m['change_corr']:.5f}  scale_ratio={m['scale_ratio']:.4f}  "
+        f"worst_rel={m['change_max_rel_diff']:.2f}  rmse={m['change_rmse_points']:.4f}pts  "
+        f"outliers={m['outlier_days']} ({m['outlier_days_per_yr']:.1f}/yr)",
+        f"        level|Δ| mean={m['level_mean_abs']:.4f} max={m['level_max_abs']:.4f}  "
+        f"rolls n/db/common={m['rolls_norgate']}/{m['rolls_databento']}/{m['rolls_common']}",
+    ]
+    if fails:
+        lines.append("        FAIL: " + "; ".join(fails))
+    return "\n".join(lines)
+
+
+def main(argv=None) -> int:
+    p = argparse.ArgumentParser(description="Validate databento backadj against Norgate.")
+    p.add_argument("--norgate-store", required=True, help="Path to the Norgate-built store.")
+    p.add_argument("--databento-store", required=True, help="Path to the databento-built store.")
+    p.add_argument("--symbols", nargs="+", default=_DEFAULT_SYMBOLS)
+    p.add_argument("--change-corr-min", type=float, default=_DEFAULT_CHANGE_CORR_MIN)
+    p.add_argument("--scale-band", type=float, default=_DEFAULT_SCALE_BAND)
+    p.add_argument("--rel-tol", type=float, default=_DEFAULT_REL_TOL)
+    p.add_argument("--max-outlier-days-per-yr", type=float, default=_DEFAULT_MAX_OUTLIER_DAYS)
+    args = p.parse_args(argv)
+
+    print(f"Validating databento vs Norgate backadj on: {', '.join(args.symbols)}")
+    any_fail, any_data = False, False
+    for sym in args.symbols:
+        ng = read_backadj(args.norgate_store, sym)
+        db = read_backadj(args.databento_store, sym)
+        if ng.empty or db.empty:
+            missing = ", ".join(s for s, df in (("norgate", ng), ("databento", db)) if df.empty)
+            print(f"  {sym:<5} SKIP  (no backadj in: {missing})")
+            continue
+        any_data = True
+        m = compare(ng, db, sym)
+        fails = evaluate(m, args.change_corr_min, args.scale_band, args.rel_tol,
+                         args.max_outlier_days_per_yr)
+        any_fail = any_fail or bool(fails)
+        print(format_report(m, fails))
+
+    if not any_data:
+        print("No comparable symbols found in both stores.")
+        return 2
+    return 1 if any_fail else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())