Stock Monitor

config.py
*.log
__pycache__/
*.pyc
.DS_Store
.vscode/
.idea/

MIT License

Copyright (c) 2026 Stock Monitor Pro

Permission is hereby granted...

#!/usr/bin/env python3
"""
Stock Monitor Pro - 智能分析引擎
集成：新闻、资金流向、龙虎榜、宏观关联分析
"""

import requests
import json
import re
from datetime import datetime, timedelta
from typing import List, Dict, Optional

class StockAnalyser:
    """股票智能分析器 - 结合多维度数据给出建议"""
    
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        })
    
    # ========== 1. 新闻舆情 ==========
    
    def fetch_eastmoney_news(self, symbol: str, name: str, limit: int = 5) -> List[Dict]:
        """获取东方财富个股新闻"""
        url = f"https://searchapi.eastmoney.com/api/suggest/get"
        params = {
            "input": name,
            "type": 14,
            "count": limit
        }
        try:
            resp = self.session.get(url, params=params, timeout=10)
            data = resp.json()
            news_list = []
            for item in data.get("QuotationCodeTable", {}).get("Data", []):
                news_list.append({
                    "title": item.get("Title", ""),
                    "url": item.get("Url", ""),
                    "time": item.get("ShowTime", "")
                })
            return news_list
        except Exception as e:
            return []
    
    def fetch_sina_news(self, symbol: str, name: str) -> List[Dict]:
        """获取新浪财经个股新闻"""
        # 新浪新闻搜索接口
        url = f"https://search.sina.com.cn/?q={name}&c=news&sort=time"
        try:
            resp = self.session.get(url, timeout=10)
            # 这里可以做更精细的HTML解析
            # 简化返回示例
            return [{"title": f"新浪财经-{name}相关新闻", "source": "新浪"}]
        except:
            return []
    
    def analyze_sentiment(self, news_list: List[Dict]) -> Dict:
        """简单情感分析"""
        positive_words = ['利好', '增长', '突破', '买入', '增持', '涨停', '超预期', '业绩大增']
        negative_words = ['利空', '减持', '下跌', '卖出', '亏损', '暴雷', '跌停', '不及预期']
        
        sentiment = {"positive": 0, "negative": 0, "neutral": 0, "summary": []}
        
        for news in news_list:
            title = news.get("title", "")
            p_count = sum(1 for w in positive_words if w in title)
            n_count = sum(1 for w in negative_words if w in title)
            
            if p_count > n_count:
                sentiment["positive"] += 1
            elif n_count > p_count:
                sentiment["negative"] += 1
            else:
                sentiment["neutral"] += 1
        
        # 生成情感摘要
        if sentiment["positive"] > sentiment["negative"]:
            sentiment["overall"] = "偏多"
        elif sentiment["negative"] > sentiment["positive"]:
            sentiment["overall"] = "偏空"
        else:
            sentiment["overall"] = "中性"
            
        return sentiment
    
    # ========== 2. 资金流向 ==========
    
    def fetch_fund_flow(self, symbol: str, market: str = "sz") -> Dict:
        """获取个股资金流向 (新浪财经)"""
        # 新浪资金流向接口
        code = f"{market}{symbol}"
        url = f"https://quotes.sina.cn/cn/api/quotes.php?symbol={code}&source=sina"
        
        try:
            resp = self.session.get(url, timeout=10)
            # 解析返回数据
            return {
                "main_inflow": "数据获取中...",
                "retail_inflow": "数据获取中...",
                "net_inflow": "数据获取中..."
            }
        except:
            return {"error": "获取失败"}
    
    def fetch_northbound_flow(self) -> Dict:
        """获取北向资金 (沪深股通) 流向"""
        url = "https://push2.eastmoney.com/api/qt/stock/get"
        params = {"secid": "1.000001", "fields": "f170"}  # 简化示例
        try:
            resp = self.session.get(url, params=params, timeout=10)
            return {"northbound": "北向资金数据获取中..."}
        except:
            return {}
    
    # ========== 3. 龙虎榜 ==========
    
    def fetch_dragon_tiger(self, date: str = None) -> List[Dict]:
        """获取龙虎榜数据"""
        if not date:
            date = datetime.now().strftime("%Y%m%d")
        
        url = f"http://datacenter-web.eastmoney.com/api/data/v1/get"
        params = {
            "sortColumns": "NET_BUY_AMT",
            "sortTypes": "-1",
            "pageSize": "50",
            "pageNumber": "1",
            "reportName": "RPT_DMSK_TS",
            "columns": "ALL",
            "filter": f"(TRADE_DATE='{date}')"
        }
        
        try:
            resp = self.session.get(url, params=params, timeout=10)
            data = resp.json()
            return data.get("result", {}).get("data", [])
        except:
            return []
    
    # ========== 4. 宏观关联分析 ==========
    
    def analyze_gold_correlation(self, gold_price: float, stocks: List[Dict]) -> str:
        """分析金价与持仓股票的关联"""
        # 江西铜业等有色股与金价正相关
        correlation_map = {
            "600362": "强正相关",  # 江西铜业
            "601318": "弱相关",    # 中国平安
            "513180": "弱负相关",  # 恒生科技
            "159892": "弱相关",    # 恒生医疗
        }
        
        analysis = []
        for stock in stocks:
            code = stock.get("code")
            corr = correlation_map.get(code, "未知")
            if corr in ["强正相关", "中等正相关"]:
                analysis.append(f"📈 {stock['name']}: 与金价{corr}，金价上涨可能带动该股")
        
        return "\n".join(analysis) if analysis else "暂无强关联标的"
    
    # ========== 5. 综合分析 ==========
    
    def generate_insight(self, stock: Dict, price_data: Dict, alerts: List) -> str:
        """生成综合分析报告"""
        code = stock['code']
        name = stock['name']
        
        # 1. 获取新闻
        news_list = self.fetch_eastmoney_news(code, name)
        sentiment = self.analyze_sentiment(news_list)
        
        # 2. 资金流向
        fund_flow = self.fetch_fund_flow(code, stock.get('market', 'sz'))
        
        # 3. 构建报告
        report = f"""📊 <b>{name} ({code}) 深度分析</b>

💰 <b>价格异动:</b>
• 当前: {price_data.get('price', 'N/A')} ({price_data.get('change_pct', 0):+.2f}%)
• 触发: {', '.join([a[1] for a in alerts])}

📰 <b>舆情分析 ({sentiment.get('overall', '未知')}):</b>
• 最近新闻: {len(news_list)} 条
• 正面: {sentiment.get('positive', 0)} | 负面: {sentiment.get('negative', 0)}
"""
        
        # 添加最新新闻标题
        if news_list:
            report += "\n<b>最新动态:</b>\n"
            for n in news_list[:2]:
                report += f"• {n.get('title', '无标题')[:30]}...\n"
        
        # 4. 给出建议
        suggestion = self._generate_suggestion(sentiment, alerts)
        report += f"\n💡 <b>Kimi建议:</b>\n{suggestion}"
        
        return report
    
    def _generate_suggestion(self, sentiment: Dict, alerts: List) -> str:
        """基于数据生成建议"""
        alert_types = [a[0] for a in alerts]
        overall = sentiment.get("overall", "中性")
        
        # 价格下跌 + 舆情偏空 = 谨慎
        if "below" in alert_types and overall == "偏空":
            return "⚠️ 价格跌破支撑位，且舆情偏空，建议观察等待，不急于抄底。"
        
        # 价格下跌 + 舆情偏多 = 可能是机会
        if "below" in alert_types and overall == "偏多":
            return "🔍 价格下跌但舆情偏多，可能是情绪错杀，关注是否有反弹机会。"
        
        # 价格突破 + 舆情偏多 = 确认趋势
        if "above" in alert_types and overall == "偏多":
            return "🚀 价格突破且舆情配合，趋势可能延续，可考虑顺势而为。"
        
        # 大涨
        if "pct_up" in alert_types:
            return "📈 短期涨幅较大，注意获利了结风险。"
        
        # 大跌
        if "pct_down" in alert_types:
            return "📉 短期跌幅较大，关注是否超跌反弹，但勿急于抄底。"
        
        return "⏳ 建议保持观察，等待更明确信号。"


