量化策略的实时监控告警系统：从开发到生产的全链路保障

┌─────────────────────────────────────────────────────────┐
│                  业务监控层                              │
│  • 策略盈亏监控                                 │
│  • 因子暴露监控                                          │
│  • 交易执行监控                                          │
│  • 风险指标监控                                          │
└─────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────┐
│                  应用监控层                              │
│  • 策略运行状态                                          │
│  • 信号生成延迟                                          │
│  • 订单执行状态                                          │
│  • 数据源健康度                                          │
└─────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────┐
│                  系统监控层                              │
│  • CPU/内存/磁盘使用率                                   │
│  • 网络延迟和带宽                                        │
│  • 进程状态监控                                          │
│  • 日志监控                                              │
└─────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────┐
│                  基础设施监控层                            │
│  • 数据库性能监控                                        │
│  • 消息队列监控                                          │
│  • 云平台资源监控                                        │
│  • 网络设备故障监控                                      │
└─────────────────────────────────────────────────────────┘

import numpy as np
import pandas as pd
from datetime import datetime
import threading
import time

class RealTimePnLMonitor:
    """实时盈亏监控系统"""
    
    def __init__(self, strategy_id, update_interval=1.0):
        """
        初始化监控系统
        
        参数:
            strategy_id: 策略ID
            update_interval: 更新间隔（秒）
        """
        self.strategy_id = strategy_id
        self.update_interval = update_interval
        self.positions = {}  # 当前持仓 {symbol: quantity}
        self.costs = {}      # 持仓成本 {symbol: avg_cost}
        self.pnl_history = []  # 盈亏历史
        
        self.is_running = False
        self.monitor_thread = None
        
        # 告警阈值
        self.pnl_alert_threshold = 0.02  # 2%盈亏告警
        self.drawdown_alert_threshold = 0.05  # 5%回撤告警
        
    def start_monitoring(self):
        """启动监控线程"""
        self.is_running = True
        self.monitor_thread = threading.Thread(target=self._monitoring_loop)
        self.monitor_thread.daemon = True
        self.monitor_thread.start()
        print(f"[{datetime.now()}] 策略 {self.strategy_id} 监控系统已启动")
    
    def stop_monitoring(self):
        """停止监控"""
        self.is_running = False
        if self.monitor_thread:
            self.monitor_thread.join()
        print(f"[{datetime.now()}] 策略 {self.strategy_id} 监控系统已停止")
    
    def _monitoring_loop(self):
        """监控主循环"""
        while self.is_running:
            try:
                # 获取最新市场数据
                current_prices = self._get_current_prices()
                
                # 计算实时盈亏
                pnl_info = self._calculate_real_time_pnl(current_prices)
                
                # 检查告警条件
                self._check_alerts(pnl_info)
                
                # 记录历史
                self.pnl_history.append({
                    'timestamp': datetime.now(),
                    'total_pnl': pnl_info['total_pnl'],
                    'total_pnl_pct': pnl_info['total_pnl_pct'],
                    'positions_value': pnl_info['positions_value'],
                    'cash': pnl_info['cash']
                })
                
                # 输出监控信息（实际中应发送到监控系统）
                self._report_metrics(pnl_info)
                
            except Exception as e:
                print(f"[{datetime.now()}] 监控错误: {str(e)}")
            
            time.sleep(self.update_interval)
    
    def _get_current_prices(self):
        """获取当前价格（模拟）"""
        # 实际应用中应从行情源获取实时数据
        # 这里返回模拟数据
        return {
            '600519.SH': 1850.00 + np.random.randn() * 10,
            '000858.SZ': 125.50 + np.random.randn() * 2,
            '00700.HK': 320.00 + np.random.randn() * 5
        }
    
    def _calculate_real_time_pnl(self, current_prices):
        """计算实时盈亏"""
        total_market_value = 0.0
        total_cost = 0.0
        
        pnl_details = {}
        
        for symbol, quantity in self.positions.items():
            if symbol in current_prices:
                current_price = current_prices[symbol]
                cost = self.costs.get(symbol, 0.0)
                
                market_value = current_price * quantity
                position_cost = cost * quantity
                
                pnl = market_value - position_cost
                pnl_pct = (current_price - cost) / cost if cost > 0 else 0.0
                
                pnl_details[symbol] = {
                    'quantity': quantity,
                    'cost': cost,
                    'current_price': current_price,
                    'market_value': market_value,
                    'pnl': pnl,
                    'pnl_pct': pnl_pct
                }
                
                total_market_value += market_value
                total_cost += position_cost
        
        total_pnl = total_market_value - total_cost
        total_pnl_pct = total_pnl / total_cost if total_cost > 0 else 0.0
        
        return {
            'total_pnl': total_pnl,
            'total_pnl_pct': total_pnl_pct,
            'positions_value': total_market_value,
            'cash': 1000000 - total_cost,  # 假设初始资金100万
            'details': pnl_details
        }
    
    def _check_alerts(self, pnl_info):
        """检查告警条件"""
        alerts = []
        
        # 检查盈亏告警
        if abs(pnl_info['total_pnl_pct']) > self.pnl_alert_threshold:
            alert_type = "亏损告警" if pnl_info['total_pnl_pct'] < 0 else "盈利告警"
            alerts.append({
                'type': alert_type,
                'message': f"策略 {self.strategy_id} {alert_type}: "
                          f"{pnl_info['total_pnl_pct']*100:.2f}%",
                'severity': 'critical' if abs(pnl_info['total_pnl_pct']) > 0.05 else 'warning'
            })
        
        # 检查回撤告警
        if len(self.pnl_history) > 0:
            peak = max([h['total_pnl_pct'] for h in self.pnl_history])
            current = pnl_info['total_pnl_pct']
            drawdown = peak - current
            
            if drawdown > self.drawdown_alert_threshold:
                alerts.append({
                    'type': '回撤告警',
                    'message': f"策略 {self.strategy_id} 回撤超过阈值: "
                              f"{drawdown*100:.2f}%",
                    'severity': 'critical'
                })
        
        # 发送告警
        for alert in alerts:
            self._send_alert(alert)
    
    def _send_alert(self, alert):
        """发送告警（模拟）"""
        print(f"\n{'='*80}")
        print(f"🚨 告警通知 🚨")
        print(f"策略ID: {self.strategy_id}")
        print(f"告警类型: {alert['type']}")
        print(f"告警级别: {alert['severity']}")
        print(f"告警信息: {alert['message']}")
        print(f"时间: {datetime.now()}")
        print(f"{'='*80}\n")
        
        # 实际应用中，这里应调用告警通知接口
        # self.notify_wechat(alert)
        # self.notify_dingtalk(alert)
        # self.notify_email(alert)
    
    def _report_metrics(self, pnl_info):
        """上报监控指标（模拟）"""
        # 实际应用中，这里应上报到Prometheus/Grafana
        print(f"[{datetime.now().strftime('%H:%M:%S')}] "
              f"策略 {self.strategy_id} - "
              f"盈亏: {pnl_info['total_pnl_pct']*100:.2f}%, "
              f"市值: {pnl_info['positions_value']:.0f}, "
              f"现金: {pnl_info['cash']:.0f}")
    
    def update_position(self, symbol, quantity, price):
        """更新持仓（模拟交易）"""
        if symbol not in self.positions:
            self.positions[symbol] = 0
            self.costs[symbol] = 0.0
        
        current_qty = self.positions[symbol]
        current_cost = self.costs[symbol]
        
        if quantity > 0:  # 买入
            # 计算新的平均成本
            total_cost = current_cost * current_qty + price * quantity
            total_qty = current_qty + quantity
            self.costs[symbol] = total_cost / total_qty if total_qty > 0 else 0.0
        elif quantity < 0:  # 卖出
            # 卖出不改变剩余持仓的成本
            pass
        
        self.positions[symbol] = current_qty + quantity

