skills/financial-report-writing/scripts/analysis_engine.py

"""
投资分析引擎
提供基本面分析、技术面分析和估值分析功能
"""

import pandas as pd
import numpy as np
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
from enum import Enum


class InvestmentRating(Enum):
    """投资评级"""
    BUY = "买入"
    ACCUMULATE = "增持"
    HOLD = "持有"
    REDUCE = "减持"
    SELL = "卖出"


class RiskLevel(Enum):
    """风险等级"""
    LOW = "低风险"
    LOW_MEDIUM = "中低风险"
    MEDIUM = "中等风险"
    MEDIUM_HIGH = "中高风险"
    HIGH = "高风险"


@dataclass
class TechnicalAnalysis:
    """技术分析结果"""
    trend: str  # 趋势判断
    support_level: float  # 支撑位
    resistance_level: float  # 阻力位
    volatility: float  # 波动率
    ma_status: str  # 均线状态
    technical_summary: str  # 技术总结


@dataclass
class FundamentalAnalysis:
    """基本面分析结果"""
    roe: float  # 净资产收益率
    roe_comment: str  # ROE评价
    pe_ttm: float  # 市盈率TTM
    pb_lf: float  # 市净率LF
    valuation_comment: str  # 估值评价
    fundamental_summary: str  # 基本面总结


@dataclass
class InvestmentRecommendation:
    """投资建议"""
    rating: InvestmentRating
    target_price: float
    risk_level: RiskLevel
    recommendation_text: str
    key_factors: List[str]


