【期货量化实战】期货程序化交易订单执行优化（Python量化）

一、前言

大单执行需要考虑市场冲击成本，直接市价单可能造成较大滑点。订单执行优化通过拆分、限价、算法执行等方式，降低执行成本，提升交易效率。本文将介绍订单执行优化的实战方法。

本文将介绍：

• 大单拆分策略
• TWAP/VWAP执行算法
• 限价单优化
• 执行质量评估

二、为什么选择天勤量化（TqSdk）

TqSdk订单执行支持：

安装方法：

pip install tqsdk

三、大单拆分策略

3.1 基础大单拆分

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：基础大单拆分
说明：本代码仅供学习参考
"""

from tqsdk import TqApi, TqAuth, TqSim
from tqsdk.lib import TargetPosTask

api = TqApi(TqSim(init_balance=500000), auth=TqAuth("快期账户", "快期密码"))

SYMBOL = "SHFE.rb2510"
TARGET_VOLUME = 50 # 目标50手
MIN_VOLUME = 2 # 每笔最小2手
MAX_VOLUME = 10 # 每笔最大10手

position = api.get_position(SYMBOL)

# 启用大单拆分模式
target_pos = TargetPosTask(api, SYMBOL, min_volume=MIN_VOLUME, max_volume=MAX_VOLUME)

print("=" * 50)
print("大单拆分执行")
print("=" * 50)

print(f"目标持仓: {TARGET_VOLUME}手")
print(f"拆分范围: {MIN_VOLUME}–{MAX_VOLUME}手/笔")

# 设置目标持仓，会自动拆分
target_pos.set_target_volume(TARGET_VOLUME)

trade_count = 0
while True:
api.wait_update()

# 监控成交
if api.is_changing(position, "pos_long"):
trade_count += 1
print(f"\\n第{trade_count}笔成交: 当前持仓={position.pos_long}手")

# 完成检查
if position.pos_long == TARGET_VOLUME:
print(f"\\n执行完成！共成交{TARGET_VOLUME}手")
break

api.close()

3.2 智能拆分策略

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：智能拆分策略
说明：本代码仅供学习参考
"""

from tqsdk import TqApi, TqAuth, TqSim
from tqsdk.lib import TargetPosTask
import random

api = TqApi(TqSim(init_balance=500000), auth=TqAuth("快期账户", "快期密码"))

SYMBOL = "SHFE.rb2510"
quote = api.get_quote(SYMBOL)

api.wait_update()

class SmartOrderSplitter:
"""智能订单拆分器"""

def __init__(self, api, symbol, base_volume=5):
self.api = api
self.symbol = symbol
self.quote = api.get_quote(symbol)
self.base_volume = base_volume

def calculate_split_size(self, total_volume, current_depth):
"""
根据订单薄深度计算拆分大小

参数:
total_volume: 总需要成交量
current_depth: 当前订单薄深度
"""
# 基础拆分大小
base_size = self.base_volume

