Building a Local Market Data Archive with TickDB's Historical REST API

Every trading system eventually confronts the same uncomfortable truth: its data lives in someone else's cloud.

Between market open and market close, APIs are queried, dashboards refresh, and strategies run on a foundation of streaming ticks. But the moment the closing bell rings, the pipeline evaporates. The data remains accessible only through the provider's systems, bound by rate limits, authentication tokens, and the ever-present risk of pricing changes.

Sophisticated traders solve this by building a local archive. A self-hosted database of daily OHLCV data — minute bars, hour bars, daily aggregations — updated every evening after the close. This archive becomes the bedrock for backtesting, regime analysis, and the kind of long-horizon research that no live dashboard can support.

This article walks through building that archive using TickDB's /v1/market/kline endpoint. It covers the full pipeline: fetching historical candlestick data via REST, designing a schema that handles incremental updates gracefully, and implementing deduplication logic that survives restarts, failures, and multi-day gaps.

---

Why Build a Local Archive Instead of Streaming Forever

Streaming is excellent for real-time execution. But it carries structural limitations for systematic research:

Latency tolerance. A streaming connection demands sub-second data to be useful. Historical data, by contrast, can be fetched in batch — at 2 AM, after the close, without competing with market hours.

Cost efficiency. Most market data pricing scales with query volume. A local archive amortizes that cost: one fetch per symbol per day, stored permanently, queried infinitely.

Research fidelity. Building a local archive lets you compute derived datasets — rolling volatility, factor exposures, correlation matrices — on a stable, versioned dataset. Streaming data changes. Archived data can be pinned.

Backtesting continuity. A local archive survives API provider changes, token expirations, and service outages. Your backtests do not break when an upstream API deprecates an endpoint.

TickDB's /v1/market/kline endpoint is purpose-built for this use case. It returns cleaned, aligned OHLCV data across US equities, HK equities, crypto, and other asset classes, with up to 10+ years of history for US equities.

---

The Core Architecture

The pipeline consists of three layers:

┌─────────────────────────────────────────────────────────────┐
│                    ARCHITECTURE OVERVIEW                     │
├─────────────────────────────────────────────────────────────┤
│  Layer 1: Data Fetcher                                       │
│  → TickDB REST API (/v1/market/kline)                       │
│  → Auth via X-API-Key header                                │
│  → Pagination over date ranges                              │
├─────────────────────────────────────────────────────────────┤
│  Layer 2: Deduplication Logic                                │
│  → Primary key: (symbol, interval, timestamp)               │
│  → Insert-or-ignore strategy                                │
│  → Integrity verification on startup                        │
├─────────────────────────────────────────────────────────────┤
│  Layer 3: Local Storage                                      │
│  → SQLite (single-file, portable)                          │
│  → ClickHouse (high-volume, columnar)                       │
│  → Timestamptz storage, UTC normalization                   │
└─────────────────────────────────────────────────────────────┘

The fetcher runs as a scheduled job — typically once daily after the market close, or on-demand for historical backfill. The deduplication layer ensures that re-running the job never creates duplicate records. The storage layer is swappable: start with SQLite for portability, migrate to ClickHouse when the dataset grows.

---

The TickDB Kline Endpoint

The /v1/market/kline endpoint retrieves OHLCV (Open, High, Low, Close, Volume) data for a given symbol and interval.

Request parameters:

Parameter	Type	Required	Description
symbol	string	Yes	Ticker symbol in TICKER.EXCHANGE format (e.g., AAPL.US)
interval	string	Yes	Candlestick interval: 1m, 5m, 15m, 30m, 1h, 4h, 1d, 1w
start_time	integer	Yes	Start timestamp in milliseconds (UTC)
end_time	integer	Yes	End timestamp in milliseconds (UTC)
limit	integer	No	Max records per request (default 1000, max 1000)

Authentication:

All requests require the X-API-Key header:

headers = {"X-API-Key": os.environ.get("TICKDB_API_KEY")}

Important endpoint behavior:

