复权与幸存者偏差：回测前必须做的两项数据修正

---

当你的策略在历史中"跑赢"了所有人

2018 年，一位量化研究员用过去 10 年美股数据回测自己的趋势跟踪策略，回测报告显示年化收益 23%，夏普比率 1.8，最大回撤仅 12%。他信心满满地上线实盘，半年后账户缩水 40%。

问题不在策略逻辑本身，而在他使用的数据：原始价格数据中，2008 年退市的雷曼兄弟、贝尔斯登、AIG 全部缺席。他的"10 年回测"实际上只包含了那批在 2008 年活下来的公司——一个被死亡数据"美颜"过的历史。

这不是孤例。几乎所有使用原始价格数据的回测都存在两类系统性偏差：价格复权偏差和幸存者偏差。它们不会让你的回测失败，但会让它变得毫无参考价值——一个被虚假高估的夏普比率，比一个亏损的策略更危险。

本文拆解这两类偏差的成因、量化它们的杀伤力，并给出生产级的数据修正方案。

---

一、价格复权：你的 K 线数据在说谎

1.1 一个被低估的复权灾难

假设你在 2020 年 1 月研究苹果公司（AAPL）过去 5 年的走势。你拿到以下数据：

日期	原始收盘价
2015-01-02	27.12
2020-01-02	75.10

心算一下：5 年涨幅 177%，年化约 22.5%。不错。

但如果你按这个逻辑买入并持有呢？2020 年初的真实感受是：苹果"已经涨了很久"，没什么便宜可捡。

问题出在哪里？ 苹果在 2015 年到 2020 年之间进行了 4 次拆股（1:7 → 1:4 → 1:4 → 1:4），你看到的 27.12 美元是拆股前的价格，而 75.10 美元是拆股后的价格。两个数字根本不在同一个度量衡下。

这不是苹果独有的问题。以下是美股历史上部分著名拆股事件：

公司	拆股日期	比例	备注
苹果	2020-08-31	1:4	5年4次拆股
特斯拉	2020-08-31	1:5	上市以来首次拆股
亚马逊	2022-06-06	1:20	上市以来首次拆股
谷歌	2022-07-18	1:20	上市以来首次拆股
伯克希尔·哈撒韦	—	从未拆股	B股曾拆分但A股从未

没有复权的价格数据，在包含拆股事件的历史区间上，会产生完全虚假的价格跳变：

# 原始数据中的"陷阱"（苹果示例）
2015-01-02: 27.12  ← 实际上市价格，历史上已发生4次拆股
2015-01-05: 26.87
...（中间省略数千条记录）...
2020-01-02: 75.10  ← 拆股后的新价格，与上方不可比

在这份"原始数据"中，2015 年买进的股票在 2020 年拆股后显示为 75 美元——但你真正持有的数量增加了 4×4×4 = 64 倍。如果用原始价格计算收益，会严重低估实际回报。更危险的反向场景是：如果你的回测买入点在拆股后、卖出点在拆股前，计算出的亏损可能完全是数据噪音。

1.2 复权的三种类型

复权不是简单地把价格"乘回去"。主流数据库（主要是 CRSP）提供了三种标准复权方式：

前复权（Adjusted Close）：以当前价格为基准，将历史上所有价格向前拉伸。

# 前复权原理示意（非精确算法）
adjusted_price_t = raw_price_t * (current_price / price_at_t)

前复权的优点是"看起来自然"——最新价格在 K 线最右侧，不需要额外标注。但致命缺陷是：任意历史时间点的调整因子不同，这意味着同一个时间序列中，不同时段的价格波动幅度不可比。在计算收益率标准差时，会引入人为的时间非平稳性。

后复权（Unadjusted / Raw）：以历史价格为基准，将后来的所有价格向后压缩。

# 后复权原理示意
adjusted_price_T = raw_price_T / (price_at_split_date)

后复权的优势在于同一时段内的价格是可加可比的，适合事件研究。但致命缺陷是历史价格数字巨大（想象一下巴菲特 A 股价格超过 60 万美元一股），容易溢出浮点精度，且不利于可视化。

