430 lines
17 KiB
Python
430 lines
17 KiB
Python
"""
|
||
分析引擎 - 核心估值和财务分析模块
|
||
"""
|
||
import pandas as pd
|
||
import numpy as np
|
||
from typing import Dict, List, Tuple, Optional
|
||
import logging
|
||
from datetime import datetime, timedelta
|
||
from config import LOW_VALUE_CRITERIA
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class StockAnalyzer:
|
||
def __init__(self):
|
||
self.criteria = LOW_VALUE_CRITERIA
|
||
|
||
def calculate_valuation_metrics(self, data: Dict) -> Dict:
|
||
"""计算估值指标"""
|
||
try:
|
||
key_metrics = data.get('key_metrics', {})
|
||
company_info = data.get('company_info', {})
|
||
|
||
# 基础估值指标
|
||
pe_ratio = key_metrics.get('pe_ratio', 0)
|
||
pb_ratio = key_metrics.get('pb_ratio', 0)
|
||
ps_ratio = key_metrics.get('ps_ratio', 0)
|
||
peg_ratio = key_metrics.get('peg_ratio', 0)
|
||
|
||
# 市值
|
||
market_cap = company_info.get('market_cap', 0)
|
||
|
||
# 计算估值分数 (0-100分，分数越高表示越被低估)
|
||
valuation_score = 0
|
||
|
||
# PE比率评分
|
||
if pe_ratio > 0 and pe_ratio <= self.criteria['max_pe_ratio']:
|
||
pe_score = max(0, 100 - (pe_ratio / self.criteria['max_pe_ratio']) * 50)
|
||
valuation_score += pe_score * 0.3
|
||
|
||
# PB比率评分
|
||
if pb_ratio > 0 and pb_ratio <= self.criteria['max_pb_ratio']:
|
||
pb_score = max(0, 100 - (pb_ratio / self.criteria['max_pb_ratio']) * 50)
|
||
valuation_score += pb_score * 0.3
|
||
|
||
# PS比率评分
|
||
if ps_ratio > 0 and ps_ratio <= self.criteria['max_ps_ratio']:
|
||
ps_score = max(0, 100 - (ps_ratio / self.criteria['max_ps_ratio']) * 50)
|
||
valuation_score += ps_score * 0.2
|
||
|
||
# PEG比率评分
|
||
if peg_ratio > 0 and peg_ratio <= 1.0:
|
||
peg_score = max(0, 100 - peg_ratio * 50)
|
||
valuation_score += peg_score * 0.2
|
||
|
||
return {
|
||
'pe_ratio': pe_ratio,
|
||
'pb_ratio': pb_ratio,
|
||
'ps_ratio': ps_ratio,
|
||
'peg_ratio': peg_ratio,
|
||
'market_cap': market_cap,
|
||
'valuation_score': min(100, max(0, valuation_score)),
|
||
'is_undervalued': pe_ratio <= self.criteria['max_pe_ratio'] and
|
||
pb_ratio <= self.criteria['max_pb_ratio'] and
|
||
ps_ratio <= self.criteria['max_ps_ratio']
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"计算估值指标失败: {e}")
|
||
return {'valuation_score': 0, 'is_undervalued': False}
|
||
|
||
def calculate_financial_health(self, data: Dict) -> Dict:
|
||
"""计算财务健康度"""
|
||
try:
|
||
key_metrics = data.get('key_metrics', {})
|
||
financial_statements = data.get('financial_statements', {})
|
||
|
||
# 获取最新财务数据
|
||
latest_financial = self._get_latest_financial_data(financial_statements)
|
||
|
||
# 流动性指标
|
||
current_ratio = key_metrics.get('current_ratio', 0)
|
||
quick_ratio = key_metrics.get('quick_ratio', 0)
|
||
|
||
# 偿债能力指标
|
||
debt_to_equity = key_metrics.get('debt_to_equity', 0)
|
||
debt_ratio = self._calculate_debt_ratio(latest_financial)
|
||
|
||
# 盈利能力指标
|
||
roe = key_metrics.get('return_on_equity', 0)
|
||
roa = key_metrics.get('return_on_assets', 0)
|
||
profit_margin = key_metrics.get('profit_margin', 0)
|
||
operating_margin = key_metrics.get('operating_margin', 0)
|
||
|
||
# 计算财务健康分数
|
||
health_score = 0
|
||
|
||
# 流动性评分 (30%)
|
||
if current_ratio >= self.criteria['min_current_ratio']:
|
||
liquidity_score = min(100, (current_ratio / 2.0) * 100)
|
||
health_score += liquidity_score * 0.3
|
||
|
||
# 偿债能力评分 (30%)
|
||
if debt_ratio <= self.criteria['max_debt_ratio']:
|
||
debt_score = max(0, 100 - (debt_ratio / self.criteria['max_debt_ratio']) * 100)
|
||
health_score += debt_score * 0.3
|
||
|
||
# 盈利能力评分 (40%)
|
||
if roe >= self.criteria['min_roe']:
|
||
profitability_score = min(100, (roe / 0.2) * 100) # 20% ROE为满分
|
||
health_score += profitability_score * 0.4
|
||
|
||
return {
|
||
'current_ratio': current_ratio,
|
||
'quick_ratio': quick_ratio,
|
||
'debt_to_equity': debt_to_equity,
|
||
'debt_ratio': debt_ratio,
|
||
'roe': roe,
|
||
'roa': roa,
|
||
'profit_margin': profit_margin,
|
||
'operating_margin': operating_margin,
|
||
'health_score': min(100, max(0, health_score)),
|
||
'is_healthy': (current_ratio >= self.criteria['min_current_ratio'] and
|
||
debt_ratio <= self.criteria['max_debt_ratio'] and
|
||
roe >= self.criteria['min_roe'])
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"计算财务健康度失败: {e}")
|
||
return {'health_score': 0, 'is_healthy': False}
|
||
|
||
def calculate_growth_metrics(self, data: Dict) -> Dict:
|
||
"""计算成长性指标"""
|
||
try:
|
||
key_metrics = data.get('key_metrics', {})
|
||
financial_statements = data.get('financial_statements', {})
|
||
|
||
# 收入增长率
|
||
revenue_growth = key_metrics.get('revenue_growth', 0)
|
||
earnings_growth = key_metrics.get('earnings_growth', 0)
|
||
|
||
# 计算历史增长率
|
||
historical_growth = self._calculate_historical_growth(financial_statements)
|
||
|
||
# 成长性评分
|
||
growth_score = 0
|
||
|
||
# 收入增长率评分
|
||
if revenue_growth > 0:
|
||
revenue_score = min(100, revenue_growth * 100)
|
||
growth_score += revenue_score * 0.4
|
||
|
||
# 利润增长率评分
|
||
if earnings_growth > 0:
|
||
earnings_score = min(100, earnings_growth * 100)
|
||
growth_score += earnings_score * 0.4
|
||
|
||
# 历史增长率评分
|
||
if historical_growth > 0:
|
||
historical_score = min(100, historical_growth * 100)
|
||
growth_score += historical_score * 0.2
|
||
|
||
return {
|
||
'revenue_growth': revenue_growth,
|
||
'earnings_growth': earnings_growth,
|
||
'historical_growth': historical_growth,
|
||
'growth_score': min(100, max(0, growth_score)),
|
||
'is_growing': revenue_growth > 0.05 and earnings_growth > 0.05 # 5%以上增长
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"计算成长性指标失败: {e}")
|
||
return {'growth_score': 0, 'is_growing': False}
|
||
|
||
def calculate_risk_metrics(self, data: Dict) -> Dict:
|
||
"""计算风险指标"""
|
||
try:
|
||
key_metrics = data.get('key_metrics', {})
|
||
stock_prices = data.get('stock_prices', pd.DataFrame())
|
||
|
||
# Beta系数
|
||
beta = key_metrics.get('beta', 1.0)
|
||
|
||
# 计算波动率
|
||
volatility = self._calculate_volatility(stock_prices)
|
||
|
||
# 计算最大回撤
|
||
max_drawdown = self._calculate_max_drawdown(stock_prices)
|
||
|
||
# 风险评分 (分数越高表示风险越低)
|
||
risk_score = 100
|
||
|
||
# Beta风险评分
|
||
if beta > 1.5:
|
||
risk_score -= (beta - 1.5) * 20
|
||
elif beta < 0.8:
|
||
risk_score -= (0.8 - beta) * 10
|
||
|
||
# 波动率风险评分
|
||
if volatility > 0.3: # 30%以上波动率
|
||
risk_score -= (volatility - 0.3) * 100
|
||
|
||
# 回撤风险评分
|
||
if max_drawdown > 0.2: # 20%以上回撤
|
||
risk_score -= (max_drawdown - 0.2) * 100
|
||
|
||
return {
|
||
'beta': beta,
|
||
'volatility': volatility,
|
||
'max_drawdown': max_drawdown,
|
||
'risk_score': max(0, min(100, risk_score)),
|
||
'risk_level': self._get_risk_level(risk_score)
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"计算风险指标失败: {e}")
|
||
return {'risk_score': 50, 'risk_level': '中等'}
|
||
|
||
def perform_dcf_valuation(self, data: Dict) -> Dict:
|
||
"""执行DCF估值分析"""
|
||
try:
|
||
financial_statements = data.get('financial_statements', {})
|
||
key_metrics = data.get('key_metrics', {})
|
||
|
||
# 获取最新财务数据
|
||
latest_financial = self._get_latest_financial_data(financial_statements)
|
||
|
||
if not latest_financial:
|
||
return {'dcf_value': 0, 'dcf_score': 0}
|
||
|
||
# 基础参数
|
||
revenue = latest_financial.get('revenue', 0)
|
||
net_income = latest_financial.get('net_income', 0)
|
||
total_assets = latest_financial.get('total_assets', 0)
|
||
|
||
if revenue <= 0 or net_income <= 0:
|
||
return {'dcf_value': 0, 'dcf_score': 0}
|
||
|
||
# 假设参数
|
||
growth_rate = min(0.1, max(0.02, key_metrics.get('revenue_growth', 0.05))) # 2%-10%
|
||
discount_rate = 0.1 # 10%折现率
|
||
terminal_growth_rate = 0.03 # 3%永续增长率
|
||
|
||
# 计算自由现金流 (简化版)
|
||
fcf = net_income * 0.8 # 假设80%的净利润为自由现金流
|
||
|
||
# 5年预测
|
||
years = 5
|
||
dcf_value = 0
|
||
|
||
for year in range(1, years + 1):
|
||
future_fcf = fcf * ((1 + growth_rate) ** year)
|
||
discounted_fcf = future_fcf / ((1 + discount_rate) ** year)
|
||
dcf_value += discounted_fcf
|
||
|
||
# 终值计算
|
||
terminal_fcf = fcf * ((1 + growth_rate) ** years) * (1 + terminal_growth_rate)
|
||
terminal_value = terminal_fcf / (discount_rate - terminal_growth_rate)
|
||
discounted_terminal_value = terminal_value / ((1 + discount_rate) ** years)
|
||
|
||
total_dcf_value = dcf_value + discounted_terminal_value
|
||
|
||
# 获取当前股价进行对比
|
||
current_price = self._get_current_price(data)
|
||
if current_price > 0:
|
||
dcf_score = min(100, (total_dcf_value / current_price) * 50) # 50%以上溢价为满分
|
||
else:
|
||
dcf_score = 0
|
||
|
||
return {
|
||
'dcf_value': total_dcf_value,
|
||
'current_price': current_price,
|
||
'dcf_score': dcf_score,
|
||
'is_undervalued_dcf': total_dcf_value > current_price if current_price > 0 else False
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"DCF估值分析失败: {e}")
|
||
return {'dcf_value': 0, 'dcf_score': 0}
|
||
|
||
def generate_investment_recommendation(self, analysis_results: Dict) -> Dict:
|
||
"""生成投资建议"""
|
||
try:
|
||
valuation_score = analysis_results.get('valuation_score', 0)
|
||
health_score = analysis_results.get('financial_health_score', 0)
|
||
growth_score = analysis_results.get('growth_score', 0)
|
||
risk_score = analysis_results.get('risk_score', 0)
|
||
|
||
# 计算综合评分
|
||
overall_score = (
|
||
valuation_score * 0.4 +
|
||
health_score * 0.3 +
|
||
growth_score * 0.2 +
|
||
risk_score * 0.1
|
||
)
|
||
|
||
# 生成建议
|
||
if overall_score >= 80:
|
||
recommendation = "强烈买入"
|
||
confidence = "高"
|
||
elif overall_score >= 65:
|
||
recommendation = "买入"
|
||
confidence = "中高"
|
||
elif overall_score >= 50:
|
||
recommendation = "持有"
|
||
confidence = "中等"
|
||
elif overall_score >= 35:
|
||
recommendation = "观望"
|
||
confidence = "中低"
|
||
else:
|
||
recommendation = "卖出"
|
||
confidence = "高"
|
||
|
||
# 风险提示
|
||
risk_warnings = []
|
||
if risk_score < 40:
|
||
risk_warnings.append("高风险投资")
|
||
if health_score < 50:
|
||
risk_warnings.append("财务健康度较低")
|
||
if growth_score < 30:
|
||
risk_warnings.append("成长性不足")
|
||
|
||
return {
|
||
'recommendation': recommendation,
|
||
'confidence': confidence,
|
||
'overall_score': overall_score,
|
||
'risk_warnings': risk_warnings,
|
||
'key_strengths': self._get_key_strengths(analysis_results),
|
||
'key_concerns': self._get_key_concerns(analysis_results)
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"生成投资建议失败: {e}")
|
||
return {'recommendation': '无法分析', 'overall_score': 0}
|
||
|
||
def _get_latest_financial_data(self, financial_statements: Dict) -> Dict:
|
||
"""获取最新财务数据"""
|
||
if not financial_statements:
|
||
return {}
|
||
|
||
# 按年份排序，获取最新的数据
|
||
sorted_periods = sorted(financial_statements.keys(), reverse=True)
|
||
return financial_statements.get(sorted_periods[0], {})
|
||
|
||
def _calculate_debt_ratio(self, financial_data: Dict) -> float:
|
||
"""计算债务比率"""
|
||
total_liabilities = financial_data.get('total_liabilities', 0)
|
||
total_assets = financial_data.get('total_assets', 0)
|
||
|
||
if total_assets > 0:
|
||
return total_liabilities / total_assets
|
||
return 0
|
||
|
||
def _calculate_historical_growth(self, financial_statements: Dict) -> float:
|
||
"""计算历史增长率"""
|
||
if len(financial_statements) < 2:
|
||
return 0
|
||
|
||
# 获取最近两年的收入数据
|
||
sorted_periods = sorted(financial_statements.keys(), reverse=True)
|
||
if len(sorted_periods) < 2:
|
||
return 0
|
||
|
||
current_revenue = financial_statements[sorted_periods[0]].get('revenue', 0)
|
||
previous_revenue = financial_statements[sorted_periods[1]].get('revenue', 0)
|
||
|
||
if previous_revenue > 0:
|
||
return (current_revenue - previous_revenue) / previous_revenue
|
||
return 0
|
||
|
||
def _calculate_volatility(self, stock_prices: pd.DataFrame) -> float:
|
||
"""计算波动率"""
|
||
if stock_prices.empty or 'close' not in stock_prices.columns:
|
||
return 0
|
||
|
||
returns = stock_prices['close'].pct_change().dropna()
|
||
return returns.std() * np.sqrt(252) # 年化波动率
|
||
|
||
def _calculate_max_drawdown(self, stock_prices: pd.DataFrame) -> float:
|
||
"""计算最大回撤"""
|
||
if stock_prices.empty or 'close' not in stock_prices.columns:
|
||
return 0
|
||
|
||
prices = stock_prices['close']
|
||
peak = prices.expanding().max()
|
||
drawdown = (prices - peak) / peak
|
||
return abs(drawdown.min())
|
||
|
||
def _get_risk_level(self, risk_score: float) -> str:
|
||
"""获取风险等级"""
|
||
if risk_score >= 80:
|
||
return "低"
|
||
elif risk_score >= 60:
|
||
return "中低"
|
||
elif risk_score >= 40:
|
||
return "中等"
|
||
elif risk_score >= 20:
|
||
return "中高"
|
||
else:
|
||
return "高"
|
||
|
||
def _get_current_price(self, data: Dict) -> float:
|
||
"""获取当前股价"""
|
||
stock_prices = data.get('stock_prices', pd.DataFrame())
|
||
if not stock_prices.empty and 'close' in stock_prices.columns:
|
||
return stock_prices['close'].iloc[-1]
|
||
return 0
|
||
|
||
def _get_key_strengths(self, analysis_results: Dict) -> List[str]:
|
||
"""获取关键优势"""
|
||
strengths = []
|
||
|
||
if analysis_results.get('valuation_score', 0) > 70:
|
||
strengths.append("估值吸引力高")
|
||
if analysis_results.get('financial_health_score', 0) > 70:
|
||
strengths.append("财务健康度良好")
|
||
if analysis_results.get('growth_score', 0) > 70:
|
||
strengths.append("成长性优秀")
|
||
if analysis_results.get('risk_score', 0) > 70:
|
||
strengths.append("风险控制良好")
|
||
|
||
return strengths
|
||
|
||
def _get_key_concerns(self, analysis_results: Dict) -> List[str]:
|
||
"""获取关键担忧"""
|
||
concerns = []
|
||
|
||
if analysis_results.get('valuation_score', 0) < 40:
|
||
concerns.append("估值偏高")
|
||
if analysis_results.get('financial_health_score', 0) < 40:
|
||
concerns.append("财务健康度不佳")
|
||
if analysis_results.get('growth_score', 0) < 40:
|
||
concerns.append("成长性不足")
|
||
if analysis_results.get('risk_score', 0) < 40:
|
||
concerns.append("风险较高")
|
||
|
||
return concerns |