核心内容摘要
长门和小南拨钢筋
前言实盘交易与回测存在很大差异实盘优化是量化策略成功的关键。
本文
总结实盘交易中的
常见问题和优化技巧帮助策略在实盘中取得更好表现。
本文将介绍回测与实盘的差异滑点与手续费处理订单执行优化风险控制优化实盘监控与调试
为什么选择天勤量化TqSdkTqSdk实盘优化支持功能说明模拟交易TqSim模拟真实交易环境实盘对接支持实盘交易订单管理支持订单状态管理风险控制支持实时风控安装方法pipinstalltqsdk pandas numpy
回测与实盘差异
1 主要差异差异项回测实盘影响滑点假设无滑点实际存在滑点收益降低手续费可能简化实际手续费成本增加数据质量可能理想化实际数据延迟信号滞后流动性假设充足实际有限成交困难心理因素无有执行偏差
2 优化方向方向说明滑点模型建立滑点模型成本控制考虑真实成本数据验证验证数据质量流动性管理考虑流动性限制
滑点处理
1 滑点模型#!/usr/bin/env python# -*- coding: utf-8 -*- 功能滑点模型 说明本代码仅供学习参考 importpandasaspdimportnumpyasnpfromtqsdkimportTqApi,TqAuthclassSlippageModel:滑点模型def__init__(self,base_slippage
0001,volatility_factor
0,volume_factor
1.
: 初始化 参数: base_slippage: 基础滑点如
01% volatility_factor: 波动率因子 volume_factor: 成交量因子 self.base_slippagebase_slippage self.volatility_factorvolatility_factor self.volume_factorvolume_factordefcalculate_slippage(self,price,volume,volatility,market_volume): 计算滑点 参数: price: 当前价格 volume: 交易量 volatility: 波动率 market_volume: 市场成交量 # 基础滑点slippageself.base_slippage# 波动率调整波动大滑点大slippage*(1volatility*self.volatility_factor)# 成交量调整量大滑点大volume_ratiovolume/(market_volume1e-
slippage*(1volume_ratio*self.volume_factor)# 计算滑点价格slippage_priceprice*(1slippage)returnslippage_price,slippagedefapply_slippage(self,order_price,order_volume,current_volatility,current_volume):应用滑点actual_price,slippageself.calculate_slippage(order_price,order_volume,current_volatility,current_volume)returnactual_price# 使用示例slippage_modelSlippageModel(base_slippage
0001,volatility_factor
0,volume_factor
1.
# 模拟订单order_price4000order_volume10volatility
02market_volume10000actual_price,slippageslippage_model.calculate_slippage(order_price,order_volume,volatility,market_volume)print(f滑点模型:)print(f 订单价格:{order_price})print(f 实际价格:{actual_price:.2f})print(f 滑点:{slippage:.4f}({slippage*100:.2%}))
2 滑点优化#!/usr/bin/env python# -*- coding: utf-8 -*- 功能滑点优化策略 说明本代码仅供学习参考 importpandasaspdimportnumpyasnpfromtqsdkimportTqApi,TqAuthclassSlippageOptimizer:滑点优化器def__init__(self):初始化passdefoptimize_order_timing(self,df,target_volume): 优化订单时机 参数: df: K线数据 target_volume: 目标交易量 # 计算流动性指标df[liquidity]df[volume]/df[volume].rolling(
.mean()df[volatility]df[close].pct_change().rolling(
.std()# 选择流动性好、波动率低的时机df[timing_score](df[liquidity]*
6(1-df[volatility]/df[volatility].max())*
0.
# 找出最佳时机best_timingdf.nlargest(1,timing_score)returnbest_timing.index[0]iflen(best_timing)0elseNonedefsplit_order(self,total_volume,max_order_size
: 拆分大单 参数: total_volume: 总交易量 max_order_size: 单笔最大量 num_ordersint(np.ceil(total_volume/max_order_size))order_sizes[max_order_size]*(num_orders-
order_sizes.append(total_volume-sum(order_sizes))returnorder_sizes# 使用示例apiTqApi(authTqAuth(快期账户,快期密码))SYMBOLSHFE.rb2510klinesapi.get_kline_serial(SYMBOL,60,
api.wait_update()optimizerSlippageOptimizer()# 优化订单时机best_timingoptimizer.optimize_order_timing(klines,target_volume
print(f最佳订单时机:{best_timing})# 拆分大单order_sizesoptimizer.split_order(total_volume20,max_order_size
print(f订单拆分:{order_sizes})api.close()
手续费优化
1 手续费计算#!/usr/bin/env python# -*- coding: utf-8 -*- 功能手续费计算 说明本代码仅供学习参考 classCommissionCalculator:手续费计算器def__init__(self,commission_rate
0001,min_commission
: 初始化 参数: commission_rate: 手续费率如
01% min_commission: 最小手续费 self.commission_ratecommission_rate self.min_commissionmin_commissiondefcalculate_commission(self,price,volume,is_openTrue): 计算手续费 参数: price: 成交价格 volume: 交易量 is_open: 是否开仓开仓和平仓都收费 commissionprice*volume*self.commission_rate commissionmax(commission,self.min_commission)# 开仓和平仓都收费ifnotis_open:commission*1# 平仓同样收费returncommissiondefcalculate_round_trip_cost(self,price,volume): 计算往返成本开仓平仓 参数: price: 成交价格 volume: 交易量 open_commissionself.calculate_commission(price,volume,is_openTrue)close_commissionself.calculate_commission(price,volume,is_openFalse)returnopen_commissionclose_commission# 使用示例calculatorCommissionCalculator(commission_rate
0001,min_commission
price4000volume10open_commissioncalculator.calculate_commission(price,volume,is_openTrue)close_commissioncalculator.calculate_commission(price,volume,is_openFalse)round_trip_costcalculator.calculate_round_trip_cost(price,volume)print(f手续费计算:)print(f 开仓手续费:{open_commission:.2f})print(f 平仓手续费:{close_commission:.2f})print(f 往返成本:{round_trip_cost:.2f})print(f 成本率:{round_trip_cost/(price*volume):.4%})
订单执行优化
1 限价单优化#!/usr/bin/env python# -*- coding: utf-8 -*- 功能限价单优化 说明本代码仅供学习参考 importpandasaspdimportnumpyasnpfromtqsdkimportTqApi,TqAuthclassLimitOrderOptimizer:限价单优化器def__init__(self,max_wait_time300,price_tolerance
0.
: 初始化 参数: max_wait_time: 最大等待时间秒 price_tolerance: 价格容忍度如
1% self.max_wait_timemax_wait_time self.price_toleranceprice_tolerancedefcalculate_limit_price(self,market_price,direction,toleranceNone): 计算限价 参数: market_price: 市场价格 direction: 方向 (1: 买入, -1: 卖出) tolerance: 价格容忍度 iftoleranceisNone:toleranceself.price_toleranceifdirection1:# 买入limit_pricemarket_price*(1-tolerance)else:# 卖出limit_pricemarket_price*(1tolerance)returnlimit_pricedefshould_convert_to_market(self,limit_price,current_price,direction,elapsed_time): 判断是否转为市价单 参数: limit_price: 限价 current_price: 当前价格 direction: 方向 elapsed_time: 已等待时间 # 时间超限ifelapsed_timeself.max_wait_time:returnTrue# 价格偏离ifdirection1:# 买入ifcurrent_pricelimit_price*(1self.price_tolerance):returnTrueelse:# 卖出ifcurrent_pricelimit_price*(1-self.price_tolerance):returnTruereturnFalse# 使用示例optimizerLimitOrderOptimizer(max_wait_time300,price_tolerance
0.
market_price4000direction1# 买入limit_priceoptimizer.calculate_limit_price(market_price,direction)print(f限价单优化:)print(f 市场价格:{market_price})print(f 限价:{limit_price:.2f})print(f 价差:{abs(market_price-limit_price):.2f})# 模拟判断current_price4005elapsed_time350should_convertoptimizer.should_convert_to_market(limit_price,current_price,direction,elapsed_time)print(f 是否转为市价:{should_convert})
风险控制优化
1 实时风控#!/usr/bin/env python# -*- coding: utf-8 -*- 功能实时风控 说明本代码仅供学习参考 importpandasaspdimportnumpyasnpfromtqsdkimportTqApi,TqAuthclassRealTimeRiskControl:实时风控def__init__(self,max_position100,max_drawdown
2,max_daily_loss
0.
: 初始化 参数: max_position: 最大持仓 max_drawdown: 最大回撤 max_daily_loss: 最大日亏损 self.max_positionmax_position self.max_drawdownmax_drawdown self.max_daily_lossmax_daily_loss self.initial_balanceNoneself.peak_balanceNoneself.daily_pnl0defcheck_position_limit(self,current_position,new_order_volume):检查持仓限制new_positioncurrent_positionnew_order_volumeifabs(new_position)self.max_position:returnFalse,f持仓超限:{new_position}{self.max_position}returnTrue,defcheck_drawdown(self,current_balance):检查回撤ifself.initial_balanceisNone:self.initial_balancecurrent_balance self.peak_balancecurrent_balance self.peak_balancemax(self.peak_balance,current_balance)drawdown(self.peak_balance-current_balance)/self.peak_balanceifdrawdownself.max_drawdown:returnFalse,f回撤超限:{drawdown:.2%}{self.max_drawdown:.2%}returnTrue,defcheck_daily_loss(self,daily_pnl,current_balance):检查日亏损daily_loss_ratioabs(daily_pnl)/current_balanceifdaily_pnl0else0ifdaily_loss_ratioself.max_daily_loss:returnFalse,f日亏损超限:{daily_loss_ratio:.2%}{self.max_daily_loss:.2%}returnTrue,defcheck_all(self,current_position,new_order_volume,current_balance,daily_pnl):综合检查checks[self.check_position_limit(current_position,new_order_volume),self.check_drawdown(current_balance),self.check_daily_loss(daily_pnl,current_balance)]forpassed,messageinchecks:ifnotpassed:returnFalse,messagereturnTrue,通过# 使用示例risk_controlRealTimeRiskControl(max_position100,max_drawdown
2,max_daily_loss
0.
# 模拟检查current_position50new_order_volume60current_balance95000# 初始100000亏损5000daily_pnl-3000passed,messagerisk_control.check_all(current_position,new_order_volume,current_balance,daily_pnl)print(f风控检查:{通过ifpassedelse拒绝})ifnotpassed:print(f 原因:{message})
实盘监控
1 性能监控#!/usr/bin/env python# -*- coding: utf-8 -*- 功能实盘性能监控 说明本代码仅供学习参考 importpandasaspdimportnumpyasnpfromdatetimeimportdatetimefromtqsdkimportTqApi,TqAuthclassLivePerformanceMonitor:实盘性能监控def__init__(self):初始化self.trades[]self.daily_pnl[]self.equity_curve[]defrecord_trade(self,symbol,direction,price,volume,pnl):记录交易trade{timestamp:datetime.now(),symbol:symbol,direction:direction,price:price,volume:volume,pnl:pnl}self.trades.append(trade)defupdate_daily_pnl(self,pnl):更新日盈亏self.daily_pnl.append({date:datetime.now().date(),pnl:pnl})defupdate_equity(self,equity):更新权益self.equity_curve.append({timestamp:datetime.now(),equity:equity})defget_statistics(self):获取统计信息ifnotself.trades:return{}df_tradespd.DataFrame(self.trades)stats{total_trades:len(self.trades),win_rate:(df_trades[pnl]
.sum()/len(df_trades)iflen(df_trades)0else0,total_pnl:df_trades[pnl].sum(),avg_pnl:df_trades[pnl].mean(),max_profit:df_trades[pnl].max(),max_loss:df_trades[pnl].min()}ifself.equity_curve:df_equitypd.DataFrame(self.equity_curve)initial_equitydf_equity[equity].iloc[0]current_equitydf_equity[equity].iloc[-1]stats[total_return](current_equity-initial_equity)/initial_equityreturnstats# 使用示例monitorLivePerformanceMonitor()# 模拟记录monitor.record_trade(SHFE.rb2510,1,4000,10,
monitor.record_trade(SHFE.rb2510,-1,4050,10,-
monitor.update_equity(
monitor.update_equity(
statsmonitor.get_statistics()print(实盘统计:)forkey,valueinstats.items():ifisinstance(value,float):ifrateinkeyorreturninkey:print(f{key}:{value:.2%})else:print(f{key}:{value:.2f})else:print(f{key}:{value})
九、
常见问题Q1: 实盘优化最重要的是什么A: 最重要的是风险控制成本管理执行质量持续监控Q2: 如何减少实盘与回测的差异A: 方法使用更真实的滑点模型考虑真实手续费验证数据质量模拟真实执行Q3: 实盘优化需要多长时间A: 建议持续优化过程定期评估和调整根据市场变化调整记录和分析问题
十、
总结要点说明滑点处理建立滑点模型手续费优化考虑真实成本订单执行优化执行策略风险控制实时风控性能监控持续监控下一步学习建议深入学习订单执行算法研究更复杂的滑点模型探索自动化优化学习实盘问题诊断免责声明本文仅供学习交流使用不构成任何投资建议。
期货交易有风险入市需谨慎。
更多资源天勤量化官网https://www.shinnytech.comGitHub开源地址https://github.com/shinnytech/tqsdk-python官方文档https://doc.shinnytech.com/tqsdk/latest