3. 高换手封板中主动买入大单的识别与统计

任务定位：把“大单”从感觉变成可统计的口径

本节要解决的是一个口径问题：高换手封板个股在封板前往往出现密集的主动买入大单，但“大单”若只凭盘口感觉判断，就无法跨样本比较。我们需要基于逐笔成交与委托明细，先定义可复算的大单阈值，再统计封板前主动买入大单的笔数、金额与占成交额比例，最后按换手率区间分组对比结构差异。

关键约定如下，全节保持一致：

• 方向判定：逐笔成交中，成交价等于卖方委托价（即买方主动吃卖盘）记为主动买入；成交价等于买方委托价记为主动卖出。若数据源直接给出买卖方向字段，以该字段为准并注明。
• 大单阈值：以单笔成交金额为口径，示例取 单笔成交金额 ≥ 100 万元 为大单。阈值必须固定后再统计，不能边看结果边调阈值。
• 统计窗口：以当日首次触及涨停价的时刻 T_seal 为基准，取 [T_seal − 30min, T_seal) 为封板前窗口，封板后成交不计入本节统计。
• 占比分母：同一窗口内全部成交金额（含主动买入、主动卖出、中性成交），保证“大单占成交额比例”可横向比较。

识别与统计的可执行流程

下面给出一段可直接运行的 Python 示例。它读取两份教学示例数据：逐笔成交表 trades.csv（字段：time, price, volume, amount, direction, order_id）与委托明细表 orders.csv（字段：order_id, order_time, order_price, order_volume, side）。示例数据为构造的教学样本，用于演示口径与计算过程，不代表任何真实个股结果。

import pandas as pd

BIG_ORDER_AMOUNT = 1_000_000      # 大单阈值：单笔成交金额 ≥ 100 万元
WINDOW_MINUTES = 30               # 封板前统计窗口

def load_data(trade_path, order_path):
    trades = pd.read_csv(trade_path, parse_dates=["time"])
    orders = pd.read_csv(order_path, parse_dates=["order_time"])
    return trades, orders

def seal_time(trades, limit_price):
    """当日首次触及涨停价的时刻"""
    hit = trades[trades["price"] >= limit_price]
    if hit.empty:
        raise ValueError("样本内未触及涨停价，请检查 limit_price")
    return hit["time"].min()

def pre_seal_window(trades, t_seal, minutes=WINDOW_MINUTES):
    start = t_seal - pd.Timedelta(minutes=minutes)
    return trades[(trades["time"] >= start) & (trades["time"] < t_seal)].copy()

def tag_big_active_buy(df, threshold=BIG_ORDER_AMOUNT):
    df["is_active_buy"] = df["direction"].eq("B")
    df["is_big"] = df["amount"] >= threshold
    df["is_big_active_buy"] = df["is_active_buy"] & df["is_big"]
    return df

def summarize(df):
    total_amount = df["amount"].sum()
    big_buy = df[df["is_big_active_buy"]]
    big_buy_amount = big_buy["amount"].sum()
    return {
        "窗口成交额": round(total_amount / 1e4, 2),          # 万元
        "主动买入大单笔数": int(len(big_buy)),
        "主动买入大单金额": round(big_buy_amount / 1e4, 2),  # 万元
        "大单占成交额比例": round(big_buy_amount / total_amount * 100, 2) if total_amount else 0.0,
        "大单单笔均额": round(big_buy["amount"].mean() / 1e4, 2) if len(big_buy) else 0.0,
    }

# 使用示例（limit_price 为该股当日涨停价）
trades, orders = load_data("trades.csv", "orders.csv")
t_seal = seal_time(trades, limit_price=12.10)
win = pre_seal_window(trades, t_seal)
win = tag_big_active_buy(win)
result = summarize(win)
print(result)

这段代码的三个要点值得注意：seal_time 用“首次触及涨停价”而非“收盘涨停”，避免把盘中反复开板的样本混入；pre_seal_window 用左闭右开区间，防止封板瞬间那笔成交被重复计入；tag_big_active_buy 把方向与大单两个条件分开打标，便于后续单独检查“大额但被动”的成交，避免把封单挂单误算成主动买入。

按换手率区间分组对比大单结构

单只个股的统计只能说明自身，要回答“高换手封板的大单结构是否不同”，需要把样本按换手率分组。示例按 换手率 < 5%、5%–10%、≥ 10% 三档分组，对每组计算大单笔数、金额与占比的中位数，减少极端样本干扰。

def group_by_turnover(samples):
    """
    samples: DataFrame，每行一只涨停股，含列
      turnover(换手率%), big_cnt, big_amt(万元), big_ratio(%), total_amt(万元)
    """
    bins = [0, 5, 10, 100]
    labels = ["低换手(<5%)", "中换手(5%-10%)", "高换手(≥10%)"]
    samples["换手区间"] = pd.cut(samples["turnover"], bins=bins,
                                 labels=labels, right=False)
    agg = samples.groupby("换手区间", observed=True).agg(
        样本数=("turnover", "size"),
        大单笔数中位数=("big_cnt", "median"),
        大单金额中位数=("big_amt", "median"),
        大单占比中位数=("big_ratio", "median"),
        窗口成交额中位数=("total_amt", "median"),
    ).round(2)
    return agg

print(group_by_turnover(samples))

判读时不要只看占比高低，要结合三项一起看：若高换手组“大单笔数多、单笔均额不增、占比反而下降”，说明成交被大量中小单摊薄，主动买入大单虽多但并未主导成交；若“笔数少、单笔均额大、占比高”，则更接近少数大资金集中扫单的结构。这两种形态在后续封板质量判断中的含义完全不同。

统计口径与分组逻辑一图看清

结果判读与常见口径陷阱

统计完成后，先做三项自检再下结论：

1. 方向字段一致性：抽样核对若干笔成交，确认 direction 与“成交价是否等于卖方委托价”的推导结果一致，避免数据源方向定义相反导致大单方向整体颠倒。
2. 阈值敏感性：用 50 万、100 万、200 万三档阈值各跑一次，若“高换手组占比更高”的结论只在某一档成立，说明结论对阈值敏感，应如实标注而非直接采信。
3. 分母口径：确认占比分母是窗口内全部成交额，而非仅主动买入成交额。两者混用会让不同样本的占比不可比。