等权复权（CRSP Total Return Index）：不仅调整价格，还累加分红再投资。严格来说，这是衡量"真实持有收益"的正确方式。

Total Return = (Price_Return) + (Dividend_Return)
其中 Dividend_Return = Σ(Dividend_i × Adjustment_Factor_i) / Price_before_dividend

结论：对于回测场景，推荐使用前复权或后复权价格（取决于你的策略类型），但必须确保整个回测区间使用同一套复权因子。如果你需要计算真实的策略收益（包含分红），使用 CRSP Total Return 指数。

1.3 分红的影响：比你想象的更隐蔽

拆股是显性的——价格会出现整数倍的跳变。但分红的影响更隐蔽。

假设某股票价格为 100 美元，宣布发放 2 美元分红。除息日当天，理论上股价应下跌 2 美元，变为 98 美元。如果你的回测使用原始价格：

• 买入时机恰好在除息日前一天，卖出在除息日后一天
• 股价从 100 跌到 98，你"亏损"了 2%
• 但你收到了 2 美元分红，实际收益为 0

如果你忽略分红再投资，你的系统会把这 2 美元的分红"误判"为价格下跌的策略亏损——或者更糟糕的，把一个简单持有策略的回报归因于错误的择时决策。

量化分红的影响规模：标普 500 成分股的平均年分红率约为 1.5%—2.0%。对于高分红行业（公用事业、消费品），年分红率可达 3%—5%。10 年回测中忽略分红，累计收益偏差可达 15%—40%。

1.4 复权数据获取：从哪里找可靠数据

数据源	复权质量	覆盖范围	成本	备注
Yahoo Finance（免费）	中等	美股为主	免费	调整因子偶有错误，对拆股边界日期处理存在争议
CRSP（学术授权）	最高	美股全量	高昂	行业标准，机构回测必备
Compustat	高	全球主要市场	高昂	含财务数据，适合多因子研究
TickDB（历史 K 线）	高	港股、数字货币、部分美股	适中	提供已复权的 10 年级别历史 K 线，含清洗和对齐

注：TickDB 的 /v1/market/kline 接口返回的 K 线数据为前复权价格，可直接用于跨周期回测，无需额外处理拆股和分红调整。但需要注意：trades 接口不支持美股逐笔成交，depth 频道在美股为 1 档深度。

---

二、幸存者偏差：死去的公司从未出现在你的"历史"里

2.1 幸存者偏差的数学本质

幸存者偏差（Survivorship Bias）的定义很简单：你用来回测的股票池，只包含了那些在回测结束时还活着的公司。那些退市的、破产的、被并购的公司，从你的"历史"中消失了。

这个"消失"意味着什么？

假设 2000 年初有 5000 只股票进入你的回测池。2020 年回测结束时，其中 800 只已经退市：

• 400 家被并购（通常有溢价，并购价 > 最后交易日价格）
• 300 家破产清算（通常损失 70%—100%）
• 100 家转板或私有化

如果你只使用"存活"到 2020 年的股票回测，你的池子自动排除了这些损失——你的夏普比率因此虚高。

量化幸存者偏差的杀伤力：

学术研究（Brown, Goetzmann & Ross, 1995; Malkiel, 1995）表明：

• 美股单只股票年均破产/退市率约 6%—8%
• 10 年期回测中，约 40%—50% 的初始股票会退市
• 忽略退市股票会使平均收益高估约 2%—4% 年化
• 对于小市值/低质量股票组合，高估幅度可达 5%—10% 年化

这不是微小的统计误差——2% 的年化收益差，在复利效应下，20 年后相当于 50% 的财富差异。

2.2 两种幸存者偏差：买入持有 vs. 滚动调仓

幸存者偏差的影响在不同策略类型中表现不同：

买入持有策略：只买不卖。初始池子中的死亡股票从未被止损，最终持有到回测结束时的价值为零（破产）或并购溢价（通常 > 最后收盘价）。忽略这些标的，你的组合真实表现应该是 -100% 或 +X%（并购溢价）。通常并购溢价会部分抵消破产损失，但净效果仍然是正偏——回测高估收益。

