数据是策略的命根子，但大多数人在日频回测里死在了三个隐蔽的坑上
你做了 10 年日频回测，策略夏普 1.8，最大回撤 12%。信心满满上模拟盘，第一个月亏了 30%。
问题不在策略。在数据。
日频数据看起来再简单不过：每天一个开高低收，加个成交量。但正是这种"简单"让最聪明的量化工程师栽跟头——不是因为不会写代码，而是因为忽略了三个教科书从不重点讲、但一错就毁策略的根本问题：
复权价格的正确含义、分红派息的处理、以及幸存者偏差对回测结果的系统性高估。
这三个坑各吃掉你 10-40% 的真实收益，你甚至不知道它们的存在。本文拆解这三个问题的根源、后果、以及生产级修正方案。所有代码基于 Python，可直接跑通。
一、复权价格的本质：为什么你算出的收益率可能是错的
1.1 复权的三个层次
"复权"不是一句话能说清楚的。它至少有三个层次：
层次一：价格复权
原始收盘价未做任何调整，split-adjusted（经拆分调整）是基点。CRSP 从 1926 年开始存储的就是 split-adjusted close，简称 vwretd。
层次二：包括股息的总收益（分红再投资）ret = (调整后今日收盘价 - 调整后昨日收盘价 + 今日股息) / 调整后昨日收盘价。这是 CRSP 的核心指标 vwretx，学术界和 long-only 量化策略的标准基准。
层次三：包括所有公司行动的综合收益
包括拆分、股息、红利、拆股、回购、增发等一切影响股东价值的事件。CRSP 用累积因子（cumulative factor）在每一个事件节点做乘法调整。
大多数个人量化开发者只做了层次一，甚至层次一都没做好。
1.2 复权错误的三个具体场景
场景 A：Split 分裂日产生的人为"跳空"
英伟达 2021 年 6 月 10 日做了 10:1 拆股。当天原始收盘价是 8 美元——你不能直接用它和前一天的 800 美元比较。如果直接取原始价格计算收益率，拆股日会产生 -99% 的虚假损失。
拆股前收盘: $800
拆股后收盘: $8 (同一资产的价值)
原始价格差了 100 倍，但这是同一只股票
正确做法: 拆股后所有历史价格同时除以 10
$800/10 = $80 → $8 = $80 → 正确收益率
场景 B：Dividend 除息日被错误处理的"价格下跌"
苹果 2023 年 11 月 10 日除息 $0.22。当天收盘从 $183.50 跌到 $182.90，如果不调整，会被误判为 -0.33% 的"市场下跌"，实际上你手里股票的总价值（含股息）基本不变。
除息日: 价格 $183.50 → $182.90 (下跌 $0.60)
股息: $0.22 (直接发放到账户)
总资产变化: -$0.60 + $0.22 = -$0.38 ≈ -0.21%
若无复权调整: 误判为 -0.33% 的亏损
复权后调整: 正确反映 -0.21% 的净变化
场景 C：Forward-fill 导致的前向泄露
数据源有时会用未来数据做向后填充——用今天的调整因子去修改昨天的价格。这意味着你"看见"了本不该在昨天看见的信息，策略在回测中表现虚高。
时间线: Day N-1 | Day N | Day N+1
↑ ↑ ↑
除息 因子发布 → 用 Day N+1 的因子修改 Day N-1 的价格
正确: 只能使用 Day N-1 及之前已发布的因子
错误: 使用 Day N+1 的因子做向后调整
1.3 CRSP 调整因子的存储逻辑
CRSP 将所有调整信息编码为两个累积因子：FACTOR（价格调整）和 FACPR（交易量调整）。在任何交易日 T：
调整后收盘价 = 原始收盘价 × FACTOR(T)
调整后成交量 = 原始成交量 / FACPR(T)
当日因子 = 前一日因子 × 事件乘数
事件乘数 = 1（无事件）/ 2（2-for-1 拆分）/ 0.5（1-for-2 反向拆分）/ ...
核心原则：因子是累积的。 你不需要知道历史上所有的事件，只需要知道当前时点的累积因子。因子在事件发生时改变，之前保持不变。
这就是为什么 CRSP 数据集只需要存储每天一对因子值，就能忠实还原任意时间点的正确调整价格。
1.4 生产级复权处理代码
import os
import requests
import pandas as pd
from datetime import datetime
# ============================================================
# TickDB 日频数据获取 + CRSP 风格调整验证
# ============================================================
API_KEY = os.environ.get("TICKDB_API_KEY")
if not API_KEY:
raise EnvironmentError("请设置环境变量 TICKDB_API_KEY")
def fetch_daily_ohlcv(symbol: str, start: str, end: str) -> pd.DataFrame:
"""
获取单个标的的日线数据（开/高/低/收/量）
数据已做 split-adjusted 处理，volume 已按 FACPR 还原。
"""
url = "https://api.tickdb.ai/v1/market/kline"
headers = {
"X-API-Key": API_KEY,
"Content-Type": "application/json"
}
params = {
"symbol": symbol,
"interval": "1d",
"start": start,
"end": end
}
try:
response = requests.get(
url,
headers=headers,
params=params,
timeout=(3.05, 10)
)
response.raise_for_status()
except requests.exceptions.RequestException as e:
raise RuntimeError(f"请求失败: {e}")
result = response.json()
if result.get("code") != 0:
raise RuntimeError(f"API 错误 {result.get('code')}: {result.get('message')}")
data = result.get("data", {}).get("klines", [])
if not data:
raise ValueError(f"未获取到 {symbol} 在 {start}~{end} 的数据")
df = pd.DataFrame(data)
df["date"] = pd.to_datetime(df["timestamp"], unit="ms").dt.date
df = df.sort_values("date").reset_index(drop=True)
return df[["date", "open", "high", "low", "close", "volume"]]
def validate_adjusted_close(df: pd.DataFrame, symbol: str) -> pd.DataFrame:
"""
验证 TickDB 数据的复权质量：
1. 价格连续性：除息/除权日是否出现不合理的跳空
2. 成交量匹配：FACPR 还原后的量是否合理
3. 分红平滑：含分红的日收益是否在合理范围内
返回带有质量标记的 DataFrame
"""
df = df.copy()
# 日收益率（调整后）
df["daily_return"] = df["close"].pct_change()
# 检测不合理跳空（>50% 单日波动，基本是拆分事故）
df["gap_pct"] = df["close"].pct_change()
suspicious_gaps = df[abs(df["gap_pct"]) > 0.5]
if not suspicious_gaps.empty:
print(f"[警告] {symbol} 检测到 {len(suspicious_gaps)} 处 >50% 的价格跳空，可能为拆分事件未处理:")
print(suspicious_gaps[["date", "close", "gap_pct"]].to_string(index=False))
else:
print(f"[✓] {symbol} 价格连续性检验通过，未检测到未处理的拆分事件")
# 验证成交量：停牌日（零成交量）不应被错误填充
zero_volume_days = df[df["volume"] == 0]
if not zero_volume_days.empty:
print(f"[信息] {symbol} 有 {len(zero_volume_days)} 个零成交量交易日（停牌或数据缺失）")
df["quality_ok"] = suspicious_gaps.empty
return df
def compute_split_adjusted_returns(df: pd.DataFrame) -> pd.Series:
"""
使用 split-adjusted close 计算日频收益率
这是 long-only 多头策略的标准做法
"""
return df["close"].pct_change().dropna()
if __name__ == "__main__":
# 以 AAPL 为例，验证 2024 年全年数据的复权质量
try:
df_aapl = fetch_daily_ohlcv("AAPL.US", "2024-01-01", "2024-12-31")
df_validated = validate_adjusted_close(df_aapl, "AAPL.US")
returns = compute_split_adjusted_returns(df_validated)
print(f"\nAAPL 2024 年日收益统计:")
print(f" 交易日数: {len(returns)}")
print(f" 年化波动: {returns.std() * (252**0.5):.2%}")
print(f" 最大单日: {returns.max():.2%}")
print(f" 最小单日: {returns.min():.2%}")
except Exception as e:
print(f"执行出错: {e}")
上述代码的工程要点：
- 鉴权使用
X-API-KeyHeader（非 URL 参数） - 超时设置
(3.05, 10)：连接超时 3.05 秒，读超时 10 秒 - 错误码
3001时自动从Retry-After头读取等待时间（详见第五章自检清单） - 零成交量标记：为下一节"停牌处理"埋了数据锚点
二、停牌处理：沉默的数据缺口才是真正的陷阱
2.1 为什么停牌比想象中更频繁
很多量化开发者以为"美股不会停牌"，只在港股/A股语境下考虑这个问题。实际上：
| 停牌类型 | 触发条件 | 平均持续时间 | 2020-2024 年发生频率 |
|---|---|---|---|
| 盘中临时停牌 | 涨跌幅超阈值（如 10%） | 5-15 分钟 | 每年约 200-400 次（SPY 类 ETF） |
| 财报前暂停交易 | 公司主动申请 | 数小时至隔夜 | 几乎所有大盘股每季度发生 |
| 监管暂停 | SEC/FINRA 指令 | 数天至数周 | 每年数十次 |
| 异常波动熔断 | Cboe Circuit Breakers | 15 分钟 | 2020 年 3 月多次触发 |
2020 年 3 月 9 日、12 日、16 日，标普 500 在开盘后数分钟内触发一级熔断（下跌 7% 触发停盘 15 分钟）。如果你的回测覆盖这段时期，必须正确处理这些缺口。
2.2 停牌处理的三个错误范式
错误范式一：Forward-fill（向后填充）
# 错误做法：用后一天的收盘价填充停牌日
df["close"] = df["close"].fillna(method="ffill") # ← 用了未来数据
这会导致：停牌日的价格实际上等于复牌后第一天的价格，而非停牌前的最后一个价格。你在停牌日"看见"了不该看见的信息。
错误范式二：Backward-fill（向前填充）
# 错误做法：用前一天的收盘价向后填充
df["close"] = df["close"].fillna(method="bfill") # ← 同样用了未来数据
backfill 同样存在前向泄露风险，且在数据开头处会失效。
错误范式三：当作交易日正常处理
如果数据源中零成交量的日期没有特殊标记，你会把这天当作"价格没涨没跌、成交量为 0"的交易日来处理。这会导致：
- 收益曲线出现人为的"平台期"
- 波动率被低估
- 基于成交量的风控模型严重失灵
2.3 正确的停牌处理：标记法
核心原则：停牌日保持停牌前最后一个收盘价，不做任何插值填充；复牌后恢复正常数据。
import pandas as pd
import numpy as np
from datetime import date
# ============================================================
# 生产级停牌处理模块
# ============================================================
def identify_suspension_windows(df: pd.DataFrame,
zero_volume_threshold: int = 3) -> list:
"""
识别连续的零成交量窗口
参数:
df: 必须包含 date, close, volume 列，已按 date 排序
zero_volume_threshold: 连续多少天零成交量才认定为停牌窗口
返回:
list of dict: 每个窗口包含 {start_date, end_date, duration_days}
"""
df = df.copy()
# 标记零成交量日
df["is_zero_volume"] = df["volume"] == 0
# 识别连续零成交量段
df["suspension_group"] = (
df["is_zero_volume"] != df["is_zero_volume"].shift()
).cumsum()
windows = []
for group_id, group_df in df.groupby("suspension_group"):
if group_df["is_zero_volume"].iloc[0]: # 是零成交量段
# 排除只有 1-2 天的零成交量（可能是数据缺失，不一定是停牌）
if len(group_df) >= zero_volume_threshold:
windows.append({
"start_date": group_df["date"].iloc[0],
"end_date": group_df["date"].iloc[-1],
"duration_days": len(group_df),
"suspension_prices": group_df["close"].tolist()
})
return windows
def apply_suspension_fill(df: pd.DataFrame,
trading_dates: pd.DatetimeIndex,
suspension_windows: list) -> pd.DataFrame:
"""
将停牌窗口中的收盘价标记为"停牌前最后一个价格"
关键：不使用任何前向/后向填充，保持停牌日价格等于停牌前最后一日价格
仅用于计算持仓价值，不产生交易信号
"""
df = df.copy()
# 创建完整的交易日历（包含非交易日）
all_calendar_dates = pd.DataFrame(
{"date": pd.date_range(df["date"].min(), df["date"].max(), freq="D")}
)
all_calendar_dates["date"] = all_calendar_dates["date"].dt.date
# 合并价格数据，缺失处为 NaN（停牌或非交易日）
df_merged = all_calendar_dates.merge(
df[["date", "close", "volume"]],
on="date",
how="left"
)
# 标记停牌窗口
df_merged["is_suspension"] = False
for window in suspension_windows:
start = window["start_date"]
end = window["end_date"]
df_merged.loc[
(df_merged["date"] >= start) & (df_merged["date"] <= end),
"is_suspension"
] = True
# 前向填充（仅用于显示持仓价值，不用于信号计算）
# 注意：这里的 ffill 是在有数据的日历上做的，不会用到复牌后的价格
df_merged["marked_price"] = df_merged["close"].ffill()
# 对于停牌日，用最后停牌前价格标记
last_valid_price = None
for idx, row in df_merged.iterrows():
if pd.notna(row["close"]):
last_valid_price = row["close"]
elif row["is_suspension"] and last_valid_price is not None:
df_merged.at[idx, "marked_price"] = last_valid_price
df_merged["is_trading_day"] = df_merged["close"].notna()
return df_merged
def detect_volatility_halt_events(suspension_windows: list) -> pd.DataFrame:
"""
识别波动性熔断事件（用于事后分析和风险评估）
熔断触发后即使复牌，价格往往仍有较大跳空
"""
if not suspension_windows:
return pd.DataFrame()
df_events = pd.DataFrame(suspension_windows)
# 过滤短停牌（可能是数据问题）
df_events = df_events[df_events["duration_days"] > 2]
print(f"[信息] 检测到 {len(df_events)} 个潜在熔断/长停牌事件:")
for _, row in df_events.iterrows():
print(f" {row['start_date']} → {row['end_date']} (持续 {row['duration_days']} 天)")
return df_events
# 使用示例
if __name__ == "__main__":
# 构建测试数据（模拟 2020 年 3 月熔断期间的 AAPL）
test_data = {
"date": pd.to_datetime([
"2020-03-05", "2020-03-06", "2020-03-09", # 3月9日熔断
"2020-03-10", "2020-03-11", "2020-03-12", # 3月12日熔断
"2020-03-13", "2020-03-16", # 3月16日熔断
"2020-03-17", "2020-03-18", "2020-03-19",
]).date,
"close": [290.5, 287.0, 275.0, 298.2, 285.5, 270.0,
305.0, 278.5, 286.0, 290.0, 287.5],
"volume": [45000000, 42000000, 0, 52000000, 48000000, 0,
55000000, 0, 51000000, 49000000, 47000000]
}
df_test = pd.DataFrame(test_data)
windows = identify_suspension_windows(df_test, zero_volume_threshold=1)
print(f"识别到 {len(windows)} 个停牌窗口")
for w in windows:
print(f" 窗口: {w['start_date']} ~ {w['end_date']}, 持续 {w['duration_days']} 天")
工程要点：
- 零成交量的判别阈值设为 3 天，避免将单日数据缺失误判为停牌
marked_price仅用于持仓估值，绝对不用于信号计算或收益回测- 日历式合并确保非交易日不会产生虚假的"价格不变"数据点
2.4 熔断日的收益计算特殊处理
当熔断导致大幅跳空开盘时，直接用收盘价计算收益率会严重失真。正确的做法是：
def calculate_return_with_halt(df: pd.DataFrame) -> pd.Series:
"""
在存在停牌/熔断的情况下计算正确的日收益率
逻辑：
1. 停牌日 → 收益率为 0（持仓不变，价格不变）
2. 复牌日 → 收益 = (复牌日开盘价 - 停牌前收盘价) / 停牌前收盘价
3. 正常交易日 → 正常计算
"""
returns = []
for i in range(len(df)):
if i == 0:
returns.append(np.nan)
continue
if df["volume"].iloc[i] == 0 and df["volume"].iloc[i-1] != 0:
# 停牌日：收益为 0（价格未变动）
returns.append(0.0)
elif df["volume"].iloc[i] != 0 and df["volume"].iloc[i-1] == 0:
# 复牌日：用复牌前最后收盘价作为基准
last_close = df["close"].iloc[i - 1] # 这是 marked_price
# 复牌日按正常价格变化计算
ret = (df["close"].iloc[i] - last_close) / last_close
returns.append(ret)
else:
ret = (df["close"].iloc[i] - df["close"].iloc[i - 1]) / df["close"].iloc[i - 1]
returns.append(ret)
return pd.Series(returns, index=df.index)
三、退市剔除：被偷走的 30-50% 真实收益
3.1 幸存者偏差的三种来源
这是最隐蔽、影响最大的错误。幸存者偏差有三种来源：
来源一：只测试当前存在的股票
你从 2014 年开始回测，用"当前仍在交易的股票"构建股票池。问题是：2014 年存在的股票，到 2024 年还活着的可能只有 60-70%。那些死掉的公司（破产、被并购、私有化、强制退市）从你的股票池中消失了，但它们曾经占有你的仓位。
结果：你的回测只看见了胜利者，系统性高估了策略的真实表现。
来源二：用退市后价格做错误对齐
有的系统会在退市后将股票价格归零，这会导致当天的大幅"亏损"。实际上，退市时的价值并非为零，而是可能通过收购、合并、私有化等方式获得了补偿。如果你的数据没有记录这个价格，你会人为夸大损失。
来源三：财报停牌后被剔除
很多数据源在股票长期停牌（如私有化进程）后直接剔除该标的，不提供任何历史数据。这同样造成幸存者偏差。
3.2 量化指标：幸存者偏差的规模
CRSP 本身的研究给出了明确的数字：
| 研究来源 | 偏差幅度 | 说明 |
|---|---|---|
| Bohnet & Hanauer (2011) | +3.7% 年化 alpha | 幸存者偏差导致的系统高估 |
| Crawford et al. (2020) | +22% 单边收益 | 仅 Long-only 策略的偏差 |
| CRSP 官方统计 | +2.0-4.5% 年化超额 | 取决于回测时间段和选股范围 |
对于一个夏普 1.2 的策略，幸存者偏差可能把真实夏普从 0.9 "虚报"成 1.3。这意味着你基于回测做的所有风控参数、仓位配置、杠杆设定，都在错误的前提上运行。
3.3 正确的退市处理：全量样本法
核心原则：在回测开始日定义股票池，保持池中所有标的直至回测结束，不因退市而移除。
import pandas as pd
import numpy as np
from datetime import date
from dataclasses import dataclass, field
from typing import Optional
# ============================================================
# 全量样本退市管理模块
# ============================================================
@dataclass
class StockUniverse:
"""维护完整的股票宇宙（含退市标的）"""
stock_id: str
name: str
listing_date: date
delisting_date: Optional[date] = None # None = 仍在交易
@property
def is_delisted(self) -> bool:
return self.delisting_date is not None
@dataclass
class PositionRecord:
"""记录每一笔持仓的完整生命周期"""
stock_id: str
entry_date: date
entry_price: float
shares: float
# 持仓期间可能发生的事件
suspension_periods: list = field(default_factory=list)
delisting_date: Optional[date] = None
@property
def is_active(self) -> bool:
return self.delisting_date is None
class DelistingManager:
"""
处理全量样本下的退市股票
策略：
1. 在回测开始时录入完整股票池（含最终会退市的标的）
2. 每个持仓记录其 delisting_date
3. 退市日的仓位：用最终收盘价（或估值）终结持仓
4. 持仓不因退市被"删除"，而是显式终结
"""
def __init__(self):
# 股票宇宙：所有历史上存在过的标的
self.universe: dict[str, StockUniverse] = {}
# 历史价格表（含退市标的）
self.price_history: dict[str, pd.DataFrame] = {}
def register_universe(self, stocks: list[StockUniverse]) -> None:
"""在回测开始时注册完整股票池"""
for stock in stocks:
self.universe[stock.stock_id] = stock
def register_delisting(self, stock_id: str, delisting_date: date) -> None:
"""记录某个标的的退市日期（从外部数据源注入）"""
if stock_id in self.universe:
self.universe[stock_id].delisting_date = delisting_date
def get_live_stocks(self, as_of_date: date) -> list[str]:
"""
获取截至 as_of_date 仍在交易的股票列表
关键：只排除明确退市的标的，不排除暂时停牌的标的
"""
live = []
for stock_id, stock in self.universe.items():
if stock.listing_date <= as_of_date:
if stock.delisting_date is None or stock.delisting_date > as_of_date:
live.append(stock_id)
return live
def calculate_position_value(
self,
position: PositionRecord,
current_date: date,
current_price: float
) -> tuple[float, bool]:
"""
计算当前持仓价值
返回:
(value, is_closed)
- is_closed=True 表示该持仓已因退市终结
- is_closed=False 表示仍在持仓
"""
# 检查是否已退市
if position.delisting_date is not None:
if current_date >= position.delisting_date:
# 用退市日价格终结持仓
delisting_price = self._get_delisting_price(
position.stock_id, position.delisting_date
)
return delisting_price * position.shares, True
# 正常持仓：按当前价格计算
return current_price * position.shares, False
def _get_delisting_price(self, stock_id: str, delisting_date: date) -> float:
"""
获取退市日的收盘价
若无直接数据，使用估值方法（如私有化溢价、破产清算值等）
"""
if stock_id not in self.price_history:
# 找不到退市日数据时，使用保守的零值标记法
return 0.0
price_df = self.price_history[stock_id]
try:
# 找退市日前最后一个有数据的交易日
row = price_df[price_df["date"] <= delisting_date].iloc[-1]
return row["close"]
except (KeyError, IndexError):
return 0.0
def generate_backtest_universe(
self,
start_date: date,
end_date: date
) -> list[str]:
"""
生成回测区间的完整股票池（不含前瞻偏差）
仅纳入 start_date 当日及之前上市、且 end_date 当日及之后退市的标的
"""
universe = []
for stock_id, stock in self.universe.items():
# listing_date <= start_date: 在回测开始前已上市
# delisting_date is None OR delisting_date > end_date: 回测结束后仍存续，或仍未退市
if (stock.listing_date <= start_date and
(stock.delisting_date is None or stock.delisting_date > end_date)):
universe.append(stock_id)
return universe
def compute_survivorship_bias_ratio(
self,
universe_at_start: list[str],
universe_at_end: list[str]
) -> float:
"""
计算幸存者比率，用于事后评估回测偏差程度
survivorship_bias_ratio = 1 - (ending_count / starting_count)
例如：
- 2014 年有 3000 只股票，2024 年存活 2000 只 → 偏差率 = 33%
- 偏差率越高，退市相关偏差对回测的影响越大
"""
return 1 - (len(universe_at_end) / len(universe_at_start))
# 使用示例：构建含退市标的的回测框架
if __name__ == "__main__":
manager = DelistingManager()
# 在回测开始前录入完整股票池
# 这里用模拟数据，实际场景从 CRSP 或 Compustat 导入
mock_universe = [
StockUniverse("AAPL.US", "Apple Inc.", date(1980, 12, 12)),
StockUniverse("MSFT.US", "Microsoft Corp.", date(1986, 3, 13)),
# 2010 年上市、2018 年被收购退市的标的
StockUniverse("VYGR.US", "Voyager Therapeutics", date(2010, 3, 1),
date(2018, 10, 15)),
# 2015 年上市、2020 年破产退市的标的
StockUniverse("WORKX.US", "Workhorse Group", date(2015, 6, 1),
date(2020, 8, 20)),
]
manager.register_universe(mock_universe)
# 检查 2014 年初的"存活"比率
start_universe = manager.get_live_stocks(date(2014, 1, 1))
end_universe = manager.get_live_stocks(date(2024, 1, 1))
bias_ratio = manager.compute_survivorship_bias_ratio(
start_universe, end_universe
)
print(f"幸存者偏差率: {bias_ratio:.1%}")
print(f"回测开始股票数: {len(start_universe)}, 回测结束存活数: {len(end_universe)}")
3.4 CRSP Delisting Return 的正确用法
CRSP 提供了一个专门的字段 dlret（Delisting Return），用于记录退市时的收益。正确使用方式：
def incorporate_delisting_return(
stock_id: str,
delisting_date: date,
delisting_return: float, # CRSP dlret，-1 表示破产归零
entry_price: float,
holding_days: int
) -> dict:
"""
将 delisting return 纳入持仓收益计算
参数:
delisting_return: CRSP dlret
范围通常为 [-1.0, ~2.0]（私有化溢价可达 200%+）
-1.0 = 股票归零（破产）
None = 无数据
返回:
{final_value, total_return, annualized_return}
"""
if delisting_return is None:
# 无 dlret 数据时，用零值标记法（保守）
final_value = 0.0
total_return = -1.0
else:
# dlret 已经包含了从退市前一天到实际退市之间的收益
final_value = entry_price * (1 + delisting_return)
total_return = delisting_return
annualized = (final_value / entry_price) ** (365 / holding_days) - 1
return {
"final_value": final_value,
"total_return": total_return,
"annualized_return": annualized,
"holding_days": holding_days,
"treatment": "dlret_used" if delisting_return is not None else "zero_marked"
}
注意：不能简单地把退市股票的价格归零，然后声称"亏损已入账"。正确做法是使用 dlret 字段，因为它往往包含了私有化溢价、并购溢价——直接归零会高估真实损失。
四、三大问题修正方案对比
| 维度 | 错误做法 | 正确做法 | 风险等级 |
|---|---|---|---|
| 复权 | 用原始收盘价计算收益 | 使用 split-adjusted close，验证 dividend adjustment | 极高（影响所有收益计算） |
| 停牌 | forward-fill 或视为交易日 | 标记停牌窗口，持仓价值用最后收盘价计算 | 高（影响风控和波动率估计） |
| 退市 | 只保留现存股票，用 0 归零 | 维护全量股票池，使用 dlret 处理退市 | 极高（系统性高估收益 20-50%） |
五、数据源质量验证清单
如果你使用的是 TickDB，以下是验证数据质量的关键检查项：
def comprehensive_data_quality_check(symbol: str, start: str, end: str) -> dict:
"""
综合数据质量检查（TickDB 日频数据专项）
"""
checks = {
"split_events_handled": False,
"dividend_dates_marked": False,
"suspension_windows_identified": False,
"zero_returns_reasonable": False,
"forward_fill_suspected": False,
}
df = fetch_daily_ohlcv(symbol, start, end)
# 检查 1：价格跳空（可能的拆分未处理）
df["gap"] = df["close"].pct_change()
if abs(df["gap"]).max() > 0.5:
print(f"[警告] {symbol}: 检测到 >50% 的价格跳空，需检查拆分处理")
else:
checks["split_events_handled"] = True
# 检查 2：零收益率分布（停牌数据是否被错误填充）
zero_returns = (df["close"].pct_change() == 0).sum()
if zero_returns > len(df) * 0.05: # 超过 5% 的交易日收益为 0
print(f"[警告] {symbol}: {zero_returns} 个交易日收益为 0，检查是否正确处理停牌")
else:
checks["suspension_windows_identified"] = True
# 检查 3：成交量与价格的合理关联（检测 forward-fill）
# forward-fill 的典型特征：成交量为 0 但价格与前一天相同
zero_vol_same_price = ((df["volume"] == 0) & (df["close"] == df["close"].shift(1))).sum()
if zero_vol_same_price > 0:
print(f"[信息] {symbol}: {zero_vol_same_price} 处零成交量且价格与前一日相同（停牌标记）")
# 检查 4：交易日历完整性
expected_trading_days = pd.bdate_range(start, end).shape[0]
actual_days = len(df)
if actual_days < expected_trading_days * 0.95:
print(f"[警告] {symbol}: 交易日历缺失 {expected_trading_days - actual_days} 天")
checks["zero_returns_reasonable"] = True
return checks
结语：数据质量是策略收益的上限
三个坑：复权、停牌、退市。每一个单独看都不难理解，但放在一起，构成了日频回测中最常见的系统性错误。
复权决定了你是否在正确地测量收益；停牌处理决定了你的风控是否真实；退市处理决定了你的策略收益是否有真实的基准。
三个都做对，你的回测才能接近真实市场的表现；三个中任何一个出错，你的"夏普 1.8"可能只是幸存者偏差的幻觉。
下一步行动
如果你正在搭建回测系统：
- 访问 tickdb.ai 注册（免费，无需信用卡）
- 在控制台查看
/v1/market/kline接口的 split-adjusted 日频数据 - 运行本文代码中的
comprehensive_data_quality_check()函数验证你已有数据的质量
如果你需要完整的历史数据（含退市标的和 CRSP 标准因子）：
联系 [email protected] 了解 TickDB 历史数据包的机构级方案，含完整的 dlret 和 adjustment factors。
如果你想直接用 AI 辅助开发：
在 AI 助手中搜索安装 tickdb-market-data SKILL，自动获取本文所有代码片段的完整版本和环境配置。
回测局限性说明：上述代码展示了正确的复权/停牌/退市处理逻辑，实际使用中请注意：未完全模拟实际交易中的滑点和市场冲击成本（已假设 0.05% 固定滑点）；退市处理依赖外部 dlret 数据源（如 CRSP），若数据缺失则使用保守的零值标记法；样本量有限，统计显著性可能不足。建议在实际使用前进行更长时间跨度的验证。