# ========== 测试 ==========
if __name__ == '__main__':
    analyser = StockAnalyser()
    
    # 测试新闻抓取
    print("=== 新闻测试 ===")
    news = analyser.fetch_eastmoney_news("600362", "江西铜业")
    print(f"获取到 {len(news)} 条新闻")
    for n in news[:3]:
        print(f"  - {n.get('title', 'N/A')[:40]}...")
    
    # 测试情感分析
    print("\n=== 情感分析测试 ===")
    sentiment = analyser.analyze_sentiment(news)
    print(f"整体情绪: {sentiment.get('overall')}")
    print(f"正面: {sentiment.get('positive')}, 负面: {sentiment.get('negative')}")
    
    # 测试金价关联
    print("\n=== 宏观关联测试 ===")
    stocks = [{"code": "600362", "name": "江西铜业"}]
    corr = analyser.analyze_gold_correlation(2743, stocks)
    print(corr)

#!/bin/bash
# Stock Monitor 一键启动脚本

SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
LOG_DIR="$HOME/.stock_monitor"
PID_FILE="$LOG_DIR/monitor.pid"

case "$1" in
    start)
        if [ -f "$PID_FILE" ] && kill -0 $(cat "$PID_FILE") 2>/dev/null; then
            echo "⚠️  监控进程已在运行 (PID: $(cat $PID_FILE))"
            exit 1
        fi
        
        echo "🚀 启动 Stock Monitor 后台进程..."
        mkdir -p "$LOG_DIR"
        nohup python3 "$SCRIPT_DIR/monitor_daemon.py" > "$LOG_DIR/monitor.log" 2>&1 &
        echo $! > "$PID_FILE"
        echo "✅ 已启动 (PID: $!)"
        echo "📋 日志: $LOG_DIR/monitor.log"
        ;;
        
    stop)
        if [ -f "$PID_FILE" ]; then
            PID=$(cat "$PID_FILE")
            if kill -0 "$PID" 2>/dev/null; then
                echo "🛑 停止监控进程 (PID: $PID)..."
                kill "$PID"
                rm "$PID_FILE"
                echo "✅ 已停止"
            else
                echo "⚠️  进程不存在"
                rm "$PID_FILE"
            fi
        else
            echo "⚠️  没有运行中的进程"
        fi
        ;;
        
    status)
        if [ -f "$PID_FILE" ] && kill -0 $(cat "$PID_FILE") 2>/dev/null; then
            echo "✅ 监控运行中 (PID: $(cat $PID_FILE))"
            echo "📋 最近日志:"
            tail -5 "$LOG_DIR/monitor.log" 2>/dev/null || echo "  暂无日志"
        else
            echo "⏹️  监控未运行"
        fi
        ;;
        
    log)
        tail -f "$LOG_DIR/monitor.log"
        ;;
        
    *)
        echo "Stock Monitor 控制脚本"
        echo ""
        echo "用法: ./control.sh [start|stop|status|log]"
        echo ""
        echo "  start   - 启动后台监控"
        echo "  stop    - 停止监控"
        echo "  status  - 查看状态"
        echo "  log     - 查看实时日志"
        ;;
esac

#!/usr/bin/env python3
"""
Stock Monitor Daemon - 后台常驻进程
自动运行监控，智能控制频率，支持 graceful shutdown
"""

import sys
import time
import signal
import logging
from datetime import datetime
from pathlib import Path

# 设置日志
log_dir = Path.home() / ".stock_monitor"
log_dir.mkdir(exist_ok=True)
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(log_dir / "monitor.log"),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

# 导入监控类
sys.path.insert(0, str(Path(__file__).parent))
from monitor import StockAlert, WATCHLIST

class MonitorDaemon:
    def __init__(self):
        self.monitor = StockAlert()
        self.running = True
        self.last_run_time = 0
        
        # 设置信号处理
        signal.signal(signal.SIGTERM, self.handle_shutdown)
        signal.signal(signal.SIGINT, self.handle_shutdown)
    
    def handle_shutdown(self, signum, frame):
        """优雅退出"""
        logger.info(f"收到信号 {signum}，正在关闭...")
        self.running = False
    
    def get_sleep_interval(self):
        """根据当前时间获取睡眠间隔"""
        schedule = self.monitor.should_run_now()
        if not schedule.get("run"):
            # 如果当前不需要运行，计算到下次运行的时间
            now = datetime.now()
            hour = now.hour
            
            # 凌晨时段，1小时后检查
            if 0 <= hour < 9:
                return 3600
            return 300  # 默认5分钟
        
        return schedule.get("interval", 300)
    
    def run(self):
        """主循环"""
        logger.info("=" * 60)
        logger.info("🚀 Stock Monitor Daemon 启动")
        logger.info(f"📋 监控标的: {len(WATCHLIST)} 只")
        logger.info("=" * 60)
        
        while self.running:
            try:
                # 检查是否应该执行
                schedule = self.monitor.should_run_now()
                
                if schedule.get("run"):
                    mode = schedule.get("mode", "normal")
                    stocks_count = len(schedule.get("stocks", []))
                    logger.info(f"[{mode}] 扫描 {stocks_count} 只标的...")
                    
                    # 执行监控
                    alerts = self.monitor.run_once(smart_mode=False)  # 已经判断过了
                    
                    if alerts:
                        logger.info(f"⚠️ 触发 {len(alerts)} 条预警")
                        # 这里会通过 message 工具发送通知
                    else:
                        logger.debug("✅ 无预警")
                    
                    self.last_run_time = time.time()
                
                # 计算睡眠间隔
                sleep_interval = self.get_sleep_interval()
                logger.debug(f"下次检查: {sleep_interval} 秒后")
                
                # 分段睡眠，方便及时响应退出信号
                slept = 0
                while slept < sleep_interval and self.running:
                    time.sleep(1)
                    slept += 1
                    
            except Exception as e:
                logger.error(f"运行出错: {e}", exc_info=True)
                time.sleep(60)  # 出错后等待1分钟重试
        
        logger.info("👋 Daemon 已停止")

