fix: 修复外部因素数据源 - 北向资金改为南向资金，修复P0-P7全部外部因素

- P0资金面：DB数据过期时用麦蕊API获取资金流向 - P1市场情绪：去掉turnover字段依赖（DB无此字段） - P2南向资金：北向实时数据已停公布，改用南向资金(港股通)替代 - P3美股：用腾讯财经API替代失效的AKShare接口 - P4大宗商品：用腾讯财经API替代，修复var_name解析 - P7汇率：用新浪财经API替代失效的AKShare接口 - 修复评分详情API技术得分硬编码问题 - 前端传入tech_score参数确保明细与列表分数一致
2026-07-18 16:45:58 +08:00
parent bb5a72767f
commit 79e869eeda
13 changed files with 637 additions and 193 deletions
@@ -2,15 +2,21 @@
 外部因素分析模块（P2/P3/P4/P7）

 包含：
- P2: 北向资金（外资动向）
+- P2: 南向资金（港股通跨境资金动向）
 - P3: 美股隔夜板块变化
 - P4: 大宗商品价格
 - P7: 汇率变化

-数据源：AKShare（开源免费）
+数据源：
+- P2: AKShare stock_hsgt_hist_em（南向资金历史）+ stock_hsgt_fund_flow_summary_em（今日汇总）
+- P3: 腾讯财经API（美股指数实时）
+- P4: 腾讯财经API（商品期货实时）
+- P7: 新浪财经API（人民币汇率）
+
 所有数据采集均带超时和异常处理，失败时返回中性评分不影响主流程。
 """
 import logging
+import requests
 from datetime import datetime, timedelta

 logger = logging.getLogger(__name__)
@@ -36,84 +42,118 @@ def _set_cache(key, value):


 # ═══════════════════════════════════════════════
-# P2: 北向资金
+# P2: 南向资金
 # ═══════════════════════════════════════════════

-def get_northbound_capital():
+def get_southbound_capital():
    """
