Automated Daily OHLCV Archival: Building a Local Market Database with TickDB's Historical API

Every evening around 4:05 PM ET, when US equity markets have closed their regular session, a quiet problem begins. The day's 390 minutes of price action — captured in 1-minute OHLCV candles for every symbol you care about — exists in the cloud, accessible via API, but not in your local system. If you run a quantitative strategy that depends on clean historical bars, or if you're building a feature store for a machine learning pipeline, that gap between cloud data and local persistence is a daily tax on your research velocity.

The naive solution is simple: write a script that fetches the day's data every night, dumps it into a spreadsheet, and moves on. That approach breaks the moment you need to run backtests across multiple years, handle exchange holidays gracefully, or support incremental updates where you only fetch the last few minutes in case your job ran late. The robust solution — the one this article walks through — is a fully automated archival pipeline that pulls minute-level OHLCV data from TickDB's /v1/market/kline endpoint, stores it with deduplication, and handles the edge cases that separate a working prototype from a production-grade system.

This is not a theoretical architecture. Every code block here is directly runnable. The system described below handles incremental updates (no full re-downloads), survives failed jobs, and stores data in either SQLite (for individual researchers) or ClickHouse (for teams running large-scale backtests).

---

Architecture Overview

Before diving into code, it helps to see the full picture. The archival pipeline consists of five logical stages:

Scheduler (cron / systemd timer)
  │
  ▼
1. Fetch  ──► TickDB /v1/market/kline  (REST, minute-level)
  │
  ▼
2. Parse  ──► Normalize response → list of OHLCV tuples
  │
  ▼
3. Deduplicate ──► Check (symbol, interval, open_time) composite key
  │
  ▼
4. Upsert ──► INSERT OR REPLACE into SQLite / bulk INSERT into ClickHouse
  │
  ▼
5. Log ──► Record run timestamp, rows affected, any errors

The pipeline is idempotent by design. Running it twice on the same day produces no duplicate rows. Running it after a failed previous attempt recovers cleanly. The scheduler fires at a fixed time each trading day, but the pipeline itself is time-agnostic — it fetches whatever date range you specify, which makes it suitable for both scheduled daily runs and ad-hoc backfills.

The key design decisions that make this production-grade are:

• Time window targeting: The pipeline fetches a specific date range (e.g., the just-completed trading day), not an unbounded "latest" query. This makes incremental updates predictable and debuggable.
• Composite deduplication key: (symbol, interval, open_time) uniquely identifies each candle. The storage layer uses this key for INSERT OR REPLACE semantics, so re-running the pipeline never corrupts data.
• Cursor-based backfill: A metadata table tracks the last successfully archived date per symbol, enabling the pipeline to resume from where it left off without manual intervention.

---

Understanding TickDB's Kline Endpoint

The /v1/market/kline endpoint is TickDB's primary interface for historical OHLCV data. Getting the parameters right is essential — the difference between kline and kline/latest, or between start_time and from, is the difference between a working pipeline and a silent data gap.

Endpoint Signature

GET https://api.tickdb.ai/v1/market/kline

Authentication uses the X-API-Key header with your API key loaded from the TICKDB_API_KEY environment variable. The key is never passed as a URL parameter for REST endpoints.

Core Parameters

Parameter	Type	Description	Notes
symbol	string	Ticker symbol with market suffix	e.g., AAPL.US, BTC.Binance
interval	string	Candle interval	1m, 5m, 15m, 1h, 4h, 1d
from	integer	Start timestamp in seconds (UTC)	Inclusive
to	integer	End timestamp in seconds (UTC)	Exclusive
limit	integer	Maximum candles per request	Up to 1000; pagination required for longer ranges

Response Structure

{
  "code": 0,
  "data": {
    "AAPL.US": [
      [1700000000, 185.20, 185.45, 185.10, 185.38, 125400],
      [1700000060, 185.38, 185.60, 185.35, 185.55, 98200]
    ]
  }
}

Each inner array represents [open_time, open, high, low, close, volume]. Timestamps are Unix seconds (UTC). The open time marks the bar's start — a 1m candle with open_time=1700000000 covers the interval [1700000000, 1700000060).

