3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在 PTrade 量化交易平台中,通过 Level-2 行情接口 get_individual_transaction,开发者可以提取当日的逐笔成交行情(Transaction Data),其中包括价格、成交量、成交方向,以及深层高频分析所需的叫买方编号(buy_no)和叫卖方编号(sell_no)。
get_individual_transaction(stocks=None, data_count=50, start_pos=0, search_direction=1, is_dict=False)is_dict=True。字典(dict)格式相比默认的 DataFrame/Panel 具有更快的解析速度和更低的内存开销。当 is_dict=True 时,接口返回格式如下:
{
"股票代码": [
[time_stamp, hq_px, business_amount, trade_index, business_direction, buy_no, sell_no, trans_flag, trans_identify_am, channel_num],
...
],
"fields": ["business_time", "hq_px", "business_amount", "trade_index", "business_direction", "buy_no", "sell_no", "trans_flag", "trans_identify_am", "channel_num"]
}
0 代表卖,1 代表买。0 代表普通成交,1 代表撤单成交。1 表示向前(返回数据按时间从最新到最旧排列),2 表示向后。以下示例展示了如何在 PTrade 策略中提取 000001.SZ 的逐笔成交数据,并解析叫买叫卖方编号:
def initialize(context):
g.security = '000001.SZ'
set_universe(g.security)
g.flag = False
def before_trading_start(context, data):
g.flag = False
def handle_data(context, data):
if not g.flag:
# 使用 is_dict=True 高效获取最近50条逐笔成交数据
res = get_individual_transaction(stocks=[g.security], data_count=50, is_dict=True)
if res and g.security in res:
transactions = res[g.security]
fields = res.get('fields', [])
# 获取各关键字段的索引
buy_no_idx = fields.index('buy_no')
sell_no_idx = fields.index('sell_no')
px_idx = fields.index('hq_px')
amt_idx = fields.index('business_amount')
dir_idx = fields.index('business_direction')
log.info(f"成功获取 {g.security} 逐笔成交数据,共 {len(transactions)} 条")
# 遍历并解析最新几笔成交
for item in transactions[:5]:
buy_no = item[buy_no_idx]
sell_no = item[sell_no_idx]
price = item[px_idx]
amount = item[amt_idx]
direction = '买' if item[dir_idx] == 1 else '卖'
log.info(f"成交价: {price}, 数量: {amount}, 方向: {direction}, 叫买方单号: {buy_no}, 叫卖方单号: {sell_no}")
g.flag = True
business_amount 大于特定阈值的记录,结合 buy_no 或 sell_no 可以识别大机构委托单的拆单主买或主卖行为。get_individual_entrust 接口返回的 order_no,可以通过 buy_no / sell_no 进行挂单与成交的关联,计算大单从挂单到撮合成交的时延。