滚动调仓策略（更常见于量化策略）：每月或每季度重新选择标的。在这个场景中，幸存者偏差有两种叠加路径：

路径1（标的死亡）：持仓中出现退市股 → 部分资金损失 → 真实组合跑输回测
路径2（选择偏差）：回测池只包含存活股 → 每次调仓都在"更优质"的选择集中筛选
                  → 选股成功率虚高 → 换手率和交易成本估算失真

路径 2 尤其容易被忽视：你的策略"选出了"那些在历史上表现良好的股票，但那些股票之所以表现良好，部分原因正是它们活了下来——这是反向因果。

2.3 历史成分股数据库：修正幸存者偏差的唯一方法

要修正幸存者偏差，你需要在每个历史时点，知道当时真实存在的股票池是什么——不是 2024 年的标普 500，而是 2014 年的标普 500、2004 年的标普 500。

这需要两类数据：

1. 历史成分股列表（Historical Constituent Lists）
标普、FTSE、罗素等指数机构会公布历史成分股调整记录。例如：

• 2008-09-22：雷曼兄弟从标普 500 中移除
• 2008-09-16：雷曼兄弟申请破产保护

如果你在 2008-01-01 构建"标普 500 等权组合"，雷曼兄弟应该在你的初始池子里。2008-09-22 之后，它应该被移除。不是你"发现"它退市了，而是按规则被动剔除。

2. 退市股票价格数据（Delisted Returns）
仅仅知道某只股票被移除了还不够。你需要它退市后的价格走势数据（通常来自 CRSP 的 hexcd 和 dlret 字段）：

# CRSP 退市收益处理逻辑（伪代码）
if company.delisted:
    # dlret: delisting return from CRSP
    # 含义：从最后交易日至实际退市日的残余收益
    final_return = (1 + price_return_before_delisting) * (1 + dlret) - 1

CRSP 的退市收益字段经过精心处理：

• 被并购：通常为正（平均约 +30%）
• 破产/纯粹退市：通常为负（平均约 -30% 至 -55%）
• 无法追溯：CRSP 使用市值加权市场收益估算（vwretx）

没有退市收益数据，你的回测中那些"消失"的股票会以最后交易日价格"凝固"在组合中——这等价于假设它们以最后收盘价被收购，严重高估了组合收益。

2.4 幸存者偏差的实际修正幅度：一个数据对比

以下是不同数据集下，美股小市值等权组合 1980—2018 年的年化收益对比（来源：Torngren & Montgomery, 2004; Dimensional Fund Advisors 研究）：

数据集	年化收益	夏普比率	最大回撤
幸存者偏差数据集（含存活股，不含退市股）	14.2%	0.71	38%
无偏差数据集（含退市股及退市收益）	11.8%	0.52	52%
偏差幅度	+2.4%	+0.19	-14%

换句话说：使用有幸存者偏差的数据，你的夏普比率被高估了约 36%（0.71/0.52 - 1）。这不是理论推演——这是可复现的经验数据。

---

三、生产级数据获取：构建无偏回测数据集

3.1 数据修正的完整流水线

原始数据获取
    ↓
[Step 1] 复权因子应用 → 统一价格基准（解决价格不可比问题）
    ↓
[Step 2] 历史成分股过滤 → 每个时间点使用正确的股票池
    ↓
[Step 3] 退市收益填补 → 不遗漏任何退市损失
    ↓
[Step 4] 数据质量验证 → 异常值检测、时间对齐、字段完整性
    ↓
可用于回测的干净数据集

下面给出 Step 1 和 Step 2 的生产级实现。Step 3 依赖 CRSP 等商业数据库，代码中给出接口调用示例。

3.2 复权处理：生产级代码

"""
TickDB 历史 K 线数据获取与复权验证
用于构建无偏回测数据集的 Step 1：价格基准统一

功能：
1. 获取多只股票的历史 K 线（已复权）
2. 验证复权因子一致性
3. 导出标准化回测格式

⚠️ 生产环境高频场景建议使用 aiohttp/asyncio 批量获取
"""

import os
import time
import json
import requests
from datetime import datetime, timedelta
from typing import Optional, Dict, List


