Stock Daily Analysis

# 用户配置 (包含 API Key，不要提交到 git)
config.json

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg

# 虚拟环境
venv/
ENV/
env/
.venv

# IDE
.vscode/
.idea/
*.swp
*.swo
*~

# 日志
*.log
logs/

# 数据缓存
.cache/
*.cache

# OS
.DS_Store
Thumbs.db

# 临时文件
tmp/
temp/
*.tmp

{
  "ai": {
    "provider": "openai",
    "api_key": "sk-替换为你的DeepSeekAPIKey",
    "base_url": "https://api.deepseek.com/v1",
    "model": "deepseek-chat",
    "temperature": 0.3,
    "max_tokens": 4096
  },
  "data": {
    "days": 60,
    "realtime_enabled": true,
    "chip_distribution_enabled": true,
    "use_market_data_skill": true,
    "market_data_skill_path": "../market-data"
  },
  "analysis": {
    "bias_threshold": 5.0,
    "volume_shrink_ratio": 0.7,
    "volume_heavy_ratio": 1.5
  },
  "logging": {
    "level": "INFO"
  }
}

MIT License

Copyright (c) 2026 Wesley Lam

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

akshare>=1.12.0
pandas>=2.0.0
numpy>=1.24.0
requests>=2.31.0
openai>=1.0.0
python-dotenv>=1.0.0

# -*- coding: utf-8 -*-
"""
AI 分析模块 - 调用 Gemini/OpenAI 进行深度分析
"""

import json
import logging
from typing import Dict, Any, Optional

try:
    from openai import OpenAI
    HAS_OPENAI = True
except ImportError:
    HAS_OPENAI = False

logger = logging.getLogger(__name__)


class AIAnalyzer:
    """AI 分析器 - 支持 Gemini 和 OpenAI"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.provider = config.get('provider', 'gemini')
        self.api_key = config.get('api_key', '')
        self.model = config.get('model', 'gemini-3-flash-preview')
        self.temperature = config.get('temperature', 0.3)
        self.max_tokens = config.get('max_tokens', 4096)
        
        if self.provider == 'openai' and HAS_OPENAI:
            base_url = config.get('base_url', 'https://api.openai.com/v1')
            self.client = OpenAI(api_key=self.api_key, base_url=base_url)
        else:
            self.client = None
    
    def analyze(self, code: str, name: str, technical_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        使用 AI 进行深度分析
        
        Args:
            code: 股票代码
            name: 股票名称
            technical_data: 技术指标数据
            
        Returns:
            AI 分析结果
        """
        if not self.api_key:
            logger.warning("未配置 API Key，跳过 AI 分析")
            return self._default_analysis()
        
        try:
            if self.provider == 'gemini':
                return self._analyze_with_gemini(code, name, technical_data)
            else:
                return self._analyze_with_openai(code, name, technical_data)
        except Exception as e:
            logger.error(f"AI 分析失败: {e}")
            return self._default_analysis()
    
    def _analyze_with_gemini(self, code: str, name: str, tech: Dict[str, Any]) -> Dict[str, Any]:
        """使用 Gemini API"""
        import requests
        import os
        
        prompt = self._build_prompt(code, name, tech)
        
        url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:generateContent"
        headers = {"Content-Type": "application/json"}
        params = {"key": self.api_key}
        
        data = {
            "contents": [{"parts": [{"text": prompt}]}],
            "generationConfig": {
                "temperature": self.temperature,
                "maxOutputTokens": self.max_tokens
            }
        }
        
        # 代理设置
        proxies = {}
        proxy_url = os.environ.get('HTTPS_PROXY') or os.environ.get('https_proxy')
        if proxy_url:
            proxies = {'https': proxy_url, 'http': proxy_url}
        
        response = requests.post(url, headers=headers, params=params, json=data, timeout=30, proxies=proxies)
        response.raise_for_status()
        
        result = response.json()
        text = result.get("candidates", [{}])[0].get("content", {}).get("parts", [{}])[0].get("text", "")
        
        return self._parse_ai_response(text, tech)
    
    def _analyze_with_openai(self, code: str, name: str, tech: Dict[str, Any]) -> Dict[str, Any]:
        """使用 OpenAI API"""
        if not HAS_OPENAI or not self.client:
            return self._default_analysis()
        
        prompt = self._build_prompt(code, name, tech)
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=self.temperature,
            max_tokens=self.max_tokens
        )
        
        text = response.choices[0].message.content
        return self._parse_ai_response(text, tech)
    
    def _build_prompt(self, code: str, name: str, tech: Dict[str, Any]) -> str:
        """构建 AI 提示词"""
        
        return f"""你是一位专业的股票分析师，请根据以下技术指标给出投资建议。

股票: {name} ({code})

技术指标数据:
- 当前价格: {tech.get('current_price', 'N/A')}
- MA5: {tech.get('ma5', 'N/A'):.2f} (乖离率: {tech.get('bias_ma5', 0):+.2f}%)
- MA10: {tech.get('ma10', 'N/A'):.2f} (乖离率: {tech.get('bias_ma10', 0):+.2f}%)
- MA20: {tech.get('ma20', 'N/A'):.2f}
- 趋势状态: {tech.get('trend_status', 'N/A')}
- MACD: {tech.get('macd_status', 'N/A')} - {tech.get('macd_signal', '')}
- RSI: {tech.get('rsi_status', 'N/A')} - {tech.get('rsi_signal', '')}
- 量能: {tech.get('volume_status', 'N/A')} - {tech.get('volume_trend', '')}
- 技术面评分: {tech.get('signal_score', 0)}/100
- 买入信号: {tech.get('buy_signal', 'N/A')}
- 买入理由: {', '.join(tech.get('signal_reasons', []))}
- 风险因素: {', '.join(tech.get('risk_factors', []))}

请输出以下 JSON 格式的分析结果:
{{
    "sentiment_score": 0-100,
    "trend_prediction": "上涨/下跌/震荡",
    "operation_advice": "买入/持有/观望/卖出",
    "confidence_level": "高/中/低",
    "analysis_summary": "一句话核心结论",
    "buy_reason": "具体买入理由",
    "risk_warning": "风险提示",
    "target_price": "目标价",
    "stop_loss": "止损价"
}}

只输出 JSON，不要其他内容。"""
    
    def _parse_ai_response(self, text: str, tech: Dict[str, Any]) -> Dict[str, Any]:
        """解析 AI 响应"""
        try:
            # 尝试提取 JSON
            import re
            json_match = re.search(r'\{.*\}', text, re.DOTALL)
            if json_match:
                result = json.loads(json_match.group())
                return {
                    'sentiment_score': result.get('sentiment_score', tech.get('signal_score', 50)),
                    'trend_prediction': result.get('trend_prediction', tech.get('trend_status', '震荡')),
                    'operation_advice': result.get('operation_advice', tech.get('buy_signal', '观望')),
                    'confidence_level': result.get('confidence_level', '中'),
                    'analysis_summary': result.get('analysis_summary', ''),
                    'buy_reason': result.get('buy_reason', ''),
                    'risk_warning': result.get('risk_warning', ''),
                    'target_price': result.get('target_price', ''),
                    'stop_loss': result.get('stop_loss', '')
                }
        except Exception as e:
            logger.warning(f"解析 AI 响应失败: {e}")
        
        # 回退到基于技术面的默认分析
        return self._default_analysis_from_tech(tech)
    
    def _default_analysis_from_tech(self, tech: Dict[str, Any]) -> Dict[str, Any]:
        """基于技术面的默认分析"""
        score = tech.get('signal_score', 50)
        buy_signal = tech.get('buy_signal', '观望')
        
        return {
            'sentiment_score': score,
            'trend_prediction': tech.get('trend_status', '震荡'),
            'operation_advice': buy_signal,
            'confidence_level': '高' if score >= 70 else '中' if score >= 50 else '低',
            'analysis_summary': ' | '.join(tech.get('signal_reasons', []))[:100],
            'buy_reason': ', '.join(tech.get('signal_reasons', [])),
            'risk_warning': ' | '.join(tech.get('risk_factors', [])),
            'target_price': '',
            'stop_loss': ''
        }
    
    def _default_analysis(self) -> Dict[str, Any]:
        """默认分析结果"""
        return {
            'sentiment_score': 50,
            'trend_prediction': '震荡',
            'operation_advice': '观望',
            'confidence_level': '低',
            'analysis_summary': 'AI 分析未启用',
            'buy_reason': '',
            'risk_warning': '',
            'target_price': '',
            'stop_loss': ''
        }

