Every quant researcher eventually hits the same wall.
You are designing a mean-reversion strategy. You need 10 years of minute-level bars for S&P 500 components. You write a simple script to fetch data from your market data provider. It works fine for a single symbol. You add 50 symbols. The script runs for 20 minutes and then crashes on a network timeout. You restart it. It starts from the beginning. You lose six hours.
This is not a hypothetical. This is the documented experience of nearly every systematic trading team that has attempted to build a historical data pipeline from scratch. The problem is not fetching data. The problem is fetching data at scale with resilience, correctness, and efficiency.
This article walks through the engineering of a production-grade batch retrieval system. We cover pagination strategies that avoid API limits, concurrent request architectures that maximize throughput without triggering rate limits, a checkpoint-and-recovery mechanism that survives crashes, and a local cache design that eliminates redundant network calls. We use Python throughout, with concrete examples against the TickDB API and analogous patterns for Polygon.io.
The target outcome is a system that can reliably pull 10 years of minute-level OHLCV data for 500 symbols — roughly 2.3 billion rows — without manual intervention, with resumable progress, and with clean separation between network failures and data corruption.
1. The Scale Problem: Why Minute-Level Data Breaks Simple Scripts
Before designing the solution, we need to understand the magnitude of the problem.
Ten years of 1-minute bars for a single US equity symbol generates approximately 2,556 trading days × 390 minutes per day = 997,000 data points. Each data point contains timestamp, open, high, low, close, and volume — six fields, roughly 80–120 bytes per row in JSON.
For 500 symbols, this becomes approximately 498 million rows. Uncompressed JSON over the wire represents roughly 50–80 GB of data transfer. Even with compressed responses and efficient binary formats, a single-threaded sequential fetch strategy runs into three compounding problems:
Problem 1: Request size limits. Most market data APIs impose a maximum number of bars per request. Polygon.io's consolidated endpoint returns at most 50,000 results per call. TickDB's /v1/market/kline endpoint supports a limit parameter with a practical maximum that varies by plan. A naive single-request approach fails immediately.
Problem 2: Rate limiting. Providers enforce request frequency limits. Polygon.io's tiered plans allow 2–5 requests per minute on historical endpoints for standard plans. TickDB enforces a 3001 error code when rate limits are exceeded, with a Retry-After header specifying the required wait time. A naive script that sends 500 sequential requests will hit the rate limit within seconds and spend most of its time waiting or failing.
Problem 3: Network unreliability. Over a 50–80 GB data transfer, the probability of at least one network failure approaches certainty. A 99.9% success rate per request means that out of 10,000 requests, 10 will fail. With a sequential strategy, any single failure aborts the entire batch unless the system is designed to recover.
The solution is not a better network connection. The solution is a system architecture that treats failure as a first-class concern.
2. Pagination Strategies: Chunking the Time Axis
The foundation of any batch retrieval system is a pagination strategy that decomposes a large time range into manageable chunks.
2.1 Time-Based Windowing
The most reliable pagination approach for time-series market data is time-based windowing: divide the full time range into contiguous, non-overlapping intervals, then fetch each interval as an independent request.
[2014-01-01 ──────────────────────────────── 2024-01-01]
↓ ↓ ↓ ↓
window_1 window_2 window_3 window_4 ...
For minute-level US equity data, a window size of 90 calendar days strikes a practical balance:
- At 390 minutes per trading day, 90 days yields approximately 35,100 bars — safely within most API limits
- Window boundaries align naturally with quarter boundaries, simplifying manual auditing
- The number of windows per symbol is predictable: (3650 days / 90) ≈ 41 windows
from datetime import datetime, timedelta
from typing import Generator, Tuple
def generate_time_windows(
start_date: datetime,
end_date: datetime,
window_days: int = 90
) -> Generator[Tuple[datetime, datetime], None, None]:
"""
Generate non-overlapping time windows from start_date to end_date.
Args:
start_date: Inclusive start of the data range
end_date: Exclusive end of the data range
window_days: Number of calendar days per window
Yields:
Tuples of (window_start, window_end) as datetime objects
"""
current = start_date
window_delta = timedelta(days=window_days)
while current < end_date:
window_end = min(current + window_delta, end_date)
yield (current, window_end)
current = window_end
2.2 Cursor-Based Pagination (When Available)
Some APIs support cursor-based pagination, where the server returns an opaque cursor with each response that points to the next page. This approach is superior to time-based windowing when available because:
- It handles variable-density data naturally (some days have more minute bars due to extended-hours activity)
- It avoids double-fetching at window boundaries
- It shifts pagination complexity to the server
def fetch_with_cursor_pagination(
api_key: str,
symbol: str,
start_ts: int, # Unix milliseconds
end_ts: int,
page_size: int = 50000
) -> list[dict]:
"""
Fetch data using cursor-based pagination.
Handles rate limiting with exponential backoff.
"""
import time
import requests
import random
base_url = "https://api.tickdb.ai/v1/market/kline"
headers = {"X-API-Key": api_key}
all_data = []
cursor = None
max_retries = 5
while True:
params = {
"symbol": symbol,
"interval": "1m",
"start": start_ts,
"end": end_ts,
"limit": page_size
}
if cursor:
params["cursor"] = cursor
for attempt in range(max_retries):
try:
response = requests.get(
base_url,
headers=headers,
params=params,
timeout=(3.05, 30)
)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"Rate limited. Waiting {retry_after}s...")
time.sleep(retry_after)
continue
result = response.json()
code = result.get("code", 0)
if code == 3001:
retry_after = int(response.headers.get("Retry-After", 5))
time.sleep(retry_after)
continue
if code != 0:
raise RuntimeError(f"API error {code}: {result.get('message')}")
data = result.get("data", [])
all_data.extend(data)
cursor = result.get("next_cursor")
if not cursor or not data:
return all_data
# Jitter to avoid thundering herd
time.sleep(random.uniform(0.05, 0.2))
break
except requests.exceptions.Timeout:
delay = min(2 ** attempt * 1.0 + random.uniform(0, 1), 60)
print(f"Timeout. Retrying in {delay:.1f}s...")
time.sleep(delay)
return all_data
2.3 Handling Symbol Lists Efficiently
When fetching data for multiple symbols (e.g., all S&P 500 components), the naive approach is to process symbols sequentially. This is safe but slow. A better approach is to interleave symbol requests with a controlled concurrency model, which we address in the next section.
For now, the key principle is that each symbol's time range should be divided into windows independently, and windows should be tracked separately in the checkpoint system.
3. Concurrent Request Architecture: Parallelism Within Rate Limits
Raw sequential fetching for 500 symbols × 41 windows = 20,500 requests. At a 1 request per second rate limit, this takes 5.7 hours — acceptable for a one-time backfill, but wasteful if you run this daily. A concurrent architecture can reduce this to 30–60 minutes while staying within rate limits.
3.1 The Token Bucket Pattern
The most effective rate-limit-aware concurrency model is the token bucket algorithm: maintain a pool of N tokens, where N is the maximum number of concurrent requests permitted. Each request consumes one token. When a request completes, the token is returned. When rate limits are hit, the bucket is temporarily emptied until the cooldown period expires.
import asyncio
import time
import random
from dataclasses import dataclass, field
from typing import Optional
from collections import deque
@dataclass
class TokenBucket:
"""
Token bucket for rate-limit-aware concurrency.
Maintains a pool of permits. Acquiring a permit blocks if none are available.
Rate limit resets are handled by tracking the cooldown_until timestamp.
"""
capacity: int
refill_rate: float # permits per second
cooldown_duration: float = 60.0 # seconds to wait after 429/3001
_tokens: float = field(init=False)
_last_refill: float = field(init=False)
_cooldown_until: float = field(init=False)
_lock: asyncio.Lock = field(init=False)
_waiters: deque = field(init=False)
def __post_init__(self):
self._tokens = float(self.capacity)
self._last_refill = time.monotonic()
self._cooldown_until = 0.0
self._lock = asyncio.Lock()
self._waiters = deque()
def _refill(self):
now = time.monotonic()
elapsed = now - self._last_refill
self._tokens = min(self.capacity, self._tokens + elapsed * self.refill_rate)
self._last_refill = now
async def acquire(self) -> None:
async with self._lock:
while True:
now = time.monotonic()
# If in cooldown, wait until cooldown expires
if now < self._cooldown_until:
wait_time = self._cooldown_until - now
await asyncio.sleep(wait_time)
continue
self._refill()
if self._tokens >= 1:
self._tokens -= 1
return
# Wait for a token to become available
wait_time = (1 - self._tokens) / self.refill_rate
await asyncio.sleep(wait_time)
def release(self) -> None:
self._tokens = min(self.capacity, self._tokens + 1)
def enter_cooldown(self, duration: Optional[float] = None) -> None:
duration = duration or self.cooldown_duration
self._cooldown_until = time.monotonic() + duration
self._tokens = 0
3.2 Worker Pool Architecture
With the token bucket in place, we implement a worker pool that processes windows from a shared queue. Workers pull windows from the queue, acquire a token, execute the request, and release the token on completion.
import asyncio
from typing import List, Callable, Any
from dataclasses import dataclass
import logging
logger = logging.getLogger(__name__)
@dataclass
class FetchWindow:
"""A single data fetch task."""
symbol: str
start_ms: int
end_ms: int
priority: int = 0
@dataclass
class FetchResult:
"""Result of a fetch operation."""
window: FetchWindow
success: bool
data: List[dict] = None
error: str = None
rows_fetched: int = 0
class WorkerPool:
"""
Manages concurrent workers that fetch data windows with rate-limit awareness.
"""
def __init__(
self,
worker_count: int,
rate_limit: float, # requests per second
fetch_func: Callable[[FetchWindow], FetchResult]
):
self.worker_count = worker_count
self.bucket = TokenBucket(
capacity=worker_count,
refill_rate=rate_limit
)
self.fetch_func = fetch_func
self.queue: asyncio.Queue = None
self.results: List[FetchResult] = []
self._shutdown = False
async def _worker(self, worker_id: int) -> None:
"""Individual worker coroutine."""
while not self._shutdown:
try:
window: FetchWindow = await asyncio.wait_for(
self.queue.get(),
timeout=1.0
)
except asyncio.TimeoutError:
continue
await self.bucket.acquire()
try:
result = await asyncio.to_thread(self.fetch_func, window)
self.results.append(result)
if result.success:
logger.info(
f"Worker {worker_id}: {window.symbol} "
f"{window.start_ms}-{window.end_ms} "
f"→ {result.rows_fetched} rows"
)
else:
logger.warning(
f"Worker {worker_id}: {window.symbol} failed: {result.error}"
)
# Re-queue for retry with exponential backoff
window.priority += 1
if window.priority < 5:
await self.queue.put(window)
except Exception as e:
logger.error(f"Worker {worker_id} exception: {e}")
finally:
self.bucket.release()
self.queue.task_done()
async def run(self, windows: List[FetchWindow]) -> List[FetchResult]:
"""
Run the worker pool until all windows are processed.
"""
self.queue = asyncio.Queue()
self.results = []
self._shutdown = False
# Enqueue all windows
for window in sorted(windows, key=lambda w: w.priority, reverse=True):
await self.queue.put(window)
# Start workers
workers = [
asyncio.create_task(self._worker(i))
for i in range(self.worker_count)
]
# Wait for queue to drain
await self.queue.join()
self._shutdown = True
# Cancel workers
for w in workers:
w.cancel()
await asyncio.gather(*workers, return_exceptions=True)
return self.results
3.3 Recommended Concurrency Settings
| Plan tier | Concurrency | Rate limit | 500 symbols × 41 windows |
|---|---|---|---|
| Free | 1 | 1 req/min | ~340 hours |
| Standard | 3 | 5 req/min | ~23 hours |
| Professional | 10 | 30 req/min | ~4 hours |
| Enterprise | 30 | 100 req/min | ~1.2 hours |
For production backfills, use the highest plan tier available. For ongoing daily updates, a standard plan with 3 workers is sufficient.
4. Checkpoint-and-Recovery: Designing Resumable Downloads
The checkpoint system is the component that transforms a fragile script into a resilient pipeline. The principle is simple: persist the state of every completed window to durable storage, and on restart, skip any window that is already persisted.
4.1 Checkpoint Data Model
Each window has three possible states:
- Pending: Not yet fetched
- In progress: Fetching (with a heartbeat timestamp)
- Complete: Data persisted to local cache
The checkpoint record for each window should include:
from dataclasses import dataclass, asdict
from datetime import datetime
from enum import Enum
import json
import sqlite3
from pathlib import Path
class WindowState(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETE = "complete"
FAILED = "failed"
@dataclass
class WindowCheckpoint:
symbol: str
start_ms: int
end_ms: int
state: WindowState
rows_fetched: int
checksum: str # SHA-256 of sorted JSON rows
updated_at: str # ISO 8601
retry_count: int = 0
def to_dict(self) -> dict:
d = asdict(self)
d["state"] = self.state.value
return d
@classmethod
def from_dict(cls, d: dict) -> "WindowCheckpoint":
d["state"] = WindowState(d["state"])
return cls(**d)
4.2 SQLite-Based Checkpoint Store
SQLite is the recommended storage backend for the checkpoint database. It is ACID-compliant, supports concurrent reads from multiple processes, and handles crash recovery correctly without external dependencies.
class CheckpointStore:
"""
SQLite-backed checkpoint store for resumable data fetching.
Schema:
checkpoints (
symbol TEXT,
start_ms INTEGER,
end_ms INTEGER,
state TEXT,
rows_fetched INTEGER,
checksum TEXT,
updated_at TEXT,
retry_count INTEGER,
PRIMARY KEY (symbol, start_ms)
)
"""
def __init__(self, db_path: str = "checkpoints.db"):
self.db_path = db_path
self._init_db()
def _init_db(self) -> None:
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS checkpoints (
symbol TEXT NOT NULL,
start_ms INTEGER NOT NULL,
end_ms INTEGER NOT NULL,
state TEXT NOT NULL DEFAULT 'pending',
rows_fetched INTEGER NOT NULL DEFAULT 0,
checksum TEXT,
updated_at TEXT NOT NULL,
retry_count INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (symbol, start_ms)
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_state
ON checkpoints(symbol, state)
""")
def get_pending_windows(
self,
symbols: List[str] = None,
limit: int = None
) -> List[WindowCheckpoint]:
"""Retrieve pending windows, optionally filtered by symbols."""
with sqlite3.connect(self.db_path) as conn:
conn.row_factory = sqlite3.Row
query = "SELECT * FROM checkpoints WHERE state = 'pending'"
params = []
if symbols:
placeholders = ",".join("?" * len(symbols))
query += f" AND symbol IN ({placeholders})"
params.extend(symbols)
query += " ORDER BY symbol, start_ms"
if limit:
query += f" LIMIT {limit}"
rows = conn.execute(query, params).fetchall()
return [WindowCheckpoint.from_dict(dict(r)) for r in rows]
def mark_in_progress(self, window: FetchWindow) -> None:
"""Mark a window as in-progress with a heartbeat timestamp."""
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
INSERT OR REPLACE INTO checkpoints
(symbol, start_ms, end_ms, state, rows_fetched, updated_at)
VALUES (?, ?, ?, 'in_progress', 0, ?)
""", (
window.symbol,
window.start_ms,
window.end_ms,
datetime.utcnow().isoformat()
))
def mark_complete(
self,
symbol: str,
start_ms: int,
rows_fetched: int,
checksum: str
) -> None:
"""Mark a window as complete."""
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
UPDATE checkpoints
SET state = 'complete',
rows_fetched = ?,
checksum = ?,
updated_at = ?
WHERE symbol = ? AND start_ms = ?
""", (
rows_fetched,
checksum,
datetime.utcnow().isoformat(),
symbol,
start_ms
))
def mark_failed(
self,
symbol: str,
start_ms: int,
error: str
) -> None:
"""Mark a window as failed after exhausting retries."""
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
UPDATE checkpoints
SET state = 'failed',
retry_count = retry_count + 1,
updated_at = ?
WHERE symbol = ? AND start_ms = ?
""", (
datetime.utcnow().isoformat(),
symbol,
start_ms
))
def reset_stale_in_progress(self, max_age_seconds: int = 300) -> int:
"""
Reset windows stuck in 'in_progress' state for too long.
Returns the count of windows reset.
"""
stale_threshold = (
datetime.utcnow() - timedelta(seconds=max_age_seconds)
).isoformat()
with sqlite3.connect(self.db_path) as conn:
cursor = conn.execute("""
UPDATE checkpoints
SET state = 'pending'
WHERE state = 'in_progress' AND updated_at < ?
""", (stale_threshold,))
return cursor.rowcount
def get_stats(self) -> dict:
"""Return fetch statistics."""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.execute("""
SELECT state, COUNT(*) as count, SUM(rows_fetched) as total_rows
FROM checkpoints
GROUP BY state
""")
rows = cursor.fetchall()
return {r[0]: {"count": r[1], "rows": r[2] or 0} for r in rows}
4.3 Stale Worker Detection
A critical edge case: what happens when a worker crashes mid-fetch? The window is left in in_progress state indefinitely. The reset_stale_in_progress() method addresses this. Call it at startup:
def resume_fetch_pipeline():
"""Initialize the pipeline, recover from any interrupted sessions."""
store = CheckpointStore()
# Recover from stale in-progress windows
reset_count = store.reset_stale_in_progress(max_age_seconds=300)
if reset_count > 0:
print(f"Recovered {reset_count} stale windows from previous run.")
# Get remaining work
pending = store.get_pending_windows()
print(f"Pending windows: {len(pending)}")
# Convert to FetchWindow objects and run
windows = [
FetchWindow(symbol=w.symbol, start_ms=w.start_ms, end_ms=w.end_ms)
for w in pending
]
# Run the worker pool (see Section 3)
# ...
This pattern — mark in-progress before requesting, mark complete after persisting — ensures that a crash at any point results in the window being re-attempted on the next run, never double-fetched.
5. Local Cache Design: Persistence and Query Performance
Data that has been fetched should be stored locally. The goals of the local cache are:
- Eliminate redundant network calls: If the same window is requested again (e.g., a daily incremental update), read from disk.
- Support backtesting queries: The cache should be queryable without re-fetching.
- Provide data integrity verification: The checksum ensures that stored data matches what the API returned.
5.1 Parquet as the Storage Format
For columnar market data, Parquet is the optimal on-disk format. It provides:
- Column pruning: A backtest that only needs
closeprices reads only theclosecolumn, not the full row. - Schema enforcement: Each file has a defined schema, preventing type mismatches.
- Compression: Typical compression ratios of 3–10× versus JSON.
- Splitability: Files can be read in chunks without loading the entire file into memory.
import pyarrow as pa
import pyarrow.parquet as pq
import hashlib
import json
def rows_to_parquet(
rows: List[dict],
symbol: str,
start_ms: int,
end_ms: int,
output_dir: str = "data"
) -> str:
"""
Convert API response rows to a Parquet file.
Returns the SHA-256 checksum of the file for integrity verification.
"""
if not rows:
return ""
# Normalize and sort rows by timestamp
normalized = sorted(rows, key=lambda r: r["t"])
# Create PyArrow table
table = pa.Table.from_pylist(normalized)
# Define output path: data/SYMBOL/START_MS_END_MS.parquet
symbol_dir = Path(output_dir) / symbol
symbol_dir.mkdir(parents=True, exist_ok=True)
file_path = symbol_dir / f"{start_ms}_{end_ms}.parquet"
# Write with ZSTD compression for best balance of speed and ratio
pq.write_table(
table,
str(file_path),
compression="zstd",
use_dictionary=True,
write_statistics=True
)
# Compute checksum of the Parquet file
sha256 = hashlib.sha256()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(65536), b""):
sha256.update(chunk)
return sha256.hexdigest()
def load_parquet_window(
symbol: str,
start_ms: int,
end_ms: int,
columns: List[str] = None,
data_dir: str = "data"
) -> pa.Table:
"""
Load a cached Parquet window, optionally filtering columns.
"""
file_path = Path(data_dir) / symbol / f"{start_ms}_{end_ms}.parquet"
if not file_path.exists():
raise FileNotFoundError(f"Cache miss: {file_path}")
table = pq.read_table(str(file_path))
if columns:
table = table.select(columns)
return table
5.2 Cache Query API for Backtesting
Once data is cached, a simple query API enables backtesting without any network calls:
from datetime import datetime
def query_cache(
symbol: str,
start_date: datetime,
end_date: datetime,
columns: List[str] = None,
data_dir: str = "data"
) -> pa.Table:
"""
Query cached data for a symbol and time range.
Automatically identifies and merges overlapping Parquet files.
"""
start_ms = int(start_date.timestamp() * 1000)
end_ms = int(end_date.timestamp() * 1000)
symbol_dir = Path(data_dir) / symbol
if not symbol_dir.exists():
raise FileNotFoundError(f"No cache for symbol: {symbol}")
# Find all parquet files that intersect with the requested range
matching_files = []
for parquet_file in symbol_dir.glob("*.parquet"):
f_start, f_end = parquet_file.stem.split("_")
f_start, f_end = int(f_start), int(f_end)
# Intersection check
if f_start <= end_ms and f_end >= start_ms:
matching_files.append(parquet_file)
if not matching_files:
raise FileNotFoundError(
f"No cached data for {symbol} in range "
f"{start_ms}-{end_ms}"
)
# Read and concatenate
tables = []
for f in sorted(matching_files):
t = pq.read_table(str(f))
# Filter to requested time range
timestamps = t.column("t").to_pylist()
mask = [start_ms <= ts <= end_ms for ts in timestamps]
filtered = t.filter(pa.array(mask))
tables.append(filtered)
return pa.concat_tables(tables)
6. Complete Pipeline: Integrating All Components
The following script integrates the time window generator, the checkpoint store, the worker pool, and the Parquet cache into a single, production-ready pipeline.
import os
import asyncio
import logging
from datetime import datetime
from pathlib import Path
# Local modules (from previous sections)
from window_generator import generate_time_windows
from checkpoint_store import CheckpointStore, WindowState
from worker_pool import WorkerPool, FetchWindow
from parquet_cache import rows_to_parquet, query_cache
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)
API_KEY = os.environ.get("TICKDB_API_KEY")
if not API_KEY:
raise EnvironmentError("Set TICKDB_API_KEY environment variable.")
def build_fetch_func(checkpoint_store: CheckpointStore):
"""Factory for the fetch function used by the worker pool."""
def fetch_window(window: FetchWindow) -> FetchResult:
from fetch_with_cursor_pagination import fetch_with_cursor_pagination
checkpoint_store.mark_in_progress(window)
try:
rows = fetch_with_cursor_pagination(
api_key=API_KEY,
symbol=window.symbol,
start_ts=window.start_ms,
end_ts=window.end_ms
)
if not rows:
return FetchResult(
window=window,
success=True,
data=[],
rows_fetched=0
)
# Persist to Parquet
checksum = rows_to_parquet(
rows=rows,
symbol=window.symbol,
start_ms=window.start_ms,
end_ms=window.end_ms
)
# Update checkpoint
checkpoint_store.mark_complete(
symbol=window.symbol,
start_ms=window.start_ms,
rows_fetched=len(rows),
checksum=checksum
)
return FetchResult(
window=window,
success=True,
data=rows,
rows_fetched=len(rows)
)
except Exception as e:
logger.error(f"Fetch failed for {window.symbol}: {e}")
checkpoint_store.mark_failed(window.symbol, window.start_ms, str(e))
return FetchResult(window=window, success=False, error=str(e))
return fetch_window
async def main():
symbols = ["AAPL.US", "MSFT.US", "GOOGL.US"] # Expand to full universe
start_date = datetime(2014, 1, 1)
end_date = datetime(2024, 1, 1)
store = CheckpointStore()
# Recover from any interrupted runs
reset_count = store.reset_stale_in_progress(max_age_seconds=300)
if reset_count:
logger.info(f"Recovered {reset_count} stale windows")
# Initialize pending windows for all symbols
all_windows = []
for symbol in symbols:
for w_start, w_end in generate_time_windows(start_date, end_date):
start_ms = int(w_start.timestamp() * 1000)
end_ms = int(w_end.timestamp() * 1000)
all_windows.append(FetchWindow(
symbol=symbol,
start_ms=start_ms,
end_ms=end_ms
))
# Filter to only pending windows (skip already-complete)
pending = store.get_pending_windows()
pending_keys = {(w.symbol, w.start_ms) for w in pending}
windows_to_fetch = [
w for w in all_windows
if (w.symbol, w.start_ms) in pending_keys
]
if not windows_to_fetch:
logger.info("All windows already fetched. Nothing to do.")
return
logger.info(f"Fetching {len(windows_to_fetch)} windows...")
# Run worker pool with 5 concurrent workers, 5 req/min rate limit
pool = WorkerPool(
worker_count=5,
rate_limit=5 / 60, # 5 requests per minute
fetch_func=build_fetch_func(store)
)
results = await pool.run(windows_to_fetch)
# Report
stats = store.get_stats()
logger.info(f"Pipeline complete. Stats: {stats}")
if __name__ == "__main__":
asyncio.run(main())
7. TickDB-Specific Considerations
For teams using TickDB as their primary data source, the following specifics apply:
Endpoint: Use GET /v1/market/kline with the following parameters:
| Parameter | Value | Notes |
|---|---|---|
symbol |
e.g., AAPL.US |
Full market suffix required |
interval |
1m |
Minute-level bars |
start |
Unix milliseconds | Inclusive |
end |
Unix milliseconds | Exclusive |
limit |
50000 |
Maximum bars per request |
adjustment |
split |
Recommended for backtesting |
Authentication: Pass the API key via the X-API-Key header. Never include the key in URL parameters for REST requests.
Rate limit response: When you receive code: 3001, read the Retry-After header and wait the specified duration before retrying. Do not implement aggressive retry loops — each failed retry wastes a request slot.
Data availability: TickDB provides 10+ years of cleaned, aligned US equity OHLCV data suitable for cross-cycle backtesting. Note that the trades endpoint does not cover US equities — use the kline endpoint for historical OHLCV retrieval.
8. Summary: The Five Principles of Resilient Data Retrieval
A production-grade historical data pipeline rests on five principles:
| Principle | Implementation | Failure mode it prevents |
|---|---|---|
| Decompose by time | 90-day windows per symbol | API request size limits |
| Limit concurrency | Token bucket with provider rate limits | 429 errors, API key suspension |
| Checkpoint every step | SQLite store: in_progress → complete | Double-fetching after crash |
| Reset stale workers | updated_at heartbeat check on startup |
Dead workers blocking the queue |
| Cache in Parquet | Columnar format with checksums | Redundant network calls, data corruption |
Applied together, these five principles produce a system that can pull 10 years of minute-level US equity data for hundreds of symbols, survive network failures, recover from crashes mid-fetch, and serve backtesting queries directly from local storage — without manual intervention.
Next Steps
If you're an individual quant researcher building your first data pipeline: Start with the sequential version of the window fetcher and checkpoint store. Add concurrency only after the basic pipeline runs successfully end-to-end for a single symbol.
If you're a team standardizing historical data infrastructure: Deploy the full pipeline with the worker pool, Parquet cache, and query API. Establish a data quality monitoring job that verifies checksums weekly and flags any corruption.
If you need 10+ years of clean, aligned US equity OHLCV data for strategy backtesting: TickDB provides the data via a simple REST API with Python client support. Sign up at tickdb.ai with no credit card required for the free tier, and generate an API key in the dashboard.
If you're building AI-assisted trading tools: Search for and install the tickdb-market-data SKILL in your AI coding assistant's marketplace. The skill provides pre-built templates for the pipeline described in this article.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results.