class TickDBKlineFetcher:
    """TickDB 历史 K 线获取器 - 生产级实现"""

    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
        if not self.api_key:
            raise ValueError("请设置环境变量 TICKDB_API_KEY")
        self.base_url = "https://api.tickdb.ai/v1/market/kline"
        self.session = requests.Session()
        self.session.headers.update({"X-API-Key": self.api_key})

        # 限频控制：TickDB 免费层限制
        self._request_interval = 0.2  # 秒，两次请求间最小间隔
        self._last_request_time = 0.0

    def _rate_limit(self):
        """自适应限频：识别 3001 错误并尊重 Retry-After"""
        elapsed = time.time() - self._last_request_time
        if elapsed < self._request_interval:
            time.sleep(self._request_interval - elapsed)
        self._last_request_time = time.time()

    def _handle_error(self, response: requests.Response, symbol: str) -> Dict:
        """TickDB 标准错误处理"""
        if response.status_code == 200:
            data = response.json()
            code = data.get("code", 0)
            if code == 0:
                return data.get("data", {})
            elif code in (1001, 1002):
                raise ValueError(f"API Key 无效 (code {code})，请检查环境变量 TICKDB_API_KEY")
            elif code == 2002:
                raise KeyError(f"交易品种 {symbol} 不存在 (code {code})")
            elif code == 3001:
                retry_after = int(response.headers.get("Retry-After", 5))
                print(f"⚠️ 限频触发 (code 3001)，等待 {retry_after} 秒")
                time.sleep(retry_after)
                return None
            else:
                raise RuntimeError(f"API 错误 {code}: {data.get('message')}")
        else:
            raise RuntimeError(f"HTTP 错误 {response.status_code}: {response.text}")

    def get_historical_klines(
        self,
        symbol: str,
        interval: str = "1d",
        start_time: Optional[int] = None,
        end_time: Optional[int] = None,
        limit: int = 1000,
    ) -> List[Dict]:
        """
        获取历史 K 线数据（前复权）

        参数：
            symbol: 交易品种，如 "AAPL.US"
            interval: K 线周期，"1m"/"5m"/"1h"/"1d"/"1w"
            start_time: 起始时间戳（毫秒）
            end_time: 结束时间戳（毫秒）
            limit: 单次请求最大条数（最大 1000）

        返回：
            K 线列表，每条包含 timestamp/open/high/low/close/vol
        """
        params = {
            "symbol": symbol,
            "interval": interval,
            "limit": limit,
        }
        if start_time:
            params["start_time"] = start_time
        if end_time:
            params["end_time"] = end_time

        self._rate_limit()
        response = self.session.get(
            self.base_url,
            params=params,
            timeout=(3.05, 10)  # 连接超时 3.05s，读取超时 10s
        )
        data = self._handle_error(response, symbol)
        return data.get("klines", []) if data else []

    def fetch_with_pagination(
        self,
        symbol: str,
        interval: str,
        start_date: datetime,
        end_date: datetime,
        limit: int = 1000,
    ) -> List[Dict]:
        """
        分页获取长周期历史数据（自动翻页）

        TickDB 每页最大 1000 条，超出范围需翻页
        翻页逻辑：从 end_date 向 start_date 迭代，每次取最新 1000 条
        """
        all_klines = []
        current_end = int(end_date.timestamp() * 1000)
        start_ts = int(start_date.timestamp() * 1000)

        max_iterations = 500  # 防止无限循环
        iteration = 0

        while current_end > start_ts and iteration < max_iterations:
            klines = self.get_historical_klines(
                symbol=symbol,
                interval=interval,
                start_time=start_ts,
                end_time=current_end,
                limit=limit,
            )

            if not klines:
                break

            all_klines.extend(klines)

            # 翻页：取本页最旧时间戳之前的数据
            oldest_ts = min(k["timestamp"] for k in klines)
            current_end = oldest_ts - 1  # 避免重复

            print(f"  [{symbol}] 已获取 {len(all_klines)} 条，"
                  f"继续获取 {datetime.fromtimestamp(oldest_ts/1000).date()} 之前的数据...")
            iteration += 1

        # 按时间升序排列
        all_klines.sort(key=lambda x: x["timestamp"])
        return all_klines


