From 2e09d1116b475bb135602032662c7c50200291b7 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 20 Jul 2026 04:07:34 -0700 Subject: [PATCH] feat(factors): ridge minute-return factor + PR-K evaluation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reproduce the FIFTH factor of the Kaiyuan market-microstructure series #27 (量岭分钟收益) and run it through the frozen StandardFactorEvaluator on the same CSI500 cell as PR-C..PR-J. Opens a FOURTH statistic family (a RETURN, after a count / a timing moment / two price levels) and is the report's only NEGATIVE peak/ridge/valley factor, so it also tests whether the SIGN transfers along with the family. The minute taxonomy is REUSED verbatim from data/clean/intraday_volume_prv.py (unmodified by this change): a ridge is an eruptive minute that is not an isolated peak. The four already-merged factors stay bit-identical, locked by hand-value tests here. Pinned choices, none of which the report specifies (all disclosed on the spec and in the config header): - minute return close_t/close_{t-1} - 1 with a WITHIN-DAY lag, so each day's first visible bar carries no return and no return crosses a day boundary; exact 60s adjacency is deliberately not required, so the post-lunch bar returns against the last bar before the gap; - RAW unadjusted closes (the adjustment factor is constant within a day and cancels in the ratio; the within-day lag excludes the one bar that could straddle an ex-date); - the daily aggregate is a SIMPLE SUM, not a compounded product: ridge minutes are non-contiguous, so a holding-period reading does not apply; - a day is valid iff >=100 classifiable bars and >=10 ridge bars carrying a valid return (counted after the guard), NaN below 10 valid days. The runner surfaces net_long_short_by_cost explicitly: this factor's sign of -1 triggers a known frozen-layer defect in aligned_spread_*, which computes sign * (gross - cost) and so adds costs back. The frozen layer is not patched. Adds: data/clean/intraday_ridge_return.py, RidgeMinuteReturnFactor, qt/eval_ridge_minute_return.py, config/phase_k_ridge_minute_return.yaml, a run-eval-ridge-minute-return subcommand, and two test files (+54 tests). --- config/phase_k_ridge_minute_return.yaml | 189 ++++ data/clean/intraday_ridge_return.py | 376 +++++++ factors/compute/intraday_derived.py | 138 +++ qt/cli.py | 65 ++ qt/eval_ridge_minute_return.py | 730 +++++++++++++ tests/test_eval_ridge_minute_return_runner.py | 435 ++++++++ tests/test_ridge_minute_return_factor.py | 997 ++++++++++++++++++ 7 files changed, 2930 insertions(+) create mode 100644 config/phase_k_ridge_minute_return.yaml create mode 100644 data/clean/intraday_ridge_return.py create mode 100644 qt/eval_ridge_minute_return.py create mode 100644 tests/test_eval_ridge_minute_return_runner.py create mode 100644 tests/test_ridge_minute_return_factor.py diff --git a/config/phase_k_ridge_minute_return.yaml b/config/phase_k_ridge_minute_return.yaml new file mode 100644 index 0000000..0557928 --- /dev/null +++ b/config/phase_k_ridge_minute_return.yaml @@ -0,0 +1,189 @@ +# PR-K — Ninth real factor evaluation: RIDGE MINUTE RETURN. +# +# Reproduces the FIFTH factor of the Kaiyuan market-microstructure series #27 (开源证券 +# 《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, reportId 4957417, §3) — the +# "量岭分钟收益因子" — as a first-class RidgeMinuteReturnFactor 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). +# +# WHY THIS FACTOR, NOW: the four prior reproductions from this report covered a COUNT +# (PR-F volume_peak_count, weak), a TIMING moment (PR-H peak_interval_kurtosis, null) and +# two PRICE LEVELS (PR-I valley_relative_vwap and PR-J valley_ridge_vwap_ratio, both +# passing). This factor is a RETURN — a fourth statistic family on the SAME classification +# machine — and it is the report's ONLY NEGATIVE peak/ridge/valley factor, so the run also +# tests whether the SIGN transfers along with the family rather than only the magnitude. +# +# SAME MACHINE as PR-F / PR-H / PR-I / PR-J. The minute classification is REUSED from +# data/clean/intraday_volume_prv.py (not re-implemented, and not modified by this PR): +# PIT-truncate each day at 14:50, classify every visible minute against its SAME-SLOT +# strictly-prior 20-day baseline (ERUPTIVE if vol > mu + sigma, else a VALLEY 量谷); a RIDGE +# 量岭 is an eruptive minute that is NOT an isolated peak. Because the classification is +# shared verbatim, a different verdict here cannot be an artefact of a differently +# implemented taxonomy. +# +# PINNED choices (the report specifies NONE of them; all disclosed on the factor spec): +# (1) the RIDGE mask is 'eruptive AND NOT an isolated peak' (PR-J's, unchanged), so +# valley|peak|ridge stays an exact partition of the classifiable bars and an isolated +# PEAK's return is counted on NEITHER side; +# (2) the minute return is close_t/close_(t-1) - 1 with a WITHIN-DAY lag against the +# previous VISIBLE bar of the same date -- each day's FIRST visible bar therefore +# carries no return and no return ever crosses a day boundary. Exact 60s adjacency is +# deliberately NOT required, so a bar opening a new session block (13:01, after the +# lunch break) returns against the last bar before the gap: a genuine price change +# rather than a silently discarded one. The same-slot baseline defuses the obvious +# selection worry, since the 13:01 slot is compared against its OWN history; +# (3) RAW (unadjusted) closes are correct here: the adjustment factor is constant within +# a day and cancels in close_t/close_(t-1), and the within-day lag already excludes +# the one bar that could straddle an ex-date (same reasoning as PR-I / PR-J / I5b); +# (4) a return is formed only when BOTH closes are finite and strictly positive; the +# guard runs at the return step only, so PR-F's same-slot baseline -- and therefore +# volume_peak_count / peak_interval_kurtosis / valley_relative_vwap / +# valley_ridge_vwap_ratio -- is bit-identical; +# (5) the daily aggregate is a SIMPLE SUM of the selected returns, NOT Pi(1+r)-1. Ridge +# minutes are NON-CONTIGUOUS within the day -- scattered eruptive moments, not a held +# position -- so a holding-period/compounding reading does not apply, and at minute +# scale the two conventions differ negligibly. The report says only "累计收益", so this +# is a disclosed interpretation, not a silent equivalence; +# (6) the trailing 20-day aggregate is LIKEWISE a sum (the factor accumulates across +# days, where PR-J's ratio averaged); +# (7) a day is VALID iff it has >= 100 classifiable bars AND >= 10 ridge bars CARRYING A +# VALID RETURN (counted AFTER the guard, so the day's first bar cannot qualify a day +# it contributes nothing to); NaN below 10 valid days. That is the SAME scarcity +# floor PR-J pinned for its ridge leg -- a ridge bar must erupt AND fail the +# isolation test, making ridges structurally far rarer than valleys -- so the two +# runs' coverage is directly comparable. The runner REPORTS the realized ridge-bar +# distribution, the return-guard attrition, the day-validity rate and the +# counterfactual valid-day count at a floor of 20, so a coverage regression shows up +# as a number rather than being hidden; +# (8) DEVIATION FROM THE REPORT, disclosed and NOT silently equated: everything spans the +# PIT-VISIBLE window 09:31-14:50 only, while the report uses the FULL day. Reading +# the closing auction would be lookahead at our 14:50 decision time. +# +# Pre-registered sign = -1 (report full-market RankIC -6.29% / RankICIR -3.55, long leg +# 7.47%/yr, long-short 14.98%/yr, IR 1.73, max drawdown 13.84%, monthly win rate 70.3%). +# The report gives NO CSI500 sub-domain figure for this factor, so none is quoted -- the +# spec must not fabricate one. Semantics per the report: ridge minutes are retail +# follow-the-crowd trading, and their accumulated return measures that crowd's +# OVER-REACTION -- the more ridge-minute return a stock has piled up, the more +# over-extended it is and the worse it performs going forward. NOTE the report is a +# MONTHLY, market-cap + industry neutral FULL-MARKET series on Wind data -- our eval cell +# is CSI500 daily with industry + size neutral, so those numbers are a LOOSE reference only +# and are NOT written in as expected values. Raw minute volume (cached as-is) has split-day +# magnitude jumps that pollute the 20-day sigma; the report (Wind) does not adjust for this +# either, so it is disclosed and NOT corrected. +# +# ⚠️ SIGN = -1 TRIGGERS A KNOWN FROZEN-LAYER DEFECT. The evaluator's aligned_spread_* fields +# compute sign * (gross - cost), which for a negative sign becomes -gross + cost -- the +# trading cost is ADDED BACK instead of deducted. The frozen layer is deliberately NOT +# patched by this PR. Read this run's tradability from net_long_short_by_cost (sign- +# agnostic and correct), which the runner extracts and logs explicitly; treat every +# aligned_spread_* field in the two reports as unreliable. +# +# Eval CELL is IDENTICAL to PR-C .. PR-J: 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. Because the cell +# matches its siblings exactly, this run is directly comparable to them. The evaluator runs +# TWICE (see qt/eval_ridge_minute_return.py): once with NO book (Incremental NOT_ASSESSED) +# and once with the confirmed book -- which matters here because a trailing SUM OF RETURNS +# is a plausible cousin of a reversal signal, and the with-book run is what says whether +# this is a genuinely new bet. + +project: + name: quantitative_trading_pr_k_ridge_minute_return + 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: ridge_minute_return_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 (ridge_minute_return_20) 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_ridge_return.py b/data/clean/intraday_ridge_return.py new file mode 100644 index 0000000..1597dea --- /dev/null +++ b/data/clean/intraday_ridge_return.py @@ -0,0 +1,376 @@ +"""RIDGE MINUTE-RETURN factor (PR-K). + +Reproduces the FIFTH factor of the Kaiyuan market-microstructure series #27 (开源证券 +《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, 2025-07-20, reportId 4957417, +§3): "计算过去 20 日量岭时点的累计收益作为量岭分钟收益因子，衡量个人投资者交易的收益贡献, +个人投资者交易的过度反应，导致量岭分钟收益因子为显著负向因子". + +Same MACHINE as PR-F / PR-H / PR-I / PR-J — what changes is the STATISTIC. The four prior +reproductions from this report covered a COUNT (weak), a TIMING moment (null) and two +PRICE LEVELS (both passed). This is the RETURN family, a fourth statistic type, and the +report's ONLY NEGATIVE peak/ridge/valley factor — so it also tests whether the SIGN +transfers along with the family. + +The classification is REUSED verbatim from :mod:`data.clean.intraday_volume_prv` +(``prepare_visible_minute_bars`` + ``peak_mask_for_symbol``, whose ``ridge`` column PR-J +already exposed) — same same-slot strictly-prior μ+kσ eruptive test, same classifiable +rule, same valid-day floor. Nothing about the taxonomy is re-implemented or modified here, +so the five factors can never drift apart. + +PINNED choices (deliberate and DISCLOSED, not tuned knobs; the report is silent on every +one of them, and each is reproduced on the factor spec so a reader sees what was assumed): + + 1. THE RIDGE MASK IS ``eruptive & ~peak`` (PR-J's, unchanged). A RIDGE (量岭) is an + eruptive minute that is NOT an isolated peak — wider than the literal "eruptive next + to an eruptive", because it also covers session-boundary eruptions and eruptions + whose neighbour is unclassifiable. It is the exact COMPLEMENT of PR-F's conservative + peak test, so ``valley | peak | ridge == classifiable`` stays an exact partition and + an isolated PEAK's return is counted on NEITHER side. + 2. THE MINUTE RETURN IS ``close_t / close_{t-1} - 1`` WITH A WITHIN-DAY LAG. The + predecessor is the previous VISIBLE bar of the SAME trade date, so each day's FIRST + visible bar has no return and never enters the sum. The lag deliberately does NOT + require exact 60s adjacency: a bar that opens a new session block (in practice 13:01, + after the lunch break) returns against the last bar before the gap, which is a + genuine price change of the stock over that interval. Dropping such bars instead + would silently discard the post-lunch minute whenever it is a ridge. The obvious + selection worry is defused by the taxonomy itself: the 13:01 slot is compared against + its OWN same-slot history, so a systematically busy post-lunch minute is not + systematically eruptive. + 3. RAW (UNADJUSTED) CLOSES. The cached minute bars are unadjusted, and that is CORRECT + here: a split/dividend adjustment factor is constant WITHIN a day, so it cancels + exactly in ``close_t / close_{t-1}``; and because the within-day lag already excludes + each day's first bar, no return ever straddles an ex-date boundary. (Same reasoning + PR-I / PR-J's ratios, PR-D's amplitude and I5b's price-limit checks rely on.) + 4. POSITIVE-CLOSE GUARD. A return is formed only when BOTH closes are finite and + strictly positive; otherwise the bar contributes nothing and is not counted towards + the ridge floor. The guard runs at the return step only — never before classification + — because PR-F's same-slot μ/σ baseline must stay bit-identical. + 5. THE DAILY AGGREGATE IS A SIMPLE SUM ``s_day = Σ r_t`` over the selected ridge bars, + NOT ``Π(1+r_t) - 1``. The report says only "累计收益" (cumulative return). Ridge + minutes are NON-CONTIGUOUS within the day — they are scattered eruptive moments, not + a held position — so a holding-period/compounding reading does not apply to them; and + at minute scale the two conventions differ negligibly anyway. The choice is therefore + the simple sum, disclosed rather than silently equated to the report's wording. + 6. THE TRAILING AGGREGATE IS ALSO A SUM. "过去 20 日累计" is read as the sum of the daily + sums over the trailing window, i.e. the factor accumulates across days rather than + averaging (which is what PR-J's ratio did). + 7. A RIDGE-BAR FLOOR OF 10, counted AFTER the return guard. Ridge bars are STRUCTURALLY + scarce: a minute must erupt (a minority event by construction, since the threshold is + its own strictly-prior same-slot μ+σ) AND fail the isolation test. This is the SAME + floor PR-J pinned for its ridge leg, so the two runs' coverage is directly + comparable, and the realized ridge-bar distribution plus the day-validity rate are + REPORTED by the runner rather than left implicit. + 8. BOTH THE SUM AND THE COUNT COVER THE PIT-VISIBLE WINDOW ONLY (09:31–14:50), not the + full session. A NECESSARY DEVIATION from the report, which uses the whole day: the + standing 14:50 decision cutoff truncates history days and the signal day identically, + and reading the closing auction would be lookahead at our decision time. + +Factor value: ``ridge_minute_return_20`` = the SUM of ``s_day`` over the symbol's most +recent ``lookback_days`` (=20) VALID trading days INCLUDING ``d``. A day is VALID iff it +has at least ``min_classifiable`` (=100) classifiable bars (PR-F's gate, unchanged) and at +least ``min_ridge_bars`` (=10) ridge bars carrying a valid return. Fewer than +``min_valid_days`` (=10) valid days in the trailing window -> NaN (honest missing; the +runner discloses the coverage). Values are emitted only on valid days. + +Pre-registered sign = -1. The report's full-market RankIC is -6.29% / RankICIR -3.55 (long +leg 7.47%/yr, long-short 14.98%/yr, IR 1.73, max drawdown 13.84%, monthly win rate 70.3%); +it gives NO CSI500 sub-domain figure for this factor, so none is quoted. Semantics per the +report: ridge minutes are retail follow-the-crowd trading, and their return contribution +measures that crowd's OVER-REACTION — the higher the accumulated ridge-minute return, the +more over-extended the stock and the worse it performs going forward. NOTE the report is a +MONTHLY, market-cap + industry neutral full-market series on Wind data while our eval cell +is CSI500 daily with industry + size neutralization, so its numbers are a LOOSE reference +only (disclosed, never mislabeled, never written in as an expected value). + +The value at ``d`` 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 +from data.clean.intraday_volume_prv import ( + VOLUME_PRV_BASELINE_DAYS, + VOLUME_PRV_BASELINE_MIN_OBS, + VOLUME_PRV_MIN_CLASSIFIABLE, + VOLUME_PRV_MIN_VALID_DAYS, + VOLUME_PRV_SIGMA_K, + peak_mask_for_symbol, + prepare_visible_minute_bars, +) + +# Factor DEFINITION constants (pinned interpretations of the report; NOT tuned knobs). +# The classification constants are IMPORTED from PR-F, never redefined. +RIDGE_RETURN_LOOKBACK_DAYS = 20 # trailing VALID trading-day SUM window, includes d +# The SAME scarcity floor PR-J pinned for its ridge leg (module docstring §7), so the two +# runs' ridge coverage is directly comparable. +RIDGE_RETURN_MIN_RIDGE_BARS = 10 # min RETURN-CARRYING ridge bars for a valid day + +# The extra 1min column this family needs on top of PR-F's (volume): the close, which +# turns consecutive bars into a minute return. +_CLOSE = "close" + +# Per-day diagnostic columns (the ridge-scarcity disclosure the task card requires). +# ``ridge_bars`` is every ridge minute; ``ridge_return_bars`` is the subset that carries a +# valid return — the one the floor actually gates on — so the attrition is visible. +DIAGNOSTIC_COLUMNS = ("classifiable_bars", "ridge_bars", "ridge_return_bars", "valid") + + +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 ridge_minute_return_by_day( + work: pd.DataFrame, + *, + min_ridge_bars: int = RIDGE_RETURN_MIN_RIDGE_BARS, + min_classifiable: int = VOLUME_PRV_MIN_CLASSIFIABLE, + with_diagnostics: bool = False, +) -> pd.Series | tuple[pd.Series, pd.DataFrame]: + """Daily summed ridge-minute return for ONE symbol, on VALID days only. + + ``work`` is one symbol's frame as returned by + :func:`~data.clean.intraday_volume_prv.peak_mask_for_symbol`, which must have been + built from bars prepared with ``extra_columns=("close",)`` so the close is available + alongside the ``ridge`` / ``classifiable`` masks. ``peak_mask_for_symbol`` returns the + rows sorted by ``(trade_date, bar_end)``, which is exactly the order the WITHIN-DAY lag + needs. + + The minute return is ``close_t / close_{t-1} - 1`` against the previous VISIBLE bar of + the SAME day (PINNED §2), guarded so both closes are finite and strictly positive + (§4), and the day's value is the SIMPLE SUM of those returns over the ridge bars (§5) + — an isolated PEAK's return is never included (§1). + + Args: + work: one symbol's classified minute frame (see above). + min_ridge_bars: minimum ridge bars CARRYING A VALID RETURN for a valid day. + min_classifiable: minimum classifiable bars for a valid day (PR-F's gate). + with_diagnostics: also return the per-day bar-count frame, so the caller can + REPORT the ridge-scarcity distribution instead of only gating on it. + + Returns: + Series indexed by ``trade_date`` (ascending) holding the summed ridge-minute + return for the days that clear both gates — invalid days are ABSENT, not NaN, so + they do not occupy a slot in the caller's trailing window (the same rule PR-F / + PR-H / PR-I / PR-J use). With ``with_diagnostics=True``, a ``(daily, diagnostics)`` + pair where ``diagnostics`` is indexed by EVERY day present in ``work`` and carries + :data:`DIAGNOSTIC_COLUMNS`. + """ + close = work[_CLOSE].to_numpy(dtype=float) + # WITHIN-DAY lag: the previous visible bar of the SAME trade date. grouping by + # trade_date is what stops a return from ever crossing a day boundary, so each day's + # first visible bar gets a NaN predecessor and drops out below. + prev_close = ( + work.groupby("trade_date", sort=False)[_CLOSE].shift(1).to_numpy(dtype=float) + ) + # Positive-close guard: a non-finite or non-positive close on EITHER side carries no + # usable return. Applied HERE, at the return step, never before classification — + # PR-F's same-slot baseline must stay bit-identical. + has_return = ( + np.isfinite(close) & (close > 0.0) + & np.isfinite(prev_close) & (prev_close > 0.0) + ) + # Denominator neutralized where the guard failed, so no divide-by-zero / inf is ever + # formed; those entries are discarded by the np.where immediately after. + safe_prev = np.where(has_return, prev_close, 1.0) + ret = np.where(has_return, close / safe_prev - 1.0, 0.0) + + ridge = work["ridge"].to_numpy(dtype=bool) + ridge_return = ridge & has_return + + per_bar = pd.DataFrame( + { + "trade_date": work["trade_date"].to_numpy(), + # SIMPLE SUM of the selected returns (PINNED §5) — not Π(1+r)-1. + "ridge_return_sum": np.where(ridge_return, ret, 0.0), + "ridge_bars": ridge.astype(np.int64), + "ridge_return_bars": ridge_return.astype(np.int64), + "classifiable_bars": work["classifiable"] + .to_numpy(dtype=bool) + .astype(np.int64), + } + ) + agg = per_bar.groupby("trade_date", sort=True).sum() + + # Two validity gates (§ module docstring / task card §1.5): PR-F's classifiable floor + # (unchanged) and enough RETURN-CARRYING ridge bars. + valid = (agg["classifiable_bars"] >= min_classifiable) & ( + agg["ridge_return_bars"] >= min_ridge_bars + ) + ok = agg.loc[valid] + if ok.empty: + daily = pd.Series([], index=pd.DatetimeIndex([], name="trade_date"), dtype=float) + else: + daily = ok["ridge_return_sum"].astype(float) + + if not with_diagnostics: + return daily + diagnostics = agg[["classifiable_bars", "ridge_bars", "ridge_return_bars"]].copy() + diagnostics["valid"] = valid + return daily, diagnostics + + +def _ridge_return_sum_for_symbol( + g: pd.DataFrame, + *, + baseline_days: int, + baseline_min_obs: int, + sigma_k: float, + lookback_days: int, + min_valid_days: int, + min_classifiable: int, + min_ridge_bars: int, + collect_diagnostics: bool = False, +) -> tuple[list[pd.Timestamp], list[float], pd.DataFrame | None]: + """Daily ridge-minute-return values for ONE symbol from its PIT-visible bars. + + Classifies the minutes with the REUSED :func:`peak_mask_for_symbol`, reduces each valid + day to its summed ridge-minute return, then takes the trailing-``lookback_days``- + valid-day SUM. No cross-symbol leakage (``g`` is one symbol's slice) and no lookahead + (the baseline is strictly prior, the window is trailing). + """ + work = peak_mask_for_symbol( + g, + baseline_days=baseline_days, + baseline_min_obs=baseline_min_obs, + sigma_k=sigma_k, + ) + result = ridge_minute_return_by_day( + work, + min_ridge_bars=min_ridge_bars, + min_classifiable=min_classifiable, + with_diagnostics=collect_diagnostics, + ) + if collect_diagnostics: + daily, diagnostics = result + else: + daily, diagnostics = result, None + if daily.empty: + return [], [], diagnostics + + # SUM over the trailing lookback_days VALID days (including d); NaN until + # min_valid_days valid days have accumulated. Emitted only on valid days. + rolled = daily.sort_index().rolling(lookback_days, min_periods=min_valid_days).sum() + days = [pd.Timestamp(d).normalize() for d in rolled.index] + return days, list(rolled.to_numpy(dtype=float)), diagnostics + + +def compute_ridge_minute_return( + bars: pd.DataFrame, + *, + lookback_days: int = RIDGE_RETURN_LOOKBACK_DAYS, + baseline_days: int = VOLUME_PRV_BASELINE_DAYS, + baseline_min_obs: int = VOLUME_PRV_BASELINE_MIN_OBS, + sigma_k: float = VOLUME_PRV_SIGMA_K, + min_valid_days: int = VOLUME_PRV_MIN_VALID_DAYS, + min_classifiable: int = VOLUME_PRV_MIN_CLASSIFIABLE, + min_ridge_bars: int = RIDGE_RETURN_MIN_RIDGE_BARS, + decision_time: str = DEFAULT_DECISION_TIME, + name: str = "ridge_minute_return", + diagnostics_out: list | None = None, +) -> pd.Series: + """PIT-safe daily "ridge minute-return" factor from 1min ``bars``. + + Takes normalized 1min ``bars``, PIT-truncates each day at ``decision_time``, classifies + every visible minute with the REUSED PR-F taxonomy, sums each valid day's ridge-minute + returns, and returns the trailing-``lookback_days``-VALID-day SUM of those daily sums. + See the module docstring for the LOCKED definition and the eight pinned choices. + + Args: + bars: normalized 1min bars (:mod:`data.clean.intraday_schema`), + ``MultiIndex(time, symbol)``. May carry one or many symbols; the grouping is + strictly per symbol (no cross-symbol leakage). + lookback_days: trailing VALID trading-day window summed (definition). + baseline_days: strictly-prior same-slot baseline window in trading days (PR-F). + baseline_min_obs: minimum same-slot observations for a classifiable bar (PR-F). + sigma_k: eruptive threshold multiplier ``k`` in ``vol > μ + k*σ`` (PR-F). + min_valid_days: minimum valid days in the trailing window for a finite value. + min_classifiable: a day needs at least this many classifiable bars (PR-F). + min_ridge_bars: a day needs at least this many ridge bars CARRYING A VALID RETURN. + decision_time: per-bar PIT cutoff time-of-day (default 14:50:00). + name: the returned Series name (the factor-panel column name). + diagnostics_out: optional list the per-symbol day-level bar-count frames are + APPENDED to (each carries a ``symbol`` column), so a caller can report the + ridge-scarcity distribution. Purely observational — supplying it does not + change the returned factor. + + 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 baseline_days < 2: + # Need >= 2 baseline observations so a ddof=1 std of the same-slot volume is + # defined. + raise ValueError(f"baseline_days must be >= 2; got {baseline_days!r}.") + if baseline_min_obs < 2: + raise ValueError(f"baseline_min_obs must be >= 2; got {baseline_min_obs!r}.") + if sigma_k < 0.0: + raise ValueError(f"sigma_k must be >= 0; got {sigma_k!r}.") + if min_valid_days < 1: + raise ValueError(f"min_valid_days must be >= 1; got {min_valid_days!r}.") + if min_classifiable < 1: + raise ValueError(f"min_classifiable must be >= 1; got {min_classifiable!r}.") + if min_ridge_bars < 1: + raise ValueError(f"min_ridge_bars must be >= 1; got {min_ridge_bars!r}.") + if len(bars) == 0: + return _empty_series(name) + + # extra_columns=("close",) is the ONLY difference from the PR-F / PR-H entry points: + # the close rides along on the surviving rows, and the truncation / volume guard / + # slot assignment are untouched. + visible = prepare_visible_minute_bars( + bars, decision_time=decision_time, extra_columns=(_CLOSE,) + ) + if visible.empty: + return _empty_series(name) + + collect = diagnostics_out is not None + index_tuples: list[tuple] = [] + values: list[float] = [] + for sym, g in visible.groupby(SYMBOL_LEVEL, sort=True): + days, vals, diagnostics = _ridge_return_sum_for_symbol( + g.reset_index(drop=True), + baseline_days=baseline_days, + baseline_min_obs=baseline_min_obs, + sigma_k=sigma_k, + lookback_days=lookback_days, + min_valid_days=min_valid_days, + min_classifiable=min_classifiable, + min_ridge_bars=min_ridge_bars, + collect_diagnostics=collect, + ) + if collect and diagnostics is not None and not diagnostics.empty: + frame = diagnostics.copy() + frame[SYMBOL_LEVEL] = str(sym) + diagnostics_out.append(frame) + 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__ = [ + "DIAGNOSTIC_COLUMNS", + "RIDGE_RETURN_LOOKBACK_DAYS", + "RIDGE_RETURN_MIN_RIDGE_BARS", + "compute_ridge_minute_return", + "ridge_minute_return_by_day", +] diff --git a/factors/compute/intraday_derived.py b/factors/compute/intraday_derived.py index 4e3744a..b3c5148 100644 --- a/factors/compute/intraday_derived.py +++ b/factors/compute/intraday_derived.py @@ -58,6 +58,10 @@ PEAK_INTERVAL_LOOKBACK_DAYS, PEAK_INTERVAL_MIN_INTERVALS, ) +from data.clean.intraday_ridge_return import ( + RIDGE_RETURN_LOOKBACK_DAYS, + RIDGE_RETURN_MIN_RIDGE_BARS, +) from data.clean.intraday_valley_ridge_vwap import ( VALLEY_RIDGE_LOOKBACK_DAYS, VALLEY_RIDGE_MIN_RIDGE_BARS, @@ -886,12 +890,146 @@ def compute(self, panel: pd.DataFrame) -> pd.Series: return panel[self.name].rename(self.name) +class RidgeMinuteReturnFactor(Factor): + """Ridge minute-return factor (daily signal, minute-derived). + + ``compute`` reads the pre-aggregated daily column the runner placed on the panel + (produced by ``data.clean.intraday_ridge_return.compute_ridge_minute_return``); it does + NO minute work of its own, mirroring :class:`JumpAmountCorrFactor` / + :class:`MinuteIdealAmplitudeFactor` / :class:`AmpMarginalAnomalyVolFactor` / + :class:`VolumePeakCountFactor` / :class:`IntradayAmpCutFactor` / + :class:`PeakIntervalKurtosisFactor` / :class:`ValleyRelativeVwapFactor` / + :class:`ValleyRidgeVwapRatioFactor` and the value / financial factors that surface an + enriched column. + + Args: + lookback_days: trailing VALID trading-day window summed; 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"ridge_minute_return_{RIDGE_RETURN_LOOKBACK_DAYS}" + + def __init__(self, lookback_days: int = RIDGE_RETURN_LOOKBACK_DAYS) -> None: + if not isinstance(lookback_days, int) or lookback_days < 1: + raise ValueError( + f"ridge-minute-return lookback_days must be a positive integer; got " + f"{lookback_days!r}." + ) + self._lookback_days = lookback_days + self.name = f"ridge_minute_return_{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: this is the report's ONLY NEGATIVE peak/ridge/valley factor + (full-market RankIC -6.29%, RankICIR -3.55, long leg 7.47%/yr, long-short + 14.98%/yr, IR 1.73, max drawdown 13.84%, monthly win rate 70.3%). The report gives + NO CSI500 sub-domain figure for this factor, so none is quoted here. Semantics per + the report: ridge minutes are retail follow-the-crowd trading and their + accumulated return measures that crowd's OVER-REACTION — the more ridge-minute + return a stock has piled up, the more over-extended it is and the worse it + performs going forward. The sign is fixed BEFORE the run (a validated prototype + must reproduce it). NOTE the report is a MONTHLY, market-cap + industry neutral + full-market series on Wind data while our eval cell is CSI500 daily with industry + + size neutral, so the report numbers are a LOOSE reference only (disclosed, never + mislabeled, never written in as an expected value). 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 enough VALID days accumulate), not a fixed leading count — the honest NaN + rate is reported by data_coverage. + + The description spells out the RELATION TO PR-F..PR-J (same reused classification, + a RETURN statistic instead of a count / timing moment / price level) plus the + pinned choices — above all the SIMPLE-SUM convention, the within-day return lag and + the raw closes, none of which the report specifies. + """ + return FactorSpec( + factor_id=self.name, + version="1.0", + description=( + f"Ridge minute-return (Kaiyuan microstructure series #27, FIFTH factor " + f"量岭分钟收益). SAME minute classification as PR-F volume_peak_count / " + f"PR-H peak_interval_kurtosis / PR-I valley_relative_vwap / PR-J " + f"valley_ridge_vwap_ratio (REUSED from data.clean.intraday_volume_prv, " + f"not re-implemented): 1min bars PIT-truncated at 14:50, a minute is " + f"ERUPTIVE if vol > μ + {VOLUME_PRV_SIGMA_K:g}σ of its SAME-SLOT " + f"strictly-prior {VOLUME_PRV_BASELINE_DAYS}-day baseline, and a RIDGE " + f"(量岭) is an eruptive minute that is NOT an isolated peak. NEW " + f"STATISTIC vs PR-F..PR-J: a RETURN rather than a count, a timing moment " + f"or a price level — each day sums the minute returns of its ridge bars " + f"and the factor sums those daily sums over the trailing " + f"{self._lookback_days} VALID days. PINNED choices (the report specifies " + f"none of them): (1) the RIDGE mask is 'eruptive AND NOT an isolated " + f"peak', keeping valley|peak|ridge an exact partition of the classifiable " + f"bars — an isolated PEAK's return is counted on NEITHER side; (2) the " + f"minute return is close_t/close_(t-1) - 1 with a WITHIN-DAY lag against " + f"the previous VISIBLE bar of the same date, so each day's FIRST visible " + f"bar carries no return and no return ever crosses a day boundary; exact " + f"60s adjacency is deliberately NOT required, so a bar opening a new " + f"session block (13:01, after lunch) returns against the last bar before " + f"the gap — a genuine price change rather than a discarded one; (3) RAW " + f"unadjusted closes are correct here because the adjustment factor is " + f"constant within a day and cancels in the ratio, and the within-day lag " + f"already excludes the one bar that could straddle an ex-date; (4) a " + f"return is formed only when both closes are finite and strictly positive " + f"(guard applied at the return step only, so PR-F's baseline is " + f"untouched); (5) the daily aggregate is a SIMPLE SUM Σr, NOT a compound " + f"Π(1+r)-1 — ridge minutes are non-contiguous within the day so a " + f"holding-period reading does not apply, and at minute scale the two " + f"differ negligibly; (6) the trailing aggregate is likewise a SUM across " + f"days; (7) a day is VALID iff it has >= {VOLUME_PRV_MIN_CLASSIFIABLE} " + f"classifiable bars AND >= {RIDGE_RETURN_MIN_RIDGE_BARS} ridge bars " + f"CARRYING A VALID RETURN (counted AFTER the guard) — the same scarcity " + f"floor PR-J pinned for its ridge leg, since a ridge bar must erupt AND " + f"fail the isolation test; the realized ridge-bar distribution and " + f"day-validity rate are REPORTED by the runner rather than left implicit; " + f"(8) DEVIATION FROM THE REPORT, disclosed: everything spans the " + f"PIT-VISIBLE window 09:31-14:50 only, not the full session — reading the " + f"close would be lookahead at our 14:50 decision time. NaN below " + f"{VOLUME_PRV_MIN_VALID_DAYS} valid days. Derived from 1min bars but a " + f"DAILY signal traded close-to-close." + ), + 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=("volume", "close"), + family="microstructure", + min_history_bars=0, + ) + + def compute(self, panel: pd.DataFrame) -> pd.Series: + """Select the pre-aggregated daily ridge-minute-return column off ``panel``. + + The runner runs ``compute_ridge_minute_return`` per symbol 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"RidgeMinuteReturnFactor needs the pre-aggregated '{self.name}' column " + f"on the panel (produced upstream by compute_ridge_minute_return and " + f"joined by the runner); panel has {list(panel.columns)}." + ) + return panel[self.name].rename(self.name) + + __all__ = [ "AmpMarginalAnomalyVolFactor", "IntradayAmpCutFactor", "JumpAmountCorrFactor", "MinuteIdealAmplitudeFactor", "PeakIntervalKurtosisFactor", + "RidgeMinuteReturnFactor", "ValleyRelativeVwapFactor", "ValleyRidgeVwapRatioFactor", "VolumePeakCountFactor", diff --git a/qt/cli.py b/qt/cli.py index 8ab71ac..55a2f0d 100644 --- a/qt/cli.py +++ b/qt/cli.py @@ -378,6 +378,64 @@ def _cmd_run_eval_valley_relative_vwap(args: argparse.Namespace) -> int: return 0 +def _fmt_metric(value: object, spec: str = ".4f") -> str: + """Format a metric that a SKIPPED section may legitimately leave absent. + + ``stability_cost`` is Skipped whenever no period has a quantile label (a coverage + degenerate case that the scarcity-gated ridge family can plausibly hit), and its + payload keys then come back as None. Rendering "n/a" keeps a thin run's summary + readable instead of raising a TypeError out of the print itself — after the reports + have already been written, and outside the command's error handling. + """ + if value is None: + return "n/a" + try: + return format(value, spec) + except (TypeError, ValueError): + return str(value) + + +def _cmd_run_eval_ridge_minute_return(args: argparse.Namespace) -> int: + """Run the two real ridge-minute-return evaluations (cache-only) + reports.""" + from qt.eval_ridge_minute_return import run_eval_ridge_minute_return + + try: + result = run_eval_ridge_minute_return(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 + net = ( + " ".join( + f"{m:g}x={v:+.6f}" for m, v in sorted(nb["net_long_short_by_cost"].items()) + ) + or "n/a" + ) + print( + f"OK run-eval-ridge-minute-return: 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" + # Ridge bars are structurally scarce and the return guard narrows them further; + # the realized distribution and the day-validity rate are surfaced so a coverage + # regression is visible. + f"{result.ridge_coverage.render()}\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" + # sign=-1 makes the frozen layer's aligned_spread_* unreliable (it adds costs back + # instead of deducting them); the sign-agnostic net spreads are printed instead. + f"net long-short by cost (aligned_spread_* UNRELIABLE at sign=-1): {net}\n" + f"turnover={_fmt_metric(nb['long_short_turnover'])} " + f"rank_autocorr_lag1={_fmt_metric(nb['rank_autocorr_lag1'])} " + f"half_life={_fmt_metric(nb['half_life_periods'], '.2f')}\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_run_eval_valley_ridge_vwap_ratio(args: argparse.Namespace) -> int: """Run the two real valley/ridge VWAP-ratio evaluations (cache-only) + reports.""" from qt.eval_valley_ridge_vwap_ratio import run_eval_valley_ridge_vwap_ratio @@ -593,6 +651,13 @@ def build_parser() -> argparse.ArgumentParser: p_vrr.add_argument("--config", required=True, help="Path to the YAML config.") p_vrr.set_defaults(func=_cmd_run_eval_valley_ridge_vwap_ratio) + p_rmr = sub.add_parser( + "run-eval-ridge-minute-return", + help="Run the ridge-minute-return factor evaluation (CSI500, cache-only).", + ) + p_rmr.add_argument("--config", required=True, help="Path to the YAML config.") + p_rmr.set_defaults(func=_cmd_run_eval_ridge_minute_return) + 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_ridge_minute_return.py b/qt/eval_ridge_minute_return.py new file mode 100644 index 0000000..9312607 --- /dev/null +++ b/qt/eval_ridge_minute_return.py @@ -0,0 +1,730 @@ +"""run-eval-ridge-minute-return: the ninth real factor evaluation (PR-K). + +Reproduces the FIFTH factor of the Kaiyuan market-microstructure series #27 (开源证券 +《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, reportId 4957417, §3) — the +"量岭分钟收益" factor — as a first-class +:class:`~factors.compute.intraday_derived.RidgeMinuteReturnFactor` 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 .. PR-J used. + +This is the FIFTH reproduction from the same report and it opens a FOURTH statistic +family. The four prior ones covered a COUNT (PR-F, weak), a TIMING moment (PR-H, null) and +two PRICE LEVELS (PR-I / PR-J, both passed). This factor is a RETURN — and it is the +report's ONLY NEGATIVE peak/ridge/valley factor, so the run also tests whether the SIGN +transfers along with the family, not just the magnitude. + +The factor is a DAILY signal derived DIRECTLY from the 1min cache (see +``data.clean.intraday_ridge_return.compute_ridge_minute_return``): PIT-truncate each day at +14:50, classify every visible minute with the taxonomy REUSED from PR-F +(``data.clean.intraday_volume_prv``: same-slot strictly-prior μ+σ eruptive test; a RIDGE is +an eruptive minute that is not an isolated peak), form each minute's return against the +previous VISIBLE bar of the SAME day, sum those returns over the day's ridge bars, and sum +the daily sums over the trailing 20 VALID trading days. It is executed CLOSE-TO-CLOSE +(daily default), so ``is_intraday=False``. The eval CELL is identical to PR-C..PR-J, so +this run is directly comparable to its siblings — same classification, different statistic. + +RIDGE SCARCITY IS MEASURED, NOT ASSUMED, exactly as in PR-J. Ridge bars are structurally +rare (a minute must erupt AND fail the isolation test), and this factor narrows them +further: only ridge bars that CARRY A VALID RETURN count, so the day's first visible bar +drops out even when it is a ridge. The runner therefore collects the realized per-day +distribution of BOTH counts plus the day-validity rate across the whole universe and logs +them, so the return-guard attrition and any coverage regression against PR-J are visible as +numbers rather than hidden behind a threshold. + +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 the accumulated +ridge-minute return adds alpha BEYOND value / low-vol. That check matters here because a +trailing sum of returns is a plausible cousin of a REVERSAL signal, and the with-book run +is the one that says whether this is a genuinely new bet. + +⚠️ THIS FACTOR'S PRE-REGISTERED SIGN IS -1, which triggers a KNOWN DEFECT in the frozen +evaluation layer: its ``aligned_spread_*`` fields compute ``sign * (gross - cost)``, which +for a negative sign becomes ``-gross + cost`` — i.e. the trading cost is ADDED BACK rather +than deducted. The frozen layer is deliberately NOT patched here. Any reading of this run's +tradability must use ``net_long_short_by_cost`` (which is sign-agnostic and correct) and +must treat ``aligned_spread_*`` as unreliable. :func:`extract_metrics` therefore surfaces +the cost-scenario net spreads explicitly. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from pathlib import Path + +import numpy as np +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_ridge_return import ( + RIDGE_RETURN_LOOKBACK_DAYS, + RIDGE_RETURN_MIN_RIDGE_BARS, + compute_ridge_minute_return, +) +from data.clean.intraday_schema import RAW_INTRADAY_FREQ, normalize_intraday_bars +from data.clean.intraday_volume_prv import ( + VOLUME_PRV_BASELINE_DAYS, + VOLUME_PRV_BASELINE_MIN_OBS, + VOLUME_PRV_MIN_CLASSIFIABLE, + VOLUME_PRV_MIN_VALID_DAYS, + VOLUME_PRV_SIGMA_K, +) +from data.clean.schema import CORE_COLUMNS, DATE_LEVEL +from factors.compute.intraday_derived import RidgeMinuteReturnFactor +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_ridge_minute_return" +_REPORT_STEM = "eval_ridge_minute_return" + +# Percentiles reported for the realized ridge-bar distribution (the scarcity disclosure). +_RIDGE_PCTL = (0, 10, 25, 50, 75, 90, 100) + +# The counterfactual floor the coverage disclosure also reports, so PR-K's ridge coverage +# is directly comparable to PR-J's (which used 20 for its VALLEY leg). +_COMPARISON_FLOOR = 20 + + +# --------------------------------------------------------------------------- # +# Ridge-scarcity coverage (measured, never assumed) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class RidgeReturnCoverage: + """Realized ridge-bar distribution + day-validity rate over the whole universe. + + Built from the per-day diagnostics the factor emits, so the numbers describe the days + the factor actually saw. ``symbol_days`` counts EVERY symbol-day with visible bars, + including the leading warm-up days that have no same-slot baseline yet; + ``classifiable_days`` counts those that clear PR-F's classifiable floor. The headline + ``validity_rate`` is taken over ``classifiable_days``, because a day with no baseline + fails for a PR-F warm-up reason rather than a ridge-scarcity one and would otherwise + make the ridge gate look worse than it is — both denominators are reported so the + reader can check that framing. + + TWO ridge counts are tracked, because this factor gates on the narrower one: total + ``ridge_bars`` and the ``ridge_return_bars`` subset that carries a valid minute return + (the day's first visible bar is excluded by the within-day lag even when it is a + ridge). Reporting both makes the return-guard attrition visible instead of implicit. + The gate-failure counts are NOT mutually exclusive and are reported for shape, not as + a partition. + """ + + symbol_days: int + classifiable_days: int + valid_days: int + ridge_return_percentiles: tuple[tuple[int, float], ...] + ridge_return_mean: float + ridge_bars_mean: float + ridge_bars_median: float + days_below_ridge_gate: int + days_below_classifiable_gate: int + # Counterfactual: how many days would survive at the higher floor PR-J used for its + # VALLEY leg. Quantifies exactly what the scarcity-driven threshold buys. + valid_days_at_comparison_floor: int + # The gates this run actually applied, so the disclosure can never describe the module + # defaults while the run used something else. + min_ridge_bars: int = RIDGE_RETURN_MIN_RIDGE_BARS + min_classifiable: int = VOLUME_PRV_MIN_CLASSIFIABLE + comparison_floor: int = _COMPARISON_FLOOR + + @property + def validity_rate(self) -> float: + """Valid days as a share of CLASSIFIABLE days (see the class docstring).""" + if not self.classifiable_days: + return float("nan") + return self.valid_days / self.classifiable_days + + @property + def return_guard_attrition(self) -> float: + """Share of ridge bars LOST to the return guard (mean over classifiable days).""" + if not np.isfinite(self.ridge_bars_mean) or self.ridge_bars_mean <= 0.0: + return float("nan") + return 1.0 - self.ridge_return_mean / self.ridge_bars_mean + + def render(self) -> str: + """One-line, secret-free summary for the run log and the CLI.""" + pctl = " ".join(f"p{p}={v:.0f}" for p, v in self.ridge_return_percentiles) + return ( + f"ridge scarcity: symbol_days={self.symbol_days} " + f"classifiable_days={self.classifiable_days} " + f"valid_days={self.valid_days} ({self.validity_rate:.1%} of classifiable) " + f"ridge_return_bars[{pctl} mean={self.ridge_return_mean:.1f}] " + f"ridge_bars_mean={self.ridge_bars_mean:.1f} " + f"ridge_bars_median={self.ridge_bars_median:.0f} " + f"return_guard_attrition={self.return_guard_attrition:.1%} " + f"below_ridge_gate({self.min_ridge_bars})={self.days_below_ridge_gate} " + f"below_classifiable_gate({self.min_classifiable})=" + f"{self.days_below_classifiable_gate} " + f"valid_if_floor_were_{self.comparison_floor}=" + f"{self.valid_days_at_comparison_floor}" + ) + + +def summarize_ridge_return_coverage( + frames: list[pd.DataFrame], + *, + min_ridge_bars: int = RIDGE_RETURN_MIN_RIDGE_BARS, + min_classifiable: int = VOLUME_PRV_MIN_CLASSIFIABLE, + comparison_floor: int = _COMPARISON_FLOOR, +) -> RidgeReturnCoverage: + """Reduce the per-symbol day-level diagnostics to the scarcity disclosure. + + The floors must be the ones the RUN applied, not the module defaults — otherwise the + disclosure would describe gates that were never enforced. + """ + gates = dict( + min_ridge_bars=min_ridge_bars, + min_classifiable=min_classifiable, + comparison_floor=comparison_floor, + ) + empty = tuple((p, float("nan")) for p in _RIDGE_PCTL) + if not frames: + return RidgeReturnCoverage( + symbol_days=0, + classifiable_days=0, + valid_days=0, + ridge_return_percentiles=empty, + ridge_return_mean=float("nan"), + ridge_bars_mean=float("nan"), + ridge_bars_median=float("nan"), + days_below_ridge_gate=0, + days_below_classifiable_gate=0, + valid_days_at_comparison_floor=0, + **gates, + ) + diag = pd.concat(frames, ignore_index=True) + classifiable = diag["classifiable_bars"].to_numpy(dtype=float) + valid = diag["valid"].to_numpy(dtype=bool) + # The bar-count distributions describe the days that had a fair chance: a warm-up day + # with no same-slot baseline has zero of everything and would only drag the + # percentiles towards zero for a reason that has nothing to do with ridge scarcity. + scored = classifiable >= min_classifiable + ridge_ret = diag.loc[scored, "ridge_return_bars"].to_numpy(dtype=float) + ridge_all = diag.loc[scored, "ridge_bars"].to_numpy(dtype=float) + # The counterfactual raises the ridge floor, leaving every other gate exactly as it was. + at_comparison = valid & ( + diag["ridge_return_bars"].to_numpy(dtype=float) >= comparison_floor + ) + return RidgeReturnCoverage( + symbol_days=int(len(diag)), + classifiable_days=int(scored.sum()), + valid_days=int(valid.sum()), + ridge_return_percentiles=( + tuple((p, float(np.percentile(ridge_ret, p))) for p in _RIDGE_PCTL) + if ridge_ret.size + else empty + ), + ridge_return_mean=float(ridge_ret.mean()) if ridge_ret.size else float("nan"), + ridge_bars_mean=float(ridge_all.mean()) if ridge_all.size else float("nan"), + ridge_bars_median=float(np.median(ridge_all)) if ridge_all.size else float("nan"), + days_below_ridge_gate=int((ridge_ret < min_ridge_bars).sum()), + days_below_classifiable_gate=int((~scored).sum()), + valid_days_at_comparison_floor=int(at_comparison.sum()), + **gates, + ) + + +# --------------------------------------------------------------------------- # +# Minute loading (cache-only, per-symbol -> memory-bounded) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _RidgeReturnMinuteLoad: + """Diagnostics from the cache-only per-symbol minute read + aggregation.""" + + factor: pd.Series # MultiIndex(date, symbol) raw ridge-minute-return sum + 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) + ridge_coverage: RidgeReturnCoverage + + +def _load_ridge_minute_return_panel( + cfg: RootConfig, + symbols: list[str], + spec: FactorSpec, + logger, + *, + lookback_days: int, + baseline_days: int, + baseline_min_obs: int, + sigma_k: float, + min_valid_days: int, + min_classifiable: int, + min_ridge_bars: int, +) -> _RidgeReturnMinuteLoad: + """Compute the raw ridge-minute-return 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). The classification (reused from PR-F) and + the return reduction run entirely inside ``compute_ridge_minute_return`` on 1min bars, + which read ``volume`` and ``close``. The per-day bar-count diagnostics are collected + alongside so the ridge scarcity can be REPORTED, not assumed; collecting them does not + change the factor. + """ + 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] = [] + diagnostics: list[pd.DataFrame] = [] + 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_ridge_minute_return( + bars, + lookback_days=lookback_days, + baseline_days=baseline_days, + baseline_min_obs=baseline_min_obs, + sigma_k=sigma_k, + min_valid_days=min_valid_days, + min_classifiable=min_classifiable, + min_ridge_bars=min_ridge_bars, + name=spec.factor_id, + diagnostics_out=diagnostics, + ) + 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), + ) + + coverage = summarize_ridge_return_coverage( + diagnostics, + min_ridge_bars=min_ridge_bars, + min_classifiable=min_classifiable, + ) + if not series: + raise ValueError( + "run-eval-ridge-minute-return blocked: no requested symbol produced a cached " + f"ridge-minute-return value over [{cfg.data.start}, {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), + ) + logger.info("%s", coverage.render()) + return _RidgeReturnMinuteLoad( + factor=factor, + requested=len(symbols), + covered=tuple(covered), + empty_symbols=tuple(empty), + raw_rows=raw_rows, + live_calls=0, + ridge_coverage=coverage, + ) + + +# --------------------------------------------------------------------------- # +# 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. + + Beyond the fields PR-C..PR-J surfaced, this also extracts the TRADABILITY comparison + quantities the PR-K analysis needs alongside PR-I / PR-J: the long-short leg turnover, + the net long-short spread at EACH cost scenario, the lag-1 rank autocorrelation with + its implied half-life, and the cross-section size. + + ⚠️ ``net_long_short_by_cost`` is deliberately the tradability field surfaced here. The + frozen layer's ``aligned_spread_*`` computes ``sign * (gross - cost)``, which for this + factor's pre-registered sign of -1 becomes ``-gross + cost`` — costs are added back + rather than deducted, so those fields are UNRELIABLE for a negative-sign factor. The + frozen layer is not patched; the correct, sign-agnostic net spreads are reported + instead. + """ + verdict = report.require_verdict() + pred = _section_payload(report, "predictive_power") + coverage = _section_payload(report, "data_coverage") + incr = _section_payload(report, "purity") + ret_risk = _section_payload(report, "return_risk") + stability = _section_payload(report, "stability_cost") + autocorr = dict(stability.get("factor_rank_autocorr_by_lag", {}) or {}) + 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": coverage.get("settled_rebalances"), + "effective_samples": coverage.get("effective_samples"), + "span_days": coverage.get("span_days"), + "cross_section_size_mean": coverage.get("cross_section_size_mean"), + "cross_section_size_median": coverage.get("cross_section_size_median"), + "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"), + "monotonicity_spearman": ret_risk.get("monotonicity_spearman"), + "gross_long_short_mean": ret_risk.get("gross_long_short_mean"), + # The correct tradability read for a negative-sign factor (see the docstring). + "net_long_short_by_cost": dict(ret_risk.get("net_long_short_by_cost", {}) or {}), + "long_short_turnover": stability.get("turnover_mean_long_short_legs"), + "rank_autocorr_lag1": autocorr.get(1), + "half_life_periods": stability.get("half_life_periods"), + } + + +# --------------------------------------------------------------------------- # +# 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-ridge-minute-return 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 RidgeMinuteReturnEvalResult: + """Immutable summary of one run-eval-ridge-minute-return 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 + ridge_coverage: RidgeReturnCoverage + 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-ridge-minute-return 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-ridge-minute-return needs data.cache.enabled=true (it reads the " + "persistent tushare cache and never warms live)." + ) + if cfg.universe.type != "index": + raise ValueError( + "run-eval-ridge-minute-return needs universe.type='index' (PIT membership, " + f"e.g. 000905.SH for CSI500); got {cfg.universe.type!r}." + ) + if not cfg.processing.neutralize.enabled: + raise ValueError( + "run-eval-ridge-minute-return expects processing.neutralize.enabled=true " + "(industry + size neutralization, matching the report's neutral column and " + "the EvalConfig declaration)." + ) + + +def run_eval_ridge_minute_return(config_path: str) -> RidgeMinuteReturnEvalResult: + """Run the two real ridge-minute-return 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 = RidgeMinuteReturnFactor(lookback_days=RIDGE_RETURN_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) + + # Ridge-minute-return factor: cache-only per-symbol aggregation -> raw -> process. + load = _load_ridge_minute_return_panel( + cfg, symbols, spec, logger, + lookback_days=RIDGE_RETURN_LOOKBACK_DAYS, + baseline_days=VOLUME_PRV_BASELINE_DAYS, + baseline_min_obs=VOLUME_PRV_BASELINE_MIN_OBS, + sigma_k=VOLUME_PRV_SIGMA_K, + min_valid_days=VOLUME_PRV_MIN_VALID_DAYS, + min_classifiable=VOLUME_PRV_MIN_CLASSIFIABLE, + min_ridge_bars=RIDGE_RETURN_MIN_RIDGE_BARS, + ) + # 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. + factor_raw = load.factor[ + load.factor.index.get_level_values(DATE_LEVEL).isin(panel_dates) + ] + factor_processed = _process_factors(cfg, factor_raw.to_frame(spec.factor_id), panel) + factor_series = factor_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"], + ) + # The tradability read for a NEGATIVE-sign factor: aligned_spread_* is unreliable here + # (frozen-layer defect, see the module docstring), so log the net spreads explicitly. + logger.info( + "net long-short by cost (aligned_spread_* is UNRELIABLE for sign=-1): %s", + no_book_metrics["net_long_short_by_cost"], + ) + + return RidgeMinuteReturnEvalResult( + 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, + ridge_coverage=load.ridge_coverage, + 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__ = [ + "RidgeMinuteReturnEvalResult", + "RidgeReturnCoverage", + "evaluate_two_runs", + "extract_metrics", + "run_eval_ridge_minute_return", + "summarize_ridge_return_coverage", +] diff --git a/tests/test_eval_ridge_minute_return_runner.py b/tests/test_eval_ridge_minute_return_runner.py new file mode 100644 index 0000000..e68a30c --- /dev/null +++ b/tests/test_eval_ridge_minute_return_runner.py @@ -0,0 +1,435 @@ +"""PR-K runner: cache-only minute loader, ridge coverage, two-run eval (no net).""" + +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 data.clean.intraday_ridge_return import RIDGE_RETURN_MIN_RIDGE_BARS +from factors.compute.intraday_derived import RidgeMinuteReturnFactor +from qt.config import ( + AlphaCfg, + BacktestCfg, + CacheCfg, + CostCfg, + DataCfg, + FactorCfg, + OutputCfg, + PortfolioCfg, + RootConfig, + UniverseCfg, +) +from qt.eval_ridge_minute_return import ( + _load_ridge_minute_return_panel, + evaluate_two_runs, + extract_metrics, + summarize_ridge_return_coverage, +) + + +# --------------------------------------------------------------------------- # +# Minute cache-only loader +# --------------------------------------------------------------------------- # +def _stored_rows(sym, day, closes, vols): + """Build STORED_COLUMNS-shaped 1min rows for one session of CONSECUTIVE minutes. + + Bar i sits at ``09:31 + i`` minutes (all inside the 14:50 PIT window). ``volume`` + drives the classification and ``close`` -- set INDEPENDENTLY -- drives the returns. + """ + base = pd.Timestamp(day) + pd.Timedelta("09:31:00") + rows = [] + for i, (c, v) in enumerate(zip(closes, vols)): + be = base + pd.Timedelta(minutes=i) + rows.append( + { + "symbol": sym, + "bar_end": be, + "source_trade_time": be, + "open": float(c), + "high": float(c), + "low": float(c), + "close": float(c), + "volume": float(v), + "amount": float(c) * float(v), + "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(), + ) + + +# Three constant background days (baseline mu=100, sigma=0 -> eruptive threshold 100), then +# the hand-computed CASE A test day: 16 slots, ridge runs at (3,4) and (8,9) with volume +# 200, an ISOLATED peak at slot 12 (volume 300, return +9.0, excluded), and closes chosen so +# the four ridge returns are +0.5 / -0.5 / +3.0 / -0.5 -> daily SUM = +2.5 (a compounded +# reading would give -0.5 instead). +_DAYS = ("2021-07-01", "2021-07-02", "2021-07-03") +_TEST_DAY = "2021-07-04" +_N_SLOTS = 16 +_RIDGES = (3, 4, 8, 9) +_PEAK = 12 +_CASE_A_SUM = 2.5 +_CASE_A_CLOSES = [ + 100.0, 100.0, 100.0, 150.0, 75.0, 75.0, 75.0, 75.0, + 300.0, 150.0, 150.0, 150.0, 1500.0, 150.0, 150.0, 150.0, +] +_BG_CLOSES = [100.0] * _N_SLOTS +_BG_VOLS = [100.0] * _N_SLOTS + + +def _case_a(): + vols = [100.0] * _N_SLOTS + for s in _RIDGES: + vols[s] = 200.0 + vols[_PEAK] = 300.0 + return list(_CASE_A_CLOSES), vols + + +# Small gates so a 4-day cache produces a value (baseline needs >= 2 prior obs; the test day +# carries 4 return-bearing ridge bars). lookback_days=1 so the test day's value IS its own +# daily sum -- the background days have no eruption at all, hence no ridge, hence no valid +# day, so they cannot accumulate in. +_LOAD_KW = dict( + lookback_days=1, baseline_days=20, baseline_min_obs=2, + sigma_k=1.0, min_valid_days=1, min_classifiable=1, min_ridge_bars=1, +) + + +def _seed_symbol(store, sym): + for day in _DAYS: + store.upsert( + INTRADAY_ENDPOINT, sym, "1min", + _stored_rows(sym, day, _BG_CLOSES, _BG_VOLS), KEY_COLS, + ) + store.upsert( + INTRADAY_ENDPOINT, sym, "1min", + _stored_rows(sym, _TEST_DAY, *_case_a()), KEY_COLS, + ) + + +def test_minute_loader_is_cache_only_and_discloses_empty(tmp_path): + root = tmp_path / "cache" + store = IntradayParquetStore(str(root)) + # Two symbols with cached minute bars; a third has NO cached minute (must be disclosed + # as empty, never fetched). + for sym in ("AAA.SZ", "BBB.SZ"): + _seed_symbol(store, sym) + + cfg = _min_config(root, "2021-07-01", "2021-07-04") + spec = RidgeMinuteReturnFactor().spec + logger = logging.getLogger("test.rmr.loader") + load = _load_ridge_minute_return_panel( + cfg, ["AAA.SZ", "BBB.SZ", "CCC.SZ"], spec, logger, **_LOAD_KW + ) + 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() + # the test day carries the hand-computed summed ridge-minute return + d = pd.Timestamp(_TEST_DAY) + assert load.factor.loc[(d, "AAA.SZ")] == pytest.approx(_CASE_A_SUM) + assert load.factor.loc[(d, "BBB.SZ")] == pytest.approx(_CASE_A_SUM) + assert set(load.factor.index.get_level_values("symbol")) == {"AAA.SZ", "BBB.SZ"} + # a day where NOTHING erupts has no ridge at all -> invalid -> no value emitted + bg = pd.Timestamp(_DAYS[-1]) + assert (bg, "AAA.SZ") not in load.factor.index + + +def test_minute_loader_reports_the_ridge_scarcity_distribution(tmp_path): + """The ridge distribution + validity rate must be MEASURED on the real days.""" + root = tmp_path / "cache" + store = IntradayParquetStore(str(root)) + _seed_symbol(store, "AAA.SZ") + cfg = _min_config(root, "2021-07-01", "2021-07-04") + load = _load_ridge_minute_return_panel( + cfg, ["AAA.SZ"], RidgeMinuteReturnFactor().spec, + logging.getLogger("test.rmr.cov"), **_LOAD_KW, + ) + cov = load.ridge_coverage + # four symbol-days seen, but the first two have no same-slot baseline yet (it needs 2 + # prior observations) -> only two are CLASSIFIABLE, and of those only the test day has a + # ridge at all -> validity rate 1/2 over the days that had a fair chance. + assert cov.symbol_days == 4 + assert cov.classifiable_days == 2 + assert cov.days_below_classifiable_gate == 2 # the warm-up days, counted separately + assert cov.valid_days == 1 + assert cov.validity_rate == pytest.approx(0.5) + # the flat background day has zero ridge bars, the test day has four; warm-up days are + # excluded from the distribution so they cannot drag it towards zero + assert dict(cov.ridge_return_percentiles)[100] == pytest.approx(4.0) + assert dict(cov.ridge_return_percentiles)[0] == pytest.approx(0.0) + assert cov.ridge_return_mean == pytest.approx(2.0) + # every ridge here sits past the day's first bar, so nothing is lost to the guard + assert cov.ridge_bars_mean == pytest.approx(2.0) + assert cov.return_guard_attrition == pytest.approx(0.0) + # the floors reported are the ones this RUN applied (_LOAD_KW uses 1, not the module + # defaults) -- otherwise the disclosure would describe gates nobody enforced + assert cov.min_ridge_bars == 1 + assert cov.days_below_ridge_gate == 1 # only the flat day (0 ridges) is below 1 + line = cov.render() + assert "ridge scarcity" in line and "valid_days=1" in line + assert "below_ridge_gate(1)" in line + + +def test_coverage_reports_the_return_guard_attrition(): + """Ridge bars lost to the within-day lag must be VISIBLE, not silently absorbed.""" + diag = pd.DataFrame( + { + "classifiable_bars": [240, 240], + "ridge_bars": [10, 30], + # one ridge on each day is the day's first visible bar -> no return + "ridge_return_bars": [9, 29], + "valid": [False, True], + }, + index=pd.DatetimeIndex(pd.bdate_range("2022-01-03", periods=2), name="trade_date"), + ) + cov = summarize_ridge_return_coverage([diag]) + assert cov.ridge_bars_mean == pytest.approx(20.0) + assert cov.ridge_return_mean == pytest.approx(19.0) + assert cov.return_guard_attrition == pytest.approx(1.0 - 19.0 / 20.0) + # the gate is applied to the RETURN-carrying count, so the 9-ridge day falls below 10 + assert cov.days_below_ridge_gate == 1 + assert "return_guard_attrition" in cov.render() + + +def test_summarize_ridge_coverage_counterfactual_at_the_comparison_floor(): + """The disclosure quantifies exactly what the scarcity floor buys (vs PR-J's 20).""" + diag = pd.DataFrame( + { + "classifiable_bars": [240, 240, 240, 240], + "ridge_bars": [4, 12, 25, 30], + "ridge_return_bars": [4, 12, 25, 30], + # as the factor would mark them under the default floor of 10 + "valid": [False, True, True, True], + }, + index=pd.DatetimeIndex(pd.bdate_range("2022-01-03", periods=4), name="trade_date"), + ) + cov = summarize_ridge_return_coverage([diag]) + assert cov.symbol_days == 4 + assert cov.classifiable_days == 4 # every day clears PR-F's classifiable floor + assert cov.valid_days == 3 + # raising the ridge floor to PR-J's 20 would keep only the 25 / 30 days + assert cov.valid_days_at_comparison_floor == 2 + assert cov.days_below_ridge_gate == 1 + assert cov.ridge_return_mean == pytest.approx((4 + 12 + 25 + 30) / 4) + # defaults are the PINNED production floors when the caller does not override + assert cov.min_ridge_bars == RIDGE_RETURN_MIN_RIDGE_BARS == 10 + assert f"below_ridge_gate({RIDGE_RETURN_MIN_RIDGE_BARS})" in cov.render() + + +def test_summarize_ridge_coverage_handles_no_frames(): + cov = summarize_ridge_return_coverage([]) + assert cov.symbol_days == 0 + assert cov.classifiable_days == 0 + assert cov.valid_days == 0 + assert np.isnan(cov.validity_rate) + assert np.isnan(cov.return_guard_attrition) + assert cov.render() # renders without dividing by zero + + +def test_minute_loader_blocks_when_nothing_cached(tmp_path): + cfg = _min_config(tmp_path / "empty", "2021-07-01", "2021-07-04") + spec = RidgeMinuteReturnFactor().spec + with pytest.raises(ValueError, match="no requested symbol produced"): + _load_ridge_minute_return_panel( + cfg, ["ZZZ.SZ"], spec, logging.getLogger("test.rmr.block"), **_LOAD_KW + ) + + +# --------------------------------------------------------------------------- # +# 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="ridge_minute_return_20" + ) + 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 = RidgeMinuteReturnFactor().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 + 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_extract_metrics_surfaces_the_pr_k_comparison_quantities(tmp_path): + """PR-K's analysis needs turnover / net-by-cost / autocorr / cross-section size. + + These are the quantities compared head-on with PR-I's and PR-J's for the same cell, so + the runner must extract them rather than leaving them buried in the JSON. + """ + factor, book, price = _synthetic_panels() + reports = evaluate_two_runs( + factor, RidgeMinuteReturnFactor().spec, _eval_cfg(), price, book, + universe_symbols=tuple(sorted(set(book.index.get_level_values("symbol")))), + fee_rate=0.001, + report_dir=tmp_path, + ) + m = extract_metrics(reports.no_book) + for key in ( + "long_short_turnover", "rank_autocorr_lag1", "half_life_periods", + "cross_section_size_mean", "cross_section_size_median", + "monotonicity_spearman", "gross_long_short_mean", + ): + assert key in m, key + assert m[key] is not None, key + # the cost gradient is a per-scenario mapping covering the declared scenarios + net = m["net_long_short_by_cost"] + assert isinstance(net, dict) and net + assert {1.0, 2.0, 4.0} <= {float(k) for k in net} + # higher costs can only reduce a gross-positive spread; the scenarios must not be equal + assert len({round(float(v), 12) for v in net.values()}) > 1 + + +def test_absent_stability_metrics_render_as_na_not_a_crash(): + """A Skipped ``stability_cost`` section must not blow up the CLI summary line. + + The section is Skipped when no period carries a quantile label, which the + scarcity-gated ridge family can plausibly reach on a thin universe. The metrics then + come back as None, and the summary is printed AFTER the reports are already on disk + and OUTSIDE the command's error handling -- so an unguarded ``:.4f`` would surface as + a raw TypeError on an otherwise successful run. + """ + from qt.cli import _fmt_metric + + assert _fmt_metric(None) == "n/a" + assert _fmt_metric(None, ".2f") == "n/a" + assert _fmt_metric(2.108441) == "2.1084" + assert _fmt_metric(8.061448, ".2f") == "8.06" + # the exact expression the summary builds, with every stability field absent + absent = {"long_short_turnover": None, "rank_autocorr_lag1": None, + "half_life_periods": None} + line = ( + f"turnover={_fmt_metric(absent['long_short_turnover'])} " + f"rank_autocorr_lag1={_fmt_metric(absent['rank_autocorr_lag1'])} " + f"half_life={_fmt_metric(absent['half_life_periods'], '.2f')}" + ) + assert line == "turnover=n/a rank_autocorr_lag1=n/a half_life=n/a" + + +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, RidgeMinuteReturnFactor().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_ridge_minute_return_factor.py b/tests/test_ridge_minute_return_factor.py new file mode 100644 index 0000000..f063ddf --- /dev/null +++ b/tests/test_ridge_minute_return_factor.py @@ -0,0 +1,997 @@ +"""PR-K: RIDGE MINUTE-RETURN factor. + +Same volume classification as PR-F / PR-H / PR-I / PR-J (REUSED verbatim, not +re-implemented): a "ridge" (量岭) is an ERUPTIVE minute that is NOT an isolated peak. What +is NEW here is the STATISTIC -- a RETURN rather than a count, a timing moment or a price +level. The factor is the trailing-20-valid-day SUM of each day's summed ridge-minute +returns. Sign is pre-registered -1 (the report's only negative peak/ridge/valley factor). + +Hand cases build a constant BACKGROUND of prior days so the same-slot baseline is exact +(mu=100, sigma=0 -> the eruptive threshold is exactly 100), then a test day whose VOLUMES +place ridge runs / an isolated peak at chosen slots and whose CLOSES are set +INDEPENDENTLY, so every minute return -- and therefore the daily sum -- is known in closed +form. The hand closes move in halvings and doublings, so each return is exactly +representable in binary and the expected sums are exact rather than approximate. + +Two properties get deliberately aggressive test data: + + * the SUM-vs-COMPOUNDING pin. At minute scale the two conventions are numerically + indistinguishable, so a test built on realistic returns could not tell them apart. The + hand days therefore use returns of +/-50%..+300%, where sum and Pi(1+r)-1 differ by + more than a factor of four, and the test asserts the SUM. + * the LEAKAGE test has TEETH: the post-14:50 block is shown to move the value materially + the moment the truncation is disabled, so "perturbation changes nothing" cannot pass + because the late bars were inert. + +Every hand day also contains an isolated PEAK whose minute return is wildly larger than +any ridge's: it must never leak into the sum, which is the one way this factor can +silently degenerate into "eruptive-minute return". +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from data.clean.intraday_ridge_return import ( + RIDGE_RETURN_LOOKBACK_DAYS, + RIDGE_RETURN_MIN_RIDGE_BARS, + compute_ridge_minute_return, + ridge_minute_return_by_day, +) +from data.clean.intraday_schema import empty_intraday_bars, normalize_intraday_bars +from data.clean.intraday_volume_prv import ( + VOLUME_PRV_BASELINE_DAYS, + VOLUME_PRV_BASELINE_MIN_OBS, + VOLUME_PRV_MIN_VALID_DAYS, + VOLUME_PRV_SIGMA_K, + peak_mask_for_symbol, + prepare_visible_minute_bars, +) +from factors.compute.intraday_derived import RidgeMinuteReturnFactor +from factors.spec import FactorSpec + +_SYM = "000001.SZ" + + +def _bars(rows): + """rows = [(time, symbol, close, volume), ...] -> normalized 1min bars. + + ``close`` is set INDEPENDENTLY of ``volume`` -- that is the point: the volume decides + WHICH minutes are ridges and the close decides WHAT they returned. ``amount`` rides + along as ``close * volume`` (this factor never reads it, but a realistic frame keeps + the reuse locks below honest). ``normalize_intraday_bars`` sets + ``available_time = bar_end + 1min``, so the 14:50 PIT cutoff excludes any bar with + ``bar_end >= 14:50``. + """ + close = [float(r[2]) for r in rows] + volume = [float(r[3]) 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": close, + "high": close, + "low": close, + "close": close, + "volume": volume, + "amount": [c * v for c, v in zip(close, volume)], + } + ) + return normalize_intraday_bars(df, freq="1min") + + +def _session(day, closes, vols, sym=_SYM, start="09:31:00"): + """One session of CONSECUTIVE 1-minute bars carrying ``closes`` / ``vols``.""" + base = pd.Timestamp(day) + pd.Timedelta(start) + return [ + (base + pd.Timedelta(minutes=i), sym, c, v) + for i, (c, v) in enumerate(zip(closes, vols)) + ] + + +def _background(n_days, n_slots, sym=_SYM, start_day="2021-07-01"): + """``n_days`` prior days of flat close-100 / volume-100 sessions. + + Same-slot baseline becomes mu=100, sigma=0 -> the eruptive threshold is exactly 100, + so on the test day a volume of 100 is a VALLEY and anything above erupts. The flat + closes also mean every background day's ridge-return sum is trivially zero -- and, + having no eruption at all, no background day is ever VALID. + """ + rows = [] + for i in range(n_days): + day = (pd.Timestamp(start_day) + pd.Timedelta(days=i)).strftime("%Y-%m-%d") + rows += _session(day, [100.0] * n_slots, [100.0] * n_slots, sym=sym) + return rows + + +_BG_DAYS = 10 +_TEST_DAY = pd.Timestamp("2021-07-11") + +# Gates small enough that the single engineered test day is the only VALID day; the +# ridge-bar and valid-day floors get their own dedicated tests below. +_KW = dict( + baseline_days=VOLUME_PRV_BASELINE_DAYS, + baseline_min_obs=VOLUME_PRV_BASELINE_MIN_OBS, + sigma_k=VOLUME_PRV_SIGMA_K, + lookback_days=RIDGE_RETURN_LOOKBACK_DAYS, + min_valid_days=1, + min_classifiable=1, + min_ridge_bars=1, +) + + +# --------------------------------------------------------------------------- # +# Hand-computed daily sums (2 non-trivial cases, exact binary fractions) +# --------------------------------------------------------------------------- # +# CASE A -- 16 slots, POSITIVE daily sum. +# ridge runs at (3,4) and (8,9): each bar volume 200. Every one of the four has an +# ERUPTIVE neighbour, so none is an isolated peak -> all four are ridges. +# closes: 100,100,100, 150, 75, 75,75,75, 300, 150, 150,150, 1500, 150,150,150 +# -> r3 = 150/100-1 = +0.5 (ridge) +# r4 = 75/150-1 = -0.5 (ridge) +# r8 = 300/75 -1 = +3.0 (ridge) +# r9 = 150/300-1 = -0.5 (ridge) +# s_day = 0.5 - 0.5 + 3.0 - 0.5 = +2.5 <- SIMPLE SUM (the pinned convention) +# Pi(1+r)-1 = 1.5*0.5*4.0*0.5 - 1 = -0.5 <- compounding would give something else +# an ISOLATED eruption at slot 12 (volume 300) IS a peak; its return is +9.0, so if it +# leaked into the sum the value would be 11.5 instead of 2.5 (the discriminator). +_CASE_A_N = 16 +_CASE_A_RIDGES = (3, 4, 8, 9) +_CASE_A_PEAK = 12 +_CASE_A_CLOSES = [ + 100.0, 100.0, 100.0, 150.0, 75.0, 75.0, 75.0, 75.0, + 300.0, 150.0, 150.0, 150.0, 1500.0, 150.0, 150.0, 150.0, +] +_CASE_A_SUM = 2.5 +_CASE_A_COMPOUNDED = 1.5 * 0.5 * 4.0 * 0.5 - 1.0 # == -0.5 +_CASE_A_PEAK_RETURN = 9.0 + + +def _case_a_day(): + vols = [100.0] * _CASE_A_N + for s in _CASE_A_RIDGES: + vols[s] = 200.0 + vols[_CASE_A_PEAK] = 300.0 + return list(_CASE_A_CLOSES), vols + + +# CASE B -- 18 slots, NEGATIVE daily sum (so a sign / absolute-value bug cannot pass both). +# ridge runs at (2,3), (7,8), (13,14); isolated peak at slot 16. +# closes halve at each ridge except the last, which doubles: +# r2=r3=r7=r8=r13=-0.5, r14=+1.0 -> s_day = -2.5 + 1.0 = -1.5 +# Pi(1+r)-1 = 0.5^5 * 2 - 1 = -0.9375 <- again distinguishable from the sum +# the peak at slot 16 returns +9.0 and must be excluded. +_CASE_B_N = 18 +_CASE_B_RIDGES = (2, 3, 7, 8, 13, 14) +_CASE_B_PEAK = 16 +_CASE_B_CLOSES = [ + 100.0, 100.0, 50.0, 25.0, 25.0, 25.0, 25.0, 12.5, + 6.25, 6.25, 6.25, 6.25, 6.25, 3.125, 6.25, 6.25, 62.5, 6.25, +] +_CASE_B_SUM = -1.5 +_CASE_B_COMPOUNDED = 0.5**5 * 2.0 - 1.0 # == -0.9375 + + +def _case_b_day(): + vols = [100.0] * _CASE_B_N + for s in _CASE_B_RIDGES: + vols[s] = 300.0 + vols[_CASE_B_PEAK] = 500.0 + return list(_CASE_B_CLOSES), vols + + +def test_hand_value_case_a_positive_sum(): + closes, vols = _case_a_day() + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", closes, vols) + out = compute_ridge_minute_return(_bars(rows), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(_CASE_A_SUM) + assert out.loc[(_TEST_DAY, _SYM)] > 0.0 + + +def test_hand_value_case_b_negative_sum(): + closes, vols = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", closes, vols) + out = compute_ridge_minute_return(_bars(rows), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(_CASE_B_SUM) + assert out.loc[(_TEST_DAY, _SYM)] < 0.0 + + +def test_daily_aggregation_is_a_simple_sum_not_a_compounded_return(): + """THE pinned-convention discriminator: Sum(r), NOT Pi(1+r)-1. + + The report says only "累计收益" (cumulative return). Ridge minutes are NON-CONTIGUOUS + within the day, so a holding-period/compounding reading does not apply, and the PINNED + choice is the simple sum. At minute scale the two are numerically indistinguishable, + so this test uses hand days with +/-50%..+300% returns where they differ by more than + a factor of four -- if the implementation ever compounded, both assertions below fail. + """ + for closes_vols, expected_sum, compounded in ( + (_case_a_day(), _CASE_A_SUM, _CASE_A_COMPOUNDED), + (_case_b_day(), _CASE_B_SUM, _CASE_B_COMPOUNDED), + ): + closes, vols = closes_vols + rows = _background(_BG_DAYS, len(closes)) + _session("2021-07-11", closes, vols) + out = compute_ridge_minute_return(_bars(rows), **_KW) + got = out.loc[(_TEST_DAY, _SYM)] + # the two conventions really are far apart on this data (not a vacuous check) + assert abs(expected_sum - compounded) > 0.4 + assert got == pytest.approx(expected_sum) + assert got != pytest.approx(compounded) + + +def test_isolated_peak_return_is_excluded_from_the_sum(): + """The sum runs over ``eruptive & ~peak``, NOT over all eruptive bars. + + Moving ONLY the isolated peak's close must leave the factor untouched; if the + implementation summed every eruptive bar the value would move. This is the single most + important discriminator in the file. + """ + closes, vols = _case_a_day() + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", closes, vols) + base = compute_ridge_minute_return(_bars(rows), **_KW) + + closes2, vols2 = _case_a_day() + closes2[_CASE_A_PEAK] = 15_000.0 # peak return +99.0 instead of +9.0 + rows2 = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", closes2, vols2) + moved = compute_ridge_minute_return(_bars(rows2), **_KW) + + key = (_TEST_DAY, _SYM) + assert base.loc[key] == pytest.approx(_CASE_A_SUM) + assert moved.loc[key] == pytest.approx(_CASE_A_SUM) + + # ...and the day really does contain an isolated peak carrying a big return, so the + # assertion above is not vacuous + work = peak_mask_for_symbol( + prepare_visible_minute_bars( + _bars(rows), extra_columns=("close",) + ).reset_index(drop=True) + ) + day = work[work["trade_date"] == _TEST_DAY].reset_index(drop=True) + assert bool(day.loc[_CASE_A_PEAK, "peak"]) is True + assert bool(day.loc[_CASE_A_PEAK, "ridge"]) is False + peak_ret = ( + day.loc[_CASE_A_PEAK, "close"] / day.loc[_CASE_A_PEAK - 1, "close"] - 1.0 + ) + assert peak_ret == pytest.approx(_CASE_A_PEAK_RETURN) + + +def test_ridge_set_on_the_hand_day_is_exactly_the_adjacent_eruption_runs(): + closes, vols = _case_a_day() + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", closes, vols) + work = peak_mask_for_symbol( + prepare_visible_minute_bars(_bars(rows)).reset_index(drop=True) + ) + day = work[work["trade_date"] == _TEST_DAY].reset_index(drop=True) + ridge_slots = tuple(day.index[day["ridge"].to_numpy(dtype=bool)]) + assert ridge_slots == _CASE_A_RIDGES + + +# --------------------------------------------------------------------------- # +# The WITHIN-DAY lag (PINNED): first visible bar of each day has no return +# --------------------------------------------------------------------------- # +def test_first_visible_bar_of_the_day_never_carries_a_return(): + """A ridge at the day's first visible bar contributes NOTHING (no previous close). + + PR-F's rule makes a session-boundary eruption a RIDGE (its isolation is unprovable), + so this is a real case: slot 0 is a ridge but has no within-day predecessor, and using + the PREVIOUS DAY's close would be a cross-day gap the definition excludes. + """ + n = 12 + closes = [100.0] * n + vols = [100.0] * n + vols[0] = 200.0 # first visible bar erupts -> ridge (no previous neighbour) + vols[1] = 200.0 # its neighbour erupts too -> also a ridge, and this one HAS a return + closes[1] = 150.0 # r1 = +0.5 + rows = _background(_BG_DAYS, n) + _session("2021-07-11", closes, vols) + bars = _bars(rows) + + work = peak_mask_for_symbol( + prepare_visible_minute_bars(bars).reset_index(drop=True) + ) + day = work[work["trade_date"] == _TEST_DAY].reset_index(drop=True) + assert bool(day.loc[0, "ridge"]) is True # slot 0 IS a ridge... + assert bool(day.loc[1, "ridge"]) is True + + out = compute_ridge_minute_return(bars, **_KW) + # ...but only slot 1's +0.5 reaches the sum + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(0.5) + + +def test_returns_never_cross_a_day_boundary(): + """Day 2's first bar does not return against day 1's close. + + Day 2 opens at half of day 1's closing level; if the lag crossed the boundary the + first bar would contribute -0.5 to day 2's sum. + """ + n = 12 + day1_closes = [100.0] * n + day2_closes = [50.0] * n + vols = [100.0] * n + vols[0] = vols[1] = 200.0 # both ridges on day 2; slot 1's return is exactly 0 + rows = _background(_BG_DAYS, n) + rows += _session("2021-07-11", day1_closes, [100.0] * n) + rows += _session("2021-07-12", day2_closes, vols) + out = compute_ridge_minute_return(_bars(rows), **{**_KW, "lookback_days": 1}) + # day 2's only return-carrying ridge is slot 1 (50/50-1 = 0) -> sum exactly 0 + assert out.loc[(pd.Timestamp("2021-07-12"), _SYM)] == pytest.approx(0.0) + + +def test_non_positive_or_missing_previous_close_is_guarded(): + """A zero / NaN close kills its own return AND the next bar's, never an inf.""" + n = 14 + closes = [100.0] * n + vols = [100.0] * n + # a ridge run at (5,6) whose PREDECESSOR (slot 4) closed at zero + closes[4] = 0.0 + closes[5] = 200.0 + closes[6] = 300.0 + vols[5] = vols[6] = 200.0 + # a second, clean ridge run at (9,10) so the day still has a finite contribution + closes[9] = 150.0 + closes[10] = 300.0 + vols[9] = vols[10] = 200.0 + rows = _background(_BG_DAYS, n) + _session("2021-07-11", closes, vols) + out = compute_ridge_minute_return(_bars(rows), **_KW) + got = out.loc[(_TEST_DAY, _SYM)] + assert np.isfinite(got) + # slot 5 is dropped (prev close 0); slot 6 = 300/200-1 = +0.5; + # slot 9 = 150/100-1 = +0.5; slot 10 = 300/150-1 = +1.0 -> 2.0 + assert got == pytest.approx(2.0) + + +def test_within_day_gap_uses_the_previous_visible_bar_of_the_same_day(): + """PINNED, and locked so it can never drift silently. + + The lag is over the previous VISIBLE bar of the SAME day, with no exact-60s adjacency + requirement. A bar that opens a new session block (the real one being 13:01, after the + lunch break) therefore returns against the last bar before the gap -- a genuine price + change of the stock over the interval, not a fabricated one. The alternative (dropping + gap-spanning bars) would silently discard the post-lunch minute whenever it is a + ridge. Note the same-slot baseline defuses the obvious selection worry: the 13:01 slot + is compared against its OWN history, so a systematically busy post-lunch minute is not + systematically eruptive. + """ + early, late = 10, 8 + late_start = "13:01:00" + rows = [] + for i in range(_BG_DAYS): + day = (pd.Timestamp("2021-07-01") + pd.Timedelta(days=i)).strftime("%Y-%m-%d") + rows += _session(day, [100.0] * early, [100.0] * early) + rows += _session(day, [100.0] * late, [100.0] * late, start=late_start) + # test day: flat morning, then the FIRST post-gap bar erupts together with its + # neighbour -> both ridges; the post-gap bar returns against the morning's last close + rows += _session("2021-07-11", [100.0] * early, [100.0] * early) + late_closes = [120.0, 120.0] + [120.0] * (late - 2) + late_vols = [200.0, 200.0] + [100.0] * (late - 2) + rows += _session("2021-07-11", late_closes, late_vols, start=late_start) + + out = compute_ridge_minute_return(_bars(rows), **_KW) + # first post-gap bar: 120/100 - 1 = +0.2 (spans the gap); its neighbour: 120/120-1 = 0 + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(0.2) + + +def test_scaling_a_days_closes_leaves_the_sum_unchanged(): + """The RAW-CLOSE pin: a within-day constant adjustment factor cancels in the ratio. + + Multiplying every close of the day by a constant (what a split/dividend adjustment + factor does, since it is constant WITHIN a day) leaves every minute return -- and + therefore the daily sum -- exactly unchanged. This is why unadjusted closes are + correct here, and it is a property, not an assumption. + """ + closes, vols = _case_a_day() + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", closes, vols) + plain = compute_ridge_minute_return(_bars(rows), **_KW) + + scaled_closes = [c * 3.0 for c in closes] + rows2 = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", scaled_closes, vols) + scaled = compute_ridge_minute_return(_bars(rows2), **_KW) + + key = (_TEST_DAY, _SYM) + assert plain.loc[key] == pytest.approx(_CASE_A_SUM) + assert scaled.loc[key] == pytest.approx(plain.loc[key]) + + +# --------------------------------------------------------------------------- # +# Ridge scarcity: the >= 10 return-carrying-ridge-bar gate +# --------------------------------------------------------------------------- # +_GATE_N = 24 + + +def _one_percent_grid(n_slots): + """Closes on which EVERY bar returns exactly +1%, so a sum reads off as a count.""" + return [100.0 * (1.01**i) for i in range(n_slots)] + + +def _ridge_runs_day(run_lengths, n_slots, first_slot=1): + """A day whose eruptive bars form CONSECUTIVE runs of the given lengths. + + Runs are separated by 2 mild bars, so every eruptive bar has an eruptive neighbour and + a run of length L contributes exactly L ridge bars (no isolated peaks at all). Closes + are the +1% grid, so the day's sum is 0.01 * (number of RETURN-CARRYING ridge bars). + """ + vols = [100.0] * n_slots + slot = first_slot + for length in run_lengths: + if slot + length > n_slots: + raise AssertionError("test day too short for the requested runs") + for k in range(length): + vols[slot + k] = 200.0 + slot += length + 2 + return _one_percent_grid(n_slots), vols + + +def _ridge_and_peak_count(closes, vols, n_slots): + rows = _background(_BG_DAYS, n_slots) + _session("2021-07-11", closes, vols) + work = peak_mask_for_symbol( + prepare_visible_minute_bars(_bars(rows)).reset_index(drop=True) + ) + day = work[work["trade_date"] == _TEST_DAY] + return int(day["ridge"].sum()), int(day["peak"].sum()) + + +def test_exactly_nine_return_carrying_ridge_bars_fails_the_default_ten_gate(): + closes, vols = _ridge_runs_day([2, 2, 2, 3], _GATE_N) # 9 ridges + assert _ridge_and_peak_count(closes, vols, _GATE_N) == (9, 0) + rows = _background(_BG_DAYS, _GATE_N) + _session("2021-07-11", closes, vols) + out = compute_ridge_minute_return( + _bars(rows), **{**_KW, "min_ridge_bars": RIDGE_RETURN_MIN_RIDGE_BARS} + ) + assert out.dropna().empty + + +def test_exactly_ten_return_carrying_ridge_bars_passes_the_default_ten_gate(): + closes, vols = _ridge_runs_day([2, 2, 2, 2, 2], _GATE_N) # 10 ridges + assert _ridge_and_peak_count(closes, vols, _GATE_N) == (10, 0) + rows = _background(_BG_DAYS, _GATE_N) + _session("2021-07-11", closes, vols) + out = compute_ridge_minute_return( + _bars(rows), **{**_KW, "min_ridge_bars": RIDGE_RETURN_MIN_RIDGE_BARS} + ) + # every bar returns +1%, and all ten ridges carry a return -> 0.10 + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(0.10) + + +def test_ridge_count_is_taken_AFTER_the_return_guard(): + """The gate counts ridges that actually CONTRIBUTE a return. + + Ten classifiable ridge bars, but one of them is the day's FIRST visible bar and so has + no return -- only nine support the sum, and the default gate must invalidate the day. + """ + closes, vols = _ridge_runs_day([2, 2, 2, 2, 2], _GATE_N, first_slot=0) + assert _ridge_and_peak_count(closes, vols, _GATE_N) == (10, 0) + rows = _background(_BG_DAYS, _GATE_N) + _session("2021-07-11", closes, vols) + out = compute_ridge_minute_return( + _bars(rows), **{**_KW, "min_ridge_bars": RIDGE_RETURN_MIN_RIDGE_BARS} + ) + assert out.dropna().empty + # ...and a floor of 9 accepts the day, proving it was the COUNT that blocked it + ok = compute_ridge_minute_return(_bars(rows), **{**_KW, "min_ridge_bars": 9}) + assert ok.loc[(_TEST_DAY, _SYM)] == pytest.approx(0.09) + + +def test_day_with_no_ridge_is_invalid(): + """Only isolated peaks erupt -> nothing to sum -> honest absence, never a fake 0.""" + n = 12 + closes = _one_percent_grid(n) + vols = [100.0] * n + for s in (3, 7): # isolated -> peaks, not ridges + vols[s] = 200.0 + rows = _background(_BG_DAYS, n) + _session("2021-07-11", closes, vols) + out = compute_ridge_minute_return(_bars(rows), **_KW) + assert out.dropna().empty + + +def test_min_classifiable_gate_invalidates_thin_days(): + closes, vols = _case_a_day() + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", closes, vols) + out = compute_ridge_minute_return(_bars(rows), **{**_KW, "min_classifiable": 100}) + assert out.dropna().empty + + +def test_baseline_insufficient_yields_no_value(): + closes, vols = _case_a_day() + rows = _background(9, _CASE_A_N) + _session("2021-07-10", closes, vols) + out = compute_ridge_minute_return(_bars(rows), **_KW) + assert out.dropna().empty + + +def test_default_gates_match_the_pinned_definition(): + assert RIDGE_RETURN_LOOKBACK_DAYS == 20 + # Same scarcity floor PR-J pinned for its ridge leg: a bar must erupt AND fail the + # isolation test, so ridges are structurally far rarer than valleys. + assert RIDGE_RETURN_MIN_RIDGE_BARS == 10 + assert VOLUME_PRV_MIN_VALID_DAYS == 10 + + +# --------------------------------------------------------------------------- # +# Window mechanics: the trailing SUM across VALID days +# --------------------------------------------------------------------------- # +def _two_case_days(): + """Background + day1 = case A pattern, day2 = case B pattern, on a shared grid.""" + n = _CASE_B_N + a_closes, a_vols = _case_a_day() + pad = n - _CASE_A_N + a_closes = a_closes + [a_closes[-1]] * pad + a_vols = a_vols + [100.0] * pad + rows = _background(_BG_DAYS, n) + rows += _session("2021-07-11", a_closes, a_vols) + rows += _session("2021-07-12", *_case_b_day()) + return rows + + +def test_factor_is_the_SUM_across_trailing_valid_days_not_the_mean(): + """The cross-day aggregation is a SUM (PR-J averaged; this factor accumulates).""" + bars = _bars(_two_case_days()) + per_day = compute_ridge_minute_return(bars, **{**_KW, "lookback_days": 1}) + d1, d2 = pd.Timestamp("2021-07-11"), pd.Timestamp("2021-07-12") + s1, s2 = per_day.loc[(d1, _SYM)], per_day.loc[(d2, _SYM)] + assert s1 == pytest.approx(_CASE_A_SUM) + assert s2 == pytest.approx(_CASE_B_SUM) + + out = compute_ridge_minute_return(bars, **{**_KW, "lookback_days": 2}) + assert out.loc[(d1, _SYM)] == pytest.approx(s1) + assert out.loc[(d2, _SYM)] == pytest.approx(s1 + s2) + # explicitly NOT the mean (the two differ here: 1.0 vs 0.5) + assert out.loc[(d2, _SYM)] != pytest.approx((s1 + s2) / 2.0) + + +def test_window_drops_days_older_than_lookback(): + rows = _two_case_days() + _session("2021-07-13", *_case_b_day()) + bars = _bars(rows) + per_day = compute_ridge_minute_return(bars, **{**_KW, "lookback_days": 1}) + d1, d2, d3 = (pd.Timestamp(f"2021-07-1{k}") for k in (1, 2, 3)) + out = compute_ridge_minute_return(bars, **{**_KW, "lookback_days": 2}) + expected = per_day.loc[(d2, _SYM)] + per_day.loc[(d3, _SYM)] + assert out.loc[(d3, _SYM)] == pytest.approx(expected) + three = compute_ridge_minute_return(bars, **{**_KW, "lookback_days": 3}) + assert three.loc[(d3, _SYM)] != pytest.approx(expected) + assert np.isfinite(per_day.loc[(d1, _SYM)]) + + +def test_invalid_days_do_not_occupy_a_slot_in_the_trailing_window(): + """A day that fails a gate is ABSENT, not zero -- it must not shorten the window.""" + n = _CASE_B_N + a_closes, a_vols = _case_a_day() + a_closes = a_closes + [a_closes[-1]] * (n - _CASE_A_N) + a_vols = a_vols + [100.0] * (n - _CASE_A_N) + rows = _background(_BG_DAYS, n) + rows += _session("2021-07-11", a_closes, a_vols) + # a day with NO eruption at all -> no ridge -> invalid, contributes nothing + rows += _session("2021-07-12", [100.0] * n, [100.0] * n) + rows += _session("2021-07-13", *_case_b_day()) + bars = _bars(rows) + out = compute_ridge_minute_return(bars, **{**_KW, "lookback_days": 2}) + per_day = compute_ridge_minute_return(bars, **{**_KW, "lookback_days": 1}) + d1, d2, d3 = (pd.Timestamp(f"2021-07-1{k}") for k in (1, 2, 3)) + assert (d2, _SYM) not in per_day.index # the flat day never produced a value + # day 3's 2-day window pools day 1 and day 3, skipping the invalid day 2 + expected = per_day.loc[(d1, _SYM)] + per_day.loc[(d3, _SYM)] + assert out.loc[(d3, _SYM)] == pytest.approx(expected) + + +def test_min_valid_days_floor_returns_nan_until_enough_valid_days(): + rows = _background(_BG_DAYS, _CASE_B_N) + for k in range(3): + d = (pd.Timestamp("2021-07-11") + pd.Timedelta(days=k)).strftime("%Y-%m-%d") + rows += _session(d, *_case_b_day()) + out = compute_ridge_minute_return(_bars(rows), **{**_KW, "min_valid_days": 3}) + assert np.isnan(out.loc[(pd.Timestamp("2021-07-11"), _SYM)]) + assert np.isnan(out.loc[(pd.Timestamp("2021-07-12"), _SYM)]) + assert out.loc[(pd.Timestamp("2021-07-13"), _SYM)] == pytest.approx(3 * _CASE_B_SUM) + + +# --------------------------------------------------------------------------- # +# PIT: no lookahead, no post-cutoff influence -- WITH TEETH +# --------------------------------------------------------------------------- # +_LATE_START = "14:50:00" +_LATE_SLOTS = 6 + + +def _two_block_background(n_days, early_slots, sym=_SYM, start_day="2021-07-01"): + """Background whose sessions cover BOTH the early block and the post-cutoff block. + + Giving the late slots a same-slot baseline is what puts TEETH in the leakage test: if + the 14:50 truncation were removed, the late bars would be fully classifiable and would + really change the factor -- so the "perturbation changes nothing" assertion is not + passing merely because the late bars are unclassifiable noise. + """ + rows = [] + for i in range(n_days): + day = (pd.Timestamp(start_day) + pd.Timedelta(days=i)).strftime("%Y-%m-%d") + rows += _session(day, [100.0] * early_slots, [100.0] * early_slots, sym=sym) + rows += _session( + day, [100.0] * _LATE_SLOTS, [100.0] * _LATE_SLOTS, + sym=sym, start=_LATE_START, + ) + return rows + + +def _late_block(day, sym=_SYM): + """A post-14:50 block with a ridge run whose returns are huge and NEGATIVE. + + Four consecutive eruptive minutes, each closing 90% below the previous bar, so if the + truncation were lifted they would add 4 * (-0.9) = -3.6 to case A's +2.5 and drive the + factor NEGATIVE. The first of them returns against the early block's last close, since + the within-day lag spans the gap (PINNED §2). + """ + closes = [15.0, 1.5, 0.15, 0.015, 0.015, 0.015] + vols = [500.0, 500.0, 500.0, 500.0, 100.0, 100.0] + return _session(day, closes, vols, sym=sym, start=_LATE_START) + + +def test_perturbing_post_1450_bars_does_not_change_factor(): + rows = _two_block_background(_BG_DAYS, _CASE_A_N) + rows += _session("2021-07-11", *_case_a_day()) + clean = compute_ridge_minute_return(_bars(rows), **_KW) + perturbed = compute_ridge_minute_return( + _bars(rows + _late_block("2021-07-11")), **_KW + ) + key = (_TEST_DAY, _SYM) + assert clean.loc[key] == pytest.approx(_CASE_A_SUM) + assert perturbed.loc[key] == pytest.approx(clean.loc[key]) + + +def test_leakage_test_has_teeth_removing_the_cutoff_changes_the_value(): + """The counter-test: with the 14:50 truncation DISABLED the value moves materially. + + Without this, the assertion above could pass for the wrong reason (post-cutoff bars + that are inert anyway). Here the same bars are shown to carry a large, real effect the + moment the cutoff stops hiding them -- so the truncation is doing the work. + """ + rows = _two_block_background(_BG_DAYS, _CASE_A_N) + rows += _session("2021-07-11", *_case_a_day()) + rows += _late_block("2021-07-11") + bars = _bars(rows) + + truncated = compute_ridge_minute_return(bars, **_KW) + untruncated = compute_ridge_minute_return(bars, decision_time="23:59:59", **_KW) + key = (_TEST_DAY, _SYM) + a, b = truncated.loc[key], untruncated.loc[key] + assert a == pytest.approx(_CASE_A_SUM) + assert np.isfinite(b) + # the late block adds two ridge bars, the second of which crashes 90% -> the value + # must move by far more than any rounding tolerance, and flip sign + assert abs(b - a) / abs(a) > 0.5 + assert b < 0.0 < a + + +def test_future_day_does_not_change_earlier_factor(): + base = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", *_case_a_day()) + a = compute_ridge_minute_return(_bars(base), **_KW) + future = base + _session( + "2021-07-12", [9_999.0] * _CASE_A_N, [9_999.0] * _CASE_A_N + ) + b = compute_ridge_minute_return(_bars(future), **_KW) + key = (_TEST_DAY, _SYM) + assert np.isfinite(a.loc[key]) + assert a.loc[key] == pytest.approx(b.loc[key]) + + +# --------------------------------------------------------------------------- # +# Per-symbol isolation +# --------------------------------------------------------------------------- # +def test_per_symbol_isolation(): + n = _CASE_B_N + a_closes, a_vols = _case_a_day() + a_closes = a_closes + [a_closes[-1]] * (n - _CASE_A_N) + a_vols = a_vols + [100.0] * (n - _CASE_A_N) + + rows = _background(_BG_DAYS, n, sym="AAA.SZ") + rows += _session("2021-07-11", a_closes, a_vols, sym="AAA.SZ") + rows += _background(_BG_DAYS, n, sym="BBB.SZ") + rows += _session("2021-07-11", *_case_b_day(), sym="BBB.SZ") + out = compute_ridge_minute_return(_bars(rows), **_KW) + + solo_a = compute_ridge_minute_return( + _bars( + _background(_BG_DAYS, n, sym="AAA.SZ") + + _session("2021-07-11", a_closes, a_vols, sym="AAA.SZ") + ), + **_KW, + ) + assert out.loc[(_TEST_DAY, "AAA.SZ")] == pytest.approx( + solo_a.loc[(_TEST_DAY, "AAA.SZ")] + ) + assert out.loc[(_TEST_DAY, "BBB.SZ")] == pytest.approx(_CASE_B_SUM) + + +# --------------------------------------------------------------------------- # +# §0 REUSE NON-DRIFT: the four MERGED factors are bit-identical +# --------------------------------------------------------------------------- # +# PR-K adds no column to and changes no line of data/clean/intraday_volume_prv.py, so the +# taxonomy the four merged factors consume is untouched. These locks assert that as a +# VALUE, not as a claim: each reproduces its own module's documented hand value. +def _prv_kw(**over): + kw = dict( + baseline_days=VOLUME_PRV_BASELINE_DAYS, + baseline_min_obs=VOLUME_PRV_BASELINE_MIN_OBS, + sigma_k=VOLUME_PRV_SIGMA_K, + lookback_days=20, + min_valid_days=1, + min_classifiable=1, + ) + kw.update(over) + return kw + + +def test_volume_peak_count_unchanged_by_pr_k(): + """PR-F's factor value is bit-identical (the PR-J lock's hand case, verbatim).""" + from data.clean.intraday_volume_prv import compute_volume_peak_count + + closes, vols = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", closes, vols) + out = compute_volume_peak_count(_bars(rows), **_prv_kw(lookback_days=1)) + # case B has exactly ONE isolated eruption (slot 16); the three adjacent runs are + # ridges, not peaks + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(1.0) + + +def test_peak_interval_kurtosis_unchanged_by_pr_k(): + """PR-H's factor value is bit-identical (the PR-H hand case, verbatim).""" + from data.clean.intraday_peak_interval import compute_peak_interval_kurtosis + + n = 25 + vols = [100.0] * n + for p in (1, 3, 6, 10, 15, 21, 23): + vols[p] = 200.0 + rows = _background(_BG_DAYS, n) + _session("2021-07-11", [100.0] * n, vols) + out = compute_peak_interval_kurtosis(_bars(rows), **_prv_kw(min_intervals=4)) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(-1.48125) + + +def test_valley_relative_vwap_unchanged_by_pr_k(): + """PR-I's factor value is bit-identical (the PR-I case-A hand value, verbatim). + + This one needs AMOUNTS set independently of the closes, so it builds its own frame + rather than going through ``_bars`` (whose amount is close * volume). + """ + from data.clean.intraday_valley_vwap import compute_valley_relative_vwap + + n = 12 + erupt = (3, 7) + vols = [100.0] * n + amts = [0.0] * n + valley_slots = [s for s in range(n) if s not in erupt] + for i, s in enumerate(valley_slots): + amts[s] = 1000.0 if i < 5 else 1200.0 + for s in erupt: + vols[s] = 200.0 + amts[s] = 4000.0 + bars = _amount_bars(n, vols, amts) + out = compute_valley_relative_vwap(bars, **_prv_kw(min_valley_bars=1)) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(77.0 / 95.0) + + +def test_valley_ridge_vwap_ratio_unchanged_by_pr_k(): + """PR-J's factor value is bit-identical (the PR-J case-A hand value, verbatim).""" + from data.clean.intraday_valley_ridge_vwap import compute_valley_ridge_vwap_ratio + + n = 16 + ridges, peak = (3, 4, 8, 9), 12 + vols = [100.0] * n + amts = [0.0] * n + for s in ridges: + vols[s] = 200.0 + amts[s] = 4000.0 + vols[peak] = 300.0 + amts[peak] = 9000.0 + valley_slots = [s for s in range(n) if s not in ridges and s != peak] + for i, s in enumerate(valley_slots): + amts[s] = 1000.0 if i < 6 else 1200.0 + bars = _amount_bars(n, vols, amts) + out = compute_valley_ridge_vwap_ratio( + bars, **_prv_kw(min_valley_bars=1, min_ridge_bars=1) + ) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(6.0 / 11.0) + + +def _amount_bars(n_slots, vols, amts, sym=_SYM): + """Bars whose AMOUNT is independent of the close (for the PR-I / PR-J locks).""" + rows = [] + for i in range(_BG_DAYS): + day = pd.Timestamp("2021-07-01") + pd.Timedelta(days=i) + base = day + pd.Timedelta("09:31:00") + for s in range(n_slots): + rows.append((base + pd.Timedelta(minutes=s), sym, 100.0, 1000.0)) + base = _TEST_DAY + pd.Timedelta("09:31:00") + for s in range(n_slots): + rows.append((base + pd.Timedelta(minutes=s), sym, vols[s], amts[s])) + df = pd.DataFrame( + { + "time": pd.to_datetime([r[0] for r in rows]), + "symbol": [r[1] for r in rows], + "open": 100.0, + "high": 100.0, + "low": 100.0, + "close": 100.0, + "volume": [float(r[2]) for r in rows], + "amount": [float(r[3]) for r in rows], + } + ) + return normalize_intraday_bars(df, freq="1min") + + +def test_the_three_masks_still_partition_the_classifiable_bars(): + """valley | peak | ridge == classifiable, pairwise disjoint (structural, unchanged).""" + closes, vols = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", closes, vols) + work = peak_mask_for_symbol( + prepare_visible_minute_bars(_bars(rows)).reset_index(drop=True) + ) + classifiable = work["classifiable"].to_numpy(dtype=bool) + valley = work["valley"].to_numpy(dtype=bool) + peak = work["peak"].to_numpy(dtype=bool) + ridge = work["ridge"].to_numpy(dtype=bool) + np.testing.assert_array_equal(valley | peak | ridge, classifiable) + assert not (valley & peak).any() + assert not (valley & ridge).any() + assert not (peak & ridge).any() + day = work["trade_date"].to_numpy() == np.datetime64(_TEST_DAY) + assert ridge[day].any() and peak[day].any() # not vacuous + + +def test_carrying_close_through_does_not_disturb_the_masks(): + """``extra_columns=("close",)`` only rides along; the taxonomy is untouched.""" + closes, vols = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", closes, vols) + bars = _bars(rows) + plain = peak_mask_for_symbol( + prepare_visible_minute_bars(bars).reset_index(drop=True) + ) + carried = peak_mask_for_symbol( + prepare_visible_minute_bars(bars, extra_columns=("close",)).reset_index(drop=True) + ) + for col in ("trade_date", "bar_end", "slot", "volume", "classifiable", "valley", + "peak", "ridge"): + pd.testing.assert_series_equal(carried[col], plain[col], check_names=False) + + +# --------------------------------------------------------------------------- # +# The per-day helper (exposed for the reuse / diagnostics path) +# --------------------------------------------------------------------------- # +def _work_for_case_a(): + closes, vols = _case_a_day() + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", closes, vols) + return peak_mask_for_symbol( + prepare_visible_minute_bars( + _bars(rows), extra_columns=("close",) + ).reset_index(drop=True) + ) + + +def test_ridge_minute_return_by_day_returns_only_valid_days(): + daily = ridge_minute_return_by_day( + _work_for_case_a(), min_ridge_bars=1, min_classifiable=1 + ) + # the background days are unclassifiable -> not valid -> absent entirely + assert list(daily.index) == [_TEST_DAY] + assert daily.loc[_TEST_DAY] == pytest.approx(_CASE_A_SUM) + + +def test_ridge_bar_counts_by_day_are_exposed_for_the_coverage_disclosure(): + """The runner must be able to REPORT the ridge distribution, not just gate on it.""" + daily, diag = ridge_minute_return_by_day( + _work_for_case_a(), min_ridge_bars=1, min_classifiable=1, with_diagnostics=True + ) + assert daily.loc[_TEST_DAY] == pytest.approx(_CASE_A_SUM) + assert int(diag.loc[_TEST_DAY, "ridge_bars"]) == 4 + # all four ridges sit past the day's first bar, so all four carry a return + assert int(diag.loc[_TEST_DAY, "ridge_return_bars"]) == 4 + assert bool(diag.loc[_TEST_DAY, "valid"]) is True + # every day present in the frame appears in the diagnostics, valid or not + assert len(diag) == _BG_DAYS + 1 + + +def test_diagnostics_separate_ridge_bars_from_return_carrying_ridge_bars(): + """The two counts must differ when a ridge opens the day (the gate uses the latter).""" + closes, vols = _ridge_runs_day([2, 2, 2, 2, 2], _GATE_N, first_slot=0) + rows = _background(_BG_DAYS, _GATE_N) + _session("2021-07-11", closes, vols) + work = peak_mask_for_symbol( + prepare_visible_minute_bars( + _bars(rows), extra_columns=("close",) + ).reset_index(drop=True) + ) + _, diag = ridge_minute_return_by_day( + work, min_ridge_bars=10, min_classifiable=1, with_diagnostics=True + ) + assert int(diag.loc[_TEST_DAY, "ridge_bars"]) == 10 + assert int(diag.loc[_TEST_DAY, "ridge_return_bars"]) == 9 + assert bool(diag.loc[_TEST_DAY, "valid"]) is False + + +# --------------------------------------------------------------------------- # +# Guards / purity +# --------------------------------------------------------------------------- # +def test_empty_bars_yield_empty_schema_series(): + out = compute_ridge_minute_return(empty_intraday_bars()) + assert out.empty + assert list(out.index.names) == ["date", "symbol"] + assert out.name == "ridge_minute_return" + + +def test_input_bars_not_mutated(): + bars = _bars( + _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", *_case_a_day()) + ) + before = bars.copy(deep=True) + compute_ridge_minute_return(bars, **_KW) + pd.testing.assert_frame_equal(bars, before) + + +def test_bad_params_raise(): + bars = _bars(_session("2021-07-01", [100.0] * 3, [100.0] * 3)) + with pytest.raises(ValueError): + compute_ridge_minute_return(bars, lookback_days=0) + with pytest.raises(ValueError): + compute_ridge_minute_return(bars, baseline_days=1) + with pytest.raises(ValueError): + compute_ridge_minute_return(bars, baseline_min_obs=1) + with pytest.raises(ValueError): + compute_ridge_minute_return(bars, sigma_k=-0.5) + with pytest.raises(ValueError): + compute_ridge_minute_return(bars, min_valid_days=0) + with pytest.raises(ValueError): + compute_ridge_minute_return(bars, min_classifiable=0) + with pytest.raises(ValueError): + compute_ridge_minute_return(bars, min_ridge_bars=0) + + +# --------------------------------------------------------------------------- # +# Spec + Factor subclass +# --------------------------------------------------------------------------- # +def test_factor_spec_is_valid_and_daily(): + spec = RidgeMinuteReturnFactor().spec + assert isinstance(spec, FactorSpec) + assert spec.factor_id == "ridge_minute_return_20" + assert spec.is_intraday is False + assert spec.expected_ic_sign == -1 # NEGATIVE (report RankIC -6.29%) + assert spec.return_basis == "close_to_close" + assert spec.forward_return_horizon == 1 + assert set(spec.input_fields) == {"volume", "close"} + for field in ( + "decision_cutoff", "data_lag", "session_open", + "execution_model", "execution_window", + ): + assert getattr(spec, field) is None + + +def test_factor_spec_discloses_the_pinned_choices(): + """A reader must see the sum-vs-compounding pin, the raw closes and the 14:50 cut.""" + desc = RidgeMinuteReturnFactor().spec.description + assert "14:50" in desc + assert "RAW" in desc.upper() + assert "sum" in desc.lower() and "compound" in desc.lower() + assert str(RIDGE_RETURN_MIN_RIDGE_BARS) in desc + assert "ridge" in desc.lower() + + +def test_factor_subclass_window_tracks_name(): + f = RidgeMinuteReturnFactor(lookback_days=10) + assert f.name == "ridge_minute_return_10" + assert f.spec.factor_id == "ridge_minute_return_10" + + +def test_factor_compute_selects_preaggregated_column(): + f = RidgeMinuteReturnFactor() + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp("2021-07-01"), _SYM)], names=["date", "symbol"] + ) + panel = pd.DataFrame({f.name: [-0.031]}, index=idx) + out = f.compute(panel) + assert out.loc[(pd.Timestamp("2021-07-01"), _SYM)] == -0.031 + assert out.name == f.name + + +def test_factor_compute_missing_column_raises(): + f = RidgeMinuteReturnFactor() + 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): + RidgeMinuteReturnFactor(lookback_days=0)