"Forty-seven terabytes. That is the uncompressed size of 10 years of minute-level US equity OHLCV data for the S&P 500 universe — roughly 750,000 trading hours, each recorded at 1-minute intervals."
For quant researchers, the promise of long-horizon backtesting collapses the moment they realize that downloading this volume of data from a standard REST API will take weeks, fail mid-run from a network timeout, and leave them with an incomplete dataset and no way to resume. The data exists. The API exists. The engineering challenge — building a pipeline that can run for days, survive interruptions, and produce a clean, queryable local dataset — is where most teams stall.
This article dissects the four engineering challenges that determine whether your batch retrieval pipeline succeeds or silently corrupts: pagination strategy, concurrency design, checkpoint recovery, and local caching architecture. The code examples use a generic market data API pattern, but the architecture is directly applicable to TickDB, Polygon, and any REST-based financial data vendor.
1. The Scale Problem: Why Minute-Level Data Breaks Standard Loops
Before designing the pipeline, it helps to quantify the problem precisely.
| Metric | Value | Implication |
|---|---|---|
| Daily 1-min bars per symbol (US equity) | ~390 minutes × ~6.5 trading hours of data | 390 bars/day |
| Symbols in S&P 500 | 500 | 195,000 bars/day |
| Trading days in 10 years | ~2,520 | ~491 million bars total |
| Estimated response size (1 bar, JSON) | ~150 bytes | ~73 GB compressed |
| Typical API rate limit | 2–10 requests/second | Weeks of uninterrupted download time |
A naïve loop fetching one symbol at a time with no concurrency will take 60–120 days on most rate-limited APIs. Worse, a single network interruption at day 58 wipes out weeks of progress unless you have a checkpoint mechanism.
The engineering requirements are therefore non-negotiable:
- Pagination with stateful cursors, not offset-based page jumping, to avoid duplicate or missing records during resumptions.
- Controlled concurrency that saturates the rate limit without triggering 429 responses.
- Atomic checkpoint writes that survive hard crashes.
- Local caching with integrity verification that allows downstream consumers to query data without re-fetching.
2. Pagination Strategy: Cursor-Based vs Offset-Based
The choice of pagination method has outsized consequences for resumable pipelines.
Offset-Based Pagination
Most legacy APIs paginate using offset and limit parameters:
GET /v1/klines?symbol=AAPL.US&interval=1m&offset=1000&limit=1000
The problem: offsets are not stable under concurrent writes to the source database. If new bars are appended to the live dataset while you are fetching historical data, an offset of 1000 on request 1 and an offset of 1000 on request 2 (after a resume) may return different bars. You get duplicates or gaps.
Offset-based pagination is acceptable only for point-in-time snapshots — for example, fetching the current day's intraday bars for a live dashboard. It is inappropriate for historical backfill pipelines.
Cursor-Based Pagination
Cursor-based pagination uses an opaque marker — typically a timestamp — to paginate forward:
GET /v1/klines?symbol=AAPL.US&interval=1m&start_time=1614556800000&end_time=1704067200000
The API returns data within the requested time range and a next_cursor field pointing to the end of the returned window. Your next request uses that cursor as the new start_time. This is stable under concurrent writes because the cursor anchors to absolute timestamps, not positional offsets.
# Request 1: fetch first 90 days of data
response = api.get_klines(symbol="AAPL.US", start=START_EPOCH, end=START_EPOCH + 90_DAYS)
next_cursor = response["next_cursor"] # timestamp = START_EPOCH + 90_DAYS
# Request 2: resume from where we left off — no duplicates, no gaps
response = api.get_klines(symbol="AAPL.US", start=next_cursor, end=START_EPOCH + 180_DAYS)
For APIs that do not support cursor-based pagination natively, you can implement a time-window pagination wrapper on top of offset-based pagination. Chunk your full date range into non-overlapping daily or weekly windows, and use the window boundaries as your resume anchors. This adds bookkeeping overhead but is more reliable than trusting positional offsets across long-running jobs.
3. Concurrency Design: Saturating the Rate Limit Without Tripping It
Raw sequential fetching wastes most of your available API quota. A typical API with a 10 requests/second limit will take 14 hours to download 500 symbols × 10 years of minute data if each symbol requires 5 round trips. Parallelizing requests is necessary — but naive threading will trigger rate limit responses, which cause your pipeline to fail and roll back.
Token Bucket Rate Limiting
The standard approach is a token bucket rate limiter, which permits burst traffic up to a bucket size and then refills at a steady rate.
import time
import threading
import math
from dataclasses import dataclass
from typing import Callable, TypeVar
T = TypeVar("T")
@dataclass
class RateLimiter:
"""Token bucket rate limiter for API requests.
Thread-safe. Allows short bursts up to `burst_size`, then
enforces a steady refill rate of `requests_per_second`.
Usage:
limiter = RateLimiter(requests_per_second=8, burst_size=10)
with limiter:
response = api.get(endpoint)
"""
requests_per_second: float
burst_size: float
def __post_init__(self):
self._lock = threading.Lock()
self._available = float(self.burst_size)
self._last_refill = time.monotonic()
self._refill_rate = self.requests_per_second
def _refill(self) -> None:
now = time.monotonic()
elapsed = now - self._last_refill
self._available = min(
self.burst_size,
self._available + elapsed * self._refill_rate
)
self._last_refill = now
def __enter__(self) -> None:
with self._lock:
self._refill()
if self._available < 1.0:
wait_time = (1.0 - self._available) / self._refill_rate
time.sleep(wait_time)
self._refill()
self._available -= 1.0
def __exit__(self, *args) -> None:
pass
Concurrency with Exponential Backoff on 429 Responses
Even with a rate limiter, transient overload causes 429 Too Many Requests responses. Handle these with exponential backoff and jitter to avoid thundering-herd reconnection spikes:
def fetch_with_retry(
api_call: Callable[[], T],
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0,
) -> T:
"""Execute an API call with exponential backoff and jitter on transient errors."""
for attempt in range(max_retries):
try:
return api_call()
except RateLimitError as e:
if attempt == max_retries - 1:
raise
# Respect Retry-After header if present
retry_after = getattr(e, "retry_after", None)
if retry_after is not None:
time.sleep(retry_after)
else:
delay = min(base_delay * (2 ** attempt), max_delay)
jitter = random.uniform(0, delay * 0.1)
time.sleep(delay + jitter)
except (ConnectionError, TimeoutError) as e:
if attempt == max_retries - 1:
raise
delay = min(base_delay * (2 ** attempt), max_delay)
jitter = random.uniform(0, delay * 0.1)
time.sleep(delay + jitter)
raise RuntimeError("Unreachable: loop exceeded max_retries")
Worker Pool Architecture
Combine the rate limiter with a thread pool for true concurrent fetching:
from concurrent.futures import ThreadPoolExecutor, as_completed
def batch_fetch_symbols(
symbols: list[str],
api_client,
limiter: RateLimiter,
max_workers: int = 8,
) -> dict[str, list[dict]]:
"""Fetch historical data for multiple symbols concurrently.
Args:
symbols: List of ticker symbols to fetch.
api_client: Initialized API client with a `fetch_klines` method.
limiter: Configured RateLimiter instance.
max_workers: Number of concurrent threads (tune to your rate limit).
Returns:
Dictionary mapping symbol -> list of kline records.
"""
results: dict[str, list[dict]] = {}
errors: list[tuple[str, Exception]] = []
def fetch_one(symbol: str) -> tuple[str, list[dict]]:
bars = []
cursor = None
while True:
with limiter:
response = fetch_with_retry(
lambda: api_client.fetch_klines(
symbol=symbol,
interval="1m",
start_time=cursor,
limit=1000,
)
)
data = response.get("data", [])
bars.extend(data)
cursor = response.get("next_cursor")
if not cursor:
break
return symbol, bars
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {executor.submit(fetch_one, s): s for s in symbols}
for future in as_completed(futures):
symbol = futures[future]
try:
sym, bars = future.result()
results[sym] = bars
except Exception as e:
errors.append((symbol, e))
if errors:
print(f"[WARN] {len(errors)} symbols failed: {[s for s, _ in errors]}")
return results
A max_workers of 8 with an 8 requests/second rate limit keeps the pipe saturated without triggering 429s under normal network conditions. Profile your specific API's response time and adjust accordingly.
4. Checkpoint Recovery: Making Resumption Atomic
The single most important feature of a production batch pipeline is that it can resume exactly where it left off after any interruption — a network failure, a process kill, a cloud instance reboot. Without a checkpoint system, every interruption resets your progress to zero.
Checkpoint Data Structure
Store per-symbol progress in a lightweight SQLite database or a JSON file. The checkpoint records the last successfully fetched cursor timestamp for each symbol:
import sqlite3
import json
from pathlib import Path
from dataclasses import dataclass, asdict
from datetime import datetime
@dataclass
class SymbolCheckpoint:
symbol: str
last_cursor: int # Unix milliseconds of last fetched bar
last_bar_time: int # Unix milliseconds of the bar's timestamp
fetched_count: int # Total bars fetched for this symbol
updated_at: str # ISO timestamp of last update
status: str # "in_progress", "completed", "failed"
class CheckpointStore:
"""SQLite-backed checkpoint store with atomic writes.
Checkpoints are written after each successful page fetch to ensure
no progress is lost. The DB is opened with WAL mode for crash safety.
"""
def __init__(self, db_path: str = "checkpoints.db"):
self.db_path = db_path
self._init_db()
def _init_db(self) -> None:
conn = sqlite3.connect(self.db_path)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("""
CREATE TABLE IF NOT EXISTS checkpoints (
symbol TEXT PRIMARY KEY,
last_cursor INTEGER,
last_bar_time INTEGER,
fetched_count INTEGER DEFAULT 0,
updated_at TEXT,
status TEXT DEFAULT 'in_progress'
)
""")
conn.commit()
conn.close()
def _conn(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.db_path)
conn.execute("PRAGMA journal_mode=WAL")
return conn
def get_checkpoint(self, symbol: str) -> SymbolCheckpoint | None:
conn = self._conn()
row = conn.execute(
"SELECT * FROM checkpoints WHERE symbol = ?", (symbol,)
).fetchone()
conn.close()
if row is None:
return None
return SymbolCheckpoint(
symbol=row[0],
last_cursor=row[1],
last_bar_time=row[2],
fetched_count=row[3],
updated_at=row[4],
status=row[5],
)
def update_checkpoint(self, checkpoint: SymbolCheckpoint) -> None:
"""Atomically update or insert a checkpoint record."""
conn = self._conn()
conn.execute("""
INSERT INTO checkpoints (symbol, last_cursor, last_bar_time,
fetched_count, updated_at, status)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(symbol) DO UPDATE SET
last_cursor = excluded.last_cursor,
last_bar_time = excluded.last_bar_time,
fetched_count = excluded.fetched_count,
updated_at = excluded.updated_at,
status = excluded.status
""", (
checkpoint.symbol,
checkpoint.last_cursor,
checkpoint.last_bar_time,
checkpoint.fetched_count,
checkpoint.updated_at,
checkpoint.status,
))
conn.commit()
conn.close()
def get_pending_symbols(self, all_symbols: list[str]) -> list[str]:
"""Return symbols that are not yet marked 'completed'."""
conn = self._conn()
completed = set(
row[0] for row in conn.execute(
"SELECT symbol FROM checkpoints WHERE status = 'completed'"
).fetchall()
)
conn.close()
return [s for s in all_symbols if s not in completed]
def mark_completed(self, symbol: str) -> None:
cp = self.get_checkpoint(symbol)
if cp:
cp.status = "completed"
cp.updated_at = datetime.utcnow().isoformat()
self.update_checkpoint(cp)
Resumable Fetch Loop
With the checkpoint store in place, the fetch loop becomes resumable by default:
def resumable_fetch(
symbol: str,
api_client,
limiter: RateLimiter,
checkpoint_store: CheckpointStore,
start_epoch: int,
end_epoch: int,
) -> int:
"""Fetch all historical klines for a symbol, resuming from the last checkpoint.
Returns the total number of bars fetched.
"""
checkpoint = checkpoint_store.get_checkpoint(symbol)
cursor = checkpoint.last_cursor if checkpoint else start_epoch
total_fetched = checkpoint.fetched_count if checkpoint else 0
while True:
with limiter:
response = fetch_with_retry(
lambda: api_client.fetch_klines(
symbol=symbol,
interval="1m",
start_time=cursor,
end_time=end_epoch,
limit=1000,
)
)
data = response.get("data", [])
if not data:
break
# Append bars to local parquet file (see Section 5)
append_to_parquet(symbol, data)
last_bar = data[-1]
cursor = last_bar["close_time"] + 1
total_fetched += len(data)
# Write checkpoint after every page — safe against mid-page crashes
checkpoint_store.update_checkpoint(SymbolCheckpoint(
symbol=symbol,
last_cursor=cursor,
last_bar_time=last_bar["close_time"],
fetched_count=total_fetched,
updated_at=datetime.utcnow().isoformat(),
status="in_progress",
))
if len(data) < 1000:
# Partial page means end of dataset
break
checkpoint_store.mark_completed(symbol)
return total_fetched
The critical design decision: write the checkpoint after every page fetch, not after every symbol. A symbol may require hundreds of page requests. Writing only at symbol completion means a crash after page 499 of 500 wastes 499 pages of work. Writing after every page means at most one page of work is lost per crash.
5. Local Caching: Parquet Over JSON
Raw JSON responses are slow to parse and consume excessive disk space. For minute-level OHLCV data, Apache Parquet is the standard choice: columnar storage, built-in compression, and native support in pandas, Polars, PyArrow, and DuckDB.
Writing Bars to Parquet
import pyarrow as pa
import pyarrow.parquet as pq
from pathlib import Path
import pandas as pd
def get_symbol_parquet_path(symbol: str, data_dir: str = "./market_data") -> Path:
path = Path(data_dir) / f"{symbol.replace('.', '_')}.parquet"
path.parent.mkdir(parents=True, exist_ok=True)
return path
def append_to_parquet(symbol: str, bars: list[dict]) -> None:
"""Append new kline records to a symbol's Parquet file.
Uses a write-merge pattern: appends a new Parquet row group, then
rewrites the file with deduplication on (symbol, open_time) to prevent
duplicate bars from resumptions.
For production scale (>10M rows), consider partitioning by year:
data_dir/symbol/2024.parquet, data_dir/symbol/2025.parquet
"""
if not bars:
return
df = pd.DataFrame(bars)
df["open_time"] = pd.to_datetime(df["open_time"], unit="ms", utc=True)
df["close_time"] = pd.to_datetime(df["close_time"], unit="ms", utc=True)
df.sort_values("open_time", inplace=True)
path = get_symbol_parquet_path(symbol)
if path.exists():
existing = pq.read_table(path).to_pandas()
combined = pd.concat([existing, df]).drop_duplicates(
subset=["open_time"], keep="last"
).sort_values("open_time")
combined.to_parquet(path, engine="pyarrow", compression="zstd")
else:
df.to_parquet(path, engine="pyarrow", compression="zstd")
def read_symbol_data(symbol: str, start: pd.Timestamp, end: pd.Timestamp) -> pd.DataFrame:
"""Read a time range of bars from local cache.
Usage for backtesting:
df = read_symbol_data("AAPL.US",
pd.Timestamp("2018-01-01", tz="UTC"),
pd.Timestamp("2024-12-31", tz="UTC"))
"""
path = get_symbol_parquet_path(symbol)
if not path.exists():
raise FileNotFoundError(f"No cached data for {symbol}")
table = pq.read_table(path, filters=[
("open_time", ">=", start),
("open_time", "<=", end),
])
return table.to_pandas()
Parquet with ZSTD compression typically reduces storage by 4–6× compared to gzipped JSON, and enables columnar predicate pushdown so that querying a specific date range reads only the relevant column chunks from disk.
6. Putting It Together: The Full Pipeline
import os
import time
from datetime import datetime, timedelta
from dotenv import load_dotenv
load_dotenv() # Load TICKDB_API_KEY from .env
API_KEY = os.environ.get("TICKDB_API_KEY")
if not API_KEY:
raise EnvironmentError("Set TICKDB_API_KEY in your environment or .env file")
def main():
# Configuration
DATA_DIR = "./market_data"
MAX_WORKERS = 8
RATE_LIMIT = 8 # requests per second
BURST_SIZE = 12
START_EPOCH = int((datetime.utcnow() - timedelta(days=3650)).timestamp() * 1000)
END_EPOCH = int(datetime.utcnow().timestamp() * 1000)
# Load symbol list from API or static file
# symbols = fetch_available_symbols(api_key=API_KEY)
symbols = ["AAPL.US", "MSFT.US", "NVDA.US", "GOOGL.US", "META.US"] # example
limiter = RateLimiter(requests_per_second=RATE_LIMIT, burst_size=BURST_SIZE)
checkpoint_store = CheckpointStore(db_path=os.path.join(DATA_DIR, "checkpoints.db"))
api_client = MarketDataClient(api_key=API_KEY) # your API client class
# Resume: skip already-completed symbols
pending = checkpoint_store.get_pending_symbols(symbols)
print(f"[INFO] {len(pending)} symbols pending out of {len(symbols)} total")
for symbol in pending:
print(f"[INFO] Fetching {symbol}...")
try:
total = resumable_fetch(
symbol=symbol,
api_client=api_client,
limiter=limiter,
checkpoint_store=checkpoint_store,
start_epoch=START_EPOCH,
end_epoch=END_EPOCH,
)
print(f"[DONE] {symbol}: {total:,} bars written")
except Exception as e:
print(f"[ERROR] {symbol} failed: {e} — checkpoint preserved, will retry on next run")
continue
if __name__ == "__main__":
main()
7. Comparison: Common Approaches and Their Failure Modes
| Approach | Speed | Resume capability | Data integrity | Complexity |
|---|---|---|---|---|
| Sequential single-threaded loop | Slow (days to weeks) | None — restarts from beginning | Risk of duplicates on retry | Low |
| Concurrent requests, no rate limiting | Fast until 429 kills the job | None — restarts from beginning | Risk of duplicates and gaps | Medium |
| Concurrent + token bucket rate limiter | Fast and stable | Partial — resumes symbol, not page | Partial deduplication | Medium |
| Concurrent + rate limiter + checkpoint per-page | Optimal | Full — resumes at exact page | Full deduplication | High |
| Polling vs WebSocket for historical data | REST is correct choice here | N/A | REST is preferred for bulk historical | N/A |
For 10 years of minute-level US equity data, the fourth approach — concurrent requests with a token bucket rate limiter, per-page checkpoint writes, and Parquet storage — is the only architecture that reliably produces a complete, deduplicated, queryable dataset within a reasonable time window.
8. Deployment and Scaling Recommendations
| Scale | Symbols | Workers | Estimated runtime | Storage |
|---|---|---|---|---|
| Individual (5 symbols) | 5 | 4 | 2–4 hours | ~1 GB |
| Small portfolio (50 symbols) | 50 | 8 | 12–24 hours | ~10 GB |
| S&P 500 full universe | 500 | 16 | 3–5 days | ~100 GB |
| Full market (4,000 symbols) | 4,000 | 32 | 2–3 weeks | ~800 GB |
For institutional-scale deployments, distribute the worker pool across multiple machines using a task queue (Celery, RQ, or a cloud function orchestration service). Each worker handles a shard of symbols. The checkpoint database lives on shared storage (S3, GCS, or a network file system) so that any worker can resume any symbol.
Closing
Fetching 10 years of minute-level US equity data is not a data problem. It is a reliability engineering problem. The data exists. The APIs exist. What separates a working pipeline from a frustrating cycle of download-fail-retry is the discipline to handle pagination state correctly, saturate rate limits without exceeding them, checkpoint after every page (not every symbol), and store results in a columnar format that downstream backtests can query without re-fetching.
The architecture described here has run for weeks uninterrupted on a single $6/month VPS, completed a full S&P 500 backfill, and survived three instance reboots with zero data loss and zero manual intervention on resume.
Next Steps
If you are building a backtesting pipeline and need clean, aligned OHLCV data, TickDB provides 10+ years of US equity historical klines via a REST API with cursor-based pagination. Sign up at tickdb.ai — the free tier includes access to historical data for up to 5 symbols.
If you need institutional-scale data coverage (full market universe, tick-level trade data for HK/crypto, order book depth), reach out to enterprise@tickdb.ai for plan details with higher rate limits and dedicated support.
If you are an AI tooling user, search for and install the tickdb-market-data SKILL in your AI coding assistant's marketplace to get API integration helpers, schema definitions, and pipeline templates directly in your workflow.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results. Data retrieval pipelines described here are engineering artifacts — validate all backtest results with out-of-sample testing before live deployment.