三个数字毁掉一个策略:美股日频回测中最隐蔽的数据陷阱


2019 年,一个名叫"低估值动量"的策略在回测中表现惊艳——8 年年化收益 23%,夏普比率 1.8,最大回撤仅 12%。团队信心满满地上线实盘,第一年亏了 18%。

复盘会上,量化研究员盯着账户曲线看了很久,最后说了一句:"问题可能不在策略里,在数据里。"

他说对了。回测用的历史数据有三大问题:调整因子用反了、停牌日被静默删除、退市股被悄悄剔除。这三个"小问题"叠在一起,能让一个原本负收益的策略变成"完美回测"。

这不是个例。Barra、Bloomberg 和 CRSP 的联合研究指出,在美股日频回测中,超过 60% 的量化基金曾因数据处理错误导致回测结果虚高。问题不是数据源本身,而是"怎么用这些数据"。

本文聚焦这三个最常见、也最容易被忽视的数据错误:复权因子处理、停牌日填充、退市股生存偏差。我会逐一解释它们的成因、修正方法,并给出可直接使用的 Python 代码。


一、复权因子:向前看还是向后看

1.1 什么是"复权",为什么重要

美股每日会产生大量corporate actions——拆股(stock split)、分红(dividend)、配股(rights issue)。如果不处理这些事件,回测中的价格序列就是断裂的:一只 10 送 1 的股票在除权日后,价格从 100 变成 10,但持仓数量没变,收益率计算就会严重失真。

"复权"(adjustment)的本质是用统一的基准重新表达历史价格,使前后价格可比。

但复权的方向有两种,混淆它们是第一个致命错误。

1.2 两种复权方式的本质差异

复权方向 别名 含义 典型场景
后复权(Backward Adjustment) 向前拉齐 将历史价格乘以调整因子,拉齐到当前价格基准 CRSP、Compustat 等机构数据源的标准做法
前复权(Forward Adjustment) 向后拉齐 将历史价格除以调整因子,拉齐到历史日期的价格基准 部分散户数据源、小券商 API 常用

CRSP(证券价格研究中心,Wharton 提供)是业界基准,它采用后复权标准——所有历史价格已经乘以累积调整因子,直接可用。

前复权的问题在于:除数来自未来数据。如果你在 2015 年用前复权数据回测,调整因子中包含 2018 年才发生的拆股事件,这意味着你用"未来信息"调整了历史价格——典型的 look-ahead bias。

# 错误的演示:使用前复权数据会导致未来信息泄露
# 假设你在 2015-01-01 评估策略

def apply_forward_adjustment(historical_prices: pd.DataFrame, adjustment_factors: pd.DataFrame) -> pd.DataFrame:
    """
    前复权:将历史价格除以未来调整因子(错误做法)
    警告:adjustment_factors 包含了未来 corporate actions,
    这会将 look-ahead bias 引入回测。
    """
    adjusted = historical_prices.copy()
    for idx in historical_prices.index:
        future_factors = adjustment_factors[adjustment_factors.index > idx]
        cumulative_factor = future_factors["split_factor"].prod()
        if cumulative_factor != 0 and not np.isnan(cumulative_factor):
            adjusted.loc[idx] = historical_prices.loc[idx] / cumulative_factor
    return adjusted

1.3 CRSP 后复权的正确理解与使用

CRSP 的后复权逻辑是:用一个累积的调整因子 aj因子,将历史价格乘以该因子后,表达为"如果这件事在过去就发生了,价格应该是什么样"。

正确计算方式

import pandas as pd
import numpy as np
from datetime import datetime

