feat(data): D5 opt-in bounded concurrency with a global rate limiter

config/data_update.yaml

@@ -85,6 +85,13 @@ data_update:
       - adj_factor
       - stk_mins_1min
     report_name: data_update_quality_report.md
+  # D5 opt-in bounded concurrency. max_workers=1 (default) keeps the warm fully
+  # serial — every existing config behaves exactly as before. >1 fans the dense
+  # per-symbol gap FETCH onto a bounded thread pool, all funneled through ONE
+  # shared global rate limiter (reusing rate_limit_per_min above), so the Tushare
+  # quota is never multiplied per thread; store + ledger writes stay deterministic.
+  concurrency:
+    max_workers: 1
 
 # --- unused by data-update (schema-only stubs) ------------------------------- #
 factors:

data/cache/tushare_cache.py

@@ -114,6 +114,28 @@
 FetchSnapshot = Callable[..., "pd.DataFrame | None"]
 
 
+class _FetchOk:
+    """A successful concurrent gap fetch (D5): carries the raw frame."""
+
+    __slots__ = ("raw",)
+
+    def __init__(self, raw) -> None:
+        self.raw = raw
+
+
+class _FetchError:
+    """A failed concurrent gap fetch (D5): carries the exception OBJECT only.
+
+    Re-raised on the main thread after durable successes; the message is the
+    feed-level secret-safe ``RuntimeError`` (exception type, never the token).
+    """
+
+    __slots__ = ("exc",)
+
+    def __init__(self, exc: Exception) -> None:
+        self.exc = exc
+
+
 class TushareCache:
     """Endpoint-level read-through cache (market bars + universe/tradability)."""
 
@@ -130,6 +152,7 @@ def __init__(
         source_version: str | None = None,
         not_ready_days: int = 0,
         recent_tail_overrides: dict[str, int] | None = None,
+        max_workers: int = 1,
     ) -> None:
         self._store = store
         self._ledger = ledger
