From 2dd10c6ffb9e62df7d6246c7d095698467a66c84 Mon Sep 17 00:00:00 2001 From: dethan3 Date: Thu, 12 Mar 2026 10:04:25 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E6=B6=A8?= =?UTF-8?q?=E8=B7=8C=E9=98=88=E5=80=BC=E5=91=8A=E8=AD=A6=E9=80=9A=E7=9F=A5?= =?UTF-8?q?=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增告警管理器模块 (alert_manager.py) - 支持邮件、飞书、企业微信、钉钉、Telegram多种通知渠道 - 添加告警规则配置系统 - 集成到主程序，自动检查告警 - 添加独立的告警检查脚本 - 支持冷却时间和每日告警限制 - 保存告警历史记录，避免重复通知 Closes #8 --- alert_manager.py | 698 +++++++++++++++++++++++++++++++++++++ alerts_config.example.json | 106 ++++++ alerts_config.json | 66 ++++ check_alerts.py | 221 ++++++++++++ main.py | 14 + 5 files changed, 1105 insertions(+) create mode 100644 alert_manager.py create mode 100644 alerts_config.example.json create mode 100644 alerts_config.json create mode 100644 check_alerts.py diff --git a/alert_manager.py b/alert_manager.py new file mode 100644 index 0000000..c86fe0a --- /dev/null +++ b/alert_manager.py @@ -0,0 +1,698 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +告警管理器 - 处理基金/指数的涨跌阈值告警 +Author: TickEye Team +Date: 2026-03-12 +""" + +import os +import json +import logging +import time +import smtplib +from email.mime.text import MIMEText +from email.mime.multipart import MIMEMultipart +from datetime import datetime, timedelta +from typing import Dict, List, Optional, Any, Tuple +import requests +from pathlib import Path + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + + +class AlertRule: + """告警规则类""" + + def __init__(self, rule_config: Dict[str, Any]): + """ + 初始化告警规则 + + Args: + rule_config: 规则配置字典 + """ + self.code = rule_config.get('code', '') + self.name = rule_config.get('name', '') + self.threshold = float(rule_config.get('threshold', 0.0)) + self.direction = rule_config.get('direction', 'up') # 'up' or 'down' + self.enabled = rule_config.get('enabled', True) + self.notification_channels = rule_config.get('notification_channels', []) + + # 验证配置 + if not self.code: + raise ValueError("告警规则必须包含代码(code)") + if self.direction not in ['up', 'down']: + raise ValueError("告警规则方向(direction)必须是'up'或'down'") + + def should_alert(self, change_pct: float) -> bool: + """ + 检查是否应该触发告警 + + Args: + change_pct: 涨跌幅百分比 + + Returns: + bool: 是否触发告警 + """ + if not self.enabled: + return False + + if self.direction == 'up': + return change_pct >= self.threshold + else: # down + return change_pct <= self.threshold + + def get_alert_message(self, change_pct: float, current_value: float) -> Dict[str, Any]: + """ + 获取告警消息 + + Args: + change_pct: 涨跌幅百分比 + current_value: 当前值 + + Returns: + Dict: 告警消息 + """ + direction_text = "上涨" if self.direction == 'up' else "下跌" + threshold_text = f"{abs(self.threshold)}%" + + return { + "code": self.code, + "name": self.name or self.code, + "current_value": current_value, + "change_pct": change_pct, + "threshold": self.threshold, + "direction": self.direction, + "direction_text": direction_text, + "threshold_text": threshold_text, + "timestamp": datetime.now().isoformat(), + "message": f"{self.name or self.code} {direction_text}超过{threshold_text}，当前涨跌幅: {change_pct:.2f}%" + } + + +class NotificationChannel: + """通知渠道基类""" + + def __init__(self, config: Dict[str, Any]): + self.config = config + self.enabled = config.get('enabled', False) + + def send(self, alert_message: Dict[str, Any]) -> bool: + """发送通知""" + raise NotImplementedError("子类必须实现send方法") + + +class EmailNotification(NotificationChannel): + """邮件通知""" + + def __init__(self, config: Dict[str, Any]): + super().__init__(config) + self.smtp_server = config.get('smtp_server', '') + self.smtp_port = config.get('smtp_port', 587) + self.smtp_username = config.get('smtp_username', '') + self.smtp_password = config.get('smtp_password', '') + self.sender_email = config.get('sender_email', '') + self.receiver_emails = config.get('receiver_emails', []) + + def send(self, alert_message: Dict[str, Any]) -> bool: + if not self.enabled: + return False + + try: + # 创建邮件内容 + subject = f"📈 基金告警: {alert_message['name']} {alert_message['direction_text']}超过{alert_message['threshold_text']}" + + body = f""" + + +

📈 基金告警通知

+

标的: {alert_message['name']} ({alert_message['code']})

+

当前值: {alert_message['current_value']}

+

涨跌幅: {alert_message['change_pct']:.2f}%

+

告警阈值: {alert_message['direction_text']} {alert_message['threshold']}%

+

触发时间: {alert_message['timestamp']}

+
+

来自 TickEye 基金监测系统

