第 8 讲：【实战】手写 A 股 PIT 检索引擎

本讲核心目标：
融会贯通第二模块前三讲的全部数据工程精髓。将动态股票池过滤（次新/ST/停牌）、财报 Point-in-Time 对齐（Notice Date As-of Join）以及申万一级行业哑变量矩阵封装为一个高度解耦的工业级参考类 AsharePITEngine。输入任意历史交易日，一键返回符合买方时点口径的特征宽表；并编写时间穿梭渗透测试套件（Penetration Test），从物理上证明系统绝无未来函数泄露！

---

1. 为什么买方机构必须拥有独立的 PIT 检索引擎？

在没有系统化封装的业余量化团队中，每个研究员在计算因子时各写一套数据读取逻辑：

• 研究员 A 忘了剔除 ST 股票，导致垃圾股污染了价值因子；
• 研究员 B 误用了报告期而非公告披露日，引入了严重的未来函数；
• 研究员 C 的行业分类使用的是 2021 年新标准跑 2016 年历史，引入了宏观前瞻偏差。

这种各自为战的混乱直接导致策略回测表现与实盘表现严重脱节。买方机构通过统一的 PIT 检索引擎，在数据接入层将所有潜在漏洞物理隔绝：

课程图示

可放大查看细节，点击百分比恢复原始大小。

---

2. API 接口规范与时间物理契约

本检索引擎查询指定交易日收盘后的静态快照截面，使用当日已完整形成的日线量价数据；生成的交易决策最早在下一交易时段（开盘或盘中）执行。

snapshot = engine.get_market_snapshot("2023-04-14", symbols=["OLD.SH", "NEW.SZ"])

严格的工程物理契约：

1. 非交易日严格拒绝：输入非交易日直接抛出明确的 ValueError，严禁隐蔽地自动向前或向后回退日期（回退会破坏时序等距性）。
2. 上市日与开市年龄对齐：输入日历必须完整覆盖标的上市开市日；停牌股票仍计入上市年龄，不可成交则标记为 in_universe = False。
3. 财报公告时间戳阻断：所有财务报表以 available_at <= cutoff（通常为收盘 15:00）为准，盘后 20:00 披露的财报绝不渗入当日截面。
4. 行业显式全覆盖：所有在市标的必须被映射到 31 个行业之一，未知标的自动计入 ind_未知，行和严格保持为 1。

---

3. 动手实操：高可靠 PIT 检索引擎实现

下面的 Python 代码严格遵循机构级买方规范，将第 5、6、7 讲的核心组件无缝组合，并内置了完备的渗透测试。

在项目根目录运行 PYTHONPATH=tutorial python3 -m examples.lecture_08_pit_engine_practice。依赖安装、共享模块和数据契约见项目说明。

"""整合第 5–7 讲的收盘截面参考实现；信号最早在下一交易时段执行。"""
from dataclasses import dataclass
from datetime import date

import polars as pl

from examples.common import close_time, require_columns, require_unique
from examples.lecture_05_dynamic_universe_builder import DynamicUniverseBuilder, UniverseFilterConfig
from examples.lecture_06_pit_financial_engine import AsharePITFinancialEngine
from examples.lecture_07_industry_engine import AshareIndustryEngine


@dataclass(frozen=True)
class EngineConfig:
    min_listed_days: int = 60
    exclude_st: bool = True
    exclude_suspended: bool = True
    taxonomy: str = "SW2021"