# === 使用示例：构建多股票回测数据集 ===

def build_backtest_dataset(symbols: List[str], start: datetime, end: datetime):
    """
    构建可用于回测的多股票历史数据集

    输出格式：{symbol: [kline_dict, ...]}，每个 symbol 的 K 线已按时间排序
    """
    fetcher = TickDBKlineFetcher()

    # ⚠️ 重要：TickDB 的 kline 接口返回的是已复权价格
    # 但你需要自行确认复权类型（前复权 vs 后复权）
    # 建议在回测文档中明确标注使用的复权方法

    dataset = {}
    for symbol in symbols:
        print(f"\n正在获取 {symbol} 的历史数据...")
        klines = fetcher.fetch_with_pagination(
            symbol=symbol,
            interval="1d",
            start_date=start,
            end_date=end,
        )
        if klines:
            # 验证：检查是否存在不连续的价格跳变（可能指示未处理的拆股事件）
            price_check = validate_continuity(klines)
            if price_check["has_gaps"]:
                print(f"  ⚠️ 警告：检测到价格不连续，跳变点：{price_check['gap_dates']}")
            dataset[symbol] = klines
            print(f"  ✅ 获取完成：{len(klines)} 条，"
                  f"范围 {datetime.fromtimestamp(klines[0]['timestamp']/1000).date()} "
                  f"至 {datetime.fromtimestamp(klines[-1]['timestamp']/1000).date()}")
        else:
            print(f"  ❌ 获取失败：{symbol} 无数据或请求失败")
        time.sleep(0.5)  # 避免对单个 IP 造成过大压力

    return dataset


def validate_continuity(klines: List[Dict], threshold: float = 0.3) -> Dict:
    """
    验证 K 线数据的价格连续性

    参数：
        klines: 按时间排序的 K 线列表
        threshold: 单日价格变动超过 30% 视为异常跳变

    返回：
        {"has_gaps": bool, "gap_dates": [(date, change_pct), ...]}
    """
    gaps = []
    for i in range(1, len(klines)):
        prev_close = klines[i - 1]["close"]
        curr_open = klines[i]["open"]
        if prev_close and curr_open and prev_close > 0:
            change = abs(curr_open - prev_close) / prev_close
            if change > threshold:
                date = datetime.fromtimestamp(klines[i]["timestamp"] / 1000).date()
                gaps.append((date, f"{change:.1%}"))

    return {"has_gaps": bool(gaps), "gap_dates": gaps}


# === 执行示例 ===
if __name__ == "__main__":
    # 设置 API Key
    os.environ["TICKDB_API_KEY"] = "your_api_key_here"

    # 示例：获取苹果、微软、谷歌 2015—2024 年的日 K 线
    symbols = ["AAPL.US", "MSFT.US", "GOOGL.US"]
    start_date = datetime(2015, 1, 1)
    end_date = datetime(2024, 12, 31)

    dataset = build_backtest_dataset(symbols, start_date, end_date)

    # 保存为 JSON，供回测引擎读取
    output_path = "backtest_dataset.json"
    with open(output_path, "w") as f:
        json.dump(dataset, f, indent=2)
    print(f"\n数据集已保存至 {output_path}")

⚠️ 复权数据使用警告：TickDB 的 /v1/market/kline 接口返回前复权价格。前复权的特性是：越接近当前时间点的数据，调整因子越大。 如果你需要计算历史上某个固定时间点（如 2015-01-01）的收益，请务必使用该时间点之后同一套调整因子计算，避免因调整因子切换导致收益率失真。

3.3 历史成分股过滤：模拟真实历史选股池

以下代码演示如何构建"滚动窗口下的历史成分股池"——这是修正幸存者偏差的核心步骤：

"""
历史成分股池构建器
用于构建无偏回测数据集的 Step 2：历史正确股票池

功能：
1. 在每个调仓时间节点，筛选出当时真实存在的股票
2. 结合退市日期，过滤掉已退市股票
3. 输出每个时间窗口的可用股票列表

⚠️ 注意：真实生产环境应使用 CRSP/Compustat 等机构数据库
以下代码使用 CSV 文件作为数据源输入格式示例
"""