# -*- coding: utf-8 -*-
"""
股票每日分析 - 主入口模块

整合数据获取、技术分析和报告生成
提供简单的调用接口
"""

import json
import logging
import os
from typing import List, Dict, Any, Optional
from pathlib import Path

# 设置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# 导入模块
from scripts.data_fetcher import get_daily_data, get_realtime_quote, get_stock_name
from scripts.trend_analyzer import StockTrendAnalyzer
from scripts.ai_analyzer import AIAnalyzer
from scripts.notifier import AnalysisReport, format_analysis_report, format_dashboard_report


def load_config(config_path: Optional[str] = None) -> Dict[str, Any]:
    """加载配置文件"""
    if config_path is None:
        skill_dir = Path(__file__).parent.parent
        config_path = skill_dir / "config.json"
    
    try:
        with open(config_path, 'r', encoding='utf-8') as f:
            return json.load(f)
    except Exception as e:
        logger.warning(f"加载配置文件失败: {e}，使用默认配置")
        return {
            "data": {"days": 60, "realtime_enabled": True},
            "analysis": {"bias_threshold": 5.0}
        }


def analyze_stock(code: str, config: Optional[Dict] = None) -> Dict[str, Any]:
    """
    分析单只股票
    
    Args:
        code: 股票代码 (如 '600519', 'AAPL', '00700')
        config: 配置字典，可选
        
    Returns:
        包含技术分析结果的字典
    """
    if config is None:
        config = load_config()
    
    logger.info(f"开始分析股票: {code}")
    
    # 获取股票名称
    name = get_stock_name(code)
    
    # 获取历史数据
    days = config.get('data', {}).get('days', 60)
    df = get_daily_data(code, days=days)
    
    if df is None or df.empty:
        logger.error(f"无法获取 {code} 的数据")
        return {
            'code': code,
            'name': name,
            'error': '数据获取失败',
            'technical_indicators': {},
            'ai_analysis': {'operation_advice': '数据不足', 'sentiment_score': 0}
        }
    
    # 技术分析
    analyzer = StockTrendAnalyzer()
    trend_result = analyzer.analyze(df, code)
    
    # 获取实时行情
    quote = get_realtime_quote(code)
    if quote:
        name = quote.name or name
    
    # AI 深度分析
    ai_config = config.get('ai', {})
    ai_analyzer = AIAnalyzer(ai_config)
    ai_result = ai_analyzer.analyze(code, name, trend_result.to_dict())
    
    # 整合结果
    result = {
        'code': code,
        'name': name,
        'technical_indicators': trend_result.to_dict(),
        'ai_analysis': ai_result
    }
    
    logger.info(f"{code} 分析完成，评分: {ai_result.get('sentiment_score', trend_result.signal_score)}")
    return result


def analyze_stocks(codes: List[str], config: Optional[Dict] = None) -> List[Dict[str, Any]]:
    """
    分析多只股票
    
    Args:
        codes: 股票代码列表
        config: 配置字典，可选
        
    Returns:
        分析结果列表
    """
    results = []
    for code in codes:
        try:
            result = analyze_stock(code, config)
            results.append(result)
        except Exception as e:
            logger.error(f"分析 {code} 时出错: {e}")
            results.append({
                'code': code,
                'name': code,
                'error': str(e),
                'ai_analysis': {'operation_advice': '分析失败', 'sentiment_score': 0}
            })
    
    return results


def print_analysis(codes: List[str]) -> None:
    """
    分析股票并打印报告
    
    Args:
        codes: 股票代码列表
    """
    results = analyze_stocks(codes)
    
    # 转换为报告格式并打印
    reports = []
    for result in results:
        if 'error' not in result:
            from scripts.notifier import create_report_from_result
            report = create_report_from_result(result)
            reports.append(report)
    
    if reports:
        print("\n" + format_dashboard_report(reports))
        
        # 打印每个股票的详细报告
        for report in reports:
            print("\n" + format_analysis_report(report))
    else:
        print("没有可显示的报告")


# 便捷函数
if __name__ == "__main__":
    # 测试
    print("=== 股票每日分析系统 ===\n")
    print("正在测试分析茅台 (600519)...\n")
    print_analysis(['600519'])

# -*- coding: utf-8 -*-
"""
数据获取模块 - 基于 akshare 的多市场数据获取
支持 A股、港股、美股行情获取
"""

import logging
import re
import time
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List

import pandas as pd
import akshare as ak

logger = logging.getLogger(__name__)


@dataclass
class StockQuote:
    """统一实时行情数据结构"""
    code: str
    name: str = ""
    price: float = 0.0
    change_pct: float = 0.0
    change_amount: float = 0.0
    volume: int = 0
    amount: float = 0.0
    open_price: float = 0.0
    high: float = 0.0
    low: float = 0.0
    pre_close: float = 0.0
    volume_ratio: Optional[float] = None
    turnover_rate: Optional[float] = None
    pe_ratio: Optional[float] = None
    pb_ratio: Optional[float] = None
    total_mv: Optional[float] = None
    circ_mv: Optional[float] = None


@dataclass
class ChipDistribution:
    """筹码分布数据"""
    profit_ratio: float = 0.0  # 获利比例
    avg_cost: float = 0.0  # 平均成本
    concentration_90: float = 0.0  # 90%筹码集中度
    concentration_70: float = 0.0  # 70%筹码集中度


def _is_us_code(stock_code: str) -> bool:
    """判断是否为美股代码（1-5个大写字母）"""
    code = stock_code.strip().upper()
    return bool(re.match(r'^[A-Z]{1,5}(\.[A-Z])?$', code))


def _is_hk_code(stock_code: str) -> bool:
    """判断是否为港股代码（5位数字）"""
    code = stock_code.lower()
    if code.startswith('hk'):
        numeric_part = code[2:]
        return numeric_part.isdigit() and 1 <= len(numeric_part) <= 5
    return code.isdigit() and len(code) == 5


def _is_etf_code(stock_code: str) -> bool:
    """判断是否为 ETF 代码"""
    etf_prefixes = ('51', '52', '56', '58', '15', '16', '18')
    return stock_code.startswith(etf_prefixes) and len(stock_code) == 6


def normalize_code(stock_code: str) -> tuple:
    """
    标准化股票代码
    
    Returns:
        tuple: (market, code)
        - market: 'a', 'hk', 'us'
        - code: 标准化后的代码
    """
    code = stock_code.strip()
    
    if _is_us_code(code):
        return 'us', code.upper()
    
    if _is_hk_code(code):
        # 去除 hk 前缀，返回5位数字
        if code.lower().startswith('hk'):
            code = code[2:]
        return 'hk', code.zfill(5)
    
    # A股默认处理
    return 'a', code


def get_daily_data(stock_code: str, days: int = 60) -> Optional[pd.DataFrame]:
    """
    获取股票日线数据
    
    Args:
        stock_code: 股票代码
        days: 获取天数
        
    Returns:
        DataFrame 包含 OHLCV 数据，失败返回 None
    """
    market, code = normalize_code(stock_code)
    
    end_date = datetime.now()
    start_date = end_date - timedelta(days=days * 2)
    
    try:
        if market == 'us':
            return _fetch_us_data(code, start_date, end_date)
        elif market == 'hk':
            return _fetch_hk_data(code, start_date, end_date)
        else:
            return _fetch_a_stock_data(code, start_date, end_date)
    except Exception as e:
        logger.error(f"获取 {stock_code} 数据失败: {e}")
        return None