# 使用示例
if __name__ == "__main__":
    # 创建监控系统
    monitor = RealTimePnLMonitor(strategy_id="momentum_strategy_v1")
    
    # 模拟初始持仓
    monitor.update_position('600519.SH', 1000, 1800.00)
    monitor.update_position('000858.SZ', 5000, 120.00)
    
    # 启动监控
    monitor.start_monitoring()
    
    # 运行一段时间
    try:
        time.sleep(10)
    except KeyboardInterrupt:
        pass
    
    # 停止监控
    monitor.stop_monitoring()

class FactorExposureMonitor:
    """因子暴露监控系统"""
    
    def __init__(self, strategy_id, target_exposures):
        """
        初始化因子暴露监控
        
        参数:
            strategy_id: 策略ID
            target_exposures: 目标因子暴露 {factor_name: target_value}
        """
        self.strategy_id = strategy_id
        self.target_exposures = target_exposures
        self.current_exposures = {}
        self.exposure_history = []
        
        # 告警阈值
        self.deviation_thresholds = {
            'market': 0.1,      # 市场因子偏离阈值
            'size': 0.15,        # 市值因子偏离阈值
            'value': 0.15,       # 价值因子偏离阈值
            'momentum': 0.2,     # 动量因子偏离阈值
            'volatility': 0.2     # 波动率因子偏离阈值
        }
    
    def update_exposures(self, holdings):
        """
        更新因子暴露
        
        参数:
            holdings: 当前持仓 {symbol: weight}
        """
        # 计算当前持仓的因子暴露
        self.current_exposures = self._calculate_factor_exposures(holdings)
        
        # 记录历史
        self.exposure_history.append({
            'timestamp': datetime.now(),
            'exposures': self.current_exposures.copy()
        })
        
        # 检查偏离
        self._check_exposure_deviation()
    
    def _calculate_factor_exposures(self, holdings):
        """计算因子暴露（简化示例）"""
        # 实际应用中应从因子数据库获取
        # 这里返回模拟数据
        
        exposures = {}
        
        for factor in self.target_exposures.keys():
            # 模拟计算因子暴露
            if factor == 'market':
                exposures[factor] = np.random.normal(0.8, 0.1)
            elif factor == 'size':
                exposures[factor] = np.random.normal(0.2, 0.15)
            elif factor == 'value':
                exposures[factor] = np.random.normal(0.3, 0.15)
            elif factor == 'momentum':
                exposures[factor] = np.random.normal(0.5, 0.2)
            elif factor == 'volatility':
                exposures[factor] = np.random.normal(-0.1, 0.2)
        
        return exposures
    
    def _check_exposure_deviation(self):
        """检查因子暴露偏离"""
        alerts = []
        
        for factor, target in self.target_exposures.items():
            current = self.current_exposures.get(factor, 0.0)
            deviation = abs(current - target)
            
            threshold = self.deviation_thresholds.get(factor, 0.2)
            
            if deviation > threshold:
                alerts.append({
                    'type': '因子暴露偏离',
                    'factor': factor,
                    'target': target,
                    'current': current,
                    'deviation': deviation,
                    'severity': 'critical' if deviation > threshold * 1.5 else 'warning'
                })
        
        # 发送告警
        for alert in alerts:
            self._send_exposure_alert(alert)
    
    def _send_exposure_alert(self, alert):
        """发送因子暴露告警"""
        print(f"\n{'='*80}")
        print(f"⚠️  因子暴露告警 ⚠️")
        print(f"策略ID: {self.strategy_id}")
        print(f"因子: {alert['factor']}")
        print(f"目标暴露: {alert['target']:.4f}")
        print(f"当前暴露: {alert['current']:.4f}")
        print(f"偏离度: {alert['deviation']:.4f}")
        print(f"告警级别: {alert['severity']}")
        print(f"时间: {datetime.now()}")
        print(f"{'='*80}\n")
    
    def plot_exposure_history(self):
        """绘制因子暴露历史（简化）"""
        if len(self.exposure_history) < 2:
            print("历史数据不足，无法绘图")
            return
        
        # 实际应用中应使用matplotlib/plotly绘图
        print(f"\n策略 {self.strategy_id} 因子暴露历史：")
        print("-" * 80)
        
        for record in self.exposure_history[-10:]:  # 显示最近10条
            timestamp = record['timestamp'].strftime('%H:%M:%S')
            exposures = record['exposures']
            
            print(f"[{timestamp}] ", end="")
            for factor, value in exposures.items():
                print(f"{factor}: {value:.3f} ", end="")
            print()