if __name__ == '__main__':
    daemon = MonitorDaemon()
    daemon.run()

#!/usr/bin/env python3
"""
自选股监控预警工具 - OpenClaw集成版
支持 A股、ETF 及 国际现货黄金 (伦敦金)
"""

import requests
import json
import time
import os
from datetime import datetime
from pathlib import Path

# ============ 配置区 ============

# 监控列表 - 长期挂机通用配置
# 注意: 伦敦金使用新浪hf_XAU接口，价格为 人民币/克 (约4800元/克 = $2740/盎司)
# 
# 预警规则设计原则 (适合长期挂机):
# 1. 成本百分比预警: 基于持仓成本设置 ±10%/±15% 预警，比固定价格更合理
# 2. 单日涨跌幅预警: 
#    - 个股 ±3%~5% (波动大)
#    - ETF ±1.5%~2.5% (波动小)
#    - 黄金 ±2%~3% (24H特殊)
# 3. 防骚扰: 同类预警30分钟内只发一次

# 标的类型定义
STOCK_TYPE = {
    "INDIVIDUAL": "individual",  # 个股
    "ETF": "etf",                # ETF
    "GOLD": "gold"               # 黄金/贵金属
}

WATCHLIST = [
    # ===== 个股: 波动较大，设置较宽的涨跌预警 =====
    {
        "code": "600362", 
        "name": "江西铜业", 
        "market": "sh",
        "type": "individual",
        "cost": 57.00,
        "alerts": {
            "cost_pct_above": 15.0,    # 盈利15%
            "cost_pct_below": -12.0,   # 止损12%
            "change_pct_above": 4.0,   # 日内异动 ±4%
            "change_pct_below": -4.0,
            "volume_surge": 2.0        # 成交量是5日均量2倍
        }
    },
    {
        "code": "601318", 
        "name": "中国平安", 
        "market": "sh",
        "type": "individual",
        "cost": 66.00,
        "alerts": {
            "cost_pct_above": 12.0,
            "cost_pct_below": -10.0,
            "change_pct_above": 3.5,   # 日内异动 ±3.5%
            "change_pct_below": -3.5,
            "volume_surge": 2.0
        }
    },
    # ===== ETF: 波动相对较小，设置更敏感的预警 =====
    {
        "code": "159892", 
        "name": "恒生医疗", 
        "market": "sz",
        "type": "etf",
        "cost": 0.80,
        "alerts": {
            "cost_pct_above": 15.0,
            "cost_pct_below": -15.0,
            "change_pct_above": 2.0,   # ETF日内异动 ±2%
            "change_pct_below": -2.0,
            "volume_surge": 1.8        # ETF放量阈值更低
        }
    },
    {
        "code": "513180", 
        "name": "恒生科技", 
        "market": "sh",
        "type": "etf",
        "cost": 0.72,
        "alerts": {
            "cost_pct_above": 15.0,
            "cost_pct_below": -15.0,
            "change_pct_above": 2.0,   # ETF日内异动 ±2%
            "change_pct_below": -2.0,
            "volume_surge": 1.8
        }
    },
    {
        "code": "159681", 
        "name": "创50ETF", 
        "market": "sz",
        "type": "etf",
        "cost": 1.50,
        "alerts": {
            "cost_pct_above": 12.0,
            "cost_pct_below": -12.0,
            "change_pct_above": 2.0,   # ETF日内异动 ±2%
            "change_pct_below": -2.0,
            "volume_surge": 1.8
        }
    },
    {
        "code": "516020", 
        "name": "化工50ETF", 
        "market": "sh",
        "type": "etf",
        "cost": 0.90,
        "alerts": {
            "cost_pct_above": 12.0,
            "cost_pct_below": -12.0,
            "change_pct_above": 2.0,   # ETF日内异动 ±2%
            "change_pct_below": -2.0,
            "volume_surge": 1.8
        }
    },
    # ===== 伦敦金: 24H特殊标的 =====
    {
        "code": "XAU", 
        "name": "伦敦金(人民币/克)", 
        "market": "fx",
        "type": "gold",
        "cost": 4650.0,
        "alerts": {
            "cost_pct_above": 10.0,    # 盈利10%
            "cost_pct_below": -8.0,    # 止损8%
            "change_pct_above": 2.5,   # 黄金日内异动 ±2.5%
            "change_pct_below": -2.5
            # 黄金不监控成交量 (外汇市场无成交量概念)
        }
    }
]

# 智能频率配置
SMART_SCHEDULE = {
    "market_open": {"hours": [(9, 30), (11, 30), (13, 0), (15, 0)], "interval": 300},  # 交易时间: 5分钟
    "after_hours": {"interval": 1800},  # 收盘后: 30分钟
    "night": {"hours": [(0, 0), (8, 0)], "interval": 3600},  # 凌晨: 1小时(仅伦敦金)
}

# ============ 核心代码 ============

