3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在 A 股高频交易与盘口微观结构分析中,撤单比例是识别主力资金“虚假申报(Spoofing)”与诱多/诱空意图的核心指标。通过 ptrade 提供的 Level-2 逐笔数据,我们可以利用委托编号(order_no)精准追踪每一笔委托的生命周期。
由于沪深两市的 Level-2 报文机制不同,我们在计算撤单时需要采用不同的处理逻辑:
get_individual_entrust)中,其 trans_kind 字段为 5(删除订单)。get_individual_transaction)中,其 trans_flag 字段为 1(撤单成交),此时的 buy_no 或 sell_no 对应原委托的 order_no。为了保证高频场景下的运行效率,建议在调用时开启 is_dict=True 参数,以获取极速的字典格式数据:
get_individual_entrust(stocks, data_count, is_dict=True)
business_time(时间)、hq_px(价格)、business_amount(委托量)、order_no(委托编号)、business_direction(方向:0-卖,1-买)、trans_kind(类型:4-增加,5-删除)。get_individual_transaction(stocks, data_count, is_dict=True)
hq_px(价格)、business_amount(成交/撤单量)、buy_no(买方编号)、sell_no(卖方编号)、trans_flag(1代表撤单成交)。以下是在 ptrade 交易环境中,兼容沪深两市、基于 is_dict=True 高效计算特定价格档位撤单比例的完整策略源码:
def initialize(context):
g.security = '600570.SS' # 恒生电子(上海市场示例)
# g.security = '000001.SZ' # 平安银行(深圳市场示例,可切换测试)
set_universe(g.security)
def handle_data(context, data):
# 获取目标价格(以当前最新价为例,实际可指定任意盘口档位价格)
snapshot = get_snapshot(g.security)
if not snapshot or g.security not in snapshot:
return
target_price = snapshot[g.security]['last_px']
# 计算该价格档位的撤单比例
ratio = calculate_cancel_ratio(g.security, target_price)
if ratio is not None:
log.info(f"代码: {g.security} | 目标价格: {target_price} | 撤单比例: {ratio:.2%}")
def calculate_cancel_ratio(security, target_price):
"""
计算指定代码在特定价格档位上的撤单比例
撤单比例 = 该价格撤单总股数 / 该价格历史总委托股数
"""
is_sh = security.endswith('.SS') or security.endswith('.XSHG')
# 1. 获取逐笔委托数据(获取最近200条)
entrust_data = get_individual_entrust([security], data_count=200, is_dict=True)
if not entrust_data or security not in entrust_data:
return None
entrust_list = entrust_data[security]
entrust_fields = entrust_data['fields']
# 字段索引映射
e_idx = {field: i for i, field in enumerate(entrust_fields)}
total_entrust_vol = 0.0 # 目标价格的总委托量
total_cancel_vol = 0.0 # 目标价格的总撤单量
# 记录目标价格的所有委托单号(用于深圳市场匹配)
target_order_nos = set()
# 遍历委托记录
for item in entrust_list:
px = item[e_idx['hq_px']]
vol = item[e_idx['business_amount']]
order_no = item[e_idx['order_no']]
kind = item[e_idx['trans_kind']]
# 仅统计目标价格
if abs(px - target_price) > 0.001:
continue
if is_sh:
# 上海市场逻辑
if kind == 4: # 增加订单
total_entrust_vol += vol
elif kind == 5: # 删除订单(撤单)
total_cancel_vol += vol
else:
# 深圳市场逻辑:先统计所有新增委托
total_entrust_vol += vol
target_order_nos.add(order_no)
# 2. 深圳市场需要结合逐笔成交数据匹配撤单
if not is_sh:
trans_data = get_individual_transaction([security], data_count=200, is_dict=True)
if trans_data and security in trans_data:
trans_list = trans_data[security]
trans_fields = trans_data['fields']
t_idx = {field: i for i, field in enumerate(trans_fields)}
for t_item in trans_list:
# trans_flag == 1 表示撤单
if t_item[t_idx['trans_flag']] == 1:
buy_no = t_item[t_idx['buy_no']]
sell_no = t_item[t_idx['sell_no']]
cancel_vol = t_item[t_idx['business_amount']]
# 如果撤单的原始委托单号在我们记录的目标价格单号中
if buy_no in target_order_nos or sell_no in target_order_nos:
total_cancel_vol += cancel_vol
# 3. 计算比例
if total_entrust_vol == 0:
return 0.0
cancel_ratio = total_cancel_vol / total_entrust_vol
return min(cancel_ratio, 1.0) # 限制最大为100%