def _fetch_a_stock_data(stock_code: str, start_date: datetime, end_date: datetime) -> pd.DataFrame:
    """获取 A 股数据"""
    start_str = start_date.strftime('%Y%m%d')
    end_str = end_date.strftime('%Y%m%d')
    
    if _is_etf_code(stock_code):
        df = ak.fund_etf_hist_em(
            symbol=stock_code,
            period="daily",
            start_date=start_str,
            end_date=end_str,
            adjust="qfq"
        )
    else:
        df = ak.stock_zh_a_hist(
            symbol=stock_code,
            period="daily",
            start_date=start_str,
            end_date=end_str,
            adjust="qfq"
        )
    
    return _standardize_columns(df)


def _fetch_hk_data(stock_code: str, start_date: datetime, end_date: datetime) -> pd.DataFrame:
    """获取港股数据"""
    start_str = start_date.strftime('%Y%m%d')
    end_str = end_date.strftime('%Y%m%d')
    
    df = ak.stock_hk_hist(
        symbol=stock_code,
        period="daily",
        start_date=start_str,
        end_date=end_str,
        adjust="qfq"
    )
    
    return _standardize_columns(df)


def _fetch_us_data(stock_code: str, start_date: datetime, end_date: datetime) -> pd.DataFrame:
    """获取美股数据"""
    df = ak.stock_us_daily(symbol=stock_code, adjust="qfq")
    
    if df is None or df.empty:
        return pd.DataFrame()
    
    # 按日期过滤
    df['date'] = pd.to_datetime(df['date'])
    df = df[(df['date'] >= start_date) & (df['date'] <= end_date)]
    
    # 标准化列名
    df = df.rename(columns={
        'date': '日期',
        'open': '开盘',
        'high': '最高',
        'low': '最低',
        'close': '收盘',
        'volume': '成交量'
    })
    
    # 计算涨跌幅和成交额
    if '收盘' in df.columns:
        df['涨跌幅'] = df['收盘'].pct_change() * 100
        df['涨跌幅'] = df['涨跌幅'].fillna(0)
    
    if '成交量' in df.columns and '收盘' in df.columns:
        df['成交额'] = df['成交量'] * df['收盘']
    
    return _standardize_columns(df)


def _standardize_columns(df: pd.DataFrame) -> pd.DataFrame:
    """标准化 DataFrame 列名"""
    if df is None or df.empty:
        return pd.DataFrame()
    
    column_mapping = {
        '日期': 'date',
        '开盘': 'open',
        '收盘': 'close',
        '最高': 'high',
        '最低': 'low',
        '成交量': 'volume',
        '成交额': 'amount',
        '涨跌幅': 'pct_chg',
    }
    
    df = df.rename(columns=column_mapping)
    
    # 确保日期格式正确
    if 'date' in df.columns:
        df['date'] = pd.to_datetime(df['date'])
    
    # 数值转换
    for col in ['open', 'high', 'low', 'close', 'volume', 'amount', 'pct_chg']:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors='coerce')
    
    # 去除空值行
    df = df.dropna(subset=['close', 'volume'])
    
    # 按日期排序
    df = df.sort_values('date', ascending=True).reset_index(drop=True)
    
    return df


def get_realtime_quote(stock_code: str) -> Optional[StockQuote]:
    """
    获取实时行情
    
    Args:
        stock_code: 股票代码
        
    Returns:
        StockQuote 对象，失败返回 None
    """
    market, code = normalize_code(stock_code)
    
    try:
        if market == 'us':
            return None  # 美股暂不支持实时行情
        elif market == 'hk':
            return _get_hk_realtime_quote(code)
        else:
            return _get_a_stock_realtime_quote(code)
    except Exception as e:
        logger.warning(f"获取实时行情失败 {stock_code}: {e}")
        return None


def _get_a_stock_realtime_quote(stock_code: str) -> Optional[StockQuote]:
    """获取 A 股实时行情"""
    try:
        df = ak.stock_zh_a_spot_em()
        row = df[df['代码'] == stock_code]
        
        if row.empty:
            return None
        
        row = row.iloc[0]
        
        return StockQuote(
            code=stock_code,
            name=str(row.get('名称', '')),
            price=float(row.get('最新价', 0)) if pd.notna(row.get('最新价')) else 0,
            change_pct=float(row.get('涨跌幅', 0)) if pd.notna(row.get('涨跌幅')) else 0,
            change_amount=float(row.get('涨跌额', 0)) if pd.notna(row.get('涨跌额')) else 0,
            volume=int(row.get('成交量', 0)) if pd.notna(row.get('成交量')) else 0,
            amount=float(row.get('成交额', 0)) if pd.notna(row.get('成交额')) else 0,
            open_price=float(row.get('今开', 0)) if pd.notna(row.get('今开')) else 0,
            high=float(row.get('最高', 0)) if pd.notna(row.get('最高')) else 0,
            low=float(row.get('最低', 0)) if pd.notna(row.get('最低')) else 0,
            volume_ratio=float(row.get('量比', 0)) if pd.notna(row.get('量比')) else None,
            turnover_rate=float(row.get('换手率', 0)) if pd.notna(row.get('换手率')) else None,
            pe_ratio=float(row.get('市盈率-动态', 0)) if pd.notna(row.get('市盈率-动态')) else None,
            pb_ratio=float(row.get('市净率', 0)) if pd.notna(row.get('市净率')) else None,
            total_mv=float(row.get('总市值', 0)) if pd.notna(row.get('总市值')) else None,
            circ_mv=float(row.get('流通市值', 0)) if pd.notna(row.get('流通市值')) else None,
        )
    except Exception as e:
        logger.warning(f"获取 A 股实时行情失败: {e}")
        return None


def _get_hk_realtime_quote(stock_code: str) -> Optional[StockQuote]:
    """获取港股实时行情"""
    try:
        df = ak.stock_hk_spot_em()
        row = df[df['代码'] == stock_code]
        
        if row.empty:
            return None
        
        row = row.iloc[0]
        
        return StockQuote(
            code=stock_code,
            name=str(row.get('名称', '')),
            price=float(row.get('最新价', 0)) if pd.notna(row.get('最新价')) else 0,
            change_pct=float(row.get('涨跌幅', 0)) if pd.notna(row.get('涨跌幅')) else 0,
            change_amount=float(row.get('涨跌额', 0)) if pd.notna(row.get('涨跌额')) else 0,
            volume=int(row.get('成交量', 0)) if pd.notna(row.get('成交量')) else 0,
            amount=float(row.get('成交额', 0)) if pd.notna(row.get('成交额')) else 0,
            volume_ratio=float(row.get('量比', 0)) if pd.notna(row.get('量比')) else None,
            turnover_rate=float(row.get('换手率', 0)) if pd.notna(row.get('换手率')) else None,
            pe_ratio=float(row.get('市盈率', 0)) if pd.notna(row.get('市盈率')) else None,
            pb_ratio=float(row.get('市净率', 0)) if pd.notna(row.get('市净率')) else None,
        )
    except Exception as e:
        logger.warning(f"获取港股实时行情失败: {e}")
        return None


def get_chip_distribution(stock_code: str) -> Optional[ChipDistribution]:
    """
    获取筹码分布数据（仅 A 股）
    
    Args:
        stock_code: 股票代码
        
    Returns:
        ChipDistribution 对象，失败返回 None
    """
    market, code = normalize_code(stock_code)
    
    if market != 'a' or _is_etf_code(code):
        return None
    
    try:
        df = ak.stock_cyq_em(symbol=code)
        
        if df is None or df.empty:
            return None
        
        latest = df.iloc[-1]
        
        return ChipDistribution(
            profit_ratio=float(latest.get('获利比例', 0)) if pd.notna(latest.get('获利比例')) else 0,
            avg_cost=float(latest.get('平均成本', 0)) if pd.notna(latest.get('平均成本')) else 0,
            concentration_90=float(latest.get('90%集中度', 0)) if pd.notna(latest.get('90%集中度')) else 0,
            concentration_70=float(latest.get('70%集中度', 0)) if pd.notna(latest.get('70%集中度')) else 0,
        )
    except Exception as e:
        logger.warning(f"获取筹码分布失败 {stock_code}: {e}")
        return None


