diff --git a/data_provider/akshare_fetcher.py b/data_provider/akshare_fetcher.py index 5b8fce716..97b53da3c 100644 --- a/data_provider/akshare_fetcher.py +++ b/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 是最低价 diff --git a/data_provider/realtime_types.py b/data_provider/realtime_types.py index 8d7d3b96c..217ba820d 100644 --- a/data_provider/realtime_types.py +++ b/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 # 换手率(%) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index f47933737..2f3623330 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/). +- [修复] 归一腾讯实时行情成交量为股口径，避免量能变化倍数被放大并误导分析报告。 - [改进] Web 路由页面改为按需加载，降低首包体积并增加路由加载失败恢复提示。 ## [3.18.0] - 2026-05-21 diff --git a/src/core/pipeline.py b/src/core/pipeline.py index 754ed5d42..98cdeaf3e 100644 --- a/src/core/pipeline.py +++ b/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, diff --git a/tests/test_akshare_realtime_logging.py b/tests/test_akshare_realtime_logging.py index 922ca3a7d..e60e6bc1b 100644 --- a/tests/test_akshare_realtime_logging.py +++ b/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) diff --git a/tests/test_pipeline_realtime_indicators.py b/tests/test_pipeline_realtime_indicators.py index d5404a48e..b466c5c3f 100644 --- a/tests/test_pipeline_realtime_indicators.py +++ b/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, "贵州茅台" )