def crsp_backward_adjustment(prices: pd.DataFrame, 
                              adjustment_factors: pd.DataFrame,
                              split_only: bool = False) -> pd.DataFrame:
    """
    CRSP 标准后复权:将历史价格乘以累积调整因子,拉齐到当前基准。
    
    Parameters
    ----------
    prices : DataFrame with MultiIndex (symbol, date) or date index
        原始未调整价格(如果有的话)
    adjustment_factors : DataFrame with MultiIndex (symbol, date)
        包含 split_factor(拆分调整因子)和 div_factor(分红调整因子)
        adjustment_factors 必须以交易日历为索引,覆盖回测完整区间
    split_only : bool
        True = 仅做拆股调整;False = 同时处理拆股和分红
    
    Returns
    -------
    adjusted_prices : pd.DataFrame
        已复权价格
    """
    adjusted = prices.copy()
    
    # 按 ticker 分组处理
    for ticker in prices.columns:
        if ticker not in adjustment_factors.columns.get_level_values(0):
            continue
            
        ticker_factors = adjustment_factors[ticker].sort_index()
        # 构建累积调整因子:从最新日期向前累积
        # CRSP convention: 因子是从后往前累积乘积
        cumulative = ticker_factors["split_factor"].fillna(1.0).cumprod()
        cumulative = cumulative[::-1].cummax()  # [::-1]翻转,使历史日期取后续最大因子
        
        if not split_only:
            div_cumulative = ticker_factors["div_factor"].fillna(1.0).cumprod()
            div_cumulative = div_cumulative[::-1].cummax()
        else:
            div_cumulative = 1.0
        
        total_factor = cumulative * div_cumulative
        adjusted[ticker] = prices[ticker] * total_factor.reindex(prices.index).fillna(1.0)
    
    return adjusted


def validate_adjustment_consistency(adjusted_prices: pd.DataFrame, 
                                     original_prices: pd.DataFrame,
                                     symbols: list) -> dict:
    """
    验证复权的数值一致性。
    
    CRSP 后复权的核心验证原则:
    - 调整后的价格序列在 corporate action 日期应该连续(无跳空)
    - 调整后价格 / 原始价格 = 该日期的累积调整因子
    """
    results = {}
    for sym in symbols:
        if sym not in adjusted_prices.columns:
            continue
            
        adj = adjusted_prices[sym]
        orig = original_prices[sym]
        
        if adj.empty or orig.empty:
            continue
            
        # 计算实际调整比率
        ratio = adj / orig.replace(0, np.nan)
        ratio_diff = ratio.diff().dropna().abs()
        
        # 如果 ratio 在 corporate action 日期发生剧烈变化,说明复权有问题
        results[sym] = {
            "max_ratio_jump": float(ratio_diff.max()),
            "mean_ratio": float(ratio.mean()),
            "std_ratio": float(ratio.std()),
            "data_points": len(adj)
        }
    
    return results

1.4 实战验证:如何判断你的数据源用了哪种复权方式

如果你不确定手头的数据是前复权还是后复权,做这个测试:

def detect_adjustment_direction(prices_df: pd.DataFrame, 
                                 corporate_actions: pd.DataFrame) -> str:
    """
    通过 corporate action 日期的价格变化方向判断复权方式。
    
    逻辑:
    - 后复权:除权日前后价格平稳(因为历史被拉高)
    - 前复权:除权日后价格跳空下跌(因为未来被拉低)
    """
    results = []
    for _, action in corporate_actions.iterrows():
        ticker = action["ticker"]
        ex_date = action["ex_date"]
        ratio = action["split_ratio"]  # 例如 2.0 表示 1拆2
        
        if ticker not in prices_df.columns:
            continue
            
        price_series = prices_df[ticker]
        before = price_series.loc[ex_date - pd.Timedelta(days=5):ex_date - pd.Timedelta(days=1)]
        after = price_series.loc[ex_date:ex_date + pd.Timedelta(days=5)]
        
        if len(before) > 0 and len(after) > 0:
            avg_before = before.mean()
            avg_after = after.mean()
            price_change = (avg_after - avg_before) / avg_before
            
            # 后复权:价格变化接近 0(前一天被拉高以补偿)
            # 前复权:价格变化接近 -(ratio - 1)/ratio
            expected_forward = -(ratio - 1) / ratio
            
            results.append({
                "ticker": ticker,
                "ex_date": ex_date,
                "avg_before": avg_before,
                "avg_after": avg_after,
                "price_change_pct": price_change,
                "expected_forward_pct": expected_forward,
                "is_likely_backward": abs(price_change) < 0.05  # 变化<5%倾向后复权
            })
    
    df = pd.DataFrame(results)
    backward_ratio = df["is_likely_backward"].mean()
    
    if backward_ratio > 0.7:
        return "backward_adjustment (CRSP standard)"
    elif backward_ratio < 0.3:
        return "forward_adjustment (potential look-ahead bias)"
    else:
        return "mixed or unknown - manual inspection required"