class OrderExecutionMonitor:
    """订单执行监控系统"""
    
    def __init__(self, strategy_id):
        self.strategy_id = strategy_id
        self.pending_orders = {}  # 待处理订单
        self.executed_orders = {}  # 已执行订单
        self.rejected_orders = {}  # 被拒绝订单
        
        # 监控指标
        self.metrics = {
            'submission_latency': [],      # 提交延迟
            'execution_latency': [],        # 执行延迟
            'fill_rate': [],                # 成交率
            'slippage': []                 # 滑点
        }
        
        # 告警阈值
        self.latency_threshold = 1.0  # 1秒延迟告警
        self.slippage_threshold = 0.01  # 1%滑点告警
    
    def monitor_order_submission(self, order_id, order_details):
        """监控订单提交"""
        start_time = time.time()
        
        # 记录订单信息
        self.pending_orders[order_id] = {
            'details': order_details,
            'submit_time': start_time,
            'status': 'PENDING'
        }
        
        # 模拟订单提交
        try:
            # 实际应用中应调用交易接口
            execution_time = time.time()
            latency = execution_time - start_time
            
            self.metrics['submission_latency'].append(latency)
            
            # 检查延迟告警
            if latency > self.latency_threshold:
                self._send_latency_alert(order_id, latency, 'submission')
            
            # 更新订单状态
            self.pending_orders[order_id]['status'] = 'SUBMITTED'
            self.pending_orders[order_id]['submit_latency'] = latency
            
        except Exception as e:
            self._handle_order_error(order_id, str(e))
    
    def monitor_order_execution(self, order_id, execution_details):
        """监控订单执行"""
        if order_id not in self.pending_orders:
            print(f"警告：订单 {order_id} 不在待处理列表中")
            return
        
        start_time = self.pending_orders[order_id]['submit_time']
        execution_time = time.time()
        
        # 计算执行延迟
        execution_latency = execution_time - start_time
        self.metrics['execution_latency'].append(execution_latency)
        
        # 计算滑点
        ordered_price = self.pending_orders[order_id]['details']['price']
        executed_price = execution_details['price']
        slippage = abs(executed_price - ordered_price) / ordered_price
        
        self.metrics['slippage'].append(slippage)
        
        # 更新订单状态
        self.pending_orders[order_id]['status'] = 'EXECUTED'
        self.pending_orders[order_id]['execution_time'] = execution_time
        self.pending_orders[order_id]['execution_latency'] = execution_latency
        self.pending_orders[order_id]['executed_price'] = executed_price
        self.pending_orders[order_id]['slippage'] = slippage
        
        # 移动到已执行订单
        self.executed_orders[order_id] = self.pending_orders.pop(order_id)
        
        # 检查告警
        if execution_latency > self.latency_threshold:
            self._send_latency_alert(order_id, execution_latency, 'execution')
        
        if slippage > self.slippage_threshold:
            self._send_slippage_alert(order_id, slippage)
    
    def _send_latency_alert(self, order_id, latency, latency_type):
        """发送延迟告警"""
        print(f"\n🚨 订单延迟告警 🚨")
        print(f"订单ID: {order_id}")
        print(f"延迟类型: {latency_type}")
        print(f"延迟时间: {latency:.4f}秒")
        print(f"阈值: {self.latency_threshold}秒")
        print(f"时间: {datetime.now()}")
    
    def _send_slippage_alert(self, order_id, slippage):
        """发送滑点告警"""
        print(f"\n🚨 订单滑点告警 🚨")
        print(f"订单ID: {order_id}")
        print(f"滑点: {slippage*100:.2f}%")
        print(f"阈值: {self.slippage_threshold*100:.2f}%")
        print(f"时间: {datetime.now()}")
    
    def _handle_order_error(self, order_id, error_msg):
        """处理订单错误"""
        self.pending_orders[order_id]['status'] = 'ERROR'
        self.pending_orders[order_id]['error'] = error_msg
        
        # 移动到拒绝订单
        self.rejected_orders[order_id] = self.pending_orders.pop(order_id)
        
        # 发送告警
        print(f"\n❌ 订单错误 ❌")
        print(f"订单ID: {order_id}")
        print(f"错误信息: {error_msg}")
        print(f"时间: {datetime.now()}")
    
    def get_execution_statistics(self):
        """获取执行统计"""
        stats = {}
        
        if self.metrics['submission_latency']:
            stats['avg_submission_latency'] = np.mean(self.metrics['submission_latency'])
            stats['max_submission_latency'] = np.max(self.metrics['submission_latency'])
        
        if self.metrics['execution_latency']:
            stats['avg_execution_latency'] = np.mean(self.metrics['execution_latency'])
            stats['max_execution_latency'] = np.max(self.metrics['execution_latency'])
        
        if self.metrics['slippage']:
            stats['avg_slippage'] = np.mean(self.metrics['slippage'])
            stats['max_slippage'] = np.max(self.metrics['slippage'])
        
        stats['total_orders'] = len(self.executed_orders) + len(self.rejected_orders)
        stats['executed_orders'] = len(self.executed_orders)
        stats['rejected_orders'] = len(self.rejected_orders)
        stats['fill_rate'] = stats['executed_orders'] / stats['total_orders'] if stats['total_orders'] > 0 else 0
        
        return stats

