feat: UI/UX全面优化 — CSS变量体系/a11y无障碍/Admin角色化/骨架屏/打印样式

2026-07-18 07:57:23 +08:00
parent 2b5a32ca1e
commit 4b42eb80fd
15 changed files with 1519 additions and 442 deletions
@@ -132,3 +132,134 @@ def compute_comprehensive_score(stock_code, stock_name, technical_score, df=None
        'all_reasons': all_reasons,
        'summary': summary,
    }
+
+
+def compute_comprehensive_score_batch(stocks_data):
+    """
+    批量计算综合评分 — 市场级因素只计算一次，个股级因素逐只计算。
+
+    优化点：
+    - P1 市场情绪、P2 北向、P3 美股、P4 商品、P7 汇率、P6 政策 → 市场级，只算一次
+    - P0 资金面 → 个股级，逐只从DB读取
+    - P5 公告/异动 → 批量模式跳过（需AKShare API + LLM，太慢），在深度分析时补充
+
+    参数:
+        stocks_data: list[dict]，每个元素包含:
+            - stock_code: str    股票代码
+            - stock_name: str    股票名称（可选）
+            - technical_score: float  技术得分（0-100）
+
+    返回:
+        dict: {stock_code: {technical_score, external_score, final_score, verdict, factors, all_reasons, summary}}
+    """
+    # ---- 市场级因素（只计算一次）----
+    market_score = 0
+    market_factors = {}
+    market_reasons = []
+    summaries = []
+
+    # P1: 市场情绪
+    try:
+        from services.market_sentiment import calc_market_sentiment
+        sentiment_result = calc_market_sentiment()
+        market_factors['market_sentiment'] = sentiment_result
+        market_score += sentiment_result.get('score', 0)
+        market_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)}:'
+                f'{sentiment_result.get("limit_down_count", 0)}）'
+            )
+    except Exception as e:
+        logger.warning(f"批量P1市场情绪分析失败: {e}")
+        market_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()
+        market_factors['external'] = ext_result
+        market_score += ext_result.get('total_score', 0)
+        market_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}")
+        market_factors['external'] = {'total_score': 0, 'summary': '分析失败', 'all_reasons': []}
+
+    # P6: 政策面（市场级，只算一次）
+    try:
+        from services.news_analyzer import get_policy_news, analyze_policy_impact
+        policy_news = get_policy_news(days=3)
+        policy_result = analyze_policy_impact(policy_news)
+        market_factors['policy'] = policy_result
+        market_score += policy_result.get('score', 0)
+        market_reasons.extend(policy_result.get('reasons', []))
+        s = policy_result.get('summary', '')
+        if s and '失败' not in s:
+            summaries.append(f'政策面：{s}')
+    except Exception as e:
+        logger.warning(f"批量P6政策面分析失败: {e}")
+        market_factors['policy'] = {'score': 0, 'summary': '分析失败', 'reasons': []}
+
+    # ---- 为每只股票计算个股级因素 ----
+    results = {}
+    for stock in stocks_data:
+        code = stock.get('stock_code', '')
+        name = stock.get('stock_name', '')
+        tech_score = stock.get('technical_score', 50)
+
+        stock_external = market_score
+        stock_factors = {
+            'market_sentiment': market_factors.get('market_sentiment', {}),
+            'external': market_factors.get('external', {}),
+            'policy': market_factors.get('policy', {}),
+        }
+        stock_reasons = list(market_reasons)
+
+        # P0: 资金面（个股级，从DB读取）
+        try:
+            from services.fund_flow_analyzer import analyze_fund_flow
+            fund_result = analyze_fund_flow(code, days=5)
+            stock_factors['fund_flow'] = fund_result
+            stock_external += fund_result.get('score', 0)
+            stock_reasons.extend(fund_result.get('reasons', []))
+        except Exception as e:
+            logger.warning(f"批量P0资金面分析失败 {code}: {e}")
+            stock_factors['fund_flow'] = {'score': 0, 'summary': '分析失败', 'reasons': []}
+
+        # P5: 公告/异动 — 批量模式跳过（需AKShare API + LLM，在深度分析时补充）
+        stock_factors['news'] = {
+            'total_score': 0, 'summary': '批量模式跳过，请使用深度分析查看',
+            'all_reasons': [],
+        }
+
+        # 外部得分上限 ±40
+        stock_external = max(-40, min(40, stock_external))
+        final_score = max(0, min(100, int(tech_score + stock_external)))
+
+        # 评级
+        if final_score >= 80:
+            verdict = '强烈看多'
+        elif final_score >= 65:
+            verdict = '看多'
+        elif final_score >= 50:
+            verdict = '中性偏多'
+        elif final_score >= 35:
+            verdict = '中性偏空'
+        else:
+            verdict = '看空'
+
+        results[code] = {
+            'technical_score': round(tech_score, 0),
+            'external_score': stock_external,
+            'final_score': final_score,
+            'verdict': verdict,
+            'factors': stock_factors,
+            'all_reasons': stock_reasons,
+            'summary': ' | '.join(summaries) if summaries else '',
+        }
+
+    return results
@@ -830,13 +830,15 @@ def compute_bull_stage(signal_status):
    }