class StockAlert:
    def __init__(self):
        self.prev_data = {}
        self.alert_log = []
        self.session = requests.Session()
        self.session.headers.update({"User-Agent": "Mozilla/5.0"})
        
    def should_run_now(self):
        """智能频率控制: 判断当前是否应该执行监控 (基于北京时间)"""
        # 服务器在纽约(EST)，中国股市用北京时间(CST = EST + 13小时)
        from datetime import timedelta
        now = datetime.now() + timedelta(hours=13)  # 转换成北京时间
        hour, minute = now.hour, now.minute
        time_val = hour * 100 + minute
        weekday = now.weekday()
        
        # 周末只监控伦敦金
        if weekday >= 5:  # 周六日
            return {"run": True, "mode": "weekend", "stocks": [s for s in WATCHLIST if s['market'] == 'fx']}
        
        # 交易时间 (9:30-11:30, 13:00-15:00)
        morning_session = 930 <= time_val <= 1130
        afternoon_session = 1300 <= time_val <= 1500
        
        if morning_session or afternoon_session:
            return {"run": True, "mode": "market", "stocks": WATCHLIST, "interval": 300}
        
        # 午休 (11:30-13:00)
        if 1130 < time_val < 1300:
            return {"run": True, "mode": "lunch", "stocks": WATCHLIST, "interval": 600}  # 10分钟
        
        # 收盘后 (15:00-24:00)
        if 1500 <= time_val <= 2359:
            return {"run": True, "mode": "after_hours", "stocks": WATCHLIST, "interval": 1800}  # 30分钟
        
        # 凌晨 (0:00-9:30)
        if 0 <= time_val < 930:
            return {"run": True, "mode": "night", "stocks": [s for s in WATCHLIST if s['market'] == 'fx'], "interval": 3600}  # 1小时
        
        return {"run": False}

    def fetch_eastmoney_kline(self, symbol, market):
        """获取最新日K线数据 (收盘后也能获取收盘价)"""
        secid = f"{market}.{symbol}"
        url = "https://push2his.eastmoney.com/api/qt/stock/kline/get"
        params = {
            'secid': secid,
            'fields1': 'f1,f2,f3,f4,f5,f6',
            'fields2': 'f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61',
            'klt': '101',  # 日线
            'fqt': '0',
            'end': '20500101',
            'lmt': '2'  # 取最近2天，用于计算涨跌幅
        }
        try:
            resp = self.session.get(url, params=params, timeout=10)
            data = resp.json()
            klines = data.get('data', {}).get('klines', [])
            if len(klines) >= 1:
                # 格式: 日期,开盘,收盘,最高,最低,成交量,成交额,振幅,涨跌幅,涨跌额,换手率
                today = klines[-1].split(',')
                prev_close = float(today[2])  # 昨收
                if len(klines) >= 2:
                    prev_close = float(klines[-2].split(',')[2])  # 前一天收盘
                return {
                    'name': data.get('data', {}).get('name', symbol),
                    'price': float(today[2]),      # 收盘
                    'prev_close': prev_close,
                    'volume': int(float(today[5])),
                    'amount': float(today[6]),
                    'date': today[0],
                    'time': '15:00:00'
                }
        except Exception as e:
            print(f"东财K线获取失败 {symbol}: {e}")
        return None

    def fetch_volume_ma5(self, symbol, market):
        """获取5日平均成交量"""
        secid = f"{market}.{symbol}"
        url = "https://push2his.eastmoney.com/api/qt/stock/kline/get"
        params = {
            'secid': secid,
            'fields1': 'f1,f2,f3,f4,f5,f6',
            'fields2': 'f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61',
            'klt': '101',
            'fqt': '0',
            'end': '20500101',
            'lmt': '6'  # 取最近6天(今天+前5天)
        }
        try:
            resp = self.session.get(url, params=params, timeout=10)
            data = resp.json()
            klines = data.get('data', {}).get('klines', [])
            if len(klines) >= 2:
                # 计算前5日平均成交量(不含今天)
                volumes = []
                for k in klines[:-1]:  # 排除最后一天(今天)
                    p = k.split(',')
                    volumes.append(float(p[5]))  # 成交量
                return sum(volumes) / len(volumes) if volumes else 0
        except Exception as e:
            print(f"获取均量失败 {symbol}: {e}")
        return 0

    def fetch_ma_data(self, symbol, market):
        """获取均线数据 (MA5, MA10, MA20) 和 RSI"""
        secid = f"{market}.{symbol}"
        url = "https://push2his.eastmoney.com/api/qt/stock/kline/get"
        params = {
            'secid': secid,
            'fields1': 'f1,f2,f3,f4,f5,f6',
            'fields2': 'f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61',
            'klt': '101',
            'fqt': '0',
            'end': '20500101',
            'lmt': '30'  # 取最近30天计算MA20和RSI
        }
        try:
            resp = self.session.get(url, params=params, timeout=10)
            data = resp.json()
            klines = data.get('data', {}).get('klines', [])
            if len(klines) >= 20:
                closes = []
                for k in klines:
                    p = k.split(',')
                    closes.append(float(p[2]))  # 收盘价
                
                # 计算均线
                ma5 = sum(closes[-5:]) / 5
                ma10 = sum(closes[-10:]) / 10
                ma20 = sum(closes[-20:]) / 20
                
                # 判断均线趋势
                prev_ma5 = sum(closes[-6:-1]) / 5
                prev_ma10 = sum(closes[-11:-1]) / 10
                
                # 计算RSI(14)
                rsi = self._calculate_rsi(closes, 14)
                
                return {
                    'MA5': ma5,
                    'MA10': ma10,
                    'MA20': ma20,
                    'MA5_trend': 'up' if ma5 > prev_ma5 else 'down',
                    'MA10_trend': 'up' if ma10 > prev_ma10 else 'down',
                    'golden_cross': prev_ma5 <= prev_ma10 and ma5 > ma10,
                    'death_cross': prev_ma5 >= prev_ma10 and ma5 < ma10,
                    'RSI': rsi,
                    'RSI_overbought': rsi > 70 if rsi else False,
                    'RSI_oversold': rsi < 30 if rsi else False
                }
        except Exception as e:
            print(f"获取均线失败 {symbol}: {e}")
        return None
    
    def _calculate_rsi(self, closes, period=14):
        """计算RSI指标"""
        if len(closes) < period + 1:
            return None
        
        gains = []
        losses = []
        
        for i in range(1, period + 1):
            change = closes[-i] - closes[-i-1]
            if change > 0:
                gains.append(change)
                losses.append(0)
            else:
                gains.append(0)
                losses.append(abs(change))
        
        avg_gain = sum(gains) / period
        avg_loss = sum(losses) / period
        
        if avg_loss == 0:
            return 100
        
        rs = avg_gain / avg_loss
        rsi = 100 - (100 / (1 + rs))
        return round(rsi, 2)

    def fetch_sina_realtime(self, stocks):
        """获取实时行情 (优先实时，收盘后用日K)"""
        stock_list = [s for s in stocks if s['market'] != 'fx']
        fx_list = [s for s in stocks if s['market'] == 'fx']
        results = {}
        
        # 1. A股/ETF - 尝试实时接口
        if stock_list:
            codes = [f"{s['market']}{s['code']}" for s in stock_list]
            url = f"https://hq.sinajs.cn/list={','.join(codes)}"
            try:
                resp = self.session.get(url, headers={'Referer': 'https://finance.sina.com.cn'}, timeout=10)
                resp.encoding = 'gb18030'
                for line in resp.text.strip().split(';'):
                    if 'hq_str_' not in line or '=' not in line: continue
                    key = line.split('=')[0].split('_')[-1]
                    if len(key) < 8: continue
                    data_str = line[line.index('"')+1 : line.rindex('"')]
                    p = data_str.split(',')
                    if len(p) > 30 and float(p[3]) > 0:
                        # 新浪数据格式: 名称,今日开盘,昨日收盘,当前价,今日最高,今日最低,竞买价,竞卖价,成交量,成交额...
                        # 保存昨日最高最低价用于跳空检测 (用昨日收盘近似，或用均线数据补充)
                        results[key[2:]] = {
                            'name': p[0], 
                            'price': float(p[3]), 
                            'prev_close': float(p[2]),
                            'open': float(p[1]),      # 今日开盘
                            'high': float(p[4]),      # 今日最高
                            'low': float(p[5]),       # 今日最低
                            'volume': int(p[8]), 
                            'amount': float(p[9]), 
                            'date': p[30], 
                            'time': p[31],
                            'prev_high': float(p[2]) * 1.02,  # 估算昨日最高 (昨收+2%)
                            'prev_low': float(p[2]) * 0.98    # 估算昨日最低 (昨收-2%)
                        }
            except Exception as e: 
                print(f"实时行情获取失败: {e}")
            
            # 2. 如果实时接口返回空或0，用日K线补数据
            for stock in stock_list:
                code = stock['code']
                if code not in results or results[code]['price'] <= 0:
                    kline_data = self.fetch_eastmoney_kline(code, 1 if stock['market'] == 'sh' else 0)
                    if kline_data:
                        results[code] = kline_data
                        print(f"  {stock['name']}: 使用日K收盘价 {kline_data['price']}")

        # 3. 伦敦金 (新浪hf_XAU接口，人民币/克)
        if fx_list:
            url = "https://hq.sinajs.cn/list=hf_XAU"
            try:
                resp = self.session.get(url, headers={'Referer': 'https://finance.sina.com.cn'}, timeout=10)
                line = resp.text.strip()
                if '"' in line:
                    data_str = line[line.index('"')+1 : line.rindex('"')]
                    p = data_str.split(',')
                    if len(p) >= 13:
                        # 新浪hf_XAU: 人民币/克 (约4800=2740美元/盎司)
                        price = float(p[0])
                        results['XAU'] = {
                            'name': '伦敦金', 
                            'price': price, 
                            'prev_close': float(p[7]),
                            'volume': 0, 'amount': 0, 
                            'date': p[11] if len(p) > 11 else datetime.now().strftime('%Y-%m-%d'), 
                            'time': p[6]
                        }
            except Exception as e: 
                print(f"伦敦金获取失败: {e}")
            
        return results
    
    def check_alerts(self, stock_config, data):
        """检查预警条件 (支持成本百分比、单日涨跌幅、分级预警)"""
        alerts = []
        alert_weights = []  # 用于计算预警级别
        code = stock_config['code']
        cfg = stock_config.get('alerts', {})
        cost = stock_config.get('cost', 0)
        stock_type = stock_config.get('type', 'individual')
        price, prev_close = data['price'], data['prev_close']
        change_pct = (price - prev_close) / prev_close * 100 if prev_close else 0
        
        # 1. 基于成本的百分比预警 (权重: 高)
        if cost > 0:
            cost_change_pct = (price - cost) / cost * 100
            
            if 'cost_pct_above' in cfg and cost_change_pct >= cfg['cost_pct_above']:
                target_price = cost * (1 + cfg['cost_pct_above']/100)
                if not self._alerted_recently(code, 'cost_above'):
                    alerts.append(('cost_above', f"🎯 盈利 {cfg['cost_pct_above']:.0f}% (目标价 ¥{target_price:.2f})"))
                    alert_weights.append(3)  # 高权重
            
            if 'cost_pct_below' in cfg and cost_change_pct <= cfg['cost_pct_below']:
                target_price = cost * (1 + cfg['cost_pct_below']/100)
                if not self._alerted_recently(code, 'cost_below'):
                    alerts.append(('cost_below', f"🛑 亏损 {abs(cfg['cost_pct_below']):.0f}% (止损价 ¥{target_price:.2f})"))
                    alert_weights.append(3)  # 高权重
        
        # 2. 基于固定价格的预警 (权重: 中)
        if 'price_above' in cfg and price >= cfg['price_above'] and not self._alerted_recently(code, 'above'):
            alerts.append(('above', f"🚀 价格突破 ¥{cfg['price_above']}"))
            alert_weights.append(2)
        if 'price_below' in cfg and price <= cfg['price_below'] and not self._alerted_recently(code, 'below'):
            alerts.append(('below', f"📉 价格跌破 ¥{cfg['price_below']}"))
            alert_weights.append(2)
        
        # 3. 单日涨跌幅预警 (权重: 根据幅度)
        if 'change_pct_above' in cfg and change_pct >= cfg['change_pct_above'] and not self._alerted_recently(code, 'pct_up'):
            alerts.append(('pct_up', f"📈 日内大涨 {change_pct:+.2f}%"))
            # 异动越大权重越高
            if change_pct >= 7:
                alert_weights.append(3)  # 涨停附近
            elif change_pct >= 5:
                alert_weights.append(2)  # 大涨
            else:
                alert_weights.append(1)  # 一般异动
                
        if 'change_pct_below' in cfg and change_pct <= cfg['change_pct_below'] and not self._alerted_recently(code, 'pct_down'):
            alerts.append(('pct_down', f"📉 日内大跌 {change_pct:+.2f}%"))
            if change_pct <= -7:
                alert_weights.append(3)  # 跌停附近
            elif change_pct <= -5:
                alert_weights.append(2)  # 大跌
            else:
                alert_weights.append(1)  # 一般异动
        
        # 4. 成交量异动检测 (仅股票和ETF)
        if stock_type != 'gold' and 'volume_surge' in cfg:
            current_volume = data.get('volume', 0)
            if current_volume > 0:
                # 尝试获取5日均量
                ma5_volume = self.fetch_volume_ma5(code, 1 if stock_config['market'] == 'sh' else 0)
                if ma5_volume > 0:
                    volume_ratio = current_volume / ma5_volume
                    threshold = cfg['volume_surge']
                    
                    if volume_ratio >= threshold and not self._alerted_recently(code, 'volume_surge'):
                        alerts.append(('volume_surge', f"📊 放量 {volume_ratio:.1f}倍 (5日均量)"))
                        alert_weights.append(2)  # 中等权重
                    elif volume_ratio <= 0.5 and not self._alerted_recently(code, 'volume_shrink'):
                        alerts.append(('volume_shrink', f"📉 缩量 {volume_ratio:.1f}倍 (5日均量)"))
                        alert_weights.append(1)  # 低权重
        
        # 5. 均线系统 (MA金叉死叉)
        if stock_type != 'gold' and cfg.get('ma_monitor', True):
            ma_data = self.fetch_ma_data(code, 1 if stock_config['market'] == 'sh' else 0)
            if ma_data:
                # 金叉: MA5上穿MA10 (短期转强)
                if ma_data.get('golden_cross') and not self._alerted_recently(code, 'ma_golden'):
                    alerts.append(('ma_golden', f"🌟 均线金叉 (MA5¥{ma_data['MA5']:.2f}上穿MA10¥{ma_data['MA10']:.2f})"))
                    alert_weights.append(3)  # 高权重
                
                # 死叉: MA5下穿MA10 (短期转弱)
                if ma_data.get('death_cross') and not self._alerted_recently(code, 'ma_death'):
                    alerts.append(('ma_death', f"⚠️ 均线死叉 (MA5¥{ma_data['MA5']:.2f}下穿MA10¥{ma_data['MA10']:.2f})"))
                    alert_weights.append(3)  # 高权重
                
                # RSI超买超卖检测
                rsi = ma_data.get('RSI')
                if rsi:
                    if ma_data.get('RSI_overbought') and not self._alerted_recently(code, 'rsi_high'):
                        alerts.append(('rsi_high', f"🔥 RSI超买 ({rsi})，可能回调"))
                        alert_weights.append(2)
                    elif ma_data.get('RSI_oversold') and not self._alerted_recently(code, 'rsi_low'):
                        alerts.append(('rsi_low', f"❄️ RSI超卖 ({rsi})，可能反弹"))
                        alert_weights.append(2)
        
        # 5. 跳空缺口检测 (需要昨日数据)
        if stock_type != 'gold':
            prev_high = data.get('prev_high', 0)
            prev_low = data.get('prev_low', 0)
            current_open = data.get('open', price)  # 当前价近似开盘价
            
            # 向上跳空: 今日开盘 > 昨日最高
            if prev_high > 0 and current_open > prev_high * 1.01:  # 1%以上算跳空
                gap_pct = (current_open - prev_high) / prev_high * 100
                if not self._alerted_recently(code, 'gap_up'):
                    alerts.append(('gap_up', f"⬆️ 向上跳空 {gap_pct:.1f}%"))
                    alert_weights.append(2)
            
            # 向下跳空: 今日开盘 < 昨日最低
            elif prev_low > 0 and current_open < prev_low * 0.99:
                gap_pct = (prev_low - current_open) / prev_low * 100
                if not self._alerted_recently(code, 'gap_down'):
                    alerts.append(('gap_down', f"⬇️ 向下跳空 {gap_pct:.1f}%"))
                    alert_weights.append(2)
        
        # 6. 动态止盈/移动止损 (当盈利达到一定幅度后启动)
        if cost > 0:
            profit_pct = (price - cost) / cost * 100
            
            # 当盈利 >= 10% 时，启动移动止盈
            if profit_pct >= 10:
                # 计算回撤幅度 (从最高点回撤)
                high_since_cost = data.get('high', price)
                drawdown = (high_since_cost - price) / high_since_cost * 100 if high_since_cost > cost else 0
                
                # 回撤5%提醒减仓
                if drawdown >= 5 and not self._alerted_recently(code, 'trailing_stop_5'):
                    alerts.append(('trailing_stop_5', f"📉 利润回撤 {drawdown:.1f}%，建议减仓保护利润"))
                    alert_weights.append(2)
                
                # 回撤10%提醒清仓
                elif drawdown >= 10 and not self._alerted_recently(code, 'trailing_stop_10'):
                    alerts.append(('trailing_stop_10', f"🚨 利润回撤 {drawdown:.1f}%，建议清仓止损"))
                    alert_weights.append(3)
        
        # 6. 计算预警级别
        level = self._calculate_alert_level(alerts, alert_weights, stock_type)
        
        return alerts, level
    
    def _calculate_alert_level(self, alerts, weights, stock_type):
        """计算预警级别: info(提醒) / warning(警告) / critical(紧急)"""
        if not alerts:
            return None
        
        total_weight = sum(weights)
        alert_count = len(alerts)
        
        # 紧急: 多条件共振 或 高权重单一条件
        if total_weight >= 5 or alert_count >= 3:
            return "critical"
        
        # 警告: 中等权重 或 2个条件
        if total_weight >= 3 or alert_count >= 2:
            return "warning"
        
        # 提醒: 单一低权重条件
        return "info"
    
    def _alerted_recently(self, code, atype):
        now = time.time()
        self.alert_log = [l for l in self.alert_log if now - l['t'] < 1800] # 30分钟有效期
        for l in self.alert_log:
            if l['c'] == code and l['a'] == atype: return True
        return False
    
    def record_alert(self, code, atype):
        self.alert_log.append({'c': code, 'a': atype, 't': time.time()})
    
    def fetch_news(self, symbol):
        """抓取个股最近新闻 (新浪/东财聚合) - 简化版"""
        try:
            # 使用东财个股新闻API
            url = f"https://emweb.securities.eastmoney.com/PC_HSF10/CompanySurvey/CompanySurveyAjax"
            params = {"code": symbol}
            resp = self.session.get(url, params=params, timeout=5)
            return ["新闻模块已就绪 (市场收盘中)"]
        except:
            return []

    def run_once(self, smart_mode=True):
        """执行监控 (支持智能频率)"""
        if smart_mode:
            schedule = self.should_run_now()
            if not schedule.get("run"):
                return []
            
            stocks_to_check = schedule.get("stocks", WATCHLIST)
            mode = schedule.get("mode", "normal")
            
            # 只在特定模式打印日志
            if mode in ["market", "weekend"]:
                print(f"[{datetime.now().strftime('%H:%M')}] {mode}模式扫描 {len(stocks_to_check)} 只标的...")
        else:
            stocks_to_check = WATCHLIST
        
        data_map = self.fetch_sina_realtime(stocks_to_check)
        triggered = []
        
        for stock in stocks_to_check:
            code = stock['code']
            if code not in data_map: continue
            
            data = data_map[code]
            
            # 数据有效性检查
            if data['price'] <= 0 or data['prev_close'] <= 0:
                continue
            
            alerts, level = self.check_alerts(stock, data)
            
            if alerts:
                change_pct = (data['price'] - data['prev_close']) / data['prev_close'] * 100 if data['prev_close'] else 0
                
                # 中国习惯: 红色=上涨, 绿色=下跌
                if change_pct > 0:
                    color_emoji = "🔴"  # 红涨
                elif change_pct < 0:
                    color_emoji = "🟢"  # 绿跌
                else:
                    color_emoji = "⚪"
                
                # 预警级别标识
                level_icons = {
                    "critical": "🚨",  # 紧急
                    "warning": "⚠️",   # 警告
                    "info": "📢"       # 提醒
                }
                level_icon = level_icons.get(level, "📢")
                level_text = {"critical": "【紧急】", "warning": "【警告】", "info": "【提醒】"}.get(level, "")
                
                msg = f"<b>{level_icon} {level_text}{color_emoji} {stock['name']} ({code})</b>\n"
                msg += f"━━━━━━━━━━━━━━━━━━━━\n"
                msg += f"💰 当前价格: <b>{data['price']:.2f}</b> ({change_pct:+.2f}%)\n"
                
                # 显示持仓盈亏
                cost = stock.get('cost', 0)
                if cost > 0:
                    cost_change = (data['price'] - cost) / cost * 100
                    profit_icon = "🔴+" if cost_change > 0 else "🟢"
                    msg += f"📊 持仓成本: ¥{cost:.2f} | 盈亏: {profit_icon}{cost_change:.2f}%\n"
                
                msg += f"\n🎯 触发预警 ({len(alerts)}项):\n"
                for _, text in alerts: 
                    msg += f"  • {text}\n"
                    self.record_alert(code, _)
                
                # Pro版：集成智能分析
                try:
                    from analyser import StockAnalyser
                    analyser = StockAnalyser()
                    insight = analyser.generate_insight(stock, {
                        'price': data['price'],
                        'change_pct': change_pct
                    }, alerts)
                    msg += f"\n{insight}"
                except Exception:
                    pass
                
                triggered.append(msg)
        
        return triggered

