At 02:13 on a research notebook, a backtest looks spectacular: a smooth equity curve, a Sharpe ratio above 2, and a maximum drawdown smaller than most index funds. Three hours later, the result disappears after the researcher changes the execution price from close to next open.

The backtest was not necessarily wrong. The research specification was incomplete.

Academic strategy replication often fails for the same reason: the paper describes the conclusion in prose, while the implementation requires every missing decision to become an executable rule. Universe membership, point-in-time data, signal timing, transaction costs, missing observations, and benchmark construction can each move the result materially.

For quant researchers and data engineers, the real work begins after the abstract. The task is to translate a research claim into a deterministic system that another researcher can inspect, run, and challenge.

This article provides that translation process. It covers paper triage, strategy specifications, data acquisition, leakage-safe backtesting, result comparison, and production handoff, with complete Python examples for historical market data and optional real-time depth monitoring.

1. From Research Claim to Testable Contract

A reproducible paper starts with a contract that defines exactly what must be implemented.

The contract prevents the researcher from changing assumptions whenever a result fails to match the publication. It also separates three elements that are often confused: the trading hypothesis, the data required to test it, and the execution rules used to simulate the hypothesis.

1.1 Extract the Claims Before the Code

Start by listing every testable claim in the paper.

Paper element Implementation question Required decision
Signal What exact formula produces the signal? Preserve variables, windows, lag rules, and thresholds
Universe Which securities were eligible? Define listing age, liquidity filters, and membership dates
Rebalancing When did positions change? Use calendar-based, monthly, or signal-based schedules
Execution When and at what price did trades occur? Specify close, next open, VWAP, or another execution rule
Costs Which frictions were included? Commission, spread, slippage, borrow cost, and market impact
Benchmark Against what was the strategy compared? Specify index, ticker, weighting, and rebalancing
Sample Which observations formed the test set? Dates, assets, frequency, and exclusion rules

A sentence such as “signals are generated using prior returns” is not yet an implementation specification. It leaves unresolved the lookback length, whether the prior observation excludes the current bar, the direction of the position, and how portfolio weights are formed.

Write these decisions down before touching market data.

1.2 Convert Ambiguous Language into Code Rules

The specification must translate economic prose into deterministic operations.

Research language Executable interpretation
“Based on information available at time (t)” Compute the signal using bars dated no later than (t)
“Trades at the close” Apply the signal generated before the close to that close
“Trades on the following day” Apply the close signal to the next bar’s open
“Monthly rebalancing” Form signals on the final eligible session of each month
“Top-decile stocks” Rank eligible assets before applying liquidity or listing filters
“Transaction costs of 10 bps” Deduct costs on position turnover, not on total portfolio value
“Excludes recent IPOs” Require a documented listing-age threshold
“Risk-free rate of 3%” Apply the rate consistently to excess-return and Sharpe calculations

The key rule is simple: every qualitative term needs an operational definition before implementation.

1.3 Record Reproducibility Metadata

A result that cannot be tied to a specific dataset and configuration is not reproducible.

Record at least the following metadata:

paper_title
paper_version_or_identifier
data_provider
symbol_universe
start_date
end_date
bar_frequency
price_adjustment_policy
signal_parameters
execution_rule
commission_bps
slippage_bps
benchmark
code_version
random_seed
result_generated_at

Keep the configuration in a version-controlled file rather than embedding values throughout the code. Python’s dataclass is sufficient for this purpose.

from dataclasses import asdict, dataclass
from datetime import date
import json


@dataclass(frozen=True)
class ResearchSpec:
    name: str
    start_date: date
    end_date: date
    fast_window: int
    slow_window: int
    commission_bps: float
    slippage_bps: float
    benchmark: str
    execution_rule: str = "next_open"

    def to_json(self) -> str:
        payload = asdict(self)
        payload["start_date"] = self.start_date.isoformat()
        payload["end_date"] = self.end_date.isoformat()
        return json.dumps(payload, indent=2, sort_keys=True)

A frozen configuration prevents accidental changes during a run. It also makes the final report reproducible from the committed spec.

2. Where Replications Quietly Diverge

Most replication failures are not caused by an incorrect signal formula. They arise from undocumented choices around data and execution.

The following table shows why two implementations can look identical while producing materially different portfolios.

