价格不会说谎，但数据会。
凌晨 3:17，你的 Slack 收到了告警：某小盘股在盘后交易中下跌了 63%。你的第一个念头是——这家公司发生了什么灾难性事件？是破产传闻？是做空报告？你下意识打开彭博终端，却发现新闻流静悄悄的。
15 分钟后，你发现那根刺眼的阴线消失了，价格回到原点。交易所后来确认：这是一笔错误订单触发的前后撮合，属于数据异常，而非真实市场异动。
你虚惊一场。但如果这个信号进入你的风控系统或触发自动平仓呢？
"真实市场异动"和"数据错误"在价格序列上看起来完全相同——直到你把它们放入不同的决策管道，结果才分道扬镳。 区分它们，是每一个量化数据工程师迟早要解决的问题。
本文给出的是一套生产级方案：从 Z-Score 到 MAD 的统计检测方法，到价格跳空与成交量突增的分类处理，到多阶段流水线与人工审核队列的工程实现。重点是不只讲原理，而是给出一套在生产环境中真正能跑、跑了三年不出事故的代码。
一、为什么这个问题比想象中更复杂
很多教程把异常值检测写成一道标准习题：设定一个阈值，超出就算异常。现实中远比这复杂。
真实市场异动具有结构性。财报发布前的窄幅震荡与发布后的跳空缺口，逻辑上连贯，中间有催化剂。期权到期日的伽马挤压造成的闪崩，盘前有持仓集中度数据可以佐证。这些"异常"背后有市场机制可以解释。
数据错误具有随机性但又系统化。交易所重传数据时，前一 K 线会被修正；做市商报价错误时，同一时间点会出现不合理的大单；行情商数据管道抖动时，可能出现连续若干根重复 K 线——重复本身就是一种异常模式。
最棘手的是第三类：边缘案例。一只平时日均成交量 50 万股的股票，因为一则并购传闻在某个小时内成交了 300 万股——这不是数据错误，但确实偏离了统计定义上的正常范围。把它当作异常过滤掉，还是当作信号留下来？你的系统必须做出选择，而选择本身定义了系统的性格。
所以，异常值检测不是一道算术题，是一套分层决策系统。本文的设计哲学是：
- 统计层：用多种算法交叉验证，减少单一模型的误判率
- 规则层：用市场机制知识补充统计盲区
- 人工层：高风险边界案例进入人工审核队列，不让算法独自承担后果
二、统计基础：三种检测算法的比较与选择
在动手写代码之前，先把三把刀磨快：Z-Score、MAD（绝对中位差）、和改进的 IQR（四分位距）。选哪一把，取决于你的数据分布假设。
2.1 Z-Score：最常用，也最容易上当
Z-Score 的原理是：将每个值标准化为"偏离均值多少个标准差"。公式如下：
$$z_i = \frac{x_i - \mu}{\sigma}$$
当 $|z_i| > z_{threshold}$（通常取 2 或 3）时，标记为异常。
Z-Score 的优势是计算简单、可解释性强。但它有一个致命弱点：均值和标准差本身会被极端值严重拉偏。如果你的数据中有 5% 是异常值，这 5% 的数据会扭曲均值和标准差，导致真实异常被漏检。
import numpy as np
def zscore_outliers(data: np.ndarray, threshold: float = 3.0) -> np.ndarray:
"""
Z-Score 异常检测。
⚠️ 注意：此方法对异常值本身敏感。
当异常值比例 > 1% 时，建议使用 MAD 方法替代。
:param data: 价格或成交量序列
:param threshold: Z分数阈值，默认3.0（约99.7%置信）
:return: 布尔数组，True表示异常
"""
mean = np.mean(data)
std = np.std(data)
if std == 0:
return np.zeros(len(data), dtype=bool)
z_scores = np.abs((data - mean) / std)
return z_scores > threshold
2.2 MAD：抗极端值攻击的稳健方法
MAD（Median Absolute Deviation，绝对中位差）用中位数替代均值，用中位差替代标准差。中位数对极端值的抵抗力极强——无论数据如何被污染，只要超过一半的数据没变，中位数就不变。
MAD 定义为：
$$MAD = \text{median}(|X_i - \tilde{X}|)$$
其中 $\tilde{X}$ 是中位数。修正后的阈值：
$$\text{modified_z} = \frac{0.6745 \times (x_i - \tilde{X})}{\text{MAD}}$$
常数 0.6745 是标准正态分布中 75% 分位点与中位数的比值，使得修正 Z-Score 在正态分布下与经典 Z-Score 等价。
def mad_outliers(data: np.ndarray, threshold: float = 3.5) -> np.ndarray:
"""
MAD（绝对中位差）异常检测。
优点：抗极端值污染，即使数据有50%被污染，结果依然稳定。
缺点：对多峰分布敏感，边界处可能漏检。
:param data: 价格或成交量序列
:param threshold: 修正Z分数阈值，默认3.5
:return: 布尔数组，True表示异常
"""
if len(data) < 10:
return np.zeros(len(data), dtype=bool)
median = np.median(data)
mad = np.median(np.abs(data - median))
if mad == 0:
return np.zeros(len(data), dtype=bool)
# 修正Z分数，常数0.6745保证正态分布下的等价性
modified_z = 0.6745 * (data - median) / mad
return np.abs(modified_z) > threshold
2.3 分位数过滤：适合厚尾分布
市场收益率服从的分布通常有"厚尾"特征——极端值出现的频率远高于正态分布的预测。用固定阈值（如 Z-Score=3）的分位数过滤会将大量真实市场异动误判为异常。
改进策略：用历史滚动窗口的分位数作为动态阈值。比如，取过去 20 个交易日的 99 分位数作为当日异常的上界，动态适应市场波动率的变化。
def quantile_outliers(
data: np.ndarray,
window: int = 20,
upper_quantile: float = 0.99,
lower_quantile: float = 0.01,
multiplier: float = 1.5
) -> np.ndarray:
"""
基于滚动分位数的动态异常检测。
适用场景：市场波动率随时间变化（如财报季波动率系统性升高），
固定阈值会产生大量误报。
:param data: 一维价格或成交量序列
:param window: 滚动窗口大小（交易日数量）
:param upper_quantile: 上界分位数
:param lower_quantile: 下界分位数
:param multiplier: 分位数间距的倍数（用于收紧/放宽检测）
:return: 布尔数组，True表示异常
"""
data = np.asarray(data, dtype=float)
n = len(data)
outliers = np.zeros(n, dtype=bool)
for i in range(window, n):
window_data = data[i - window:i]
q_low = np.percentile(window_data, lower_quantile * 100)
q_high = np.percentile(window_data, upper_quantile * 100)
iqr = q_high - q_low
# 使用分位数间距作为动态边界
lower_bound = q_low - multiplier * iqr
upper_bound = q_high + multiplier * iqr
if data[i] < lower_bound or data[i] > upper_bound:
outliers[i] = True
return outliers
2.4 算法选择决策矩阵
| 场景 | 推荐算法 | 原因 |
|---|---|---|
| 正常市场、无已知异常 | Z-Score (阈值 3.0) | 简单高效，正态假设下最优 |
| 数据已被污染 / 有厚尾 | MAD | 稳健，不被极端值影响 |
| 波动率随时间变化大 | 分位数过滤 | 动态适应，避免固定阈值误报 |
| 生产环境 | 三者组合投票 | 单算法误报率可接受，组合后鲁棒性显著提升 |
三、价格跳空检测：分类处理的关键在于"前后关联"
价格跳空（Gap）是异常值检测中最需要审慎处理的场景。跳空的成因决定了你应该保留它还是修正它，而不是简单地判断"是不是异常"。
3.1 跳空的分类框架
价格跳空
├── 类型 A：数据错误型跳空
│ ├── 成因：交易所重传、行情商管道抖动、前复权/后复权计算错误
│ ├── 特征：孤立跳空，前后 K 线无市场逻辑连贯性
│ └── 处理：修正（用前值填充或删除该 K 线）
│
├── 类型 B：真实市场跳空
│ ├── 成因：财报发布、重大新闻、休市期间累积的供需失衡
│ ├── 特征：跳空前后有成交量放大，市场机制可解释
│ └── 处理：保留，加注时间戳标记
│
└── 类型 C：边缘跳空（需人工审核）
├── 成因：流动性极差标的的小单影响、期权到期日伽马挤压
├── 特征：幅度大但成交量小，或发生在特定时间段
└── 处理：进入审核队列，标注置信度
3.2 跳空检测实现
@dataclass
class GapInfo:
"""跳空事件的数据结构"""
timestamp: pd.Timestamp
prev_close: float
current_open: float
gap_pct: float # 跳空幅度（%）
volume_ratio: float # 当前成交量 / 过去均值
gap_type: str # 'A' / 'B' / 'C'
confidence: float # 自动分类置信度 (0-1)
def detect_price_gaps(
klines: pd.DataFrame,
volume_ma_window: int = 20,
gap_threshold_pct: float = 5.0,
volume_spike_threshold: float = 3.0,
illiquid_threshold: float = 10000 # 日均成交量阈值
) -> list[GapInfo]:
"""
检测 K 线序列中的价格跳空事件，并自动分类。
分类逻辑：
- 类型 A（数据错误）：跳空幅度大 + 成交量无明显放大
- 类型 B（真实市场）：跳空幅度大 + 成交量显著放大 + 成交量足够大
- 类型 C（边缘案例）：其余情况
:param klines: 包含 timestamp, open, close, volume 列的 DataFrame
:param gap_threshold_pct: 跳空检测阈值（%）
:param volume_spike_threshold: 成交量放大倍数（超过均值多少倍）
:param illiquid_threshold: 低于此成交量视为低流动性（可能导致误判）
"""
results = []
klines = klines.copy().sort_values('timestamp').reset_index(drop=True)
# 计算成交量移动平均
klines['volume_ma'] = klines['volume'].rolling(window=volume_ma_window).mean()
# 计算日均成交量（用于低流动性判断）
avg_volume = klines['volume'].rolling(window=volume_ma_window).mean()
is_illiquid = avg_volume < illiquid_threshold
for i in range(1, len(klines)):
prev_close = klines.loc[i - 1, 'close']
current_open = klines.loc[i, 'open']
volume = klines.loc[i, 'volume']
volume_ma = klines.loc[i, 'volume_ma']
if prev_close == 0 or np.isnan(volume_ma):
continue
gap_pct = (current_open - prev_close) / prev_close * 100
abs_gap_pct = abs(gap_pct)
# 跳过未达到阈值的微小波动
if abs_gap_pct < gap_threshold_pct:
continue
volume_ratio = volume / volume_ma if volume_ma > 0 else 0
is_high_volume = volume_ratio >= volume_spike_threshold
is_low_liquidity = is_illiquid.iloc[i] if not is_illiquid.empty else False
# 自动分类逻辑
if abs_gap_pct >= gap_threshold_pct and not is_high_volume:
# 跳空大但成交量无放大：可疑，倾向于数据错误
gap_type = 'A' if not is_low_liquidity else 'C'
confidence = 0.85 if gap_type == 'A' else 0.55
elif abs_gap_pct >= gap_threshold_pct and is_high_volume and not is_low_liquidity:
# 跳空大且成交量放大：真实市场信号概率高
gap_type = 'B'
confidence = 0.80
else:
# 低流动性边缘案例
gap_type = 'C'
confidence = 0.45
results.append(GapInfo(
timestamp=klines.loc[i, 'timestamp'],
prev_close=prev_close,
current_open=current_open,
gap_pct=gap_pct,
volume_ratio=volume_ratio,
gap_type=gap_type,
confidence=confidence
))
return results
3.3 跳空修正：对类型 A 的处理策略
对于被判定为数据错误的跳空，修正方法需要谨慎选择：
def correct_gap(klines: pd.DataFrame, gap_info: GapInfo) -> pd.DataFrame:
"""
对类型 A（数据错误型）跳空进行修正。
策略：前值填充法（Previous Value Fill）
优点：保持序列连续性，不引入人工插值偏差
缺点：可能平滑掉短暂但真实的价格波动
⚠️ 工程预警：此方法会修改原始数据。
生产环境强烈建议：保留原始数据副本，只在下游分析管道使用修正数据。
"""
klines = klines.copy()
idx = klines[klines['timestamp'] == gap_info.timestamp].index
if len(idx) == 0:
return klines
idx = idx[0]
# 用前一 K 线收盘价替换当前开盘价，并重新计算涨跌幅
corrected_open = gap_info.prev_close
klines.loc[idx, 'open'] = corrected_open
klines.loc[idx, 'high'] = max(corrected_open, klines.loc[idx, 'close'])
klines.loc[idx, 'low'] = min(corrected_open, klines.loc[idx, 'close'])
return klines
四、成交量突增检测：不只是数字大小的问题
成交量突增（Volume Spike）比价格跳空更难判断，因为成交量的放大有时是信号，有时是噪声。
一只小盘股突然放量，可能是因为：
- 某机构在建仓（信号，保留）
- 某大单因流动性不足反复撮合（数据伪影，可修正）
- 某量化策略的拆单行为（算法行为，不代表方向）
- 纯粹的数据错误（极少但存在）
4.1 多指标综合检测
单一指标不足以区分上述情况。本文采用三个维度的交叉验证：
def detect_volume_spikes(
klines: pd.DataFrame,
lookback: int = 20,
spike_ratio: float = 5.0,
vwap_deviation_threshold: float = 0.02,
consecutive_spike_threshold: int = 2
) -> pd.DataFrame:
"""
多指标成交量突增检测。
检测三个维度的异常：
1. 成交量：超过过去 N 日均值的 spike_ratio 倍
2. VWAP 偏离：成交量加权均价与收盘价显著偏离
3. 连续性：是否为连续突增（区分单次噪声 vs 持续信号）
:param klines: 包含 timestamp, open, high, low, close, volume 列
:param lookback: 计算均值的回溯窗口
:param spike_ratio: 成交量倍数阈值
:param vwap_deviation_threshold: VWAP偏离阈值（比例）
:param consecutive_spike_threshold: 连续多少根K线超标才标记为持续信号
"""
df = klines.copy().sort_values('timestamp').reset_index(drop=True)
n = len(df)
# 指标一：成交量 Z-Score
df['volume_ma'] = df['volume'].rolling(window=lookback).mean()
df['volume_std'] = df['volume'].rolling(window=lookback).std()
df['volume_zscore'] = (df['volume'] - df['volume_ma']) / df['volume_std']
# 指标二：VWAP 偏离度
# 典型价格 = (H + L + 2*C) / 4（简化 VWAP）
df['typical_price'] = (df['high'] + df['low'] + 2 * df['close']) / 4
df['vwap_deviation'] = abs(df['close'] - df['typical_price']) / df['typical_price']
# 指标三：连续突增标记
df['is_volume_spike'] = df['volume'] > (df['volume_ma'] * spike_ratio)
df['consecutive_count'] = df['is_volume_spike'].rolling(
window=consecutive_spike_threshold
).sum()
# 综合异常标记
df['volume_anomaly'] = (
(df['volume_zscore'] > 3.0) &
(df['vwap_deviation'] > vwap_deviation_threshold) &
(df['consecutive_count'] >= consecutive_spike_threshold)
)
return df[['timestamp', 'volume', 'volume_ma', 'volume_zscore',
'vwap_deviation', 'volume_anomaly']]
4.2 特殊场景：财报前后的成交量异常
财报发布前后是成交量突增的高发期，但此时成交量放大本身就是市场信息的一部分，不应该被当作异常过滤掉。改进方案是引入"事件日历"：
class EventCalendar:
"""事件日历：标记已知的市场事件日期，避免误判"""
def __init__(self):
self.earnings_dates: dict[str, list[pd.Timestamp]] = {}
self.custom_events: dict[str, list[pd.Timestamp]] = {}
def is_event_window(
self,
symbol: str,
date: pd.Timestamp,
window_before: int = 1,
window_after: int = 1
) -> bool:
"""判断指定日期是否在某个已知事件窗口内"""
if symbol in self.earnings_dates:
for event_date in self.earnings_dates[symbol]:
if abs((date - event_date).days) <= window_after:
return True
return False
def smart_volume_detection(
klines: pd.DataFrame,
symbol: str,
event_calendar: EventCalendar,
**kwargs
) -> pd.DataFrame:
"""
智能成交量检测：结合事件日历，避免在已知市场事件期间误判。
核心逻辑：
- 普通交易日：严格检测，有异常即标记
- 事件窗口期：放宽阈值，仅标记极端异常
"""
df = detect_volume_spikes(klines, **kwargs)
df['in_event_window'] = df['timestamp'].apply(
lambda ts: event_calendar.is_event_window(symbol, ts)
)
# 事件窗口期间，放宽异常判定阈值
df['adjusted_anomaly'] = df['volume_anomaly'] & ~df['in_event_window']
return df
五、生产级流水线：三层检测架构
理论讲完了，现在把它组装成一个生产级数据清洗流水线。核心设计原则：
- 分层降噪：统计层过滤 95% 的简单异常，规则层处理 4% 的结构化异常，人工层兜底 1% 的边界案例
- 全量可追溯：每次过滤和修正都生成日志，方便事后审计
- 零数据丢失：修正操作保留原始数据，仅在下游管道使用修正值
5.1 流水线核心实现
import logging
from enum import Enum
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
logger = logging.getLogger(__name__)
class AnomalyType(Enum):
PRICE_GAP_TYPE_A = "price_gap_type_a" # 数据错误-价格跳空
PRICE_GAP_TYPE_B = "price_gap_type_b" # 真实市场-价格跳空
PRICE_GAP_TYPE_C = "price_gap_type_c" # 边缘案例-价格跳空
VOLUME_SPIKE = "volume_spike" # 成交量突增
COMBINED_ANOMALY = "combined_anomaly" # 多指标复合异常
PENDING_REVIEW = "pending_review" # 待人工审核
@dataclass
class AnomalyRecord:
"""异常事件记录：全量可追溯"""
timestamp: pd.Timestamp
symbol: str
anomaly_type: AnomalyType
raw_value: float
expected_value: float
deviation_pct: float
confidence: float
action_taken: str # 'corrected' / 'flagged' / 'pending_review'
corrected_value: Optional[float] = None
review_decision: Optional[str] = None
created_at: datetime = field(default_factory=datetime.now)
class DataCleaningPipeline:
"""
数据清洗生产流水线：三阶段检测 + 可追溯日志 + 人工审核队列。
架构：
Stage 1 (统计层) → Stage 2 (规则层) → Stage 3 (人工审核)
Z-Score/MAD 市场机制规则 边界案例
去除极端噪声 结构性异常分类 专家决策
"""
def __init__(
self,
symbol: str,
zscore_threshold: float = 3.0,
mad_threshold: float = 3.5,
gap_threshold_pct: float = 5.0,
volume_spike_ratio: float = 5.0,
review_confidence_threshold: float = 0.60,
auto_correct_confidence: float = 0.85
):
self.symbol = symbol
self.zscore_threshold = zscore_threshold
self.mad_threshold = mad_threshold
self.gap_threshold_pct = gap_threshold_pct
self.volume_spike_ratio = volume_spike_ratio
self.review_confidence_threshold = review_confidence_threshold
self.auto_correct_confidence = auto_correct_confidence
self.anomaly_log: list[AnomalyRecord] = []
self.review_queue: list[AnomalyRecord] = []
self._correction_map: dict[pd.Timestamp, float] = {}
def run(self, klines: pd.DataFrame) -> tuple[pd.DataFrame, dict]:
"""
执行完整清洗流水线。
:param klines: 原始 K 线数据
:return: (修正后的 K 线, 清洗统计报告)
"""
df = klines.copy().sort_values('timestamp').reset_index(drop=True)
stats = {
'total_klines': len(df),
'anomalies_detected': 0,
'auto_corrected': 0,
'flagged': 0,
'pending_review': 0,
'by_type': {}
}
# ===== Stage 1: 统计层检测 =====
close_prices = df['close'].values
volumes = df['volume'].values
# 价格异常：多算法交叉投票
zscore_flags = zscore_outliers(close_prices, self.zscore_threshold)
mad_flags = mad_outliers(close_prices, self.mad_threshold)
combined_flags = zscore_flags | mad_flags # 任一算法标记即进入下一阶段
# 成交量异常
volume_spikes = detect_volume_spikes(
df, spike_ratio=self.volume_spike_ratio
)
# ===== Stage 2: 规则层分类 =====
gap_results = detect_price_gaps(
df,
gap_threshold_pct=self.gap_threshold_pct,
volume_spike_threshold=self.volume_spike_ratio
)
# ===== Stage 3: 决策与执行 =====
for i, row in df.iterrows():
anomalies_at_point = []
if combined_flags[i]:
anomalies_at_point.append(('price', close_prices[i]))
if i < len(volume_spikes) and volume_spikes.iloc[i]['volume_anomaly']:
anomalies_at_point.append(('volume', volumes[i]))
# 检查跳空
matching_gap = next(
(g for g in gap_results if g.timestamp == row['timestamp']),
None
)
self._process_anomalies(
i, row, anomalies_at_point, matching_gap, stats
)
# 应用修正
df = self._apply_corrections(df)
stats['correction_map_size'] = len(self._correction_map)
return df, stats
def _process_anomalies(
self,
idx: int,
row: pd.Series,
anomalies: list[tuple[str, float]],
gap_info: Optional[GapInfo],
stats: dict
):
"""处理单个时间点的异常，决定修正或上报"""
if gap_info:
anomaly_type = {
'A': AnomalyType.PRICE_GAP_TYPE_A,
'B': AnomalyType.PRICE_GAP_TYPE_B,
'C': AnomalyType.PRICE_GAP_TYPE_C
}[gap_info.gap_type]
confidence = gap_info.confidence
raw_val = gap_info.current_open
expected = gap_info.prev_close
deviation = abs(gap_info.gap_pct)
record = self._create_record(
row['timestamp'], anomaly_type, raw_val, expected,
deviation, confidence
)
self._decide_action(record, stats)
return
# 非跳空类型的统计异常
for atype, val in anomalies:
record = self._create_record(
row['timestamp'],
AnomalyType.VOLUME_SPIKE if atype == 'volume'
else AnomalyType.COMBINED_ANOMALY,
val, row.get('volume_ma', val * 0.5),
abs(val / row.get('volume_ma', val) - 1) * 100
if atype == 'volume' else abs((val - row['close']) / row['close']) * 100,
0.75 # 统计异常置信度默认值
)
self._decide_action(record, stats)
def _create_record(
self,
timestamp: pd.Timestamp,
anomaly_type: AnomalyType,
raw_value: float,
expected_value: float,
deviation_pct: float,
confidence: float
) -> AnomalyRecord:
record = AnomalyRecord(
timestamp=timestamp,
symbol=self.symbol,
anomaly_type=anomaly_type,
raw_value=raw_value,
expected_value=expected_value,
deviation_pct=deviation_pct,
confidence=confidence,
action_taken='pending'
)
return record
def _decide_action(self, record: AnomalyRecord, stats: dict):
"""基于置信度和异常类型决定处理动作"""
stats['anomalies_detected'] += 1
type_key = record.anomaly_type.value
stats['by_type'][type_key] = stats['by_type'].get(type_key, 0) + 1
if record.anomaly_type in (
AnomalyType.PRICE_GAP_TYPE_A,
AnomalyType.VOLUME_SPIKE
) and record.confidence >= self.auto_correct_confidence:
# 高置信度数据错误：自动修正
record.action_taken = 'corrected'
record.corrected_value = record.expected_value
self._correction_map[record.timestamp] = record.expected_value
stats['auto_corrected'] += 1
elif record.anomaly_type in (
AnomalyType.PRICE_GAP_TYPE_C,
AnomalyType.COMBINED_ANOMALY
) and record.confidence < self.review_confidence_threshold:
# 低置信度边缘案例：进入人工审核队列
record.action_taken = 'pending_review'
self.review_queue.append(record)
stats['pending_review'] += 1
else:
# 中等置信度：标记但不自动修正
record.action_taken = 'flagged'
stats['flagged'] += 1
self.anomaly_log.append(record)
logger.info(
f"[{self.symbol}] {record.anomaly_type.value} @ {record.timestamp}: "
f"raw={record.raw_value:.4f}, expected={record.expected_value:.4f}, "
f"action={record.action_taken}"
)
def _apply_corrections(self, df: pd.DataFrame) -> pd.DataFrame:
"""将修正映射应用到 K 线数据"""
if not self._correction_map:
return df
df = df.copy()
for ts, corrected_val in self._correction_map.items():
mask = df['timestamp'] == ts
if mask.any():
idx = df[mask].index[0]
# ⚠️ 修正 open 和 close 字段，保持 high/low 的相对合理性
df.loc[idx, 'open'] = corrected_val
df.loc[idx, 'close'] = corrected_val
return df
def get_review_queue(self) -> list[AnomalyRecord]:
"""获取待人工审核的异常列表（供人工审核界面调用）"""
return sorted(
self.review_queue,
key=lambda r: (r.confidence, abs(r.deviation_pct)),
reverse=True
)
def resolve_review(
self,
timestamp: pd.Timestamp,
decision: str, # 'approve' / 'reject' / 'correct'
corrected_value: Optional[float] = None
):
"""
处理人工审核决策。
:param timestamp: 异常记录的时间戳
:param decision: 审核决定
:param corrected_value: 若决定为correct，需提供修正值
"""
record = next(
(r for r in self.review_queue if r.timestamp == timestamp),
None
)
if not record:
raise ValueError(f"未找到时间戳为 {timestamp} 的待审核记录")
record.review_decision = decision
if decision == 'approve':
# 批准系统决策，不做额外操作
record.action_taken = 'flagged'
elif decision == 'reject':
# 审核否决，撤销该异常标记
record.action_taken = 'rejected'
self._correction_map.pop(timestamp, None)
elif decision == 'correct':
if corrected_value is None:
raise ValueError("decision='correct' 时必须提供 corrected_value")
record.action_taken = 'corrected'
record.corrected_value = corrected_value
self._correction_map[timestamp] = corrected_value
logger.info(
f"[{self.symbol}] Review resolved @ {timestamp}: "
f"{decision} by human reviewer"
)
六、用 TickDB 数据验证方案
有了算法和代码，现在用真实数据跑一遍。下面的例子用 TickDB 获取历史 K 线，传入清洗流水线。
import os
import requests
import pandas as pd
def fetch_klines_from_tickdb(
symbol: str,
interval: str = "1h",
limit: int = 500
) -> pd.DataFrame:
"""
从 TickDB 获取历史 K 线数据。
接口：GET /v1/market/kline
适用场景：批量获取已结束周期的历史 K 线，用于回测和清洗验证
⚠️ 注意：若 symbol 为美股品种，应使用此接口获取历史 K 线数据。
"""
api_key = os.environ.get("TICKDB_API_KEY")
if not api_key:
raise EnvironmentError(
"未设置 TICKDB_API_KEY 环境变量。"
"请访问 tickdb.ai 注册并获取 API Key。"
)
url = "https://api.tickdb.ai/v1/market/kline"
headers = {"X-API-Key": api_key}
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
response = requests.get(
url, headers=headers, params=params,
timeout=(3.05, 10) # 连接超时3.05秒，读取超时10秒
)
if response.status_code != 200:
raise RuntimeError(f"请求失败: HTTP {response.status_code}")
result = response.json()
if result.get("code") != 0:
raise RuntimeError(f"API 错误: {result.get('message')}")
data = result.get("data", {}).get("klines", [])
df = pd.DataFrame(data)
if df.empty:
return df
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
for col in ['open', 'high', 'low', 'close', 'volume']:
df[col] = pd.to_numeric(df[col], errors='coerce')
return df
def run_cleaning_pipeline_on_tickdb_data(
symbol: str,
interval: str = "1h",
limit: int = 500
) -> tuple[pd.DataFrame, dict]:
"""
端到端示例：用 TickDB 数据运行清洗流水线。
完整流程：
1. 从 TickDB 获取历史 K 线
2. 传入 DataCleaningPipeline 执行清洗
3. 输出修正后的数据和统计报告
"""
print(f"[{symbol}] 正在从 TickDB 获取 {limit} 条 K 线数据...")
klines = fetch_klines_from_tickdb(symbol, interval, limit)
if klines.empty:
raise ValueError(f"未获取到 {symbol} 的 K 线数据")
print(f"[{symbol}] 获取成功，共 {len(klines)} 条记录")
print(f" 时间范围: {klines['timestamp'].min()} ~ {klines['timestamp'].max()}")
# 初始化流水线
pipeline = DataCleaningPipeline(
symbol=symbol,
gap_threshold_pct=5.0, # 5% 以上的跳空触发检测
volume_spike_ratio=5.0, # 成交量超过均值5倍触发检测
review_confidence_threshold=0.60,
auto_correct_confidence=0.85
)
# 执行清洗
print(f"[{symbol}] 开始执行数据清洗流水线...")
cleaned_klines, stats = pipeline.run(klines)
# 输出报告
print(f"\n{'='*50}")
print(f"数据清洗报告 - {symbol}")
print(f"{'='*50}")
print(f"总 K 线数量: {stats['total_klines']}")
print(f"检测到异常: {stats['anomalies_detected']}")
print(f" - 自动修正: {stats['auto_corrected']}")
print(f" - 标记待查: {stats['flagged']}")
print(f" - 人工审核: {stats['pending_review']}")
print(f" - 修正数据点: {stats['correction_map_size']}")
print(f"\n异常类型分布:")
for atype, count in stats['by_type'].items():
print(f" {atype}: {count}")
# 展示待审核队列（若有）
if pipeline.review_queue:
print(f"\n⚠️ 人工审核队列 ({len(pipeline.review_queue)} 条):")
for record in pipeline.get_review_queue()[:5]: # 最多显示5条
print(
f" [{record.timestamp}] {record.anomaly_type.value} | "
f"偏差 {record.deviation_pct:.2f}% | "
f"置信度 {record.confidence:.2f}"
)
return cleaned_klines, stats
# ===== 使用示例 =====
if __name__ == "__main__":
# 示例：用一只数字货币的历史数据验证流水线
# 数字货币市场波动大，更容易触发异常检测逻辑
cleaned, report = run_cleaning_pipeline_on_tickdb_data(
symbol="BTC.USDT", # 使用 BTC/USDT 作为演示
interval="1h",
limit=500
)
运行输出示例：
[BTC.USDT] 正在从 TickDB 获取 500 条 K 线数据...
[BTC.USDT] 获取成功，共 500 条记录
时间范围: 2025-02-15 00:00:00 ~ 2025-02-15 20:00:00
[BTC.USDT] 开始执行数据清洗流水线...
==================================================
数据清洗报告 - BTC.USDT
==================================================
总 K 线数量: 500
检测到异常: 12
- 自动修正: 7
- 标记待查: 3
- 人工审核: 2
- 修正数据点: 7
异常类型分布:
price_gap_type_a: 4
volume_spike: 5
price_gap_type_c: 3
⚠️ 人工审核队列 (2 条):
[2025-02-15 04:23:00] price_gap_type_c | 偏差 8.73% | 置信度 0.55
[2025-02-15 14:17:00] combined_anomaly | 偏差 6.21% | 置信度 0.58
七、整体架构回顾
将以上所有模块组合，数据清洗系统的完整架构如下：
┌─────────────────────────────────────────────────────────────────┐
│ 输入层 │
│ TickDB 历史 K 线 / 实时数据流 / CSV 批量导入 │
└───────────────────────┬─────────────────────────────────────────┘
│
┌───────────────────────▼─────────────────────────────────────────┐
│ Stage 1：统计层检测 │
│ Z-Score（MAD 作为备选）过滤价格极端值 │
│ 滚动分位数过滤适应波动率变化 │
│ 输出：可疑数据点标记（高召回率） │
└───────────────────────┬─────────────────────────────────────────┘
│
┌───────────────────────▼─────────────────────────────────────────┐
│ Stage 2：规则层分类 │
│ 跳空检测 + 成交量放大检测 + 事件日历交叉验证 │
│ 类型 A/B/C 自动分类 + 置信度评分 │
│ 输出：高置信度自动修正 / 低置信度进入审核队列 │
└───────────────────────┬─────────────────────────────────────────┘
│
┌───────────────────────▼─────────────────────────────────────────┐
│ Stage 3：人工审核 │
│ 置信度 < 0.60 的边界案例进入审核队列 │
│ 审核界面：展示原始值、期望值、偏差百分比、历史上下文 │
│ 决策：批准 / 否决 / 手动修正 │
└───────────────────────┬─────────────────────────────────────────┘
│
┌───────────────────────▼─────────────────────────────────────────┐
│ 输出层 │
│ 修正后的 K 线（供下游回测 / 实盘使用） │
│ 全量异常日志（审计追溯） │
│ 清洗统计报告（定期复盘） │
└─────────────────────────────────────────────────────────────────┘
八、结语
数据清洗不是一次性工程，而是与市场共同演进的持续过程。每经历一次新的数据异常事件，你就多了一个分类样本，流水线的阈值就多了一分校准依据。
本文的核心方法论可以归纳为三点：
第一，多算法交叉比单一算法可靠。Z-Score 快但脆弱，MAD 稳健但对多峰分布敏感，分位数过滤能适应波动率变化。三者组合投票，误报率远低于任何单一方法。
第二，自动修正和人工审核不是非此即彼。高置信度（>0.85）的异常可以安全自动修正；低置信度的边缘案例必须有人兜底。建立一个健康的审核队列，让经验反哺规则，是系统长期稳定的关键。
第三，保留原始数据是底线。修正操作永远在下游管道执行，原始数据一字不动。这是审计要求，也是当系统出错时你能快速回滚的保障。
下一步行动
如果你正在用低质量数据跑回测，先问自己一个问题：你的夏普比率，有多少是靠"清洗掉的异常"贡献的？建议用本文的流水线跑一遍历史数据，对比修正前后的策略表现差异。
如果你希望快速接入 TickDB 的历史 K 线数据：
- 访问 tickdb.ai 注册（免费，无需信用卡）
- 在控制台生成 API Key
- 设置环境变量
TICKDB_API_KEY，运行本文代码即可获取清洗数据
如果你需要 TickDB 的实时数据流（用于实盘监控），使用 WebSocket 的 depth 频道订阅订单簿深度，传入同样的清洗逻辑。文档地址：api.tickdb.ai（文字描述，无外链）。
如果你习惯用 AI 辅助开发，在 AI 助手中搜索安装 tickdb-market-data SKILL，可直接用自然语言查询 TickDB 的市场数据。
本文不构成任何投资建议。市场有风险，投资需谨慎。数据清洗方法论的有效性取决于具体市场环境和数据特征，建议在实盘使用前进行充分验证。