class AnalysisEngine:
    """
    投资分析引擎
    """

    # ROE评价标准
    ROE_THRESHOLDS = {
        'excellent': 20,
        'good': 15,
        'average': 10,
        'poor': 5
    }

    # PE评价标准（参考值，实际应根据行业调整）
    PE_THRESHOLDS = {
        'undervalued': 15,
        'fair': 25,
        'overvalued': 35
    }

    # PB评价标准
    PB_THRESHOLDS = {
        'undervalued': 1.5,
        'fair': 3.0,
        'overvalued': 5.0
    }

    def __init__(self, price_df: pd.DataFrame, financial_data: Dict):
        """
        初始化分析引擎

        Args:
            price_df: 价格数据DataFrame
            financial_data: 财务数据字典
        """
        self.price_df = price_df.copy()
        self.financial_data = financial_data

        # 计算涨跌幅和收益率
        if 'pct_chg' not in self.price_df.columns and 'close' in self.price_df.columns:
            self.price_df['pct_chg'] = self.price_df['close'].pct_change() * 100

    def analyze_technical(self) -> TechnicalAnalysis:
        """
        技术分析

        Returns:
            TechnicalAnalysis对象
        """
        df = self.price_df

        # 计算均线
        if 'ma5' not in df.columns:
            df['ma5'] = df['close'].rolling(window=5).mean()
        if 'ma10' not in df.columns:
            df['ma10'] = df['close'].rolling(window=10).mean()
        if 'ma20' not in df.columns:
            df['ma20'] = df['close'].rolling(window=20).mean()

        latest = df.iloc[-1]

        # 趋势判断
        if latest['close'] > latest['ma5'] > latest['ma10'] > latest['ma20']:
            trend = "强势上涨"
        elif latest['close'] > latest['ma5'] > latest['ma10']:
            trend = "震荡上行"
        elif latest['close'] < latest['ma5'] < latest['ma10'] < latest['ma20']:
            trend = "强势下跌"
        elif latest['close'] < latest['ma5'] < latest['ma10']:
            trend = "震荡下行"
        else:
            trend = "震荡整理"

        # 计算支撑位和阻力位（基于近期高低点）
        recent_data = df.tail(20)
        support_level = recent_data['low'].min()
        resistance_level = recent_data['high'].max()

        # 计算波动率
        returns = df['close'].pct_change().dropna()
        volatility = returns.std() * np.sqrt(52) * 100  # 年化波动率

        # 均线状态
        if latest['ma5'] > latest['ma20']:
            ma_status = "多头排列"
        elif latest['ma5'] < latest['ma20']:
            ma_status = "空头排列"
        else:
            ma_status = "均线粘合"

        # 技术总结
        if trend in ["强势上涨", "震荡上行"] and ma_status == "多头排列":
            technical_summary = "技术面呈上升趋势，建议关注回调买入机会。"
        elif trend in ["强势下跌", "震荡下行"]:
            technical_summary = "技术面呈下降趋势，建议观望或减仓。"
        else:
            technical_summary = "技术面呈震荡格局，建议高抛低吸。"

        return TechnicalAnalysis(
            trend=trend,
            support_level=round(support_level, 2),
            resistance_level=round(resistance_level, 2),
            volatility=round(volatility, 2),
            ma_status=ma_status,
            technical_summary=technical_summary
        )

    def analyze_fundamental(self) -> FundamentalAnalysis:
        """
        基本面分析

        Returns:
            FundamentalAnalysis对象
        """
        # 获取ROE
        roe = self.financial_data.get('roe_wgt', 0) or self.financial_data.get('roe', 0)

        # ROE评价
        if roe >= self.ROE_THRESHOLDS['excellent']:
            roe_comment = "ROE表现优异，公司盈利能力强劲"
        elif roe >= self.ROE_THRESHOLDS['good']:
            roe_comment = "ROE表现良好，盈利能力较强"
        elif roe >= self.ROE_THRESHOLDS['average']:
            roe_comment = "ROE处于行业平均水平"
        else:
            roe_comment = "ROE偏低，需关注盈利能力改善"

        # 获取PE和PB
        pe_ttm = self.financial_data.get('pe_ttm', 0) or self.price_df['pe_ttm'].iloc[-1] if 'pe_ttm' in self.price_df.columns else 0
        pb_lf = self.financial_data.get('pb_lf', 0) or self.price_df['pb_lf'].iloc[-1] if 'pb_lf' in self.price_df.columns else 0

        # 估值评价
        if pe_ttm <= self.PE_THRESHOLDS['undervalued'] and pb_lf <= self.PB_THRESHOLDS['undervalued']:
            valuation_comment = "估值处于历史低位，具备较高安全边际"
        elif pe_ttm <= self.PE_THRESHOLDS['fair']:
            valuation_comment = "估值处于合理区间"
        elif pe_ttm <= self.PE_THRESHOLDS['overvalued']:
            valuation_comment = "估值略高，需关注业绩成长性"
        else:
            valuation_comment = "估值偏高，注意追高风险"

        # 基本面总结
        if roe >= self.ROE_THRESHOLDS['good'] and pe_ttm <= self.PE_THRESHOLDS['fair']:
            fundamental_summary = "公司基本面健康，盈利能力强，估值合理，具备投资价值。"
        elif roe >= self.ROE_THRESHOLDS['good']:
            fundamental_summary = "公司盈利能力较强，但估值偏高，建议等待回调机会。"
        else:
            fundamental_summary = "公司基本面一般，建议谨慎观望。"

        return FundamentalAnalysis(
            roe=round(roe, 2) if roe else 0,
            roe_comment=roe_comment,
            pe_ttm=round(pe_ttm, 2) if pe_ttm else 0,
            pb_lf=round(pb_lf, 2) if pb_lf else 0,
            valuation_comment=valuation_comment,
            fundamental_summary=fundamental_summary
        )

    def generate_recommendation(self,
                               current_price: float,
                               fundamental: FundamentalAnalysis = None,
                               technical: TechnicalAnalysis = None) -> InvestmentRecommendation:
        """
        生成投资建议

        Args:
            current_price: 当前价格
            fundamental: 基本面分析结果
            technical: 技术面分析结果

        Returns:
            InvestmentRecommendation对象
        """
        if fundamental is None:
            fundamental = self.analyze_fundamental()
        if technical is None:
            technical = self.analyze_technical()

        # 综合评分
        score = 0

        # 基本面评分
        if fundamental.roe >= self.ROE_THRESHOLDS['excellent']:
            score += 3
        elif fundamental.roe >= self.ROE_THRESHOLDS['good']:
            score += 2
        elif fundamental.roe >= self.ROE_THRESHOLDS['average']:
            score += 1

        # 估值评分
        if fundamental.pe_ttm <= self.PE_THRESHOLDS['undervalued']:
            score += 3
        elif fundamental.pe_ttm <= self.PE_THRESHOLDS['fair']:
            score += 2
        elif fundamental.pe_ttm <= self.PE_THRESHOLDS['overvalued']:
            score += 1

        # 技术面评分
        if technical.trend in ["强势上涨", "震荡上行"]:
            score += 2
        elif technical.trend == "震荡整理":
            score += 1

        # 风险等级
        if technical.volatility > 50:
            risk_level = RiskLevel.HIGH
        elif technical.volatility > 35:
            risk_level = RiskLevel.MEDIUM_HIGH
        elif technical.volatility > 25:
            risk_level = RiskLevel.MEDIUM
        elif technical.volatility > 15:
            risk_level = RiskLevel.LOW_MEDIUM
        else:
            risk_level = RiskLevel.LOW

        # 目标价格计算（基于PE和支撑阻力）
        target_price_pe = current_price * (self.PE_THRESHOLDS['fair'] / max(fundamental.pe_ttm, 1))
        target_price_resistance = technical.resistance_level * 1.1
        target_price = round((target_price_pe + target_price_resistance) / 2, 2)

        # 投资建议
        key_factors = []

        if score >= 6:
            rating = InvestmentRating.BUY
            recommendation_text = f"综合基本面和技术面分析，该股投资价值较高。目标价{target_price}元，建议逢低买入。"
            key_factors.append("盈利能力强劲，ROE表现优异")
        elif score >= 4:
            rating = InvestmentRating.ACCUMULATE
            recommendation_text = f"公司基本面良好，建议关注回调机会逐步建仓。目标价{target_price}元。"
            key_factors.append("基本面稳健，估值合理")
        elif score >= 2:
            rating = InvestmentRating.HOLD
            recommendation_text = "公司基本面一般，当前估值合理，建议持有观望。"
            key_factors.append("基本面中性，估值合理")
        else:
            rating = InvestmentRating.REDUCE
            recommendation_text = "公司基本面较弱或估值偏高，建议谨慎或减仓。"
            key_factors.append("基本面一般或估值偏高")

        if technical.trend in ["强势上涨", "震荡上行"]:
            key_factors.append("技术面呈上升趋势")
        elif technical.trend in ["强势下跌", "震荡下行"]:
            key_factors.append("技术面呈下降趋势")

        if fundamental.pe_ttm <= self.PE_THRESHOLDS['undervalued']:
            key_factors.append("估值处于历史低位")

        return InvestmentRecommendation(
            rating=rating,
            target_price=target_price,
            risk_level=risk_level,
            recommendation_text=recommendation_text,
            key_factors=key_factors
        )

    def generate_core_summary(self,
                             fundamental: FundamentalAnalysis = None,
                             technical: TechnicalAnalysis = None,
                             recommendation: InvestmentRecommendation = None) -> List[str]:
        """
        生成核心提要要点

        Returns:
            要点列表
        """
        if fundamental is None:
            fundamental = self.analyze_fundamental()
        if technical is None:
            technical = self.analyze_technical()
        if recommendation is None:
            # 获取最新价格
            latest_price = self.price_df['close'].iloc[-1] if 'close' in self.price_df.columns else 0
            recommendation = self.generate_recommendation(latest_price, fundamental, technical)

        summary_points = []

        # ROE要点
        if fundamental.roe > 0:
            summary_points.append(
                f"该公司最新净资产收益率（ROE）为{fundamental.roe}%，{fundamental.roe_comment}"
            )

        # 估值要点
        if fundamental.pe_ttm > 0:
            summary_points.append(
                f"当前市盈率（TTM）为{fundamental.pe_ttm}倍，{fundamental.valuation_comment}"
            )

        # 技术面要点
        summary_points.append(
            f"技术面显示股价呈现{technical.trend}趋势，{technical.technical_summary}"
        )

        # 投资建议要点
        summary_points.append(
            f"给予\"{recommendation.rating.value}\"评级，目标价格{recommendation.target_price}元，风险等级为{recommendation.risk_level.value}"
        )

        return summary_points

    def get_price_statistics(self) -> Dict:
        """
        获取价格统计数据

        Returns:
            统计数据字典
        """
        df = self.price_df

        if len(df) == 0:
            return {}

        return {
            'latest_price': round(df['close'].iloc[-1], 2),
            'highest_price': round(df['high'].max(), 2),
            'lowest_price': round(df['low'].min(), 2),
            'avg_price': round(df['close'].mean(), 2),
            'total_change': round(((df['close'].iloc[-1] / df['close'].iloc[0]) - 1) * 100, 2),
            'max_daily_gain': round(df['pct_chg'].max(), 2) if 'pct_chg' in df.columns else 0,
            'max_daily_loss': round(df['pct_chg'].min(), 2) if 'pct_chg' in df.columns else 0,
            'volatility': round(df['close'].pct_change().std() * np.sqrt(52) * 100, 2)
        }