Specification gap Interpretation A Interpretation B Consequence
Return timestamp Current close Previous completed close Signal changes by one bar
Execution price Signal-day close Next-session open Large overnight returns enter results
Universe Current constituents Point-in-time constituents Survivorship and selection bias
Delisted securities Excluded Included at terminal value Return distribution changes
Missing bars Forward-filled Dropped Signals use stale or nonexistent prices
Corporate actions Raw prices Adjusted prices Returns and trading thresholds change
Cash return Ignored Included Sharpe and drawdown differ
Costs 10 bps round trip 10 bps per side Turnover cost doubles
Rebalancing Every day Monthly Trade count and exposure differ
Benchmark SPY total return SPY price return Dividend treatment changes
Statistical uncertainty Single point estimate Confidence interval Weak samples appear conclusive

These differences are not academic details. They are part of the strategy.

2.1 Separate Three Layers of Validation

A reliable replication process validates three layers independently.

The signal layer checks whether the implementation computes the same value as the paper’s formula. This layer should use small fixtures with manually calculated expected outputs.

The portfolio layer checks how signals become positions. It verifies weight calculation, turnover, cash, and execution timing.

The metrics layer checks annualization, benchmark alignment, risk statistics, and cost accounting.

Testing the final Sharpe ratio alone cannot identify which layer failed. Separate tests make debugging much faster.

2.2 Establish Acceptance Criteria in Advance

Define acceptable differences before viewing the replication results. Otherwise, a researcher may redefine “correct” after each failed comparison.

Example acceptance criteria might include:

  • Zero look-ahead violations in automated tests.
  • At least 99.9% timestamp alignment between the paper dataset and the replication dataset.
  • Trade-count difference below 5% when the paper reports its universe and rebalancing rule.
  • Annualized return difference below two percentage points.
  • Sharpe ratio difference below 0.20.
  • Maximum drawdown difference below three percentage points.

These thresholds are examples, not universal standards. Tight tolerances may be appropriate for deterministic daily strategies and inappropriate for intraday execution with vendor-specific fills.

3. Data Architecture for Paper Reproduction

The production system should separate market-data acquisition, signal computation, portfolio simulation, and reporting.

flowchart LR
    P[Paper Specification] --> C[Configuration Registry]
    H[Historical REST API] --> D[Data Adapter]
    O[Optional Depth WebSocket] --> D
    D --> V[Schema and Quality Checks]
    V --> S[Signal Engine]
    C --> S
    S --> B[Portfolio Simulator]
    V --> B
    B --> M[Metrics and Trade Statistics]
    M --> R[Paper Comparison Report]
    M --> A[Paper Trading Monitor]
    O --> A

3.1 Historical Data and Live Data Serve Different Purposes

Historical bars support backtests and parameter studies. A live endpoint supports current monitoring but should not replace historical data in a reproducible backtest.

For US equities, a useful separation is:

  • Use the historical GET /v1/market/kline endpoint for completed research periods.
  • Use GET /v1/market/kline/latest for a current candle or live dashboard.
  • Use the depth channel only when the paper’s hypothesis depends on order-book state.

The product’s capability boundaries matter. Historical US equity OHLCV data covers more than 10 years of cleaned and aligned bars. The trades endpoint does not cover US equities or A-shares. US depth coverage is L1; HK equities and crypto support L1 through L10. Forex, precious metals, and indices do not support the depth channel.

A paper requiring full US equity trade history therefore needs a different data source. A strategy requiring only OHLCV can use the historical kline workflow below.

3.2 Normalize Provider Data Into a Canonical Schema

The backtester should not depend on a provider’s raw response format. Introduce an adapter that converts vendor records into a stable internal schema:

timestamp, open, high, low, close, volume

The adapter should perform four checks:

  1. Parse timestamps and normalize them to UTC.
  2. Sort records by timestamp and reject duplicate timestamps.
  3. Validate that high >= max(open, close) and low <= min(open, close).
  4. Identify missing expected bars without silently fabricating returns.

Provider-specific field names belong in the adapter, not in the strategy engine.

3.3 Download Historical Bars with Authentication and Failure Handling

The following script uses TickDB’s REST authentication through the X-API-Key header. It validates the API key, applies request timeouts, honors rate-limit responses, retries transient failures with exponential backoff and jitter, and writes canonical CSV data.