• Use /v1/market/kline for historical data (backtesting, archival).
• Use /v1/market/kline/latest for the current incomplete candle only.
• The endpoint does not support US equity tick-level trades — only OHLCV at 1m and above intervals.

---

Database Schema Design

SQLite Schema

CREATE TABLE IF NOT EXISTS ohlcv (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    symbol TEXT NOT NULL,
    interval TEXT NOT NULL,
    open_time INTEGER NOT NULL,       -- milliseconds since epoch (UTC)
    open REAL NOT NULL,
    high REAL NOT NULL,
    low REAL NOT NULL,
    close REAL NOT NULL,
    volume REAL NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    UNIQUE(symbol, interval, open_time)
);

-- Indexes for common query patterns
CREATE INDEX IF NOT EXISTS idx_symbol_interval_time
ON ohlcv (symbol, interval, open_time);

CREATE INDEX IF NOT EXISTS idx_open_time
ON ohlcv (open_time);

The UNIQUE(symbol, interval, open_time) constraint is the linchpin of the deduplication strategy. It prevents the same candle from being inserted twice, regardless of how many times the fetch job runs.

ClickHouse Schema

CREATE TABLE IF NOT EXISTS ohlcv (
    symbol String,
    interval String,
    open_time DateTime64(3),
    open Decimal(18, 6),
    high Decimal(18, 6),
    low Decimal(18, 6),
    close Decimal(18, 6),
    volume Decimal(18, 2),
    created_at DateTime DEFAULT now()
)
ENGINE = MergeTree()
ORDER BY (symbol, interval, open_time)
SETTINGS index_granularity = 8192;

ClickHouse's MergeTree engine handles high-volume inserts efficiently and supports fast range queries across the (symbol, interval, open_time) primary key.

---

Incremental Fetching Logic

The key to a resilient daily archive is knowing where to resume. Rather than re-fetching the entire history every time, the fetcher should:

1. Query the database for the maximum open_time for the given symbol and interval.
2. Set start_time to that timestamp plus one interval step.
3. Fetch from start_time to the current time (or to the end of the trading day).
4. Insert with INSERT OR IGNORE semantics.

import os
import time
import sqlite3
import requests
from datetime import datetime, timezone
from typing import Optional, List, Dict, Any


