Datos y rollovers de futuros CME

Código Machine Learning for Trading

Resumen

Esta guía de datos describe una colección de contratos de futuros continuos CME que abarca índices bursátiles, tipos de interés, energía, metales, divisas, agricultura y ganadería. Explica los datos de origen horarios y la frecuencia diaria derivada, los distintos vencimientos, la configuración de productos y el filtrado mediante un cargador. El proceso de descarga divide los archivos por producto y año, permite estimar costes y hacer simulaciones de descarga, y recomienda consultar las tarifas del proveedor antes de solicitar datos.

Los contratos usan rollovers basados en volumen: la transición ocurre cuando el volumen del día anterior del siguiente contrato supera al del contrato actual, usando información disponible antes de la siguiente sesión. La guía presenta esto como una convención de rollover orientada al trading y distingue los vencimientos del contrato más próximo, el segundo y el tercero. Es una referencia para adquirir datos y conocer su cobertura, no una evaluación de estrategias; no aporta pruebas sobre el comportamiento de los rendimientos ni sobre el desempeño de los rollovers. El proveedor de datos es de pago, y las descargas sustituyen los historiales completos en lugar de actualizarlos de forma incremental, lo que afecta al coste y mantenimiento de los conjuntos de datos de investigación.

Ideas clave

El conjunto de datos abarca futuros continuos de varios grupos de activos y ofrece datos horarios y diarios derivados.
Un rollover basado en volumen utiliza el volumen del día anterior para determinar cuándo cambiar de contrato.
Los distintos vencimientos representan los meses del contrato más próximo, el segundo y el tercero.
La estimación de costes y las descargas de prueba ayudan a gestionar el acceso de pago a los datos.
Como las descargas sustituyen el historial completo, la frecuencia de actualización debe tener en cuenta el coste.

Etiquetas

Futuros Multiactivo Materias primas Pruebas retrospectivas

Texto completo

# dataset_card.py


```py
# ---
# jupyter:
#   jupytext:
#     cell_metadata_filter: -all
#     text_representation:
#       extension: .py
#       format_name: percent
#       format_version: '1.3'
#       jupytext_version: 1.19.3
#   kernelspec:
#     display_name: Python 3 (ipykernel)
#     language: python
#     name: python3
# ---

# %% [markdown]
# # CME Futures Dataset
#
# Continuous futures contracts from CME Group via Databento.
#
# | Property | Value |
# |----------|-------|
# | **Provider** | Databento |
# | **Asset Class** | Futures (Equity, Rates, Energy, Metals, FX, Ags) |
# | **Frequency** | Hourly, Daily (derived) |
# | **Products** | 30 core + 6 extension |
# | **Coverage** | 2011-2025 |
# | **Size** | ~500 MB |
# | **API Key** | `DATABENTO_API_KEY` (**PAID**) |
# | **Loader** | `load_cme_futures()` |
#
# **WARNING**: Databento is a paid data provider. Always estimate costs before downloading.

# %%
"""CME Futures - download, explore, and update workflow."""

import json
import os
from pathlib import Path

import polars as pl
import yaml
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# %% [markdown]
# ## 1. Configuration
#
# The futures universe is defined in `config.yaml`. Includes 30 core
# products across equity indices, treasuries, energy, metals, currencies, and
# agriculture.

# %%
# Load and display configuration
config_path = Path("config.yaml")
config = yaml.safe_load(config_path.read_text())

print("=== CME Futures Configuration ===")
print(f"Dataset: {config['dataset']}")
print(f"Schema: {config['schema']}")
print(f"Roll type: {config['roll_type']} (volume-based)")
print(f"Tenors: {config['tenors']} (front, second, third month)")
print(f"Date range: {config['default_start']} to {config['default_end']}")
print("\nProduct categories:")

# Count by category
categories = {}
for product, info in config["products"].items():
    cat = info.get("category", "unknown")
    categories[cat] = categories.get(cat, 0) + 1

for cat, count in sorted(categories.items()):
    print(f"  {cat}: {count} products")

print(f"\nTotal core products: {len(config['products'])}")
print(f"Extension products: {len(config.get('extension_products', {}))}")

# %% [markdown]
# ## 2. API Key Setup
#
# **Databento is a paid data provider.** New accounts receive $125 free credit.
#
# ### Getting a Databento API Key
#
# 1. Sign up at [Databento](https://databento.com/signup) ($125 free credit)
# 2. Navigate to **API Keys** in your dashboard
# 3. Create a new API key
# 4. Add to your `.env` file in the repository root:
#
# ```bash
# DATABENTO_API_KEY=db-your-api-key-here
# ```
#
# ### Cost Reference
#
# | Data Type | Cost Estimate |
# |-----------|---------------|
# | Hourly OHLCV | ~$0.50-1.00 per product per year |
# | Daily OHLCV | ~$0.05-0.10 per product per year |
# | Full 30 products x 15 years | ~$75-100 |
#
# **ALWAYS run cost estimation before downloading!**