import psutil
import shutil

class SystemHealthMonitor:
    """系统健康度监控"""
    
    def __init__(self, strategy_id):
        self.strategy_id = strategy_id
        self.monitoring_interval = 5  # 5秒检查一次
        self.is_running = False
        
        # 告警阈值
        self.cpu_threshold = 80  # CPU使用率阈值（%）
        self.memory_threshold = 85  # 内存使用率阈值（%）
        self.disk_threshold = 90  # 磁盘使用率阈值（%）
        
        # 监控历史
        self.resource_history = []
    
    def start_monitoring(self):
        """启动系统监控"""
        self.is_running = True
        
        def monitoring_loop():
            while self.is_running:
                try:
                    # 收集系统指标
                    metrics = self._collect_system_metrics()
                    
                    # 记录历史
                    self.resource_history.append({
                        'timestamp': datetime.now(),
                        'metrics': metrics
                    })
                    
                    # 检查告警条件
                    self._check_system_alerts(metrics)
                    
                    # 上报指标
                    self._report_system_metrics(metrics)
                    
                except Exception as e:
                    print(f"系统监控错误: {str(e)}")
                
                time.sleep(self.monitoring_interval)
        
        monitor_thread = threading.Thread(target=monitoring_loop)
        monitor_thread.daemon = True
        monitor_thread.start()
        
        print(f"[{datetime.now()}] 系统监控已启动")
    
    def _collect_system_metrics(self):
        """收集系统指标"""
        metrics = {}
        
        # CPU使用率
        metrics['cpu_percent'] = psutil.cpu_percent(interval=1)
        
        # 内存使用率
        memory = psutil.virtual_memory()
        metrics['memory_percent'] = memory.percent
        metrics['memory_used_gb'] = memory.used / (1024**3)
        metrics['memory_total_gb'] = memory.total / (1024**3)
        
        # 磁盘使用率
        disk = psutil.disk_usage('/')
        metrics['disk_percent'] = disk.percent
        metrics['disk_used_gb'] = disk.used / (1024**3)
        metrics['disk_total_gb'] = disk.total / (1024**3)
        
        # 网络I/O
        net_io = psutil.net_io_counters()
        metrics['bytes_sent'] = net_io.bytes_sent
        metrics['bytes_recv'] = net_io.bytes_recv
        
        # 进程数
        metrics['process_count'] = len(psutil.pids())
        
        return metrics
    
    def _check_system_alerts(self, metrics):
        """检查系统告警"""
        alerts = []
        
        # CPU告警
        if metrics['cpu_percent'] > self.cpu_threshold:
            alerts.append({
                'type': 'CPU使用率过高',
                'value': metrics['cpu_percent'],
                'threshold': self.cpu_threshold,
                'severity': 'critical' if metrics['cpu_percent'] > 95 else 'warning'
            })
        
        # 内存告警
        if metrics['memory_percent'] > self.memory_threshold:
            alerts.append({
                'type': '内存使用率过高',
                'value': metrics['memory_percent'],
                'threshold': self.memory_threshold,
                'severity': 'critical' if metrics['memory_percent'] > 95 else 'warning'
            })
        
        # 磁盘告警
        if metrics['disk_percent'] > self.disk_threshold:
            alerts.append({
                'type': '磁盘使用率过高',
                'value': metrics['disk_percent'],
                'threshold': self.disk_threshold,
                'severity': 'critical' if metrics['disk_percent'] > 98 else 'warning'
            })
        
        # 发送告警
        for alert in alerts:
            self._send_system_alert(alert)
    
    def _send_system_alert(self, alert):
        """发送系统告警"""
        print(f"\n⚠️  系统资源告警 ⚠️")
        print(f"策略ID: {self.strategy_id}")
        print(f"告警类型: {alert['type']}")
        print(f"当前值: {alert['value']:.1f}%")
        print(f"告警阈值: {alert['threshold']}%")
        print(f"告警级别: {alert['severity']}")
        print(f"时间: {datetime.now()}")
        print(f"建议操作: 立即检查系统资源使用情况，必要时扩容或优化代码")
    
    def _report_system_metrics(self, metrics):
        """上报系统指标"""
        # 简化输出，实际应上报到监控系统
        if metrics['cpu_percent'] > 70 or metrics['memory_percent'] > 70:
            print(f"[{datetime.now().strftime('%H:%M:%S')}] "
                  f"CPU: {metrics['cpu_percent']:.1f}%, "
                  f"内存: {metrics['memory_percent']:.1f}%, "
                  f"磁盘: {metrics['disk_percent']:.1f}%")

