perf(market): coalesce concurrent tw-institutional fetches (cache-stampede guard) (#1841)

Hardens the merged #1777 TwInstitutionalFetcher. Under concurrent callers for the
same (market, date) the previous check-then-fetch could issue N duplicate upstream
requests -- which for the TWSE T86 RWD endpoint risks tripping its informal
~3 req/5s ban. A per-(market,date) in-flight lock now coalesces same-key callers
into a single fetch (double-checked against the cache); different keys still fetch
in parallel, and the master lock is never held across network I/O.

- data_provider/tw_institutional_fetcher.py: per-key in-flight lock + _read_cache /
  _key_lock helpers. Behavior is otherwise unchanged -- still fail-open, still
  caches only non-empty results, tw-only, no data_provider/base.py change.
- tests: + concurrent-same-key-coalesces-to-single-fetch (8 threads -> 1 request),
  different-keys-not-coalesced, and HTTP-error (429) fail-open
- docs/CHANGELOG.md: [改进] entry

Refs #1777

Co-authored-by: zhulinsen <42829555+ZhuLinsen@users.noreply.github.com>

data_provider/tw_institutional_fetcher.py

@@ -119,6 +119,9 @@ class TwInstitutionalFetcher:
         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] = {}
 
     # ------------------------------------------------------------------ public
     def get_institutional_net(
@@ -174,23 +177,46 @@ class TwInstitutionalFetcher:
         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.
+        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)
-        now = time.time()
+        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
+            table = self._fetch_twse(ad_date) if market == "twse" else self._fetch_tpex()
+            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 (now - self._cache_at.get(key, 0.0)) < self._cache_ttl:
+            if cached is not None and (time.time() - 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
+        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:

docs/CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+- [改进] #1777 台股三大法人 fetcher（`TwInstitutionalFetcher`）增加缓存防击穿：并发同 (市场, 日期) 调用合并为单次上游请求，保护 TWSE T86 ~3 req/5s 限流额度；不同 key 仍并行；新增并发单次抓取、不同 key 各抓一次、HTTP 错误 fail-open 回归测试。
 - [修复] A 股个股分析遇到空 `belong_boards` 占位时会继续补查所属板块，关联板块模块在已有板块时稳定展示；对应涨跌幅缺失时只显示板块，不再输出占位涨跌幅。
 - [修复] 大盘复盘在 LLM 标题漂移或正文缺少板块段时，会从结构化 `sectors` 兜底渲染板块表，避免 Web 与推送报告偶发缺少板块主线。
 

tests/test_tw_institutional_fetcher.py

@@ -258,5 +258,49 @@ class TestStructureRobustness(unittest.TestCase):
             self.assertIsNone(_fetcher().get_institutional_net("3105.TWO"))
 
 
+class TestConcurrencyAndHttpError(unittest.TestCase):
+    """Cache-stampede guard: concurrent same-key callers coalesce into one upstream
+    fetch (protects the T86 ~3 req/5 s budget); HTTP errors fail open."""
+
+    def test_concurrent_same_key_coalesces_to_single_fetch(self):
+        import threading
+        import time as _t
+        calls = []
+        barrier = threading.Barrier(8)
+
+        def slow_get(*a, **k):
+            calls.append(1)
+            _t.sleep(0.05)   # window for the other threads to pile up on the key lock
+            return _resp(T86_FIXTURE)
+
+        f = _fetcher()
+
+        def caller():
+            barrier.wait()   # release all 8 threads together so they truly race
+            f.get_institutional_net("2330.TW", "20260626")
+
+        with patch("data_provider.tw_institutional_fetcher.requests.get", side_effect=slow_get):
+            threads = [threading.Thread(target=caller) for _ in range(8)]
+            for th in threads:
+                th.start()
+            for th in threads:
+                th.join()
+        self.assertEqual(len(calls), 1)   # 8 concurrent same-key callers -> ONE fetch
+
+    def test_different_keys_are_not_coalesced(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("2330.TW", "20260625")   # different date -> different key
+            self.assertEqual(mock_get.call_count, 2)
+
+    def test_http_error_fails_open(self):
+        import requests as _rq
+        resp = MagicMock()
+        resp.raise_for_status.side_effect = _rq.HTTPError("429 Too Many Requests")
+        with patch("data_provider.tw_institutional_fetcher.requests.get", return_value=resp):
+            self.assertIsNone(_fetcher().get_institutional_net("2330.TW", "20260626"))
+
+
 if __name__ == "__main__":
     unittest.main()