2. 买卖价差统计与做市成本估算

任务目标

基于历史逐笔成交或盘口数据，统计买卖价差（Bid-Ask Spread）的分布特征，并估算做市商在给定交易量下的平均做市成本。本案例使用模拟的限价订单簿快照数据，计算相对价差、价差分布，并给出按交易量加权的平均成本估算。

环境与依赖

Python 3.8+，仅需标准库 csv、statistics、collections，无需第三方库。

数据说明

输入文件 orderbook_snapshots.csv，每行代表一个时间点的订单簿快照，字段如下：

timestamp,bid_price,bid_size,ask_price,ask_size
09:30:00.123,100.10,500,100.15,800
09:30:00.456,100.11,300,100.16,1200
...

其中 bid_price 为最高买价，ask_price 为最低卖价，bid_size/ask_size 为对应档位的挂单量（股）。

核心实现

以下代码读取快照数据，计算每个快照的价差与相对价差（以中价归一化），统计分布分位数，并按指定交易量估算做市成本。做市成本定义为：为满足给定交易量 Q，做市商在买盘吃单（以 ask 价买入）或卖盘吃单（以 bid 价卖出）时，相对中价产生的平均不利偏移。

import csv
from statistics import median, quantiles
from collections import defaultdict

def load_snapshots(path):
    """读取订单簿快照，返回列表[(bid_price, bid_size, ask_price, ask_size)]"""
    data = []
    with open(path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            bid_p = float(row['bid_price'])
            ask_p = float(row['ask_price'])
            bid_s = float(row['bid_size'])
            ask_s = float(row['ask_size'])
            if bid_p > 0 and ask_p > bid_p:  # 过滤无效快照
                data.append((bid_p, bid_s, ask_p, ask_s))
    return data

def spread_stats(snapshots):
    """计算价差与相对价差的分布统计"""
    abs_spreads = [ask - bid for bid, _, ask, _ in snapshots]
    rel_spreads = []
    for bid, _, ask, _ in snapshots:
        mid = (bid + ask) / 2.0
        rel_spreads.append((ask - bid) / mid)
    
    def pct(lst, p):
        lst_sorted = sorted(lst)
        idx = int(p * (len(lst_sorted) - 1))
        return lst_sorted[idx]
    
    return {
        'abs_mean': sum(abs_spreads) / len(abs_spreads),
        'abs_median': median(abs_spreads),
        'abs_p25': pct(abs_spreads, 0.25),
        'abs_p75': pct(abs_spreads, 0.75),
        'rel_mean': sum(rel_spreads) / len(rel_spreads),
        'rel_median': median(rel_spreads),
        'rel_p25': pct(rel_spreads, 0.25),
        'rel_p75': pct(rel_spreads, 0.75),
        'count': len(snapshots)
    }

def estimate_cost(snapshots, trade_qty):
    """
    估算做市成本：对每个快照，若 trade_qty 不超过 ask_size，则按 ask 价买入；
    若不超过 bid_size，则按 bid 价卖出。计算成交价相对中价的偏移。
    返回平均成本（以中价百分比表示）和样本数。
    """
    total_cost_bps = 0.0
    valid_samples = 0
    for bid, bid_sz, ask, ask_sz in snapshots:
        mid = (bid + ask) / 2.0
        if trade_qty <= ask_sz:
            # 买入方向：付出 ask 价
            cost = (ask - mid) / mid
            total_cost_bps += cost
            valid_samples += 1
        elif trade_qty <= bid_sz:
            # 卖出方向：获得 bid 价
            cost = (mid - bid) / mid
            total_cost_bps += cost
            valid_samples += 1
        # 若两侧深度均不足，则跳过该快照（无法在该价位完成交易量）
    
    if valid_samples == 0:
        return None, 0
    avg_cost = total_cost_bps / valid_samples
    return avg_cost * 10000, valid_samples  # 转为基点(bps)

def main():
    snapshots = load_snapshots('orderbook_snapshots.csv')
    if not snapshots:
        print("无有效快照数据")
        return
    
    stats = spread_stats(snapshots)
    print(f"快照数量: {stats['count']}")
    print(f"价差(元): 均值={stats['abs_mean']:.4f}, 中位={stats['abs_median']:.4f}, "
          f"P25={stats['abs_p25']:.4f}, P75={stats['abs_p75']:.4f}")
    print(f"相对价差(%): 均值={stats['rel_mean']*100:.3f}, 中位={stats['rel_median']*100:.3f}, "
          f"P25={stats['rel_p25']*100:.3f}, P75={stats['rel_p75']*100:.3f}")
    
    # 估算不同交易量下的平均成本
    for qty in [100, 500, 1000]:
        avg_cost_bps, n = estimate_cost(snapshots, qty)
        if n > 0:
            print(f"交易量={qty}: 平均成本={avg_cost_bps:.2f} bps (基于{n}个可成交快照)")
        else:
            print(f"交易量={qty}: 无可成交快照")

if __name__ == '__main__':
    main()

运行方式

将上述代码保存为 spread_analysis.py。
准备数据文件 orderbook_snapshots.csv 于同目录，格式如上。
执行 python spread_analysis.py。

验证方法与判定标准

程序输出应包含：快照总数、价差与相对价差的均值/中位数/四分位数，以及三个指定交易量下的平均成本（以bps为单位）。判定通过的准则：

程序能无错误读取CSV并输出统计结果；
价差均值应等于所有快照 (ask - bid) 的算术平均，可手工抽样核对；
相对价差均值应等于所有快照 (ask - bid) / mid 的算术平均；
对于交易量估算，若某快照的 ask_size 和 bid_size 均小于交易量，则该快照被跳过，输出中会注明有效样本数；
平均成本数值应介于0与最大相对价差之间（即不超过该交易量下可成交快照的最大半价差），否则说明计算有误。

预期输出示例

使用包含1000条有效快照的模拟数据（价差在0.02～0.08元之间随机分布，深度在200～2000股之间），预期输出形如：

快照数量: 1000
价差(元): 均值=0.0501, 中位=0.0500, P25=0.0250, P75=0.0750
相对价差(%): 均值=0.0500, 中位=0.0499, P25=0.0250, P75=0.0750
交易量=100: 平均成本=12.50 bps (基于1000个可成交快照)
交易量=500: 平均成本=12.48 bps (基于998个可成交快照)
交易量=1000: 平均成本=12.45 bps (基于850个可成交快照)

注意：实际数值取决于输入数据，上述仅为格式示例。当交易量增大时，部分深度不足的快照会被跳过，有效样本数相应减少。

常见问题

CSV编码问题：若文件含BOM头，可改用 encoding='utf-8-sig' 读取。
深度不足处理：本实现仅当单侧深度足以完全成交时才计入成本样本。若需部分成交场景，可扩展为按可成交比例加权计算。
价差分布可视化：如需观察分布形态，可自行用 matplotlib 绘制直方图，但本案例核心统计量已由代码输出。

📚 课程目录