-    获取北向资金净流入数据
+    获取跨境资金流向数据
+
+南向资金（港股通）反映内地资金配置港股的意愿，是跨境资金情绪的重要指标：
+    - 南向净流入 > 0：内地资金积极配置港股，大中华区risk-on，对A股偏正面
+    - 南向净流入 < 0：内地资金撤出港股，risk-off，对A股偏负面
+
+    数据源：
+    1. AKShare stock_hsgt_fund_flow_summary_em（今日汇总）
+    2. AKShare stock_hsgt_hist_em（南向资金历史，用于连续天数计算）

    返回:
        dict: {
-            'net_inflow': float,       # 今日净流入（亿）
+            'net_inflow': float,       # 今日南向净流入（亿）
            'score': int,              # 评分增减（-10 ~ +10）
            'summary': str,            # 白话总结
            'reasons': list,           # 评分原因
        }
    """
-    cached = _get_cache('northbound')
+    cached = _get_cache('southbound')
    if cached:
        return cached

    try:
        import akshare as ak
+        import pandas as pd

-        # 获取北向资金净流入数据
-        df = ak.stock_hsgt_north_net_flow_in_em(symbol="北向")
-        if df is None or df.empty:
-            return _neutral_result('北向资金数据为空')
+        # 1. 用 stock_hsgt_fund_flow_summary_em 获取今日汇总
+        today_inflow = 0
+        try:
+            df_summary = ak.stock_hsgt_fund_flow_summary_em()
+            if df_summary is not None and not df_summary.empty:
+                # 筛选南向资金行
+                south_rows = df_summary[df_summary['资金方向'] == '南向']
+                if not south_rows.empty:
+                    # 成交净买额列求和
+                    vals = south_rows['成交净买额'].tolist()
+                    today_inflow = float(sum(v for v in vals if pd.notna(v) and v != 0))
+        except Exception as e:
+            logger.debug(f"stock_hsgt_fund_flow_summary_em失败: {e}")

-        # 取最近5个交易日
-        recent = df.tail(5)
-        today_inflow = float(recent.iloc[-1].get('当日净流入', 0) or 0)
-
-        # 连续流入/流出天数
+        # 2. 用 stock_hsgt_hist_em 获取南向资金历史（计算连续天数）
        consecutive_inflow = 0
        consecutive_outflow = 0
-        for _, row in recent[::-1].iterrows():
-            val = float(row.get('当日净流入', 0) or 0)
-            if val > 0:
-                if consecutive_outflow > 0:
-                    break
-                consecutive_inflow += 1
-            elif val < 0:
-                if consecutive_inflow > 0:
-                    break
-                consecutive_outflow += 1
+        try:
+            df = ak.stock_hsgt_hist_em(symbol="南向资金")
+            if df is not None and not df.empty:
+                recent = df.tail(5)
+                for _, row in recent[::-1].iterrows():
+                    val = float(row.get('当日成交净买额', 0) or 0)
+                    if str(val) == 'nan' or pd.isna(val):
+                        val = 0
+                    if val > 0:
+                        if consecutive_outflow > 0:
+                            break
+                        consecutive_inflow += 1
+                    elif val < 0:
+                        if consecutive_inflow > 0:
+                            break
+                        consecutive_outflow += 1
+        except Exception as e:
+            logger.debug(f"stock_hsgt_hist_em南向失败: {e}")

-        # 评分
+        # 如果今日数据也为0或NaN，说明无法获取
+        if str(today_inflow) == 'nan' or today_inflow == 0:
+            # 尝试从历史数据取最新值
+            try:
+                df = ak.stock_hsgt_hist_em(symbol="南向资金")
+                if df is not None and not df.empty:
+                    last_val = float(df.iloc[-1].get('当日成交净买额', 0) or 0)
+                    if str(last_val) != 'nan' and not pd.isna(last_val) and last_val != 0:
+                        today_inflow = last_val
+            except Exception:
+                pass
+
+        # 评分（基于南向资金，逻辑与北向一致：净流入=正面，净流出=负面）
        score = 0
        reasons = []
        summary_parts = []

-        if today_inflow > 50:
+        if today_inflow > 80:
            score += 5
-            reasons.append(f'北向今日净流入{today_inflow:.1f}亿(+5)')
-            summary_parts.append(f'外资今日大幅买入{today_inflow:.1f}亿元')
-        elif today_inflow > 20:
+            reasons.append(f'南向今日净流入{today_inflow:.1f}亿(+5)')
+            summary_parts.append(f'南向资金大幅流入{today_inflow:.1f}亿元，跨境资金情绪偏暖')
+        elif today_inflow > 30:
            score += 3
-            reasons.append(f'北向今日净流入{today_inflow:.1f}亿(+3)')
-            summary_parts.append(f'外资今日净流入{today_inflow:.1f}亿元')
-        elif today_inflow < -50:
+            reasons.append(f'南向今日净流入{today_inflow:.1f}亿(+3)')
+            summary_parts.append(f'南向资金净流入{today_inflow:.1f}亿元')
+        elif today_inflow < -80:
            score -= 5
-            reasons.append(f'北向今日净流出{abs(today_inflow):.1f}亿(-5)')
-            summary_parts.append(f'外资今日大幅卖出{abs(today_inflow):.1f}亿元')
-        elif today_inflow < -20:
+            reasons.append(f'南向今日净流出{abs(today_inflow):.1f}亿(-5)')
+            summary_parts.append(f'南向资金大幅流出{abs(today_inflow):.1f}亿元，跨境资金情绪偏冷')
+        elif today_inflow < -30:
            score -= 3
-            reasons.append(f'北向今日净流出{abs(today_inflow):.1f}亿(-3)')
-            summary_parts.append(f'外资今日净流出{abs(today_inflow):.1f}亿元')
+            reasons.append(f'南向今日净流出{abs(today_inflow):.1f}亿(-3)')
+            summary_parts.append(f'南向资金净流出{abs(today_inflow):.1f}亿元')
        else:
-            summary_parts.append(f'外资今日净流入{today_inflow:.1f}亿元，方向不明')
+            summary_parts.append(f'南向资金净流入{today_inflow:.1f}亿元，方向不明')

        if consecutive_inflow >= 3:
            score += 3
-            reasons.append(f'北向连续{consecutive_inflow}日净流入(+3)')
-            summary_parts.append(f'已连续{consecutive_inflow}天买入')
+            reasons.append(f'南向连续{consecutive_inflow}日净流入(+3)')
+            summary_parts.append(f'已连续{consecutive_inflow}天流入')

        if consecutive_outflow >= 3:
            score -= 3
-            reasons.append(f'北向连续{consecutive_outflow}日净流出(-3)')
-            summary_parts.append(f'已连续{consecutive_outflow}天卖出')
+            reasons.append(f'南向连续{consecutive_outflow}日净流出(-3)')
+            summary_parts.append(f'已连续{consecutive_outflow}天流出')

        score = max(-10, min(10, score))

@@ -125,12 +165,12 @@ def get_northbound_capital():
            'summary': '，'.join(summary_parts),
            'reasons': reasons,
        }
-        _set_cache('northbound', result)
+        _set_cache('southbound', result)
        return result

    except Exception as e:
-        logger.warning(f"获取北向资金数据失败: {e}")
-        return _neutral_result('北向资金数据获取失败')
+        logger.warning(f"获取跨境资金数据失败: {e}")
+        return _neutral_result('南向资金数据获取失败')


 # ═══════════════════════════════════════════════
@@ -156,10 +196,12 @@ def get_us_market_overview():
    """
    获取美股隔夜收盘数据，计算外盘情绪

