2. 数据读取与清洗实战

逐笔成交数据是交易所撮合引擎在每一笔成交发生时推送的原始记录，通常以 CSV、Parquet 或压缩文本形式按日落地。它比 K 线更细、更"脏"：同一毫秒内可能有多笔成交，字段可能因交易所或数据商不同而缺失，时间戳可能是毫秒也可能是微秒，甚至存在重复推送。本节的任务是把一份原始逐笔成交文件，处理成可直接用于特征提取的干净 DataFrame。我们以常见的 CSV 格式为例，字段包括：trade_id（成交编号）、symbol（证券代码）、trade_time（成交时间）、price（成交价）、volume（成交量）、amount（成交额）、bs_flag（买卖方向）。

一、读取阶段：先看清数据长什么样

不要一上来就 read_csv 然后直接清洗。第一步是"侦察"：读入后立即检查行数、列名、dtype 和前几行。逐笔数据常见的坑是时间列被自动推断成字符串、成交量被读成 object、代码列前导零丢失。因此读取时应显式指定 dtype 与时间解析方式：

import pandas as pd

df = pd.read_csv(
    "trades_20240102.csv",
    dtype={"trade_id": "string", "symbol": "string",
           "price": "float64", "volume": "int64",
           "amount": "float64", "bs_flag": "string"},
    parse_dates=["trade_time"],
    encoding="utf-8"
)
print(df.shape, df.dtypes)
print(df.head(5))
print(df["trade_time"].min(), df["trade_time"].max())

如果文件很大（单日逐笔可达数百万行），可用 chunksize 分块读取，或改用 Parquet 列式存储。读取后先确认时间范围是否落在当日交易时段内（例如 A 股 09:30–11:30、13:00–15:00），这是判断时间戳是否错位的第一道关卡。

二、缺失值：区分"真缺失"与"业务无意义"

逐笔数据的缺失值不能一律删除或一律填充，要按字段语义处理。trade_id、trade_time、price、volume 是核心字段，缺失即该记录不可用，应直接剔除；amount 缺失可由 price × volume 重算补全；bs_flag 缺失则不能随意填，因为买卖方向直接影响主动买卖量特征，应标记为"未知"并单独统计占比，若占比过高（如超过 5%）需回查数据源。

core = ["trade_id", "trade_time", "price", "volume"]
df = df.dropna(subset=core)

df["amount"] = df["amount"].fillna(df["price"] * df["volume"])
df["bs_flag"] = df["bs_flag"].fillna("UNKNOWN")
print(df["bs_flag"].value_counts(normalize=True))

三、异常值：用业务规则而非纯统计阈值

逐笔数据的异常值主要有三类：价格越界（如价格为 0 或负、超出当日涨跌停价）、成交量异常（如单笔量远超流通盘）、成交额与价量不匹配。纯用 3σ 或 IQR 会误杀真实的大单，因此应优先用业务规则：价格必须落在当日涨跌停区间内；volume 必须为正整数；amount 与 price × volume 的相对误差应在容差内（如 1%）。

# 示例：涨跌停价由外部行情表提供，此处假设已合并
mask_price = (df["price"] > 0) & \
             (df["price"] >= df["limit_down"]) & \
             (df["price"] <= df["limit_up"])
mask_vol = df["volume"] > 0
mask_amt = ((df["amount"] - df["price"] * df["volume"]).abs()
            / (df["price"] * df["volume"]).clip(lower=1e-8)) < 0.01

df_clean = df[mask_price & mask_vol & mask_amt].copy()
print("剔除异常:", len(df) - len(df_clean))

被剔除的记录不要直接丢弃，建议写入 rejected.csv 并附上剔除原因，便于复盘数据源质量。

四、重复记录：按成交编号去重，而非按整行

逐笔数据重复有两种：完全重复（整行相同）和"同 trade_id 不同字段"（推送重发或修正）。正确做法是先按 trade_id 去重并保留最后一条（通常是最新修正），再检查是否仍有整行重复。注意：不同 trade_id 但时间、价格、量完全相同的记录可能是真实的连续成交，不能误删。

df_clean = (df_clean
    .sort_values("trade_time")
    .drop_duplicates(subset=["trade_id"], keep="last")
    .drop_duplicates()
    .reset_index(drop=True))

五、时间戳对齐：统一时区、精度与排序

时间戳对齐是逐笔清洗中最容易被忽视、却对后续特征影响最大的一步。需要统一三件事：时区（全部转为交易所本地时间或统一 UTC）、精度（毫秒/微秒统一到同一单位）、单调性（按时间升序排列，同一时间戳内按 trade_id 稳定排序）。若数据来自多个文件拼接，还要检查跨文件边界是否有重叠时段。

df_clean["trade_time"] = (pd.to_datetime(df_clean["trade_time"], utc=True)
                          .dt.tz_convert("Asia/Shanghai"))
df_clean = df_clean.sort_values(["trade_time", "trade_id"]) \
                   .reset_index(drop=True)
# 检查时间是否单调不减
assert df_clean["trade_time"].is_monotonic_increasing

下图概括了从原始文件到干净样本的完整处理链，以及每个环节的判断依据与输出物：

六、清洗后的验收：三个必须通过的检查

清洗完成不等于可以交付，建议在进入特征提取前跑一遍验收检查：其一，trade_time 单调不减且全部落在交易时段内；其二，核心字段无缺失、volume 全为正、price 全在涨跌停区间；其三，trade_id 唯一。三项任一不通过，说明清洗逻辑有漏洞，应回到对应环节排查，而不是带着脏数据继续往下走。

需要强调的是，清洗规则必须与后续特征口径一致。例如若特征要计算"主动买入量"，那么 bs_flag 为 UNKNOWN 的记录就不能简单计入买方或卖方，而应在特征层单独处理；若特征依赖精确到毫秒的成交间隔，那么时间戳精度不统一会直接导致间隔计算失真。清洗不是孤立的预处理，而是特征可信度的第一道保障。