mirrors / zhulinsen--daily_stock_analysis

项目文件夹

zhulinsen--daily_stock_analysis
文件
main
T
wehub-resource-sync 084fa65ec4
Auto Tag / tag (push) Has been skipped
详情
chore: import upstream snapshot with attribution
2026-07-13 12:05:10 +08:00

399 行
17 KiB
Python
原始文件 永久链接 Blame 文件历史

# -*- 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
from data_provider.realtime_types import CircuitBreaker
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()
# One lock per unique (market, ad_date) key; bounded by tw markets x
# distinct dates queried -- low thousands at most, negligible memory.
self._inflight: Dict[Any, threading.Lock] = {}
# Per-market circuit breaker (keyed "twse"/"tpex"): when an endpoint is down
# (>= 3 consecutive failures) skip the network round-trip for ~5 min and fail
# open, instead of paying timeout + throttle on every stock during an outage.
# Reuses the repo's CircuitBreaker (same one DataFetcherManager uses).
self._breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=300.0)
# ------------------------------------------------------------------ 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
record = table.get(base)
# TPEx OpenAPI serves only the LATEST trading day (no date param). If a caller
# asked for a specific date, never silently return a different-day record --
# fail open (None) so a date-mismatched 上櫃 figure can't reach a report.
if record is not None and date and market == "tpex":
requested = self._norm_ad_date(date)
if requested and record.get("date") != requested:
logger.info(
"[tw-inst] TPEx %s requested date %s != served %s -> fail-open",
base, requested, record.get("date"),
)
return None
return record
# ------------------------------------------------------------------ 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.
Concurrent callers for the SAME (market, date) coalesce into a single
upstream fetch (cache-stampede guard) -- this keeps the T86 ~3 req/5 s
budget intact under parallel callers; different keys still fetch in
parallel, and the master lock is never held across network I/O. On a fetch
error the key-lock is released and waiting callers each retry independently
(serialized only by _throttle), since failures are deliberately not cached.
"""
ad_date = self._norm_ad_date(date) if market == "twse" else None
key = (market, ad_date)
cached = self._read_cache(key)
if cached is not None:
return cached
# Serialize same-key fetches so a burst of callers issues ONE request, not N.
with self._key_lock(key):
cached = self._read_cache(key) # double-check: a prior holder may have filled it
if cached is not None:
return cached
# Circuit breaker: if this endpoint has been failing (>= 3 in a row), skip
# the network round-trip and fail open (empty) until the ~5 min cooldown
# half-opens -- so a TWSE/TPEx outage costs ~0 per stock, not timeout+throttle.
if not self._breaker.is_available(market):
logger.info("[tw-inst] %s circuit OPEN -> skip fetch, fail-open", market)
return {}
try:
table = self._fetch_twse(ad_date) if market == "twse" else self._fetch_tpex()
except Exception as exc: # network / HTTP error -> trip the breaker, then re-raise
self._breaker.record_failure(market, str(exc))
raise
# The breaker tracks REACHABILITY (open only on hard network/HTTP errors).
# An empty / stat!=OK body still means the endpoint RESPONDED, so it counts
# as success: it resets the failure streak and, during HALF_OPEN recovery,
# closes the breaker instead of re-opening it (so a no-data day mid-recovery
# can never strand the breaker open). Only non-empty tables are cached.
self._breaker.record_success(market)
if table: # never cache an empty / failed fetch -> no TTL-long blackout
with self._lock:
self._cache[key] = table
self._cache_at[key] = time.time()
return table
def _read_cache(self, key: Any) -> Optional[Dict[str, dict]]:
with self._lock:
cached = self._cache.get(key)
if cached is not None and (time.time() - self._cache_at.get(key, 0.0)) < self._cache_ttl:
return cached
return None
def _key_lock(self, key: Any) -> threading.Lock:
with self._lock:
lock = self._inflight.get(key)
if lock is None:
lock = threading.Lock()
self._inflight[key] = lock
return lock
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,
}
在新工单中引用 查看 Git Blame 复制永久链接