Futures History

import json
import re
import time
import requests
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Literal
from urllib.parse import quote

BASE_URL = 'https://api.insightsentry.com'
START_YEAR = 2025
BASE_SYMBOL = 'CME_MINI:NQ1!'
API_KEY = 'YOUR_KEY'
BarType = Literal['second', 'minute', 'hour']
BAR_TYPE: BarType = 'minute'
BAR_INTERVAL = 1
CONTRACT_LOOKBACK_MONTHS = 6
HISTORY_PARAMS = {
    'bar_interval': str(BAR_INTERVAL),
    # Optional /history params:
    # 'extended': 'true',
    # 'badj': 'true',
    # 'dadj': 'false',
    # 'settlement': 'false',
}
MAX_RETRIES = 5
REQUEST_TIMEOUT_SECONDS = 120
CONTRACT_CODE_PATTERN = re.compile(r'^(?P<month_code>[A-Z])(?P<year>\d{4})$')

FetchStatus = Literal['success', 'skip', 'retry']


@dataclass(frozen=True)
class FetchResult:
    status: FetchStatus
    data: dict[str, Any] | None = None
    message: str | None = None


@dataclass(frozen=True)
class ContractListing:
    month_code: str
    settlement_month: int
    settlement_date: datetime


def continuous_to_base_symbol(symbol_code):
    if symbol_code.endswith(('1!', '2!')):
        return symbol_code[:-2]
    return symbol_code


def extract_contract_month_code(base_code, contract_code):
    if not contract_code.startswith(base_code):
        raise ValueError(f'{contract_code} does not start with base code {base_code}')

    suffix = contract_code[len(base_code):]
    match = CONTRACT_CODE_PATTERN.fullmatch(suffix)
    if not match:
        raise ValueError(
            f'Could not parse {contract_code}. Expected format like {base_code}X2026'
        )

    return match.group('month_code')


def build_contract_schedule(contract_listings):
    month_by_code = {}
    for listing in sorted(contract_listings, key=lambda item: item.settlement_date):
        existing_month = month_by_code.setdefault(listing.month_code, listing.settlement_month)
        if existing_month != listing.settlement_month:
            raise ValueError(
                f'Contract month {listing.month_code} maps to both '
                f'{existing_month} and {listing.settlement_month}'
            )

    ordered_codes = tuple(
        code for code, _month in sorted(month_by_code.items(), key=lambda item: item[1])
    )

    return ordered_codes, month_by_code