# %%
# Verify API key is configured
api_key = os.getenv("DATABENTO_API_KEY")
if api_key:
    # Show partial key for verification
    print(f"DATABENTO_API_KEY: {api_key[:8]}... (configured)")
else:
    print("WARNING: DATABENTO_API_KEY not set in environment")
    print("Sign up at: https://databento.com/signup ($125 free credit)")
    print("Add to .env file: DATABENTO_API_KEY=db-your-key-here")

# %% [markdown]
# ## 3. Download Data
#
# **IMPORTANT**: Always run cost estimation before downloading!
#
# The download:
# - Uses Hive partitioning by product/year for efficient updates
# - Downloads full date range per product in one API call (cost efficient)
# - Stores V0, V1, V2 tenors (front, second, third month) stacked


# %%
def estimate_futures_cost(products: list[str] | None = None) -> float:
    """Estimate download cost from Databento.

    Args:
        products: Specific products to estimate (default: all from config)

    Returns:
        Estimated cost in USD
    """
    import databento as db

    api_key = os.getenv("DATABENTO_API_KEY")
    if not api_key:
        raise ValueError("DATABENTO_API_KEY not set. See API Key Setup section.")

    # Load config
    config = yaml.safe_load(config_path.read_text())

    if products is None:
        products = list(config["products"].keys())

    client = db.Historical()
    total_cost = 0.0

    print("=== Cost Estimation ===")
    print(f"Products: {len(products)}")
    print(f"Tenors: {config['tenors']}")
    print(f"Date range: {config['default_start']} to {config['default_end']}")
    print()

    for product in products:
        product_info = config["products"].get(product, {})
        start = product_info.get("start", config["default_start"])

        # Build symbols for continuous contracts
        symbols = [f"{product}.{config['roll_type']}.{pos}" for pos in config["tenors"]]

        try:
            cost = client.metadata.get_cost(
                dataset=config["dataset"],
                symbols=symbols,
                schema=config["schema"],
                start=start,
                end=config["default_end"],
                stype_in="continuous",
            )
            total_cost += cost
            print(f"  {product}: ${cost:.2f}")
        except Exception as e:
            print(f"  {product}: ERROR - {e}")

    print(f"\n{'=' * 40}")
    print(f"TOTAL ESTIMATED COST: ${total_cost:.2f}")
    print(f"{'=' * 40}")

    return total_cost