from datetime import datetime
from typing import Dict, List, Tuple, Optional
import pandas as pd


class HistoricalConstituentPool:
    """
    历史成分股池管理器

    数据输入格式（CSV，字段说明）：
        symbol: 股票代码
        add_date: 加入日期（YYYY-MM-DD）
        remove_date: 移除日期（YYYY-MM-DD），空表示至今仍存活
        remove_reason: 移除原因（delisted/merged/acquired/spinoff/other）
    """

    def __init__(self, constituent_csv_path: str):
        self.df = pd.read_csv(constituent_csv_path, parse_dates=["add_date", "remove_date"])
        # 处理空值：NaT 表示股票至今仍存活
        self.df["remove_date"] = self.df["remove_date"].fillna(pd.Timestamp.max)
        print(f"✅ 成分股数据库加载完成：{len(self.df)} 条记录")

    def get_pool_at_date(self, date: datetime) -> List[str]:
        """
        获取指定日期的存活股票池

        逻辑：add_date <= date < remove_date
        即：在 date 当天，该股票仍然在成分股列表中
        """
        mask = (self.df["add_date"] <= date) & (self.df["remove_date"] > date)
        active = self.df.loc[mask, "symbol"].tolist()
        return active

    def get_pool_range(
        self,
        start_date: datetime,
        end_date: datetime,
        rebalance_frequency: str = "M",
    ) -> Dict[str, List[str]]:
        """
        按指定频率生成历史调仓日期对应的股票池

        参数：
            start_date: 回测开始日期
            end_date: 回测结束日期
            rebalance_frequency: 调仓频率，
                'M' 每月初，'Q' 每季度初，'Y' 每年初

        返回：
            {"2015-01-01": ["AAPL.US", "MSFT.US", ...], ...}
        """
        # 生成调仓日期序列
        date_range = pd.date_range(start=start_date, end=end_date, freq=rebalance_frequency)
        pools = {}

        for date in date_range:
            pool = self.get_pool_at_date(date)
            date_str = date.strftime("%Y-%m-%d")
            pools[date_str] = pool

            # 打印每期股票池大小，用于监控
            print(f"  [{date_str}] 股票池：{len(pool)} 只")

        return pools

    def get_delisted_in_range(
        self, start_date: datetime, end_date: datetime
    ) -> List[Dict]:
        """
        获取回测区间内退市股票列表

        用于 Step 3：退市收益填补
        返回退市股票及其退市原因，供 CRSP 退市收益查询使用
        """
        mask = (
            (self.df["remove_date"] > start_date)
            & (self.df["remove_date"] <= end_date)
            & (self.df["remove_date"] < pd.Timestamp.max)
        )
        delisted = self.df.loc[mask, ["symbol", "remove_date", "remove_reason"]].copy()
        delisted["remove_date"] = delisted["remove_date"].dt.strftime("%Y-%m-%d")
        return delisted.to_dict("records")


# === 使用示例 ===
def run_backtest_with_correct_pool():
    """演示：使用历史成分股池运行回测"""

    pool_manager = HistoricalConstituentPool("sp500_constituents.csv")

    # 生成 2015—2024 每年初的股票池
    pools = pool_manager.get_pool_range(
        start_date=datetime(2015, 1, 1),
        end_date=datetime(2024, 1, 1),
        rebalance_frequency="Y",  # 每年调仓一次
    )

    # 获取退市股票列表（供后续 Step 3 使用）
    delisted = pool_manager.get_delisted_in_range(
        start_date=datetime(2015, 1, 1),
        end_date=datetime(2024, 1, 1),
    )
    print(f"\n回测区间内退市股票：{len(delisted)} 只")
    for item in delisted[:5]:  # 打印前 5 个示例
        print(f"  {item['symbol']}: {item['remove_date']} ({item['remove_reason']})")

    return pools, delisted


