一、前言
订单执行质量直接影响交易成本和策略收益。优化订单执行可以减少滑点、降低交易成本、提高策略表现。本文将介绍各种订单执行优化方法。
本文将介绍：
- 订单执行类型
- 滑点控制
- 执行算法
- 时机选择
- 成本优化
二、为什么选择天勤量化（TqSdk）
TqSdk订单执行支持：
| 功能 | 说明 |
|---|---|
| 多种订单类型 | 支持限价、市价等 |
| 快速执行 | 支持快速下单 |
| 订单管理 | 支持订单查询和管理 |
| 灵活扩展 | 支持自定义执行逻辑 |
安装方法：
pipinstalltqsdk pandas numpy三、订单执行基础
3.1 订单类型
| 类型 | 说明 | 适用场景 |
|---|---|---|
| 市价单 | 立即成交 | 快速执行 |
| 限价单 | 指定价格 | 控制成本 |
| 止损单 | 触发止损 | 风险控制 |
| 止盈单 | 触发止盈 | 锁定利润 |
3.2 执行成本
| 成本类型 | 说明 | 影响因素 |
|---|---|---|
| 手续费 | 交易手续费 | 交易所规定 |
| 滑点 | 成交价差 | 流动性、订单大小 |
| 冲击成本 | 市场冲击 | 订单大小、市场深度 |
四、滑点控制
4.1 滑点估算
#!/usr/bin/env python# -*- coding: utf-8 -*-""" 功能：订单执行优化 说明：本代码仅供学习参考 """fromtqsdkimportTqApi,TqAuthimportpandasaspdimportnumpyasnpdefestimate_slippage(api,symbol,volume,direction="BUY"):"""估算滑点"""quote=api.get_quote(symbol)api.wait_update()ifdirection=="BUY":target_price=quote.ask_price1 available_volume=quote.ask_volume1# 如果订单量小于可用量，滑点较小ifvolume<=available_volume:slippage=0.0001# 0.01%else:# 需要吃多个价位slippage=0.0005# 0.05%else:target_price=quote.bid_price1 available_volume=quote.bid_volume1ifvolume<=available_volume:slippage=0.0001else:slippage=0.0005returnslippage,target_price# 使用示例api=TqApi(auth=TqAuth("快期账户","快期密码"))slippage,price=estimate_slippage(api,"SHFE.rb2510",1,"BUY")print(f"预计滑点:{slippage:.4%}, 目标价格:{price:.2f}")api.close()4.2 滑点控制策略
defcontrol_slippage(api,symbol,direction,volume,max_slippage=0.001):"""控制滑点"""slippage,target_price=estimate_slippage(api,symbol,volume,direction)ifslippage>max_slippage:# 滑点过大，分批执行batch_size=int(volume*max_slippage/slippage)returnbatch_sizeelse:# 滑点可接受，直接执行returnvolume# 使用示例optimal_volume=control_slippage(api,"SHFE.rb2510","BUY",10,max_slippage=0.001)print(f"建议分批执行，每批:{optimal_volume}手")五、执行算法
5.1 TWAP算法
deftwap_execution(api,symbol,direction,total_volume,duration_minutes=60):""" TWAP（时间加权平均价格）执行 参数: duration_minutes: 执行时间（分钟） """intervals=duration_minutes volume_per_interval=total_volume/intervals orders=[]foriinrange(intervals):quote=api.get_quote(symbol)api.wait_update()# 以中间价下单mid_price=(quote.bid_price1+quote.ask_price1)/2# 限价单order=api.insert_order(symbol,direction,"OPEN",int(volume_per_interval),limit_price=mid_price)orders.append(order)api.wait_update()ifi<intervals-1:time.sleep(60)# 等待1分钟returnorders# 使用示例orders=twap_execution(api,"SHFE.rb2510","BUY",10,duration_minutes=30)print(f"TWAP执行完成，共{len(orders)}笔订单")5.2 VWAP算法
defvwap_execution(api,symbol,direction,total_volume,klines):""" VWAP（成交量加权平均价格）执行 参数: klines: 历史K线数据 """# 计算历史VWAPvwap=(klines['close']*klines['volume']).sum()/klines['volume'].sum()quote=api.get_quote(symbol)api.wait_update()current_price=quote.last_price# 如果当前价格低于VWAP，买入ifdirection=="BUY"andcurrent_price<vwap:# 根据价格偏离程度决定执行量price_deviation=(vwap-current_price)/vwap execution_ratio=min(price_deviation*10,1.0)# 最多100%volume=int(total_volume*execution_ratio)order=api.insert_order(symbol,direction,"OPEN",volume)returnorderreturnNone# 使用示例klines=api.get_kline_serial("SHFE.rb2510",3600,100)api.wait_update()order=vwap_execution(api,"SHFE.rb2510","BUY",10,klines)5.3 冰山算法
deficeberg_execution(api,symbol,direction,total_volume,visible_size=1):""" 冰山算法（隐藏大单） 参数: visible_size: 可见订单大小 """remaining_volume=total_volume orders=[]whileremaining_volume>0:current_volume=min(visible_size,remaining_volume)# 限价单quote=api.get_quote(symbol)api.wait_update()ifdirection=="BUY":limit_price=quote.ask_price1else:limit_price=quote.bid_price1 order=api.insert_order(symbol,direction,"OPEN",current_volume,limit_price=limit_price)orders.append(order)remaining_volume-=current_volume# 等待订单成交api.wait_update()time.sleep(1)returnorders六、时机选择
6.1 流动性时机
deffind_liquidity_window(klines,window_hours=24):"""寻找流动性窗口"""# 按小时统计成交量klines['hour']=klines.index.hour hourly_volume=klines.groupby('hour')['volume'].mean()# 找出成交量最大的时段best_hours=hourly_volume.nlargest(3).index.tolist()returnbest_hours# 使用示例best_hours=find_liquidity_window(klines)print(f"最佳交易时段:{best_hours}")6.2 波动率时机
deffind_low_volatility_window(klines):"""寻找低波动率窗口"""returns=klines['close'].pct_change()# 按小时统计波动率klines['hour']=klines.index.hour hourly_vol=klines.groupby('hour').apply(lambdax:x['close'].pct_change().std())# 找出波动率最低的时段best_hours=hourly_vol.nsmallest(3).index.tolist()returnbest_hours# 使用示例low_vol_hours=find_low_volatility_window(klines)print(f"低波动率时段:{low_vol_hours}")七、成本优化
7.1 手续费优化
defoptimize_commission(api,symbol,direction,volume,commission_rate=0.0001):"""优化手续费"""quote=api.get_quote(symbol)api.wait_update()price=quote.last_price contract_value=price*volume*10# 假设每手10吨commission=contract_value*commission_rate# 如果手续费过高，考虑分批执行ifcommission>100:# 假设阈值100元# 分批执行以降低单笔手续费batches=int(np.ceil(volume/5))# 每批5手returnbatcheselse:return1# 使用示例batches=optimize_commission(api,"SHFE.rb2510","BUY",20)print(f"建议分{batches}批执行")7.2 总成本优化
defoptimize_total_cost(api,symbol,direction,volume):"""优化总成本"""# 估算各项成本slippage,target_price=estimate_slippage(api,symbol,volume,direction)slippage_cost=target_price*volume*10*slippage commission_rate=0.0001contract_value=target_price*volume*10commission_cost=contract_value*commission_rate total_cost=slippage_cost+commission_cost# 如果成本过高，优化执行方式iftotal_cost>200:# 假设阈值200元# 使用TWAP执行return"TWAP",total_cost*0.8# TWAP可降低20%成本else:return"MARKET",total_cost# 使用示例execution_method,cost=optimize_total_cost(api,"SHFE.rb2510","BUY",10)print(f"推荐执行方式:{execution_method}, 预计成本:{cost:.2f}元")八、执行监控
8.1 执行质量监控
classExecutionMonitor:"""执行监控"""def__init__(self,api):self.api=api self.executions=[]deftrack_execution(self,order_id,expected_price,expected_volume):"""跟踪执行"""order=self.api.get_order(order_id)self.api.wait_update()iforder.status=="FINISHED":# 计算执行质量actual_price=order.trade_priceifhasattr(order,'trade_price')elseexpected_price actual_volume=order.trade_volume slippage=abs(actual_price-expected_price)/expected_price fill_rate=actual_volume/expected_volume execution_record={'order_id':order_id,'expected_price':expected_price,'actual_price':actual_price,'slippage':slippage,'fill_rate':fill_rate}self.executions.append(execution_record)returnexecution_recordreturnNonedefget_execution_stats(self):"""获取执行统计"""ifnotself.executions:returnNoneavg_slippage=np.mean([e['slippage']foreinself.executions])avg_fill_rate=np.mean([e['fill_rate']foreinself.executions])return{'avg_slippage':avg_slippage,'avg_fill_rate':avg_fill_rate,'total_executions':len(self.executions)}# 使用示例monitor=ExecutionMonitor(api)stats=monitor.get_execution_stats()ifstats:print(f"平均滑点:{stats['avg_slippage']:.4%}")print(f"平均成交率:{stats['avg_fill_rate']:.2%}")九、总结
9.1 执行优化要点
| 要点 | 说明 |
|---|---|
| 滑点控制 | 控制执行滑点 |
| 算法选择 | 选择合适的执行算法 |
| 时机选择 | 选择最佳执行时机 |
| 成本优化 | 优化总执行成本 |
9.2 注意事项
- 流动性- 考虑市场流动性
- 成本平衡- 平衡滑点和手续费
- 执行速度- 平衡执行速度和成本
- 持续监控- 持续监控执行质量
免责声明：本文仅供学习交流使用，不构成任何投资建议。期货交易有风险，入市需谨慎。
更多资源：
- 天勤量化官网：https://www.shinnytech.com
- GitHub开源地址：https://github.com/shinnytech/tqsdk-python
- 官方文档：https://doc.shinnytech.com/tqsdk/tqsdk/latest