def get_stock_name(stock_code: str) -> str:
    """获取股票名称"""
    quote = get_realtime_quote(stock_code)
    if quote and quote.name:
        return quote.name
    
    # 默认返回代码
    return stock_code

# -*- coding: utf-8 -*-
"""
Market Data Skill 集成模块
使用已有的 market-data skill 获取行情数据
"""

import json
import logging
import subprocess
import pandas as pd
from pathlib import Path
from typing import Optional, Dict, Any, List

logger = logging.getLogger(__name__)


class MarketDataFetcher:
    """
    集成 market-data skill 的数据获取器
    """
    
    def __init__(self, skill_path: Optional[str] = None):
        if skill_path is None:
            # 默认相对路径
            skill_path = Path(__file__).parent.parent.parent / "market-data"
        else:
            skill_path = Path(skill_path)
        
        self.skill_path = skill_path
        self.script_path = skill_path / "scripts" / "quote_cn_pro.py"
        
    def is_available(self) -> bool:
        """检查 market-data skill 是否可用"""
        return self.script_path.exists()
    
    def get_kline_data(self, code: str, count: int = 60, period: str = "day") -> Optional[pd.DataFrame]:
        """
        获取 K 线数据
        
        Args:
            code: 股票代码
            count: 数据条数
            period: 周期 (day/week/month/5min)
            
        Returns:
            DataFrame 包含 OHLCV 数据
        """
        if not self.is_available():
            logger.warning(f"market-data skill 不可用: {self.script_path}")
            return None
        
        try:
            cmd = [
                "python3", str(self.script_path),
                code,
                "--kline",
                "--count", str(count),
                "--period", period
            ]
            
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=30,
                cwd=str(self.skill_path)
            )
            
            if result.returncode != 0:
                logger.warning(f"获取 K 线数据失败: {result.stderr}")
                return None
            
            # 解析输出
            return self._parse_kline_output(result.stdout)
            
        except Exception as e:
            logger.error(f"调用 market-data skill 失败: {e}")
            return None
    
    def _parse_kline_output(self, output: str) -> Optional[pd.DataFrame]:
        """解析 K 线输出"""
        lines = output.strip().split('\n')
        data = []
        
        for line in lines:
            line = line.strip()
            if not line or line.startswith('📈') or line.startswith('📡') or line.startswith('='):
                continue
            
            # 尝试解析 K 线数据行
            # 格式: 日期 开盘 收盘 最高 最低 成交量 成交额 ...
            parts = line.split()
            if len(parts) >= 6:
                try:
                    data.append({
                        'date': parts[0],
                        'open': float(parts[1]),
                        'close': float(parts[2]),
                        'high': float(parts[3]),
                        'low': float(parts[4]),
                        'volume': float(parts[5]),
                    })
                except (ValueError, IndexError):
                    continue
        
        if data:
            df = pd.DataFrame(data)
            df['date'] = pd.to_datetime(df['date'])
            return df
        
        return None
    
    def get_daily_summary(self, code: str) -> Optional[Dict[str, Any]]:
        """
        获取日线总结
        
        Returns:
            包含价格、均线等数据的字典
        """
        if not self.is_available():
            return None
        
        try:
            cmd = [
                "python3", str(self.script_path),
                code,
                "--daily"
            ]
            
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=30,
                cwd=str(self.skill_path)
            )
            
            if result.returncode != 0:
                return None
            
            return self._parse_daily_output(result.stdout)
            
        except Exception as e:
            logger.error(f"获取日线数据失败: {e}")
            return None
    
    def _parse_daily_output(self, output: str) -> Dict[str, Any]:
        """解析日线输出"""
        result = {}
        lines = output.split('\n')
        
        for line in lines:
            line = line.strip()
            
            # 提取当前价格
            if '当前价格:' in line:
                try:
                    result['price'] = float(line.split(':')[1].split()[0])
                except:
                    pass
            
            # 提取涨跌幅
            elif '涨跌:' in line:
                try:
                    parts = line.split(':')[1].strip()
                    if '+' in parts:
                        result['change_pct'] = float(parts.split('%')[0].replace('+', ''))
                    elif '-' in parts:
                        result['change_pct'] = -float(parts.split('%')[0].replace('-', ''))
                except:
                    pass
            
            # 提取均线
            elif 'MA5:' in line:
                try:
                    result['ma5'] = float(line.split(':')[1].strip())
                except:
                    pass
            elif 'MA10:' in line:
                try:
                    result['ma10'] = float(line.split(':')[1].strip())
                except:
                    pass
            elif 'MA20:' in line:
                try:
                    result['ma20'] = float(line.split(':')[1].strip())
                except:
                    pass
        
        return result


def create_data_fetcher(config: Optional[Dict] = None) -> Any:
    """
    工厂函数：创建合适的数据获取器
    
    优先使用 market-data skill，如果不可用则回退到 akshare
    """
    if config is None:
        config = {}
    
    use_market_data = config.get('data', {}).get('use_market_data_skill', True)
    
    if use_market_data:
        skill_path = config.get('data', {}).get('market_data_skill_path', '../market-data')
        fetcher = MarketDataFetcher(skill_path)
        
        if fetcher.is_available():
            logger.info("使用 market-data skill 获取数据")
            return fetcher
        else:
            logger.warning(f"market-data skill 不可用，回退到 akshare")
    
    # 回退到 akshare
    from scripts.data_fetcher import get_daily_data, get_realtime_quote
    return None  # 使用 None 表示使用默认的 akshare 方式

# -*- coding: utf-8 -*-
"""
通知/输出处理模块
负责格式化分析报告并输出结果
"""

import logging
from typing import Dict, Any, List, Optional
from dataclasses import dataclass

logger = logging.getLogger(__name__)


@dataclass
class AnalysisReport:
    """分析报告数据结构"""
    code: str
    name: str
    sentiment_score: int
    trend_prediction: str
    operation_advice: str
    decision_type: str
    confidence_level: str
    technical_summary: Dict[str, Any]
    ai_analysis: Optional[str] = None
    risk_warning: str = ""
    buy_reason: str = ""
    support_levels: List[float] = None
    resistance_levels: List[float] = None
    
    def __post_init__(self):
        if self.support_levels is None:
            self.support_levels = []
        if self.resistance_levels is None:
            self.resistance_levels = []


def format_analysis_report(report: AnalysisReport) -> str:
    """
    格式化分析报告为文本
    
    Args:
        report: 分析报告数据
        
    Returns:
        格式化后的报告文本
    """
    lines = [
        f"{'='*50}",
        f"📊 {report.name} ({report.code}) 分析报告",
        f"{'='*50}",
        "",
        f"【核心结论】",
        f"  操作建议: {report.operation_advice}",
        f"  趋势预测: {report.trend_prediction}",
        f"  情绪评分: {report.sentiment_score}/100",
        f"  置信度: {report.confidence_level}",
        "",
        f"【技术面分析】",
    ]
    
    # 技术指标
    tech = report.technical_summary
    if 'current_price' in tech:
        lines.append(f"  当前价格: {tech.get('current_price', 'N/A')}")
    
    if 'ma5' in tech:
        lines.append(f"  MA5: {tech.get('ma5', 'N/A'):.2f} (乖离率: {tech.get('bias_ma5', 0):+.2f}%)")
    if 'ma10' in tech:
        lines.append(f"  MA10: {tech.get('ma10', 'N/A'):.2f} (乖离率: {tech.get('bias_ma10', 0):+.2f}%)")
    if 'ma20' in tech:
        lines.append(f"  MA20: {tech.get('ma20', 'N/A'):.2f}")
    
    if 'trend_status' in tech:
        lines.append(f"  趋势状态: {tech.get('trend_status', 'N/A')}")
    
    if 'volume_status' in tech:
        lines.append(f"  量能状态: {tech.get('volume_status', 'N/A')}")
    
    if 'macd_status' in tech:
        lines.append(f"  MACD: {tech.get('macd_status', 'N/A')}")
    
    if 'rsi_status' in tech:
        lines.append(f"  RSI: {tech.get('rsi_status', 'N/A')}")
    
    lines.append("")
    
    # 支撑压力位
    if report.support_levels:
        lines.append(f"【支撑位】")
        for level in report.support_levels[:3]:
            lines.append(f"  - {level:.2f}")
        lines.append("")
    
    if report.resistance_levels:
        lines.append(f"【压力位】")
        for level in report.resistance_levels[:3]:
            lines.append(f"  - {level:.2f}")
        lines.append("")
    
    # 买入理由
    if report.buy_reason:
        lines.append(f"【买入理由】")
        lines.append(f"  {report.buy_reason}")
        lines.append("")
    
    # 风险警告
    if report.risk_warning:
        lines.append(f"【风险提示】")
        lines.append(f"  {report.risk_warning}")
        lines.append("")
    
    # AI 分析
    if report.ai_analysis:
        lines.append(f"【AI 分析】")
        lines.append(f"  {report.ai_analysis}")
        lines.append("")
    
    lines.append(f"{'='*50}")
    
    return "\n".join(lines)