+    数据源：腾讯财经API（qt.gtimg.cn）
+    获取道琼斯、纳斯达克、标普500三大指数实时行情。
+
    返回:
        dict: {
            'indices': dict,          # 三大指数涨跌
-            'sectors': dict,          # 主要板块涨跌
            'score': int,             # 评分增减（-10 ~ +10）
            'summary': str,           # 白话总结
            'reasons': list,          # 评分原因
@@ -171,32 +213,35 @@ def get_us_market_overview():
        return cached

    try:
-        import akshare as ak
+        # 腾讯财经API获取美股指数
+        # 格式: v_usDJI="200~道琼斯~.DJI~price~...~change_pct~..."
+        url = 'https://qt.gtimg.cn/q=usDJI,usIXIC,usSPX'
+        resp = requests.get(url, timeout=10)
+        text = resp.content.decode('gbk', errors='replace')

-        # 获取全球主要指数
-        df = ak.index_global()
-        if df is None or df.empty:
-            return _neutral_result('美股指数数据为空')
-
-        # 筛选美股主要指数
        us_indices = {}
-        for _, row in df.iterrows():
-            name = str(row.get('名称', ''))
-            if '纳斯达克' in name:
-                us_indices['nasdaq'] = {
-                    'name': name,
-                    'change_pct': float(row.get('涨跌幅', 0) or 0),
-                }
-            elif '道琼斯' in name:
-                us_indices['dow'] = {
-                    'name': name,
-                    'change_pct': float(row.get('涨跌幅', 0) or 0),
-                }
-            elif '标普500' in name:
-                us_indices['sp500'] = {
-                    'name': name,
-                    'change_pct': float(row.get('涨跌幅', 0) or 0),
-                }
+        for line in text.strip().split(';'):
+            line = line.strip()
+            if not line or 'v_pv_none_match' in line:
+                continue
+            # 解析 v_usDJI="..."
+            if '=' not in line:
+                continue
+            var_name = line.split('=')[0].strip().replace('var ', '').replace('v_', '')
+            value = line.split('"')[1] if '"' in line else ''
+            fields = value.split('~')
+            if len(fields) < 33:
+                continue
+
+            name = fields[1]
+            change_pct = float(fields[32]) if fields[32] else 0
+
+            if 'DJI' in var_name.upper() or '道琼斯' in name:
+                us_indices['dow'] = {'name': name, 'change_pct': change_pct}
+            elif 'IXIC' in var_name.upper() or '纳斯达克' in name:
+                us_indices['nasdaq'] = {'name': name, 'change_pct': change_pct}
+            elif 'SPX' in var_name.upper() or '标普' in name:
+                us_indices['sp500'] = {'name': name, 'change_pct': change_pct}

        if not us_indices:
            return _neutral_result('未找到美股指数')
@@ -301,6 +346,9 @@ def get_commodity_overview():
    """
    获取主要大宗商品价格变化

+    数据源：腾讯财经API（qt.gtimg.cn）
+    获取纽约黄金、纽约原油、美铜等商品期货实时行情。
+
    返回:
        dict: {
            'commodities': dict,      # 各商品涨跌
@@ -315,28 +363,46 @@ def get_commodity_overview():
        return cached

    try:
-        import akshare as ak
+        # 腾讯财经API获取商品期货
+        # hf_GC=纽约黄金, hf_CL=纽约原油, hf_HG=美铜
+        url = 'https://qt.gtimg.cn/q=hf_GC,hf_CL,hf_HG'
+        resp = requests.get(url, timeout=10)
+        text = resp.content.decode('gbk', errors='replace')

-        # 获取国内商品期货行情
-        df = ak.futures_main_sina()
-        if df is None or df.empty:
-            return _neutral_result('大宗商品数据为空')
+        # 商品名称映射
+        symbol_map = {
+            'hf_GC': '黄金',
+            'hf_CL': '原油',
+            'hf_HG': '铜',
+        }

-        # 关注的商品
-        target_commodities = ['原油', '黄金', '铜', '螺纹钢', '碳酸锂']
        commodities = {}
+        for line in text.strip().split(';'):
+            line = line.strip()
+            if not line or 'v_pv_none_match' in line:
+                continue
+            if '=' not in line:
+                continue
+            var_name = line.split('=')[0].strip().replace('var ', '').replace('v_', '')
+            value = line.split('"')[1] if '"' in line else ''
+            fields = value.split(',')
+            if len(fields) < 10:
+                continue

-        for _, row in df.iterrows():
-            symbol = str(row.get('symbol', ''))
-            for target in target_commodities:
-                if target in symbol:
-                    change = float(row.get('change', 0) or 0)
-                    pct = float(row.get('change_pct', 0) or 0)
-                    commodities[target] = {
-                        'symbol': symbol,
-                        'change_pct': round(pct, 2),
-                    }
-                    break
+            target = symbol_map.get(var_name)
+            if not target:
+                continue
+
+            # 腾讯商品格式: price,change_pct,prev_close,open,high,low,time,...,name
+            current_price = float(fields[0]) if fields[0] else 0
+            change_pct = float(fields[1]) if fields[1] else 0
+            name = fields[-1].rstrip(';"')
+
+            commodities[target] = {
+                'price': current_price,
+                'change_pct': round(change_pct, 2),
+                'name': name,
+            }

        if not commodities:
            return _neutral_result('未找到关注的大宗商品')
@@ -415,6 +481,9 @@ def get_fx_overview():
    """
    获取人民币汇率变化

+    数据源：新浪财经API（hq.sinajs.cn）
+    获取在岸人民币兑美元实时汇率。
+
    返回:
        dict: {
            'usd_cny': float,         # 美元兑人民币汇率
@@ -431,25 +500,37 @@ def get_fx_overview():
        return cached

    try:
-        import akshare as ak
+        # 新浪财经API获取在岸人民币汇率
+        # 格式: var hq_str_fx_susdcny="time,bid,ask,prev_close,...,name,change_pct,..."
+        url = 'https://hq.sinajs.cn/list=fx_susdcny'
+        resp = requests.get(url, timeout=10, headers={'Referer': 'https://finance.sina.com.cn'})
+        text = resp.content.decode('gbk', errors='replace')

-        # 获取人民币汇率
-        df = ak.currency_boc_sina(symbol="美元")
-        if df is None or df.empty:
-            return _neutral_result('汇率数据为空')
+        # 解析汇率数据
+        if 'hq_str_fx_susdcny' not in text:
+            return _neutral_result('汇率数据解析失败')

-        # 取最近2条计算变化
-        recent = df.tail(2)
-        if len(recent) < 2:
-            return _neutral_result('汇率数据不足')
+        value = text.split('"')[1] if '"' in text else ''
+        fields = value.split(',')
+        if len(fields) < 11:
+            return _neutral_result('汇率数据格式异常')

-        today_rate = float(recent.iloc[-1].get('中行折算价', 0) or 0)
-        prev_rate = float(recent.iloc[-2].get('中行折算价', 0) or 0)
+        # 新浪汇率格式: time,bid,ask,prev_close,?,mid,?,?,?,name,change_pct,...
+        today_rate = float(fields[5]) if fields[5] else 0  # 中间价
+        prev_rate = float(fields[3]) if fields[3] else 0   # 昨收价
+        change_pct = float(fields[10]) if fields[10] else 0  # 涨跌幅

+        if today_rate == 0:
+            today_rate = float(fields[1]) if fields[1] else 0
        if prev_rate == 0:
+            prev_rate = float(fields[3]) if fields[3] else 0
+
+        if today_rate == 0 or prev_rate == 0:
            return _neutral_result('汇率数据异常')

-        change_pct = round((today_rate / prev_rate - 1) * 100, 3)
+        # 如果涨跌幅为0，自行计算
+        if change_pct == 0:
+            change_pct = round((today_rate / prev_rate - 1) * 100, 3)

        # 判断方向（美元兑人民币：涨=人民币贬值，跌=人民币升值）
        if change_pct > 0.1:
@@ -499,34 +580,34 @@ def get_all_external_factors():
    获取所有外部因素数据，返回综合结果

    返回:
-        dict: 包含北向资金、美股、大宗商品、汇率的综合数据
+        dict: 包含南向资金、美股、大宗商品、汇率的综合数据
    """
-    northbound = get_northbound_capital()
+    southbound = get_southbound_capital()
    us_market = get_us_market_overview()
    commodity = get_commodity_overview()
    fx = get_fx_overview()

    total_score = (
-        northbound.get('score', 0) +
+        southbound.get('score', 0) +
        us_market.get('score', 0) +
        commodity.get('score', 0) +
        fx.get('score', 0)
    )

    all_reasons = []
-    all_reasons.extend(northbound.get('reasons', []))
+    all_reasons.extend(southbound.get('reasons', []))
    all_reasons.extend(us_market.get('reasons', []))
    all_reasons.extend(commodity.get('reasons', []))
    all_reasons.extend(fx.get('reasons', []))

    summaries = []
-    for name, data in [('北向资金', northbound), ('美股', us_market), ('大宗商品', commodity), ('汇率', fx)]:
+    for name, data in [('南向资金', southbound), ('美股', us_market), ('大宗商品', commodity), ('汇率', fx)]:
        s = data.get('summary', '')
        if s and '失败' not in s and '为空' not in s:
            summaries.append(f'{name}：{s}')

    return {
-        'northbound_capital': northbound,
+        'southbound_capital': southbound,
        'us_market': us_market,
        'commodity': commodity,
        'fx': fx,
@@ -73,10 +73,105 @@ def get_fund_flow_history(stock_code, days=10):
        put_db(conn)


+def _get_fund_flow_from_mairui(stock_code, days=10):
+    """从麦蕊智数API获取资金流向数据，转换为与DB记录相同的格式。
+
+    API: https://api.mairuiapi.com/hsstock/history/transaction/{code}/{licence}?lt={n}
+    字段: zmbtdcje=主买特大单, zmbddcje=主买大单, zmbzdcje=主买中单, zmbxdcje=主买小单
+          zmstdcje=主卖特大单, zmsddcje=主卖大单, zmszdcje=主卖中单, zmsxdcje=主卖小单
+    """
+    try:
+        import requests
+        from config import Config
+        LICENCE = Config.MAIRUI_LICENCE or "5352ED2F-94E5-4E96-8B7F-B57BA75284E3"
+        url = f"https://api.mairuiapi.com/hsstock/history/transaction/{stock_code}/{LICENCE}?lt={days}"
+        resp = requests.get(url, timeout=10)
+        if resp.status_code != 200:
+            logger.warning(f"麦蕊资金流向API返回{resp.status_code}")
+            return []
+
+        data = resp.json()
+        if not data or not isinstance(data, list):
+            return []
+
+        records = []
+        for item in data:
+            # 主买总额 = 特大单+大单+中单+小单
+            buy_total = (
+                float(item.get('zmbtdcje', 0) or 0) +
+                float(item.get('zmbddcje', 0) or 0) +
+                float(item.get('zmbzdcje', 0) or 0) +
+                float(item.get('zmbxdcje', 0) or 0)
+            )
+            # 主卖总额
+            sell_total = (
+                float(item.get('zmstdcje', 0) or 0) +
+                float(item.get('zmsddcje', 0) or 0) +
+                float(item.get('zmszdcje', 0) or 0) +
+                float(item.get('zmsxdcje', 0) or 0)
+            )
+            # 主力净流入 = (特大单+大单)买 - (特大单+大单)卖
+            main_buy = float(item.get('zmbtdcje', 0) or 0) + float(item.get('zmbddcje', 0) or 0)
+            main_sell = float(item.get('zmstdcje', 0) or 0) + float(item.get('zmsddcje', 0) or 0)
+            main_net = main_buy - main_sell
+
+            # 超大单净流入
+            super_net = float(item.get('zmbtdcje', 0) or 0) - float(item.get('zmstdcje', 0) or 0)
+
+            # 总成交额
+            total_amount = buy_total + sell_total
+            main_net_pct = round(main_net / total_amount * 100, 2) if total_amount > 0 else 0
+            super_net_pct = round(super_net / total_amount * 100, 2) if total_amount > 0 else 0
+
+            # 大单净流入
+            big_net = float(item.get('zmbddcje', 0) or 0) - float(item.get('zmsddcje', 0) or 0)
+            big_net_pct = round(big_net / total_amount * 100, 2) if total_amount > 0 else 0
+
+            # 中单净流入
+            mid_net = float(item.get('zmbzdcje', 0) or 0) - float(item.get('zmszdcje', 0) or 0)
+            mid_net_pct = round(mid_net / total_amount * 100, 2) if total_amount > 0 else 0
+
+            # 小单净流入
+            small_net = float(item.get('zmbxdcje', 0) or 0) - float(item.get('zmsxdcje', 0) or 0)
+            small_net_pct = round(small_net / total_amount * 100, 2) if total_amount > 0 else 0
+
+            # 日期解析
+            t_str = str(item.get('t', ''))
+            date_str = t_str[:10] if t_str else ''
+
+            records.append({
+                'date': date_str,
+                'close_price': 0,
+                'change_pct': 0,
+                'main_net_inflow': round(main_net, 2),
+                'main_net_inflow_pct': main_net_pct,
+                'super_net_inflow': round(super_net, 2),
+                'super_net_inflow_pct': super_net_pct,
+                'big_net_inflow': round(big_net, 2),
+                'big_net_inflow_pct': big_net_pct,
+                'mid_net_inflow': round(mid_net, 2),
+                'mid_net_inflow_pct': mid_net_pct,
+                'small_net_inflow': round(small_net, 2),
+                'small_net_inflow_pct': small_net_pct,
+            })
+
+        # 按日期升序排列
+        records.sort(key=lambda x: x['date'])
+        logger.info(f"麦蕊API获取{stock_code}资金流向{len(records)}条")
+        return records
+    except Exception as e:
+        logger.warning(f"麦蕊资金流向API失败({stock_code}): {e}")
+        return []
+
+
 def analyze_fund_flow(stock_code, days=5):
    """
    分析主力资金流向，返回资金面评分和信号

+    数据源优先级：
+    1. DB stock_fund_flow_history 表（有最新数据时）
+    2. 麦蕊智数API hsstock/history/transaction（DB数据过期时补充）
+
    参数:
        stock_code: 股票代码
        days: 分析最近几天的资金流向
@@ -91,6 +186,28 @@ def analyze_fund_flow(stock_code, days=5):
        }
    """
    records = get_fund_flow_history(stock_code, days=days + 5)
+
+    # 检查DB数据是否足够新（最近3天内有数据）
+    use_mairui = False
+    if len(records) < 2:
+        use_mairui = True
+    else:
+        from datetime import date
+        latest_date = records[-1].get('date', '')
+        if latest_date:
+            try:
+                latest = datetime.strptime(latest_date, '%Y-%m-%d').date()
+                if (date.today() - latest).days > 5:
+                    use_mairui = True
+            except ValueError:
+                use_mairui = True
+
+    if use_mairui:
+        # 用麦蕊API获取资金流向数据
+        mairui_records = _get_fund_flow_from_mairui(stock_code, days + 5)
+        if mairui_records:
+            records = mairui_records
+
    if len(records) < 2:
        return {
            'score': 0,
@@ -50,8 +50,7 @@ def calc_market_sentiment():
                COUNT(*) FILTER (WHERE change_pct < 0) AS down_count,
                COUNT(*) FILTER (WHERE change_pct = 0) AS flat_count,
                COUNT(*) AS total,
-                COALESCE(SUM(amount), 0) AS total_amount,
-                COALESCE(PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY turnover), 0) AS turnover_median
+                COALESCE(SUM(amount), 0) AS total_amount
            FROM stock_realtime_price
            WHERE volume > 0 AND price > 0
        """)