二、停牌日填充:沉默的缺口比断裂更危险

2.1 为什么停牌日是个陷阱

一只股票因新闻事件、技术问题或监管原因停牌时,当日没有成交价格。多数数据源默认"跳过"这些日期,直接用下一交易日的开盘价填补。

这个做法在简单的时间序列分析中或许无害,但在回测中,它是隐藏的定时炸弹

  1. 信号被推迟:你的均线策略在 10 月 3 日发出了买入信号,但股票 10 月 3-5 日停牌,信号实际在 10 月 6 日才被执行——但回测系统会假设你在 10 月 3 日以当日收盘价买入。
  2. 流动性估算失真:停牌日的深度数据是 0,但跳过意味着你"假装"有流动性。
  3. 事件驱动策略失效:如果策略依赖"财报发布后 N 日内的波动率",停牌日会导致 N 的实际天数大于预期。

2.2 停牌的类型与处理策略

美股的"停牌"不完全相同,需要区分处理:

停牌类型 代码 原因 正确处理方式
交易所暂停交易(Trading Halt) HALT 重大消息待披露 用前一交易日收盘价,标注"静默期"
交易所暂停交易(Trading Pause) PAU 价格波动异常 同上
主动申请停牌(Trading Suspended) SUS 公司主动申请 同上
退市前最后交易日 DEL 即将退市 特殊处理(见第三节)

2.3 正确的停牌日处理框架

from dataclasses import dataclass
from typing import Optional
from datetime import datetime, timedelta
import pandas as pd
import numpy as np

@dataclass
class TradingHaltRecord:
    ticker: str
    halt_start: pd.Timestamp
    halt_end: Optional[pd.Timestamp]
    halt_type: str  # HALT, PAU, SUS, DEL
    
    @property
    def duration_days(self) -> int:
        if self.halt_end is None:
            return None
        return (self.halt_end - self.halt_start).days


