免责声明：本文基于个人使用体验，与任何厂商无商业关系。内容仅供技术交流参考，不构成投资建议。
一、前言
二十年的期货交易生涯，让我深刻认识到一个道理：风控是量化交易的生命线。
很多人把精力都放在策略开发上，却忽视了风控系统。结果一次黑天鹅事件，就把多年的利润吃光。
今天这篇文章，我来分享一下期货量化风控系统的设计思路和实践经验。
二、风控系统的核心要素
一个完整的风控系统包含以下要素：
| 要素 | 说明 |
|---|---|
| 仓位控制 | 控制单次交易的资金比例 |
| 止损策略 | 限制单笔交易的最大亏损 |
| 回撤控制 | 限制账户的最大回撤 |
| 频率限制 | 避免过度交易 |
| 异常处理 | 处理各种异常情况 |
三、仓位控制
3.1 固定手数法
最简单的方法，每次固定交易N手：
FIXED_VOLUME=1# 固定每次交易1手defget_volume():returnFIXED_VOLUME优点：简单
缺点：没有考虑账户资金变化
3.2 固定比例法
每次使用账户资金的固定比例：
defget_volume_by_ratio(account,quote,ratio=0.1):""" 按资金比例计算仓位 ratio: 使用资金比例，如0.1表示使用10%资金 """available=account.available*ratio margin_per_hand=quote.margin volume=int(available/margin_per_hand)returnmax(1,volume)# 至少1手3.3 ATR仓位法（推荐）
基于ATR（真实波动幅度）控制风险：
defget_volume_by_atr(account,atr_value,contract_multiplier,risk_per_trade=0.02):""" 基于ATR的仓位计算 risk_per_trade: 单笔交易风险比例，如0.02表示2% """# 单笔最大可承受亏损risk_amount=account.balance*risk_per_trade# 每手的风险（2倍ATR作为止损距离）risk_per_hand=atr_value*2*contract_multiplier# 计算手数volume=int(risk_amount/risk_per_hand)returnmax(1,volume)3.4 最大持仓限制
MAX_POSITION=5# 最大持仓手数defcheck_position_limit(position,target_volume):"""检查持仓限制"""current_pos=position.pos_long+position.pos_shortifcurrent_pos+target_volume>MAX_POSITION:allowed=MAX_POSITION-current_posprint(f"超出持仓限制，调整为{allowed}手")returnmax(0,allowed)returntarget_volume四、止损策略
4.1 固定点数止损
STOP_LOSS_POINTS=50# 止损50点defcheck_fixed_stop(entry_price,current_price,direction):"""固定点数止损"""ifdirection=="LONG":ifcurrent_price<=entry_price-STOP_LOSS_POINTS:returnTrueelifdirection=="SHORT":ifcurrent_price>=entry_price+STOP_LOSS_POINTS:returnTruereturnFalse4.2 百分比止损
STOP_LOSS_PERCENT=0.02# 止损2%defcheck_percent_stop(entry_price,current_price,direction):"""百分比止损"""ifdirection=="LONG":ifcurrent_price<=entry_price*(1-STOP_LOSS_PERCENT):returnTrueelifdirection=="SHORT":ifcurrent_price>=entry_price*(1+STOP_LOSS_PERCENT):returnTruereturnFalse4.3 ATR跟踪止损（推荐）
classATRTrailingStop:"""ATR跟踪止损"""def__init__(self,atr_multiplier=2.0):self.atr_multiplier=atr_multiplier self.stop_price=Noneself.highest_price=Noneself.lowest_price=Nonedefupdate(self,current_price,atr_value,direction):"""更新止损价"""ifdirection=="LONG":ifself.highest_priceisNone:self.highest_price=current_priceelse:self.highest_price=max(self.highest_price,current_price)# 跟踪止损new_stop=self.highest_price-self.atr_multiplier*atr_valueifself.stop_priceisNoneornew_stop>self.stop_price:self.stop_price=new_stopelifdirection=="SHORT":ifself.lowest_priceisNone:self.lowest_price=current_priceelse:self.lowest_price=min(self.lowest_price,current_price)new_stop=self.lowest_price+self.atr_multiplier*atr_valueifself.stop_priceisNoneornew_stop<self.stop_price:self.stop_price=new_stopreturnself.stop_pricedefshould_stop(self,current_price,direction):"""判断是否触发止损"""ifself.stop_priceisNone:returnFalseifdirection=="LONG":returncurrent_price<=self.stop_priceelifdirection=="SHORT":returncurrent_price>=self.stop_pricereturnFalse五、回撤控制
5.1 最大回撤限制
classDrawdownController:"""回撤控制器"""def__init__(self,max_drawdown=0.15,daily_max_loss=0.03):self.max_drawdown=max_drawdown self.daily_max_loss=daily_max_loss self.peak_balance=Noneself.daily_start_balance=Nonedefupdate(self,current_balance,is_new_day=False):"""更新状态"""ifis_new_dayorself.daily_start_balanceisNone:self.daily_start_balance=current_balanceifself.peak_balanceisNoneorcurrent_balance>self.peak_balance:self.peak_balance=current_balancedefcheck(self,current_balance):"""检查是否超出限制"""# 检查总回撤ifself.peak_balance:drawdown=(self.peak_balance-current_balance)/self.peak_balanceifdrawdown>self.max_drawdown:returnFalse,f"总回撤{drawdown:.2%}超过限制{self.max_drawdown:.2%}"# 检查日亏损ifself.daily_start_balance:daily_loss=(self.daily_start_balance-current_balance)/self.daily_start_balanceifdaily_loss>self.daily_max_loss:returnFalse,f"日亏损{daily_loss:.2%}超过限制{self.daily_max_loss:.2%}"returnTrue,"正常"5.2 使用示例
fromtqsdkimportTqApi,TqAuth,TqSim api=TqApi(TqSim(),auth=TqAuth("账户","密码"))account=api.get_account()# 初始化回撤控制器dd_controller=DrawdownController(max_drawdown=0.15,daily_max_loss=0.03)dd_controller.update(account.balance)whileTrue:api.wait_update()# 检查回撤is_ok,msg=dd_controller.check(account.balance)ifnotis_ok:print(f"风控触发:{msg}")# 平掉所有仓位# 停止交易break# 正常交易逻辑...六、完整风控系统示例
fromtqsdkimportTqApi,TqAuth,TqSimfromtqsdk.taimportATRimportloggingfromdatetimeimportdatetime# 配置日志logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(message)s')classRiskManager:"""风控管理器"""def__init__(self,config):self.config=config self.trailing_stop=ATRTrailingStop(config['atr_multiplier'])self.dd_controller=DrawdownController(config['max_drawdown'],config['daily_max_loss'])self.entry_price=Noneself.direction=Nonedefcalc_volume(self,account,quote,atr_value):"""计算交易手数"""# 基于ATR计算risk_amount=account.balance*self.config['risk_per_trade']risk_per_hand=atr_value*self.config['atr_multiplier']*10# 假设合约乘数10volume=int(risk_amount/risk_per_hand)# 限制最大手数volume=min(volume,self.config['max_volume'])returnmax(1,volume)defon_open_position(self,price,direction):"""开仓时调用"""self.entry_price=price self.direction=direction self.trailing_stop=ATRTrailingStop(self.config['atr_multiplier'])logging.info(f"开仓 | 方向:{direction}价格:{price}")defon_close_position(self):"""平仓时调用"""self.entry_price=Noneself.direction=Nonelogging.info("平仓")defcheck_risk(self,account,current_price,atr_value):"""检查风险"""# 1. 检查回撤self.dd_controller.update(account.balance)is_ok,msg=self.dd_controller.check(account.balance)ifnotis_ok:logging.warning(f"回撤风控触发:{msg}")returnFalse,"CLOSE_ALL"# 2. 检查止损ifself.entry_priceandself.direction:self.trailing_stop.update(current_price,atr_value,self.direction)ifself.trailing_stop.should_stop(current_price,self.direction):logging.warning(f"止损触发 | 止损价:{self.trailing_stop.stop_price}")returnFalse,"STOP_LOSS"returnTrue,"OK"# ========== 策略主程序 ==========CONFIG={'symbol':'SHFE.rb2505','risk_per_trade':0.02,'atr_multiplier':2.0,'max_drawdown':0.15,'daily_max_loss':0.03,'max_volume':5,}api=TqApi(TqSim(init_balance=100000),auth=TqAuth("账户","密码"))risk_mgr=RiskManager(CONFIG)quote=api.get_quote(CONFIG['symbol'])klines=api.get_kline_serial(CONFIG['symbol'],60,200)position=api.get_position(CONFIG['symbol'])account=api.get_account()whileTrue:api.wait_update()ifapi.is_changing(klines):# 计算ATRatr=ATR(klines,20)atr_value=atr.atr.iloc[-1]current_price=klines.close.iloc[-1]# 风险检查is_ok,action=risk_mgr.check_risk(account,current_price,atr_value)ifnotis_ok:ifaction=="CLOSE_ALL"oraction=="STOP_LOSS":# 平掉所有仓位ifposition.pos_long>0:api.insert_order(CONFIG['symbol'],"SELL","CLOSE",position.pos_long)ifposition.pos_short>0:api.insert_order(CONFIG['symbol'],"BUY","CLOSE",position.pos_short)risk_mgr.on_close_position()ifaction=="CLOSE_ALL":logging.error("回撤超限，停止交易")breakcontinue# 正常交易逻辑...# ...七、风控参数建议
| 参数 | 保守 | 中等 | 激进 |
|---|---|---|---|
| 单笔风险比例 | 1% | 2% | 3% |
| 最大持仓 | 3手 | 5手 | 10手 |
| ATR止损倍数 | 3.0 | 2.0 | 1.5 |
| 最大回撤 | 10% | 15% | 20% |
| 日最大亏损 | 2% | 3% | 5% |
八、总结
期货量化风控系统的核心原则：
- 仓位控制：永远不要满仓，控制单笔风险
- 止损纪律：止损必须执行，不要心存侥幸
- 回撤限制：设置明确的回撤上限
- 持续监控：风控不是一次性的，需要持续监控
我目前使用TqSdk做风控，主要因为API简洁，容易实现各种风控逻辑。当然，这只是我个人的选择，每个人需求不同。
希望这篇文章对你有所帮助！
声明：本文基于个人学习经验整理，仅供技术交流参考，不构成任何投资建议。
