"Price is the effect. The order book is the cause."

A quant researcher runs a mean-reversion strategy on three years of historical data. The strategy returns 23% annualized with a Sharpe of 1.8. They paper trade it for six months. It loses 12%.

The strategy did not change. The market regime did not shift dramatically. The culprit is almost always the same: dirty data.

Data quality is the invisible tax on quantitative research. A single corporate action — a stock split, a dividend, a rights issue — can generate phantom returns of 50% or more if not handled correctly. A 200-millisecond timestamp misalignment between two exchanges can manufacture an arbitrage signal that evaporates the moment you deploy capital. A single outlier — a print error from a data vendor, a fat-finger trade — can distort your volatility estimate by an order of magnitude.

This article dissects what TickDB's "清洗对齐" (cleaning and alignment) process actually does. We compare raw market data against processed data across five dimensions: split and dividend adjustments, timestamp normalization, anomaly detection, missing data interpolation, and cross-market temporal alignment. We show the algorithms used, the edge cases handled, and the code that lets you verify the process yourself.

The Five-Dimension Comparison Framework

Before diving into implementation, we need a common framework for evaluating data quality. TickDB's cleaning pipeline addresses five distinct problems that appear in raw market data feeds. Not every data source suffers from all five, but the ones that do not still benefit from explicit processing.

Dimension Problem in raw data TickDB's solution Verification endpoint
Corporate actions Splits, dividends, and mergers corrupt price continuity Split-adjusted close; dividend-adjusted returns /v1/reference/splits
Timestamp alignment Exchange-reported times vary by venue, protocol, and timezone UTC-normalized timestamps with millisecond precision Raw vs. adjusted comparison
Anomaly detection Print errors, fat-finger trades, exchange corrections Statistical outlier detection with configurable thresholds /v1/market/anomalies
Missing data Gaps from exchange outages, data vendor failures, or holiday schedules Intelligent gap detection with session-boundary awareness /v1/market/kline with ignore_gap parameter
Cross-market sync US equity and futures markets open at different times; crypto trades 24/7 Session-window tagging and phase-aware gap filling Symbol metadata endpoint

Each dimension has its own engineering complexity. We will examine each in turn.

Dimension 1: Corporate Action Adjustments (复权)

The Problem

Corporate actions are events that change a company's share count, price, or capital structure. They include stock splits, reverse splits, stock dividends, cash dividends, mergers, acquisitions, and rights offerings. Without adjustment, historical price series exhibit discontinuities that have nothing to do with market forces.

Consider a 2-for-1 stock split on January 15. Before the split, the stock trades at $200. After the split, it opens at $100. A backtest that uses raw prices will show a 50% drop on January 15 — not because the market crashed, but because the accounting changed.

Dividends are subtler. A $3 quarterly dividend represents a real transfer of value from the company to shareholders. When a stock goes ex-dividend, the price drops by approximately the dividend amount. A naive return calculation that does not account for this will systematically overstate returns on dividend-paying stocks, especially over long holding periods.

TickDB's Adjustment Methodology

TickDB applies split-adjusted closes for price continuity and dividend-adjusted returns for return calculations. The distinction matters.

For price series, TickDB multiplies all pre-split prices by the split ratio. For a 2-for-1 split, every price before January 15 is multiplied by 0.5. This makes the series continuous — a $200 price and a $100 price represent the same proportional ownership.

For return calculations, TickDB uses adjusted close prices that account for both splits and dividends. The formula:

adjusted_close[t] = raw_close[t] * cum_adj_factor[t]

where cum_adj_factor[t] incorporates all split and dividend adjustments from time t forward.

import os
import requests

def fetch_splits_and_dividends(symbol: str, start_date: str, end_date: str) -> dict:
    """
    Fetch corporate action adjustments for a given symbol and date range.
    Returns split events and dividend events with their adjustment factors.
    
    API reference: GET /v1/reference/corporate-actions
    Authentication: X-API-Key header
    """
    api_key = os.environ.get("TICKDB_API_KEY")
    if not api_key:
        raise ValueError("TICKDB_API_KEY environment variable not set")
    
    headers = {"X-API-Key": api_key}
    params = {
        "symbol": symbol,
        "start_date": start_date,
        "end_date": end_date,
        "action_type": "all"  # Options: all, split, dividend, merger
    }
    
    response = requests.get(
        "https://api.tickdb.ai/v1/reference/corporate-actions",
        headers=headers,
        params=params,
        timeout=(3.05, 10)
    )
    
    response.raise_for_status()
    data = response.json()
    
    if data.get("code") != 0:
        raise RuntimeError(f"API error {data.get('code')}: {data.get('message')}")
    
    return data.get("data", {})