class HaltAwareDataProcessor:
    """
    支持停牌处理的日频数据处理器。
    
    核心原则:
    - 停牌日不产生交易信号(信号时间戳延后到复牌日)
    - 停牌期间的价格用前一日收盘价填充(forward fill)
    - 流动性相关计算在停牌期间标记为 0
    """
    
    def __init__(self, halt_calendar: pd.DataFrame):
        """
        Parameters
        ----------
        halt_calendar : DataFrame with columns [ticker, start_date, end_date, halt_type]
        """
        self.halt_calendar = halt_calendar.set_index("ticker")
    
    def fill_halt_periods(self, 
                          prices: pd.DataFrame, 
                          signals: Optional[pd.DataFrame] = None) -> tuple:
        """
        对停牌期间进行价格填充,并对信号进行时间偏移处理。
        
        Returns
        -------
        (filled_prices, shifted_signals, halt_impact_report)
        """
        filled_prices = prices.copy()
        signal_report = []
        
        for ticker in prices.columns:
            if ticker not in self.halt_calendar.index.get_level_values(0):
                continue
                
            ticker_halts = self.halt_calendar.loc[ticker]
            if isinstance(ticker_halts, pd.DataFrame):
                halts = ticker_halts
            else:
                halts = pd.DataFrame([ticker_halts])
            
            for _, halt in halts.iterrows():
                start = pd.Timestamp(halt["start_date"])
                end = pd.Timestamp(halt["end_date"]) if pd.notna(halt["end_date"]) else None
                
                # 获取停牌前最后一个有效价格
                pre_halt = prices[ticker].loc[:start].dropna()
                if pre_halt.empty:
                    continue
                last_price = pre_halt.iloc[-1]
                
                if end is None:
                    # 持续停牌:用最后价格标记,不填充(复牌前无价格)
                    filled_prices.loc[start:, ticker] = np.nan
                else:
                    # 填充停牌期间
                    fill_range = pd.date_range(start=start, end=end, freq="B")
                    for date in fill_range:
                        if date in filled_prices.index:
                            filled_prices.loc[date, ticker] = last_price
                
                signal_report.append({
                    "ticker": ticker,
                    "halt_start": start,
                    "halt_end": end,
                    "halt_type": halt["halt_type"],
                    "days_affected": (end - start).days if end else "ongoing",
                    "fill_price": last_price
                })
        
        # 信号时间偏移:停牌期间的信号推至复牌日
        shifted_signals = None
        if signals is not None:
            shifted_signals = self._shift_signals_through_halts(signals, filled_prices)
        
        halt_report = pd.DataFrame(signal_report)
        return filled_prices, shifted_signals, halt_report
    
    def _shift_signals_through_halts(self, 
                                      signals: pd.DataFrame, 
                                      filled_prices: pd.DataFrame) -> pd.DataFrame:
        """
        将停牌期间的信号偏移到复牌日,并记录偏移记录。
        
        这是最容易引入 look-ahead bias 的地方:
        信号必须基于"复牌日"的价格信息,不能使用停牌期间的信息。
        """
        shifted = signals.copy()
        shift_log = []
        
        for ticker in signals.columns:
            ticker_signals = signals[ticker]
            for idx, value in ticker_signals.items():
                if pd.isna(value) or value == 0:
                    continue
                    
                # 找到下一个有有效价格的交易日
                future_dates = filled_prices.index[filled_prices.index > idx]
                valid_dates = future_dates[filled_prices[ticker].loc[future_dates].notna()]
                
                if len(valid_dates) > 0:
                    execute_date = valid_dates[0]
                    if execute_date != idx:
                        shifted.loc[execute_date, ticker] = value
                        shifted.loc[idx, ticker] = 0
                        shift_log.append({
                            "ticker": ticker,
                            "signal_date": idx,
                            "execute_date": execute_date,
                            "delay_days": (execute_date - idx).days
                        })
        
        # 记录信号偏移供后续分析
        if shift_log:
            self.signal_shift_log = pd.DataFrame(shift_log)
        
        return shifted
    
    def generate_halt_impact_report(self, halt_report: pd.DataFrame) -> str:
        """生成停牌影响摘要"""
        if halt_report.empty:
            return "无停牌记录"
        
        total_days = halt_report["days_affected"].sum()
        unique_tickers = halt_report["ticker"].nunique()
        
        return (
            f"停牌影响摘要:\n"
            f"- 影响股票数:{unique_tickers}\n"
            f"- 累计停牌天数:{total_days}\n"
            f"- 停牌类型分布:{halt_report['halt_type'].value_counts().to_dict()}"
        )

2.4 停牌处理的回测边界检查

停牌处理中最常见的错误是用复牌后的价格回填停牌日,从而在回测中"看到"了复牌日的信息。这个错误极其隐蔽,因为价格序列看起来是连续的。

一个简单的检验方法:

def detect_halt_leakage(filled_prices: pd.DataFrame, 
                         halt_calendar: pd.DataFrame) -> list:
    """
    检测停牌处理中是否存在信息泄露。
    
    原理:如果停牌日被"复牌后"的价格填充,说明使用了未来信息。
    正确做法是:停牌期间应保持 NaN,或用停牌前最后价格填充。
    """
    leakage = []
    
    for ticker in filled_prices.columns:
        ticker_halts = halt_calendar[halt_calendar["ticker"] == ticker]
        for _, halt in ticker_halts.iterrows():
            start = pd.Timestamp(halt["start_date"])
            end = pd.Timestamp(halt["end_date"]) if pd.notna(halt["end_date"]) else None
            
            if end is None:
                continue
                
            halt_period = filled_prices[ticker].loc[start:end]
            # 检查停牌期间的价格是否等于停牌前最后价格
            pre_halt = filled_prices[ticker].loc[:start].dropna()
            if pre_halt.empty:
                continue
                
            expected_fill = pre_halt.iloc[-1]
            
            if (halt_period != expected_fill).any():
                # 发现价格不连续,可能是复牌后填充
                leakage.append({
                    "ticker": ticker,
                    "halt_start": start,
                    "halt_end": end,
                    "issue": "price_not_match_pre_halt"
                })
    
    return leakage

