From 1b750a59f13f3cd955493e0f683f3b95ad036814 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 14 Jun 2026 16:17:04 -0400 Subject: [PATCH 1/6] Add daily_contracts table and dated-fetch config Store the dated contract strip per commodity (daily settlements) in a new daily_contracts table keyed by (Commodity, ContractMonth, Datetime), with an idempotent ON CONFLICT upsert mirroring intraday_prices. Add DataFetchConfig knobs for the dated strip (enabled, lookback, contract count). --- src/energex/config.py | 9 +++++ src/energex/database.py | 74 +++++++++++++++++++++++++++++++++++++++-- 2 files changed, 81 insertions(+), 2 deletions(-) diff --git a/src/energex/config.py b/src/energex/config.py index 7c8ef99..242b78a 100644 --- a/src/energex/config.py +++ b/src/energex/config.py @@ -85,6 +85,15 @@ class DataFetchConfig(BaseSettings): yfinance_timeout: int = Field(default=30, description="Yahoo Finance timeout in seconds") data_fetch_retries: int = Field(default=3, description="Number of download attempts", ge=0) + dated_enabled: bool = Field( + default=True, description="Enable ingestion of the dated futures contract strip" + ) + dated_lookback_days: int = Field( + default=45, description="Daily-bar lookback window per dated contract", ge=1 + ) + dated_contract_count: int = Field( + default=12, description="Number of forward monthly contracts to fetch per commodity", ge=1 + ) model_config = SettingsConfigDict(env_prefix="DATAFETCH_", case_sensitive=False) diff --git a/src/energex/database.py b/src/energex/database.py index e30c75f..d4b8fd8 100644 --- a/src/energex/database.py +++ b/src/energex/database.py @@ -24,6 +24,20 @@ class EnergyDatabase: #: Canonical column order for the ``intraday_prices`` table. COLUMNS = ("Datetime", "Symbol", "Open", "High", "Low", "Close", "Volume") + #: Canonical column order for the ``daily_contracts`` table (dated contract strip). + DAILY_CONTRACTS_COLUMNS = ( + "Datetime", + "Commodity", + "ContractMonth", + "Symbol", + "Open", + "High", + "Low", + "Close", + "Volume", + "OpenInterest", + ) + def __init__(self, db_path: str | None = None, read_only: bool = False): # Resolve the path from configuration (ENERGEX_DB_PATH) when not given. if db_path is None: @@ -45,7 +59,7 @@ def __init__(self, db_path: str | None = None, read_only: bool = False): self._migrate_to_timestamptz() def _init_tables(self) -> None: - """Create the intraday_prices table if it does not already exist (idempotent).""" + """Create the storage tables if they do not already exist (idempotent).""" self.conn.execute(""" CREATE TABLE IF NOT EXISTS intraday_prices ( Datetime TIMESTAMPTZ, @@ -58,6 +72,21 @@ def _init_tables(self) -> None: CONSTRAINT pk_intraday PRIMARY KEY (Symbol, Datetime) ) """) + self.conn.execute(""" + CREATE TABLE IF NOT EXISTS daily_contracts ( + Datetime TIMESTAMPTZ, + Commodity VARCHAR, + ContractMonth DATE, + Symbol VARCHAR, + Open DOUBLE, + High DOUBLE, + Low DOUBLE, + Close DOUBLE, + Volume BIGINT, + OpenInterest BIGINT, + CONSTRAINT pk_daily_contracts PRIMARY KEY (Commodity, ContractMonth, Datetime) + ) + """) def _migrate_to_timestamptz(self) -> None: """Migrate a legacy naive-TIMESTAMP Datetime column to TIMESTAMPTZ (UTC). @@ -99,10 +128,11 @@ def _migrate_to_timestamptz(self) -> None: raise def reset(self) -> None: - """Drop and recreate the table. Destructive — for explicit bootstrap only.""" + """Drop and recreate the tables. Destructive — for explicit bootstrap only.""" if self.read_only: raise DatabaseError("Cannot reset a read-only database connection") self.conn.execute("DROP TABLE IF EXISTS intraday_prices") + self.conn.execute("DROP TABLE IF EXISTS daily_contracts") self._init_tables() def insert_intraday_data(self, df: pl.DataFrame) -> None: @@ -141,6 +171,46 @@ def insert_intraday_data(self, df: pl.DataFrame) -> None: logger.error("Error inserting data: %s", e) raise + def insert_daily_contracts(self, df: pl.DataFrame) -> None: + """Idempotently upsert dated-contract daily bars keyed by + (Commodity, ContractMonth, Datetime).""" + if df.is_empty(): + logger.info("No daily contracts to insert") + return + + missing = [c for c in self.DAILY_CONTRACTS_COLUMNS if c not in df.columns] + if missing: + raise DatabaseError( + f"DataFrame is missing required columns {missing}; got {df.columns}" + ) + + # Select named columns in canonical order so the INSERT never binds positionally. + df = df.select(self.DAILY_CONTRACTS_COLUMNS) + + try: + self.conn.execute("BEGIN TRANSACTION") + self.conn.execute(""" + INSERT INTO daily_contracts + (Datetime, Commodity, ContractMonth, Symbol, + Open, High, Low, Close, Volume, OpenInterest) + SELECT Datetime, Commodity, ContractMonth, Symbol, + Open, High, Low, Close, Volume, OpenInterest FROM df + ON CONFLICT (Commodity, ContractMonth, Datetime) DO UPDATE SET + Symbol = excluded.Symbol, + Open = excluded.Open, + High = excluded.High, + Low = excluded.Low, + Close = excluded.Close, + Volume = excluded.Volume, + OpenInterest = excluded.OpenInterest + """) + self.conn.execute("COMMIT") + logger.info("Upserted %d daily contract rows", len(df)) + except Exception as e: + self.conn.execute("ROLLBACK") + logger.error("Error inserting daily contracts: %s", e) + raise + def query(self, sql: str, params: list[Any] | None = None) -> pl.DataFrame: """Run a read query on a separate cursor and return a Polars DataFrame.""" result = pl.from_arrow(self.conn.cursor().execute(sql, params or []).arrow()) From c4d055edef0b01cce48aa70dad8319f6ed357821 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 14 Jun 2026 16:17:10 -0400 Subject: [PATCH 2/6] Fetch dated futures contract strip via yfinance source Enumerate forward monthly NYMEX contracts (CL/BZ/NG roots, month codes F-Z) and fetch each contract's recent daily bars, tagging Commodity and ContractMonth. A single empty/failed contract is logged and skipped. Expose a pure _dated_tickers helper for unit testing and a fetch_dated method on the DataSource adapter. --- src/energex/data_fetcher.py | 106 ++++++++++++++++++++++++- src/energex/sources/base.py | 4 + src/energex/sources/stubs.py | 3 + src/energex/sources/yfinance_source.py | 3 + 4 files changed, 115 insertions(+), 1 deletion(-) diff --git a/src/energex/data_fetcher.py b/src/energex/data_fetcher.py index 43a4e08..af0a9ad 100644 --- a/src/energex/data_fetcher.py +++ b/src/energex/data_fetcher.py @@ -1,7 +1,7 @@ # src/energex/data_fetcher.py import logging import time -from datetime import datetime, timedelta +from datetime import date, datetime, timedelta from typing import Any import polars as pl @@ -9,10 +9,34 @@ import yfinance as yf from energex.config import get_settings +from energex.database import EnergyDatabase from energex.exceptions import DataFetchError logger = logging.getLogger(__name__) +#: CME/NYMEX delivery-month codes, January..December. +MONTH_CODES = "FGHJKMNQUVXZ" + + +def _dated_tickers(root: str, start: date, count: int) -> list[tuple[str, date]]: + """Enumerate the next ``count`` monthly contracts from ``start``'s month. + + Returns ``(ticker, contract_month)`` pairs where ``ticker`` is the NYMEX symbol + (e.g. ``CLZ26.NYM``) and ``contract_month`` is the first day of the delivery month. + Pure and network-free for unit testing. + """ + out: list[tuple[str, date]] = [] + year, month = start.year, start.month + for _ in range(count): + contract_month = date(year, month, 1) + ticker = f"{root}{MONTH_CODES[month - 1]}{year % 100:02d}.NYM" + out.append((ticker, contract_month)) + month += 1 + if month > 12: + month = 1 + year += 1 + return out + def normalize_datetime_to_utc(df: pl.DataFrame) -> pl.DataFrame: """Convert a tz-aware Datetime column to a UTC instant. @@ -42,6 +66,8 @@ def __init__(self) -> None: self.retries = max(1, settings.data_fetch.data_fetch_retries) self.end_time = datetime.now(pytz.UTC) self.start_time = self.end_time - timedelta(days=1) + self.dated_lookback_days = settings.data_fetch.dated_lookback_days + self.dated_contract_count = settings.data_fetch.dated_contract_count def _download_with_retry(self, ticker: str) -> Any: """Download with a timeout and exponential backoff; raise on real failure.""" @@ -66,6 +92,84 @@ def _download_with_retry(self, ticker: str) -> Any: f"Failed to download {ticker} after {self.retries} attempts" ) from last_exc + def _download_daily_with_retry(self, ticker: str, period_days: int) -> Any: + """Download daily bars for a dated contract with timeout and backoff. + + Mirrors ``_download_with_retry`` but uses a daily interval over a recent window. + """ + period = f"{period_days}d" + last_exc: Exception | None = None + for attempt in range(1, self.retries + 1): + try: + return yf.download( + ticker, + period=period, + interval="1d", + timeout=self.timeout, + ) + except Exception as e: + last_exc = e + logger.warning( + "Download attempt %d/%d for %s failed: %s", attempt, self.retries, ticker, e + ) + if attempt < self.retries: + time.sleep(min(2 ** (attempt - 1), 10)) + raise DataFetchError( + f"Failed to download {ticker} after {self.retries} attempts" + ) from last_exc + + def fetch_dated_contracts(self, commodity: str) -> pl.DataFrame: + """Fetch the dated forward contract strip for one commodity. + + Enumerates the next ``dated_contract_count`` monthly NYMEX contracts from the + commodity root (CL/BZ/NG) and fetches each contract's recent daily bars. A + single contract returning empty (transient yfinance miss) is logged and skipped; + it never raises. Returns a frame with ``DAILY_CONTRACTS_COLUMNS``. + """ + root = self.ENERGY_SYMBOLS[commodity]["ticker"].removesuffix("=F") + today = datetime.now(pytz.UTC).date() + dfs: list[pl.DataFrame] = [] + for ticker, contract_month in _dated_tickers(root, today, self.dated_contract_count): + try: + data = self._download_daily_with_retry(ticker, self.dated_lookback_days) + except DataFetchError as e: + logger.warning("Skipping contract %s: %s", ticker, e) + continue + if data.empty: + logger.info("No data for contract %s (%s)", ticker, commodity) + continue + + pdf = data.reset_index() + pdf.columns = [col[0] if isinstance(col, tuple) else col for col in pdf.columns] + df = pl.from_pandas(pdf).rename({"Date": "Datetime"}) + if "OpenInterest" not in df.columns: + df = df.with_columns(pl.lit(None, dtype=pl.Int64).alias("OpenInterest")) + df = df.with_columns( + pl.lit(commodity).alias("Commodity"), + pl.lit(contract_month).alias("ContractMonth"), + pl.lit(ticker).alias("Symbol"), + pl.col("OpenInterest").cast(pl.Int64), + ) + df = normalize_datetime_to_utc(df) + df = df.select(EnergyDatabase.DAILY_CONTRACTS_COLUMNS) + dfs.append(df) + + if not dfs: + logger.warning("No dated contracts fetched for %s", commodity) + return pl.DataFrame() + return pl.concat(dfs) + + def fetch_all_dated(self) -> pl.DataFrame: + """Fetch and concatenate the dated contract strip for all commodities.""" + dfs = [ + df + for commodity in self.ENERGY_SYMBOLS + if not (df := self.fetch_dated_contracts(commodity)).is_empty() + ] + if not dfs: + return pl.DataFrame() + return pl.concat(dfs) + def get_commodity_data(self, commodity: str) -> pl.DataFrame: """Fetch intraday data for one commodity key (crude, brent, gas). diff --git a/src/energex/sources/base.py b/src/energex/sources/base.py index ca1cb7c..97138dd 100644 --- a/src/energex/sources/base.py +++ b/src/energex/sources/base.py @@ -25,3 +25,7 @@ def fetch(self, commodity: str) -> pl.DataFrame: @abstractmethod def fetch_all(self) -> pl.DataFrame: """Fetch and combine all supported commodities.""" + + @abstractmethod + def fetch_dated(self) -> pl.DataFrame: + """Fetch the dated forward contract strip for all supported commodities.""" diff --git a/src/energex/sources/stubs.py b/src/energex/sources/stubs.py index 30ee264..1116cdd 100644 --- a/src/energex/sources/stubs.py +++ b/src/energex/sources/stubs.py @@ -19,6 +19,9 @@ def fetch(self, commodity: str) -> pl.DataFrame: def fetch_all(self) -> pl.DataFrame: raise NotImplementedError(self.setup) + def fetch_dated(self) -> pl.DataFrame: + raise NotImplementedError(self.setup) + class DatabentoDataSource(_NotConfiguredSource): """Primary intraday source for CME CL/NG (per-contract symbology + expiry + OI).""" diff --git a/src/energex/sources/yfinance_source.py b/src/energex/sources/yfinance_source.py index 7d0a956..bbb4329 100644 --- a/src/energex/sources/yfinance_source.py +++ b/src/energex/sources/yfinance_source.py @@ -23,3 +23,6 @@ def fetch(self, commodity: str) -> pl.DataFrame: def fetch_all(self) -> pl.DataFrame: return self._fetcher.fetch_all_commodities() + + def fetch_dated(self) -> pl.DataFrame: + return self._fetcher.fetch_all_dated() From b7a7b29c9790ac7e67f0a655c9afa2e37e363d9c Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 14 Jun 2026 16:17:16 -0400 Subject: [PATCH 3/6] Add DatedFuturesAnalyzer term-structure analytics Real forward-curve analytics over dated contracts: curve_as_of, annualized term_structure_slope, per-pair roll_yield, and curve shape classification. Point the gated continuous-proxy FuturesAnalyzer methods at this new capability. --- src/energex/analysis/__init__.py | 2 + src/energex/analysis/dated_futures.py | 103 ++++++++++++++++++++++++++ src/energex/analysis/futures.py | 10 ++- 3 files changed, 111 insertions(+), 4 deletions(-) create mode 100644 src/energex/analysis/dated_futures.py diff --git a/src/energex/analysis/__init__.py b/src/energex/analysis/__init__.py index 6b041ae..d71438d 100644 --- a/src/energex/analysis/__init__.py +++ b/src/energex/analysis/__init__.py @@ -1,5 +1,6 @@ """Analysis modules for energy derivatives trading.""" +from energex.analysis.dated_futures import DatedFuturesAnalyzer from energex.analysis.futures import FuturesAnalyzer from energex.analysis.quality import DataQualityChecker from energex.analysis.volatility import VolatilityAnalyzer @@ -32,6 +33,7 @@ def check_sentiment_available() -> bool: "DataQualityChecker", "VolatilityAnalyzer", "FuturesAnalyzer", + "DatedFuturesAnalyzer", "MarketSentimentAnalyzer", "check_sentiment_available", ] diff --git a/src/energex/analysis/dated_futures.py b/src/energex/analysis/dated_futures.py new file mode 100644 index 0000000..9ab7037 --- /dev/null +++ b/src/energex/analysis/dated_futures.py @@ -0,0 +1,103 @@ +# src/energex/analysis/dated_futures.py +"""Term-structure analytics over the dated futures contract strip. + +Unlike :class:`energex.analysis.futures.FuturesAnalyzer` (which only has continuous +front-month proxies), this operates on real dated contracts (``daily_contracts`` rows) +so the forward curve, slope, roll yield, and curve shape are genuine cross-sectional +quantities at a single point in time (ASSESSMENT R8). +""" + +from datetime import date + +import polars as pl + + +class DatedFuturesAnalyzer: + """Forward-curve / term-structure analytics over dated contracts. + + Args: + df: Polars frame of ``daily_contracts`` rows (columns include Commodity, + ContractMonth, Datetime, Close). + """ + + def __init__(self, df: pl.DataFrame): + self.df = df + + def curve_as_of(self, commodity: str, asof: date) -> pl.DataFrame: + """The dated forward curve for ``commodity`` observed on ``asof``. + + One row per ContractMonth (sorted ascending) with the settlement Close and + ``days_to_maturity`` = (ContractMonth - asof).days. Returns the rows whose + ``Datetime`` falls on ``asof``. + """ + curve = ( + self.df.filter( + (pl.col("Commodity") == commodity) & (pl.col("Datetime").dt.date() == asof) + ) + .select(["ContractMonth", "Close"]) + .unique(subset=["ContractMonth"]) + .sort("ContractMonth") + .with_columns( + (pl.col("ContractMonth") - pl.lit(asof)).dt.total_days().alias("days_to_maturity") + ) + ) + return curve + + def term_structure_slope(self, commodity: str, asof: date) -> float: + """Annualized front-to-back slope: (back/front - 1) * 365 / (back_dtm - front_dtm). + + Returns ``nan`` when the curve has fewer than two points or the maturity span / + front price is degenerate. + """ + curve = self.curve_as_of(commodity, asof) + if curve.height < 2: + return float("nan") + front = curve.row(0, named=True) + back = curve.row(-1, named=True) + span = back["days_to_maturity"] - front["days_to_maturity"] + if span == 0 or front["Close"] == 0: + return float("nan") + return float((back["Close"] / front["Close"] - 1.0) * 365.0 / span) + + def roll_yield(self, commodity: str, asof: date) -> pl.DataFrame: + """Annualized roll yield between each adjacent contract pair. + + For neighbours (near, far): (near/far - 1) * 365 / (far_dtm - near_dtm). Positive + in backwardation (near > far). Returns one row per adjacent pair. + """ + curve = self.curve_as_of(commodity, asof) + if curve.height < 2: + return pl.DataFrame( + schema={ + "ContractMonth": pl.Date, + "ContractMonth_far": pl.Date, + "roll_yield": pl.Float64, + } + ) + return ( + curve.with_columns( + pl.col("ContractMonth").shift(-1).alias("ContractMonth_far"), + pl.col("Close").shift(-1).alias("Close_far"), + pl.col("days_to_maturity").shift(-1).alias("dtm_far"), + ) + .drop_nulls("Close_far") + .with_columns( + ( + (pl.col("Close") / pl.col("Close_far") - 1.0) + * 365.0 + / (pl.col("dtm_far") - pl.col("days_to_maturity")) + ).alias("roll_yield") + ) + .select(["ContractMonth", "ContractMonth_far", "roll_yield"]) + ) + + def shape(self, commodity: str, asof: date) -> str: + """Curve shape from the sign of the slope. + + Returns "backwardation" (downward-sloping), "contango" (upward-sloping), or + "flat" (zero / undefined slope). + """ + slope = self.term_structure_slope(commodity, asof) + if slope != slope or slope == 0.0: # nan or exactly flat + return "flat" + return "contango" if slope > 0 else "backwardation" diff --git a/src/energex/analysis/futures.py b/src/energex/analysis/futures.py index 3486f8d..8aa5157 100644 --- a/src/energex/analysis/futures.py +++ b/src/energex/analysis/futures.py @@ -8,8 +8,10 @@ # clearly instead of crashing on a missing 'expiry' column or invalid Polars APIs. _NEEDS_CONTRACT_MODEL = ( "{name} requires a contract-month/expiry data model (dated contracts, and for " - "carry/implied-rate a spot leg + risk-free curve). The store only holds continuous " - "front-month proxies (CL=F/BZ=F/NG=F). See ASSESSMENT.md R8." + "carry/implied-rate a spot leg + risk-free curve); this FuturesAnalyzer only has " + "continuous front-month proxies (CL=F/BZ=F/NG=F). For real dated-contract term " + "structure, slope, and roll yield use energex.analysis.dated_futures." + "DatedFuturesAnalyzer over the daily_contracts table. See ASSESSMENT.md R8." ) @@ -59,11 +61,11 @@ def calculate_basis_risk(self, spot_symbol: str, futures_symbol: str) -> pl.Data def analyze_roll_yield( self, front_month: str, back_month: str, window_minutes: int = 30 ) -> pl.DataFrame: - """Annualized roll yield — gated (needs contract expiries).""" + """Annualized roll yield — gated. Use DatedFuturesAnalyzer.roll_yield instead.""" raise NotImplementedError(_NEEDS_CONTRACT_MODEL.format(name="analyze_roll_yield")) def analyze_futures_curve(self, symbols: list[str]) -> pl.DataFrame: - """Cross-sectional curve by expiry — gated (needs contract expiries).""" + """Cross-sectional curve by expiry — gated. Use DatedFuturesAnalyzer.curve_as_of.""" raise NotImplementedError(_NEEDS_CONTRACT_MODEL.format(name="analyze_futures_curve")) def calculate_implied_rates( From ced930b0b42e497032d0197e800e38155f4321cc Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 14 Jun 2026 16:17:16 -0400 Subject: [PATCH 4/6] Wire dated ingestion and /curve endpoint into the service Add run_dated_ingestion (fetch -> upsert -> checkpoint), run it alongside intraday ingestion in the scheduler gated on dated_enabled, expose a /curve route, and report a daily_contracts row count from /healthz. --- src/energex/service/app.py | 42 ++++++++++++++++++++++++++++++--- src/energex/service/pipeline.py | 17 +++++++++++++ 2 files changed, 56 insertions(+), 3 deletions(-) diff --git a/src/energex/service/app.py b/src/energex/service/app.py index 556926e..6f05dcb 100644 --- a/src/energex/service/app.py +++ b/src/energex/service/app.py @@ -8,20 +8,21 @@ import os from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from datetime import datetime, timezone +from datetime import date, datetime, timezone from typing import Any import polars as pl from fastapi import FastAPI, HTTPException, Query from fastapi.responses import HTMLResponse +from energex.analysis.dated_futures import DatedFuturesAnalyzer from energex.analysis.futures import FuturesAnalyzer from energex.analysis.market_sentiment import MarketSentimentAnalyzer from energex.analysis.volatility import VolatilityAnalyzer from energex.config import get_settings from energex.database import EnergyDatabase from energex.logging_config import setup_logging -from energex.service.pipeline import run_ingestion +from energex.service.pipeline import run_dated_ingestion, run_ingestion from energex.service.scheduler import build_scheduler from energex.visualization.charts import MarketVisualizer @@ -54,7 +55,16 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: app.state.db = db app.state.scheduler = None if start_scheduler: - scheduler = build_scheduler(lambda: run_ingestion(db), cron=DEFAULT_CRON) + + def _ingest_job() -> None: + run_ingestion(db) + if settings.data_fetch.dated_enabled: + try: + run_dated_ingestion(db) + except Exception as e: # pragma: no cover - defensive + logger.error("Dated ingestion failed: %s", e) + + scheduler = build_scheduler(_ingest_job, cron=DEFAULT_CRON) scheduler.start() app.state.scheduler = scheduler logger.info("energex service started (db=%s)", resolved_db_path) @@ -77,13 +87,18 @@ def healthz() -> dict[str, Any]: .execute("SELECT COUNT(*), MAX(Datetime) FROM intraday_prices") .fetchone() ) + contracts_row = ( + db.conn.cursor().execute("SELECT COUNT(*) FROM daily_contracts").fetchone() + ) except Exception as e: # pragma: no cover - defensive raise HTTPException(status_code=503, detail=f"database error: {e}") from e count, latest = row if row is not None else (0, None) + contract_count = contracts_row[0] if contracts_row is not None else 0 return { "status": "ok", "rows": count, "latest": latest.isoformat() if latest else None, + "contract_rows": contract_count, } @app.get("/prices") @@ -119,6 +134,27 @@ def futures(front: str = Query(...), back: str = Query(...)) -> list[dict[str, A out = FuturesAnalyzer(df).calculate_term_structure(front, back) return out.select(["Datetime", "spread", "spread_pct"]).to_dicts() + @app.get("/curve") + def curve( + commodity: str = Query(...), asof: str | None = Query(default=None) + ) -> list[dict[str, Any]]: + df = _read_df( + app.state.db, + "SELECT * FROM daily_contracts WHERE Commodity = ? ORDER BY Datetime, ContractMonth", + [commodity], + ) + if df.is_empty(): + raise HTTPException(status_code=404, detail=f"no dated contracts for {commodity}") + asof_date = ( + date.fromisoformat(asof) + if asof is not None + else df.select(pl.col("Datetime").dt.date().max()).item() + ) + out = DatedFuturesAnalyzer(df).curve_as_of(commodity, asof_date) + if out.is_empty(): + raise HTTPException(status_code=404, detail=f"no curve for {commodity} on {asof_date}") + return out.select(["ContractMonth", "Close", "days_to_maturity"]).to_dicts() + @app.get("/sentiment") def sentiment(headline: str = Query(...), summary: str | None = None) -> dict[str, Any]: # The provider comes from config (set DEFAULT_LLM_PROVIDER=ollama + diff --git a/src/energex/service/pipeline.py b/src/energex/service/pipeline.py index fc831ef..505c4d4 100644 --- a/src/energex/service/pipeline.py +++ b/src/energex/service/pipeline.py @@ -25,3 +25,20 @@ def run_ingestion(db: EnergyDatabase, source_name: str | None = None) -> int: db.conn.execute("CHECKPOINT") logger.info("Ingestion upserted %d rows (source=%s)", df.height, source.name) return df.height + + +def run_dated_ingestion(db: EnergyDatabase, source_name: str | None = None) -> int: + """Fetch the dated contract strip and upsert into ``daily_contracts``. + + Mirrors :func:`run_ingestion`; the upsert on (Commodity, ContractMonth, Datetime) + makes repeated runs idempotent. + """ + source = get_data_source(source_name or os.environ.get("ENERGEX_DATA_SOURCE", "yfinance")) + df = source.fetch_dated() + if df.is_empty(): + logger.info("Dated ingestion produced no rows (source=%s)", source.name) + return 0 + db.insert_daily_contracts(df) + db.conn.execute("CHECKPOINT") + logger.info("Dated ingestion upserted %d rows (source=%s)", df.height, source.name) + return df.height From 4f716a6285ab6caa26007186e50ab10a8dbde2a7 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 14 Jun 2026 16:17:21 -0400 Subject: [PATCH 5/6] Test dated contracts, curve analytics, and ingestion; document feature Add storage idempotence/PK tests, DatedFuturesAnalyzer backwardation/contango cases, pure ticker-enumeration unit tests, source/pipeline fetch_dated coverage, a sample_daily_contracts fixture, and README notes. --- README.md | 7 +++ tests/conftest.py | 42 ++++++++++++- tests/test_data_fetcher.py | 21 ++++++- tests/test_dated_contracts.py | 110 ++++++++++++++++++++++++++++++++++ tests/test_dated_futures.py | 48 +++++++++++++++ tests/test_service.py | 35 +++++++++++ tests/test_sources.py | 9 +++ 7 files changed, 270 insertions(+), 2 deletions(-) create mode 100644 tests/test_dated_contracts.py create mode 100644 tests/test_dated_futures.py diff --git a/README.md b/README.md index cc3104c..01684fc 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,7 @@ Energy derivatives data collection and analysis system with advanced analytics f ### Data Collection - Real-time intraday data fetching (1-minute intervals) - Support for major energy futures contracts (CL=F, BZ=F, NG=F) +- Dated futures contract curves: the forward strip of monthly contracts per commodity (daily settlements), enabling real term-structure analytics - Efficient storage using DuckDB - Robust error handling and validation @@ -28,6 +29,12 @@ Energy derivatives data collection and analysis system with advanced analytics f - Basis risk measurement - Implied interest rates +- **Dated Futures Curves** (`DatedFuturesAnalyzer`): + - Forward curve as of a date (Close + days-to-maturity per contract month) + - Annualized term-structure slope + - Per-adjacent-pair annualized roll yield + - Curve shape classification (backwardation / contango / flat) + - **Market Sentiment Analysis** ✨ *New in v0.3.0*: - AI-powered news sentiment analysis - Multi-LLM provider support (OpenAI, Anthropic, Ollama) diff --git a/tests/conftest.py b/tests/conftest.py index c732c23..adea478 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,7 +2,7 @@ from __future__ import annotations -from datetime import datetime, timedelta +from datetime import date, datetime, timedelta, timezone import polars as pl import pytest @@ -34,6 +34,46 @@ def sample_ohlcv() -> pl.DataFrame: return pl.DataFrame(rows) +@pytest.fixture +def sample_daily_contracts() -> pl.DataFrame: + """Dated-contract daily settlements for two commodities over two snapshot days. + + ``crude`` is backwardated (near > far); ``gas`` is contango (near < far). Datetimes + are tz-aware UTC; ContractMonth is the first of the delivery month. + """ + rows = [] + snapshots = [ + datetime(2024, 1, 2, 0, 0, tzinfo=timezone.utc), + datetime(2024, 1, 3, 0, 0, tzinfo=timezone.utc), + ] + # (commodity, root, base price, per-month step) — negative step = backwardation. + specs = [ + ("crude", "CL", 80.0, -0.5), + ("gas", "NG", 2.0, 0.05), + ] + month_codes = "FGHJKMNQUVXZ" + for snap_i, snap in enumerate(snapshots): + for commodity, root, base, step in specs: + for m in range(6): + contract_month = date(2024, 2 + m, 1) + close = base + step * m + snap_i * 0.1 + rows.append( + { + "Datetime": snap, + "Commodity": commodity, + "ContractMonth": contract_month, + "Symbol": f"{root}{month_codes[1 + m]}24.NYM", + "Open": close - 0.1, + "High": close + 0.2, + "Low": close - 0.2, + "Close": close, + "Volume": 1000 + m * 10, + "OpenInterest": 5000 + m * 100, + } + ) + return pl.DataFrame(rows) + + @pytest.fixture def tmp_db_path(tmp_path) -> str: """An isolated DuckDB file path under tmp — never the repo's energy.db.""" diff --git a/tests/test_data_fetcher.py b/tests/test_data_fetcher.py index c1e1190..15cc705 100644 --- a/tests/test_data_fetcher.py +++ b/tests/test_data_fetcher.py @@ -1,16 +1,35 @@ """Tests for ingestion resilience: retries, error classification, logging (R10).""" import logging +from datetime import date import pandas as pd import polars as pl import pytest from energex import data_fetcher as df_mod -from energex.data_fetcher import EnergyDataFetcher +from energex.data_fetcher import MONTH_CODES, EnergyDataFetcher, _dated_tickers from energex.exceptions import DataFetchError +def test_dated_tickers_count_codes_and_months(): + out = _dated_tickers("CL", date(2024, 1, 15), 12) + assert len(out) == 12 + # First contract: January code 'F', year 24, first of month. + assert out[0] == ("CLF24.NYM", date(2024, 1, 1)) + # Codes follow F G H J K M N Q U V X Z over the year. + assert [t[0][2] for t in out] == list(MONTH_CODES) + assert [t[1] for t in out] == [date(2024, m, 1) for m in range(1, 13)] + + +def test_dated_tickers_year_rollover(): + out = _dated_tickers("NG", date(2024, 11, 1), 3) + assert out[0] == ("NGX24.NYM", date(2024, 11, 1)) + assert out[1] == ("NGZ24.NYM", date(2024, 12, 1)) + # Rolls into the next year with the January code. + assert out[2] == ("NGF25.NYM", date(2025, 1, 1)) + + @pytest.fixture(autouse=True) def _no_sleep(monkeypatch): # Keep retry/backoff tests fast. diff --git a/tests/test_dated_contracts.py b/tests/test_dated_contracts.py new file mode 100644 index 0000000..087d0e6 --- /dev/null +++ b/tests/test_dated_contracts.py @@ -0,0 +1,110 @@ +"""Tests for daily_contracts storage: idempotent upsert + PK integrity.""" + +from datetime import date, datetime, timezone + +import polars as pl +import pytest + +from energex.database import EnergyDatabase +from energex.exceptions import DatabaseError + + +def test_history_survives_reinstantiation(sample_daily_contracts, tmp_db_path): + db = EnergyDatabase(tmp_db_path) + db.insert_daily_contracts(sample_daily_contracts) + db.conn.close() + + db2 = EnergyDatabase(tmp_db_path) + count = db2.conn.execute("SELECT COUNT(*) FROM daily_contracts").fetchone()[0] + db2.conn.close() + assert count == sample_daily_contracts.height + + +def test_reinserting_same_batch_does_not_duplicate(sample_daily_contracts, tmp_db_path): + db = EnergyDatabase(tmp_db_path) + db.insert_daily_contracts(sample_daily_contracts) + db.insert_daily_contracts(sample_daily_contracts) + count = db.conn.execute("SELECT COUNT(*) FROM daily_contracts").fetchone()[0] + db.conn.close() + assert count == sample_daily_contracts.height + + +def test_upsert_updates_close_on_conflict(sample_daily_contracts, tmp_db_path): + db = EnergyDatabase(tmp_db_path) + db.insert_daily_contracts(sample_daily_contracts) + + revised = sample_daily_contracts.with_columns((pl.col("Close") + 1.0).alias("Close")) + db.insert_daily_contracts(revised) + + count = db.conn.execute("SELECT COUNT(*) FROM daily_contracts").fetchone()[0] + first = db.conn.execute( + "SELECT Close FROM daily_contracts " + "WHERE Commodity = 'crude' ORDER BY ContractMonth, Datetime LIMIT 1" + ).fetchone()[0] + db.conn.close() + + expected = revised.filter(pl.col("Commodity") == "crude").sort(["ContractMonth", "Datetime"])[ + "Close" + ][0] + assert count == sample_daily_contracts.height + assert first == pytest.approx(expected) + + +def test_read_only_mode_rejects_writes(sample_daily_contracts, tmp_db_path): + db = EnergyDatabase(tmp_db_path) + db.insert_daily_contracts(sample_daily_contracts) + db.conn.close() + + ro = EnergyDatabase(tmp_db_path, read_only=True) + count = ro.conn.execute("SELECT COUNT(*) FROM daily_contracts").fetchone()[0] + assert count == sample_daily_contracts.height + with pytest.raises(Exception): # noqa: B017 - duckdb raises on write to a read-only conn + ro.conn.execute( + "INSERT INTO daily_contracts VALUES " + "(NULL, 'x', DATE '2024-02-01', 'X', 1, 1, 1, 1, 1, 1)" + ) + ro.conn.close() + + +def test_reset_clears_daily_contracts(sample_daily_contracts, tmp_db_path): + db = EnergyDatabase(tmp_db_path) + db.insert_daily_contracts(sample_daily_contracts) + db.reset() + count = db.conn.execute("SELECT COUNT(*) FROM daily_contracts").fetchone()[0] + db.conn.close() + assert count == 0 + + +def test_insert_rejects_frame_missing_required_columns(tmp_db_path): + db = EnergyDatabase(tmp_db_path) + bad = pl.DataFrame( + { + "Datetime": [datetime(2024, 1, 2, tzinfo=timezone.utc)], + "Commodity": ["crude"], + "ContractMonth": [date(2024, 2, 1)], + } + ) + with pytest.raises(DatabaseError): + db.insert_daily_contracts(bad) + db.conn.close() + + +def test_primary_key_is_commodity_month_datetime(sample_daily_contracts, tmp_db_path): + db = EnergyDatabase(tmp_db_path) + db.insert_daily_contracts(sample_daily_contracts) + # Same (Commodity, ContractMonth, Datetime) but different Symbol must upsert, not add. + dup = sample_daily_contracts.head(1).with_columns(pl.lit("OTHER.NYM").alias("Symbol")) + db.insert_daily_contracts(dup) + count = db.conn.execute("SELECT COUNT(*) FROM daily_contracts").fetchone()[0] + sym = db.conn.execute( + "SELECT Symbol FROM daily_contracts WHERE Commodity = ? AND ContractMonth = ? " + "AND Datetime = ?", + [ + dup["Commodity"][0], + dup["ContractMonth"][0], + dup["Datetime"][0], + ], + ).fetchone()[0] + db.conn.close() + assert count == sample_daily_contracts.height + assert sym == "OTHER.NYM" diff --git a/tests/test_dated_futures.py b/tests/test_dated_futures.py new file mode 100644 index 0000000..00a83bd --- /dev/null +++ b/tests/test_dated_futures.py @@ -0,0 +1,48 @@ +"""Tests for DatedFuturesAnalyzer term-structure analytics.""" + +from datetime import date + +from energex.analysis.dated_futures import DatedFuturesAnalyzer + +ASOF = date(2024, 1, 2) + + +def test_curve_as_of_sorted_with_days_to_maturity(sample_daily_contracts): + curve = DatedFuturesAnalyzer(sample_daily_contracts).curve_as_of("crude", ASOF) + assert curve.height == 6 + months = curve["ContractMonth"].to_list() + assert months == sorted(months) + # First contract month is 2024-02-01 -> 30 days out from 2024-01-02. + assert curve["days_to_maturity"][0] == (date(2024, 2, 1) - ASOF).days + + +def test_backwardation_case(sample_daily_contracts): + analyzer = DatedFuturesAnalyzer(sample_daily_contracts) + slope = analyzer.term_structure_slope("crude", ASOF) + assert slope < 0 + assert analyzer.shape("crude", ASOF) == "backwardation" + + +def test_contango_case(sample_daily_contracts): + analyzer = DatedFuturesAnalyzer(sample_daily_contracts) + slope = analyzer.term_structure_slope("gas", ASOF) + assert slope > 0 + assert analyzer.shape("gas", ASOF) == "contango" + + +def test_roll_yield_signs(sample_daily_contracts): + analyzer = DatedFuturesAnalyzer(sample_daily_contracts) + crude = analyzer.roll_yield("crude", ASOF) + assert crude.height == 5 + # Backwardation: near > far -> positive roll yield. + assert all(v > 0 for v in crude["roll_yield"].to_list()) + + gas = analyzer.roll_yield("gas", ASOF) + assert all(v < 0 for v in gas["roll_yield"].to_list()) + + +def test_slope_nan_safe_when_single_contract(sample_daily_contracts): + one = sample_daily_contracts.filter(sample_daily_contracts["ContractMonth"] == date(2024, 2, 1)) + slope = DatedFuturesAnalyzer(one).term_structure_slope("crude", ASOF) + assert slope != slope # nan + assert DatedFuturesAnalyzer(one).shape("crude", ASOF) == "flat" diff --git a/tests/test_service.py b/tests/test_service.py index ef0227f..b1c9ac2 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -120,6 +120,41 @@ def test_run_ingestion_upserts(monkeypatch, tmp_db_path): assert count == 1 +def test_run_dated_ingestion_upserts(monkeypatch, tmp_db_path, sample_daily_contracts): + from energex.service import pipeline + from energex.sources.yfinance_source import YFinanceDataSource + + monkeypatch.setattr(YFinanceDataSource, "fetch_dated", lambda self: sample_daily_contracts) + db = EnergyDatabase(tmp_db_path) + n = pipeline.run_dated_ingestion(db) + count = db.conn.execute("SELECT COUNT(*) FROM daily_contracts").fetchone()[0] + db.conn.close() + assert n == sample_daily_contracts.height + assert count == sample_daily_contracts.height + + +def test_curve_endpoint(tmp_db_path, sample_daily_contracts): + db = EnergyDatabase(tmp_db_path) + db.insert_daily_contracts(sample_daily_contracts) + db.conn.close() + app = create_app(db_path=tmp_db_path, start_scheduler=False) + with TestClient(app) as c: + r = c.get("/curve", params={"commodity": "crude", "asof": "2024-01-02"}) + assert r.status_code == 200 + data = r.json() + assert len(data) == 6 + assert "days_to_maturity" in data[0] + # backwardation: closes decline along the curve + closes = [row["Close"] for row in data] + assert closes == sorted(closes, reverse=True) + + miss = c.get("/curve", params={"commodity": "nope"}) + assert miss.status_code == 404 + + health = c.get("/healthz").json() + assert health["contract_rows"] == sample_daily_contracts.height + + def test_sentiment_endpoint_returns_analysis(client, monkeypatch): from energex.analysis.market_sentiment import MarketSentimentAnalyzer diff --git a/tests/test_sources.py b/tests/test_sources.py index 0a6b128..3535f24 100644 --- a/tests/test_sources.py +++ b/tests/test_sources.py @@ -32,6 +32,13 @@ def test_yfinance_source_delegates_to_fetcher(monkeypatch): assert out.equals(fake) +def test_yfinance_source_fetch_dated_delegates(monkeypatch): + fake = pl.DataFrame({"Commodity": ["crude"]}) + monkeypatch.setattr(df_mod.EnergyDataFetcher, "fetch_all_dated", lambda self: fake) + out = get_data_source("yfinance").fetch_dated() + assert out.equals(fake) + + @pytest.mark.parametrize("name", ["databento", "ice-brent", "eia-spot"]) def test_stub_sources_raise_not_implemented(name): src = get_data_source(name) @@ -40,3 +47,5 @@ def test_stub_sources_raise_not_implemented(name): src.fetch_all() with pytest.raises(NotImplementedError): src.fetch("crude") + with pytest.raises(NotImplementedError): + src.fetch_dated() From cd3fa78f60d089b2b5337849fa8aecab257c2a89 Mon Sep 17 00:00:00 2001 From: oldhero5 Date: Sun, 14 Jun 2026 16:20:53 -0400 Subject: [PATCH 6/6] Guard roll yield against degenerate maturity spans; stamp daily bars UTC roll_yield now emits null (not silent inf) for same- or inverted-maturity adjacent pairs. Daily contract bars are tz-naive from the vendor, so stamp them UTC explicitly rather than relying on the session TimeZone at insert. --- src/energex/analysis/dated_futures.py | 10 +++++----- src/energex/data_fetcher.py | 6 ++++++ 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/src/energex/analysis/dated_futures.py b/src/energex/analysis/dated_futures.py index 9ab7037..57064b4 100644 --- a/src/energex/analysis/dated_futures.py +++ b/src/energex/analysis/dated_futures.py @@ -81,12 +81,12 @@ def roll_yield(self, commodity: str, asof: date) -> pl.DataFrame: pl.col("days_to_maturity").shift(-1).alias("dtm_far"), ) .drop_nulls("Close_far") + .with_columns((pl.col("dtm_far") - pl.col("days_to_maturity")).alias("_span")) .with_columns( - ( - (pl.col("Close") / pl.col("Close_far") - 1.0) - * 365.0 - / (pl.col("dtm_far") - pl.col("days_to_maturity")) - ).alias("roll_yield") + pl.when(pl.col("_span") > 0) + .then((pl.col("Close") / pl.col("Close_far") - 1.0) * 365.0 / pl.col("_span")) + .otherwise(None) # degenerate same-/inverted-maturity pair: no roll yield + .alias("roll_yield") ) .select(["ContractMonth", "ContractMonth_far", "roll_yield"]) ) diff --git a/src/energex/data_fetcher.py b/src/energex/data_fetcher.py index af0a9ad..371f4aa 100644 --- a/src/energex/data_fetcher.py +++ b/src/energex/data_fetcher.py @@ -150,6 +150,12 @@ def fetch_dated_contracts(self, commodity: str) -> pl.DataFrame: pl.lit(ticker).alias("Symbol"), pl.col("OpenInterest").cast(pl.Int64), ) + # Daily bars come back tz-naive (date-only); stamp them as UTC + # explicitly so the TIMESTAMPTZ column is host-independent rather + # than relying on the session TimeZone at insert time. + dt = df.schema["Datetime"] + if isinstance(dt, pl.Datetime) and dt.time_zone is None: + df = df.with_columns(pl.col("Datetime").dt.replace_time_zone("UTC")) df = normalize_datetime_to_utc(df) df = df.select(EnergyDatabase.DAILY_CONTRACTS_COLUMNS) dfs.append(df)