3. 主动买卖方向推断的经典规则实现

在逐笔成交数据中，交易所通常只给出成交价、成交量和成交笔数，并不直接标注这笔成交是买方主动还是卖方主动。经典推断规则的核心思想非常朴素：把成交价与成交瞬间的买一价、卖一价做比较，判断这笔成交是“吃掉了卖盘”还是“砸向了买盘”。本节围绕这条规则，讲清楚如何编码、参数如何设置、边界条件如何处理，并在真实逐笔数据上完成一次初步标注。

一、规则的形式化定义

先约定参照系：以成交价 P、成交前最优买价 B1、最优卖价 A1 为输入，输出方向标签 D ∈ {+1, -1, 0}，其中 +1 表示主动买入（买方发起），-1 表示主动卖出（卖方发起），0 表示无法判定。经典规则可写成：

if P >= A1:        D = +1   # 成交价打到或越过卖一，买方主动
elif P <= B1:      D = -1   # 成交价打到或跌破买一，卖方主动
else:              D = 0    # 成交价落在买卖价之间，方向不明

这里有两个容易踩坑的细节。第一，比较对象必须是成交发生前的盘口快照，而不是成交后的盘口，否则价格已经被这笔成交推动，规则会系统性失真。第二，等号归属要明确：当 P 恰好等于 A1 时归为主动买入，恰好等于 B1 时归为主动卖出，这是行业里最常见的约定，但必须在代码里写死并保持全流程一致。

二、参数设置与可调项

规则本身只有三个输入，但工程实现中需要设置几类参数，它们直接影响标注覆盖率与准确率：

• 盘口快照对齐方式：逐笔成交与逐笔委托/快照的时间戳精度不同。常见做法是取“成交时间戳之前最近一条快照”，并设置最大容忍延迟 max_lag_ms（示例取 500ms），超过则标记为不可判定。
• 价格容差 eps：用于处理浮点比较和最小价位跳动。示例中按标的最小变动单位设置，如 A 股取 0.01 元，比较时用 P >= A1 - eps 与 P <= B1 + eps。
• 零股/异常量过滤：成交量小于阈值或价格明显偏离盘口的记录，先隔离再人工复核，不直接进入统计。
• 集合竞价时段开关：开盘/收盘集合竞价期间没有连续双边报价，规则不适用，应单独打标而非强行推断。

三、边界条件的处理策略

真实数据里，最麻烦的不是规则本身，而是规则覆盖不到的情形。下面这张图把成交价与盘口的关系、对应的判定结果以及需要特殊处理的边界情形放在同一张关系图里，便于对照实现：

图中三类主判定之外，底部列出的边界情形必须在代码里显式分支，而不是靠默认值兜底。例如涨跌停封板时，卖一价可能不存在（涨停），此时 P ≥ A1 无法成立，但成交价等于涨停价且成交方向多为买方主动，需要按“单边报价规则”单独处理并打上 limit_up 标记，供后续误差验证时区分统计。

四、在真实逐笔数据上的初步标注

假设我们拿到某只股票一个交易日的逐笔成交表（字段：trade_time, price, volume, buy_order_id, sell_order_id）和逐笔快照表（字段：snap_time, bid1, ask1）。标注流程分三步：

1. 时间对齐：对每条成交，用二分查找在快照序列中定位 snap_time ≤ trade_time 的最后一条，计算延迟 lag = trade_time - snap_time，超过 max_lag_ms 的置为不可判定。
2. 规则判定：按第一节的三分支逻辑打标，同时写入 rule_version 与 eps，保证可追溯。
3. 结果汇总：按方向统计笔数与成交量占比，并单独统计 D=0 与不可判定的比例，作为标注质量的第一个观察指标。

下面是一段可直接运行的 Python 示例（示例参数与结果均为教学假设，非实测）：

import bisect

def label_trades(trades, snaps, eps=0.01, max_lag_ms=500):
    snap_times = [s["snap_time"] for s in snaps]
    out = []
    for t in trades:
        i = bisect.bisect_right(snap_times, t["trade_time"]) - 1
        if i < 0 or t["trade_time"] - snaps[i]["snap_time"] > max_lag_ms:
            out.append({**t, "dir": 0, "flag": "no_snapshot"}); continue
        b1, a1, p = snaps[i]["bid1"], snaps[i]["ask1"], t["price"]
        if a1 and p >= a1 - eps:
            d, flag = 1, "active_buy"
        elif b1 and p <= b1 + eps:
            d, flag = -1, "active_sell"
        else:
            d, flag = 0, "between"
        out.append({**t, "dir": d, "flag": flag})
    return out

在一段示例数据上运行后，可能得到类似“主动买入 46%、主动卖出 41%、方向不明 9%、不可判定 4%”的分布。这个分布本身不是结论，而是下一步误差验证的起点：方向不明占比过高，往往说明快照对齐延迟偏大或 eps 设置不当；不可判定占比过高，则要检查快照频率是否足够。

把规则、参数、边界分支和标注结果的可追溯字段一次性设计到位，后续无论调整 eps、更换快照频率，还是引入更精细的推断模型，都能在同一套标注框架下对比效果，而不必推倒重来。