From 7715896fe4cc14f856a3e769dfdbf5c51cd1957d Mon Sep 17 00:00:00 2001 From: Wenyu Chiou <162016108+WenyuChiou@users.noreply.github.com> Date: Sun, 28 Jun 2026 18:49:09 +0800 Subject: [PATCH] =?UTF-8?q?feat(market):=20tw=20institutional-flows=20(?= =?UTF-8?q?=E4=B8=89=E5=A4=A7=E6=B3=95=E4=BA=BA)=20data-layer=20fetcher=20?= =?UTF-8?q?(#1829)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the #1777 maintainer-greenlit Phase-2 data layer: a self-contained, tw-only fetcher for Taiwan per-stock institutional (外資/投信/自營商) net buy/sell. Strictly additive -- no change to the existing cn/hk/us/jp/kr flows in data_provider/base.py, and not yet wired into the report/Web/scoring path (a deliberate follow-up per the maintainer's scope). - data_provider/tw_institutional_fetcher.py (NEW): TwInstitutionalFetcher - 上市 .TW -> TWSE T86 legacy rwd JSON endpoint (西元 date, comma values) - 上櫃 .TWO -> TPEx OpenAPI tpex_3insti_daily_trading (民國 date, plain ints) - T86 columns are read by NAME (validated against the payload `fields` header), so a TWSE column rename / reorder fails open instead of silently shipping misaligned numbers under stale indices - foreign_net = 外陸資 (ex 外資自營商, T86) / dealer-excluded foreign (TPEx) so the breakdown matches the official 三大法人 total; total_net is the official figure; unit = shares, signs preserved - whole-market single-day cache keyed by (market, date), filtered per stock; ~3 req/5s throttle (own lock) for the T86 endpoint - fail-open: any network/rate-limit/empty/unknown-stock returns None; a missing or renamed column drops the row (never a fabricated 0); a row whose trading date cannot be attributed (TPEx 民國 unconvertible) is dropped; empty/failed fetches are not cached (no TTL-long blackout) - tests/test_tw_institutional_fetcher.py (NEW): 20 offline tests with fixtures trimmed from real T86 (2330) / TPEx (3105) responses; pins the net breakdown + sign, 民國->西元 conversion, routing, caching, and fail-open -- including column reorder (parsed by name), column rename / missing header (fail-open), the missing-column-vs-genuine-zero distinction, and unconvertible TPEx dates - docs/market-support.md + docs/CHANGELOG.md: data-source capability boundary + OGDL v1 license note; no new config (.env.example untouched) Addresses the #1829 review (read T86 by field name; drop TPEx rows with an unconvertible date; foreign_net excludes foreign-dealers). Sources are 政府開放資料 under 政府資料開放授權條款第 1 版 (OGDL v1, commercial-safe). Refs #1777 --- data_provider/tw_institutional_fetcher.py | 337 ++++++++++++++++++++++ docs/CHANGELOG.md | 1 + docs/market-support.md | 1 + tests/test_tw_institutional_fetcher.py | 262 +++++++++++++++++ 4 files changed, 601 insertions(+) create mode 100644 data_provider/tw_institutional_fetcher.py create mode 100644 tests/test_tw_institutional_fetcher.py diff --git a/data_provider/tw_institutional_fetcher.py b/data_provider/tw_institutional_fetcher.py new file mode 100644 index 000000000..48bc14074 --- /dev/null +++ b/data_provider/tw_institutional_fetcher.py @@ -0,0 +1,337 @@ +# -*- coding: utf-8 -*- +"""TwInstitutionalFetcher — Taiwan 三大法人 (institutional-investor) daily net buy/sell. + +Data-layer only, ``tw``-only, strictly additive. This module is a self-contained +data-access building block: it fetches, parses, caches and fail-opens. It is NOT +wired into the analysis report / Web / scoring path — that is a deliberate +follow-up (per #1777). It does not touch the existing A-share / HK / US / JP / KR +flows in ``data_provider/base.py``. + +Sources (政府開放資料, 政府資料開放授權條款第 1 版 / OGDL v1, commercial-safe, no key): + - 上市 TWSE T86 「三大法人買賣超日報」 (per-stock), legacy RWD JSON endpoint + https://www.twse.com.tw/rwd/zh/fund/T86?response=json&date=YYYYMMDD&selectType=ALLBUT0999 + (date is 西元 ``YYYYMMDD``; numeric values are comma-formatted strings) + - 上櫃 TPEx ``tpex_3insti_daily_trading``, OpenAPI + https://www.tpex.org.tw/openapi/v1/tpex_3insti_daily_trading + (date is 民國 ``1150626``; numeric values are plain integer strings) + +Fail-open contract: any network error, rate-limit, empty response, unexpected +shape or missing field returns ``None`` (no data) — it never raises into the +caller, so the analysis main flow is never interrupted. + +Units are **shares (股)**, not lots (張). Buy/sell-net signs are preserved +(negative = net sell). +""" + +from __future__ import annotations + +import logging +import threading +import time +from typing import Any, Dict, Optional + +import requests + +logger = logging.getLogger(__name__) + +_T86_URL = "https://www.twse.com.tw/rwd/zh/fund/T86" +_TPEX_URL = "https://www.tpex.org.tw/openapi/v1/tpex_3insti_daily_trading" +_UA = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/120.0 Safari/537.36" +) + +# TWSE T86 core column NAMES. Read by name (not a fixed index) so a TWSE column +# rename / reorder fails open instead of silently shipping misaligned numbers. +# foreign = 外陸資 (NOT incl 外資自營商): foreign-dealer sits outside the 外資 +# category in the official 三大法人 total. +_T86_CODE = "證券代號" +_T86_FOREIGN = "外陸資買賣超股數(不含外資自營商)" +_T86_TRUST = "投信買賣超股數" +_T86_DEALER = "自營商買賣超股數" +_T86_TOTAL = "三大法人買賣超股數" +_T86_CORE = (_T86_CODE, _T86_FOREIGN, _T86_TRUST, _T86_DEALER, _T86_TOTAL) + +# TPEx OpenAPI column keys (verified live 2026-06; note the inconsistent spacing in +# the official feed). foreign = dealer-excluded, matching TotalDifference = +# foreign-excl + trust + dealer. +_TPEX_FOREIGN_EXCL = ( + "Foreign Investors include Mainland Area Investors " + "(Foreign Dealers excluded)-Difference" +) +_TPEX_TRUST = "SecuritiesInvestmentTrustCompanies-Difference" +_TPEX_DEALER = "Dealers-Difference" +_TPEX_TOTAL = "TotalDifference" + + +def _to_int(value: Any) -> Optional[int]: + """Parse a TWSE/TPEx numeric cell to int, preserving sign. + + Handles comma grouping (T86) and plain ints (TPEx). Empty / ``--`` / ``-`` / + non-numeric -> ``None`` (treated as missing, never a fabricated 0). + """ + try: + text = str(value).replace(",", "").replace(" ", "").strip() + except (TypeError, ValueError): + return None + if text in ("", "-", "--", "—"): + return None + try: + return int(text) + except ValueError: + try: + return int(float(text)) + except ValueError: + return None + + +def minguo_to_ad(date_str: Any) -> Optional[str]: + """Convert a TPEx 民國 date ``YYYMMDD`` (e.g. ``1150626``) to 西元 ``YYYYMMDD``. + + ``1150626`` -> ``20260626`` (民國 115 + 1911 = 西元 2026). Returns ``None`` for + anything that is not a 7-digit 民國 date, so a format change fails open. + """ + text = str(date_str).strip() + if not (text.isdigit() and len(text) == 7): + return None + return f"{int(text[:3]) + 1911}{text[3:]}" + + +class TwInstitutionalFetcher: + """Fetch Taiwan per-stock 三大法人 net buy/sell, ``.TW`` (上市) / ``.TWO`` (上櫃) only.""" + + name = "TwInstitutionalFetcher" + + def __init__( + self, + *, + cache_ttl_seconds: int = 900, + min_request_interval: float = 1.8, + timeout: int = 15, + ) -> None: + # Whole-market single-day cache keyed by (market, ad_date); filtered per stock. + self._cache: Dict[Any, Dict[str, dict]] = {} + self._cache_at: Dict[Any, float] = {} + self._cache_ttl = cache_ttl_seconds + self._timeout = timeout + # TWSE T86 RWD endpoint has an informal ~3 req / 5 s ban; throttle requests. + self._min_interval = min_request_interval + self._last_request_at = 0.0 + self._lock = threading.Lock() + self._throttle_lock = threading.Lock() + + # ------------------------------------------------------------------ public + def get_institutional_net( + self, stock_code: str, date: Optional[str] = None + ) -> Optional[dict]: + """Return the normalized 三大法人 record for one TW stock, or ``None``. + + ``stock_code`` must carry an explicit ``.TW`` / ``.TWO`` suffix; a bare or + non-TW code returns ``None`` (not applicable). ``date`` (西元 ``YYYYMMDD``) + only applies to 上市/T86; 上櫃/TPEx OpenAPI serves the latest trading day. + Fail-open: any error returns ``None``. + """ + market = self._market_of(stock_code) + if market is None: + return None + base = self._base_code(stock_code) + try: + table = self._whole_market(market, date) + except Exception as exc: # noqa: BLE001 - fail-open by contract + logger.info( + "[tw-inst] fetch failed market=%s code=%s: %s", market, stock_code, exc + ) + return None + if not table: + return None + return table.get(base) + + # ------------------------------------------------------------------ routing + @staticmethod + def _market_of(stock_code: Any) -> Optional[str]: + upper = str(stock_code or "").strip().upper() + if upper.endswith(".TWO"): + return "tpex" + if upper.endswith(".TW"): + return "twse" + return None + + @staticmethod + def _base_code(stock_code: Any) -> str: + return str(stock_code or "").strip().upper().rsplit(".", 1)[0] + + @staticmethod + def _norm_ad_date(date: Any) -> Optional[str]: + if not date: + return None + text = str(date).strip().replace("-", "").replace("/", "") + return text if (text.isdigit() and len(text) == 8) else None + + # -------------------------------------------------- whole-market cached fetch + def _whole_market(self, market: str, date: Optional[str]) -> Dict[str, dict]: + """Whole-market single-day table {code: record}, cached per (market, date). + + May raise on network / HTTP errors -- the public get_institutional_net wraps + this in a fail-open try/except. Only non-empty results are cached, so a + transient rate-limit / empty response is retried on the next call rather + than serving an empty table for the whole TTL. A benign check-then-fetch + race may issue a duplicate request under concurrent callers (this v1 is not + called concurrently); it never corrupts data -- last write wins. + """ + ad_date = self._norm_ad_date(date) if market == "twse" else None + key = (market, ad_date) + now = time.time() + with self._lock: + cached = self._cache.get(key) + if cached is not None and (now - self._cache_at.get(key, 0.0)) < self._cache_ttl: + return cached + table = self._fetch_twse(ad_date) if market == "twse" else self._fetch_tpex() + if table: # never cache an empty / failed fetch -> avoid a TTL-long silent blackout + with self._lock: + self._cache[key] = table + self._cache_at[key] = time.time() + return table + + def _throttle(self) -> None: + with self._throttle_lock: + wait = self._min_interval - (time.time() - self._last_request_at) + if wait > 0: + time.sleep(wait) + self._last_request_at = time.time() + + def _get_json(self, url: str, params: Optional[dict] = None) -> Any: + self._throttle() + resp = requests.get( + url, + params=params, + headers={"User-Agent": _UA, "Accept": "application/json"}, + timeout=self._timeout, + ) + resp.raise_for_status() + return resp.json() + + # ------------------------------------------------------------- TWSE T86 (上市) + def _fetch_twse(self, ad_date: Optional[str]) -> Dict[str, dict]: + params = {"response": "json", "selectType": "ALLBUT0999"} + if ad_date: + params["date"] = ad_date + payload = self._get_json(_T86_URL, params) + if not isinstance(payload, dict) or payload.get("stat") != "OK": + return {} + rows = payload.get("data") + if not isinstance(rows, list) or not rows: + return {} + idx = self._t86_index_map(payload.get("fields")) + if idx is None: # header missing or a core column renamed/removed -> fail-open + logger.info("[tw-inst] T86 fields header missing/renamed -> fail-open") + return {} + payload_date = self._norm_ad_date(payload.get("date")) or ad_date + table: Dict[str, dict] = {} + for row in rows: + record = self._parse_t86_row(row, payload_date, idx) + if record is not None: + table[record["stock_code"]] = record + return table + + @staticmethod + def _t86_index_map(fields: Any) -> Optional[Dict[str, int]]: + """Map each core T86 column NAME to its index, or None if any is missing. + + Reading by name (not a fixed index) means a TWSE column rename / reorder + fails open rather than silently shipping misaligned foreign/trust/dealer + numbers under stale indices. + """ + if not isinstance(fields, list): + return None + idx: Dict[str, int] = {} + for name in _T86_CORE: + try: + idx[name] = fields.index(name) + except ValueError: + return None + return idx + + @staticmethod + def _parse_t86_row( + row: Any, ad_date: Optional[str], idx: Dict[str, int] + ) -> Optional[dict]: + if not isinstance(row, (list, tuple)) or any(i >= len(row) for i in idx.values()): + return None + if ad_date is None: # data with no attributable trading date -> fail-open + return None + code = str(row[idx[_T86_CODE]]).strip() + if not code: + return None + foreign = _to_int(row[idx[_T86_FOREIGN]]) # 外陸資 (ex 外資自營商) + trust = _to_int(row[idx[_T86_TRUST]]) + dealer = _to_int(row[idx[_T86_DEALER]]) + total = _to_int(row[idx[_T86_TOTAL]]) + # A None core component means a missing / unparseable column (NOT genuine 0, + # which parses to 0) -> drop the row so a report never reads a fabricated zero. + if foreign is None or trust is None or dealer is None: + return None + return TwInstitutionalFetcher._build_record( + code, ad_date, "上市", "TWSE-T86", foreign, trust, dealer, total + ) + + # -------------------------------------------------------------- TPEx (上櫃) + def _fetch_tpex(self) -> Dict[str, dict]: + payload = self._get_json(_TPEX_URL) + if not isinstance(payload, list) or not payload: + return {} + table: Dict[str, dict] = {} + for raw in payload: + record = self._parse_tpex_row(raw) + if record is not None: + table[record["stock_code"]] = record + return table + + @staticmethod + def _parse_tpex_row(raw: Any) -> Optional[dict]: + if not isinstance(raw, dict): + return None + code = str(raw.get("SecuritiesCompanyCode", "")).strip() + if not code: + return None + ad_date = minguo_to_ad(raw.get("Date", "")) + if ad_date is None: # 民國 date unconvertible -> no attributable day -> fail-open + return None + foreign = _to_int(raw.get(_TPEX_FOREIGN_EXCL)) # dealer-excluded foreign + trust = _to_int(raw.get(_TPEX_TRUST)) + dealer = _to_int(raw.get(_TPEX_DEALER)) + total = _to_int(raw.get(_TPEX_TOTAL)) + # A None core component means a missing / renamed column -> fail-open (never a + # fabricated 0). Genuine zero activity parses to 0 and is kept. + if foreign is None or trust is None or dealer is None: + return None + return TwInstitutionalFetcher._build_record( + code, ad_date, "上櫃", "TPEx-OpenAPI", foreign, trust, dealer, total + ) + + # -------------------------------------------------------------- normalize + @staticmethod + def _build_record( + code: str, + ad_date: Optional[str], + market_label: str, + source: str, + foreign: int, + trust: int, + dealer: int, + total: Optional[int], + ) -> dict: + # foreign / trust / dealer are guaranteed non-None by the parsers (a missing + # component fails the row open upstream), so a genuine 0 is preserved as 0 + # and is never confused with a missing column. + return { + "stock_code": code, + "date": ad_date, + "market": market_label, + "source": source, + "unit": "shares", + "foreign_net": foreign, + "trust_net": trust, + "dealer_net": dealer, + # Official total when present; otherwise the component sum (kept consistent). + "total_net": total if total is not None else foreign + trust + dealer, + } diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index acdd2a693..90d18fe8d 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] - [改进] TickFlow 扩展为可选 A 股日 K、实时行情、股票列表/名称数据源，并为日 K 请求增加 count、完整性校验和批量预取缓存保护。 +- [新功能] #1777 新增台股三大法人（institutional flows）资料层 fetcher `TwInstitutionalFetcher`：上市走 TWSE T86 legacy `rwd` 端点、上柜走 TPEx OpenAPI，正规化外资/投信/自营商/三大法人每日买卖超（单位股数，民国↔西元日期转换有单测），按日期+市场单日缓存，失败/限流/空响应一律 fail-open；仅 `.TW`/`.TWO` 生效、严格 additive，不改动现有市场流程、不接报告/Web/评分/`capital_flow_signal`。资料来源为政府开放资料（OGDL v1）。 - [修复] API 异步批量分析共享概念板块排行缓存，避免同批多股重复拉取全市场概念排行。 - [文档] 补齐概念板块排行字段契约与通知报告行业/概念类型列展示说明。 - [新功能] #1742 新增信号归因分析功能（dashboard.signal_attribution），解释推荐理由的构成（技术指标、新闻舆情、基本面、市场环境的贡献度，以及最强看多/看空信号）。支持默认通知报告和 Jinja2 模板渲染，包含中英文国际化标签。归一化函数在 _parse_response() 和 parse_dashboard_json() 中显式调用，确保有效非零贡献度归一化到 100，all-zero 保留为 0（表示无有效信号）。 diff --git a/docs/market-support.md b/docs/market-support.md index 4e7349465..e4696da9e 100644 --- a/docs/market-support.md +++ b/docs/market-support.md @@ -107,6 +107,7 @@ PY - 报告 Prompt 已增加台股市场语义（新台币、三大法人、TWSE/TPEx ±10% 涨跌停），避免套用 A 股北向资金、龙虎榜等概念。 - 交易日历注册 `tw: XTAI / Asia/Taipei`。TWSE 为 09:00–13:30 连续交易、无午休；收盘集合竞价暂不建模，与 jp/kr 一致。若本地 `exchange-calendars` 版本缺少对应日历，既有 fail-open/fail-closed 语义保持不变。 - 主要指数提供加权指数 `^TWII` 与柜买指数 `^TWOII`。 +- 三大法人买卖超（institutional flows）资料层：`TwInstitutionalFetcher`（`data_provider/tw_institutional_fetcher.py`）提供上市（TWSE T86，legacy `rwd` 端点）/ 上柜（TPEx OpenAPI）每日外资·投信·自营商·三大法人买卖超（单位：**股数**；按日期+市场做单日全市场缓存再过滤个股，TPEx 民国年转西元有单测覆盖）。接口失败/限流/空响应/字段缺失一律 **fail-open** 返回无数据，不中断分析；仅对 `.TW`/`.TWO` 生效，不改动现有市场流程。资料来源为政府开放资料，采「政府资料开放授权条款第 1 版」(OGDL v1，允许商用与再散布，需标示来源)。**本次仅资料层 fetcher/parser/cache/tests，尚未接入报告展示、Web 展示、评分权重或 `capital_flow_signal` 派生。** 不承诺项： diff --git a/tests/test_tw_institutional_fetcher.py b/tests/test_tw_institutional_fetcher.py new file mode 100644 index 000000000..a3dd8b431 --- /dev/null +++ b/tests/test_tw_institutional_fetcher.py @@ -0,0 +1,262 @@ +# -*- coding: utf-8 -*- +"""Offline unit tests for TwInstitutionalFetcher (台股三大法人 data-layer fetcher). + +Fixtures are trimmed from real TWSE T86 / TPEx OpenAPI responses (captured +2026-06-26) so the parser is pinned to the actual field layout, date formats, +units and buy/sell-net signs — no network is touched. +""" + +import os +import sys +import unittest +from unittest.mock import patch, MagicMock + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from data_provider.tw_institutional_fetcher import ( # noqa: E402 + TwInstitutionalFetcher, + minguo_to_ad, + _to_int, +) + +# --- real TWSE T86 row for 2330 台積電 @ 20260626 (西元 date, comma-grouped values) --- +T86_FIXTURE = { + "stat": "OK", + "date": "20260626", + "fields": [ + "證券代號", "證券名稱", + "外陸資買進股數(不含外資自營商)", "外陸資賣出股數(不含外資自營商)", "外陸資買賣超股數(不含外資自營商)", + "外資自營商買進股數", "外資自營商賣出股數", "外資自營商買賣超股數", + "投信買進股數", "投信賣出股數", "投信買賣超股數", + "自營商買賣超股數", + "自營商買進股數(自行買賣)", "自營商賣出股數(自行買賣)", "自營商買賣超股數(自行買賣)", + "自營商買進股數(避險)", "自營商賣出股數(避險)", "自營商買賣超股數(避險)", + "三大法人買賣超股數", + ], + "data": [ + ["2330", "台積電 ", "22,676,018", "36,957,173", "-14,281,155", + "0", "0", "0", "1,034,258", "299,860", "734,398", + "1,009,368", "226,100", "769,604", "-543,504", + "3,424,484", "1,871,612", "1,552,872", "-12,537,389"], + ["2337", "旺宏 ", "99,038,413", "41,711,072", "57,327,341", + "0", "0", "0", "345,000", "3,850,000", "-3,505,000", + "1,914,924", "1,683,980", "1,636,000", "47,980", + "2,954,044", "1,087,100", "1,866,944", "55,737,265"], + ], +} + +# --- real TPEx OpenAPI row for 3105 穩懋 @ 民國 1150626 (plain ints, messy keys) --- +TPEX_FIXTURE = [ + { + "Date": "1150626", + "SecuritiesCompanyCode": "3105", + "CompanyName": "穩懋", + "Foreign Investors include Mainland Area Investors (Foreign Dealers excluded)-Total Buy": "11888101", + " Foreign Investors include Mainland Area Investors (Foreign Dealers excluded)-Total Sell": "12871054", + "Foreign Investors include Mainland Area Investors (Foreign Dealers excluded)-Difference": "-982953", + "Foreign Dealers-Total Buy": "0", + "Foreign Dealers-TotalSell": "0", + "ForeignDealers-Difference": "0", + "ForeignInvestorsIncludeMainlandAreaInvestors-TotalBuy": "11888101", + "ForeignInvestorsIncludeMainlandAreaInvestors-TotalSell": "12871054", + "ForeignInvestorsInclude MainlandAreaInvestors-Difference": "-982953", + "SecuritiesInvestmentTrustCompanies-TotalBuy": "29737", + "SecuritiesInvestmentTrustCompanies-TotalSell": "2924000", + "SecuritiesInvestmentTrustCompanies-Difference": "-2894263", + "Dealers-TotalBuy": "1228357", + "Dealers-TotalSell": "1726131", + "Dealers-Difference": "-497774", + "Dealers -TotalSell": "853696", + "TotalDifference": "-4374990", + }, +] + + +def _resp(json_data): + resp = MagicMock() + resp.status_code = 200 + resp.json.return_value = json_data + resp.raise_for_status.return_value = None + return resp + + +def _fetcher(): + # min_request_interval=0 disables the throttle sleep in tests. + return TwInstitutionalFetcher(min_request_interval=0) + + +class TestPureHelpers(unittest.TestCase): + def test_minguo_to_ad(self): + self.assertEqual(minguo_to_ad("1150626"), "20260626") # 民國115 -> 西元2026 + self.assertEqual(minguo_to_ad("1010101"), "20120101") # 民國101 -> 西元2012 + self.assertEqual(minguo_to_ad("0010101"), "19120101") # 民國1 -> 西元1912 + self.assertEqual(minguo_to_ad("0990101"), "20100101") # 民國99 -> 西元2010 + for bad in ("", "115062", "20260626", "abcdefg", None): + self.assertIsNone(minguo_to_ad(bad), bad) + + def test_to_int_preserves_sign_and_strips_commas(self): + self.assertEqual(_to_int("22,676,018"), 22676018) + self.assertEqual(_to_int("-14,281,155"), -14281155) # sign preserved + self.assertEqual(_to_int("0"), 0) + self.assertEqual(_to_int("10178972"), 10178972) # plain TPEx int + for blank in ("", "--", "-", "—", None, "n/a"): + self.assertIsNone(_to_int(blank), blank) + + +class TestT86Parsing(unittest.TestCase): + def test_twse_2330_net_breakdown(self): + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp(T86_FIXTURE)): + rec = _fetcher().get_institutional_net("2330.TW", "20260626") + self.assertIsNotNone(rec) + self.assertEqual(rec["stock_code"], "2330") + self.assertEqual(rec["market"], "上市") + self.assertEqual(rec["source"], "TWSE-T86") + self.assertEqual(rec["unit"], "shares") + self.assertEqual(rec["date"], "20260626") + # foreign = 外陸資 (不含外資自營商) = -14,281,155 + self.assertEqual(rec["foreign_net"], -14281155) + self.assertEqual(rec["trust_net"], 734398) + self.assertEqual(rec["dealer_net"], 1009368) + self.assertEqual(rec["total_net"], -12537389) + # the official total equals the component sum (sanity, sign-correct) + self.assertEqual(rec["total_net"], rec["foreign_net"] + rec["trust_net"] + rec["dealer_net"]) + + def test_twse_lowercase_suffix_and_other_row(self): + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp(T86_FIXTURE)): + rec = _fetcher().get_institutional_net("2337.tw") + self.assertEqual(rec["stock_code"], "2337") + self.assertEqual(rec["trust_net"], -3505000) # negative net preserved + + def test_twse_stat_not_ok_fails_open(self): + bad = {"stat": "很抱歉，沒有符合條件的資料!", "data": []} + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp(bad)): + self.assertIsNone(_fetcher().get_institutional_net("2330.TW", "20260101")) + + def test_twse_empty_data_fails_open(self): + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp({"stat": "OK", "data": []})): + self.assertIsNone(_fetcher().get_institutional_net("2330.TW")) + + +class TestTpexParsing(unittest.TestCase): + def test_tpex_3105_net_breakdown_and_minguo_date(self): + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp(TPEX_FIXTURE)): + rec = _fetcher().get_institutional_net("3105.TWO") + self.assertIsNotNone(rec) + self.assertEqual(rec["stock_code"], "3105") + self.assertEqual(rec["market"], "上櫃") + self.assertEqual(rec["source"], "TPEx-OpenAPI") + self.assertEqual(rec["date"], "20260626") # 民國 1150626 -> 西元 + self.assertEqual(rec["foreign_net"], -982953) + self.assertEqual(rec["trust_net"], -2894263) + self.assertEqual(rec["dealer_net"], -497774) + self.assertEqual(rec["total_net"], -4374990) + self.assertEqual(rec["total_net"], rec["foreign_net"] + rec["trust_net"] + rec["dealer_net"]) + + def test_tpex_non_list_fails_open(self): + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp({})): + self.assertIsNone(_fetcher().get_institutional_net("6488.TWO")) + + +class TestRoutingAndFailOpen(unittest.TestCase): + def test_bare_or_non_tw_code_returns_none_without_fetching(self): + with patch("data_provider.tw_institutional_fetcher.requests.get") as mock_get: + f = _fetcher() + self.assertIsNone(f.get_institutional_net("2330")) # bare -> not applicable + self.assertIsNone(f.get_institutional_net("AAPL")) + self.assertIsNone(f.get_institutional_net("600519.SH")) + mock_get.assert_not_called() + + def test_network_error_fails_open(self): + with patch("data_provider.tw_institutional_fetcher.requests.get", side_effect=ConnectionError("boom")): + self.assertIsNone(_fetcher().get_institutional_net("2330.TW", "20260626")) + + def test_unknown_stock_returns_none(self): + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp(T86_FIXTURE)): + self.assertIsNone(_fetcher().get_institutional_net("9999.TW")) + + def test_whole_market_cached_single_fetch(self): + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp(T86_FIXTURE)) as mock_get: + f = _fetcher() + f.get_institutional_net("2330.TW", "20260626") + f.get_institutional_net("2337.TW", "20260626") # same (market, date) -> cache hit + self.assertEqual(mock_get.call_count, 1) + + +class TestMissingFieldAndEmptyCacheFailOpen(unittest.TestCase): + """Dual-review P0/P1 guards: a missing/renamed column must NOT become a + fabricated 0, and an empty/failed fetch must NOT be cached for the TTL.""" + + def test_tpex_missing_core_field_drops_row(self): + import copy + row = copy.deepcopy(TPEX_FIXTURE[0]) + del row["SecuritiesInvestmentTrustCompanies-Difference"] # trust column renamed/missing + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp([row])): + self.assertIsNone(_fetcher().get_institutional_net("3105.TWO")) + + def test_tpex_genuine_zero_is_kept(self): + import copy + row = copy.deepcopy(TPEX_FIXTURE[0]) + row["SecuritiesInvestmentTrustCompanies-Difference"] = "0" # 投信 truly net-zero today + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp([row])): + rec = _fetcher().get_institutional_net("3105.TWO") + self.assertIsNotNone(rec) + self.assertEqual(rec["trust_net"], 0) # genuine 0 preserved + self.assertEqual(rec["foreign_net"], -982953) + + def test_twse_missing_core_field_drops_row(self): + import copy + fix = copy.deepcopy(T86_FIXTURE) + trust_idx = fix["fields"].index("投信買賣超股數") + fix["data"][0][trust_idx] = "" # 投信買賣超 cell blank -> missing -> drop 2330, not 0 + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp(fix)): + self.assertIsNone(_fetcher().get_institutional_net("2330.TW", "20260626")) + + def test_empty_result_not_cached_and_retried(self): + empty = {"stat": "OK", "data": []} + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp(empty)) as mock_get: + f = _fetcher() + self.assertIsNone(f.get_institutional_net("2330.TW", "20260626")) + self.assertIsNone(f.get_institutional_net("2330.TW", "20260626")) + self.assertEqual(mock_get.call_count, 2) # empty not cached -> re-fetched + + +class TestStructureRobustness(unittest.TestCase): + """Maintainer blockers: T86 read by column NAME (rename/reorder -> fail-open), + and TPEx date-required (an un-attributable trading day drops the row).""" + + def test_twse_reordered_fields_parsed_by_name(self): + import copy + fix = copy.deepcopy(T86_FIXTURE) + perm = list(range(len(fix["fields"])))[::-1] # reverse the column order + fix["fields"] = [fix["fields"][p] for p in perm] + fix["data"] = [[row[p] for p in perm] for row in fix["data"]] + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp(fix)): + rec = _fetcher().get_institutional_net("2330.TW", "20260626") + self.assertIsNotNone(rec) # parsed correctly despite reorder + self.assertEqual(rec["foreign_net"], -14281155) + self.assertEqual(rec["trust_net"], 734398) + self.assertEqual(rec["dealer_net"], 1009368) + self.assertEqual(rec["total_net"], -12537389) + + def test_twse_renamed_core_field_fails_open(self): + import copy + fix = copy.deepcopy(T86_FIXTURE) + fix["fields"][10] = "投信買賣超股數_v2" # 投信 column renamed + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp(fix)): + self.assertIsNone(_fetcher().get_institutional_net("2330.TW", "20260626")) + + def test_twse_missing_fields_header_fails_open(self): + fix = {"stat": "OK", "date": "20260626", "data": [["2330", "x", "1", "2", "3"]]} + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp(fix)): + self.assertIsNone(_fetcher().get_institutional_net("2330.TW", "20260626")) + + def test_tpex_unconvertible_date_drops_row(self): + import copy + row = copy.deepcopy(TPEX_FIXTURE[0]) + row["Date"] = "bad-date" # not a 7-digit 民國 -> drop + with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=_resp([row])): + self.assertIsNone(_fetcher().get_institutional_net("3105.TWO")) + + +if __name__ == "__main__": + unittest.main()