+ + + """ + + # 创建邮件 + msg = MIMEMultipart() + msg['From'] = self.sender_email + msg['To'] = ', '.join(self.receiver_emails) + msg['Subject'] = subject + msg.attach(MIMEText(body, 'html')) + + # 发送邮件 + with smtplib.SMTP(self.smtp_server, self.smtp_port) as server: + server.starttls() + server.login(self.smtp_username, self.smtp_password) + server.send_message(msg) + + logger.info(f"邮件告警发送成功: {alert_message['code']}") + return True + + except Exception as e: + logger.error(f"邮件告警发送失败: {e}") + return False + + +class FeishuNotification(NotificationChannel): + """飞书通知""" + + def __init__(self, config: Dict[str, Any]): + super().__init__(config) + self.webhook_url = config.get('webhook_url', '') + self.template = config.get('template', 'default') + + def send(self, alert_message: Dict[str, Any]) -> bool: + if not self.enabled or not self.webhook_url: + return False + + try: + # 创建飞书消息卡片 + card = { + "msg_type": "interactive", + "card": { + "config": { + "wide_screen_mode": True + }, + "header": { + "title": { + "tag": "plain_text", + "content": f"📈 基金告警: {alert_message['name']}" + }, + "template": "red" if alert_message['direction'] == 'down' else "green" + }, + "elements": [ + { + "tag": "div", + "text": { + "tag": "lark_md", + "content": f"**标的:** {alert_message['name']} ({alert_message['code']})\n" + f"**当前值:** {alert_message['current_value']}\n" + f"**涨跌幅:** {alert_message['change_pct']:.2f}%\n" + f"**告警阈值:** {alert_message['direction_text']} {alert_message['threshold']}%\n" + f"**触发时间:** {alert_message['timestamp']}" + } + }, + { + "tag": "hr" + }, + { + "tag": "note", + "elements": [ + { + "tag": "plain_text", + "content": "来自 TickEye 基金监测系统" + } + ] + } + ] + } + } + + # 发送请求 + response = requests.post( + self.webhook_url, + json=card, + headers={'Content-Type': 'application/json'}, + timeout=10 + ) + + if response.status_code == 200: + logger.info(f"飞书告警发送成功: {alert_message['code']}") + return True + else: + logger.error(f"飞书告警发送失败: {response.status_code} - {response.text}") + return False + + except Exception as e: + logger.error(f"飞书告警发送失败: {e}") + return False + + +class WechatWorkNotification(NotificationChannel): + """企业微信通知""" + + def __init__(self, config: Dict[str, Any]): + super().__init__(config) + self.webhook_url = config.get('webhook_url', '') + self.key = config.get('key', '') + + def send(self, alert_message: Dict[str, Any]) -> bool: + if not self.enabled or not self.webhook_url: + return False + + try: + # 创建企业微信消息 + message = { + "msgtype": "markdown", + "markdown": { + "content": f"## 📈 基金告警通知\n" + f"> **标的:** {alert_message['name']} ({alert_message['code']})\n" + f"> **当前值:** {alert_message['current_value']}\n" + f"> **涨跌幅:** {alert_message['change_pct']:.2f}%\n" + f"> **告警阈值:** {alert_message['direction_text']} {alert_message['threshold']}%\n" + f"> **触发时间:** {alert_message['timestamp']}\n" + f"> \n" + f"> 来自 TickEye 基金监测系统" + } + } + + # 发送请求 + response = requests.post( + self.webhook_url, + json=message, + headers={'Content-Type': 'application/json'}, + timeout=10 + ) + + if response.status_code == 200: + logger.info(f"企业微信告警发送成功: {alert_message['code']}") + return True + else: + logger.error(f"企业微信告警发送失败: {response.status_code} - {response.text}") + return False + + except Exception as e: + logger.error(f"企业微信告警发送失败: {e}") + return False + + +class DingTalkNotification(NotificationChannel): + """钉钉通知""" + + def __init__(self, config: Dict[str, Any]): + super().__init__(config) + self.webhook_url = config.get('webhook_url', '') + self.secret = config.get('secret', '') + + def send(self, alert_message: Dict[str, Any]) -> bool: + if not self.enabled or not self.webhook_url: + return False + + try: + import hmac + import hashlib + import base64 + import urllib.parse + + # 如果有secret，计算签名 + timestamp = str(round(time.time() * 1000)) + sign = '' + + if self.secret: + secret_enc = self.secret.encode('utf-8') + string_to_sign = f"{timestamp}\n{self.secret}" + string_to_sign_enc = string_to_sign.encode('utf-8') + hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest() + sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) + webhook_url = f"{self.webhook_url}×tamp={timestamp}&sign={sign}" + else: + webhook_url = self.webhook_url + + # 创建钉钉消息 + message = { + "msgtype": "markdown", + "markdown": { + "title": f"📈 基金告警: {alert_message['name']}", + "text": f"### 📈 基金告警通知\n" + f"**标的:** {alert_message['name']} ({alert_message['code']})\n\n" + f"**当前值:** {alert_message['current_value']}\n\n" + f"**涨跌幅:** {alert_message['change_pct']:.2f}%\n\n" + f"**告警阈值:** {alert_message['direction_text']} {alert_message['threshold']}%\n\n" + f"**触发时间:** {alert_message['timestamp']}\n\n" + f"---\n\n" + f"来自 TickEye 基金监测系统" + } + } + + # 发送请求 + response = requests.post( + webhook_url, + json=message, + headers={'Content-Type': 'application/json'}, + timeout=10 + ) + + if response.status_code == 200: + logger.info(f"钉钉告警发送成功: {alert_message['code']}") + return True + else: + logger.error(f"钉钉告警发送失败: {response.status_code} - {response.text}") + return False + + except Exception as e: + logger.error(f"钉钉告警发送失败: {e}") + return False + + +class TelegramNotification(NotificationChannel): + """Telegram通知""" + + def __init__(self, config: Dict[str, Any]): + super().__init__(config) + self.bot_token = config.get('bot_token', '') + self.chat_id = config.get('chat_id', '') + + def send(self, alert_message: Dict[str, Any]) -> bool: + if not self.enabled or not self.bot_token or not self.chat_id: + return False + + try: + # 创建Telegram消息 + text = f"📈 *基金告警通知*\n\n" + text += f"*标的:* {alert_message['name']} ({alert_message['code']})\n" + text += f"*当前值:* {alert_message['current_value']}\n" + text += f"*涨跌幅:* {alert_message['change_pct']:.2f}%\n" + text += f"*告警阈值:* {alert_message['direction_text']} {alert_message['threshold']}%\n" + text += f"*触发时间:* {alert_message['timestamp']}\n\n" + text += f"来自 TickEye 基金监测系统" + + # 发送请求 + url = f"https://api.telegram.org/bot{self.bot_token}/sendMessage" + payload = { + "chat_id": self.chat_id, + "text": text, + "parse_mode": "Markdown" + } + + response = requests.post(url, json=payload, timeout=10) + + if response.status_code == 200: + logger.info(f"Telegram告警发送成功: {alert_message['code']}") + return True + else: + logger.error(f"Telegram告警发送失败: {response.status_code} - {response.text}") + return False + + except Exception as e: + logger.error(f"Telegram告警发送失败: {e}") + return False + + +class AlertHistory: + """告警历史记录""" + + def __init__(self, history_file: str = "alert_history.json"): + self.history_file = history_file + self.history = self._load_history() + + def _load_history(self) -> Dict[str, List[Dict]]: + """加载历史记录""" + if os.path.exists(self.history_file): + try: + with open(self.history_file, 'r', encoding='utf-8') as f: + return json.load(f) + except Exception as e: + logger.error(f"加载告警历史失败: {e}") + return {} + return {} + + def _save_history(self): + """保存历史记录""" + try: + with open(self.history_file, 'w', encoding='utf-8') as f: + json.dump(self.history, f, ensure_ascii=False, indent=2) + except Exception as e: + logger.error(f"保存告警历史失败: {e}") + + def add_alert(self, code: str, alert_message: Dict[str, Any]): + """添加告警记录""" + if code not in self.history: + self.history[code] = [] + + self.history[code].append({ + "timestamp": alert_message['timestamp'], + "change_pct": alert_message['change_pct'], + "threshold": alert_message['threshold'], + "direction": alert_message['direction'] + }) + + # 只保留最近100条记录 + if len(self.history[code]) > 100: + self.history[code] = self.history[code][-100:] + + self._save_history() + + def should_send_alert(self, code: str, cooldown_minutes: int = 30) -> bool: + """ + 检查是否应该发送告警（避免重复通知） + + Args: + code: 基金/指数代码 + cooldown_minutes: 冷却时间（分钟） + + Returns: + bool: 是否应该发送告警 + """ + if code not in self.history or not self.history[code]: + return True + + last_alert = self.history[code][-1] + last_time = datetime.fromisoformat(last_alert['timestamp']) + now = datetime.now() + + # 检查是否在冷却时间内 + cooldown_seconds = cooldown_minutes * 60 + time_diff = (now - last_time).total_seconds() + + return time_diff > cooldown_seconds + + def get_today_alerts_count(self, code: str) -> int: + """获取今天该代码的告警次数""" + if code not in self.history: + return 0 + + today = datetime.now().date() + count = 0 + + for alert in self.history[code]: + alert_date = datetime.fromisoformat(alert['timestamp']).date() + if alert_date == today: + count += 1 + + return count + + +class AlertManager: + """告警管理器""" + + def __init__(self, config_file: str = "alerts_config.json"): + self.config_file = config_file + self.config = self._load_config() + self.rules = self._load_rules() + self.channels = self._load_channels() + self.history = AlertHistory() + self.global_settings = self.config.get('global_settings', {}) + + def _load_config(self) -> Dict[str, Any]: + """加载配置文件""" + if not os.path.exists(self.config_file): + logger.warning(f"告警配置文件不存在: {self.config_file}") + return {"enabled": False} + + try: + with open(self.config_file, 'r', encoding='utf-8') as f: + return json.load(f) + except Exception as e: + logger.error(f"加载告警配置失败: {e}") + return {"enabled": False} + + def _load_rules(self) -> List[AlertRule]: + """加载告警规则""" + rules_config = self.config.get('rules', []) + rules = [] + + for rule_config in rules_config: + try: + rule = AlertRule(rule_config) + if rule.enabled: + rules.append(rule) + except Exception as e: + logger.error(f"加载告警规则失败: {e}") + + logger.info(f"加载了 {len(rules)} 个告警规则") + return rules + + def _load_channels(self) -> Dict[str, NotificationChannel]: + """加载通知渠道""" + channels_config = self.config.get('notification_channels', {}) + channels = {} + + # 邮件渠道 + if 'email' in channels_config: + channels['email'] = EmailNotification(channels_config['email']) + + # 飞书渠道 + if 'feishu' in channels_config: + channels['feishu'] = FeishuNotification(channels_config['feishu']) + + # 企业微信渠道 + if 'wechat_work' in channels_config: + channels['wechat_work'] = WechatWorkNotification(channels_config['wechat_work']) + + # 钉钉渠道 + if 'dingtalk' in channels_config: + channels['dingtalk'] = DingTalkNotification(channels_config['dingtalk']) + + # Telegram渠道 + if 'telegram' in channels_config: + channels['telegram'] = TelegramNotification(channels_config['telegram']) + + logger.info(f"加载了 {len(channels)} 个通知渠道") + return channels + + def check_alerts(self, fund_data: Dict[str, Dict]) -> List[Dict[str, Any]]: + """ + 检查所有基金数据，触发告警 + + Args: + fund_data: 基金数据字典，格式为 {code: {name, price, change}} + + Returns: + List: 触发的告警列表 + """ + if not self.config.get('enabled', False): + logger.info("告警功能已禁用") + return [] + + triggered_alerts = [] + + for rule in self.rules: + code = rule.code + + if code not in fund_data: + logger.debug(f"规则 {code} 对应的基金数据不存在") + continue + + data = fund_data[code] + change_pct = data.get('change', 0.0) + current_value = data.get('price', 0.0) + + # 检查是否应该触发告警 + if rule.should_alert(change_pct): + # 检查冷却时间 + cooldown = self.global_settings.get('cooldown_minutes', 30) + if not self.history.should_send_alert(code, cooldown): + logger.info(f"代码 {code} 仍在冷却时间内，跳过告警") + continue + + # 检查每日最大告警次数 + max_per_day = self.global_settings.get('max_alerts_per_day', 10) + today_count = self.history.get_today_alerts_count(code) + if today_count >= max_per_day: + logger.warning(f"代码 {code} 今日告警次数已达上限 ({max_per_day}次)") + continue + + # 创建告警消息 + alert_message = rule.get_alert_message(change_pct, current_value) + + # 发送通知 + sent_channels = [] + for channel_name in rule.notification_channels: + if channel_name in self.channels: + channel = self.channels[channel_name] + if channel.send(alert_message): + sent_channels.append(channel_name) + + if sent_channels: + alert_message['sent_channels'] = sent_channels + triggered_alerts.append(alert_message) + + # 记录到历史 + self.history.add_alert(code, alert_message) + + logger.info(f"代码 {code} 触发告警，已通过 {', '.join(sent_channels)} 发送通知") + + return triggered_alerts + + def send_daily_summary(self, fund_data: Dict[str, Dict], triggered_alerts: List[Dict[str, Any]]) -> bool: + """ + 发送每日汇总报告 + + Args: + fund_data: 所有基金数据 + triggered_alerts: 今日触发的告警列表 + + Returns: + bool: 是否发送成功 + """ + if not self.global_settings.get('enable_daily_summary', False): + return False + + # 检查是否应该发送每日汇总 + summary_time = self.global_settings.get('summary_time', '18:00') + current_time = datetime.now().strftime('%H:%M') + + if current_time != summary_time: + return False + + # 创建汇总消息 + total_funds = len(fund_data) + total_alerts = len(triggered_alerts) + + summary_message = { + "type": "daily_summary", + "date": datetime.now().strftime('%Y-%m-%d'), + "total_funds": total_funds, + "total_alerts": total_alerts, + "triggered_alerts": triggered_alerts, + "timestamp": datetime.now().isoformat() + } + + # 发送到所有启用的渠道 + success = False + for channel_name, channel in self.channels.items(): + if channel.enabled: + # 这里需要为每个渠道实现汇总消息的发送 + # 暂时只记录日志 + logger.info(f"发送每日汇总到 {channel_name}") + + return success + + +# 全局告警管理器实例 +_alert_manager_instance = None + +def get_alert_manager() -> AlertManager: + """获取全局告警管理器实例""" + global _alert_manager_instance + if _alert_manager_instance is None: + _alert_manager_instance = AlertManager() + return _alert_manager_instance + + +def check_and_send_alerts(fund_data: Dict[str, Dict]) -> List[Dict[str, Any]]: + """ + 检查并发送告警（便捷函数） + + Args: + fund_data: 基金数据字典 + + Returns: + List: 触发的告警列表 + """ + manager = get_alert_manager() + return manager.check_alerts(fund_data) + + +if __name__ == "__main__": + # 测试代码 + test_data = { + "000001": {"name": "上证指数", "price": 3200.0, "change": 5.5}, + "NDX": {"name": "纳斯达克", "price": 18000.0, "change": 2.8}, + "007360": {"name": "测试基金", "price": 1.2, "change": -3.5} + } + + alerts = check_and_send_alerts(test_data) + print(f"触发了 {len(alerts)} 个告警") + for alert in alerts: + print(f"告警: {alert['message']}") \ No newline at end of file diff --git a/alerts_config.example.json b/alerts_config.example.json new file mode 100644 index 0000000..322ed6e --- /dev/null +++ b/alerts_config.example.json @@ -0,0 +1,106 @@ +{ + "enabled": true, + "notification_channels": { + "email": { + "enabled": false, + "smtp_server": "smtp.gmail.com", + "smtp_port": 587, + "smtp_username": "your-email@gmail.com", + "smtp_password": "your-app-password", + "sender_email": "your-email@gmail.com", + "receiver_emails": ["recipient@example.com"] + }, + "feishu": { + "enabled": true, + "webhook_url": "https://open.feishu.cn/open-apis/bot/v2/hook/your-webhook-token", + "template": "default" + }, + "wechat_work": { + "enabled": false, + "webhook_url": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your-key", + "key": "your-key" + }, + "dingtalk": { + "enabled": false, + "webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=your-token", + "secret": "your-secret-if-has" + }, + "telegram": { + "enabled": false, + "bot_token": "your-bot-token", + "chat_id": "your-chat-id" + } + }, + "rules": [ + { + "code": "000001", + "name": "上证指数", + "threshold": 5.0, + "direction": "up", + "enabled": true, + "notification_channels": ["feishu"] + }, + { + "code": "000001", + "name": "上证指数", + "threshold": -3.0, + "direction": "down", + "enabled": true, + "notification_channels": ["feishu"] + }, + { + "code": "HSI", + "name": "恒生指数", + "threshold": 4.0, + "direction": "up", + "enabled": true, + "notification_channels": ["feishu"] + }, + { + "code": "HSI", + "name": "恒生指数", + "threshold": -2.5, + "direction": "down", + "enabled": true, + "notification_channels": ["feishu"] + }, + { + "code": "NDX", + "name": "纳斯达克", + "threshold": 3.0, + "direction": "up", + "enabled": true, + "notification_channels": ["feishu"] + }, + { + "code": "NDX", + "name": "纳斯达克", + "threshold": -2.0, + "direction": "down", + "enabled": true, + "notification_channels": ["feishu"] + }, + { + "code": "015596", + "name": "国泰国证有色金属行业指数(LOF)C", + "threshold": 5.0, + "direction": "up", + "enabled": true, + "notification_channels": ["feishu"] + }, + { + "code": "015596", + "name": "国泰国证有色金属行业指数(LOF)C", + "threshold": -3.0, + "direction": "down", + "enabled": true, + "notification_channels": ["feishu"] + } + ], + "global_settings": { + "cooldown_minutes": 30, + "max_alerts_per_day": 10, + "enable_daily_summary": true, + "summary_time": "18:00" + } +} \ No newline at end of file diff --git a/alerts_config.json b/alerts_config.json new file mode 100644 index 0000000..08c5868 --- /dev/null +++ b/alerts_config.json @@ -0,0 +1,66 @@ +{ + "enabled": true, + "notification_channels": { + "email": { + "enabled": false, + "smtp_server": "smtp.example.com", + "smtp_port": 587, + "smtp_username": "user@example.com", + "smtp_password": "", + "sender_email": "user@example.com", + "receiver_emails": ["user@example.com"] + }, + "feishu": { + "enabled": true, + "webhook_url": "", + "template": "default" + }, + "wechat_work": { + "enabled": false, + "webhook_url": "", + "key": "" + }, + "dingtalk": { + "enabled": false, + "webhook_url": "", + "secret": "" + }, + "telegram": { + "enabled": false, + "bot_token": "", + "chat_id": "" + } + }, + "rules": [ + { + "code": "000001", + "name": "上证指数", + "threshold": 5.0, + "direction": "up", + "enabled": true, + "notification_channels": ["feishu"] + }, + { + "code": "000001", + "name": "上证指数", + "threshold": -3.0, + "direction": "down", + "enabled": true, + "notification_channels": ["feishu"] + }, + { + "code": "NDX", + "name": "纳斯达克", + "threshold": 3.0, + "direction": "up", + "enabled": true, + "notification_channels": ["feishu"] + } + ], + "global_settings": { + "cooldown_minutes": 30, + "max_alerts_per_day": 10, + "enable_daily_summary": true, + "summary_time": "18:00" + } +} \ No newline at end of file diff --git a/check_alerts.py b/check_alerts.py new file mode 100644 index 0000000..b9c7180 --- /dev/null +++ b/check_alerts.py @@ -0,0 +1,221 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +独立的告警检查脚本 +可以单独运行来检查告警规则 +Author: TickEye Team +Date: 2026-03-12 +""" + +import sys +import os +import logging +from datetime import datetime + +# 添加项目路径 +sys.path.append(os.path.dirname(os.path.abspath(__file__))) + +from fund_analysis import get_fund_summary, get_owned_funds +from alert_manager import check_and_send_alerts, get_alert_manager + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + + +def convert_fund_data_for_alerts(fund_summaries): + """ + 将 fund_analysis.py 的数据格式转换为告警管理器期望的格式 + + Args: + fund_summaries: fund_analysis.py 返回的基金数据列表 + + Returns: + dict: 适合告警检查的数据格式 + """ + alert_data = {} + + for summary in fund_summaries: + # 只处理状态正常的基金 + if summary['status'] == '正常': + fund_code = summary['fund_code'] + + # 提取净值（去掉字符串格式） + try: + price = float(summary['net_value']) + except (ValueError, TypeError): + price = 0.0 + + # 提取涨跌幅（去掉%符号） + try: + change_str = summary['change_pct'] + if change_str != 'N/A' and '%' in change_str: + change = float(change_str.replace('%', '')) + else: + change = 0.0 + except (ValueError, TypeError): + change = 0.0 + + alert_data[fund_code] = { + "name": summary['fund_name'], + "price": price, + "change": change + } + + return alert_data + + +def get_all_fund_data(days=1): + """ + 获取所有配置基金的分析数据 + + Args: + days: 分析最近多少天的数据 + + Returns: + dict: 基金分析数据字典 + """ + owned_funds, _ = get_owned_funds() + if not owned_funds: + logger.error("未配置任何基金代码！") + return {} + + logger.info(f"开始获取 {len(owned_funds)} 个标的的数据...") + + fund_summaries = [] + for fund_code in owned_funds: + logger.info(f"正在获取基金 {fund_code} 的数据...") + summary = get_fund_summary(fund_code, days) + fund_summaries.append(summary) + + return convert_fund_data_for_alerts(fund_summaries) + + +def check_alerts_standalone(days=1): + """ + 独立的告警检查函数 + + Args: + days: 分析最近多少天的数据 + + Returns: + bool: 是否成功执行 + """ + print("🔔 TickEye 告警检查") + print("=" * 60) + + # 检查告警配置 + alert_manager = get_alert_manager() + if not alert_manager.config.get('enabled', False): + print("⚠️ 告警功能已禁用") + print("请在 alerts_config.json 中启用告警功能") + return False + + print("📊 告警配置已加载") + print(f"📅 数据分析天数: {days} 天") + print(f"📋 告警规则数: {len(alert_manager.rules)}") + print(f"📱 通知渠道数: {len(alert_manager.channels)}") + print() + + # 获取基金数据 + print("🔍 正在获取基金数据...") + fund_data = get_all_fund_data(days) + + if not fund_data: + print("❌ 未获取到任何基金数据") + return False + + print(f"✅ 成功获取 {len(fund_data)} 个标的的数据") + + # 检查告警 + print("\n🔔 正在检查告警规则...") + alerts = check_and_send_alerts(fund_data) + + if alerts: + print(f"⚠️ 触发了 {len(alerts)} 个告警:") + for i, alert in enumerate(alerts, 1): + print(f"\n {i}. {alert['message']}") + print(f" 代码: {alert['code']}") + print(f" 当前值: {alert['current_value']}") + print(f" 涨跌幅: {alert['change_pct']:.2f}%") + print(f" 阈值: {alert['direction_text']} {alert['threshold']}%") + print(f" 通知渠道: {', '.join(alert.get('sent_channels', []))}") + print(f" 触发时间: {alert['timestamp']}") + else: + print("✅ 未触发任何告警") + + # 显示告警规则统计 + print("\n📊 告警规则统计:") + for rule in alert_manager.rules: + code = rule.code + name = rule.name or code + threshold = rule.threshold + direction = "上涨" if rule.direction == 'up' else "下跌" + channels = ', '.join(rule.notification_channels) if rule.notification_channels else "无" + + # 检查该代码是否有数据 + has_data = code in fund_data + current_change = fund_data.get(code, {}).get('change', 'N/A') if has_data else 'N/A' + + status = "✅ 正常" if rule.enabled else "❌ 禁用" + data_status = "📊 有数据" if has_data else "❌ 无数据" + + print(f" • {name} ({code}): {direction} {threshold}% → {channels}") + print(f" 状态: {status} | {data_status} | 当前涨跌幅: {current_change}%") + + return True + + +def show_alert_history(): + """显示告警历史""" + alert_manager = get_alert_manager() + history = alert_manager.history.history + + if not history: + print("📝 告警历史为空") + return + + print("📝 告警历史记录:") + for code, alerts in history.items(): + print(f"\n {code} ({len(alerts)} 条记录):") + for alert in alerts[-5:]: # 只显示最近5条 + timestamp = alert['timestamp'] + change_pct = alert['change_pct'] + threshold = alert['threshold'] + direction = "上涨" if alert['direction'] == 'up' else "下跌" + print(f" • {timestamp}: {direction} {threshold}% (实际: {change_pct:.2f}%)") + + +def main(): + """主函数""" + # 默认参数 + days = 1 + show_history = False + + # 解析命令行参数 + if len(sys.argv) > 1: + if sys.argv[1] == 'history': + show_history = True + else: + try: + days = int(sys.argv[1]) + if days <= 0: + print("❌ 天数必须大于0，使用默认值1天") + days = 1 + except ValueError: + print("❌ 请输入有效的天数，使用默认值1天") + days = 1 + + if show_history: + show_alert_history() + else: + success = check_alerts_standalone(days) + if not success: + sys.exit(1) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/main.py b/main.py index b473655..2dbfcd2 100644 --- a/main.py +++ b/main.py @@ -22,6 +22,7 @@ from fund_analysis import get_fund_summary, get_owned_funds from feishu_notifier import FeishuNotifier from feishu_config import get_config +from alert_manager import check_and_send_alerts, get_alert_manager # 配置日志 logging.basicConfig( @@ -188,6 +189,19 @@ def send_fund_analysis_to_feishu(days=1, send_summary=True, send_table=False): print(f" 📊 基金数据: {success_count}/{total_funds} 只成功") print(f" 📤 飞书通知: {success_notifications}/{total_notifications} 条成功") + # 检查并发送告警 + print("\n🔔 正在检查告警规则...") + try: + alerts = check_and_send_alerts(feishu_data) + if alerts: + print(f" ⚠️ 触发了 {len(alerts)} 个告警:") + for alert in alerts: + print(f" • {alert['message']}") + else: + print(" ✅ 未触发任何告警") + except Exception as e: + print(f" ❌ 告警检查失败: {e}") + if success_notifications == total_notifications: print("🎉 基金监测与通知任务完成！") return True From e7a8850e460c6b53ec8bde323653b5a5665f9be9 Mon Sep 17 00:00:00 2001 From: dethan3 Date: Thu, 19 Mar 2026 13:06:49 +0800 Subject: [PATCH 2/2] refactor(alerts): harden alert manager implementation Improve alert rule and notification channel typing, centralize defaults, and tighten alert history handling. Also keep alert dispatch behavior explicit so only successfully delivered notifications are recorded and returned. --- alert_manager.py | 929 ++++++++++++++++++++++++++++------------------- 1 file changed, 555 insertions(+), 374 deletions(-) diff --git a/alert_manager.py b/alert_manager.py index c86fe0a..b13b79f 100644 --- a/alert_manager.py +++ b/alert_manager.py @@ -6,81 +6,113 @@ Date: 2026-03-12 """ -import os +import base64 +import hashlib +import hmac import json import logging -import time +import os import smtplib -from email.mime.text import MIMEText +import time +import urllib.parse +from datetime import datetime from email.mime.multipart import MIMEMultipart -from datetime import datetime, timedelta -from typing import Dict, List, Optional, Any, Tuple +from email.mime.text import MIMEText +from typing import Any, Dict, List, Optional, Union + import requests -from pathlib import Path + +# 类型别名 +ConfigDict = Dict[str, Any] +AlertMessage = Dict[str, Any] +HistoryRecord = Dict[str, Union[str, float]] +FundItem = Dict[str, Union[str, float, int]] +FundData = Dict[str, FundItem] # 配置日志 logging.basicConfig( level=logging.INFO, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", ) logger = logging.getLogger(__name__) class AlertRule: """告警规则类""" - - def __init__(self, rule_config: Dict[str, Any]): + + DIRECTION_UP = "up" + DIRECTION_DOWN = "down" + VALID_DIRECTIONS = {DIRECTION_UP, DIRECTION_DOWN} + DEFAULT_DIRECTION = DIRECTION_UP + DEFAULT_THRESHOLD = 0.0 + DEFAULT_ENABLED = True + DEFAULT_CODE = "" + DEFAULT_NAME = "" + DEFAULT_NOTIFICATION_CHANNELS: List[str] = [] + + def __init__(self, rule_config: ConfigDict) -> None: """ 初始化告警规则 - + Args: rule_config: 规则配置字典 """ - self.code = rule_config.get('code', '') - self.name = rule_config.get('name', '') - self.threshold = float(rule_config.get('threshold', 0.0)) - self.direction = rule_config.get('direction', 'up') # 'up' or 'down' - self.enabled = rule_config.get('enabled', True) - self.notification_channels = rule_config.get('notification_channels', []) - - # 验证配置 + self.code: str = str(rule_config.get("code", self.DEFAULT_CODE)) + self.name: str = str(rule_config.get("name", self.DEFAULT_NAME)) + self.threshold: float = float( + rule_config.get("threshold", self.DEFAULT_THRESHOLD) + ) + self.direction: str = str( + rule_config.get("direction", self.DEFAULT_DIRECTION) + ) + self.enabled: bool = bool(rule_config.get("enabled", self.DEFAULT_ENABLED)) + self.notification_channels: List[str] = list( + rule_config.get( + "notification_channels", + self.DEFAULT_NOTIFICATION_CHANNELS, + ) + ) + if not self.code: raise ValueError("告警规则必须包含代码(code)") - if self.direction not in ['up', 'down']: + if self.direction not in self.VALID_DIRECTIONS: raise ValueError("告警规则方向(direction)必须是'up'或'down'") - + def should_alert(self, change_pct: float) -> bool: """ 检查是否应该触发告警 - + Args: change_pct: 涨跌幅百分比 - + Returns: bool: 是否触发告警 """ if not self.enabled: return False - - if self.direction == 'up': + + if self.direction == self.DIRECTION_UP: return change_pct >= self.threshold - else: # down - return change_pct <= self.threshold - - def get_alert_message(self, change_pct: float, current_value: float) -> Dict[str, Any]: + return change_pct <= self.threshold + + def get_alert_message( + self, + change_pct: float, + current_value: Union[int, float], + ) -> AlertMessage: """ 获取告警消息 - + Args: change_pct: 涨跌幅百分比 current_value: 当前值 - + Returns: - Dict: 告警消息 + AlertMessage: 告警消息 """ - direction_text = "上涨" if self.direction == 'up' else "下跌" + direction_text = "上涨" if self.direction == self.DIRECTION_UP else "下跌" threshold_text = f"{abs(self.threshold)}%" - + return { "code": self.code, "name": self.name or self.code, @@ -91,576 +123,726 @@ def get_alert_message(self, change_pct: float, current_value: float) -> Dict[str "direction_text": direction_text, "threshold_text": threshold_text, "timestamp": datetime.now().isoformat(), - "message": f"{self.name or self.code} {direction_text}超过{threshold_text}，当前涨跌幅: {change_pct:.2f}%" + "message": ( + f"{self.name or self.code} {direction_text}超过{threshold_text}，" + f"当前涨跌幅: {change_pct:.2f}%" + ), } class NotificationChannel: """通知渠道基类""" - - def __init__(self, config: Dict[str, Any]): - self.config = config - self.enabled = config.get('enabled', False) - - def send(self, alert_message: Dict[str, Any]) -> bool: + + DEFAULT_ENABLED = False + + def __init__(self, config: ConfigDict) -> None: + self.config: ConfigDict = config + self.enabled: bool = bool(config.get("enabled", self.DEFAULT_ENABLED)) + + def send(self, alert_message: AlertMessage) -> bool: """发送通知""" raise NotImplementedError("子类必须实现send方法") class EmailNotification(NotificationChannel): """邮件通知""" - - def __init__(self, config: Dict[str, Any]): - super().__init__(config) - self.smtp_server = config.get('smtp_server', '') - self.smtp_port = config.get('smtp_port', 587) - self.smtp_username = config.get('smtp_username', '') - self.smtp_password = config.get('smtp_password', '') - self.sender_email = config.get('sender_email', '') - self.receiver_emails = config.get('receiver_emails', []) - - def send(self, alert_message: Dict[str, Any]) -> bool: - if not self.enabled: - return False - - try: - # 创建邮件内容 - subject = f"📈 基金告警: {alert_message['name']} {alert_message['direction_text']}超过{alert_message['threshold_text']}" - - body = f""" + + DEFAULT_SMTP_SERVER = "" + DEFAULT_SMTP_PORT = 587 + DEFAULT_SMTP_USERNAME = "" + DEFAULT_SMTP_PASSWORD = "" + DEFAULT_SENDER_EMAIL = "" + DEFAULT_RECEIVER_EMAILS: List[str] = [] + SUBJECT_TEMPLATE = "📈 基金告警: {name} {direction_text}超过{threshold_text}" + HTML_TEMPLATE = """