-def find_bull_stocks(scan_rows, holding_codes=None):
+def find_bull_stocks(scan_rows, holding_codes=None, scores_map=None):
    """
    从扫描结果中找出潜在牛股，按阶段分组排序。
    
    参数:
        scan_rows: list[dict] 扫描结果列表 (含 code, name, signal_status, indicators, triggered_count)
        holding_codes: set 持仓代码集合
+        scores_map: dict 综合评分映射 {code: {technical_score, external_score, final_score, verdict}}
+                    当提供时，每只股票附加三项得分，并按综合得分排序
    
    返回:
        dict: {
@@ -872,8 +874,21 @@ def find_bull_stocks(scan_rows, holding_codes=None):
            row.get('triggered_count'), is_holding,
        )

+        # 附加综合评分（如果提供了 scores_map）
+        code = row.get('code', '')
+        technical_score = rate
+        external_score = 0
+        final_score = rate
+        verdict = ''
+        if scores_map and code in scores_map:
+            sc = scores_map[code]
+            technical_score = sc.get('technical_score', rate)
+            external_score = sc.get('external_score', 0)
+            final_score = sc.get('final_score', rate)
+            verdict = sc.get('verdict', '')
+
        item = {
-            'code': row.get('code', ''),
+            'code': code,
            'name': row.get('name', ''),
            'stage': stage,
            'stage_name': bull['stage_name'],
@@ -887,16 +902,26 @@ def find_bull_stocks(scan_rows, holding_codes=None):
            'recommend_type': st,
            'recommend_text': disp,
            'recommend_reason': reason,
-            'recommend_rate': rate,
+            'recommend_rate': final_score,
            'is_holding': is_holding,
            'triggered_count': row.get('triggered_count', 0),
+            'technical_score': technical_score,
+            'external_score': external_score,
+            'final_score': final_score,
+            'verdict': verdict,
        }

        stages[stage].append(item)

-    # 每个阶段内按推荐评分降序排序
-    for stage_num in stages:
-        stages[stage_num].sort(key=lambda x: (-x['recommend_rate'], -x['progress']))
+    # 每个阶段内排序：有综合评分时按综合得分→技术得分→进度，否则按推荐评分→进度
+    if scores_map:
+        for stage_num in stages:
+            stages[stage_num].sort(
+                key=lambda x: (-x.get('final_score', 0), -x.get('technical_score', 0), -x['progress'])
+            )
+    else:
+        for stage_num in stages:
+            stages[stage_num].sort(key=lambda x: (-x['recommend_rate'], -x['progress']))

    total = sum(len(v) for v in stages.values())