""" 综合评分引擎 — 整合所有影响因素到统一评分体系 将技术面(基础50%+)与外部因素(加减分项)整合为最终评分。 权重分配: - 技术面评分(compute_deep_analysis 原始分):基础分(0-100) - P0 资金面:±20 - P1 市场情绪:±10 - P2 北向资金:±10 - P3 美股外盘:±10 - P4 大宗商品:±5 - P5 公告/异动:±15 - P6 政策面:±10 - P7 汇率:±3 最终评分 = 技术面基础分 + 外部因素加减分(上限100,下限0) """ import logging logger = logging.getLogger(__name__) def compute_comprehensive_score(stock_code, stock_name, technical_score, df=None): """ 综合评分引擎 — 整合技术面和所有外部因素 参数: stock_code: 股票代码 stock_name: 股票名称 technical_score: float — 技术面基础评分(0-100,来自 compute_deep_analysis) df: K线DataFrame(用于异动检测,可选) 返回: dict: { 'technical_score': float, # 技术面基础分 'external_score': int, # 外部因素总加减分 'final_score': int, # 最终综合评分(0-100) 'verdict': str, # 最终评级 'factors': dict, # 各因素详细数据 'all_reasons': list, # 所有评分原因 'summary': str, # 综合白话总结 } """ factors = {} all_reasons = [] external_score = 0 summaries = [] # ---- P0: 主力资金进出 ---- try: from services.fund_flow_analyzer import analyze_fund_flow fund_result = analyze_fund_flow(stock_code, days=5) factors['fund_flow'] = fund_result external_score += fund_result.get('score', 0) all_reasons.extend(fund_result.get('reasons', [])) s = fund_result.get('summary', '') if s and '暂无' not in s: summaries.append(f'资金面:{s}') except Exception as e: logger.warning(f"P0资金面分析失败: {e}") factors['fund_flow'] = {'score': 0, 'summary': '分析失败', 'reasons': []} # ---- P1: 市场情绪指标 ---- try: from services.market_sentiment import calc_market_sentiment sentiment_result = calc_market_sentiment() factors['market_sentiment'] = sentiment_result external_score += sentiment_result.get('score', 0) all_reasons.extend(sentiment_result.get('reasons', [])) s = sentiment_result.get('sentiment', '') if s and '无数据' not in s: summaries.append(f'市场情绪:{s}(涨跌停{sentiment_result.get("limit_up_count",0)}:{sentiment_result.get("limit_down_count",0)})') except Exception as e: logger.warning(f"P1市场情绪分析失败: {e}") factors['market_sentiment'] = {'score': 0, 'summary': '分析失败', 'reasons': []} # ---- P2/P3/P4/P7: 外部因素(北向/美股/商品/汇率)---- try: from services.external_factors import get_all_external_factors ext_result = get_all_external_factors() factors['external'] = ext_result external_score += ext_result.get('total_score', 0) all_reasons.extend(ext_result.get('all_reasons', [])) s = ext_result.get('summary', '') if s: summaries.append(s) except Exception as e: logger.warning(f"P2-P7外部因素分析失败: {e}") factors['external'] = {'total_score': 0, 'summary': '分析失败', 'all_reasons': []} # ---- P5/P6: 公告/异动/政策 ---- try: from services.news_analyzer import analyze_news_factors news_result = analyze_news_factors(stock_code, stock_name, df) factors['news'] = news_result external_score += news_result.get('total_score', 0) all_reasons.extend(news_result.get('all_reasons', [])) s = news_result.get('summary', '') if s: summaries.append(s) except Exception as e: logger.warning(f"P5/P6消息面分析失败: {e}") factors['news'] = {'total_score': 0, 'summary': '分析失败', 'all_reasons': []} # ---- 最终评分 ---- # 外部因素加减分上限:±40(避免喧宾夺主) external_score = max(-40, min(40, external_score)) final_score = max(0, min(100, int(technical_score + external_score))) # 最终评级 if final_score >= 80: verdict = '强烈看多' elif final_score >= 65: verdict = '看多' elif final_score >= 50: verdict = '中性偏多' elif final_score >= 35: verdict = '中性偏空' else: verdict = '看空' # 综合总结 summary = ' | '.join(summaries) if summaries else '暂无外部因素数据' return { 'technical_score': round(technical_score, 0), 'external_score': external_score, 'final_score': final_score, 'verdict': verdict, 'factors': factors, 'all_reasons': all_reasons, 'summary': summary, }