def compute_split_adjusted_prices(raw_prices: list[dict], splits: list[dict]) -> list[dict]:
    """
    Compute split-adjusted price series.
    
    For each price point, traverse splits forward in time and accumulate
    the adjustment factor. A split of 2:1 means pre-split prices get multiplied
    by 0.5.
    """
    # Sort splits chronologically
    sorted_splits = sorted(splits, key=lambda x: x["effective_date"])
    
    adjusted_prices = []
    for price_record in raw_prices:
        price_date = price_record["timestamp"][:10]  # Extract YYYY-MM-DD
        
        # Calculate cumulative adjustment factor
        cum_factor = 1.0
        for split in sorted_splits:
            if split["effective_date"] <= price_date:
                # For a n:m split, adjustment factor is m/n
                cum_factor *= (split["new_shares"] / split["old_shares"])
        
        adjusted_prices.append({
            "timestamp": price_record["timestamp"],
            "raw_close": price_record["close"],
            "adjusted_close": round(price_record["close"] * cum_factor, 4),
            "adjustment_factor": round(cum_factor, 6)
        })
    
    return adjusted_prices

# Example usage
if __name__ == "__main__":
    try:
        actions = fetch_splits_and_dividends("AAPL.US", "2023-01-01", "2024-01-01")
        splits = [a for a in actions.get("events", []) if a["type"] == "split"]
        
        print(f"Found {len(splits)} split events for AAPL.US")
        for split in splits:
            print(f"  {split['effective_date']}: {split['old_shares']}-for-{split['new_shares']} split")
    except Exception as e:
        print(f"Error: {e}")

Edge Cases Handled

  1. Reverse splits: A 1-for-10 reverse split means pre-split prices are multiplied by 10. TickDB handles this explicitly rather than treating it as a split with a fractional ratio.

  2. Stock dividends: A 5% stock dividend is equivalent to a 21-for-20 split. TickDB converts all stock dividends to equivalent split ratios.

  3. Merger-adjusted prices: For acquisitions, TickDB uses the acquisition price as the adjustment anchor. Pre-merger prices are adjusted so that the merger announcement date shows a clean discontinuity rather than a spurious gap.

  4. Spinoffs: When a company spins off a subsidiary, both the parent and the subsidiary receive adjustment factors. The combined value of parent plus subsidiary post-spinoff should equal the pre-spinoff adjusted value of the parent.

Dimension 2: Timestamp Normalization (时间戳对齐)

The Problem

Market data arrives with timestamps from three different worlds:

  1. Exchange timestamps: When the trade or quote actually occurred on the exchange matching engine.
  2. Vendor timestamps: When the data vendor's system received and processed the data.
  3. Local timestamps: When your application processes the data.

The gap between exchange timestamp and vendor timestamp can range from milliseconds to seconds. The gap between vendor timestamp and your local timestamp depends on your network path and processing latency.

More critically, different exchanges report in different timezones and different precisions. The NYSE reports in US Eastern Time. The Tokyo Stock Exchange reports in Japan Standard Time (UTC+9). Crypto exchanges report in UTC. A naive join of data from multiple sources will produce phantom arbitrage opportunities that do not exist.

TickDB's Timestamp Strategy

TickDB normalizes all timestamps to UTC at millisecond precision. The normalization process follows three rules:

  1. Exchange-native normalization: TickDB maintains a per-symbol timezone database. Every symbol is tagged with its exchange's local timezone and trading session hours.

  2. DST-aware conversion: US equity markets observe daylight saving time. A timestamp of "09:30:00 EST" and "09:30:00 EDT" represent the same clock time but different UTC offsets. TickDB converts both to UTC correctly.

  3. Precision alignment: If a data source provides second-level precision, TickDB pads to millisecond precision with 000. If a source provides sub-millisecond precision, TickDB truncates to milliseconds.

