盘口下单算法（难度：中级）
一、盘口下单算法详解
在主动型量化交易策略体系中，盘口下单算法是一种独特而有效的订单执行方法。相较于以成交量为驱动的成交量加权平均价格（POV）策略，盘口下单算法对市场的挂单结构更为敏感。在市场深度良好但成交清淡的情况下，POV 策略可能会因为缺乏成交量而难以有效执行；而盘口下单算法则可以直接利用丰富的挂单信息，更快速地推进订单的执行。 尤其是在临近收盘或高频交易活跃时段，盘口下单算法可以通过快速而频繁的挂单和撤单操作，更好地把握执行节奏。
盘口下单算法的基本原理是：根据预先设定的参考盘口（可以是对手盘的挂单，也可以是本方的挂单），并结合用户设定的量比比例，计算出当前应提交的订单手数。 同时，订单的报价方式也由用户指定，可以选择“对价”（即以对手盘的价格进行申报）或“挂价”（即以本方更优的价格进行申报）。例如，配置为“参考对手盘口、20%比例、对价申报”时，若卖一档挂有 500 手卖单，算法会自动计算出本次应买入 100 手，并以当前卖一的价格提交买入限价单。 这种机制能够根据市场挂单量的变化灵活调整每次下单的规模，避免因下单量过小而影响执行效率，或因下单量过大而对市场价格造成不必要的冲击。
在实际执行过程中，这种动态调整机制使得策略能够紧密跟随市场变化，从而在提高执行效率的同时，也增强了交易的隐蔽性。 尽管盘口下单算法在流动性充足的市场中表现出色，但也存在一定的局限性。例如，当市场出现大量的虚假挂单（即大额挂单并非真实的成交意愿）或者挂单频繁被撤销的情况下，算法的执行效果可能会受到影响。 因此，在实际应用盘口下单算法时，建议结合时间限制、滑点容忍度以及合理的价格容错机制进行综合控制，以应对潜在的市场风险。
二、天勤介绍
天勤平台概述
天勤（TqSdk）是一个由信易科技开发的开源量化交易系统，为期货、期权等衍生品交易提供专业的量化交易解决方案。平台具有以下特 点：
- 丰富的行情数据： 提供所有可交易合约的全部Tick和K线数据，基于内存数据库实现零延迟访问。
- 一站式的解决方案： 从历史数据分析到实盘交易的完整工具链，打通开发、回测、模拟到实盘的全流程。
- 专业的技术支持： 近百个技术指标源码，深度集成pandas和numpy，采用单线程异步模型保证性能。
策略开发流程
- 环境准备
- 安装Python环境（推荐Python 3.6或以上版本）
- 安装tqsdk包：pip install tqsdk
- 注册天勤账户获取访问密钥
- 数据准备
- 订阅近月与远月合约行情
- 获取历史K线或者Tick数据（用于分析与行情推进）
- 策略编写
- 设计信号生成逻辑（基于价差、均值和标准差）
- 编写交易执行模块（开仓、平仓逻辑）
- 实现风险控制措施（止损、资金管理）
- 回测验证
- 设置回测时间区间和初始资金
- 运行策略获取回测结果
- 分析绩效指标（胜率、收益率、夏普率等）
- 策略优化
- 调整参数（标准差倍数、窗口大小等）
- 添加过滤条件（成交量、波动率等）
- 完善风险控制机制
三、跟量下单算法代码实现
本节将介绍一个基于天勤量化交易平台（TqSdk）并使用 Python 语言实现的盘口下单算法策略。该策略通过实时分析市场盘口数据，并根据用户预设的量比比例和报单方式，动态地控制每轮的委托手数和价格，从而在保证交易执行效率的同时，降低对市场价格的扰动。
从策略设计的角度来看，本策略的核心思想是“以市场挂单量为基准确定下单量”，这与以市场成交量为基准控制交易速度的 POV 策略有着本质的区别，使其更适合应用于那些成交频率不高但挂单量相对活跃的市场环境。 在策略运行过程中，系统会持续跟踪市场盘口的变化，并根据价格的波动和当前订单的状态自动进行撤单和重新挂单的操作。
该盘口下单算法的实现主要可以分为以下关键步骤：
1. 设定策略初始参数
配置策略运行所需的关键参数，包括交易合约代码、交易方向（买入/卖出）、开平仓类型、目标交易总手数、参考盘口的类型、跟盘口量的比例，以及报单方式（如“对价”或“挂价”）等。
2. 初始化环境与数据
连接 TqSdk 接口，订阅行情数据，初始化记录已成交手数等关键变量，为后续的交易循环做准备。
3. 执行基于盘口监控的交易循环
在策略主循环中，实时获取并分析市场盘口数据。根据设定的参考盘口和比例计算本轮应下单手数，并根据报单类型确定委托价格，提交限价订单。同时监控订单状态，并在价格发生不利变动或订单未成交时，进行撤单和重新挂单操作。
4. 统计成交与结束策略
每轮交易完成后，更新累计成交手数。当累计成交手数达到或超过预设的目标总手数时，结束主循环，策略执行完毕，并关闭与交易平台的连接。
天勤实现算法代码
#!/usr/bin/env python
# coding=utf-8
__author__ = "Chaos"
from tqsdk import TqApi, TqAuth, TqKq
import math
# === 用户参数 ===
SYMBOL = "SHFE.ag2506" # 交易合约
DIRECTION = "BUY" # "BUY"为买入，"SELL"为卖出
OFFSET = "OPEN" # "OPEN"为开仓，"CLOSE"为平仓,"CLOSETODAY"为平今仓
TOTAL_VOLUME = 20 # 目标总手数
ORDERBOOK_RATIO = 0.2 # 盘口量比比例（如0.2表示20%）
ORDER_TYPE = "对价" # "对价"为对价报单，"挂价"为挂价报单
ORDERBOOK_TYPE = "对手盘口" # "对手盘口"或"挂单盘口"
# === 初始化API ===
acc = TqKq()
api = TqApi(account=acc, auth=TqAuth("快期账号", "快期密码"))
quote = api.get_quote(SYMBOL)
# === 初始化变量 ===
traded_volume = 0 # 已成交手数
last_printed_volume = 0 # 上次打印的成交手数
current_order = None # 当前订单
print(f"盘口下单算法启动，合约: {SYMBOL}，目标: {TOTAL_VOLUME}手，方向: {DIRECTION}，量比比例: {ORDERBOOK_RATIO*100}%")
def get_orderbook_volume():
"""获取盘口数量"""
if ORDERBOOK_TYPE == "对手盘口":
# 对手盘口：买入时看卖一量，卖出时看买一量
return quote.ask_volume1 if DIRECTION == "BUY" else quote.bid_volume1
else:
# 挂单盘口：买入时看买一量，卖出时看卖一量
return quote.bid_volume1 if DIRECTION == "BUY" else quote.ask_volume1
def get_order_price():
"""获取下单价格"""
if ORDER_TYPE == "对价":
# 对价报单
return quote.ask_price1 if DIRECTION == "BUY" else quote.bid_price1
else:
# 挂价报单
return quote.bid_price1 if DIRECTION == "BUY" else quote.ask_price1
try:
while traded_volume < TOTAL_VOLUME:
api.wait_update()
# 获取当前盘口数量
orderbook_volume = get_orderbook_volume()
# 计算本轮应下单手数
order_volume = int(math.floor(orderbook_volume * ORDERBOOK_RATIO))
# 不能超过剩余目标
order_volume = min(order_volume, TOTAL_VOLUME - traded_volume)
if order_volume > 0:
print(f"\n当前盘口数量: {orderbook_volume}手")
print(f"计算下单手数: {orderbook_volume} * {ORDERBOOK_RATIO} = {order_volume}手")
# 下新单
price = get_order_price()
current_order = api.insert_order(
symbol=SYMBOL,
direction=DIRECTION,
offset=OFFSET,
volume=order_volume,
limit_price=price
)
print(f"下单: {order_volume}手，价格: {price}，报单类型: {ORDER_TYPE}")
# 记录上一次的状态和剩余量
last_status = current_order.status
last_volume_left = current_order.volume_left
# 等待订单状态更新
while current_order.status == "ALIVE":
api.wait_update()
# 只在状态或剩余量发生变化时打印
if current_order.status != last_status or current_order.volume_left != last_volume_left:
print(f"订单状态更新: {current_order.status}, 剩余量: {current_order.volume_left}")
last_status = current_order.status
last_volume_left = current_order.volume_left
# 检查价格是否变化
new_price = get_order_price()
if new_price != price:
print(f"价格发生变化: {price} -> {new_price}")
# 如果还有未成交部分，先撤单
if current_order.volume_left > 0:
print(f"撤单: {current_order.volume_left}手")
api.cancel_order(current_order.order_id)
# 等待撤单完成
while current_order.status == "ALIVE":
api.wait_update()
# 重新下单
current_order = api.insert_order(
symbol=SYMBOL,
direction=DIRECTION,
offset=OFFSET,
volume=current_order.volume_left,
limit_price=new_price
)
print(f"重新下单: {current_order.volume_left}手，价格: {new_price}")
price = new_price
last_status = current_order.status
last_volume_left = current_order.volume_left
else:
# 如果订单已经完成，跳出循环
break
# 检查订单是否出错
if current_order.is_error:
print(f"下单失败: {current_order.last_msg}")
break
# 计算实际成交量
actual_trade = current_order.volume_orign - current_order.volume_left
if actual_trade > 0:
print(f"本轮成交: {actual_trade}手")
traded_volume += actual_trade
# 只在成交手数变化时打印进度
if traded_volume > last_printed_volume:
print(f"当前进度: {traded_volume} / {TOTAL_VOLUME}")
last_printed_volume = traded_volume
else:
print(f"订单未成交: {current_order.last_msg}")
print("盘口下单算法执行完毕")
except Exception as e:
print(f"算法执行异常: {e}")
finally:
api.close()