def add_months(value, months):
    month_index = value.year * 12 + value.month - 1 + months
    return datetime(month_index // 12, month_index % 12 + 1, 1)


def iter_months(start_date, end_date):
    current = start_date
    while current <= end_date:
        yield current
        current = add_months(current, 1)


def fetch_contract_schedule(session):
    encoded_symbol = quote(BASE_SYMBOL, safe='')
    url = f'{BASE_URL}/v3/symbols/{encoded_symbol}/contracts'
    response = session.get(url, timeout=REQUEST_TIMEOUT_SECONDS)
    response.raise_for_status()
    data = response.json()

    base_code = data.get('base_code') or continuous_to_base_symbol(BASE_SYMBOL)
    contract_listings = []
    for contract in data.get('contracts', []):
        contract_code = contract.get('code')
        settlement_date = contract.get('settlement_date')
        if not contract_code or not settlement_date:
            continue

        parsed_settlement_date = datetime.strptime(settlement_date, '%Y%m%d')
        contract_listings.append(
            ContractListing(
                month_code=extract_contract_month_code(base_code, contract_code),
                settlement_month=parsed_settlement_date.month,
                settlement_date=parsed_settlement_date,
            )
        )

    if not contract_listings:
        raise ValueError(f'No contracts returned for {BASE_SYMBOL}')

    base_contracts, settlement_months = build_contract_schedule(contract_listings)
    return base_code, base_contracts, settlement_months


def get_contract_months(settlement_months, contract, year):
    settlement_date = datetime(year, settlement_months[contract], 1)
    start_date = add_months(settlement_date, -CONTRACT_LOOKBACK_MONTHS + 1)
    return list(iter_months(start_date, settlement_date))


def iter_archive_start_dates(start_date, bar_type):
    if bar_type == 'second':
        # The archive endpoint serves second bars by day. Iterate every day in the
        # contract month to cover the same monthly range as minute/hour requests.
        current = start_date
        now = datetime.now()
        dates = []
        while current.month == start_date.month and current <= now:
            dates.append(current.strftime('%Y-%m-%d'))
            current += timedelta(days=1)
        return dates

    # Minute and hour archive requests use a monthly YYYY-MM bucket.
    return [start_date.strftime('%Y-%m')]


def timeframe_label(bar_type):
    suffixes = {
        'second': 's',
        'minute': 'm',
        'hour': 'h',
    }
    return f'{BAR_INTERVAL}{suffixes[bar_type]}'


def output_path(base_code, symbol_code, formatted_start_date_str, bar_type):
    return (
        Path('data')
        / base_code
        / symbol_code
        / timeframe_label(bar_type)
        / f'{formatted_start_date_str}.json'
    )


def fetch_and_save(session, base_code, symbol_code, start_date, bar_type=BAR_TYPE):
    archive_start_dates = iter_archive_start_dates(start_date, bar_type)

    for formatted_start_date_str in archive_start_dates:
        fetch_and_save_archive_period(
            session,
            base_code,
            symbol_code,
            formatted_start_date_str,
            bar_type,
        )


def fetch_and_save_archive_period(session, base_code, symbol_code, formatted_start_date_str, bar_type):
    file_path = output_path(base_code, symbol_code, formatted_start_date_str, bar_type)
    if file_path.exists():
        print(f'Already exists: {file_path}')
        return

    result = get_ohlcv(session, symbol_code, formatted_start_date_str, bar_type)
    if result.status == 'skip':
        print(f'Skipping {symbol_code} {formatted_start_date_str}: {result.message}')
        return
    if result.status != 'success' or result.data is None:
        print(f'Failed to fetch {symbol_code} {formatted_start_date_str}: {result.message}')
        return

    series_len = len(result.data.get('series', []))
    print(f'Symbol: {symbol_code} | Date: {formatted_start_date_str} | Bars: {series_len}')
    file_path.parent.mkdir(parents=True, exist_ok=True)
    with open(file_path, 'w') as f:
        json.dump(result.data, f)


def get_ohlcv(session, symbol_code, formatted_start_date_str, bar_type):
    for attempt in range(1, MAX_RETRIES + 1):
        result = fetch_archive_period(session, symbol_code, formatted_start_date_str, bar_type)
        if result.status in ('success', 'skip'):
            return result

        print(f'Retrying {symbol_code} {formatted_start_date_str}: {result.message}')
        time.sleep(attempt * 0.5)

    return FetchResult('retry', message='exhausted retries')


def fetch_archive_period(session, symbol_code, formatted_start_date_str, bar_type):
    try:
        encoded_symbol = quote(symbol_code, safe='')
        url = f'{BASE_URL}/v3/symbols/{encoded_symbol}/history'
        params = {
            **HISTORY_PARAMS,
            'bar_type': bar_type,
            'start_date': formatted_start_date_str,
        }
        response = session.get(url, params=params, timeout=REQUEST_TIMEOUT_SECONDS)
    except requests.RequestException as e:
        return FetchResult('retry', message=str(e))

    if response.status_code == 429 or response.status_code >= 500:
        return FetchResult('retry', message=response.text)

    if response.status_code != 200:
        return FetchResult('skip', message=response.text)

    try:
        data = response.json()
    except ValueError as e:
        return FetchResult('retry', message=f'Invalid JSON response: {e}')

    message = data.get('message') or data.get('error')
    if message:
        message = str(message)
        if 'No data is available' in message:
            return FetchResult('skip', message=message)
        return FetchResult('retry', message=message)

    return FetchResult('success', data=data)


def build_session():
    session = requests.Session()
    session.headers.update({
        'Authorization': f'Bearer {API_KEY}',
        'Content-Type': 'application/json',
    })
    return session


def main():
    session = build_session()
    now = datetime.now()
    current_year = now.year
    if BAR_TYPE not in ('second', 'minute', 'hour'):
        raise ValueError('BAR_TYPE must be second, minute, or hour')
    if not isinstance(BAR_INTERVAL, int) or BAR_INTERVAL < 1:
        raise ValueError('BAR_INTERVAL must be a positive integer')

    base_code, base_contracts, settlement_months = fetch_contract_schedule(session)
    print(f"Discovered contract months for {base_code}: {', '.join(base_contracts)}")

    for year in range(START_YEAR, current_year + 1):
        for contract in base_contracts:
            symbol_code = f'{base_code}{contract}{year}'
            months = get_contract_months(settlement_months, contract, year)

            for start_date in months:
                if start_date > now:
                    continue

                print(
                    f"Fetching {BAR_TYPE} data for {symbol_code} "
                    f"for {start_date.strftime('%Y-%m')}"
                )
                fetch_and_save(session, base_code, symbol_code, start_date)

if __name__ == '__main__':
    main()