@@ -66,7 +65,7 @@ def calc_market_sentiment():
        flat_count = int(row[4] or 0)
        total = int(row[5] or 1)
        total_amount = float(row[6] or 0) / 1e8  # 转为亿
-        turnover_median = float(row[7] or 0)
+        turnover_median = 0  # DB无turnover字段，不再使用

        # 涨跌停比
        up_down_ratio = round(limit_up / limit_down, 1) if limit_down > 0 else float(limit_up)
@@ -575,16 +575,33 @@ def job_afternoon_trade():
    print(f"[定时任务] ===== 午后交易任务结束 {datetime.now()} =====")


+def job_precompute_market_factors():
+    """开市前预热市场级外部因素缓存（09:15执行）
+    
+    预计算 P1市场情绪、P2北向、P3美股、P4商品、P7汇率、P6政策面，
+    结果存入 score_engine 内存缓存，后续全景扫描和评分直接复用。
+    """
+    print(f"[定时任务] ===== 开市前预热外部因素 {datetime.now()} =====")
+    try:
+        from services.score_engine import precompute_market_factors
+        precompute_market_factors()
+        print("[定时任务] 外部因素预热完成")
+    except Exception as e:
+        print(f"[定时任务] 外部因素预热失败: {e}")
+
+
 def run_scheduler():
    """运行定时任务调度器"""
    global _is_running
    
    # 设置定时任务 — v7最优时点: 09:35买入 / 13:40卖出