from datetime import datetime, timezone, timedelta
from typing import Optional

# Exchange timezone database (simplified)
EXCHANGE_TIMEZONES = {
    "US": "America/New_York",      # NYSE, NASDAQ
    "HK": "Asia/Hong_Kong",         # HKEX
    "CN": "Asia/Shanghai",          # SSE, SZSE
    "JP": "Asia/Tokyo",             # TSE
    "UK": "Europe/London",          # LSE
    "CRYPTO": "UTC",                # Binance, Coinbase, etc.
}

def normalize_to_utc(
    timestamp: str,
    exchange_code: str,
    original_format: str = "%Y-%m-%d %H:%M:%S"
) -> str:
    """
    Normalize a timestamp to UTC, handling exchange-specific timezones
    and daylight saving time transitions.
    
    Args:
        timestamp: Original timestamp string from the data source
        exchange_code: Exchange identifier (US, HK, CN, JP, UK, CRYPTO)
        original_format: strftime format of the input timestamp
    
    Returns:
        UTC-normalized timestamp in ISO 8601 format with milliseconds
    """
    tz_name = EXCHANGE_TIMEZONES.get(exchange_code, "UTC")
    
    # Parse the original timestamp
    naive_dt = datetime.strptime(timestamp, original_format)
    
    # For non-UTC exchanges, we need the timezone-aware version
    if tz_name != "UTC":
        import zoneinfo
        tz = zoneinfo.ZoneInfo(tz_name)
        local_dt = naive_dt.replace(tzinfo=tz)
        utc_dt = local_dt.astimezone(timezone.utc)
    else:
        utc_dt = naive_dt.replace(tzinfo=timezone.utc)
    
    # Return in ISO 8601 format with milliseconds
    return utc_dt.strftime("%Y-%m-%dT%H:%M:%S.000+00:00")

def validate_timestamp_monotonicity(kline_data: list[dict]) -> list[dict]:
    """
    Verify that a kline (candlestick) series has monotonically increasing
    timestamps. Returns a list of gaps or overlaps detected.
    
    This is critical for backtesting — non-monotonic timestamps can cause
    look-ahead bias or data leakage.
    """
    anomalies = []
    
    for i in range(1, len(kline_data)):
        prev_ts = datetime.fromisoformat(kline_data[i-1]["timestamp"].replace("Z", "+00:00"))
        curr_ts = datetime.fromisoformat(kline_data[i]["timestamp"].replace("Z", "+00:00"))
        
        if curr_ts <= prev_ts:
            anomalies.append({
                "index": i,
                "type": "non_monotonic",
                "previous": kline_data[i-1]["timestamp"],
                "current": kline_data[i]["timestamp"],
                "delta_ms": (curr_ts - prev_ts).total_seconds() * 1000
            })
        
        # Check for gaps larger than expected interval
        expected_interval = timedelta(hours=1)  # Example: 1h klines
        actual_delta = curr_ts - prev_ts
        
        if actual_delta > expected_interval * 1.5:  # 50% tolerance
            anomalies.append({
                "index": i,
                "type": "gap",
                "previous": kline_data[i-1]["timestamp"],
                "current": kline_data[i]["timestamp"],
                "gap_ms": (actual_delta - expected_interval).total_seconds() * 1000
            })
    
    return anomalies

# Example validation
sample_klines = [
    {"timestamp": "2024-01-15T09:30:00.000-05:00", "close": 185.42},
    {"timestamp": "2024-01-15T10:30:00.000-05:00", "close": 186.10},
    {"timestamp": "2024-01-15T11:30:00.000-05:00", "close": 185.88},
    # Simulate an anomaly
    {"timestamp": "2024-01-15T11:30:00.000-05:00", "close": 185.75},  # Duplicate timestamp
]

anomalies = validate_timestamp_monotonicity(sample_klines)
print(f"Found {len(anomalies)} timestamp anomalies")
for a in anomalies:
    print(f"  {a['type']}: index={a['index']}, delta={a['delta_ms']}ms")

DST Edge Case: The Spring Forward Problem

US equity markets open at 09:30 ET every regular trading day. In March, when DST begins, the market opens at 09:30 EDT instead of 09:30 EST. The UTC time of the open shifts from 14:30 UTC to 13:30 UTC.

A strategy that expects the market to open at exactly 14:30 UTC will fail silently for two weeks in March. TickDB's session tagging explicitly marks which timestamps are affected by DST transitions, so you can filter or adjust accordingly.

Dimension 3: Anomaly Detection (异常值检测)

The Problem

Raw market data contains several categories of anomalous prints:

  1. Exchange errors: Fat-finger trades, matching engine glitches, test prints that leaked into production feeds.
  2. Data vendor errors: Misaligned fields, duplicate records, records with wrong symbols.
  3. Corporate events: Trading halts, limit-up/limit-down hits, options expiration pinning.
  4. Liquidity artifacts: Bid-ask bounce at market open, end-of-day cleanup prints.

Anomalous data points distort statistical estimates. A single 10-standard-deviation print can inflate your volatility estimate by 300%. A stale quote that never gets updated can trigger false breakout signals.

TickDB's Detection Algorithms

TickDB uses a multi-stage anomaly detection pipeline:

Stage 1: Range checks

  • Price outside [0.01 * prev_close, 100 * prev_close]
  • Volume of 0 or negative
  • Bid >= Ask (invalid quote)

Stage 2: Statistical outlier detection

  • Price deviation from rolling median exceeds 5 standard deviations
  • Volume exceeds 20x the rolling median volume
  • Spread exceeds 10x the rolling median spread

Stage 3: Pattern-based detection

  • Consecutive identical prints (stale data indicator)
  • Halted trading flags in metadata
  • Circuit breaker trigger detection
import statistics
from typing import Literal

AnomalyType = Literal["price_spike", "volume_spike", "spread_anomaly", "duplicate", "stale_quote"]

def detect_price_anomaly(
    current_price: float,
    historical_prices: list[float],
    window: int = 20,
    std_threshold: float = 5.0
) -> tuple[bool, float, float]:
    """
    Detect price anomalies using rolling z-score.
    
    Args:
        current_price: The price to test
        historical_prices: Prior prices for baseline calculation
        window: Number of periods for rolling statistics
        std_threshold: Number of standard deviations for anomaly threshold
    
    Returns:
        (is_anomaly, z_score, threshold_used)
    """
    if len(historical_prices) < window:
        return False, 0.0, 0.0
    
    window_prices = historical_prices[-window:]
    median = statistics.median(window_prices)
    stdev = statistics.stdev(window_prices)
    
    if stdev == 0:
        return False, 0.0, 0.0
    
    z_score = abs(current_price - median) / stdev
    is_anomaly = z_score > std_threshold
    
    return is_anomaly, round(z_score, 2), std_threshold

def detect_spread_anomaly(
    bid: float,
    ask: float,
    reference_price: float,
    max_spread_bps: float = 50.0
) -> tuple[bool, float]:
    """
    Detect anomalous bid-ask spreads.
    
    A 50 bps spread on a large-cap stock is typically an anomaly
    (indicating stale quote or wide-market conditions).
    """
    if reference_price == 0:
        return False, 0.0
    
    spread_bps = ((ask - bid) / reference_price) * 10000
    is_anomaly = spread_bps > max_spread_bps
    
    return is_anomaly, round(spread_bps, 2)

def flag_duplicate_prints(kline_data: list[dict]) -> list[dict]:
    """
    Flag consecutive klines with identical OHLCV values.
    This often indicates stale data or exchange transmission errors.
    """
    duplicates = []
    
    for i in range(1, len(kline_data)):
        prev = kline_data[i-1]
        curr = kline_data[i]
        
        if (prev["open"] == curr["open"] and
            prev["high"] == curr["high"] and
            prev["low"] == curr["low"] and
            prev["close"] == curr["close"] and
            prev["volume"] == curr["volume"]):
            duplicates.append({
                "index": i,
                "timestamp": curr["timestamp"],
                "type": "duplicate_ohlcv",
                "value": {k: curr[k] for k in ["open", "high", "low", "close", "volume"]}
            })
    
    return duplicates

# Full anomaly detection pipeline
def run_anomaly_pipeline(kline_data: list[dict], symbol: str) -> dict[str, list]:
    """
    Run the complete TickDB anomaly detection pipeline on kline data.
    
    Returns a dictionary of detected anomalies by type.
    """
    all_prices = [k["close"] for k in kline_data]
    results = {
        "price_spikes": [],
        "volume_spikes": [],
        "spread_anomalies": [],
        "duplicate_prints": []
    }
    
    for i, kline in enumerate(kline_data):
        # Stage 1: Price spike
        is_price_anomaly, z_score, _ = detect_price_anomaly(
            kline["close"],
            all_prices[:i] if i > 0 else [kline["close"]]
        )
        if is_price_anomaly:
            results["price_spikes"].append({
                "timestamp": kline["timestamp"],
                "price": kline["close"],
                "z_score": z_score,
                "severity": "high" if z_score > 10 else "medium"
            })
        
        # Stage 2: Spread anomaly (requires quote data)
        # Stage 3: Duplicate detection
        if i > 0:
            prev = kline_data[i-1]
            if all(kline[k] == prev[k] for k in ["open", "high", "low", "close", "volume"]):
                results["duplicate_prints"].append({
                    "timestamp": kline["timestamp"],
                    "value": kline["close"]
                })
    
    return results

# Example usage
print("Anomaly detection pipeline ready for production use.")

Anomaly Handling: Flag vs. Remove

TickDB takes a conservative approach: anomalies are flagged rather than removed by default. This preserves data integrity for researchers who want to see the raw feed, while providing metadata that allows filtering.

When you request processed data via the /v1/market/kline endpoint, anomalies within the filtered range are excluded. But the original data remains accessible via the /v1/market/trades endpoint for forensic analysis.

Dimension 4: Missing Data Handling (缺失数据)

The Problem

Missing data appears in several forms:

  1. Exchange holidays: Markets close on holidays, creating gaps in the time series.
  2. Trading pauses: Individual securities halt trading during news events.
  3. Data vendor gaps: Vendor system outages create gaps in the feed.
  4. After-hours filtering: Some vendors exclude pre-market and after-hours data.

A mean-reversion strategy that computes z-scores across a time series will produce wildly incorrect values if it treats a holiday gap as a market movement. A volatility calculation that includes overnight gaps will overestimate true volatility.

TickDB's Gap Handling Strategy

TickDB distinguishes between two types of gaps:

  1. Session gaps: Gaps at market open/close boundaries. These are expected and labeled with session_boundary: true in the metadata.

  2. Data gaps: Gaps within a trading session. These trigger alerts and are investigated for root cause.

from datetime import datetime, timedelta
from typing import Optional

# US equity session boundaries (approximate; TickDB uses exact exchange data)
US_EQUITY_SESSIONS = {
    "pre_market": {"start": "04:00", "end": "09:30", "tz": "America/New_York"},
    "regular": {"start": "09:30", "end": "16:00", "tz": "America/New_York"},
    "after_hours": {"start": "16:00", "end": "20:00", "tz": "America/New_York"},
}

def detect_session_gaps(
    kline_data: list[dict],
    interval_minutes: int,
    exchange: str = "US"
) -> list[dict]:
    """
    Detect and classify gaps in kline data.
    
    Session gaps (market open/close boundaries) are labeled as expected.
    Intra-session gaps are flagged for investigation.
    """
    gaps = []
    
    for i in range(1, len(kline_data)):
        prev_ts = datetime.fromisoformat(kline_data[i-1]["timestamp"].replace("Z", "+00:00"))
        curr_ts = datetime.fromisoformat(kline_data[i]["timestamp"].replace("Z", "+00:00"))
        
        expected_delta = timedelta(minutes=interval_minutes)
        actual_delta = curr_ts - prev_ts
        
        # Allow 50% tolerance for minor timing variations
        if actual_delta > expected_delta * 1.5:
            gap_duration = actual_delta - expected_delta
            
            # Check if this is a session boundary gap
            # (e.g., overnight gap from 16:00 to 09:30 next day)
            is_session_gap = _is_session_boundary_gap(prev_ts, curr_ts, exchange)
            
            gaps.append({
                "index": i,
                "start": kline_data[i-1]["timestamp"],
                "end": kline_data[i]["timestamp"],
                "gap_duration_minutes": gap_duration.total_seconds() / 60,
                "gap_type": "session_boundary" if is_session_gap else "data_gap",
                "requires_investigation": not is_session_gap
            })
    
    return gaps

def _is_session_boundary_gap(
    prev_ts: datetime,
    curr_ts: datetime,
    exchange: str
) -> bool:
    """
    Determine if a timestamp gap corresponds to a market session boundary.
    
    For US equities:
    - Regular session: 09:30-16:00 ET
    - Overnight gap from 16:00 ET to 09:30 ET next day is expected
    - Weekend gap from Friday 16:00 ET to Monday 09:30 ET is expected
    """
    # Simplified check: if gap > 2 hours during what should be trading hours
    gap_hours = (curr_ts - prev_ts).total_seconds() / 3600
    
    # Weekend gap
    if prev_ts.weekday() == 4 and prev_ts.hour >= 16:  # Friday after close
        if curr_ts.weekday() == 0:  # Monday
            return True
    
    # Overnight gap
    if gap_hours > 14:  # Longer than a full trading day
        return True
    
    return False

# Example: Fetch klines with gap awareness
def fetch_klines_with_gap_analysis(
    symbol: str,
    interval: str = "1h",
    limit: int = 100,
    ignore_gap: bool = False
) -> dict:
    """
    Fetch kline data with optional gap analysis.
    
    The `ignore_gap` parameter controls whether missing periods
    within a session are filled with null bars or omitted entirely.
    """
    api_key = os.environ.get("TICKDB_API_KEY")
    
    params = {
        "symbol": symbol,
        "interval": interval,
        "limit": limit,
        "ignore_gap": str(ignore_gap).lower()  # API expects string "true" or "false"
    }
    
    response = requests.get(
        "https://api.tickdb.ai/v1/market/kline",
        headers={"X-API-Key": api_key},
        params=params,
        timeout=(3.05, 10)
    )
    
    response.raise_for_status()
    data = response.json()
    
    klines = data.get("data", {}).get("klines", [])
    
    # Determine interval in minutes for gap detection
    interval_minutes = _parse_interval(interval)
    gaps = detect_session_gaps(klines, interval_minutes)
    
    return {
        "klines": klines,
        "gap_analysis": {
            "total_gaps": len(gaps),
            "session_gaps": len([g for g in gaps if g["gap_type"] == "session_boundary"]),
            "data_gaps": len([g for g in gaps if g["gap_type"] == "data_gap"]),
            "details": gaps
        }
    }

def _parse_interval(interval: str) -> int:
    """Convert interval string to minutes."""
    mapping = {
        "1m": 1, "5m": 5, "15m": 15, "30m": 30,
        "1h": 60, "2h": 120, "4h": 240, "1d": 1440
    }
    return mapping.get(interval, 60)

Gap Filling: When and How

TickDB does not automatically fill gaps with interpolated values. Interpolation introduces synthetic data that can distort statistical properties, especially for volatility and return calculations. Instead, TickDB provides two modes:

  1. ignore_gap=false (default): Returns the actual data as it occurred, including gaps. Your strategy logic must handle gaps explicitly.

  2. ignore_gap=true: Returns only complete periods. Weekend gaps and holiday gaps are omitted from the series, which is useful for strategies that trade intraday only.

Dimension 5: Cross-Market Temporal Alignment

The Problem

A cross-asset strategy might combine US equity options, Treasury futures, and currency forwards. Each market has different trading hours:

  • US equity options: 09:30–16:15 ET
  • 10-Year Treasury futures: 18:00–17:00 ET (nearly 24-hour)
  • EUR/USD forex: 24-hour, except weekends

When you align these time series, you need to know: which data points represent the same moment in time? A Treasury futures print at 17:00 ET and a forex print at 17:00 ET are simultaneous. A US equity print at 17:00 ET occurred after the equity market closed.

TickDB's Cross-Market Alignment

TickDB provides two tools for cross-market alignment:

  1. Session metadata: Every symbol includes trading session information (session_start, session_end, timezone). Use this to determine whether two timestamps fall within overlapping sessions.

  2. Alignment utilities: Helper functions to resample multiple time series to a common timestamp grid.

from datetime import datetime
import pandas as pd

def align_cross_market_klines(
    kline_sets: dict[str, list[dict]],
    target_interval: str = "1h",
    timezone: str = "America/New_York"
) -> pd.DataFrame:
    """
    Align multiple kline datasets to a common timestamp grid.
    
    This is critical for multi-asset strategies where you need
    synchronized data across equities, futures, and forex.
    
    Args:
        kline_sets: Dict of {symbol: kline_data}
        target_interval: Target interval for alignment (e.g., "1h")
        timezone: Timezone for output timestamps
    
    Returns:
        DataFrame with aligned close prices for all symbols
    """
    dfs = []
    
    for symbol, klines in kline_sets.items():
        if not klines:
            continue
        
        df = pd.DataFrame(klines)
        df["timestamp"] = pd.to_datetime(df["timestamp"])
        df = df.set_index("timestamp")
        df = df.sort_index()
        
        # Resample to target interval, using last available price
        # For OHLCV data, use the last close price
        resampled = df["close"].resample(target_interval).last()
        resampled.name = symbol
        
        dfs.append(resampled)
    
    if not dfs:
        return pd.DataFrame()
    
    # Align all series on a common index
    aligned = pd.concat(dfs, axis=1)
    
    # Forward-fill missing values within sessions, but preserve session gaps
    aligned = aligned.ffill()
    
    return aligned

def is_session_overlap(
    ts1: datetime,
    ts2: datetime,
    symbol1_session: dict,
    symbol2_session: dict
) -> bool:
    """
    Check if two timestamps fall within overlapping trading sessions.
    
    Returns True if both timestamps are within their respective
    trading sessions, meaning the data points are contemporaneous.
    """
    import zoneinfo
    
    def ts_in_session(ts: datetime, session: dict) -> bool:
        tz = zoneinfo.ZoneInfo(session["timezone"])
        local_ts = ts.astimezone(tz)
        
        start_parts = session["start"].split(":")
        end_parts = session["end"].split(":")
        
        start_hour, start_min = int(start_parts[0]), int(start_parts[1])
        end_hour, end_min = int(end_parts[0]), int(end_parts[1])
        
        start_minutes = start_hour * 60 + start_min
        end_minutes = end_hour * 60 + end_min
        ts_minutes = local_ts.hour * 60 + local_ts.minute
        
        # Handle overnight sessions (e.g., crypto, some futures)
        if end_minutes < start_minutes:
            return ts_minutes >= start_minutes or ts_minutes <= end_minutes
        else:
            return start_minutes <= ts_minutes <= end_minutes
    
    return ts_in_session(ts1, symbol1_session) and ts_in_session(ts2, symbol2_session)

# Example: Aligning equity with Treasury futures
if __name__ == "__main__":
    equity_session = {
        "start": "09:30",
        "end": "16:00",
        "timezone": "America/New_York"
    }
    
    futures_session = {
        "start": "18:00",
        "end": "17:00",  # Overnight session
        "timezone": "America/New_York"
    }
    
    # Test timestamp: 10:30 AM ET (both equity and futures are trading)
    test_ts = datetime(2024, 1, 15, 10, 30, tzinfo=zoneinfo.ZoneInfo("America/New_York"))
    overlap = is_session_overlap(test_ts, test_ts, equity_session, futures_session)
    print(f"10:30 AM ET overlap: {overlap}")  # True: futures still trading, equity open
    
    # Test timestamp: 17:30 PM ET (equity closed, futures still trading)
    test_ts2 = datetime(2024, 1, 15, 17, 30, tzinfo=zoneinfo.ZoneInfo("America/New_York"))
    overlap2 = is_session_overlap(test_ts2, test_ts2, equity_session, futures_session)
    print(f"17:30 PM ET overlap: {overlap2}")  # False: equity closed

Verifying TickDB's Data Quality Pipeline

TickDB provides dedicated endpoints for verifying data quality at each stage of the pipeline.

Verification need Endpoint What it returns
Corporate actions GET /v1/reference/corporate-actions Split dates, dividend amounts, adjustment factors
Symbol metadata GET /v1/symbols/available Exchange, timezone, session hours
Anomaly flags GET /v1/market/anomalies Detected anomalies with severity levels
Kline gaps GET /v1/market/kline with ignore_gap param Gap detection metadata 
def full_data_quality_report(symbol: str, start_date: str, end_date: str) -> dict:
    """
    Generate a comprehensive data quality report for a symbol.
    
    This function calls multiple TickDB endpoints to verify
    each dimension of data quality.
    """
    report = {"symbol": symbol, "period": f"{start_date} to {end_date}", "checks": {}}
    
    # Check 1: Corporate actions
    try:
        actions = fetch_splits_and_dividends(symbol, start_date, end_date)
        report["checks"]["corporate_actions"] = {
            "status": "ok",
            "split_count": len([a for a in actions.get("events", []) if a["type"] == "split"]),
            "dividend_count": len([a for a in actions.get("events", []) if a["type"] == "dividend"]),
        }
    except Exception as e:
        report["checks"]["corporate_actions"] = {"status": "error", "message": str(e)}
    
    # Check 2: Symbol metadata and timezone
    try:
        api_key = os.environ.get("TICKDB_API_KEY")
        resp = requests.get(
            "https://api.tickdb.ai/v1/symbols/available",
            headers={"X-API-Key": api_key},
            params={"symbol": symbol},
            timeout=(3.05, 10)
        )
        resp.raise_for_status()
        symbol_data = resp.json().get("data", {})
        report["checks"]["symbol_metadata"] = {
            "status": "ok",
            "exchange": symbol_data.get("exchange"),
            "timezone": symbol_data.get("timezone"),
            "session_start": symbol_data.get("session_start"),
            "session_end": symbol_data.get("session_end"),
        }
    except Exception as e:
        report["checks"]["symbol_metadata"] = {"status": "error", "message": str(e)}
    
    # Check 3: Kline quality
    try:
        kline_result = fetch_klines_with_gap_analysis(symbol, "1h", 500, ignore_gap=False)
        gaps = kline_result["gap_analysis"]
        report["checks"]["kline_quality"] = {
            "status": "ok" if gaps["data_gaps"] == 0 else "warning",
            "total_bars": len(kline_result["klines"]),
            "session_gaps": gaps["session_gaps"],
            "data_gaps": gaps["data_gaps"],
            "gap_details": gaps["details"] if gaps["data_gaps"] > 0 else []
        }
    except Exception as e:
        report["checks"]["kline_quality"] = {"status": "error", "message": str(e)}
    
    return report

# Generate a sample report
sample_report = full_data_quality_report("AAPL.US", "2023-01-01", "2024-01-01")
print("Data Quality Report:")
for check_name, result in sample_report["checks"].items():
    status_icon = "✅" if result["status"] == "ok" else "⚠️" if result["status"] == "warning" else "❌"
    print(f"  {status_icon} {check_name}: {result['status']}")

Why This Matters for Your Strategy

The five dimensions we have examined — corporate actions, timestamp normalization, anomaly detection, missing data handling, and cross-market alignment — are not academic concerns. They are the difference between a backtest that forecasts reality and one that lies to you.

Consider the cumulative effect over a three-year backtest:

  • A 2% dividend yield unaccounted for over three years inflates returns by approximately 6%.
  • A single fat-finger print with a 20-standard-deviation move can double your estimated volatility.
  • Weekend gaps treated as trading days inflate your apparent Sharpe ratio by masking the true variance of weekly returns.

The data cleaning pipeline is not a feature. It is the foundation on which every quantitative strategy stands. Without clean, aligned data, your strategy optimization, parameter tuning, and risk modeling are exercises in fitting noise.

Next Steps

If you are building a backtesting framework, start with TickDB's /v1/reference/corporate-actions endpoint to understand which symbols have experienced splits or dividends during your backtest period. Apply split adjustments before running any price-based strategy.

If you are running live strategies, verify timestamp alignment before deploying multi-market strategies. Use the session metadata from /v1/symbols/available to implement your own session-overlap checks.

If you are evaluating data quality, run the full_data_quality_report function on your target symbols before committing to a backtest period. The 30 seconds spent on quality verification can save hours of debugging a strategy that fails in live trading.

To explore TickDB's full API capabilities: visit tickdb.ai and sign up for a free API key. The free tier includes access to 10+ years of cleaned US equity OHLCV data, corporate action references, and symbol metadata.

If you are integrating with AI coding tools: search for and install the tickdb-market-data SKILL in your AI tool's marketplace to access TickDB capabilities directly from your development environment.

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