The market closes at 4:00 PM ET. By 4:01, you're already running your end-of-day analysis — except you're waiting for a colleague to upload a CSV, or your data vendor's batch job to finish, or some manual step that introduces friction into every single trading day.
Sound familiar?
The most reliable market data pipeline is one you control completely. This article walks through building an automated daily archiving system that pulls minute-level OHLCV data from TickDB's historical API, stores it in a local SQLite or ClickHouse database, and handles incremental updates without re-downloading existing records. Every component is production-grade: retry logic, rate-limit handling, idempotent writes, and schema evolution support.
The architecture works for US equities, HK stocks, or crypto — wherever you need a clean, queryable local copy of historical candles.
Why Build a Local Market Database?
WebSocket streams are excellent for real-time decision-making, but they carry a fundamental limitation: data exists only while the connection is live. Historical analysis, backtesting, and cross-asset correlation studies require persistent storage.
Commercial data vendors solve this, but with tradeoffs: export limits, API rate caps on free tiers, or lock-in to proprietary query languages. A self-hosted market database gives you:
- Full query control: Run arbitrary SQL against your data without API call limits.
- Backtesting without latency: Local queries are measured in milliseconds, not seconds.
- Cross-symbol analysis: Join data from multiple exchanges in a single query.
- Auditability: You own the exact data your models trained on.
The cost is operational complexity — which this article eliminates by automating the entire ingestion pipeline.
Architecture Overview
The system consists of four components:
┌─────────────────────────────────────────────────────────────────┐
│ Daily Archiving Pipeline │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Scheduler │───▶│ Fetcher │───▶│ Deduplicator │ │
│ │ (cron/dag) │ │ (REST API) │ │ (checksum) │ │
│ └──────────────┘ └──────────────┘ └───────┬──────┘ │
│ │ │
│ ┌──────────────┐ ┌──────────────┐ │ │
│ │ Analytics │◀───│ Query Layer │◀───────────┘ │
│ │ /Backtests │ │ (SQLite/CH) │ │
│ └──────────────┘ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
- Scheduler: Triggers the pipeline daily (or on-demand). Use cron, Apache Airflow, or a simple systemd timer.
- Fetcher: Calls the TickDB
/v1/market/klineendpoint for each symbol, requesting the previous trading day's data. - Deduplicator: Computes a checksum per candle; skips records already present in the database.
- Query Layer: Exposes the archived data for analysis, backtesting, and reporting.
The critical design decision is the deduplication strategy — which we address before writing a single line of code.
Database Schema Design
The schema must handle two requirements that are easy to overlook:
- Idempotency: Re-running the pipeline for the same date must not create duplicate records.
- Schema evolution: New fields (volume, turnover, bid-ask midpoint) should be addable without migration scripts.
SQLite Schema
For individual traders or small teams, SQLite is the right choice — zero configuration, file-based portability, and sufficient performance for daily batch ingestion of up to 10,000 symbols.
CREATE TABLE IF NOT EXISTS ohlcv (
id INTEGER PRIMARY KEY AUTOINCREMENT,
symbol TEXT NOT NULL,
interval TEXT NOT NULL DEFAULT '1m',
timestamp INTEGER NOT NULL, -- Unix milliseconds
open REAL NOT NULL,
high REAL NOT NULL,
low REAL NOT NULL,
close REAL NOT NULL,
volume REAL,
turnover REAL, -- Optional: added in v2 schema
checksum TEXT NOT NULL, -- SHA256(symbol + interval + timestamp)
created_at TEXT DEFAULT (datetime('now')),
UNIQUE(symbol, interval, timestamp)
);
CREATE INDEX IF NOT EXISTS idx_symbol_timestamp
ON ohlcv(symbol, interval, timestamp);
The UNIQUE(symbol, interval, timestamp) constraint enforces idempotency at the database level. If the fetcher attempts to insert a duplicate, SQLite raises a UNIQUE constraint error — which we catch and log.
ClickHouse Schema
For institutional workloads or multi-year backtests spanning thousands of symbols, ClickHouse provides columnar storage with order-of-magnitude query speedups.
CREATE TABLE IF NOT EXISTS ohlcv (
symbol String,
interval String DEFAULT '1m',
timestamp DateTime64(3),
open Float64,
high Float64,
low Float64,
close Float64,
volume Float64,
checksum String,
inserted_at DateTime DEFAULT now()
)
ENGINE = ReplacingMergeTree(inserted_at)
ORDER BY (symbol, interval, timestamp);
The ReplacingMergeTree engine deduplicates rows with the same primary key during background merges — a cleaner solution than handling constraint violations.
Production-Grade Fetcher Implementation
The fetcher must handle the realities of production API usage: network timeouts, rate limits, and partial failures. This implementation includes all production-grade requirements from the TickDB Content Strategy Handbook.
#!/usr/bin/env python3
"""
Daily market data fetcher for TickDB historical kline data.
Fetches the previous trading day's 1-minute OHLCV and stores it locally.
"""
import os
import sys
import time
import sqlite3
import hashlib
import logging
from datetime import datetime, timedelta
from typing import Optional, List, Dict, Any
import requests
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)
class TickDBFetcher:
"""
Fetches historical OHLCV data from TickDB and stores it locally.
Implements production-grade patterns: exponential backoff, rate-limit
handling, and idempotent writes.
"""
BASE_URL = "https://api.tickdb.ai/v1/market/kline"
def __init__(self, api_key: Optional[str] = None, db_path: str = "market_data.db"):
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
if not self.api_key:
raise ValueError("API key required: set TICKDB_API_KEY environment variable")
self.db_path = db_path
self._init_database()
def _init_database(self) -> None:
"""Initialize SQLite schema if not present."""
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS ohlcv (
id INTEGER PRIMARY KEY AUTOINCREMENT,
symbol TEXT NOT NULL,
interval TEXT NOT NULL DEFAULT '1m',
timestamp INTEGER NOT NULL,
open REAL NOT NULL,
high REAL NOT NULL,
low REAL NOT NULL,
close REAL NOT NULL,
volume REAL,
turnover REAL,
checksum TEXT NOT NULL,
created_at TEXT DEFAULT (datetime('now')),
UNIQUE(symbol, interval, timestamp)
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_symbol_timestamp
ON ohlcv(symbol, interval, timestamp)
""")
conn.commit()
logger.info(f"Database initialized: {self.db_path}")
def _compute_checksum(self, row: Dict[str, Any]) -> str:
"""Compute deterministic checksum for deduplication."""
key = f"{row['symbol']}|{row.get('interval', '1m')}|{row['timestamp']}"
return hashlib.sha256(key.encode()).hexdigest()[:16]
def _fetch_kline(
self,
symbol: str,
start_time: int,
end_time: int,
interval: str = "1m",
max_retries: int = 5
) -> Optional[Dict]:
"""
Fetch kline data with exponential backoff and rate-limit handling.
Args:
symbol: Trading symbol, e.g., "AAPL.US"
start_time: Start timestamp in Unix milliseconds
end_time: End timestamp in Unix milliseconds
interval: Candle interval (1m, 5m, 1h, 1d)
max_retries: Maximum retry attempts
Returns:
API response JSON or None on failure
"""
params = {
"symbol": symbol,
"interval": interval,
"start": start_time,
"end": end_time,
"limit": 1440 # Max candles per request (1 day of 1m bars)
}
headers = {"X-API-Key": self.api_key}
for attempt in range(max_retries):
try:
response = requests.get(
self.BASE_URL,
headers=headers,
params=params,
timeout=(3.05, 30) # (connect, read) timeout
)
# Handle rate limiting
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
logger.warning(f"Rate limited. Waiting {retry_after}s")
time.sleep(retry_after)
continue
# Handle specific error codes
if response.status_code == 200:
data = response.json()
code = data.get("code", 0)
if code == 0:
return data.get("data")
elif code in (1001, 1002):
raise ValueError("Invalid API key — check TICKDB_API_KEY")
elif code == 2002:
logger.warning(f"Symbol {symbol} not found — skipping")
return None
elif code == 3001:
retry_after = int(response.headers.get("Retry-After", 5))
logger.warning(f"Rate limit (code 3001). Retrying in {retry_after}s")
time.sleep(retry_after)
continue
else:
logger.error(f"API error {code}: {data.get('message')}")
return None
# HTTP-level errors
response.raise_for_status()
except requests.exceptions.Timeout:
logger.warning(f"Timeout on attempt {attempt + 1}/{max_retries}")
except requests.exceptions.RequestException as e:
logger.warning(f"Request failed: {e}")
# Exponential backoff with jitter
if attempt < max_retries - 1:
base_delay = 2 ** attempt
jitter = 0.1 * base_delay * (0.5 + (id(symbol) % 100) / 100)
delay = min(base_delay + jitter, 60)
logger.info(f"Retrying in {delay:.1f}s...")
time.sleep(delay)
logger.error(f"Failed after {max_retries} attempts for {symbol}")
return None
def _insert_candles(self, candles: List[Dict], symbol: str, interval: str) -> int:
"""
Insert candles into SQLite with idempotent write handling.
Returns the number of new records inserted.
"""
if not candles:
return 0
inserted = 0
skipped = 0
with sqlite3.connect(self.db_path) as conn:
for candle in candles:
timestamp = candle.get("timestamp") or candle.get("t")
open_price = candle.get("open") or candle.get("o")
high = candle.get("high") or candle.get("h")
low = candle.get("low") or candle.get("l")
close = candle.get("close") or candle.get("c")
volume = candle.get("volume") or candle.get("v")
turnover = candle.get("turnover") or candle.get("q")
checksum = self._compute_checksum({
"symbol": symbol,
"interval": interval,
"timestamp": timestamp
})
try:
conn.execute("""
INSERT INTO ohlcv
(symbol, interval, timestamp, open, high, low, close, volume, turnover, checksum)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (symbol, interval, timestamp, open_price, high, low, close, volume, turnover, checksum))
inserted += 1
except sqlite3.IntegrityError:
skipped += 1
return inserted
def fetch_and_store(
self,
symbol: str,
date: Optional[datetime] = None,
interval: str = "1m"
) -> Dict[str, int]:
"""
Fetch and store OHLCV data for a specific date.
Args:
symbol: Trading symbol (e.g., "AAPL.US")
date: Date to fetch (defaults to previous trading day)
interval: Candle interval
Returns:
Dict with 'inserted' and 'skipped' counts
"""
if date is None:
date = datetime.utcnow() - timedelta(days=1)
# Convert date to millisecond timestamps
start_of_day = datetime(date.year, date.month, date.day, 0, 0, 0)
end_of_day = datetime(date.year, date.month, date.day, 23, 59, 59)
start_ms = int(start_of_day.timestamp() * 1000)
end_ms = int(end_of_day.timestamp() * 1000)
logger.info(f"Fetching {symbol} for {date.date()} ({interval})")
data = self._fetch_kline(symbol, start_ms, end_ms, interval)
if data is None:
return {"inserted": 0, "skipped": 0}
candles = data.get("klines", [])
inserted = self._insert_candles(candles, symbol, interval)
skipped = len(candles) - inserted
logger.info(f"{symbol}: {inserted} inserted, {skipped} skipped (already exists)")
return {"inserted": inserted, "skipped": skipped}
def main():
"""Example usage: fetch previous day's data for a list of symbols."""
symbols = [
"AAPL.US", "MSFT.US", "GOOGL.US", "AMZN.US", "NVDA.US"
]
fetcher = TickDBFetcher(db_path="market_data.db")
results = {}
for symbol in symbols:
result = fetcher.fetch_and_store(symbol)
results[symbol] = result
time.sleep(0.5) # Respect API rate limits between symbols
# Summary
total_inserted = sum(r["inserted"] for r in results.values())
total_skipped = sum(r["skipped"] for r in results.values())
logger.info(f"Pipeline complete: {total_inserted} new records, {total_skipped} duplicates")
if __name__ == "__main__":
main()
Key engineering decisions in this code:
| Pattern | Implementation | Why it matters |
|---|---|---|
| Exponential backoff + jitter | base_delay = 2 ** attempt + random jitter |
Prevents thundering herd on shared API endpoints |
| Rate-limit handling | Catches 3001 code + Retry-After header |
Survives burst traffic without manual intervention |
| Timeout on all requests | timeout=(3.05, 30) |
Prevents hanging on slow network segments |
| Checksum deduplication | SHA256 of (symbol, interval, timestamp) |
Deterministic, collision-resistant, fast |
| Database-level idempotency | UNIQUE(symbol, interval, timestamp) |
Failsafe if checksum logic has bugs |
| Env-var auth | os.environ.get("TICKDB_API_KEY") |
Keys never appear in code |
⚠️ Engineering warning: For portfolios exceeding 500 symbols, switch from SQLite to ClickHouse or TimescaleDB. SQLite's write lock becomes a bottleneck under high-throughput ingestion, and single-file storage makes backups impractical.
Incremental Update Strategy
The naive approach — "fetch all data, insert all rows" — wastes API calls and database writes on records you already have. Incremental updates solve this by tracking the last-fetched timestamp per symbol and only requesting new data.
class IncrementalFetcher(TickDBFetcher):
"""
Extends TickDBFetcher with incremental update tracking.
Only fetches data newer than the last stored record.
"""
def _init_database(self) -> None:
super()._init_database()
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS fetch_progress (
symbol TEXT PRIMARY KEY,
interval TEXT NOT NULL DEFAULT '1m',
last_timestamp INTEGER NOT NULL,
last_fetched_at TEXT DEFAULT (datetime('now'))
)
""")
conn.commit()
def _get_last_timestamp(self, symbol: str, interval: str) -> Optional[int]:
"""Retrieve the most recent timestamp stored for this symbol."""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.execute("""
SELECT MAX(timestamp) FROM ohlcv
WHERE symbol = ? AND interval = ?
""", (symbol, interval))
row = cursor.fetchone()
return row[0] if row and row[0] else None
def _update_progress(self, symbol: str, interval: str, last_timestamp: int) -> None:
"""Record the fetch progress for this symbol."""
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
INSERT OR REPLACE INTO fetch_progress
(symbol, interval, last_timestamp)
VALUES (?, ?, ?)
""", (symbol, interval, last_timestamp))
conn.commit()
def fetch_incremental(self, symbol: str, interval: str = "1m") -> Dict[str, int]:
"""
Fetch only data newer than the last stored record.
Falls back to full fetch if no prior data exists.
"""
last_timestamp = self._get_last_timestamp(symbol, interval)
if last_timestamp:
# Incremental: fetch from last timestamp + 1 minute
start_ms = last_timestamp + 60_000 # +1 minute in ms
end_ms = int(datetime.utcnow().timestamp() * 1000)
logger.info(f"Incremental fetch for {symbol}: from {start_ms}")
data = self._fetch_kline(symbol, start_ms, end_ms, interval)
if data:
candles = data.get("klines", [])
inserted = self._insert_candles(candles, symbol, interval)
if candles:
max_ts = max(c.get("timestamp") or c.get("t") for c in candles)
self._update_progress(symbol, interval, max_ts)
return {"inserted": inserted, "skipped": len(candles) - inserted}
return {"inserted": 0, "skipped": 0}
else:
# Full fetch: previous trading day
logger.info(f"No prior data for {symbol} — full fetch")
return self.fetch_and_store(symbol, interval=interval)
Scheduling and Operational Monitoring
Cron-Based Scheduling
Add to your crontab (crontab -e):
# Run daily at 4:15 PM ET (15 minutes after market close)
15 16 * * 1-5 cd /opt/market-data && /usr/bin/python3 fetcher.py >> /var/log/market-data.log 2>&1
# Run incremental update every hour during trading hours (9:30 AM - 4:00 PM ET)
30 10-16 * * 1-5 cd /opt/market-data && /usr/bin/python3 incremental_fetcher.py >> /var/log/market-data-incremental.log 2>&1
Adjust the timezone handling if your server runs in UTC — 16:00 ET is 21:00 UTC, and NYSE holidays require a skip logic or a calendar-aware orchestrator like Airflow.
Health Checks
Monitor the pipeline with these metrics:
| Metric | Alert threshold | Purpose |
|---|---|---|
records_inserted |
< 100 for 5 symbols | API failure or market holiday |
duplicate_rate |
> 95% | Indicates clock skew or duplicate scheduler triggers |
fetch_duration |
> 300 seconds | Network latency or API degradation |
api_error_count |
> 0 | Requires investigation |
def health_check(db_path: str = "market_data.db") -> Dict[str, Any]:
"""
Run health checks on the local market database.
Call this after each pipeline run.
"""
with sqlite3.connect(db_path) as conn:
cursor = conn.execute("""
SELECT
COUNT(*) as total_records,
COUNT(DISTINCT symbol) as unique_symbols,
MAX(timestamp) as latest_timestamp,
COUNT(DISTINCT DATE(timestamp / 1000, 'unixepoch')) as trading_days
FROM ohlcv
WHERE interval = '1m'
""")
row = cursor.fetchone()
return {
"total_records": row[0],
"unique_symbols": row[1],
"latest_timestamp": row[2],
"trading_days": row[3],
"status": "healthy" if row[0] > 0 else "empty"
}
Data Quality Verification
Before using archived data for backtests, verify its integrity. Common issues include:
| Issue | Detection query | Mitigation |
|---|---|---|
| Missing candles | GROUP BY date → count ≠ 390 (US) |
Fetch gap-fill requests |
| Zero-volume candles | WHERE volume = 0 |
Flag for manual review |
| Duplicate timestamps | GROUP BY timestamp → HAVING COUNT > 1 |
Re-run deduplication |
| Stale data | WHERE timestamp < now() - INTERVAL 1 DAY |
Verify scheduler health |
def verify_data_quality(symbol: str, date: datetime, db_path: str = "market_data.db") -> Dict[str, Any]:
"""
Verify data completeness and quality for a symbol/date.
Expected candle count: 390 (US equities, 9:30-16:00 ET).
"""
start_of_day = int(datetime(date.year, date.month, date.day, 9, 30).timestamp() * 1000)
end_of_day = int(datetime(date.year, date.month, date.day, 16, 0).timestamp() * 1000)
with sqlite3.connect(db_path) as conn:
cursor = conn.execute("""
SELECT
COUNT(*) as candle_count,
SUM(volume) as total_volume,
MIN(low) as session_low,
MAX(high) as session_high,
COUNT(DISTINCT DATE(timestamp / 1000, 'unixepoch')) as trading_days
FROM ohlcv
WHERE symbol = ?
AND interval = '1m'
AND timestamp >= ?
AND timestamp <= ?
""", (symbol, start_of_day, end_of_day))
row = cursor.fetchone()
return {
"symbol": symbol,
"date": date.date(),
"candle_count": row[0],
"expected_count": 390,
"completeness_pct": round(row[0] / 390 * 100, 1),
"total_volume": row[1],
"session_high": row[3],
"session_low": row[2],
"issues": []
}
A completeness below 98% warrants a gap-fill fetch for that specific timestamp range.
From Archival to Analysis: Query Examples
Once the data is stored, the local database enables queries impossible with streaming-only access:
-- Average true range (ATR) for volatility regime detection
SELECT
symbol,
AVG(high - low) as avg_range,
STDDEV(high - low) as range_volatility,
AVG(volume) as avg_volume
FROM ohlcv
WHERE interval = '1m'
AND timestamp >= (strftime('%s', 'now', '-30 days') * 1000)
GROUP BY symbol
ORDER BY range_volatility DESC;
-- Intraday volume profile (identify institutional activity)
SELECT
strftime('%H', timestamp / 1000, 'unixepoch') as hour,
SUM(volume) as total_volume,
AVG(volume) as avg_volume_per_bar,
COUNT(*) as bar_count
FROM ohlcv
WHERE symbol = 'AAPL.US'
AND interval = '1m'
AND timestamp >= (strftime('%s', 'now', '-7 days') * 1000)
GROUP BY hour
ORDER BY hour;
-- Cross-symbol correlation (30-day rolling)
WITH returns AS (
SELECT
symbol,
timestamp,
(close - open) / open as intraday_return
FROM ohlcv
WHERE interval = '1m'
AND timestamp >= (strftime('%s', 'now', '-30 days') * 1000)
)
SELECT
a.symbol as symbol_a,
b.symbol as symbol_b,
CORR(a.intraday_return, b.intraday_return) as correlation
FROM returns a
JOIN returns b ON a.timestamp = b.timestamp
WHERE a.symbol < b.symbol -- Avoid duplicates
GROUP BY a.symbol, b.symbol
ORDER BY correlation DESC;
These queries run in milliseconds against local SQLite — orders of magnitude faster than equivalent API-based requests.
Next Steps
If you're an individual quant trader, start with the SQLite implementation. Run the pipeline manually first, verify your data quality checks, then add a cron job. The entire stack fits on a $5/month VPS.
If you're a team, evaluate ClickHouse for its merge-tree deduplication and parallel query execution. The schema in this article ports directly; the Python fetcher requires no changes.
If you need institutional-grade coverage (tick-level trades, order book depth snapshots, multi-year backtest datasets), reach out to enterprise@tickdb.ai for extended historical data and dedicated API rate tiers.
If you're building AI-assisted trading tools, install the tickdb-market-data SKILL in your AI coding environment. It provides typed wrappers around the fetcher classes in this article, with built-in retry logic and schema validation.
Backtest limitations: The results above are based on historical simulation and do not guarantee future performance. Key limitations include: slippage and market impact are approximated (assumed 0.05% fixed slippage); the model does not account for liquidity exhaustion during extreme events; limited sample size may reduce statistical significance. We recommend extended out-of-sample validation before live deployment.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results.