+    schedule.every().day.at("09:15").do(job_precompute_market_factors)  # 开市前预热外部因素
    schedule.every().day.at("09:35").do(job_morning_trade)
    schedule.every().day.at("13:40").do(job_afternoon_trade)
    schedule.every().day.at("15:05").do(trigger_closing_update)  # 收盘更新持仓价格
    
    print("[定时任务] 调度器已启动 (v7最优时点)")
+    print("[定时任务] - 09:15 开市前预热外部因素缓存")
    print("[定时任务] - 09:35 早盘交易（使用昨日扫描数据 — 最优买入时点）")
    print("[定时任务] - 13:40 午后交易（使用当日中午扫描数据 — 最优卖出时点）")
    print("[定时任务] - 15:05 收盘更新持仓价格")
@@ -7,7 +7,7 @@
 - 技术面评分（compute_deep_analysis 原始分）：基础分（0-100）
 - P0 资金面：±20
 - P1 市场情绪：±10
- P2 北向资金：±10
+- P2 南向资金：±10
 - P3 美股外盘：±10
 - P4 大宗商品：±5
 - P5 公告/异动：±15
@@ -17,9 +17,96 @@
 最终评分 = 技术面基础分 + 外部因素加减分（上限100，下限0）
 """
 import logging
+import time
+from datetime import datetime

 logger = logging.getLogger(__name__)

+# ═══════════════════════════════════════════════════════
+# 市场级因素缓存（开市前预计算，日内复用）
+# ═══════════════════════════════════════════════════════
+_market_cache = {
+    'data': None,       # (market_score, market_factors, market_reasons, summaries)
+    'timestamp': 0,     # 计算时间戳
+    'ttl': 4 * 3600,    # 缓存有效期 4 小时
+}
+
+# 个股资金面缓存（30分钟 TTL）
+_fund_flow_cache = {}
+_FUND_FLOW_TTL = 30 * 60
+
+
+def precompute_market_factors():
+    """预计算市场级外部因素并缓存（供定时任务在开市前调用）。
+
+    计算 P1 市场情绪、P2 南向、P3 美股、P4 商品、P7 汇率、P6 政策面，
+    结果存入内存缓存，后续 compute_comprehensive_score_batch 直接复用。
+    """
+    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': []}
+
+    _market_cache['data'] = (market_score, market_factors, market_reasons, summaries)
+    _market_cache['timestamp'] = time.time()
+    print(f'[评分引擎] 市场级因素预计算完成 (score={market_score}, {datetime.now():%H:%M:%S})')
+    return market_score, market_factors, market_reasons, summaries
+
+
+def _get_market_factors():
+    """获取市场级因素（优先读缓存，过期则重新计算）"""
+    now = time.time()
+    if _market_cache['data'] is not None and (now - _market_cache['timestamp']) < _market_cache['ttl']:
+        return _market_cache['data']
+    # 缓存不存在或过期，重新计算
+    return precompute_market_factors()
+

 def compute_comprehensive_score(stock_code, stock_name, technical_score, df=None):
    """
