3. 数据诊断环境搭建与工具链

逐笔特征漂移识别不是一次性的脚本任务，而是一条需要反复运行、对比、回溯的诊断链路。环境搭建的目标，是让“取数—算特征—看漂移—定位异常”这四个动作能够稳定复现，而不是每次换一台机器、换一个时间窗口就得到不同结论。本节围绕四个组成部分展开：Python 数据分析栈、逐笔数据存储与查询、特征计算流水线、可视化工具配置，并给出可直接落地的配置示例。

一、Python 数据分析栈：锁定版本，隔离环境

逐笔数据体量大、字段多，特征计算又涉及滚动窗口与分组聚合，因此对数值计算库的版本一致性要求较高。建议使用 conda 建立独立环境，并显式锁定核心依赖版本，避免 pandas 大版本升级带来的行为差异（例如分组聚合默认值、时间频率别名变更）导致同一份特征代码在不同环境算出不同结果。

conda create -n tickdrift python=3.10 -y
conda activate tickdrift
pip install "pandas==2.1.4" "numpy==1.26.4" "pyarrow==14.0.2" \
            "duckdb==0.9.2" "polars==0.20.5" \
            "matplotlib==3.8.2" "plotly==5.18.0" "pyarrow-hotfix"
pip freeze > requirements.lock

这里选择 pandas 作为特征计算主框架，polars 用于大文件预筛，duckdb 承担查询层，pyarrow 负责列式存储读写。把 pip freeze 的结果固化为 requirements.lock，是保证诊断结论可复现的第一道防线。若团队多人协作，还应把 Python 小版本一并写入文档，因为不同小版本在浮点运算与哈希实现上也可能存在细微差异。

二、逐笔数据存储与查询：列式分区 + SQL 直查

逐笔成交/委托数据通常按“交易日 + 标的”组织。推荐使用 Parquet 列式存储，并按 trade_date=YYYYMMDD 分区目录存放，这样查询某一天或某几天时可以直接做分区裁剪，避免全量扫描。字段上至少保留：证券代码、交易所时间戳（精确到毫秒或微秒）、逐笔序号、成交价、成交量、买卖方向标识。时间戳建议统一转换为带时区的标准时间，并在文档中明确“以交易所时间为准、不做本地时区换算”的约定，全文与图示保持一致。

import duckdb

con = duckdb.connect()
con.execute("SET threads=8;")

q = """
SELECT code, ts, seq, price, volume, side
FROM read_parquet('data/ticks/trade_date=2024011*/part-*.parquet')
WHERE code = '600000'
  AND ts BETWEEN TIMESTAMP '2024-01-15 09:30:00'
             AND TIMESTAMP '2024-01-15 11:30:00'
ORDER BY ts, seq
"""
df = con.execute(q).df()

用 DuckDB 直接对 Parquet 做 SQL 查询，好处是不需要先把数据全量载入内存，且能利用分区与列裁剪。查询结果再交给 pandas 做特征计算，形成“查询层—计算层”的清晰分工。需要注意：逐笔数据存在同一毫秒内多笔记录的情况，排序键必须是 (ts, seq) 组合，仅按时间戳排序会打乱真实成交顺序，进而影响后续滚动特征。

三、特征计算流水线：从逐笔到分钟级特征

漂移识别的核心是“同一特征在不同时间窗口的分布是否发生偏移”。因此流水线应把逐笔数据聚合为固定频率（如 1 分钟）的特征序列，并保留可追溯的中间结果。一个最小可用的流水线包含三步：清洗（去重、剔除异常价格与零量）、聚合（按分钟计算价量特征）、落盘（写入特征表并记录计算参数）。

import pandas as pd

def build_features(df: pd.DataFrame) -> pd.DataFrame:
    df = df.drop_duplicates(subset=["ts", "seq"])
    df = df[(df["price"] > 0) & (df["volume"] > 0)].copy()
    df["minute"] = df["ts"].dt.floor("1min")
    df["amt"] = df["price"] * df["volume"]
    df["signed_vol"] = df["volume"] * df["side"].map({"B": 1, "S": -1})

    g = df.groupby(["code", "minute"], as_index=False)
    feat = g.agg(
        vwap=("amt", lambda s: s.sum()),
        vol=("volume", "sum"),
        n_trades=("seq", "count"),
        net_vol=("signed_vol", "sum"),
    )
    feat["vwap"] = feat["vwap"] / feat["vol"]
    feat["avg_trade_size"] = feat["vol"] / feat["n_trades"]
    feat["order_imbalance"] = feat["net_vol"] / feat["vol"]
    return feat

这段示例中，order_imbalance 定义为净买入量除以总成交量，取值在 [-1, 1] 之间，正值表示买方主动成交占优。该定义必须在文档与图示中统一，后续所有漂移判断都基于同一口径。流水线落盘时建议同时写入 calc_version 与 param_json 两列，记录本次计算使用的代码版本与参数，便于日后复现某次诊断结论。

四、可视化工具配置：静态图与交互图分工

可视化承担两类任务：一是快速浏览特征分布随时间的变化，二是对可疑窗口做下钻定位。建议 matplotlib 负责生成可归档的静态图（用于日报与复盘留痕），plotly 负责交互式下钻（缩放时间轴、悬停查看具体分钟值）。中文字体需要显式配置，否则图中中文会显示为方块。

import matplotlib
matplotlib.rcParams["font.sans-serif"] = ["Noto Sans CJK SC", "SimHei"]
matplotlib.rcParams["axes.unicode_minus"] = False

import plotly.io as pio
pio.templates.default = "plotly_white"

下图展示了本节工具链的整体数据流：逐笔 Parquet 分区经 DuckDB 查询后进入 pandas 特征流水线，产出分钟级特征表，再分别供给静态图与交互图，最终服务于漂移判断。图中同时标出了各环节需要固化的版本与参数，这些正是保证诊断可复现的关键控制点。

五、环境自检：用一条命令验证链路通畅

环境搭好后，应做一次端到端自检：取一个交易日、一只标的、一个上午时段，跑通查询、特征计算与出图，并核对关键数值。例如检查某分钟 vwap 是否落在该分钟最高价与最低价之间、order_imbalance 是否在 [-1, 1] 内、n_trades 是否为正整数。这些约束能快速暴露排序错误、方向映射错误或除零问题。自检通过后，把查询 SQL、特征函数与绘图脚本一并纳入版本管理，后续每次诊断只需替换日期与标的参数，即可在一致的环境中复现结果。