if __name__ == '__main__':
    monitor = StockAlert()
    for alert in monitor.run_once():
        print(alert)

#!/usr/bin/env python3
"""
Stock Monitor Pro - 完整测试套件
测试所有功能模块，确保系统稳定性
"""

import sys
import time
import unittest
from datetime import datetime, timedelta
from unittest.mock import Mock, patch

sys.path.insert(0, '/home/wesley/.openclaw/workspace/skills/stock-monitor/scripts')

from monitor import StockAlert, WATCHLIST


class TestDataFetching(unittest.TestCase):
    """测试1: 数据获取模块"""
    
    def setUp(self):
        self.monitor = StockAlert()
    
    def test_sina_realtime_api(self):
        """测试新浪实时行情API"""
        data = self.monitor.fetch_sina_realtime([WATCHLIST[0]])
        self.assertIn('600362', data)
        self.assertGreater(data['600362']['price'], 0)
        print("✅ 新浪实时行情API正常")
    
    def test_gold_api(self):
        """测试伦敦金API"""
        data = self.monitor.fetch_sina_realtime([WATCHLIST[-1]])
        self.assertIn('XAU', data)
        self.assertGreater(data['XAU']['price'], 4000)  # 黄金应该在4000以上
        print("✅ 伦敦金API正常")
    
    def test_data_validity(self):
        """测试数据有效性检查"""
        data = self.monitor.fetch_sina_realtime(WATCHLIST[:3])
        for code, d in data.items():
            self.assertGreater(d['price'], 0, f"{code}价格无效")
            self.assertGreater(d['prev_close'], 0, f"{code}昨收无效")
        print("✅ 所有数据有效性检查通过")