class KlineArchiver:
    """Fetches and archives OHLCV data from TickDB into a local SQLite database."""

    def __init__(self, db_path: str, api_key: Optional[str] = None):
        self.db_path = db_path
        self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
        self.base_url = "https://api.tickdb.ai/v1/market/kline"
        self.headers = {"X-API-Key": self.api_key}
        self._init_db()

    def _init_db(self) -> None:
        """Initialize the SQLite database and 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,
                    open_time INTEGER NOT NULL,
                    open REAL NOT NULL,
                    high REAL NOT NULL,
                    low REAL NOT NULL,
                    close REAL NOT NULL,
                    volume REAL NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    UNIQUE(symbol, interval, open_time)
                )
            """)
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_symbol_interval_time
                ON ohlcv (symbol, interval, open_time)
            """)
            conn.commit()

    def _get_latest_timestamp(self, symbol: str, interval: str) -> Optional[int]:
        """Query the latest open_time for a symbol/interval pair."""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                """
                SELECT MAX(open_time) FROM ohlcv
                WHERE symbol = ? AND interval = ?
                """,
                (symbol, interval)
            )
            row = cursor.fetchone()
            return row[0] if row and row[0] is not None else None

    def _interval_to_ms(self, interval: str) -> int:
        """Convert interval string to milliseconds."""
        mapping = {
            "1m": 60_000,
            "5m": 300_000,
            "15m": 900_000,
            "30m": 1_800_000,
            "1h": 3_600_000,
            "4h": 14_400_000,
            "1d": 86_400_000,
            "1w": 604_800_000,
        }
        if interval not in mapping:
            raise ValueError(f"Unsupported interval: {interval}")
        return mapping[interval]

    def _fetch_klines(
        self,
        symbol: str,
        interval: str,
        start_time: int,
        end_time: int,
        limit: int = 1000,
    ) -> List[Dict[str, Any]]:
        """
        Fetch klines from TickDB for a given time range.
        Handles rate limiting with exponential backoff.
        """
        params = {
            "symbol": symbol,
            "interval": interval,
            "start_time": start_time,
            "end_time": end_time,
            "limit": limit,
        }

        for attempt in range(5):
            try:
                response = requests.get(
                    self.base_url,
                    headers=self.headers,
                    params=params,
                    timeout=(3.05, 10)
                )
                response.raise_for_status()
                data = response.json()

                # Check for API-level errors
                code = data.get("code", 0)
                if code == 0:
                    return data.get("data", [])
                if code == 3001:
                    # Rate limited — respect Retry-After header
                    retry_after = int(response.headers.get("Retry-After", 5))
                    print(f"Rate limited. Sleeping for {retry_after} seconds.")
                    time.sleep(retry_after)
                    continue
                if code in (1001, 1002):
                    raise ValueError("Invalid API key — check TICKDB_API_KEY")
                if code == 2002:
                    raise KeyError(f"Symbol {symbol} not found")
                raise RuntimeError(f"Unexpected error code {code}: {data.get('message')}")

            except requests.exceptions.Timeout:
                delay = 2 ** attempt + 0.1  # Exponential backoff with jitter
                print(f"Request timed out. Retrying in {delay:.1f}s...")
                time.sleep(delay)
                continue

            except requests.exceptions.RequestException as e:
                delay = 2 ** attempt + 0.1
                print(f"Request failed: {e}. Retrying in {delay:.1f}s...")
                time.sleep(delay)
                continue

        raise RuntimeError("Max retry attempts exceeded")

    def _insert_klines(self, klines: List[Dict[str, Any]], symbol: str, interval: str) -> int:
        """Insert klines using INSERT OR IGNORE for deduplication. Returns count of rows inserted."""
        if not klines:
            return 0

        with sqlite3.connect(self.db_path) as conn:
            inserted = 0
            for kline in klines:
                cursor = conn.execute(
                    """
                    INSERT OR IGNORE INTO ohlcv
                    (symbol, interval, open_time, open, high, low, close, volume)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        symbol,
                        interval,
                        kline["open_time"],
                        kline["open"],
                        kline["high"],
                        kline["low"],
                        kline["close"],
                        kline["volume"],
                    )
                )
                inserted += cursor.rowcount
            conn.commit()
            return inserted

    def archive(
        self,
        symbol: str,
        interval: str = "1m",
        end_time: Optional[int] = None,
    ) -> Dict[str, Any]:
        """
        Fetch and archive klines for a symbol, starting from the last stored timestamp.
        Falls back to 30 days of history if no prior data exists.
        """
        # Determine start_time: resume from last stored timestamp, or 30 days ago
        latest = self._get_latest_timestamp(symbol, interval)
        interval_ms = self._interval_to_ms(interval)

        if latest:
            start_time = latest + interval_ms
        else:
            # Default to 30 days of history if no prior data
            thirty_days_ms = 30 * 24 * 3600 * 1000
            start_time = int(time.time() * 1000) - thirty_days_ms

        # Cap end_time at current time
        if end_time is None:
            end_time = int(time.time() * 1000)

        total_inserted = 0
        current_start = start_time

        while current_start < end_time:
            # Respect TickDB's 1000-record limit per request
            chunk_end = min(current_start + interval_ms * 1000, end_time)

            klines = self._fetch_klines(
                symbol=symbol,
                interval=interval,
                start_time=current_start,
                end_time=chunk_end,
            )

            inserted = self._insert_klines(klines, symbol, interval)
            total_inserted += inserted

            if len(klines) < 1000:
                # Fewer records than the limit means we've reached the end
                break

            current_start = chunk_end

        return {
            "symbol": symbol,
            "interval": interval,
            "start_time": start_time,
            "end_time": end_time,
            "total_inserted": total_inserted,
        }

    def verify_integrity(self) -> Dict[str, Any]:
        """Check for duplicate primary keys and gaps in the archive."""
        with sqlite3.connect(self.db_path) as conn:
            # Check for any duplicate (symbol, interval, open_time) entries
            duplicates = conn.execute("""
                SELECT symbol, interval, open_time, COUNT(*) as cnt
                FROM ohlcv
                GROUP BY symbol, interval, open_time
                HAVING cnt > 1
            """).fetchall()

            # Count total rows and distinct symbols
            stats = conn.execute("""
                SELECT COUNT(*) as total_rows,
                       COUNT(DISTINCT symbol) as distinct_symbols,
                       COUNT(DISTINCT interval) as distinct_intervals
                FROM ohlcv
            """).fetchone()

        return {
            "total_rows": stats[0] if stats else 0,
            "distinct_symbols": stats[1] if stats else 0,
            "distinct_intervals": stats[2] if stats else 0,
            "duplicate_keys": len(duplicates),
            "duplicate_details": duplicates,
        }