def download_futures_data(
    products: list[str] | None = None,
    dry_run: bool = True,  # Default to dry_run=True for safety!
    force: bool = False,
):
    """Download CME futures data from Databento.

    Args:
        products: Specific products to download (default: all from config)
        dry_run: If True, show what would be downloaded without doing it (DEFAULT: True)
        force: If True, re-download even if data exists
    """
    import databento as db

    from utils import ML4T_DATA_PATH

    api_key = os.getenv("DATABENTO_API_KEY")
    if not api_key:
        raise ValueError("DATABENTO_API_KEY not set. See API Key Setup section.")

    # Load config
    config = yaml.safe_load(config_path.read_text())

    if products is None:
        products = list(config["products"].keys())

    output_dir = ML4T_DATA_PATH / "futures" / "market" / "continuous" / "hourly"

    print("=== CME Futures Download ===")
    print(f"Products: {len(products)}")
    print(f"Tenors: {config['tenors']}")
    print(f"Date range: {config['default_start']} to {config['default_end']}")
    print(f"Output: {output_dir}")

    if dry_run:
        print("\n[DRY RUN] Would download:")
        for product in products:
            product_info = config["products"].get(product, {})
            start = product_info.get("start", config["default_start"])
            print(f"  {product}: {start} to {config['default_end']}")
        print("\nRun estimate_futures_cost() to see cost estimate.")
        print("Set dry_run=False to actually download.")
        return

    # Initialize client
    client = db.Historical()
    total_rows = 0

    print(f"\nDownloading {len(products)} products...")
    for product in products:
        product_info = config["products"].get(product, {})
        start = product_info.get("start", config["default_start"])

        # Check existing
        product_dir = output_dir / f"product={product}"
        if product_dir.exists() and not force:
            existing_years = list(product_dir.glob("year=*/data.parquet"))
            if existing_years:
                print(
                    f"  {product}: Already exists ({len(existing_years)} years). Use force=True to re-download."
                )
                continue

        # Build symbols for continuous contracts
        symbols = [f"{product}.{config['roll_type']}.{pos}" for pos in config["tenors"]]

        print(f"  {product}...", end=" ", flush=True)
        try:
            data = client.timeseries.get_range(
                dataset=config["dataset"],
                symbols=symbols,
                schema=config["schema"],
                start=start,
                end=config["default_end"],
                stype_in="continuous",
            )

            df = data.to_df()
            if len(df) == 0:
                print("WARNING (no data)")
                continue

            # Convert to polars and add metadata
            df_pl = pl.from_pandas(df.reset_index())
            df_pl = df_pl.with_columns(pl.lit(product).alias("product"))

            # Extract tenor from symbol (ES.v.0 -> 0)
            if "symbol" in df_pl.columns:
                df_pl = df_pl.with_columns(
                    pl.col("symbol")
                    .str.extract(rf"\.{config['roll_type']}\.(\d+)$", 1)
                    .cast(pl.Int8)
                    .alias("tenor")
                )

            # Partition by year
            df_pl = df_pl.with_columns(pl.col("ts_event").dt.year().alias("year"))

            for year in df_pl["year"].unique().sort().to_list():
                year_data = df_pl.filter(pl.col("year") == year)
                year_dir = output_dir / f"product={product}" / f"year={year}"
                year_dir.mkdir(parents=True, exist_ok=True)
                year_data.sort(["ts_event", "symbol"]).write_parquet(year_dir / "data.parquet")

            total_rows += len(df_pl)
            print(f"OK ({len(df_pl):,} rows)")

        except Exception as e:
            print(f"ERROR: {e}")

    print("\n=== Complete ===")
    print(f"Total rows: {total_rows:,}")
    print(f"Output: {output_dir}")


# %% [markdown]
# ### Estimate Cost (ALWAYS DO THIS FIRST!)

# %%
# Uncomment to estimate cost for all products
# estimate_futures_cost()

# Estimate for specific products
# estimate_futures_cost(products=["ES", "NQ", "CL", "GC"])

# %% [markdown]
# ### Download Data
#
# **WARNING**: This will consume Databento credits!

# %%
# Dry run (default) - shows what would be downloaded
download_futures_data(dry_run=True)

# %%
# Uncomment to actually download (after reviewing cost estimate!)
# download_futures_data(dry_run=False)

# Download specific products only
# download_futures_data(products=["ES", "NQ"], dry_run=False)

# %% [markdown]
# ## 4. Load and Explore
#
# Once downloaded, use the loader throughout the book:

# %%
from data import load_cme_futures

# Load daily continuous contracts (default)
df = load_cme_futures(frequency="daily")

print(f"Shape: {df.shape}")
print(f"Products: {df['product'].n_unique()}")
print(f"Date range: {df['session_date'].min()} to {df['session_date'].max()}")
print(f"Memory: {df.estimated_size('mb'):.1f} MB")

# %%
# Schema
df.schema

# %%
# Preview
df.head(10)

# %% [markdown]
# ### Coverage by Product

# %%
# Coverage and basic stats by product
coverage = (
    df.group_by("product")
    .agg(
        pl.col("session_date").min().alias("first_date"),
        pl.col("session_date").max().alias("last_date"),
        pl.len().alias("n_bars"),
        pl.col("volume").mean().alias("avg_volume"),
    )
    .sort("avg_volume", descending=True)
)
coverage

# %% [markdown]
# ### Hourly Data

# %%
# Load hourly data for specific products
hourly = load_cme_futures(frequency="hourly", products=["ES", "NQ"])
print(f"Hourly ES/NQ: {hourly.shape}")
print(f"Date range: {hourly['timestamp'].min()} to {hourly['timestamp'].max()}")

# %% [markdown]
# ## 5. Data Profile
#
# Profiles document the dataset structure, statistics, and quality metrics.

# %%
from ml4t.data.storage.data_profile import load_profile

from utils import ML4T_DATA_PATH