# 根据深度调整
if current_depth < 10:
# 深度不足，减小单笔
split_size = max(1, base_size // 2)
elif current_depth < 50:
split_size = base_size
else:
# 深度充足，可以稍大
split_size = min(base_size * 2, total_volume)

return split_size

def get_orderbook_depth(self):
"""获取订单薄深度"""
bid_depth = sum([
self.quote.bid_volume1, self.quote.bid_volume2, self.quote.bid_volume3,
self.quote.bid_volume4, self.quote.bid_volume5
])

ask_depth = sum([
self.quote.ask_volume1, self.quote.ask_volume2, self.quote.ask_volume3,
self.quote.ask_volume4, self.quote.ask_volume5
])

return min(bid_depth, ask_depth)

splitter = SmartOrderSplitter(api, SYMBOL, base_volume=5)

print("=" * 50)
print("智能拆分策略")
print("=" * 50)

total_volume = 30
remaining = total_volume

while remaining > 0:
api.wait_update()

depth = splitter.get_orderbook_depth()
split_size = splitter.calculate_split_size(remaining, depth)
split_size = min(split_size, remaining)

print(f"\\n剩余: {remaining}手, 订单薄深度: {depth}手")
print(f"本次拆分: {split_size}手")

# 这里简化，实际应该使用TargetPosTask
remaining -= split_size

if remaining == 0:
print("拆分完成")

api.close()

四、TWAP/VWAP执行算法

4.1 TWAP执行

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：TWAP执行算法
说明：本代码仅供学习参考
"""

from tqsdk import TqApi, TqAuth, TqSim
from tqsdk.lib import TargetPosScheduler
from tqsdk.algorithm import twap_table
from pandas import DataFrame

api = TqApi(TqSim(init_balance=500000), auth=TqAuth("快期账户", "快期密码"))

SYMBOL = "SHFE.rb2510"
TARGET_VOLUME = 100 # 目标100手
DURATION = 600 # 执行时间600秒（10分钟）
MIN_INTERVAL = 60 # 最小间隔60秒
MAX_INTERVAL = 120 # 最大间隔120秒

print("=" * 50)
print("TWAP执行算法")
print("=" * 50)

# 生成TWAP时间表
time_table = twap_table(api, SYMBOL, TARGET_VOLUME, DURATION, MIN_INTERVAL, MAX_INTERVAL)

print(f"目标持仓: {TARGET_VOLUME}手")
print(f"执行时间: {DURATION}秒")
print(f"时间表行数: {len(time_table)}")

print("\\n时间表示例（前5行）:")
print(time_table.head().to_string(index=False))

# 创建执行器
scheduler = TargetPosScheduler(api, SYMBOL, time_table)

# 执行
while not scheduler.is_finished():
api.wait_update()

print("\\n执行完成！")
print(f"成交记录数: {len(scheduler.trades_df)}")

# 计算成交均价
if len(scheduler.trades_df) > 0:
avg_price = (scheduler.trades_df['price'] * scheduler.trades_df['volume']).sum() / \\
scheduler.trades_df['volume'].sum()
print(f"成交均价: {avg_price:.2f}")

api.close()

4.2 VWAP执行

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：VWAP执行算法
说明：本代码仅供学习参考
"""

from tqsdk import TqApi, TqAuth, TqSim
from tqsdk.lib import TargetPosScheduler
from tqsdk.algorithm import vwap_table
from pandas import DataFrame

api = TqApi(TqSim(init_balance=500000), auth=TqAuth("快期账户", "快期密码"))

SYMBOL = "SHFE.rb2510"
TARGET_VOLUME = 100 # 目标100手
DURATION = 600 # 执行时间600秒

print("=" * 50)
print("VWAP执行算法")
print("=" * 50)

# 生成VWAP时间表（需要历史成交量数据）
# 注意：vwap_table需要历史数据，这里简化示例
klines = api.get_kline_serial(SYMBOL, 60, 100) # 获取分钟K线
api.wait_update()

# 生成VWAP时间表
time_table = vwap_table(api, SYMBOL, TARGET_VOLUME, DURATION)

print(f"目标持仓: {TARGET_VOLUME}手")
print(f"执行时间: {DURATION}秒")
print(f"时间表行数: {len(time_table)}")

# 创建执行器
scheduler = TargetPosScheduler(api, SYMBOL, time_table)

# 执行
while not scheduler.is_finished():
api.wait_update()

print("\\n执行完成！")

api.close()

4.3 自定义执行算法

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：自定义执行算法
说明：本代码仅供学习参考
"""

from tqsdk import TqApi, TqAuth, TqSim
from tqsdk.lib import TargetPosScheduler
from pandas import DataFrame
import time

api = TqApi(TqSim(init_balance=500000), auth=TqAuth("快期账户", "快期密码"))

SYMBOL = "SHFE.rb2510"
TARGET_VOLUME = 50
DURATION = 300 # 5分钟

quote = api.get_quote(SYMBOL)
api.wait_update()

def create_adaptive_time_table(total_volume, duration, quote):
"""
创建自适应时间表（根据订单薄深度调整）

参数:
total_volume: 总成交量
duration: 执行时间（秒）
quote: 行情对象
"""
# 评估订单薄深度
bid_depth = sum([
quote.bid_volume1, quote.bid_volume2, quote.bid_volume3,
quote.bid_volume4, quote.bid_volume5
])

ask_depth = sum([
quote.ask_volume1, quote.ask_volume2, quote.ask_volume3,
quote.ask_volume4, quote.ask_volume5
])

avg_depth = (bid_depth + ask_depth) / 2

# 根据深度决定执行策略
if avg_depth < 20:
# 深度不足，慢速执行
intervals = [30] * (duration // 30)
volumes = [total_volume // len(intervals)] * len(intervals)
prices = ["PASSIVE"] * len(intervals) # 排队价
else:
# 深度充足，快速执行
intervals = [10] * (duration // 10)
volumes = [total_volume // len(intervals)] * len(intervals)
prices = ["ACTIVE"] * len(intervals) # 对价

time_table = DataFrame({
'interval': intervals,
'target_pos': volumes,
'price': prices
})

return time_table

print("=" * 50)
print("自适应执行算法")
print("=" * 50)

time_table = create_adaptive_time_table(TARGET_VOLUME, DURATION, quote)

print(f"目标持仓: {TARGET_VOLUME}手")
print(f"执行时间: {DURATION}秒")
print(f"时间表行数: {len(time_table)}")
print(f"执行模式: {time_table['price'].iloc[0]}")

scheduler = TargetPosScheduler(api, SYMBOL, time_table)

while not scheduler.is_finished():
api.wait_update()

print("\\n执行完成！")

api.close()

五、限价单优化

5.1 动态限价策略

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：动态限价策略
说明：本代码仅供学习参考
"""

from tqsdk import TqApi, TqAuth, TqSim
from tqsdk.lib import TargetPosTask

api = TqApi(TqSim(init_balance=200000), auth=TqAuth("快期账户", "快期密码"))

SYMBOL = "SHFE.rb2510"
quote = api.get_quote(SYMBOL)

api.wait_update()

def dynamic_limit_price(direction):
"""
动态限价函数

参数:
direction: "BUY" 或 "SELL"
"""
tick = quote.price_tick

if direction == "BUY":
# 买入：在买一价基础上加1-3档
base_price = quote.bid_price1 if quote.bid_price1 > 0 else quote.last_price
# 根据订单薄深度调整
ask_depth = quote.ask_volume1 + quote.ask_volume2

if ask_depth < 10:
# 深度不足，加价更多
limit_price = base_price + tick * 3
elif ask_depth < 50:
limit_price = base_price + tick * 2
else:
limit_price = base_price + tick * 1
else:
# 卖出：在卖一价基础上减1-3档
base_price = quote.ask_price1 if quote.ask_price1 > 0 else quote.last_price
bid_depth = quote.bid_volume1 + quote.bid_volume2

if bid_depth < 10:
limit_price = base_price – tick * 3
elif bid_depth < 50:
limit_price = base_price – tick * 2
else:
limit_price = base_price – tick * 1

# 价格无效时使用最新价
if limit_price != limit_price: # 判断nan
limit_price = quote.last_price

return limit_price

print("=" * 50)
print("动态限价策略")
print("=" * 50)

target_pos = TargetPosTask(api, SYMBOL, price=dynamic_limit_price)

# 测试
target_pos.set_target_volume(1)

count = 0
while count < 10:
api.wait_update()

if api.is_changing(quote, "last_price"):
buy_price = dynamic_limit_price("BUY")
sell_price = dynamic_limit_price("SELL")

print(f"\\n最新价: {quote.last_price:.0f}")
print(f"买入限价: {buy_price:.0f}")
print(f"卖出限价: {sell_price:.0f}")

count += 1

api.close()

5.2 冰山订单

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：冰山订单（隐藏大单）
说明：本代码仅供学习参考
"""

from tqsdk import TqApi, TqAuth, TqSim
from tqsdk.lib import TargetPosTask
import time

api = TqApi(TqSim(init_balance=500000), auth=TqAuth("快期账户", "快期密码"))

SYMBOL = "SHFE.rb2510"
TOTAL_VOLUME = 100 # 总目标100手
VISIBLE_VOLUME = 5 # 每次显示5手

quote = api.get_quote(SYMBOL)
position = api.get_position(SYMBOL)

api.wait_update()

class IcebergOrder:
"""冰山订单管理器"""

def __init__(self, api, symbol, total_volume, visible_volume):
self.api = api
self.symbol = symbol
self.total_volume = total_volume
self.visible_volume = visible_volume
self.target_pos = TargetPosTask(api, symbol)
self.executed = 0
self.last_execute_time = 0

def execute_step(self):
"""执行一步"""
current_time = time.time()

# 控制执行频率（每10秒执行一次）
if current_time – self.last_execute_time < 10:
return

remaining = self.total_volume – self.executed

if remaining > 0:
# 本次执行量
step_volume = min(self.visible_volume, remaining)
target = self.executed + step_volume

self.target_pos.set_target_volume(target)
self.executed = target
self.last_execute_time = current_time

return step_volume

return 0

print("=" * 50)
print("冰山订单执行")
print("=" * 50)

iceberg = IcebergOrder(api, SYMBOL, TOTAL_VOLUME, VISIBLE_VOLUME)

print(f"总目标: {TOTAL_VOLUME}手")
print(f"每次显示: {VISIBLE_VOLUME}手")

while iceberg.executed < TOTAL_VOLUME:
api.wait_update()

step_vol = iceberg.execute_step()
if step_vol > 0:
print(f"\\n执行 {step_vol}手, 累计 {iceberg.executed}/{TOTAL_VOLUME}手")

print("\\n执行完成！")

api.close()

六、执行质量评估

6.1 执行成本分析

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：执行成本分析
说明：本代码仅供学习参考
"""

import pandas as pd
import numpy as np

def analyze_execution_cost(trades_df, benchmark_price):
"""
分析执行成本

参数:
trades_df: 成交记录DataFrame，需包含price和volume列
benchmark_price: 基准价格（如VWAP或开盘价）
"""
if len(trades_df) == 0:
return None

# 计算加权平均成交价
total_value = (trades_df['price'] * trades_df['volume']).sum()
total_volume = trades_df['volume'].sum()
vwap = total_value / total_volume if total_volume > 0 else 0

# 执行成本（相对于基准价）
execution_cost = vwap – benchmark_price

# 执行成本比例
cost_ratio = execution_cost / benchmark_price if benchmark_price > 0 else 0

# 价格改善（如果使用限价单）
# 这里简化，实际需要记录限价和成交价

# 执行时间
if 'datetime' in trades_df.columns:
start_time = trades_df['datetime'].min()
end_time = trades_df['datetime'].max()
duration = (end_time – start_time).total_seconds()
else:
duration = 0

return {
'vwap': vwap,
'benchmark_price': benchmark_price,
'execution_cost': execution_cost,
'cost_ratio': cost_ratio,
'duration': duration,
'total_volume': total_volume,
'trade_count': len(trades_df)
}

# 示例使用
from tqsdk import TqApi, TqAuth, TqSim
from tqsdk.lib import TargetPosScheduler
from tqsdk.algorithm import twap_table

api = TqApi(TqSim(init_balance=500000), auth=TqAuth("快期账户", "快期密码"))

SYMBOL = "SHFE.rb2510"
quote = api.get_quote(SYMBOL)
api.wait_update()

benchmark = quote.last_price # 使用当前价作为基准

time_table = twap_table(api, SYMBOL, 50, 300, 30, 60)
scheduler = TargetPosScheduler(api, SYMBOL, time_table)

while not scheduler.is_finished():
api.wait_update()

# 分析执行成本
if len(scheduler.trades_df) > 0:
cost_analysis = analyze_execution_cost(scheduler.trades_df, benchmark)

print("=" * 50)
print("执行成本分析")
print("=" * 50)

print(f"成交均价(VWAP): {cost_analysis['vwap']:.2f}")
print(f"基准价格: {cost_analysis['benchmark_price']:.2f}")
print(f"执行成本: {cost_analysis['execution_cost']:.2f}")
print(f"成本比例: {cost_analysis['cost_ratio']:.4%}")
print(f"执行时间: {cost_analysis['duration']:.0f}秒")
print(f"成交笔数: {cost_analysis['trade_count']}")

api.close()

6.2 执行效率评估

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
功能：执行效率评估
说明：本代码仅供学习参考
"""

def evaluate_execution_efficiency(trades_df, target_volume, target_duration):
"""
评估执行效率

参数:
trades_df: 成交记录
target_volume: 目标成交量
target_duration: 目标执行时间（秒）
"""
if len(trades_df) == 0:
return None

# 完成度
actual_volume = trades_df['volume'].sum()
completion_rate = actual_volume / target_volume if target_volume > 0 else 0

# 时间效率
if 'datetime' in trades_df.columns:
actual_duration = (trades_df['datetime'].max() – trades_df['datetime'].min()).total_seconds()
time_efficiency = target_duration / actual_duration if actual_duration > 0 else 0
else:
actual_duration = 0
time_efficiency = 0

# 执行均匀度（理想情况下应该均匀分布）
if len(trades_df) > 1 and 'datetime' in trades_df.columns:
time_intervals = trades_df['datetime'].diff().dt.total_seconds().dropna()
if len(time_intervals) > 0:
interval_std = time_intervals.std()
interval_mean = time_intervals.mean()
uniformity = 1 – (interval_std / interval_mean) if interval_mean > 0 else 0
else:
uniformity = 0
else:
uniformity = 0

# 综合效率得分
efficiency_score = (completion_rate * 0.4 +
min(time_efficiency, 1.0) * 0.3 +
max(uniformity, 0) * 0.3)

return {
'completion_rate': completion_rate,
'time_efficiency': time_efficiency,
'uniformity': uniformity,
'efficiency_score': efficiency_score,
'actual_volume': actual_volume,
'actual_duration': actual_duration
}

# 示例使用
from tqsdk import TqApi, TqAuth, TqSim
from tqsdk.lib import TargetPosScheduler
from tqsdk.algorithm import twap_table

api = TqApi(TqSim(init_balance=500000), auth=TqAuth("快期账户", "快期密码"))

SYMBOL = "SHFE.rb2510"
TARGET_VOLUME = 50
TARGET_DURATION = 300

time_table = twap_table(api, SYMBOL, TARGET_VOLUME, TARGET_DURATION, 30, 60)
scheduler = TargetPosScheduler(api, SYMBOL, time_table)

while not scheduler.is_finished():
api.wait_update()

if len(scheduler.trades_df) > 0:
efficiency = evaluate_execution_efficiency(scheduler.trades_df,
TARGET_VOLUME,
TARGET_DURATION)

print("=" * 50)
print("执行效率评估")
print("=" * 50)

print(f"完成度: {efficiency['completion_rate']:.2%}")
print(f"时间效率: {efficiency['time_efficiency']:.2f}")
print(f"执行均匀度: {efficiency['uniformity']:.2f}")
print(f"综合效率得分: {efficiency['efficiency_score']:.2f}")

api.close()

七、总结

执行优化速查

# 大单拆分
target_pos = TargetPosTask(api, symbol, min_volume=2, max_volume=10)

# TWAP执行
time_table = twap_table(api, symbol, volume, duration, min_interval, max_interval)

# 动态限价
def get_price(direction):
return quote.bid_price1 + tick # 自定义价格

免责声明：本文仅供学习交流使用，不构成任何投资建议。期货交易有风险，入市需谨慎。

更多资源：

• 天勤量化官网：https://www.shinnytech.com
• GitHub开源地址：https://github.com/shinnytech/tqsdk-python
• 官方文档：https://doc.shinnytech.com/tqsdk/latest