From 88411efa1c77393b5e8065567c571700070af9dd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=80=BB=E7=AE=A1?= Date: Wed, 9 Sep 2026 17:59:58 +0800 Subject: [PATCH] fix(HEL-356): complete Shenwan industry signal Co-authored-by: multica-agent --- backend/data/providers/tushare_industries.py | 21 +++++++ backend/features/heaven/engine.py | 13 +++- backend/features/heaven/manual.py | 2 +- config/architecture-inventory.json | 8 +-- tests/test_hel494_regressions.py | 36 +++++++++++ xiaobai-datahub/datahub/realtime_serve.py | 60 +++++++++++++++---- .../tests/test_realtime_intraday.py | 29 ++++++++- 7 files changed, 151 insertions(+), 18 deletions(-) diff --git a/backend/data/providers/tushare_industries.py b/backend/data/providers/tushare_industries.py index bbecaeb..80412ca 100644 --- a/backend/data/providers/tushare_industries.py +++ b/backend/data/providers/tushare_industries.py @@ -169,6 +169,16 @@ class ShenwanIndustryMixin: ) else: outer_error = rt_error or outer_error + if not outer_precise: + # The exact SW L2 member set and every member's final daily bar + # are already available. Use their equal-weight move as an + # explicitly labelled provisional outer value until Tushare's + # official sw_daily row is published. + daily = {"change": equal_change} + actual_trade_date = trade_date + outer_precise = True + outer_source = "sw_member_equal_daily" + outer_error = "" else: try: snapshot = self._sw_realtime_sector_snapshot( @@ -520,6 +530,12 @@ class ShenwanIndustryMixin: for item in members } equal_change = sum(item["change"] for item in valid) / len(valid) if valid else 0 + if not outer_precise and inner_precise: + official_change = equal_change + outer_precise = True + outer_source = "sw_member_equal_realtime" + outer_error = "" + quote_date = trade_date amount_billion = sum(_number(item.get("amount")) for item in valid) / 100000000 market_rows: list[dict[str, Any]] = [] try: @@ -642,6 +658,11 @@ class ShenwanIndustryMixin: return {}, "", f"申万行业 {sector_code} 盘后正式数据待入库" return {}, "", f"数据中枢申万实时暂不可用：{message[:180]}" if row: + actual_name = str(row.get("name") or "").strip() + if sector_name and actual_name != str(sector_name).strip(): + return {}, "", ( + f"数据中枢返回的行业名称不匹配：期望 {sector_name}，实际 {actual_name or '--'}" + ) return dict(row), str(row.get("source") or "datahub"), "" if finalized: return {}, "", f"申万行业 {sector_code} 当日盘后正式数据尚未入库" diff --git a/backend/features/heaven/engine.py b/backend/features/heaven/engine.py index 488bea0..8d1506c 100644 --- a/backend/features/heaven/engine.py +++ b/backend/features/heaven/engine.py @@ -802,7 +802,11 @@ def _market_line_scores( {}, ) sector_quantitative_mode = str(sector.get("_quantitative_mode") or "") - actual_sector_source = str(sector.get("source") or "").startswith("tushare_") + sector_source = str(sector.get("source") or "") + actual_sector_source = ( + sector_source.startswith("tushare_") + or "sw_member_equal_" in sector_source + ) if (sector.get("realtime") and actual_sector_source) or sector_quantitative_mode == "intraday": sector_change = float(sector.get("change") or 0) sector_change_score = _clamp(sector_change / 5, -1, 1) @@ -831,8 +835,13 @@ def _market_line_scores( f"成分上涨 {int(sector_up)} 家、下跌 {int(sector_down)} 家", f"平均换手 {float(sector.get('turnover_rate') or 0):.2f}%，相对市场 {float(sector.get('relative_turnover') or 0):.2f} 倍", ] + sector_change_label = ( + "申万二级行业成分等权涨跌" + if "sw_member_equal_" in sector_source + else "申万二级行业官方涨跌" + ) sector_outer_evidence = [ - f"申万二级行业官方涨跌 {sector_change:+.2f}%", + f"{sector_change_label} {sector_change:+.2f}%", f"领涨 {sector.get('leader') or '--'} {float(sector.get('leading_pct') or 0):+.2f}%", ] elif actual_sector_source or sector_quantitative_mode == "historical": diff --git a/backend/features/heaven/manual.py b/backend/features/heaven/manual.py index bbf27ce..b2f73c3 100644 --- a/backend/features/heaven/manual.py +++ b/backend/features/heaven/manual.py @@ -28,7 +28,7 @@ class HeavenManualMixin: "sector_coverage": {"line": 3, "label": "成分行情覆盖率", "unit": "%", "min": 0, "max": 100}, "sector_relative_turnover": {"line": 3, "label": "行业相对市场换手", "unit": "倍", "min": 0, "max": 20}, "sector_member_equal_change": {"line": 3, "label": "成分等权涨跌幅", "unit": "%", "min": -100, "max": 100}, - "sector_change": {"line": 4, "label": "申万官方涨跌幅", "unit": "%", "min": -100, "max": 100}, + "sector_change": {"line": 4, "label": "申万行业涨跌幅", "unit": "%", "min": -100, "max": 100}, "sector_leading_pct": {"line": [3, 4], "label": "行业领涨股涨跌幅", "unit": "%", "min": -100, "max": 100}, "market_sentiment_score": {"line": 5, "label": "市场情绪温度", "unit": "分", "min": 0, "max": 100}, "market_seal_rate": {"line": 5, "label": "封板率", "unit": "%", "min": 0, "max": 100}, diff --git a/config/architecture-inventory.json b/config/architecture-inventory.json index 129b7ca..88b7a64 100644 --- a/config/architecture-inventory.json +++ b/config/architecture-inventory.json @@ -473,8 +473,8 @@ }, { "path": "backend/features/heaven/engine.py", - "bytes": 51764, - "lines": 1183 + "bytes": 52046, + "lines": 1192 }, { "path": "frontend/index.html", @@ -483,8 +483,8 @@ }, { "path": "backend/data/providers/tushare_industries.py", - "bytes": 37786, - "lines": 873 + "bytes": 38936, + "lines": 894 }, { "path": "backend/features/screener/catalog.py", diff --git a/tests/test_hel494_regressions.py b/tests/test_hel494_regressions.py index 185ade1..69a4fca 100644 --- a/tests/test_hel494_regressions.py +++ b/tests/test_hel494_regressions.py @@ -384,3 +384,39 @@ class MemberQuoteCoverageTests(unittest.TestCase): self.assertEqual(snapshot["change"], 1.5) self.assertNotIn("权限", snapshot.get("outer_error") or "") self.assertNotIn("rt_sw_k", snapshot.get("outer_error") or "") + + def test_closed_uses_complete_member_daily_when_sector_quote_is_wrong(self) -> None: + client = TushareClient(token="demo") + client.resolve_trade_context = lambda _date: ("20260909", "20260908") + client.sw_stock_industry = MagicMock( + return_value={"l2_code": "801074.SI", "l2_name": "专用设备"} + ) + client._sw_sector_members = MagicMock(return_value=[ + {"ts_code": "603011.SH", "name": "合锻智能"}, + {"ts_code": "000821.SZ", "name": "京山轻机"}, + ]) + client._stock_listing_reference = MagicMock(return_value={}) + client._load_daily = MagicMock(return_value=[ + {"ts_code": "603011.SH", "name": "合锻智能", "pct_chg": 2, "amount": 1e8}, + {"ts_code": "000821.SZ", "name": "京山轻机", "pct_chg": -1, "amount": 1e8}, + ]) + client._confirmed_suspended_members = MagicMock(return_value=[]) + client.query = MagicMock(return_value=[]) + client.try_sector_quote = MagicMock(return_value={ + "code": "801074.SI", + "name": "托育服务", + "change": -2.19, + "quote_date": "20260909", + "quote_time": "2026-09-09T15:00:00+08:00", + "source": "eastmoney_sw", + }) + + snapshot = client.sw_sector_snapshot( + "603011.SH", "20260909", allow_realtime_close=True + ) + + self.assertTrue(snapshot["precise"]) + self.assertEqual(snapshot["name"], "专用设备") + self.assertEqual(snapshot["change"], 0.5) + self.assertEqual(snapshot["outer_source"], "sw_member_equal_daily") + self.assertEqual(snapshot["outer_error"], "") diff --git a/xiaobai-datahub/datahub/realtime_serve.py b/xiaobai-datahub/datahub/realtime_serve.py index 16b6235..bcd49b1 100644 --- a/xiaobai-datahub/datahub/realtime_serve.py +++ b/xiaobai-datahub/datahub/realtime_serve.py @@ -197,17 +197,22 @@ def fetch_sector_quote(db: HubDB, code: str, expected_date: str = "") -> dict[st if ts_code.isdigit(): ts_code = f"{ts_code}.SI" cache_key = f"sector:{ts_code}" + canonical_name = _canonical_sw_name(db, ts_code) cached = _read_cache(db, cache_key) - if cached is not None: + if cached is not None and _sector_payload_matches(cached, canonical_name): return cached errors: list[str] = [] try: row = EastmoneyAdapter().fetch_shenwan_quote(ts_code) + if not _sector_row_matches(row, canonical_name): + raise AdapterError( + f"industry name mismatch: expected {canonical_name}, got {row.get('name') or '--'}" + ) source = str(row.get("source") or "eastmoney_sw") except Exception as exc: errors.append(f"eastmoney:{exc}") recovered = _load_quotes_lkg(db, cache_key, expected_date) - if recovered is not None: + if recovered is not None and _sector_payload_matches(recovered, canonical_name): return recovered raise RealtimeApiError( "SOURCE_UNAVAILABLE", @@ -217,7 +222,7 @@ def fetch_sector_quote(db: HubDB, code: str, expected_date: str = "") -> dict[st quote_date = str(row.get("quote_date") or "") if want and quote_date and quote_date != want: recovered = _load_quotes_lkg(db, cache_key, want) - if recovered is not None: + if recovered is not None and _sector_payload_matches(recovered, canonical_name): return recovered raise RealtimeApiError("SOURCE_UNAVAILABLE", f"sector quote date {quote_date} != {want}") payload = _envelope( @@ -235,6 +240,29 @@ def fetch_sector_quote(db: HubDB, code: str, expected_date: str = "") -> dict[st return payload +def _canonical_sw_name(db: HubDB, ts_code: str) -> str: + row = db.fetchone( + "SELECT name FROM sector_master WHERE ts_code = ? AND family = 'sw'", + (ts_code,), + ) + if not row: + row = db.fetchone( + "SELECT name FROM eod_sector_daily WHERE ts_code = ? AND family = 'sw' " + "ORDER BY trade_date DESC LIMIT 1", + (ts_code,), + ) + return str((row or {}).get("name") or "").strip() + + +def _sector_row_matches(row: dict[str, Any], canonical_name: str) -> bool: + return not canonical_name or str(row.get("name") or "").strip() == canonical_name + + +def _sector_payload_matches(payload: dict[str, Any], canonical_name: str) -> bool: + data = payload.get("data") if isinstance(payload, dict) else None + return isinstance(data, dict) and _sector_row_matches(data, canonical_name) + + def fetch_limit_pool(db: HubDB, trade_date: str = "") -> dict[str, Any]: day = yyyymmdd(trade_date or now_shanghai()) cache_key = f"limit-pool:{day}" @@ -489,18 +517,29 @@ def warm_realtime(db: HubDB) -> dict[str, Any]: try: masters = db.fetchall( """ - SELECT ts_code FROM sector_master WHERE family = 'sw' - UNION - SELECT ts_code FROM eod_sector_daily WHERE family = 'sw' - ORDER BY ts_code + SELECT ts_code, name, 0 AS priority FROM sector_master WHERE family = 'sw' + UNION ALL + SELECT ts_code, name, 1 AS priority FROM eod_sector_daily WHERE family = 'sw' + ORDER BY priority, ts_code """ ) - codes = [str(row.get("ts_code") or "") for row in masters if row.get("ts_code")] - sector_rows = _eastmoney_sector_quotes(codes) - for row in sector_rows: + canonical_names: dict[str, str] = {} + for master in masters: + master_code = str(master.get("ts_code") or "") + master_name = str(master.get("name") or "").strip() + if master_code and master_name: + canonical_names.setdefault(master_code, master_name) + codes = list(canonical_names) + fetched_sector_rows = _eastmoney_sector_quotes(codes) + for row in fetched_sector_rows: if _row_quote_date(row, today) != today: continue code = str(row.get("ts_code") or "").upper() + if not _sector_row_matches(row, canonical_names.get(code, "")): + result["errors"].append( + f"sector:{code}:industry name mismatch" + ) + continue payload = _envelope( row, { @@ -513,6 +552,7 @@ def warm_realtime(db: HubDB) -> dict[str, Any]: }, ) _write_cache(db, f"sector:{code}", payload, INDEX_TTL, "eastmoney_sw") + sector_rows.append(row) result["sectors"] = len(sector_rows) result["rows"] += len(sector_rows) except Exception as exc: diff --git a/xiaobai-datahub/tests/test_realtime_intraday.py b/xiaobai-datahub/tests/test_realtime_intraday.py index 3364fb2..00420ea 100644 --- a/xiaobai-datahub/tests/test_realtime_intraday.py +++ b/xiaobai-datahub/tests/test_realtime_intraday.py @@ -9,7 +9,7 @@ from datahub.adapters.base import AdapterError from datahub.adapters.eastmoney import HIS_TRENDS_URL, TRENDS_URL, EastmoneyAdapter, _normalize_market_quote from datahub.adapters.tencent import _tencent_symbol from datahub.db import HubDB -from datahub.realtime_serve import fetch_intraday +from datahub.realtime_serve import fetch_intraday, fetch_sector_quote from datahub.serving import ApiError, V1API from datahub.timeutil import now_shanghai, yyyymmdd @@ -110,6 +110,33 @@ class IntradayLkgTests(unittest.TestCase): self.assertIn("intraday unavailable", str(ctx.exception)) +class SectorQuoteIdentityTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.db = HubDB(Path(self.tmp.name) / "hub.db") + self.db.execute( + "INSERT INTO sector_master(ts_code,name,family,updated_at) VALUES (?,?,?,?)", + ("801074.SI", "专用设备", "sw", "2026-09-09T10:00:00+08:00"), + ) + + def tearDown(self) -> None: + self.tmp.cleanup() + + def test_rejects_same_numeric_code_from_different_taxonomy(self) -> None: + wrong = { + "ts_code": "801074.SI", + "name": "托育服务", + "change": -2.19, + "quote_date": "20260909", + "source": "eastmoney_sw", + } + with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: + mocked.return_value.fetch_shenwan_quote.return_value = wrong + with self.assertRaises(Exception) as ctx: + fetch_sector_quote(self.db, "801074.SI", "20260909") + self.assertIn("industry name mismatch", str(ctx.exception)) + + class ServingIntradayDateTests(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory()