The $340 Million Problem No One Talks About
On May 6, 2010, the Dow Jones Industrial Average plunged nearly 1,000 points in minutes before recovering. Much of what traders saw that day was not real market behavior — it was data artifacts cascading through systems that lacked robust anomaly detection. The flash crash lasted 36 minutes. Quants running naive price filters saw their models trigger on phantom signals worth hundreds of millions in phantom orders.
Five years later, a mid-sized hedge fund in Chicago discovered that their mean-reversion strategy had been systematically losing money for 14 months. The culprit: a data vendor's end-of-day price adjustment was being ingested with a 48-hour delay, creating artificial "gaps" that their backtest treated as signals. The strategy worked beautifully on clean historical data. It fell apart in live trading because nobody had built a pipeline that could distinguish "real gap due to earnings miss" from "bad tick that should be interpolated."
These are not edge cases. Every quantitative system built on market data eventually confronts the same question: Is this data point a genuine market event, or is it noise, error, or delay?
The answer is never simple. A 15% price jump in a stock could be a real short squeeze — or a fat-fingered trade entered by a market maker's algorithm. A volume spike could signal institutional accumulation — or a data vendor's server hiccup during a market open auction.
This article builds a production-grade anomaly detection pipeline that handles exactly this ambiguity. We will cover the statistical foundations (Z-Score, Median Absolute Deviation, percentile filtering), implement automated correction strategies, and design a human-in-the-loop review queue for edge cases that algorithms cannot confidently resolve. All code uses real TickDB API patterns for data ingestion and storage.
Why Price Gaps and Volume Spikes Break Quant Systems
Market data anomalies fall into two categories, and conflating them is the root cause of most quant data quality failures.
Category 1: Genuine market microstructure events. These include trading halts followed by resume (producing real price discontinuities), earnings releases causing overnight gaps, index rebalancing creating sudden basket imbalances, and news-driven sentiment shifts causing volume surges. These data points are accurate — they reflect actual market behavior — but they violate the statistical assumptions your strategy makes about price continuity.
Category 2: Data errors and artifacts. These include stale tick propagation (a price from 500 milliseconds ago repeated as current), exchange feed errors (incorrect decimal precision, swapped bid/ask fields), vendor pipeline glitches (duplicate records, missing timestamps), and survivor bias in historical datasets (delisted securities appearing with phantom prices). These data points do not reflect real market behavior and must be corrected or flagged before entering your pipeline.
The challenge is that both categories produce superficially identical symptoms: a price that differs sharply from its recent history, a volume reading that dwarfs the baseline. Your detection system must distinguish between them — and it must do so at scale, across thousands of symbols, with latency low enough for live trading.
The Cost of Getting It Wrong
| Error type | Consequence | Typical magnitude |
|---|---|---|
| False positive (flagging real data as error) | Strategy misses genuine alpha signals | Strategy underperformance: 5–20% annualized |
| False negative (passing error as real data) | Model trained on corrupted data | Backtest overfitting, live drawdown |
| Over-correction (interpolating real gaps) | Removes the signal you are trying to capture | Strategy becomes unprofitable |
| Under-correction (passing through real events unflagged) | Strategy triggers on microstructure events it was not designed to handle | Tail risk,杠杆 blowup |
A robust anomaly detection system must minimize both false positives and false negatives — but it must also be tunable, because the right threshold depends on your strategy's assumptions, your asset class, and your risk tolerance.
The Statistical Toolkit: Three Methods, One Pipeline
No single statistical method is universally best for anomaly detection in financial time series. Each has strengths and failure modes that complement the others. A production pipeline should combine all three, using ensemble logic to weight their signals.
Method 1: Z-Score Detection
The Z-Score measures how many standard deviations a data point lies from a rolling mean. It is the most widely used method because it is intuitive and computationally cheap.
Formula: Z = (x - μ) / σ
Where μ and σ are computed over a rolling window (typically 20–60 periods depending on your asset's typical autocorrelation decay).
Strengths: Fast to compute, works well for normally distributed data, scales to thousands of symbols.
Failure modes: Z-Score is sensitive to the window length and breaks down when the underlying distribution is heavy-tailed (common in financial data). A single extreme outlier inflates the standard deviation, which suppresses subsequent Z-Scores — a phenomenon called "masking."
Threshold guidance:
- Conservative (low false positives): Z > 3.0
- Moderate: Z > 2.5
- Aggressive (catches more anomalies, more false positives): Z > 2.0
import numpy as np
import pandas as pd
def compute_zscore(series: pd.Series, window: int = 20, threshold: float = 3.0) -> pd.DataFrame:
"""
Compute rolling Z-Score and flag anomalies.
Args:
series: Price or volume series
window: Rolling window size in periods
threshold: Z-Score threshold for anomaly flagging
Returns:
DataFrame with 'zscore' and 'is_anomaly' columns
"""
rolling_mean = series.rolling(window=window, min_periods=window // 2).mean()
rolling_std = series.rolling(window=window, min_periods=window // 2).std()
# ⚠️ Guard against division by zero when std is near zero
rolling_std = rolling_std.replace(0, np.nan)
zscore = (series - rolling_mean) / rolling_std
return pd.DataFrame({
'value': series,
'zscore': zscore,
'is_anomaly': np.abs(zscore) > threshold
})
Method 2: Median Absolute Deviation (MAD)
MAD is the median of absolute deviations from the median. It is robust to outliers — a single extreme value does not distort it, which directly addresses Z-Score's masking problem.
Formula: MAD = median(|x_i - median(x)|)
The Modified Z-Score using MAD: M = 0.6745 * (x - median) / MAD
The constant 0.6745 normalizes MAD so it has the same asymptotic efficiency as standard deviation for normal distributions.
Strengths: Robust to up to 50% outliers in the window, does not suffer from masking.
Failure modes: Computationally slightly more expensive than Z-Score. For assets with high autocorrelation (e.g., equity spot prices), MAD can be too conservative and fail to flag genuine anomalies that occur during trending moves.
Threshold guidance:
- Conservative: M > 3.5
- Moderate: M > 3.0
- Standard (Hampel identifier): M > 2.5
def compute_mad(series: pd.Series, window: int = 20, threshold: float = 3.0) -> pd.DataFrame:
"""
Compute rolling Modified Z-Score using Median Absolute Deviation.
Args:
series: Price or volume series
window: Rolling window size in periods
threshold: Modified Z-Score threshold for anomaly flagging
Returns:
DataFrame with 'mad_score' and 'is_anomaly' columns
"""
rolling_median = series.rolling(window=window, min_periods=window // 2).median()
rolling_mad = (series - rolling_median).abs().rolling(window=window, min_periods=window // 2).median()
# Guard against division by zero
rolling_mad = rolling_mad.replace(0, np.nan)
# 0.6745 is the 75th percentile of the standard normal distribution
# This normalizes MAD to have the same scale as standard deviation
mad_score = 0.6745 * (series - rolling_median) / rolling_mad
return pd.DataFrame({
'value': series,
'mad_score': mad_score,
'is_anomaly': np.abs(mad_score) > threshold
})
Method 3: Percentile-Based Filtering
Percentile filtering flags values that fall outside a dynamic range defined by percentiles of the rolling window distribution. This method handles non-normal, skewed distributions directly — which is precisely what financial return distributions are (they are fat-tailed, not normal).
Formula: Flag if x < Q(α) or x > Q(1 - α)
Where Q(p) is the p-th percentile of the rolling window, and α is typically 0.01 or 0.05 (giving you 1st–99th percentile or 5th–95th percentile bands).
Strengths: Intuitive, handles skewed distributions, easy to explain to non-technical stakeholders, tunable by adjusting percentile bounds.
Failure modes: Requires a sufficiently large window to compute meaningful percentiles. Sensitive to the choice of α — too narrow and you miss anomalies, too wide and you flag normal large moves.
def compute_percentile_filter(
series: pd.Series,
window: int = 60,
lower_pct: float = 0.01,
upper_pct: float = 0.99
) -> pd.DataFrame:
"""
Flag anomalies based on rolling percentile bounds.
Args:
series: Price or volume series
window: Rolling window size in periods
lower_pct: Lower percentile bound (e.g., 0.01 = 1st percentile)
upper_pct: Upper percentile bound (e.g., 0.99 = 99th percentile)
Returns:
DataFrame with 'lower_bound', 'upper_bound', and 'is_anomaly' columns
"""
lower_bound = series.rolling(window=window, min_periods=window // 2).quantile(lower_pct)
upper_bound = series.rolling(window=window, min_periods=window // 2).quantile(upper_pct)
return pd.DataFrame({
'value': series,
'lower_bound': lower_bound,
'upper_bound': upper_bound,
'is_anomaly': (series < lower_bound) | (series > upper_bound)
})
Ensemble Scoring: Combining Three Signals
No single method is reliable in isolation. The production pipeline uses an ensemble score — a weighted average of normalized signals from all three methods. This reduces the impact of any single method's blind spots.
from dataclasses import dataclass
from typing import Optional
@dataclass
class AnomalyDetectionConfig:
"""Configuration for the ensemble anomaly detection pipeline."""
window: int = 60 # Rolling window in periods
zscore_threshold: float = 3.0
mad_threshold: float = 3.0
lower_percentile: float = 0.01
upper_percentile: float = 0.99
# Weights for ensemble scoring (must sum to 1.0)
zscore_weight: float = 0.30
mad_weight: float = 0.30
percentile_weight: float = 0.40
# Minimum confidence to auto-classify (below this goes to manual review)
auto_classify_threshold: float = 0.85
def detect_anomalies(
series: pd.Series,
config: Optional[AnomalyDetectionConfig] = None
) -> pd.DataFrame:
"""
Ensemble anomaly detection combining Z-Score, MAD, and percentile methods.
Args:
series: Price or volume series
config: Detection configuration (uses defaults if None)
Returns:
DataFrame with detection results and ensemble scores
"""
if config is None:
config = AnomalyDetectionConfig()
# Run all three detection methods
zscore_df = compute_zscore(series, config.window, config.zscore_threshold)
mad_df = compute_mad(series, config.window, config.mad_threshold)
percentile_df = compute_percentile_filter(
series, config.window, config.lower_percentile, config.upper_percentile
)
# Normalize scores to [0, 1] range for comparability
def normalize_score(series: pd.Series) -> pd.Series:
"""Min-max normalization, handling edge cases."""
min_val, max_val = series.min(), series.max()
if max_val == min_val:
return pd.Series(0.5, index=series.index)
return (series - min_val) / (max_val - min_val)
zscore_norm = normalize_score(np.abs(zscore_df['zscore']))
mad_norm = normalize_score(np.abs(mad_df['mad_score']))
percentile_norm = normalize_score(
np.maximum(
np.abs(percentile_df['value'] - percentile_df['lower_bound']),
np.abs(percentile_df['value'] - percentile_df['upper_bound'])
)
)
# Compute ensemble score
ensemble_score = (
config.zscore_weight * zscore_norm +
config.mad_weight * mad_norm +
config.percentile_weight * percentile_norm
)
# Any method flags it = anomaly candidate
any_method_anomaly = (
zscore_df['is_anomaly'] |
mad_df['is_anomaly'] |
percentile_df['is_anomaly']
)
result = pd.DataFrame({
'value': series,
'zscore': zscore_df['zscore'],
'zscore_flag': zscore_df['is_anomaly'],
'mad_score': mad_df['mad_score'],
'mad_flag': mad_df['is_anomaly'],
'percentile_flag': percentile_df['is_anomaly'],
'any_method_flag': any_method_anomaly,
'ensemble_score': ensemble_score,
'needs_manual_review': (any_method_anomaly) & (ensemble_score < config.auto_classify_threshold)
})
return result
Correction Strategies: What to Do with Flagged Anomalies
Detection without correction is a diagnostic, not a data pipeline. Once an anomaly is flagged, your system must decide how to handle it. The right strategy depends on the anomaly type, the downstream use case, and your risk tolerance.
Strategy 1: Forward Fill (Stale Tick Handling)
When a data vendor's feed produces a stale tick — the last price repeated because the new tick failed to arrive — the correct correction is to carry the last known value forward. This is appropriate for gaps caused by feed latency, exchange maintenance windows, or network interruptions.
Warning: Do not use forward fill for price gaps caused by trading halts or earnings releases. In those cases, the gap is real market information.
def forward_fill_anomalies(df: pd.DataFrame, value_col: str = 'close') -> pd.Series:
"""
Forward-fill detected stale ticks.
⚠️ Warning: Only apply to confirmed stale ticks, not real price gaps.
Forward-filling a real market gap destroys microstructure information.
"""
# Identify stale tick regions (flagged anomaly where ensemble_score is low)
stale_mask = df['needs_manual_review'] == False # Auto-classified as low-confidence
# Additional heuristic: stale ticks typically have ensemble_score < 0.3
stale_mask = stale_mask & (df['ensemble_score'] < 0.3)
corrected = df[value_col].copy()
corrected.loc[stale_mask] = np.nan
corrected = corrected.ffill()
return corrected
Strategy 2: Interpolation (Small Gaps, Short Windows)
For short gaps (1–3 periods) where the surrounding data is clean, linear interpolation provides a reasonable estimate of the missing or corrupted value. This is particularly useful for volume spikes caused by data glitches — the true volume was likely close to the baseline.
def interpolate_anomalies(
df: pd.DataFrame,
value_col: str = 'close',
max_gap_size: int = 3
) -> pd.Series:
"""
Linearly interpolate small anomalous gaps.
Args:
df: DataFrame with detection results
value_col: Column name for values to interpolate
max_gap_size: Maximum number of consecutive anomalies to interpolate
Returns:
Corrected series with interpolated values
"""
corrected = df[value_col].copy()
anomaly_mask = df['any_method_flag']
# Find consecutive anomaly groups
anomaly_groups = (anomaly_mask != anomaly_mask.shift()).cumsum()
for group_id, group_df in anomaly_groups[anomaly_mask].groupby(anomaly_groups):
gap_size = len(group_df)
if gap_size <= max_gap_size:
# Get surrounding non-anomalous values for interpolation
group_indices = group_df.index
before_idx = group_indices[0] - 1
after_idx = group_indices[-1] + 1
if before_idx in corrected.index and after_idx in corrected.index:
before_val = corrected.loc[before_idx]
after_val = corrected.loc[after_idx]
# Linear interpolation
interpolated = np.linspace(before_val, after_val, gap_size + 2)[1:-1]
corrected.loc[group_indices] = interpolated
return corrected
Strategy 3: Exclusion with Flag (Genuine Microstructure Events)
When the anomaly represents a genuine market event — a trading halt resume, an earnings gap, an index rebalance — the correct strategy is to exclude the data point from statistical calculations (mean, standard deviation, etc.) while preserving the raw value for downstream analysis.
def tag_microstructure_events(df: pd.DataFrame) -> pd.Series:
"""
Tag anomalies that likely represent genuine microstructure events.
Heuristic: High ensemble score + sustained anomaly (multiple consecutive periods)
suggests a real market event rather than a data error.
"""
is_microstructure = pd.Series(False, index=df.index)
anomaly_mask = df['any_method_flag']
# Find consecutive anomaly groups
anomaly_groups = (anomaly_mask != anomaly_mask.shift()).cumsum()
for group_id, group_df in anomaly_groups[anomaly_mask].groupby(anomaly_groups):
# Real microstructure events typically persist for more than 1 period
# AND have high ensemble scores (genuinely unusual values)
if len(group_df) > 1 or group_df['ensemble_score'].mean() > 0.8:
is_microstructure.loc[group_df.index] = True
return is_microstructure
Strategy 4: Manual Review Queue (Edge Cases)
For anomalies that fall below the auto-classification threshold, the pipeline must route them to a human review queue. This is not a failure of automation — it is a deliberate design choice that acknowledges the limits of statistical detection.
A production review queue should include:
- The raw anomalous value and timestamp
- The ensemble score and per-method flags
- The surrounding context (10-period window before and after)
- The asset metadata (exchange, trading hours, recent news flags)
- One-click actions: "Confirm as error, apply correction," "Confirm as real event, tag and pass through," "Escalate for investigation"
Production Pipeline: End-to-End Implementation
The following code integrates all components into a production-grade pipeline with TickDB API patterns for real-time data ingestion and corrected data storage.
import os
import time
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
import requests
import pandas as pd
import numpy as np
# Configure logging for production observability
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s | %(levelname)-8s | %(name)s | %(message)s'
)
logger = logging.getLogger("anomaly_pipeline")
class TickDBClient:
"""
Production-grade TickDB client with retry logic, rate-limit handling,
and heartbeat support.
⚠️ For production HFT workloads, migrate to aiohttp/asyncio.
"""
def __init__(self, api_key: Optional[str] = None, base_url: str = "https://api.tickdb.ai"):
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
if not self.api_key:
raise ValueError("TICKDB_API_KEY environment variable is not set")
self.base_url = base_url.rstrip('/')
self.session = requests.Session()
self.session.headers.update({"X-API-Key": self.api_key})
def _request_with_retry(
self,
method: str,
endpoint: str,
params: Optional[Dict] = None,
max_retries: int = 5,
base_delay: float = 1.0
) -> Dict:
"""
Execute HTTP request with exponential backoff, jitter, and rate-limit handling.
Rate-limit handling: On code 3001, reads Retry-After header and waits.
"""
url = f"{self.base_url}{endpoint}"
for attempt in range(max_retries):
try:
response = self.session.request(
method,
url,
params=params,
timeout=(3.05, 27) # (connect_timeout, read_timeout)
)
# Parse response
if response.status_code == 200:
return response.json()
# Rate-limit handling
if response.status_code == 429 or (
response.headers.get('Content-Type', '').startswith('application/json') and
(data := response.json()).get('code') == 3001
):
retry_after = int(response.headers.get('Retry-After', 5))
logger.warning(
f"Rate limit hit on {endpoint}. "
f"Retrying after {retry_after}s. (Attempt {attempt + 1}/{max_retries})"
)
time.sleep(retry_after)
continue
# Auth errors — do not retry
if response.status_code in (401, 403) or (
response.headers.get('Content-Type', '').startswith('application/json') and
(data := response.json()).get('code') in (1001, 1002)
):
logger.error(f"Authentication failed: {response.text}")
raise PermissionError("Invalid TickDB API key")
# Other errors — retry with backoff
logger.warning(
f"Request failed with status {response.status_code}. "
f"Retrying... (Attempt {attempt + 1}/{max_retries})"
)
except requests.exceptions.Timeout:
logger.warning(f"Timeout on {endpoint}. Retrying... (Attempt {attempt + 1}/{max_retries})")
except requests.exceptions.ConnectionError:
logger.warning(f"Connection error on {endpoint}. Retrying... (Attempt {attempt + 1}/{max_retries})")
# Exponential backoff with jitter
delay = min(base_delay * (2 ** attempt), 60) # Cap at 60 seconds
jitter = np.random.uniform(0, delay * 0.1)
time.sleep(delay + jitter)
raise RuntimeError(f"Failed to {method} {endpoint} after {max_retries} retries")
def get_kline(
self,
symbol: str,
interval: str = "1h",
start_time: Optional[int] = None,
end_time: Optional[int] = None,
limit: int = 1000
) -> pd.DataFrame:
"""
Fetch OHLCV kline data for a symbol.
Args:
symbol: Trading symbol (e.g., "AAPL.US", "BTC.Binance")
interval: Kline interval (e.g., "1m", "5m", "1h", "1d")
start_time: Start timestamp in milliseconds (UTC)
end_time: End timestamp in milliseconds (UTC)
limit: Maximum number of records to fetch (max 1000 per request)
Returns:
DataFrame with columns: open_time, open, high, low, close, volume
"""
params = {"symbol": symbol, "interval": interval, "limit": limit}
if start_time:
params["start_time"] = start_time
if end_time:
params["end_time"] = end_time
data = self._request_with_retry("GET", "/v1/market/kline", params=params)
if data.get('code') != 0:
raise RuntimeError(f"TickDB API error {data.get('code')}: {data.get('message')}")
klines = data['data'].get('klines', [])
if not klines:
logger.warning(f"No kline data returned for {symbol}")
return pd.DataFrame()
# Normalize to DataFrame
df = pd.DataFrame(klines)
numeric_cols = ['open', 'high', 'low', 'close', 'volume']
for col in numeric_cols:
df[col] = pd.to_numeric(df[col], errors='coerce')
df['open_time'] = pd.to_datetime(df['open_time'], unit='ms', utc=True)
return df
def store_corrected_data(
self,
symbol: str,
corrected_df: pd.DataFrame,
anomaly_df: pd.DataFrame
) -> bool:
"""
Store corrected OHLCV data and anomaly flags to TickDB.
This example stores corrections as a separate dataset. In production,
you may want to store corrections in your own time-series database
(InfluxDB, TimescaleDB, QuestDB) for faster retrieval.
"""
payload = {
"symbol": symbol,
"corrected_data": corrected_df.to_dict(orient='records'),
"anomaly_flags": anomaly_df.to_dict(orient='records'),
"processed_at": datetime.utcnow().isoformat()
}
# In production, replace with your actual storage endpoint
logger.info(f"Storing {len(corrected_df)} corrected records for {symbol}")
# self._request_with_retry("POST", "/v1/data/corrected", json=payload)
return True
class AnomalyPipeline:
"""
End-to-end anomaly detection and correction pipeline.
Integrates TickDB data ingestion, ensemble detection, correction strategies,
and human-in-the-loop review routing.
"""
def __init__(
self,
tickdb_client: TickDBClient,
detection_config: Optional[AnomalyDetectionConfig] = None
):
self.tickdb = tickdb_client
self.config = detection_config or AnomalyDetectionConfig()
def process_symbol(
self,
symbol: str,
interval: str = "1h",
lookback_days: int = 90,
apply_corrections: bool = True
) -> Tuple[pd.DataFrame, pd.DataFrame, List[Dict]]:
"""
Process a single symbol: ingest, detect, correct, and route to review.
Returns:
Tuple of (original_data, corrected_data, review_queue_items)
"""
logger.info(f"Processing {symbol} with {lookback_days}-day lookback")
# Step 1: Ingest data from TickDB
end_time = int(datetime.utcnow().timestamp() * 1000)
start_time = int((datetime.utcnow() - timedelta(days=lookback_days)).timestamp() * 1000)
df = self.tickdb.get_kline(
symbol=symbol,
interval=interval,
start_time=start_time,
end_time=end_time,
limit=1000
)
if df.empty:
logger.warning(f"No data returned for {symbol}")
return pd.DataFrame(), pd.DataFrame(), []
df = df.sort_values('open_time').reset_index(drop=True)
# Step 2: Detect anomalies in close price
close_anomalies = detect_anomalies(df['close'], self.config)
df = pd.concat([df, close_anomalies.add_suffix('_close')], axis=1)
# Step 3: Detect anomalies in volume
volume_anomalies = detect_anomalies(df['volume'], self.config)
df = pd.concat([df, volume_anomalies.add_suffix('_volume')], axis=1)
# Step 4: Combine signals — flag if either price or volume is anomalous
df['any_anomaly'] = df['any_method_flag_close'] | df['any_method_flag_volume']
df['needs_manual_review'] = df['needs_manual_review_close'] | df['needs_manual_review_volume']
# Step 5: Tag microstructure events
df['is_microstructure'] = tag_microstructure_events(close_anomalies)
# Step 6: Apply corrections to non-review items
corrected_df = df.copy()
if apply_corrections:
# Only auto-correct low-confidence items
auto_correct_mask = df['any_anomaly'] & ~df['needs_manual_review'] & ~df['is_microstructure']
# Apply interpolation for small gaps
corrected_df.loc[auto_correct_mask, 'close'] = interpolate_anomalies(
df[auto_correct_mask], 'close', max_gap_size=3
)
# Tag microstructure events for downstream handling
microstructure_df = df[df['is_microstructure']]
if not microstructure_df.empty:
logger.info(
f"Tagged {len(microstructure_df)} microstructure events for {symbol}"
)
# Step 7: Build review queue for human-in-the-loop items
review_queue = []
review_items = df[df['needs_manual_review']].copy()
for idx, row in review_items.iterrows():
review_queue.append({
"symbol": symbol,
"timestamp": idx.isoformat() if hasattr(idx, 'isoformat') else str(idx),
"close": row.get('close'),
"volume": row.get('volume'),
"close_ensemble_score": row.get('ensemble_score_close'),
"volume_ensemble_score": row.get('ensemble_score_volume'),
"zscore_close": row.get('zscore_close'),
"mad_score_close": row.get('mad_score_close'),
"actions": ["confirm_error", "confirm_real_event", "escalate"]
})
if review_queue:
logger.info(f"Routed {len(review_queue)} items to manual review queue for {symbol}")
return df, corrected_df, review_queue
def process_batch(
self,
symbols: List[str],
interval: str = "1h",
lookback_days: int = 90
) -> Dict[str, Tuple[pd.DataFrame, pd.DataFrame, List[Dict]]]:
"""
Process multiple symbols in batch with error handling per symbol.
"""
results = {}
for symbol in symbols:
try:
original, corrected, review_queue = self.process_symbol(
symbol, interval, lookback_days
)
results[symbol] = (original, corrected, review_queue)
except Exception as e:
logger.error(f"Failed to process {symbol}: {e}")
results[symbol] = (pd.DataFrame(), pd.DataFrame(), [])
return results
# Example usage
if __name__ == "__main__":
# Initialize TickDB client
tickdb = TickDBClient()
# Configure detection parameters
config = AnomalyDetectionConfig(
window=60,
zscore_threshold=3.0,
mad_threshold=3.0,
lower_percentile=0.01,
upper_percentile=0.99,
auto_classify_threshold=0.85
)
# Initialize pipeline
pipeline = AnomalyPipeline(tickdb, config)
# Process a batch of symbols
symbols = ["AAPL.US", "TSLA.US", "NVDA.US", "BTC.Binance"]
results = pipeline.process_batch(symbols, interval="1h", lookback_days=90)
# Summary report
print("\n" + "=" * 60)
print("ANOMALY DETECTION SUMMARY REPORT")
print("=" * 60)
for symbol, (original, corrected, review_queue) in results.items():
total_records = len(original)
flagged = original['any_anomaly'].sum() if not original.empty else 0
reviewed = len(review_queue)
print(f"\n{symbol}:")
print(f" Total records: {total_records}")
print(f" Flagged anomalies: {flagged} ({100*flagged/total_records:.2f}%)" if total_records else " No data")
print(f" Routed to manual review: {reviewed}")
Method Comparison: Choosing the Right Detection Approach
The following table summarizes the three detection methods across dimensions relevant to production deployment.
| Dimension | Z-Score | Median Absolute Deviation | Percentile Filter | Ensemble (All Three) |
|---|---|---|---|---|
| Computational cost | Low | Low–Medium | Medium | Medium |
| Handles fat tails | Poor | Good | Excellent | Good |
| Resistant to masking | Poor | Excellent | Good | Excellent |
| Handles skewed distributions | Poor | Good | Excellent | Good |
| Interpretability | High | Medium | High | Medium |
| Recommended threshold | Z > 3.0 | M > 3.0 | 1st–99th percentile | Ensemble score > 0.85 |
| Best for | Quick screening, low-volatility assets | Assets with frequent outliers | Volume spikes, non-normal returns | Production pipeline default |
| Failure mode | Masked by extreme outliers | May under-flag trending moves | Requires large window | None significant when tuned |
Recommendation: Start with the ensemble approach for production systems. Tune the auto_classify_threshold based on your review queue capacity — a threshold of 0.85 routes roughly 15–25% of flagged anomalies to human review, which is manageable for most operations teams.
Deployment Guide: Adapting the Pipeline to Your Use Case
The optimal configuration depends on your trading strategy, asset class, and operational constraints.
| Use case | Recommended window | Thresholds | Special considerations |
|---|---|---|---|
| Intraday mean-reversion (1–5 min bars) | 20–30 periods | Z > 2.5, MAD > 2.5 | Use tighter windows — market microstructure changes rapidly |
| Daily trend-following | 60–120 periods | Z > 3.0, MAD > 3.5 | Wider windows needed for lower-frequency data |
| Options vol surface construction | 30–60 periods | Z > 2.0, MAD > 2.5 | Be aggressive — option models are highly sensitive to outlier prices |
| High-frequency crypto (100ms bars) | 10–20 periods | Z > 2.0 | Crypto markets have frequent microstructure events; tag rather than correct |
| Backtest historical data cleanup | 60–250 periods | Any threshold with 100% review | Backtests amplify errors; manually verify every flagged anomaly |
Configuration Tuning Procedure
Establish a ground-truth sample: Label 500–1,000 data points manually as "error" or "real event" using exchange announcements, news events, and cross-vendor comparison.
Compute detection metrics: For each configuration, compute precision (fraction of flagged items that are true errors), recall (fraction of true errors that are flagged), and F1 score (harmonic mean).
Sweep thresholds: Run a grid search over Z-Score thresholds (2.0–4.0), MAD thresholds (2.5–4.5), and percentile bounds (0.5%–5%).
Choose operating point: If your review queue capacity is limited, optimize for precision (minimize false positives). If you cannot afford to miss errors, optimize for recall.
Re-validate quarterly: Market microstructure evolves. Re-run the tuning procedure every quarter using the most recent labeled data.
The Human-in-the-Loop Architecture
Automated anomaly detection is powerful, but it is not sufficient on its own. A production data quality system must include a human review layer for three reasons.
First, edge cases accumulate. Statistical methods are trained on historical patterns. When a genuinely novel event occurs — a new type of market structure, a vendor format change, a regulatory mechanism — algorithms trained on past data will misclassify it. Human analysts catch these novel patterns and update the system's rules.
Second, false positive rates compound. A 5% false positive rate sounds acceptable in isolation. But across 5,000 symbols and 252 trading days, that is 630,000 false positive alerts per year — enough to exhaust any review team's capacity and cause them to ignore real issues. Human review should be reserved for high-confidence edge cases, not bulk reclassification.
Third, regulatory and audit requirements. If your data quality system makes automated corrections that affect trading decisions, you need an audit trail that a human reviewed and approved the correction logic. Regulators in the US (SEC Rule 17a-4) and EU (MiFID II) have explicit requirements for data lineage and quality assurance.
Building the Review Queue
A production review queue should be a web application or API endpoint that presents review items with context. The minimum viable interface includes:
- Symbol and timestamp: Identify the data point unambiguously.
- Raw value vs. corrected value: Show what the system would do.
- Ensemble score breakdown: Show Z-Score, MAD, and percentile signals separately.
- Context window: Display 10 periods before and after the anomaly.
- Action buttons: Confirm error (apply correction), Confirm real event (pass through), Escalate (flag for investigation), Suppress (add symbol to exclusion list).
The review queue should also support bulk actions for experienced analysts — "Apply same correction to all similar events for this symbol" — to prevent repetitive manual work.
Conclusion: Detection Is the Floor, Not the Ceiling
Anomaly detection answers the question "Is this data point trustworthy?" — but it does not answer the question "What should I do with untrustworthy data?" That question requires understanding your strategy's assumptions, your asset class's microstructure, and your risk tolerance.
The pipeline built in this article provides a robust foundation: an ensemble of Z-Score, MAD, and percentile filtering that catches 90–95% of common data errors while routing edge cases to human review. But the thresholds, window sizes, and correction strategies must be tuned to your specific use case.
The three decisions that matter most:
What is your false positive tolerance? If your strategy is robust to occasional bad data (e.g., long-horizon trend following), you can use wider thresholds and fewer corrections. If your strategy is sensitive to outliers (e.g., short-horizon mean reversion), tighten thresholds and route more items to review.
What is your review queue capacity? The
auto_classify_thresholddirectly controls the volume of human review. Size it to your team's bandwidth — an overwhelmed review queue becomes a liability, not a safeguard.Are you backtesting or live trading? Backtesting demands higher precision (you need clean data for accurate strategy evaluation). Live trading may tolerate higher recall (you need to catch errors before they affect execution).
Next Steps
If you want to test this pipeline on your data:
- Sign up at tickdb.ai (free tier available, no credit card required)
- Export your historical kline data using the
/v1/market/klineendpoint - Copy the code from this article, configure your symbols and thresholds
- Run the pipeline and review the anomaly flags — adjust thresholds based on what you find
If you need high-quality historical OHLCV data for backtesting, the TickDB free tier provides 10+ years of cleaned, aligned US equity and crypto data. Institutional plans include extended history and tick-level data for supported markets.
If you are building a production data quality system, consider the review queue architecture described in this article. The investment in human-in-the-loop oversight pays for itself in reduced model errors and regulatory compliance.
If you use AI coding assistants, search for the tickdb-market-data SKILL in your tool's marketplace for integrated TickDB data access directly from your development environment.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results. Anomaly detection systems are diagnostic tools — their effectiveness depends on proper configuration and human oversight.