📈 基金告警通知

-

标的: {alert_message['name']} ({alert_message['code']})

-

当前值: {alert_message['current_value']}

-

涨跌幅: {alert_message['change_pct']:.2f}%

-

告警阈值: {alert_message['direction_text']} {alert_message['threshold']}%

-

触发时间: {alert_message['timestamp']}

+

标的: {name} ({code})

+

当前值: {current_value}

+

涨跌幅: {change_pct:.2f}%

+

告警阈值: {direction_text} {threshold}%

+

触发时间: {timestamp}

来自 TickEye 基金监测系统

- """ - - # 创建邮件 + """ + MIME_SUBTYPE_HTML = "html" + + def __init__(self, config: ConfigDict) -> None: + super().__init__(config) + self.smtp_server: str = str( + config.get("smtp_server", self.DEFAULT_SMTP_SERVER) + ) + self.smtp_port: int = int(config.get("smtp_port", self.DEFAULT_SMTP_PORT)) + self.smtp_username: str = str( + config.get("smtp_username", self.DEFAULT_SMTP_USERNAME) + ) + self.smtp_password: str = str( + config.get("smtp_password", self.DEFAULT_SMTP_PASSWORD) + ) + self.sender_email: str = str( + config.get("sender_email", self.DEFAULT_SENDER_EMAIL) + ) + self.receiver_emails: List[str] = list( + config.get("receiver_emails", self.DEFAULT_RECEIVER_EMAILS) + ) + + def send(self, alert_message: AlertMessage) -> bool: + if not self.enabled: + return False + + try: + subject = self.SUBJECT_TEMPLATE.format( + name=alert_message["name"], + direction_text=alert_message["direction_text"], + threshold_text=alert_message["threshold_text"], + ) + body = self.HTML_TEMPLATE.format(**alert_message) + msg = MIMEMultipart() - msg['From'] = self.sender_email - msg['To'] = ', '.join(self.receiver_emails) - msg['Subject'] = subject - msg.attach(MIMEText(body, 'html')) - - # 发送邮件 + msg["From"] = self.sender_email + msg["To"] = ", ".join(self.receiver_emails) + msg["Subject"] = subject + msg.attach(MIMEText(body, self.MIME_SUBTYPE_HTML)) + with smtplib.SMTP(self.smtp_server, self.smtp_port) as server: server.starttls() server.login(self.smtp_username, self.smtp_password) server.send_message(msg) - + logger.info(f"邮件告警发送成功: {alert_message['code']}") return True - - except Exception as e: - logger.error(f"邮件告警发送失败: {e}") + + except Exception as exc: + logger.error(f"邮件告警发送失败: {exc}") return False class FeishuNotification(NotificationChannel): """飞书通知""" - - def __init__(self, config: Dict[str, Any]): + + DEFAULT_WEBHOOK_URL = "" + DEFAULT_TEMPLATE = "default" + REQUEST_TIMEOUT = 10 + CONTENT_TYPE_JSON = "application/json" + HEADER_TEMPLATE_UP = "green" + HEADER_TEMPLATE_DOWN = "red" + SOURCE_TEXT = "来自 TickEye 基金监测系统" + + def __init__(self, config: ConfigDict) -> None: super().__init__(config) - self.webhook_url = config.get('webhook_url', '') - self.template = config.get('template', 'default') - - def send(self, alert_message: Dict[str, Any]) -> bool: + self.webhook_url: str = str( + config.get("webhook_url", self.DEFAULT_WEBHOOK_URL) + ) + self.template: str = str(config.get("template", self.DEFAULT_TEMPLATE)) + + def send(self, alert_message: AlertMessage) -> bool: if not self.enabled or not self.webhook_url: return False - + try: - # 创建飞书消息卡片 card = { "msg_type": "interactive", "card": { "config": { - "wide_screen_mode": True + "wide_screen_mode": True, }, "header": { "title": { "tag": "plain_text", - "content": f"📈 基金告警: {alert_message['name']}" + "content": f"📈 基金告警: {alert_message['name']}", }, - "template": "red" if alert_message['direction'] == 'down' else "green" + "template": ( + self.HEADER_TEMPLATE_DOWN + if alert_message["direction"] == AlertRule.DIRECTION_DOWN + else self.HEADER_TEMPLATE_UP + ), }, "elements": [ { "tag": "div", "text": { "tag": "lark_md", - "content": f"**标的:** {alert_message['name']} ({alert_message['code']})\n" - f"**当前值:** {alert_message['current_value']}\n" - f"**涨跌幅:** {alert_message['change_pct']:.2f}%\n" - f"**告警阈值:** {alert_message['direction_text']} {alert_message['threshold']}%\n" - f"**触发时间:** {alert_message['timestamp']}" - } + "content": ( + f"**标的:** {alert_message['name']} " + f"({alert_message['code']})\n" + f"**当前值:** {alert_message['current_value']}\n" + f"**涨跌幅:** {alert_message['change_pct']:.2f}%\n" + f"**告警阈值:** " + f"{alert_message['direction_text']} " + f"{alert_message['threshold']}%\n" + f"**触发时间:** {alert_message['timestamp']}" + ), + }, }, { - "tag": "hr" + "tag": "hr", }, { "tag": "note", "elements": [ { "tag": "plain_text", - "content": "来自 TickEye 基金监测系统" + "content": self.SOURCE_TEXT, } - ] - } - ] - } + ], + }, + ], + }, } - - # 发送请求 + response = requests.post( self.webhook_url, json=card, - headers={'Content-Type': 'application/json'}, - timeout=10 + headers={"Content-Type": self.CONTENT_TYPE_JSON}, + timeout=self.REQUEST_TIMEOUT, ) - + if response.status_code == 200: logger.info(f"飞书告警发送成功: {alert_message['code']}") return True - else: - logger.error(f"飞书告警发送失败: {response.status_code} - {response.text}") - return False - - except Exception as e: - logger.error(f"飞书告警发送失败: {e}") + + logger.error( + f"飞书告警发送失败: {response.status_code} - {response.text}" + ) + return False + + except Exception as exc: + logger.error(f"飞书告警发送失败: {exc}") return False class WechatWorkNotification(NotificationChannel): """企业微信通知""" - - def __init__(self, config: Dict[str, Any]): + + DEFAULT_WEBHOOK_URL = "" + DEFAULT_KEY = "" + REQUEST_TIMEOUT = 10 + CONTENT_TYPE_JSON = "application/json" + MESSAGE_TEMPLATE = ( + "## 📈 基金告警通知\n" + "> **标的:** {name} ({code})\n" + "> **当前值:** {current_value}\n" + "> **涨跌幅:** {change_pct:.2f}%\n" + "> **告警阈值:** {direction_text} {threshold}%\n" + "> **触发时间:** {timestamp}\n" + "> \n" + "> 来自 TickEye 基金监测系统" + ) + + def __init__(self, config: ConfigDict) -> None: super().__init__(config) - self.webhook_url = config.get('webhook_url', '') - self.key = config.get('key', '') - - def send(self, alert_message: Dict[str, Any]) -> bool: + self.webhook_url: str = str( + config.get("webhook_url", self.DEFAULT_WEBHOOK_URL) + ) + self.key: str = str(config.get("key", self.DEFAULT_KEY)) + + def send(self, alert_message: AlertMessage) -> bool: if not self.enabled or not self.webhook_url: return False - + try: - # 创建企业微信消息 message = { "msgtype": "markdown", "markdown": { - "content": f"## 📈 基金告警通知\n" - f"> **标的:** {alert_message['name']} ({alert_message['code']})\n" - f"> **当前值:** {alert_message['current_value']}\n" - f"> **涨跌幅:** {alert_message['change_pct']:.2f}%\n" - f"> **告警阈值:** {alert_message['direction_text']} {alert_message['threshold']}%\n" - f"> **触发时间:** {alert_message['timestamp']}\n" - f"> \n" - f"> 来自 TickEye 基金监测系统" - } + "content": self.MESSAGE_TEMPLATE.format(**alert_message), + }, } - - # 发送请求 + response = requests.post( self.webhook_url, json=message, - headers={'Content-Type': 'application/json'}, - timeout=10 + headers={"Content-Type": self.CONTENT_TYPE_JSON}, + timeout=self.REQUEST_TIMEOUT, ) - + if response.status_code == 200: logger.info(f"企业微信告警发送成功: {alert_message['code']}") return True - else: - logger.error(f"企业微信告警发送失败: {response.status_code} - {response.text}") - return False - - except Exception as e: - logger.error(f"企业微信告警发送失败: {e}") + + logger.error( + f"企业微信告警发送失败: {response.status_code} - {response.text}" + ) + return False + + except Exception as exc: + logger.error(f"企业微信告警发送失败: {exc}") return False class DingTalkNotification(NotificationChannel): """钉钉通知""" - - def __init__(self, config: Dict[str, Any]): + + DEFAULT_WEBHOOK_URL = "" + DEFAULT_SECRET = "" + REQUEST_TIMEOUT = 10 + CONTENT_TYPE_JSON = "application/json" + MESSAGE_TITLE_TEMPLATE = "📈 基金告警: {name}" + MESSAGE_TEXT_TEMPLATE = ( + "### 📈 基金告警通知\n" + "**标的:** {name} ({code})\n\n" + "**当前值:** {current_value}\n\n" + "**涨跌幅:** {change_pct:.2f}%\n\n" + "**告警阈值:** {direction_text} {threshold}%\n\n" + "**触发时间:** {timestamp}\n\n" + "---\n\n" + "来自 TickEye 基金监测系统" + ) + TIMESTAMP_MULTIPLIER = 1000 + ENCODING = "utf-8" + SIGN_SEPARATOR = "\n" + SIGN_TIMESTAMP_PARAM = "timestamp" + SIGN_PARAM = "sign" + + def __init__(self, config: ConfigDict) -> None: super().__init__(config) - self.webhook_url = config.get('webhook_url', '') - self.secret = config.get('secret', '') - - def send(self, alert_message: Dict[str, Any]) -> bool: + self.webhook_url: str = str( + config.get("webhook_url", self.DEFAULT_WEBHOOK_URL) + ) + self.secret: str = str(config.get("secret", self.DEFAULT_SECRET)) + + def send(self, alert_message: AlertMessage) -> bool: if not self.enabled or not self.webhook_url: return False - + try: - import hmac - import hashlib - import base64 - import urllib.parse - - # 如果有secret，计算签名 - timestamp = str(round(time.time() * 1000)) - sign = '' - + timestamp = str(round(time.time() * self.TIMESTAMP_MULTIPLIER)) + webhook_url = self.webhook_url + if self.secret: - secret_enc = self.secret.encode('utf-8') - string_to_sign = f"{timestamp}\n{self.secret}" - string_to_sign_enc = string_to_sign.encode('utf-8') - hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest() + secret_enc = self.secret.encode(self.ENCODING) + string_to_sign = ( + f"{timestamp}{self.SIGN_SEPARATOR}{self.secret}" + ).encode(self.ENCODING) + hmac_code = hmac.new( + secret_enc, + string_to_sign, + digestmod=hashlib.sha256, + ).digest() sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) - webhook_url = f"{self.webhook_url}×tamp={timestamp}&sign={sign}" - else: - webhook_url = self.webhook_url - - # 创建钉钉消息 + webhook_url = ( + f"{self.webhook_url}&{self.SIGN_TIMESTAMP_PARAM}={timestamp}" + f"&{self.SIGN_PARAM}={sign}" + ) + message = { "msgtype": "markdown", "markdown": { - "title": f"📈 基金告警: {alert_message['name']}", - "text": f"### 📈 基金告警通知\n" - f"**标的:** {alert_message['name']} ({alert_message['code']})\n\n" - f"**当前值:** {alert_message['current_value']}\n\n" - f"**涨跌幅:** {alert_message['change_pct']:.2f}%\n\n" - f"**告警阈值:** {alert_message['direction_text']} {alert_message['threshold']}%\n\n" - f"**触发时间:** {alert_message['timestamp']}\n\n" - f"---\n\n" - f"来自 TickEye 基金监测系统" - } + "title": self.MESSAGE_TITLE_TEMPLATE.format( + name=alert_message["name"] + ), + "text": self.MESSAGE_TEXT_TEMPLATE.format(**alert_message), + }, } - - # 发送请求 + response = requests.post( webhook_url, json=message, - headers={'Content-Type': 'application/json'}, - timeout=10 + headers={"Content-Type": self.CONTENT_TYPE_JSON}, + timeout=self.REQUEST_TIMEOUT, ) - + if response.status_code == 200: logger.info(f"钉钉告警发送成功: {alert_message['code']}") return True - else: - logger.error(f"钉钉告警发送失败: {response.status_code} - {response.text}") - return False - - except Exception as e: - logger.error(f"钉钉告警发送失败: {e}") + + logger.error( + f"钉钉告警发送失败: {response.status_code} - {response.text}" + ) + return False + + except Exception as exc: + logger.error(f"钉钉告警发送失败: {exc}") return False class TelegramNotification(NotificationChannel): """Telegram通知""" - - def __init__(self, config: Dict[str, Any]): + + DEFAULT_BOT_TOKEN = "" + DEFAULT_CHAT_ID = "" + REQUEST_TIMEOUT = 10 + API_URL_TEMPLATE = "https://api.telegram.org/bot{bot_token}/sendMessage" + PARSE_MODE = "Markdown" + MESSAGE_TEMPLATE = ( + "📈 *基金告警通知*\n\n" + "*标的:* {name} ({code})\n" + "*当前值:* {current_value}\n" + "*涨跌幅:* {change_pct:.2f}%\n" + "*告警阈值:* {direction_text} {threshold}%\n" + "*触发时间:* {timestamp}\n\n" + "来自 TickEye 基金监测系统" + ) + + def __init__(self, config: ConfigDict) -> None: super().__init__(config) - self.bot_token = config.get('bot_token', '') - self.chat_id = config.get('chat_id', '') - - def send(self, alert_message: Dict[str, Any]) -> bool: + self.bot_token: str = str( + config.get("bot_token", self.DEFAULT_BOT_TOKEN) + ) + self.chat_id: str = str(config.get("chat_id", self.DEFAULT_CHAT_ID)) + + def send(self, alert_message: AlertMessage) -> bool: if not self.enabled or not self.bot_token or not self.chat_id: return False - + try: - # 创建Telegram消息 - text = f"📈 *基金告警通知*\n\n" - text += f"*标的:* {alert_message['name']} ({alert_message['code']})\n" - text += f"*当前值:* {alert_message['current_value']}\n" - text += f"*涨跌幅:* {alert_message['change_pct']:.2f}%\n" - text += f"*告警阈值:* {alert_message['direction_text']} {alert_message['threshold']}%\n" - text += f"*触发时间:* {alert_message['timestamp']}\n\n" - text += f"来自 TickEye 基金监测系统" - - # 发送请求 - url = f"https://api.telegram.org/bot{self.bot_token}/sendMessage" + text = self.MESSAGE_TEMPLATE.format(**alert_message) + url = self.API_URL_TEMPLATE.format(bot_token=self.bot_token) payload = { "chat_id": self.chat_id, "text": text, - "parse_mode": "Markdown" + "parse_mode": self.PARSE_MODE, } - - response = requests.post(url, json=payload, timeout=10) - + + response = requests.post( + url, + json=payload, + timeout=self.REQUEST_TIMEOUT, + ) + if response.status_code == 200: logger.info(f"Telegram告警发送成功: {alert_message['code']}") return True - else: - logger.error(f"Telegram告警发送失败: {response.status_code} - {response.text}") - return False - - except Exception as e: - logger.error(f"Telegram告警发送失败: {e}") + + logger.error( + f"Telegram告警发送失败: {response.status_code} - {response.text}" + ) + return False + + except Exception as exc: + logger.error(f"Telegram告警发送失败: {exc}") return False class AlertHistory: """告警历史记录""" - - def __init__(self, history_file: str = "alert_history.json"): - self.history_file = history_file - self.history = self._load_history() - - def _load_history(self) -> Dict[str, List[Dict]]: + + DEFAULT_HISTORY_FILE = "alert_history.json" + DEFAULT_HISTORY: Dict[str, List[HistoryRecord]] = {} + MAX_HISTORY_PER_CODE = 100 + DEFAULT_COOLDOWN_MINUTES = 30 + + def __init__(self, history_file: str = DEFAULT_HISTORY_FILE) -> None: + self.history_file: str = history_file + self.history: Dict[str, List[HistoryRecord]] = self._load_history() + + def _load_history(self) -> Dict[str, List[HistoryRecord]]: """加载历史记录""" if os.path.exists(self.history_file): try: - with open(self.history_file, 'r', encoding='utf-8') as f: - return json.load(f) - except Exception as e: - logger.error(f"加载告警历史失败: {e}") + with open(self.history_file, "r", encoding="utf-8") as file: + loaded = json.load(file) + return dict(loaded) + except Exception as exc: + logger.error(f"加载告警历史失败: {exc}") return {} return {} - - def _save_history(self): + + def _save_history(self) -> None: """保存历史记录""" try: - with open(self.history_file, 'w', encoding='utf-8') as f: - json.dump(self.history, f, ensure_ascii=False, indent=2) - except Exception as e: - logger.error(f"保存告警历史失败: {e}") - - def add_alert(self, code: str, alert_message: Dict[str, Any]): + with open(self.history_file, "w", encoding="utf-8") as file: + json.dump(self.history, file, ensure_ascii=False, indent=2) + except Exception as exc: + logger.error(f"保存告警历史失败: {exc}") + + def add_alert(self, code: str, alert_message: AlertMessage) -> None: """添加告警记录""" if code not in self.history: self.history[code] = [] - - self.history[code].append({ - "timestamp": alert_message['timestamp'], - "change_pct": alert_message['change_pct'], - "threshold": alert_message['threshold'], - "direction": alert_message['direction'] - }) - - # 只保留最近100条记录 - if len(self.history[code]) > 100: - self.history[code] = self.history[code][-100:] - + + self.history[code].append( + { + "timestamp": str(alert_message["timestamp"]), + "change_pct": float(alert_message["change_pct"]), + "threshold": float(alert_message["threshold"]), + "direction": str(alert_message["direction"]), + } + ) + + if len(self.history[code]) > self.MAX_HISTORY_PER_CODE: + self.history[code] = self.history[code][-self.MAX_HISTORY_PER_CODE :] + self._save_history() - - def should_send_alert(self, code: str, cooldown_minutes: int = 30) -> bool: + + def should_send_alert( + self, + code: str, + cooldown_minutes: int = DEFAULT_COOLDOWN_MINUTES, + ) -> bool: """ 检查是否应该发送告警（避免重复通知） - + Args: code: 基金/指数代码 cooldown_minutes: 冷却时间（分钟） - + Returns: bool: 是否应该发送告警 """ if code not in self.history or not self.history[code]: return True - + last_alert = self.history[code][-1] - last_time = datetime.fromisoformat(last_alert['timestamp']) + last_timestamp = last_alert.get("timestamp") + if not isinstance(last_timestamp, str): + return True + + last_time = datetime.fromisoformat(last_timestamp) now = datetime.now() - - # 检查是否在冷却时间内 cooldown_seconds = cooldown_minutes * 60 time_diff = (now - last_time).total_seconds() - + return time_diff > cooldown_seconds - + def get_today_alerts_count(self, code: str) -> int: """获取今天该代码的告警次数""" if code not in self.history: return 0 - + today = datetime.now().date() count = 0 - + for alert in self.history[code]: - alert_date = datetime.fromisoformat(alert['timestamp']).date() + timestamp = alert.get("timestamp") + if not isinstance(timestamp, str): + continue + alert_date = datetime.fromisoformat(timestamp).date() if alert_date == today: count += 1 - + return count class AlertManager: """告警管理器""" - - def __init__(self, config_file: str = "alerts_config.json"): - self.config_file = config_file - self.config = self._load_config() - self.rules = self._load_rules() - self.channels = self._load_channels() - self.history = AlertHistory() - self.global_settings = self.config.get('global_settings', {}) - - def _load_config(self) -> Dict[str, Any]: + + DEFAULT_CONFIG_FILE = "alerts_config.json" + DEFAULT_ENABLED = False + DEFAULT_CONFIG = {"enabled": False} + DEFAULT_GLOBAL_SETTINGS: ConfigDict = {} + DEFAULT_COOLDOWN_MINUTES = 30 + DEFAULT_MAX_ALERTS_PER_DAY = 10 + DEFAULT_SUMMARY_TIME = "18:00" + GLOBAL_SETTINGS_KEY = "global_settings" + RULES_KEY = "rules" + NOTIFICATION_CHANNELS_KEY = "notification_channels" + SUMMARY_ENABLED_KEY = "enable_daily_summary" + SUMMARY_TIME_KEY = "summary_time" + COOLDOWN_KEY = "cooldown_minutes" + MAX_ALERTS_PER_DAY_KEY = "max_alerts_per_day" + + CHANNEL_EMAIL = "email" + CHANNEL_FEISHU = "feishu" + CHANNEL_WECHAT_WORK = "wechat_work" + CHANNEL_DINGTALK = "dingtalk" + CHANNEL_TELEGRAM = "telegram" + + def __init__(self, config_file: str = DEFAULT_CONFIG_FILE) -> None: + self.config_file: str = config_file + self.config: ConfigDict = self._load_config() + self.rules: List[AlertRule] = self._load_rules() + self.channels: Dict[str, NotificationChannel] = self._load_channels() + self.history: AlertHistory = AlertHistory() + self.global_settings: ConfigDict = self.config.get( + self.GLOBAL_SETTINGS_KEY, + self.DEFAULT_GLOBAL_SETTINGS, + ) + + def _load_config(self) -> ConfigDict: """加载配置文件""" if not os.path.exists(self.config_file): logger.warning(f"告警配置文件不存在: {self.config_file}") - return {"enabled": False} - + return dict(self.DEFAULT_CONFIG) + try: - with open(self.config_file, 'r', encoding='utf-8') as f: - return json.load(f) - except Exception as e: - logger.error(f"加载告警配置失败: {e}") - return {"enabled": False} - + with open(self.config_file, "r", encoding="utf-8") as file: + return json.load(file) + except Exception as exc: + logger.error(f"加载告警配置失败: {exc}") + return dict(self.DEFAULT_CONFIG) + def _load_rules(self) -> List[AlertRule]: """加载告警规则""" - rules_config = self.config.get('rules', []) - rules = [] - + rules_config: List[ConfigDict] = list( + self.config.get(self.RULES_KEY, []) + ) + rules: List[AlertRule] = [] + for rule_config in rules_config: try: rule = AlertRule(rule_config) if rule.enabled: rules.append(rule) - except Exception as e: - logger.error(f"加载告警规则失败: {e}") - + except Exception as exc: + logger.error(f"加载告警规则失败: {exc}") + logger.info(f"加载了 {len(rules)} 个告警规则") return rules - + def _load_channels(self) -> Dict[str, NotificationChannel]: """加载通知渠道""" - channels_config = self.config.get('notification_channels', {}) - channels = {} - - # 邮件渠道 - if 'email' in channels_config: - channels['email'] = EmailNotification(channels_config['email']) - - # 飞书渠道 - if 'feishu' in channels_config: - channels['feishu'] = FeishuNotification(channels_config['feishu']) - - # 企业微信渠道 - if 'wechat_work' in channels_config: - channels['wechat_work'] = WechatWorkNotification(channels_config['wechat_work']) - - # 钉钉渠道 - if 'dingtalk' in channels_config: - channels['dingtalk'] = DingTalkNotification(channels_config['dingtalk']) - - # Telegram渠道 - if 'telegram' in channels_config: - channels['telegram'] = TelegramNotification(channels_config['telegram']) - + channels_config: ConfigDict = self.config.get( + self.NOTIFICATION_CHANNELS_KEY, + {}, + ) + channels: Dict[str, NotificationChannel] = {} + + if self.CHANNEL_EMAIL in channels_config: + channels[self.CHANNEL_EMAIL] = EmailNotification( + channels_config[self.CHANNEL_EMAIL] + ) + + if self.CHANNEL_FEISHU in channels_config: + channels[self.CHANNEL_FEISHU] = FeishuNotification( + channels_config[self.CHANNEL_FEISHU] + ) + + if self.CHANNEL_WECHAT_WORK in channels_config: + channels[self.CHANNEL_WECHAT_WORK] = WechatWorkNotification( + channels_config[self.CHANNEL_WECHAT_WORK] + ) + + if self.CHANNEL_DINGTALK in channels_config: + channels[self.CHANNEL_DINGTALK] = DingTalkNotification( + channels_config[self.CHANNEL_DINGTALK] + ) + + if self.CHANNEL_TELEGRAM in channels_config: + channels[self.CHANNEL_TELEGRAM] = TelegramNotification( + channels_config[self.CHANNEL_TELEGRAM] + ) + logger.info(f"加载了 {len(channels)} 个通知渠道") return channels - - def check_alerts(self, fund_data: Dict[str, Dict]) -> List[Dict[str, Any]]: + + def check_alerts(self, fund_data: FundData) -> List[AlertMessage]: """ 检查所有基金数据，触发告警 - + Args: fund_data: 基金数据字典，格式为 {code: {name, price, change}} - + Returns: - List: 触发的告警列表 + List[AlertMessage]: 触发的告警列表 """ - if not self.config.get('enabled', False): + if not self.config.get("enabled", self.DEFAULT_ENABLED): logger.info("告警功能已禁用") return [] - - triggered_alerts = [] - + + triggered_alerts: List[AlertMessage] = [] + for rule in self.rules: code = rule.code - + if code not in fund_data: logger.debug(f"规则 {code} 对应的基金数据不存在") continue - + data = fund_data[code] - change_pct = data.get('change', 0.0) - current_value = data.get('price', 0.0) - - # 检查是否应该触发告警 + change_pct = float(data.get("change", 0.0)) + current_value = float(data.get("price", 0.0)) + if rule.should_alert(change_pct): - # 检查冷却时间 - cooldown = self.global_settings.get('cooldown_minutes', 30) + cooldown = int( + self.global_settings.get( + self.COOLDOWN_KEY, + self.DEFAULT_COOLDOWN_MINUTES, + ) + ) if not self.history.should_send_alert(code, cooldown): logger.info(f"代码 {code} 仍在冷却时间内，跳过告警") continue - - # 检查每日最大告警次数 - max_per_day = self.global_settings.get('max_alerts_per_day', 10) + + max_per_day = int( + self.global_settings.get( + self.MAX_ALERTS_PER_DAY_KEY, + self.DEFAULT_MAX_ALERTS_PER_DAY, + ) + ) today_count = self.history.get_today_alerts_count(code) if today_count >= max_per_day: - logger.warning(f"代码 {code} 今日告警次数已达上限 ({max_per_day}次)") + logger.warning( + f"代码 {code} 今日告警次数已达上限 ({max_per_day}次)" + ) continue - - # 创建告警消息 + alert_message = rule.get_alert_message(change_pct, current_value) - - # 发送通知 - sent_channels = [] + + sent_channels: List[str] = [] for channel_name in rule.notification_channels: - if channel_name in self.channels: - channel = self.channels[channel_name] - if channel.send(alert_message): - sent_channels.append(channel_name) - + channel: Optional[NotificationChannel] = self.channels.get( + channel_name + ) + if channel and channel.send(alert_message): + sent_channels.append(channel_name) + if sent_channels: - alert_message['sent_channels'] = sent_channels + alert_message["sent_channels"] = sent_channels triggered_alerts.append(alert_message) - - # 记录到历史 self.history.add_alert(code, alert_message) - - logger.info(f"代码 {code} 触发告警，已通过 {', '.join(sent_channels)} 发送通知") - + + logger.info( + f"代码 {code} 触发告警，已通过 " + f"{', '.join(sent_channels)} 发送通知" + ) + return triggered_alerts - - def send_daily_summary(self, fund_data: Dict[str, Dict], triggered_alerts: List[Dict[str, Any]]) -> bool: + + def send_daily_summary( + self, + fund_data: FundData, + triggered_alerts: List[AlertMessage], + ) -> bool: """ 发送每日汇总报告 - + Args: fund_data: 所有基金数据 triggered_alerts: 今日触发的告警列表 - + Returns: bool: 是否发送成功 """ - if not self.global_settings.get('enable_daily_summary', False): + if not self.global_settings.get(self.SUMMARY_ENABLED_KEY, False): return False - - # 检查是否应该发送每日汇总 - summary_time = self.global_settings.get('summary_time', '18:00') - current_time = datetime.now().strftime('%H:%M') - + + summary_time = str( + self.global_settings.get( + self.SUMMARY_TIME_KEY, + self.DEFAULT_SUMMARY_TIME, + ) + ) + current_time = datetime.now().strftime("%H:%M") + if current_time != summary_time: return False - - # 创建汇总消息 + total_funds = len(fund_data) total_alerts = len(triggered_alerts) - - summary_message = { + + summary_message: AlertMessage = { "type": "daily_summary", - "date": datetime.now().strftime('%Y-%m-%d'), + "date": datetime.now().strftime("%Y-%m-%d"), "total_funds": total_funds, "total_alerts": total_alerts, "triggered_alerts": triggered_alerts, - "timestamp": datetime.now().isoformat() + "timestamp": datetime.now().isoformat(), } - - # 发送到所有启用的渠道 + success = False for channel_name, channel in self.channels.items(): if channel.enabled: - # 这里需要为每个渠道实现汇总消息的发送 - # 暂时只记录日志 logger.info(f"发送每日汇总到 {channel_name}") - + + _ = summary_message return success -# 全局告警管理器实例 -_alert_manager_instance = None +_alert_manager_instance: Optional[AlertManager] = None + def get_alert_manager() -> AlertManager: """获取全局告警管理器实例""" @@ -670,29 +852,28 @@ def get_alert_manager() -> AlertManager: return _alert_manager_instance -def check_and_send_alerts(fund_data: Dict[str, Dict]) -> List[Dict[str, Any]]: +def check_and_send_alerts(fund_data: FundData) -> List[AlertMessage]: """ 检查并发送告警（便捷函数） - + Args: fund_data: 基金数据字典 - + Returns: - List: 触发的告警列表 + List[AlertMessage]: 触发的告警列表 """ manager = get_alert_manager() return manager.check_alerts(fund_data) if __name__ == "__main__": - # 测试代码 - test_data = { + test_data: FundData = { "000001": {"name": "上证指数", "price": 3200.0, "change": 5.5}, "NDX": {"name": "纳斯达克", "price": 18000.0, "change": 2.8}, - "007360": {"name": "测试基金", "price": 1.2, "change": -3.5} + "007360": {"name": "测试基金", "price": 1.2, "change": -3.5}, } - + alerts = check_and_send_alerts(test_data) print(f"触发了 {len(alerts)} 个告警") for alert in alerts: - print(f"告警: {alert['message']}") \ No newline at end of file + print(f"告警: {alert['message']}")