#!/usr/bin/env python3
"""Download historical bars and normalize them for reproducible research."""

from __future__ import annotations

import argparse
import os
import random
import time
from pathlib import Path
from typing import Any

import requests


API_BASE = "https://api.tickdb.ai"
API_KEY = os.environ.get("TICKDB_API_KEY")

FIELD_ALIASES = {
    "timestamp": ("timestamp", "time", "datetime", "date", "t"),
    "open": ("open", "o"),
    "high": ("high", "h"),
    "low": ("low", "l"),
    "close": ("close", "c"),
    "volume": ("volume", "vol", "v"),
}


def retry_delay(response: requests.Response | None, attempt: int) -> float:
    """Use Retry-After when available; otherwise use exponential backoff."""
    if response is not None:
        retry_after = response.headers.get("Retry-After")
        if retry_after:
            try:
                return max(0.0, float(retry_after))
            except ValueError:
                pass

    base = min(2**attempt, 30)
    return base + random.uniform(0, base * 0.1)


def extract_rows(payload: dict[str, Any]) -> list[dict[str, Any]]:
    """Extract rows from common envelope shapes without hiding schema errors."""
    data = payload.get("data")

    if isinstance(data, list):
        return data

    if isinstance(data, dict):
        for key in ("items", "list", "bars", "klines", "records"):
            rows = data.get(key)
            if isinstance(rows, list):
                return rows

    raise ValueError(
        "Could not locate a bar list in the API response. "
        "Inspect the response and update extract_rows() for the provider schema."
    )