import requests
import json

class MultiChannelAlerter:
    """多通道告警通知系统"""
    
    def __init__(self, strategy_id):
        self.strategy_id = strategy_id
        
        # 告警通道配置
        self.channels = {
            'wechat': {
                'enabled': True,
                'webhook_url': 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY'
            },
            'dingtalk': {
                'enabled': True,
                'webhook_url': 'https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN'
            },
            'email': {
                'enabled': True,
                'smtp_server': 'smtp.example.com',
                'sender': 'alert@example.com',
                'password': 'password',
                'receivers': ['admin@example.com']
            },
            'slack': {
                'enabled': False,
                'webhook_url': 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL'
            }
        }
        
        # 告警级别与通道映射
        self.severity_channels = {
            'critical': ['wechat', 'dingtalk', 'email'],
            'warning': ['wechat', 'dingtalk'],
            'info': ['wechat']
        }
    
    def send_alert(self, alert_info):
        """
        发送告警
        
        参数:
            alert_info: 告警信息字典
                {
                    'type': '盈亏告警',
                    'severity': 'critical',
                    'message': '具体告警信息',
                    'timestamp': datetime.now()
                }
        """
        severity = alert_info.get('severity', 'warning')
        
        # 根据告警级别选择通知通道
        channels_to_notify = self.severity_channels.get(severity, ['wechat'])
        
        # 并行发送通知
        import concurrent.futures
        
        with concurrent.futures.ThreadPoolExecutor() as executor:
            futures = []
            
            for channel in channels_to_notify:
                if self.channels[channel]['enabled']:
                    future = executor.submit(
                        self._send_to_channel,
                        channel,
                        alert_info
                    )
                    futures.append(future)
            
            # 等待所有通知发送完成
            concurrent.futures.wait(futures)
    
    def _send_to_channel(self, channel, alert_info):
        """发送告警到指定通道"""
        try:
            if channel == 'wechat':
                self._send_wechat(alert_info)
            elif channel == 'dingtalk':
                self._send_dingtalk(alert_info)
            elif channel == 'email':
                self._send_email(alert_info)
            elif channel == 'slack':
                self._send_slack(alert_info)
            
            print(f"✅ 告警已发送到 {channel}: {alert_info['type']}")
            
        except Exception as e:
            print(f"❌ 发送告警到 {channel} 失败: {str(e)}")
    
    def _send_wechat(self, alert_info):
        """发送企业微信通知"""
        webhook_url = self.channels['wechat']['webhook_url']
        
        # 构建消息内容
        message = {
            "msgtype": "markdown",
            "markdown": {
                "content": f"""### 🚨 量化策略告警
                
**策略ID**: {self.strategy_id}
**告警类型**: {alert_info['type']}
**告警级别**: {alert_info['severity']}
**告警信息**: {alert_info['message']}
**时间**: {alert_info['timestamp'].strftime('%Y-%m-%d %H:%M:%S')}
                
> 请及时检查策略运行状态
                """
            }
        }
        
        # 发送请求
        response = requests.post(
            webhook_url,
            headers={'Content-Type': 'application/json'},
            data=json.dumps(message),
            timeout=5
        )
        
        if response.status_code != 200:
            raise Exception(f"企业微信通知失败: {response.text}")
    
    def _send_dingtalk(self, alert_info):
        """发送钉钉通知"""
        webhook_url = self.channels['dingtalk']['webhook_url']
        
        # 构建消息内容
        message = {
            "msgtype": "markdown",
            "markdown": {
                "title": "量化策略告警",
                "text": f"""### 🚨 量化策略告警
                
**策略ID**: {self.strategy_id}
**告警类型**: {alert_info['type']}
**告警级别**: {alert_info['severity']}
**告警信息**: {alert_info['message']}
**时间**: {alert_info['timestamp'].strftime('%Y-%m-%d %H:%M:%S')}
                
> 请及时检查策略运行状态
                """
            }
        }
        
        # 发送请求
        response = requests.post(
            webhook_url,
            headers={'Content-Type': 'application/json'},
            data=json.dumps(message),
            timeout=5
        )
        
        if response.status_code != 200:
            raise Exception(f"钉钉通知失败: {response.text}")
    
    def _send_email(self, alert_info):
        """发送邮件通知"""
        # 简化示例，实际应用应使用smtplib
        print(f"📧 邮件通知: {alert_info['type']} - {alert_info['message']}")
    
    def _send_slack(self, alert_info):
        """发送Slack通知"""
        webhook_url = self.channels['slack']['webhook_url']
        
        # 构建消息内容
        message = {
            "text": f"""🚨 量化策略告警
            
策略ID: {self.strategy_id}
告警类型: {alert_info['type']}
告警级别: {alert_info['severity']}
告警信息: {alert_info['message']}
时间: {alert_info['timestamp'].strftime('%Y-%m-%d %H:%M:%S')}
            """
        }
        
        # 发送请求
        response = requests.post(
            webhook_url,
            headers={'Content-Type': 'application/json'},
            data=json.dumps(message),
            timeout=5
        )
        
        if response.status_code != 200:
            raise Exception(f"Slack通知失败: {response.text}")