Critical Constraint: Pagination

For a full trading day at 1-minute resolution, US equities generate 390 bars per symbol. With a limit=1000, a single request covers this comfortably for one symbol. However, if you are backfilling 5 years of daily data for 500 symbols, you need to paginate through responses using the from parameter, advancing it to the last received open_time + interval_seconds for each subsequent page.

Common Mistakes to Avoid

Passing from as a date string instead of a Unix timestamp is the most frequent error. Another common mistake is using kline/latest for historical data — that endpoint returns only the current incomplete candle and is intended for live dashboards, not archival. The third trap is forgetting that to is exclusive, which means requesting from=1700000000, to=1700000000 returns zero candles. Always set to to the first timestamp after the last candle you want.

---

Designing the Storage Layer

The choice of database depends on your scale and query patterns. Both SQLite and ClickHouse are covered here; the code is structured so you can swap one for the other without rewriting the fetch logic.

SQLite: Lightweight and Portable

SQLite is the right choice if you are an individual researcher, you need portability (the database is a single file you can copy anywhere), and your dataset fits comfortably on a single machine. A 10-year archive of 1-minute OHLCV for 200 US equity symbols is approximately 4.5 billion rows, which at roughly 60 bytes per row compresses to about 270 GB — at the upper end of what SQLite handles well, but workable on modern hardware with appropriate indexing.

The schema uses a composite primary key for deduplication:

CREATE TABLE IF NOT EXISTS ohlcv_1m (
    symbol         TEXT    NOT NULL,
    open_time      INTEGER NOT NULL,
    interval_      TEXT    NOT NULL DEFAULT '1m',
    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,
    PRIMARY KEY (symbol, interval_, open_time)
);

CREATE INDEX IF NOT EXISTS idx_ohlcv_symbol_time
    ON ohlcv_1m (symbol, open_time DESC);

