fix: normalize Tencent realtime volume (#1409)

* fix: normalize Tencent realtime volume

* fix: harden Tencent realtime normalization

* docs: localize Tencent realtime comments

---------

Co-authored-by: mumu <42829555+ZhuLinsen@users.noreply.github.com>

data_provider/akshare_fetcher.py

@@ -137,6 +137,56 @@ def _is_hk_code(stock_code: str) -> bool:
     return code.isdigit() and len(code) == 5
 
 
+def _normalize_tencent_volume(fields: List[str]) -> Optional[int]:
+    """
+    将腾讯实时行情成交量归一为股。
+
+    腾讯返回内容对字段 6 的公开说明和实际返回不完全一致。优先使用
+    换手率、价格、流通市值交叉校验，在原值和旧的“手转股”结果中选择
+    更接近的一方。若无法交叉校验，则保留旧的“手转股”兜底逻辑，避免
+    传统腾讯返回内容回归为原成交量的 1/100。
+    """
+    if len(fields) <= 6 or not fields[6]:
+        return None
+
+    raw_volume = safe_int(fields[6])
+    if raw_volume is None:
+        return None
+
+    price = safe_float(fields[3]) if len(fields) > 3 else None
+    turnover_rate = safe_float(fields[38]) if len(fields) > 38 else None
+    circ_mv_yi = safe_float(fields[44]) if len(fields) > 44 and fields[44] else None
+    circ_mv = circ_mv_yi * 100000000 if circ_mv_yi is not None else None
+
+    if price and price > 0 and turnover_rate and turnover_rate > 0 and circ_mv and circ_mv > 0:
+        expected_volume = (circ_mv / price) * (turnover_rate / 100)
+        if expected_volume > 0:
+            raw_delta = abs(raw_volume - expected_volume)
+            hand_to_share_volume = raw_volume * 100
+            hand_delta = abs(hand_to_share_volume - expected_volume)
+            return raw_volume if raw_delta <= hand_delta else hand_to_share_volume
+
+    return raw_volume * 100
+
+
+def _parse_tencent_amount(fields: List[str]) -> Optional[float]:
+    """
+    解析腾讯实时行情成交额，单位为元。
+
+    观测到的返回内容中，字段 35 包含更精确的“价格/成交量/成交额”
+    三元组。字段 37 是旧的“万元”口径兜底字段。
+    """
+    if len(fields) > 35 and fields[35]:
+        parts = fields[35].split("/")
+        if len(parts) >= 3:
+            precise_amount = safe_float(parts[2])
+            if precise_amount is not None:
+                return precise_amount
+
+    amount_wan = safe_float(fields[37]) if len(fields) > 37 and fields[37] else None
+    return amount_wan * 10000 if amount_wan is not None else None
+
+
 def is_hk_stock_code(stock_code: str) -> bool:
     """
     Public API: determine if a stock code is a Hong Kong stock.
@@ -1175,11 +1225,12 @@ class AkshareFetcher(BaseFetcher):
             circuit_breaker.record_success(source_key)
             
             # 腾讯数据字段顺序（完整）：
-            # 1:名称 2:代码 3:最新价 4:昨收 5:今开 6:成交量(手) 7:外盘 8:内盘
+            # 1:名称 2:代码 3:最新价 4:昨收 5:今开 6:成交量 7:外盘 8:内盘
             # 9-28:买卖五档 30:时间戳 31:涨跌额 32:涨跌幅(%) 33:最高 34:最低 35:收盘/成交量/成交额
-            # 36:成交量(手) 37:成交额(万) 38:换手率(%) 39:市盈率 43:振幅(%)
+            # 36:成交量(口径随 payload 变化) 37:成交额(万) 38:换手率(%) 39:市盈率 43:振幅(%)
             # 44:流通市值(亿) 45:总市值(亿) 46:市净率 47:涨停价 48:跌停价 49:量比
             # 使用 realtime_types.py 中的统一转换函数
+            amount = _parse_tencent_amount(fields)
             quote = UnifiedRealtimeQuote(
                 code=stock_code,
                 name=fields[1] if len(fields) > 1 else "",
@@ -1187,7 +1238,8 @@ class AkshareFetcher(BaseFetcher):
                 price=safe_float(fields[3]),
                 change_pct=safe_float(fields[32]),
                 change_amount=safe_float(fields[31]) if len(fields) > 31 else None,
-                volume=safe_int(fields[6]) * 100 if fields[6] else None,  # 腾讯返回的是手，转为股
+                volume=_normalize_tencent_volume(fields),
+                amount=amount,
                 open_price=safe_float(fields[5]),
                 high=safe_float(fields[33]) if len(fields) > 33 else None,  # 修正：字段 33 是最高价
                 low=safe_float(fields[34]) if len(fields) > 34 else None,  # 修正：字段 34 是最低价

data_provider/realtime_types.py

@@ -125,7 +125,7 @@ class UnifiedRealtimeQuote:
     change_amount: Optional[float] = None   # 涨跌额
     
     # === 量价指标（部分源可能缺失）===
-    volume: Optional[int] = None            # 成交量（手）
+    volume: Optional[int] = None            # 成交量（股，与历史日线口径一致）
     amount: Optional[float] = None          # 成交额（元）
     volume_ratio: Optional[float] = None    # 量比
     turnover_rate: Optional[float] = None   # 换手率(%)

docs/CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 
 <!-- 新条目格式：- [类型] 描述（类型取值：新功能/改进/修复/文档/测试/chore）-->
 <!-- 每条独立一行追加到本段末尾，无需分类标题，合并时冲突最小 -->
+- [修复] 归一腾讯实时行情成交量为股口径，避免量能变化倍数被放大并误导分析报告。
 - [改进] Web 路由页面改为按需加载，降低首包体积并增加路由加载失败恢复提示。
 
 ## [3.18.0] - 2026-05-21

src/core/pipeline.py

@@ -627,8 +627,8 @@ class StockAnalysisPipeline:
                 'risk_factors': trend_result.risk_factors,
             }
 
-        # Issue #234: Override today with realtime OHLC + trend MA for intraday analysis
-        # Guard: trend_result.ma5 > 0 ensures MA calculation succeeded (data sufficient)
+        # Issue #234：盘中分析使用实时 OHLC 与趋势 MA 覆盖 today。
+        # 防护条件：trend_result.ma5 > 0 表示 MA 计算已成功且数据量充足。
         if realtime_quote and trend_result and trend_result.ma5 > 0:
             price = getattr(realtime_quote, 'price', None)
             if price is not None and price > 0:
@@ -636,6 +636,12 @@ class StockAnalysisPipeline:
                 if enhanced.get('yesterday') and isinstance(enhanced['yesterday'], dict):
                     yesterday_close = enhanced['yesterday'].get('close')
                 orig_today = enhanced.get('today') or {}
+                market_today = get_market_now(
+                    get_market_for_stock(normalize_stock_code(enhanced.get('code', '')))
+                ).date().isoformat()
+                source = getattr(realtime_quote, 'source', None)
+                source_name = getattr(source, 'value', source)
+                source_name = str(source_name) if source_name is not None else 'unknown'
                 open_p = getattr(realtime_quote, 'open_price', None) or getattr(
                     realtime_quote, 'pre_close', None
                 ) or yesterday_close or orig_today.get('open') or price
@@ -652,6 +658,9 @@ class StockAnalysisPipeline:
                     'ma5': trend_result.ma5,
                     'ma10': trend_result.ma10,
                     'ma20': trend_result.ma20,
+                    'date': market_today,
+                    'data_source': f"realtime:{source_name}",
+                    'realtime_source': source_name,
                 }
                 if vol is not None:
                     realtime_today['volume'] = vol
@@ -659,16 +668,20 @@ class StockAnalysisPipeline:
                     realtime_today['amount'] = amt
                 if pct is not None:
                     realtime_today['pct_chg'] = pct
+                realtime_owned_fields = {
+                    'open', 'high', 'low', 'close',
+                    'volume', 'amount', 'pct_chg', 'pctChg',
+                    'date', 'data_source', 'dataSource', 'source',
+                    'realtime_source', 'realtimeSource',
+                }
                 for k, v in orig_today.items():
-                    if k not in realtime_today and v is not None:
+                    if k not in realtime_today and k not in realtime_owned_fields and v is not None:
                         realtime_today[k] = v
                 enhanced['today'] = realtime_today
                 enhanced['ma_status'] = self._compute_ma_status(
                     price, trend_result.ma5, trend_result.ma10, trend_result.ma20
                 )
-                enhanced['date'] = get_market_now(
-                    get_market_for_stock(normalize_stock_code(enhanced.get('code', '')))
-                ).date().isoformat()
+                enhanced['date'] = market_today
                 if yesterday_close is not None:
                     try:
                         yc = float(yesterday_close)
@@ -1458,8 +1471,8 @@ class StockAnalysisPipeline:
         self, df: pd.DataFrame, realtime_quote: Any, code: str
     ) -> pd.DataFrame:
         """
-        Augment historical OHLCV with today's realtime quote for intraday MA calculation.
-        Issue #234: Use realtime price instead of yesterday's close for technical indicators.
+        使用当日实时行情补齐历史 OHLCV，用于盘中 MA 计算。
+        Issue #234：技术指标使用实时价格，而不是沿用昨日收盘价。
         """
         if df is None or df.empty or 'close' not in df.columns:
             return df
@@ -1469,7 +1482,7 @@ class StockAnalysisPipeline:
         if price is None or not (isinstance(price, (int, float)) and price > 0):
             return df
 
-        # Optional: skip augmentation on non-trading days (fail-open)
+        # 非交易日可跳过实时补齐；异常情况下保持失败开放。
         enable_realtime_tech = getattr(
             self.config, 'enable_realtime_technical_indicators', True
         )
@@ -1496,7 +1509,7 @@ class StockAnalysisPipeline:
         pct = getattr(realtime_quote, 'change_pct', None)
 
         if last_date >= market_today:
-            # Update last row with realtime close (copy to avoid mutating caller's df)
+            # 使用实时收盘价更新最后一行；先复制，避免修改调用方传入的 df。
             df = df.copy()
             idx = df.index[-1]
             df.loc[idx, 'close'] = price
@@ -1513,7 +1526,7 @@ class StockAnalysisPipeline:
             if pct is not None:
                 df.loc[idx, 'pct_chg'] = pct
         else:
-            # Append virtual today row
+            # 追加一行虚拟的当日实时 K 线。
             new_row = {
                 'code': code,
                 'date': market_today,

tests/test_akshare_realtime_logging.py

@@ -45,23 +45,35 @@ def _make_sina_payload() -> str:
     return f'var hq_str_sh601006="{",".join(fields)}";'
 
 
-def _make_tencent_payload() -> str:
+def _make_tencent_payload(
+    *,
+    price: str = "5.19",
+    volume: str = "1234",
+    amount_triplet: str = "",
+    amount_wan: str = "640.45",
+    turnover_rate: str = "0.69",
+    circ_mv_yi: str = "0.93",
+    total_mv_yi: str = "1.20",
+) -> str:
     fields = ["0"] * 50
     fields[1] = "大秦铁路"
     fields[2] = "601006"
-    fields[3] = "5.19"
+    fields[3] = price
     fields[4] = "5.00"
     fields[5] = "5.10"
-    fields[6] = "1234"
+    fields[6] = volume
     fields[31] = "0.19"
     fields[32] = "3.80"
-    fields[34] = "5.20"
-    fields[35] = "5.05"
-    fields[38] = "0.69"
+    fields[33] = "5.20"
+    fields[34] = "5.05"
+    if amount_triplet:
+        fields[35] = amount_triplet
+    fields[37] = amount_wan
+    fields[38] = turnover_rate
     fields[39] = "12.3"
     fields[43] = "2.00"
-    fields[44] = "1000"
-    fields[45] = "1200"
+    fields[44] = circ_mv_yi
+    fields[45] = total_mv_yi
     fields[46] = "1.20"
     fields[49] = "0.63"
     return f'v_sh601006="{"~".join(fields)}";'
@@ -148,11 +160,62 @@ def test_tencent_realtime_success_logs_endpoint(caplog, monkeypatch, akshare_fet
     assert quote is not None
     assert quote.name == "大秦铁路"
     assert quote.price == 5.19
+    assert quote.volume == 123400
+    assert quote.amount == 6404500
     assert breaker.successes == ["akshare_tencent"]
     assert f"endpoint={TENCENT_REALTIME_ENDPOINT}" in caplog.text
     assert "[实时行情-腾讯] 601006 大秦铁路:" in caplog.text
 
 
+def test_tencent_realtime_volume_keeps_share_unit_when_turnover_matches(monkeypatch, akshare_fetcher):
+    breaker = _DummyCircuitBreaker()
+    monkeypatch.setattr("data_provider.akshare_fetcher.get_realtime_circuit_breaker", lambda: breaker)
+    monkeypatch.setattr(
+        "data_provider.akshare_fetcher.requests.get",
+        lambda *args, **kwargs: _DummyResponse(
+            200,
+            _make_tencent_payload(
+                price="122.70",
+                volume="10931723",
+                amount_triplet="122.70/10931723/1327404280",
+                amount_wan="168369.8131",
+                turnover_rate="14.98",
+                circ_mv_yi="89.53",
+                total_mv_yi="147.24",
+            ),
+        ),
+    )
+
+    quote = akshare_fetcher._get_stock_realtime_quote_tencent("688691")
+
+    assert quote is not None
+    assert quote.volume == 10931723
+    assert quote.amount == 1327404280
+
+
+def test_tencent_realtime_volume_falls_back_to_legacy_hand_unit_when_not_cross_checkable(
+    monkeypatch, akshare_fetcher
+):
+    breaker = _DummyCircuitBreaker()
+    monkeypatch.setattr("data_provider.akshare_fetcher.get_realtime_circuit_breaker", lambda: breaker)
+    monkeypatch.setattr(
+        "data_provider.akshare_fetcher.requests.get",
+        lambda *args, **kwargs: _DummyResponse(
+            200,
+            _make_tencent_payload(
+                volume="1234",
+                turnover_rate="",
+                circ_mv_yi="",
+            ),
+        ),
+    )
+
+    quote = akshare_fetcher._get_stock_realtime_quote_tencent("601006")
+
+    assert quote is not None
+    assert quote.volume == 123400
+
+
 def test_hot_stocks_uses_eastmoney_hot_ranking_when_available(monkeypatch, akshare_fetcher):
     fake_akshare = SimpleNamespace()
     monkeypatch.setitem(sys.modules, "akshare", fake_akshare)

tests/test_pipeline_realtime_indicators.py

@@ -1,11 +1,11 @@
 # -*- coding: utf-8 -*-
 """
-Unit tests for Issue #234: intraday realtime technical indicators.
+Issue #234 盘中实时技术指标的单元测试。
 
-Covers:
-- _augment_historical_with_realtime: append/update logic, guards
-- _compute_ma_status: MA alignment string
-- _enhance_context: today override with realtime + trend_result
+覆盖范围：
+- _augment_historical_with_realtime：追加/更新逻辑和防护条件
+- _compute_ma_status：均线排列文案
+- _enhance_context：使用 realtime + trend_result 覆盖 today
 """
 
 import os
@@ -29,6 +29,7 @@ def _make_realtime_quote(
     high: float = 16.29,
     low: float = 15.55,
     volume: int = 13995600,
+    amount: float = None,
     change_pct: float = 0.96,
 ) -> UnifiedRealtimeQuote:
     return UnifiedRealtimeQuote(
@@ -40,12 +41,13 @@ def _make_realtime_quote(
         high=high,
         low=low,
         volume=volume,
+        amount=amount,
         change_pct=change_pct,
     )
 
 
 def _make_historical_df(days: int = 25, last_date: date = None) -> pd.DataFrame:
-    """Build historical OHLCV DataFrame."""
+    """构造历史 OHLCV DataFrame。"""
     if last_date is None:
         last_date = date.today() - timedelta(days=1)
     dates = [last_date - timedelta(days=i) for i in range(days - 1, -1, -1)]
@@ -72,7 +74,7 @@ def _make_historical_df(days: int = 25, last_date: date = None) -> pd.DataFrame:
 
 
 class TestAugmentHistoricalWithRealtime(unittest.TestCase):
-    """Tests for _augment_historical_with_realtime."""
+    """_augment_historical_with_realtime 的测试。"""
 
     def setUp(self) -> None:
         self._db_path = os.path.join(
@@ -121,8 +123,8 @@ class TestAugmentHistoricalWithRealtime(unittest.TestCase):
         self, _mock_market, _mock_open, mock_now
     ) -> None:
         today = date.today()
-        # Pin market clock to today (UTC) so the pipeline's market_today == date.today(),
-        # regardless of which timezone get_market_now would normally use (e.g. CST=UTC+8).
+        # 固定市场时钟为 UTC 当日，使 pipeline 的 market_today 等于 date.today()，
+        # 不受 get_market_now 通常使用的市场时区影响（例如 CST=UTC+8）。
         mock_now.return_value = datetime(
             today.year, today.month, today.day, 10, 0, tzinfo=timezone.utc
         )
@@ -141,8 +143,8 @@ class TestAugmentHistoricalWithRealtime(unittest.TestCase):
         self, _mock_market, _mock_open, mock_now
     ) -> None:
         today = date.today()
-        # Pin market clock to today so last_date >= market_today and the row is updated
-        # rather than appended (avoids off-by-one when CI runs after market closes in CST).
+        # 固定市场时钟为当日，使 last_date >= market_today，从而更新最后一行而不是追加。
+        # 这可以避免 CI 在 CST 收盘后运行时出现日期边界偏移。
         mock_now.return_value = datetime(
             today.year, today.month, today.day, 10, 0, tzinfo=timezone.utc
         )
@@ -155,7 +157,7 @@ class TestAugmentHistoricalWithRealtime(unittest.TestCase):
 
 
 class TestComputeMaStatus(unittest.TestCase):
-    """Tests for _compute_ma_status."""
+    """_compute_ma_status 的测试。"""
 
     def test_bullish_alignment(self) -> None:
         status = StockAnalysisPipeline._compute_ma_status(11, 10, 9.5, 9)
@@ -171,7 +173,7 @@ class TestComputeMaStatus(unittest.TestCase):
 
 
 class TestEnhanceContextRealtimeOverride(unittest.TestCase):
-    """Tests for _enhance_context today override with realtime + trend."""
+    """_enhance_context 使用实时行情和趋势结果覆盖 today 的测试。"""
 
     def setUp(self) -> None:
         self._db_path = os.path.join(
@@ -190,8 +192,8 @@ class TestEnhanceContextRealtimeOverride(unittest.TestCase):
         self, _mock_market, mock_now
     ) -> None:
         today = date.today()
-        # Pin market clock so _enhance_context sets enhanced['date'] == date.today().isoformat()
-        # regardless of which timezone get_market_now would normally use (e.g. CST=UTC+8).
+        # 固定市场时钟，使 _enhance_context 设置 enhanced['date'] == date.today().isoformat()，
+        # 不受 get_market_now 通常使用的市场时区影响（例如 CST=UTC+8）。
         mock_now.return_value = datetime(
             today.year, today.month, today.day, 10, 0, tzinfo=timezone.utc
         )
@@ -218,9 +220,105 @@ class TestEnhanceContextRealtimeOverride(unittest.TestCase):
         self.assertEqual(enhanced["today"]["ma20"], 14.9)
         self.assertIn("多头", enhanced["ma_status"])
         self.assertEqual(enhanced["date"], today.isoformat())
+        self.assertEqual(enhanced["today"]["date"], today.isoformat())
+        self.assertEqual(enhanced["today"]["data_source"], "realtime:tencent")
+        self.assertEqual(enhanced["today"]["realtime_source"], "tencent")
         self.assertIn("price_change_ratio", enhanced)
         self.assertIn("volume_change_ratio", enhanced)
 
+    @patch("src.core.pipeline.get_market_now")
+    @patch("src.core.pipeline.get_market_for_stock", return_value="cn")
+    def test_tencent_688691_volume_change_ratio_uses_normalized_share_volume(
+        self, _mock_market, mock_now
+    ) -> None:
+        today = date.today()
+        mock_now.return_value = datetime(
+            today.year, today.month, today.day, 10, 0, tzinfo=timezone.utc
+        )
+        context = {
+            "code": "688691",
+            "date": (today - timedelta(days=1)).isoformat(),
+            "today": {
+                "close": 128.46,
+                "volume": 19512753,
+                "amount": 2487341983,
+                "date": (today - timedelta(days=1)).isoformat(),
+                "dataSource": "AkshareFetcher",
+            },
+            "yesterday": {"close": 128.46, "volume": 19512753},
+        }
+        quote = UnifiedRealtimeQuote(
+            code="688691",
+            name="灿芯股份",
+            source=RealtimeSource.TENCENT,
+            price=122.70,
+            open_price=120.09,
+            high=125.96,
+            low=116.20,
+            volume=10931723,
+            amount=1327404280,
+            change_pct=3.40,
+        )
+        trend = TrendAnalysisResult(
+            code="688691",
+            trend_status=TrendStatus.BULL,
+            ma5=120.014,
+            ma10=119.425,
+            ma20=115.8305,
+        )
+
+        enhanced = self.pipeline._enhance_context(
+            context, quote, None, trend, "灿芯股份"
+        )
+
+        self.assertEqual(enhanced["today"]["volume"], 10931723)
+        self.assertEqual(enhanced["today"]["amount"], 1327404280)
+        self.assertEqual(enhanced["volume_change_ratio"], 0.56)
+        self.assertEqual(enhanced["today"]["date"], today.isoformat())
+        self.assertEqual(enhanced["today"]["data_source"], "realtime:tencent")
+        self.assertEqual(enhanced["today"]["realtime_source"], "tencent")
+        self.assertNotIn("dataSource", enhanced["today"])
+
+    @patch("src.core.pipeline.get_market_now")
+    @patch("src.core.pipeline.get_market_for_stock", return_value="cn")
+    def test_realtime_today_does_not_backfill_historical_amount_or_source(
+        self, _mock_market, mock_now
+    ) -> None:
+        today = date.today()
+        mock_now.return_value = datetime(
+            today.year, today.month, today.day, 10, 0, tzinfo=timezone.utc
+        )
+        context = {
+            "code": "600519",
+            "date": (today - timedelta(days=1)).isoformat(),
+            "today": {
+                "close": 15.0,
+                "amount": 999999,
+                "date": (today - timedelta(days=1)).isoformat(),
+                "dataSource": "AkshareFetcher",
+                "code": "600519",
+            },
+            "yesterday": {"close": 14.5, "volume": 1000000},
+        }
+        quote = _make_realtime_quote(price=15.72, amount=None)
+        trend = TrendAnalysisResult(
+            code="600519",
+            trend_status=TrendStatus.BULL,
+            ma5=15.5,
+            ma10=15.2,
+            ma20=14.9,
+        )
+
+        enhanced = self.pipeline._enhance_context(
+            context, quote, None, trend, "贵州茅台"
+        )
+
+        self.assertNotIn("amount", enhanced["today"])
+        self.assertNotIn("dataSource", enhanced["today"])
+        self.assertEqual(enhanced["today"]["date"], today.isoformat())
+        self.assertEqual(enhanced["today"]["data_source"], "realtime:tencent")
+        self.assertEqual(enhanced["today"]["code"], "600519")
+
     def test_enhance_context_injects_runtime_news_window_days(self) -> None:
         context = {"code": "600519", "today": {"close": 15.0}}
         enhanced = self.pipeline._enhance_context(
@@ -248,10 +346,10 @@ class TestEnhanceContextRealtimeOverride(unittest.TestCase):
         self.assertEqual(enhanced["today"]["close"], 15.0)
 
     def test_today_not_overridden_when_trend_ma_zero(self) -> None:
-        """When StockTrendAnalyzer returns early (data insufficient), ma5=0.0. Must not override."""
+        """StockTrendAnalyzer 因数据不足提前返回 ma5=0.0 时，不应覆盖 today。"""
         context = {"code": "600519", "today": {"close": 15.0, "ma5": 14.8}}
         quote = _make_realtime_quote(price=15.72)
-        trend = TrendAnalysisResult(code="600519")  # defaults: ma5=ma10=ma20=0.0
+        trend = TrendAnalysisResult(code="600519")  # 默认 ma5=ma10=ma20=0.0
         enhanced = self.pipeline._enhance_context(
             context, quote, None, trend, "贵州茅台"
         )