当前位置：首页 > news >正文

【期货量化实战】量化交易策略的实盘监控系统（完整教程）

news 2026/3/26 21:37:41

一、前言

实盘监控是量化交易系统的重要组成部分。完善的监控系统可以及时发现异常、预警风险、保障策略稳定运行。本文将介绍如何构建完整的实盘监控系统。

监控系统架构
账户监控
策略监控
异常预警
监控面板

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

功能	说明
实时数据	支持获取实时账户和持仓数据
订单查询	支持查询订单状态
灵活扩展	支持自定义监控逻辑
稳定可靠	系统稳定可靠

安装方法：

pipinstalltqsdk pandas numpy

三、监控系统架构

3.1 监控模块

模块	功能
账户监控	监控账户资金、盈亏
持仓监控	监控持仓情况
订单监控	监控订单执行
策略监控	监控策略运行状态
异常监控	监控异常情况

3.2 监控指标

指标	说明
账户余额	账户资金余额
持仓盈亏	持仓盈亏情况
交易次数	交易次数统计
策略信号	策略信号生成
系统状态	系统运行状态

四、账户监控

4.1 基础监控

#!/usr/bin/env python# -*- coding: utf-8 -*-""" 功能：实盘监控系统 说明：本代码仅供学习参考 """fromtqsdkimportTqApi,TqAuthimportpandasaspdimporttimefromdatetimeimportdatetimeclassAccountMonitor:"""账户监控"""def__init__(self,api):self.api=api self.account_history=[]defmonitor_account(self):"""监控账户"""account=self.api.get_account()self.api.wait_update()account_info={'timestamp':datetime.now(),'balance':account.balance,'available':account.available,'margin':account.margin,'profit':account.profit,'risk_ratio':account.margin/account.balanceifaccount.balance>0else0}self.account_history.append(account_info)returnaccount_infodefcheck_alerts(self,account_info):"""检查预警"""alerts=[]# 资金使用率预警ifaccount_info['risk_ratio']>0.8:alerts.append("资金使用率过高")# 亏损预警ifaccount_info['profit']<-10000:alerts.append("亏损超过阈值")# 可用资金不足ifaccount_info['available']<10000:alerts.append("可用资金不足")returnalerts# 使用示例api=TqApi(auth=TqAuth("快期账户","快期密码"))monitor=AccountMonitor(api)whileTrue:account_info=monitor.monitor_account()alerts=monitor.check_alerts(account_info)ifalerts:foralertinalerts:print(f"预警:{alert}")print(f"账户余额:{account_info['balance']:.2f}, 盈亏:{account_info['profit']:.2f}")time.sleep(60)

4.2 持仓监控

classPositionMonitor:"""持仓监控"""def__init__(self,api,symbols):self.api=api self.symbols=symbols self.position_history=[]defmonitor_positions(self):"""监控持仓"""positions_info={}forsymbolinself.symbols:position=self.api.get_position(symbol)self.api.wait_update()positions_info[symbol]={'pos_long':position.pos_long,'pos_short':position.pos_short,'pos':position.pos_long-position.pos_short,'profit':position.profit}self.position_history.append({'timestamp':datetime.now(),'positions':positions_info})returnpositions_infodefcheck_position_alerts(self,positions_info):"""检查持仓预警"""alerts=[]forsymbol,pos_infoinpositions_info.items():# 持仓过大ifabs(pos_info['pos'])>10:alerts.append(f"{symbol}持仓过大:{pos_info['pos']}")# 持仓亏损过大ifpos_info['profit']<-5000:alerts.append(f"{symbol}持仓亏损过大:{pos_info['profit']:.2f}")returnalerts# 使用示例symbols=["SHFE.rb2510","SHFE.hc2510"]pos_monitor=PositionMonitor(api,symbols)positions=pos_monitor.monitor_positions()alerts=pos_monitor.check_position_alerts(positions)

五、策略监控

5.1 策略状态监控

classStrategyMonitor:"""策略监控"""def__init__(self,api,strategy_func,symbol):self.api=api self.strategy_func=strategy_func self.symbol=symbol self.signal_history=[]self.running=Truedefmonitor_strategy(self):"""监控策略"""klines=self.api.get_kline_serial(self.symbol,3600,200)self.api.wait_update()# 生成信号signal=self.strategy_func(klines)signal_info={'timestamp':datetime.now(),'signal':signal,'price':klines['close'].iloc[-1]}self.signal_history.append(signal_info)returnsignal_infodefcheck_strategy_alerts(self,signal_info):"""检查策略预警"""alerts=[]# 信号异常ifabs(signal_info['signal'])>10:alerts.append("策略信号异常")# 长时间无信号iflen(self.signal_history)>100:recent_signals=[s['signal']forsinself.signal_history[-100:]]ifall(s==0forsinrecent_signals):alerts.append("策略长时间无信号")returnalerts# 使用示例deftest_strategy(klines):fromtqsdk.tafuncimportma ma5=ma(klines['close'],5)ma20=ma(klines['close'],20)ifma5.iloc[-1]>ma20.iloc[-1]:return1elifma5.iloc[-1]<ma20.iloc[-1]:return-1return0strategy_monitor=StrategyMonitor(api,test_strategy,"SHFE.rb2510")signal_info=strategy_monitor.monitor_strategy()alerts=strategy_monitor.check_strategy_alerts(signal_info)

六、异常预警

6.1 预警系统

classAlertSystem:"""预警系统"""def__init__(self):self.alerts=[]defadd_alert(self,level,message,source):"""添加预警"""alert={'timestamp':datetime.now(),'level':level,# INFO, WARNING, ERROR'message':message,'source':source}self.alerts.append(alert)# 打印预警print(f"[{level}]{source}:{message}")defget_recent_alerts(self,count=10):"""获取最近预警"""returnself.alerts[-count:]# 使用示例alert_system=AlertSystem()alert_system.add_alert("WARNING","资金使用率过高","账户监控")alert_system.add_alert("ERROR","策略异常","策略监控")

6.2 通知系统

defsend_notification(message,level="INFO"):"""发送通知（可扩展为邮件、微信等）"""print(f"[{level}]{datetime.now()}:{message}")# 可以扩展为：# - 邮件通知# - 微信通知# - 短信通知# - 桌面通知# 使用示例send_notification("账户余额不足","WARNING")

七、监控面板

7.1 监控面板

classMonitoringDashboard:"""监控面板"""def__init__(self,api,monitors):self.api=api self.monitors=monitors self.alert_system=AlertSystem()defupdate(self):"""更新监控面板"""# 更新所有监控formonitorinself.monitors:try:monitor.update()exceptExceptionase:self.alert_system.add_alert("ERROR",f"监控异常:{e}",monitor.__class__.__name__)# 检查预警formonitorinself.monitors:alerts=monitor.check_alerts()foralertinalerts:self.alert_system.add_alert("WARNING",alert,monitor.__class__.__name__)defget_status(self):"""获取状态"""status={'timestamp':datetime.now(),'monitors':{},'alerts':self.alert_system.get_recent_alerts(5)}formonitorinself.monitors:status['monitors'][monitor.__class__.__name__]=monitor.get_status()returnstatus# 使用示例dashboard=MonitoringDashboard(api,[monitor,pos_monitor,strategy_monitor])whileTrue:dashboard.update()status=dashboard.get_status()print(f"监控状态:{status}")time.sleep(60)