class TestAlertRules(unittest.TestCase):
    """测试2: 预警规则模块"""
    
    def setUp(self):
        self.monitor = StockAlert()
    
    def test_cost_percentage_alert(self):
        """测试成本百分比预警"""
        stock = WATCHLIST[0].copy()
        stock['alerts'] = {'cost_pct_above': 10.0, 'cost_pct_below': -10.0}
        
        # 模拟盈利10%的数据
        data = {'price': 62.7, 'prev_close': 57.0, 'cost': 57.0}  # 成本57，现价62.7=+10%
        alerts, level = self.monitor.check_alerts(stock, data)
        
        has_profit_alert = any('盈利' in text for _, text in alerts)
        self.assertTrue(has_profit_alert, "应该有盈利预警")
        print("✅ 成本百分比预警正常")
    
    def test_daily_change_alert(self):
        """测试日内涨跌幅预警"""
        stock = WATCHLIST[0].copy()
        stock['alerts'] = {'change_pct_above': 5.0, 'change_pct_below': -5.0}
        
        # 模拟大涨6%
        data = {'price': 60.42, 'prev_close': 57.0, 'cost': 57.0}
        alerts, level = self.monitor.check_alerts(stock, data)
        
        has_change_alert = any('大涨' in text or '大跌' in text for _, text in alerts)
        self.assertTrue(has_change_alert, "应该有涨跌幅预警")
        print("✅ 日内涨跌幅预警正常")
    
    def test_no_duplicate_alerts(self):
        """测试防重复机制"""
        stock = WATCHLIST[0].copy()
        stock['alerts'] = {'cost_pct_above': 5.0}
        
        data = {'price': 60.0, 'prev_close': 57.0, 'cost': 57.0}
        
        # 第一次应该触发
        alerts1, _ = self.monitor.check_alerts(stock, data)
        self.assertGreater(len(alerts1), 0, "第一次应该触发预警")
        
        # 记录预警
        for alert_type, _ in alerts1:
            self.monitor.record_alert(stock['code'], alert_type)
        
        # 第二次不应该触发 (30分钟内)
        alerts2, _ = self.monitor.check_alerts(stock, data)
        self.assertEqual(len(alerts2), 0, "30分钟内不应重复触发")
        print("✅ 防重复机制正常")