class AsharePITEngine:
    def __init__(self, daily_bars, stock_basics, pit_financials, industry_mapping,
                 trade_calendar, status_history, config: EngineConfig | None = None):
        self.cfg = config or EngineConfig()
        self.bars = daily_bars
        self.basics = stock_basics
        self.fins = pit_financials
        self.industries = industry_mapping
        self.calendar = trade_calendar
        self.status = status_history
        self.universe = DynamicUniverseBuilder(UniverseFilterConfig(
            self.cfg.min_listed_days, self.cfg.exclude_st, self.cfg.exclude_suspended))
        self.financial_engine = AsharePITFinancialEngine()
        self.industry_engine = AshareIndustryEngine(self.cfg.taxonomy)

    def get_market_snapshot(self, trade_date: str, symbols: list[str] | None = None) -> pl.DataFrame:
        day = date.fromisoformat(trade_date)
        if day not in self.calendar["trade_date"].to_list():
            raise ValueError(f"{day} 不在输入交易日历中；不会自动回退到前一日")
        require_columns(self.bars, ["symbol", "trade_date", "close_hfq", "volume"])
        basics = self.basics.filter((pl.col("list_date") <= day) &
                                   (pl.col("delist_date").is_null() | (pl.col("delist_date") > day)))
        if symbols is not None:
            missing = set(symbols) - set(basics["symbol"])
            if missing:
                raise ValueError(f"查询日不存在有效上市档案: {sorted(missing)}")
            basics = basics.filter(pl.col("symbol").is_in(symbols))
        bars = self.bars.filter((pl.col("trade_date") <= day) &
                               pl.col("symbol").is_in(basics["symbol"].to_list()))
        require_unique(bars, ["symbol", "trade_date"])
        today = bars.filter(pl.col("trade_date") == day)
        if today.is_empty():
            raise ValueError(f"{day} 没有行情数据")
        if basics.join(today, on="symbol", how="anti").height:
            raise ValueError("上市股票缺少当日行情；停牌应显式提供零成交量记录")
        universe = self.universe.build_dynamic_universe(today, basics, self.calendar, self.status)
        features = (
            bars.sort(["symbol", "trade_date"]).with_columns(
                (pl.col("close_hfq") / pl.col("close_hfq").shift(20).over("symbol") - 1).alias("mom_20d"),
                (pl.col("close_hfq").pct_change().rolling_std(20).over("symbol") * 252**.5).alias("vol_20d"),
            ).filter(pl.col("trade_date") == day)
            .select("symbol", "mom_20d", "vol_20d")
        )
        financials = self.financial_engine.snapshot(self.fins, close_time(day))
        industries = self.industry_engine.build_industry_dummy_matrix(
            today.select("symbol"), self.industries, trade_date)
        result = (universe.join(features, on="symbol", validate="1:1")
                  .join(financials, on="symbol", how="left", validate="1:1")
                  .join(industries, on="symbol", how="left", validate="1:1")
                  .with_columns(pl.lit(close_time(day)).alias("decision_at")))
        self.financial_engine.assert_zero_future_leakage(result)
        return result.sort("symbol")

    @staticmethod
    def run_lookahead_penetration_tests(engine: "AsharePITEngine"):
        snapshot = engine.get_market_snapshot("2023-04-14")
        old = snapshot.filter(pl.col("symbol") == "OLD.SH")
        assert old["report_period"][0] == date(2022, 12, 31)
        assert old["in_universe"][0]
        assert not snapshot.filter(pl.col("symbol") == "HALT.SZ")["in_universe"][0]
        try:
            engine.get_market_snapshot("2023-04-15")
        except ValueError as error:
            assert "交易日历" in str(error)
        else:
            raise AssertionError("非交易日必须显式拒绝")
        print(snapshot.select("symbol", "trade_date", "in_universe", "report_period", "industry_name"))
        print("合成样例通过；完整边界验证请运行 python -m unittest discover -s tests -v")


if __name__ == "__main__":
    from examples.fixtures import sample_data
    data = sample_data()
    engine = AsharePITEngine(data["bars"], data["basics"], data["financials"],
                            data["industries"], data["calendar"], data["status"],
                            EngineConfig(taxonomy="demo-v1"))
    AsharePITEngine.run_lookahead_penetration_tests(engine)

核心运行输出与实盘对齐验证：

shape: (3, 5)
┌─────────┬────────────┬─────────────┬───────────────┬───────────────┐
│ symbol  ┆ trade_date ┆ in_universe ┆ report_period ┆ industry_name │
│ ---     ┆ ---        ┆ ---         ┆ ---           ┆ ---           │
│ str     ┆ date       ┆ bool        ┆ date          ┆ str           │
╞═════════╪════════════╪═════════════╪═══════════════╪═══════════════╡
│ HALT.SZ ┆ 2023-04-14 ┆ false       ┆ null          ┆ 房地产        │
│ NEW.SZ  ┆ 2023-04-14 ┆ false       ┆ null          ┆ 电子          │
│ OLD.SH  ┆ 2023-04-14 ┆ true        ┆ 2022-12-31    ┆ 食品饮料      │
└─────────┴────────────┴─────────────┴───────────────┴───────────────┘
合成样例通过；完整边界验证请运行 python -m unittest discover -s tests -v

1. OLD.SH：上市时间充足、无风险警示、正常交易，成功入池（in_universe = true），正确对齐 2022 年报（食品饮料行业）；
2. NEW.SZ：上市不足 60 个交易日，被次新股机制自动拦截（in_universe = false）；
3. HALT.SZ：全天成交量为 0（停牌），被流动性保护机制拦截（in_universe = false）。

---

4. 时间穿梭渗透测试（Time-Travel Penetration Test）

在买方持续集成（CI/CD）流水线中，如何证明一个回测系统 100% 不存在未来函数？答案就是时间穿梭渗透测试：

课程图示

可放大查看细节，点击百分比恢复原始大小。

运行全局测试套件：

PYTHONPATH=tutorial python3 -m unittest discover -s tutorial/tests -v

---

5. 本讲小结与第二模块知识图谱总结

至此，第二模块：数据工程与无泄漏清洗（PIT 基石） 全部四大核心讲座构建完毕！

• 我们彻底厘清了 A 股除权除息与前/后复权的本质差异（第 4 讲与第 5 讲）；
• 掌握了次新股、ST 警示与停牌股票的动态交易掩码漏斗（第 5 讲）；
• 攻克了报告期与公告日错位、同日双报与历史更正的 PIT 财报算法（第 6 讲）；
• 解决了申万 31 行业跨期改版与虚拟变量陷阱数学约束（第 7 讲）；
• 打造了高可靠的 AsharePITEngine 并通过了时间穿梭渗透测试（第 8 讲）。

从下一模块开始，我们将正式进入：第三模块：多因子挖掘与经典特征工程，基于这一坚不可摧的数据底座，系统构建时序量价因子、基本面质量因子、高频微观结构因子以及生产级因子正交化流水线！