diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index e2968a1d9..9ebf4dfa9 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/). - [新功能] 报告输出语言新增韩语（`REPORT_LANGUAGE=ko`），覆盖个股报告、大盘复盘、提示词输出语言、决策护栏、通知模板标签与 Web 报告详情页文案；`ko` 复用英文结构骨架并约束模型用韩文输出，`zh`/`en` 行为保持不变 (#1614) - [修复] 修复 Web 首页个股栏在 stock-bar 摘要字段缺失或动作建议无法归类时隐藏情绪分与建议标识的问题。 +- [测试] 台股三大法人 fetcher（TwInstitutionalFetcher）新增真实端点 live-smoke 脚本（tests/tw_institutional_live_smoke.py，非 pytest）与 @pytest.mark.network 漂移检测测试：核对 TWSE T86 / TPEx 核心字段名仍在、解析结果与原始字段一致；仅在非阻断的 network-smoke 定时任务运行，阻断门（pytest -m "not network"）不收集，离线 fixtures 无法察觉的上游字段改名/端点变动由此告警。 - [修复] 修复 Web 设置页定时任务“立即执行一次”后台线程未传 `stock_codes` 导致任务崩溃的问题。 ## [3.24.1] - 2026-06-28 diff --git a/tests/test_tw_institutional_network.py b/tests/test_tw_institutional_network.py new file mode 100644 index 000000000..7306f801c --- /dev/null +++ b/tests/test_tw_institutional_network.py @@ -0,0 +1,152 @@ +# -*- coding: utf-8 -*- +"""Live-network drift tests for TwInstitutionalFetcher, gated by @pytest.mark.network. + +These hit the REAL TWSE T86 + TPEx OpenAPI endpoints, so they run ONLY in the +non-blocking "Network Smoke" cron (`pytest -m network`). The blocking backend gate +runs `pytest -m "not network"` (scripts/ci_gate.sh), so these never gate a PR. + +The offline suite (tests/test_tw_institutional_fetcher.py) pins the parser to frozen +fixtures and therefore cannot detect upstream feed drift; this file is that detector. +Each test is self-contained (one raw fetch + the fetcher, in the same test) so a column +rename is caught LOUD and a narrow connectivity window cannot split-skip two corroborating +tests. Drift fails LOUD; a transport error, non-trading day, or transient blip skips QUIET +so the cron is not noisy. A 200 that is NOT JSON (maintenance page / URL migration) is +DRIFT, not a blip, so it fails — never skipped. Both feeds are public 政府開放資料, no creds. + +For the richer human-readable cross-check, see tests/tw_institutional_live_smoke.py. +""" + +import os +import sys +import unittest + +import pytest + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from data_provider.tw_institutional_fetcher import ( # noqa: E402 + TwInstitutionalFetcher, + _to_int, + _T86_CORE, + _T86_CODE, + _T86_FOREIGN, + _T86_TRUST, + _T86_DEALER, + _T86_TOTAL, + _T86_URL, + _TPEX_URL, + _TPEX_FOREIGN_EXCL, + _TPEX_TRUST, + _TPEX_DEALER, + _TPEX_TOTAL, + _UA, +) + +import requests # noqa: E402 + +_HEADERS = {"User-Agent": _UA, "Accept": "application/json"} +_NET_FIELDS = ("foreign_net", "trust_net", "dealer_net", "total_net") + + +def _fetch_with_retry(fetcher, code, tries=3): + """Transient upstream blips are real (observed live); retry before giving up.""" + rec = None + for _ in range(tries): + rec = fetcher.get_institutional_net(code) + if rec is not None: + return rec + return rec + + +@pytest.mark.network +class TestTwInstitutionalLiveNetwork(unittest.TestCase): + """Cron-only smoke: assert the live feeds still match the fetcher's contract.""" + + def _get_feed_or_skip(self, url, params=None): + """Transport error -> skip (can't judge drift from an unreachable feed); a 200 + that is not valid JSON (HTML maintenance page / URL migration) -> fail LOUD.""" + try: + resp = requests.get(url, params=params, headers=_HEADERS, timeout=20) + except requests.exceptions.RequestException as exc: + self.skipTest(f"endpoint unreachable: {exc}") + try: + return resp.json() + except ValueError as exc: # non-JSON body is feed drift, not a transient blip + self.fail(f"{url} returned non-JSON (maintenance page / URL migration?): {exc}") + + def _assert_record_shape(self, rec, market_label): + for field in _NET_FIELDS: + self.assertIsInstance(rec[field], int, f"{field} not an int: {rec[field]!r}") + self.assertEqual(rec["market"], market_label) + self.assertEqual(rec["unit"], "shares") + self.assertTrue(rec["date"].isdigit() and len(rec["date"]) == 8, f"bad date {rec['date']!r}") + + def test_t86_live_columns_and_fetcher_match_raw(self): + """T86 core columns still named as expected AND the fetcher's parsed net figures + equal the raw columns for a liquid stock (catches a fabricated fallback total).""" + payload = self._get_feed_or_skip( + _T86_URL, {"response": "json", "selectType": "ALLBUT0999"}) + if not isinstance(payload, dict): + self.fail(f"T86 response not a JSON object: {type(payload).__name__} (feed shape drift)") + if payload.get("stat") != "OK": + self.skipTest(f"T86 stat={payload.get('stat')} (likely non-trading day)") + fields = payload.get("fields") or [] + missing = [name for name in _T86_CORE if name not in fields] + self.assertEqual(missing, [], f"TWSE T86 core columns renamed/removed: {missing}") + + idx = {name: fields.index(name) for name in fields} + row = next((r for r in (payload.get("data") or []) + if isinstance(r, (list, tuple)) and str(r[idx[_T86_CODE]]).strip() == "2330"), None) + rec = _fetch_with_retry(TwInstitutionalFetcher(), "2330.TW") + if rec is None: + # row present in the raw feed but the fetcher returned None => parse/date drift + # (the exact fail-open this test exists to catch) -> FAIL, never a soft-skip. + if row is not None: + self.fail("2330 is present in the raw T86 feed but the fetcher returned None after " + "retries — parse/date drift (e.g. a column/date-format change)") + self.skipTest("2330.TW None and absent from the raw feed (transient / suspended)") + if row is None: + self.skipTest("2330 not in the raw T86 snapshot (cross-check unavailable)") + self._assert_record_shape(rec, "上市") + self.assertEqual(rec["foreign_net"], _to_int(row[idx[_T86_FOREIGN]])) + self.assertEqual(rec["trust_net"], _to_int(row[idx[_T86_TRUST]])) + self.assertEqual(rec["dealer_net"], _to_int(row[idx[_T86_DEALER]])) + # raw total present (it is in _T86_CORE, asserted above) -> the fetcher must echo it, + # never the foreign+trust+dealer fallback synthesised when the column is absent. + self.assertEqual(rec["total_net"], _to_int(row[idx[_T86_TOTAL]])) + + def test_tpex_live_columns_and_fetcher_match_raw(self): + """TPEx core keys still present AND the fetcher's parsed net figures equal the raw + columns for a liquid stock (catches a fabricated fallback total).""" + arr = self._get_feed_or_skip(_TPEX_URL) + if not isinstance(arr, list): + self.fail(f"TPEx response not a JSON array: {type(arr).__name__} (feed shape drift)") + if not arr: + self.skipTest("TPEx returned an empty list (likely non-trading day)") + if not isinstance(arr[0], dict): + self.fail(f"TPEx arr[0] not a dict: {type(arr[0]).__name__} (feed shape drift)") + core_keys = (_TPEX_FOREIGN_EXCL, _TPEX_TRUST, _TPEX_DEALER, _TPEX_TOTAL) + missing = [k for k in core_keys if k not in arr[0]] + self.assertEqual(missing, [], f"TPEx core keys renamed/removed: {missing}") + + raw = next((r for r in arr + if isinstance(r, dict) and str(r.get("SecuritiesCompanyCode", "")).strip() == "5483"), None) + rec = _fetch_with_retry(TwInstitutionalFetcher(), "5483.TWO") + if rec is None: + # row present in the raw feed but the fetcher returned None => parse/date drift + # (e.g. a 民國 date-format change _parse_tpex_row can't convert) -> FAIL, not soft-skip. + if raw is not None: + self.fail("5483 is present in the raw TPEx feed but the fetcher returned None after " + "retries — parse/date drift (e.g. a 民國 date-format change)") + self.skipTest("5483.TWO None and absent from the raw feed (transient / suspended)") + if raw is None: + self.skipTest("5483 not in the raw TPEx snapshot (cross-check unavailable)") + self._assert_record_shape(rec, "上櫃") + self.assertEqual(rec["foreign_net"], _to_int(raw.get(_TPEX_FOREIGN_EXCL))) + self.assertEqual(rec["trust_net"], _to_int(raw.get(_TPEX_TRUST))) + self.assertEqual(rec["dealer_net"], _to_int(raw.get(_TPEX_DEALER))) + self.assertEqual(rec["total_net"], _to_int(raw.get(_TPEX_TOTAL))) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/tw_institutional_live_smoke.py b/tests/tw_institutional_live_smoke.py new file mode 100644 index 000000000..f834fe415 --- /dev/null +++ b/tests/tw_institutional_live_smoke.py @@ -0,0 +1,276 @@ +# -*- coding: utf-8 -*- +"""Manual live smoke for TwInstitutionalFetcher (NOT run by pytest — no test_ prefix). + +The offline unit tests (tests/test_tw_institutional_fetcher.py) pin the parser to +FROZEN fixtures, so by construction they can never notice an upstream feed change. +This script hits the REAL TWSE T86 + TPEx OpenAPI endpoints to surface that drift: +an endpoint move, a core-column rename (which makes the fetcher fail-open SILENTLY, +returning None for every stock), a 民國->ISO date-format switch, or a schema change. +Both feeds are public 政府開放資料 — no credentials, no key. + +Skip vs drift (a drift detector must not report a feed change as "PASS"): + - A transport error (endpoint unreachable / SSL / timeout) -> SOFT SKIP: you cannot + detect drift from an unreachable feed. + - A non-trading-day response (T86 stat != "OK" / empty TPEx list) -> SOFT SKIP. + - A 200 that is NOT valid JSON of the expected shape (an HTML maintenance page or a + URL migration) -> DRIFT, reported LOUD ([x], exit 1). This is exactly the class of + endpoint change the script exists to catch, so it must never be swallowed as a blip. + - A core / foreign-dealer column rename -> DRIFT, reported LOUD. + +What each level asserts (for stocks present in the feed that day): + the fetcher's foreign/trust/dealer/total equal the raw columns it claims to read, and + the ALWAYS-TRUE reconstruction `total == foreign + foreign_dealer + trust + dealer` + holds (the 3-term identity `total == foreign + trust + dealer` only holds when the + foreign-dealer sub-component is 0, so this reads the raw foreign-dealer column instead). + +Usage: + python tests/tw_institutional_live_smoke.py # default 2330.TW 0050.TW 5483.TWO 6488.TWO + python tests/tw_institutional_live_smoke.py 2330.TW 5483.TWO # custom codes + +Exit code 0 = all checks pass / soft-skipped; 1 = a drift / parse mismatch was detected. +""" + +import argparse +import os +import sys + +_PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +sys.path.insert(0, _PROJECT_ROOT) + +try: # Windows cp950 console mangles the Chinese feed messages / 上市·上櫃 labels + sys.stdout.reconfigure(encoding="utf-8") +except Exception: # noqa: BLE001 - cosmetic only; never block the smoke on console encoding + pass + +import requests # noqa: E402 + +from data_provider.tw_institutional_fetcher import ( # noqa: E402 + TwInstitutionalFetcher, + _to_int, + _T86_URL, + _T86_CORE, + _T86_CODE, + _T86_FOREIGN, + _T86_TRUST, + _T86_DEALER, + _T86_TOTAL, + _TPEX_URL, + _TPEX_FOREIGN_EXCL, + _TPEX_TRUST, + _TPEX_DEALER, + _TPEX_TOTAL, + _UA, +) + +# Raw foreign-dealer columns — NOT read by the fetcher (foreign deliberately excludes +# them), but needed here to verify the always-true total reconstruction. A rename of +# these is drift relative to the reconstruction check, so their presence is asserted. +_T86_FOREIGN_DEALER = "外資自營商買賣超股數" +_TPEX_FOREIGN_DEALER = "ForeignDealers-Difference" + +_HEADERS = {"User-Agent": _UA, "Accept": "application/json"} + + +def _print_header(title: str) -> None: + print(f"\n{'=' * 64}") + print(f" {title}") + print(f"{'=' * 64}") + + +def _check(label: str, ok: bool, detail: str = "") -> bool: + mark = "[+]" if ok else "[x]" + suffix = f" {detail}" if detail else "" + print(f" {mark} {label}{suffix}") + return ok + + +def _get_feed(url: str, params=None): + """Fetch + JSON-parse a feed, classifying the outcome for a drift detector. + + Returns (payload, status, message) where status is one of: + 'ok' -> payload is the parsed JSON + 'skip' -> transport error; cannot judge drift from an unreachable feed + 'drift' -> a 200 that is not valid JSON (HTML maintenance page / URL migration) + """ + try: + resp = requests.get(url, params=params, headers=_HEADERS, timeout=20) + except requests.exceptions.RequestException as exc: + return None, "skip", f"endpoint unreachable: {exc}" + try: + return resp.json(), "ok", "" + except ValueError as exc: # json.JSONDecodeError subclasses ValueError; non-JSON body = drift + return None, "drift", f"non-JSON response (maintenance page / URL migration?): {exc}" + + +def _fetch_with_retry(fetcher: TwInstitutionalFetcher, code: str, tries: int = 3): + """Transient upstream blips are real (observed live: a single TPEx call returning + empty, the next succeeding). Retry before treating a None as a hard miss.""" + rec = None + for _ in range(tries): + rec = fetcher.get_institutional_net(code) + if rec is not None: + return rec + return rec + + +def _cross_check_record(rec: dict, raw_foreign, raw_trust, raw_dealer, raw_total, + raw_foreign_dealer) -> bool: + """Assert the fetcher parsed the columns it claims to, and the total reconstructs. + + raw_foreign_dealer may be None when the foreign-dealer column is renamed/absent (a + drift already flagged up-front) or genuinely missing for this stock — in that case the + always-true reconstruction cannot be computed, so it is reported as skipped rather than + silently passed with a fabricated 0; the other four column cross-checks still run. + """ + ok = True + ok &= _check("foreign_net == raw foreign(ex-dealer) col", rec["foreign_net"] == raw_foreign, + f"fetcher={rec['foreign_net']:,} raw={raw_foreign}") + ok &= _check("trust_net == raw trust col", rec["trust_net"] == raw_trust, + f"fetcher={rec['trust_net']:,} raw={raw_trust}") + ok &= _check("dealer_net == raw dealer col", rec["dealer_net"] == raw_dealer, + f"fetcher={rec['dealer_net']:,} raw={raw_dealer}") + ok &= _check("total_net == raw total col", rec["total_net"] == raw_total, + f"fetcher={rec['total_net']:,} raw={raw_total}") + if raw_foreign_dealer is None: + print(" [~] total reconstruction skipped — foreign-dealer value unavailable for this stock") + return ok + recon = rec["foreign_net"] + raw_foreign_dealer + rec["trust_net"] + rec["dealer_net"] + ok &= _check("total == foreign + foreign_dealer + trust + dealer (always-true)", + recon == rec["total_net"], + f"recon={recon:,} total={rec['total_net']:,} (foreign_dealer={raw_foreign_dealer:,})") + return ok + + +def level_twse(fetcher: TwInstitutionalFetcher, codes) -> bool: + _print_header(f"Level TWSE / T86 (上市): {codes or '(none)'}") + payload, status, msg = _get_feed(_T86_URL, {"response": "json", "selectType": "ALLBUT0999"}) + if status == "skip": + print(f" [!] T86 {msg} — soft skip") + return True + if status == "drift": + return _check("T86 returned JSON", False, msg) + if not isinstance(payload, dict): + return _check("T86 response is a JSON object", False, f"got {type(payload).__name__} — feed shape drift") + if payload.get("stat") != "OK": + print(f" [!] T86 stat={payload.get('stat')} (likely non-trading day) — soft skip") + return True + + fields = payload.get("fields") or [] + missing = [n for n in _T86_CORE if n not in fields] + ok = _check("T86 core column names present", not missing, + f"missing={missing}" if missing else f"all {len(_T86_CORE)} present") + if missing: + print(" -> a renamed/removed core column makes the fetcher fail-open SILENTLY") + return ok # core rename => fetcher fail-opens (rec=None); stop before indexing absent columns + fd_present = _T86_FOREIGN_DEALER in fields + ok &= _check("T86 foreign-dealer column present (for reconstruction)", fd_present, + "" if fd_present else f"'{_T86_FOREIGN_DEALER}' renamed/removed — reconstruction unavailable") + + idx = {n: fields.index(n) for n in fields} + fd_idx = idx.get(_T86_FOREIGN_DEALER) + rows = {str(r[idx[_T86_CODE]]).strip(): r for r in (payload.get("data") or []) + if isinstance(r, (list, tuple)) and _T86_CODE in idx} + + for code in codes: + print(f"\n -- {code} --") + base = code.upper().rsplit(".", 1)[0] + raw = rows.get(base) + rec = _fetch_with_retry(fetcher, code) + if rec is None: + # raw row present but the fetcher returned None => a parse/date drift the fetcher + # fail-opened on (e.g. a 民國->ISO date switch) — fail LOUD, never a soft-skip + # (that conflation is exactly what would let the drift this script exists to catch slip). + if raw is not None: + ok &= _check(f"{base}: fetcher must parse a row that exists in the raw feed", False, + "raw row present but get_institutional_net() returned None after retries — parse/date drift") + else: + print(f" [!] {base} not in T86 feed today — soft skip") + continue + if raw is None: + print(f" [!] {base} parsed by the fetcher but absent from this raw snapshot — soft-skip cross-check") + continue + fd_val = _to_int(raw[fd_idx]) if (fd_idx is not None and fd_idx < len(raw)) else None + ok &= _cross_check_record( + rec, + _to_int(raw[idx[_T86_FOREIGN]]), _to_int(raw[idx[_T86_TRUST]]), + _to_int(raw[idx[_T86_DEALER]]), _to_int(raw[idx[_T86_TOTAL]]), fd_val, + ) + return ok + + +def level_tpex(fetcher: TwInstitutionalFetcher, codes) -> bool: + _print_header(f"Level TPEx (上櫃): {codes or '(none)'}") + arr, status, msg = _get_feed(_TPEX_URL) + if status == "skip": + print(f" [!] TPEx {msg} — soft skip") + return True + if status == "drift": + return _check("TPEx returned JSON", False, msg) + if not isinstance(arr, list): + return _check("TPEx response is a JSON array", False, f"got {type(arr).__name__} — feed shape drift") + if not arr: + print(" [!] TPEx returned an empty list (likely non-trading day) — soft skip") + return True + + sample = arr[0] + core_keys = (_TPEX_FOREIGN_EXCL, _TPEX_TRUST, _TPEX_DEALER, _TPEX_TOTAL) + missing = [k for k in core_keys if k not in sample] + ok = _check("TPEx core column keys present", not missing, + f"missing={missing}" if missing else f"all {len(core_keys)} present") + if missing: + print(" -> a renamed/removed core key makes the fetcher fail-open SILENTLY") + fd_present = _TPEX_FOREIGN_DEALER in sample + ok &= _check("TPEx foreign-dealer key present (for reconstruction)", fd_present, + "" if fd_present else f"'{_TPEX_FOREIGN_DEALER}' renamed/removed — reconstruction unavailable") + + by_code = {str(r.get("SecuritiesCompanyCode", "")).strip(): r for r in arr if isinstance(r, dict)} + for code in codes: + print(f"\n -- {code} --") + base = code.upper().rsplit(".", 1)[0] + raw = by_code.get(base) + rec = _fetch_with_retry(fetcher, code) + if rec is None: + # raw row present but fetcher None => parse/date drift (e.g. a 民國 date-format change + # _parse_tpex_row can't convert) — fail LOUD, not a soft-skip. + if raw is not None: + ok &= _check(f"{base}: fetcher must parse a row that exists in the raw feed", False, + "raw row present but get_institutional_net() returned None after retries — parse/date drift") + else: + print(f" [!] {base} not in TPEx feed today — soft skip") + continue + if raw is None: + print(f" [!] {base} parsed by the fetcher but absent from this raw snapshot — soft-skip cross-check") + continue + fd_val = _to_int(raw.get(_TPEX_FOREIGN_DEALER)) if fd_present else None + ok &= _cross_check_record( + rec, + _to_int(raw.get(_TPEX_FOREIGN_EXCL)), _to_int(raw.get(_TPEX_TRUST)), + _to_int(raw.get(_TPEX_DEALER)), _to_int(raw.get(_TPEX_TOTAL)), fd_val, + ) + return ok + + +def main() -> int: + parser = argparse.ArgumentParser(description="TwInstitutionalFetcher live smoke (real TWSE/TPEx)") + parser.add_argument("codes", nargs="*", + default=["2330.TW", "0050.TW", "5483.TWO", "6488.TWO"], + help="stock codes with .TW / .TWO suffix") + args = parser.parse_args() + codes = args.codes or ["2330.TW", "0050.TW", "5483.TWO", "6488.TWO"] + tw = [c for c in codes if c.upper().endswith(".TW")] + two = [c for c in codes if c.upper().endswith(".TWO")] + + print("TwInstitutionalFetcher live smoke (public TWSE T86 + TPEx OpenAPI, no creds)") + fetcher = TwInstitutionalFetcher() + results = {"TWSE": level_twse(fetcher, tw), "TPEx": level_tpex(fetcher, two)} + + _print_header("Summary") + for name, passed in results.items(): + print(f" {'[+]' if passed else '[x]'} {name}: {'PASS' if passed else 'FAIL'}") + all_ok = all(results.values()) + print(f"\n {'All checks passed.' if all_ok else 'DRIFT/MISMATCH detected — see [x] above.'}") + return 0 if all_ok else 1 + + +if __name__ == "__main__": + sys.exit(main())