diff --git a/README.md b/README.md index fa134cd..6983d13 100644 --- a/README.md +++ b/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: `` = your store, `` = your virtualenv, `` = your Databento key, `` = 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= +export COTDATA_PRICE_SOURCE=databento +export DATABENTO_API_KEY= +BIN=/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= +/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 /run-prices.sh >> /prices.log 2>&1 + +# COT — daily morning catch-up (holiday-delayed releases and a safety net). +10 8 * * * flock -n /tmp/cotdata-cot.lock /run-cot.sh >> /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 /run-cot.sh >> /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.` 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 diff --git a/pyproject.toml b/pyproject.toml index f7a01fa..5ad3d11 100644 --- a/pyproject.toml +++ b/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"] diff --git a/scripts/validate_databento_vs_norgate.py b/scripts/validate_databento_vs_norgate.py new file mode 100644 index 0000000..85f5871 --- /dev/null +++ b/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()) diff --git a/src/cotdata/providers/cftc.py b/src/cotdata/providers/cftc.py index de670c4..9b18604 100644 --- a/src/cotdata/providers/cftc.py +++ b/src/cotdata/providers/cftc.py @@ -95,7 +95,9 @@ def _parse_zip(zip_path: Path) -> pd.DataFrame: data = fh.read() df = pd.read_excel(io.BytesIO(data), usecols=TARGET_COLS) # .xls → needs xlrd df[CONTRACT_CODE] = df[CONTRACT_CODE].apply(_standardize_code) - df[REPORT_DATE] = pd.to_datetime(df[REPORT_DATE]).dt.tz_localize(None) + df[REPORT_DATE] = pd.to_datetime(df[REPORT_DATE], format='mixed').dt.tz_localize(None) + if not df.empty and df[REPORT_DATE].max() > pd.Timestamp.today().normalize() + pd.Timedelta(days=7): + raise ValueError(f"Date parsing sanity check failed: found future date {df[REPORT_DATE].max()}") return df diff --git a/src/cotdata/providers/cftc_disagg.py b/src/cotdata/providers/cftc_disagg.py index e7963d1..379149b 100644 --- a/src/cotdata/providers/cftc_disagg.py +++ b/src/cotdata/providers/cftc_disagg.py @@ -73,7 +73,9 @@ def _parse_zip(zip_path: Path) -> pd.DataFrame: # Coerce the key schema columns to match the Legacy schema format df[CONTRACT_CODE] = df[CONTRACT_CODE].apply(_standardize_code) - df[REPORT_DATE] = pd.to_datetime(df[REPORT_DATE]).dt.tz_localize(None) + df[REPORT_DATE] = pd.to_datetime(df[REPORT_DATE], format='mixed').dt.tz_localize(None) + if not df.empty and df[REPORT_DATE].max() > pd.Timestamp.today().normalize() + pd.Timedelta(days=7): + raise ValueError(f"Date parsing sanity check failed: found future date {df[REPORT_DATE].max()}") # Parquet cannot serialize mixed-type object columns (e.g. Traders_Tot_Old contains ints and strings) for col in df.select_dtypes(include=['object']).columns: diff --git a/src/cotdata/providers/cftc_tff.py b/src/cotdata/providers/cftc_tff.py index 35ebc7d..27f25fe 100644 --- a/src/cotdata/providers/cftc_tff.py +++ b/src/cotdata/providers/cftc_tff.py @@ -73,7 +73,9 @@ def _parse_zip(zip_path: Path) -> pd.DataFrame: # Coerce the key schema columns to match the Legacy schema format df[CONTRACT_CODE] = df[CONTRACT_CODE].apply(_standardize_code) - df[REPORT_DATE] = pd.to_datetime(df[REPORT_DATE]).dt.tz_localize(None) + df[REPORT_DATE] = pd.to_datetime(df[REPORT_DATE], format='mixed').dt.tz_localize(None) + if not df.empty and df[REPORT_DATE].max() > pd.Timestamp.today().normalize() + pd.Timedelta(days=7): + raise ValueError(f"Date parsing sanity check failed: found future date {df[REPORT_DATE].max()}") # Parquet cannot serialize mixed-type object columns (e.g. Traders_Tot_Old contains ints and strings) for col in df.select_dtypes(include=['object']).columns: diff --git a/src/cotdata/providers/databento.py b/src/cotdata/providers/databento.py index caf3559..8be6060 100644 --- a/src/cotdata/providers/databento.py +++ b/src/cotdata/providers/databento.py @@ -16,6 +16,8 @@ Requires DATABENTO_API_KEY. The daily path below is superseded by Norgate; the intraday work will reuse this statistics logic against ohlcv-1h / trades schemas. """ +import datetime as dt +import json import logging import os import sys @@ -26,7 +28,7 @@ import pandas as pd -from .. import config +from .. import config, store from ..registry import all_symbols logger = logging.getLogger(__name__) @@ -318,3 +320,329 @@ def update_all_daily_prices(force_refresh: bool = False) -> None: run_batch_backfill(to_batch) except Exception as e: # noqa: BLE001 logger.error("Batch backfill failed: %s", e) + + +# ── Raw ingest (Stage 1): the paid, append-only landing store ──────────────── +# ADR-0006: databento is a two-stage producer. Stage 1 (here) is the ONLY step +# that hits the paid API. It pulls raw .n.0 / .n.1 ohlcv-1d + statistics into an +# immutable, append-only raw store, keyed by fetched date range in a manifest so a +# re-run resumes from last_date+1 and never re-pulls a range already held. The free +# Stage 2 `build` (see ADR item 4) re-derives back-adjusted prices from these local +# files with no API cost. The raw store is PRODUCER-INTERNAL, not the consumer +# contract — keep it out of any store sync to consumers. + +GLBX_HISTORY_FLOOR = "2010-06-06" # earliest GLBX.MDP3 history +_FEEDS = (".n.0", ".n.1") # front + second continuous (second gives the roll gap) +_SCHEMAS = ("ohlcv-1d", "statistics") + + +def raw_root() -> Path: + """Producer-internal databento raw store: $COTDATA_DATABENTO_RAW if set, else a + ``_raw/databento`` namespace under the cotdata store (leading underscore = not a + consumer domain; exclude it from any consumer sync).""" + env = os.environ.get("COTDATA_DATABENTO_RAW", "").strip() + return Path(env) if env else (config.store_root() / "_raw" / "databento") + + +def _raw_path(symbol: str, feed: str, schema: str) -> Path: + sub = "ohlcv" if schema == "ohlcv-1d" else "statistics" + return raw_root() / sub / f"{symbol}{feed}.parquet" + + +def _ingest_manifest_path() -> Path: + return raw_root() / "ingest_manifest.json" + + +def _load_ingest_manifest() -> dict: + p = _ingest_manifest_path() + return json.loads(p.read_text()) if p.exists() else {} + + +def _write_ingest_manifest(m: dict) -> None: + p = _ingest_manifest_path() + p.parent.mkdir(parents=True, exist_ok=True) + tmp = p.with_suffix(".json.tmp") + tmp.write_text(json.dumps(m, indent=2, sort_keys=True)) + os.replace(tmp, p) + + +def _to_naive(x): + """tz-naive UTC for either tz-aware or naive datetime input (databento returns UTC). + Handles both a DatetimeIndex (``.tz_convert``) and a Series (``.dt.tz_convert``).""" + ts = pd.to_datetime(x, utc=True) + return ts.dt.tz_convert(None) if isinstance(ts, pd.Series) else ts.tz_convert(None) + + +def _normalize(raw: pd.DataFrame, schema: str) -> pd.DataFrame: + """Light bronze normalization: tz-naive timestamps, one row per day for ohlcv. + Columns are otherwise preserved as databento returns them, so Stage 2 can + re-extract settlement/OI/etc. without a re-fetch.""" + raw = raw.copy() + if schema == "ohlcv-1d": + raw.index = _to_naive(raw.index).normalize() + raw.index.name = "Date" + return raw[~raw.index.duplicated(keep="last")].sort_index() + # statistics: flatten ts_event out of the index, keep every stat row. + raw = raw.reset_index() + for c in ("ts_event", "ts_ref"): + if c in raw.columns: + raw[c] = _to_naive(raw[c]) + return raw.drop_duplicates().reset_index(drop=True) + + +def _date_bounds(raw: pd.DataFrame, schema: str): + if schema == "ohlcv-1d": + return str(raw.index.min().date()), str(raw.index.max().date()) + if "ts_event" in raw.columns and len(raw): + d = pd.to_datetime(raw["ts_event"]) + return str(d.min().date()), str(d.max().date()) + return None, None + + +def _append_raw(symbol: str, feed: str, schema: str, new_df: pd.DataFrame) -> None: + path = _raw_path(symbol, feed, schema) + existing = pd.read_parquet(path) if path.exists() else pd.DataFrame() + combined = pd.concat([existing, new_df]) if not existing.empty else new_df + if schema == "ohlcv-1d": + combined = combined[~combined.index.duplicated(keep="last")].sort_index() + else: + combined = combined.drop_duplicates().reset_index(drop=True) + store._atomic_write_parquet(combined, path) + + +def _client_from_env(): + import databento as db # lazy — behind the [databento] extra + key = os.environ.get("DATABENTO_API_KEY") + if not key: + raise RuntimeError("DATABENTO_API_KEY is not set; cannot ingest from databento.") + return db.Historical(key=key) + + +def _fetch(client, dataset: str, dbsym: str, schema: str, start: str, end: str) -> pd.DataFrame: + with warnings.catch_warnings(): + warnings.filterwarnings("ignore", message=".*No data found.*") + warnings.filterwarnings("ignore", message=".*did not resolve.*") + # Databento flags a handful of historically "degraded" sessions (e.g. 2014-06-11) + # on every request. It's a known data-condition note, not an error, and it floods + # the ingest/cron logs — quiet it. + warnings.filterwarnings("ignore", message=".*reduced quality.*") + data = client.timeseries.get_range( + dataset=dataset, symbols=[dbsym], stype_in="continuous", + schema=schema, start=start, end=end) + return data.to_df() + + +def ingest(symbols=None, *, client=None, dataset="GLBX.MDP3", end=None, + cold_start=GLBX_HISTORY_FLOOR) -> dict: + """Fetch raw databento daily bars (.n.0 + .n.1) and statistics into the raw + store, append-only. Resumes each (symbol, feed, schema) from its manifest + last_date+1. Scoped to registry symbols that databento can serve (a non-null + ``databento`` mapping); pass ``symbols`` to narrow further. + + `client` is injectable (databento.Historical-shaped) for tests; the default + builds one from DATABENTO_API_KEY. Returns {kind, ok, symbols, rows}.""" + targets = [s for s in all_symbols() + if s.databento and s.price_source in (None, "databento") + and (symbols is None or s.internal in symbols)] + if not targets: + print("databento ingest: no databento-capable symbols" + + (f" among {symbols}" if symbols else "")) + return {"kind": "ingest_databento", "ok": True, "symbols": 0, "rows": 0} + + if client is None: + client = _client_from_env() + + end = end or (pd.Timestamp.now().normalize() - pd.Timedelta(days=1)).strftime("%Y-%m-%d") + manifest = _load_ingest_manifest() + total_rows, failed = 0, 0 + + for s in targets: + for feed in _FEEDS: + dbsym = f"{s.databento}{feed}" + for schema in _SCHEMAS: + key = f"{s.internal}{feed}:{schema}" + rec = manifest.get(key, {}) + last = rec.get("last_date") + start = ((pd.Timestamp(last) + pd.Timedelta(days=1)).strftime("%Y-%m-%d") + if last else cold_start) + if pd.Timestamp(start) < pd.Timestamp(GLBX_HISTORY_FLOOR): + start = GLBX_HISTORY_FLOOR + if pd.Timestamp(start) > pd.Timestamp(end): + continue # already current + try: + raw = _fetch(client, dataset, dbsym, schema, start, end) + except Exception as e: # noqa: BLE001 — databento/network is flaky + print(f"{s.internal}{feed} {schema}: databento fetch failed ({start}..{end}) — {e}") + failed += 1 + continue + if raw is None or raw.empty: + continue + raw = _normalize(raw, schema) + if raw.empty: + continue + _append_raw(s.internal, feed, schema, raw) + first, newest = _date_bounds(raw, schema) + manifest[key] = { + "first_date": rec.get("first_date") or first, + "last_date": newest, + "n_rows": int(rec.get("n_rows", 0)) + len(raw), + "fetched_at": dt.datetime.utcnow().isoformat(timespec="seconds") + "Z", + } + total_rows += len(raw) + print(f"{s.internal}{feed} {schema}: +{len(raw):5d} rows ({start}..{newest}) -> raw store") + + _write_ingest_manifest(manifest) + return {"kind": "ingest_databento", "ok": failed == 0, "symbols": len(targets), "rows": total_rows} + + +# ── Build (Stage 2): free, raw store -> back-adjusted store prices ──────────── +# ADR-0006. Derives two series per symbol from the local raw store (no API cost): +# unadj — raw front continuous (.n.0), settlement close, roll gaps intact. +# backadj — additive back-adjustment, Norgate's method exactly: at each roll the +# gap = new_close - old_close on the roll date = n1_settle - n0_settle; +# every price up to AND INCLUDING the roll date is shifted by the gap; +# gaps accumulate back-to-front so the newest segment stays at real prices. +# Then store.write_prices(..., source='databento'). propadj is derived on read from +# unadj + backadj by the consumer API, unchanged. + +_OHLCV_COLMAP = {"open": "Open", "high": "High", "low": "Low", + "close": "Close", "volume": "Volume"} +_OUT_COLS = ["Open", "High", "Low", "Close", "Volume", "Open Interest"] + + +def _read_ohlcv(symbol: str, feed: str) -> pd.DataFrame: + p = _raw_path(symbol, feed, "ohlcv-1d") + if not p.exists(): + return pd.DataFrame() + df = pd.read_parquet(p).rename(columns=_OHLCV_COLMAP) + # instrument_id is the roll signal: for a continuous series databento's `symbol` + # column is the constant alias ("ES.n.0"), while instrument_id changes to the new + # contract at each roll. Keep both. + keep = [c for c in ("Open", "High", "Low", "Close", "Volume", "instrument_id", "symbol") + if c in df.columns] + df = df[keep].copy() + df.index = pd.to_datetime(df.index).normalize() + df.index.name = "Date" + return df[~df.index.duplicated(keep="last")].sort_index() + + +def _stat_series(symbol: str, feed: str, stat_type: int, date_col: str, value_col: str) -> pd.Series: + """Daily series of a databento statistic — settlement (stat_type 3, dated by + ts_ref, the session it applies to) or Open Interest (stat_type 9, by ts_event).""" + p = _raw_path(symbol, feed, "statistics") + if not p.exists(): + return pd.Series(dtype="float64") + st = pd.read_parquet(p) + if not {"stat_type", date_col, value_col}.issubset(st.columns): + return pd.Series(dtype="float64") + sel = st[st["stat_type"] == stat_type].copy() + if sel.empty: + return pd.Series(dtype="float64") + sel["D"] = pd.to_datetime(sel[date_col]).dt.normalize() + return sel.groupby("D")[value_col].last() + + +def _with_settlement(ohlcv: pd.DataFrame, settle: pd.Series) -> pd.DataFrame: + """Override Close with exchange settlement (stat_type 3) where present, so the + series is settlement-based like Norgate rather than the ohlcv last-trade close.""" + if ohlcv.empty or settle.empty: + return ohlcv + out = ohlcv.copy() + out["Close"] = settle.reindex(out.index).combine_first(out["Close"]) + return out + + +def _roll_key(df: pd.DataFrame) -> Optional[str]: + """The column that identifies the active contract, for roll detection. Prefer + ``instrument_id`` (changes at each roll); ``symbol`` is only useful when it carries + resolved contracts rather than databento's constant continuous alias.""" + return "instrument_id" if "instrument_id" in df.columns else ( + "symbol" if "symbol" in df.columns else None) + + +def _cumulative_offset(n0: pd.DataFrame, n1: pd.DataFrame): + """Additive back-adjust offset per date: the sum of the roll gap + (n1_close - n0_close measured ON each roll date) over all rolls at or after that + date. A roll date is the last session a front contract is active — its active + contract (instrument_id) differs from the next day's. Returns (offset Series on + n0.index, n_rolls, n_missing).""" + offset = pd.Series(0.0, index=n0.index) + key = _roll_key(n0) + if key is None or n1.empty or "Close" not in n1.columns: + return offset, 0, 0 + sym = n0[key] + is_roll = sym.ne(sym.shift(-1)) & sym.shift(-1).notna() + roll_dates = list(n0.index[is_roll]) + if not roll_dates: + return offset, 0, 0 + n0_close, n1_close = n0["Close"], n1["Close"].reindex(n0.index) + gaps, missing = {}, 0 + for d in roll_dates: + g = n1_close.get(d, float("nan")) - n0_close.get(d, float("nan")) + if pd.isna(g): + missing += 1 + continue + gaps[d] = float(g) + # Walk dates newest->oldest; add each roll's gap as we pass it (inclusive of the + # roll date), so every earlier price carries the cumulative offset. + running, out = 0.0, {} + for d in reversed(list(n0.index)): + if d in gaps: + running += gaps[d] + out[d] = running + return pd.Series(out).reindex(n0.index), len(gaps), missing + + +def build(symbols=None) -> dict: + """Stage 2 (free): read the raw store and write unadj + backadj daily bars to the + cotdata store for every databento-capable symbol. Requires the raw store populated + by ingest(). Returns {kind, ok, symbols, wrote}.""" + targets = [s for s in all_symbols() + if s.databento and s.price_source in (None, "databento") + and (symbols is None or s.internal in symbols)] + wrote, skipped = 0, 0 + for s in targets: + n0 = _read_ohlcv(s.internal, ".n.0") + if n0.empty: + print(f"{s.internal}: no raw .n.0 ohlcv — run ingest first; skipping") + skipped += 1 + continue + n1 = _read_ohlcv(s.internal, ".n.1") + n0 = _with_settlement(n0, _stat_series(s.internal, ".n.0", 3, "ts_ref", "price")) + if not n1.empty: + n1 = _with_settlement(n1, _stat_series(s.internal, ".n.1", 3, "ts_ref", "price")) + oi = _stat_series(s.internal, ".n.0", 9, "ts_event", "quantity") + + unadj = n0.copy() + unadj["Open Interest"] = oi.reindex(unadj.index) if not oi.empty else float("nan") + extra = [] + key = _roll_key(unadj) + if key: + # The active-contract id, as a string, so roll_dates() can detect the change. + # (databento gives no calendar month here, so this is the instrument_id.) + unadj["Delivery Month"] = unadj[key].astype(str) + extra = ["Delivery Month"] + + offset, n_rolls, n_missing = _cumulative_offset(n0, n1) + if n_rolls == 0: + print(f"{s.internal}: no rolls detected (backadj == unadj) — verify the raw ohlcv " + f"carries `instrument_id` (the roll signal; `symbol` is the constant alias)") + if n_missing: + print(f"{s.internal}: {n_missing} roll gap(s) unmeasurable (no .n.1 close) — treated as 0") + + backadj = unadj.copy() + for c in ("Open", "High", "Low", "Close"): + if c in backadj.columns: + backadj[c] = backadj[c] + offset + + for c in _OUT_COLS: + if c not in unadj.columns: + unadj[c] = float("nan") + backadj[c] = float("nan") + cols = _OUT_COLS + extra + store.write_prices(s.internal, "unadj", unadj[cols], source="databento") + store.write_prices(s.internal, "backadj", backadj[cols], source="databento") + wrote += 1 + print(f"{s.internal}: built unadj+backadj ({len(unadj)} bars, {n_rolls} rolls) -> store") + + return {"kind": "build_databento", "ok": skipped == 0, "symbols": len(targets), "wrote": wrote} diff --git a/src/cotdata/providers/yfinance.py b/src/cotdata/providers/yfinance.py index 76b7aff..e31180a 100644 --- a/src/cotdata/providers/yfinance.py +++ b/src/cotdata/providers/yfinance.py @@ -15,7 +15,7 @@ import pandas as pd from .. import store -from ..registry import all_symbols +from ..registry import all_symbols, default_price_source, resolve_source def _fetch(ticker: str) -> pd.DataFrame: @@ -35,10 +35,16 @@ def _fetch(ticker: str) -> pd.DataFrame: def update(symbols=None) -> dict: - """Fetch Yahoo OHLCV for registry symbols with a ``yahoo`` ticker (default: all - that have one; pass ``symbols`` to scope). Returns {kind, ok, wrote}.""" + """Fetch Yahoo OHLCV for registry symbols that RESOLVE to yfinance on this + deployment (see registry.resolve_source): markets the default vendor can't serve + (e.g. MSCI ETF proxies always; ICE softs when the default is databento) plus any + explicit ``price_source: yfinance`` override. Keyed on $COTDATA_PRICE_SOURCE, so + the same softs stay on Norgate locally and fall to Yahoo on a databento server. + Pass ``symbols`` to scope. Returns {kind, ok, wrote}.""" + default = default_price_source() targets = [s for s in all_symbols() - if s.yahoo and (symbols is None or s.internal in symbols)] + if s.yahoo and resolve_source(s, default) == "yfinance" + and (symbols is None or s.internal in symbols)] if not targets: print("yfinance: no registry symbols with a 'yahoo' ticker" + (f" among {symbols}" if symbols else "")) diff --git a/src/cotdata/registry.py b/src/cotdata/registry.py index 3ec4aae..e295149 100644 --- a/src/cotdata/registry.py +++ b/src/cotdata/registry.py @@ -37,6 +37,17 @@ class Symbol: # Optional Yahoo Finance ticker for markets Norgate/databento don't cover (the # yfinance provider prices these; e.g. "EEM"/"EFA" as ETF proxies for MSCI EM/EAFE). yahoo: Optional[str] = None + # Databento GLBX.MDP3 continuous root — the databento producer queries + # ".n.0". Defaults to the internal symbol; None (registry + # `databento: null`) when GLBX carries no series for the market (ICE softs, + # lumber, MSCI intl). None is the capability signal the databento producer + # filters on, and the deployment falls back to yfinance where a yahoo ticker exists. + databento: Optional[str] = None + # Optional per-symbol price-source override (norgate | databento | yfinance). + # Normally unset: the source is the deployment default (COTDATA_PRICE_SOURCE) + # resolved against capability (see resolve_source). Set only to pin one market + # to a specific vendor regardless of the deployment default. + price_source: Optional[str] = None # Predecessor CFTC codes from earlier exchange/contract listings of the SAME # instrument, stitched in chronologically behind cftc_code by get_cot. Each # entry is a bare code string (scale 1.0) or a (code, scale) tuple. Kept as a @@ -58,6 +69,23 @@ def _coerce_hist_codes(raw) -> tuple: return tuple(tuple(h) if isinstance(h, list) else h for h in (raw or [])) +# The price vendors a symbol can be sourced from. 'yfinance' is the universal +# research-grade fallback when the deployment's preferred vendor can't serve a market. +PRICE_SOURCES = ("norgate", "databento", "yfinance") + + +def _validate_source(value, internal) -> Optional[str]: + """Validate an explicit price_source override from the YAML (None if unset).""" + if value is None: + return None + v = str(value).strip().lower() + if v not in PRICE_SOURCES: + raise ValueError( + f"cotdata registry: symbol '{internal}' has price_source={value!r}; " + f"expected one of {PRICE_SOURCES}.") + return v + + def load_registry(yaml_path=None) -> Dict[str, Symbol]: """Build the symbol registry from YAML. @@ -114,6 +142,10 @@ def load_registry(yaml_path=None) -> Dict[str, Symbol]: cftc_code=attrs["cftc_code"], hist_codes=_coerce_hist_codes(attrs.get("hist_codes")), yahoo=attrs.get("yahoo"), + # Defaults to the internal root; explicit `databento: null` marks a + # market GLBX doesn't carry (capability signal for the producer). + databento=attrs.get("databento", internal), + price_source=_validate_source(attrs.get("price_source"), internal), ) return registry @@ -132,3 +164,47 @@ def all_symbols() -> List[Symbol]: def by_asset_class(asset_class: str) -> List[Symbol]: return [s for s in REGISTRY.values() if s.asset_class == asset_class] + + +# ── Price-source selection ─────────────────────────────────────────────────── +# Which vendor prices a symbol is a DEPLOYMENT choice, not a fixed identity fact: +# the same ES is Norgate for local research and databento on the public-dash server. +# So it is resolved at runtime from three inputs, not baked per-symbol in the shared +# registry: (1) a deployment default (COTDATA_PRICE_SOURCE), (2) per-symbol capability +# (which vendors carry a series — the norgate/databento/yahoo mappings), and (3) an +# optional per-symbol override. See ADR-0006. + +def _can_serve(sym: Symbol, source: str) -> bool: + """Whether `source` has a price series for `sym` (its vendor mapping is present).""" + if source == "norgate": + return sym.norgate is not None + if source == "databento": + return sym.databento is not None + if source == "yfinance": + return sym.yahoo is not None + raise ValueError(f"unknown price source {source!r}; expected one of {PRICE_SOURCES}") + + +def resolve_source(sym: Symbol, default: str = "norgate") -> Optional[str]: + """The vendor that prices `sym` on a deployment whose default source is `default`. + + An explicit `sym.price_source` override wins. Otherwise use `default` when that + vendor can serve the symbol, else fall back to yfinance where a yahoo ticker + exists. Returns None when nothing can price it (the caller skips the symbol).""" + if sym.price_source: + return sym.price_source + if _can_serve(sym, default): + return default + if _can_serve(sym, "yfinance"): + return "yfinance" + return None + + +def default_price_source() -> str: + """Deployment-wide default price vendor from $COTDATA_PRICE_SOURCE ('norgate' if + unset). Local research leaves it unset; the databento server sets it to 'databento'.""" + src = os.environ.get("COTDATA_PRICE_SOURCE", "norgate").strip().lower() + if src not in PRICE_SOURCES: + raise ValueError( + f"COTDATA_PRICE_SOURCE={src!r} is invalid; expected one of {PRICE_SOURCES}.") + return src diff --git a/src/cotdata/registry.yaml b/src/cotdata/registry.yaml index b9a1a04..ce86734 100644 --- a/src/cotdata/registry.yaml +++ b/src/cotdata/registry.yaml @@ -21,10 +21,12 @@ Equities: MME: # MSCI Emerging Markets Index (ICE) — priced via EEM cftc_code: "244042" norgate: null # no Norgate continuous series — priced off the ETF proxy + databento: null # ICE product — not on GLBX.MDP3 yahoo: "EEM" MFS: # MSCI EAFE Index (ICE) — priced via EFA cftc_code: "244041" norgate: null # (Norgate doesn't carry MSCI intl indices) — skip in the + databento: null # ICE product — not on GLBX.MDP3 yahoo: "EFA" # Norgate producer; the yfinance provider prices these Metals: @@ -103,19 +105,35 @@ Fixed Income: ZB: cftc_code: "020601" +# Softs trade on ICE, not CME Globex, so databento GLBX.MDP3 carries no series for +# them (see providers/databento.py _DATABENTO_UNSUPPORTED). Norgate covers them for +# local research; a databento deployment falls back to the Yahoo continuous futures +# below (research-grade). LBR's Yahoo symbol is unverified, confirm before relying on it. Softs: SB: cftc_code: "080732" + databento: null + yahoo: "SB=F" CT: cftc_code: "033661" + databento: null + yahoo: "CT=F" CC: cftc_code: "073732" + databento: null + yahoo: "CC=F" KC: cftc_code: "083731" + databento: null + yahoo: "KC=F" OJ: cftc_code: "040701" + databento: null + yahoo: "OJ=F" LBR: cftc_code: "058644" + databento: null + yahoo: "LBS=F" # unverified Yahoo lumber symbol — confirm before relying hist_codes: - ["058643", 4.0] diff --git a/src/cotdata/update.py b/src/cotdata/update.py index 7d26a81..58ab577 100644 --- a/src/cotdata/update.py +++ b/src/cotdata/update.py @@ -15,8 +15,15 @@ def main() -> None: p.add_argument("--prices", action="store_true", help="Update Norgate price bars (Windows).") p.add_argument("--metadata", action="store_true", help="Update Norgate contract metadata (Windows).") p.add_argument("--prices-yahoo", action="store_true", - help="Update prices from Yahoo Finance for registry symbols with a " - "'yahoo' ticker (cross-platform; research-grade).") + help="Update prices from Yahoo Finance for registry symbols that resolve to " + "yfinance on this deployment (cross-platform; research-grade).") + p.add_argument("--ingest-databento", action="store_true", + help="Databento Stage 1 (paid API, cross-platform): fetch raw .n.0/.n.1 " + "ohlcv-1d + statistics into the append-only raw store ($COTDATA_DATABENTO_RAW). " + "Resumable — re-runs only pull new dates. Needs DATABENTO_API_KEY.") + p.add_argument("--build-databento", action="store_true", + help="Databento Stage 2 (free, no API): build back-adjusted prices from the " + "raw store into the cotdata store. Run after --ingest-databento.") p.add_argument("--cot-legacy", action="store_true", help="Update CFTC COT Legacy (cross-platform).") p.add_argument("--cot-disagg", action="store_true", help="Update CFTC COT Disaggregated Futures-Only (cross-platform).") p.add_argument("--cot-tff", action="store_true", help="Update Traders in Financial Futures (TFF) COT (cross-platform).") @@ -64,8 +71,10 @@ def main() -> None: "at": _dt.datetime.utcnow().isoformat(timespec="seconds") + "Z"}) return - if not (args.prices or args.metadata or args.prices_yahoo or args.cot_legacy or args.cot_disagg or args.cot_tff or args.cot_all): - p.error("nothing to do — pass --check, --prices, --prices-yahoo, --metadata, --cot-legacy, --cot-disagg, --cot-tff, or --cot-all") + if not (args.prices or args.metadata or args.prices_yahoo or args.ingest_databento + or args.build_databento or args.cot_legacy or args.cot_disagg or args.cot_tff or args.cot_all): + p.error("nothing to do — pass --check, --prices, --prices-yahoo, --ingest-databento, " + "--build-databento, --metadata, --cot-legacy, --cot-disagg, --cot-tff, or --cot-all") if args.prices or args.metadata: from .providers import norgate @@ -99,6 +108,20 @@ def main() -> None: if not (r or {}).get("ok", True): failed_kinds.append("prices_yahoo") + if args.ingest_databento: + from .providers import databento + r = databento.ingest(symbols=args.symbols) + kinds.append("ingest_databento") + if not (r or {}).get("ok", True): + failed_kinds.append("ingest_databento") + + if args.build_databento: + from .providers import databento + r = databento.build(symbols=args.symbols) + kinds.append("build_databento") + if not (r or {}).get("ok", True): + failed_kinds.append("build_databento") + if args.cot_legacy or args.cot_all: from .providers import cftc r = cftc.update() diff --git a/tests/test_databento_build.py b/tests/test_databento_build.py new file mode 100644 index 0000000..c2475b8 --- /dev/null +++ b/tests/test_databento_build.py @@ -0,0 +1,136 @@ +"""Stage-2 databento build (ADR-0006): additive back-adjustment from the raw store. + +Populates a raw store directly (as ingest would leave it), runs build(), and reads +the result back through the real consumer API (cotdata.get_prices) to verify unadj, +settlement override, Open Interest, and the Norgate-style additive back-adjustment. +""" +from pathlib import Path + +import pandas as pd +import pytest + +import cotdata +from cotdata.providers.databento import build + + +def _write_ohlcv(raw, symbol, feed, dates, close, sym, instrument_id=None): + idx = pd.DatetimeIndex(pd.to_datetime(dates), name="Date") + n = len(close) + data = {"open": close, "high": [c + 0.5 for c in close], "low": [c - 0.5 for c in close], + "close": close, "volume": [1000] * n, + "symbol": sym if isinstance(sym, list) else [sym] * n} + if instrument_id is not None: + data["instrument_id"] = instrument_id + df = pd.DataFrame(data, index=idx) + p = Path(raw) / "ohlcv" / f"{symbol}{feed}.parquet" + p.parent.mkdir(parents=True, exist_ok=True) + df.to_parquet(p) + + +def _write_stats(raw, symbol, feed, dates, settle=None, oi=None): + idx = pd.to_datetime(dates) + frames = [] + if settle is not None: + frames.append(pd.DataFrame( + {"ts_event": idx, "ts_ref": idx, "stat_type": 3, "price": settle, + "quantity": float("nan")})) + if oi is not None: + frames.append(pd.DataFrame( + {"ts_event": idx, "ts_ref": pd.NaT, "stat_type": 9, "price": float("nan"), + "quantity": oi})) + df = pd.concat(frames, ignore_index=True) + p = Path(raw) / "statistics" / f"{symbol}{feed}.parquet" + p.parent.mkdir(parents=True, exist_ok=True) + df.to_parquet(p) + + +@pytest.fixture +def stores(tmp_path, monkeypatch): + raw, store_dir = tmp_path / "raw", tmp_path / "store" + monkeypatch.setenv("COTDATA_DATABENTO_RAW", str(raw)) + monkeypatch.setenv("COTDATA_STORE", str(store_dir)) + return raw, store_dir + + +def test_build_back_adjusts_the_roll_gap(stores): + raw, _ = stores + dates = pd.date_range("2020-01-01", periods=6, freq="D") + # Front contract A on d1-3, rolls to B at d3's close; B on d4-6. + _write_ohlcv(raw, "ES", ".n.0", dates, [100, 101, 102, 110, 111, 112], ["A", "A", "A", "B", "B", "B"]) + # Second contract carries B on d1-3 (so n1[d3] is B's price the day of the roll). + _write_ohlcv(raw, "ES", ".n.1", dates, [103, 104, 105, 113, 114, 115], ["B", "B", "B", "C", "C", "C"]) + + res = build(["ES"]) + assert res["ok"] and res["wrote"] == 1 + + unadj = cotdata.get_prices("ES", adjustment="unadj") + backadj = cotdata.get_prices("ES", adjustment="backadj") + + # unadj keeps the raw front prices (roll gap intact: 102 -> 110). + assert list(unadj["Close"]) == [100, 101, 102, 110, 111, 112] + # Roll gap on d3 = n1 - n0 = 105 - 102 = +3, so every price up to & incl. d3 shifts +3. + assert list(backadj["Close"]) == [103, 104, 105, 110, 111, 112] + # The whole bar shifts by the same offset; the newest segment is untouched. + assert backadj["High"].iloc[0] == unadj["High"].iloc[0] + 3 + assert backadj["Low"].iloc[2] == unadj["Low"].iloc[2] + 3 + assert backadj["Close"].iloc[-1] == unadj["Close"].iloc[-1] + + +def test_build_detects_rolls_from_instrument_id_not_symbol(stores): + # The real databento shape: for a continuous series the `symbol` column is a + # CONSTANT alias ("ES.n.0"), and instrument_id is the resolved contract that + # changes at the roll. Roll detection must key on instrument_id, not symbol. + raw, _ = stores + dates = pd.date_range("2020-01-01", periods=6, freq="D") + _write_ohlcv(raw, "ES", ".n.0", dates, [100, 101, 102, 110, 111, 112], + sym="ES.n.0", instrument_id=[10, 10, 10, 20, 20, 20]) + _write_ohlcv(raw, "ES", ".n.1", dates, [103, 104, 105, 113, 114, 115], + sym="ES.n.1", instrument_id=[20, 20, 20, 30, 30, 30]) + + build(["ES"]) + unadj = cotdata.get_prices("ES", adjustment="unadj") + backadj = cotdata.get_prices("ES", adjustment="backadj") + # The constant `symbol` alias would find no rolls; instrument_id finds the d3 roll, + # gap = 105 - 102 = +3, applied to the pre-roll segment. + assert list(unadj["Close"]) == [100, 101, 102, 110, 111, 112] + assert list(backadj["Close"]) == [103, 104, 105, 110, 111, 112] + + +def test_build_uses_settlement_and_open_interest(stores): + raw, _ = stores + dates = pd.date_range("2020-01-01", periods=6, freq="D") + _write_ohlcv(raw, "ES", ".n.0", dates, [100, 101, 102, 110, 111, 112], ["A", "A", "A", "B", "B", "B"]) + _write_ohlcv(raw, "ES", ".n.1", dates, [103, 104, 105, 113, 114, 115], ["B", "B", "B", "C", "C", "C"]) + # Settlement (stat_type 3) sits 0.5 above the last-trade close; OI (stat_type 9) = 5000. + _write_stats(raw, "ES", ".n.0", dates, settle=[100.5, 101.5, 102.5, 110.5, 111.5, 112.5], oi=[5000] * 6) + _write_stats(raw, "ES", ".n.1", dates, settle=[103.5, 104.5, 105.5, 113.5, 114.5, 115.5]) + + build(["ES"]) + unadj = cotdata.get_prices("ES", adjustment="unadj") + backadj = cotdata.get_prices("ES", adjustment="backadj") + + # Close is the settlement, not the ohlcv last trade; OI comes from stat_type 9. + assert list(unadj["Close"]) == [100.5, 101.5, 102.5, 110.5, 111.5, 112.5] + assert list(unadj["Open Interest"]) == [5000] * 6 + # Gap measured on settlements: 105.5 - 102.5 = +3.0. + assert list(backadj["Close"]) == [103.5, 104.5, 105.5, 110.5, 111.5, 112.5] + + +def test_build_no_rolls_leaves_series_unadjusted(stores, capsys): + raw, _ = stores + dates = pd.date_range("2020-01-01", periods=5, freq="D") + _write_ohlcv(raw, "ES", ".n.0", dates, [100, 101, 102, 103, 104], ["A"] * 5) + _write_ohlcv(raw, "ES", ".n.1", dates, [110, 111, 112, 113, 114], ["B"] * 5) + + build(["ES"]) + unadj = cotdata.get_prices("ES", adjustment="unadj") + backadj = cotdata.get_prices("ES", adjustment="backadj") + + assert list(backadj["Close"]) == list(unadj["Close"]) # no roll → no adjustment + assert "no rolls detected" in capsys.readouterr().out + + +def test_build_skips_symbol_missing_from_raw_store(stores): + # Nothing written for CL → build reports it skipped (and it's not 'ok'). + res = build(["CL"]) + assert res["wrote"] == 0 and res["ok"] is False diff --git a/tests/test_databento_ingest.py b/tests/test_databento_ingest.py new file mode 100644 index 0000000..94d55e8 --- /dev/null +++ b/tests/test_databento_ingest.py @@ -0,0 +1,138 @@ +"""Stage-1 databento ingest (ADR-0006): raw landing store + resume manifest. + +Exercised with an injected fake client shaped like ``databento.Historical`` — no +API key, no network. Verifies the raw files, the fetched-range manifest, the +resume-from-last_date behaviour, and that databento-null symbols are skipped. +""" +import json + +import pandas as pd + +from cotdata.providers.databento import ingest + + +# ── a databento.Historical-shaped fake ─────────────────────────────────────── +class _FakeResp: + def __init__(self, df): + self._df = df + + def to_df(self): + return self._df + + +class _FakeTS: + def __init__(self, owner): + self.owner = owner + + def get_range(self, *, dataset, symbols, stype_in, schema, start, end): + self.owner.calls.append((symbols[0], schema, start, end)) + frame = self.owner.frames.get((symbols[0], schema)) + if frame is None or frame.empty: + return _FakeResp(pd.DataFrame()) + s, e = pd.Timestamp(start), pd.Timestamp(end) + # databento returns ts_event as the index for both schemas. + naive = frame.index.tz_convert(None) + mask = (naive >= s) & (naive <= e) + return _FakeResp(frame[mask]) + + +class _FakeClient: + def __init__(self, frames): + self.frames = frames + self.calls = [] + + @property + def timeseries(self): + return _FakeTS(self) + + +def _ohlcv(dates, base): + idx = pd.to_datetime(dates).tz_localize("UTC") + idx.name = "ts_event" + n = len(idx) + return pd.DataFrame( + {"open": [base] * n, "high": [base + 1] * n, "low": [base - 1] * n, + "close": [base + 0.5] * n, "volume": [1000] * n, "symbol": ["ES.FUT"] * n}, + index=idx) + + +def _stats(dates, price): + idx = pd.to_datetime(dates).tz_localize("UTC") + idx.name = "ts_event" + n = len(idx) + return pd.DataFrame( + {"ts_ref": idx, "stat_type": [3] * n, "price": [price] * n, "quantity": [0] * n}, + index=idx) + + +def _frames(dates): + return { + ("ES.n.0", "ohlcv-1d"): _ohlcv(dates, 100), + ("ES.n.1", "ohlcv-1d"): _ohlcv(dates, 101), + ("ES.n.0", "statistics"): _stats(dates, 100.5), + ("ES.n.1", "statistics"): _stats(dates, 101.5), + } + + +# ── tests ───────────────────────────────────────────────────────────────────── +def test_ingest_writes_raw_files_and_manifest(tmp_path, monkeypatch): + monkeypatch.setenv("COTDATA_DATABENTO_RAW", str(tmp_path)) + dates = pd.date_range("2020-01-01", periods=5, freq="D") + client = _FakeClient(_frames(dates)) + + res = ingest(symbols=["ES"], client=client, end="2020-01-05", cold_start="2020-01-01") + + assert res["ok"] and res["symbols"] == 1 + for feed in (".n.0", ".n.1"): + assert (tmp_path / "ohlcv" / f"ES{feed}.parquet").exists() + assert (tmp_path / "statistics" / f"ES{feed}.parquet").exists() + + ohlcv = pd.read_parquet(tmp_path / "ohlcv" / "ES.n.0.parquet") + assert len(ohlcv) == 5 + assert ohlcv.index.tz is None # bronze is tz-naive + assert ohlcv.index.is_monotonic_increasing + + man = json.loads((tmp_path / "ingest_manifest.json").read_text()) + assert man["ES.n.0:ohlcv-1d"]["last_date"] == "2020-01-05" + assert man["ES.n.0:ohlcv-1d"]["first_date"] == "2020-01-01" + assert man["ES.n.0:ohlcv-1d"]["n_rows"] == 5 + + +def test_ingest_resumes_from_last_date(tmp_path, monkeypatch): + monkeypatch.setenv("COTDATA_DATABENTO_RAW", str(tmp_path)) + + # First pull: 5 days. + ingest(symbols=["ES"], client=_FakeClient(_frames(pd.date_range("2020-01-01", periods=5))), + end="2020-01-05", cold_start="2020-01-01") + + # Second pull: source now has 8 days; a fresh client so we can inspect its calls. + client2 = _FakeClient(_frames(pd.date_range("2020-01-01", periods=8))) + ingest(symbols=["ES"], client=client2, end="2020-01-08", cold_start="2020-01-01") + + # Resume: the ohlcv .n.0 call must start the day AFTER the stored last_date. + ohlcv_calls = [c for c in client2.calls if c[0] == "ES.n.0" and c[1] == "ohlcv-1d"] + assert ohlcv_calls and ohlcv_calls[0][2] == "2020-01-06" + + combined = pd.read_parquet(tmp_path / "ohlcv" / "ES.n.0.parquet") + assert len(combined) == 8 # 5 + 3 appended, no dupes + man = json.loads((tmp_path / "ingest_manifest.json").read_text()) + assert man["ES.n.0:ohlcv-1d"]["last_date"] == "2020-01-08" + assert man["ES.n.0:ohlcv-1d"]["n_rows"] == 8 + + +def test_ingest_noop_when_already_current(tmp_path, monkeypatch): + monkeypatch.setenv("COTDATA_DATABENTO_RAW", str(tmp_path)) + ingest(symbols=["ES"], client=_FakeClient(_frames(pd.date_range("2020-01-01", periods=5))), + end="2020-01-05", cold_start="2020-01-01") + + client2 = _FakeClient(_frames(pd.date_range("2020-01-01", periods=5))) + ingest(symbols=["ES"], client=client2, end="2020-01-05", cold_start="2020-01-01") + assert client2.calls == [] # start would be > end → nothing fetched + + +def test_ingest_skips_databento_null_symbol(tmp_path, monkeypatch): + monkeypatch.setenv("COTDATA_DATABENTO_RAW", str(tmp_path)) + client = _FakeClient({}) + res = ingest(symbols=["CC"], client=client, end="2020-01-05") # CC is databento: null + assert res["symbols"] == 0 + assert client.calls == [] diff --git a/tests/test_registry.py b/tests/test_registry.py index 99e5eeb..072352d 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -6,8 +6,10 @@ from cotdata.registry import ( all_symbols, by_asset_class, + default_price_source, hist_code_scales, load_registry, + resolve_source, symbol, ) @@ -91,6 +93,84 @@ def test_is_equity_derived_from_asset_class(): assert s.is_equity == (s.asset_class == "Equities"), s.internal +# ── price-source selection: capability + deployment default + override ──────── +def test_databento_mapping_defaults_to_internal_root(): + # Norgate-covered CME/CBOT/NYMEX/COMEX markets default databento to the root. + assert symbol("ES").databento == "ES" + assert symbol("CL").databento == "CL" + assert symbol("GC").databento == "GC" + + +def test_databento_null_for_non_glbx_markets(): + # ICE softs + lumber (not on GLBX Globex) and MSCI intl are databento: null. + for s in ("SB", "CT", "CC", "KC", "OJ", "LBR", "MME", "MFS"): + assert symbol(s).databento is None, s + + +def test_resolve_source_uses_deployment_default_when_capable(): + assert resolve_source(symbol("ES"), "norgate") == "norgate" + assert resolve_source(symbol("ES"), "databento") == "databento" + assert resolve_source(symbol("CL"), "databento") == "databento" + + +def test_resolve_source_falls_back_to_yfinance_when_default_cannot_serve(): + # MME/MFS have neither a norgate nor a databento series, but do have a yahoo + # ETF proxy — so either deployment default resolves to yfinance. + assert resolve_source(symbol("MME"), "norgate") == "yfinance" + assert resolve_source(symbol("MME"), "databento") == "yfinance" + + +def test_softs_have_yahoo_fallback_and_resolve_to_yfinance_on_databento(): + # ICE softs aren't on GLBX but carry a Yahoo continuous fallback, so a databento + # deployment resolves them to yfinance; Norgate still covers them locally. + for s in ("SB", "CT", "CC", "KC", "OJ", "LBR"): + assert symbol(s).yahoo, s + assert resolve_source(symbol(s), "databento") == "yfinance", s + assert resolve_source(symbol(s), "norgate") == "norgate", s + + +def test_resolve_source_none_when_no_vendor_can_serve(tmp_path): + # A synthetic market with no vendor mapping at all → nothing can price it. + reg = load_registry(_write(tmp_path, textwrap.dedent(""" + Metals: + XX: + cftc_code: "000000" + norgate: null + databento: null + """))) + assert resolve_source(reg["XX"], "databento") is None + assert resolve_source(reg["XX"], "norgate") is None + + +def test_price_source_override_wins(tmp_path): + reg = load_registry(_write(tmp_path, textwrap.dedent(""" + Metals: + GC: + cftc_code: "088691" + price_source: yfinance + yahoo: "GLD" + """))) + assert reg["GC"].price_source == "yfinance" + # Override beats the deployment default even though Norgate could serve GC. + assert resolve_source(reg["GC"], "norgate") == "yfinance" + + +def test_invalid_price_source_override_raises(tmp_path): + with pytest.raises(ValueError, match="price_source"): + load_registry(_write( + tmp_path, 'Metals:\n GC:\n cftc_code: "1"\n price_source: bloomberg\n')) + + +def test_default_price_source_env(monkeypatch): + monkeypatch.delenv("COTDATA_PRICE_SOURCE", raising=False) + assert default_price_source() == "norgate" + monkeypatch.setenv("COTDATA_PRICE_SOURCE", "databento") + assert default_price_source() == "databento" + monkeypatch.setenv("COTDATA_PRICE_SOURCE", "nope") + with pytest.raises(ValueError, match="COTDATA_PRICE_SOURCE"): + default_price_source() + + # ── $COTDATA_REGISTRY / explicit-path override (the point of the refactor) ──── _MINI_YAML = textwrap.dedent(""" Metals: diff --git a/tests/test_validate_databento.py b/tests/test_validate_databento.py new file mode 100644 index 0000000..5a88778 --- /dev/null +++ b/tests/test_validate_databento.py @@ -0,0 +1,107 @@ +"""Unit tests for the ADR-0006 item-6 validation harness (scripts/validate_databento_vs_norgate.py). + +The harness itself needs real Norgate + databento stores to run; here we verify its +comparison logic against synthetic backadj frames: that an anchor-only difference +passes, a scale (unit) mismatch fails, per-day outliers are counted, roll dates are +read from Delivery Month, and read_backadj round-trips a store parquet. +""" +import importlib.util +from pathlib import Path + +import numpy as np +import pandas as pd +import pytest + +_SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "validate_databento_vs_norgate.py" +_spec = importlib.util.spec_from_file_location("validate_databento_vs_norgate", _SCRIPT) +val = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(val) + + +_DATES = pd.date_range("2020-01-01", periods=250, freq="B") +_RNG = np.random.default_rng(0) +_CHANGES = _RNG.normal(0, 1.0, len(_DATES)) +_NG_CLOSE = 100 + np.cumsum(_CHANGES) + + +def _frame(close, dm=None): + df = pd.DataFrame({"Open": close, "High": close, "Low": close, "Close": close}, + index=pd.DatetimeIndex(_DATES, name="Date")) + if dm is not None: + df["Delivery Month"] = dm + return df + + +def _pass_thresholds(m): + return val.evaluate(m, corr_min=0.999, scale_band=0.02, rel_tol=0.10, max_outliers_per_yr=8) + + +def test_anchor_only_difference_passes(): + # databento = norgate + a constant (a floating back-adjust anchor) → identical + # daily changes, different level. This is the expected clean case. + ng = _frame(_NG_CLOSE) + db = _frame(_NG_CLOSE + 7.5) + m = val.compare(ng, db, "ES") + + assert m["change_corr"] > 0.99999 + assert m["scale_ratio"] == pytest.approx(1.0, abs=1e-9) + assert m["change_max_rel_diff"] < 1e-9 + assert m["level_mean_abs"] == pytest.approx(7.5) + assert _pass_thresholds(m) == [] + + +def test_scale_mismatch_fails(): + # databento daily changes are 1.5x norgate's (e.g. a settlement/unit scale bug): + # perfectly correlated but the wrong size → caught by scale_ratio, not corr. + ng = _frame(_NG_CLOSE) + db = _frame(100 + np.cumsum(_CHANGES * 1.5)) + m = val.compare(ng, db, "ES") + + assert m["change_corr"] > 0.999 # still correlated + assert m["scale_ratio"] == pytest.approx(1.5, abs=0.05) + fails = _pass_thresholds(m) + assert any("scale_ratio" in f for f in fails) + + +def test_per_day_outliers_are_counted_and_gate_fails(): + # Inject a 4-point discrepancy every 12th day (~20/yr, each ~4x a typical daily + # move) — the kind of thing a roll-date mismatch or a bad settlement would cause. + ng = _frame(_NG_CLOSE) + db_changes = _CHANGES.copy() + db_changes[::12] += 4.0 + db = _frame(100 + np.cumsum(db_changes)) + m = val.compare(ng, db, "ES") + + assert m["outlier_days"] >= 15 + assert m["change_max_rel_diff"] > 1.0 # worst day dwarfs a typical move + assert _pass_thresholds(m) != [] # too many outlier days → fail + # Loosening the outlier budget past what occurred clears that specific gate. + loose = val.evaluate(m, corr_min=0.0, scale_band=1.0, rel_tol=0.10, max_outliers_per_yr=1000) + assert loose == [] + + +def test_insufficient_overlap_reports_error(): + ng = _frame(_NG_CLOSE).iloc[:2] + db = _frame(_NG_CLOSE).iloc[100:102] # no common dates + m = val.compare(ng, db, "ES") + assert m.get("error") == "insufficient overlap" + assert _pass_thresholds(m) == ["insufficient overlap"] + + +def test_roll_dates_from_delivery_month(): + dm = ["A"] * 100 + ["B"] * 150 + f = _frame(_NG_CLOSE, dm=dm) + rolls = val._roll_dates(f) + assert len(rolls) == 1 and rolls[0] == _DATES[100] + # roll counts surface in the comparison metrics + m = val.compare(f, _frame(_NG_CLOSE + 1, dm=dm), "ES") + assert m["rolls_norgate"] == 1 and m["rolls_common"] == 1 + + +def test_read_backadj_roundtrip(tmp_path): + prices = tmp_path / "prices" + prices.mkdir() + _frame(_NG_CLOSE).to_parquet(prices / "ES_backadj.parquet") + got = val.read_backadj(str(tmp_path), "ES") + assert not got.empty and got.index.name == "Date" and got.index.tz is None + assert val.read_backadj(str(tmp_path), "NOPE").empty