mirror of
https://github.com/ZhuLinsen/daily_stock_analysis
synced 2026-09-20 10:53:33 +08:00
feat(market): tw institutional-flows (三大法人) data-layer fetcher (#1829)
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
This commit is contained in:
Wenyu Chiou
337
data_provider/tw_institutional_fetcher.py
Normal file
337
data_provider/tw_institutional_fetcher.py
Normal file
@@ -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,
|
||||
}
|
||||
@@ -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（表示无有效信号）。
|
||||
|
||||
@@ -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` 派生。**
|
||||
|
||||
不承诺项：
|
||||
|
||||
|
||||
262
tests/test_tw_institutional_fetcher.py
Normal file
262
tests/test_tw_institutional_fetcher.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user