def format_dashboard_report(reports: List[AnalysisReport]) -> str:
    """
    格式化决策仪表盘报告（多股票汇总）
    
    Args:
        reports: 分析报告列表
        
    Returns:
        格式化的仪表盘报告
    """
    if not reports:
        return "暂无分析报告"
    
    # 统计
    buy_count = sum(1 for r in reports if r.decision_type == 'buy')
    hold_count = sum(1 for r in reports if r.decision_type == 'hold')
    sell_count = sum(1 for r in reports if r.decision_type == 'sell')
    
    lines = [
        f"{'='*60}",
        f"📊 股票分析决策仪表盘",
        f"{'='*60}",
        "",
        f"分析股票数: {len(reports)} 只",
        f"🟢 买入: {buy_count}  🟡 观望: {hold_count}  🔴 卖出: {sell_count}",
        "",
        f"{'='*60}",
    ]
    
    for report in reports:
        emoji = "🟢" if report.decision_type == 'buy' else "🟡" if report.decision_type == 'hold' else "🔴"
        lines.append(f"{emoji} {report.name} ({report.code})")
        lines.append(f"   建议: {report.operation_advice} | 评分: {report.sentiment_score}/100")
        lines.append(f"   趋势: {report.trend_prediction}")
        
        # 添加关键技术指标
        tech = report.technical_summary
        key_info = []
        
        if 'bias_ma5' in tech:
            key_info.append(f"乖离率: {tech['bias_ma5']:+.1f}%")
        if 'macd_status' in tech:
            key_info.append(f"MACD: {tech['macd_status']}")
        
        if key_info:
            lines.append(f"   关键指标: {' | '.join(key_info)}")
        
        lines.append("")
    
    lines.append(f"{'='*60}")
    
    return "\n".join(lines)


def create_report_from_result(result: Dict[str, Any]) -> AnalysisReport:
    """
    从分析结果字典创建报告对象
    
    Args:
        result: 分析结果字典
        
    Returns:
        AnalysisReport 对象
    """
    technical = result.get('technical_indicators', {})
    ai_result = result.get('ai_analysis', {})
    
    # 确定决策类型
    advice = ai_result.get('operation_advice', '观望')
    if advice in ['买入', '加仓', '强烈买入']:
        decision_type = 'buy'
    elif advice in ['卖出', '减仓', '强烈卖出']:
        decision_type = 'sell'
    else:
        decision_type = 'hold'
    
    return AnalysisReport(
        code=result.get('code', ''),
        name=result.get('name', ''),
        sentiment_score=ai_result.get('sentiment_score', 50),
        trend_prediction=ai_result.get('trend_prediction', '震荡'),
        operation_advice=advice,
        decision_type=decision_type,
        confidence_level=ai_result.get('confidence_level', '中'),
        technical_summary=technical,
        ai_analysis=ai_result.get('analysis_summary', ''),
        risk_warning=ai_result.get('risk_warning', ''),
        buy_reason=ai_result.get('buy_reason', ''),
        support_levels=technical.get('support_levels', []),
        resistance_levels=technical.get('resistance_levels', []),
    )


def print_report(report: AnalysisReport) -> None:
    """打印分析报告到控制台"""
    print(format_analysis_report(report))


def print_dashboard(reports: List[AnalysisReport]) -> None:
    """打印决策仪表盘到控制台"""
    print(format_dashboard_report(reports))

#!/bin/bash
# 每日股票分析 - 运行脚本

set -e

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PROJECT_DIR="$SCRIPT_DIR/daily_stock_analysis"
VENV_DIR="$PROJECT_DIR/.venv"

echo "=== 每日股票智能分析 ==="
echo "时间: $(date '+%Y-%m-%d %H:%M:%S')"
echo ""

# 检查项目是否存在
if [ ! -d "$PROJECT_DIR" ]; then
    echo "❌ 错误: 项目未安装"
    echo "请先运行: ./setup.sh"
    exit 1
fi

# 检查环境变量配置
if [ ! -f "$PROJECT_DIR/.env" ]; then
    echo "❌ 错误: 未找到 .env 配置文件"
    echo "请执行: cd $PROJECT_DIR && cp .env.example .env"
    echo "然后编辑 .env 配置 API Key 和股票列表"
    exit 1
fi

cd "$PROJECT_DIR"

# 运行分析
echo "→ 开始分析..."
"$VENV_DIR/bin/python" main.py

echo ""
echo "✅ 分析完成!"
echo ""

#!/bin/bash
# 每日股票分析 - 安装脚本

set -e

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PROJECT_DIR="$SCRIPT_DIR/daily_stock_analysis"
VENV_DIR="$PROJECT_DIR/.venv"
REPO_URL="https://github.com/ZhuLinsen/daily_stock_analysis.git"

echo "=== 每日股票智能分析系统 - 安装 ==="
echo ""

# 检查 Python
if ! command -v python3 &> /dev/null; then
    echo "❌ 错误: 需要 Python 3.8+"
    exit 1
fi

PYTHON_VERSION=$(python3 --version | cut -d' ' -f2 | cut -d'.' -f1,2)
echo "✓ Python 版本: $PYTHON_VERSION"

# 克隆或更新项目
if [ -d "$PROJECT_DIR" ]; then
    echo "→ 项目已存在，更新代码..."
    cd "$PROJECT_DIR"
    git pull origin main
else
    echo "→ 克隆项目仓库..."
    git clone "$REPO_URL" "$PROJECT_DIR"
    cd "$PROJECT_DIR"
fi

# 创建虚拟环境
if [ ! -d "$VENV_DIR" ]; then
    echo "→ 创建虚拟环境..."
    python3 -m venv "$VENV_DIR"
fi

# 安装依赖
echo "→ 安装 Python 依赖..."
"$VENV_DIR/bin/pip" install -q -r requirements.txt

echo ""
echo "✅ 安装完成!"
echo ""
echo "下一步:"
echo "  1. cd $PROJECT_DIR"
echo "  2. cp .env.example .env"
echo "  3. 编辑 .env 配置 API Key 和股票列表"
echo "  4. cd $SCRIPT_DIR && ./run.sh"
echo ""

# -*- coding: utf-8 -*-
"""
趋势交易分析器 - 基于交易理念的技术分析

核心原则：
1. 严进策略 - 不追高，追求每笔交易成功率
2. 趋势交易 - MA5>MA10>MA20 多头排列，顺势而为
3. 买点偏好 - 在 MA5/MA10 附近回踩买入
"""

import logging
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Any
from enum import Enum

import pandas as pd
import numpy as np

logger = logging.getLogger(__name__)


class TrendStatus(Enum):
    """趋势状态枚举"""
    STRONG_BULL = "强势多头"
    BULL = "多头排列"
    WEAK_BULL = "弱势多头"
    CONSOLIDATION = "盘整"
    WEAK_BEAR = "弱势空头"
    BEAR = "空头排列"
    STRONG_BEAR = "强势空头"


class VolumeStatus(Enum):
    """量能状态枚举"""
    HEAVY_VOLUME_UP = "放量上涨"
    HEAVY_VOLUME_DOWN = "放量下跌"
    SHRINK_VOLUME_UP = "缩量上涨"
    SHRINK_VOLUME_DOWN = "缩量回调"
    NORMAL = "量能正常"


class BuySignal(Enum):
    """买入信号枚举"""
    STRONG_BUY = "强烈买入"
    BUY = "买入"
    HOLD = "持有"
    WAIT = "观望"
    SELL = "卖出"
    STRONG_SELL = "强烈卖出"


class MACDStatus(Enum):
    """MACD状态枚举"""
    GOLDEN_CROSS_ZERO = "零轴上金叉"
    GOLDEN_CROSS = "金叉"
    BULLISH = "多头"
    CROSSING_UP = "上穿零轴"
    CROSSING_DOWN = "下穿零轴"
    BEARISH = "空头"
    DEATH_CROSS = "死叉"


class RSIStatus(Enum):
    """RSI状态枚举"""
    OVERBOUGHT = "超买"
    STRONG_BUY = "强势买入"
    NEUTRAL = "中性"
    WEAK = "弱势"
    OVERSOLD = "超卖"


@dataclass
class TrendAnalysisResult:
    """趋势分析结果"""
    code: str
    
    # 趋势判断
    trend_status: TrendStatus = TrendStatus.CONSOLIDATION
    ma_alignment: str = ""
    trend_strength: float = 0.0
    
    # 均线数据
    ma5: float = 0.0
    ma10: float = 0.0
    ma20: float = 0.0
    ma60: float = 0.0
    current_price: float = 0.0
    
    # 乖离率
    bias_ma5: float = 0.0
    bias_ma10: float = 0.0
    bias_ma20: float = 0.0
    
    # 量能分析
    volume_status: VolumeStatus = VolumeStatus.NORMAL
    volume_ratio_5d: float = 0.0
    volume_trend: str = ""
    
    # 支撑压力
    support_ma5: bool = False
    support_ma10: bool = False
    resistance_levels: List[float] = field(default_factory=list)
    support_levels: List[float] = field(default_factory=list)
    
    # MACD 指标
    macd_dif: float = 0.0
    macd_dea: float = 0.0
    macd_bar: float = 0.0
    macd_status: MACDStatus = MACDStatus.BULLISH
    macd_signal: str = ""
    
    # RSI 指标
    rsi_6: float = 0.0
    rsi_12: float = 0.0
    rsi_24: float = 0.0
    rsi_status: RSIStatus = RSIStatus.NEUTRAL
    rsi_signal: str = ""
    
    # 买入信号
    buy_signal: BuySignal = BuySignal.WAIT
    signal_score: int = 0
    signal_reasons: List[str] = field(default_factory=list)
    risk_factors: List[str] = field(default_factory=list)
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return {
            'code': self.code,
            'trend_status': self.trend_status.value,
            'ma_alignment': self.ma_alignment,
            'trend_strength': self.trend_strength,
            'ma5': self.ma5,
            'ma10': self.ma10,
            'ma20': self.ma20,
            'ma60': self.ma60,
            'current_price': self.current_price,
            'bias_ma5': self.bias_ma5,
            'bias_ma10': self.bias_ma10,
            'bias_ma20': self.bias_ma20,
            'volume_status': self.volume_status.value,
            'volume_ratio_5d': self.volume_ratio_5d,
            'volume_trend': self.volume_trend,
            'support_ma5': self.support_ma5,
            'support_ma10': self.support_ma10,
            'buy_signal': self.buy_signal.value,
            'signal_score': self.signal_score,
            'signal_reasons': self.signal_reasons,
            'risk_factors': self.risk_factors,
            'macd_status': self.macd_status.value,
            'macd_signal': self.macd_signal,
            'rsi_status': self.rsi_status.value,
            'rsi_signal': self.rsi_signal,
        }


