Dated futures curves: ingest, store, and analyze real term structure

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)

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",
 ]

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("dtm_far") - pl.col("days_to_maturity")).alias("_span"))
+            .with_columns(
+                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"])
+        )
+
+    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"

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(

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)
 

src/energex/data_fetcher.py

@@ -1,18 +1,42 @@
 # 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
 import pytz
 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,90 @@ 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),
+            )
+            # 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)
+
+        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).