diff --git a/config/data_update.yaml b/config/data_update.yaml new file mode 100644 index 0000000..c4430ea --- /dev/null +++ b/config/data_update.yaml @@ -0,0 +1,126 @@ +# P4-3 data updater config — the daily 21:00 (Asia/Shanghai) incremental warm. +# +# Used ONLY by: python -m qt.cli data-update --config config/data_update.yaml +# It warms/updates the read-through caches and runs NOTHING else (no factors, +# no alpha, no portfolio, no backtest, no PanelStore). Scheduling is external: +# a systemd timer / cron fires the CLI at 21:00 Asia/Shanghai (see the +# stage_p4_3 acceptance doc for an example unit). +# +# The factors/alpha/portfolio/backtest/cost/output sections below are present +# ONLY to satisfy the shared RootConfig schema (so `validate-config` accepts this +# file); the data-update command ignores them. + +project: + name: quantitative_trading_data_update + timezone: Asia/Shanghai + +data: + source: tushare + freq: D + start: "2024-01-01" + end: "2024-12-31" + external_secret_file: "/home/shaofl/Projects/financial_projects/.config.json" + tushare_token_key: "tushare.token" + output_name: data_update + cache: + enabled: true + root_dir: artifacts/cache/tushare/v1 + refresh_recent_days: 14 + refresh_dimension_days: 30 + force_refresh: [] + +universe: + type: index + index_code: "000300.SH" + symbols: [] + min_listing_days: 60 + filters: + missing_close: true + suspended: true + st: true + limit_up_down: true + +# The actual data-update job parameters. +data_update: + timezone: Asia/Shanghai + scheduled_start: "21:00:00" + endpoints: + - market_daily + - adj_factor + - index_weight + - suspend_d + - namechange + - stk_limit + - stock_basic + - daily_basic + - fina_indicator + - index_member_all + index_codes: + - "000300.SH" + lookback_days: 400 # dense endpoints: re-warm [today - lookback, today] + tail_refresh_days: 14 # recent-tail always refetched (corrections/delays) + not_ready_days: 1 # today may be unpublished at 21:00 -> not_ready, not covered + fina_tail_days: 400 # refetch recent report periods to catch LATE disclosures + # NOTE: the cache ALWAYS stores the canonical fina superset (roe / netprofit_yoy + # / grossprofit_margin) regardless of this list, so a warm here never blocks a + # later config that needs a different field. grossprofit_margin is listed + # explicitly anyway (real configs use it) for a self-documenting summary. + fina_fields: + - roe + - netprofit_yoy + - grossprofit_margin + rate_limit_per_min: 450 # conservative (official stk_mins ceiling is 500/min) + force_refresh: [] + +# --- unused by data-update (schema-only stubs) ------------------------------- # +factors: + - name: momentum_20 + enabled: true + params: + window: 20 + price_col: close + +processing: + drop_missing: true + standardize: + enabled: true + method: zscore + winsorize: + enabled: false + method: mad + n: 3.0 + neutralize: + enabled: true + industry_col: industry + size_col: market_cap + industry_level: L1 + +alpha: + model: equal_weight + params: {} + +portfolio: + constructor: topn_equal_weight + top_n: 30 + long_only: true + max_weight: null + turnover_cap: null + +backtest: + initial_nav: 1.0 + rebalance: monthly + event_order: close_to_next_period + cash_return: 0.0 + +cost: + fee_rate: 0.001 + slippage_rate: 0.0 + turnover_formula: l1 + +output: + root_dir: artifacts + data_dir: artifacts/data + factor_dir: artifacts/factors + report_dir: artifacts/reports + log_dir: artifacts/logs + overwrite: true diff --git a/data/cache/__init__.py b/data/cache/__init__.py index 85e43f4..64e3406 100644 --- a/data/cache/__init__.py +++ b/data/cache/__init__.py @@ -12,8 +12,11 @@ * PanelStore remains a per-run artifact layer, NOT the cache source of truth; * the cache and its ledger never store a token or any secret-file content. -P4-1 implements ``market_daily`` and ``adj_factor`` only. Universe/tradability -(P4-2) and factor-support endpoints (P4-3) are deliberately out of scope here. +Cached endpoints: ``market_daily`` / ``adj_factor`` (P4-1); ``index_weight`` / +``suspend_d`` / ``namechange`` / ``stk_limit`` / ``stock_basic`` (P4-2); +``daily_basic`` / ``fina_indicator`` / ``index_member_all`` (P4-3, factor +support). The separate intraday 1min cache (``stk_mins``) lives in +``data.cache.intraday_*`` with its own timestamp-interval ledger. """ from __future__ import annotations diff --git a/data/cache/tushare_cache.py b/data/cache/tushare_cache.py index 10cc531..da57059 100644 --- a/data/cache/tushare_cache.py +++ b/data/cache/tushare_cache.py @@ -1,21 +1,29 @@ -"""Read-through cache for tushare endpoints (P4-1 market bars + P4-2 universe/tradability). +"""Read-through cache for tushare endpoints (P4-1 market bars + P4-2 universe/ +tradability + P4-3 factor-support: daily_basic / fina_indicator / index_member_all). ``TushareCache`` turns a requested range into ONLY the uncovered gaps (or a stale snapshot), fetches those via a caller-supplied ``fetch`` callable (the feed wraps its own retry/throttle there, so the cache stays transport-agnostic), -upserts the raw rows, records coverage (including empty returns), then returns -the full requested data read back from the cache. +upserts the raw rows, records coverage (including empty / not-ready returns), +then returns the full requested data read back from the cache. Three planning shapes share one engine: * dense per-symbol date-range (``market_daily``, ``adj_factor``, ``suspend_d``, - ``stk_limit``): coverage key = symbol, gaps subtracted per symbol, a recent - tail refetched within ``refresh_recent_days``; + ``stk_limit``, ``daily_basic``, ``fina_indicator``): coverage key = symbol, + gaps subtracted per symbol, a trailing tail refetched (today-based + ``refresh_recent_days``, or a per-endpoint range-trailing override — fina + refetches recent report periods to catch LATE disclosures); * index-keyed date-range (``index_weight``): coverage key = index_code, gaps subtracted over the whole index, each uncovered gap paged in <=90-day windows (tushare's per-call row cap), raw snapshots stored; - * snapshot / dimension (``namechange`` per-symbol, ``stock_basic`` global): - no date range — refetched only when never fetched, stale beyond - ``refresh_dimension_days``, or force-refreshed. + * snapshot / dimension (``namechange`` per-symbol, ``index_member_all`` + per-symbol, ``stock_basic`` global): no date range — refetched only when + never fetched, stale beyond ``refresh_dimension_days``, or force-refreshed. + +``fina_indicator`` is field-set dependent and ALWAYS stores the canonical +``FINA_FIELDS`` superset (one schema, one coverage), so a warm for one config's +fields never blocks another's. P4-3 also adds a ``not_ready`` pending window +(today's unpublished data is not frozen as covered). Behaviour the acceptance pins down (each endpoint): * full cache miss -> fetch the uncovered range, populate cache; @@ -53,12 +61,16 @@ NAMECHANGE = "namechange" STK_LIMIT = "stk_limit" STOCK_BASIC = "stock_basic" +# P4-3 factor-support endpoints. +DAILY_BASIC = "daily_basic" # dense per-symbol date-range (pe/pb/total_mv) +FINA_INDICATOR = "fina_indicator" # per-symbol, report-period range, carries ann_date +INDEX_MEMBER_ALL = "index_member_all" # per-symbol dimension (SW in/out intervals) # every endpoint this cache knows (fetch_counts is seeded with all of them so a # warm run reports an explicit 0 for an endpoint it never had to touch). ALL_ENDPOINTS = ( MARKET_DAILY, ADJ_FACTOR, INDEX_WEIGHT, SUSPEND_D, NAMECHANGE, - STK_LIMIT, STOCK_BASIC, + STK_LIMIT, STOCK_BASIC, DAILY_BASIC, FINA_INDICATOR, INDEX_MEMBER_ALL, ) # sentinel key for a global (whole-market) snapshot endpoint (stock_basic). @@ -84,6 +96,32 @@ _STOCK_BASIC_COLUMNS = ["symbol", "list_date"] _STOCK_BASIC_KEY = ["symbol"] +# P4-3 endpoints. daily_basic is a dense per-symbol date-range (like market bars); +# the stored ``date`` is the trade_date. fina_indicator's stored ``date`` is the +# REPORT-PERIOD end_date (the axis tushare filters on); ann_date is kept as a raw +# column for the downstream PIT as-of (ann_date <= trade_date) — never as the +# coverage axis. index_member_all is a per-symbol dimension of SW in/out intervals. +_DAILY_BASIC_COLUMNS = ["date", "symbol", "pe", "pb", "total_mv"] +_DAILY_BASIC_KEY = ["date", "symbol"] + +# fina_indicator is field-set dependent: a per-(symbol) parquet keyed by +# (symbol, end_date, ann_date) CANNOT hold two different field sets for the same +# report (a later subset upsert would overwrite an earlier one and a covered +# interval would be reused with the wrong columns). So the cache ALWAYS fetches + +# stores the CANONICAL SUPERSET of every financial field the project supports; a +# caller (the feed) selects its requested subset on read. Coverage is therefore +# uniform (one field set) and a subset warm never blocks a later different-subset +# request. Must stay a superset of factors.compute.financial.SUPPORTED_FIELDS +# (guarded by a drift test). +FINA_FIELDS: tuple[str, ...] = ("roe", "netprofit_yoy", "grossprofit_margin") +_FINA_COLUMNS = ["date", "symbol", "ann_date", "end_date", *FINA_FIELDS] +_FINA_KEY = ["symbol", "end_date", "ann_date"] + +_INDEX_MEMBER_COLUMNS = [ + "symbol", "l1_name", "l2_name", "l3_name", "in_date", "out_date" +] +_INDEX_MEMBER_KEY = ["symbol", "in_date", "out_date", "l1_name", "l2_name", "l3_name"] + # tushare per-call row cap forces index_weight to be paged in <=90-day windows. _INDEX_WINDOW_DAYS = 90 @@ -225,6 +263,75 @@ def _parse_stock_basic(raw: pd.DataFrame | None) -> pd.DataFrame: return df[_STOCK_BASIC_COLUMNS] +def _parse_daily_basic(raw: pd.DataFrame | None) -> pd.DataFrame: + """tushare ``daily_basic`` frame -> canonical-raw rows (or empty). + + Stores pe / pb / total_mv RAW (published same-day, PIT-safe by construction); + the value-ratio inversion and log-market-cap stay downstream, unchanged. + """ + if raw is None or len(raw) == 0: + return pd.DataFrame(columns=_DAILY_BASIC_COLUMNS) + df = raw.rename(columns={"ts_code": "symbol", "trade_date": "date"}).copy() + df["date"] = pd.to_datetime(df["date"].astype(str), format="%Y%m%d") + df["symbol"] = df["symbol"].astype(str) + for col in ("pe", "pb", "total_mv"): + if col not in df.columns: + df[col] = float("nan") + return df[_DAILY_BASIC_COLUMNS] + + +def _parse_fina(raw: pd.DataFrame | None) -> pd.DataFrame: + """tushare ``fina_indicator`` frame -> canonical-raw SUPERSET rows (or empty). + + The stored ``date`` is the report-period ``end_date`` (the coverage axis); + ``ann_date`` (disclosure) and ``end_date`` are kept RAW so the downstream + ``ann_date <= trade_date`` as-of alignment is byte-identical to the direct + path. ALL of :data:`FINA_FIELDS` are stored (missing -> NaN) so the schema is + field-set independent. + """ + if raw is None or len(raw) == 0: + return pd.DataFrame(columns=_FINA_COLUMNS) + df = raw.rename(columns={"ts_code": "symbol"}).copy() + df["symbol"] = df["symbol"].astype(str) + df["ann_date"] = df["ann_date"].astype(str) if "ann_date" in df.columns else None + df["end_date"] = df["end_date"].astype(str) if "end_date" in df.columns else None + df["date"] = pd.to_datetime(df["end_date"], format="%Y%m%d", errors="coerce") + for f in FINA_FIELDS: + if f not in df.columns: + df[f] = float("nan") + return df[_FINA_COLUMNS] + + +def _parse_index_member(raw: pd.DataFrame | None, symbol: str) -> pd.DataFrame: + """tushare ``index_member_all`` frame -> canonical-raw SW interval rows (or empty). + + Stores the L1/L2/L3 names + ``in_date``/``out_date`` (out NaT for an active + membership). The level-name selection and the as-of interval lookup stay + downstream (the feed builds {symbol: [(name, in, out)]}, unchanged). + """ + if raw is None or len(raw) == 0: + return pd.DataFrame(columns=_INDEX_MEMBER_COLUMNS) + df = raw.copy() + df["symbol"] = ( + df["ts_code"].astype(str) if "ts_code" in df.columns else str(symbol) + ) + for col in ("l1_name", "l2_name", "l3_name"): + if col not in df.columns: + df[col] = None + else: + df[col] = df[col].astype(object) + df["in_date"] = pd.to_datetime( + df["in_date"].astype(str), format="%Y%m%d", errors="coerce" + ) + if "out_date" in df.columns: + df["out_date"] = pd.to_datetime( + df["out_date"].astype(str), format="%Y%m%d", errors="coerce" + ) + else: + df["out_date"] = pd.NaT + return df[_INDEX_MEMBER_COLUMNS] + + class TushareCache: """Endpoint-level read-through cache (market bars + universe/tradability).""" @@ -239,6 +346,8 @@ def __init__( today: pd.Timestamp | None = None, clock: Callable[[], pd.Timestamp] | None = None, source_version: str | None = None, + not_ready_days: int = 0, + recent_tail_overrides: dict[str, int] | None = None, ) -> None: self._store = store self._ledger = ledger @@ -248,17 +357,43 @@ def __init__( self._today = pd.Timestamp(today).normalize() if today is not None else None self._clock = clock or pd.Timestamp.now self._source_version = source_version + # P4-3: how many trailing calendar days (ending today) are treated as + # "may not be published yet". An EMPTY return inside this pending window is + # recorded as ``not_ready`` (NOT coverage) so it is retried on a later run + # — a 21:00 updater must not freeze today's unpublished row as covered. + # 0 (default) disables this, keeping every existing endpoint's behaviour + # byte-identical. + self._not_ready_days = max(0, int(not_ready_days)) + # P4-3: per-endpoint override of the trailing-refetch window over the + # 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 {}) # 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 # the run log. Seeded with every known endpoint so a warm run reports an # explicit 0 even for endpoints it never had to touch. self.fetch_counts: dict[str, int] = {ep: 0 for ep in ALL_ENDPOINTS} + # P4-3 summary counters (for the data-update summary): rows upserted and + # not-ready (pending) records, per endpoint. + self.written_counts: dict[str, int] = {ep: 0 for ep in ALL_ENDPOINTS} + self.not_ready_counts: dict[str, int] = {ep: 0 for ep in ALL_ENDPOINTS} def stats(self) -> dict[str, int]: """Endpoint -> number of gap fetches sent to the API this instance.""" return dict(self.fetch_counts) + def update_summary(self) -> dict[str, dict[str, int]]: + """Per-endpoint {requests, rows_written, not_ready} for the data updater.""" + return { + ep: { + "requests": int(self.fetch_counts.get(ep, 0)), + "rows_written": int(self.written_counts.get(ep, 0)), + "not_ready": int(self.not_ready_counts.get(ep, 0)), + } + for ep in ALL_ENDPOINTS + } + # -- dense per-symbol date-range endpoints ----------------------------- # def daily_bars( self, symbols: list[str], start: str, end: str, fetch: FetchOne @@ -294,6 +429,77 @@ def stk_limit( _STK_LIMIT_COLUMNS, _STK_LIMIT_KEY, ) + # -- P4-3 factor-support endpoints ------------------------------------- # + def daily_basic( + self, symbols: list[str], start: str, end: str, fetch: FetchOne + ) -> pd.DataFrame: + """Canonical-raw daily_basic rows [date, symbol, pe, pb, total_mv]. + + Dense per-symbol date-range over the trade_date, exactly like the market + bars (recent-tail refetch + not-ready pending window apply). The feed's + ``market_cap`` / ``value_ratios`` select their columns from this one cache. + """ + return self._read_through( + DAILY_BASIC, symbols, start, end, fetch, _parse_daily_basic, + _DAILY_BASIC_COLUMNS, _DAILY_BASIC_KEY, + ) + + def fina_indicator( + self, + symbols: list[str], + start: str, + end: str, + fetch: FetchOne, + ) -> pd.DataFrame: + """Canonical-raw fina_indicator rows [symbol, ann_date, end_date, *FINA_FIELDS]. + + ALWAYS the SUPERSET (``FINA_FIELDS``) — the ``fetch`` closure MUST request + every superset field, so a subset warm never blocks a later different-subset + request (the caller selects its subset on read). Coverage is planned over + the REPORT-PERIOD (``end_date``) range; to catch LATE disclosures of recent + periods, a long trailing window is always refetched + (``recent_tail_overrides[fina_indicator]``). ``ann_date`` is stored raw for + the downstream as-of and is NEVER the coverage axis. + """ + out = self._read_through( + FINA_INDICATOR, symbols, start, end, fetch, _parse_fina, + _FINA_COLUMNS, _FINA_KEY, + ) + keep = ["symbol", "ann_date", "end_date", *FINA_FIELDS] + present = [c for c in keep if c in out.columns] + return out[present].reset_index(drop=True) + + def index_member_all( + self, symbols: list[str], fetch: FetchSnapshot + ) -> pd.DataFrame: + """Canonical-raw SW interval rows for ``symbols`` (per-symbol dimension). + + Each symbol's membership history is refetched only when never fetched, + stale beyond ``refresh_dimension_days``, or force-refreshed (slow-moving, + like namechange). Returns the stored rows for the requested symbols; the + level-name selection + as-of interval lookup stay the feed's job. + """ + forced = INDEX_MEMBER_ALL in self._force_refresh + fields_hash = _fields_hash(_INDEX_MEMBER_COLUMNS) + out: list[pd.DataFrame] = [] + for sym in symbols: + if forced or self._snapshot_stale(INDEX_MEMBER_ALL, sym): + self._fetch_snapshot( + INDEX_MEMBER_ALL, sym, "symbol", lambda s=sym: fetch(s), + lambda raw, s=sym: _parse_index_member(raw, s), + _INDEX_MEMBER_KEY, fields_hash, + ) + cached = self._store.read_symbol(INDEX_MEMBER_ALL, sym) + if not cached.empty: + out.append(cached[_INDEX_MEMBER_COLUMNS]) + _LOGGER.info( + "cache %s: %d symbols (api calls=%d)", + INDEX_MEMBER_ALL, len(symbols), self.fetch_counts[INDEX_MEMBER_ALL], + ) + if not out: + return pd.DataFrame(columns=_INDEX_MEMBER_COLUMNS) + return pd.concat(out, ignore_index=True) + # -- index-keyed date-range endpoint (index_weight, 90-day paged) ------- # def index_weight( self, index_code: str, start: str, end: str, fetch: FetchOne @@ -422,16 +628,31 @@ def _read_through( return pd.concat(out, ignore_index=True).reset_index(drop=True) def _gaps_for(self, endpoint, symbol, req_start, req_end, forced): - """The uncovered sub-intervals to fetch (+ a forced recent tail).""" + """The uncovered sub-intervals to fetch (+ a forced trailing tail).""" if forced: return [(req_start, req_end)] covered = self._ledger.covered_intervals(endpoint, symbol) gaps = subtract_intervals(req_start, req_end, covered) - recent = self._recent_tail(req_start, req_end) - if recent is not None: - gaps = merge_intervals(gaps + [recent]) + tail = self._tail_for(endpoint, req_start, req_end) + if tail is not None: + gaps = merge_intervals(gaps + [tail]) return gaps + def _tail_for(self, endpoint, req_start, req_end): + """The trailing window always refetched for ``endpoint``. + + A per-endpoint override (``recent_tail_overrides``) refetches the trailing + N days of the REQUESTED range — used by fina_indicator to catch LATE + disclosures of recent report periods, independent of today. Otherwise the + today-based recent tail (``refresh_recent_days``) applies. + """ + override = self._recent_tail_overrides.get(endpoint) + if override is not None: + if override <= 0: + return None + return (max(req_start, req_end - pd.Timedelta(days=override - 1)), req_end) + return self._recent_tail(req_start, req_end) + def _recent_tail(self, req_start, req_end): """Force-refetch the recent tail within ``refresh_recent_days`` of today.""" if self._refresh_recent_days <= 0: @@ -442,25 +663,75 @@ def _recent_tail(self, req_start, req_end): return None # whole request is safely historical return (max(req_start, threshold), req_end) + def _pending_start(self): + """First calendar day treated as 'may not be published yet' (or None).""" + if self._not_ready_days <= 0: + return None + today = self._today if self._today is not None else self._clock().normalize() + return today - pd.Timedelta(days=self._not_ready_days - 1) + def _fetch_gap( self, endpoint, symbol, gap_start, gap_end, fetch, parse, key_cols, fields_hash ): - """Fetch one gap, upsert raw rows, record coverage (incl. empty).""" + """Fetch one gap, upsert raw rows, record coverage (incl. empty/not-ready).""" raw = fetch(symbol, _compact(gap_start), _compact(gap_end)) self.fetch_counts[endpoint] = self.fetch_counts.get(endpoint, 0) + 1 parsed = parse(raw) - row_count = len(parsed) - if row_count: + 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 + ) + + def _record_gap_coverage( + self, endpoint, symbol, gap_start, gap_end, parsed, fields_hash + ): + """Record coverage for a fetched gap, carving out a not-ready pending tail. + + With ``not_ready_days == 0`` (default) this records ONE row for the whole + gap (ok if any row else empty) — byte-identical to the historical + behaviour. Otherwise the gap is split at the pending boundary: the + historical part records ok/empty; the pending part records ``not_ready`` + when it returned no row (so a 21:00 updater never freezes today's + unpublished data as covered). + """ + pending_start = self._pending_start() + if pending_start is None or gap_end < pending_start: + self._record_one(endpoint, symbol, gap_start, gap_end, parsed, + "empty", fields_hash) + return + has_date = "date" in parsed.columns + hist_end = pending_start - pd.Timedelta(days=1) + if gap_start <= hist_end: + hist = parsed[parsed["date"] <= hist_end] if has_date else parsed.iloc[0:0] + self._record_one(endpoint, symbol, gap_start, hist_end, hist, + "empty", fields_hash) + pend = parsed[parsed["date"] >= pending_start] if has_date else parsed + self._record_one(endpoint, symbol, pending_start, gap_end, pend, + "not_ready", fields_hash) + + def _record_one( + self, endpoint, symbol, start, end, rows, empty_status, fields_hash + ): + """Append one coverage row; ``empty_status`` is used when ``rows`` is empty.""" + n = len(rows) + status = "ok" if n else empty_status + if status == "not_ready": + self.not_ready_counts[endpoint] = ( + self.not_ready_counts.get(endpoint, 0) + 1 + ) self._ledger.record( endpoint=endpoint, key_type="symbol", key=symbol, - start_date=gap_start, - end_date=gap_end, + start_date=start, + end_date=end, fields_hash=fields_hash, - row_count=row_count, - status="ok" if row_count else "empty", + row_count=n, + status=status, fetched_at=self._clock(), source_version=self._source_version, ) @@ -526,6 +797,9 @@ def _fetch_snapshot( row_count = len(parsed) if row_count: self._store.upsert_symbol(endpoint, key, parsed, key_cols) + self.written_counts[endpoint] = ( + self.written_counts.get(endpoint, 0) + row_count + ) self._ledger.record( endpoint=endpoint, key_type=key_type, diff --git a/data/feed/tushare_covariates.py b/data/feed/tushare_covariates.py index 1bc9abc..7fa7487 100644 --- a/data/feed/tushare_covariates.py +++ b/data/feed/tushare_covariates.py @@ -48,9 +48,9 @@ def __init__( self._rate_limit = rate_limit self._max_retries = max(1, int(max_retries)) self._pro = None - # P4-2: optional shared read-through cache. Only ``listing_dates`` - # (stock_basic.list_date, for min_listing_days) reads through it; the - # daily_basic / index_member_all covariates are P4-3 and stay direct. + # Optional shared read-through cache: ``listing_dates`` (stock_basic, P4-2), + # ``market_cap`` / ``value_ratios`` (daily_basic, P4-3) and + # ``pit_sw_intervals`` (index_member_all, P4-3) all read through it. # None keeps the historical direct fetch EXACTLY. self._cache = cache @@ -141,6 +141,9 @@ def pit_sw_intervals( raise ValueError( f"industry level must be one of {list(self._SW_LEVEL_COLUMN)}; got {level!r}." ) + if self._cache is not None: + frame = self._cache.index_member_all(symbols, self._index_member_fetch()) + return self._intervals_from_member(frame, col) pro = self._client() out: dict[str, list[tuple]] = {} for sym in symbols: @@ -161,6 +164,39 @@ def pit_sw_intervals( out[str(sym)] = rows return out + def _index_member_fetch(self): + """`(symbol) -> raw index_member_all frame` (per-symbol dimension).""" + + def fetch(symbol): + return self._call(self._client().index_member_all, ts_code=symbol) + + return fetch + + def _intervals_from_member( + self, frame: pd.DataFrame, col: str + ) -> dict[str, list[tuple]]: + """Build {symbol: [(name, in_date, out_date|None)]} from cached raw rows. + + Same shape as the direct path: ``out_date`` NaT -> None, a symbol with no + usable row is simply absent. The cached rows already carry datetime + in_date/out_date, so the per-symbol interval set is identical to a direct + per-symbol fetch. + """ + out: dict[str, list[tuple]] = {} + if frame is None or frame.empty or col not in frame.columns: + return out + for sym, sub in frame.groupby(frame["symbol"].astype(str), sort=False): + rows: list[tuple] = [] + for r in sub.itertuples(): + name = getattr(r, col) + in_d = pd.Timestamp(r.in_date) if pd.notna(r.in_date) else None + out_raw = getattr(r, "out_date", None) + out_d = pd.Timestamp(out_raw) if pd.notna(out_raw) else None + rows.append((name, in_d, out_d)) + if rows: + out[str(sym)] = rows + return out + def value_ratios(self, symbols: list[str], start: str, end: str) -> pd.DataFrame: """Return DataFrame[date, symbol, pe, pb] from daily_basic (P3-5). @@ -168,6 +204,13 @@ def value_ratios(self, symbols: list[str], start: str, end: str) -> pd.DataFrame inversion to value_ep / value_bp (with non-positive guards) happens in the pipeline's value enrichment, not here. """ + if self._cache is not None: + df = self._cache.daily_basic(symbols, start, end, self._daily_basic_fetch()) + if df.empty: + return self._empty_value_ratios() + df = df.copy() + df["symbol"] = df["symbol"].astype(str) + return df[["date", "symbol", "pe", "pb"]].reset_index(drop=True) pro = self._client() s = pd.Timestamp(start).strftime("%Y%m%d") e = pd.Timestamp(end).strftime("%Y%m%d") @@ -183,12 +226,7 @@ def value_ratios(self, symbols: list[str], start: str, end: str) -> pd.DataFrame if df is not None and len(df) > 0: frames.append(df) if not frames: - return pd.DataFrame( - {"date": pd.Series([], dtype="datetime64[ns]"), - "symbol": pd.Series([], dtype=object), - "pe": pd.Series([], dtype=float), - "pb": pd.Series([], dtype=float)} - ) + return self._empty_value_ratios() out = pd.concat(frames, ignore_index=True).rename(columns={"ts_code": "symbol"}) out["date"] = pd.to_datetime(out["trade_date"].astype(str), format="%Y%m%d") out["symbol"] = out["symbol"].astype(str) @@ -196,6 +234,13 @@ def value_ratios(self, symbols: list[str], start: str, end: str) -> pd.DataFrame def market_cap(self, symbols: list[str], start: str, end: str) -> pd.DataFrame: """Return DataFrame[date, symbol, market_cap] from daily_basic.total_mv.""" + if self._cache is not None: + df = self._cache.daily_basic(symbols, start, end, self._daily_basic_fetch()) + if df.empty: + return self._empty_market_cap() + df = df.rename(columns={"total_mv": "market_cap"}).copy() + df["symbol"] = df["symbol"].astype(str) + return df[["date", "symbol", "market_cap"]].reset_index(drop=True) pro = self._client() s = pd.Timestamp(start).strftime("%Y%m%d") e = pd.Timestamp(end).strftime("%Y%m%d") @@ -211,14 +256,43 @@ def market_cap(self, symbols: list[str], start: str, end: str) -> pd.DataFrame: if df is not None and len(df) > 0: frames.append(df) if not frames: - return pd.DataFrame( - {"date": pd.Series([], dtype="datetime64[ns]"), - "symbol": pd.Series([], dtype=object), - "market_cap": pd.Series([], dtype=float)} - ) + return self._empty_market_cap() out = pd.concat(frames, ignore_index=True).rename( columns={"ts_code": "symbol", "total_mv": "market_cap"} ) out["date"] = pd.to_datetime(out["trade_date"].astype(str), format="%Y%m%d") out["symbol"] = out["symbol"].astype(str) return out[["date", "symbol", "market_cap"]] + + def _daily_basic_fetch(self): + """`(symbol, s_compact, e_compact) -> raw daily_basic frame` (pe/pb/total_mv). + + One cached daily_basic call serves BOTH value_ratios (pe/pb) and + market_cap (total_mv); the client is built lazily inside the closure. + """ + + def fetch(symbol, start_compact, end_compact): + return self._call( + self._client().daily_basic, ts_code=symbol, + start_date=start_compact, end_date=end_compact, + fields="ts_code,trade_date,pe,pb,total_mv", + ) + + return fetch + + @staticmethod + def _empty_value_ratios() -> pd.DataFrame: + return pd.DataFrame( + {"date": pd.Series([], dtype="datetime64[ns]"), + "symbol": pd.Series([], dtype=object), + "pe": pd.Series([], dtype=float), + "pb": pd.Series([], dtype=float)} + ) + + @staticmethod + def _empty_market_cap() -> pd.DataFrame: + return pd.DataFrame( + {"date": pd.Series([], dtype="datetime64[ns]"), + "symbol": pd.Series([], dtype=object), + "market_cap": pd.Series([], dtype=float)} + ) diff --git a/data/feed/tushare_fina.py b/data/feed/tushare_fina.py index 333cabd..ff78da2 100644 --- a/data/feed/tushare_fina.py +++ b/data/feed/tushare_fina.py @@ -14,10 +14,13 @@ import pandas as pd +from data.cache.tushare_cache import FINA_FIELDS from data.feed.secret import read_token from data.feed.throttle import request_with_retry -# financial fields supported as P1 factors. +# financial fields requested by default when a caller names none. The cache +# stores the full ``FINA_FIELDS`` superset regardless (so a subset warm never +# blocks a later different-subset request); this is only the default REQUEST set. DEFAULT_FIELDS: tuple[str, ...] = ("roe", "netprofit_yoy") @@ -30,12 +33,18 @@ def __init__( token_key: str = "tushare.token", rate_limit: int | None = None, max_retries: int = 6, + cache=None, ) -> None: self._secret_file = str(secret_file) self._token_key = token_key self._rate_limit = rate_limit self._max_retries = max(1, int(max_retries)) self._pro = None + # P4-3: optional shared read-through cache. None keeps the historical + # direct per-symbol fetch EXACTLY; the cache stores RAW fina_indicator + # rows (with ann_date) only — the ann_date<=trade_date as-of alignment + # stays downstream, byte-identical. + self._cache = cache def _client(self): if self._pro is None: @@ -44,6 +53,11 @@ def _client(self): self._pro = ts.pro_api(read_token(self._secret_file, self._token_key)) return self._pro + def _call(self, fn, **kwargs): + return request_with_retry( + fn, max_retries=self._max_retries, rate_limit=self._rate_limit, **kwargs + ) + def get_fina_indicator( self, symbols: list[str], @@ -66,19 +80,30 @@ def get_fina_indicator( """ wanted = list(fields) if fields else list(DEFAULT_FIELDS) col_spec = ",".join(["ts_code", "ann_date", "end_date", *wanted]) + + if self._cache is not None: + # The cache stores the CANONICAL SUPERSET (field-set independent), so a + # warm for one config's fields never blocks another's. Fetch the + # superset; select the requested subset on read. + super_spec = ",".join(["ts_code", "ann_date", "end_date", *FINA_FIELDS]) + cached = self._cache.fina_indicator( + symbols, start, end, self._fina_fetch(super_spec) + ) + if cached.empty: + return self._empty(wanted) + cached = cached.copy() + cached["symbol"] = cached["symbol"].astype(str) + keep = ["symbol", "ann_date", "end_date", *wanted] + return cached[[c for c in keep if c in cached.columns]].reset_index(drop=True) + pro = self._client() s = pd.Timestamp(start).strftime("%Y%m%d") e = pd.Timestamp(end).strftime("%Y%m%d") frames: list[pd.DataFrame] = [] for sym in symbols: - df = request_with_retry( - pro.fina_indicator, - max_retries=self._max_retries, - rate_limit=self._rate_limit, - ts_code=sym, - start_date=s, - end_date=e, + df = self._call( + pro.fina_indicator, ts_code=sym, start_date=s, end_date=e, fields=col_spec, ) if df is not None and len(df) > 0: @@ -92,6 +117,22 @@ def get_fina_indicator( keep = ["symbol", "ann_date", "end_date", *wanted] return out[[c for c in keep if c in out.columns]] + def _fina_fetch(self, col_spec: str): + """`(symbol, s_compact, e_compact) -> raw fina_indicator frame` (period range). + + The client is built lazily inside the closure, so a fully-covered warm run + reads no token and constructs no client. ``start_date``/``end_date`` filter + by the REPORT PERIOD (end_date), matching the direct path exactly. + """ + + def fetch(symbol, start_compact, end_compact): + return self._call( + self._client().fina_indicator, ts_code=symbol, + start_date=start_compact, end_date=end_compact, fields=col_spec, + ) + + return fetch + @staticmethod def _empty(fields: list[str]) -> pd.DataFrame: cols = {"symbol": pd.Series([], dtype=object), diff --git a/docs/data/tushare_permissions.md b/docs/data/tushare_permissions.md index c8e6dc5..7d79cfa 100644 --- a/docs/data/tushare_permissions.md +++ b/docs/data/tushare_permissions.md @@ -46,9 +46,9 @@ | `namechange` | P2-2 ST 区间 | ts | authorized (8) | ✅ | **已证** | dimension（in/out 区间）| P4-2 已缓存 | | `stock_basic` | P2-2 list_date（min_listing_days）| ts | authorized (1) | ✅ | **已证** | dimension（全局快照）| P4-2 已缓存 | | `stk_limit` | P2-2 **raw** 涨跌停价 | ts+d | authorized (1) | ✅ | **已证** | T 日 **raw 价**（front-adjust 前比对）| P4-2 已缓存 | -| `daily_basic` | P2-3/P3-5 pe/pb/total_mv | ts+d | authorized (1) | ✅ | **已证** | **当日发布，same-day PIT-safe** | P4-3 待缓存 | -| `fina_indicator` | P3-1 roe/np_yoy（ann_date as-of）| ts+period | authorized (1) | ✅ | **已证** | **按 ann_date 披露后才可用** | 单次最多 100 行；P4-3 待缓存 | -| `index_member_all` | P2-3 PIT SW 行业 | ts | authorized (1) | ✅ | **已证** | in/out 区间 as-of | P4-3 待缓存 | +| `daily_basic` | P2-3/P3-5 pe/pb/total_mv | ts+d | authorized (1) | ✅ | **已证** | **当日发布，same-day PIT-safe** | P4-3 已缓存 | +| `fina_indicator` | P3-1 roe/np_yoy（ann_date as-of）| ts+period | authorized (1) | ✅ | **已证** | **按 ann_date 披露后才可用** | 单次最多 100 行；P4-3 已缓存 | +| `index_member_all` | P2-3 PIT SW 行业 | ts | authorized (1) | ✅ | **已证** | in/out 区间 as-of | P4-3 已缓存 | | `income` | 财务源（ann_date）| ts+period | authorized (1) | ✅ | 同 fina | 按 ann_date 披露 | 目前未进因子，留作财务源 | ### EXPLORATORY 候选源（权限有，但**批量额度未核实**——接入前各自限频体检） diff --git a/qt/cli.py b/qt/cli.py index 8a1f170..5b2aace 100644 --- a/qt/cli.py +++ b/qt/cli.py @@ -156,6 +156,23 @@ def _cmd_run_phase3_subset(args: argparse.Namespace) -> int: return 0 +def _cmd_data_update(args: argparse.Namespace) -> int: + """Warm/update the tushare caches (P4-3); never runs a backtest.""" + from qt.data_updater import format_summary, run_data_update + + try: + result = run_data_update(args.config) + except (ConfigError, ValueError, FileNotFoundError) as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 1 + print( + f"OK data-update: {len(result.endpoints)} endpoints, " + f"{len(result.symbols)} symbols ({result.elapsed_seconds:.1f}s)\n" + f"{format_summary(result)}" + ) + return 0 + + def _cmd_fetch_data(args: argparse.Namespace) -> int: """Stage helper: run the spine and report the data-fetch stage.""" return _run_pipeline_cmd(args.config, "fetch-data") @@ -215,6 +232,13 @@ def build_parser() -> argparse.ArgumentParser: p_sub.add_argument("--config", required=True, help="Path to the YAML config.") p_sub.set_defaults(func=_cmd_run_phase3_subset) + p_du = sub.add_parser( + "data-update", + help="Warm/update the tushare raw caches (P4-3); no backtest.", + ) + p_du.add_argument("--config", required=True, help="Path to the YAML config.") + p_du.set_defaults(func=_cmd_data_update) + for name, func, help_text in ( ("fetch-data", _cmd_fetch_data, "Run the spine, report data fetch."), ("compute-factors", _cmd_compute_factors, "Run the spine, report factor compute."), diff --git a/qt/config.py b/qt/config.py index 10eecfc..464a01d 100644 --- a/qt/config.py +++ b/qt/config.py @@ -48,8 +48,9 @@ class CacheCfg(_Strict): downstream transforms (``front_adjust``, raw price-limit checks, PIT as-of membership / industry / financials) still run in memory, unchanged. - P4-2 caches: index_weight, suspend_d, namechange, stk_limit, stock_basic. - Not yet cached (P4-3): daily_basic, fina_indicator, index_member_all. + Cached: market bars (P4-1); index_weight / suspend_d / namechange / stk_limit + / stock_basic (P4-2); daily_basic / fina_indicator / index_member_all (P4-3). + The 21:00 ``data-update`` job warms these incrementally (see DataUpdateCfg). """ enabled: bool = False @@ -480,6 +481,64 @@ def _check_sections(self) -> "SubsetValidationCfg": return self +_DATA_UPDATE_ENDPOINTS = frozenset({ + "market_daily", "adj_factor", "index_weight", "suspend_d", "namechange", + "stk_limit", "stock_basic", "daily_basic", "fina_indicator", + "index_member_all", "stk_mins_1min", +}) + + +class DataUpdateCfg(_Strict): + """Standalone data-updater section (P4-3) — consumed ONLY by ``data-update``. + + Declares the daily 21:00 (Asia/Shanghai) incremental warm: which endpoints to + refresh, the lookback/tail policy, the not-ready pending window, and a + conservative rate limit. Scheduling is external (systemd timer / cron calls + the CLI); this is purely the job's parameters. Optional on RootConfig, so + every existing config still validates unchanged. + """ + + timezone: str = "Asia/Shanghai" + scheduled_start: str = "21:00:00" + endpoints: list[str] = Field(default_factory=list) + index_codes: list[str] = Field(default_factory=list) + lookback_days: int = 400 + tail_refresh_days: int = 14 + not_ready_days: int = 1 + fina_tail_days: int = 400 + fina_fields: list[str] = Field(default_factory=lambda: ["roe", "netprofit_yoy"]) + rate_limit_per_min: int = 450 + force_refresh: list[str] = Field(default_factory=list) + + @field_validator("endpoints", "force_refresh") + @classmethod + def _check_endpoints(cls, v: list[str]) -> list[str]: + unknown = [e for e in v if e not in _DATA_UPDATE_ENDPOINTS] + if unknown: + raise ValueError( + f"data_update endpoint(s) {unknown} unknown; " + f"must be in {sorted(_DATA_UPDATE_ENDPOINTS)}." + ) + return v + + @field_validator( + "lookback_days", "tail_refresh_days", "not_ready_days", + "fina_tail_days", "rate_limit_per_min", + ) + @classmethod + def _check_non_negative(cls, v: int) -> int: + if v < 0: + raise ValueError(f"data_update integer fields must be >= 0; got {v}.") + return v + + @field_validator("rate_limit_per_min") + @classmethod + def _check_rate_positive(cls, v: int) -> int: + if v <= 0: + raise ValueError(f"data_update.rate_limit_per_min must be > 0; got {v}.") + return v + + class RootConfig(_Strict): """Top-level config composing every section. @@ -505,6 +564,8 @@ class RootConfig(_Strict): robustness: RobustnessCfg | None = None # P3-6 subset validation (consumed only by run-phase3-subset). subset_validation: SubsetValidationCfg | None = None + # P4-3 data updater (consumed only by data-update). + data_update: DataUpdateCfg | None = None @model_validator(mode="after") def _check_subset_groups_reference_enabled_factors(self) -> "RootConfig": diff --git a/qt/data_updater.py b/qt/data_updater.py new file mode 100644 index 0000000..c49f3ff --- /dev/null +++ b/qt/data_updater.py @@ -0,0 +1,247 @@ +"""Standalone Tushare cache updater (P4-3) — the 21:00 incremental warm. + +A SEPARATE entry point from the backtest pipeline: it only WARMS / UPDATES the +read-through caches (daily endpoint cache + the I2 intraday 1min cache). It never +computes factors, never builds an alpha/portfolio, never runs a backtest, and +never writes a ``PanelStore`` — the backtest still does its own read-through to +fill gaps. Scheduling is external (a systemd timer / cron fires the CLI at 21:00 +Asia/Shanghai); this module is just the job body + an explainable summary. + +Incremental semantics come from the cache layer: a fully-covered historical run +makes ~0 API calls; a new trading day fetches only the new dates or the recent +tail; a failed fetch records no coverage (retried later); an EMPTY return inside +the not-ready pending window (today, unpublished at 21:00) is recorded as +``not_ready`` — never frozen as permanent coverage. Daily and intraday coverage +use their own ledgers/stores (never mixed). No token / qfq / factor result is +ever stored. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field + +import pandas as pd + +from qt.config import RootConfig, load_config + +# the intraday window the 21:00 job warms (the bulk historical minute backfill is +# a separate manual run; the daily job only tops up the recent tail). +_INTRADAY_TAIL_DAYS = 7 + + +@dataclass +class UpdateFeeds: + """The feed objects the updater drives (all share the daily cache).""" + + market: object | None = None + index: object | None = None + flags: object | None = None + covariates: object | None = None + fina: object | None = None + intraday: object | None = None + + +@dataclass(frozen=True) +class UpdateResult: + """Outcome of a data-update run (immutable).""" + + window_start: pd.Timestamp + window_end: pd.Timestamp + symbols: list[str] + endpoints: list[str] + summary: dict[str, dict[str, int]] + elapsed_seconds: float = 0.0 + notes: list[str] = field(default_factory=list) + + +def update_endpoints( + cache, + feeds: UpdateFeeds, + symbols: list[str], + *, + start: str, + end: str, + endpoints: list[str], + index_codes: list[str], + fina_fields: list[str], + sw_level: str = "L1", + intraday_cache=None, + intraday_window: tuple[str, str] | None = None, +) -> dict[str, dict[str, int]]: + """Warm each requested endpoint through its feed; return the per-endpoint summary. + + Pure orchestration: every feed shares the read-through ``cache``, so only + uncovered ranges hit the API. Calls NOTHING but the cache-warming feed + methods — no factor / alpha / portfolio / backtest / PanelStore. + """ + eps = set(endpoints) + if ({"market_daily", "adj_factor"} & eps) and feeds.market is not None: + feeds.market.get_bars(symbols, start, end) + if "index_weight" in eps and feeds.index is not None: + for code in index_codes: + feeds.index.get_constituents(code, start, end) + if "suspend_d" in eps and feeds.flags is not None: + feeds.flags.suspended(symbols, start, end) + if "namechange" in eps and feeds.flags is not None: + feeds.flags.st_intervals(symbols) + if "stk_limit" in eps and feeds.flags is not None: + feeds.flags.limits(symbols, start, end) + if "stock_basic" in eps and feeds.covariates is not None: + feeds.covariates.listing_dates(symbols) + if "daily_basic" in eps and feeds.covariates is not None: + feeds.covariates.market_cap(symbols, start, end) + if "fina_indicator" in eps and feeds.fina is not None: + feeds.fina.get_fina_indicator(symbols, start, end, fields=fina_fields) + if "index_member_all" in eps and feeds.covariates is not None: + feeds.covariates.pit_sw_intervals(symbols, sw_level) + + summary = cache.update_summary() + if "stk_mins_1min" in eps and feeds.intraday is not None and intraday_window: + s, e = intraday_window + feeds.intraday.get_minutes(symbols, s, e) + st = intraday_cache.stats() if intraday_cache is not None else {} + summary["stk_mins_1min"] = { + "requests": int(st.get("stk_mins_1min", 0)), + "rows_written": 0, + "not_ready": 0, + } + return summary + + +def _resolve_today(cfg: RootConfig, today) -> pd.Timestamp: + if today is not None: + return pd.Timestamp(today).normalize() + return pd.Timestamp.now(tz=cfg.data_update.timezone).tz_localize(None).normalize() + + +def _resolve_symbols(cfg: RootConfig, feeds: UpdateFeeds, start: str, end: str) -> list[str]: + """Universe to warm: static symbols, or the union of index constituents.""" + if cfg.universe.type == "static": + return list(cfg.universe.symbols) + codes = cfg.data_update.index_codes or ( + [cfg.universe.index_code] if cfg.universe.index_code else [] + ) + syms: set[str] = set() + if feeds.index is not None: + for code in codes: + cons = feeds.index.get_constituents(code, start, end) + if not cons.empty: + syms.update(cons["symbol"].astype(str).tolist()) + return sorted(syms) + + +def _build_feeds(cfg: RootConfig, cache, intraday_cache, rate_limit: int) -> UpdateFeeds: + """Construct the real tushare feeds, all sharing the read-through caches.""" + from data.feed.index_feed import IndexConstituentsFeed + from data.feed.tushare_covariates import TushareCovariatesFeed + from data.feed.tushare_feed import TushareFeed + from data.feed.tushare_fina import TushareFinancialFeed + from data.feed.tushare_flags import TushareFlagsFeed + from data.feed.tushare_intraday import TushareIntradayFeed + + secret = cfg.data.external_secret_file + key = cfg.data.tushare_token_key + return UpdateFeeds( + market=TushareFeed(secret, token_key=key, rate_limit=rate_limit, cache=cache), + index=IndexConstituentsFeed(secret, token_key=key, cache=cache), + flags=TushareFlagsFeed(secret, token_key=key, rate_limit=rate_limit, cache=cache), + covariates=TushareCovariatesFeed( + secret, token_key=key, rate_limit=rate_limit, cache=cache + ), + fina=TushareFinancialFeed( + secret, token_key=key, rate_limit=rate_limit, cache=cache + ), + intraday=TushareIntradayFeed( + secret, token_key=key, rate_limit=rate_limit, cache=intraday_cache + ), + ) + + +def run_data_update(config_path: str, *, today=None) -> UpdateResult: + """Run the 21:00 incremental cache warm from ``config_path``. + + Requires ``data.source == 'tushare'``, an external secret file, and + ``data.cache.enabled``. Builds the daily + intraday read-through caches with + the not-ready pending window and the fina late-disclosure tail, warms each + configured endpoint, and returns the per-endpoint summary. Never runs a + backtest / writes a PanelStore. + """ + t0 = time.perf_counter() + cfg = load_config(config_path) + du = cfg.data_update + if du is None: + raise ValueError("data-update requires a 'data_update' config section.") + if cfg.data.source != "tushare": + raise ValueError("data-update requires data.source='tushare' (it pulls real data).") + if not cfg.data.external_secret_file: + raise ValueError("data-update requires data.external_secret_file (token).") + if not cfg.data.cache.enabled: + raise ValueError("data-update requires data.cache.enabled=true.") + + today_ts = _resolve_today(cfg, today) + end = today_ts.strftime("%Y-%m-%d") + start = (today_ts - pd.Timedelta(days=du.lookback_days)).strftime("%Y-%m-%d") + + from data.cache import ( + CacheParquetStore, + CoverageLedger, + IntradayCoverageLedger, + IntradayParquetStore, + TushareCache, + TushareIntradayCache, + ) + + root = cfg.data.cache.root_dir + cache = TushareCache( + CacheParquetStore(root), + CoverageLedger(root), + refresh_recent_days=du.tail_refresh_days, + refresh_dimension_days=cfg.data.cache.refresh_dimension_days, + force_refresh=tuple(du.force_refresh), + today=today_ts, + not_ready_days=du.not_ready_days, + recent_tail_overrides={"fina_indicator": du.fina_tail_days}, + ) + intraday_cache = TushareIntradayCache( + IntradayParquetStore(root), IntradayCoverageLedger(root) + ) + feeds = _build_feeds(cfg, cache, intraday_cache, du.rate_limit_per_min) + + symbols = _resolve_symbols(cfg, feeds, start, end) + intraday_window = ( + (today_ts - pd.Timedelta(days=_INTRADAY_TAIL_DAYS)).strftime("%Y-%m-%d 00:00:00"), + today_ts.strftime("%Y-%m-%d 23:59:59"), + ) + summary = update_endpoints( + cache, feeds, symbols, + start=start, end=end, + endpoints=du.endpoints, index_codes=du.index_codes, + fina_fields=du.fina_fields, + sw_level=cfg.processing.neutralize.industry_level, + intraday_cache=intraday_cache, intraday_window=intraday_window, + ) + return UpdateResult( + window_start=pd.Timestamp(start), + window_end=pd.Timestamp(end), + symbols=symbols, + endpoints=list(du.endpoints), + summary=summary, + elapsed_seconds=time.perf_counter() - t0, + ) + + +def format_summary(result: UpdateResult) -> str: + """One human line per endpoint: requests / rows_written / not_ready.""" + lines = [ + f"data-update window [{result.window_start.date()} .. " + f"{result.window_end.date()}], {len(result.symbols)} symbols" + ] + for ep in result.endpoints: + s = result.summary.get(ep, {}) + lines.append( + f" {ep}: requests={s.get('requests', 0)} " + f"rows_written={s.get('rows_written', 0)} " + f"not_ready={s.get('not_ready', 0)}" + ) + return "\n".join(lines) diff --git a/qt/pipeline.py b/qt/pipeline.py index 7ea8318..11b4541 100644 --- a/qt/pipeline.py +++ b/qt/pipeline.py @@ -440,9 +440,9 @@ def run_phase0(config_path: str) -> Phase0Result: panel = _load_panel(cfg, symbols, logger, cache) factors = _build_factors(cfg) primary = factors[0] - panel = _maybe_enrich_financials(cfg, panel, symbols, factors, logger) - panel = _maybe_enrich_value(cfg, panel, symbols, factors, logger) - panel = _maybe_enrich_covariates(cfg, panel, symbols, logger) + panel = _maybe_enrich_financials(cfg, panel, symbols, factors, logger, cache) + panel = _maybe_enrich_value(cfg, panel, symbols, factors, logger, cache) + panel = _maybe_enrich_covariates(cfg, panel, symbols, logger, cache) panel = _maybe_enrich_listing(cfg, panel, symbols, logger, cache) # P4-1/P4-2: one concise cache-stats line after every cached endpoint has run # (market bars + universe/tradability). A warm historical rerun shows all 0s. @@ -640,6 +640,7 @@ def _build_universe( _CACHED_ENDPOINTS: tuple[str, ...] = ( "market_daily", "adj_factor", "index_weight", "suspend_d", "namechange", "stk_limit", "stock_basic", + "daily_basic", "fina_indicator", "index_member_all", ) @@ -826,6 +827,7 @@ def _maybe_enrich_financials( symbols: list[str], factors: list, logger: logging.Logger, + cache=None, ) -> pd.DataFrame: """Attach ann_date-aligned columns for ALL financial factors (single fetch). @@ -844,7 +846,9 @@ def _maybe_enrich_financials( f"data (data.source='tushare'); they cannot run on demo data." ) feed = TushareFinancialFeed( - cfg.data.external_secret_file, token_key=cfg.data.tushare_token_key + cfg.data.external_secret_file, + token_key=cfg.data.tushare_token_key, + cache=cache, ) # Look back before start so the prior already-disclosed report is fetched and # can be as-of carried forward onto the early trade dates (no NaN gap). @@ -869,6 +873,7 @@ def _maybe_enrich_value( symbols: list[str], factors: list, logger: logging.Logger, + cache=None, ) -> pd.DataFrame: """Attach value_ep / value_bp columns when value factors are enabled (P3-5). @@ -886,7 +891,9 @@ def _maybe_enrich_value( "they cannot run on demo data." ) feed = TushareCovariatesFeed( - cfg.data.external_secret_file, token_key=cfg.data.tushare_token_key + cfg.data.external_secret_file, + token_key=cfg.data.tushare_token_key, + cache=cache, ) ratios = feed.value_ratios(symbols, cfg.data.start, cfg.data.end) enriched = panel.copy() @@ -1000,7 +1007,8 @@ def _process_factors( def _maybe_enrich_covariates( - cfg: RootConfig, panel: pd.DataFrame, symbols: list[str], logger: logging.Logger + cfg: RootConfig, panel: pd.DataFrame, symbols: list[str], + logger: logging.Logger, cache=None, ) -> pd.DataFrame: """Attach industry + market_cap when neutralization is enabled (tushare only).""" if not cfg.processing.neutralize.enabled: @@ -1011,7 +1019,9 @@ def _maybe_enrich_covariates( "industry + market_cap need real data (demo has neither)." ) feed = TushareCovariatesFeed( - cfg.data.external_secret_file, token_key=cfg.data.tushare_token_key + cfg.data.external_secret_file, + token_key=cfg.data.tushare_token_key, + cache=cache, ) market_cap = feed.market_cap(symbols, cfg.data.start, cfg.data.end) panel = enrich_covariates(panel, market_cap=market_cap) diff --git a/tests/test_data_updater.py b/tests/test_data_updater.py new file mode 100644 index 0000000..250a81e --- /dev/null +++ b/tests/test_data_updater.py @@ -0,0 +1,222 @@ +"""P4-3 feed cached==direct + data-updater orchestration tests (network-free).""" + +from __future__ import annotations + +import pandas as pd + +from data.cache.coverage import CoverageLedger +from data.cache.parquet_store import CacheParquetStore +from data.cache.tushare_cache import DAILY_BASIC, TushareCache +from data.clean.pit_financials import asof_financials +from data.feed.tushare_covariates import TushareCovariatesFeed +from data.feed.tushare_fina import TushareFinancialFeed +from qt.data_updater import UpdateFeeds, update_endpoints + +_CLK = lambda: pd.Timestamp("2026-06-13 21:00:00") # noqa: E731 + + +class _FakePro: + """Window-aware fake tushare client for daily_basic / fina / index_member_all.""" + + def daily_basic(self, ts_code, start_date, end_date, fields=None): # noqa: ARG002 + cat = { + "000001.SZ": [("20240102", 10.0, 1.0, 1000.0), ("20240103", 11.0, 1.1, 1100.0)], + "000002.SZ": [("20240102", 20.0, 2.0, 2000.0)], + } + S, E = pd.Timestamp(start_date), pd.Timestamp(end_date) + rows = [r for r in cat.get(ts_code, []) if S <= pd.Timestamp(r[0]) <= E] + if not rows: + return pd.DataFrame() + return pd.DataFrame({ + "ts_code": [ts_code] * len(rows), + "trade_date": [r[0] for r in rows], + "pe": [r[1] for r in rows], "pb": [r[2] for r in rows], + "total_mv": [r[3] for r in rows], + }) + + def fina_indicator(self, ts_code, start_date, end_date, fields=None): # noqa: ARG002 + # returns the full superset (roe / netprofit_yoy / grossprofit_margin) + cat = {"000001.SZ": [("20230331", "20230420", 5.0, 10.0, 30.0), + ("20230630", "20230815", 6.0, 12.0, 31.0)]} + S, E = pd.Timestamp(start_date), pd.Timestamp(end_date) + rows = [r for r in cat.get(ts_code, []) if S <= pd.Timestamp(r[0]) <= E] + if not rows: + return pd.DataFrame() + return pd.DataFrame({ + "ts_code": [ts_code] * len(rows), + "end_date": [r[0] for r in rows], "ann_date": [r[1] for r in rows], + "roe": [r[2] for r in rows], "netprofit_yoy": [r[3] for r in rows], + "grossprofit_margin": [r[4] for r in rows], + }) + + def index_member_all(self, ts_code): + cat = {"000001.SZ": [("Banks", "B2", "B3", "20200101", None)], + "000002.SZ": [("RealEstate", "R2", "R3", "20200101", "20221231")]} + rows = cat.get(ts_code, []) + if not rows: + return pd.DataFrame() + return pd.DataFrame({ + "ts_code": [ts_code] * len(rows), + "l1_name": [r[0] for r in rows], "l2_name": [r[1] for r in rows], + "l3_name": [r[2] for r in rows], "in_date": [r[3] for r in rows], + "out_date": [r[4] for r in rows], + }) + + +def _covariates(tmp_path, monkeypatch, with_cache): + feed = TushareCovariatesFeed("x.json", cache=_mk_cache(tmp_path) if with_cache else None) + monkeypatch.setattr(feed, "_client", lambda: _FakePro()) + return feed + + +def _mk_cache(tmp_path): + root = str(tmp_path / "cache") + return TushareCache(CacheParquetStore(root), CoverageLedger(root), clock=_CLK, + refresh_recent_days=0) + + +_SYMS = ["000001.SZ", "000002.SZ"] + + +def _sorted(df): + return df.sort_values(list(df.columns)).reset_index(drop=True) + + +def test_market_cap_cached_equals_direct(tmp_path, monkeypatch): + direct = _covariates(tmp_path / "d", monkeypatch, with_cache=False) + cached = _covariates(tmp_path / "c", monkeypatch, with_cache=True) + a = direct.market_cap(_SYMS, "2024-01-01", "2024-01-31") + b = cached.market_cap(_SYMS, "2024-01-01", "2024-01-31") + pd.testing.assert_frame_equal(_sorted(a), _sorted(b)) + + +def test_value_ratios_cached_equals_direct(tmp_path, monkeypatch): + direct = _covariates(tmp_path / "d", monkeypatch, with_cache=False) + cached = _covariates(tmp_path / "c", monkeypatch, with_cache=True) + a = direct.value_ratios(_SYMS, "2024-01-01", "2024-01-31") + b = cached.value_ratios(_SYMS, "2024-01-01", "2024-01-31") + pd.testing.assert_frame_equal(_sorted(a), _sorted(b)) + + +def test_pit_sw_intervals_cached_equals_direct(tmp_path, monkeypatch): + direct = _covariates(tmp_path / "d", monkeypatch, with_cache=False) + cached = _covariates(tmp_path / "c", monkeypatch, with_cache=True) + a = direct.pit_sw_intervals(_SYMS, "L1") + b = cached.pit_sw_intervals(_SYMS, "L1") + assert set(a) == set(b) + for sym in a: + assert set(a[sym]) == set(b[sym]) # identical interval set per symbol + + +def test_fina_subset_warm_does_not_block_other_subset(tmp_path, monkeypatch): + # REGRESSION (codex acceptance blocker): a data-update / backtest that warms + # fina for ["roe"] must NOT poison a later config needing grossprofit_margin. + cache = _mk_cache(tmp_path) + f1 = TushareFinancialFeed("x.json", cache=cache) + monkeypatch.setattr(f1, "_client", lambda: _FakePro()) + # warm requesting only roe (e.g. the default data_update fields) + f1.get_fina_indicator(["000001.SZ"], "2023-01-01", "2023-12-31", ["roe"]) + # a later feed over the SAME cache requests grossprofit_margin + f2 = TushareFinancialFeed("x.json", cache=cache) + monkeypatch.setattr(f2, "_client", lambda: _FakePro()) + out = f2.get_fina_indicator( + ["000001.SZ"], "2023-01-01", "2023-12-31", ["grossprofit_margin"] + ) + assert "grossprofit_margin" in out.columns # no KeyError, real column + assert set(out["grossprofit_margin"]) == {30.0, 31.0} # real data, not NaN + + +def test_fina_asof_cached_equals_direct(tmp_path, monkeypatch): + fields = ["roe", "netprofit_yoy"] + direct = TushareFinancialFeed("x.json") + monkeypatch.setattr(direct, "_client", lambda: _FakePro()) + cached = TushareFinancialFeed("x.json", cache=_mk_cache(tmp_path)) + monkeypatch.setattr(cached, "_client", lambda: _FakePro()) + a = direct.get_fina_indicator(["000001.SZ"], "2023-01-01", "2023-12-31", fields) + b = cached.get_fina_indicator(["000001.SZ"], "2023-01-01", "2023-12-31", fields) + pd.testing.assert_frame_equal(_sorted(a), _sorted(b)) + # the as-of result is identical too + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp("2023-09-01"), "000001.SZ")], names=["date", "symbol"] + ) + pd.testing.assert_frame_equal( + asof_financials(idx, a, fields), asof_financials(idx, b, fields) + ) + + +# --------------------------------------------------------------------------- # +# updater orchestration: only warm methods are called, summary is produced +# --------------------------------------------------------------------------- # +class _RecFeed: + def __init__(self): + self.calls: list[str] = [] + + def _rec(self, name, value): + self.calls.append(name) + return value + + def get_bars(self, symbols, s, e): # noqa: ARG002 + return self._rec("get_bars", None) + + def get_constituents(self, code, s, e): # noqa: ARG002 + return self._rec(f"constituents:{code}", pd.DataFrame()) + + def suspended(self, symbols, s, e): # noqa: ARG002 + return self._rec("suspended", set()) + + def st_intervals(self, symbols): # noqa: ARG002 + return self._rec("st_intervals", {}) + + def limits(self, symbols, s, e): # noqa: ARG002 + return self._rec("limits", pd.DataFrame()) + + def listing_dates(self, symbols): # noqa: ARG002 + return self._rec("listing_dates", {}) + + def market_cap(self, symbols, s, e): # noqa: ARG002 + return self._rec("market_cap", pd.DataFrame()) + + def get_fina_indicator(self, symbols, s, e, fields=None): # noqa: ARG002 + return self._rec("fina", pd.DataFrame()) + + def pit_sw_intervals(self, symbols, level): # noqa: ARG002 + return self._rec("pit_sw", {}) + + def get_minutes(self, symbols, s, e): # noqa: ARG002 + return self._rec("get_minutes", pd.DataFrame()) + + +def test_update_endpoints_warms_only_configured(tmp_path): + root = str(tmp_path / "cache") + cache = TushareCache(CacheParquetStore(root), CoverageLedger(root), clock=_CLK) + rec = _RecFeed() + feeds = UpdateFeeds(market=rec, index=rec, flags=rec, covariates=rec, + fina=rec, intraday=rec) + + class _IC: + def stats(self): + return {"stk_mins_1min": 0} + + summary = update_endpoints( + cache, feeds, _SYMS, + start="2024-01-01", end="2024-01-31", + endpoints=["daily_basic", "fina_indicator", "index_member_all", "stk_mins_1min"], + index_codes=["000300.SH"], fina_fields=["roe"], + intraday_cache=_IC(), intraday_window=("2024-01-24 00:00:00", "2024-01-31 23:59:59"), + ) + # exactly the warm methods for the requested endpoints fired + assert rec.calls == ["market_cap", "fina", "pit_sw", "get_minutes"] + # summary has every daily endpoint (seeded) + the intraday entry + assert DAILY_BASIC in summary and "stk_mins_1min" in summary + + +def test_update_endpoints_skips_unconfigured(tmp_path): + root = str(tmp_path / "cache") + cache = TushareCache(CacheParquetStore(root), CoverageLedger(root), clock=_CLK) + rec = _RecFeed() + feeds = UpdateFeeds(market=rec, index=rec, flags=rec, covariates=rec, fina=rec) + update_endpoints( + cache, feeds, _SYMS, start="2024-01-01", end="2024-01-31", + endpoints=["market_daily", "adj_factor"], index_codes=[], fina_fields=["roe"], + ) + assert rec.calls == ["get_bars"] # one call warms both market endpoints diff --git a/tests/test_tushare_cache_p4_3.py b/tests/test_tushare_cache_p4_3.py new file mode 100644 index 0000000..7c68e2c --- /dev/null +++ b/tests/test_tushare_cache_p4_3.py @@ -0,0 +1,271 @@ +"""P4-3 cache endpoint tests: daily_basic / fina_indicator / index_member_all. + +Covers cold/warm/partial/failed read-through, the not-ready pending window, the +fina late-disclosure tail, and the per-endpoint update summary. Fake fetch +callables; network-free; no token. +""" + +from __future__ import annotations + +import pandas as pd +import pytest + +from data.cache.coverage import CoverageLedger +from data.cache.parquet_store import CacheParquetStore +from data.cache.tushare_cache import ( + DAILY_BASIC, + FINA_INDICATOR, + INDEX_MEMBER_ALL, + TushareCache, +) + +_CLK = lambda: pd.Timestamp("2026-06-13 21:00:00") # noqa: E731 + + +def _cache(tmp_path, **kw): + root = str(tmp_path / "cache") + return TushareCache(CacheParquetStore(root), CoverageLedger(root), clock=_CLK, **kw) + + +# --------------------------------------------------------------------------- # +# daily_basic — dense per-symbol date-range +# --------------------------------------------------------------------------- # +class _DBFetch: + def __init__(self, catalog): + self.catalog = catalog + self.calls: list[tuple] = [] + + def __call__(self, symbol, s, e): + self.calls.append((symbol, s, e)) + S, E = pd.Timestamp(s), pd.Timestamp(e) + rows = [r for r in self.catalog.get(symbol, []) if S <= pd.Timestamp(r[0]) <= E] + if not rows: + return pd.DataFrame() + return pd.DataFrame({ + "ts_code": [symbol] * len(rows), + "trade_date": [r[0].replace("-", "") for r in rows], + "pe": [r[1] for r in rows], + "pb": [r[2] for r in rows], + "total_mv": [r[3] for r in rows], + }) + + +_DB_CAT = {"000001.SZ": [ + ("2024-01-02", 10.0, 1.0, 1000.0), + ("2024-01-03", 11.0, 1.1, 1100.0), + ("2024-01-04", 12.0, 1.2, 1200.0), +]} + + +def test_daily_basic_cold_warm_partial(tmp_path): + cache = _cache(tmp_path, refresh_recent_days=0) + f = _DBFetch(_DB_CAT) + cold = cache.daily_basic(["000001.SZ"], "2024-01-02", "2024-01-03", f) + assert list(cold.columns) == ["date", "symbol", "pe", "pb", "total_mv"] + assert len(cold) == 2 + assert f.calls # cold hit the API + # warm identical -> zero calls + f.calls.clear() + warm = cache.daily_basic(["000001.SZ"], "2024-01-02", "2024-01-03", f) + assert f.calls == [] + pd.testing.assert_frame_equal( + cold.sort_values(["date", "symbol"]).reset_index(drop=True), + warm.sort_values(["date", "symbol"]).reset_index(drop=True), + ) + # partial: extend by one day -> only the new day fetched + f.calls.clear() + cache.daily_basic(["000001.SZ"], "2024-01-02", "2024-01-04", f) + assert len(f.calls) == 1 + assert f.calls[0][1] == "20240104" and f.calls[0][2] == "20240104" + + +def test_daily_basic_failed_fetch_records_no_coverage(tmp_path): + cache = _cache(tmp_path, refresh_recent_days=0) + + def boom(symbol, s, e): + raise ConnectionError("transient") + + with pytest.raises(ConnectionError): + cache.daily_basic(["000001.SZ"], "2024-01-02", "2024-01-03", boom) + assert cache._ledger.read().empty # failed fetch is not coverage + # retried successfully later + good = _DBFetch(_DB_CAT) + out = cache.daily_basic(["000001.SZ"], "2024-01-02", "2024-01-03", good) + assert len(out) == 2 and good.calls + + +def test_daily_basic_not_ready_today_is_retried(tmp_path): + # today = 2024-01-05; data published only through 01-04 at 21:00. + cat = {"000001.SZ": _DB_CAT["000001.SZ"]} # no 01-05 row yet + today = pd.Timestamp("2024-01-05") + cache = _cache(tmp_path, refresh_recent_days=0, not_ready_days=1, today=today) + f = _DBFetch(cat) + cache.daily_basic(["000001.SZ"], "2024-01-02", "2024-01-05", f) + led = cache._ledger.read() + # the pending day 01-05 returned nothing -> recorded not_ready (NOT coverage) + assert (led["status"] == "not_ready").any() + # a later run still sees 01-05 as a gap and refetches it + f.calls.clear() + cache.daily_basic(["000001.SZ"], "2024-01-02", "2024-01-05", f) + assert any(c[1] == "20240105" for c in f.calls) + + +def test_not_ready_disabled_covers_today(tmp_path): + # not_ready_days=0 (default): today's empty would be covered -> NOT retried. + today = pd.Timestamp("2024-01-05") + cache = _cache(tmp_path, refresh_recent_days=0, not_ready_days=0, today=today) + f = _DBFetch(_DB_CAT) + cache.daily_basic(["000001.SZ"], "2024-01-02", "2024-01-05", f) + f.calls.clear() + cache.daily_basic(["000001.SZ"], "2024-01-02", "2024-01-05", f) + assert f.calls == [] # whole range (incl 01-05) recorded as covered + + +# --------------------------------------------------------------------------- # +# fina_indicator — report-period range, keeps ann_date, late-disclosure tail +# --------------------------------------------------------------------------- # +class _FinaFetch: + """Fake fina fetch: returns the full superset (roe / netprofit_yoy / gpm).""" + + def __init__(self, catalog): + self.catalog = catalog + self.calls: list[tuple] = [] + + def __call__(self, symbol, s, e): + self.calls.append((symbol, s, e)) + S, E = pd.Timestamp(s), pd.Timestamp(e) + rows = [r for r in self.catalog.get(symbol, []) if S <= pd.Timestamp(r[0]) <= E] + if not rows: + return pd.DataFrame() + return pd.DataFrame({ + "ts_code": [symbol] * len(rows), + "end_date": [r[0].replace("-", "") for r in rows], + "ann_date": [r[1].replace("-", "") for r in rows], + "roe": [r[2] for r in rows], + "netprofit_yoy": [r[3] for r in rows], + "grossprofit_margin": [r[4] for r in rows], + }) + + +# (end_date, ann_date, roe, np_yoy, grossprofit_margin) +_FINA_CAT = {"000001.SZ": [ + ("2023-03-31", "2023-04-20", 5.0, 10.0, 30.0), + ("2023-06-30", "2023-08-15", 6.0, 12.0, 31.0), +]} + + +def test_fina_stores_superset_keeps_ann_date(tmp_path): + cache = _cache(tmp_path) + f = _FinaFetch(_FINA_CAT) + cold = cache.fina_indicator(["000001.SZ"], "2023-01-01", "2023-12-31", f) + # the cache ALWAYS stores the canonical superset (field-set independent) + assert list(cold.columns) == [ + "symbol", "ann_date", "end_date", "roe", "netprofit_yoy", "grossprofit_margin", + ] + assert set(cold["ann_date"]) == {"20230420", "20230815"} # disclosure dates kept + assert len(cold) == 2 + # warm identical request (fina tail off) -> zero calls + f.calls.clear() + cache2 = TushareCache( + cache._store, cache._ledger, clock=_CLK, + recent_tail_overrides={FINA_INDICATOR: 0}, + ) + warm = cache2.fina_indicator(["000001.SZ"], "2023-01-01", "2023-12-31", f) + assert f.calls == [] + assert len(warm) == 2 + + +def test_fina_subset_warm_then_superset_no_keyerror(tmp_path): + # REGRESSION (codex acceptance blocker): a warm that only NEEDED roe must not + # block a later request that needs grossprofit_margin. Because the cache stores + # the superset, the second read finds the column (no KeyError) with real data. + cache = TushareCache( + cache_store := CacheParquetStore(str(tmp_path / "c")), + cache_led := CoverageLedger(str(tmp_path / "c")), + clock=_CLK, recent_tail_overrides={FINA_INDICATOR: 0}, + ) + f = _FinaFetch(_FINA_CAT) + cache.fina_indicator(["000001.SZ"], "2023-01-01", "2023-12-31", f) # cold warm + f.calls.clear() + # a DIFFERENT instance over the SAME store/ledger (a later config's warm cache) + cache2 = TushareCache( + cache_store, cache_led, clock=_CLK, recent_tail_overrides={FINA_INDICATOR: 0} + ) + out = cache2.fina_indicator(["000001.SZ"], "2023-01-01", "2023-12-31", f) + assert f.calls == [] # warm hit (no refetch) + assert "grossprofit_margin" in out.columns + assert set(out["grossprofit_margin"]) == {30.0, 31.0} # real data, not NaN + + +def test_fina_tail_refetches_recent_periods(tmp_path): + # a long fina tail re-fetches the trailing window of the requested range every + # run, so a LATE disclosure of a recent period is caught. + cache = _cache(tmp_path, recent_tail_overrides={FINA_INDICATOR: 400}) + f = _FinaFetch(_FINA_CAT) + cache.fina_indicator(["000001.SZ"], "2023-01-01", "2023-12-31", f) + f.calls.clear() + # second identical run still refetches the trailing window (late disclosures) + cache.fina_indicator(["000001.SZ"], "2023-01-01", "2023-12-31", f) + assert f.calls # NOT zero — the tail is always refetched + + +def test_fina_fields_cover_all_financial_factors(tmp_path): # noqa: ARG001 + # drift guard: the cache superset must cover every project financial factor. + from data.cache.tushare_cache import FINA_FIELDS + from factors.compute.financial import SUPPORTED_FIELDS + + assert set(FINA_FIELDS) >= set(SUPPORTED_FIELDS) + + +# --------------------------------------------------------------------------- # +# index_member_all — per-symbol dimension (SW in/out intervals) +# --------------------------------------------------------------------------- # +class _MemberFetch: + def __init__(self, catalog): + self.catalog = catalog + self.calls: list[str] = [] + + def __call__(self, symbol): + self.calls.append(symbol) + rows = self.catalog.get(symbol, []) + if not rows: + return pd.DataFrame() + return pd.DataFrame({ + "ts_code": [symbol] * len(rows), + "l1_name": [r[0] for r in rows], + "l2_name": [r[1] for r in rows], + "l3_name": [r[2] for r in rows], + "in_date": [r[3] for r in rows], + "out_date": [r[4] for r in rows], + }) + + +_MEM_CAT = {"000001.SZ": [ + ("Banks", "Banks2", "Banks3", "20200101", None), +]} + + +def test_index_member_all_cold_warm(tmp_path): + cache = _cache(tmp_path, refresh_dimension_days=30) + f = _MemberFetch(_MEM_CAT) + cold = cache.index_member_all(["000001.SZ"], f) + assert list(cold.columns) == [ + "symbol", "l1_name", "l2_name", "l3_name", "in_date", "out_date", + ] + assert cold.iloc[0]["l1_name"] == "Banks" + assert pd.isna(cold.iloc[0]["out_date"]) # active membership + assert f.calls == ["000001.SZ"] + # warm: snapshot fresh -> zero calls + f.calls.clear() + warm = cache.index_member_all(["000001.SZ"], f) + assert f.calls == [] + assert len(warm) == 1 + + +def test_update_summary_counts(tmp_path): + cache = _cache(tmp_path, refresh_recent_days=0) + cache.daily_basic(["000001.SZ"], "2024-01-02", "2024-01-03", _DBFetch(_DB_CAT)) + summ = cache.update_summary() + assert summ[DAILY_BASIC]["requests"] >= 1 + assert summ[DAILY_BASIC]["rows_written"] == 2 + assert summ[INDEX_MEMBER_ALL]["requests"] == 0 # untouched endpoint seeds 0