from prometheus_client import Counter, Gauge, Histogram, start_http_server
import time

class PrometheusMetricsExporter:
    """Prometheus指标导出器"""
    
    def __init__(self, strategy_id, port=8000):
        self.strategy_id = strategy_id
        
        # 定义Prometheus指标
        self.pnl_gauge = Gauge(
            'strategy_pnl_percent',
            'Strategy PnL percentage',
            ['strategy_id']
        )
        
        self.position_gauge = Gauge(
            'strategy_position_value',
            'Strategy position value',
            ['strategy_id', 'symbol']
        )
        
        self.order_counter = Counter(
            'strategy_orders_total',
            'Total number of orders',
            ['strategy_id', 'status']
        )
        
        self.execution_latency_histogram = Histogram(
            'strategy_execution_latency_seconds',
            'Order execution latency in seconds',
            ['strategy_id']
        )
        
        self.factor_exposure_gauge = Gauge(
            'strategy_factor_exposure',
            'Strategy factor exposure',
            ['strategy_id', 'factor_name']
        )
        
        # 启动Prometheus HTTP服务器
        start_http_server(port)
        print(f"Prometheus指标服务器已启动，端口: {port}")
    
    def update_pnl(self, pnl_percent):
        """更新盈亏指标"""
        self.pnl_gauge.labels(strategy_id=self.strategy_id).set(pnl_percent)
    
    def update_position(self, symbol, value):
        """更新持仓指标"""
        self.position_gauge.labels(
            strategy_id=self.strategy_id,
            symbol=symbol
        ).set(value)
    
    def record_order(self, status):
        """记录订单"""
        self.order_counter.labels(
            strategy_id=self.strategy_id,
            status=status
        ).inc()
    
    def record_execution_latency(self, latency):
        """记录执行延迟"""
        self.execution_latency_histogram.labels(
            strategy_id=self.strategy_id
        ).observe(latency)
    
    def update_factor_exposure(self, factor_name, exposure):
        """更新因子暴露"""
        self.factor_exposure_gauge.labels(
            strategy_id=self.strategy_id,
            factor_name=factor_name
        ).set(exposure)

