智能股票分析系统 - 主要分析脚本

stock_analyzer.py

#!/usr/bin/env python3
"""
智能股票分析系统
- 获取VIX指数和美股数据
- MACD技术指标分析
- DeepSeek AI智能分析
- 自动邮件推送
运行环境：GitHub Actions
运行时间：东八区早上8:30 (UTC 00:30)
"""

import json
import requests
import smtplib
import logging
import os
from datetime import datetime, timedelta
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import Dict, List, Any
import time

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('stock_analysis.log'),
        logging.StreamHandler()
    ]
)

class StockAnalyzer:
    def __init__(self, config_file: str = 'config.json'):
        """初始化股票分析器"""
        self.config = self.load_config(config_file)
        self.alpha_vantage_api_key = self.config['alpha_vantage']['api_key']
        self.deepseek_api_key = self.config['deepseek']['api_key']
        self.email_config = self.config['email']
        self.stocks = self.config['stocks']
        
    def load_config(self, config_file: str) -> Dict:
        """加载配置文件"""
        try:
            with open(config_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        except FileNotFoundError:
            logging.error(f"配置文件 {config_file} 不存在")
            raise
        except json.JSONDecodeError:
            logging.error(f"配置文件 {config_file} 格式错误")
            raise

    def get_stock_data(self, symbol: str) -> Dict[str, Any]:
        """获取股票的基本数据"""
        url = "https://www.alphavantage.co/query"
        params = {
            'function': 'TIME_SERIES_DAILY',
            'symbol': symbol,
            'apikey': self.alpha_vantage_api_key,
            'outputsize': 'compact'
        }
        
        try:
            response = requests.get(url, params=params)
            response.raise_for_status()
            data = response.json()
            
            if 'Error Message' in data:
                logging.error(f"获取 {symbol} 数据失败: {data['Error Message']}")
                return None
            
            if 'Note' in data:
                logging.warning(f"API调用频率限制: {data['Note']}")
                time.sleep(12)  # Alpha Vantage免费版限制每分钟5次调用
                return self.get_stock_data(symbol)
            
            time_series = data.get('Time Series (Daily)', {})
            if not time_series:
                logging.error(f"无法获取 {symbol} 的时间序列数据")
                return None
            
            # 获取最新交易日数据
            latest_date = max(time_series.keys())
            latest_data = time_series[latest_date]
            
            return {
                'symbol': symbol,
                'date': latest_date,
                'open': float(latest_data['1. open']),
                'high': float(latest_data['2. high']),
                'low': float(latest_data['3. low']),
                'close': float(latest_data['4. close']),
                'volume': int(latest_data['5. volume'])
            }
            
        except requests.RequestException as e:
            logging.error(f"请求 {symbol} 数据时出错: {e}")
            return None

    def get_macd_data(self, symbol: str) -> Dict[str, Any]:
        """获取MACD指标数据"""
        url = "https://www.alphavantage.co/query"
        params = {
            'function': 'MACD',
            'symbol': symbol,
            'interval': 'daily',
            'series_type': 'close',
            'apikey': self.alpha_vantage_api_key
        }
        
        try:
            response = requests.get(url, params=params)
            response.raise_for_status()
            data = response.json()
            
            if 'Error Message' in data:
                logging.error(f"获取 {symbol} MACD数据失败: {data['Error Message']}")
                return None
            
            if 'Note' in data:
                logging.warning(f"API调用频率限制: {data['Note']}")
                time.sleep(12)
                return self.get_macd_data(symbol)
            
            macd_data = data.get('Technical Analysis: MACD', {})
            if not macd_data:
                logging.error(f"无法获取 {symbol} 的MACD数据")
                return None
            
            # 获取最新MACD数据
            latest_date = max(macd_data.keys())
            latest_macd = macd_data[latest_date]
            
            return {
                'symbol': symbol,
                'date': latest_date,
                'macd': float(latest_macd['MACD']),
                'macd_signal': float(latest_macd['MACD_Signal']),
                'macd_hist': float(latest_macd['MACD_Hist'])
            }
            
        except requests.RequestException as e:
            logging.error(f"请求 {symbol} MACD数据时出错: {e}")
            return None

    def get_vix_data(self) -> Dict[str, Any]:
        """获取VIX恐慌指数"""
        url = "https://www.alphavantage.co/query"
        params = {
            'function': 'TIME_SERIES_DAILY',
            'symbol': 'VIX',
            'apikey': self.alpha_vantage_api_key,
            'outputsize': 'compact'
        }
        
        try:
            response = requests.get(url, params=params)
            response.raise_for_status()
            data = response.json()
            
            if 'Error Message' in data:
                logging.error(f"获取VIX数据失败: {data['Error Message']}")
                return None
            
            if 'Note' in data:
                logging.warning(f"API调用频率限制: {data['Note']}")
                time.sleep(12)
                return self.get_vix_data()
            
            time_series = data.get('Time Series (Daily)', {})
            if not time_series:
                logging.error("无法获取VIX时间序列数据")
                return None
            
            latest_date = max(time_series.keys())
            latest_data = time_series[latest_date]
            
            return {
                'symbol': 'VIX',
                'date': latest_date,
                'close': float(latest_data['4. close']),
                'open': float(latest_data['1. open']),
                'high': float(latest_data['2. high']),
                'low': float(latest_data['3. low'])
            }
            
        except requests.RequestException as e:
            logging.error(f"请求VIX数据时出错: {e}")
            return None

    def analyze_with_deepseek(self, market_data: Dict) -> str:
        """使用DeepSeek API分析市场数据"""
        url = "https://api.deepseek.com/v1/chat/completions"
        
        # 构建分析提示
        prompt = self.build_analysis_prompt(market_data)
        
        headers = {
            'Authorization': f'Bearer {self.deepseek_api_key}',
            'Content-Type': 'application/json'
        }
        
        payload = {
            'model': 'deepseek-chat',
            'messages': [
                {
                    'role': 'system',
                    'content': '你是一位专业的股票分析师，请基于提供的市场数据进行客观、专业的分析。'
                },
                {
                    'role': 'user',
                    'content': prompt
                }
            ],
            'temperature': 0.7,
            'max_tokens': 2000
        }
        
        try:
            response = requests.post(url, headers=headers, json=payload)
            response.raise_for_status()
            result = response.json()
            
            return result['choices'][0]['message']['content']
            
        except requests.RequestException as e:
            logging.error(f"DeepSeek API请求失败: {e}")
            return "分析服务暂时不可用，请稍后重试。"

    def build_analysis_prompt(self, market_data: Dict) -> str:
        """构建分析提示"""
        prompt = f"""请分析以下美股市场数据（数据日期：{datetime.now().strftime('%Y-%m-%d')} 东八区早晨获取）：

## VIX恐慌指数
- 当前值：{market_data['vix']['close']:.2f}
- 开盘：{market_data['vix']['open']:.2f}
- 最高：{market_data['vix']['high']:.2f}
- 最低：{market_data['vix']['low']:.2f}

## 个股数据与技术指标

"""
        
        for stock_data in market_data['stocks']:
            if stock_data and stock_data['basic_data'] and stock_data['macd_data']:
                basic = stock_data['basic_data']
                macd = stock_data['macd_data']
                
                prompt += f"""### {basic['symbol']}
- 收盘价：${basic['close']:.2f}
- 涨跌幅：{((basic['close'] - basic['open']) / basic['open'] * 100):+.2f}%
- 成交量：{basic['volume']:,}
- MACD：{macd['macd']:.4f}
- MACD信号线：{macd['macd_signal']:.4f}
- MACD柱状图：{macd['macd_hist']:.4f}

"""
        
        prompt += """
请从以下角度进行分析：
1. 市场整体情绪（基于VIX指数）
2. 个股技术面分析（价格走势、MACD信号）
3. 风险评估
4. 短期投资建议

请用中文回复，分析要客观专业，避免过于乐观或悲观的表述。
"""
        
        return prompt

    def send_email(self, subject: str, content: str):
        """发送邮件"""
        try:
            msg = MIMEMultipart()
            msg['From'] = self.email_config['from_email']
            msg['To'] = self.email_config['to_email']
            msg['Subject'] = subject
            
            msg.attach(MIMEText(content, 'plain', 'utf-8'))
            
            server = smtplib.SMTP(self.email_config['smtp_server'], self.email_config['smtp_port'])
            server.starttls()
            server.login(self.email_config['from_email'], self.email_config['password'])
            
            text = msg.as_string()
            server.sendmail(self.email_config['from_email'], self.email_config['to_email'], text)
            server.quit()
            
            logging.info("邮件发送成功")
            
        except Exception as e:
            logging.error(f"邮件发送失败: {e}")

    def run_analysis(self):
        """运行完整的分析流程"""
        logging.info("开始股票分析流程")
        
        # 收集所有数据
        market_data = {
            'vix': None,
            'stocks': []
        }
        
        # 获取VIX数据
        logging.info("获取VIX数据...")
        market_data['vix'] = self.get_vix_data()
        
        # 获取股票数据
        for symbol in self.stocks:
            logging.info(f"获取 {symbol} 数据...")
            
            basic_data = self.get_stock_data(symbol)
            time.sleep(12)  # API调用间隔
            
            macd_data = self.get_macd_data(symbol)
            time.sleep(12)  # API调用间隔
            
            market_data['stocks'].append({
                'symbol': symbol,
                'basic_data': basic_data,
                'macd_data': macd_data
            })
        
        # 使用DeepSeek分析
        logging.info("使用DeepSeek进行市场分析...")
        analysis_result = self.analyze_with_deepseek(market_data)
        
        # 准备邮件内容
        email_subject = f"📈 美股智能分析报告 - {datetime.now().strftime('%Y-%m-%d')}"
        
        email_content = f"""
📈 美股市场智能分析报告
=======================================

🕘 报告时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} (东八区)
📊 数据来源：Alpha Vantage | 🤖 分析引擎：DeepSeek AI

{analysis_result}

=======================================
⚠️  投资提醒：
本报告由AI自动生成，仅供参考学习，不构成投资建议。
投资有风险，入市需谨慎。请根据自身风险承受能力谨慎决策。

📞 技术支持：GitHub Issues
🔗 项目地址：https://github.com/onlinefchen/auto-stock-analysis
"""
        
        # 发送邮件
        logging.info("发送分析报告邮件...")
        self.send_email(email_subject, email_content)
        
        logging.info("股票分析流程完成")

def main():
    """主函数"""
    try:
        analyzer = StockAnalyzer()
        analyzer.run_analysis()
    except Exception as e:
        logging.error(f"程序执行失败: {e}")
        raise

if __name__ == "__main__":
    main()