def perform_comprehensive_analysis(price_df: pd.DataFrame,
                                   financial_data: Dict,
                                   current_price: float = None) -> Dict:
    """
    执行综合分析并返回所有结果

    Args:
        price_df: 价格数据
        financial_data: 财务数据
        current_price: 当前价格（可选）

    Returns:
        包含所有分析结果的字典
    """
    engine = AnalysisEngine(price_df, financial_data)

    if current_price is None:
        current_price = price_df['close'].iloc[-1]

    technical = engine.analyze_technical()
    fundamental = engine.analyze_fundamental()
    recommendation = engine.generate_recommendation(current_price, fundamental, technical)
    summary_points = engine.generate_core_summary(fundamental, technical, recommendation)
    price_stats = engine.get_price_statistics()

    return {
        'technical': technical,
        'fundamental': fundamental,
        'recommendation': recommendation,
        'summary_points': summary_points,
        'price_stats': price_stats
    }


if __name__ == '__main__':
    # 测试代码
    import numpy as np

    # 创建模拟数据
    dates = pd.date_range(start='2025-01-01', periods=60, freq='W')
    np.random.seed(42)
    prices = 30 + np.cumsum(np.random.randn(60) * 0.5)

    price_df = pd.DataFrame({
        'date': dates,
        'open': prices * 0.98,
        'high': prices * 1.03,
        'low': prices * 0.97,
        'close': prices,
        'volume': np.random.randint(1000000, 5000000, 60),
        'pe_ttm': 25 + np.random.randn(60) * 2,
        'pb_lf': 3 + np.random.randn(60) * 0.2
    })

    financial_data = {
        'roe_wgt': 16.5,
        'pe_ttm': 25.3,
        'pb_lf': 3.1,
        'sec_name': '测试公司'
    }

    # 执行分析
    results = perform_comprehensive_analysis(price_df, financial_data)

    print("=== 技术分析 ===")
    print(f"趋势: {results['technical'].trend}")
    print(f"支撑位: {results['technical'].support_level}")
    print(f"阻力位: {results['technical'].resistance_level}")

    print("\n=== 基本面分析 ===")
    print(f"ROE: {results['fundamental'].roe}%")
    print(f"PE TTM: {results['fundamental'].pe_ttm}")
    print(f"PB LF: {results['fundamental'].pb_lf}")

    print("\n=== 投资建议 ===")
    print(f"评级: {results['recommendation'].rating.value}")
    print(f"目标价: {results['recommendation'].target_price}元")
    print(f"风险等级: {results['recommendation'].risk_level.value}")

    print("\n=== 核心提要 ===")
    for i, point in enumerate(results['summary_points'], 1):
        print(f"{i}. {point}")