CREATE TABLE IF NOT EXISTS archive_cursor (
    symbol         TEXT PRIMARY KEY,
    last_archived  INTEGER NOT NULL,
    updated_at     TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

The archive_cursor table is the mechanism for incremental updates. Instead of re-fetching the entire trading day every run, the pipeline reads the last_archived timestamp for each symbol and requests only candles with open_time > last_archived.

ClickHouse: For Teams and Large Backtests

If you are running multi-year backtests across hundreds of symbols with concurrent queries, ClickHouse provides columnar storage with partition pruning and vectorized execution that SQLite cannot match. The equivalent table definition:

CREATE TABLE IF NOT EXISTS ohlcv_1m (
    symbol     String,
    open_time  UInt32,
    interval_  String DEFAULT '1m',
    open       Float64,
    high       Float64,
    low        Float64,
    close      Float64,
    volume     Float64,
    created_at DateTime DEFAULT now()
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(fromUnixTimestamp64Milli(open_time))
ORDER BY (symbol, open_time)
SETTINGS index_granularity = 8192;

ClickHouse's MergeTree engine stores data sorted by (symbol, open_time), which matches the access pattern of virtually every quantitative strategy. The partition by month allows efficient pruning when querying a specific date range.

---

The Archival Pipeline in Python

The following implementation is a complete, production-grade pipeline. It handles authentication, request pagination, response parsing, deduplication, and storage. It is designed to be called from a scheduler or imported as a module.

import os
import time
import json
import sqlite3
import logging
from datetime import datetime, timezone, timedelta
from typing import Generator, Optional
from dataclasses import dataclass

import requests

# ─────────────────────────────────────────────────────────────────
# Configuration
# ─────────────────────────────────────────────────────────────────
API_KEY = os.environ.get("TICKDB_API_KEY")
if not API_KEY:
    raise EnvironmentError("TICKDB_API_KEY environment variable is not set")

BASE_URL = "https://api.tickdb.ai/v1/market/kline"
DB_PATH = os.environ.get("ARCHIVE_DB_PATH", "market_data.db")

# Symbols to archive — in production, load from a config file or database
SYMBOLS = ["AAPL.US", "MSFT.US", "NVDA.US", "SPY.US", "QQQ.US"]
INTERVAL = "1m"
CANDLES_PER_PAGE = 1000  # Max allowed by TickDB

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger(__name__)


@dataclass
class KlineCandle:
    """Represents a single OHLCV candle."""
    symbol: str
    open_time: int
    open: float
    high: float
    low: float
    close: float
    volume: float


# ─────────────────────────────────────────────────────────────────
# TickDB API Client
# ─────────────────────────────────────────────────────────────────
class TickDBClient:
    """Low-level client for the TickDB kline endpoint."""

    def __init__(self, api_key: str, base_url: str = BASE_URL):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({"X-API-Key": api_key})

    def fetch_klines(
        self,
        symbol: str,
        interval: str,
        from_ts: int,
        to_ts: int,
        limit: int = CANDLES_PER_PAGE,
    ) -> list[KlineCandle]:
        """
        Fetch a page of kline candles for a single symbol and time range.
        Handles rate limiting (code 3001) with Retry-After.
        """
        params = {
            "symbol": symbol,
            "interval": interval,
            "from": from_ts,
            "to": to_ts,
            "limit": limit,
        }

        response = self.session.get(self.base_url, params=params, timeout=(3.05, 10))
        data = response.json()

        code = data.get("code", -1)

        if code == 3001:
            # Rate limited — respect Retry-After header
            retry_after = int(response.headers.get("Retry-After", 5))
            logger.warning(f"Rate limited. Sleeping {retry_after}s before retry.")
            time.sleep(retry_after)
            return self.fetch_klines(symbol, interval, from_ts, to_ts, limit)

        if code != 0:
            raise RuntimeError(
                f"TickDB API error {code}: {data.get('message', 'unknown')}"
            )

        raw_candles = data.get("data", {}).get(symbol, [])
        return [
            KlineCandle(
                symbol=symbol,
                open_time=candle[0],
                open=float(candle[1]),
                high=float(candle[2]),
                low=float(candle[3]),
                close=float(candle[4]),
                volume=float(candle[5]),
            )
            for candle in raw_candles
        ]

    def fetch_all_candles(
        self, symbol: str, interval: str, from_ts: int, to_ts: int
    ) -> Generator[KlineCandle, None, int]:
        """
        Paginate through all available candles for the given time range.
        Yields candles one at a time and yields the total count on completion.
        """
        current_from = from_ts
        total_fetched = 0

        while True:
            candles = self.fetch_klines(symbol, interval, current_from, to_ts)

            if not candles:
                break

            for candle in candles:
                yield candle

            total_fetched += len(candles)
            last_time = candles[-1].open_time

            # Advance to the next bar after the last received
            current_from = last_time + 60  # 60 seconds for 1m interval
            if current_from >= to_ts:
                break

            # Brief pause to avoid hammering rate limits during backfills
            time.sleep(0.1)

        yield total_fetched


# ─────────────────────────────────────────────────────────────────
# Storage Layer (SQLite)
# ─────────────────────────────────────────────────────────────────
class SQLiteArchiver:
    """Handles persistence to SQLite with deduplication."""

    def __init__(self, db_path: str):
        self.db_path = db_path
        self._init_schema()

    def _init_schema(self):
        """Create tables if they do not exist."""
        with sqlite3.connect(self.db_path) as conn:
            conn.executescript("""
                CREATE TABLE IF NOT EXISTS ohlcv_1m (
                    symbol     TEXT    NOT NULL,
                    open_time  INTEGER NOT NULL,
                    interval_  TEXT    NOT NULL DEFAULT '1m',
                    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,
                    PRIMARY KEY (symbol, interval_, open_time)
                );

                CREATE INDEX IF NOT EXISTS idx_ohlcv_symbol_time
                    ON ohlcv_1m (symbol, open_time DESC);

                CREATE TABLE IF NOT EXISTS archive_cursor (
                    symbol        TEXT PRIMARY KEY,
                    last_archived INTEGER NOT NULL,
                    updated_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                );
            """)

    def get_cursor(self, symbol: str) -> Optional[int]:
        """Return the last archived open_time for a symbol, or None if never archived."""
        with sqlite3.connect(self.db_path) as conn:
            row = conn.execute(
                "SELECT last_archived FROM archive_cursor WHERE symbol = ?",
                (symbol,),
            ).fetchone()
            return row[0] if row else None

    def upsert_candles(self, candles: list[KlineCandle]) -> int:
        """
        Insert or replace candles. Returns the number of rows affected.
        SQLite's INSERT OR REPLACE handles the deduplication via the PRIMARY KEY.
        """
        if not candles:
            return 0

        with sqlite3.connect(self.db_path) as conn:
            conn.executemany(
                """
                INSERT OR REPLACE INTO ohlcv_1m
                    (symbol, open_time, interval_, open, high, low, close, volume)
                VALUES (?, ?, '1m', ?, ?, ?, ?, ?)
                """,
                [
                    (c.symbol, c.open_time, c.open, c.high, c.low, c.close, c.volume)
                    for c in candles
                ],
            )
            return conn.total_changes

    def update_cursor(self, symbol: str, last_archived: int):
        """Record the most recent successfully archived open_time for a symbol."""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                """
                INSERT OR REPLACE INTO archive_cursor (symbol, last_archived, updated_at)
                VALUES (?, ?, CURRENT_TIMESTAMP)
                """,
                (symbol, last_archived),
            )


# ─────────────────────────────────────────────────────────────────
# Date Helpers
# ─────────────────────────────────────────────────────────────────
def trading_day_offset(offset: int = 0) -> datetime:
    """
    Return the trading day that is `offset` days before the most recent weekday.
    offset=0  → today (if weekday) or Friday (if weekend)
    offset=1  → previous trading day
    """
    now = datetime.now(timezone.utc)
    # Shift to US Eastern for trading day boundary
    today = now - timedelta(hours=5)  # Approximate ET offset
    days_back = offset + (1 if today.weekday() == 6 else 0)  # Sunday = Monday offset
    if today.weekday() == 5:  # Saturday
        days_back += 1
    trading_day = today - timedelta(days=days_back)
    return trading_day.replace(hour=0, minute=0, second=0, microsecond=0)


def day_range_ts(day: datetime) -> tuple[int, int]:
    """Convert a trading day to from/to Unix timestamps (US equity, 9:30–16:00 ET)."""
    # Market open: 9:30 AM ET = 14:30 UTC (or 13:30 UTC during DST)
    # Use a conservative UTC offset; in production, use a proper holiday calendar
    from_ts = int((day + timedelta(hours=14, minutes=30)).timestamp())
    to_ts = int((day + timedelta(hours=21)).timestamp())  # Safe upper bound
    return from_ts, to_ts


# ─────────────────────────────────────────────────────────────────
# Main Pipeline
# ─────────────────────────────────────────────────────────────────
def run_daily_archive(symbols: list[str] = SYMBOLS, lookback_days: int = 1):
    """
    Main entry point. Archives minute-level OHLCV for the specified symbols.

    Args:
        symbols: List of tickers to archive.
        lookback_days: How many trading days to backfill (1 = just yesterday).
    """
    client = TickDBClient(API_KEY)
    archiver = SQLiteArchiver(DB_PATH)

    for day_offset in range(lookback_days):
        trading_day = trading_day_offset(day_offset)
        from_ts, to_ts = day_range_ts(trading_day)

        logger.info(
            f"Archiving {trading_day.date()}  "
            f"({datetime.utcfromtimestamp(from_ts)} → {datetime.utcfromtimestamp(to_ts)} UTC)"
        )

        for symbol in symbols:
            try:
                # Incremental: start from where we last left off
                cursor = archiver.get_cursor(symbol)
                start_ts = cursor + 60 if cursor else from_ts

                if start_ts >= to_ts:
                    logger.info(f"  {symbol}: already up to date (cursor={cursor})")
                    continue

                all_candles = []
                total = 0

                candle_gen = client.fetch_all_candles(symbol, INTERVAL, start_ts, to_ts)
                for item in candle_gen:
                    if isinstance(item, int):
                        total = item
                    else:
                        all_candles.append(item)

                    # Batch insert every 500 candles to bound memory usage
                    if len(all_candles) >= 500:
                        archiver.upsert_candles(all_candles)
                        all_candles = []

                # Flush remaining
                if all_candles:
                    archiver.upsert_candles(all_candles)

                # Update cursor to the latest candle
                if all_candles:
                    latest = max(c.open_time for c in all_candles)
                    archiver.update_cursor(symbol, latest)
                    logger.info(
                        f"  {symbol}: archived {total} candles "
                        f"(open_time {all_candles[0].open_time} → {latest})"
                    )
                else:
                    logger.info(f"  {symbol}: no new candles in range")

            except Exception as e:
                logger.error(f"  {symbol}: archive failed — {e}")
                # Continue with next symbol; failed symbols retry on next run


if __name__ == "__main__":
    run_daily_archive()

Design Decisions Worth Noting

The pipeline uses a generator-based fetcher (fetch_all_candles) rather than loading all candles into memory at once. For a full backfill covering 5 years of daily data across 200 symbols, holding everything in memory would require gigabytes of RAM. The generator yields candles one at a time, the main loop batches them in groups of 500, and each batch is written to the database before the next batch is fetched. This keeps memory usage bounded regardless of how much data you're archiving.

The lookback_days parameter defaults to 1, which archives only the previous trading day. Setting it to 5 handles long weekends where a holiday skips one or more trading days. Setting it to 365 triggers a full backfill. The cursor table ensures that even if a backfill is interrupted midway through the symbol list, it resumes correctly on the next run.

The date helper (trading_day_offset) is intentionally conservative — it uses a simple weekday check rather than a full holiday calendar. For US equities, the major holidays (New Year's Day, Independence Day, Thanksgiving, Christmas) are the primary non-trading days. A production deployment should integrate a holiday calendar library such as pandas_market_calendars to avoid requesting data for days when the exchange was closed.

---

Incremental Update Logic in Detail

The cursor table is the core of the incremental update mechanism, but its behavior deserves careful explanation because it interacts with several subtle edge cases.

Normal daily run: The cursor for AAPL.US points to the last archived open_time from the previous trading day (e.g., 1700072940, the 3:59 PM ET bar). The pipeline requests from_ts = 1700072940 + 60 = 1700073000, which is the first bar of the new trading day. Only the new day's candles are fetched.

Missed run (weekend gap): If the job failed on Friday and ran successfully on Monday, the cursor still points to Friday's last bar. The pipeline fetches from Friday's close through Monday's open, capturing any weekend-aligned data. For US equities, there is no weekend trading, so this resolves to a single-day fetch, but the logic handles it correctly regardless.

Partial failure mid-symbol: If the pipeline fetches 800 candles for AAPL.US, inserts 500, and then crashes before updating the cursor, the next run starts from the cursor (Friday's last bar), re-fetches the 800 candles, and INSERT OR REPLACE overwrites the 500 already stored. No data is lost. The only cost is a small amount of redundant API calls — an acceptable tradeoff for simplicity.

Missed symbol in a multi-symbol run: If the pipeline successfully archives AAPL.US and MSFT.US but fails on NVDA.US, the cursor for NVDA.US is not updated. The next run fetches NVDA.US's missing candles without re-fetching the others. This is the correct behavior.

---

Scheduling the Pipeline

With the pipeline written, the next step is automation. Two common approaches are covered here.

Option 1: cron (Unix/Linux/macOS)

Add the following entry to your crontab (crontab -e):

# Run at 4:05 AM ET every weekday — markets close at 4:00 PM ET
5 4 * * 1-5  /usr/bin/python3 /opt/archiver/run_archive.py >> /var/log/archiver.log 2>&1

Adjust the path to match your installation. The output redirection captures both stdout and stderr to a log file, which is essential for debugging failed runs. A common gotcha with cron is that the environment differs from your interactive shell — the TICKDB_API_KEY and ARCHIVE_DB_PATH environment variables must be set in the crontab entry or in a wrapper script:

5 4 * * 1-5 \
    TICKDB_API_KEY=tk_live_xxxxxxxxxxxx \
    ARCHIVE_DB_PATH=/data/market_data.db \
    /usr/bin/python3 /opt/archiver/run_archive.py \
    >> /var/log/archiver.log 2>&1

Option 2: systemd Timer (Linux)

For systems running systemd, a timer provides more reliable scheduling, better logging integration, and dependency management:

# /etc/systemd/system/tickdb-archiver.service
[Unit]
Description=TickDB Daily OHLCV Archiver

[Service]
Type=oneshot
ExecStart=/usr/bin/python3 /opt/archiver/run_archive.py
Environment="TICKDB_API_KEY=tk_live_xxxxxxxxxxxx"
Environment="ARCHIVE_DB_PATH=/data/market_data.db"
StandardOutput=journal
StandardError=journal

# /etc/systemd/system/tickdb-archiver.timer
[Unit]
Description=Run TickDB archiver at 4:05 AM ET every weekday

[Timer]
OnCalendar=Mon..Fri 04:05:00
Persistent=true
RandomizedDelaySec=300

[Install]
WantedBy=timers.target

Enable and start the timer:

sudo systemctl daemon-reload
sudo systemctl enable --now tickdb-archiver.timer

The RandomizedDelaySec=300 adds up to 5 minutes of random jitter to the scheduled time. This prevents the thundering herd problem where every user of a public API hits it simultaneously at exactly 4:05 AM. TickDB's rate limits are generous, but jitter is good practice for any external API dependency.

---

Production Considerations

Holiday and Weekend Gaps

The trading_day_offset helper returns the correct base day, but it does not currently cross-reference against an exchange holiday calendar. For a fully robust production system, replace the manual weekday check with pandas_market_calendars:

import pandas_market_calendars as mcal

def is_trading_day(date: datetime) -> bool:
    nyse = mcal.get_calendar("NYSE")
    schedule = nyse.valid_days(start_date=date, end_date=date)
    return len(schedule) > 0

This library handles US federal holiday rules, early closes (e.g., day before Independence Day), and timezone conversions automatically.

Data Validation

Before inserting candles into the database, validate a few invariants:

• high >= open, high >= close, high >= low
• low <= open, low <= close
• volume >= 0
• open_time falls within the requested range

Any candle failing these checks should be logged as an anomaly and skipped, not inserted. A corrupted bar in your historical database can silently corrupt every backtest that depends on it.

Monitoring the Pipeline

A pipeline that runs silently is a pipeline you don't trust. Add three monitoring hooks:

1. Row count alert: If the number of archived candles for a symbol deviates significantly from the expected count (e.g., more than 20% below the 390-minute trading day), send an alert. This catches API response failures that return an empty data object without raising an exception.

2. Cursor staleness alert: If the last_archived timestamp for any symbol is more than two trading days old, alert on-call. This catches scheduler failures where the cron job silently stopped running.

3. API error rate: Track the frequency of non-rate-limit API errors. A spike may indicate a TickDB endpoint issue.

A simple Slack webhook integration:

def send_alert(message: str):
    webhook_url = os.environ.get("SLACK_WEBHOOK_URL")
    if not webhook_url:
        return
    requests.post(
        webhook_url,
        json={"text": f"[Archiver Alert] {message}"},
        timeout=5,
    )

ClickHouse Variant

For teams using ClickHouse, the storage layer (SQLiteArchiver) can be replaced with a ClickHouse client using the same interface. The INSERT INTO ... ON CLUSTER syntax handles deduplication via the ORDER BY key:

from clickhouse_driver import Client

class ClickHouseArchiver:
    def __init__(self, host: str, database: str = "market_data"):
        self.client = Client(host=host, database=database)

    def upsert_candles(self, candles: list[KlineCandle]) -> int:
        if not candles:
            return 0
        rows = [
            (c.symbol, c.open_time, "1m", c.open, c.high, c.low, c.close, c.volume)
            for c in candles
        ]
        self.client.execute(
            """
            INSERT INTO ohlcv_1m
                (symbol, open_time, interval_, open, high, low, close, volume)
            VALUES
            """,
            rows,
        )
        return len(rows)

    def get_cursor(self, symbol: str) -> Optional[int]:
        result = self.client.execute(
            "SELECT last_archived FROM archive_cursor WHERE symbol = %s",
            (symbol,),
        )
        return result[0][0] if result else None

    def update_cursor(self, symbol: str, last_archived: int):
        self.client.execute(
            """
            INSERT INTO archive_cursor (symbol, last_archived, updated_at)
            VALUES (%s, %s, now())
            """,
            (symbol, last_archived),
        )

The pipeline code itself does not change — pass a ClickHouseArchiver instance to run_daily_archive instead of SQLiteArchiver, and the rest of the logic is identical.

---

Querying Your Local Archive

With data flowing into your local database, the natural next step is using it for analysis. A few representative queries demonstrate what becomes possible.

Calculate daily returns from 1-minute bars:

SELECT
    symbol,
    DATE(FROM_UNIXTIME(open_time)) AS trade_date,
    ROUND((MAX(close) - MIN(open)) / MIN(open) * 100, 2) AS intraday_range_pct,
    ROUND((MAX(close) - MIN(open)) / MIN(open) * 100, 2) AS intraday_range_pct,
    SUM(volume) AS total_volume,
    COUNT(*)    AS bar_count
FROM ohlcv_1m
WHERE open_time >= 1700000000
  AND open_time <  1700100000
GROUP BY symbol, trade_date
ORDER BY intraday_range_pct DESC
LIMIT 20;

Detect unusual volume on a per-symbol basis:

WITH avg_volume AS (
    SELECT symbol, AVG(volume) AS mean_vol
    FROM ohlcv_1m
    WHERE open_time >= 1700000000 - 86400 * 30  -- last 30 days
    GROUP BY symbol
)
SELECT
    o.symbol,
    DATE(FROM_UNIXTIME(o.open_time)) AS trade_date,
    o.volume,
    a.mean_vol,
    ROUND(o.volume / a.mean_vol, 2) AS vol_ratio
FROM ohlcv_1m o
JOIN avg_volume a ON o.symbol = a.symbol
WHERE o.volume > a.mean_vol * 5
ORDER BY vol_ratio DESC
LIMIT 20;

These queries run in under a second on a SQLite database with years of data and a proper index. The ORDER BY (symbol, open_time) index on the table means range scans on a single symbol are index-only operations — SQLite does not need to read the table data itself.

---

Closing

The pipeline described here transforms a daily manual task — "go fetch yesterday's data" — into a self-sustaining system that runs unattended, survives failures gracefully, and builds a compounding asset: a clean, local OHLCV database that grows more valuable with every trading day.

The core design principles are not unique to this use case. Any time you are pulling structured data from an external API and need it locally, the same pattern applies: fetch with pagination, store with a deduplication key, track progress with a cursor, and make the pipeline idempotent so that rerunning it never causes data corruption. TickDB's /v1/market/kline endpoint is a well-structured source — consistent timestamps, paginated responses, a clean error code system — which makes implementing this pattern straightforward.

If your data needs grow beyond what a single-machine SQLite database can handle efficiently, the same pipeline architecture swaps cleanly into ClickHouse, PostgreSQL, or any other database that supports INSERT OR REPLACE semantics. The fetch logic, the cursor management, and the deduplication strategy remain identical.

Next Steps

If you want to start archiving today, sign up at tickdb.ai to get a free API key (no credit card required), set the TICKDB_API_KEY environment variable, and run the script in this article. The free tier is sufficient for archiving a handful of symbols with daily 1-minute data.

If you need 10+ years of historical OHLCV for backtesting across a broad universe of symbols, the same archival pipeline handles full backfills — set lookback_days=3650 and let it run. For institutional-grade backfill performance and coverage, contact enterprise@tickdb.ai for plans that include optimized API access and historical data bundles.

If you use AI coding assistants, search for and install the tickdb-market-data SKILL in your AI tool's marketplace. It includes pre-built templates for this archival pipeline, the WebSocket streaming client, and common quantitative analysis queries.

---

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