class StockTrendAnalyzer:
    """
    股票趋势分析器
    
    基于交易理念：
    1. 趋势判断 - MA5>MA10>MA20 多头排列
    2. 乖离率检测 - 不追高，偏离 MA5 超过 5% 不买
    3. 量能分析 - 偏好缩量回调
    4. MACD/RSI 指标分析
    """
    
    BIAS_THRESHOLD = 5.0  # 乖离率阈值
    VOLUME_SHRINK_RATIO = 0.7
    VOLUME_HEAVY_RATIO = 1.5
    MA_SUPPORT_TOLERANCE = 0.02
    
    MACD_FAST = 12
    MACD_SLOW = 26
    MACD_SIGNAL = 9
    
    RSI_SHORT = 6
    RSI_MID = 12
    RSI_LONG = 24
    RSI_OVERBOUGHT = 70
    RSI_OVERSOLD = 30
    
    def analyze(self, df: pd.DataFrame, code: str) -> TrendAnalysisResult:
        """
        分析股票趋势
        
        Args:
            df: 包含 OHLCV 数据的 DataFrame
            code: 股票代码
            
        Returns:
            TrendAnalysisResult 分析结果
        """
        result = TrendAnalysisResult(code=code)
        
        if df is None or df.empty or len(df) < 20:
            logger.warning(f"{code} 数据不足，无法进行趋势分析")
            result.risk_factors.append("数据不足，无法完成分析")
            return result
        
        # 确保数据按日期排序
        df = df.sort_values('date').reset_index(drop=True)
        
        # 计算指标
        df = self._calculate_mas(df)
        df = self._calculate_macd(df)
        df = self._calculate_rsi(df)
        
        # 获取最新数据
        latest = df.iloc[-1]
        result.current_price = float(latest['close'])
        result.ma5 = float(latest['MA5'])
        result.ma10 = float(latest['MA10'])
        result.ma20 = float(latest['MA20'])
        result.ma60 = float(latest.get('MA60', 0))
        
        # 分析各项
        self._analyze_trend(df, result)
        self._calculate_bias(result)
        self._analyze_volume(df, result)
        self._analyze_support_resistance(df, result)
        self._analyze_macd(df, result)
        self._analyze_rsi(df, result)
        self._generate_signal(result)
        
        return result
    
    def _calculate_mas(self, df: pd.DataFrame) -> pd.DataFrame:
        """计算均线"""
        df = df.copy()
        df['MA5'] = df['close'].rolling(window=5, min_periods=1).mean()
        df['MA10'] = df['close'].rolling(window=10, min_periods=1).mean()
        df['MA20'] = df['close'].rolling(window=20, min_periods=1).mean()
        if len(df) >= 60:
            df['MA60'] = df['close'].rolling(window=60, min_periods=1).mean()
        else:
            df['MA60'] = df['MA20']
        return df
    
    def _calculate_macd(self, df: pd.DataFrame) -> pd.DataFrame:
        """计算 MACD 指标"""
        df = df.copy()
        
        ema_fast = df['close'].ewm(span=self.MACD_FAST, adjust=False).mean()
        ema_slow = df['close'].ewm(span=self.MACD_SLOW, adjust=False).mean()
        
        df['MACD_DIF'] = ema_fast - ema_slow
        df['MACD_DEA'] = df['MACD_DIF'].ewm(span=self.MACD_SIGNAL, adjust=False).mean()
        df['MACD_BAR'] = (df['MACD_DIF'] - df['MACD_DEA']) * 2
        
        return df
    
    def _calculate_rsi(self, df: pd.DataFrame) -> pd.DataFrame:
        """计算 RSI 指标"""
        df = df.copy()
        
        for period in [self.RSI_SHORT, self.RSI_MID, self.RSI_LONG]:
            delta = df['close'].diff()
            gain = delta.where(delta > 0, 0)
            loss = -delta.where(delta < 0, 0)
            
            avg_gain = gain.rolling(window=period, min_periods=1).mean()
            avg_loss = loss.rolling(window=period, min_periods=1).mean()
            
            rs = avg_gain / avg_loss
            rsi = 100 - (100 / (1 + rs))
            
            df[f'RSI_{period}'] = rsi.fillna(50)
        
        return df
    
    def _analyze_trend(self, df: pd.DataFrame, result: TrendAnalysisResult) -> None:
        """分析趋势状态"""
        ma5, ma10, ma20 = result.ma5, result.ma10, result.ma20
        
        if ma5 > ma10 > ma20:
            # 检查趋势强度
            if len(df) >= 5:
                prev = df.iloc[-5]
                prev_spread = (prev['MA5'] - prev['MA20']) / prev['MA20'] * 100 if prev['MA20'] > 0 else 0
                curr_spread = (ma5 - ma20) / ma20 * 100 if ma20 > 0 else 0
                
                if curr_spread > prev_spread and curr_spread > 5:
                    result.trend_status = TrendStatus.STRONG_BULL
                    result.ma_alignment = "强势多头排列，均线发散上行"
                    result.trend_strength = 90
                else:
                    result.trend_status = TrendStatus.BULL
                    result.ma_alignment = "多头排列 MA5>MA10>MA20"
                    result.trend_strength = 75
            else:
                result.trend_status = TrendStatus.BULL
                result.ma_alignment = "多头排列 MA5>MA10>MA20"
                result.trend_strength = 75
                
        elif ma5 > ma10 and ma10 <= ma20:
            result.trend_status = TrendStatus.WEAK_BULL
            result.ma_alignment = "弱势多头，MA5>MA10 但 MA10≤MA20"
            result.trend_strength = 55
            
        elif ma5 < ma10 < ma20:
            if len(df) >= 5:
                prev = df.iloc[-5]
                prev_spread = (prev['MA20'] - prev['MA5']) / prev['MA5'] * 100 if prev['MA5'] > 0 else 0
                curr_spread = (ma20 - ma5) / ma5 * 100 if ma5 > 0 else 0
                
                if curr_spread > prev_spread and curr_spread > 5:
                    result.trend_status = TrendStatus.STRONG_BEAR
                    result.ma_alignment = "强势空头排列，均线发散下行"
                    result.trend_strength = 10
                else:
                    result.trend_status = TrendStatus.BEAR
                    result.ma_alignment = "空头排列 MA5<MA10<MA20"
                    result.trend_strength = 25
            else:
                result.trend_status = TrendStatus.BEAR
                result.ma_alignment = "空头排列 MA5<MA10<MA20"
                result.trend_strength = 25
                
        elif ma5 < ma10 and ma10 >= ma20:
            result.trend_status = TrendStatus.WEAK_BEAR
            result.ma_alignment = "弱势空头，MA5<MA10 但 MA10≥MA20"
            result.trend_strength = 40
            
        else:
            result.trend_status = TrendStatus.CONSOLIDATION
            result.ma_alignment = "均线缠绕，趋势不明"
            result.trend_strength = 50
    
    def _calculate_bias(self, result: TrendAnalysisResult) -> None:
        """计算乖离率"""
        price = result.current_price
        
        if result.ma5 > 0:
            result.bias_ma5 = (price - result.ma5) / result.ma5 * 100
        if result.ma10 > 0:
            result.bias_ma10 = (price - result.ma10) / result.ma10 * 100
        if result.ma20 > 0:
            result.bias_ma20 = (price - result.ma20) / result.ma20 * 100
    
    def _analyze_volume(self, df: pd.DataFrame, result: TrendAnalysisResult) -> None:
        """分析量能"""
        if len(df) < 5:
            return
        
        latest = df.iloc[-1]
        vol_5d_avg = df['volume'].iloc[-6:-1].mean()
        
        if vol_5d_avg > 0:
            result.volume_ratio_5d = float(latest['volume']) / vol_5d_avg
        
        # 判断价格变化
        if len(df) >= 2:
            prev_close = df.iloc[-2]['close']
            price_change = (latest['close'] - prev_close) / prev_close * 100
            
            # 量能状态判断
            if result.volume_ratio_5d >= self.VOLUME_HEAVY_RATIO:
                if price_change > 0:
                    result.volume_status = VolumeStatus.HEAVY_VOLUME_UP
                    result.volume_trend = "放量上涨，多头力量强劲"
                else:
                    result.volume_status = VolumeStatus.HEAVY_VOLUME_DOWN
                    result.volume_trend = "放量下跌，注意风险"
            elif result.volume_ratio_5d <= self.VOLUME_SHRINK_RATIO:
                if price_change > 0:
                    result.volume_status = VolumeStatus.SHRINK_VOLUME_UP
                    result.volume_trend = "缩量上涨，上攻动能不足"
                else:
                    result.volume_status = VolumeStatus.SHRINK_VOLUME_DOWN
                    result.volume_trend = "缩量回调，洗盘特征明显"
            else:
                result.volume_status = VolumeStatus.NORMAL
                result.volume_trend = "量能正常"
    
    def _analyze_support_resistance(self, df: pd.DataFrame, result: TrendAnalysisResult) -> None:
        """分析支撑压力位"""
        price = result.current_price
        
        # 检查 MA5 支撑
        if result.ma5 > 0:
            ma5_distance = abs(price - result.ma5) / result.ma5
            if ma5_distance <= self.MA_SUPPORT_TOLERANCE and price >= result.ma5:
                result.support_ma5 = True
                result.support_levels.append(result.ma5)
        
        # 检查 MA10 支撑
        if result.ma10 > 0:
            ma10_distance = abs(price - result.ma10) / result.ma10
            if ma10_distance <= self.MA_SUPPORT_TOLERANCE and price >= result.ma10:
                result.support_ma10 = True
                if result.ma10 not in result.support_levels:
                    result.support_levels.append(result.ma10)
        
        # MA20 作为重要支撑
        if result.ma20 > 0 and price >= result.ma20:
            result.support_levels.append(result.ma20)
        
        # 近期高点作为压力
        if len(df) >= 20:
            recent_high = df['high'].iloc[-20:].max()
            if recent_high > price:
                result.resistance_levels.append(recent_high)
    
    def _analyze_macd(self, df: pd.DataFrame, result: TrendAnalysisResult) -> None:
        """分析 MACD 指标"""
        if len(df) < self.MACD_SLOW:
            result.macd_signal = "数据不足"
            return
        
        latest = df.iloc[-1]
        prev = df.iloc[-2]
        
        result.macd_dif = float(latest['MACD_DIF'])
        result.macd_dea = float(latest['MACD_DEA'])
        result.macd_bar = float(latest['MACD_BAR'])
        
        # 判断金叉死叉
        prev_dif_dea = prev['MACD_DIF'] - prev['MACD_DEA']
        curr_dif_dea = result.macd_dif - result.macd_dea
        
        is_golden_cross = prev_dif_dea <= 0 and curr_dif_dea > 0
        is_death_cross = prev_dif_dea >= 0 and curr_dif_dea < 0
        is_crossing_up = prev['MACD_DIF'] <= 0 and result.macd_dif > 0
        is_crossing_down = prev['MACD_DIF'] >= 0 and result.macd_dif < 0
        
        if is_golden_cross and result.macd_dif > 0:
            result.macd_status = MACDStatus.GOLDEN_CROSS_ZERO
            result.macd_signal = "⭐ 零轴上金叉，强烈买入信号！"
        elif is_crossing_up:
            result.macd_status = MACDStatus.CROSSING_UP
            result.macd_signal = "⚡ DIF上穿零轴，趋势转强"
        elif is_golden_cross:
            result.macd_status = MACDStatus.GOLDEN_CROSS
            result.macd_signal = "✅ 金叉，趋势向上"
        elif is_death_cross:
            result.macd_status = MACDStatus.DEATH_CROSS
            result.macd_signal = "❌ 死叉，趋势向下"
        elif is_crossing_down:
            result.macd_status = MACDStatus.CROSSING_DOWN
            result.macd_signal = "⚠️ DIF下穿零轴，趋势转弱"
        elif result.macd_dif > 0 and result.macd_dea > 0:
            result.macd_status = MACDStatus.BULLISH
            result.macd_signal = "✓ 多头排列，持续上涨"
        elif result.macd_dif < 0 and result.macd_dea < 0:
            result.macd_status = MACDStatus.BEARISH
            result.macd_signal = "⚠ 空头排列，持续下跌"
        else:
            result.macd_status = MACDStatus.BULLISH
            result.macd_signal = "MACD 中性区域"
    
    def _analyze_rsi(self, df: pd.DataFrame, result: TrendAnalysisResult) -> None:
        """分析 RSI 指标"""
        if len(df) < self.RSI_LONG:
            result.rsi_signal = "数据不足"
            return
        
        latest = df.iloc[-1]
        
        result.rsi_6 = float(latest[f'RSI_{self.RSI_SHORT}'])
        result.rsi_12 = float(latest[f'RSI_{self.RSI_MID}'])
        result.rsi_24 = float(latest[f'RSI_{self.RSI_LONG}'])
        
        rsi_mid = result.rsi_12
        
        if rsi_mid > self.RSI_OVERBOUGHT:
            result.rsi_status = RSIStatus.OVERBOUGHT
            result.rsi_signal = f"⚠️ RSI超买({rsi_mid:.1f}>70)，短期回调风险高"
        elif rsi_mid > 60:
            result.rsi_status = RSIStatus.STRONG_BUY
            result.rsi_signal = f"✅ RSI强势({rsi_mid:.1f})，多头力量充足"
        elif rsi_mid >= 40:
            result.rsi_status = RSIStatus.NEUTRAL
            result.rsi_signal = f"RSI中性({rsi_mid:.1f})，震荡整理中"
        elif rsi_mid >= self.RSI_OVERSOLD:
            result.rsi_status = RSIStatus.WEAK
            result.rsi_signal = f"⚡ RSI弱势({rsi_mid:.1f})，关注反弹"
        else:
            result.rsi_status = RSIStatus.OVERSOLD
            result.rsi_signal = f"⭐ RSI超卖({rsi_mid:.1f}<30)，反弹机会大"
    
    def _generate_signal(self, result: TrendAnalysisResult) -> None:
        """生成买入信号和综合评分"""
        score = 0
        reasons = []
        risks = []
        
        # 趋势评分（30分）
        trend_scores = {
            TrendStatus.STRONG_BULL: 30,
            TrendStatus.BULL: 26,
            TrendStatus.WEAK_BULL: 18,
            TrendStatus.CONSOLIDATION: 12,
            TrendStatus.WEAK_BEAR: 8,
            TrendStatus.BEAR: 4,
            TrendStatus.STRONG_BEAR: 0,
        }
        trend_score = trend_scores.get(result.trend_status, 12)
        score += trend_score
        
        if result.trend_status in [TrendStatus.STRONG_BULL, TrendStatus.BULL]:
            reasons.append(f"✅ {result.trend_status.value}，顺势做多")
        elif result.trend_status in [TrendStatus.BEAR, TrendStatus.STRONG_BEAR]:
            risks.append(f"⚠️ {result.trend_status.value}，不宜做多")
        
        # 乖离率评分（20分）
        bias = result.bias_ma5
        if bias < 0:
            if bias > -3:
                score += 20
                reasons.append(f"✅ 价格略低于MA5({bias:.1f}%)，回踩买点")
            elif bias > -5:
                score += 16
                reasons.append(f"✅ 价格回踩MA5({bias:.1f}%)，观察支撑")
            else:
                score += 8
                risks.append(f"⚠️ 乖离率过大({bias:.1f}%)，可能破位")
        elif bias < 2:
            score += 18
            reasons.append(f"✅ 价格贴近MA5({bias:.1f}%)，介入好时机")
        elif bias < self.BIAS_THRESHOLD:
            score += 14
            reasons.append(f"⚡ 价格略高于MA5({bias:.1f}%)，可小仓介入")
        else:
            score += 4
            risks.append(f"❌ 乖离率过高({bias:.1f}%>5%)，严禁追高！")
        
        # 量能评分（15分）
        volume_scores = {
            VolumeStatus.SHRINK_VOLUME_DOWN: 15,
            VolumeStatus.HEAVY_VOLUME_UP: 12,
            VolumeStatus.NORMAL: 10,
            VolumeStatus.SHRINK_VOLUME_UP: 6,
            VolumeStatus.HEAVY_VOLUME_DOWN: 0,
        }
        vol_score = volume_scores.get(result.volume_status, 8)
        score += vol_score
        
        if result.volume_status == VolumeStatus.SHRINK_VOLUME_DOWN:
            reasons.append("✅ 缩量回调，主力洗盘")
        elif result.volume_status == VolumeStatus.HEAVY_VOLUME_DOWN:
            risks.append("⚠️ 放量下跌，注意风险")
        
        # 支撑评分（10分）
        if result.support_ma5:
            score += 5
            reasons.append("✅ MA5支撑有效")
        if result.support_ma10:
            score += 5
            reasons.append("✅ MA10支撑有效")
        
        # MACD 评分（15分）
        macd_scores = {
            MACDStatus.GOLDEN_CROSS_ZERO: 15,
            MACDStatus.GOLDEN_CROSS: 12,
            MACDStatus.CROSSING_UP: 10,
            MACDStatus.BULLISH: 8,
            MACDStatus.BEARISH: 2,
            MACDStatus.CROSSING_DOWN: 0,
            MACDStatus.DEATH_CROSS: 0,
        }
        macd_score = macd_scores.get(result.macd_status, 5)
        score += macd_score
        
        if result.macd_status in [MACDStatus.GOLDEN_CROSS_ZERO, MACDStatus.GOLDEN_CROSS]:
            reasons.append(f"✅ {result.macd_signal}")
        elif result.macd_status in [MACDStatus.DEATH_CROSS, MACDStatus.CROSSING_DOWN]:
            risks.append(f"⚠️ {result.macd_signal}")
        else:
            reasons.append(result.macd_signal)
        
        # RSI 评分（10分）
        rsi_scores = {
            RSIStatus.OVERSOLD: 10,
            RSIStatus.STRONG_BUY: 8,
            RSIStatus.NEUTRAL: 5,
            RSIStatus.WEAK: 3,
            RSIStatus.OVERBOUGHT: 0,
        }
        rsi_score = rsi_scores.get(result.rsi_status, 5)
        score += rsi_score
        
        if result.rsi_status in [RSIStatus.OVERSOLD, RSIStatus.STRONG_BUY]:
            reasons.append(f"✅ {result.rsi_signal}")
        elif result.rsi_status == RSIStatus.OVERBOUGHT:
            risks.append(f"⚠️ {result.rsi_signal}")
        else:
            reasons.append(result.rsi_signal)
        
        # 综合判断
        result.signal_score = score
        result.signal_reasons = reasons
        result.risk_factors = risks
        
        if score >= 75 and result.trend_status in [TrendStatus.STRONG_BULL, TrendStatus.BULL]:
            result.buy_signal = BuySignal.STRONG_BUY
        elif score >= 60 and result.trend_status in [TrendStatus.STRONG_BULL, TrendStatus.BULL, TrendStatus.WEAK_BULL]:
            result.buy_signal = BuySignal.BUY
        elif score >= 45:
            result.buy_signal = BuySignal.HOLD
        elif score >= 30:
            result.buy_signal = BuySignal.WAIT
        elif result.trend_status in [TrendStatus.BEAR, TrendStatus.STRONG_BEAR]:
            result.buy_signal = BuySignal.STRONG_SELL
        else:
            result.buy_signal = BuySignal.SELL


def analyze_stock(df: pd.DataFrame, code: str) -> TrendAnalysisResult:
    """便捷函数：分析单只股票"""
    analyzer = StockTrendAnalyzer()
    return analyzer.analyze(df, code)