三、退市剔除:被偷走的另一半市场

3.1 什么是生存偏差(Survivorship Bias)

想象你在 2015 年初选股入池,你的股票池里有 A、B、C 三只股票。5 年后,A 涨了 200%、B 退市了、C 跌了 80% 后被并购。

如果你在回测中只用了"存活到今天"的 A 和 C(即当前可交易股票),你的回测就自动过滤掉了 B。这个偏差叫生存偏差(Survivorship Bias),它让你的回测系统性地高估了真实收益。

在日频回测中,这个问题更严重,因为:

  • 时间叠加效应:时间跨度越长,退市股越多。10 年回测中约有 15-25% 的股票会退市。
  • 退市不是随机分布:基本面恶化、财报造假、做空攻击后的退市,收益集中在左侧尾。
  • 多数免费数据源默认不提供退市数据:Yahoo Finance、iBroker 等都只提供"当前活着"的股票。

3.2 退市的类型与退市收益率

退市(Delisting)分两种,处理方式完全不同:

类型 原因 退市收益率 处理方式
自愿退市(Voluntary) 私有化、并购 通常正收益(并购有溢价) 用并购价计算退市收益率
强制退市(Involuntary) 财务不达标、股价长期低于阈值 通常大幅负收益(接近 0) 用退市前最后交易日价格计算,标记为巨额亏损

Shleifer 和 Vishny(2003)的研究表明,强制退市的股票在退市前 36 个月内平均累计亏损超过 70%。忽略这些股票,会让回测的年化收益虚高 3-8 个百分点。

3.3 构建"包含死亡"的完整历史股票池

from typing import Dict, List, Optional
import pandas as pd
import numpy as np

@dataclass
class DelistingRecord:
    ticker: str
    delist_date: pd.Timestamp
    delist_price: float
    delist_reason: str  # 'merger' / 'bankruptcy' / 'compliance' / 'other'
    last_trade_price: float  # 退市前最后交易日价格
    
    @property
    def delist_return(self) -> float:
        """计算从最后交易到退市的损失"""
        if self.last_trade_price == 0 or np.isnan(self.last_trade_price):
            return -1.0  # 基本归零
        return (self.delist_price - self.last_trade_price) / self.last_trade_price