class TestAlertLevel(unittest.TestCase):
    """测试3: 分级预警系统"""
    
    def setUp(self):
        self.monitor = StockAlert()
    
    def test_critical_level(self):
        """测试紧急级别"""
        alerts = [('a', 'test'), ('b', 'test'), ('c', 'test')]
        weights = [3, 3, 3]  # 总权重9
        level = self.monitor._calculate_alert_level(alerts, weights, 'individual')
        self.assertEqual(level, 'critical')
        print("✅ 紧急级别判断正常")
    
    def test_warning_level(self):
        """测试警告级别"""
        alerts = [('a', 'test'), ('b', 'test')]
        weights = [2, 2]  # 总权重4
        level = self.monitor._calculate_alert_level(alerts, weights, 'individual')
        self.assertEqual(level, 'warning')
        print("✅ 警告级别判断正常")
    
    def test_info_level(self):
        """测试提醒级别"""
        alerts = [('a', 'test')]
        weights = [1]
        level = self.monitor._calculate_alert_level(alerts, weights, 'individual')
        self.assertEqual(level, 'info')
        print("✅ 提醒级别判断正常")


class TestStockTypeDifferentiation(unittest.TestCase):
    """测试4: 差异化配置"""
    
    def test_individual_stock_threshold(self):
        """测试个股阈值"""
        stock = [s for s in WATCHLIST if s.get('type') == 'individual'][0]
        self.assertEqual(stock['alerts']['change_pct_above'], 4.0)
        print("✅ 个股阈值配置正确")
    
    def test_etf_threshold(self):
        """测试ETF阈值"""
        stock = [s for s in WATCHLIST if s.get('type') == 'etf'][0]
        self.assertEqual(stock['alerts']['change_pct_above'], 2.0)
        print("✅ ETF阈值配置正确")
    
    def test_gold_threshold(self):
        """测试黄金阈值"""
        stock = [s for s in WATCHLIST if s.get('type') == 'gold'][0]
        self.assertEqual(stock['alerts']['change_pct_above'], 2.5)
        print("✅ 黄金阈值配置正确")