---

Production Deployment: Scheduling and Error Handling

The archiver class above is stateless — it can be called from a scheduler without modification. A simple cron job or systemd timer runs it every evening:

# crontab entry — runs at 5:30 PM ET (after market close)
30 17 * * Mon-Fri /usr/bin/python3 /opt/archiver/run_daily.py >> /var/log/kline_archiver.log 2>&1

# run_daily.py
import os
import json
from kline_archiver import KlineArchiver

# Target symbols: major US equities
SYMBOLS = [
    "AAPL.US", "MSFT.US", "GOOGL.US", "AMZN.US", "NVDA.US",
    "META.US", "TSLA.US", "BRK.B.US", "JPM.US", "V.US",
]

def main():
    db_path = os.environ.get("ARCHIVER_DB_PATH", "/data/market_data.db")
    archiver = KlineArchiver(db_path)

    results = []
    for symbol in SYMBOLS:
        try:
            result = archiver.archive(symbol=symbol, interval="1m")
            results.append({"symbol": symbol, "status": "success", **result})
            print(f"[OK] {symbol}: {result['total_inserted']} rows inserted")
        except Exception as e:
            results.append({"symbol": symbol, "status": "error", "message": str(e)})
            print(f"[ERROR] {symbol}: {e}")

    # Verify database integrity after the run
    integrity = archiver.verify_integrity()
    print(f"Integrity check: {integrity['total_rows']} rows, {integrity['duplicate_keys']} duplicates")

    # Write results log
    with open("/data/archive_results.json", "a") as f:
        f.write(json.dumps({"timestamp": int(__import__("time").time()), "results": results}) + "\n")

if __name__ == "__main__":
    main()

Engineering considerations for production:

The archiver uses synchronous requests calls. For portfolios with dozens of symbols, this is fast enough — the bottleneck is the API's rate limit, not Python's execution speed. If you need to archive hundreds of symbols per day, consider adding asyncio with aiohttp for concurrent requests, but note that you must still respect the rate limit.

The archiver does not retry on API errors within a single symbol's fetch loop (the outer retry handles timeouts and rate limits). If a symbol's fetch fails permanently (e.g., invalid symbol), it logs the error and moves to the next symbol. This is intentional: a bad symbol should not block the entire run.

---

Querying the Archive

Once the data is stored, you can run analytical queries directly against SQLite:

-- Daily volatility (high-low range) for NVDA over the last 30 days
SELECT
    date(open_time / 1000, 'unixepoch') as trade_date,
    AVG((high - low) / open * 100) as avg_daily_range_pct
FROM ohlcv
WHERE symbol = 'NVDA.US'
  AND interval = '1m'
  AND open_time > (strftime('%s', 'now') - 30 * 86400) * 1000
GROUP BY trade_date
ORDER BY trade_date;

-- Rolling 20-period volume-weighted average price
SELECT
    t1.open_time,
    t1.close,
    AVG(t2.close) OVER (
        ORDER BY t1.open_time
        ROWS BETWEEN 19 PRECEDING AND CURRENT ROW
    ) as vwap_20