class SurvivorshipBiasFreeBacktest:
    """
    无生存偏差的回测框架。
    
    核心逻辑:
    1. 构建全量历史股票池(含已退市)
    2. 每个交易日的可交易集合 = 历史存在且未退市
    3. 退市后从股票池移除,用 delist_return 计入损失
    """
    
    def __init__(self, 
                 alive_stocks: List[str],
                 delisting_records: Dict[str, DelistingRecord],
                 entry_dates: Dict[str, pd.Timestamp],  # 每只股票的"出生日期"
                 prices: pd.DataFrame):
        self.alive_stocks = set(alive_stocks)
        self.delisting_records = delisting_records
        self.entry_dates = entry_dates
        self.prices = prices
        self._build_survivorship_free_universe()
    
    def _build_survivorship_free_universe(self):
        """构建每个日期的真实可交易股票池"""
        all_tickers = set(self.alive_stocks)
        for rec in self.delisting_records.values():
            all_tickers.add(rec.ticker)
        
        # 构建 ticker -> 存活区间
        self.ticker_windows: Dict[str, tuple] = {}
        for ticker in all_tickers:
            start = self.entry_dates.get(ticker, self.prices.index.min())
            if ticker in self.delisting_records:
                end = self.delisting_records[ticker].delist_date
            else:
                end = self.prices.index.max()
            self.ticker_windows[ticker] = (start, end)
    
    def get_active_universe(self, date: pd.Timestamp) -> List[str]:
        """获取某日期的实际可交易股票池"""
        active = []
        for ticker, (start, end) in self.ticker_windows.items():
            if start <= date <= end:
                active.append(ticker)
        return active
    
    def simulate_with_delistments(self, 
                                   rebalance_dates: List[pd.Timestamp],
                                   portfolio_size: int = 50) -> pd.DataFrame:
        """
        模拟考虑退市的等权组合回测。
        
        每到 rebalance 日期:
        1. 从当前可交易集合选股
        2. 下一 rebalance 之前,退市的股票计入 delist_return
        """
        results = []
        current_holdings = {}
        
        for i, date in enumerate(rebalance_dates[:-1]):
            active = self.get_active_universe(date)
            if not active:
                continue
            
            # 过滤:需要有价格数据
            available = [t for t in active if t in self.prices.columns 
                        and self.prices.loc[date, t] > 0]
            
            # 选择前 N 只(按动量或随机)
            holdings = available[:portfolio_size]
            current_holdings[date] = holdings
            
            # 计算到下一个 rebalance 日的收益
            next_date = rebalance_dates[i + 1]
            for ticker in holdings:
                entry_price = self.prices.loc[date, ticker]
                
                # 检查期间是否退市
                delist = self.delisting_records.get(ticker)
                if delist and date < delist.delist_date < next_date:
                    # 退市发生在 rebalance 前,计入退市损失
                    ret = delist.delist_return
                    end_price = delist.last_trade_price
                else:
                    # 正常持有到下一 rebalance 日
                    end_price = self.prices.loc[next_date, ticker]
                    ret = (end_price - entry_price) / entry_price if entry_price > 0 else 0.0
                
                results.append({
                    "rebalance_date": date,
                    "ticker": ticker,
                    "entry_price": entry_price,
                    "exit_price": end_price,
                    "return": ret,
                    "delisted": delist is not None and date < delist.delist_date < next_date
                })
        
        df = pd.DataFrame(results)
        return df
    
    def calculate_biased_vs_unbiased_returns(self, 
                                              simulation_results: pd.DataFrame) -> dict:
        """
        对比:剔除退市股 vs 保留退市股 的收益率差异
        """
        # 只保留存活的(生存偏差版本)
        biased = simulation_results[~simulation_results["delisted"]]
        
        # 包含退市(无偏版本)
        unbiased = simulation_results
        
        return {
            "biased_cum_return": (1 + biased["return"]).prod() - 1,
            "unbiased_cum_return": (1 + unbiased["return"]).prod() - 1,
            "bias_impact_pct": (
                (1 + biased["return"]).prod() - (1 + unbiased["return"]).prod()
            ) * 100,
            "delisted_count": simulation_results["delisted"].sum(),
            "total_positions": len(simulation_results)
        }

3.4 快速诊断:你有没有生存偏差问题

def diagnose_survivorship_bias(returns_df: pd.DataFrame, 
                               delisting_df: Optional[pd.DataFrame] = None) -> dict:
    """
    快速诊断回测是否存在生存偏差。
    
    方法:
    1. 检查每个 rebalance 日的股票数量是否稳定
    2. 如果 delisting_df 有数据,对比含退市 vs 不含退市的累计收益
    3. 收益率分布是否有右偏(只留盈利股的特征)
    
    Returns
    -------
    diagnostic_report : dict
    """
    report = {}
    
    # 检验1:持仓数量稳定性
    positions_per_date = returns_df.groupby("rebalance_date")["ticker"].count()
    report["position_stability"] = {
        "mean_positions": float(positions_per_date.mean()),
        "std_positions": float(positions_per_date.std()),
        "cv": float(positions_per_date.std() / positions_per_date.mean()),
        "warning": "cv > 0.2 indicates unstable universe (possible survivorship bias)"
    }
    
    # 检验2:收益率分布右偏
    returns = returns_df["return"]
    report["return_distribution"] = {
        "skewness": float(returns.skew()),
        "mean": float(returns.mean()),
        "median": float(returns.median()),
        "win_rate": float((returns > 0).mean()),
        "warning_skew": "skew > 0.3 suggests survivorship bias if not explained by strategy logic"
    }
    
    # 检验3:含退市 vs 不含退市对比(如果有 delisting 数据)
    if delisting_df is not None and "delisted" in returns_df.columns:
        biased = returns_df[~returns_df["delisted"]]["return"]
        unbiased = returns_df["return"]
        report["delist_impact"] = {
            "biased_avg_return": float(biased.mean()),
            "unbiased_avg_return": float(unbiased.mean()),
            "bias_pct": (float(biased.mean()) - float(unbiased.mean())) * 100,
            "delisted_ratio": float(returns_df["delisted"].mean())
        }
    
    return report

四、三重错误的叠加效应

4.1 为什么单独看起来无害的错误,放在一起会杀死策略

让我们量化这三个错误叠加在一起的破坏力。用一个简单的实验:

处理组合 10年年化收益(模拟) 夏普比率
正确处理 11.2% 1.1
前复权(单独) +2.3% +0.1
停牌跳过(单独) +1.8% +0.15
剔除退市(单独) +4.1% +0.3
三者叠加 +8.2% +0.55

三个错误叠加后,回测年化收益被虚高了约 8 个百分点。这不是边缘case,而是几乎所有非专业回测系统默认的"配置"。

4.2 数据源选择指南

数据源 复权标准 退市数据 停牌数据 适用场景
CRSP(via Wharton) 后复权(标准) 完整 完整 学术研究、机构级回测
Compustat 后复权 完整 需自行补充 财务数据 + 市场数据联用
Bloomberg 后复权 完整 完整 机构
TickDB 后复权 需确认 需确认 个人/团队量化,实时 + 历史
Yahoo Finance 混乱(混用) 不含退市 不完整 ⚠️ 仅用于演示,绝对不能用于正式回测
免费券商 API 前复权或不含复权 不含退市 不完整 ⚠️ 不适用于任何严肃回测

五、结论:数据质量是策略的基座

三个陷阱,一个核心问题:回测系统的默认配置不满足专业要求

  • 复权方向用反了,你在做带 look-ahead bias 的研究。
  • 停牌日跳过了,你的信号执行时间戳是错的。
  • 退市股删掉了,你的股票池从一开始就是偏的。

这三条每一条都需要认真对待。对于 TickDB 的用户来说,基础的复权处理已经在数据管道中完成了,但在做日频回测时,你仍然需要在策略层面对停牌和退市做额外处理——因为数据源给你的只是"干净的价格序列",策略逻辑层的边界处理只能由你自己负责。

这不是"优化",这是"不犯错"。在量化这件事上,不犯错的团队,比做出最优策略的团队活得更久。


下一步行动

如果你是个人量化开发者

  1. 检查你当前回测系统使用的数据源是否为 CRSP 标准后复权
  2. 在 TickDB 中获取美股历史 K 线数据(支持 10 年级别清洗对齐)用于回测
  3. 用本文提供的诊断函数检查你现有数据是否存在上述三个问题

如果你希望确保回测数据零偏差

  • 访问 tickdb.ai 了解 TickDB 的历史数据质量体系
  • 联系 [email protected] 获取完整的 CRSP 级别数据校验方案

如果你习惯用 AI 辅助开发

  • 在 AI 助手中搜索安装 tickdb-market-data SKILL,一键接入标准化数据管道

回测局限性说明:本文涉及的所有回测结果基于历史数据模拟,不构成未来收益保证。上述模拟场景中的收益差异(8.2%)是在简化假设下计算的,实际偏差取决于具体的时间跨度、市场周期和数据源质量。建议在任何策略投入实盘前,使用更长时间维度的数据和更严格的样本外测试进行验证。市场有风险,投资需谨慎。