InsightSentry - Real-time Financial Data API

1. Overview

This guide demonstrates how to fetch historical data for Futures contracts. The example below uses NASDAQ Futures (NQ) to illustrate how to handle contract month logic, date ranges, and pagination.

Note that continuous futures do not support more than 1 year of historical data. Therefore, for extensive Futures history, you must use specific contracts. This guide explains how to fetch data for each individual contract.

Tip: For most futures backfills, use the insight CLI first. download_history can expand continuous futures such as CME_MINI:NQ1! into specific contracts and write local JSON/CSV files. See the Insight Tool guide for setup.

BASH

insight download_history --symbol "CME_MINI:NQ1!" --bar_type hour \
  --from 2025-01 --to 2025-06 --output_dir ./futures --format both

Important Caution

Currently, concurrent requests are not encouraged and may fail. Please run requests sequentially as demonstrated in this script. This code is for demonstration purposes and provides a basic implementation.

2. Implementation

The following Python script demonstrates how to iterate through contract months and fetch second, minute, or hour data. Set BASE_SYMBOL to the continuous futures code you use, such as COMEX:GC1!. The script reads the product-specific contract month letters and settlement months from the contracts endpoint, then projects that schedule across historical years because the endpoint returns current and future contracts only. By default, each generated contract fetches the 6 monthly buckets ending in its settlement month.

For second-level bars, the archive endpoint is date-based, so the script automatically expands each contract month into daily requests to cover the whole month. The script will create a data directory and save JSON files organized by symbol, bar type, interval, and date. You can also store the data in SQL database or in different format.

To change intervals, history depth, or archive options, edit BAR_INTERVAL, CONTRACT_LOOKBACK_MONTHS, and HISTORY_PARAMS in the script. The request logic fills bar_type and start_date automatically for each archive period.

PYTHON

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()

Historical Data for Futures

1. Overview

Important Caution

2. Implementation