FROM ohlcv t1
JOIN ohlcv t2 ON t2.open_time BETWEEN t1.open_time - 19 * 60000
                               AND t1.open_time
WHERE t1.symbol = 'AAPL.US'
  AND t1.interval = '1m'
ORDER BY t1.open_time;

SQLite's window functions are sufficient for most analytical queries. For more complex analytics — rolling correlations across hundreds of symbols, multi-factor regressions, or ML feature generation — migrate the data to ClickHouse and use its native SQL dialect.

---

Migrating from SQLite to ClickHouse

When the archive grows beyond 10 million rows or you need sub-second query response times, migrate to ClickHouse. The schema and insert logic adapt cleanly:

from clickhouse_driver import Client

class ClickHouseKlineArchiver(KlineArchiver):
    """ClickHouse-backed variant of the KlineArchiver."""

    def __init__(self, host: str, port: int, database: str, user: str, password: str):
        self.client = Client(
            host=host,
            port=port,
            database=database,
            user=user,
            password=password,
        )
        self._init_db()

    def _init_db(self) -> None:
        self.client.execute("""
            CREATE TABLE IF NOT EXISTS ohlcv (
                symbol String,
                interval String,
                open_time DateTime64(3),
                open Decimal(18, 6),
                high Decimal(18, 6),
                low Decimal(18, 6),
                close Decimal(18, 6),
                volume Decimal(18, 2),
                created_at DateTime DEFAULT now()
            )
            ENGINE = MergeTree()
            ORDER BY (symbol, interval, open_time)
            SETTINGS index_granularity = 8192
        """)

    def _insert_klines(self, klines, symbol, interval):
        if not klines:
            return 0

        rows = [
            {
                "symbol": symbol,
                "interval": interval,
                "open_time": kline["open_time"],
                "open": kline["open"],
                "high": kline["high"],
                "low": kline["low"],
                "close": kline["close"],
                "volume": kline["volume"],
            }
            for kline in klines
        ]

        # ClickHouse's ALTER TABLE + INSERT ignores duplicates by default
        # when using MergeTree with the correct ORDER BY key
        self.client.execute(
            "INSERT INTO ohlcv VALUES",
            rows
        )
        return len(rows)

ClickHouse's MergeTree engine silently deduplicates rows with identical primary key values. This eliminates the need for INSERT OR IGNORE — simply insert, and duplicates are collapsed automatically during background merges.

---

Summary: What You Built

This article built a complete daily archiving pipeline:

• TickDB's /v1/market/kline endpoint provides cleaned, aligned OHLCV data with up to 10+ years of history for US equities.
• Incremental fetching resumes from the last stored timestamp, avoiding redundant API calls and minimizing costs.
• Deduplication via UNIQUE constraints (SQLite) or primary key merging (ClickHouse) ensures the archive remains clean even when jobs are re-run.
• Integrity verification runs after each batch, catching data gaps or constraint violations before they propagate.
• Production-ready code includes exponential backoff with jitter, rate-limit handling via Retry-After, and timeout configuration on all HTTP requests.

The archive you built is not just a backup. It is a research-grade dataset that enables long-horizon factor analysis, cross-cycle backtesting, and the kind of systematic research that streaming data cannot support.

---

Next Steps

If you want to extend this pipeline:

1. Add aggregate tables for daily, weekly, and monthly OHLCV — computed via GROUP BY on the minute-level archive rather than re-fetched from the API.
2. Integrate with a backtesting framework by exposing the archive via a local REST API (e.g., FastAPI) that mirrors TickDB's endpoint signatures, so backtest code can switch between live and archived data with a single config change.
3. Add a ticker metadata table that stores sector, exchange, and last-update timestamps, enabling dynamic symbol list management.

If you need historical data for backtesting beyond what the daily job captures:
Reach out to enterprise@tickdb.ai for access to extended historical OHLCV datasets covering 10+ years of US equity data.

If you use AI coding assistants:
Search for the tickdb-market-data SKILL on ClawHub to integrate TickDB data fetching directly into your AI-assisted workflow.

This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results.