def normalize_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Convert provider rows into the canonical research schema."""
    normalized: list[dict[str, Any]] = []

    for row in rows:
        if not isinstance(row, dict):
            continue

        result: dict[str, Any] = {}

        for canonical, aliases in FIELD_ALIASES.items():
            value = next(
                (row[alias] for alias in aliases if alias in row and row[alias] is not None),
                None,
            )
            if value is None:
                raise ValueError(f"Missing {canonical!r} in bar record: {row}")
            result[canonical] = value

        normalized.append(result)

    normalized.sort(key=lambda item: str(item["timestamp"]))

    seen: set[str] = set()
    for row in normalized:
        key = str(row["timestamp"])
        if key in seen:
            raise ValueError(f"Duplicate timestamp found: {key}")
        seen.add(key)

    return normalized


def fetch_bars(
    session: requests.Session,
    symbol: str,
    interval: str,
    limit: int,
    max_attempts: int = 6,
) -> list[dict[str, Any]]:
    if not API_KEY:
        raise RuntimeError("Set TICKDB_API_KEY before running this script.")

    params = {
        "symbol": symbol,
        "interval": interval,
        "limit": limit,
    }
    headers = {"X-API-Key": API_KEY}

    for attempt in range(max_attempts):
        response = None

        try:
            response = session.get(
                f"{API_BASE}/v1/market/kline",
                headers=headers,
                params=params,
                timeout=(3.05, 15),
            )
        except (requests.Timeout, requests.ConnectionError) as exc:
            wait = retry_delay(None, attempt)
            print(f"Transient network error: {exc}. Retrying in {wait:.2f}s.")
            time.sleep(wait)
            continue

        if response.status_code == 429:
            wait = retry_delay(response, attempt)
            print(f"HTTP rate limit. Retrying in {wait:.2f}s.")
            time.sleep(wait)
            continue

        if 500 <= response.status_code < 600:
            wait = retry_delay(response, attempt)
            print(f"HTTP {response.status_code}. Retrying in {wait:.2f}s.")
            time.sleep(wait)
            continue

        try:
            payload = response.json()
        except ValueError as exc:
            raise RuntimeError(f"API returned non-JSON content: {response.text[:200]}") from exc

        code = payload.get("code", 0)

        if code == 0:
            return normalize_rows(extract_rows(payload))

        if code in (1001, 1002):
            raise RuntimeError("Invalid or missing API key. Check TICKDB_API_KEY.")

        if code == 2002:
            raise KeyError(f"Symbol {symbol!r} was not found. Verify the symbol identifier.")

        if code == 3001:
            wait = retry_delay(response, attempt)
            print(f"API rate limit. Retrying in {wait:.2f}s.")
            time.sleep(wait)
            continue

        raise RuntimeError(
            f"Unexpected API error {code}: {payload.get('message', 'unknown error')}"
        )

    raise RuntimeError(f"Maximum attempts reached while downloading {symbol}.")


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--symbol", required=True, help="Example: AAPL.US")
    parser.add_argument("--interval", default="1d")
    parser.add_argument("--limit", type=int, default=10000)
    parser.add_argument("--output", required=True, help="Destination CSV path")
    args = parser.parse_args()

    session = requests.Session()
    rows = fetch_bars(session, args.symbol, args.interval, args.limit)

    if not rows:
        raise RuntimeError("The API returned no usable bars.")

    output = Path(args.output)
    output.parent.mkdir(parents=True, exist_ok=True)

    fieldnames = ["timestamp", "open", "high", "low", "close", "volume"]
    with output.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)

    print(f"Wrote {len(rows)} bars to {output}")


if __name__ == "__main__":
    import csv
    main()

Set the key before running the script:

export TICKDB_API_KEY="your_api_key"
python download_bars.py \
  --symbol AAPL.US \
  --interval 1d \
  --limit 10000 \
  --output data/AAPL_US_1d.csv

The script intentionally fails when it cannot normalize the response. A missing-field error is preferable to a backtest running on silently malformed data.

3.4 Validate Data Before Calculating Returns

A basic quality gate should answer these questions:

  • Does the date range match the paper?
  • Are timestamps sorted and unique?
  • Are there unexpected non-trading days?
  • Are zero-volume bars present?
  • Are adjusted and unadjusted prices mixed?
  • Are missing values represented as null, zero, or forward-filled values?
  • Does the benchmark share the same calendar as the strategy universe?

At minimum, report the number of observations and the number of missing expected bars.

import pandas as pd


def validate_bars(path: str) -> pd.DataFrame:
    df = pd.read_csv(path, parse_dates=["timestamp"])
    df = df.sort_values("timestamp").reset_index(drop=True)

    if df["timestamp"].duplicated().any():
        raise ValueError("Duplicate timestamps detected.")

    numeric = ["open", "high", "low", "close", "volume"]
    if df[numeric].isna().any().any():
        raise ValueError("Missing values detected in canonical OHLCV fields.")

    invalid = (
        (df["high"] < df[["open", "close"]].max(axis=1))
        | (df["low"] > df[["open", "close"]].min(axis=1))
        | (df["high"] < df["low"])
    )
    if invalid.any():
        raise ValueError(f"Invalid OHLC relationships on {invalid.sum()} rows.")

    return df

4. Implement the Signal Without Look-Ahead

The signal engine should operate on information available at the decision timestamp. The portfolio engine then applies an explicit execution rule.

A common error is to calculate a moving average using the current close, use it to trade at the current close, and describe the strategy as if the trader knew the closing price before trading at that price.

4.1 Separate Signal Time from Execution Time

For a daily strategy:

  1. Calculate the signal using data through bar (t).
  2. Shift the desired position forward by one bar.
  3. Apply the position to returns beginning at (t+1).
  4. Charge turnover costs when the position changes.

This is only one possible convention. If the paper explicitly trades at the next open, calculate the portfolio return using the next open-to-close return instead. The implementation must match the paper rather than the convenient default.

4.2 Use an Explicit Reference Strategy

The following example implements a moving-average crossover as a complete reference backtest. It is not presented as a reproduction of any specific academic paper; it demonstrates the mechanics required when translating a paper’s signal into code.

The script reports net return, annualized return, Sharpe ratio, Sortino ratio, maximum drawdown, profit factor, win rate, trade count, and a buy-and-hold benchmark.

#!/usr/bin/env python3
"""Leakage-aware daily moving-average crossover backtest."""

from __future__ import annotations

import argparse
import json
from pathlib import Path

import numpy as np
import pandas as pd


TRADING_DAYS = 252


def load_csv(path: str) -> pd.DataFrame:
    df = pd.read_csv(path, parse_dates=["timestamp"])
    df = df.sort_values("timestamp").reset_index(drop=True)

    required = {"timestamp", "open", "high", "low", "close"}
    missing = required.difference(df.columns)
    if missing:
        raise ValueError(f"Missing required columns: {sorted(missing)}")

    return df


def calculate_metrics(returns: pd.Series) -> dict[str, float]:
    returns = returns.dropna()
    if returns.empty:
        raise ValueError("No returns available.")

    equity = (1.0 + returns).cumprod()
    drawdown = equity / equity.cummax() - 1.0

    periods = len(returns)
    total_return = float(equity.iloc[-1] - 1.0)
    annualized_return = float((1.0 + total_return) ** (TRADING_DAYS / periods) - 1.0)

    volatility = float(returns.std(ddof=1))
    sharpe = (
        float(returns.mean() / volatility * np.sqrt(TRADING_DAYS))
        if volatility > 0
        else 0.0
    )

    downside = returns[returns < 0]
    downside_deviation = float(downside.std(ddof=1)) if len(downside) > 1 else 0.0
    sortino = (
        float(returns.mean() / downside_deviation * np.sqrt(TRADING_DAYS))
        if downside_deviation > 0
        else 0.0
    )

    gains = returns[returns > 0].sum()
    losses = abs(returns[returns < 0].sum())
    profit_factor = float(gains / losses) if losses > 0 else float("inf")
    win_rate = float((returns > 0).mean())

    return {
        "total_return": total_return,
        "annualized_return": annualized_return,
        "annualized_volatility": volatility * np.sqrt(TRADING_DAYS),
        "sharpe": sharpe,
        "sortino": sortino,
        "max_drawdown": float(drawdown.min()),
        "profit_factor": profit_factor,
        "win_rate": win_rate,
        "observations": int(periods),
    }


def run_backtest(
    data: pd.DataFrame,
    benchmark: pd.DataFrame | None,
    fast_window: int,
    slow_window: int,
    cost_bps: float,
) -> dict:
    if fast_window >= slow_window:
        raise ValueError("fast_window must be smaller than slow_window.")

    df = data.copy()
    df["fast_ma"] = df["close"].rolling(fast_window).mean()
    df["slow_ma"] = df["close"].rolling(slow_window).mean()

    # A signal becomes tradable only after the current bar closes.
    df["raw_position"] = (df["fast_ma"] > df["slow_ma"]).astype(float)
    df["position"] = df["raw_position"].shift(1).fillna(0.0)

    df["asset_return"] = df["close"].pct_change().fillna(0.0)
    df["turnover"] = df["position"].diff().abs().fillna(df["position"].abs())
    df["cost"] = df["turnover"] * (cost_bps / 10_000.0)
    df["strategy_return"] = df["position"] * df["asset_return"] - df["cost"]

    results = {
        "strategy": calculate_metrics(df["strategy_return"]),
        "trade_count": int((df["position"].diff().fillna(df["position"]) != 0).sum()),
        "parameters": {
            "fast_window": fast_window,
            "slow_window": slow_window,
            "cost_bps": cost_bps,
        },
    }

    if benchmark is not None:
        bench = benchmark[["timestamp", "close"]].copy()
        merged = df[["timestamp", "strategy_return"]].merge(
            bench,
            on="timestamp",
            how="inner",
        )
        benchmark_returns = merged["close"].pct_change().fillna(0.0)
        results["benchmark"] = calculate_metrics(benchmark_returns)
        results["benchmark_observations"] = int(len(merged))

    return results


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--data", required=True)
    parser.add_argument("--benchmark")
    parser.add_argument("--fast", type=int, default=20)
    parser.add_argument("--slow", type=int, default=100)
    parser.add_argument("--cost-bps", type=float, default=10.0)
    parser.add_argument("--output", default="results/backtest.json")
    args = parser.parse_args()

    data = load_csv(args.data)
    benchmark = load_csv(args.benchmark) if args.benchmark else None

    results = run_backtest(
        data=data,
        benchmark=benchmark,
        fast_window=args.fast,
        slow_window=args.slow,
        cost_bps=args.cost_bps,
    )

    output = Path(args.output)
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(json.dumps(results, indent=2), encoding="utf-8")

    print(json.dumps(results, indent=2))


if __name__ == "__main__":
    main()

Example execution:

python backtest.py \
  --data data/AAPL_US_1d.csv \
  --fast 20 \
  --slow 100 \
  --cost-bps 10 \
  --output results/AAPL_crossover.json

The important engineering decision is not the crossover itself. It is the separation between a signal calculated at the end of a bar and a position applied only after that information becomes actionable.

5. Reproduce the Portfolio Mechanics

A signal is not a portfolio. The portfolio layer converts signals into positions and cash.

5.1 Model Position Sizing Explicitly

For a multi-asset paper, determine whether returns are:

  • Equal-weighted.
  • Inverse-volatility weighted.
  • Market-cap weighted.
  • Signal-strength weighted.
  • Ranked into fixed quantiles.
  • Subject to per-asset caps.
  • Subject to turnover or leverage constraints.

A simple equal-weight implementation might divide capital equally at each rebalance:

def equal_weight_weights(symbols: list[str]) -> dict[str, float]:
    if not symbols:
        return {}

    weight = 1.0 / len(symbols)
    return {symbol: weight for symbol in symbols}

That function is correct only if the paper uses equal weighting. Do not introduce equal weighting because it is easy; introduce it because the specification requires it.

5.2 Make Costs Directional

Costs should depend on the position transition:

[
\text{Turnover}t = |w_t - w{t-1}|
]

A practical cost model is:

[
\text{Cost}_t

\text{Turnover}t
\times
\left(
\frac{\text{commission}{\text{bps}}}{10{,}000}
+
\frac{\text{slippage}_{\text{bps}}}{10{,}000}
\right)
]

Separate these concepts in the configuration even if the first implementation combines them. Commission, spread, and market impact behave differently under stress, so researchers must be able to change them independently.

5.3 Preserve Corporate-Action Assumptions

A daily strategy using adjusted prices needs a consistent interpretation of adjusted volume and turnover. A futures strategy needs contract rolls and collateral treatment. A cross-sectional equity strategy needs point-in-time membership and delisting returns.

The principle is universal: state what the return represents. If the paper measures close-to-close total return, reproduce that convention. Do not mix adjusted close-to-close returns with unadjusted execution prices.

6. Compare the Reproduction with the Paper

Result comparison should use a structured report rather than a visual inspection of two equity curves.

6.1 Compare Like With Like

At minimum, report the following dimensions:

Dimension Paper Reproduction Difference
Start date — — —
End date — — —
Observations — — —
Universe — — —
Rebalancing frequency — — —
Total return — — —
Annualized return — — —
Annualized volatility — — —
Sharpe ratio — — —
Sortino ratio — — —
Profit factor — — —
Win rate — — —
Maximum drawdown — — —
Number of trades — — —
Total costs — — —
Benchmark — — —

Do not fill this table until the dataset, period, benchmark, and cost assumptions have been aligned.

6.2 Classify the Mismatch

A useful diagnostic process categorizes each discrepancy:

  • Data mismatch: different assets, dates, timestamps, or adjustment policies.
  • Signal mismatch: different calculations, windows, lags, or thresholds.
  • Portfolio mismatch: different weighting, rebalancing, or constraints.
  • Execution mismatch: different fill timing or price.
  • Cost mismatch: different commission, spread, or slippage.
  • Statistical mismatch: a small sample or parameter instability despite equivalent implementation.

This classification prevents endless code changes. If the datasets differ, modifying the signal is not a valid response.

6.3 Use a Small Set of Diagnostic Tests

Useful diagnostic comparisons include:

  • Signal correlation by timestamp.
  • Position overlap between paper and replication.
  • Daily return correlation.
  • Monthly return correlation.
  • Cumulative return divergence by quarter.
  • Turnover and cost comparison.
  • Results with and without costs.
  • Results across subperiods.
  • Results under alternative but reasonable execution rules.

If the signal matches but returns do not, inspect the portfolio construction layer. If both signal and positions match, inspect execution and metric calculations.

7. Add Real-Time Depth Only When the Paper Requires It

Many academic strategies require only historical prices. Adding a WebSocket feed to every project increases operational complexity without improving the replication.

Depth becomes necessary when the hypothesis explicitly references:

  • Bid-ask spread.
  • Top-of-book imbalance.
  • Queue depth.
  • Liquidity withdrawal.
  • Trade-to-book imbalance.
  • Order cancellation behavior.
  • Short-horizon execution conditions.

7.1 Use Depth as an Optional Ingestion Path

The following example maintains a WebSocket connection, sends application-level heartbeats, detects silent connections, handles API rate-limit messages, and reconnects with exponential backoff and jitter.

Set TICKDB_WS_URL to the WebSocket endpoint shown in the account documentation, and set TICKDB_SUBSCRIBE to the required subscription message.

#!/usr/bin/env python3
"""Monitor depth updates with heartbeat and resilient reconnection."""

from __future__ import annotations

import asyncio
import json
import os
import random
from urllib.parse import quote

import websockets


API_KEY = os.environ.get("TICKDB_API_KEY")
WS_URL = os.environ.get("TICKDB_WS_URL")
SUBSCRIBE = json.loads(
    os.environ.get(
        "TICKDB_SUBSCRIBE",
        json.dumps(
            {
                "cmd": "subscribe",
                "channel": "depth",
                "symbol": "AAPL.US",
            }
        ),
    )
)

HEARTBEAT_SECONDS = 15
RECEIVE_TIMEOUT_SECONDS = 45


async def receive_with_heartbeat(websocket) -> str:
    await websocket.send(json.dumps({"cmd": "ping"}))

    response = await asyncio.wait_for(
        websocket.recv(),
        timeout=RECEIVE_TIMEOUT_SECONDS,
    )

    try:
        payload = json.loads(response)
    except (TypeError, json.JSONDecodeError):
        return response

    if isinstance(payload, dict) and payload.get("code") == 3001:
        retry_after = payload.get("retry_after", 5)
        try:
            delay = float(retry_after)
        except (TypeError, ValueError):
            delay = 5.0
        await asyncio.sleep(delay)
        return response

    return response


async def stream_depth() -> None:
    if not API_KEY:
        raise RuntimeError("Set TICKDB_API_KEY before running this script.")
    if not WS_URL:
        raise RuntimeError("Set TICKDB_WS_URL to the TickDB WebSocket endpoint.")

    separator = "&" if "?" in WS_URL else "?"
    authenticated_url = f"{WS_URL}{separator}api_key={quote(API_KEY, safe='')}"

    attempt = 0

    while True:
        try:
            async with websockets.connect(
                authenticated_url,
                open_timeout=10,
                ping_interval=20,
                ping_timeout=10,
                close_timeout=5,
                max_queue=1000,
            ) as websocket:
                attempt = 0

                await websocket.send(json.dumps(SUBSCRIBE))
                print(f"Subscribed: {SUBSCRIBE}")

                while True:
                    message = await receive_with_heartbeat(websocket)
                    print(message)

        except asyncio.CancelledError:
            raise
        except Exception as exc:
            base = min(2**attempt, 30)
            delay = base + random.uniform(0, base * 0.1)
            attempt += 1

            print(f"WebSocket error: {exc}. Reconnecting in {delay:.2f}s.")
            await asyncio.sleep(delay)


if __name__ == "__main__":
    try:
        asyncio.run(stream_depth())
    except KeyboardInterrupt:
        print("Depth monitor stopped.")

The example does not transform the depth payload into a strategy signal. That separation is deliberate. First verify that the live data matches the paper’s definitions. Only then calculate imbalance or pressure ratios.

For example, a top-of-book pressure ratio can be defined as:

[
\text{PressureRatio}_t

\frac{\text{Bid Size}{t,1}}
{\text{Ask Size}{t,1}}
]

A higher ratio represents more displayed size on the bid than the ask. It does not prove that buyers control the market, and it can be spoofed. Treat it as an observable variable, not a causal explanation.

8. Document and Promote the Replication

A replication becomes research infrastructure only when another researcher can rerun it without asking the author for missing context.

8.1 Package the Project

A minimal reproducible repository should contain:

quant-replication/
├── README.md
├── pyproject.toml
├── config/
│   └── experiment.yaml
├── data/
│   └── README.md
├── src/
│   ├── data_adapter.py
│   ├── signals.py
│   ├── portfolio.py
│   ├── backtest.py
│   └── reporting.py
├── tests/
│   ├── test_signals.py
│   ├── test_costs.py
│   └── test_data_adapter.py
├── scripts/
│   └── download_bars.py
└── results/
    └── manifest.json

Avoid committing API keys or licensed data. Store only symbols, date ranges, checksums, and reproducible metadata.

8.2 Add Data Provenance

For every downloaded dataset, record:

  • Retrieval timestamp.
  • API endpoint and parameters.
  • Symbol list.
  • Date range.
  • Bar frequency.
  • Row count.
  • Schema version.
  • SHA-256 checksum.
  • Adjustment policy.
  • Missing-bar policy.

A checksum turns “the data changed” into a testable statement.

import hashlib
from pathlib import Path


def sha256_file(path: str) -> str:
    digest = hashlib.sha256()

    with Path(path).open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)

    return digest.hexdigest()

8.3 Automate the Acceptance Pipeline

A publication-grade replication should run in this order:

Install dependencies
      ↓
Load committed configuration
      ↓
Validate dataset checksums
      ↓
Run unit tests
      ↓
Generate signals
      ↓
Simulate portfolio
      ↓
Calculate metrics
      ↓
Compare with acceptance thresholds
      ↓
Write result manifest

Use a fixed environment for authoritative runs. A container or locked Python dependency set reduces the chance that a library upgrade changes a result.

8.4 Match Deployment Scope to the Reader

User segment Recommended environment Historical data Live monitoring Primary priority
Individual researcher Local Python environment CSV or Parquet cache Optional dashboard Specification clarity
Quant team Version-controlled repository and CI Versioned data warehouse Shared WebSocket service Reproducible runs
Institutional platform Containerized jobs and immutable artifacts Central data lake Low-latency monitoring service Auditability and controls

The same principle applies at every level: the research specification should be more durable than the code that executes it.

9. A Practical Replication Workflow

The complete process can be condensed into eight steps.

Step 1: Classify the Paper

Determine whether the strategy requires bars, trades, order-book depth, fundamentals, options data, or multiple data types. This prevents selecting an inadequate dataset.

Step 2: Extract the Testable Claims

Record the signal, universe, period, execution rule, costs, benchmark, and reported statistics.

Step 3: Freeze the Specification

Resolve ambiguous language before running the implementation.

Step 4: Build a Minimal Baseline

Implement a simple version of the signal and verify a handful of observations manually.

Step 5: Acquire and Validate Data

Download the required history, normalize the schema, document provenance, and run quality checks.

Step 6: Reproduce the Portfolio

Implement position sizing, rebalancing, execution timing, and costs.

Step 7: Compare and Diagnose

Align the sample, benchmark, and assumptions before comparing metrics.

Step 8: Stress the Result

Test alternative costs, parameter neighborhoods, subperiods, and execution conventions. A result that collapses under small assumption changes should be described as fragile.

Backtest Disclosure Checklist

Before publishing or presenting a replication, disclose:

Item Required value
Period Exact start and end dates
Sample size Number of assets, observations, and events
Return basis Price return or total return
Execution assumption Exact fill timing and price
Commission Basis points per side
Slippage Basis points and whether spread is separate
Benchmark Exact security or index
Win rate Gross and net of costs
Profit factor Gross and net of costs
Sharpe ratio Annualization factor and risk-free assumption
Sortino ratio Downside definition
Maximum drawdown Percentage and duration
Limitations Look-ahead, survivorship, liquidity, and sample concerns

A suitable disclosure reads:

Backtest limitations: The results are based on historical simulation and do not guarantee future performance. The model approximates transaction costs using fixed commission and slippage assumptions, does not fully model liquidity exhaustion or market impact, and remains subject to data availability and sample-selection limitations. Validate the strategy out of sample before live deployment.

10. The Real Deliverable Is Traceability

The best paper replication is not the one whose equity curve most closely resembles the published chart. It is the one that allows another researcher to identify exactly why each result occurs.

Traceability requires five things:

  1. A frozen research specification.
  2. Versioned, validated data.
  3. A signal implementation with deterministic tests.
  4. A portfolio simulator that models timing and costs.
  5. A comparison report that distinguishes data, signal, execution, and statistical differences.

Once those elements exist, the strategy becomes more than a notebook result. It becomes an inspectable engineering asset that can move into paper trading, live monitoring, or further research without losing its assumptions.

Next Steps

If you are an individual quantitative researcher, begin with a paper that uses daily OHLCV data and write a one-page execution contract before writing the strategy code.

If you want to reproduce a strategy with live market data, create a TickDB API key at tickdb.ai, configure TICKDB_API_KEY, and use the historical downloader to create a canonical dataset.

If you need more than 10 years of aligned US equity OHLCV for institutional research, contact [email protected] for Professional and Enterprise data options.

If you use AI coding assistants, search for and install the tickdb-market-data SKILL in your tool’s marketplace to accelerate market-data integration while retaining the same authentication and production safeguards.

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