# 集成示例
if __name__ == "__main__":
    # 创建指标导出器
    exporter = PrometheusMetricsExporter(strategy_id="momentum_v1", port=8000)
    
    # 模拟更新指标
    exporter.update_pnl(2.5)
    exporter.update_position('600519.SH', 1850000)
    exporter.record_order('executed')
    exporter.record_execution_latency(0.35)
    exporter.update_factor_exposure('market', 0.85)
    
    print("指标已更新，可在 http://localhost:8000/metrics 查看")
    
    # 保持程序运行
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("程序已停止")

class CompleteMonitoringSystem:
    """完整的量化策略监控系统"""
    
    def __init__(self, strategy_id):
        self.strategy_id = strategy_id
        
        # 初始化各个监控模块
        self.pnl_monitor = RealTimePnLMonitor(strategy_id)
        self.factor_monitor = FactorExposureMonitor(
            strategy_id,
            target_exposures={
                'market': 0.8,
                'size': 0.2,
                'value': 0.3,
                'momentum': 0.5
            }
        )
        self.execution_monitor = OrderExecutionMonitor(strategy_id)
        self.system_monitor = SystemHealthMonitor(strategy_id)
        self.alerter = MultiChannelAlerter(strategy_id)
        
        # 初始化Prometheus指标导出器
        self.metrics_exporter = PrometheusMetricsExporter(strategy_id, port=8000)
        
        print(f"策略 {strategy_id} 的完整监控系统已初始化")
    
    def start_all_monitors(self):
        """启动所有监控"""
        print(f"\n{'='*80}")
        print(f"启动策略 {self.strategy_id} 的所有监控模块")
        print(f"{'='*80}\n")
        
        # 启动盈亏监控
        self.pnl_monitor.start_monitoring()
        
        # 启动系统监控
        self.system_monitor.start_monitoring()
        
        print(f"\n✅ 所有监控模块已启动")
        print(f"📊 Grafana仪表盘: http://localhost:3000")
        print(f"📈 Prometheus指标: http://localhost:8000/metrics\n")
    
    def stop_all_monitors(self):
        """停止所有监控"""
        print(f"\n{'='*80}")
        print(f"停止策略 {self.strategy_id} 的所有监控模块")
        print(f"{'='*80}\n")
        
        self.pnl_monitor.stop_monitoring()
        
        print(f"\n✅ 所有监控模块已停止")
    
    def integrated_alert_handler(self, alert):
        """集成告警处理"""
        # 1. 记录告警日志
        self._log_alert(alert)
        
        # 2. 更新监控指标
        self._update_metrics_from_alert(alert)
        
        # 3. 发送告警通知
        self.alerter.send_alert(alert)
        
        # 4. 触发自动响应（可选）
        self._trigger_auto_response(alert)
    
    def _log_alert(self, alert):
        """记录告警日志"""
        log_file = f"logs/{self.strategy_id}_alerts.log"
        os.makedirs('logs', exist_ok=True)
        
        with open(log_file, 'a') as f:
            log_entry = {
                'timestamp': alert['timestamp'].isoformat(),
                'strategy_id': self.strategy_id,
                'alert_type': alert['type'],
                'severity': alert['severity'],
                'message': alert['message']
            }
            f.write(json.dumps(log_entry) + '\n')
    
    def _update_metrics_from_alert(self, alert):
        """根据告警更新指标"""
        if alert['type'] == '盈亏告警':
            # 提取盈亏百分比
            pnl_pct = self._extract_pnl_from_message(alert['message'])
            self.metrics_exporter.update_pnl(pnl_pct)
    
    def _trigger_auto_response(self, alert):
        """触发自动响应"""
        # 例如：严重亏损时自动停止策略
        if alert['type'] == '盈亏告警' and alert['severity'] == 'critical':
            if '亏损' in alert['message'] and '5%' in alert['message']:
                print(f"\n⚠️  自动响应触发 ⚠️")
                print(f"策略 {self.strategy_id} 亏损超过5%，自动停止交易")
                # 实际应用中应调用策略控制接口
                # self.stop_trading()
    
    def _extract_pnl_from_message(self, message):
        """从告警信息中提取盈亏百分比"""
        # 简化实现
        import re
        match = re.search(r'([-+]?\d+\.\d+)%', message)
        if match:
            return float(match.group(1))
        return 0.0
    
    def generate_monitoring_report(self):
        """生成监控报告"""
        report = {
            'strategy_id': self.strategy_id,
            'report_time': datetime.now().isoformat(),
            'pnl_summary': self._get_pnl_summary(),
            'execution_summary': self.execution_monitor.get_execution_statistics(),
            'system_health': self._get_system_health_summary(),
            'alert_count': self._get_alert_count()
        }
        
        # 保存报告
        report_file = f"reports/{self.strategy_id}_monitoring_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        os.makedirs('reports', exist_ok=True)
        
        with open(report_file, 'w') as f:
            json.dump(report, f, indent=2)
        
        print(f"\n📊 监控报告已生成: {report_file}")
        
        return report
    
    def _get_pnl_summary(self):
        """获取盈亏摘要"""
        if len(self.pnl_monitor.pnl_history) == 0:
            return {}
        
        pnl_history = self.pnl_monitor.pnl_history
        
        return {
            'current_pnl_pct': pnl_history[-1]['total_pnl_pct'],
            'max_pnl_pct': max([h['total_pnl_pct'] for h in pnl_history]),
            'min_pnl_pct': min([h['total_pnl_pct'] for h in pnl_history]),
            'avg_pnl_pct': np.mean([h['total_pnl_pct'] for h in pnl_history])
        }
    
    def _get_system_health_summary(self):
        """获取系统健康摘要"""
        if len(self.system_monitor.resource_history) == 0:
            return {}
        
        recent_metrics = [h['metrics'] for h in self.system_monitor.resource_history[-10:]]
        
        return {
            'avg_cpu_percent': np.mean([m['cpu_percent'] for m in recent_metrics]),
            'avg_memory_percent': np.mean([m['memory_percent'] for m in recent_metrics]),
            'avg_disk_percent': np.mean([m['disk_percent'] for m in recent_metrics])
        }
    
    def _get_alert_count(self):
        """获取告警计数"""
        # 简化实现，实际应从日志或数据库统计
        return {
            'critical': 0,
            'warning': 0,
            'info': 0
        }

# 完整系统使用示例
if __name__ == "__main__":
    # 创建完整监控系统
    monitoring_system = CompleteMonitoringSystem(strategy_id="momentum_strategy_v1")
    
    # 启动所有监控
    monitoring_system.start_all_monitors()
    
    try:
        # 运行一段时间
        time.sleep(60)
        
        # 生成监控报告
        monitoring_system.generate_monitoring_report()
        
    except KeyboardInterrupt:
        print("\n接收到中断信号")
    
    finally:
        # 停止所有监控
        monitoring_system.stop_all_monitors()