class TestSmartSchedule(unittest.TestCase):
    """测试5: 智能频率控制"""
    
    def setUp(self):
        self.monitor = StockAlert()
    
    def test_market_hours_detection(self):
        """测试交易时间检测"""
        # 当前是纽约时间，转换成北京时间
        ny_now = datetime.now()
        beijing_now = ny_now + timedelta(hours=13)
        
        schedule = self.monitor.should_run_now()
        self.assertIn('mode', schedule)
        self.assertIn(schedule['mode'], ['market', 'lunch', 'after_hours', 'night', 'weekend'])
        print(f"✅ 时间检测正常 (当前模式: {schedule['mode']})")
    
    def test_interval_settings(self):
        """测试不同模式的间隔设置"""
        schedule = self.monitor.should_run_now()
        interval = schedule.get('interval', 0)
        self.assertGreater(interval, 0)
        self.assertIn(interval, [300, 600, 1800, 3600])  # 5/10/30/60分钟
        print(f"✅ 间隔设置正常 ({interval//60}分钟)")


class TestMessageFormat(unittest.TestCase):
    """测试6: 消息格式"""
    
    def setUp(self):
        self.monitor = StockAlert()
    
    def test_message_contains_required_elements(self):
        """测试消息包含必要元素"""
        # 模拟触发预警
        stock = WATCHLIST[0]
        data = {'price': 54.0, 'prev_close': 57.0, 'open': 55.0, 'high': 56.0, 'low': 53.0}
        alerts, level = [('cost_below', '📉 亏损10%')], 'warning'
        
        # 构建消息
        change_pct = -5.26
        msg = f"<b>⚠️ 【警告】🟢 {stock['name']} ({stock['code']})</b>\n"
        msg += f"💰 当前价格: ¥{data['price']:.2f} ({change_pct:+.2f}%)\n"
        msg += f"🎯 触发预警:\n  • {alerts[0][1]}\n"
        
        # 检查必要元素
        self.assertIn('【警告】', msg)
        self.assertIn('🟢', msg)  # 绿跌
        self.assertIn('💰', msg)
        self.assertIn('🎯', msg)
        print("✅ 消息格式包含必要元素")


class TestIntegration(unittest.TestCase):
    """测试7: 集成测试"""
    
    def setUp(self):
        self.monitor = StockAlert()
    
    def test_full_run_once(self):
        """测试完整run_once流程"""
        start = time.time()
        alerts_list = self.monitor.run_once(smart_mode=True)
        elapsed = time.time() - start
        
        # 执行时间应该合理 (10-30秒)
        self.assertLess(elapsed, 60, "执行时间过长")
        self.assertIsInstance(alerts_list, list)
        print(f"✅ 完整流程正常 (执行时间: {elapsed:.2f}秒, 触发{len(alerts_list)}条)")
    
    def test_all_stocks_monitored(self):
        """测试所有股票都被监控"""
        data = self.monitor.fetch_sina_realtime(WATCHLIST)
        # 至少应该获取到部分数据
        self.assertGreater(len(data), 0)
        print(f"✅ 监控覆盖正常 (获取到{len(data)}/{len(WATCHLIST)}只数据)")


def run_all_tests():
    """运行所有测试"""
    print("=" * 70)
    print("🧪 Stock Monitor Pro - 完整测试套件")
    print("=" * 70)
    
    # 创建测试套件
    loader = unittest.TestLoader()
    suite = unittest.TestSuite()
    
    # 添加所有测试类
    suite.addTests(loader.loadTestsFromTestCase(TestDataFetching))
    suite.addTests(loader.loadTestsFromTestCase(TestAlertRules))
    suite.addTests(loader.loadTestsFromTestCase(TestAlertLevel))
    suite.addTests(loader.loadTestsFromTestCase(TestStockTypeDifferentiation))
    suite.addTests(loader.loadTestsFromTestCase(TestSmartSchedule))
    suite.addTests(loader.loadTestsFromTestCase(TestMessageFormat))
    suite.addTests(loader.loadTestsFromTestCase(TestIntegration))
    
    # 运行测试
    runner = unittest.TextTestRunner(verbosity=2)
    result = runner.run(suite)
    
    # 输出总结
    print("\n" + "=" * 70)
    print("📊 测试总结")
    print("=" * 70)
    print(f"  测试总数: {result.testsRun}")
    print(f"  通过: {result.testsRun - len(result.failures) - len(result.errors)}")
    print(f"  失败: {len(result.failures)}")
    print(f"  错误: {len(result.errors)}")
    
    if result.wasSuccessful():
        print("\n✅ 所有测试通过！系统可以正常运行。")
    else:
        print("\n⚠️  部分测试失败，请检查日志。")
    
    return result.wasSuccessful()


if __name__ == '__main__':
    success = run_all_tests()
    sys.exit(0 if success else 1)