diff --git a/config/phase_d_minute_ideal_amp.yaml b/config/phase_d_minute_ideal_amp.yaml new file mode 100644 index 0000000..8e52a22 --- /dev/null +++ b/config/phase_d_minute_ideal_amp.yaml @@ -0,0 +1,121 @@ +# PR-D — Second real factor evaluation: minute ideal amplitude. +# +# Reproduces the Kaiyuan report §30 factor (分钟理想振幅因子, 市场微观结构系列 30, +# full-market IC -0.059 / ICIR -3.1 / rankIC -0.076, market-cap + industry neutral) +# as a first-class MinuteIdealAmplitudeFactor and runs it through the FROZEN +# StandardFactorEvaluator on REAL cached A-share data. CACHE-ONLY: the minute read is +# provably live-call-free; daily / universe / covariate endpoints go through the +# read-through cache (warm -> 0 gap fetches). +# +# The subject factor pools the trailing 10 trading days of PIT-truncated (14:50) +# minutes per symbol, ranks them by RAW close, and takes the mean-amplitude spread +# (high/low - 1) between the top / bottom 25% by close (N=10, lambda=25%). NOTE: the +# report's CSI500 numbers (rankICIR -2.13, long-short 10.6%) are for the COMBINED +# factor, NOT this single factor — treat them only as a loose reference. +# +# Eval CELL is IDENTICAL to PR-C (phase_c_jump_amount_corr.yaml): universe = CSI500 +# (000905.SH), PIT membership; window = 2021-07-01 .. 2026-06-30 (the project's +# minute-coverage window); daily rebalance; OOS split 2024-01-01; book = +# value_ep/value_bp/volatility_20; fee 0.001. The ONLY substantive difference is the +# subject factor, which is minute-derived and computed by the runner from the +# intraday cache — it is NOT a config-listed daily factor. The evaluator runs TWICE +# (see qt/eval_minute_ideal_amplitude.py): once with NO book (Incremental +# NOT_ASSESSED) and once with the confirmed book to measure whether minute ideal +# amplitude adds alpha BEYOND value / low-vol. + +project: + name: quantitative_trading_pr_d_minute_ideal_amp + timezone: Asia/Shanghai +data: + source: tushare + freq: D + start: '2021-07-01' + end: '2026-06-30' + external_secret_file: /home/shaofl/Projects/financial_projects/.config.json + tushare_token_key: tushare.token + output_name: minute_ideal_amp_daily + cache: + enabled: true + root_dir: artifacts/cache/tushare/v1 + refresh_recent_days: 14 + refresh_dimension_days: 30 + force_refresh: [] +universe: + type: index + index_code: 000905.SH + symbols: [] + min_listing_days: 60 + filters: + missing_close: true + suspended: false + st: false + limit_up_down: false +# factors = the confirmed BOOK used for the Incremental axis (value + low-vol). The +# SUBJECT factor (minute_ideal_amp_10) is minute-derived and computed by the runner +# from the intraday cache — it is NOT a config-listed daily factor. +factors: +- name: value_ep + enabled: true +- name: value_bp + enabled: true +- name: volatility_20 + enabled: true + params: + window: 20 + price_col: close +processing: + drop_missing: true + standardize: + enabled: true + method: zscore + # winsorize is a P0 no-op in this codebase; the EvalConfig declares winsorize=None + # accordingly (nothing is clipped), so the report never overstates preprocessing. + winsorize: + enabled: false + method: mad + n: 3.0 + # Industry + market-cap neutralization (SW-L1), matching the report's neutral + # column and the EvalConfig neutralization declaration. + neutralize: + enabled: true + industry_col: industry + size_col: market_cap + industry_level: L1 +alpha: + model: equal_weight + params: {} +portfolio: + # Required by the schema but unused: this runner evaluates a factor, it does not + # build/execute a portfolio. + constructor: topn_equal_weight + top_n: 50 + long_only: true + max_weight: null + turnover_cap: null +backtest: + # Required by the schema but unused (no backtest is run). The EvalConfig uses a + # DAILY rebalance (contract default), independent of this field. + initial_nav: 1.0 + rebalance: monthly + event_order: close_to_next_period + cash_return: 0.0 +cost: + fee_rate: 0.001 + slippage_rate: 0.0 + turnover_formula: l1 +analytics: + forward_return_periods: + - 1 + quantiles: 5 + benchmark: null +# OOS split (window midpoint) so the OOS section runs and the Predictive axis can be +# assessed (sign consistency across both holdout subperiods). +oos: + split_date: '2024-01-01' +output: + root_dir: artifacts + data_dir: artifacts/data + factor_dir: artifacts/factors + report_dir: artifacts/reports + log_dir: artifacts/logs + overwrite: true diff --git a/data/clean/intraday_amplitude.py b/data/clean/intraday_amplitude.py new file mode 100644 index 0000000..23aa4f8 --- /dev/null +++ b/data/clean/intraday_amplitude.py @@ -0,0 +1,200 @@ +"""Minute "ideal amplitude" factor (PR-D): trailing-window price-ranked amplitude. + +Reproduces the Kaiyuan report §30 (市场微观结构系列 30) 分钟理想振幅因子 as a daily +PIT-safe column derived from 1min bars. Kept in the DATA-clean layer (like +:func:`data.clean.intraday_aggregate.compute_jump_amount_corr`) so the ``factors`` +layer only SELECTS the pre-aggregated column and never fetches or sees a forward +return. + +Definition (LOCKED — reproduced from the report; N/lambda/min-minutes are part of +the factor DEFINITION, not tuned knobs). For each symbol and each panel date ``d``: + + 1. Take the symbol's most recent ``N`` (=10) trading days INCLUDING ``d`` (the + symbol's own minute-trading days — mirrors the jump factor's trailing window). + 2. PIT truncation (standing authorization): keep only bars with + ``available_time <= (that bar's trade_date + decision_time)`` (default 14:50). + This reuses the I3 per-bar cutoff path, so EVERY day in the window is truncated + to its own [session-open, 14:50] and post-14:50 / close data is never touched. + 3. Per-bar minute amplitude ``amp = high/low - 1``; a bar is dropped unless + ``low > 0`` and ``high >= low``. + 4. Pool ALL surviving bars of the window into ONE set ("merged cut", not a + per-day cut). If the pool has fewer than ``min_minutes`` (=1150 ≈ half of + 10 x ~230) valid minutes, the value is NaN (honest missing — no fabricated + warm-up; coverage is disclosed by the runner). + 5. Rank the pooled minutes by RAW minute close (unadjusted — amplitude is a ratio + so it needs no adjustment, and the report ranks on the raw minute price), with + ``(close, bar_end)`` as a stable total order. With ``k = floor(lambda * n)`` + (lambda=0.25): + V_high = mean amp of the ``k`` HIGHEST-close minutes + V_low = mean amp of the ``k`` LOWEST-close minutes + 6. factor(d, s) = ``V_high - V_low`` (column ``minute_ideal_amp_{N}``). + +Pre-registered sign = -1 (report full-market IC -0.059 / ICIR -3.1 / rankIC -0.076). +The value at ``(d, s)`` uses only bars at dates <= d, so a factor value never sees a +future bar (invariant #1); it is a DAILY signal traded close-to-close from d+1. + +This module is DATA-layer only: it does not fetch, does not touch factors / alpha / +portfolio / runtime, and never sees a token. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + +from data.clean.intraday_aggregate import DAILY_INDEX_NAMES, DEFAULT_DECISION_TIME +from data.clean.intraday_schema import SYMBOL_LEVEL, validate_intraday_bars + +# Factor DEFINITION constants (report terminal parameters; NOT tuned knobs). +IDEAL_AMP_LOOKBACK_DAYS = 10 # trailing trading-day window (N), includes date d +IDEAL_AMP_LAMBDA = 0.25 # top/bottom fraction by close (lambda) that forms V_high/V_low +IDEAL_AMP_MIN_MINUTES = 1150 # minimum valid pooled minutes for a finite value + + +def _empty_series(name: str) -> pd.Series: + """Schema-shaped empty ``MultiIndex(date, symbol)`` factor Series.""" + index = pd.MultiIndex.from_arrays( + [pd.DatetimeIndex([]), pd.Index([], dtype=object)], + names=DAILY_INDEX_NAMES, + ) + return pd.Series([], index=index, dtype=float, name=name) + + +def _rank_cut( + closes: np.ndarray, amps: np.ndarray, bar_ends: np.ndarray, lam: float, min_minutes: int +) -> float: + """V_high - V_low over one pooled window; NaN if too few minutes or k < 1. + + ``closes``/``amps``/``bar_ends`` are equal-length arrays for ONE pooled window. + Ranking is a stable total order ``(close, bar_end)`` (lexsort with close as the + primary key), so the top-k / bottom-k selection is fully deterministic even when + two minutes share a close. + """ + n = closes.size + if n < min_minutes: + return float("nan") + k = int(np.floor(lam * n)) + if k < 1: + return float("nan") + order = np.lexsort((bar_ends, closes)) # close primary, bar_end tie-break + a = amps[order] + return float(a[-k:].mean() - a[:k].mean()) + + +def _amplitude_for_symbol( + g: pd.DataFrame, lookback_days: int, lam: float, min_minutes: int +) -> tuple[list[pd.Timestamp], list[float]]: + """Daily factor values for ONE symbol from its PIT-filtered, guarded bars. + + ``g`` holds columns ``trade_date`` / ``close`` / ``amp`` / ``bar_end_ns`` for a + single symbol. Per-day arrays are built once, then each date pools the trailing + ``lookback_days`` days (including that date) — no cross-symbol leakage because + ``g`` is a single symbol's slice. + """ + days: list[pd.Timestamp] = [] + day_close: list[np.ndarray] = [] + day_amp: list[np.ndarray] = [] + day_be: list[np.ndarray] = [] + for day, sub in g.groupby("trade_date", sort=True): + days.append(pd.Timestamp(day).normalize()) + day_close.append(sub["close"].to_numpy(dtype=float)) + day_amp.append(sub["amp"].to_numpy(dtype=float)) + day_be.append(sub["bar_end_ns"].to_numpy(dtype="int64")) + + values: list[float] = [] + for j in range(len(days)): + lo = max(0, j - lookback_days + 1) + closes = np.concatenate(day_close[lo : j + 1]) + amps = np.concatenate(day_amp[lo : j + 1]) + bes = np.concatenate(day_be[lo : j + 1]) + values.append(_rank_cut(closes, amps, bes, lam, min_minutes)) + return days, values + + +def compute_minute_ideal_amplitude( + bars: pd.DataFrame, + *, + lookback_days: int = IDEAL_AMP_LOOKBACK_DAYS, + lam: float = IDEAL_AMP_LAMBDA, + min_minutes: int = IDEAL_AMP_MIN_MINUTES, + decision_time: str = DEFAULT_DECISION_TIME, + name: str = "minute_ideal_amp", +) -> pd.Series: + """PIT-safe daily "minute ideal amplitude" factor from 1min ``bars``. + + See the module docstring for the LOCKED definition. The heavy per-symbol loop is + memory-bounded (the runner feeds one symbol at a time), but this function also + accepts a multi-symbol frame and keeps symbols strictly isolated. + + Args: + bars: normalized 1min bars (:mod:`data.clean.intraday_schema`), + ``MultiIndex(time, symbol)``. + lookback_days: trailing trading-day window length (part of the definition). + lam: top/bottom close fraction defining V_high/V_low (0 < lam <= 0.5). + min_minutes: minimum valid pooled minutes for a finite value. + decision_time: per-bar PIT cutoff time-of-day (default 14:50:00). + name: the returned Series name (the factor-panel column name). + + Returns: + ``MultiIndex(date, symbol)`` Series (midnight-normalized dates) of the daily + factor value, sorted, named ``name``. Pure: never mutates ``bars``. + """ + validate_intraday_bars(bars) + if lookback_days < 1: + raise ValueError(f"lookback_days must be >= 1; got {lookback_days!r}.") + if not (0.0 < lam <= 0.5): + raise ValueError(f"lam must be in (0, 0.5]; got {lam!r}.") + if min_minutes < 2: + # Need at least 2 minutes so a non-empty top/bottom cut can exist. + raise ValueError(f"min_minutes must be >= 2; got {min_minutes!r}.") + if len(bars) == 0: + return _empty_series(name) + + work = bars.reset_index()[ + [SYMBOL_LEVEL, "bar_end", "available_time", "high", "low", "close"] + ].copy() + work["trade_date"] = work["bar_end"].dt.normalize() + # PIT truncation FIRST (per-bar timestamps): each bar's cutoff is its own + # trade_date + decision_time, so every day is truncated to [open, 14:50]. + cutoff = work["trade_date"] + pd.Timedelta(decision_time) + visible = work.loc[work["available_time"] <= cutoff].copy() + if visible.empty: + return _empty_series(name) + + low = visible["low"].to_numpy(dtype=float) + high = visible["high"].to_numpy(dtype=float) + guard = (low > 0.0) & (high >= low) + visible = visible.loc[guard].copy() + if visible.empty: + return _empty_series(name) + + visible["amp"] = visible["high"].to_numpy(dtype=float) / visible["low"].to_numpy( + dtype=float + ) - 1.0 + # int64 nanoseconds for a deterministic lexsort tie-break (view avoids the + # datetime->int astype deprecation). + visible["bar_end_ns"] = visible["bar_end"].to_numpy(dtype="datetime64[ns]").astype( + "int64" + ) + visible = visible.sort_values([SYMBOL_LEVEL, "bar_end"], kind="mergesort") + + index_tuples: list[tuple] = [] + values: list[float] = [] + for sym, g in visible.groupby(SYMBOL_LEVEL, sort=True): + days, vals = _amplitude_for_symbol(g, lookback_days, lam, min_minutes) + for day, val in zip(days, vals): + index_tuples.append((day, str(sym))) + values.append(val) + + if not index_tuples: + return _empty_series(name) + index = pd.MultiIndex.from_tuples(index_tuples, names=DAILY_INDEX_NAMES) + return pd.Series(values, index=index, name=name).sort_index() + + +__all__ = [ + "IDEAL_AMP_LAMBDA", + "IDEAL_AMP_LOOKBACK_DAYS", + "IDEAL_AMP_MIN_MINUTES", + "compute_minute_ideal_amplitude", +] diff --git a/factors/compute/intraday_derived.py b/factors/compute/intraday_derived.py index 1f0947c..ce8a445 100644 --- a/factors/compute/intraday_derived.py +++ b/factors/compute/intraday_derived.py @@ -1,9 +1,10 @@ """Daily factors DERIVED from intraday (minute) bars but executed close-to-close. -The single member today is :class:`JumpAmountCorrFactor` (PR-C), the Kaiyuan -report §6 "price-jump turnover correlation" factor. Like the value / MMP factors, -the heavy computation runs UPSTREAM (``data.clean.intraday_aggregate. -compute_jump_amount_corr`` aggregates 1min bars into a daily +The members today are :class:`JumpAmountCorrFactor` (PR-C, the Kaiyuan report §6 +"price-jump turnover correlation" factor) and :class:`MinuteIdealAmplitudeFactor` +(PR-D, the Kaiyuan report §30 "minute ideal amplitude" factor). Like the value / +MMP factors, the heavy computation runs UPSTREAM (``data.clean.intraday_aggregate`` +/ ``data.clean.intraday_amplitude`` aggregate 1min bars into a daily ``MultiIndex(date, symbol)`` column) and the Factor here simply SELECTS its column off the panel the runner already enriched. Keeping the minute aggregation in the data-clean layer preserves the layering: ``factors`` never fetches and never sees a @@ -30,6 +31,11 @@ JUMP_LOOKBACK_DAYS, JUMP_MIN_PAIRS, ) +from data.clean.intraday_amplitude import ( + IDEAL_AMP_LAMBDA, + IDEAL_AMP_LOOKBACK_DAYS, + IDEAL_AMP_MIN_MINUTES, +) from factors.base import Factor from factors.spec import FactorSpec @@ -115,4 +121,88 @@ def compute(self, panel: pd.DataFrame) -> pd.Series: return panel[self.name].rename(self.name) -__all__ = ["JumpAmountCorrFactor"] +class MinuteIdealAmplitudeFactor(Factor): + """Minute ideal-amplitude factor (daily signal, minute-derived). + + ``compute`` reads the pre-aggregated daily column the runner placed on the panel + (produced by ``data.clean.intraday_amplitude.compute_minute_ideal_amplitude``); + it does NO minute work of its own, mirroring :class:`JumpAmountCorrFactor` and + the value / financial factors that surface an enriched column. + + Args: + lookback_days: trailing trading-day window; part of the factor DEFINITION + (reproduced from the report), not a tuned knob. It only names the column + so a non-default window cannot silently mislabel it. + """ + + name: str = f"minute_ideal_amp_{IDEAL_AMP_LOOKBACK_DAYS}" + + def __init__(self, lookback_days: int = IDEAL_AMP_LOOKBACK_DAYS) -> None: + if not isinstance(lookback_days, int) or lookback_days < 1: + raise ValueError( + f"minute-ideal-amplitude lookback_days must be a positive integer; " + f"got {lookback_days!r}." + ) + self._lookback_days = lookback_days + self.name = f"minute_ideal_amp_{lookback_days}" + + @property + def lookback_days(self) -> int: + return self._lookback_days + + @property + def spec(self) -> FactorSpec: + """Evaluation contract; a property so ``factor_id`` tracks the window. + + expected_ic_sign=-1: the report's RankIC mean is -7.6% (full market, N=10, + lambda=25%) — a HIGH minute ideal amplitude predicts LOWER forward returns. + The sign is fixed BEFORE the run (a validated prototype must reproduce it). + is_intraday=False by the module docstring's reasoning: minute INPUT but a + DAILY signal traded close-to-close. min_history_bars=0: the warm-up is + DATA-dependent (a value appears once >= ``IDEAL_AMP_MIN_MINUTES`` valid + pooled minutes accumulate in the trailing window), not a fixed leading + count — the honest NaN rate is reported by data_coverage. + """ + return FactorSpec( + factor_id=self.name, + version="1.0", + description=( + f"Minute ideal amplitude (Kaiyuan report §30): pool the 1min bars of " + f"the trailing {self._lookback_days} trading days (PIT-truncated at " + f"14:50 per bar), rank the pooled minutes by RAW close, and return " + f"V_high - V_low where V_high/V_low are the mean per-minute amplitude " + f"(high/low - 1) of the top / bottom floor({IDEAL_AMP_LAMBDA:g}*n) " + f"minutes by close. Derived from 1min bars but a DAILY signal traded " + f"close-to-close; >= {IDEAL_AMP_MIN_MINUTES} valid pooled minutes " + f"required else NaN." + ), + expected_ic_sign=-1, + is_intraday=False, + forward_return_horizon=1, + return_basis="close_to_close", + # The 1min bar fields the upstream aggregation is derived from. Declared + # for honest provenance disclosure (data_coverage lists them); the daily + # panel surfaces the pre-aggregated column itself. + input_fields=("high", "low", "close"), + family="microstructure", + min_history_bars=0, + ) + + def compute(self, panel: pd.DataFrame) -> pd.Series: + """Select the pre-aggregated daily minute-ideal-amplitude column off ``panel``. + + The runner runs ``compute_minute_ideal_amplitude`` on the minute cache + upstream and joins the result as ``self.name``; here we only surface it, so + this factor does no temporal logic and cannot introduce lookahead. + """ + if self.name not in panel.columns: + raise ValueError( + f"MinuteIdealAmplitudeFactor needs the pre-aggregated '{self.name}' " + f"column on the panel (produced upstream by " + f"compute_minute_ideal_amplitude and joined by the runner); panel has " + f"{list(panel.columns)}." + ) + return panel[self.name].rename(self.name) + + +__all__ = ["JumpAmountCorrFactor", "MinuteIdealAmplitudeFactor"] diff --git a/qt/cli.py b/qt/cli.py index ab0ae9e..757c88f 100644 --- a/qt/cli.py +++ b/qt/cli.py @@ -228,6 +228,31 @@ def _cmd_run_eval_jump_amount_corr(args: argparse.Namespace) -> int: return 0 +def _cmd_run_eval_minute_ideal_amplitude(args: argparse.Namespace) -> int: + """Run the two real minute-ideal-amplitude factor evaluations (cache-only) + reports.""" + from qt.eval_minute_ideal_amplitude import run_eval_minute_ideal_amplitude + + try: + result = run_eval_minute_ideal_amplitude(args.config) + except (ConfigError, ValueError, FileNotFoundError) as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 1 + nb, wb = result.no_book_metrics, result.with_book_metrics + print( + f"OK run-eval-minute-ideal-amplitude: covered={result.covered_symbols}/" + f"{result.requested_symbols}, stk_mins_live_calls={result.minute_live_calls}, " + f"factor_rows={result.factor_rows} ({result.elapsed:.1f}s)\n" + f"no-book: {nb['deployment']} (predictive={nb['predictive']}) " + f"ic_mean={nb['ic_mean']:.4f} ic_ir={nb['ic_ir']:.3f} N_eff={nb['effective_samples']:.1f}\n" + f"with-book: {wb['deployment']} (incremental={wb['incremental']}) " + f"incr_ic_ir={wb['incremental_ic_ir']:.3f}\n" + f"reports: {result.reports.no_book_md} | {result.reports.with_book_md}\n" + f"dashboards: {result.reports.no_book_dashboard} | " + f"{result.reports.with_book_dashboard}" + ) + return 0 + + def _cmd_data_update(args: argparse.Namespace) -> int: """Warm/update the tushare caches (P4-3); never runs a backtest.""" from qt.data_updater import format_summary, run_data_update @@ -366,6 +391,13 @@ def build_parser() -> argparse.ArgumentParser: p_jac.add_argument("--config", required=True, help="Path to the YAML config.") p_jac.set_defaults(func=_cmd_run_eval_jump_amount_corr) + p_mia = sub.add_parser( + "run-eval-minute-ideal-amplitude", + help="Run the minute-ideal-amplitude factor evaluation (CSI500, cache-only).", + ) + p_mia.add_argument("--config", required=True, help="Path to the YAML config.") + p_mia.set_defaults(func=_cmd_run_eval_minute_ideal_amplitude) + for name, func, help_text in ( ("fetch-data", _cmd_fetch_data, "Run the spine, report data fetch."), ("compute-factors", _cmd_compute_factors, "Run the spine, report factor compute."), diff --git a/qt/eval_minute_ideal_amplitude.py b/qt/eval_minute_ideal_amplitude.py new file mode 100644 index 0000000..682d813 --- /dev/null +++ b/qt/eval_minute_ideal_amplitude.py @@ -0,0 +1,489 @@ +"""run-eval-minute-ideal-amplitude: the second real factor evaluation (PR-D). + +Reproduces the Kaiyuan report §30 "minute ideal amplitude" factor +(市场微观结构系列 30) as a first-class +:class:`~factors.compute.intraday_derived.MinuteIdealAmplitudeFactor` and runs it +through the FROZEN :class:`~analytics.eval.StandardFactorEvaluator` on REAL cached +A-share data (CSI500, PIT membership) — the same contract-driven loop PR-C used. + +The factor is a DAILY signal derived from 1min bars (see +``data.clean.intraday_amplitude.compute_minute_ideal_amplitude``): pool the trailing +10 trading days of PIT-truncated (14:50) minutes, rank them by RAW close, and take +the mean-amplitude spread between the top / bottom 25% by close. It is executed +CLOSE-TO-CLOSE (daily default), so ``is_intraday=False`` (the reasoning is +documented on the factor's spec). + +CACHE-ONLY: every input is read from the persistent tushare cache +(``artifacts/cache/tushare/v1``). The minute read is provably live-call-free (the +minute store has no fetch closure — a miss simply yields no rows); the daily / +universe / covariate endpoints go through the shared read-through cache, which on a +fully-warmed cache does zero gap fetches (disclosed via the run-log cache-stats +line). Forward returns are computed ONLY at the evaluator/analytics boundary from +``ctx.price_panel`` — the factor computation never sees a future return. + +The evaluator is run TWICE: once with NO known-factor book (the Incremental axis is +NOT_ASSESSED) and once with the project's independently-confirmed book (value_ep / +value_bp / volatility_20) so the Incremental axis measures whether minute ideal +amplitude adds alpha BEYOND value / low-vol. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from pathlib import Path + +import pandas as pd + +from analytics.eval import ( + EvalConfig, + EvalContext, + FactorEvalReport, + StandardFactorEvaluator, +) +from analytics.eval.figures import render_factor_dashboard +from data.cache.intraday_cache import ENDPOINT as INTRADAY_ENDPOINT +from data.cache.intraday_cache import READ_COLUMNS +from data.cache.intraday_parquet_store import IntradayParquetStore +from data.clean.intraday_amplitude import ( + IDEAL_AMP_LAMBDA, + IDEAL_AMP_LOOKBACK_DAYS, + IDEAL_AMP_MIN_MINUTES, + compute_minute_ideal_amplitude, +) +from data.clean.intraday_schema import RAW_INTRADAY_FREQ, normalize_intraday_bars +from data.clean.schema import CORE_COLUMNS, DATE_LEVEL +from factors.compute.intraday_derived import MinuteIdealAmplitudeFactor +from factors.spec import FactorSpec +from qt.config import RootConfig, load_config +from qt.pipeline import ( + _build_cache, + _build_universe, + _load_panel, + _log_run_cache_stats, + _make_logger, + _maybe_enrich_covariates, + _maybe_enrich_value, + _process_factors, +) + +_LOGGER_NAME = "qt.eval_minute_ideal_amplitude" +_REPORT_STEM = "eval_minute_ideal_amplitude" + + +# --------------------------------------------------------------------------- # +# Minute loading (cache-only, per-symbol -> memory-bounded) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _AmpMinuteLoad: + """Diagnostics from the cache-only per-symbol minute read + aggregation.""" + + factor: pd.Series # MultiIndex(date, symbol) raw minute-ideal-amplitude + requested: int + covered: tuple[str, ...] # symbols that produced >= 1 finite factor value + empty_symbols: tuple[str, ...] # requested but no cached minute / no value + raw_rows: int + live_calls: int # provably 0 (store read has no fetch closure) + + +def _load_minute_ideal_amp_panel( + cfg: RootConfig, + symbols: list[str], + spec: FactorSpec, + logger, + *, + lookback_days: int, + lam: float, + min_minutes: int, +) -> _AmpMinuteLoad: + """Compute the raw minute-ideal-amplitude panel per symbol from the minute cache. + + Memory-bounded: one symbol's minute history is read, aggregated to its daily + factor series, and discarded before the next — the multi-year all-symbol minute + panel is NEVER materialized. Read-only: :meth:`IntradayParquetStore.read_range` + has no fetch closure, so ``stk_mins`` live calls are provably zero (a symbol with + no cached minute simply yields no rows and is disclosed as empty). + """ + root = cfg.data.cache.root_dir + store = IntradayParquetStore(root) + start = pd.Timestamp(cfg.data.start).normalize() + end = pd.Timestamp(cfg.data.end).normalize() + pd.Timedelta("23:59:59") + + series: list[pd.Series] = [] + covered: list[str] = [] + empty: list[str] = [] + raw_rows = 0 + for i, sym in enumerate(symbols): + part = store.read_range(INTRADAY_ENDPOINT, sym, RAW_INTRADAY_FREQ, start, end) + if part.empty: + empty.append(sym) + continue + raw_rows += len(part) + bars = normalize_intraday_bars( + part.rename(columns={"bar_end": "time"})[READ_COLUMNS], + freq=RAW_INTRADAY_FREQ, + ) + s = compute_minute_ideal_amplitude( + bars, + lookback_days=lookback_days, + lam=lam, + min_minutes=min_minutes, + name=spec.factor_id, + ) + if s.notna().any(): + series.append(s) + covered.append(sym) + else: + empty.append(sym) + if (i + 1) % 100 == 0: + logger.info( + "minute aggregation: %d/%d symbols processed (%d with a value)", + i + 1, len(symbols), len(covered), + ) + + if not series: + raise ValueError( + "run-eval-minute-ideal-amplitude blocked: no requested symbol produced a " + f"cached minute ideal-amplitude value over [{cfg.data.start}, " + f"{cfg.data.end}]. The minute cache is required (this runner never warms " + "it); check coverage." + ) + factor = pd.concat(series).sort_index() + logger.info( + "minute aggregation (cache-only): %d/%d symbols with a value, %d raw 1min " + "rows read, %d factor rows, stk_mins_live_calls=0", + len(covered), len(symbols), raw_rows, len(factor), + ) + return _AmpMinuteLoad( + factor=factor, + requested=len(symbols), + covered=tuple(covered), + empty_symbols=tuple(empty), + raw_rows=raw_rows, + live_calls=0, + ) + + +# --------------------------------------------------------------------------- # +# Evaluation core (network-free seam: given panels, run the two evaluations) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _RunReports: + """The two evaluation reports (no-book / with-book) + their file paths.""" + + no_book: FactorEvalReport + with_book: FactorEvalReport + no_book_md: Path + no_book_json: Path + with_book_md: Path + with_book_json: Path + no_book_dashboard: Path + with_book_dashboard: Path + + +def evaluate_two_runs( + factor_panel: pd.Series | pd.DataFrame, + spec: FactorSpec, + eval_cfg: EvalConfig, + price_panel: pd.DataFrame, + book: pd.DataFrame, + *, + universe_symbols: tuple[str, ...], + fee_rate: float, + report_dir: Path, + stem: str = _REPORT_STEM, +) -> _RunReports: + """Run the StandardFactorEvaluator TWICE (no book / with book) and write reports. + + This is the network-free seam: given the PROCESSED factor panel, the qfq price + panel (for forward returns), and the PROCESSED known-factor book, it does the two + ``evaluate`` calls and writes ``{stem}_no_book`` / ``{stem}_with_book`` as + Markdown + JSON. Run 1 omits ``known_factors`` (Incremental NOT_ASSESSED); run 2 + supplies the book (Incremental measured). + """ + evaluator = StandardFactorEvaluator() + report_dir.mkdir(parents=True, exist_ok=True) + + ctx_no_book = EvalContext( + price_panel=price_panel, + universe_symbols=universe_symbols, + fee_rate=fee_rate, + ) + # evaluate_with_ir yields the SAME report as evaluate() plus the IR the + # research-style dashboard needs (per-period IC + quantile return series). + report_no_book, ir_no_book = evaluator.evaluate_with_ir( + factor_panel, spec, eval_cfg, ctx_no_book + ) + + ctx_with_book = EvalContext( + price_panel=price_panel, + universe_symbols=universe_symbols, + fee_rate=fee_rate, + known_factors=book, + ) + report_with_book, ir_with_book = evaluator.evaluate_with_ir( + factor_panel, spec, eval_cfg, ctx_with_book + ) + + nb_md, nb_json = _write_report(report_no_book, report_dir, f"{stem}_no_book") + wb_md, wb_json = _write_report(report_with_book, report_dir, f"{stem}_with_book") + nb_png = render_factor_dashboard( + report_no_book, ir_no_book, report_dir / f"{stem}_no_book_dashboard.png" + ) + wb_png = render_factor_dashboard( + report_with_book, ir_with_book, report_dir / f"{stem}_with_book_dashboard.png" + ) + return _RunReports( + no_book=report_no_book, + with_book=report_with_book, + no_book_md=nb_md, + no_book_json=nb_json, + with_book_md=wb_md, + with_book_json=wb_json, + no_book_dashboard=nb_png, + with_book_dashboard=wb_png, + ) + + +def _write_report(report: FactorEvalReport, report_dir: Path, stem: str) -> tuple[Path, Path]: + """Write ``report`` as deterministic Markdown + JSON; return the two paths.""" + md_path = report_dir / f"{stem}.md" + json_path = report_dir / f"{stem}.json" + md_path.write_text(report.render(), encoding="utf-8") + json_path.write_text(report.to_json(), encoding="utf-8") + return md_path, json_path + + +# --------------------------------------------------------------------------- # +# Metric extraction (for the CLI line + the handoff) +# --------------------------------------------------------------------------- # +def _section_payload(report: FactorEvalReport, name: str) -> dict: + section = report.by_name().get(name) + return dict(getattr(section, "payload", {}) or {}) + + +def extract_metrics(report: FactorEvalReport) -> dict: + """Pull the headline verdict + gated metrics out of a finished report.""" + verdict = report.require_verdict() + pred = _section_payload(report, "predictive_power") + purity = _section_payload(report, "data_coverage") + incr = _section_payload(report, "purity") + return { + "deployment": verdict.verdict, + "predictive": verdict.predictive.verdict, + "incremental": verdict.incremental.verdict, + "tradable": verdict.tradable.verdict, + "ic_mean": pred.get("ic_mean"), + "ic_ir": pred.get("ic_ir"), + "ic_ir_ci_low": pred.get("ic_ir_ci_low"), + "ic_ir_ci_high": pred.get("ic_ir_ci_high"), + "ic_ir_ci_n_eff": pred.get("ic_ir_ci_n_eff"), + "ic_win_rate": pred.get("ic_win_rate"), + "ic_nw_t": pred.get("ic_nw_t"), + "settled_rebalances": purity.get("settled_rebalances"), + "effective_samples": purity.get("effective_samples"), + "span_days": purity.get("span_days"), + "incremental_ic_ir": incr.get("incremental_ic_ir"), + "incremental_ic_ir_ci_low": incr.get("incremental_ic_ir_ci_low"), + "incremental_ic_ir_ci_high": incr.get("incremental_ic_ir_ci_high"), + "incremental_ic_ir_ci_n_eff": incr.get("incremental_ic_ir_ci_n_eff"), + "incremental_ic_mean": incr.get("incremental_ic_mean"), + } + + +# --------------------------------------------------------------------------- # +# EvalConfig construction (HONEST provenance of what the runner actually did) +# --------------------------------------------------------------------------- # +def _build_eval_config(cfg: RootConfig) -> EvalConfig: + """Build the per-run EvalConfig, declaring EXACTLY what the pipeline applied. + + The declarations must not overstate: this codebase's winsorize step is a P0 + no-op, so ``winsorize`` is declared None (nothing was clipped) even though the + config may toggle it. z-score + industry/size neutralization ARE applied, so they + are declared. ``oos_split`` (from the config's ``oos`` block) makes the OOS + section run so the Predictive axis can be assessed. ``is_exploratory=True``: this + is a reproduction on a shorter window / narrower neutralization than the report, + not a return claim (it caps the deployment label at Watch). + """ + if cfg.oos is None: + raise ValueError( + "run-eval-minute-ideal-amplitude requires an 'oos' section (split_date) " + "so the Predictive axis has an out-of-sample split to assess; add e.g. " + "oos: {split_date: '2024-01-01'}." + ) + return EvalConfig( + universe=cfg.universe.index_code or cfg.universe.type, + universe_is_pit=cfg.universe.type == "index", + start=cfg.data.start, + end=cfg.data.end, + is_exploratory=True, + post_hoc_selected=False, + rebalance="daily", + n_quantiles=int(cfg.analytics.quantiles), + cost_scenarios=(1.0, 2.0, 4.0), + oos_split=cfg.oos.split_date, + # winsorize is a P0 no-op in this codebase -> declare None (nothing clipped). + winsorize=None, + standardize="zscore" if cfg.processing.standardize.enabled else None, + neutralization=("industry", "size") if cfg.processing.neutralize.enabled else (), + industry_level=cfg.processing.neutralize.industry_level, + tuned=False, + # We evaluated ONE pre-registered factor whose sign came from the report (not + # a screen of our own); the report's own factor screen is a caveat noted in + # the run's prose, not our multiple-testing background. + n_factors_screened=1, + data_snapshot_id=cfg.data.cache.root_dir, + ) + + +# --------------------------------------------------------------------------- # +# Result container + the full glue +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class MinuteIdealAmpEvalResult: + """Immutable summary of one run-eval-minute-ideal-amplitude run.""" + + config: RootConfig + spec: FactorSpec + requested_symbols: int + covered_symbols: int + empty_symbols: int + factor_rows: int + minute_raw_rows: int + minute_live_calls: int + no_book_metrics: dict + with_book_metrics: dict + reports: _RunReports + log_path: Path + elapsed: float + + +def _check_preconditions(cfg: RootConfig) -> None: + """Fail readably if the config cannot drive a real, cache-only CSI500 eval.""" + if cfg.data.source != "tushare": + raise ValueError( + "run-eval-minute-ideal-amplitude needs data.source='tushare' (real cached " + f"A-share data); got {cfg.data.source!r}." + ) + if not cfg.data.cache.enabled: + raise ValueError( + "run-eval-minute-ideal-amplitude needs data.cache.enabled=true (it reads " + "the persistent tushare cache and never warms live)." + ) + if cfg.universe.type != "index": + raise ValueError( + "run-eval-minute-ideal-amplitude needs universe.type='index' (PIT " + f"membership, e.g. 000905.SH for CSI500); got {cfg.universe.type!r}." + ) + if not cfg.processing.neutralize.enabled: + raise ValueError( + "run-eval-minute-ideal-amplitude expects processing.neutralize.enabled=" + "true (industry + size neutralization, matching the report's neutral " + "column and the EvalConfig declaration)." + ) + + +def run_eval_minute_ideal_amplitude(config_path: str) -> MinuteIdealAmpEvalResult: + """Run the two real minute-ideal-amplitude evaluations (cache-only) + reports.""" + cfg = load_config(config_path) + _check_preconditions(cfg) + + log_path = Path(cfg.output.log_dir) / f"{_REPORT_STEM}.log" + logger = _make_logger(log_path, name=_LOGGER_NAME) + started = time.monotonic() + + factor = MinuteIdealAmplitudeFactor(lookback_days=IDEAL_AMP_LOOKBACK_DAYS) + spec = factor.spec + eval_cfg = _build_eval_config(cfg) + logger.info( + "eval config: %s rebalance=daily oos_split=%s", eval_cfg.universe, eval_cfg.oos_split + ) + + cache = _build_cache(cfg) + universe, symbols = _build_universe(cfg, logger, cache) + panel = _load_panel(cfg, symbols, logger, cache) + + # Book (value_ep / value_bp / volatility_20): enrich, compute, process. The value + # factors need daily_basic pe/pb; volatility_20 needs close. + book_factors = _build_book_factors() + panel = _maybe_enrich_value(cfg, panel, symbols, book_factors, logger, cache) + panel = _maybe_enrich_covariates(cfg, panel, symbols, logger, cache) + _log_run_cache_stats(cache, logger) # daily/universe/covariate gap-fetches (warm -> 0) + + panel_dates = pd.Index( + pd.unique(panel.index.get_level_values(DATE_LEVEL)), name=DATE_LEVEL + ) + book_raw = pd.concat( + [f.compute(panel).rename(f.name) for f in book_factors], axis=1 + ) + book_processed = _process_factors(cfg, book_raw, panel) + + # Minute ideal-amplitude factor: cache-only per-symbol aggregation -> raw -> process. + load = _load_minute_ideal_amp_panel( + cfg, symbols, spec, logger, + lookback_days=IDEAL_AMP_LOOKBACK_DAYS, + lam=IDEAL_AMP_LAMBDA, + min_minutes=IDEAL_AMP_MIN_MINUTES, + ) + # A minute date with no daily bar cannot have a forward return; keep the factor on + # the daily trading grid so the analytics boundary has a price for every date. + amp_raw = load.factor[ + load.factor.index.get_level_values(DATE_LEVEL).isin(panel_dates) + ] + amp_processed = _process_factors(cfg, amp_raw.to_frame(spec.factor_id), panel) + factor_series = amp_processed[spec.factor_id] + + price_panel = panel[CORE_COLUMNS] + reports = evaluate_two_runs( + factor_series, + spec, + eval_cfg, + price_panel, + book_processed, + universe_symbols=tuple(symbols), + fee_rate=float(cfg.cost.fee_rate), + report_dir=Path(cfg.output.report_dir), + ) + + no_book_metrics = extract_metrics(reports.no_book) + with_book_metrics = extract_metrics(reports.with_book) + logger.info( + "verdict no-book: %s (predictive=%s); with-book: %s (incremental=%s)", + no_book_metrics["deployment"], no_book_metrics["predictive"], + with_book_metrics["deployment"], with_book_metrics["incremental"], + ) + + return MinuteIdealAmpEvalResult( + config=cfg, + spec=spec, + requested_symbols=load.requested, + covered_symbols=len(load.covered), + empty_symbols=len(load.empty_symbols), + factor_rows=int(len(factor_series)), + minute_raw_rows=load.raw_rows, + minute_live_calls=load.live_calls, + no_book_metrics=no_book_metrics, + with_book_metrics=with_book_metrics, + reports=reports, + log_path=log_path, + elapsed=time.monotonic() - started, + ) + + +def _build_book_factors() -> list: + """Instantiate the confirmed book (value_ep / value_bp / volatility_20).""" + from factors.compute.candidates import ValueFactor, VolatilityFactor + + return [ + ValueFactor("value_ep"), + ValueFactor("value_bp"), + VolatilityFactor(window=20), + ] + + +__all__ = [ + "MinuteIdealAmpEvalResult", + "evaluate_two_runs", + "extract_metrics", + "run_eval_minute_ideal_amplitude", +] diff --git a/tests/test_eval_minute_ideal_amplitude_runner.py b/tests/test_eval_minute_ideal_amplitude_runner.py new file mode 100644 index 0000000..81bae7e --- /dev/null +++ b/tests/test_eval_minute_ideal_amplitude_runner.py @@ -0,0 +1,239 @@ +"""PR-D runner: cache-only minute loader + the two-run evaluation core (no network).""" + +from __future__ import annotations + +import logging + +import numpy as np +import pandas as pd +import pytest + +from analytics.eval import EvalConfig, MANDATORY_SECTIONS, Section, Skipped +from analytics.eval.verdict import AXIS_NOT_ASSESSED, AXIS_VERDICTS +from data.cache.intraday_cache import ENDPOINT as INTRADAY_ENDPOINT +from data.cache.intraday_parquet_store import KEY_COLS, IntradayParquetStore +from factors.compute.intraday_derived import MinuteIdealAmplitudeFactor +from qt.config import ( + AlphaCfg, + BacktestCfg, + CacheCfg, + CostCfg, + DataCfg, + FactorCfg, + OutputCfg, + PortfolioCfg, + RootConfig, + UniverseCfg, +) +from qt.eval_minute_ideal_amplitude import ( + _load_minute_ideal_amp_panel, + evaluate_two_runs, + extract_metrics, +) + + +# --------------------------------------------------------------------------- # +# Minute cache-only loader +# --------------------------------------------------------------------------- # +def _stored_rows(sym, day, closes, amps): + """Build STORED_COLUMNS-shaped 1min rows for one session. + + Minute i: ``low=100``, ``high=100*(1+amp_i)`` (amplitude ``amp_i`` exactly), + ``close=close_i`` (controls the price rank). Bars start at 09:31 so all sit + inside the 14:50 PIT window. + """ + base = pd.Timestamp(day) + pd.Timedelta("09:31:00") + rows = [] + for i, (c, a) in enumerate(zip(closes, amps)): + be = base + pd.Timedelta(minutes=i) + rows.append( + { + "symbol": sym, + "bar_end": be, + "source_trade_time": be, + "open": c, + "high": 100.0 * (1.0 + a), + "low": 100.0, + "close": c, + "volume": 1.0, + "amount": 1.0, + "freq": "1min", + } + ) + return pd.DataFrame(rows) + + +def _min_config(root, start, end): + return RootConfig( + data=DataCfg( + source="tushare", + start=start, + end=end, + external_secret_file="/nonexistent.json", + cache=CacheCfg(enabled=True, root_dir=str(root)), + ), + universe=UniverseCfg(type="static", symbols=["A.SZ"]), + factors=[FactorCfg(name="momentum_20")], + alpha=AlphaCfg(), + portfolio=PortfolioCfg(top_n=1), + backtest=BacktestCfg(), + cost=CostCfg(), + output=OutputCfg(), + ) + + +def test_minute_loader_is_cache_only_and_discloses_empty(tmp_path): + root = tmp_path / "cache" + store = IntradayParquetStore(str(root)) + closes = [100.0, 101.0, 102.0, 103.0, 104.0, 105.0] + amps = [0.01, 0.02, 0.03, 0.04, 0.05, 0.06] + # Two symbols with cached minute bars across two trading days; a third symbol has + # NO cached minute at all (must be disclosed as empty, never fetched). + for sym in ("AAA.SZ", "BBB.SZ"): + for day in ("2021-07-01", "2021-07-02"): + store.upsert(INTRADAY_ENDPOINT, sym, "1min", + _stored_rows(sym, day, closes, amps), KEY_COLS) + + cfg = _min_config(root, "2021-07-01", "2021-07-02") + spec = MinuteIdealAmplitudeFactor().spec + logger = logging.getLogger("test.mia.loader") + load = _load_minute_ideal_amp_panel( + cfg, ["AAA.SZ", "BBB.SZ", "CCC.SZ"], spec, logger, + lookback_days=20, lam=0.25, min_minutes=4, + ) + assert load.live_calls == 0 # store read has no fetch closure + assert set(load.covered) == {"AAA.SZ", "BBB.SZ"} # both produced a value + assert load.empty_symbols == ("CCC.SZ",) # uncovered disclosed, not fetched + assert load.factor.name == spec.factor_id + assert load.factor.notna().any() + # values live only on the covered symbols + assert set(load.factor.index.get_level_values("symbol")) == {"AAA.SZ", "BBB.SZ"} + + +def test_minute_loader_blocks_when_nothing_cached(tmp_path): + cfg = _min_config(tmp_path / "empty", "2021-07-01", "2021-07-02") + spec = MinuteIdealAmplitudeFactor().spec + with pytest.raises(ValueError, match="no requested symbol produced"): + _load_minute_ideal_amp_panel( + cfg, ["ZZZ.SZ"], spec, logging.getLogger("test.mia.block"), + lookback_days=20, lam=0.25, min_minutes=4, + ) + + +# --------------------------------------------------------------------------- # +# Evaluation core (two runs) — synthetic processed panels, no network +# --------------------------------------------------------------------------- # +def _synthetic_panels(n_days=90, n_symbols=15, seed=7): + rng = np.random.default_rng(seed) + dates = pd.bdate_range("2022-01-03", periods=n_days) + symbols = [f"{i:06d}.SZ" for i in range(n_symbols)] + idx = pd.MultiIndex.from_product([dates, symbols], names=["date", "symbol"]) + + # processed (z-scored-ish) subject factor + a 3-column book on the same grid + factor = pd.Series( + rng.standard_normal(len(idx)), index=idx, name="minute_ideal_amp_10" + ) + book = pd.DataFrame( + { + "value_ep": rng.standard_normal(len(idx)), + "value_bp": rng.standard_normal(len(idx)), + "volatility_20": rng.standard_normal(len(idx)), + }, + index=idx, + ) + # a qfq-style price panel with the CORE_COLUMNS the forward-return boundary needs + close = pd.Series( + 100.0 * np.exp(np.cumsum(0.001 * rng.standard_normal(len(idx)))), index=idx + ) + price = pd.DataFrame( + { + "open": close.to_numpy(), + "high": close.to_numpy() * 1.01, + "low": close.to_numpy() * 0.99, + "close": close.to_numpy(), + "volume": 1_000.0, + "amount": 100_000.0, + "adj_factor": 1.0, + }, + index=idx, + ) + return factor, book, price + + +def _eval_cfg(): + return EvalConfig( + universe="000905.SH", + universe_is_pit=True, + start="2022-01-03", + end="2022-05-10", + is_exploratory=True, + post_hoc_selected=False, + rebalance="daily", + n_quantiles=5, + oos_split="2022-03-15", + winsorize=None, + standardize="zscore", + neutralization=("industry", "size"), + industry_level="L1", + ) + + +def test_evaluate_two_runs_produces_full_reports_and_incremental_axis(tmp_path): + factor, book, price = _synthetic_panels() + spec = MinuteIdealAmplitudeFactor().spec + reports = evaluate_two_runs( + factor, spec, _eval_cfg(), price, book, + universe_symbols=tuple(sorted(set(book.index.get_level_values("symbol")))), + fee_rate=0.001, + report_dir=tmp_path, + ) + + for report in (reports.no_book, reports.with_book): + by = report.by_name() + assert set(by) == set(MANDATORY_SECTIONS) # all 8 present + assert all(isinstance(s, (Section, Skipped)) for s in by.values()) + assert report.verdict is not None # a verdict was produced + + # no-book: purity Skipped (no book) -> Incremental NOT_ASSESSED + assert isinstance(reports.no_book.by_name()["purity"], Skipped) + assert reports.no_book.verdict.incremental.verdict == AXIS_NOT_ASSESSED + + # with-book: purity is a real Section that populated the Incremental facts + purity = reports.with_book.by_name()["purity"] + assert isinstance(purity, Section) + assert purity.payload["known_factors_supplied"] is True + assert "incremental_ic_ir" in purity.payload + # the axis is now assessed (not NOT_ASSESSED) and is a valid axis state + incr = reports.with_book.verdict.incremental.verdict + assert incr in AXIS_VERDICTS and incr != AXIS_NOT_ASSESSED + + # reports were written to disk (md + json), both runs + for p in (reports.no_book_md, reports.no_book_json, + reports.with_book_md, reports.with_book_json): + assert p.exists() and p.stat().st_size > 0 + + # the research-style dashboard PNG is emitted for both runs (mandatory report + # artifact alongside md/json) + for p in (reports.no_book_dashboard, reports.with_book_dashboard): + assert p.exists() and p.stat().st_size > 20_000 + with open(p, "rb") as fh: + assert fh.read(8) == b"\x89PNG\r\n\x1a\n" + + # metrics extraction surfaces the gated fields + m = extract_metrics(reports.no_book) + assert m["deployment"] in {"Adopt", "Watch", "Reject", "INSUFFICIENT-DATA"} + assert "effective_samples" in m and "ic_ir" in m + + +def test_no_book_run_never_reaches_adopt(tmp_path): + # Structural guarantee (design §6): with no book + no execution facts, at most + # Watch — regardless of the signal (exploratory cap + NOT_ASSESSED axes). + factor, book, price = _synthetic_panels(seed=3) + reports = evaluate_two_runs( + factor, MinuteIdealAmplitudeFactor().spec, _eval_cfg(), price, book, + universe_symbols=(), + fee_rate=0.001, + report_dir=tmp_path, + ) + assert reports.no_book.verdict.verdict != "Adopt" + assert reports.no_book.verdict.tradable.verdict == AXIS_NOT_ASSESSED diff --git a/tests/test_minute_ideal_amplitude_factor.py b/tests/test_minute_ideal_amplitude_factor.py new file mode 100644 index 0000000..7d57d9b --- /dev/null +++ b/tests/test_minute_ideal_amplitude_factor.py @@ -0,0 +1,370 @@ +"""PR-D: minute ideal-amplitude factor (aggregation + spec + PIT truncation). + +The factor pools the 1min bars of a symbol's trailing ``N`` trading days (PIT- +truncated at 14:50 per bar), ranks the pooled minutes by their raw close price, +and returns ``V_high - V_low`` where ``V_high``/``V_low`` are the mean per-minute +amplitude (``high/low - 1``) of the top / bottom ``floor(lambda*n)`` minutes by +close. Sign is pre-registered -1 (high ideal amplitude -> lower forward return). +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from data.clean.intraday_amplitude import ( + IDEAL_AMP_LAMBDA, + IDEAL_AMP_LOOKBACK_DAYS, + IDEAL_AMP_MIN_MINUTES, + compute_minute_ideal_amplitude, +) +from data.clean.intraday_schema import empty_intraday_bars, normalize_intraday_bars +from factors.compute.intraday_derived import MinuteIdealAmplitudeFactor +from factors.spec import FactorSpec + +_SYM = "000001.SZ" + + +def _bars(rows): + """rows = [(time, symbol, high, low, close), ...] -> normalized 1min bars. + + ``open``/``volume``/``amount`` are filled harmlessly; the factor reads + ``high``/``low`` (per-minute amplitude ``high/low - 1``) and ``close`` (the + price-rank cut). ``close`` is set INDEPENDENTLY of ``high``/``low`` because the + factor decouples them: rank by close, measure amplitude by high/low. + """ + df = pd.DataFrame( + { + "time": pd.to_datetime([r[0] for r in rows]), + "symbol": [r[1] for r in rows], + "open": [r[4] for r in rows], + "high": [r[2] for r in rows], + "low": [r[3] for r in rows], + "close": [r[4] for r in rows], + "volume": [1.0] * len(rows), + "amount": [1.0] * len(rows), + } + ) + return normalize_intraday_bars(df, freq="1min") + + +def _session(day, closes, amps, sym=_SYM, start="09:31:00"): + """One session: minute i at ``low=100``, ``high=100*(1+amp_i)`` (amp exact), + ``close=close_i`` (controls the price rank). Starts at ``start`` so every bar + stays inside the 14:50 PIT window when ``start`` is early enough.""" + base = pd.Timestamp(day) + pd.Timedelta(start) + return [ + (base + pd.Timedelta(minutes=i), sym, 100.0 * (1.0 + a), 100.0, c) + for i, (c, a) in enumerate(zip(closes, amps)) + ] + + +def _ideal_amp_ref(closes, amps, bar_ends, lam, min_minutes): + """Hand reference: rank pooled minutes by (close, bar_end); V_high - V_low.""" + n = len(closes) + if n < min_minutes: + return float("nan") + k = int(np.floor(lam * n)) + if k < 1: + return float("nan") + be = np.asarray([pd.Timestamp(b).value for b in bar_ends], dtype="int64") + order = np.lexsort((be, np.asarray(closes, dtype=float))) # close primary + a = np.asarray(amps, dtype=float)[order] + return float(a[-k:].mean() - a[:k].mean()) + + +# --------------------------------------------------------------------------- # +# Correctness vs a hand-computed reference +# --------------------------------------------------------------------------- # +def test_single_day_hand_value_n4_k1(): + # n=4, k=floor(0.25*4)=1: V_high = amp of the single highest-close minute, + # V_low = amp of the single lowest-close minute. + closes = [100.0, 101.0, 102.0, 103.0] + amps = [0.010, 0.040, 0.020, 0.030] + bars = _bars(_session("2021-07-01", closes, amps)) + out = compute_minute_ideal_amplitude( + bars, lookback_days=20, lam=0.25, min_minutes=4 + ) + d1 = pd.Timestamp("2021-07-01") + # highest close 103 -> amp 0.030; lowest close 100 -> amp 0.010; diff 0.020. + assert out.loc[(d1, _SYM)] == pytest.approx(0.030 - 0.010) + # bare default column name (the window suffix lives on the Factor, like jump). + assert out.name == "minute_ideal_amp" + + +def test_single_day_hand_value_n8_k2(): + closes = [100.0, 101.0, 102.0, 103.0, 104.0, 105.0, 106.0, 107.0] + amps = [0.01, 0.02, 0.03, 0.04, 0.05, 0.06, 0.07, 0.08] + bars = _bars(_session("2021-07-01", closes, amps)) + out = compute_minute_ideal_amplitude( + bars, lookback_days=20, lam=0.25, min_minutes=8 + ) + d1 = pd.Timestamp("2021-07-01") + # k=2: V_high = mean(amp of closes 106,107) = mean(0.07,0.08) = 0.075; + # V_low = mean(amp of closes 100,101) = mean(0.01,0.02) = 0.015. + assert out.loc[(d1, _SYM)] == pytest.approx(0.075 - 0.015) + + +def test_trailing_window_pools_two_days(): + # lookback_days=2, min_minutes=8: day2 pools BOTH days' 4 bars (n=8, k=2); + # day1 alone has only 4 bars (< 8) -> NaN (warm-up handled by the min gate). + closes1 = [100.0, 101.0, 102.0, 103.0] + amps1 = [0.01, 0.05, 0.02, 0.06] + closes2 = [104.0, 105.0, 106.0, 107.0] + amps2 = [0.03, 0.07, 0.04, 0.08] + rows = _session("2021-07-01", closes1, amps1) + _session( + "2021-07-02", closes2, amps2 + ) + bars = _bars(rows) + out = compute_minute_ideal_amplitude( + bars, lookback_days=2, lam=0.25, min_minutes=8 + ) + d1, d2 = pd.Timestamp("2021-07-01"), pd.Timestamp("2021-07-02") + assert np.isnan(out.loc[(d1, _SYM)]) # only 4 bars in day1's window + # day2 pool: closes 100..107, amps as pooled. 2 lowest closes 100(0.01),101(0.05) + # -> V_low 0.03; 2 highest 106(0.04),107(0.08) -> V_high 0.06; diff 0.03. + pooled_closes = closes1 + closes2 + pooled_amps = amps1 + amps2 + assert out.loc[(d2, _SYM)] == pytest.approx( + (0.04 + 0.08) / 2 - (0.01 + 0.05) / 2 + ) + # cross-check against the reference helper on the raw pool ordering. + be = [ + pd.Timestamp("2021-07-01 09:31") + pd.Timedelta(minutes=i) for i in range(4) + ] + [pd.Timestamp("2021-07-02 09:31") + pd.Timedelta(minutes=i) for i in range(4)] + assert out.loc[(d2, _SYM)] == pytest.approx( + _ideal_amp_ref(pooled_closes, pooled_amps, be, 0.25, 8) + ) + + +def test_trailing_window_drops_days_older_than_lookback(): + # 3 days, lookback_days=2, min_minutes=4: day3's 2-day window is {day2, day3}; + # day1's bars must NOT enter day3's pool. Give day1 an extreme amp so leakage + # would be visible, then confirm day3 == pool of day2+day3 only. + closes = [100.0, 101.0, 102.0, 103.0] + amps_hi = [0.90, 0.90, 0.90, 0.90] # day1 extreme (should be dropped at day3) + amps2 = [0.01, 0.05, 0.02, 0.06] + amps3 = [0.03, 0.07, 0.04, 0.08] + rows = ( + _session("2021-07-01", closes, amps_hi) + + _session("2021-07-02", [104.0, 105.0, 106.0, 107.0], amps2) + + _session("2021-07-05", [108.0, 109.0, 110.0, 111.0], amps3) + ) + out = compute_minute_ideal_amplitude( + _bars(rows), lookback_days=2, lam=0.25, min_minutes=8 + ) + d3 = pd.Timestamp("2021-07-05") + pooled_closes = [104.0, 105.0, 106.0, 107.0, 108.0, 109.0, 110.0, 111.0] + pooled_amps = amps2 + amps3 + be = [ + pd.Timestamp("2021-07-02 09:31") + pd.Timedelta(minutes=i) for i in range(4) + ] + [pd.Timestamp("2021-07-05 09:31") + pd.Timedelta(minutes=i) for i in range(4)] + assert out.loc[(d3, _SYM)] == pytest.approx( + _ideal_amp_ref(pooled_closes, pooled_amps, be, 0.25, 8) + ) + + +# --------------------------------------------------------------------------- # +# PIT truncation at 14:50 (leakage) +# --------------------------------------------------------------------------- # +def test_pit_truncation_excludes_post_1450_bars(): + # Bars spanning the 14:50 cutoff: only bar_end with available_time (=bar_end+1min) + # <= 14:50 survive. Build a small session where the post-14:50 minutes carry an + # extreme amplitude that would dominate the cut if it leaked in. + day = "2021-07-01" + pre = [ + (pd.Timestamp(f"{day} 14:46"), _SYM, 101.0, 100.0, 100.0), # amp 0.01, close 100 + (pd.Timestamp(f"{day} 14:47"), _SYM, 104.0, 100.0, 101.0), # amp 0.04, close 101 + (pd.Timestamp(f"{day} 14:48"), _SYM, 102.0, 100.0, 102.0), # amp 0.02, close 102 + (pd.Timestamp(f"{day} 14:49"), _SYM, 103.0, 100.0, 103.0), # amp 0.03, close 103 (last incl.) + ] + post = [ + (pd.Timestamp(f"{day} 14:50"), _SYM, 190.0, 100.0, 104.0), # available 14:51 -> excluded + (pd.Timestamp(f"{day} 14:51"), _SYM, 195.0, 100.0, 105.0), # excluded + (pd.Timestamp(f"{day} 15:00"), _SYM, 199.0, 100.0, 106.0), # excluded + ] + out = compute_minute_ideal_amplitude( + _bars(pre + post), lookback_days=20, lam=0.25, min_minutes=4 + ) + d1 = pd.Timestamp("2021-07-01") + # Only the 4 pre-14:50 bars enter: k=1 -> V_high amp(close 103)=0.03, + # V_low amp(close 100)=0.01 -> 0.02. The extreme post bars never appear. + assert out.loc[(d1, _SYM)] == pytest.approx(0.02) + + +def test_perturbing_post_1450_bars_does_not_change_factor(): + day = "2021-07-01" + pre = [ + (pd.Timestamp(f"{day} 14:46"), _SYM, 101.0, 100.0, 100.0), + (pd.Timestamp(f"{day} 14:47"), _SYM, 104.0, 100.0, 101.0), + (pd.Timestamp(f"{day} 14:48"), _SYM, 102.0, 100.0, 102.0), + (pd.Timestamp(f"{day} 14:49"), _SYM, 103.0, 100.0, 103.0), + ] + a = compute_minute_ideal_amplitude( + _bars(pre + [(pd.Timestamp(f"{day} 14:50"), _SYM, 190.0, 100.0, 104.0)]), + lookback_days=20, lam=0.25, min_minutes=4, + ) + # wildly perturb every post-14:50 bar (amplitude AND close) + b = compute_minute_ideal_amplitude( + _bars(pre + [ + (pd.Timestamp(f"{day} 14:50"), _SYM, 900.0, 100.0, 500.0), + (pd.Timestamp(f"{day} 15:00"), _SYM, 950.0, 100.0, 999.0), + ]), + lookback_days=20, lam=0.25, min_minutes=4, + ) + d1 = pd.Timestamp("2021-07-01") + assert a.loc[(d1, _SYM)] == pytest.approx(b.loc[(d1, _SYM)]) + + +# --------------------------------------------------------------------------- # +# Gates, ties, isolation, guards +# --------------------------------------------------------------------------- # +def test_min_minutes_gate_default_1150_returns_nan_on_small_pool(): + # A handful of bars with the REAL default gate (1150) -> NaN (honest missing). + closes = [100.0 + i for i in range(10)] + amps = [0.01 * (i + 1) for i in range(10)] + bars = _bars(_session("2021-07-01", closes, amps)) + out = compute_minute_ideal_amplitude(bars) # defaults: N=10, lam=0.25, min=1150 + assert out.dropna().empty + + +def test_tie_break_is_deterministic_by_bar_end(): + # Two minutes share close=100; the earlier bar_end must be the "lowest close" + # pick at k=1. Swapping their amps would change the result, so asserting the + # earlier-bar amp locks the (close, bar_end) tie order. + rows = [ + (pd.Timestamp("2021-07-01 09:31"), _SYM, 101.0, 100.0, 100.0), # amp 0.01, close 100 (t0) + (pd.Timestamp("2021-07-01 09:32"), _SYM, 110.0, 100.0, 100.0), # amp 0.10, close 100 (t1) + (pd.Timestamp("2021-07-01 09:33"), _SYM, 102.0, 100.0, 101.0), # amp 0.02, close 101 + (pd.Timestamp("2021-07-01 09:34"), _SYM, 103.0, 100.0, 102.0), # amp 0.03, close 102 (high) + ] + out = compute_minute_ideal_amplitude( + _bars(rows), lookback_days=20, lam=0.25, min_minutes=4 + ) + d1 = pd.Timestamp("2021-07-01") + # k=1: lowest close is the 09:31 bar (earlier bar_end wins the tie) -> V_low 0.01; + # highest close 102 -> V_high 0.03; factor 0.02. (If the 09:32 bar won the tie, + # V_low would be 0.10 and factor -0.07.) + assert out.loc[(d1, _SYM)] == pytest.approx(0.02) + + +def test_per_symbol_isolation(): + closes = [100.0, 101.0, 102.0, 103.0] + amps_a = [0.01, 0.02, 0.03, 0.04] + amps_b = [0.08, 0.07, 0.06, 0.05] + rows = _session("2021-07-01", closes, amps_a, sym="AAA.SZ") + _session( + "2021-07-01", closes, amps_b, sym="BBB.SZ" + ) + out = compute_minute_ideal_amplitude( + _bars(rows), lookback_days=20, lam=0.25, min_minutes=4 + ) + d1 = pd.Timestamp("2021-07-01") + # k=1 each: A -> amp(close103)=0.04 - amp(close100)=0.01 = 0.03; + # B -> amp(close103)=0.05 - amp(close100)=0.08 = -0.03. + assert out.loc[(d1, "AAA.SZ")] == pytest.approx(0.04 - 0.01) + assert out.loc[(d1, "BBB.SZ")] == pytest.approx(0.05 - 0.08) + + +def test_amplitude_guards_drop_bad_bars(): + # low<=0 and high the whole minute block MUST be None (validated by FactorSpec). + for field in ( + "decision_cutoff", "data_lag", "session_open", + "execution_model", "execution_window", + ): + assert getattr(spec, field) is None + + +def test_factor_defaults_match_module_constants(): + assert IDEAL_AMP_LOOKBACK_DAYS == 10 + assert IDEAL_AMP_LAMBDA == 0.25 + assert IDEAL_AMP_MIN_MINUTES == 1150 + f = MinuteIdealAmplitudeFactor() + assert f.lookback_days == IDEAL_AMP_LOOKBACK_DAYS + assert f.name == "minute_ideal_amp_10" + + +def test_factor_subclass_window_tracks_name(): + f = MinuteIdealAmplitudeFactor(lookback_days=5) + assert f.name == "minute_ideal_amp_5" + assert f.spec.factor_id == "minute_ideal_amp_5" + + +def test_factor_compute_selects_preaggregated_column(): + f = MinuteIdealAmplitudeFactor() + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp("2021-07-01"), _SYM)], names=["date", "symbol"] + ) + panel = pd.DataFrame({f.name: [0.25]}, index=idx) + out = f.compute(panel) + assert out.loc[(pd.Timestamp("2021-07-01"), _SYM)] == 0.25 + assert out.name == f.name + + +def test_factor_compute_missing_column_raises(): + f = MinuteIdealAmplitudeFactor() + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp("2021-07-01"), _SYM)], names=["date", "symbol"] + ) + with pytest.raises(ValueError, match="pre-aggregated"): + f.compute(pd.DataFrame({"other": [1.0]}, index=idx)) + + +def test_factor_bad_params_raise(): + with pytest.raises(ValueError): + MinuteIdealAmplitudeFactor(lookback_days=0)