# === CSV 数据格式示例（用于初始化） ===
"""
symbol,add_date,remove_date,remove_reason
AAPL.US,1980-12-12,,alive
MSFT.US,1986-03-13,,alive
GOOGL.US,2004-08-19,2024-04-01,spinoff
LEH.US,1994-05-04,2008-09-17,delisted
BEAS.US,1998-06-17,2007-05-31,merged
"""

# === 完整回测数据流水线整合 ===
def build_unbiased_backtest_data(
    symbols: List[str],
    start_date: datetime,
    end_date: datetime,
    tickdb_api_key: str,
) -> pd.DataFrame:
    """
    完整流水线：获取数据 → 复权验证 → 历史成分股过滤

    返回：
        干净的 DataFrame，columns = [date, symbol, open, high, low, close, volume]
        仅包含在每个交易日在真实成分股池中的标的
    """
    import os
    os.environ["TICKDB_API_KEY"] = tickdb_api_key

    # Step 1: 获取 TickDB 复权 K 线数据
    dataset = build_backtest_dataset(symbols, start_date, end_date)

    # Step 2: 加载历史成分股池（需要提前准备的 CSV）
    # 在真实场景中，应使用 CRSP 数据库或订阅的历史成分股服务
    pool_manager = HistoricalConstituentPool("constituents/constituents.csv")

    # Step 3: 合并，输出干净的 DataFrame
    rows = []
    for symbol, klines in dataset.items():
        for k in klines:
            date = datetime.fromtimestamp(k["timestamp"] / 1000)
            # 过滤：只保留在该日期存在于成分股池中的股票
            pool_on_date = pool_manager.get_pool_at_date(date)
            if symbol in pool_on_date:
                rows.append({
                    "date": date,
                    "symbol": symbol,
                    "open": k["open"],
                    "high": k["high"],
                    "low": k["low"],
                    "close": k["close"],
                    "volume": k["vol"],
                })

    df = pd.DataFrame(rows)
    print(f"\n✅ 干净数据集构建完成：{len(df)} 条记录，"
          f"{df['symbol'].nunique()} 只股票，"
          f"{df['date'].dt.date.nunique()} 个交易日")
    return df

3.4 退市收益填补：CRSP 数据接入示意

Step 3 需要 CRSP 商业数据库。以下给出数据接入的示意结构，帮助理解完整流水线：

"""
退市收益填补（CRSP 接口示意）

⚠️ 这是示意代码。真实 CRSP 数据需要机构订阅授权。
代码仅展示数据格式和处理逻辑。
"""

def apply_delisting_returns(
    portfolio_df: pd.DataFrame,
    crsp_delist_path: str,  # CRSP delisting returns CSV
    crsp_link_path: str,   # CRSP stock-event link table
) -> pd.DataFrame:
    """
    对回测组合中的退市股票应用退市收益

    CRSP 关键字段（crsp_delist.csv）：
        permno: CRSP 内部编号
        dlstdt: delisting date
        dlret: delisting return
        dlretx: delisting return excluding dividends

    处理逻辑：
    1. 找到组合中每只股票的退市日期
    2. 在退市日后，持股比例按 dlret 调整
    3. 若 dlret 为空（CRSP 无法追溯），使用 vwretx 估算
    """
    crsp_delist = pd.read_csv(crsp_delist_path)

    df = portfolio_df.copy()

    for _, row in crsp_delist.iterrows():
        permno = row["permno"]
        dlret = row["dlret"]
        dl_date = pd.to_datetime(row["dlstdt"])

        # 找到该股票在组合中的持仓
        mask = (df["permno"] == permno) & (df["date"] >= dl_date)

        if dlret is not None and not pd.isna(dlret):
            # 应用退市收益：最终清算收益 = 最后收盘价 × (1 + dlret)
            adjustment = 1 + dlret
        else:
            # CRSP 无法追溯时，使用市场收益估算（偏保守）
            adjustment = 1 + row.get("vwretx", 0)
            print(f"  ⚠️ 退市收益不可追溯 (permno {permno})，"
                  f"使用市场收益估算: {adjustment:.4f}")

        # 更新持仓市值
        df.loc[mask, "position_value"] *= adjustment

    return df

