diff --git a/config/phase_m_peak_ridge_amount_ratio.yaml b/config/phase_m_peak_ridge_amount_ratio.yaml new file mode 100644 index 0000000..ec9875b --- /dev/null +++ b/config/phase_m_peak_ridge_amount_ratio.yaml @@ -0,0 +1,190 @@ +# PR-M — Eleventh real factor evaluation: PEAK/RIDGE AMOUNT RATIO. THE CLOSING FACTOR (10/10) +# of this reproduction loop. +# +# Reproduces the SEVENTH factor of the Kaiyuan market-microstructure series #27 (开源证券 +# 《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, reportId 4957417, §7.2) — the +# "峰岭成交比因子" — as a first-class PeakRidgeAmountRatioFactor 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, LAST: the nine prior reproductions from this report covered a COUNT (PR-F +# volume_peak_count, weak), a TIMING moment (PR-H peak_interval_kurtosis, null), two price +# RATIOS (PR-I valley_relative_vwap and PR-J valley_ridge_vwap_ratio, both PASSING), a price +# POSITION (PR-L valley_price_quantile, PASSING and strongest) and a RETURN (PR-K +# ridge_minute_return, sign transferred but Reject). EVERY signal found so far has been a PRICE +# signal. This factor is the AMOUNT family — the last untested one — and it carries no price +# information at all: both legs are pure traded VALUE. It is therefore the single test that +# separates "only price information survives in this taxonomy" from "the peak/ridge behavioural +# split itself carries alpha". A PASS widens the finding; a REJECT sharpens it. Both are results. +# +# SAME MACHINE as PR-F / PR-H / PR-I / PR-J / PR-K / PR-L. 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), a PEAK (量峰) being an ISOLATED +# eruption and a RIDGE (量岭) an eruption that is not isolated. Because the classification is +# shared verbatim, a different verdict here cannot be an artefact of a differently implemented +# taxonomy. +# +# THE AGGREGATION IS A RATIO OF SUMS, AND THAT IS THE REPORT'S OWN WORDING — not the mean of +# daily ratios PR-J used. §7.2 says "计算 20 日量峰总成交额与量岭总成交额，二者做比": pool each +# leg over 20 days, then divide ONCE. §7.1 (PR-J) instead says "计算 20 日价格比均值", a mean of +# daily ratios. The report draws the distinction itself in adjacent sections, so each is +# followed as written. The difference is real, not cosmetic: a ratio of sums is AMOUNT-WEIGHTED +# (busy days dominate) while a mean of ratios weights every valid day equally — and it is also +# far better behaved here, since a single day whose ridge amount nearly vanishes would dominate +# a mean of daily ratios. NOTE: the task card for this PR pinned the mean-of-ratios form; the +# report overrides it and the deviation is recorded in the PR report. +# +# PINNED choices (all disclosed on the factor spec; the report specifies only the shape): +# (1) PEAK is the NUMERATOR and RIDGE the DENOMINATOR, fixed both by the name 峰岭 (peak +# first) and, independently, by the report's stated semantics — the factor measures +# "知情交易相对个人投资者交易的相对参与程度", informed trading RELATIVE TO retail; +# (2) the masks are PR-J's exactly, so valley|peak|ridge stays an exact partition of the +# classifiable bars. A VALLEY bar enters NEITHER leg: this factor reads only the two +# ERUPTIVE groups, which is the one way it could silently degenerate and is the subject +# of a dedicated invariance test; +# (3) a bar with non-finite or non-positive amount is dropped from BOTH legs AND from both +# bar counts, at the summation step only, so PR-F's same-slot baseline stays +# bit-identical. VOLUME is deliberately NOT part of this guard (unlike PR-I/PR-J), because +# this factor never divides by volume; +# (4) RAW unadjusted amounts are exactly correct here. Traded VALUE in RMB is not rescaled by +# any split/dividend adjustment factor — the adjustment moves price and share count in +# compensating directions. So unlike PR-L there is no ex-date caveat to disclose, and +# unlike PR-I/PR-J not even a "cancels within the day" argument is needed; +# (5) DEVIATION FROM THE REPORT, disclosed: both legs span 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; +# (6) an ASYMMETRIC bar floor — >= 5 tradable PEAK bars against >= 10 tradable RIDGE bars. +# This asymmetry RUNS THE OPPOSITE WAY from PR-J's: there ridges were the scarce leg, here +# PEAKS are, because a peak must erupt AND be ISOLATED, and isolation is a strong +# condition on a name whose eruptions cluster. The floor is set low deliberately, and the +# realized peak-bar distribution, the day-validity rate and the COUNTERFACTUAL valid-day +# count at a peak floor of 10 are all REPORTED by the runner rather than left implicit; +# (7) a day is VALID iff it also has >= 100 classifiable bars and strictly positive amount in +# both legs; NaN below 10 valid days in the trailing window. +# +# Pre-registered sign = +1, READ FROM THE REPORT rather than inferred from data or argued from +# semantics: §7.2 states "峰岭成交比因子 RankIC 均值 10.28%，RankICIR 4.07" — an explicitly +# POSITIVE RankIC — with a long leg of 16.06%/yr, long-short 27.13%/yr, IR 2.89, max drawdown +# 7.88%, 74.3% monthly win rate, and every year positive across 2013-2025. The report's §1 +# taxonomy summary agrees independently: "对于量峰时点，分钟数、成交额类因子更加有效，反映知情 +# 交易参与度，为正向因子", with the ridge leg's amount factors marked negative-alpha, so +# peak-over-ridge is positive on both counts. 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. The report gives no CSI500 sub-domain breakdown for this factor, so none is quoted. +# +# SIGN = +1 means this factor is NOT affected by the frozen layer's aligned-spread cost-sign +# defect (which computes sign * (gross - cost) and therefore ADDS costs back only when the sign +# is negative). As with PR-L the aligned_spread_* fields may be read directly here. +# net_long_short_by_cost remains the primary tradability read, for sibling comparability. +# +# Eval CELL is IDENTICAL to PR-C .. PR-L: 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_peak_ridge_amount_ratio.py): once with NO book (Incremental NOT_ASSESSED) and once +# with the confirmed book. That second run carries a different weight here than for the price +# factors: a traded-VALUE mix has no obvious kinship with a value or low-vol bet, so an +# Incremental PASS would be a genuinely NEW exposure rather than a repackaged one. + +project: + name: quantitative_trading_pr_m_peak_ridge_amount_ratio + 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: peak_ridge_amount_ratio_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 (peak_ridge_amount_ratio_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. Unlike PR-L this factor has no + # neutralization of its own — the raw ratio goes straight into the standard pipeline. + 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_amount_ratio.py b/data/clean/intraday_amount_ratio.py new file mode 100644 index 0000000..e4aeeea --- /dev/null +++ b/data/clean/intraday_amount_ratio.py @@ -0,0 +1,426 @@ +"""PEAK/RIDGE AMOUNT-RATIO factor (PR-M). + +Reproduces the SEVENTH factor of the Kaiyuan market-microstructure series #27 (开源证券 +《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, 2025-07-20, reportId 4957417, +§7.2 「峰岭成交比因子多空年化收益 27.13%」): "本小节通过计算 20 日量峰总成交额与量岭总成交额， +二者做比作为峰岭成交比因子，衡量知情交易相对个人投资者交易的相对参与程度". + +This is the last untested FAMILY of the reproduction loop. The nine prior factors covered a +COUNT (PR-F volume_peak_count, weak), a TIMING moment (PR-H peak_interval_kurtosis, null), +two price RATIOS (PR-I valley_relative_vwap, PR-J valley_ridge_vwap_ratio, both PASSING), a +price POSITION (PR-L valley_price_quantile, PASSING and strongest) and a RETURN (PR-K +ridge_minute_return, sign transferred but Reject). Every signal found so far has been a +PRICE signal. This factor carries NO price information at all — it is a pure TRADED-VALUE +mix between the two eruptive groups — so it is the one test that separates "only price +information survives in this taxonomy" from "the peak/ridge split itself carries alpha". + +The classification is REUSED verbatim from :mod:`data.clean.intraday_volume_prv` +(``prepare_visible_minute_bars`` + ``peak_mask_for_symbol``) — same same-slot strictly-prior +μ+kσ eruptive test, same classifiable rule. Nothing about the taxonomy is re-implemented +here, so the seven factors can never drift apart. + +AGGREGATION — THE REPORT'S FORM, WHICH IS *NOT* THE MEAN OF DAILY RATIOS +------------------------------------------------------------------------ +The report says "计算 20 日量峰总成交额与量岭总成交额，二者做比": sum the peak amount over +20 days, sum the ridge amount over 20 days, THEN divide. That is a RATIO OF SUMS. Contrast +§7.1 (PR-J), which explicitly says "计算 20 日价格比均值" — a MEAN OF RATIOS. The report +draws the distinction itself in adjacent sections, so it is a deliberate difference in the +source and is followed here. + +The two forms are genuinely different, not a rounding detail: a ratio of sums is +AMOUNT-WEIGHTED (busy days dominate) while a mean of ratios weights every valid day +equally. The ratio of sums is also structurally far better behaved for THIS quantity — +``peak_amt / ridge_amt`` on a single day has a small, noisy denominator and a heavy right +tail, and a single day whose ridge amount nearly vanishes would dominate a 20-day mean of +ratios. Pooling both legs first is the natural estimator of a 20-day participation MIX, +which is exactly what the report says the factor measures. + +PINNED choices (deliberate and DISCLOSED, not tuned knobs; reproduced on the factor spec): + + 1. PEAK IS THE NUMERATOR, RIDGE THE DENOMINATOR — "峰岭成交比", peak first, and the + report's stated semantics ("衡量知情交易相对个人投资者交易的相对参与程度", informed + trading RELATIVE TO retail) fix the direction independently of the name's word order. + 2. THE RIDGE MASK IS ``eruptive & ~peak`` and the PEAK MASK is PR-F's isolated-eruption + test — the identical masks PR-J used, so ``valley | peak | ridge == classifiable`` + stays an exact partition and no eruptive bar is silently counted on both sides. A + VALLEY bar enters NEITHER leg: this factor reads only the two ERUPTIVE groups. + 3. POSITIVE-TRADE GUARD. A bar with non-finite or non-positive ``amount`` carries no + traded value, so it is dropped from both legs and from both bar counts. The guard runs + at the summation step only — never before classification — because the same-slot μ/σ + baseline is PR-F's and must stay bit-identical. Unlike PR-I/PR-J this factor never + divides by volume, so ``volume`` is NOT part of the guard: a bar with a real amount + contributes its amount regardless of how its volume is recorded. + 4. RAW (UNADJUSTED) AMOUNTS. ``amount`` is traded VALUE in RMB, which no split or + dividend adjustment factor rescales — the adjustment moves prices and share counts in + compensating directions. Both legs are therefore free of the ex-date caveat PR-L had + to disclose, and free even of PR-I/PR-J's weaker "cancels within the day" argument: + there is nothing to cancel. + 5. BOTH LEGS 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. + 6. AN ASYMMETRIC BAR FLOOR — ``min_peak_bars`` (=5) against ``min_ridge_bars`` (=10). + PEAKS ARE THE SCARCER LEG HERE, the reverse of PR-J's valley/ridge asymmetry: a peak + must erupt AND be ISOLATED, and isolation is a strong condition on a liquid name whose + eruptions cluster. Holding the peak leg to the ridge floor would discard sound days + and bias the surviving sample toward names whose eruptions happen to arrive alone. + The floor is therefore set LOW, deliberately, and both the realized peak-bar + distribution and the resulting day-validity rate are REPORTED + (``with_diagnostics=True``) rather than left implicit — together with the + counterfactual valid-day count at a peak floor of 10. + 7. BOTH BAR COUNTS ARE TAKEN AFTER THE GUARD, so a day cannot qualify on the strength of + bars that traded nothing. + +Factor value: ``peak_ridge_amount_ratio_20`` = ``Σ peak_amt / Σ ridge_amt`` where both sums +run over the symbol's most recent ``lookback_days`` (=20) VALID trading days INCLUDING +``d``. A day is VALID iff it clears four gates: at least ``min_classifiable`` (=100) +classifiable bars (PR-F's gate, unchanged), at least ``min_peak_bars`` (=5) tradable peak +bars, at least ``min_ridge_bars`` (=10) tradable ridge bars, and strictly positive amount in +BOTH legs. 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, READ FROM THE REPORT, not from data and not from semantics: §7.2 +states "峰岭成交比因子 RankIC 均值 10.28%，RankICIR 4.07" — an explicitly POSITIVE RankIC — +with a long leg of 16.06%/yr, long-short 27.13%/yr, IR 2.89, max drawdown 7.88% and a 74.3% +monthly win rate (every year positive 2013–2025). The report's §1 taxonomy summary agrees +independently: "对于量峰时点，分钟数、成交额类因子更加有效，反映知情交易参与度，为正向因子" +— amount factors at PEAK minutes are POSITIVE factors — while the same sentence marks the +ridge leg's amount factors as negative-alpha, so peak-over-ridge is positive on both counts. +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. +PEAK_RIDGE_LOOKBACK_DAYS = 20 # trailing VALID trading-day window for BOTH sums, includes d +# PINNED LOWER than the ridge floor (module docstring §6): a peak must erupt AND be +# ISOLATED, which makes peaks the structurally scarcer leg of this pair. +PEAK_RIDGE_MIN_PEAK_BARS = 5 # min TRADABLE peak bars for a valid day +PEAK_RIDGE_MIN_RIDGE_BARS = 10 # min TRADABLE ridge bars for a valid day + +# The extra 1min column this factor needs on top of PR-F's (volume): the traded value. +# Unlike PR-I / PR-J this factor never divides by volume — amount is the whole quantity. +_AMOUNT = "amount" + +# Per-day diagnostic columns (the peak-scarcity disclosure the task card requires). +DIAGNOSTIC_COLUMNS = ("classifiable_bars", "peak_bars", "ridge_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) + + +# --------------------------------------------------------------------------- # +# The three load-bearing steps, each a NAMED function. +# +# They are small enough to inline, but each one carries a correctness property that the +# test-suite has to be able to BREAK on purpose: the positive-trade guard, the trailing +# (never forward-looking) window, and the per-symbol split. A test that asserts "perturbing +# X changes nothing" proves nothing unless the defective implementation can be substituted +# and shown to FAIL it, so each property gets its own substitutable seam. +# --------------------------------------------------------------------------- # +def _tradable_amount(amt: np.ndarray) -> np.ndarray: + """Positive-trade guard (module docstring §3): finite, strictly positive amount. + + ``volume`` is deliberately absent — this factor never divides by volume, so a bar with + a real traded value contributes it regardless of how its volume is recorded. + """ + return np.isfinite(amt) & (amt > 0.0) + + +def _trailing_ratio_of_sums( + legs: pd.DataFrame, *, lookback_days: int, min_valid_days: int +) -> pd.Series: + """The report's ratio of 20-day sums over a STRICTLY TRAILING valid-day window. + + Both legs are pooled with the SAME window and the SAME ``min_periods``, so numerator + and denominator always cover exactly the same days. The window is trailing and the + index is sorted ascending first, so a value at ``d`` can never absorb a later day. + """ + ordered = legs.sort_index() + roll = ordered.rolling(lookback_days, min_periods=min_valid_days).sum() + return roll["peak_amt"] / roll["ridge_amt"] + + +def _symbol_frames(visible: pd.DataFrame): + """Split the visible bars into ONE FRAME PER SYMBOL, in sorted symbol order. + + The whole cross-symbol isolation guarantee lives here: every downstream step (the + same-slot baseline, the classification, both amount legs, the trailing window) runs on + a single symbol's rows, so one symbol's bars can never reach another's factor value. + """ + for sym, g in visible.groupby(SYMBOL_LEVEL, sort=True): + yield str(sym), g.reset_index(drop=True) + + +def peak_ridge_amount_by_day( + work: pd.DataFrame, + *, + min_peak_bars: int = PEAK_RIDGE_MIN_PEAK_BARS, + min_ridge_bars: int = PEAK_RIDGE_MIN_RIDGE_BARS, + min_classifiable: int = VOLUME_PRV_MIN_CLASSIFIABLE, + with_diagnostics: bool = False, +) -> pd.DataFrame | tuple[pd.DataFrame, pd.DataFrame]: + """Daily peak / ridge traded-AMOUNT totals 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=("amount",)`` so the traded value is + available alongside the ``peak`` / ``ridge`` / ``classifiable`` masks. + + Returns the two LEGS rather than their daily ratio, because the factor is the report's + RATIO OF 20-DAY SUMS (module docstring), not the mean of daily ratios — the caller + pools each leg over the trailing window first and divides once at the end. + + Args: + work: one symbol's classified minute frame (see above). + min_peak_bars: minimum TRADABLE peak bars for a valid day (PINNED lower than the + ridge floor — see §6; peaks are the scarcer leg of this pair). + min_ridge_bars: minimum TRADABLE ridge bars 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 peak-scarcity distribution instead of only gating on it. + + Returns: + DataFrame indexed by ``trade_date`` (ascending) with ``peak_amt`` / ``ridge_amt`` + columns for the days that clear all 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 + ``(legs, diagnostics)`` pair where ``diagnostics`` is indexed by EVERY day present + in ``work`` and carries :data:`DIAGNOSTIC_COLUMNS`. + """ + amt = work[_AMOUNT].to_numpy(dtype=float) + # Positive-trade guard: a bar with no finite positive traded value contributes nothing. + # Applied HERE, at the summation step, never before classification — PR-F's same-slot + # baseline must stay bit-identical. + tradable = _tradable_amount(amt) + peak = work["peak"].to_numpy(dtype=bool) & tradable + ridge = work["ridge"].to_numpy(dtype=bool) & tradable + + per_bar = pd.DataFrame( + { + "trade_date": work["trade_date"].to_numpy(), + "peak_amt": np.where(peak, amt, 0.0), + "ridge_amt": np.where(ridge, amt, 0.0), + "peak_bars": peak.astype(np.int64), + "ridge_bars": ridge.astype(np.int64), + "classifiable_bars": work["classifiable"] + .to_numpy(dtype=bool) + .astype(np.int64), + } + ) + agg = per_bar.groupby("trade_date", sort=True).sum() + + # Four validity gates (module docstring): PR-F's classifiable floor (unchanged), + # enough TRADABLE peak bars, enough TRADABLE ridge bars, and strictly positive amount + # on both legs (a day contributing 0 to the denominator pool is never fabricated). + valid = ( + (agg["classifiable_bars"] >= min_classifiable) + & (agg["peak_bars"] >= min_peak_bars) + & (agg["ridge_bars"] >= min_ridge_bars) + & (agg["peak_amt"] > 0.0) + & (agg["ridge_amt"] > 0.0) + ) + legs = agg.loc[valid, ["peak_amt", "ridge_amt"]].astype(float) + + if not with_diagnostics: + return legs + diagnostics = agg[["classifiable_bars", "peak_bars", "ridge_bars"]].copy() + diagnostics["valid"] = valid + return legs, diagnostics + + +def _peak_ridge_amount_ratio_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_peak_bars: int, + min_ridge_bars: int, + collect_diagnostics: bool = False, +) -> tuple[list[pd.Timestamp], list[float], pd.DataFrame | None]: + """Daily peak/ridge amount-ratio 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 two amount legs, then divides the trailing-``lookback_days``-valid-day SUM of + the peak leg by that of the ridge leg (the report's ratio-of-sums form). No cross-symbol + leakage (``g`` is one symbol's slice) and no lookahead (the baseline is strictly prior, + both rolling windows are trailing). + """ + work = peak_mask_for_symbol( + g, + baseline_days=baseline_days, + baseline_min_obs=baseline_min_obs, + sigma_k=sigma_k, + ) + result = peak_ridge_amount_by_day( + work, + min_peak_bars=min_peak_bars, + min_ridge_bars=min_ridge_bars, + min_classifiable=min_classifiable, + with_diagnostics=collect_diagnostics, + ) + if collect_diagnostics: + legs, diagnostics = result + else: + legs, diagnostics = result, None + if legs.empty: + return [], [], diagnostics + + # RATIO OF SUMS over the trailing lookback_days VALID days (including d); NaN until + # min_valid_days valid days have accumulated. + ratio = _trailing_ratio_of_sums( + legs, lookback_days=lookback_days, min_valid_days=min_valid_days + ) + days = [pd.Timestamp(d).normalize() for d in ratio.index] + return days, list(ratio.to_numpy(dtype=float)), diagnostics + + +def compute_peak_ridge_amount_ratio( + bars: pd.DataFrame, + *, + lookback_days: int = PEAK_RIDGE_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_peak_bars: int = PEAK_RIDGE_MIN_PEAK_BARS, + min_ridge_bars: int = PEAK_RIDGE_MIN_RIDGE_BARS, + decision_time: str = DEFAULT_DECISION_TIME, + name: str = "peak_ridge_amount_ratio", + diagnostics_out: list | None = None, +) -> pd.Series: + """PIT-safe daily "peak/ridge traded-amount ratio" 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, totals each valid day's peak and + ridge traded amount, and returns the ratio of the trailing-``lookback_days``-VALID-day + SUMS of the two legs. See the module docstring for the LOCKED definition, the report's + ratio-of-sums wording, and the seven 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 POOLED by both legs (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_peak_bars: a day needs at least this many TRADABLE peak bars (PINNED lower than + the ridge floor — see the module docstring §6). + min_ridge_bars: a day needs at least this many TRADABLE ridge bars. + 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 + peak-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_peak_bars < 1: + raise ValueError(f"min_peak_bars must be >= 1; got {min_peak_bars!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=("amount",) is the ONLY difference from the PR-F / PR-H entry points: + # the traded value 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=(_AMOUNT,) + ) + if visible.empty: + return _empty_series(name) + + collect = diagnostics_out is not None + index_tuples: list[tuple] = [] + values: list[float] = [] + for sym, g in _symbol_frames(visible): + days, vals, diagnostics = _peak_ridge_amount_ratio_for_symbol( + g, + 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_peak_bars=min_peak_bars, + 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] = sym + diagnostics_out.append(frame) + for day, val in zip(days, vals): + index_tuples.append((day, 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", + "PEAK_RIDGE_LOOKBACK_DAYS", + "PEAK_RIDGE_MIN_PEAK_BARS", + "PEAK_RIDGE_MIN_RIDGE_BARS", + "compute_peak_ridge_amount_ratio", + "peak_ridge_amount_by_day", +] diff --git a/factors/compute/intraday_derived.py b/factors/compute/intraday_derived.py index 3493115..26c8a76 100644 --- a/factors/compute/intraday_derived.py +++ b/factors/compute/intraday_derived.py @@ -35,6 +35,11 @@ JUMP_LOOKBACK_DAYS, JUMP_MIN_PAIRS, ) +from data.clean.intraday_amount_ratio import ( + PEAK_RIDGE_LOOKBACK_DAYS, + PEAK_RIDGE_MIN_PEAK_BARS, + PEAK_RIDGE_MIN_RIDGE_BARS, +) from data.clean.intraday_amp_anomaly import ( AMP_ANOMALY_FREQ, AMP_ANOMALY_LOOKBACK_DAYS, @@ -1175,12 +1180,151 @@ def compute(self, panel: pd.DataFrame) -> pd.Series: return panel[self.name].rename(self.name) +class PeakRidgeAmountRatioFactor(Factor): + """Peak/ridge traded-amount ratio factor (daily signal, minute-derived). + + ``compute`` reads the pre-aggregated daily column the runner placed on the panel + (produced by + ``data.clean.intraday_amount_ratio.compute_peak_ridge_amount_ratio``); 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` / :class:`RidgeMinuteReturnFactor` / + :class:`ValleyPriceQuantileFactor` and the value / financial factors that surface an + enriched column. + + Args: + lookback_days: trailing VALID trading-day window POOLED by both amount legs; 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"peak_ridge_amount_ratio_{PEAK_RIDGE_LOOKBACK_DAYS}" + + def __init__(self, lookback_days: int = PEAK_RIDGE_LOOKBACK_DAYS) -> None: + if not isinstance(lookback_days, int) or lookback_days < 1: + raise ValueError( + f"peak-ridge-amount-ratio lookback_days must be a positive integer; got " + f"{lookback_days!r}." + ) + self._lookback_days = lookback_days + self.name = f"peak_ridge_amount_ratio_{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 READ FROM THE REPORT, not inferred from data and not merely + argued from semantics: §7.2 states "峰岭成交比因子 RankIC 均值 10.28%，RankICIR + 4.07" — an explicitly POSITIVE RankIC — with a long leg of 16.06%/yr, long-short + 27.13%/yr, IR 2.89, max drawdown 7.88%, 74.3% monthly win rate and every year + positive 2013–2025. The report's §1 taxonomy summary agrees independently ("对于 + 量峰时点，分钟数、成交额类因子更加有效，反映知情交易参与度，为正向因子", with the + ridge leg's amount factors marked negative-alpha), so peak-over-ridge is positive + on both counts. 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 AGGREGATION FORM, which is the one place this + factor's definition departs from its siblings: the report specifies a RATIO OF + 20-DAY SUMS, not the mean of daily ratios that §7.1 (PR-J) specifies. + """ + return FactorSpec( + factor_id=self.name, + version="1.0", + description=( + f"Peak/ridge traded-amount ratio (Kaiyuan microstructure series #27, " + f"SEVENTH factor 峰岭成交比, §7.2). SAME minute classification as PR-F " + f"volume_peak_count / PR-H peak_interval_kurtosis / PR-I " + f"valley_relative_vwap / PR-J valley_ridge_vwap_ratio (REUSED from " + f"data.clean.intraday_volume_prv, not re-implemented): 1min bars " + f"PIT-truncated at 14:50, a minute is ERUPTIVE if vol > μ + " + f"{VOLUME_PRV_SIGMA_K:g}σ of its SAME-SLOT strictly-prior " + f"{VOLUME_PRV_BASELINE_DAYS}-day baseline; a PEAK is an ISOLATED eruption " + f"and a RIDGE is 'eruptive AND NOT a peak'. This is the first factor of " + f"the loop carrying NO price information: both legs are pure traded VALUE, " + f"so it separates 'only price information survives in this taxonomy' from " + f"'the peak/ridge split itself carries alpha'. AGGREGATION = RATIO OF " + f"SUMS: Σ peak amount over the trailing {self._lookback_days} VALID days " + f"divided by Σ ridge amount over the SAME days — the report's literal " + f"wording ('计算 20 日量峰总成交额与量岭总成交额，二者做比'), which " + f"DIFFERS from §7.1's mean-of-daily-ratios and is followed as written; it " + f"is also the better-behaved estimator, since a single day with a nearly " + f"vanishing ridge amount would dominate a mean of daily ratios. PINNED " + f"choices: (1) PEAK is the numerator, RIDGE the denominator, fixed both by " + f"the name 峰岭 and by the report's stated semantics (informed trading " + f"RELATIVE TO retail participation); (2) the masks are PR-J's exactly, so " + f"valley|peak|ridge stays an exact partition of the classifiable bars and " + f"a VALLEY bar enters NEITHER leg — this factor reads only the two " + f"ERUPTIVE groups; (3) bars with non-finite or non-positive amount are " + f"dropped from both legs and both bar counts (guard applied at summation " + f"only, so PR-F's baseline is untouched); volume is deliberately NOT in " + f"the guard because this factor never divides by volume; (4) RAW " + f"unadjusted amounts are exactly correct: traded VALUE in RMB is not " + f"rescaled by any split/dividend adjustment factor, so unlike PR-L there " + f"is no ex-date caveat and unlike PR-I/PR-J no within-day cancellation " + f"argument is even needed; (5) DEVIATION FROM THE REPORT, disclosed: both " + f"legs span the PIT-VISIBLE window 09:31-14:50 only, not the full session " + f"— reading the close would be lookahead at our 14:50 decision time; (6) a " + f"day is VALID iff it has >= {VOLUME_PRV_MIN_CLASSIFIABLE} classifiable " + f"bars AND >= {PEAK_RIDGE_MIN_PEAK_BARS} TRADABLE peak bars AND >= " + f"{PEAK_RIDGE_MIN_RIDGE_BARS} TRADABLE ridge bars (both counted AFTER the " + f"guard) AND positive amount in both legs — the PEAK floor is deliberately " + f"LOWER, the REVERSE of PR-J's asymmetry, because a peak must erupt AND be " + f"ISOLATED, making peaks the scarcer leg of THIS pair; the realized " + f"peak-bar distribution, the day-validity rate and the counterfactual " + f"valid-day count at a peak floor of 10 are all REPORTED by the runner " + f"rather than left implicit; NaN below {VOLUME_PRV_MIN_VALID_DAYS} valid " + f"days. Derived from 1min bars but a 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: volume drives + # the REUSED classification, amount is the factor's whole quantity. Declared + # for honest provenance disclosure (data_coverage lists them); the daily panel + # surfaces the pre-aggregated column itself. + input_fields=("volume", "amount"), + family="microstructure", + min_history_bars=0, + ) + + def compute(self, panel: pd.DataFrame) -> pd.Series: + """Select the pre-aggregated daily peak/ridge amount-ratio column off ``panel``. + + The runner runs ``compute_peak_ridge_amount_ratio`` 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"PeakRidgeAmountRatioFactor needs the pre-aggregated '{self.name}' " + f"column on the panel (produced upstream by " + f"compute_peak_ridge_amount_ratio and joined by the runner); panel has " + f"{list(panel.columns)}." + ) + return panel[self.name].rename(self.name) + + __all__ = [ "AmpMarginalAnomalyVolFactor", "IntradayAmpCutFactor", "JumpAmountCorrFactor", "MinuteIdealAmplitudeFactor", "PeakIntervalKurtosisFactor", + "PeakRidgeAmountRatioFactor", "RidgeMinuteReturnFactor", "ValleyPriceQuantileFactor", "ValleyRelativeVwapFactor", diff --git a/qt/cli.py b/qt/cli.py index eea9e36..63751e8 100644 --- a/qt/cli.py +++ b/qt/cli.py @@ -436,6 +436,50 @@ def _cmd_run_eval_ridge_minute_return(args: argparse.Namespace) -> int: return 0 +def _cmd_run_eval_peak_ridge_amount_ratio(args: argparse.Namespace) -> int: + """Run the two real peak/ridge amount-ratio evaluations (cache-only) + reports.""" + from qt.eval_peak_ridge_amount_ratio import run_eval_peak_ridge_amount_ratio + + try: + result = run_eval_peak_ridge_amount_ratio(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-peak-ridge-amount-ratio: 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" + # PEAK bars are the structurally scarce leg of THIS pair (the reverse of PR-J); + # the realized distribution, the day-validity rate and the counterfactual at a + # peak floor of 10 are surfaced so the lowered gate is a number, not a claim. + f"{result.peak_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, so the frozen layer's aligned-spread cost-sign defect does NOT apply + # to this factor; net spreads are still printed for sibling comparability. + f"net long-short by cost: {net}\n" + f"ic_rank={_fmt_metric(nb['ic_mean'])} " + f"ic_pearson={_fmt_metric(nb['ic_pearson_mean'])} " + f"monotonicity={_fmt_metric(nb['monotonicity_spearman'])}\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_price_quantile(args: argparse.Namespace) -> int: """Run the two real valley-price-quantile evaluations (cache-only) + reports.""" from qt.eval_valley_price_quantile import run_eval_valley_price_quantile @@ -715,6 +759,13 @@ def build_parser() -> argparse.ArgumentParser: p_vpq.add_argument("--config", required=True, help="Path to the YAML config.") p_vpq.set_defaults(func=_cmd_run_eval_valley_price_quantile) + p_pra = sub.add_parser( + "run-eval-peak-ridge-amount-ratio", + help="Run the peak/ridge amount-ratio factor evaluation (CSI500, cache-only).", + ) + p_pra.add_argument("--config", required=True, help="Path to the YAML config.") + p_pra.set_defaults(func=_cmd_run_eval_peak_ridge_amount_ratio) + 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_peak_ridge_amount_ratio.py b/qt/eval_peak_ridge_amount_ratio.py new file mode 100644 index 0000000..472b76a --- /dev/null +++ b/qt/eval_peak_ridge_amount_ratio.py @@ -0,0 +1,724 @@ +"""run-eval-peak-ridge-amount-ratio: the eleventh real factor evaluation (PR-M). + +Reproduces the SEVENTH factor of the Kaiyuan market-microstructure series #27 (开源证券 +《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, reportId 4957417, §7.2) — the +"峰岭成交比" factor — as a first-class +:class:`~factors.compute.intraday_derived.PeakRidgeAmountRatioFactor` 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-L used. + +THIS IS THE CLOSING FACTOR of the reproduction loop, and it is the one that closes the +FAMILY MAP. The nine prior reproductions covered a COUNT (PR-F volume_peak_count, weak), a +TIMING moment (PR-H peak_interval_kurtosis, null), two price RATIOS (PR-I +valley_relative_vwap, PR-J valley_ridge_vwap_ratio, both PASSING), a price POSITION (PR-L +valley_price_quantile, PASSING and strongest) and a RETURN (PR-K ridge_minute_return, sign +transferred but Reject). Every signal found so far has been a PRICE signal. This factor +carries NO price information whatsoever — both legs are pure traded VALUE — so it is the +one test that distinguishes "only price information survives in this taxonomy" from "the +peak/ridge behavioural split itself carries alpha". A PASS widens the finding to the +taxonomy; a REJECT sharpens it to prices. Either outcome is a legitimate result and is +reported as such. + +The factor is a DAILY signal derived DIRECTLY from the 1min cache (see +``data.clean.intraday_amount_ratio.compute_peak_ridge_amount_ratio``): 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 PEAK is +an ISOLATED eruption, a RIDGE is an eruptive minute that is not isolated), total each valid +day's peak and ridge traded amount, and divide the trailing-20-VALID-day SUM of the peak +leg by that of the ridge leg. NOTE THE AGGREGATION: the report specifies a RATIO OF SUMS +here ("计算 20 日量峰总成交额与量岭总成交额，二者做比"), NOT the mean of daily ratios that +§7.1 (PR-J) specifies — the two forms are followed as written in each section. It is +executed CLOSE-TO-CLOSE (daily default), so ``is_intraday=False``. The eval CELL is +identical to PR-C..PR-L, so this run is directly comparable to its siblings. + +PEAK SCARCITY IS MEASURED, NOT ASSUMED — and the asymmetry RUNS THE OTHER WAY THAN PR-J's. +There, ridges were the scarce leg; here PEAKS are, because a peak must erupt AND be +ISOLATED. The valid-day gate therefore asks for >= 5 tradable peak bars against the ridge +leg's >= 10. The runner collects the REALIZED per-day peak-bar distribution, the day +validity rate across the whole universe, and the COUNTERFACTUAL valid-day count at a peak +floor of 10, so the cost of the lowered threshold is a number rather than an assumption. + +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 ratio adds alpha +BEYOND value / low-vol. The with-book run carries a different weight here than for the +price factors: a traded-VALUE mix has no obvious kinship with a value or low-vol bet, so an +Incremental PASS would be a genuinely new exposure rather than a repackaged one — and an +Incremental failure would be the more surprising outcome. +""" + +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_amount_ratio import ( + PEAK_RIDGE_LOOKBACK_DAYS, + PEAK_RIDGE_MIN_PEAK_BARS, + PEAK_RIDGE_MIN_RIDGE_BARS, + compute_peak_ridge_amount_ratio, +) +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 PeakRidgeAmountRatioFactor +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_peak_ridge_amount_ratio" +_REPORT_STEM = "eval_peak_ridge_amount_ratio" + +# Percentiles reported for the realized peak-bar distribution (the scarcity disclosure). +_PEAK_PCTL = (0, 10, 25, 50, 75, 90, 100) + +# The counterfactual peak floor the task card asks to quantify: how many days would still +# be valid if the PEAK leg were held to the RIDGE leg's floor instead of its own. +_COUNTERFACTUAL_PEAK_FLOOR = PEAK_RIDGE_MIN_RIDGE_BARS + + +# --------------------------------------------------------------------------- # +# Peak-scarcity coverage (measured, never assumed) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class PeakCoverage: + """Realized peak-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 peak-scarcity one and would otherwise + make the peak gate look worse than it is — both denominators are reported so the reader + can check that framing. The gate-failure counts are NOT mutually exclusive (a thin day + can fail several gates at once) and are reported for shape, not as a partition. + """ + + symbol_days: int + classifiable_days: int + valid_days: int + peak_percentiles: tuple[tuple[int, float], ...] + peak_mean: float + ridge_median: float + days_below_peak_gate: int + days_below_ridge_gate: int + days_below_classifiable_gate: int + # Counterfactual: how many days would survive if the PEAK leg were held to the RIDGE + # floor. Quantifies exactly what the lowered threshold buys. + valid_days_at_ridge_floor: int + # The gates this run actually applied, so the disclosure can never describe the module + # defaults while the run used something else. + min_peak_bars: int = PEAK_RIDGE_MIN_PEAK_BARS + min_ridge_bars: int = PEAK_RIDGE_MIN_RIDGE_BARS + min_classifiable: int = VOLUME_PRV_MIN_CLASSIFIABLE + counterfactual_peak_floor: int = _COUNTERFACTUAL_PEAK_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 + + 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.peak_percentiles) + return ( + f"peak 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"peak_bars[{pctl} mean={self.peak_mean:.1f}] " + f"ridge_bars_median={self.ridge_median:.0f} " + f"below_peak_gate({self.min_peak_bars})={self.days_below_peak_gate} " + 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_peak_floor_were_{self.counterfactual_peak_floor}=" + f"{self.valid_days_at_ridge_floor}" + ) + + +def summarize_peak_coverage( + frames: list[pd.DataFrame], + *, + min_peak_bars: int = PEAK_RIDGE_MIN_PEAK_BARS, + min_ridge_bars: int = PEAK_RIDGE_MIN_RIDGE_BARS, + min_classifiable: int = VOLUME_PRV_MIN_CLASSIFIABLE, + counterfactual_peak_floor: int = _COUNTERFACTUAL_PEAK_FLOOR, +) -> PeakCoverage: + """Reduce the per-symbol day-level diagnostics to the scarcity disclosure. + + The three 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_peak_bars=min_peak_bars, + min_ridge_bars=min_ridge_bars, + min_classifiable=min_classifiable, + counterfactual_peak_floor=counterfactual_peak_floor, + ) + empty = tuple((p, float("nan")) for p in _PEAK_PCTL) + if not frames: + return PeakCoverage( + symbol_days=0, + classifiable_days=0, + valid_days=0, + peak_percentiles=empty, + peak_mean=float("nan"), + ridge_median=float("nan"), + days_below_peak_gate=0, + days_below_ridge_gate=0, + days_below_classifiable_gate=0, + valid_days_at_ridge_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 peak scarcity. + scored = classifiable >= min_classifiable + peak = diag.loc[scored, "peak_bars"].to_numpy(dtype=float) + ridge = diag.loc[scored, "ridge_bars"].to_numpy(dtype=float) + # The counterfactual raises the PEAK floor, leaving every other gate exactly as it was. + at_ridge_floor = valid & ( + diag["peak_bars"].to_numpy(dtype=float) >= counterfactual_peak_floor + ) + return PeakCoverage( + symbol_days=int(len(diag)), + classifiable_days=int(scored.sum()), + valid_days=int(valid.sum()), + peak_percentiles=( + tuple((p, float(np.percentile(peak, p))) for p in _PEAK_PCTL) + if peak.size + else empty + ), + peak_mean=float(peak.mean()) if peak.size else float("nan"), + ridge_median=float(np.median(ridge)) if ridge.size else float("nan"), + days_below_peak_gate=int((peak < min_peak_bars).sum()), + days_below_ridge_gate=int((ridge < min_ridge_bars).sum()), + days_below_classifiable_gate=int((~scored).sum()), + valid_days_at_ridge_floor=int(at_ridge_floor.sum()), + **gates, + ) + + +# --------------------------------------------------------------------------- # +# Minute loading (cache-only, per-symbol -> memory-bounded) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _PeakRidgeMinuteLoad: + """Diagnostics from the cache-only per-symbol minute read + aggregation.""" + + factor: pd.Series # MultiIndex(date, symbol) raw peak/ridge amount ratio + 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) + peak_coverage: PeakCoverage + + +def _load_peak_ridge_amount_ratio_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_peak_bars: int, + min_ridge_bars: int, +) -> _PeakRidgeMinuteLoad: + """Compute the raw peak/ridge amount-ratio 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 amount-ratio reduction run entirely inside ``compute_peak_ridge_amount_ratio`` on + 1min bars, which read ``volume`` and ``amount``. The per-day bar-count diagnostics are + collected alongside so the peak 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_peak_ridge_amount_ratio( + 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_peak_bars=min_peak_bars, + 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_peak_coverage( + diagnostics, + min_peak_bars=min_peak_bars, + min_ridge_bars=min_ridge_bars, + min_classifiable=min_classifiable, + ) + if not series: + raise ValueError( + "run-eval-peak-ridge-amount-ratio blocked: no requested symbol produced a " + f"cached peak/ridge amount-ratio value over [{cfg.data.start}, " + f"{cfg.data.end}]. The minute cache is required (this runner never warms it); " + "check coverage." + ) + factor = pd.concat(series).sort_index() + logger.info( + "minute aggregation (cache-only): %d/%d symbols with a value, %d raw 1min rows " + "read, %d factor rows, stk_mins_live_calls=0", + len(covered), len(symbols), raw_rows, len(factor), + ) + logger.info("%s", coverage.render()) + return _PeakRidgeMinuteLoad( + factor=factor, + requested=len(symbols), + covered=tuple(covered), + empty_symbols=tuple(empty), + raw_rows=raw_rows, + live_calls=0, + peak_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. + + Surfaces the same comparison quantities PR-K / PR-L extracted (turnover, net long-short + by cost scenario, lag-1 rank autocorrelation + half-life, cross-section size) so PR-M + sits in ONE table with all nine siblings — which matters more for this run than any + other, since the closing deliverable of the loop is the full ten-factor comparison. + ``ic_pearson_mean`` rides alongside the rank ``ic_mean`` to keep testing the PR-K + review's regularity, that a divergence between the two predicts the monotonicity gate + failing. + + ``aligned_spread_*`` is NOT unreliable here: the frozen layer's cost-sign defect only + mis-signs negative-sign factors, and this factor's pre-registered sign is +1. + ``net_long_short_by_cost`` is still surfaced as the primary read, for comparability. + """ + 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_pearson_mean": pred.get("ic_pearson_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"), + "net_long_short_by_cost": dict(ret_risk.get("net_long_short_by_cost", {}) or {}), + # sign=+1 -> the aligned spreads are NOT mis-signed for this factor. + "aligned_spread_by_cost": dict(ret_risk.get("aligned_spread_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-peak-ridge-amount-ratio 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 PeakRidgeAmountRatioEvalResult: + """Immutable summary of one run-eval-peak-ridge-amount-ratio 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 + peak_coverage: PeakCoverage + 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-peak-ridge-amount-ratio 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-peak-ridge-amount-ratio needs data.cache.enabled=true (it reads the " + "persistent tushare cache and never warms live)." + ) + if cfg.universe.type != "index": + raise ValueError( + "run-eval-peak-ridge-amount-ratio needs universe.type='index' (PIT " + f"membership, e.g. 000905.SH for CSI500); got {cfg.universe.type!r}." + ) + if not cfg.processing.neutralize.enabled: + raise ValueError( + "run-eval-peak-ridge-amount-ratio expects processing.neutralize.enabled=true " + "(industry + size neutralization, matching the report's neutral column and " + "the EvalConfig declaration)." + ) + + +def run_eval_peak_ridge_amount_ratio(config_path: str) -> PeakRidgeAmountRatioEvalResult: + """Run the two real peak/ridge amount-ratio 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 = PeakRidgeAmountRatioFactor(lookback_days=PEAK_RIDGE_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) + + # Peak/ridge amount-ratio factor: cache-only per-symbol aggregation -> raw -> process. + load = _load_peak_ridge_amount_ratio_panel( + cfg, symbols, spec, logger, + lookback_days=PEAK_RIDGE_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_peak_bars=PEAK_RIDGE_MIN_PEAK_BARS, + min_ridge_bars=PEAK_RIDGE_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"], + ) + + return PeakRidgeAmountRatioEvalResult( + 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, + peak_coverage=load.peak_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__ = [ + "PeakCoverage", + "PeakRidgeAmountRatioEvalResult", + "evaluate_two_runs", + "extract_metrics", + "run_eval_peak_ridge_amount_ratio", + "summarize_peak_coverage", +] diff --git a/tests/test_peak_ridge_amount_ratio_factor.py b/tests/test_peak_ridge_amount_ratio_factor.py new file mode 100644 index 0000000..a3ec4d5 --- /dev/null +++ b/tests/test_peak_ridge_amount_ratio_factor.py @@ -0,0 +1,1003 @@ +"""PR-M: PEAK/RIDGE AMOUNT-RATIO factor. + +Same volume classification as PR-F / PR-H / PR-I / PR-J / PR-K / PR-L (REUSED, not +re-implemented). What is NEW here is twofold: + + 1. The factor carries NO PRICE INFORMATION -- both legs are pure traded VALUE. The one + way it can silently degenerate is by letting the wrong bar class into a leg, so every + hand day deliberately contains VALLEY bars whose amounts are enormous and distinctive: + they must never reach either leg. + 2. The AGGREGATION IS A RATIO OF 20-DAY SUMS, not the mean of daily ratios that PR-J + used. The report specifies the two forms differently in adjacent sections (§7.2 vs + §7.1), so ``test_factor_is_the_ratio_of_trailing_sums_not_the_mean_of_daily_ratios`` + pins the distinction with a two-day case whose two forms differ numerically. + +DEFECT-INJECTION DISCIPLINE (the PR-L lesson, applied to EVERY invariance test). +A test of the shape "perturbing X leaves the factor unchanged" is worthless until the +defective implementation has been substituted and the test shown to FAIL against it -- +PR-L shipped an anti-lookahead test that passed even under the buggy code because the +perturbation shape was invisible to the math. Each such test here therefore has a +``_defect_*`` context manager that monkeypatches the production seam into the specific bug +it guards, plus a companion ``test_*_has_teeth_*`` that RUNS the same assertions under the +defect and asserts they blow up. The teeth tests are part of the shipped suite, so the +guarantee cannot be quietly lost by someone "simplifying" a perturbation later. + +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 valleys / isolated peaks / ridge runs at chosen slots and whose AMOUNTS are set +independently, so both legs -- and therefore the ratio -- are known in closed form. +""" + +from __future__ import annotations + +import contextlib + +import numpy as np +import pandas as pd +import pytest + +import data.clean.intraday_amount_ratio as amount_ratio_mod +from data.clean.intraday_amount_ratio import ( + PEAK_RIDGE_LOOKBACK_DAYS, + PEAK_RIDGE_MIN_PEAK_BARS, + PEAK_RIDGE_MIN_RIDGE_BARS, + compute_peak_ridge_amount_ratio, + peak_ridge_amount_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_CLASSIFIABLE, + VOLUME_PRV_MIN_VALID_DAYS, + VOLUME_PRV_SIGMA_K, + peak_mask_for_symbol, + prepare_visible_minute_bars, +) +from factors.compute.intraday_derived import PeakRidgeAmountRatioFactor +from factors.spec import FactorSpec + +_SYM = "000001.SZ" + + +def _bars(rows): + """rows = [(time, symbol, volume, amount), ...] -> normalized 1min bars. + + ``amount`` is set INDEPENDENTLY of ``volume``: volume drives the classification, amount + is the factor's whole quantity. OHLC are dummy constants (this factor reads neither). + ``normalize_intraday_bars`` sets ``available_time = bar_end + 1min``, so the 14:50 PIT + cutoff excludes any bar with ``bar_end >= 14:50``. + """ + 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 _session(day, vols, amts, sym=_SYM, start="09:31:00"): + """One session of CONSECUTIVE 1-minute bars carrying ``vols`` / ``amts``.""" + base = pd.Timestamp(day) + pd.Timedelta(start) + return [ + (base + pd.Timedelta(minutes=i), sym, v, a) + for i, (v, a) in enumerate(zip(vols, amts)) + ] + + +# A VALLEY amount large enough that leaking even ONE valley bar into either leg would move +# the ratio by orders of magnitude. Used on every hand day. +_VALLEY_AMT = 500_000.0 + + +def _background(n_days, n_slots, sym=_SYM, start_day="2021-07-01"): + """``n_days`` prior days of flat volume-100 sessions. + + Same-slot baseline becomes mu=100, sigma=0 -> the eruptive threshold is exactly 100, + so on a test day a volume of 100 is a VALLEY and anything above erupts. + """ + 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, [_VALLEY_AMT] * n_slots, sym=sym) + return rows + + +_BG_DAYS = 10 +_TEST_DAY = pd.Timestamp("2021-07-11") +_DAY_1 = "2021-07-11" +_DAY_2 = "2021-07-12" + +# Gates small enough that the engineered test day(s) are the only VALID days; the +# peak-bar / ridge-bar / 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=PEAK_RIDGE_LOOKBACK_DAYS, + min_valid_days=1, + min_classifiable=1, + min_peak_bars=1, + min_ridge_bars=1, +) + + +# --------------------------------------------------------------------------- # +# Hand-computed amount ratios (2 non-trivial cases, closed-form fractions) +# --------------------------------------------------------------------------- # +# Both cases use 20 slots with the SAME geometry, so only the AMOUNTS differ: +# ISOLATED PEAKS at slots 2, 5, 8 -- each flanked by valleys, so each is a true peak. +# RIDGE RUNS at (11, 12) and (15, 16) -- adjacent eruptions, so none is isolated. +# Every other slot is a VALLEY carrying the huge _VALLEY_AMT. +_N = 20 +_PEAKS = (2, 5, 8) +_RIDGES = (11, 12, 15, 16) + +# CASE A -- peak-heavy: peaks 1000 + 2000 + 3000 = 6000; ridges 500 + 700 + 800 + 1000 = +# 3000 -> ratio 2.0 (informed participation dominates). +_CASE_A_PEAK_AMTS = (1000.0, 2000.0, 3000.0) +_CASE_A_RIDGE_AMTS = (500.0, 700.0, 800.0, 1000.0) +_CASE_A_PEAK_SUM = 6000.0 +_CASE_A_RIDGE_SUM = 3000.0 +_CASE_A_RATIO = 2.0 + +# CASE B -- ridge-heavy: peaks 300 + 400 + 500 = 1200; ridges 1000 + 1100 + 1200 + 1300 = +# 4600 -> ratio 6/23 = 0.26087. The OPPOSITE side of 1.0 from case A, so an inverted +# numerator/denominator cannot pass both. +_CASE_B_PEAK_AMTS = (300.0, 400.0, 500.0) +_CASE_B_RIDGE_AMTS = (1000.0, 1100.0, 1200.0, 1300.0) +_CASE_B_PEAK_SUM = 1200.0 +_CASE_B_RIDGE_SUM = 4600.0 +_CASE_B_RATIO = 1200.0 / 4600.0 + + +def _day(peak_amts, ridge_amts, *, peak_vol=300.0, ridge_vol=200.0): + """Build the (vols, amts) of one hand day from its peak / ridge amounts.""" + vols = [100.0] * _N + amts = [_VALLEY_AMT] * _N + for slot, amt in zip(_PEAKS, peak_amts): + vols[slot] = peak_vol + amts[slot] = amt + for slot, amt in zip(_RIDGES, ridge_amts): + vols[slot] = ridge_vol + amts[slot] = amt + return vols, amts + + +def _case_a_day(): + return _day(_CASE_A_PEAK_AMTS, _CASE_A_RIDGE_AMTS) + + +def _case_b_day(): + return _day(_CASE_B_PEAK_AMTS, _CASE_B_RIDGE_AMTS) + + +def _one_day_bars(day=_DAY_1, case=_case_a_day): + return _bars(_background(_BG_DAYS, _N) + _session(day, *case())) + + +def test_hand_value_case_a_ratio_above_one(): + out = compute_peak_ridge_amount_ratio(_one_day_bars(case=_case_a_day), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(_CASE_A_RATIO) + assert out.loc[(_TEST_DAY, _SYM)] > 1.0 + + +def test_hand_value_case_b_ratio_below_one(): + out = compute_peak_ridge_amount_ratio(_one_day_bars(case=_case_b_day), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(_CASE_B_RATIO) + assert out.loc[(_TEST_DAY, _SYM)] < 1.0 + + +def test_peak_is_the_numerator_and_ridge_the_denominator(): + """Scaling one leg must move the ratio in the DIRECTION that leg sits on. + + Doubling the peak amounts doubles the factor; doubling the ridge amounts halves it. + A swapped numerator/denominator fails both assertions. + """ + base = compute_peak_ridge_amount_ratio(_one_day_bars(), **_KW).loc[(_TEST_DAY, _SYM)] + + doubled_peaks = _bars( + _background(_BG_DAYS, _N) + + _session( + _DAY_1, + *_day([2 * a for a in _CASE_A_PEAK_AMTS], _CASE_A_RIDGE_AMTS), + ) + ) + doubled_ridges = _bars( + _background(_BG_DAYS, _N) + + _session( + _DAY_1, + *_day(_CASE_A_PEAK_AMTS, [2 * a for a in _CASE_A_RIDGE_AMTS]), + ) + ) + key = (_TEST_DAY, _SYM) + assert compute_peak_ridge_amount_ratio(doubled_peaks, **_KW).loc[key] == pytest.approx( + 2.0 * base + ) + assert compute_peak_ridge_amount_ratio( + doubled_ridges, **_KW + ).loc[key] == pytest.approx(base / 2.0) + + +def test_legs_by_day_match_the_hand_sums(): + """The per-day leg totals themselves, not just their ratio.""" + visible = prepare_visible_minute_bars( + _one_day_bars(), decision_time="14:50:00", extra_columns=("amount",) + ) + work = peak_mask_for_symbol(visible.reset_index(drop=True)) + legs = peak_ridge_amount_by_day( + work, min_peak_bars=1, min_ridge_bars=1, min_classifiable=1 + ) + row = legs.loc[_TEST_DAY] + assert row["peak_amt"] == pytest.approx(_CASE_A_PEAK_SUM) + assert row["ridge_amt"] == pytest.approx(_CASE_A_RIDGE_SUM) + + +def test_ridge_run_bars_are_all_ridges_and_flanked_eruptions_are_all_peaks(): + """The hand geometry is what the REUSED classifier actually produces.""" + visible = prepare_visible_minute_bars( + _one_day_bars(), decision_time="14:50:00", extra_columns=("amount",) + ) + work = peak_mask_for_symbol(visible.reset_index(drop=True)) + day = work[work["trade_date"] == _TEST_DAY].reset_index(drop=True) + assert list(np.flatnonzero(day["peak"].to_numpy())) == list(_PEAKS) + assert list(np.flatnonzero(day["ridge"].to_numpy())) == list(_RIDGES) + + +# --------------------------------------------------------------------------- # +# THE AGGREGATION FORM: ratio of sums, NOT mean of daily ratios (report §7.2 vs §7.1) +# --------------------------------------------------------------------------- # +_TWO_DAY_RATIO_OF_SUMS = (_CASE_A_PEAK_SUM + _CASE_B_PEAK_SUM) / ( + _CASE_A_RIDGE_SUM + _CASE_B_RIDGE_SUM +) # 7200 / 7600 = 18/19 = 0.947368... +_TWO_DAY_MEAN_OF_RATIOS = (_CASE_A_RATIO + _CASE_B_RATIO) / 2.0 # 1.130434... + + +def _two_day_bars(): + return _bars( + _background(_BG_DAYS, _N) + + _session(_DAY_1, *_case_a_day()) + + _session(_DAY_2, *_case_b_day()) + ) + + +def test_factor_is_the_ratio_of_trailing_sums_not_the_mean_of_daily_ratios(): + """§7.2 says "20 日量峰总成交额与量岭总成交额，二者做比" -- pool first, divide once. + + The two candidate forms are numerically far apart on this pair of days (0.947 vs + 1.130) and sit on OPPOSITE sides of 1.0, so the assertion below cannot be satisfied by + the mean-of-ratios form that PR-J's §7.1 wording specifies. + """ + kw = dict(_KW, min_valid_days=2) + out = compute_peak_ridge_amount_ratio(_two_day_bars(), **kw) + value = out.loc[(pd.Timestamp(_DAY_2), _SYM)] + assert value == pytest.approx(_TWO_DAY_RATIO_OF_SUMS) + assert value != pytest.approx(_TWO_DAY_MEAN_OF_RATIOS) + # And the two forms really are distinguishable here (guards the fixture itself). + assert abs(_TWO_DAY_RATIO_OF_SUMS - _TWO_DAY_MEAN_OF_RATIOS) > 0.15 + + +def test_busy_days_dominate_the_pool_which_is_what_a_ratio_of_sums_means(): + """A ratio of sums is AMOUNT-weighted: scaling one day's whole activity moves it. + + Multiplying day 1's BOTH legs by 10 leaves day 1's own daily ratio untouched (2.0), so + a mean of daily ratios would not move at all. The ratio of sums must move, because day + 1 now carries ten times the weight in the pool. + """ + kw = dict(_KW, min_valid_days=2) + flat = compute_peak_ridge_amount_ratio(_two_day_bars(), **kw) + scaled_rows = ( + _background(_BG_DAYS, _N) + + _session( + _DAY_1, + *_day( + [10 * a for a in _CASE_A_PEAK_AMTS], + [10 * a for a in _CASE_A_RIDGE_AMTS], + ), + ) + + _session(_DAY_2, *_case_b_day()) + ) + scaled = compute_peak_ridge_amount_ratio(_bars(scaled_rows), **kw) + key = (pd.Timestamp(_DAY_2), _SYM) + expected = (10 * _CASE_A_PEAK_SUM + _CASE_B_PEAK_SUM) / ( + 10 * _CASE_A_RIDGE_SUM + _CASE_B_RIDGE_SUM + ) + assert scaled.loc[key] == pytest.approx(expected) + assert scaled.loc[key] != pytest.approx(flat.loc[key]) + # Day 1's own daily ratio is unchanged at 2.0 -- read off a single-day window -- so a + # MEAN of daily ratios would have been identical in both runs. + solo = compute_peak_ridge_amount_ratio( + _bars(scaled_rows), **dict(_KW, lookback_days=1, min_valid_days=1) + ) + assert solo.loc[(pd.Timestamp(_DAY_1), _SYM)] == pytest.approx(_CASE_A_RATIO) + + +# --------------------------------------------------------------------------- # +# Validity gates: >=5 peak bars (PINNED, the scarce leg), >=10 ridge bars, classifiable +# --------------------------------------------------------------------------- # +def _geometry_day(n_peaks, n_ridges, n_slots=140): + """A day with EXACTLY ``n_peaks`` isolated peaks and ``n_ridges`` ridge bars. + + Peaks are laid down first at stride 3, so each is flanked by valleys and is genuinely + ISOLATED. Ridge RUNS follow well clear of them, separated by two valleys so adjacent + runs cannot merge. A run of length 1 would itself be isolated and classify as a PEAK, + so every run is >= 2 and an ODD ridge count is built from one run of 3 plus pairs. + """ + if n_ridges == 1 or n_ridges < 0: + raise ValueError(f"n_ridges must be 0 or >= 2; got {n_ridges}.") + vols = [100.0] * n_slots + amts = [_VALLEY_AMT] * n_slots + for i in range(n_peaks): + slot = 1 + 3 * i + vols[slot] = 300.0 + amts[slot] = 1000.0 + + runs = [] + remaining = n_ridges + if remaining % 2 == 1: + runs.append(3) + remaining -= 3 + runs += [2] * (remaining // 2) + + cursor = 3 * n_peaks + 4 + for run in runs: + for k in range(run): + vols[cursor + k] = 200.0 + amts[cursor + k] = 500.0 + cursor += run + 2 + return vols, amts + + +def _geometry_counts(vols, amts, n_slots): + """The realized (peak, ridge) bar counts of a geometry day, via the real classifier.""" + rows = _background(_BG_DAYS, n_slots) + _session(_DAY_1, vols, amts) + visible = prepare_visible_minute_bars( + _bars(rows), decision_time="14:50:00", extra_columns=("amount",) + ) + work = peak_mask_for_symbol(visible.reset_index(drop=True)) + day = work[work["trade_date"] == _TEST_DAY] + return int(day["peak"].sum()), int(day["ridge"].sum()) + + +@pytest.mark.parametrize("n_peaks", [4, 5, 6]) +def test_peak_bar_gate_boundary_at_five(n_peaks): + """4 peak bars FAIL the pinned floor of 5; 5 and 6 PASS. The 4-vs-5 boundary.""" + n_slots = 140 + vols, amts = _geometry_day(n_peaks, 12, n_slots=n_slots) + peaks, ridges = _geometry_counts(vols, amts, n_slots) + assert peaks == n_peaks and ridges == 12 # the fixture is what it claims + rows = _background(_BG_DAYS, n_slots) + _session(_DAY_1, vols, amts) + out = compute_peak_ridge_amount_ratio( + _bars(rows), + **dict( + _KW, + min_peak_bars=PEAK_RIDGE_MIN_PEAK_BARS, + min_ridge_bars=PEAK_RIDGE_MIN_RIDGE_BARS, + min_classifiable=1, + ), + ) + key = (_TEST_DAY, _SYM) + if n_peaks < PEAK_RIDGE_MIN_PEAK_BARS: + assert key not in out.index or not np.isfinite(out.loc[key]) + else: + assert np.isfinite(out.loc[key]) + + +@pytest.mark.parametrize("n_ridges", [8, 9, 10, 11]) +def test_ridge_bar_gate_boundary_at_ten(n_ridges): + """9 ridge bars FAIL the floor of 10; 10 and 11 PASS.""" + n_slots = 140 + vols, amts = _geometry_day(6, n_ridges, n_slots=n_slots) + peaks, ridges = _geometry_counts(vols, amts, n_slots) + assert peaks == 6 and ridges == n_ridges + rows = _background(_BG_DAYS, n_slots) + _session(_DAY_1, vols, amts) + out = compute_peak_ridge_amount_ratio( + _bars(rows), + **dict( + _KW, + min_peak_bars=PEAK_RIDGE_MIN_PEAK_BARS, + min_ridge_bars=PEAK_RIDGE_MIN_RIDGE_BARS, + min_classifiable=1, + ), + ) + key = (_TEST_DAY, _SYM) + if n_ridges < PEAK_RIDGE_MIN_RIDGE_BARS: + assert key not in out.index or not np.isfinite(out.loc[key]) + else: + assert np.isfinite(out.loc[key]) + + +def test_bar_counts_are_taken_AFTER_the_positive_trade_guard(): + """A day with 5 peak bars, one of which traded nothing, has only 4 COUNTABLE peaks.""" + n_slots = 140 + vols, amts = _geometry_day(5, 12, n_slots=n_slots) + peak_slots = [1 + 3 * i for i in range(5)] + amts[peak_slots[0]] = 0.0 # traded nothing -> dropped by the guard AND uncounted + rows = _background(_BG_DAYS, n_slots) + _session(_DAY_1, vols, amts) + out = compute_peak_ridge_amount_ratio( + _bars(rows), + **dict( + _KW, + min_peak_bars=PEAK_RIDGE_MIN_PEAK_BARS, + min_ridge_bars=PEAK_RIDGE_MIN_RIDGE_BARS, + min_classifiable=1, + ), + ) + key = (_TEST_DAY, _SYM) + assert key not in out.index or not np.isfinite(out.loc[key]) + + +def test_day_with_no_ridge_is_invalid(): + vols = [100.0] * _N + amts = [_VALLEY_AMT] * _N + for slot, amt in zip(_PEAKS, _CASE_A_PEAK_AMTS): + vols[slot] = 300.0 + amts[slot] = amt + rows = _background(_BG_DAYS, _N) + _session(_DAY_1, vols, amts) + out = compute_peak_ridge_amount_ratio(_bars(rows), **_KW) + assert (_TEST_DAY, _SYM) not in out.index + + +def test_day_with_no_peak_is_invalid(): + vols = [100.0] * _N + amts = [_VALLEY_AMT] * _N + for slot, amt in zip(_RIDGES, _CASE_A_RIDGE_AMTS): + vols[slot] = 200.0 + amts[slot] = amt + rows = _background(_BG_DAYS, _N) + _session(_DAY_1, vols, amts) + out = compute_peak_ridge_amount_ratio(_bars(rows), **_KW) + assert (_TEST_DAY, _SYM) not in out.index + + +def test_min_classifiable_gate_invalidates_thin_days(): + out = compute_peak_ridge_amount_ratio( + _one_day_bars(), **dict(_KW, min_classifiable=_N + 1) + ) + assert (_TEST_DAY, _SYM) not in out.index + + +def test_min_valid_days_floor_returns_nan_until_enough_valid_days(): + out = compute_peak_ridge_amount_ratio(_two_day_bars(), **dict(_KW, min_valid_days=3)) + assert out.empty or not np.isfinite(out.to_numpy()).any() + + +def test_window_drops_days_older_than_lookback(): + """With lookback_days=1 only day d itself is pooled, so day 2 == its own ratio.""" + out = compute_peak_ridge_amount_ratio( + _two_day_bars(), **dict(_KW, lookback_days=1, min_valid_days=1) + ) + assert out.loc[(pd.Timestamp(_DAY_2), _SYM)] == pytest.approx(_CASE_B_RATIO) + assert out.loc[(pd.Timestamp(_DAY_1), _SYM)] == pytest.approx(_CASE_A_RATIO) + + +def test_default_gates_match_the_pinned_definition(): + assert PEAK_RIDGE_LOOKBACK_DAYS == 20 + assert PEAK_RIDGE_MIN_PEAK_BARS == 5 + assert PEAK_RIDGE_MIN_RIDGE_BARS == 10 + # The PEAK floor is the LOWER one -- the REVERSE of PR-J's valley/ridge asymmetry. + assert PEAK_RIDGE_MIN_PEAK_BARS < PEAK_RIDGE_MIN_RIDGE_BARS + assert VOLUME_PRV_MIN_VALID_DAYS == 10 + assert VOLUME_PRV_MIN_CLASSIFIABLE == 100 + + +# --------------------------------------------------------------------------- # +# The positive-trade guard: two-sided (dropped, AND would have mattered) +# --------------------------------------------------------------------------- # +@contextlib.contextmanager +def _defect_no_trade_guard(monkeypatch): + """DEFECT: the positive-trade guard admits every bar.""" + monkeypatch.setattr( + amount_ratio_mod, + "_tradable_amount", + lambda amt: np.ones(len(amt), dtype=bool), + ) + yield + + +@pytest.mark.parametrize("bad", [0.0, -5000.0, float("nan"), float("inf")]) +def test_non_positive_amount_is_dropped_from_the_peak_leg(bad): + """A degenerate peak bar leaves the OTHER peaks' sum exactly intact.""" + vols, amts = _case_a_day() + amts[_PEAKS[0]] = bad + rows = _background(_BG_DAYS, _N) + _session(_DAY_1, vols, amts) + out = compute_peak_ridge_amount_ratio(_bars(rows), **_KW) + expected = (_CASE_A_PEAK_SUM - _CASE_A_PEAK_AMTS[0]) / _CASE_A_RIDGE_SUM + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(expected) + + +@pytest.mark.parametrize("bad", [0.0, -5000.0, float("nan"), float("inf")]) +def test_non_positive_amount_is_dropped_from_the_ridge_leg(bad): + vols, amts = _case_a_day() + amts[_RIDGES[0]] = bad + rows = _background(_BG_DAYS, _N) + _session(_DAY_1, vols, amts) + out = compute_peak_ridge_amount_ratio(_bars(rows), **_KW) + expected = _CASE_A_PEAK_SUM / (_CASE_A_RIDGE_SUM - _CASE_A_RIDGE_AMTS[0]) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(expected) + + +def test_guard_test_has_teeth_without_the_guard_the_value_changes(monkeypatch): + """The counter-test: the dropped bar really would have mattered. + + Without this, the two tests above could pass because the bar was inert anyway. Under + the defect the negative amount is summed into the peak leg and the ratio drops, so the + guard is doing the work. The magnitude is chosen to keep the defective peak sum + POSITIVE, so the day still clears the validity gate and the two values are directly + comparable rather than one of them simply vanishing. + """ + bad = -500.0 + vols, amts = _case_a_day() + amts[_PEAKS[0]] = bad + rows = _background(_BG_DAYS, _N) + _session(_DAY_1, vols, amts) + guarded = compute_peak_ridge_amount_ratio(_bars(rows), **_KW).loc[(_TEST_DAY, _SYM)] + with _defect_no_trade_guard(monkeypatch): + broken = compute_peak_ridge_amount_ratio(_bars(rows), **_KW).loc[ + (_TEST_DAY, _SYM) + ] + assert guarded == pytest.approx( + (_CASE_A_PEAK_SUM - _CASE_A_PEAK_AMTS[0]) / _CASE_A_RIDGE_SUM + ) + # The defect sums -500 in place of dropping it: 5000 -> 4500 in the numerator. + assert broken == pytest.approx( + (_CASE_A_PEAK_SUM - _CASE_A_PEAK_AMTS[0] + bad) / _CASE_A_RIDGE_SUM + ) + assert broken != pytest.approx(guarded) + + +# --------------------------------------------------------------------------- # +# INVARIANCE 1 -- VALLEY bars reach NEITHER leg (this factor reads only eruptions) +# --------------------------------------------------------------------------- # +@contextlib.contextmanager +def _defect_valleys_in_the_peak_leg(monkeypatch): + """DEFECT: the peak leg is ``peak | valley`` -- i.e. "everything that is not a ridge". + + A plausible mis-reading of the taxonomy, and the exact bug the invariance test below + exists to catch. + """ + real = amount_ratio_mod.peak_mask_for_symbol + + def broken(g, **kwargs): + work = real(g, **kwargs) + work = work.copy() + work["peak"] = work["peak"] | work["valley"] + return work + + monkeypatch.setattr(amount_ratio_mod, "peak_mask_for_symbol", broken) + yield + + +def _valley_perturbed_bars(): + """Case A with EVERY valley bar's amount changed (peaks / ridges untouched).""" + vols, amts = _case_a_day() + for slot in range(_N): + if slot not in _PEAKS and slot not in _RIDGES: + amts[slot] = 7_777_777.0 + return _bars(_background(_BG_DAYS, _N) + _session(_DAY_1, vols, amts)) + + +def test_perturbing_valley_amounts_does_not_change_the_factor(): + """VALLEY minutes are not part of either leg, so their traded value is irrelevant.""" + key = (_TEST_DAY, _SYM) + clean = compute_peak_ridge_amount_ratio(_one_day_bars(), **_KW).loc[key] + perturbed = compute_peak_ridge_amount_ratio(_valley_perturbed_bars(), **_KW).loc[key] + assert clean == pytest.approx(_CASE_A_RATIO) + assert perturbed == pytest.approx(clean) + + +def test_valley_invariance_has_teeth_under_the_valleys_in_leg_defect(monkeypatch): + """The SAME assertions, run against the defective mask, must FAIL. + + This is the PR-L discipline: an invariance test is only evidence once the bug it + claims to exclude has been shown to trip it. + """ + key = (_TEST_DAY, _SYM) + with _defect_valleys_in_the_peak_leg(monkeypatch): + clean = compute_peak_ridge_amount_ratio(_one_day_bars(), **_KW).loc[key] + perturbed = compute_peak_ridge_amount_ratio(_valley_perturbed_bars(), **_KW).loc[ + key + ] + assert clean != pytest.approx(_CASE_A_RATIO) + assert perturbed != pytest.approx(clean) + + +# --------------------------------------------------------------------------- # +# INVARIANCE 2 -- PIT: post-14:50 bars cannot influence the value +# --------------------------------------------------------------------------- # +_LATE_START = "14:50:00" +_LATE_SLOTS = 8 + + +@contextlib.contextmanager +def _defect_no_pit_cutoff(monkeypatch): + """DEFECT: the 14:50 truncation is gone (the whole session is 'visible').""" + real = amount_ratio_mod.prepare_visible_minute_bars + + def broken(bars, *, decision_time, extra_columns=()): + return real(bars, decision_time="23:59:59", extra_columns=extra_columns) + + monkeypatch.setattr(amount_ratio_mod, "prepare_visible_minute_bars", broken) + yield + + +def _two_block_background(n_days, early_slots, sym=_SYM, start_day="2021-07-01"): + """Background covering 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 truncation were removed, the late bars would be fully CLASSIFIABLE and would really + change the factor -- so "the perturbation changes nothing" 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, [_VALLEY_AMT] * early_slots, sym=sym) + rows += _session( + day, [100.0] * _LATE_SLOTS, [_VALLEY_AMT] * _LATE_SLOTS, + sym=sym, start=_LATE_START, + ) + return rows + + +def _late_block(day, sym=_SYM): + """A post-14:50 block carrying a huge RIDGE run (slots 0,1 and 4,5) and a peak at 3. + + The amounts are orders of magnitude above the early block's, so if these bars ever + reached either leg the ratio would move enormously. + """ + vols = [100.0] * _LATE_SLOTS + amts = [_VALLEY_AMT] * _LATE_SLOTS + for s in (0, 1, 4, 5): + vols[s] = 200.0 + amts[s] = 900_000.0 + vols[3] = 300.0 + amts[3] = 400_000.0 + return _session(day, vols, amts, sym=sym, start=_LATE_START) + + +def _pit_pair(): + """(clean bars, bars + a post-14:50 block) sharing the same early session.""" + rows = _two_block_background(_BG_DAYS, _N) + _session(_DAY_1, *_case_a_day()) + return _bars(rows), _bars(rows + _late_block(_DAY_1)) + + +def test_perturbing_post_1450_bars_does_not_change_the_factor(): + clean_bars, perturbed_bars = _pit_pair() + key = (_TEST_DAY, _SYM) + clean = compute_peak_ridge_amount_ratio(clean_bars, **_KW).loc[key] + perturbed = compute_peak_ridge_amount_ratio(perturbed_bars, **_KW).loc[key] + assert clean == pytest.approx(_CASE_A_RATIO) + assert perturbed == pytest.approx(clean) + + +def test_pit_invariance_has_teeth_under_the_no_cutoff_defect(monkeypatch): + """The SAME assertions, run with the 14:50 truncation removed, must FAIL.""" + clean_bars, perturbed_bars = _pit_pair() + key = (_TEST_DAY, _SYM) + with _defect_no_pit_cutoff(monkeypatch): + clean = compute_peak_ridge_amount_ratio(clean_bars, **_KW).loc[key] + perturbed = compute_peak_ridge_amount_ratio(perturbed_bars, **_KW).loc[key] + # The late bars are inert only because they are HIDDEN; expose them and they dominate. + assert perturbed != pytest.approx(clean) + assert abs(perturbed - clean) / clean > 0.5 + + +def test_public_decision_time_also_exposes_the_late_block(): + """A defect-free restatement of the same fact, via the public ``decision_time`` arg.""" + _, perturbed_bars = _pit_pair() + key = (_TEST_DAY, _SYM) + truncated = compute_peak_ridge_amount_ratio(perturbed_bars, **_KW).loc[key] + untruncated = compute_peak_ridge_amount_ratio( + perturbed_bars, decision_time="23:59:59", **_KW + ).loc[key] + assert truncated == pytest.approx(_CASE_A_RATIO) + assert np.isfinite(untruncated) + assert abs(untruncated - truncated) / truncated > 0.5 + + +# --------------------------------------------------------------------------- # +# INVARIANCE 3 -- a FUTURE day cannot change an earlier value (trailing window) +# --------------------------------------------------------------------------- # +@contextlib.contextmanager +def _defect_forward_looking_window(monkeypatch): + """DEFECT: the rolling window LEADS instead of trails (classic lookahead).""" + + def broken(legs, *, lookback_days, min_valid_days): + ordered = legs.sort_index() + roll = ( + ordered[::-1].rolling(lookback_days, min_periods=min_valid_days).sum()[::-1] + ) + return roll["peak_amt"] / roll["ridge_amt"] + + monkeypatch.setattr(amount_ratio_mod, "_trailing_ratio_of_sums", broken) + yield + + +def _future_pair(): + """(bars ending on day 1, the same bars plus a wildly different day 2).""" + base = _background(_BG_DAYS, _N) + _session(_DAY_1, *_case_a_day()) + future = base + _session(_DAY_2, *_case_b_day()) + return _bars(base), _bars(future) + + +def test_future_day_does_not_change_the_earlier_factor(): + base_bars, future_bars = _future_pair() + key = (_TEST_DAY, _SYM) + a = compute_peak_ridge_amount_ratio(base_bars, **_KW).loc[key] + b = compute_peak_ridge_amount_ratio(future_bars, **_KW).loc[key] + assert a == pytest.approx(_CASE_A_RATIO) + assert b == pytest.approx(a) + + +def test_future_day_invariance_has_teeth_under_the_forward_window_defect(monkeypatch): + """The SAME assertions, run against a LEADING window, must FAIL.""" + base_bars, future_bars = _future_pair() + key = (_TEST_DAY, _SYM) + with _defect_forward_looking_window(monkeypatch): + a = compute_peak_ridge_amount_ratio(base_bars, **_KW).loc[key] + b = compute_peak_ridge_amount_ratio(future_bars, **_KW).loc[key] + # Day 1's value now absorbs day 2 (case B), so adding day 2 moves it. + assert b != pytest.approx(a) + assert b == pytest.approx(_TWO_DAY_RATIO_OF_SUMS) + + +# --------------------------------------------------------------------------- # +# INVARIANCE 4 -- cross-symbol isolation +# --------------------------------------------------------------------------- # +@contextlib.contextmanager +def _defect_symbol_frames_mislabeled(monkeypatch): + """DEFECT: each symbol's LABEL is paired with a DIFFERENT symbol's bars. + + Note which bug this is and why. LITERAL pooling -- running the classifier over two + symbols at once -- cannot happen silently: ``peak_mask_for_symbol`` unstacks by + ``(trade_date, slot)`` and pandas raises "Index contains duplicate entries" the moment + two symbols share a timestamp, which is a structural guarantee rather than a test + result. The failure mode that CAN pass unnoticed is a mislabeled pairing in the split + itself, so that is what the invariance test is made to face. + """ + + def broken(visible): + frames = [ + (str(sym), g.reset_index(drop=True)) + for sym, g in visible.groupby("symbol", sort=True) + ] + rotated = frames[1:] + frames[:1] + for (sym, _), (_, g) in zip(frames, rotated): + yield sym, g + + monkeypatch.setattr(amount_ratio_mod, "_symbol_frames", broken) + yield + + +def _two_symbol_bars(): + """AAA on case A, BBB on case B -- deliberately different ratios.""" + rows = _background(_BG_DAYS, _N, sym="AAA.SZ") + rows += _session(_DAY_1, *_case_a_day(), sym="AAA.SZ") + rows += _background(_BG_DAYS, _N, sym="BBB.SZ") + rows += _session(_DAY_1, *_case_b_day(), sym="BBB.SZ") + return _bars(rows) + + +def _solo_bars(sym, case): + return _bars( + _background(_BG_DAYS, _N, sym=sym) + _session(_DAY_1, *case(), sym=sym) + ) + + +def test_per_symbol_isolation(): + """Each symbol's value equals the value it has when computed entirely alone.""" + together = compute_peak_ridge_amount_ratio(_two_symbol_bars(), **_KW) + solo_a = compute_peak_ridge_amount_ratio(_solo_bars("AAA.SZ", _case_a_day), **_KW) + solo_b = compute_peak_ridge_amount_ratio(_solo_bars("BBB.SZ", _case_b_day), **_KW) + assert together.loc[(_TEST_DAY, "AAA.SZ")] == pytest.approx( + solo_a.loc[(_TEST_DAY, "AAA.SZ")] + ) + assert together.loc[(_TEST_DAY, "BBB.SZ")] == pytest.approx( + solo_b.loc[(_TEST_DAY, "BBB.SZ")] + ) + assert together.loc[(_TEST_DAY, "AAA.SZ")] == pytest.approx(_CASE_A_RATIO) + assert together.loc[(_TEST_DAY, "BBB.SZ")] == pytest.approx(_CASE_B_RATIO) + + +def test_isolation_has_teeth_under_the_mislabeled_frames_defect(monkeypatch): + """The SAME assertions, run against a mislabeled split, must FAIL.""" + solo_a = compute_peak_ridge_amount_ratio(_solo_bars("AAA.SZ", _case_a_day), **_KW) + solo_b = compute_peak_ridge_amount_ratio(_solo_bars("BBB.SZ", _case_b_day), **_KW) + with _defect_symbol_frames_mislabeled(monkeypatch): + together = compute_peak_ridge_amount_ratio(_two_symbol_bars(), **_KW) + assert together.loc[(_TEST_DAY, "AAA.SZ")] != pytest.approx( + solo_a.loc[(_TEST_DAY, "AAA.SZ")] + ) + assert together.loc[(_TEST_DAY, "BBB.SZ")] != pytest.approx( + solo_b.loc[(_TEST_DAY, "BBB.SZ")] + ) + # Each name now carries the OTHER's ratio -- the swap is exact, not just noisy. + assert together.loc[(_TEST_DAY, "AAA.SZ")] == pytest.approx(_CASE_B_RATIO) + assert together.loc[(_TEST_DAY, "BBB.SZ")] == pytest.approx(_CASE_A_RATIO) + + +def test_literal_pooling_cannot_happen_silently(monkeypatch): + """Records the structural guarantee the mislabel defect stands in for. + + Handing the shared classifier two symbols at once RAISES rather than returning a + quietly-wrong frame, so cross-symbol contamination has no silent path at that layer. + """ + + def pooled(visible): + frame = visible.reset_index(drop=True) + for sym in sorted(set(visible["symbol"])): + yield str(sym), frame.copy() + + monkeypatch.setattr(amount_ratio_mod, "_symbol_frames", pooled) + with pytest.raises(ValueError, match="duplicate entries"): + compute_peak_ridge_amount_ratio(_two_symbol_bars(), **_KW) + + +# --------------------------------------------------------------------------- # +# REUSE NON-DRIFT: PR-M adds nothing to the shared taxonomy +# --------------------------------------------------------------------------- # +def test_the_three_masks_still_partition_the_classifiable_bars(): + visible = prepare_visible_minute_bars( + _one_day_bars(), decision_time="14:50:00", extra_columns=("amount",) + ) + work = peak_mask_for_symbol(visible.reset_index(drop=True)) + valley = work["valley"].to_numpy(dtype=bool) + peak = work["peak"].to_numpy(dtype=bool) + ridge = work["ridge"].to_numpy(dtype=bool) + classifiable = work["classifiable"].to_numpy(dtype=bool) + assert np.array_equal(valley | peak | ridge, classifiable) + assert not (valley & peak).any() + assert not (valley & ridge).any() + assert not (peak & ridge).any() + + +def test_peak_and_ridge_legs_use_the_shared_masks_verbatim(): + """The legs are exactly ``mask & tradable`` on the SHARED columns, nothing else.""" + visible = prepare_visible_minute_bars( + _one_day_bars(), decision_time="14:50:00", extra_columns=("amount",) + ) + work = peak_mask_for_symbol(visible.reset_index(drop=True)) + legs = peak_ridge_amount_by_day( + work, min_peak_bars=1, min_ridge_bars=1, min_classifiable=1 + ) + amt = work["amount"].to_numpy(dtype=float) + day = work["trade_date"].to_numpy() == np.datetime64(_TEST_DAY) + tradable = np.isfinite(amt) & (amt > 0.0) + expected_peak = amt[day & work["peak"].to_numpy(dtype=bool) & tradable].sum() + expected_ridge = amt[day & work["ridge"].to_numpy(dtype=bool) & tradable].sum() + assert legs.loc[_TEST_DAY, "peak_amt"] == pytest.approx(expected_peak) + assert legs.loc[_TEST_DAY, "ridge_amt"] == pytest.approx(expected_ridge) + + +# --------------------------------------------------------------------------- # +# Diagnostics channel (the peak-scarcity disclosure) +# --------------------------------------------------------------------------- # +def test_diagnostics_expose_the_per_day_bar_counts(): + visible = prepare_visible_minute_bars( + _one_day_bars(), decision_time="14:50:00", extra_columns=("amount",) + ) + work = peak_mask_for_symbol(visible.reset_index(drop=True)) + legs, diag = peak_ridge_amount_by_day( + work, + min_peak_bars=1, + min_ridge_bars=1, + min_classifiable=1, + with_diagnostics=True, + ) + assert set(amount_ratio_mod.DIAGNOSTIC_COLUMNS) <= set(diag.columns) + assert int(diag.loc[_TEST_DAY, "peak_bars"]) == len(_PEAKS) + assert int(diag.loc[_TEST_DAY, "ridge_bars"]) == len(_RIDGES) + assert bool(diag.loc[_TEST_DAY, "valid"]) + # Diagnostics cover EVERY day present, valid or not; the legs only the valid ones. + assert len(diag) >= len(legs) + + +def test_diagnostics_out_does_not_change_the_factor(): + plain = compute_peak_ridge_amount_ratio(_one_day_bars(), **_KW) + sink: list = [] + with_diag = compute_peak_ridge_amount_ratio( + _one_day_bars(), diagnostics_out=sink, **_KW + ) + pd.testing.assert_series_equal(plain, with_diag) + assert sink and "symbol" in sink[0].columns + + +# --------------------------------------------------------------------------- # +# Edges / purity +# --------------------------------------------------------------------------- # +def test_empty_bars_yield_empty_schema_series(): + out = compute_peak_ridge_amount_ratio(empty_intraday_bars(), name="f") + assert out.empty + assert list(out.index.names) == ["date", "symbol"] + assert out.name == "f" + + +def test_input_bars_not_mutated(): + bars = _one_day_bars() + before = bars.copy(deep=True) + compute_peak_ridge_amount_ratio(bars, **_KW) + pd.testing.assert_frame_equal(bars, before) + + +def test_baseline_insufficient_yields_no_value(): + rows = _session(_DAY_1, *_case_a_day()) + out = compute_peak_ridge_amount_ratio(_bars(rows), **_KW) + assert out.empty or not np.isfinite(out.to_numpy()).any() + + +@pytest.mark.parametrize( + "kwargs", + [ + dict(lookback_days=0), + dict(baseline_days=1), + dict(baseline_min_obs=1), + dict(sigma_k=-1.0), + dict(min_valid_days=0), + dict(min_classifiable=0), + dict(min_peak_bars=0), + dict(min_ridge_bars=0), + ], +) +def test_bad_params_raise(kwargs): + with pytest.raises(ValueError): + compute_peak_ridge_amount_ratio(_one_day_bars(), **kwargs) + + +# --------------------------------------------------------------------------- # +# The Factor class + spec +# --------------------------------------------------------------------------- # +def test_factor_spec_is_valid_and_daily(): + spec = PeakRidgeAmountRatioFactor().spec + assert isinstance(spec, FactorSpec) + assert spec.factor_id == "peak_ridge_amount_ratio_20" + assert spec.expected_ic_sign == 1 + assert spec.is_intraday is False + assert spec.return_basis == "close_to_close" + assert spec.forward_return_horizon == 1 + assert spec.family == "microstructure" + assert set(spec.input_fields) == {"volume", "amount"} + + +def test_factor_spec_records_the_pre_registered_sign_source_and_aggregation(): + """The spec must state the report's RankIC and the ratio-of-sums form.""" + text = PeakRidgeAmountRatioFactor().spec.description + assert "RATIO OF" in text and "SUMS" in text + assert "峰岭成交比" in text + assert str(PEAK_RIDGE_MIN_PEAK_BARS) in text + assert str(PEAK_RIDGE_MIN_RIDGE_BARS) in text + + +def test_factor_subclass_window_tracks_name(): + assert PeakRidgeAmountRatioFactor(10).name == "peak_ridge_amount_ratio_10" + assert PeakRidgeAmountRatioFactor(10).spec.factor_id == "peak_ridge_amount_ratio_10" + + +def test_factor_compute_selects_preaggregated_column(): + factor = PeakRidgeAmountRatioFactor() + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp("2024-01-02"), "000001.SZ")], names=["date", "symbol"] + ) + panel = pd.DataFrame({factor.name: [1.25], "close": [10.0]}, index=idx) + out = factor.compute(panel) + assert out.name == factor.name + assert out.iloc[0] == pytest.approx(1.25) + + +def test_factor_compute_missing_column_raises(): + factor = PeakRidgeAmountRatioFactor() + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp("2024-01-02"), "000001.SZ")], names=["date", "symbol"] + ) + with pytest.raises(ValueError, match="pre-aggregated"): + factor.compute(pd.DataFrame({"close": [10.0]}, index=idx)) + + +@pytest.mark.parametrize("bad", [0, -1, 2.5, "20", None]) +def test_factor_bad_params_raise(bad): + with pytest.raises(ValueError): + PeakRidgeAmountRatioFactor(bad)