diff --git a/data/cache/__init__.py b/data/cache/__init__.py index 3c8666f..85e43f4 100644 --- a/data/cache/__init__.py +++ b/data/cache/__init__.py @@ -19,7 +19,17 @@ from __future__ import annotations from data.cache.coverage import CoverageLedger +from data.cache.intraday_cache import TushareIntradayCache +from data.cache.intraday_coverage import IntradayCoverageLedger +from data.cache.intraday_parquet_store import IntradayParquetStore from data.cache.parquet_store import CacheParquetStore from data.cache.tushare_cache import TushareCache -__all__ = ["CacheParquetStore", "CoverageLedger", "TushareCache"] +__all__ = [ + "CacheParquetStore", + "CoverageLedger", + "IntradayCoverageLedger", + "IntradayParquetStore", + "TushareCache", + "TushareIntradayCache", +] diff --git a/data/cache/intraday_cache.py b/data/cache/intraday_cache.py new file mode 100644 index 0000000..b805aa1 --- /dev/null +++ b/data/cache/intraday_cache.py @@ -0,0 +1,198 @@ +"""Read-through cache for tushare ``stk_mins`` raw 1min bars (I2). + +``TushareIntradayCache`` turns a requested minute window into ONLY the uncovered +trading-day gaps, 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 bars into the month-partitioned store, records coverage in the +intraday ledger (including empty returns), then returns the requested bars read +back from the cache. + +Raw intraday source of truth is ``freq="1min"`` ONLY. Coarser bars (5/15/30/60min) +are derived later from cached 1min data, never fetched/cached here — a non-1min +request fails fast (before any fetch). Natural key is ``(symbol, freq, bar_end)``. + +Behaviour the acceptance pins down: + * full cache miss -> fetch the uncovered day-range, populate cache; + * full cache hit -> ZERO fetch calls; + * partial gap -> fetch only the missing days; + * empty return -> still recorded as coverage (no needless refetch); + * failed fetch -> NOT recorded as coverage (a later run retries); + * duplicate upsert -> one row per (symbol, freq, bar_end). + +Stored bars are RAW (``bar_end``/OHLCV/volume/amount/``source_trade_time``/``freq``); +the derived PIT fields (``bar_start``/``available_time``) are recomputed by +``normalize_intraday_bars`` after read. No token or secret ever reaches this +layer; the ledger holds endpoint metadata only. +""" + +from __future__ import annotations + +import hashlib +import logging +from collections.abc import Callable + +import pandas as pd + +from data.cache.intervals import subtract_intervals +from data.cache.intraday_coverage import IntradayCoverageLedger +from data.cache.intraday_parquet_store import KEY_COLS, STORED_COLUMNS, IntradayParquetStore +from data.clean.intraday_schema import RAW_INTRADAY_FREQ, ensure_raw_intraday_freq + +_LOGGER = logging.getLogger("data.cache.intraday") + +ENDPOINT = "stk_mins_1min" + +# Columns returned to the feed (== feed._to_canonical output, so the cached path +# normalizes to a frame byte-identical to the direct path). +READ_COLUMNS: list[str] = [ + "time", + "symbol", + "open", + "high", + "low", + "close", + "volume", + "amount", + "source_trade_time", +] + +# 1min A-share bars are ~240 rows/trading-day; <=23 calendar days (~16 trading +# days, ~3840 rows) keeps each call well under tushare's 8000-row single-call cap. +_FETCH_WINDOW_DAYS = 23 + +# A fetch callable: (symbol, start_dt_str, end_dt_str) -> raw stk_mins frame|None, +# where the datetime strings are 'YYYY-MM-DD HH:MM:SS'. +FetchIntraday = Callable[[str, str, str], "pd.DataFrame | None"] + + +def _fields_hash(columns: list[str]) -> str: + return hashlib.sha1(",".join(sorted(columns)).encode("utf-8")).hexdigest()[:16] + + +def _parse_stk_mins(raw: pd.DataFrame | None) -> pd.DataFrame: + """tushare ``stk_mins`` frame -> raw-canonical stored rows (or empty). + + ``ts_code`` -> symbol, ``vol`` -> volume, ``trade_time`` -> ``bar_end`` (the + bar's END) + ``source_trade_time`` (raw string, audit). ``freq`` is fixed to + the only raw freq, 1min. Mirrors the feed's direct ``_to_canonical`` so the + stored bars rebuild the exact same intraday frame. + """ + if raw is None or len(raw) == 0: + return pd.DataFrame(columns=STORED_COLUMNS) + df = raw.rename(columns={"ts_code": "symbol", "vol": "volume"}).copy() + df["bar_end"] = pd.to_datetime(df["trade_time"]) + df["source_trade_time"] = df["trade_time"].astype(str) + df["symbol"] = df["symbol"].astype(str) + df["freq"] = RAW_INTRADAY_FREQ + for col in STORED_COLUMNS: + if col not in df.columns: + df[col] = float("nan") + return df[STORED_COLUMNS] + + +class TushareIntradayCache: + """Endpoint-level read-through cache for stk_mins raw 1min bars.""" + + def __init__( + self, + store: IntradayParquetStore, + ledger: IntradayCoverageLedger, + *, + force_refresh: bool = False, + fetch_window_days: int = _FETCH_WINDOW_DAYS, + clock: Callable[[], pd.Timestamp] | None = None, + ) -> None: + self._store = store + self._ledger = ledger + self._force_refresh = bool(force_refresh) + self._window_days = max(1, int(fetch_window_days)) + self._clock = clock or pd.Timestamp.now + self.fetch_counts: dict[str, int] = {ENDPOINT: 0} + + def stats(self) -> dict[str, int]: + """Endpoint -> number of gap-window fetches sent to the API this instance.""" + return dict(self.fetch_counts) + + # -- read-through API --------------------------------------------------- # + def stk_mins_1min( + self, + symbols: list[str], + start: str, + end: str, + fetch: FetchIntraday, + *, + freq: str = RAW_INTRADAY_FREQ, + ) -> pd.DataFrame: + """Return raw 1min bars for ``symbols`` over [start, end] (read-through). + + ``freq`` MUST be ``"1min"`` (checked before any fetch). Returns columns + ``READ_COLUMNS`` (``time`` == ``bar_end``); the feed normalizes them into + the canonical intraday panel. An all-miss-empty result is an empty frame. + """ + ensure_raw_intraday_freq(freq) + req_start = pd.Timestamp(start) + req_end = pd.Timestamp(end) + fields_hash = _fields_hash(STORED_COLUMNS) + + out: list[pd.DataFrame] = [] + for symbol in symbols: + for gap_start, gap_end in self._day_gaps(symbol, req_start, req_end, freq): + self._fetch_gap(symbol, freq, gap_start, gap_end, fetch, fields_hash) + cached = self._store.read_range(ENDPOINT, symbol, freq, req_start, req_end) + if not cached.empty: + hit = cached.rename(columns={"bar_end": "time"}) + out.append(hit[READ_COLUMNS]) + _LOGGER.info( + "cache %s: %d symbols (api window-calls=%d)", + ENDPOINT, len(symbols), self.fetch_counts[ENDPOINT], + ) + if not out: + return pd.DataFrame(columns=READ_COLUMNS) + return pd.concat(out, ignore_index=True).reset_index(drop=True) + + # -- planning ----------------------------------------------------------- # + def _day_gaps(self, symbol, req_start, req_end, freq): + """Uncovered trading-day intervals to fetch for ``symbol``.""" + if self._force_refresh: + return [(req_start.normalize(), req_end.normalize())] + covered = self._ledger.covered_day_intervals(ENDPOINT, symbol, freq) + return subtract_intervals(req_start, req_end, covered) + + def _fetch_gap(self, symbol, freq, gap_start, gap_end, fetch, fields_hash): + """Fetch one day-gap (paged by window), upsert raw bars, record coverage. + + Coverage is recorded for the WHOLE gap ONCE, AFTER every window of the gap + succeeds (ok if any bar landed, else empty). If a window fetch raises, the + exception propagates and coverage is NOT recorded, so the gap is retried + on a later run (rows already upserted are deduped idempotently). + """ + total_rows = 0 + win_start = pd.Timestamp(gap_start).normalize() + gap_end_day = pd.Timestamp(gap_end).normalize() + while win_start <= gap_end_day: + win_end = min( + win_start + pd.Timedelta(days=self._window_days - 1), gap_end_day + ) + raw = fetch( + symbol, + win_start.strftime("%Y-%m-%d 00:00:00"), + win_end.strftime("%Y-%m-%d 23:59:59"), + ) + self.fetch_counts[ENDPOINT] = self.fetch_counts.get(ENDPOINT, 0) + 1 + parsed = _parse_stk_mins(raw) + if not parsed.empty: + self._store.upsert(ENDPOINT, symbol, freq, parsed, KEY_COLS) + total_rows += len(parsed) + win_start = win_end + pd.Timedelta(days=1) + self._ledger.record( + endpoint=ENDPOINT, + key_type="symbol", + key=symbol, + raw_freq=freq, + start_time=pd.Timestamp(gap_start).normalize(), + end_time=gap_end_day, + fields_hash=fields_hash, + row_count=total_rows, + status="ok" if total_rows else "empty", + fetched_at=self._clock(), + ) diff --git a/data/cache/intraday_coverage.py b/data/cache/intraday_coverage.py new file mode 100644 index 0000000..63436d4 --- /dev/null +++ b/data/cache/intraday_coverage.py @@ -0,0 +1,140 @@ +"""Intraday-aware coverage ledger for the minute cache (I2: stk_mins 1min). + +The daily :class:`data.cache.coverage.CoverageLedger` tracks coverage as +``start_date``/``end_date`` (day resolution). Minute coverage is a TIMESTAMP +interval over a single raw frequency, so reusing the daily date-range columns +would silently conflate a date range with a timestamp range. This is a SEPARATE +ledger (its own parquet file) with explicit ``start_time``/``end_time`` and a +``raw_freq`` column. + +Schema (one append-only row per fetch attempt): + + endpoint stk_mins_1min + key_type symbol + key e.g. 000001.SZ + raw_freq always "1min" (raw intraday SoT; coarser bars are derived) + start_time closed interval start (Timestamp, minute precision possible) + end_time closed interval end (Timestamp) + fields_hash stable hash of the stored field set + fetched_at local timestamp of the fetch + row_count rows the endpoint returned + status ok | empty | failed + schema_version cache schema version + +Only ``ok`` / ``empty`` rows count as COVERAGE (a ``failed`` fetch leaves the +range uncovered so a later run retries). The planner consumes coverage at +trading-day granularity (a minute fetch unit is a whole trading day), but the +ledger preserves the true timestamp span. The ledger NEVER stores a token or any +secret-file content — it holds endpoint metadata only. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pandas as pd + +from data.cache.intervals import Interval + +INTRADAY_LEDGER_COLUMNS: list[str] = [ + "endpoint", + "key_type", + "key", + "raw_freq", + "start_time", + "end_time", + "fields_hash", + "fetched_at", + "row_count", + "status", + "schema_version", +] + +_COVERING_STATUSES = ("ok", "empty") + + +class IntradayCoverageLedger: + """Append-only parquet ledger of fetched (endpoint, symbol, time-range) coverage.""" + + def __init__(self, root: str, schema_version: str = "v1") -> None: + self._root = Path(root) + self._schema_version = schema_version + + @property + def path(self) -> Path: + return self._root / "manifest" / "coverage_intraday.parquet" + + # -- read --------------------------------------------------------------- # + def read(self) -> pd.DataFrame: + """Return the full ledger (empty, correctly-typed frame if absent).""" + if not self.path.exists(): + return pd.DataFrame(columns=INTRADAY_LEDGER_COLUMNS) + return pd.read_parquet(self.path) + + def covered_day_intervals( + self, endpoint: str, key: str, raw_freq: str + ) -> list[Interval]: + """Covered closed DAY intervals for ``(endpoint, key, raw_freq)``. + + The stored span is a true timestamp interval; the planner consumes it at + trading-day granularity (a minute fetch unit is a whole trading day), so + each covered span is reported day-normalized for the day-interval algebra + in :mod:`data.cache.intervals`. Only ``ok``/``empty`` rows count. + """ + ledger = self.read() + if ledger.empty: + return [] + mask = ( + (ledger["endpoint"] == endpoint) + & (ledger["key"] == str(key)) + & (ledger["raw_freq"] == str(raw_freq)) + & (ledger["status"].isin(_COVERING_STATUSES)) + & ledger["start_time"].notna() + & ledger["end_time"].notna() + ) + sub = ledger[mask] + return [ + (pd.Timestamp(s).normalize(), pd.Timestamp(e).normalize()) + for s, e in zip(sub["start_time"], sub["end_time"]) + ] + + # -- write -------------------------------------------------------------- # + def record( + self, + *, + endpoint: str, + key_type: str, + key: str, + raw_freq: str, + start_time, + end_time, + fields_hash: str, + row_count: int, + status: str, + fetched_at: pd.Timestamp, + ) -> None: + """Append one coverage row (atomic). No secret is ever written here.""" + row = { + "endpoint": endpoint, + "key_type": key_type, + "key": str(key), + "raw_freq": str(raw_freq), + "start_time": pd.Timestamp(start_time) if start_time is not None else pd.NaT, + "end_time": pd.Timestamp(end_time) if end_time is not None else pd.NaT, + "fields_hash": str(fields_hash), + "fetched_at": pd.Timestamp(fetched_at), + "row_count": int(row_count), + "status": str(status), + "schema_version": self._schema_version, + } + existing = self.read() + new_row = pd.DataFrame([row]) + combined = new_row if existing.empty else pd.concat( + [existing, new_row], ignore_index=True + ) + combined = combined.reindex(columns=INTRADAY_LEDGER_COLUMNS) + self.path.parent.mkdir(parents=True, exist_ok=True) + tmp = self.path.with_suffix(".parquet.tmp") + combined.to_parquet(tmp, engine="pyarrow", index=False) + os.replace(tmp, self.path) diff --git a/data/cache/intraday_parquet_store.py b/data/cache/intraday_parquet_store.py new file mode 100644 index 0000000..89170df --- /dev/null +++ b/data/cache/intraday_parquet_store.py @@ -0,0 +1,177 @@ +"""Month-partitioned parquet store for raw 1min bars (I2: stk_mins 1min). + +The daily :class:`data.cache.parquet_store.CacheParquetStore` keeps ONE file per +``(endpoint, symbol)``. That is too coarse for multi-year 1min data (one symbol +is hundreds of thousands of rows), so the intraday store partitions by +``year``/``month`` of ``bar_end``: + + /stk_mins_1min/freq=1min/symbol_prefix=000/symbol=000001.SZ/year=2024/month=01.parquet + +Natural key is ``(symbol, freq, bar_end)`` with ``freq`` always ``1min``. Writes +are ATOMIC (write ``*.tmp`` then ``os.replace``) and idempotent: an upsert drops +duplicates by the natural key keeping the latest fetched row, so re-fetching an +overlapping window never doubles a ``bar_end``. A best-effort per-file lock +guards the read-modify-write. + +The store holds RAW bars only (``bar_end``/OHLCV/volume/amount/``source_trade_time``/ +``freq``) and never sees a token — nothing secret is ever written here. The +derived PIT fields (``bar_start``/``available_time``) are NOT stored: they are +recomputed by ``normalize_intraday_bars`` after read. +""" + +from __future__ import annotations + +import os +import time +from contextlib import contextmanager +from pathlib import Path + +import pandas as pd + +# Raw-canonical columns persisted per 1min bar (natural key = symbol, freq, bar_end). +STORED_COLUMNS: list[str] = [ + "symbol", + "bar_end", + "source_trade_time", + "open", + "high", + "low", + "close", + "volume", + "amount", + "freq", +] +KEY_COLS: list[str] = ["symbol", "freq", "bar_end"] + + +def _symbol_prefix(symbol: str) -> str: + s = str(symbol) + return s[:3] if len(s) >= 3 else s + + +def _months_between(start: pd.Timestamp, end: pd.Timestamp) -> list[tuple[int, int]]: + """Inclusive list of (year, month) tuples spanning [start, end].""" + s = pd.Timestamp(start).normalize().replace(day=1) + e = pd.Timestamp(end).normalize().replace(day=1) + out: list[tuple[int, int]] = [] + cur = s + while cur <= e: + out.append((cur.year, cur.month)) + cur = (cur + pd.Timedelta(days=32)).replace(day=1) + return out + + +class IntradayParquetStore: + """Persist raw 1min bars as per-(symbol, freq, year, month) parquet files.""" + + def __init__(self, root: str) -> None: + self._root = Path(root) + + # -- paths -------------------------------------------------------------- # + def month_path( + self, endpoint: str, symbol: str, freq: str, year: int, month: int + ) -> Path: + return ( + self._root + / endpoint + / f"freq={freq}" + / f"symbol_prefix={_symbol_prefix(symbol)}" + / f"symbol={symbol}" + / f"year={year}" + / f"month={month:02d}.parquet" + ) + + def _lock_path(self, endpoint: str, symbol: str, freq: str, year: int, month: int) -> Path: + return ( + self._root / ".locks" + / f"{endpoint}__{symbol}__{freq}__{year}-{month:02d}.lock" + ) + + # -- locking (best-effort, per file) ------------------------------------ # + @contextmanager + def _locked(self, endpoint, symbol, freq, year, month, timeout: float = 10.0): + lock = self._lock_path(endpoint, symbol, freq, year, month) + lock.parent.mkdir(parents=True, exist_ok=True) + deadline = time.monotonic() + timeout + fd = None + while True: + try: + fd = os.open(str(lock), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + break + except FileExistsError: + if time.monotonic() > deadline: + break + time.sleep(0.05) + try: + yield + finally: + if fd is not None: + os.close(fd) + try: + lock.unlink() + except FileNotFoundError: + pass + + # -- read --------------------------------------------------------------- # + def read_range( + self, endpoint: str, symbol: str, freq: str, start, end + ) -> pd.DataFrame: + """Return stored raw bars for ``symbol`` with ``bar_end`` in [start, end]. + + Reads only the month partitions the window spans; an absent month is + simply skipped. The result is sorted by ``bar_end`` (empty if nothing + cached). Both bounds are inclusive. + """ + req_start = pd.Timestamp(start) + req_end = pd.Timestamp(end) + frames: list[pd.DataFrame] = [] + for year, month in _months_between(req_start, req_end): + path = self.month_path(endpoint, symbol, freq, year, month) + if path.exists(): + frames.append(pd.read_parquet(path)) + if not frames: + return pd.DataFrame(columns=STORED_COLUMNS) + df = pd.concat(frames, ignore_index=True) + mask = (df["bar_end"] >= req_start) & (df["bar_end"] <= req_end) + return df.loc[mask, STORED_COLUMNS].sort_values("bar_end").reset_index(drop=True) + + # -- upsert ------------------------------------------------------------- # + def upsert( + self, endpoint: str, symbol: str, freq: str, rows: pd.DataFrame, key_cols: list[str] + ) -> int: + """Merge ``rows`` into their month partitions, dedup by ``key_cols``. + + Returns the number of rows written across the touched months. New rows + win on a key collision (a re-fetched overlap replaces the stale row). + Atomic per-month write; never mutates the caller's frame. + """ + if rows is None or rows.empty: + return 0 + rows = rows.copy() + rows["bar_end"] = pd.to_datetime(rows["bar_end"]) + bar_end = rows["bar_end"] + written = 0 + for (year, month), part in rows.groupby([bar_end.dt.year, bar_end.dt.month]): + written += self._upsert_month( + endpoint, symbol, freq, int(year), int(month), part, key_cols + ) + return written + + def _upsert_month( + self, endpoint, symbol, freq, year, month, rows, key_cols + ) -> int: + path = self.month_path(endpoint, symbol, freq, year, month) + with self._locked(endpoint, symbol, freq, year, month): + if path.exists(): + existing = pd.read_parquet(path) + combined = pd.concat([existing, rows], ignore_index=True) + else: + combined = rows.copy() + combined = combined.drop_duplicates(subset=key_cols, keep="last") + combined = combined.sort_values("bar_end").reset_index(drop=True) + combined = combined.reindex(columns=STORED_COLUMNS) + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".parquet.tmp") + combined.to_parquet(tmp, engine="pyarrow", index=False) + os.replace(tmp, path) + return len(combined) diff --git a/data/clean/intraday_aggregate.py b/data/clean/intraday_aggregate.py new file mode 100644 index 0000000..3d62308 --- /dev/null +++ b/data/clean/intraday_aggregate.py @@ -0,0 +1,248 @@ +"""Minute -> daily PIT-safe aggregation (I3): 14:50 tail-rebalance features. + +Turns normalized 1min bars (:mod:`data.clean.intraday_schema`) into a daily +``(date, symbol)`` feature table the factor layer can join. The single hard rule +(``02_minute_pit_semantics.md``): a bar may inform a decision iff +``available_time <= decision_time``. Concretely, for each bar the cutoff is its +own trading date plus ``decision_time`` (default ``14:50:00``), and ONLY bars +with ``available_time <= cutoff`` enter the aggregation. + +Ordering matters and is enforced: the PIT filter runs on per-bar TIMESTAMPS +FIRST; only after filtering are the surviving bars grouped by ``(trade_date, +symbol)``. Bars are never date-normalized into a daily bucket before the cutoff +is applied — doing so would discard the very timestamps the filter needs, and a +post-14:50 bar could leak into a 14:50 decision. + +There is no ``data_lag`` parameter here: ``available_time`` already bakes in the +lag (it was set to ``bar_end + data_lag`` at normalize time), so the cutoff +compares against ``available_time`` directly — passing a second lag would +double-count it. + +Feature columns ENCODE the cutoff so an ambiguous name can never hide a leak: + + intraday_ret_0930_1450 close/open return, session open -> cutoff + intraday_realized_vol_0930_1450 sqrt(sum of 1min squared log-returns) + intraday_vwap_0930_1450 sum(amount) / sum(volume) + intraday_last30m_ret_1420_1450 return over the last 30m before the cutoff + +This module is DATA-layer only: it does not fetch, does not touch factors / alpha +/ portfolio / runtime, and never sees a token. Coarser intraday bars, if needed, +are DERIVED here from 1min via :func:`resample_intraday_bars` (never raw-fetched), +and a derived bar inherits ``available_time = max(source_1min.available_time)``. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + +from data.clean.intraday_schema import ( + INTRADAY_CORE_COLUMNS, + INTRADAY_INDEX_NAMES, + SYMBOL_LEVEL, + ensure_supported_freq, + validate_intraday_bars, +) + +DATE_LEVEL = "date" +DAILY_INDEX_NAMES: list[str] = [DATE_LEVEL, SYMBOL_LEVEL] + +# Feature keys (selectable via the ``features`` argument); column NAMES below +# encode the cutoff and are derived from the time arguments. +INTRADAY_FEATURE_KEYS: tuple[str, ...] = ( + "ret", + "realized_vol", + "vwap", + "last30m_ret", +) + +DEFAULT_DECISION_TIME = "14:50:00" +DEFAULT_SESSION_OPEN = "09:30:00" +DEFAULT_LAST_WINDOW_MINUTES = 30 + + +def _hhmm(time_str: str) -> str: + """'14:50:00' -> '1450' (label for a within-day time).""" + total = int(pd.Timedelta(time_str).total_seconds() // 60) + return f"{total // 60:02d}{total % 60:02d}" + + +def _hhmm_minus(time_str: str, minutes: int) -> str: + """Label for ``time_str`` shifted back ``minutes`` (e.g. 14:50 - 30 -> '1420').""" + total = int(pd.Timedelta(time_str).total_seconds() // 60) - int(minutes) + return f"{total // 60:02d}{total % 60:02d}" + + +def _resolve_feature_keys(features: list[str] | None) -> list[str]: + if features is None: + return list(INTRADAY_FEATURE_KEYS) + unknown = [f for f in features if f not in INTRADAY_FEATURE_KEYS] + if unknown: + raise ValueError( + f"Unknown intraday feature(s): {unknown}. " + f"Known: {list(INTRADAY_FEATURE_KEYS)}." + ) + return list(features) + + +def _column_name( + key: str, session_open: str, decision_time: str, last_window_minutes: int +) -> str: + o, c = _hhmm(session_open), _hhmm(decision_time) + if key == "ret": + return f"intraday_ret_{o}_{c}" + if key == "realized_vol": + return f"intraday_realized_vol_{o}_{c}" + if key == "vwap": + return f"intraday_vwap_{o}_{c}" + if key == "last30m_ret": + start = _hhmm_minus(decision_time, last_window_minutes) + return f"intraday_last{int(last_window_minutes)}m_ret_{start}_{c}" + raise ValueError(f"Unhandled intraday feature key: {key!r}.") + + +def _empty_daily(colnames: list[str]) -> pd.DataFrame: + index = pd.MultiIndex.from_arrays( + [pd.DatetimeIndex([]), pd.Index([], dtype=object)], + names=DAILY_INDEX_NAMES, + ) + out = pd.DataFrame({c: pd.Series([], dtype=float) for c in colnames}) + out.index = index + return out + + +def _compute_group( + g: pd.DataFrame, decision_time: str, last_window_minutes: int +) -> dict[str, float]: + """Per-(date, symbol) features from already PIT-filtered, bar_end-sorted bars.""" + closes = g["close"].to_numpy(dtype=float) + opens = g["open"].to_numpy(dtype=float) + vols = g["volume"].to_numpy(dtype=float) + amts = g["amount"].to_numpy(dtype=float) + bar_end = g["bar_end"] + trade_date = pd.Timestamp(bar_end.iloc[0]).normalize() + cutoff = trade_date + pd.Timedelta(decision_time) + window_start = cutoff - pd.Timedelta(minutes=int(last_window_minutes)) + + first_open, last_close = opens[0], closes[-1] + ret = (last_close / first_open - 1.0) if first_open else float("nan") + + if len(closes) >= 2 and np.all(closes > 0): + logret = np.diff(np.log(closes)) + realized_vol = float(np.sqrt(np.sum(logret**2))) + else: + realized_vol = float("nan") + + tot_vol = float(np.nansum(vols)) + vwap = float(np.nansum(amts) / tot_vol) if tot_vol > 0 else float("nan") + + ref_mask = (bar_end <= window_start).to_numpy() + if ref_mask.any(): + ref_close = closes[ref_mask][-1] + last30 = (last_close / ref_close - 1.0) if ref_close else float("nan") + else: + last30 = float("nan") + + return { + "ret": ret, + "realized_vol": realized_vol, + "vwap": vwap, + "last30m_ret": last30, + } + + +def asof_daily_features( + bars: pd.DataFrame, + *, + decision_time: str = DEFAULT_DECISION_TIME, + session_open: str = DEFAULT_SESSION_OPEN, + last_window_minutes: int = DEFAULT_LAST_WINDOW_MINUTES, + features: list[str] | None = None, +) -> pd.DataFrame: + """PIT-safe daily features from normalized 1min ``bars``. + + For each bar the cutoff is ``bar's trade_date + decision_time``; only bars + with ``available_time <= cutoff`` are used (the filter runs on timestamps + BEFORE any daily grouping). Returns a ``MultiIndex(date, symbol)`` frame whose + columns encode the cutoff. An empty input, or an input with no visible bar + before any cutoff, yields a schema-shaped empty daily frame. + """ + validate_intraday_bars(bars) + keys = _resolve_feature_keys(features) + colnames = [ + _column_name(k, session_open, decision_time, last_window_minutes) + for k in keys + ] + + if len(bars) == 0: + return _empty_daily(colnames) + + work = bars.reset_index() + work["trade_date"] = work["bar_end"].dt.normalize() + cutoff = work["trade_date"] + pd.Timedelta(decision_time) + # PIT filter FIRST (per-bar timestamps), THEN group by day. + visible = work.loc[work["available_time"] <= cutoff].copy() + if visible.empty: + return _empty_daily(colnames) + + visible = visible.sort_values([SYMBOL_LEVEL, "bar_end"]) + index_tuples: list[tuple] = [] + data: dict[str, list[float]] = {c: [] for c in colnames} + for (date, sym), g in visible.groupby(["trade_date", SYMBOL_LEVEL], sort=True): + feats = _compute_group(g, decision_time, last_window_minutes) + index_tuples.append((date, str(sym))) + for key, col in zip(keys, colnames): + data[col].append(feats[key]) + + index = pd.MultiIndex.from_tuples(index_tuples, names=DAILY_INDEX_NAMES) + return pd.DataFrame(data, index=index)[colnames].sort_index() + + +def resample_intraday_bars(bars: pd.DataFrame, freq: str) -> pd.DataFrame: + """Derive coarser intraday bars from normalized 1min ``bars``. + + Coarser bars are DERIVED from 1min, never raw-fetched. Each coarse bar covers + a ``freq``-aligned window ending at ``bar_end``; OHLC = first-open / max-high / + min-low / last-close, volume/amount summed. Critically, the coarse bar's + ``available_time = max(source_1min.available_time)`` — it is only usable once + EVERY constituent 1min bar is available — and is NOT recomputed as + ``bar_end + data_lag`` (which would understate availability). The result + passes :func:`validate_intraday_bars`. + """ + ensure_supported_freq(freq) + validate_intraday_bars(bars) + + if len(bars) == 0: + # empty 1min -> empty coarse (same schema, just relabelled freq) + return bars.copy() + + # Bucket each 1min bar by its freq-aligned window (ceil bar_end to the grid), + # then aggregate WITHIN the bucket from the SOURCE bars only: bar_start/bar_end/ + # available_time are min/max/max over the constituents, never the nominal grid + # boundary. A partial bucket therefore ends at its real last 1min bar, keeping + # available_time (= max source) >= bar_end and never claiming data it lacks. + work = bars.reset_index().sort_values([SYMBOL_LEVEL, "bar_end"]) + work["bucket"] = work["bar_end"].dt.ceil(freq) + grouped = work.groupby([SYMBOL_LEVEL, "bucket"], sort=True).agg( + open=("open", "first"), + high=("high", "max"), + low=("low", "min"), + close=("close", "last"), + volume=("volume", "sum"), + amount=("amount", "sum"), + bar_start=("bar_start", "min"), + bar_end=("bar_end", "max"), + available_time=("available_time", "max"), + ).reset_index() + grouped["freq"] = freq + + index = pd.MultiIndex.from_arrays( + [grouped["bar_end"].to_numpy(), grouped[SYMBOL_LEVEL].astype(str).to_numpy()], + names=INTRADAY_INDEX_NAMES, + ) + ordered = [*INTRADAY_CORE_COLUMNS, "freq", "bar_start", "bar_end", "available_time"] + out = grouped[ordered].copy() + out.index = index + out = out.sort_index() + validate_intraday_bars(out) + return out diff --git a/data/clean/intraday_schema.py b/data/clean/intraday_schema.py new file mode 100644 index 0000000..d716941 --- /dev/null +++ b/data/clean/intraday_schema.py @@ -0,0 +1,282 @@ +"""Intraday (minute) bar schema — SEPARATE from the daily panel contract. + +The daily canonical panel (:mod:`data.clean.schema`) normalizes its ``date`` +level to midnight; minute bars must NOT reuse that — for intraday data the +intra-day ``time`` is meaningful and must be preserved. This module defines the +minute-bar shape and the helpers that enforce it. It is pure: every function +returns a new object and never mutates its input. It is also source-agnostic: +raw-vendor field renaming (e.g. tushare ``vol`` -> ``volume``) belongs in the +feed, not here. + +Canonical intraday shape:: + + index: MultiIndex(time, symbol) + time -> pandas.Timestamp at minute precision, NOT normalized to midnight + symbol -> str, tushare style, e.g. "000001.SZ" + sorted by (time, symbol), no duplicate (time, symbol) pairs. + +Required columns: open, high, low, close, volume, amount, freq, bar_start, +bar_end, available_time. Raw persisted intraday bars use ``freq="1min"``; coarser +intraday bars, if needed later, are derived views rebuilt from 1min data. Extra +columns such as ``source_trade_time`` are kept for audit. + +Minute-level PIT time rules (see tmp/context/intraday_pit_checkpoints/ +02_minute_pit_semantics.md) — the upstream ``trade_time`` is the END of the +interval the bar represents:: + + raw_freq = 1min + bar_end = source trade_time + bar_start = bar_end - freq + available_time = bar_end + data_lag (earliest a strategy may use the bar) + +``available_time`` is what enforces "an observation may be used for a decision +iff available_time <= decision_time". A conservative default ``data_lag`` of +``1min`` is used for backtests unless a tighter measured feed delay is known. + +IMPORTANT (design invariant, CLAUDE.md #1): this module is intraday *shape* only. +It must NOT compute forward returns or anything that touches the future. +""" + +from __future__ import annotations + +import pandas as pd + +# Minimal required OHLCV columns for an intraday bar. +INTRADAY_CORE_COLUMNS: list[str] = [ + "open", + "high", + "low", + "close", + "volume", + "amount", +] + +# Derived PIT/time columns the normalizer adds. +INTRADAY_TIME_COLUMNS: list[str] = [ + "freq", + "bar_start", + "bar_end", + "available_time", +] + +# Canonical index level names. +TIME_LEVEL = "time" +SYMBOL_LEVEL = "symbol" +INTRADAY_INDEX_NAMES: list[str] = [TIME_LEVEL, SYMBOL_LEVEL] + +RAW_INTRADAY_FREQ = "1min" + +# The schema can represent derived coarser bars, but raw feeds/cache should only +# persist 1min bars. Coarser bars are built from the 1min cache in a later +# resampling stage, not fetched as separate upstream products. +DERIVED_INTRADAY_FREQS: tuple[str, ...] = ("5min", "15min", "30min", "60min") +SUPPORTED_INTRADAY_FREQS: tuple[str, ...] = ( + RAW_INTRADAY_FREQ, + *DERIVED_INTRADAY_FREQS, +) + + +def ensure_supported_freq(freq: str) -> None: + """Raise a readable ValueError unless ``freq`` is a supported schema freq.""" + if freq not in SUPPORTED_INTRADAY_FREQS: + raise ValueError( + f"Unsupported intraday freq {freq!r}; must be one of " + f"{SUPPORTED_INTRADAY_FREQS}." + ) + + +def ensure_raw_intraday_freq(freq: str) -> None: + """Raise unless a raw upstream/cache request uses the only raw freq: 1min.""" + if freq != RAW_INTRADAY_FREQ: + raise ValueError( + f"Raw intraday bars support only freq={RAW_INTRADAY_FREQ!r}; " + "build coarser bars from cached 1min data via intraday resampling." + ) + + +def _freq_to_timedelta(freq: str) -> pd.Timedelta: + """Convert a supported minute ``freq`` string into a ``pd.Timedelta``.""" + ensure_supported_freq(freq) + return pd.Timedelta(freq) + + +def _extract_time_symbol( + df: pd.DataFrame, +) -> tuple[pd.Series, pd.Series, pd.DataFrame]: + """Pull (time, symbol) out of a frame whether they are columns or the index. + + Returns (time_series, symbol_series, payload_frame_without_keys). Raises a + readable ValueError if neither layout is present. Mirrors + :func:`data.clean.schema._extract_date_symbol` but never normalizes ``time``. + """ + if isinstance(df.index, pd.MultiIndex) and df.index.nlevels == 2: + names = list(df.index.names) + if names == INTRADAY_INDEX_NAMES or set(names) == set(INTRADAY_INDEX_NAMES): + reset = df.reset_index() + return ( + reset[TIME_LEVEL], + reset[SYMBOL_LEVEL], + reset.drop(columns=INTRADAY_INDEX_NAMES), + ) + if all(n is None for n in names): + time = df.index.get_level_values(0).to_series(index=range(len(df))) + symbol = df.index.get_level_values(1).to_series(index=range(len(df))) + return time, symbol, df.reset_index(drop=True) + + if TIME_LEVEL in df.columns and SYMBOL_LEVEL in df.columns: + reset = df.reset_index(drop=True) + return ( + reset[TIME_LEVEL], + reset[SYMBOL_LEVEL], + reset.drop(columns=INTRADAY_INDEX_NAMES), + ) + + raise ValueError( + "Intraday bars must provide 'time' and 'symbol' either as a " + "MultiIndex(time, symbol) or as two columns named 'time' and 'symbol'. " + f"Got index names={list(df.index.names)} and columns={list(df.columns)}." + ) + + +def normalize_intraday_bars( + df: pd.DataFrame, freq: str, data_lag: str = "1min" +) -> pd.DataFrame: + """Return a new intraday panel in canonical shape. + + Accepts ``df`` with (time, symbol) either as columns or a MultiIndex, plus + the OHLCV columns. Derives ``freq``/``bar_start``/``bar_end``/ + ``available_time`` from ``time`` (treated as the bar END) and produces: + + - MultiIndex(time, symbol), ``time`` at its original minute precision + (NEVER normalized to midnight), ``symbol`` str; + - rows sorted ascending by (time, symbol); + - duplicates collapsed by the natural key (symbol, freq, bar_end), with the + LATER input row winning (upstream may return reverse-chronological rows). + + Raises ValueError for an unsupported ``freq`` or missing OHLCV columns. Extra + input columns (e.g. ``source_trade_time``) are preserved for audit. The input + is never mutated. + """ + bar_span = _freq_to_timedelta(freq) + lag = pd.Timedelta(data_lag) + + time, symbol, payload = _extract_time_symbol(df) + + missing = [c for c in INTRADAY_CORE_COLUMNS if c not in payload.columns] + if missing: + raise ValueError( + f"Intraday bars are missing required OHLCV columns: {missing}. " + f"INTRADAY_CORE_COLUMNS = {INTRADAY_CORE_COLUMNS}." + ) + + bar_end = pd.to_datetime(time).reset_index(drop=True) + + out = payload.copy() + out[TIME_LEVEL] = bar_end.to_numpy() + out[SYMBOL_LEVEL] = symbol.astype(str).to_numpy() + out["freq"] = freq + out["bar_end"] = bar_end.to_numpy() + out["bar_start"] = (bar_end - bar_span).to_numpy() + out["available_time"] = (bar_end + lag).to_numpy() + + # Dedup BEFORE sorting so "later input row wins" is by input order. + out = out.drop_duplicates( + subset=[SYMBOL_LEVEL, "freq", "bar_end"], keep="last" + ) + out = out.sort_values([TIME_LEVEL, SYMBOL_LEVEL], kind="mergesort") + + index = pd.MultiIndex.from_arrays( + [out[TIME_LEVEL].to_numpy(), out[SYMBOL_LEVEL].to_numpy()], + names=INTRADAY_INDEX_NAMES, + ) + out = out.drop(columns=[TIME_LEVEL, SYMBOL_LEVEL]) + out.index = index + + ordered = [*INTRADAY_CORE_COLUMNS, *INTRADAY_TIME_COLUMNS] + extras = [c for c in out.columns if c not in ordered] + return out[[*ordered, *extras]] + + +def empty_intraday_bars() -> pd.DataFrame: + """An empty but schema-shaped intraday panel (used when a feed returns none). + + Built with proper dtypes so :func:`validate_intraday_bars` accepts it. + """ + index = pd.MultiIndex.from_arrays( + [pd.DatetimeIndex([]), pd.Index([], dtype=object)], + names=INTRADAY_INDEX_NAMES, + ) + data = { + "open": pd.Series([], dtype=float), + "high": pd.Series([], dtype=float), + "low": pd.Series([], dtype=float), + "close": pd.Series([], dtype=float), + "volume": pd.Series([], dtype=float), + "amount": pd.Series([], dtype=float), + "freq": pd.Series([], dtype=object), + "bar_start": pd.Series([], dtype="datetime64[ns]"), + "bar_end": pd.Series([], dtype="datetime64[ns]"), + "available_time": pd.Series([], dtype="datetime64[ns]"), + } + out = pd.DataFrame(data) + out.index = index + return out + + +def validate_intraday_bars(df: pd.DataFrame) -> None: + """Assert that ``df`` already satisfies the canonical intraday contract. + + Raises a readable ValueError on any violation; returns None on success. Does + not mutate or return a frame — use :func:`normalize_intraday_bars` to fix + shape. + """ + if not isinstance(df.index, pd.MultiIndex) or df.index.nlevels != 2: + raise ValueError( + "Intraday index must be a 2-level MultiIndex(time, symbol); " + f"got {type(df.index).__name__} with names={list(df.index.names)}." + ) + if list(df.index.names) != INTRADAY_INDEX_NAMES: + raise ValueError( + f"Intraday index level names must be {INTRADAY_INDEX_NAMES}; " + f"got {list(df.index.names)}." + ) + + time_level = df.index.get_level_values(TIME_LEVEL) + if not pd.api.types.is_datetime64_any_dtype(time_level): + raise ValueError( + f"Intraday 'time' level must be datetime (pandas.Timestamp); " + f"got dtype {time_level.dtype}." + ) + + symbol_level = df.index.get_level_values(SYMBOL_LEVEL) + if not all(isinstance(s, str) for s in symbol_level): + raise ValueError("Intraday 'symbol' level must contain only str values.") + + required = [*INTRADAY_CORE_COLUMNS, *INTRADAY_TIME_COLUMNS] + missing = [c for c in required if c not in df.columns] + if missing: + raise ValueError(f"Intraday bars are missing required columns: {missing}.") + + if df.index.duplicated().any(): + raise ValueError("Intraday bars have duplicate (time, symbol) index entries.") + + if not df.index.is_monotonic_increasing: + raise ValueError("Intraday bars must be sorted by (time, symbol).") + + if len(df) == 0: + return + + for col in ("bar_start", "bar_end", "available_time"): + if not pd.api.types.is_datetime64_any_dtype(df[col]): + raise ValueError( + f"Intraday '{col}' column must be datetime; got dtype {df[col].dtype}." + ) + if (df["available_time"] < df["bar_end"]).any(): + raise ValueError( + "Intraday available_time must be >= bar_end " + "(PIT data lag cannot be negative)." + ) + if (df["bar_start"] >= df["bar_end"]).any(): + raise ValueError( + "Intraday bar_start must be < bar_end (a bar spans a positive interval)." + ) diff --git a/data/feed/tushare_intraday.py b/data/feed/tushare_intraday.py new file mode 100644 index 0000000..1be0987 --- /dev/null +++ b/data/feed/tushare_intraday.py @@ -0,0 +1,208 @@ +"""TushareIntradayFeed: 1min bars from tushare ``stk_mins`` (raw, PIT-ready). + +A SEPARATE feed from :class:`data.feed.tushare_feed.TushareFeed` — it does NOT +touch the daily ``get_bars`` path and does not compute factors or aggregate to +daily. It pulls raw minute bars and maps them onto the intraday schema +(:mod:`data.clean.intraday_schema`); minute-level PIT time fields (bar_start / +bar_end / available_time) are derived there. + +Token handling mirrors the other tushare feeds: read from the EXTERNAL json +config (never hardcoded, never committed), handed straight to the SDK, never +printed/logged/exposed via repr. The client is built lazily, so constructing +this class needs no network and no credentials. + +Field mapping (tushare ``stk_mins`` with ``freq=1min`` -> intraday schema):: + + ts_code -> symbol + trade_time -> source_trade_time (kept for audit) + time (== bar_end) + vol -> volume + open/high/low/close/amount -> same + +The upstream ``stk_mins`` may return rows in reverse-chronological order and has +a single-call row cap; the schema normalizer sorts and deduplicates. Bulk / +multi-day windowing and a read-through cache are deferred (stages I2+); this feed +is the minimal network-free-testable raw shape. Coarser intraday bars +(5/15/30/60min) are intentionally NOT fetched here; they are derived from cached +1min bars in a later clean/resample layer. +""" + +from __future__ import annotations + +import pandas as pd + +from data.clean.intraday_schema import ( + INTRADAY_CORE_COLUMNS, + RAW_INTRADAY_FREQ, + empty_intraday_bars, + ensure_raw_intraday_freq, + normalize_intraday_bars, +) +from data.feed.secret import read_token +from data.feed.throttle import request_with_retry + +# tushare raw column -> intraday-schema column. +_FIELD_MAP: dict[str, str] = { + "ts_code": "symbol", + "vol": "volume", +} + + +class TushareIntradayFeed: + """Loads raw 1min bars from tushare ``stk_mins``.""" + + def __init__( + self, + secret_file: str, + 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 + # SEC-004: per-minute call cap + retry on transient errors. stk_mins has + # an exact per-minute ceiling that was NOT stress-tested; batch use must + # set a conservative rate_limit (see docs/data/tushare_permissions.md). + self._rate_limit = rate_limit + self._max_retries = max(1, int(max_retries)) + self._pro = None # lazily built tushare pro client + # I2: optional read-through 1min cache (TushareIntradayCache). None keeps + # the historical direct per-symbol fetch path EXACTLY; only an opted-in + # caller injects one. The cache stores RAW 1min bars only; normalization + # still runs downstream, unchanged. + self._cache = cache + + # -- secret handling / client ------------------------------------------ # + def _client(self): + """Build (once) and return the tushare pro client. Lazy + no logging.""" + if self._pro is None: + import tushare as ts + + # Do NOT log/print the token. Hand it directly to the SDK. + self._pro = ts.pro_api(read_token(self._secret_file, self._token_key)) + return self._pro + + # -- rate limit + retry (SEC-004) -------------------------------------- # + def _call(self, fn, **kwargs): + """Invoke a tushare endpoint with the shared retry + per-minute throttle.""" + return request_with_retry( + fn, + max_retries=self._max_retries, + rate_limit=self._rate_limit, + **kwargs, + ) + + # -- intraday API ------------------------------------------------------- # + def get_minutes( + self, + symbols: list[str], + start: str, + end: str, + *, + freq: str = RAW_INTRADAY_FREQ, + data_lag: str = "1min", + ) -> pd.DataFrame: + """Return normalized raw 1min bars for ``symbols`` over [start, end]. + + ``start``/``end`` are minute-window bounds — pass full datetime strings + (e.g. ``"2024-01-02 09:30:00"``); a date-only value resolves to that + day's 00:00:00. ``freq`` is kept as an explicit raw-frequency guard and + must be ``"1min"``. Coarser bars are derived from cached 1min data. + ``data_lag`` sets ``available_time = bar_end + data_lag`` for PIT safety. + An empty upstream return yields a schema-shaped empty frame (not an error). + """ + if not symbols: + raise ValueError( + "TushareIntradayFeed.get_minutes requires a non-empty symbol list." + ) + ensure_raw_intraday_freq(freq) + + if self._cache is not None: + return self._get_minutes_cached(symbols, start, end, freq, data_lag) + + pro = self._client() + start_fmt = self._fmt_dt(start) + end_fmt = self._fmt_dt(end) + + frames: list[pd.DataFrame] = [] + for symbol in symbols: + raw = self._call( + pro.stk_mins, + ts_code=symbol, + freq=freq, + start_date=start_fmt, + end_date=end_fmt, + ) + if raw is None or len(raw) == 0: + continue + frames.append(self._to_canonical(raw)) + + if not frames: + return empty_intraday_bars() + + combined = pd.concat(frames, ignore_index=True) + return normalize_intraday_bars(combined, freq=freq, data_lag=data_lag) + + # -- read-through cache path (I2) --------------------------------------- # + def _get_minutes_cached(self, symbols, start, end, freq, data_lag) -> pd.DataFrame: + """Build the panel from the read-through 1min cache. + + The cache fetches only uncovered trading-day gaps (a warm identical + request makes zero ``stk_mins`` calls); the per-symbol throttle + retry + stay HERE via the ``self._call`` closure. The cache returns the SAME + columns as ``_to_canonical``, so after ``normalize_intraday_bars`` the + result is byte-identical to the direct path. + """ + combined = self._cache.stk_mins_1min( + symbols, start, end, self._stk_mins_fetch(), freq=freq + ) + if combined.empty: + return empty_intraday_bars() + return normalize_intraday_bars(combined, freq=freq, data_lag=data_lag) + + def _stk_mins_fetch(self): + """`(symbol, start_dt, end_dt) -> raw stk_mins frame` (1min, lazy client). + + The tushare client is built lazily INSIDE the closure (on the first gap + actually fetched), so a fully-covered warm cache run reads no token and + constructs no client. ``start_dt``/``end_dt`` are 'YYYY-MM-DD HH:MM:SS'. + """ + + def fetch(symbol, start_dt, end_dt): + return self._call( + self._client().stk_mins, + ts_code=symbol, + freq=RAW_INTRADAY_FREQ, + start_date=start_dt, + end_date=end_dt, + ) + + return fetch + + # -- mapping helpers ---------------------------------------------------- # + def _to_canonical(self, raw: pd.DataFrame) -> pd.DataFrame: + """Rename tushare columns and expose ``time`` + audit ``source_trade_time``. + + Returns columns (time, symbol, OHLCV, source_trade_time); the schema + normalizer derives freq/bar_start/bar_end/available_time. ``time`` keeps + minute precision — it is NOT normalized to midnight. + """ + df = raw.rename(columns=_FIELD_MAP).copy() + df["source_trade_time"] = df["trade_time"].astype(str) + df["time"] = pd.to_datetime(df["trade_time"]) + df["symbol"] = df["symbol"].astype(str) + + keep = ["time", "symbol", *INTRADAY_CORE_COLUMNS, "source_trade_time"] + present = [c for c in keep if c in df.columns] + return df[present] + + @staticmethod + def _fmt_dt(value: str) -> str: + """Format a window bound as the ``'YYYY-MM-DD HH:MM:SS'`` stk_mins expects.""" + return pd.Timestamp(value).strftime("%Y-%m-%d %H:%M:%S") + + def __repr__(self) -> str: # never leak the token + return ( + f"TushareIntradayFeed(secret_file={self._secret_file!r}, " + f"token_key={self._token_key!r}, rate_limit={self._rate_limit!r})" + ) diff --git a/runtime/intraday_execution.py b/runtime/intraday_execution.py new file mode 100644 index 0000000..9a48158 --- /dev/null +++ b/runtime/intraday_execution.py @@ -0,0 +1,227 @@ +"""Intraday tail-rebalance execution semantics (I4) — SEPARATE from daily. + +The daily backtest decides at T's close and measures holding returns +close(T)->close(T+1) (``runtime/backtest`` + ``settle(holding_returns)``). A +14:50 tail rebalance is a DIFFERENT event model and must NOT silently reuse that: + + signal cutoff T 14:50:00 — the decision may use only bars with + available_time <= cutoff (I3's job). + execution timestamp T 14:51:00 — the trade fills at the NEXT minute bar + (``next_minute_close``), i.e. on data that + is INTENTIONALLY after the signal cutoff. + holding period exec(T) -> exec(T_next) — return is measured from this + rebalance's execution price to the next + rebalance's execution price, NEVER + close-to-close. + +This module is a runtime/execution skeleton: pure functions + small frozen +dataclasses turning (target weights per rebalance date, 1min bars, exec config) +into execution prices, execution-to-execution holding returns, and an explainable +blocked log. It does NOT touch factors/alpha math, the daily backtest, or the +config models; a missing minute bar / NaN price / no bar in the execution window +produces a BLOCKED record — never a silent daily-close fallback. Daily EOD data +(``daily_basic`` etc.) is by construction not consulted here: only intraday bars +are read, so T-day EOD values cannot leak into a 14:50 decision/execution. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +import numpy as np +import pandas as pd + +from data.clean.intraday_schema import validate_intraday_bars + +# Only the conservative first model is implemented; the others are declared in +# the roadmap (02_minute_pit_semantics.md) and rejected with a clear message +# until they land, so a config can never silently fall back to a wrong model. +SUPPORTED_EXECUTION_MODELS: tuple[str, ...] = ("next_minute_close",) +_FUTURE_EXECUTION_MODELS: tuple[str, ...] = ("tail_vwap", "closing_call_proxy") + +# blocked reasons (explainable; never a silent daily-close substitution). +REASON_NO_BAR = "no_execution_bar" +REASON_MISSING_PRICE = "missing_price" + + +@dataclass(frozen=True) +class IntradayExecutionConfig: + """A strategy's tail-rebalance execution declaration. + + ``decision_time`` is the signal cutoff (T 14:50). ``data_lag`` documents the + feature-availability lag consumed by the I3 cutoff (``available_time = + bar_end + data_lag``); it is NOT re-applied to execution pricing. Execution + fills at the earliest 1min bar whose ``bar_end`` falls in + ``[execution_window_start, execution_window_end]`` (default 14:51–14:56:59). + """ + + decision_time: str = "14:50:00" + data_lag: str = "1min" + execution_model: str = "next_minute_close" + execution_window: tuple[str, str] = ("14:51:00", "14:56:59") + + def __post_init__(self) -> None: + if self.execution_model not in SUPPORTED_EXECUTION_MODELS: + future = ( + " (planned, not yet implemented)" + if self.execution_model in _FUTURE_EXECUTION_MODELS + else "" + ) + raise ValueError( + f"Unsupported execution_model {self.execution_model!r}{future}; " + f"supported: {SUPPORTED_EXECUTION_MODELS}." + ) + start = pd.Timedelta(self.execution_window[0]) + end = pd.Timedelta(self.execution_window[1]) + if start > pd.Timedelta(self.decision_time) and start <= end: + return + raise ValueError( + "execution_window must satisfy decision_time < start <= end " + f"(got decision={self.decision_time}, window={self.execution_window})." + ) + + +@dataclass(frozen=True) +class ExecutionFill: + """One symbol's execution outcome on one rebalance date (immutable).""" + + symbol: str + date: pd.Timestamp + exec_time: pd.Timestamp | None + exec_price: float | None + blocked: bool + reason: str | None = None + + +@dataclass(frozen=True) +class TailRebalanceResult: + """Outcome of an execution-priced tail-rebalance simulation (immutable).""" + + period_returns: pd.Series # entry date -> gross portfolio return + exec_prices: pd.DataFrame # (date x symbol) execution prices + holding_returns: pd.DataFrame # (entry date x symbol) exec-to-exec return + fills: list[ExecutionFill] = field(default_factory=list) + + @property + def blocked(self) -> list[ExecutionFill]: + """All blocked fills (no execution bar / missing price).""" + return [f for f in self.fills if f.blocked] + + +def resolve_fill( + symbol: str, + date: pd.Timestamp, + day_bars: pd.DataFrame, + cfg: IntradayExecutionConfig, +) -> ExecutionFill: + """Resolve the execution fill for ``symbol`` on ``date`` from its 1min bars. + + ``day_bars`` are the 1min bars for THIS symbol on THIS date (must carry + ``bar_end`` + ``close``). ``next_minute_close`` takes the EARLIEST bar whose + ``bar_end`` is in the execution window; a missing bar or NaN close yields a + BLOCKED fill with an explainable reason (never a daily-close fallback). + """ + day = pd.Timestamp(date).normalize() + win_start = day + pd.Timedelta(cfg.execution_window[0]) + win_end = day + pd.Timedelta(cfg.execution_window[1]) + + if day_bars is None or day_bars.empty: + return ExecutionFill(symbol, day, None, None, True, REASON_NO_BAR) + cand = day_bars[ + (day_bars["bar_end"] >= win_start) & (day_bars["bar_end"] <= win_end) + ].sort_values("bar_end") + if cand.empty: + return ExecutionFill(symbol, day, None, None, True, REASON_NO_BAR) + + bar = cand.iloc[0] # next_minute_close: earliest available bar in the window + price = bar["close"] + if pd.isna(price): + return ExecutionFill( + symbol, day, pd.Timestamp(bar["bar_end"]), None, True, REASON_MISSING_PRICE + ) + return ExecutionFill( + symbol, day, pd.Timestamp(bar["bar_end"]), float(price), False, None + ) + + +def build_execution_prices( + bars: pd.DataFrame, + rebalance_dates: list[pd.Timestamp], + symbols: list[str], + cfg: IntradayExecutionConfig, +) -> tuple[pd.DataFrame, list[ExecutionFill]]: + """Execution price matrix (date x symbol) + the per-(date, symbol) fill log. + + Reads ONLY intraday bars (no daily/EOD data). A blocked fill leaves a NaN in + the matrix and an explained entry in the returned fills list. + """ + validate_intraday_bars(bars) + dates = [pd.Timestamp(d).normalize() for d in rebalance_dates] + syms = [str(s) for s in symbols] + + work = bars.reset_index() + work["date"] = work["bar_end"].dt.normalize() + + fills: list[ExecutionFill] = [] + prices = pd.DataFrame(index=pd.Index(dates, name="date"), columns=syms, dtype=float) + for date in dates: + on_day = work[work["date"] == date] + for sym in syms: + day_bars = on_day[on_day["symbol"] == sym] + fill = resolve_fill(sym, date, day_bars, cfg) + fills.append(fill) + prices.loc[date, sym] = ( + np.nan if fill.exec_price is None else fill.exec_price + ) + return prices, fills + + +def simulate_tail_rebalance( + weights_by_date: dict[pd.Timestamp, pd.Series], + bars: pd.DataFrame, + cfg: IntradayExecutionConfig | None = None, +) -> TailRebalanceResult: + """Simulate a tail rebalance, pricing fills at execution timestamps. + + ``weights_by_date`` maps each rebalance date to its target weights + (symbol-indexed). For each period [T, T_next] the per-symbol holding return is + ``exec_price(T_next) / exec_price(T) - 1`` (execution-to-execution, NOT + close-to-close), and the gross portfolio return is ``sum_sym w_T * r``. A + symbol missing either execution price is BLOCKED — excluded from that period's + return (its weight earns nothing, i.e. cash) and recorded in the fills log. + The final rebalance date has no subsequent period and no return. + """ + cfg = cfg or IntradayExecutionConfig() + dates = sorted(weights_by_date) + symbols = sorted({str(s) for w in weights_by_date.values() for s in w.index}) + prices, fills = build_execution_prices(bars, dates, symbols, cfg) + + period_returns: dict[pd.Timestamp, float] = {} + holding_rows: dict[pd.Timestamp, dict[str, float]] = {} + for i in range(len(dates) - 1): + entry, exit_ = dates[i], dates[i + 1] + weights = weights_by_date[dates[i]] + p_entry, p_exit = prices.loc[entry], prices.loc[exit_] + gross = 0.0 + per_symbol: dict[str, float] = {} + for sym, wt in weights.items(): + a, b = p_entry.get(str(sym)), p_exit.get(str(sym)) + if pd.notna(a) and pd.notna(b) and a != 0: + r = float(b) / float(a) - 1.0 + per_symbol[str(sym)] = r + gross += float(wt) * r + # else: blocked at entry or exit -> excluded (recorded in `fills`) + period_returns[entry] = gross + holding_rows[entry] = per_symbol + + pr = pd.Series(period_returns, dtype=float) + pr.index.name = "date" + holding = pd.DataFrame(holding_rows).T if holding_rows else pd.DataFrame() + if not holding.empty: + holding.index.name = "date" + return TailRebalanceResult( + period_returns=pr, + exec_prices=prices, + holding_returns=holding, + fills=fills, + ) diff --git a/tests/test_intraday_aggregate.py b/tests/test_intraday_aggregate.py new file mode 100644 index 0000000..f78112b --- /dev/null +++ b/tests/test_intraday_aggregate.py @@ -0,0 +1,229 @@ +"""Intraday -> daily PIT aggregation tests (I3): cutoff, leakage, isolation.""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from data.clean.intraday_aggregate import ( + asof_daily_features, + resample_intraday_bars, +) +from data.clean.intraday_schema import ( + empty_intraday_bars, + normalize_intraday_bars, + validate_intraday_bars, +) + +_RET = "intraday_ret_0930_1450" +_VOL = "intraday_realized_vol_0930_1450" +_VWAP = "intraday_vwap_0930_1450" +_L30 = "intraday_last30m_ret_1420_1450" + + +def _norm(rows, data_lag="1min"): + """rows = [(time_str, symbol, close), ...] -> normalized 1min bars.""" + cl = [r[2] for r in rows] + df = pd.DataFrame( + { + "time": pd.to_datetime([r[0] for r in rows]), + "symbol": [r[1] for r in rows], + "open": cl, + "high": [c + 0.5 for c in cl], + "low": [c - 0.5 for c in cl], + "close": cl, + "volume": [100.0] * len(rows), + "amount": [10.0 * c for c in cl], + } + ) + return normalize_intraday_bars(df, freq="1min", data_lag=data_lag) + + +_DAY = "2024-01-02" + + +def _one_symbol_day(extra_after_cutoff=False): + rows = [ + (f"{_DAY} 09:31:00", "000001.SZ", 10.0), + (f"{_DAY} 09:32:00", "000001.SZ", 10.5), + (f"{_DAY} 14:00:00", "000001.SZ", 11.0), + (f"{_DAY} 14:30:00", "000001.SZ", 11.5), + (f"{_DAY} 14:48:00", "000001.SZ", 12.0), + (f"{_DAY} 14:49:00", "000001.SZ", 12.5), + ] + if extra_after_cutoff: + rows += [ + (f"{_DAY} 14:51:00", "000001.SZ", 99.0), + (f"{_DAY} 14:55:00", "000001.SZ", 99.0), + ] + return rows + + +# --------------------------------------------------------------------------- # +# basic shape + column names +# --------------------------------------------------------------------------- # +def test_columns_encode_cutoff_and_index_is_daily(): + out = asof_daily_features(_norm(_one_symbol_day())) + assert list(out.columns) == [_RET, _VOL, _VWAP, _L30] + assert list(out.index.names) == ["date", "symbol"] + assert len(out) == 1 + (date, sym) = out.index[0] + assert sym == "000001.SZ" + assert date == pd.Timestamp("2024-01-02") # midnight, not a minute timestamp + assert date.hour == 0 and date.minute == 0 + + +def test_ret_value_open_to_last_visible(): + out = asof_daily_features(_norm(_one_symbol_day())) + # last visible close 12.5 (14:49) / first open 10.0 - 1 + assert out[_RET].iloc[0] == pytest.approx(12.5 / 10.0 - 1.0) + # last30m: ref = last bar with bar_end <= 14:20 -> 14:00 close 11.0; last 12.5 + assert out[_L30].iloc[0] == pytest.approx(12.5 / 11.0 - 1.0) + assert out[_VOL].iloc[0] > 0 + + +# --------------------------------------------------------------------------- # +# PIT: cutoff leakage + availability +# --------------------------------------------------------------------------- # +def test_post_cutoff_bars_do_not_leak(): + base = asof_daily_features(_norm(_one_symbol_day(extra_after_cutoff=False))) + # identical pre-14:50 bars, plus post-14:50 bars with garbage prices + perturbed = asof_daily_features(_norm(_one_symbol_day(extra_after_cutoff=True))) + pd.testing.assert_frame_equal(base, perturbed) + + +def test_delayed_availability_excludes_bar(): + bars = _norm(_one_symbol_day()) + base = asof_daily_features(bars) + # push the 14:00 bar's availability past the 14:50 cutoff + delayed = bars.copy() + mask = delayed["bar_end"] == pd.Timestamp("2024-01-02 14:00:00") + delayed.loc[mask, "available_time"] = pd.Timestamp("2024-01-02 15:00:00") + got = asof_daily_features(delayed) + # equals computing on bars with that bar physically removed + dropped = bars[bars["bar_end"] != pd.Timestamp("2024-01-02 14:00:00")] + expected = asof_daily_features(dropped) + pd.testing.assert_frame_equal(got, expected) + # and the exclusion actually changed something (vwap sums fewer bars) + assert got[_VWAP].iloc[0] != base[_VWAP].iloc[0] + + +def test_all_bars_after_cutoff_returns_empty(): + rows = [ + (f"{_DAY} 14:51:00", "000001.SZ", 10.0), + (f"{_DAY} 14:52:00", "000001.SZ", 11.0), + ] + out = asof_daily_features(_norm(rows)) + assert len(out) == 0 + assert list(out.index.names) == ["date", "symbol"] + assert list(out.columns) == [_RET, _VOL, _VWAP, _L30] + + +# --------------------------------------------------------------------------- # +# multi-symbol / multi-day isolation +# --------------------------------------------------------------------------- # +def test_multi_symbol_multi_day_no_contamination(): + rows = [ + ("2024-01-02 09:31:00", "000001.SZ", 10.0), + ("2024-01-02 09:32:00", "000001.SZ", 11.0), # ret 0.10 + ("2024-01-03 09:31:00", "000001.SZ", 20.0), + ("2024-01-03 09:32:00", "000001.SZ", 22.0), # ret 0.10 + ("2024-01-02 09:31:00", "000002.SZ", 5.0), + ("2024-01-02 09:32:00", "000002.SZ", 6.0), # ret 0.20 + ] + out = asof_daily_features(_norm(rows)) + assert len(out) == 3 + assert out.loc[(pd.Timestamp("2024-01-02"), "000001.SZ"), _RET] == pytest.approx(0.10) + assert out.loc[(pd.Timestamp("2024-01-03"), "000001.SZ"), _RET] == pytest.approx(0.10) + assert out.loc[(pd.Timestamp("2024-01-02"), "000002.SZ"), _RET] == pytest.approx(0.20) + + +# --------------------------------------------------------------------------- # +# empty input / feature selection +# --------------------------------------------------------------------------- # +def test_empty_input_returns_schema_shaped_empty(): + out = asof_daily_features(empty_intraday_bars()) + assert len(out) == 0 + assert list(out.index.names) == ["date", "symbol"] + assert list(out.columns) == [_RET, _VOL, _VWAP, _L30] + + +def test_feature_subset_and_unknown(): + out = asof_daily_features(_norm(_one_symbol_day()), features=["vwap"]) + assert list(out.columns) == [_VWAP] + with pytest.raises(ValueError, match="Unknown intraday feature"): + asof_daily_features(_norm(_one_symbol_day()), features=["bogus"]) + + +def test_custom_decision_time_relabels_columns(): + out = asof_daily_features(_norm(_one_symbol_day()), decision_time="14:30:00") + assert "intraday_ret_0930_1430" in out.columns + assert "intraday_last30m_ret_1400_1430" in out.columns + # cutoff 14:30: the 14:30 bar's availability is 14:31 (1min lag) > 14:30, so it + # is excluded too -> last visible bar is 14:00 (close 11.0). + assert out["intraday_ret_0930_1430"].iloc[0] == pytest.approx(11.0 / 10.0 - 1.0) + + +# --------------------------------------------------------------------------- # +# derived coarse bars: available_time = max(source 1min) +# --------------------------------------------------------------------------- # +def test_resample_available_time_is_source_max(): + rows = [ + (f"{_DAY} 09:31:00", "000001.SZ", 10.0), + (f"{_DAY} 09:32:00", "000001.SZ", 11.0), + (f"{_DAY} 09:33:00", "000001.SZ", 12.0), + (f"{_DAY} 09:34:00", "000001.SZ", 9.0), + (f"{_DAY} 09:35:00", "000001.SZ", 13.0), + ] + bars = _norm(rows, data_lag="1min") + coarse = resample_intraday_bars(bars, "5min") + validate_intraday_bars(coarse) + assert len(coarse) == 1 + row = coarse.iloc[0] + # one 5min bar ending 09:35 + assert coarse.index.get_level_values("time")[0] == pd.Timestamp("2024-01-02 09:35:00") + assert row["bar_end"] == pd.Timestamp("2024-01-02 09:35:00") + assert row["bar_start"] == pd.Timestamp("2024-01-02 09:30:00") + assert row["freq"] == "5min" + # OHLC aggregation + assert row["open"] == 10.0 # first + assert row["close"] == 13.0 # last + assert row["high"] == 13.0 + 0.5 # max high + assert row["low"] == 9.0 - 0.5 # min low + assert row["volume"] == 500.0 # 5 * 100 + # availability inherits the MAX source 1min available_time (09:35 + 1min lag) + src_max = bars["available_time"].max() + assert row["available_time"] == src_max == pd.Timestamp("2024-01-02 09:36:00") + + +def test_resample_keeps_symbols_separate(): + rows = [ + (f"{_DAY} 09:31:00", "000001.SZ", 10.0), + (f"{_DAY} 09:32:00", "000001.SZ", 11.0), + (f"{_DAY} 09:31:00", "000002.SZ", 50.0), + (f"{_DAY} 09:32:00", "000002.SZ", 52.0), + ] + coarse = resample_intraday_bars(_norm(rows), "5min") + syms = sorted(set(coarse.index.get_level_values("symbol"))) + assert syms == ["000001.SZ", "000002.SZ"] + # each symbol's coarse close is its own last 1min close + a = coarse.xs("000001.SZ", level="symbol").iloc[0] + b = coarse.xs("000002.SZ", level="symbol").iloc[0] + assert a["close"] == 11.0 and b["close"] == 52.0 + + +def test_resample_rejects_unsupported_freq(): + with pytest.raises(ValueError, match="freq"): + resample_intraday_bars(_norm(_one_symbol_day()), "7min") + + +def test_resample_realized_vol_uses_log_returns(): + # sanity: a known 2-bar vol so the feature is not silently zero + rows = [ + (f"{_DAY} 09:31:00", "000001.SZ", 10.0), + (f"{_DAY} 09:32:00", "000001.SZ", 11.0), + ] + out = asof_daily_features(_norm(rows)) + expected = abs(np.log(11.0 / 10.0)) + assert out[_VOL].iloc[0] == pytest.approx(expected) diff --git a/tests/test_intraday_execution.py b/tests/test_intraday_execution.py new file mode 100644 index 0000000..b46f74f --- /dev/null +++ b/tests/test_intraday_execution.py @@ -0,0 +1,208 @@ +"""Intraday tail-rebalance execution tests (I4): cutoff vs exec, exec-to-exec.""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from data.clean.intraday_aggregate import asof_daily_features +from data.clean.intraday_schema import normalize_intraday_bars +from runtime.intraday_execution import ( + REASON_MISSING_PRICE, + REASON_NO_BAR, + ExecutionFill, + IntradayExecutionConfig, + build_execution_prices, + resolve_fill, + simulate_tail_rebalance, +) + + +def _norm(rows, data_lag="1min"): + """rows = [(time_str, symbol, close), ...] -> normalized 1min bars.""" + cl = [r[2] for r in rows] + df = pd.DataFrame( + { + "time": pd.to_datetime([r[0] for r in rows]), + "symbol": [r[1] for r in rows], + "open": cl, + "high": [c + 0.5 if pd.notna(c) else c for c in cl], + "low": [c - 0.5 if pd.notna(c) else c for c in cl], + "close": cl, + "volume": [100.0] * len(rows), + "amount": [10.0 * c if pd.notna(c) else c for c in cl], + } + ) + return normalize_intraday_bars(df, freq="1min", data_lag=data_lag) + + +def _day(bars, sym, date="2024-01-02"): + w = bars.reset_index() + w["date"] = w["bar_end"].dt.normalize() + return w[(w["symbol"] == sym) & (w["date"] == pd.Timestamp(date))] + + +# --------------------------------------------------------------------------- # +# config validation +# --------------------------------------------------------------------------- # +def test_unsupported_execution_model_rejected(): + with pytest.raises(ValueError, match="execution_model"): + IntradayExecutionConfig(execution_model="tail_vwap") + + +def test_bad_execution_window_rejected(): + # window start must be strictly after the decision time + with pytest.raises(ValueError, match="execution_window"): + IntradayExecutionConfig(execution_window=("14:49:00", "14:56:59")) + + +# --------------------------------------------------------------------------- # +# signal cutoff vs execution timestamp (the core separation) +# --------------------------------------------------------------------------- # +def test_signal_cutoff_and_execution_timestamp_are_separate(): + rows = [ + ("2024-01-02 09:31:00", "000001.SZ", 10.0), + ("2024-01-02 14:49:00", "000001.SZ", 11.0), + ("2024-01-02 14:51:00", "000001.SZ", 12.0), # execution bar (post-cutoff) + ("2024-01-02 14:55:00", "000001.SZ", 99.0), + ] + bars = _norm(rows) + fill = resolve_fill( + "000001.SZ", pd.Timestamp("2024-01-02"), + _day(bars, "000001.SZ"), IntradayExecutionConfig(), + ) + # execution INTENTIONALLY uses the 14:51 bar, which is AFTER the 14:50 cutoff + assert not fill.blocked + assert fill.exec_time == pd.Timestamp("2024-01-02 14:51:00") + assert fill.exec_price == 12.0 + + # the same 14:51 (and 14:55) bars are NOT visible to the 14:50 signal: + # perturbing them leaves the I3 daily feature byte-identical. + base = asof_daily_features(bars) + perturbed_rows = rows[:2] + [ + ("2024-01-02 14:51:00", "000001.SZ", -123.0), + ("2024-01-02 14:55:00", "000001.SZ", -456.0), + ] + perturbed = asof_daily_features(_norm(perturbed_rows)) + pd.testing.assert_frame_equal(base, perturbed) + + +# --------------------------------------------------------------------------- # +# next_minute_close model +# --------------------------------------------------------------------------- # +def test_next_minute_close_uses_1451_bar(): + rows = [ + ("2024-01-02 14:51:00", "000001.SZ", 10.0), + ("2024-01-02 14:52:00", "000001.SZ", 10.5), + ] + fill = resolve_fill( + "000001.SZ", pd.Timestamp("2024-01-02"), + _day(_norm(rows), "000001.SZ"), IntradayExecutionConfig(), + ) + assert fill.exec_time == pd.Timestamp("2024-01-02 14:51:00") + assert fill.exec_price == 10.0 + + +def test_missing_1451_uses_first_bar_in_window(): + rows = [ + ("2024-01-02 14:53:00", "000001.SZ", 11.0), # no 14:51/14:52 + ("2024-01-02 14:55:00", "000001.SZ", 11.5), + ] + fill = resolve_fill( + "000001.SZ", pd.Timestamp("2024-01-02"), + _day(_norm(rows), "000001.SZ"), IntradayExecutionConfig(), + ) + assert fill.exec_time == pd.Timestamp("2024-01-02 14:53:00") # first in window + assert fill.exec_price == 11.0 + + +def test_no_bar_in_window_is_blocked(): + rows = [("2024-01-02 14:00:00", "000001.SZ", 10.0)] # nothing in 14:51-14:56:59 + fill = resolve_fill( + "000001.SZ", pd.Timestamp("2024-01-02"), + _day(_norm(rows), "000001.SZ"), IntradayExecutionConfig(), + ) + assert fill.blocked + assert fill.reason == REASON_NO_BAR + assert fill.exec_price is None + + +def test_nan_price_is_blocked_missing_price(): + rows = [("2024-01-02 14:51:00", "000001.SZ", np.nan)] + fill = resolve_fill( + "000001.SZ", pd.Timestamp("2024-01-02"), + _day(_norm(rows), "000001.SZ"), IntradayExecutionConfig(), + ) + assert fill.blocked + assert fill.reason == REASON_MISSING_PRICE + assert fill.exec_time == pd.Timestamp("2024-01-02 14:51:00") # the bar existed + + +# --------------------------------------------------------------------------- # +# holding return = exec-to-exec, NOT close-to-close +# --------------------------------------------------------------------------- # +def test_holding_return_is_execution_to_execution_not_close(): + rows = [ + # T: exec(14:51)=10.0, daily close(15:00)=10.5 + ("2024-01-02 14:51:00", "000001.SZ", 10.0), + ("2024-01-02 15:00:00", "000001.SZ", 10.5), + # T_next: exec(14:51)=12.0, daily close(15:00)=11.0 + ("2024-01-03 14:51:00", "000001.SZ", 12.0), + ("2024-01-03 15:00:00", "000001.SZ", 11.0), + ] + bars = _norm(rows) + weights = { + pd.Timestamp("2024-01-02"): pd.Series({"000001.SZ": 1.0}), + pd.Timestamp("2024-01-03"): pd.Series({"000001.SZ": 1.0}), + } + res = simulate_tail_rebalance(weights, bars) + entry = pd.Timestamp("2024-01-02") + exec_to_exec = 12.0 / 10.0 - 1.0 # 0.20 + close_to_close = 11.0 / 10.5 - 1.0 # ~0.0476 + assert res.period_returns[entry] == pytest.approx(exec_to_exec) + assert res.period_returns[entry] != pytest.approx(close_to_close) + assert res.holding_returns.loc[entry, "000001.SZ"] == pytest.approx(0.20) + # exec prices are the 14:51 bars, not the 15:00 closes + assert res.exec_prices.loc[entry, "000001.SZ"] == 10.0 + + +# --------------------------------------------------------------------------- # +# portfolio period return excludes blocked symbols (no silent fallback) +# --------------------------------------------------------------------------- # +def test_blocked_symbol_excluded_and_logged(): + rows = [ + # A: tradable both days (entry 10 -> exit 11, r=0.10) + ("2024-01-02 14:51:00", "000001.SZ", 10.0), + ("2024-01-03 14:51:00", "000001.SZ", 11.0), + # B: entry ok at T, but NO bar in the window at T_next -> blocked at exit + ("2024-01-02 14:51:00", "000002.SZ", 20.0), + ("2024-01-03 14:00:00", "000002.SZ", 25.0), # outside execution window + ] + bars = _norm(rows) + weights = { + pd.Timestamp("2024-01-02"): pd.Series({"000001.SZ": 0.5, "000002.SZ": 0.5}), + pd.Timestamp("2024-01-03"): pd.Series({"000001.SZ": 1.0}), + } + res = simulate_tail_rebalance(weights, bars) + entry = pd.Timestamp("2024-01-02") + # only A contributes: 0.5 * 0.10 ; B excluded (its weight earns nothing) + assert res.period_returns[entry] == pytest.approx(0.5 * 0.10) + blocked_syms = {(f.symbol, f.reason) for f in res.blocked} + assert ("000002.SZ", REASON_NO_BAR) in blocked_syms + + +def test_build_execution_prices_matrix_and_log(): + rows = [ + ("2024-01-02 14:51:00", "000001.SZ", 10.0), + ("2024-01-02 14:00:00", "000002.SZ", 20.0), # not in window -> blocked NaN + ] + bars = _norm(rows) + prices, fills = build_execution_prices( + bars, [pd.Timestamp("2024-01-02")], ["000001.SZ", "000002.SZ"], + IntradayExecutionConfig(), + ) + assert prices.loc[pd.Timestamp("2024-01-02"), "000001.SZ"] == 10.0 + assert pd.isna(prices.loc[pd.Timestamp("2024-01-02"), "000002.SZ"]) + assert any(f.blocked and f.symbol == "000002.SZ" for f in fills) + assert all(isinstance(f, ExecutionFill) for f in fills) diff --git a/tests/test_intraday_schema.py b/tests/test_intraday_schema.py new file mode 100644 index 0000000..e1b5792 --- /dev/null +++ b/tests/test_intraday_schema.py @@ -0,0 +1,178 @@ +"""Intraday schema tests — minute-precision time, PIT derivation, dedup, sort.""" + +from __future__ import annotations + +import pandas as pd +import pytest + +from data.clean.intraday_schema import ( + empty_intraday_bars, + normalize_intraday_bars, + validate_intraday_bars, +) + + +def _bars(times, symbol="000001.SZ", close=10.0): + """Build a raw intraday input frame (columns time, symbol, OHLCV).""" + return pd.DataFrame( + { + "time": pd.to_datetime(list(times)), + "symbol": symbol, + "open": close, + "high": close, + "low": close, + "close": close, + "volume": 1000.0, + "amount": 10000.0, + } + ) + + +def test_time_preserves_minute_precision(): + # The daily contract normalizes to midnight; intraday must NOT. + out = normalize_intraday_bars( + _bars(["2024-01-02 09:31:00", "2024-01-02 09:32:00"]), freq="1min" + ) + times = out.index.get_level_values("time") + assert times[0] == pd.Timestamp("2024-01-02 09:31:00") + assert times[0].hour == 9 and times[0].minute == 31 # not midnight + + +def test_raw_1min_bar_window_and_available_time(): + out = normalize_intraday_bars( + _bars(["2024-01-02 09:31:00"]), freq="1min", data_lag="1min" + ) + row = out.iloc[0] + assert row["bar_end"] == pd.Timestamp("2024-01-02 09:31:00") + assert row["bar_start"] == pd.Timestamp("2024-01-02 09:30:00") # end - 1min + assert row["available_time"] == pd.Timestamp("2024-01-02 09:32:00") # end + lag + assert row["freq"] == "1min" + + +def test_schema_can_represent_future_derived_coarser_bars(): + out = normalize_intraday_bars( + _bars(["2024-01-02 09:35:00"]), freq="5min", data_lag="1min" + ) + row = out.iloc[0] + assert row["bar_start"] == pd.Timestamp("2024-01-02 09:30:00") + assert row["freq"] == "5min" + + +def test_data_lag_shifts_available_time(): + out = normalize_intraday_bars( + _bars(["2024-01-02 09:31:00"]), freq="1min", data_lag="3min" + ) + assert out.iloc[0]["available_time"] == pd.Timestamp("2024-01-02 09:34:00") + + +def test_reverse_chronological_input_sorted_ascending(): + out = normalize_intraday_bars( + _bars( + [ + "2024-01-02 09:33:00", + "2024-01-02 09:31:00", + "2024-01-02 09:32:00", + ] + ), + freq="1min", + ) + times = list(out.index.get_level_values("time")) + assert times == sorted(times) + + +def test_sorted_by_time_then_symbol(): + df = pd.DataFrame( + { + "time": pd.to_datetime(["2024-01-02 09:31:00"] * 2), + "symbol": ["000002.SZ", "000001.SZ"], + "open": 1.0, + "high": 1.0, + "low": 1.0, + "close": 1.0, + "volume": 1.0, + "amount": 1.0, + } + ) + out = normalize_intraday_bars(df, freq="1min") + assert list(out.index.get_level_values("symbol")) == ["000001.SZ", "000002.SZ"] + + +def test_duplicate_key_last_row_wins(): + df = pd.DataFrame( + { + "time": pd.to_datetime( + ["2024-01-02 09:31:00", "2024-01-02 09:31:00"] + ), + "symbol": ["000001.SZ", "000001.SZ"], + "open": [1.0, 2.0], + "high": [1.0, 2.0], + "low": [1.0, 2.0], + "close": [10.0, 20.0], # later row should win + "volume": [100.0, 200.0], + "amount": [1.0, 2.0], + } + ) + out = normalize_intraday_bars(df, freq="1min") + assert len(out) == 1 + assert out.iloc[0]["close"] == 20.0 + assert out.iloc[0]["volume"] == 200.0 + + +def test_missing_core_column_readable_error(): + df = _bars(["2024-01-02 09:31:00"]).drop(columns=["amount"]) + with pytest.raises(ValueError, match="amount"): + normalize_intraday_bars(df, freq="1min") + + +def test_missing_time_symbol_readable_error(): + df = pd.DataFrame({"open": [1.0]}) + with pytest.raises(ValueError, match="time"): + normalize_intraday_bars(df, freq="1min") + + +def test_unsupported_freq_readable_error(): + with pytest.raises(ValueError, match="freq"): + normalize_intraday_bars(_bars(["2024-01-02 09:31:00"]), freq="2min") + + +def test_source_trade_time_preserved_as_extra(): + df = _bars(["2024-01-02 09:31:00"]) + df["source_trade_time"] = "2024-01-02 09:31:00" + out = normalize_intraday_bars(df, freq="1min") + assert "source_trade_time" in out.columns + + +def test_normalize_does_not_mutate_input(): + df = _bars(["2024-01-02 09:31:00", "2024-01-02 09:30:00"]) + before = df.copy(deep=True) + normalize_intraday_bars(df, freq="1min") + pd.testing.assert_frame_equal(df, before) + + +def test_empty_intraday_bars_schema_and_validate(): + e = empty_intraday_bars() + assert list(e.index.names) == ["time", "symbol"] + for c in ( + "open", + "high", + "low", + "close", + "volume", + "amount", + "freq", + "bar_start", + "bar_end", + "available_time", + ): + assert c in e.columns + assert len(e) == 0 + validate_intraday_bars(e) # must not raise + + +def test_validate_accepts_normalized_and_rejects_raw_frame(): + out = normalize_intraday_bars( + _bars(["2024-01-02 09:31:00", "2024-01-02 09:32:00"]), freq="1min" + ) + validate_intraday_bars(out) # passes + with pytest.raises(ValueError): + validate_intraday_bars(_bars(["2024-01-02 09:31:00"])) # not indexed diff --git a/tests/test_tushare_cache_intraday.py b/tests/test_tushare_cache_intraday.py new file mode 100644 index 0000000..fee1cf3 --- /dev/null +++ b/tests/test_tushare_cache_intraday.py @@ -0,0 +1,269 @@ +"""TushareIntradayCache (I2) tests — fake SDK, network-free, no token. + +Covers cold/warm/partial/empty/failed read-through, idempotent upsert, the +1min-only raw guard, ledger/store schema (no secret), and cached==direct +equality after normalization. +""" + +from __future__ import annotations + +import pandas as pd +import pytest + +from data.cache.intraday_cache import ENDPOINT, TushareIntradayCache +from data.cache.intraday_coverage import ( + INTRADAY_LEDGER_COLUMNS, + IntradayCoverageLedger, +) +from data.cache.intraday_parquet_store import STORED_COLUMNS, IntradayParquetStore +from data.clean.intraday_schema import validate_intraday_bars +from data.feed.tushare_intraday import TushareIntradayFeed + +_FIXED_CLOCK = lambda: pd.Timestamp("2026-06-13 10:00:00") # noqa: E731 + + +# --------------------------------------------------------------------------- # +# fakes +# --------------------------------------------------------------------------- # +class _FetchRecorder: + """A window-aware fake fetch callable; records the (symbol, start, end) calls.""" + + def __init__(self, catalog: dict[str, list[tuple[str, float]]]): + self.catalog = catalog + self.calls: list[tuple[str, str, str]] = [] + + def __call__(self, symbol, start_dt, end_dt): + self.calls.append((symbol, start_dt, end_dt)) + s, e = pd.Timestamp(start_dt), pd.Timestamp(end_dt) + rows = [ + (tt, close) + for tt, close in self.catalog.get(symbol, []) + if s <= pd.Timestamp(tt) <= e + ] + if not rows: + return pd.DataFrame() + return pd.DataFrame( + { + "ts_code": [symbol] * len(rows), + "trade_time": [r[0] for r in rows], + "open": [r[1] for r in rows], + "high": [r[1] for r in rows], + "low": [r[1] for r in rows], + "close": [r[1] for r in rows], + "vol": [100.0] * len(rows), + "amount": [1000.0] * len(rows), + } + ) + + +class _FakePro: + """Fake tushare client exposing window-aware ``stk_mins``.""" + + def __init__(self, catalog): + self._rec = _FetchRecorder(catalog) + + def stk_mins(self, ts_code, freq, start_date, end_date): # noqa: ARG002 + return self._rec(ts_code, start_date, end_date) + + +def _cache(tmp_path, **kw): + root = str(tmp_path / "cache") + store = IntradayParquetStore(root) + ledger = IntradayCoverageLedger(root) + return TushareIntradayCache(store, ledger, clock=_FIXED_CLOCK, **kw), store, ledger + + +_CATALOG = { + "000001.SZ": [ + ("2024-01-02 09:31:00", 11.0), + ("2024-01-02 09:32:00", 12.0), + ("2024-01-03 09:31:00", 13.0), + ("2024-01-04 09:31:00", 14.0), + ] +} + + +# --------------------------------------------------------------------------- # +# cold / warm / partial +# --------------------------------------------------------------------------- # +def test_cold_miss_writes_raw_1min(tmp_path): + cache, store, ledger = _cache(tmp_path) + fetch = _FetchRecorder(_CATALOG) + out = cache.stk_mins_1min( + ["000001.SZ"], "2024-01-02 09:30:00", "2024-01-02 14:55:00", fetch + ) + assert not out.empty + assert list(out.columns) == [ + "time", "symbol", "open", "high", "low", "close", + "volume", "amount", "source_trade_time", + ] + # raw bars persisted + stored = store.read_range(ENDPOINT, "000001.SZ", "1min", + "2024-01-02 00:00:00", "2024-01-02 23:59:59") + assert list(stored.columns) == STORED_COLUMNS + assert len(stored) == 2 # 09:31 + 09:32 + # coverage recorded ok + led = ledger.read() + assert (led["status"] == "ok").any() + assert fetch.calls # the API was hit on a cold miss + + +def test_warm_identical_zero_sdk_calls(tmp_path): + cache, _, _ = _cache(tmp_path) + fetch = _FetchRecorder(_CATALOG) + cold = cache.stk_mins_1min( + ["000001.SZ"], "2024-01-02 09:30:00", "2024-01-02 14:55:00", fetch + ) + fetch.calls.clear() + warm = cache.stk_mins_1min( + ["000001.SZ"], "2024-01-02 09:30:00", "2024-01-02 14:55:00", fetch + ) + assert fetch.calls == [] # zero SDK calls on a fully-covered warm request + pd.testing.assert_frame_equal(cold, warm) + + +def test_partial_gap_fetches_only_uncovered_days(tmp_path): + cache, _, _ = _cache(tmp_path) + fetch = _FetchRecorder(_CATALOG) + cache.stk_mins_1min( + ["000001.SZ"], "2024-01-02 09:00:00", "2024-01-03 15:00:00", fetch + ) + fetch.calls.clear() + cache.stk_mins_1min( + ["000001.SZ"], "2024-01-02 09:00:00", "2024-01-04 15:00:00", fetch + ) + # only 2024-01-04 was uncovered -> a single window starting that day + assert len(fetch.calls) == 1 + _, start_dt, end_dt = fetch.calls[0] + assert start_dt == "2024-01-04 00:00:00" + assert end_dt == "2024-01-04 23:59:59" + + +def test_empty_records_coverage_and_avoids_refetch(tmp_path): + cache, _, ledger = _cache(tmp_path) + fetch = _FetchRecorder(_CATALOG) # 999999.SZ absent -> empty + out1 = cache.stk_mins_1min( + ["999999.SZ"], "2024-01-02 09:30:00", "2024-01-02 14:55:00", fetch + ) + assert out1.empty + led = ledger.read() + assert (led["status"] == "empty").any() + fetch.calls.clear() + out2 = cache.stk_mins_1min( + ["999999.SZ"], "2024-01-02 09:30:00", "2024-01-02 14:55:00", fetch + ) + assert fetch.calls == [] # empty coverage prevents a needless refetch + assert out2.empty + + +def test_failed_fetch_records_no_coverage_and_retries(tmp_path): + cache, _, ledger = _cache(tmp_path) + + def raising_fetch(symbol, start_dt, end_dt): + raise ConnectionError("transient") + + with pytest.raises(ConnectionError): + cache.stk_mins_1min( + ["000001.SZ"], "2024-01-02 09:30:00", "2024-01-02 14:55:00", + raising_fetch, + ) + assert ledger.read().empty # a failed fetch is NOT coverage + # a later run with a working fetch still sees the gap and retries + good = _FetchRecorder(_CATALOG) + out = cache.stk_mins_1min( + ["000001.SZ"], "2024-01-02 09:30:00", "2024-01-02 14:55:00", good + ) + assert not out.empty + assert good.calls # the gap was retried + + +# --------------------------------------------------------------------------- # +# upsert idempotency / guard / schema +# --------------------------------------------------------------------------- # +def test_duplicate_upsert_keeps_one_row_per_key(tmp_path): + # direct store idempotency + store = IntradayParquetStore(str(tmp_path / "c")) + rows = pd.DataFrame( + { + "symbol": ["000001.SZ"], + "bar_end": [pd.Timestamp("2024-01-02 09:31:00")], + "source_trade_time": ["2024-01-02 09:31:00"], + "open": [1.0], "high": [1.0], "low": [1.0], "close": [11.0], + "volume": [100.0], "amount": [1000.0], "freq": ["1min"], + } + ) + store.upsert(ENDPOINT, "000001.SZ", "1min", rows, ["symbol", "freq", "bar_end"]) + store.upsert(ENDPOINT, "000001.SZ", "1min", rows, ["symbol", "freq", "bar_end"]) + got = store.read_range(ENDPOINT, "000001.SZ", "1min", + "2024-01-02 00:00:00", "2024-01-02 23:59:59") + assert len(got) == 1 # one row per (symbol, freq, bar_end) + + # also via cache force_refresh (re-fetches but does not double rows) + cache, _, _ = _cache(tmp_path, force_refresh=True) + fetch = _FetchRecorder(_CATALOG) + cache.stk_mins_1min(["000001.SZ"], "2024-01-02 09:30:00", + "2024-01-02 14:55:00", fetch) + cache.stk_mins_1min(["000001.SZ"], "2024-01-02 09:30:00", + "2024-01-02 14:55:00", fetch) + cstore = cache._store + got2 = cstore.read_range(ENDPOINT, "000001.SZ", "1min", + "2024-01-02 00:00:00", "2024-01-02 23:59:59") + assert len(got2) == 2 # 09:31 + 09:32, no duplicates despite two fetches + + +def test_non_1min_raw_request_cannot_hit_fetch(tmp_path): + cache, _, _ = _cache(tmp_path) + fetch = _FetchRecorder(_CATALOG) + with pytest.raises(ValueError, match="freq"): + cache.stk_mins_1min( + ["000001.SZ"], "2024-01-02 09:30:00", "2024-01-02 14:55:00", + fetch, freq="5min", + ) + assert fetch.calls == [] # rejected before any fetch + + +def test_ledger_and_store_carry_no_secret(tmp_path): + cache, store, ledger = _cache(tmp_path) + fetch = _FetchRecorder(_CATALOG) + cache.stk_mins_1min( + ["000001.SZ"], "2024-01-02 09:30:00", "2024-01-02 14:55:00", fetch + ) + # ledger holds only endpoint metadata; no token-bearing column + assert list(ledger.read().columns) == INTRADAY_LEDGER_COLUMNS + assert not any("token" in c for c in INTRADAY_LEDGER_COLUMNS) + # stored bars hold only raw market columns + stored = store.read_range(ENDPOINT, "000001.SZ", "1min", + "2024-01-02 00:00:00", "2024-01-02 23:59:59") + assert list(stored.columns) == STORED_COLUMNS + assert not any("token" in c for c in STORED_COLUMNS) + + +# --------------------------------------------------------------------------- # +# cached path == direct path +# --------------------------------------------------------------------------- # +def test_cached_output_equals_direct_feed_output(tmp_path, monkeypatch): + pro = _FakePro(_CATALOG) + start, end = "2024-01-02 09:30:00", "2024-01-02 09:35:00" + + direct = TushareIntradayFeed("unused.json") + monkeypatch.setattr(direct, "_client", lambda: pro) + out_direct = direct.get_minutes(["000001.SZ"], start, end) + + cache, _, _ = _cache(tmp_path) + cached = TushareIntradayFeed("unused.json", cache=cache) + monkeypatch.setattr(cached, "_client", lambda: pro) + out_cached = cached.get_minutes(["000001.SZ"], start, end) + + validate_intraday_bars(out_cached) + pd.testing.assert_frame_equal(out_direct, out_cached) + + +def test_cached_feed_empty_window_is_schema_shaped(tmp_path, monkeypatch): + pro = _FakePro(_CATALOG) + cache, _, _ = _cache(tmp_path) + feed = TushareIntradayFeed("unused.json", cache=cache) + monkeypatch.setattr(feed, "_client", lambda: pro) + out = feed.get_minutes(["999999.SZ"], "2024-01-02 09:30:00", "2024-01-02 09:35:00") + assert len(out) == 0 + assert list(out.index.names) == ["time", "symbol"] + validate_intraday_bars(out) diff --git a/tests/test_tushare_intraday_feed.py b/tests/test_tushare_intraday_feed.py new file mode 100644 index 0000000..5fe9e41 --- /dev/null +++ b/tests/test_tushare_intraday_feed.py @@ -0,0 +1,149 @@ +"""TushareIntradayFeed tests — no network, fake SDK, no token leak.""" + +from __future__ import annotations + +import json + +import pandas as pd +import pytest + +from data.clean.intraday_schema import validate_intraday_bars +from data.feed.tushare_intraday import TushareIntradayFeed + + +class _Pro: + """Fake tushare client. stk_mins returns reverse-chronological raw rows.""" + + def __init__(self): + self.calls: list[tuple] = [] + + def stk_mins(self, ts_code, freq, start_date, end_date): + self.calls.append((ts_code, freq, start_date, end_date)) + if ts_code == "000001.SZ": + return pd.DataFrame( + { + "ts_code": ["000001.SZ", "000001.SZ"], + "trade_time": [ + "2024-01-02 09:32:00", # reverse order on purpose + "2024-01-02 09:31:00", + ], + "open": [1.0, 1.0], + "high": [1.0, 1.0], + "low": [1.0, 1.0], + "close": [12.0, 11.0], + "vol": [200.0, 100.0], + "amount": [2.0, 1.0], + } + ) + return pd.DataFrame() # empty for any other symbol + + +def _feed(monkeypatch, pro=None): + pro = pro or _Pro() + feed = TushareIntradayFeed("unused.json") + monkeypatch.setattr(feed, "_client", lambda: pro) + return feed, pro + + +def test_get_minutes_maps_and_shapes(monkeypatch): + feed, _ = _feed(monkeypatch) + out = feed.get_minutes( + ["000001.SZ"], "2024-01-02 09:30:00", "2024-01-02 09:35:00", freq="1min" + ) + assert list(out.index.names) == ["time", "symbol"] + assert "volume" in out.columns and "vol" not in out.columns + # reverse-chronological input is sorted ascending + times = list(out.index.get_level_values("time")) + assert times == sorted(times) + assert out.iloc[0]["close"] == 11.0 # 09:31 comes first + + +def test_vol_and_ts_code_mapping_and_bar_end(monkeypatch): + feed, _ = _feed(monkeypatch) + out = feed.get_minutes( + ["000001.SZ"], "2024-01-02 09:30:00", "2024-01-02 09:35:00" + ) + assert out.iloc[0]["volume"] == 100.0 # vol -> volume + assert out.index.get_level_values("symbol")[0] == "000001.SZ" # ts_code -> symbol + assert out.iloc[0]["bar_end"] == pd.Timestamp("2024-01-02 09:31:00") # trade_time + assert out.iloc[0]["available_time"] == pd.Timestamp("2024-01-02 09:32:00") + + +def test_empty_return_is_schema_shaped(monkeypatch): + feed, _ = _feed(monkeypatch) + out = feed.get_minutes( + ["999999.SZ"], "2024-01-02 09:30:00", "2024-01-02 09:35:00" + ) + assert len(out) == 0 + assert list(out.index.names) == ["time", "symbol"] + validate_intraday_bars(out) # must not raise + + +def test_multi_symbol_concat_and_sort(monkeypatch): + class _P: + def stk_mins(self, ts_code, freq, start_date, end_date): # noqa: ARG002 + return pd.DataFrame( + { + "ts_code": [ts_code], + "trade_time": ["2024-01-02 09:31:00"], + "open": [1.0], + "high": [1.0], + "low": [1.0], + "close": [1.0], + "vol": [1.0], + "amount": [1.0], + } + ) + + feed, _ = _feed(monkeypatch, _P()) + out = feed.get_minutes( + ["000002.SZ", "000001.SZ"], "2024-01-02 09:30:00", "2024-01-02 09:35:00" + ) + assert sorted(set(out.index.get_level_values("symbol"))) == [ + "000001.SZ", + "000002.SZ", + ] + + +def test_stk_mins_called_with_datetime_window(monkeypatch): + feed, pro = _feed(monkeypatch) + feed.get_minutes( + ["000001.SZ"], "2024-01-02 09:30:00", "2024-01-02 09:35:00", freq="1min" + ) + ts_code, freq, start_date, end_date = pro.calls[0] + assert ts_code == "000001.SZ" + assert freq == "1min" + assert start_date == "2024-01-02 09:30:00" + assert end_date == "2024-01-02 09:35:00" + + +@pytest.mark.parametrize("freq", ["5min", "2min"]) +def test_non_1min_raw_freq_rejected_before_sdk_call(monkeypatch, freq): + feed, pro = _feed(monkeypatch) + with pytest.raises(ValueError, match="freq"): + feed.get_minutes( + ["000001.SZ"], + "2024-01-02 09:30:00", + "2024-01-02 09:35:00", + freq=freq, + ) + assert pro.calls == [] + + +def test_empty_symbols_rejected(monkeypatch): + feed, _ = _feed(monkeypatch) + with pytest.raises(ValueError, match="symbol"): + feed.get_minutes([], "2024-01-02 09:30:00", "2024-01-02 09:35:00") + + +def test_no_token_leak(tmp_path, monkeypatch): + secret = tmp_path / "config.json" + fake_token = "FAKE_TOKEN_DO_NOT_LEAK_abcdef0123456789" + secret.write_text(json.dumps({"tushare": {"token": fake_token}})) + feed = TushareIntradayFeed(str(secret)) + monkeypatch.setattr(feed, "_client", lambda: _Pro()) # never reads the token + out = feed.get_minutes( + ["000001.SZ"], "2024-01-02 09:30:00", "2024-01-02 09:35:00" + ) + assert fake_token not in repr(feed) + assert fake_token not in out.to_csv()