profile_path = ML4T_DATA_PATH / "futures" / "market" / "continuous" / "hourly" / "_profile.json"
profile = load_profile(profile_path)

if profile is None:
    print(f"No profile at {profile_path}")
    print(
        "Profiles are written next to the data by whatever builds the dataset - the\n"
        "download script in this directory, or the ml4t-data loader it drives - through\n"
        "ml4t.data.storage.data_profile. There is no separate profile-generating script,\n"
        "and nothing in this notebook writes one."
    )
else:
    print("=== Futures Profile ===")
    print(f"Written by {profile.source}")
    print(profile.summary())

# %% [markdown]
# ## 6. Loader Options
#
# The loader supports filtering by frequency, products, tenors, and date range:

# %%
# Daily frequency (default)
daily = load_cme_futures(frequency="daily")
print(f"Daily data: {daily.shape}")

# %%
# Specific products
equities = load_cme_futures(products=["ES", "NQ", "YM", "RTY"])
print(f"Equity indices only: {equities.shape}")

# %%
# Specific tenor (front month only)
front_month = load_cme_futures(tenors=[0])
print(f"Front month only: {front_month.shape}")

# %%
# Date range
recent = load_cme_futures(start_date="2024-01-01")
print(f"2024 onwards: {recent.shape}")

# %%
# Combined filters
filtered = load_cme_futures(
    frequency="daily",
    products=["ES", "CL", "GC"],
    tenors=[0],
    start_date="2020-01-01",
    end_date="2023-12-31",
)
print(f"ES/CL/GC front month 2020-2023: {filtered.shape}")

# %% [markdown]
# ## 7. Documentation
#
# ### Databento
# - [Databento Documentation](https://databento.com/docs/)
# - [CME Globex Dataset](https://databento.com/docs/datasets/cme)
# - [Continuous Contracts](https://databento.com/docs/schemas/continuous)
#
# ### Continuous Contract Construction
#
# The data uses **volume-based roll** (`.v.` suffix):
# - Roll occurs when previous day's volume shows next contract > current
# - This is realistic for trading (you know yesterday's volume at today's open)
#
# Available tenors:
# - **V0**: Front month (nearest expiry)
# - **V1**: Second month
# - **V2**: Third month
#
# ### Product Categories
#
# | Category | Products | Description |
# |----------|----------|-------------|
# | Equity Index | ES, NQ, YM, RTY | S&P 500, NASDAQ-100, Dow, Russell 2000 |
# | Treasury | ZN, ZB, ZF, ZT | 10Y, 30Y, 5Y, 2Y notes/bonds |
# | Energy | CL, NG, HO, RB | Crude, natural gas, heating oil, gasoline |
# | Metals | GC, SI, HG, PL | Gold, silver, copper, platinum |
# | FX | 6E, 6J, 6B, 6A, 6C, 6S | EUR, JPY, GBP, AUD, CAD, CHF |
# | Agriculture | ZC, ZS, ZW, ZM, ZL | Corn, soybeans, wheat, meal, oil |
# | Livestock | LE, HE, GF | Live cattle, lean hogs, feeder cattle |

# %% [markdown]
# ## 8. Updating Data
#
# To update with the latest data:
#
# ```python
# # Estimate cost first!
# estimate_futures_cost()
#
# # Download updates (re-downloads full history)
# download_futures_data(dry_run=False)
#
# # Force re-download specific products
# download_futures_data(products=["ES", "NQ"], force=True, dry_run=False)
# ```
#
# **Note**: Databento downloads replace full history (no incremental updates).
# Plan updates strategically to minimize cost.

# %% [markdown]
# ## Summary
#
# | Item | Value |
# |------|-------|
# | Products | 30 core (+ 6 extension) |
# | Frequencies | Hourly (raw), Daily (derived) |
# | Coverage | 2011-2025 |
# | Provider | Databento (**PAID** - $125 free credit) |
# | Config | `config.yaml` |
# | Loader | `load_cme_futures(frequency, products, tenors, start_date, end_date)` |
#
# **CRITICAL**: Always run `estimate_futures_cost()` before downloading!

```

Se muestra íntegramente con atribución según la licencia de la fuente. Licencia: MIT

Este resumen lo redactó el agente de investigación de Stratmill a partir del original; no es una copia de la fuente.

Datos de futuros continuos CME, rollovers por volumen y descarga

Resumen

Ideas clave

Etiquetas