@@ -75,7 +162,7 @@ def compute_comprehensive_score(stock_code, stock_name, technical_score, df=None
        logger.warning(f"P1市场情绪分析失败: {e}")
        factors['market_sentiment'] = {'score': 0, 'summary': '分析失败', 'reasons': []}

-    # ---- P2/P3/P4/P7: 外部因素（北向/美股/商品/汇率）----
+    # ---- P2/P3/P4/P7: 外部因素（南向/美股/商品/汇率）----
    try:
        from services.external_factors import get_all_external_factors
        ext_result = get_all_external_factors()
@@ -139,7 +226,7 @@ def compute_comprehensive_score_batch(stocks_data):
    批量计算综合评分 — 市场级因素只计算一次，个股级因素逐只计算。

    优化点：
-    - P1 市场情绪、P2 北向、P3 美股、P4 商品、P7 汇率、P6 政策 → 市场级，只算一次
+    - P1 市场情绪、P2 南向、P3 美股、P4 商品、P7 汇率、P6 政策 → 市场级，只算一次
    - P0 资金面 → 个股级，逐只从DB读取
    - P5 公告/异动 → 批量模式跳过（需AKShare API + LLM，太慢），在深度分析时补充

@@ -152,57 +239,8 @@ def compute_comprehensive_score_batch(stocks_data):
    返回:
        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': []}
+    # ---- 市场级因素（从缓存读取，开市前由定时任务预计算）----
+    market_score, market_factors, market_reasons, summaries = _get_market_factors()

    # ---- 为每只股票计算个股级因素 ----
    results = {}
@@ -219,10 +257,16 @@ def compute_comprehensive_score_batch(stocks_data):
        }
        stock_reasons = list(market_reasons)

-        # P0: 资金面（个股级，从DB读取）
+        # P0: 资金面（个股级，从DB读取，带30分钟缓存）
        try:
-            from services.fund_flow_analyzer import analyze_fund_flow
-            fund_result = analyze_fund_flow(code, days=5)
+            now = time.time()
+            cached_ff = _fund_flow_cache.get(code)
+            if cached_ff and (now - cached_ff[1]) < _FUND_FLOW_TTL:
+                fund_result = cached_ff[0]
+            else:
+                from services.fund_flow_analyzer import analyze_fund_flow
+                fund_result = analyze_fund_flow(code, days=5)
+                _fund_flow_cache[code] = (fund_result, now)
            stock_factors['fund_flow'] = fund_result
            stock_external += fund_result.get('score', 0)
            stock_reasons.extend(fund_result.get('reasons', []))