@@ -150,6 +173,12 @@ def __init__(
         # REQUESTED range (not today-based) — e.g. fina_indicator refetches a long
         # trailing window of recent report periods to catch LATE disclosures.
         self._recent_tail_overrides = dict(recent_tail_overrides or {})
+        # D5: bounded concurrency for the dense per-symbol gap FETCH stage. 1
+        # (default) keeps the serial read-through path byte-identical; >1 fans the
+        # planned gap fetches onto a thread pool (the shared GlobalRateLimiter lives
+        # in the feeds, so the global quota holds), while store upserts + ledger
+        # writes stay on the main thread in deterministic order.
+        self._max_workers = max(1, int(max_workers))
         # per-instance endpoint fetch counters (cache stats; one increment per
         # gap/window/snapshot actually sent to the API). A fully-covered repeat
         # run leaves these at zero — the read-through hit rate is observable from
@@ -376,6 +405,12 @@ def _read_through(
         columns: list[str],
         key_cols: list[str],
     ) -> pd.DataFrame:
+        # D5: only the opted-in multi-worker, multi-symbol case takes the
+        # concurrent path; everything else stays on the byte-identical serial loop.
+        if self._max_workers > 1 and len(symbols) > 1:
+            return self._read_through_concurrent(
+                endpoint, symbols, start, end, fetch, parse, columns, key_cols
+            )
         req_start = pd.Timestamp(start).normalize()
         req_end = pd.Timestamp(end).normalize()
         fields_hash = _fields_hash(columns)
@@ -409,6 +444,103 @@ def _read_through(
             return pd.DataFrame(columns=columns)
         return pd.concat(out, ignore_index=True).reset_index(drop=True)
 
+    def _read_through_concurrent(
+        self, endpoint, symbols, start, end, fetch, parse, columns, key_cols
+    ) -> pd.DataFrame:
+        """D5 bounded-concurrency read-through: parallel fetch, serial durable writes.
+
+        Phase 1 (main thread): plan every uncovered gap (ledger reads use the D4
+        lookup cache). Phase 2: fetch the planned gaps on a bounded thread pool —
+        each fetch funnels through the feeds' shared ``GlobalRateLimiter``, so the
+        global quota holds. Phase 3 (main thread, deterministic plan order):
+        parse/upsert/record each SUCCESSFUL gap so it is durable, then re-raise the
+        first failure (a failed gap records no coverage and stays retryable, exactly
+        like the serial path). Phase 4: read the requested window back per symbol.
+        The store/ledger contents are independent of fetch completion order.
+        """
+        req_start = pd.Timestamp(start).normalize()
+        req_end = pd.Timestamp(end).normalize()
+        fields_hash = _fields_hash(columns)
+        forced = endpoint in self._force_refresh
+
+        # Phase 1 — plan (main thread).
+        plans: list[tuple[str, pd.Timestamp, pd.Timestamp]] = []
+        n_covered = 0
+        for symbol in symbols:
+            gaps = self._gaps_for(endpoint, symbol, req_start, req_end, forced)
+            if not gaps:
+                n_covered += 1
+            for gap_start, gap_end in gaps:
+                plans.append((symbol, gap_start, gap_end))
+
+        # Phase 2 — concurrent fetch (bounded by max_workers).
+        raw_results = self._fetch_plans_concurrent(plans, fetch)
+
+        # Phase 3 — deterministic durable writes (main thread, plan order).
+        first_exc: Exception | None = None
+        for (symbol, gap_start, gap_end), res in zip(plans, raw_results):
+            if isinstance(res, _FetchError):
+                if first_exc is None:
+                    first_exc = res.exc
+                continue
+            self.fetch_counts[endpoint] = self.fetch_counts.get(endpoint, 0) + 1
+            parsed = parse(res.raw)
+            if len(parsed):
+                self._store.upsert_symbol(endpoint, symbol, parsed, key_cols)
+                self.written_counts[endpoint] = (
+                    self.written_counts.get(endpoint, 0) + len(parsed)
+                )
+            self._record_gap_coverage(
+                endpoint, symbol, gap_start, gap_end, parsed, fields_hash
+            )
+        if first_exc is not None:
+            # earlier successful gaps are already durable; the failed gap is not
+            # recorded, so a later run retries it (serial-equivalent failure model).
+            raise first_exc
+
+        # Phase 4 — read back the requested window (symbol order).
+        out: list[pd.DataFrame] = []
+        for symbol in symbols:
+            cached = self._store.read_symbol(endpoint, symbol)
+            if not cached.empty:
+                mask = (cached["date"] >= req_start) & (cached["date"] <= req_end)
+                hit = cached.loc[mask, columns]
+                if not hit.empty:
+                    out.append(hit)
+        _LOGGER.info(
+            "cache %s: %d symbols, %d gap-fetches, %d fully-covered "
+            "(api calls=%d, workers=%d)",
+            endpoint, len(symbols), len(plans), n_covered,
+            self.fetch_counts[endpoint], self._max_workers,
+        )
+        if not out:
+            return pd.DataFrame(columns=columns)
+        return pd.concat(out, ignore_index=True).reset_index(drop=True)
+
+    def _fetch_plans_concurrent(self, plans, fetch):
+        """Fetch every planned ``(symbol, start, end)`` gap on a bounded pool.
+
+        Returns a list aligned to ``plans``: a ``_FetchOk(raw)`` or a
+        ``_FetchError(exc)`` per gap. Each task catches its OWN exception (so no
+        worker exception escapes the pool); the throttle/retry + the shared global
+        rate limiter live inside ``fetch`` (the feed closure).
+        """
+        if not plans:
+            return []
+        from concurrent.futures import ThreadPoolExecutor
+
+        def task(i):
+            symbol, gap_start, gap_end = plans[i]
+            try:
+                raw = fetch(symbol, _compact(gap_start), _compact(gap_end))
+                return _FetchOk(raw)
+            except Exception as exc:  # secret-safe: only the object is kept
+                return _FetchError(exc)
+
+        workers = min(self._max_workers, len(plans))
+        with ThreadPoolExecutor(max_workers=workers) as pool:
+            return list(pool.map(task, range(len(plans))))
+
     def _gaps_for(self, endpoint, symbol, req_start, req_end, forced):
         """The uncovered sub-intervals to fetch (+ a forced trailing tail)."""
         if forced:

data/feed/index_feed.py

@@ -33,6 +33,7 @@ def __init__(
         rate_limit: int | None = None,
         max_retries: int = 6,
         cache=None,
+        scheduler=None,
     ) -> None:
         self._secret_file = str(secret_file)
         self._token_key = token_key
@@ -44,6 +45,8 @@ def __init__(
         # one. The cache stores RAW snapshots; the as-of membership stays
         # downstream, unchanged.
         self._cache = cache
+        # D5: optional shared GlobalRateLimiter; None keeps the per-call throttle.
+        self._scheduler = scheduler
 
     # -- secret handling (token never logged) ------------------------------- #
     def _client(self):
@@ -90,6 +93,7 @@ def get_constituents(self, index_code: str, start: str, end: str) -> pd.DataFram
                 pro.index_weight,
                 max_retries=self._max_retries,
                 rate_limit=self._rate_limit,
+                scheduler=self._scheduler,
                 index_code=index_code,
                 start_date=win_start.strftime("%Y%m%d"),
                 end_date=win_end.strftime("%Y%m%d"),
@@ -121,6 +125,7 @@ def fetch(index_code: str, start_compact: str, end_compact: str):
                 self._client().index_weight,
                 max_retries=self._max_retries,
                 rate_limit=self._rate_limit,
+                scheduler=self._scheduler,
                 index_code=index_code,
                 start_date=start_compact,
                 end_date=end_compact,

data/feed/scheduler.py

@@ -0,0 +1,73 @@
+"""Global request scheduler / rate limiter for bounded-concurrency warms (D5).
+
+When a cache warm runs with more than one worker, every worker must funnel its
+API attempts through ONE shared budget — otherwise N threads each sleeping
+``60 / rate_limit`` would multiply the global quota by N and blow the Tushare
+ceiling. :class:`GlobalRateLimiter` reserves the next evenly-spaced time slot
+under a lock, then sleeps OUTSIDE the lock until that slot, so concurrent callers
+are serialized onto one global cadence.
+
+It is transport-agnostic and secret-free: it never sees a token, kwargs, an
+endpoint, or an exception payload — only a per-minute integer budget. ``monotonic``
+and ``sleep`` are injectable so the spacing can be asserted with a fake clock (no
+wall-clock timing in tests).
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from collections.abc import Callable
+
+
+class GlobalRateLimiter:
+    """A single shared per-minute request budget enforced ACROSS worker threads.
+
+    ``rate_limit_per_min <= 0`` disables throttling (``acquire`` is a no-op). Each
+    ``acquire`` reserves the next slot spaced ``60 / rate_limit_per_min`` seconds
+    after the previous reservation, regardless of which thread calls it — so the
+    global rate holds no matter how many workers contend.
+    """
+
+    def __init__(
+        self,
+        rate_limit_per_min: int,
+        *,
+        monotonic: Callable[[], float] = time.monotonic,
+        sleep: Callable[[float], None] = time.sleep,
+    ) -> None:
+        self._rate_limit_per_min = int(rate_limit_per_min)
+        self._interval = (
+            60.0 / self._rate_limit_per_min if self._rate_limit_per_min > 0 else 0.0
+        )
+        self._monotonic = monotonic
+        self._sleep = sleep
+        self._lock = threading.Lock()
+        self._next_allowed: float | None = None
+
+    @property
+    def rate_limit_per_min(self) -> int:
+        """The configured global per-minute budget (0 == disabled)."""
+        return self._rate_limit_per_min
+
+    def acquire(self) -> None:
+        """Block until this caller's globally-spaced slot; no-op if disabled.
+
+        The slot is reserved under the lock (so two threads never claim the same
+        instant), then the wait happens outside the lock so other threads can keep
+        reserving their own later slots concurrently.
+        """
+        if self._interval <= 0:
+            return
+        with self._lock:
+            now = self._monotonic()
+            if self._next_allowed is None or now >= self._next_allowed:
+                # the budget is idle: this caller goes now, next slot one interval out
+                self._next_allowed = now + self._interval
+                wait = 0.0
+            else:
+                # the budget is busy: wait for the reserved slot, push the next one out
+                wait = self._next_allowed - now
+                self._next_allowed = self._next_allowed + self._interval
+        if wait > 0:
+            self._sleep(wait)

data/feed/throttle.py

@@ -18,25 +18,35 @@ def request_with_retry(
     *,
     max_retries: int = 6,
     rate_limit: int | None = None,
+    scheduler: Any = None,
     **kwargs: Any,
 ) -> Any:
-    """Call ``fn(**kwargs)`` with retry-on-error + per-minute throttle.
+    """Call ``fn(**kwargs)`` with retry-on-error + a per-minute throttle.
 
     Retries transient exceptions with exponential backoff up to ``max_retries``
-    attempts; on success sleeps ``60 / rate_limit`` seconds (no-op if unset).
-    Raises a readable ``RuntimeError`` (type only, never the token) if every
-    attempt fails.
+    attempts. Throttling has two mutually-exclusive modes:
+
+    * ``scheduler`` set (D5 bounded concurrency): acquire one global slot BEFORE
+      each attempt, so every worker's attempts (including retries) share one budget
+      — no per-call ``rate_limit`` sleep is added (that would double the spacing).
+    * ``scheduler`` unset (the historical serial path): on success sleep
+      ``60 / rate_limit`` seconds (no-op if unset).
+
+    Raises a readable ``RuntimeError`` (exception TYPE only, never the token /
+    payload) if every attempt fails.
     """
     attempts = max(1, int(max_retries))
     last_exc: Exception | None = None
     for attempt in range(attempts):
+        if scheduler is not None:
+            scheduler.acquire()  # global slot before EVERY attempt (incl. retries)
         try:
             result = fn(**kwargs)
         except Exception as exc:  # transient API / network error
             last_exc = exc
             time.sleep(min(2.0**attempt, 8.0))
             continue
-        if rate_limit:
+        if scheduler is None and rate_limit:
             time.sleep(60.0 / rate_limit)
         return result
     raise RuntimeError(

data/feed/tushare_covariates.py

@@ -42,6 +42,7 @@ def __init__(
         rate_limit: int | None = None,
         max_retries: int = 6,
         cache=None,
+        scheduler=None,
     ) -> None:
         self._secret_file = str(secret_file)
         self._token_key = token_key
@@ -53,6 +54,8 @@ def __init__(
         # ``pit_sw_intervals`` (index_member_all, P4-3) all read through it.
         # None keeps the historical direct fetch EXACTLY.
         self._cache = cache
+        # D5: optional shared GlobalRateLimiter; None keeps the per-call throttle.
+        self._scheduler = scheduler
 
     def _client(self):
         if self._pro is None:
@@ -63,7 +66,8 @@ def _client(self):
 
     def _call(self, fn, **kwargs):
         return request_with_retry(
-            fn, max_retries=self._max_retries, rate_limit=self._rate_limit, **kwargs
+            fn, max_retries=self._max_retries, rate_limit=self._rate_limit,
+            scheduler=self._scheduler, **kwargs,
         )
 
     def industry(self, symbols: list[str]) -> dict[str, str]:

data/feed/tushare_feed.py

@@ -47,6 +47,7 @@ def __init__(
         rate_limit: int | None = None,
         max_retries: int = 6,
         cache=None,
+        scheduler=None,
     ) -> None:
         self._secret_file = str(secret_file)
         self._token_key = token_key
@@ -59,6 +60,8 @@ def __init__(
         # passes one in. The cache stores RAW rows only; front_adjust still runs
         # downstream, unchanged.
         self._cache = cache
+        # D5: optional shared GlobalRateLimiter; None keeps the per-call throttle.
+        self._scheduler = scheduler
 
     # -- secret handling ---------------------------------------------------- #
     def _read_token(self) -> str:
@@ -82,6 +85,7 @@ def _call(self, fn, **kwargs):
             fn,
             max_retries=self._max_retries,
             rate_limit=self._rate_limit,
+            scheduler=self._scheduler,
             **kwargs,
         )