---

四、数据修正效果的实证对比

4.1 修正前后收益对比

为直观展示两项修正的效果，以下引用 Dimensional Fund Advisors 的公开研究数据（Maggian, 2021），模拟一个 5 年期小市值价值策略（年换手率 30%）在 2015—2020 的表现：

数据处理情况	年化收益	夏普比率	最大回撤	胜率（月度）
原始价格 + 存活股池（无修正）	14.8%	0.83	28%	62%
仅复权处理（保留存活股池）	13.2%	0.74	31%	59%
仅幸存者偏差修正（原始价格）	12.1%	0.65	39%	56%
两项修正均完成	10.7%	0.58	47%	54%

关键洞察：无修正回测将年化收益高估了 38%（14.8 vs 10.7），夏普比率高估了 43%（0.83 vs 0.58）。最大回撤被低估了 19 个百分点——实盘中你会经历的痛苦程度，远超回测的预警。

4.2 修正量级的行业差异

行业/风格	平均年分红率	5 年退市率	综合偏差（年化）
科技成长	0.5%	25%	约 1.0%—1.5%
标普 500 全市场	1.8%	35%	约 2.0%—3.0%
小市值罗素 2000	1.2%	45%	约 3.0%—5.0%
低质量/僵尸股	2.5%	60%	约 5.0%—8.0%

结论很清楚：你的策略越偏向小市值、低质量、长期持有，数据偏差越致命。

---

五、TickDB 在数据修正体系中的位置

理解完两项数据修正，你可能会问：TickDB 在这个体系中扮演什么角色？

TickDB 的定位是 Step 1 的可靠数据源。

数据修正环节	TickDB 能做什么	TickDB 不能做什么
复权处理	✅ /kline 接口提供已复权历史 K 线，10 年级别	❌ 不提供原始未复权价格
历史成分股池	❌ 不提供历史成分股调整记录	❌ 需要自备 CRSP/标普历史数据
退市收益填补	❌ 不提供退市股票价格和收益数据	❌ 需要 CRSP delist 文件
实时监控与信号	✅ depth/trades 频道支持港股和数字货币实时数据	❌ 美股 depth 仅 1 档，不支持 trades

推荐工作流：

TickDB /kline（历史 K 线，已复权）
    ↓
与 CRSP 历史成分股池合并
    ↓
与 CRSP 退市收益数据合并
    ↓
导入回测引擎（如 Backtrader、Zipline、VectorBT）

---

结语

一个策略的回测年化收益是 18% 还是 11%，不是"跑得准不准"的问题——这是策略本身的评判标准问题。用了错误的数据，你的"18%"本质上是在回测一个不存在的历史。

复权和幸存者偏差修正是回测准确性的地基。 没有地基，再精密的因子模型、再复杂的机器学习架构，结果都是沙滩上的高楼。

下次看到一份夏普比率超过 1.5 的回测报告，先问两个问题：用的价格是复权的吗？用的股票池是当时真实存在的吗？

---

下一步行动

如果你在构建回测系统：

1. 访问 tickdb.ai 注册，获取 API Key（免费层无需信用卡）
2. 使用本文提供的代码模板，通过 /kline 接口获取已复权历史数据
3. 自行准备或订阅 CRSP 历史成分股数据，完成 Step 2 和 Step 3

如果你需要完整的历史成分股和退市数据库：
联系 [email protected] 了解 TickDB 机构数据合作方案，或获取 CRSP 学术授权渠道指引。

如果你习惯用 AI 辅助开发：
在 AI 助手中搜索安装 tickdb-market-data SKILL，快速接入历史 K 线数据获取能力。

---

回测局限性说明：本文引用的回测数据基于公开研究文献，使用的模拟参数（年化收益、夏普比率等）反映的是典型偏差幅度区间而非精确预测。实际偏差幅度受策略类型、市场环境、数据来源质量等多因素影响。建议在任何实盘部署前，使用完整数据（CRSP + 复权 + 退市收益）进行独立验证。

风险提示：本文不构成任何投资建议。市场有风险，投资需谨慎。