From 2521ab92e19231b21ebb99f84769ad21ce22ca28 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 20 Jul 2026 02:57:18 -0700 Subject: [PATCH 1/2] refactor(data): expose the ridge mask on the shared peak taxonomy The peak/ridge/valley classification already computed `eruptive` internally but exposed only `classifiable` / `valley` / `peak`. Add `ridge` = eruptive & ~peak so the ridge-price family (PR-J) consumes THE SAME classification instead of re-deriving it and drifting apart from PR-F / PR-H / PR-I. Purely additive: the three pre-existing masks are computed exactly as before and no consumer reads columns positionally. The three masks now partition the classifiable bars exactly (valley | peak | ridge == classifiable, pairwise disjoint). The rule is PINNED as the exact complement of the conservative peak test, so it is wider than the literal 'eruptive next to an eruptive': it also covers session-boundary eruptions and eruptions whose neighbour is unclassifiable, i.e. every eruptive bar whose isolation is not provable. Verified bit-identical on real cached minute data: volume_peak_count_20, peak_interval_kurtosis_20 and valley_relative_vwap_20 over 30 CSI500 names x 2022-2024 (21142 symbol-days) give assert_frame_equal(check_exact=True) and the same SHA256 b3e68795... before and after. --- data/clean/intraday_volume_prv.py | 22 +++++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/data/clean/intraday_volume_prv.py b/data/clean/intraday_volume_prv.py index 5726a44..3f0756b 100644 --- a/data/clean/intraday_volume_prv.py +++ b/data/clean/intraday_volume_prv.py @@ -196,9 +196,12 @@ def peak_mask_for_symbol( Returns: ``g`` sorted by ``(trade_date, bar_end)`` on a fresh ``RangeIndex`` with the added boolean columns ``classifiable`` (a strictly-prior baseline existed), - ``valley`` (classifiable and NOT eruptive — the report's 量谷, what PR-I prices) - and ``peak``. ``valley`` and ``peak`` are disjoint: a peak is eruptive by - construction. Pure: never mutates ``g``. + ``valley`` (classifiable and NOT eruptive — the report's 量谷, what PR-I prices), + ``peak`` and ``ridge`` (eruptive but NOT an isolated peak — the report's 量岭, + what PR-J prices). The three masks PARTITION the classifiable bars exactly: + ``valley`` holds the mild ones and ``peak`` / ``ridge`` split the eruptive ones, + so ``valley | peak | ridge == classifiable`` and no two overlap. Pure: never + mutates ``g``. """ # day x slot volume matrix: rows are the symbol's trading days, columns are the # minute-of-day slots; a missing (day, slot) cell is NaN (no bar that minute). @@ -261,6 +264,19 @@ def peak_mask_for_symbol( & next_mild ) work["peak"] = peak + # RIDGE (量岭) == eruptive AND NOT an isolated peak. Exposed as a first-class boolean + # so the ridge-PRICE family (PR-J) consumes THE SAME classification instead of + # re-deriving it and drifting. Purely additive: ``classifiable`` / ``valley`` / + # ``peak`` above are computed exactly as before and no consumer of this frame reads + # columns positionally. + # + # PINNED, and wider than the literal "eruptive next to an eruptive": an eruptive bar + # is a ridge whenever its isolation is NOT PROVABLE, which also covers the + # session-boundary bars (a neighbour missing across the lunch break / the cutoff / + # a gap) and the bars whose neighbour is unclassifiable. That is the exact complement + # of PR-F's deliberately conservative peak rule, so the taxonomy stays a clean + # partition (valley | peak | ridge == classifiable) with no bar silently dropped. + work["ridge"] = eruptive & ~peak return work From 8de14fc04586253d0c3d063b18aa7267fc1ccf3b Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 20 Jul 2026 02:57:41 -0700 Subject: [PATCH 2/2] feat(factors): valley/ridge VWAP-ratio factor + PR-J evaluation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reproduces the FOURTH factor of the Kaiyuan microstructure series #27 (谷岭加权价格比, reportId 4957417 §7.1) as ValleyRidgeVwapRatioFactor and runs it through the frozen StandardFactorEvaluator on real cached CSI500 data. Direct robustness test of PR-I: same reused classification, same Sigma-amount/Sigma-volume VWAP identity, same eval cell — only the DENOMINATOR changes, from the whole visible day's VWAP to the RIDGE VWAP, so the two behavioural groups collide head-on. Definition (PINNED, disclosed on the spec): daily ratio = valley VWAP / ridge VWAP, averaged over the trailing 20 VALID days. A day is valid iff it has >=100 classifiable bars, >=20 tradable valley bars, >=10 tradable ridge bars and positive volume in both denominators. The ridge floor is deliberately LOWER because a ridge bar must erupt AND fail the isolation test; the runner MEASURES and logs the realized ridge-bar distribution, the day-validity rate and the counterfactual valid-day count at a 20-bar floor, so the scarcity is a reported number rather than a hidden threshold. Prototype sanity (tmp/design/pr_j_prototype_sanity.txt): mean daily cross-sectional RankIC +0.0438 over 692 days on 30 CSI500 names, 2022-2024 — POSITIVE, matching the pre-registered sign +1. Full run (995/996 symbols, stk_mins_live_calls=0, 438s): both axes PASS as in PR-I — no-book Watch (predictive PASS, IC 0.0339, ICIR 0.425, CI low +0.355, NW-t 14.11, N_eff 854, win 66.8%, monotonicity 0.90); with-book Watch (incremental PASS, incremental ICIR 0.289). Day validity 51.5% of classifiable symbol-days, materially below PR-I's ~99.8% — disclosed, not hidden. Net long-short spread is NEGATIVE at every cost scenario (gross +0.0004/period vs 2.10 turnover x 0.001 fee); reported as-is, the frozen evaluator's known aligned-spread cost-sign issue is untouched. --- config/phase_j_valley_ridge_vwap_ratio.yaml | 170 ++++ data/clean/intraday_valley_ridge_vwap.py | 380 +++++++ factors/compute/intraday_derived.py | 132 +++ qt/cli.py | 35 + qt/eval_valley_ridge_vwap_ratio.py | 684 +++++++++++++ ...est_eval_valley_ridge_vwap_ratio_runner.py | 365 +++++++ tests/test_valley_ridge_vwap_ratio_factor.py | 923 ++++++++++++++++++ 7 files changed, 2689 insertions(+) create mode 100644 config/phase_j_valley_ridge_vwap_ratio.yaml create mode 100644 data/clean/intraday_valley_ridge_vwap.py create mode 100644 qt/eval_valley_ridge_vwap_ratio.py create mode 100644 tests/test_eval_valley_ridge_vwap_ratio_runner.py create mode 100644 tests/test_valley_ridge_vwap_ratio_factor.py diff --git a/config/phase_j_valley_ridge_vwap_ratio.yaml b/config/phase_j_valley_ridge_vwap_ratio.yaml new file mode 100644 index 0000000..acd3e23 --- /dev/null +++ b/config/phase_j_valley_ridge_vwap_ratio.yaml @@ -0,0 +1,170 @@ +# PR-J — Eighth real factor evaluation: VALLEY/RIDGE VWAP RATIO. +# +# Reproduces the FOURTH factor of the Kaiyuan market-microstructure series #27 (开源证券 +# 《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, reportId 4957417, §7.1) — the +# "谷岭加权价格比因子" — as a first-class ValleyRidgeVwapRatioFactor and runs it through the +# FROZEN StandardFactorEvaluator on REAL cached A-share data. CACHE-ONLY: the minute read +# is provably live-call-free; daily / universe / covariate endpoints go through the +# read-through cache (warm -> 0 gap fetches). +# +# WHY THIS FACTOR, NOW: PR-I (valley VWAP / WHOLE-VISIBLE-DAY VWAP) is the strongest factor +# of this loop -- both assessed axes PASS. PR-J holds everything else fixed and swaps ONLY +# the denominator to the RIDGE VWAP, so the two behavioural groups of the report's +# peak/ridge/valley taxonomy collide head-on. If the price-level signal survives the swap +# it is a property of the FAMILY; if it does not, PR-I depended on that specific +# denominator. Either outcome is a legitimate result and is reported as such. +# +# SAME MACHINE as PR-F / PR-H / PR-I. The minute classification is REUSED from +# data/clean/intraday_volume_prv.py (not re-implemented): PIT-truncate each day at 14:50, +# classify every visible minute against its SAME-SLOT strictly-prior 20-day baseline +# (ERUPTIVE if vol > mu + sigma, else a VALLEY 量谷); a RIDGE 量岭 is an eruptive minute that +# is NOT an isolated peak. Because the classification is shared verbatim, a different +# verdict here cannot be an artefact of a differently-implemented taxonomy. +# +# PINNED choices (disclosed on the factor spec): +# (1) the RIDGE mask is 'eruptive AND NOT an isolated peak'. This is WIDER than the +# literal "eruptive next to an eruptive": it also covers session-boundary eruptions +# and eruptions whose neighbour is unclassifiable, i.e. every eruptive bar whose +# isolation is UNPROVABLE. It is the exact complement of PR-F's conservative peak +# rule, so valley|peak|ridge stays an exact partition of the classifiable bars and no +# eruptive bar is silently dropped. An isolated PEAK contributes to NEITHER leg; +# (2) each VWAP uses the aggregation identity sum(p*v)/sum(v) == sum(amount)/sum(volume); +# (3) bars with non-finite or non-positive volume OR amount are dropped from BOTH sums; +# the guard runs at the summation step only, so PR-F's same-slot baseline -- and +# therefore volume_peak_count / peak_interval_kurtosis / valley_relative_vwap -- is +# bit-identical; +# (4) RAW (unadjusted) prices are correct here: the adjustment factor is constant within +# a day and cancels exactly in the ratio (same reasoning as PR-I / PR-D / I5b); +# (5) DEVIATION FROM THE REPORT, disclosed and NOT silently equated: 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) ASYMMETRIC BAR FLOOR -- a day is VALID iff it has >= 100 classifiable bars AND +# >= 20 TRADABLE valley bars AND >= 10 TRADABLE ridge bars (both counted AFTER the +# positive-trade guard) AND positive volume in both denominators; NaN below 10 valid +# days. The ridge floor is deliberately LOWER than the valley floor because a ridge +# bar must erupt AND fail the isolation test, making ridges structurally far scarcer; +# holding both legs to 20 would discard sound days and bias the surviving sample +# towards unusually turbulent sessions. This threshold is a PINNED interpretation, +# not a tuned knob, and the runner REPORTS the realized ridge-bar distribution, the +# day-validity rate and the counterfactual valid-day count at a 20-ridge floor, so a +# coverage regression against PR-I shows up as a number rather than being hidden. +# +# Pre-registered sign = +1 (report full-market RankIC +6.98% / RankICIR 3.56, long-short +# 15.83%/yr, IR 1.83, monthly win rate 72.3%, only 2023 negative in 13 years; CSI500 +# sub-domain long-short 10.49% / IR 1.34, the closest comparable to this cell). Semantics +# per the report: a HIGH valley/ridge price ratio means retail over-reaction pushed the +# eruptive minutes' price DOWN relative to the calm ones, so the stock is depressed and +# performs better going forward. NOTE the report is a MONTHLY, market-cap + industry +# neutral FULL-MARKET series on Wind data -- our eval cell is CSI500 daily with industry + +# size neutral, so those numbers are a LOOSE reference only and are NOT written in as +# expected values. Raw minute volume (cached as-is) has split-day magnitude jumps that +# pollute the 20-day sigma; the report (Wind) does not adjust for this either, so it is +# disclosed and NOT corrected. +# +# Eval CELL is IDENTICAL to PR-C .. PR-I: 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 PR-I exactly, this run is directly comparable to the factor it is meant +# to stress. The evaluator runs TWICE (see qt/eval_valley_ridge_vwap_ratio.py): once with +# NO book (Incremental NOT_ASSESSED) and once with the confirmed book -- which matters more +# than usual here, since a relative PRICE LEVEL is a plausible cousin of a value signal and +# the with-book run is what says whether this is a genuinely new bet. + +project: + name: quantitative_trading_pr_j_valley_ridge_vwap_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: valley_ridge_vwap_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 (valley_ridge_vwap_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. + 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_valley_ridge_vwap.py b/data/clean/intraday_valley_ridge_vwap.py new file mode 100644 index 0000000..4234957 --- /dev/null +++ b/data/clean/intraday_valley_ridge_vwap.py @@ -0,0 +1,380 @@ +"""VALLEY/RIDGE VWAP-RATIO factor (PR-J). + +Reproduces the FOURTH factor of the Kaiyuan market-microstructure series #27 (开源证券 +《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, 2025-07-20, reportId 4957417, +§7.1): "计算每日量谷成交量加权价格，与量岭成交量加权价格做比，并计算 20 日价格比均值作为 +谷岭加权价格比因子". + +Same MACHINE as PR-F / PR-H / PR-I and the same VWAP identity as PR-I — what changes is +the DENOMINATOR. PR-I divided the valley VWAP by the WHOLE VISIBLE DAY's VWAP; this +factor divides it by the RIDGE VWAP, so the two behavioural groups of the report's +peak/ridge/valley taxonomy are contrasted head-on rather than one being compared against +an aggregate that contains it. That makes PR-J the natural robustness test of PR-I: if +the price-level signal survives the swap, it is a property of the FAMILY; if it does not, +PR-I depended on its specific denominator. Either outcome is a legitimate result. + +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, same valid-day floor. Nothing +about the taxonomy is re-implemented here, so the four factors can never drift apart. + +PINNED choices (deliberate and DISCLOSED, not tuned knobs; reproduced on the factor spec +so a reader sees exactly what was assumed): + + 1. THE RIDGE MASK IS ``eruptive & ~peak``. A RIDGE (量岭) is an eruptive minute that is + NOT an isolated peak. This is WIDER than the literal "eruptive next to an eruptive": + it also covers the session-boundary eruptions (a neighbour missing across the lunch + break / the cutoff / a data gap) and the eruptions whose neighbour is unclassifiable. + The rule is the exact COMPLEMENT of PR-F's deliberately conservative peak test, so + the taxonomy stays a clean partition — ``valley | peak | ridge == classifiable`` — + and no eruptive bar is silently dropped on either side. Disclosed rather than + silently equated to the report's wording. + 2. VWAP VIA THE AGGREGATION IDENTITY (same as PR-I). A bar's volume-weighted price is + ``p = amount / volume``, so a set's volume-weighted price ``Σ(p_i·v_i) / Σv_i`` + collapses EXACTLY to ``Σamount / Σvolume``. Both legs use that identity: valley VWAP + = Σamount(valley bars)/Σvolume(valley bars), ridge VWAP = Σamount(ridge + bars)/Σvolume(ridge bars). + 3. POSITIVE-TRADE GUARD. A bar with non-finite or non-positive ``volume`` OR ``amount`` + carries no price information, so it is dropped from BOTH sums. 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. A zero-volume minute is therefore + still classified (and still a valley) but contributes nothing to either VWAP. + 4. RAW (UNADJUSTED) PRICES. The cached minute bars are unadjusted. Both legs are sums + over the SAME trading day, and a split/dividend adjustment factor is constant within + a day, so it cancels exactly in the ratio — no adjustment is needed (the same + reasoning PR-I, PR-D's amplitude and I5b's price-limit checks rely on). + 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_valley_bars`` (=20) but ``min_ridge_bars`` (=10). + Ridge bars are STRUCTURALLY far scarcer than valley bars: a minute must erupt (a + minority event by construction, since the threshold is its own strictly-prior + same-slot μ+σ) AND fail the isolation test. Holding the ridge leg to the valley + leg's floor would throw away a large share of otherwise sound days and bias the + surviving sample towards unusually turbulent sessions. The floor is therefore + LOWERED, deliberately, and both the realized ridge-bar distribution and the + resulting day-validity rate are REPORTED (``with_diagnostics=True``) rather than + left implicit — if coverage comes out materially worse than PR-I's, that is a + finding to surface, not to hide. + 7. BOTH BAR COUNTS ARE TAKEN AFTER THE GUARD, so a day cannot qualify on the strength + of bars that traded nothing. + +Factor value: ``valley_ridge_vwap_ratio_20`` = the MEAN of the daily ratio +``valley VWAP / ridge VWAP`` 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_valley_bars`` (=20) tradable valley bars, at least ``min_ridge_bars`` (=10) +tradable ridge bars, and strictly positive volume in BOTH denominators. Fewer than +``min_valid_days`` (=10) valid days in the trailing window -> NaN (honest missing; the +runner discloses the coverage). Values are emitted only on valid days. + +Pre-registered sign = +1. The report's full-market RankIC is +6.98% / RankICIR 3.56 +(long-short 15.83%/yr, IR 1.83, monthly win rate 72.3%, only 2023 negative in 13 years); +its CSI500 sub-domain long-short is 10.49% / IR 1.34, the closest comparable to our eval +cell. Semantics per the report: a HIGH valley/ridge price ratio means retail +over-reaction pushed the eruptive minutes' price DOWN relative to the calm ones, so the +stock is depressed and performs better going forward. NOTE the report is a MONTHLY, +market-cap + industry neutral full-market series on Wind data while our eval cell is +CSI500 daily with industry + size neutralization, so its numbers are a LOOSE reference +only (disclosed, never mislabeled, never written in as an expected value). + +The value at ``d`` uses only bars at dates <= d, so a factor value never sees a future +bar (invariant #1); it is a DAILY signal traded close-to-close from d+1. This module is +DATA-layer only: it does not fetch, does not touch factors / alpha / portfolio / +runtime, and never sees a token. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + +from data.clean.intraday_aggregate import DAILY_INDEX_NAMES, DEFAULT_DECISION_TIME +from data.clean.intraday_schema import SYMBOL_LEVEL, validate_intraday_bars +from data.clean.intraday_volume_prv import ( + VOLUME_PRV_BASELINE_DAYS, + VOLUME_PRV_BASELINE_MIN_OBS, + VOLUME_PRV_MIN_CLASSIFIABLE, + VOLUME_PRV_MIN_VALID_DAYS, + VOLUME_PRV_SIGMA_K, + peak_mask_for_symbol, + prepare_visible_minute_bars, +) + +# Factor DEFINITION constants (pinned interpretations of the report; NOT tuned knobs). +# The classification constants are IMPORTED from PR-F, never redefined. +VALLEY_RIDGE_LOOKBACK_DAYS = 20 # trailing VALID trading-day mean window, includes d +VALLEY_RIDGE_MIN_VALLEY_BARS = 20 # min TRADABLE valley bars for a valid day +# PINNED LOWER than the valley floor (module docstring §6): ridge bars are structurally +# far scarcer, so holding both legs to 20 would bias the surviving sample. +VALLEY_RIDGE_MIN_RIDGE_BARS = 10 # min TRADABLE ridge bars for a valid day + +# The extra 1min column this family needs on top of PR-F's (volume): the traded value, +# which turns the bar set into a volume-weighted price via Σamount/Σvolume. +_AMOUNT = "amount" + +# Per-day diagnostic columns (the ridge-scarcity disclosure the task card requires). +DIAGNOSTIC_COLUMNS = ("classifiable_bars", "valley_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) + + +def valley_ridge_vwap_ratio_by_day( + work: pd.DataFrame, + *, + min_valley_bars: int = VALLEY_RIDGE_MIN_VALLEY_BARS, + min_ridge_bars: int = VALLEY_RIDGE_MIN_RIDGE_BARS, + min_classifiable: int = VOLUME_PRV_MIN_CLASSIFIABLE, + with_diagnostics: bool = False, +) -> pd.Series | tuple[pd.Series, pd.DataFrame]: + """Daily ``valley VWAP / ridge VWAP`` ratio 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 ``valley`` / ``ridge`` / ``classifiable`` masks. + + Both VWAPs use the ``Σamount / Σvolume`` aggregation identity (PINNED §2 of the + module docstring). The POSITIVE-TRADE GUARD (§3) drops non-finite / non-positive + volume or amount bars from both sums. Unlike PR-I, the denominator is NOT the whole + visible day but the RIDGE bars alone (§1: ``eruptive & ~peak``), so an isolated PEAK + contributes to NEITHER leg. + + Args: + work: one symbol's classified minute frame (see above). + min_valley_bars: minimum TRADABLE valley bars for a valid day. + min_ridge_bars: minimum TRADABLE ridge bars for a valid day (PINNED lower than + the valley floor — see §6). + min_classifiable: minimum classifiable bars for a valid day (PR-F's gate). + with_diagnostics: also return the per-day bar-count frame, so the caller can + REPORT the ridge-scarcity distribution instead of only gating on it. + + Returns: + Series indexed by ``trade_date`` (ascending) holding the ratio 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 use). With + ``with_diagnostics=True``, a ``(ratio, diagnostics)`` pair where ``diagnostics`` + is indexed by EVERY day present in ``work`` and carries + :data:`DIAGNOSTIC_COLUMNS`. + """ + vol = work["volume"].to_numpy(dtype=float) + amt = work[_AMOUNT].to_numpy(dtype=float) + # Positive-trade guard: no price information in a bar that traded nothing (or whose + # fields are non-finite / negative). Applied HERE, at the summation step, never + # before classification — PR-F's same-slot baseline must stay bit-identical. + tradable = np.isfinite(vol) & (vol > 0.0) & np.isfinite(amt) & (amt > 0.0) + valley = work["valley"].to_numpy(dtype=bool) & tradable + ridge = work["ridge"].to_numpy(dtype=bool) & tradable + + per_bar = pd.DataFrame( + { + "trade_date": work["trade_date"].to_numpy(), + "valley_amt": np.where(valley, amt, 0.0), + "valley_vol": np.where(valley, vol, 0.0), + "ridge_amt": np.where(ridge, amt, 0.0), + "ridge_vol": np.where(ridge, vol, 0.0), + "valley_bars": valley.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 / task card §1.4): PR-F's classifiable + # floor (unchanged), enough TRADABLE valley bars, enough TRADABLE ridge bars, and a + # positive denominator on both legs (a 0/0 ratio is never fabricated). + valid = ( + (agg["classifiable_bars"] >= min_classifiable) + & (agg["valley_bars"] >= min_valley_bars) + & (agg["ridge_bars"] >= min_ridge_bars) + & (agg["valley_vol"] > 0.0) + & (agg["ridge_vol"] > 0.0) + ) + ok = agg.loc[valid] + if ok.empty: + ratio = pd.Series([], index=pd.DatetimeIndex([], name="trade_date"), dtype=float) + else: + valley_vwap = ok["valley_amt"] / ok["valley_vol"] + ridge_vwap = ok["ridge_amt"] / ok["ridge_vol"] + ratio = (valley_vwap / ridge_vwap).astype(float) + + if not with_diagnostics: + return ratio + diagnostics = agg[["classifiable_bars", "valley_bars", "ridge_bars"]].copy() + diagnostics["valid"] = valid + return ratio, diagnostics + + +def _valley_ridge_ratio_mean_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_valley_bars: int, + min_ridge_bars: int, + collect_diagnostics: bool = False, +) -> tuple[list[pd.Timestamp], list[float], pd.DataFrame | None]: + """Daily valley/ridge VWAP-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 VWAP ratio, then takes the trailing-``lookback_days``-valid-day + mean. No cross-symbol leakage (``g`` is one symbol's slice) and no lookahead (the + baseline is strictly prior, the mean window is trailing). + """ + work = peak_mask_for_symbol( + g, + baseline_days=baseline_days, + baseline_min_obs=baseline_min_obs, + sigma_k=sigma_k, + ) + result = valley_ridge_vwap_ratio_by_day( + work, + min_valley_bars=min_valley_bars, + min_ridge_bars=min_ridge_bars, + min_classifiable=min_classifiable, + with_diagnostics=collect_diagnostics, + ) + if collect_diagnostics: + ratio, diagnostics = result + else: + ratio, diagnostics = result, None + if ratio.empty: + return [], [], diagnostics + + # Mean over the trailing lookback_days VALID days (including d); NaN until + # min_valid_days valid days have accumulated. Emitted only on valid days. + rolled = ratio.sort_index().rolling(lookback_days, min_periods=min_valid_days).mean() + days = [pd.Timestamp(d).normalize() for d in rolled.index] + return days, list(rolled.to_numpy(dtype=float)), diagnostics + + +def compute_valley_ridge_vwap_ratio( + bars: pd.DataFrame, + *, + lookback_days: int = VALLEY_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_valley_bars: int = VALLEY_RIDGE_MIN_VALLEY_BARS, + min_ridge_bars: int = VALLEY_RIDGE_MIN_RIDGE_BARS, + decision_time: str = DEFAULT_DECISION_TIME, + name: str = "valley_ridge_vwap_ratio", + diagnostics_out: list | None = None, +) -> pd.Series: + """PIT-safe daily "valley/ridge VWAP 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, computes each valid + day's ``valley VWAP / ridge VWAP`` ratio via the ``Σamount/Σvolume`` identity, and + returns the trailing-``lookback_days``-VALID-day mean of that ratio. See the module + docstring for the LOCKED definition 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 averaged (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_valley_bars: a day needs at least this many TRADABLE valley bars. + min_ridge_bars: a day needs at least this many TRADABLE ridge bars (PINNED lower + than the valley floor — see the module docstring §6). + decision_time: per-bar PIT cutoff time-of-day (default 14:50:00). + name: the returned Series name (the factor-panel column name). + diagnostics_out: optional list the per-symbol day-level bar-count frames are + APPENDED to (each carries a ``symbol`` column), so a caller can report the + ridge-scarcity distribution. Purely observational — supplying it does not + change the returned factor. + + Returns: + ``MultiIndex(date, symbol)`` Series (midnight-normalized dates) of the daily + factor value, sorted, named ``name``. Pure: never mutates ``bars``. + """ + validate_intraday_bars(bars) + if lookback_days < 1: + raise ValueError(f"lookback_days must be >= 1; got {lookback_days!r}.") + if baseline_days < 2: + # Need >= 2 baseline observations so a ddof=1 std of the same-slot volume is + # defined. + raise ValueError(f"baseline_days must be >= 2; got {baseline_days!r}.") + if baseline_min_obs < 2: + raise ValueError(f"baseline_min_obs must be >= 2; got {baseline_min_obs!r}.") + if sigma_k < 0.0: + raise ValueError(f"sigma_k must be >= 0; got {sigma_k!r}.") + if min_valid_days < 1: + raise ValueError(f"min_valid_days must be >= 1; got {min_valid_days!r}.") + if min_classifiable < 1: + raise ValueError(f"min_classifiable must be >= 1; got {min_classifiable!r}.") + if min_valley_bars < 1: + raise ValueError(f"min_valley_bars must be >= 1; got {min_valley_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 visible.groupby(SYMBOL_LEVEL, sort=True): + days, vals, diagnostics = _valley_ridge_ratio_mean_for_symbol( + g.reset_index(drop=True), + baseline_days=baseline_days, + baseline_min_obs=baseline_min_obs, + sigma_k=sigma_k, + lookback_days=lookback_days, + min_valid_days=min_valid_days, + min_classifiable=min_classifiable, + min_valley_bars=min_valley_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] = str(sym) + diagnostics_out.append(frame) + for day, val in zip(days, vals): + index_tuples.append((day, str(sym))) + values.append(val) + + if not index_tuples: + return _empty_series(name) + index = pd.MultiIndex.from_tuples(index_tuples, names=DAILY_INDEX_NAMES) + return pd.Series(values, index=index, name=name).sort_index() + + +__all__ = [ + "DIAGNOSTIC_COLUMNS", + "VALLEY_RIDGE_LOOKBACK_DAYS", + "VALLEY_RIDGE_MIN_RIDGE_BARS", + "VALLEY_RIDGE_MIN_VALLEY_BARS", + "compute_valley_ridge_vwap_ratio", + "valley_ridge_vwap_ratio_by_day", +] diff --git a/factors/compute/intraday_derived.py b/factors/compute/intraday_derived.py index 160c797..4e3744a 100644 --- a/factors/compute/intraday_derived.py +++ b/factors/compute/intraday_derived.py @@ -58,6 +58,11 @@ PEAK_INTERVAL_LOOKBACK_DAYS, PEAK_INTERVAL_MIN_INTERVALS, ) +from data.clean.intraday_valley_ridge_vwap import ( + VALLEY_RIDGE_LOOKBACK_DAYS, + VALLEY_RIDGE_MIN_RIDGE_BARS, + VALLEY_RIDGE_MIN_VALLEY_BARS, +) from data.clean.intraday_valley_vwap import ( VALLEY_VWAP_LOOKBACK_DAYS, VALLEY_VWAP_MIN_VALLEY_BARS, @@ -755,6 +760,132 @@ def compute(self, panel: pd.DataFrame) -> pd.Series: return panel[self.name].rename(self.name) +class ValleyRidgeVwapRatioFactor(Factor): + """Valley/ridge VWAP-ratio factor (daily signal, minute-derived). + + ``compute`` reads the pre-aggregated daily column the runner placed on the panel + (produced by + ``data.clean.intraday_valley_ridge_vwap.compute_valley_ridge_vwap_ratio``); it does + NO minute work of its own, mirroring :class:`JumpAmountCorrFactor` / + :class:`MinuteIdealAmplitudeFactor` / :class:`AmpMarginalAnomalyVolFactor` / + :class:`VolumePeakCountFactor` / :class:`IntradayAmpCutFactor` / + :class:`PeakIntervalKurtosisFactor` / :class:`ValleyRelativeVwapFactor` and the value + / financial factors that surface an enriched column. + + Args: + lookback_days: trailing VALID trading-day window averaged; 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"valley_ridge_vwap_ratio_{VALLEY_RIDGE_LOOKBACK_DAYS}" + + def __init__(self, lookback_days: int = VALLEY_RIDGE_LOOKBACK_DAYS) -> None: + if not isinstance(lookback_days, int) or lookback_days < 1: + raise ValueError( + f"valley-ridge-vwap-ratio lookback_days must be a positive integer; got " + f"{lookback_days!r}." + ) + self._lookback_days = lookback_days + self.name = f"valley_ridge_vwap_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: the report's full-market RankIC is +6.98% (RankICIR 3.56, + long-short 15.83%/yr, IR 1.83, monthly win rate 72.3%, only 2023 negative across + 13 years), and its CSI500 sub-domain long-short is 10.49% / IR 1.34, the closest + comparable to our eval cell. Semantics per the report: a HIGH valley/ridge price + ratio means retail over-reaction pushed the eruptive minutes' price DOWN relative + to the calm ones, so the stock is depressed and performs better going forward. + The sign is fixed BEFORE the run (a validated prototype must reproduce it). NOTE + the report is a MONTHLY, market-cap + industry neutral full-market series on Wind + data while our eval cell is CSI500 daily with industry + size neutral, so the + report numbers are a LOOSE reference only (disclosed, never mislabeled, never + written in as an expected value). is_intraday=False by the module docstring's + reasoning: minute INPUT but a DAILY signal traded close-to-close. + min_history_bars=0: the warm-up is DATA-dependent (a value appears once enough + VALID days accumulate), not a fixed leading count — the honest NaN rate is + reported by data_coverage. + + The description spells out the RELATION TO PR-I (same reused classification, same + VWAP identity, DENOMINATOR swapped from the whole visible day to the ridge bars) + plus the pinned choices — above all the ASYMMETRIC bar floor, which exists + because ridge bars are structurally far scarcer than valley bars. + """ + return FactorSpec( + factor_id=self.name, + version="1.0", + description=( + f"Valley/ridge VWAP ratio (Kaiyuan microstructure series #27, FOURTH " + f"factor 谷岭加权价格比). SAME minute classification as PR-F " + f"volume_peak_count / PR-H peak_interval_kurtosis / PR-I " + f"valley_relative_vwap (REUSED from data.clean.intraday_volume_prv, not " + f"re-implemented): 1min bars PIT-truncated at 14:50, a minute is ERUPTIVE " + f"if vol > μ + {VOLUME_PRV_SIGMA_K:g}σ of its SAME-SLOT strictly-prior " + f"{VOLUME_PRV_BASELINE_DAYS}-day baseline, else it is a VALLEY (量谷). " + f"DENOMINATOR SWAPPED vs PR-I: instead of the whole visible day's VWAP " + f"the divisor is the RIDGE (量岭) VWAP, so the two behavioural groups are " + f"contrasted head-on — daily ratio = (valley VWAP) / (ridge VWAP), " + f"averaged over the trailing {self._lookback_days} VALID days. PINNED " + f"choices: (1) the RIDGE mask is 'eruptive AND NOT an isolated peak', " + f"which is WIDER than 'eruptive next to an eruptive' — it also covers " + f"session-boundary eruptions and eruptions with an unclassifiable " + f"neighbour, keeping valley|peak|ridge an exact partition of the " + f"classifiable bars; an isolated PEAK contributes to NEITHER leg; " + f"(2) each VWAP uses the aggregation identity Σ(p·v)/Σv = Σamount/Σvolume; " + f"(3) bars with non-finite or non-positive volume or amount are dropped " + f"from BOTH sums (guard applied at summation only, so PR-F's baseline is " + f"untouched); (4) RAW unadjusted prices are correct here because the " + f"adjustment factor is constant within a day and cancels in the ratio; " + f"(5) DEVIATION FROM THE REPORT, disclosed: both legs span the " + f"PIT-VISIBLE window 09:31-14:50 only, not the full session — reading the " + f"close would be lookahead at our 14:50 decision time; (6) a day is VALID " + f"iff it has >= {VOLUME_PRV_MIN_CLASSIFIABLE} classifiable bars AND >= " + f"{VALLEY_RIDGE_MIN_VALLEY_BARS} TRADABLE valley bars AND >= " + f"{VALLEY_RIDGE_MIN_RIDGE_BARS} TRADABLE ridge bars (both counted AFTER " + f"the guard) AND positive volume in both denominators — the ridge floor " + f"is deliberately LOWER because a ridge bar must erupt AND fail the " + f"isolation test, making ridges structurally far scarcer than valleys, " + f"and the realized ridge-bar distribution plus day-validity rate are " + f"REPORTED by the runner rather than left implicit; NaN below " + f"{VOLUME_PRV_MIN_VALID_DAYS} valid days. Derived from 1min bars but a " + f"DAILY signal traded close-to-close." + ), + expected_ic_sign=1, + is_intraday=False, + forward_return_horizon=1, + return_basis="close_to_close", + # The 1min bar fields the upstream aggregation is derived from. Declared for + # honest provenance disclosure (data_coverage lists them); the daily panel + # surfaces the pre-aggregated column itself. + input_fields=("volume", "amount"), + family="microstructure", + min_history_bars=0, + ) + + def compute(self, panel: pd.DataFrame) -> pd.Series: + """Select the pre-aggregated daily valley/ridge VWAP-ratio column off ``panel``. + + The runner runs ``compute_valley_ridge_vwap_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"ValleyRidgeVwapRatioFactor needs the pre-aggregated '{self.name}' " + f"column on the panel (produced upstream by " + f"compute_valley_ridge_vwap_ratio and joined by the runner); panel has " + f"{list(panel.columns)}." + ) + return panel[self.name].rename(self.name) + + __all__ = [ "AmpMarginalAnomalyVolFactor", "IntradayAmpCutFactor", @@ -762,5 +893,6 @@ def compute(self, panel: pd.DataFrame) -> pd.Series: "MinuteIdealAmplitudeFactor", "PeakIntervalKurtosisFactor", "ValleyRelativeVwapFactor", + "ValleyRidgeVwapRatioFactor", "VolumePeakCountFactor", ] diff --git a/qt/cli.py b/qt/cli.py index d6ae5dd..8ab71ac 100644 --- a/qt/cli.py +++ b/qt/cli.py @@ -378,6 +378,34 @@ def _cmd_run_eval_valley_relative_vwap(args: argparse.Namespace) -> int: return 0 +def _cmd_run_eval_valley_ridge_vwap_ratio(args: argparse.Namespace) -> int: + """Run the two real valley/ridge VWAP-ratio evaluations (cache-only) + reports.""" + from qt.eval_valley_ridge_vwap_ratio import run_eval_valley_ridge_vwap_ratio + + try: + result = run_eval_valley_ridge_vwap_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 + print( + f"OK run-eval-valley-ridge-vwap-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" + # Ridge bars are structurally scarce; the realized distribution and the + # day-validity rate are surfaced so a coverage regression is visible. + f"{result.ridge_coverage.render()}\n" + f"no-book: {nb['deployment']} (predictive={nb['predictive']}) " + f"ic_mean={nb['ic_mean']:.4f} ic_ir={nb['ic_ir']:.3f} N_eff={nb['effective_samples']:.1f}\n" + f"with-book: {wb['deployment']} (incremental={wb['incremental']}) " + f"incr_ic_ir={wb['incremental_ic_ir']:.3f}\n" + f"reports: {result.reports.no_book_md} | {result.reports.with_book_md}\n" + f"dashboards: {result.reports.no_book_dashboard} | " + f"{result.reports.with_book_dashboard}" + ) + return 0 + + def _cmd_data_update(args: argparse.Namespace) -> int: """Warm/update the tushare caches (P4-3); never runs a backtest.""" from qt.data_updater import format_summary, run_data_update @@ -558,6 +586,13 @@ def build_parser() -> argparse.ArgumentParser: p_vrv.add_argument("--config", required=True, help="Path to the YAML config.") p_vrv.set_defaults(func=_cmd_run_eval_valley_relative_vwap) + p_vrr = sub.add_parser( + "run-eval-valley-ridge-vwap-ratio", + help="Run the valley/ridge VWAP-ratio factor evaluation (CSI500, cache-only).", + ) + p_vrr.add_argument("--config", required=True, help="Path to the YAML config.") + p_vrr.set_defaults(func=_cmd_run_eval_valley_ridge_vwap_ratio) + 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_valley_ridge_vwap_ratio.py b/qt/eval_valley_ridge_vwap_ratio.py new file mode 100644 index 0000000..c225595 --- /dev/null +++ b/qt/eval_valley_ridge_vwap_ratio.py @@ -0,0 +1,684 @@ +"""run-eval-valley-ridge-vwap-ratio: the eighth real factor evaluation (PR-J). + +Reproduces the FOURTH factor of the Kaiyuan market-microstructure series #27 (开源证券 +《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, reportId 4957417, §7.1) — the +"谷岭加权价格比" factor — as a first-class +:class:`~factors.compute.intraday_derived.ValleyRidgeVwapRatioFactor` 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-I used. + +This is the FOURTH reproduction from the same report and the direct ROBUSTNESS TEST of +PR-I. PR-I (valley VWAP / whole-visible-day VWAP) is the strongest factor of this loop — +both assessed axes PASS. PR-J keeps everything else fixed and swaps only the DENOMINATOR +to the RIDGE VWAP, contrasting the two behavioural groups head-on. If the price-level +signal survives the swap it is a property of the FAMILY; if it does not, PR-I depended on +its specific denominator. 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_valley_ridge_vwap.compute_valley_ridge_vwap_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 VALLEY +is a classifiable non-eruptive minute, a RIDGE is an eruptive minute that is not an +isolated peak), take each valid day's ratio of the valley VWAP to the ridge VWAP (both +via the Σamount/Σvolume identity), and average that ratio over the trailing 20 VALID +trading days. It is executed CLOSE-TO-CLOSE (daily default), so ``is_intraday=False``. +The eval CELL is identical to PR-C..PR-I, so this run is directly comparable to its +siblings — same classification, same VWAP identity, different denominator. + +RIDGE SCARCITY IS MEASURED, NOT ASSUMED. Ridge bars are structurally far rarer than +valley bars (a minute must erupt AND fail the isolation test), which is why the valid-day +gate asks for >= 10 tradable ridge bars against the valley leg's >= 20. The runner +therefore collects the REALIZED per-day ridge-bar distribution and the day-validity rate +across the whole universe and logs them, so a coverage regression against PR-I is visible +as a number rather than hidden behind a lower threshold. + +CACHE-ONLY: every input is read from the persistent tushare cache +(``artifacts/cache/tushare/v1``). The minute read is provably live-call-free (the +minute store has no fetch closure — a miss simply yields no rows); the daily / +universe / covariate endpoints go through the shared read-through cache, which on a +fully-warmed cache does zero gap fetches (disclosed via the run-log cache-stats line). +Forward returns are computed ONLY at the evaluator/analytics boundary from +``ctx.price_panel`` — the factor computation never sees a future return. + +The evaluator is run TWICE: once with NO known-factor book (the Incremental axis is +NOT_ASSESSED) and once with the project's independently-confirmed book (value_ep / +value_bp / volatility_20) so the Incremental axis measures whether the ratio adds alpha +BEYOND value / low-vol. As with PR-I that check matters more than usual: a relative PRICE +LEVEL is a plausible cousin of a value signal, so the with-book run is the one that says +whether this is a new bet. +""" + +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_schema import RAW_INTRADAY_FREQ, normalize_intraday_bars +from data.clean.intraday_valley_ridge_vwap import ( + VALLEY_RIDGE_LOOKBACK_DAYS, + VALLEY_RIDGE_MIN_RIDGE_BARS, + VALLEY_RIDGE_MIN_VALLEY_BARS, + compute_valley_ridge_vwap_ratio, +) +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 ValleyRidgeVwapRatioFactor +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_valley_ridge_vwap_ratio" +_REPORT_STEM = "eval_valley_ridge_vwap_ratio" + +# Percentiles reported for the realized ridge-bar distribution (the scarcity disclosure). +_RIDGE_PCTL = (0, 10, 25, 50, 75, 90, 100) + + +# --------------------------------------------------------------------------- # +# Ridge-scarcity coverage (measured, never assumed) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class RidgeCoverage: + """Realized ridge-bar distribution + day-validity rate over the whole universe. + + Built from the per-day diagnostics the factor emits, so the numbers describe the + days the factor actually saw. ``symbol_days`` counts EVERY symbol-day with visible + bars, including the leading warm-up days that have no same-slot baseline yet; + ``classifiable_days`` counts those that clear PR-F's classifiable floor. The + headline ``validity_rate`` is taken over ``classifiable_days``, because a day with + no baseline fails for a PR-F warm-up reason rather than a ridge-scarcity one and + would otherwise make the ridge gate look worse than it is — both denominators are + reported so the reader can check that framing. 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 + ridge_percentiles: tuple[tuple[int, float], ...] + ridge_mean: float + valley_median: float + days_below_ridge_gate: int + days_below_valley_gate: int + days_below_classifiable_gate: int + # Counterfactual: how many days would survive if the ridge leg were held to the + # VALLEY floor. Quantifies exactly what the lowered threshold buys. + valid_days_at_valley_floor: int + # The gates this run actually applied, so the disclosure can never describe the + # module defaults while the run used something else. + min_ridge_bars: int = VALLEY_RIDGE_MIN_RIDGE_BARS + min_valley_bars: int = VALLEY_RIDGE_MIN_VALLEY_BARS + min_classifiable: int = VOLUME_PRV_MIN_CLASSIFIABLE + + @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.ridge_percentiles) + return ( + f"ridge scarcity: symbol_days={self.symbol_days} " + f"classifiable_days={self.classifiable_days} " + f"valid_days={self.valid_days} ({self.validity_rate:.1%} of classifiable) " + f"ridge_bars[{pctl} mean={self.ridge_mean:.1f}] " + f"valley_bars_median={self.valley_median:.0f} " + f"below_ridge_gate({self.min_ridge_bars})=" + f"{self.days_below_ridge_gate} " + f"below_valley_gate({self.min_valley_bars})=" + f"{self.days_below_valley_gate} " + f"below_classifiable_gate({self.min_classifiable})=" + f"{self.days_below_classifiable_gate} " + f"valid_if_ridge_floor_were_{self.min_valley_bars}=" + f"{self.valid_days_at_valley_floor}" + ) + + +def summarize_ridge_coverage( + frames: list[pd.DataFrame], + *, + min_ridge_bars: int = VALLEY_RIDGE_MIN_RIDGE_BARS, + min_valley_bars: int = VALLEY_RIDGE_MIN_VALLEY_BARS, + min_classifiable: int = VOLUME_PRV_MIN_CLASSIFIABLE, +) -> RidgeCoverage: + """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_ridge_bars=min_ridge_bars, + min_valley_bars=min_valley_bars, + min_classifiable=min_classifiable, + ) + empty = tuple((p, float("nan")) for p in _RIDGE_PCTL) + if not frames: + return RidgeCoverage( + symbol_days=0, + classifiable_days=0, + valid_days=0, + ridge_percentiles=empty, + ridge_mean=float("nan"), + valley_median=float("nan"), + days_below_ridge_gate=0, + days_below_valley_gate=0, + days_below_classifiable_gate=0, + valid_days_at_valley_floor=0, + **gates, + ) + diag = pd.concat(frames, ignore_index=True) + classifiable = diag["classifiable_bars"].to_numpy(dtype=float) + valid = diag["valid"].to_numpy(dtype=bool) + # The bar-count distributions describe the days that had a fair chance: a warm-up day + # with no same-slot baseline has zero of everything and would only drag the + # percentiles towards zero for a reason that has nothing to do with ridge scarcity. + scored = classifiable >= min_classifiable + ridge = diag.loc[scored, "ridge_bars"].to_numpy(dtype=float) + valley = diag.loc[scored, "valley_bars"].to_numpy(dtype=float) + # The counterfactual holds the ridge leg to the valley floor, leaving every other + # gate exactly as it was. + at_valley_floor = valid & ( + diag["ridge_bars"].to_numpy(dtype=float) >= min_valley_bars + ) + return RidgeCoverage( + symbol_days=int(len(diag)), + classifiable_days=int(scored.sum()), + valid_days=int(valid.sum()), + ridge_percentiles=( + tuple((p, float(np.percentile(ridge, p))) for p in _RIDGE_PCTL) + if ridge.size + else empty + ), + ridge_mean=float(ridge.mean()) if ridge.size else float("nan"), + valley_median=float(np.median(valley)) if valley.size else float("nan"), + days_below_ridge_gate=int((ridge < min_ridge_bars).sum()), + days_below_valley_gate=int((valley < min_valley_bars).sum()), + days_below_classifiable_gate=int((~scored).sum()), + valid_days_at_valley_floor=int(at_valley_floor.sum()), + **gates, + ) + + +# --------------------------------------------------------------------------- # +# Minute loading (cache-only, per-symbol -> memory-bounded) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _ValleyRidgeMinuteLoad: + """Diagnostics from the cache-only per-symbol minute read + aggregation.""" + + factor: pd.Series # MultiIndex(date, symbol) raw valley/ridge VWAP 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) + ridge_coverage: RidgeCoverage + + +def _load_valley_ridge_vwap_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_valley_bars: int, + min_ridge_bars: int, +) -> _ValleyRidgeMinuteLoad: + """Compute the raw valley/ridge VWAP-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 VWAP-ratio reduction run entirely inside + ``compute_valley_ridge_vwap_ratio`` on 1min bars, which read ``volume`` and + ``amount``. The per-day bar-count diagnostics are collected alongside so the ridge + scarcity can be REPORTED, not assumed; collecting them does not change the factor. + """ + root = cfg.data.cache.root_dir + store = IntradayParquetStore(root) + start = pd.Timestamp(cfg.data.start).normalize() + end = pd.Timestamp(cfg.data.end).normalize() + pd.Timedelta("23:59:59") + + series: list[pd.Series] = [] + covered: list[str] = [] + empty: list[str] = [] + diagnostics: list[pd.DataFrame] = [] + raw_rows = 0 + for i, sym in enumerate(symbols): + part = store.read_range(INTRADAY_ENDPOINT, sym, RAW_INTRADAY_FREQ, start, end) + if part.empty: + empty.append(sym) + continue + raw_rows += len(part) + bars = normalize_intraday_bars( + part.rename(columns={"bar_end": "time"})[READ_COLUMNS], + freq=RAW_INTRADAY_FREQ, + ) + s = compute_valley_ridge_vwap_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_valley_bars=min_valley_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_ridge_coverage( + diagnostics, + min_ridge_bars=min_ridge_bars, + min_valley_bars=min_valley_bars, + min_classifiable=min_classifiable, + ) + if not series: + raise ValueError( + "run-eval-valley-ridge-vwap-ratio blocked: no requested symbol produced a " + f"cached valley/ridge VWAP-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 _ValleyRidgeMinuteLoad( + factor=factor, + requested=len(symbols), + covered=tuple(covered), + empty_symbols=tuple(empty), + raw_rows=raw_rows, + live_calls=0, + ridge_coverage=coverage, + ) + + +# --------------------------------------------------------------------------- # +# Evaluation core (network-free seam: given panels, run the two evaluations) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _RunReports: + """The two evaluation reports (no-book / with-book) + their file paths.""" + + no_book: FactorEvalReport + with_book: FactorEvalReport + no_book_md: Path + no_book_json: Path + with_book_md: Path + with_book_json: Path + no_book_dashboard: Path + with_book_dashboard: Path + + +def evaluate_two_runs( + factor_panel: pd.Series | pd.DataFrame, + spec: FactorSpec, + eval_cfg: EvalConfig, + price_panel: pd.DataFrame, + book: pd.DataFrame, + *, + universe_symbols: tuple[str, ...], + fee_rate: float, + report_dir: Path, + stem: str = _REPORT_STEM, +) -> _RunReports: + """Run the StandardFactorEvaluator TWICE (no book / with book) and write reports. + + This is the network-free seam: given the PROCESSED factor panel, the qfq price + panel (for forward returns), and the PROCESSED known-factor book, it does the two + ``evaluate`` calls and writes ``{stem}_no_book`` / ``{stem}_with_book`` as + Markdown + JSON. Run 1 omits ``known_factors`` (Incremental NOT_ASSESSED); run 2 + supplies the book (Incremental measured). + """ + evaluator = StandardFactorEvaluator() + report_dir.mkdir(parents=True, exist_ok=True) + + ctx_no_book = EvalContext( + price_panel=price_panel, + universe_symbols=universe_symbols, + fee_rate=fee_rate, + ) + # evaluate_with_ir yields the SAME report as evaluate() plus the IR the + # research-style dashboard needs (per-period IC + quantile return series). + report_no_book, ir_no_book = evaluator.evaluate_with_ir( + factor_panel, spec, eval_cfg, ctx_no_book + ) + + ctx_with_book = EvalContext( + price_panel=price_panel, + universe_symbols=universe_symbols, + fee_rate=fee_rate, + known_factors=book, + ) + report_with_book, ir_with_book = evaluator.evaluate_with_ir( + factor_panel, spec, eval_cfg, ctx_with_book + ) + + nb_md, nb_json = _write_report(report_no_book, report_dir, f"{stem}_no_book") + wb_md, wb_json = _write_report(report_with_book, report_dir, f"{stem}_with_book") + nb_png = render_factor_dashboard( + report_no_book, ir_no_book, report_dir / f"{stem}_no_book_dashboard.png" + ) + wb_png = render_factor_dashboard( + report_with_book, ir_with_book, report_dir / f"{stem}_with_book_dashboard.png" + ) + return _RunReports( + no_book=report_no_book, + with_book=report_with_book, + no_book_md=nb_md, + no_book_json=nb_json, + with_book_md=wb_md, + with_book_json=wb_json, + no_book_dashboard=nb_png, + with_book_dashboard=wb_png, + ) + + +def _write_report(report: FactorEvalReport, report_dir: Path, stem: str) -> tuple[Path, Path]: + """Write ``report`` as deterministic Markdown + JSON; return the two paths.""" + md_path = report_dir / f"{stem}.md" + json_path = report_dir / f"{stem}.json" + md_path.write_text(report.render(), encoding="utf-8") + json_path.write_text(report.to_json(), encoding="utf-8") + return md_path, json_path + + +# --------------------------------------------------------------------------- # +# Metric extraction (for the CLI line + the handoff) +# --------------------------------------------------------------------------- # +def _section_payload(report: FactorEvalReport, name: str) -> dict: + section = report.by_name().get(name) + return dict(getattr(section, "payload", {}) or {}) + + +def extract_metrics(report: FactorEvalReport) -> dict: + """Pull the headline verdict + gated metrics out of a finished report.""" + verdict = report.require_verdict() + pred = _section_payload(report, "predictive_power") + purity = _section_payload(report, "data_coverage") + incr = _section_payload(report, "purity") + return { + "deployment": verdict.verdict, + "predictive": verdict.predictive.verdict, + "incremental": verdict.incremental.verdict, + "tradable": verdict.tradable.verdict, + "ic_mean": pred.get("ic_mean"), + "ic_ir": pred.get("ic_ir"), + "ic_ir_ci_low": pred.get("ic_ir_ci_low"), + "ic_ir_ci_high": pred.get("ic_ir_ci_high"), + "ic_ir_ci_n_eff": pred.get("ic_ir_ci_n_eff"), + "ic_win_rate": pred.get("ic_win_rate"), + "ic_nw_t": pred.get("ic_nw_t"), + "settled_rebalances": purity.get("settled_rebalances"), + "effective_samples": purity.get("effective_samples"), + "span_days": purity.get("span_days"), + "incremental_ic_ir": incr.get("incremental_ic_ir"), + "incremental_ic_ir_ci_low": incr.get("incremental_ic_ir_ci_low"), + "incremental_ic_ir_ci_high": incr.get("incremental_ic_ir_ci_high"), + "incremental_ic_ir_ci_n_eff": incr.get("incremental_ic_ir_ci_n_eff"), + "incremental_ic_mean": incr.get("incremental_ic_mean"), + } + + +# --------------------------------------------------------------------------- # +# EvalConfig construction (HONEST provenance of what the runner actually did) +# --------------------------------------------------------------------------- # +def _build_eval_config(cfg: RootConfig) -> EvalConfig: + """Build the per-run EvalConfig, declaring EXACTLY what the pipeline applied. + + The declarations must not overstate: this codebase's winsorize step is a P0 no-op, + so ``winsorize`` is declared None (nothing was clipped) even though the config may + toggle it. z-score + industry/size neutralization ARE applied, so they are + declared. ``oos_split`` (from the config's ``oos`` block) makes the OOS section run + so the Predictive axis can be assessed. ``is_exploratory=True``: this is a + reproduction on a shorter window / narrower neutralization than the report, not a + return claim (it caps the deployment label at Watch). + """ + if cfg.oos is None: + raise ValueError( + "run-eval-valley-ridge-vwap-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 ValleyRidgeVwapRatioEvalResult: + """Immutable summary of one run-eval-valley-ridge-vwap-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 + ridge_coverage: RidgeCoverage + 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-valley-ridge-vwap-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-valley-ridge-vwap-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-valley-ridge-vwap-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-valley-ridge-vwap-ratio expects " + "processing.neutralize.enabled=true (industry + size neutralization, " + "matching the report's neutral column and the EvalConfig declaration)." + ) + + +def run_eval_valley_ridge_vwap_ratio(config_path: str) -> ValleyRidgeVwapRatioEvalResult: + """Run the two real valley/ridge VWAP-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 = ValleyRidgeVwapRatioFactor(lookback_days=VALLEY_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) + + # Valley/ridge VWAP-ratio factor: cache-only per-symbol aggregation -> raw -> process. + load = _load_valley_ridge_vwap_ratio_panel( + cfg, symbols, spec, logger, + lookback_days=VALLEY_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_valley_bars=VALLEY_RIDGE_MIN_VALLEY_BARS, + min_ridge_bars=VALLEY_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 ValleyRidgeVwapRatioEvalResult( + config=cfg, + spec=spec, + requested_symbols=load.requested, + covered_symbols=len(load.covered), + empty_symbols=len(load.empty_symbols), + factor_rows=int(len(factor_series)), + minute_raw_rows=load.raw_rows, + minute_live_calls=load.live_calls, + ridge_coverage=load.ridge_coverage, + no_book_metrics=no_book_metrics, + with_book_metrics=with_book_metrics, + reports=reports, + log_path=log_path, + elapsed=time.monotonic() - started, + ) + + +def _build_book_factors() -> list: + """Instantiate the confirmed book (value_ep / value_bp / volatility_20).""" + from factors.compute.candidates import ValueFactor, VolatilityFactor + + return [ + ValueFactor("value_ep"), + ValueFactor("value_bp"), + VolatilityFactor(window=20), + ] + + +__all__ = [ + "RidgeCoverage", + "ValleyRidgeVwapRatioEvalResult", + "evaluate_two_runs", + "extract_metrics", + "run_eval_valley_ridge_vwap_ratio", + "summarize_ridge_coverage", +] diff --git a/tests/test_eval_valley_ridge_vwap_ratio_runner.py b/tests/test_eval_valley_ridge_vwap_ratio_runner.py new file mode 100644 index 0000000..79097ad --- /dev/null +++ b/tests/test_eval_valley_ridge_vwap_ratio_runner.py @@ -0,0 +1,365 @@ +"""PR-J runner: cache-only minute loader, ridge-scarcity coverage, two-run eval (no net).""" + +from __future__ import annotations + +import logging + +import numpy as np +import pandas as pd +import pytest + +from analytics.eval import EvalConfig, MANDATORY_SECTIONS, Section, Skipped +from analytics.eval.verdict import AXIS_NOT_ASSESSED, AXIS_VERDICTS +from data.cache.intraday_cache import ENDPOINT as INTRADAY_ENDPOINT +from data.cache.intraday_parquet_store import KEY_COLS, IntradayParquetStore +from data.clean.intraday_valley_ridge_vwap import ( + VALLEY_RIDGE_MIN_RIDGE_BARS, + VALLEY_RIDGE_MIN_VALLEY_BARS, +) +from factors.compute.intraday_derived import ValleyRidgeVwapRatioFactor +from qt.config import ( + AlphaCfg, + BacktestCfg, + CacheCfg, + CostCfg, + DataCfg, + FactorCfg, + OutputCfg, + PortfolioCfg, + RootConfig, + UniverseCfg, +) +from qt.eval_valley_ridge_vwap_ratio import ( + _load_valley_ridge_vwap_ratio_panel, + evaluate_two_runs, + extract_metrics, + summarize_ridge_coverage, +) + + +# --------------------------------------------------------------------------- # +# Minute cache-only loader +# --------------------------------------------------------------------------- # +def _stored_rows(sym, day, vols, amts): + """Build STORED_COLUMNS-shaped 1min rows for one session of CONSECUTIVE minutes. + + Bar i sits at ``09:31 + i`` minutes (all inside the 14:50 PIT window). OHLC are dummy + constants; ``volume`` drives the classification and ``amount`` -- set INDEPENDENTLY -- + drives the VWAPs. + """ + base = pd.Timestamp(day) + pd.Timedelta("09:31:00") + rows = [] + for i, (v, a) in enumerate(zip(vols, amts)): + be = base + pd.Timedelta(minutes=i) + rows.append( + { + "symbol": sym, + "bar_end": be, + "source_trade_time": be, + "open": 100.0, + "high": 100.0, + "low": 100.0, + "close": 100.0, + "volume": float(v), + "amount": float(a), + "freq": "1min", + } + ) + return pd.DataFrame(rows) + + +def _min_config(root, start, end): + return RootConfig( + data=DataCfg( + source="tushare", + start=start, + end=end, + external_secret_file="/nonexistent.json", + cache=CacheCfg(enabled=True, root_dir=str(root)), + ), + universe=UniverseCfg(type="static", symbols=["A.SZ"]), + factors=[FactorCfg(name="momentum_20")], + alpha=AlphaCfg(), + portfolio=PortfolioCfg(top_n=1), + backtest=BacktestCfg(), + cost=CostCfg(), + output=OutputCfg(), + ) + + +# Three constant background days (baseline mu=100, sigma=0 -> eruptive threshold 100), +# then the hand-computed CASE A test day: 16 slots, ridge runs at (3,4) and (8,9) with +# volume 200 / amount 4000 (price 20), an ISOLATED peak at slot 12 (volume 300 / amount +# 9000 -> price 30, excluded from BOTH legs), and 11 valley bars at volume 100 carrying +# amount 1000 (x6) and 1200 (x5) -> valley VWAP 12000/1100, ridge VWAP 20, ratio 6/11. +_DAYS = ("2021-07-01", "2021-07-02", "2021-07-03") +_TEST_DAY = "2021-07-04" +_N_SLOTS = 16 +_RIDGES = (3, 4, 8, 9) +_PEAK = 12 +_CASE_A_RATIO = 6.0 / 11.0 +_BG_VOLS = [100.0] * _N_SLOTS +_BG_AMTS = [1000.0] * _N_SLOTS + + +def _case_a(): + vols = [100.0] * _N_SLOTS + amts = [0.0] * _N_SLOTS + for s in _RIDGES: + vols[s] = 200.0 + amts[s] = 4000.0 + vols[_PEAK] = 300.0 + amts[_PEAK] = 9000.0 + valley_slots = [s for s in range(_N_SLOTS) if s not in _RIDGES and s != _PEAK] + for i, s in enumerate(valley_slots): + amts[s] = 1000.0 if i < 6 else 1200.0 + return vols, amts + + +# Small gates so a 4-day cache produces a value (baseline needs >= 2 prior obs; the test +# day carries 4 tradable ridge bars and 11 tradable valley bars). lookback_days=1 so the +# test day's value IS its own ratio -- the background days have no eruption at all, hence +# no ridge, hence no valid day, so they cannot average in. +_LOAD_KW = dict( + lookback_days=1, baseline_days=20, baseline_min_obs=2, + sigma_k=1.0, min_valid_days=1, min_classifiable=1, + min_valley_bars=1, min_ridge_bars=1, +) + + +def _seed_symbol(store, sym): + for day in _DAYS: + store.upsert( + INTRADAY_ENDPOINT, sym, "1min", + _stored_rows(sym, day, _BG_VOLS, _BG_AMTS), KEY_COLS, + ) + store.upsert( + INTRADAY_ENDPOINT, sym, "1min", + _stored_rows(sym, _TEST_DAY, *_case_a()), KEY_COLS, + ) + + +def test_minute_loader_is_cache_only_and_discloses_empty(tmp_path): + root = tmp_path / "cache" + store = IntradayParquetStore(str(root)) + # Two symbols with cached minute bars; a third has NO cached minute (must be + # disclosed as empty, never fetched). + for sym in ("AAA.SZ", "BBB.SZ"): + _seed_symbol(store, sym) + + cfg = _min_config(root, "2021-07-01", "2021-07-04") + spec = ValleyRidgeVwapRatioFactor().spec + logger = logging.getLogger("test.vrr.loader") + load = _load_valley_ridge_vwap_ratio_panel( + cfg, ["AAA.SZ", "BBB.SZ", "CCC.SZ"], spec, logger, **_LOAD_KW + ) + assert load.live_calls == 0 # store read has no fetch closure + assert set(load.covered) == {"AAA.SZ", "BBB.SZ"} # both produced a value + assert load.empty_symbols == ("CCC.SZ",) # uncovered disclosed, not fetched + assert load.factor.name == spec.factor_id + assert load.factor.notna().any() + # the test day carries the hand-computed valley/ridge VWAP ratio + d = pd.Timestamp(_TEST_DAY) + assert load.factor.loc[(d, "AAA.SZ")] == pytest.approx(_CASE_A_RATIO) + assert load.factor.loc[(d, "BBB.SZ")] == pytest.approx(_CASE_A_RATIO) + assert set(load.factor.index.get_level_values("symbol")) == {"AAA.SZ", "BBB.SZ"} + # a day where NOTHING erupts has no ridge leg at all -> invalid -> no value emitted + bg = pd.Timestamp(_DAYS[-1]) + assert (bg, "AAA.SZ") not in load.factor.index + + +def test_minute_loader_reports_the_ridge_scarcity_distribution(tmp_path): + """The ridge-bar distribution + validity rate must be MEASURED on the real days.""" + root = tmp_path / "cache" + store = IntradayParquetStore(str(root)) + _seed_symbol(store, "AAA.SZ") + cfg = _min_config(root, "2021-07-01", "2021-07-04") + load = _load_valley_ridge_vwap_ratio_panel( + cfg, ["AAA.SZ"], ValleyRidgeVwapRatioFactor().spec, + logging.getLogger("test.vrr.cov"), **_LOAD_KW, + ) + cov = load.ridge_coverage + # four symbol-days seen, but the first two have no same-slot baseline yet (it needs + # 2 prior observations) -> only two are CLASSIFIABLE, and of those only the test day + # has a ridge at all -> validity rate 1/2 over the days that had a fair chance. + assert cov.symbol_days == 4 + assert cov.classifiable_days == 2 + assert cov.days_below_classifiable_gate == 2 # the warm-up days, counted separately + assert cov.valid_days == 1 + assert cov.validity_rate == pytest.approx(0.5) + # the flat background day has zero ridge bars, the test day has four; warm-up days + # are excluded from the distribution so they cannot drag it towards zero + assert dict(cov.ridge_percentiles)[100] == pytest.approx(4.0) + assert dict(cov.ridge_percentiles)[0] == pytest.approx(0.0) + assert cov.ridge_mean == pytest.approx(2.0) + # the floors reported are the ones this RUN applied (_LOAD_KW uses 1, not the module + # defaults) -- otherwise the disclosure would describe gates nobody enforced + assert cov.min_ridge_bars == 1 + assert cov.min_valley_bars == 1 + assert cov.days_below_ridge_gate == 1 # only the flat day (0 ridges) is below 1 + line = cov.render() + assert "ridge scarcity" in line and "valid_days=1" in line + assert "below_ridge_gate(1)" in line and "below_valley_gate(1)" in line + + +def test_summarize_ridge_coverage_counterfactual_at_the_valley_floor(): + """The disclosure quantifies exactly what the LOWERED ridge floor buys.""" + diag = pd.DataFrame( + { + "classifiable_bars": [240, 240, 240, 240], + "valley_bars": [200, 200, 200, 200], + "ridge_bars": [4, 12, 25, 30], + # as the factor would mark them under the default floor of 10 + "valid": [False, True, True, True], + }, + index=pd.DatetimeIndex(pd.bdate_range("2022-01-03", periods=4), name="trade_date"), + ) + cov = summarize_ridge_coverage([diag]) + assert cov.symbol_days == 4 + assert cov.classifiable_days == 4 # every day clears PR-F's classifiable floor + assert cov.valid_days == 3 + # holding the ridge leg to the VALLEY floor (20) would keep only the 25 / 30 days + assert cov.valid_days_at_valley_floor == 2 + assert cov.days_below_ridge_gate == 1 + assert cov.days_below_valley_gate == 0 + assert cov.ridge_mean == pytest.approx((4 + 12 + 25 + 30) / 4) + # defaults are the PINNED production floors when the caller does not override + assert cov.min_ridge_bars == VALLEY_RIDGE_MIN_RIDGE_BARS == 10 + assert cov.min_valley_bars == VALLEY_RIDGE_MIN_VALLEY_BARS == 20 + assert f"below_ridge_gate({VALLEY_RIDGE_MIN_RIDGE_BARS})" in cov.render() + + +def test_summarize_ridge_coverage_handles_no_frames(): + cov = summarize_ridge_coverage([]) + assert cov.symbol_days == 0 + assert cov.classifiable_days == 0 + assert cov.valid_days == 0 + assert np.isnan(cov.validity_rate) + assert cov.render() # renders without dividing by zero + + +def test_minute_loader_blocks_when_nothing_cached(tmp_path): + cfg = _min_config(tmp_path / "empty", "2021-07-01", "2021-07-04") + spec = ValleyRidgeVwapRatioFactor().spec + with pytest.raises(ValueError, match="no requested symbol produced"): + _load_valley_ridge_vwap_ratio_panel( + cfg, ["ZZZ.SZ"], spec, logging.getLogger("test.vrr.block"), **_LOAD_KW + ) + + +# --------------------------------------------------------------------------- # +# Evaluation core (two runs) — synthetic processed panels, no network +# --------------------------------------------------------------------------- # +def _synthetic_panels(n_days=90, n_symbols=15, seed=7): + rng = np.random.default_rng(seed) + dates = pd.bdate_range("2022-01-03", periods=n_days) + symbols = [f"{i:06d}.SZ" for i in range(n_symbols)] + idx = pd.MultiIndex.from_product([dates, symbols], names=["date", "symbol"]) + + # processed (z-scored-ish) subject factor + a 3-column book on the same grid + factor = pd.Series( + rng.standard_normal(len(idx)), index=idx, name="valley_ridge_vwap_ratio_20" + ) + book = pd.DataFrame( + { + "value_ep": rng.standard_normal(len(idx)), + "value_bp": rng.standard_normal(len(idx)), + "volatility_20": rng.standard_normal(len(idx)), + }, + index=idx, + ) + # a qfq-style price panel with the CORE_COLUMNS the forward-return boundary needs + close = pd.Series( + 100.0 * np.exp(np.cumsum(0.001 * rng.standard_normal(len(idx)))), index=idx + ) + price = pd.DataFrame( + { + "open": close.to_numpy(), + "high": close.to_numpy() * 1.01, + "low": close.to_numpy() * 0.99, + "close": close.to_numpy(), + "volume": 1_000.0, + "amount": 100_000.0, + "adj_factor": 1.0, + }, + index=idx, + ) + return factor, book, price + + +def _eval_cfg(): + return EvalConfig( + universe="000905.SH", + universe_is_pit=True, + start="2022-01-03", + end="2022-05-10", + is_exploratory=True, + post_hoc_selected=False, + rebalance="daily", + n_quantiles=5, + oos_split="2022-03-15", + winsorize=None, + standardize="zscore", + neutralization=("industry", "size"), + industry_level="L1", + ) + + +def test_evaluate_two_runs_produces_full_reports_and_incremental_axis(tmp_path): + factor, book, price = _synthetic_panels() + spec = ValleyRidgeVwapRatioFactor().spec + reports = evaluate_two_runs( + factor, spec, _eval_cfg(), price, book, + universe_symbols=tuple(sorted(set(book.index.get_level_values("symbol")))), + fee_rate=0.001, + report_dir=tmp_path, + ) + + for report in (reports.no_book, reports.with_book): + by = report.by_name() + assert set(by) == set(MANDATORY_SECTIONS) # all 8 present + assert all(isinstance(s, (Section, Skipped)) for s in by.values()) + assert report.verdict is not None # a verdict was produced + + # no-book: purity Skipped (no book) -> Incremental NOT_ASSESSED + assert isinstance(reports.no_book.by_name()["purity"], Skipped) + assert reports.no_book.verdict.incremental.verdict == AXIS_NOT_ASSESSED + + # with-book: purity is a real Section that populated the Incremental facts + purity = reports.with_book.by_name()["purity"] + assert isinstance(purity, Section) + assert purity.payload["known_factors_supplied"] is True + assert "incremental_ic_ir" in purity.payload + incr = reports.with_book.verdict.incremental.verdict + assert incr in AXIS_VERDICTS and incr != AXIS_NOT_ASSESSED + + # reports were written to disk (md + json), both runs + for p in (reports.no_book_md, reports.no_book_json, + reports.with_book_md, reports.with_book_json): + assert p.exists() and p.stat().st_size > 0 + + # the research-style dashboard PNG is emitted for both runs (mandatory report + # artifact alongside md/json) + for p in (reports.no_book_dashboard, reports.with_book_dashboard): + assert p.exists() and p.stat().st_size > 20_000 + with open(p, "rb") as fh: + assert fh.read(8) == b"\x89PNG\r\n\x1a\n" + + # metrics extraction surfaces the gated fields + m = extract_metrics(reports.no_book) + assert m["deployment"] in {"Adopt", "Watch", "Reject", "INSUFFICIENT-DATA"} + assert "effective_samples" in m and "ic_ir" in m + + +def test_no_book_run_never_reaches_adopt(tmp_path): + # Structural guarantee (design §6): with no book + no execution facts, at most + # Watch — regardless of the signal (exploratory cap + NOT_ASSESSED axes). + factor, book, price = _synthetic_panels(seed=3) + reports = evaluate_two_runs( + factor, ValleyRidgeVwapRatioFactor().spec, _eval_cfg(), price, book, + universe_symbols=(), + fee_rate=0.001, + report_dir=tmp_path, + ) + assert reports.no_book.verdict.verdict != "Adopt" + assert reports.no_book.verdict.tradable.verdict == AXIS_NOT_ASSESSED diff --git a/tests/test_valley_ridge_vwap_ratio_factor.py b/tests/test_valley_ridge_vwap_ratio_factor.py new file mode 100644 index 0000000..b6f9349 --- /dev/null +++ b/tests/test_valley_ridge_vwap_ratio_factor.py @@ -0,0 +1,923 @@ +"""PR-J: VALLEY/RIDGE VWAP-RATIO factor. + +Same volume classification as PR-F / PR-H / PR-I (REUSED, not re-implemented), and the +same VWAP identity as PR-I -- but the DENOMINATOR is swapped from the whole visible day +to the RIDGE bars, so the factor contrasts the two behavioural groups head-on. A "ridge" +(量岭) is an ERUPTIVE minute that is NOT an isolated peak: PR-F's internal +``eruptive & ~peak``, now exposed as a first-class ``ridge`` column. The factor is the +trailing-20-valid-day mean of ``valley VWAP / ridge VWAP``. Sign is pre-registered +1. + +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 / ridge runs / an isolated peak at chosen slots and whose AMOUNTS are set +independently, so both VWAPs -- and therefore the ratio -- are known in closed form. +Every hand day deliberately contains an isolated PEAK whose price differs sharply from +the ridges: it must never leak into the ridge leg, which is the one way this factor can +silently degenerate into "eruptive VWAP". +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from data.clean.intraday_schema import empty_intraday_bars, normalize_intraday_bars +from data.clean.intraday_valley_ridge_vwap import ( + VALLEY_RIDGE_LOOKBACK_DAYS, + VALLEY_RIDGE_MIN_RIDGE_BARS, + VALLEY_RIDGE_MIN_VALLEY_BARS, + compute_valley_ridge_vwap_ratio, + valley_ridge_vwap_ratio_by_day, +) +from data.clean.intraday_volume_prv import ( + VOLUME_PRV_BASELINE_DAYS, + VOLUME_PRV_BASELINE_MIN_OBS, + VOLUME_PRV_MIN_VALID_DAYS, + VOLUME_PRV_SIGMA_K, + peak_mask_for_symbol, + prepare_visible_minute_bars, +) +from factors.compute.intraday_derived import ValleyRidgeVwapRatioFactor +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`` -- that is the point: the per-bar price + is ``amount / volume``. OHLC are dummy constants (this factor reads only volume + + amount). ``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)) + ] + + +def _background(n_days, n_slots, sym=_SYM, start_day="2021-07-01"): + """``n_days`` prior days of flat volume-100 / amount-1000 sessions. + + Same-slot baseline becomes mu=100, sigma=0 -> the eruptive threshold is exactly 100, + so on the test day a volume of 100 is a VALLEY and anything above erupts. + """ + 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, [1000.0] * n_slots, sym=sym) + return rows + + +_BG_DAYS = 10 +_TEST_DAY = pd.Timestamp("2021-07-11") + +# Gates small enough that the single engineered test day is the only VALID day; the +# valley-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=VALLEY_RIDGE_LOOKBACK_DAYS, + min_valid_days=1, + min_classifiable=1, + min_valley_bars=1, + min_ridge_bars=1, +) + + +# --------------------------------------------------------------------------- # +# Hand-computed VWAP ratios (2 non-trivial cases, closed-form fractions) +# --------------------------------------------------------------------------- # +# CASE A -- 16 slots. +# ridge runs at slots (3,4) and (8,9): each bar volume 200 / amount 4000 -> price 20. +# Every one of the four has an ERUPTIVE neighbour, so none is an isolated peak -> +# all four are ridges. ridge VWAP = 16000 / 800 = 20.0 +# an ISOLATED eruption at slot 12 (volume 300 / amount 9000 -> price 30) IS a peak, +# so it must be excluded from the ridge leg (the discriminating case). +# valley bars: 11 x volume 100; six carry amount 1000 (price 10), five amount 1200 +# (price 12) -> valley VWAP = 12000 / 1100 +# ratio = (12000/1100) / 20 = 6/11 = 0.545454... +# The ridges traded far ABOVE the calm minutes, so the ratio sits below 1. +_CASE_A_N = 16 +_CASE_A_RIDGES = (3, 4, 8, 9) +_CASE_A_PEAK = 12 +_CASE_A_RATIO = 6.0 / 11.0 + + +def _case_a_day(): + n = _CASE_A_N + vols = [100.0] * n + amts = [0.0] * n + for s in _CASE_A_RIDGES: + vols[s] = 200.0 + amts[s] = 4000.0 + vols[_CASE_A_PEAK] = 300.0 + amts[_CASE_A_PEAK] = 9000.0 + valley_slots = [ + s for s in range(n) if s not in _CASE_A_RIDGES and s != _CASE_A_PEAK + ] + for i, s in enumerate(valley_slots): + amts[s] = 1000.0 if i < 6 else 1200.0 + return vols, amts + + +# CASE B -- 18 slots, ridge runs at (2,3), (7,8), (13,14): volume 300 / amount 2400 -> +# price 8 -> ridge VWAP = 14400 / 1800 = 8.0 +# isolated peak at slot 16 (volume 500 / amount 25000 -> price 50), excluded. +# valley bars: 11 x volume 100; six amount 1000, five amount 1500 -> 13500 / 1100 +# ratio = (13500/1100) / 8 = 135/88 = 1.534090... +# Here the ridges traded BELOW the calm minutes -> ratio > 1, the opposite direction from +# case A, so a sign / inversion bug cannot pass both. +_CASE_B_N = 18 +_CASE_B_RIDGES = (2, 3, 7, 8, 13, 14) +_CASE_B_PEAK = 16 +_CASE_B_RATIO = 135.0 / 88.0 + + +def _case_b_day(): + n = _CASE_B_N + vols = [100.0] * n + amts = [0.0] * n + for s in _CASE_B_RIDGES: + vols[s] = 300.0 + amts[s] = 2400.0 + vols[_CASE_B_PEAK] = 500.0 + amts[_CASE_B_PEAK] = 25_000.0 + valley_slots = [ + s for s in range(n) if s not in _CASE_B_RIDGES and s != _CASE_B_PEAK + ] + for i, s in enumerate(valley_slots): + amts[s] = 1000.0 if i < 6 else 1500.0 + return vols, amts + + +def test_hand_value_case_a_ratio_below_one(): + vols, amts = _case_a_day() + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", vols, amts) + out = compute_valley_ridge_vwap_ratio(_bars(rows), **_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_above_one(): + vols, amts = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", vols, amts) + out = compute_valley_ridge_vwap_ratio(_bars(rows), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(_CASE_B_RATIO) + assert out.loc[(_TEST_DAY, _SYM)] > 1.0 + + +def test_isolated_peak_is_excluded_from_the_ridge_leg(): + """The ridge denominator is ``eruptive & ~peak``, NOT all eruptive bars. + + Moving ONLY the isolated peak's price must leave the factor untouched; if the + implementation summed every eruptive bar the ratio would move. This is the single + most important discriminator in the file. + """ + vols, amts = _case_a_day() + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", vols, amts) + base = compute_valley_ridge_vwap_ratio(_bars(rows), **_KW) + + vols2, amts2 = _case_a_day() + amts2[_CASE_A_PEAK] = 90_000.0 # price 300 instead of 30 + rows2 = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", vols2, amts2) + moved = compute_valley_ridge_vwap_ratio(_bars(rows2), **_KW) + + key = (_TEST_DAY, _SYM) + assert base.loc[key] == pytest.approx(_CASE_A_RATIO) + assert moved.loc[key] == pytest.approx(_CASE_A_RATIO) + + # ...and the day really does contain an isolated peak (the assertion is not vacuous) + work = peak_mask_for_symbol( + prepare_visible_minute_bars(_bars(rows)).reset_index(drop=True) + ) + day = work[work["trade_date"] == _TEST_DAY].reset_index(drop=True) + assert bool(day.loc[_CASE_A_PEAK, "peak"]) is True + assert bool(day.loc[_CASE_A_PEAK, "ridge"]) is False + + +def test_ridge_mask_is_eruptive_with_an_eruptive_neighbour_on_the_hand_day(): + """The ridge set on case A is EXACTLY the two adjacent-eruption runs.""" + vols, amts = _case_a_day() + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", vols, amts) + work = peak_mask_for_symbol( + prepare_visible_minute_bars(_bars(rows)).reset_index(drop=True) + ) + day = work[work["trade_date"] == _TEST_DAY].reset_index(drop=True) + ridge_slots = tuple(day.index[day["ridge"].to_numpy(dtype=bool)]) + assert ridge_slots == _CASE_A_RIDGES + + +def test_vwap_uses_the_amount_over_volume_aggregation_identity(): + """sum(p_i * v_i) / sum(v_i) with p_i = amount_i/volume_i IS sum(amount)/sum(volume). + + The PINNED VWAP definition, recomputed the LONG way from per-bar prices for BOTH + legs (valley bars and ridge bars) and checked against the closed-form ratio the + factor returns -- so the identity is verified, not assumed. + """ + vols, amts = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", vols, amts) + bars = _bars(rows) + + visible = prepare_visible_minute_bars(bars, extra_columns=("amount",)) + work = peak_mask_for_symbol(visible.reset_index(drop=True)) + day = work[work["trade_date"] == _TEST_DAY] + + v = day["volume"].to_numpy(dtype=float) + a = day["amount"].to_numpy(dtype=float) + price = a / v # per-bar VWAP + is_valley = day["valley"].to_numpy(dtype=bool) + is_ridge = day["ridge"].to_numpy(dtype=bool) + + valley_long = float((price[is_valley] * v[is_valley]).sum() / v[is_valley].sum()) + assert valley_long == pytest.approx(float(a[is_valley].sum() / v[is_valley].sum())) + assert valley_long == pytest.approx(13500.0 / 1100.0) + + ridge_long = float((price[is_ridge] * v[is_ridge]).sum() / v[is_ridge].sum()) + assert ridge_long == pytest.approx(float(a[is_ridge].sum() / v[is_ridge].sum())) + assert ridge_long == pytest.approx(8.0) + + out = compute_valley_ridge_vwap_ratio(bars, **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(valley_long / ridge_long) + + +def test_moving_the_ridge_price_moves_the_ratio_inversely(): + """The ridge leg really is the DENOMINATOR (guard against a swapped ratio).""" + vols, amts = _case_a_day() + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", vols, amts) + base = compute_valley_ridge_vwap_ratio(_bars(rows), **_KW) + + vols2, amts2 = _case_a_day() + for s in _CASE_A_RIDGES: + amts2[s] = 8000.0 # ridge price 40 instead of 20 -> denominator doubles + rows2 = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", vols2, amts2) + dearer = compute_valley_ridge_vwap_ratio(_bars(rows2), **_KW) + + key = (_TEST_DAY, _SYM) + assert dearer.loc[key] == pytest.approx(base.loc[key] / 2.0) + + +def test_day_where_valleys_and_ridges_share_one_price_has_ratio_exactly_one(): + """An exact structural identity: the factor is centred on 1. + + Every bar trades at price 10, so both VWAPs are 10 whatever the volume split. + """ + n = _CASE_A_N + vols, _ = _case_a_day() + amts = [10.0 * v for v in vols] # price == 10 everywhere + rows = _background(_BG_DAYS, n) + _session("2021-07-11", vols, amts) + out = compute_valley_ridge_vwap_ratio(_bars(rows), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(1.0) + + +# --------------------------------------------------------------------------- # +# Ridge scarcity: the >= 10 tradable-ridge-bar gate (PINNED lower than the valley +# floor because ridges are structurally far rarer) +# --------------------------------------------------------------------------- # +def _ridge_runs_day(run_lengths, n_slots): + """A day whose eruptive bars form CONSECUTIVE runs of the given lengths. + + Runs are separated by 2 mild bars, so every eruptive bar has an eruptive neighbour + and a run of length L contributes exactly L ridge bars (no isolated peaks at all). + Ridge bars: volume 200 / amount 4000 (price 20); valleys: volume 100 / amount 1000. + """ + vols = [100.0] * n_slots + amts = [1000.0] * n_slots + slot = 1 + for length in run_lengths: + if slot + length > n_slots: + raise AssertionError("test day too short for the requested runs") + for k in range(length): + vols[slot + k] = 200.0 + amts[slot + k] = 4000.0 + slot += length + 2 + return vols, amts + + +def _ridge_count(vols, amts, n_slots): + """Ridge-bar count on the engineered day, straight off the reused mask.""" + rows = _background(_BG_DAYS, n_slots) + _session("2021-07-11", vols, amts) + work = peak_mask_for_symbol( + prepare_visible_minute_bars(_bars(rows)).reset_index(drop=True) + ) + day = work[work["trade_date"] == _TEST_DAY] + return int(day["ridge"].sum()), int(day["peak"].sum()) + + +_RIDGE_GATE_N = 24 + + +def test_exactly_nine_ridge_bars_fails_the_default_ten_gate(): + vols, amts = _ridge_runs_day([2, 2, 2, 3], _RIDGE_GATE_N) # 9 ridges + assert _ridge_count(vols, amts, _RIDGE_GATE_N) == (9, 0) + rows = _background(_BG_DAYS, _RIDGE_GATE_N) + _session("2021-07-11", vols, amts) + out = compute_valley_ridge_vwap_ratio( + _bars(rows), **{**_KW, "min_ridge_bars": VALLEY_RIDGE_MIN_RIDGE_BARS} + ) + assert out.dropna().empty + + +def test_exactly_ten_ridge_bars_passes_the_default_ten_gate(): + vols, amts = _ridge_runs_day([2, 2, 2, 2, 2], _RIDGE_GATE_N) # 10 ridges + assert _ridge_count(vols, amts, _RIDGE_GATE_N) == (10, 0) + rows = _background(_BG_DAYS, _RIDGE_GATE_N) + _session("2021-07-11", vols, amts) + out = compute_valley_ridge_vwap_ratio( + _bars(rows), **{**_KW, "min_ridge_bars": VALLEY_RIDGE_MIN_RIDGE_BARS} + ) + # 14 valleys @ price 10, 10 ridges @ price 20 -> ratio exactly 0.5 + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(0.5) + + +def test_ridge_bar_count_is_taken_AFTER_the_positive_trade_guard(): + """The >= min_ridge_bars gate counts ridges that actually CONTRIBUTE to the VWAP. + + Ten classifiable ridge bars, but one traded nothing -> only nine support the + denominator, so the default gate must invalidate the day. + """ + vols, amts = _ridge_runs_day([2, 2, 2, 2, 2], _RIDGE_GATE_N) + # A ZERO-VOLUME bar could not erupt at all, so the untradable ridge is made with a + # zero AMOUNT: the bar still clears the volume threshold (200 > 100) and is still + # classified as a ridge, but it carries no traded value for the VWAP. + amts[1] = 0.0 + assert _ridge_count(vols, amts, _RIDGE_GATE_N) == (10, 0) + rows = _background(_BG_DAYS, _RIDGE_GATE_N) + _session("2021-07-11", vols, amts) + out = compute_valley_ridge_vwap_ratio( + _bars(rows), **{**_KW, "min_ridge_bars": VALLEY_RIDGE_MIN_RIDGE_BARS} + ) + assert out.dropna().empty + # ...and a floor of 9 accepts the day, proving it was the COUNT that blocked it + ok = compute_valley_ridge_vwap_ratio(_bars(rows), **{**_KW, "min_ridge_bars": 9}) + assert np.isfinite(ok.loc[(_TEST_DAY, _SYM)]) + + +def test_day_with_no_ridge_is_invalid(): + """Only isolated peaks erupt -> no ridge denominator -> honest NaN, never 0/0.""" + n = 12 + vols = [100.0] * n + amts = [1000.0] * n + for s in (3, 7): # isolated -> peaks, not ridges + vols[s] = 200.0 + amts[s] = 4000.0 + rows = _background(_BG_DAYS, n) + _session("2021-07-11", vols, amts) + out = compute_valley_ridge_vwap_ratio(_bars(rows), **_KW) + assert out.dropna().empty + + +def test_day_with_no_tradable_ridge_is_invalid(): + vols, amts = _case_a_day() + for s in _CASE_A_RIDGES: + amts[s] = 0.0 # classifiable ridges that traded no value + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", vols, amts) + out = compute_valley_ridge_vwap_ratio(_bars(rows), **_KW) + assert out.dropna().empty + + +def test_day_with_no_tradable_valley_is_invalid(): + vols, amts = _case_a_day() + for s in range(_CASE_A_N): + if s not in _CASE_A_RIDGES and s != _CASE_A_PEAK: + vols[s] = 0.0 + amts[s] = 0.0 + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", vols, amts) + out = compute_valley_ridge_vwap_ratio(_bars(rows), **_KW) + assert out.dropna().empty + + +def test_degenerate_bars_are_dropped_from_both_legs_without_polluting_either(): + """Untradable bars on BOTH sides of the taxonomy leave the hand value exact.""" + n = _CASE_A_N + 3 + vols, amts = _case_a_day() + # slot 16: zero-volume valley; slot 17: zero-amount valley; slot 18: negative amount + vols = vols + [0.0, 50.0, 50.0] + amts = amts + [500.0, 0.0, -9_999.0] + rows = _background(_BG_DAYS, n) + _session("2021-07-11", vols, amts) + out = compute_valley_ridge_vwap_ratio(_bars(rows), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(_CASE_A_RATIO) + + +def test_valley_bar_gate_still_applies_independently_of_the_ridge_gate(): + """The two floors are separate gates; the valley one is unchanged from PR-I.""" + vols, amts = _case_a_day() # 11 valley bars + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", vols, amts) + bars = _bars(rows) + ok = compute_valley_ridge_vwap_ratio(bars, **{**_KW, "min_valley_bars": 11}) + assert ok.loc[(_TEST_DAY, _SYM)] == pytest.approx(_CASE_A_RATIO) + too_few = compute_valley_ridge_vwap_ratio(bars, **{**_KW, "min_valley_bars": 12}) + assert too_few.dropna().empty + + +# --------------------------------------------------------------------------- # +# Window mechanics: the trailing mean over VALID days +# --------------------------------------------------------------------------- # +def _two_case_days(): + """Background + day1 = case A pattern, day2 = case B pattern, on a shared grid.""" + n = _CASE_B_N + a_vols, a_amts = _case_a_day() + pad = n - _CASE_A_N + a_vols = a_vols + [100.0] * pad + a_amts = a_amts + [1000.0] * pad + rows = _background(_BG_DAYS, n) + rows += _session("2021-07-11", a_vols, a_amts) + rows += _session("2021-07-12", *_case_b_day()) + return rows + + +def test_factor_is_the_mean_of_the_trailing_valid_day_ratios(): + bars = _bars(_two_case_days()) + ratios = compute_valley_ridge_vwap_ratio(bars, **{**_KW, "lookback_days": 1}) + d1, d2 = pd.Timestamp("2021-07-11"), pd.Timestamp("2021-07-12") + r1, r2 = ratios.loc[(d1, _SYM)], ratios.loc[(d2, _SYM)] + assert r1 != pytest.approx(r2) + + out = compute_valley_ridge_vwap_ratio(bars, **{**_KW, "lookback_days": 2}) + assert out.loc[(d1, _SYM)] == pytest.approx(r1) + assert out.loc[(d2, _SYM)] == pytest.approx((r1 + r2) / 2.0) + + +def test_window_drops_days_older_than_lookback(): + rows = _two_case_days() + _session("2021-07-13", *_case_b_day()) + bars = _bars(rows) + per_day = compute_valley_ridge_vwap_ratio(bars, **{**_KW, "lookback_days": 1}) + d1, d2, d3 = (pd.Timestamp(f"2021-07-1{k}") for k in (1, 2, 3)) + out = compute_valley_ridge_vwap_ratio(bars, **{**_KW, "lookback_days": 2}) + expected = (per_day.loc[(d2, _SYM)] + per_day.loc[(d3, _SYM)]) / 2.0 + assert out.loc[(d3, _SYM)] == pytest.approx(expected) + three = compute_valley_ridge_vwap_ratio(bars, **{**_KW, "lookback_days": 3}) + assert three.loc[(d3, _SYM)] != pytest.approx(expected) + assert np.isfinite(per_day.loc[(d1, _SYM)]) + + +def test_invalid_days_do_not_occupy_a_slot_in_the_trailing_window(): + """A day that fails a gate is ABSENT, not NaN -- it must not shorten the window.""" + n = _CASE_B_N + a_vols, a_amts = _case_a_day() + a_vols = a_vols + [100.0] * (n - _CASE_A_N) + a_amts = a_amts + [1000.0] * (n - _CASE_A_N) + rows = _background(_BG_DAYS, n) + rows += _session("2021-07-11", a_vols, a_amts) + # a day with NO ridge at all -> invalid, contributes nothing + rows += _session("2021-07-12", [100.0] * n, [1000.0] * n) + rows += _session("2021-07-13", *_case_b_day()) + out = compute_valley_ridge_vwap_ratio(_bars(rows), **{**_KW, "lookback_days": 2}) + per_day = compute_valley_ridge_vwap_ratio(_bars(rows), **{**_KW, "lookback_days": 1}) + d1, d2, d3 = (pd.Timestamp(f"2021-07-1{k}") for k in (1, 2, 3)) + assert (d2, _SYM) not in per_day.index # the flat day never produced a value + # day 3's 2-day window pools day 1 and day 3, skipping the invalid day 2 + expected = (per_day.loc[(d1, _SYM)] + per_day.loc[(d3, _SYM)]) / 2.0 + assert out.loc[(d3, _SYM)] == pytest.approx(expected) + + +def test_min_valid_days_floor_returns_nan_until_enough_valid_days(): + rows = _background(_BG_DAYS, _CASE_B_N) + for k in range(3): + d = (pd.Timestamp("2021-07-11") + pd.Timedelta(days=k)).strftime("%Y-%m-%d") + rows += _session(d, *_case_b_day()) + out = compute_valley_ridge_vwap_ratio(_bars(rows), **{**_KW, "min_valid_days": 3}) + assert np.isnan(out.loc[(pd.Timestamp("2021-07-11"), _SYM)]) + assert np.isnan(out.loc[(pd.Timestamp("2021-07-12"), _SYM)]) + assert np.isfinite(out.loc[(pd.Timestamp("2021-07-13"), _SYM)]) + + +def test_min_classifiable_gate_invalidates_thin_days(): + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", *_case_a_day()) + out = compute_valley_ridge_vwap_ratio(_bars(rows), **{**_KW, "min_classifiable": 100}) + assert out.dropna().empty + + +def test_baseline_insufficient_yields_no_value(): + rows = _background(9, _CASE_A_N) + _session("2021-07-10", *_case_a_day()) + out = compute_valley_ridge_vwap_ratio(_bars(rows), **_KW) + assert out.dropna().empty + + +def test_default_gates_match_the_pinned_definition(): + assert VALLEY_RIDGE_LOOKBACK_DAYS == 20 + assert VALLEY_RIDGE_MIN_VALLEY_BARS == 20 + # PINNED LOWER than the valley floor: ridges are structurally far rarer (a bar must + # erupt AND have an eruptive neighbour). + assert VALLEY_RIDGE_MIN_RIDGE_BARS == 10 + assert VOLUME_PRV_MIN_VALID_DAYS == 10 + + +# --------------------------------------------------------------------------- # +# PIT: no lookahead, no post-cutoff influence -- WITH TEETH +# --------------------------------------------------------------------------- # +_LATE_START = "14:50:00" +_LATE_SLOTS = 6 + + +def _two_block_background(n_days, early_slots, sym=_SYM, start_day="2021-07-01"): + """Background whose sessions cover BOTH the early block and the post-cutoff block. + + Giving the late slots a same-slot baseline is what puts TEETH in the leakage test: + if the 14:50 truncation were removed, the late bars would be fully classifiable and + would really change the factor -- so the "perturbation changes nothing" assertion is + not passing merely because the late bars are unclassifiable noise. + """ + rows = [] + for i in range(n_days): + day = (pd.Timestamp(start_day) + pd.Timedelta(days=i)).strftime("%Y-%m-%d") + rows += _session(day, [100.0] * early_slots, [1000.0] * early_slots, sym=sym) + rows += _session( + day, [100.0] * _LATE_SLOTS, [1000.0] * _LATE_SLOTS, + sym=sym, start=_LATE_START, + ) + return rows + + +def _late_block(day, sym=_SYM): + """A post-14:50 block with two ridge runs priced ONE (wildly off the early block).""" + vols = [100.0] * _LATE_SLOTS + amts = [1000.0] * _LATE_SLOTS + for s in (0, 1, 3, 4): + vols[s] = 500.0 + amts[s] = 500.0 # price 1 + return _session(day, vols, amts, sym=sym, start=_LATE_START) + + +def test_perturbing_post_1450_bars_does_not_change_factor(): + rows = _two_block_background(_BG_DAYS, _CASE_A_N) + rows += _session("2021-07-11", *_case_a_day()) + clean = compute_valley_ridge_vwap_ratio(_bars(rows), **_KW) + perturbed = compute_valley_ridge_vwap_ratio( + _bars(rows + _late_block("2021-07-11")), **_KW + ) + key = (_TEST_DAY, _SYM) + assert clean.loc[key] == pytest.approx(_CASE_A_RATIO) + assert perturbed.loc[key] == pytest.approx(clean.loc[key]) + + +def test_leakage_test_has_teeth_removing_the_cutoff_changes_the_value(): + """The counter-test: with the 14:50 truncation DISABLED the value moves materially. + + Without this, the assertion above could pass for the wrong reason (post-cutoff bars + that are inert anyway). Here the same bars are shown to carry a large, real effect + the moment the cutoff stops hiding them -- so the truncation is doing the work. + """ + rows = _two_block_background(_BG_DAYS, _CASE_A_N) + rows += _session("2021-07-11", *_case_a_day()) + rows += _late_block("2021-07-11") + bars = _bars(rows) + + truncated = compute_valley_ridge_vwap_ratio(bars, **_KW) + untruncated = compute_valley_ridge_vwap_ratio( + bars, decision_time="23:59:59", **_KW + ) + key = (_TEST_DAY, _SYM) + a, b = truncated.loc[key], untruncated.loc[key] + assert a == pytest.approx(_CASE_A_RATIO) + assert np.isfinite(b) + # the late block adds four cheap ridge bars, collapsing the denominator: the value + # must move by far more than any rounding tolerance + assert abs(b - a) / a > 0.5 + assert b > 1.0 > a + + +def test_future_day_does_not_change_earlier_factor(): + base = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", *_case_a_day()) + a = compute_valley_ridge_vwap_ratio(_bars(base), **_KW) + future = base + _session( + "2021-07-12", [9_999.0] * _CASE_A_N, [1.0] * _CASE_A_N + ) + b = compute_valley_ridge_vwap_ratio(_bars(future), **_KW) + key = (_TEST_DAY, _SYM) + assert np.isfinite(a.loc[key]) + assert a.loc[key] == pytest.approx(b.loc[key]) + + +# --------------------------------------------------------------------------- # +# Per-symbol isolation +# --------------------------------------------------------------------------- # +def test_per_symbol_isolation(): + n = _CASE_B_N + a_vols, a_amts = _case_a_day() + a_vols = a_vols + [100.0] * (n - _CASE_A_N) + a_amts = a_amts + [1000.0] * (n - _CASE_A_N) + + rows = _background(_BG_DAYS, n, sym="AAA.SZ") + rows += _session("2021-07-11", a_vols, a_amts, sym="AAA.SZ") + rows += _background(_BG_DAYS, n, sym="BBB.SZ") + rows += _session("2021-07-11", *_case_b_day(), sym="BBB.SZ") + out = compute_valley_ridge_vwap_ratio(_bars(rows), **_KW) + + solo_a = compute_valley_ridge_vwap_ratio( + _bars( + _background(_BG_DAYS, n, sym="AAA.SZ") + + _session("2021-07-11", a_vols, a_amts, sym="AAA.SZ") + ), + **_KW, + ) + assert out.loc[(_TEST_DAY, "AAA.SZ")] == pytest.approx( + solo_a.loc[(_TEST_DAY, "AAA.SZ")] + ) + assert out.loc[(_TEST_DAY, "BBB.SZ")] == pytest.approx(_CASE_B_RATIO) + + +# --------------------------------------------------------------------------- # +# §0 REUSE NON-DRIFT: the ridge exposure changed nothing for PR-F / PR-H / PR-I +# --------------------------------------------------------------------------- # +def test_ridge_is_exactly_eruptive_and_not_peak(): + vols, amts = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", vols, amts) + work = peak_mask_for_symbol( + prepare_visible_minute_bars(_bars(rows)).reset_index(drop=True) + ) + classifiable = work["classifiable"].to_numpy(dtype=bool) + ridge = work["ridge"].to_numpy(dtype=bool) + peak = work["peak"].to_numpy(dtype=bool) + # reconstruct "eruptive" independently from the raw threshold rule + eruptive = classifiable & ( + work["volume"].to_numpy(dtype=float) > work["thr"].to_numpy(dtype=float) + ) + np.testing.assert_array_equal(ridge, eruptive & ~peak) + day = work["trade_date"].to_numpy() == np.datetime64(_TEST_DAY) + assert ridge[day].any() and peak[day].any() # not vacuous + + +def test_the_three_masks_partition_the_classifiable_bars(): + """valley | peak | ridge == classifiable, pairwise disjoint. + + The structural guarantee that adding ``ridge`` did not carve anything out of the + existing masks: every classifiable bar lands in exactly one of the three. + """ + vols, amts = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", vols, amts) + work = peak_mask_for_symbol( + prepare_visible_minute_bars(_bars(rows)).reset_index(drop=True) + ) + classifiable = work["classifiable"].to_numpy(dtype=bool) + valley = work["valley"].to_numpy(dtype=bool) + peak = work["peak"].to_numpy(dtype=bool) + ridge = work["ridge"].to_numpy(dtype=bool) + np.testing.assert_array_equal(valley | peak | ridge, classifiable) + assert not (valley & peak).any() + assert not (valley & ridge).any() + assert not (peak & ridge).any() + + +def test_session_boundary_eruption_is_a_ridge_not_a_peak(): + """PINNED: an eruptive bar whose isolation is UNPROVABLE counts as a ridge. + + The first visible bar of the day has no previous neighbour, so PR-F conservatively + refuses to call it a peak; the complement therefore makes it a ridge. Disclosed + behaviour, locked here so it can never drift silently. + """ + n = 12 + vols = [100.0] * n + amts = [1000.0] * n + vols[0] = 200.0 # first visible bar erupts -> no previous neighbour + amts[0] = 4000.0 + vols[5] = 200.0 # an isolated eruption in the interior -> a real peak + amts[5] = 4000.0 + rows = _background(_BG_DAYS, n) + _session("2021-07-11", vols, amts) + work = peak_mask_for_symbol( + prepare_visible_minute_bars(_bars(rows)).reset_index(drop=True) + ) + day = work[work["trade_date"] == _TEST_DAY].reset_index(drop=True) + assert bool(day.loc[0, "ridge"]) is True + assert bool(day.loc[0, "peak"]) is False + assert bool(day.loc[5, "peak"]) is True + assert bool(day.loc[5, "ridge"]) is False + + +def test_valley_and_peak_and_classifiable_unchanged_by_the_ridge_exposure(): + """The three PRE-EXISTING masks are bit-identical with the ridge column present.""" + vols, amts = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", vols, amts) + bars = _bars(rows) + plain = peak_mask_for_symbol( + prepare_visible_minute_bars(bars).reset_index(drop=True) + ) + carried = peak_mask_for_symbol( + prepare_visible_minute_bars(bars, extra_columns=("amount",)).reset_index(drop=True) + ) + for col in ("trade_date", "bar_end", "slot", "volume", "classifiable", "valley", "peak"): + pd.testing.assert_series_equal(carried[col], plain[col], check_names=False) + + +def test_volume_peak_count_unchanged_by_the_ridge_exposure(): + """PR-F's factor value is bit-identical (same hand case as the PR-I lock).""" + from data.clean.intraday_volume_prv import compute_volume_peak_count + + vols, amts = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", vols, amts) + kw = dict( + baseline_days=VOLUME_PRV_BASELINE_DAYS, + baseline_min_obs=VOLUME_PRV_BASELINE_MIN_OBS, + sigma_k=VOLUME_PRV_SIGMA_K, + lookback_days=1, + min_valid_days=1, + min_classifiable=1, + ) + out = compute_volume_peak_count(_bars(rows), **kw) + # case B has exactly ONE isolated eruption (slot 16); the three adjacent runs are + # ridges, not peaks + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(1.0) + + +def test_peak_interval_kurtosis_unchanged_by_the_ridge_exposure(): + """PR-H's factor value is bit-identical (the PR-H hand case, verbatim).""" + from data.clean.intraday_peak_interval import compute_peak_interval_kurtosis + + n = 25 + vols = [100.0] * n + for p in (1, 3, 6, 10, 15, 21, 23): + vols[p] = 200.0 + rows = _background(_BG_DAYS, n) + _session("2021-07-11", vols, [1000.0] * n) + out = compute_peak_interval_kurtosis( + _bars(rows), + baseline_days=VOLUME_PRV_BASELINE_DAYS, + baseline_min_obs=VOLUME_PRV_BASELINE_MIN_OBS, + sigma_k=VOLUME_PRV_SIGMA_K, + lookback_days=20, + min_valid_days=1, + min_classifiable=1, + min_intervals=4, + ) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(-1.48125) + + +def test_valley_relative_vwap_unchanged_by_the_ridge_exposure(): + """PR-I's factor value is bit-identical (the PR-I case-A hand value, verbatim).""" + from data.clean.intraday_valley_vwap import compute_valley_relative_vwap + + n = 12 + erupt = (3, 7) + vols = [100.0] * n + amts = [0.0] * n + valley_slots = [s for s in range(n) if s not in erupt] + for i, s in enumerate(valley_slots): + amts[s] = 1000.0 if i < 5 else 1200.0 + for s in erupt: + vols[s] = 200.0 + amts[s] = 4000.0 + rows = _background(_BG_DAYS, n) + _session("2021-07-11", vols, amts) + out = compute_valley_relative_vwap( + _bars(rows), + baseline_days=VOLUME_PRV_BASELINE_DAYS, + baseline_min_obs=VOLUME_PRV_BASELINE_MIN_OBS, + sigma_k=VOLUME_PRV_SIGMA_K, + lookback_days=20, + min_valid_days=1, + min_classifiable=1, + min_valley_bars=1, + ) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(77.0 / 95.0) + + +# --------------------------------------------------------------------------- # +# The per-day ratio helper (exposed for the reuse / diagnostics path) +# --------------------------------------------------------------------------- # +def test_valley_ridge_vwap_ratio_by_day_returns_only_valid_days(): + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", *_case_a_day()) + work = peak_mask_for_symbol( + prepare_visible_minute_bars( + _bars(rows), extra_columns=("amount",) + ).reset_index(drop=True) + ) + ratio = valley_ridge_vwap_ratio_by_day( + work, min_valley_bars=1, min_ridge_bars=1, min_classifiable=1 + ) + # the background days are unclassifiable -> not valid -> absent entirely + assert list(ratio.index) == [_TEST_DAY] + assert ratio.loc[_TEST_DAY] == pytest.approx(_CASE_A_RATIO) + + +def test_ridge_bar_counts_by_day_are_exposed_for_the_coverage_disclosure(): + """The runner must be able to REPORT the ridge-bar distribution, not just gate on it.""" + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", *_case_a_day()) + work = peak_mask_for_symbol( + prepare_visible_minute_bars( + _bars(rows), extra_columns=("amount",) + ).reset_index(drop=True) + ) + ratio, diag = valley_ridge_vwap_ratio_by_day( + work, min_valley_bars=1, min_ridge_bars=1, min_classifiable=1, + with_diagnostics=True, + ) + assert ratio.loc[_TEST_DAY] == pytest.approx(_CASE_A_RATIO) + assert int(diag.loc[_TEST_DAY, "ridge_bars"]) == 4 + assert int(diag.loc[_TEST_DAY, "valley_bars"]) == 11 + assert bool(diag.loc[_TEST_DAY, "valid"]) is True + # every classifiable day appears in the diagnostics, valid or not + assert len(diag) >= 1 + + +# --------------------------------------------------------------------------- # +# Guards / purity +# --------------------------------------------------------------------------- # +def test_empty_bars_yield_empty_schema_series(): + out = compute_valley_ridge_vwap_ratio(empty_intraday_bars()) + assert out.empty + assert list(out.index.names) == ["date", "symbol"] + assert out.name == "valley_ridge_vwap_ratio" + + +def test_input_bars_not_mutated(): + bars = _bars( + _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", *_case_a_day()) + ) + before = bars.copy(deep=True) + compute_valley_ridge_vwap_ratio(bars, **_KW) + pd.testing.assert_frame_equal(bars, before) + + +def test_bad_params_raise(): + bars = _bars(_session("2021-07-01", [100.0] * 3, [1000.0] * 3)) + with pytest.raises(ValueError): + compute_valley_ridge_vwap_ratio(bars, lookback_days=0) + with pytest.raises(ValueError): + compute_valley_ridge_vwap_ratio(bars, baseline_days=1) + with pytest.raises(ValueError): + compute_valley_ridge_vwap_ratio(bars, baseline_min_obs=1) + with pytest.raises(ValueError): + compute_valley_ridge_vwap_ratio(bars, sigma_k=-0.5) + with pytest.raises(ValueError): + compute_valley_ridge_vwap_ratio(bars, min_valid_days=0) + with pytest.raises(ValueError): + compute_valley_ridge_vwap_ratio(bars, min_classifiable=0) + with pytest.raises(ValueError): + compute_valley_ridge_vwap_ratio(bars, min_valley_bars=0) + with pytest.raises(ValueError): + compute_valley_ridge_vwap_ratio(bars, min_ridge_bars=0) + + +# --------------------------------------------------------------------------- # +# Spec + Factor subclass +# --------------------------------------------------------------------------- # +def test_factor_spec_is_valid_and_daily(): + spec = ValleyRidgeVwapRatioFactor().spec + assert isinstance(spec, FactorSpec) + assert spec.factor_id == "valley_ridge_vwap_ratio_20" + assert spec.is_intraday is False + assert spec.expected_ic_sign == 1 # POSITIVE (report RankIC +6.98%) + assert spec.return_basis == "close_to_close" + assert spec.forward_return_horizon == 1 + assert set(spec.input_fields) == {"volume", "amount"} + for field in ( + "decision_cutoff", "data_lag", "session_open", + "execution_model", "execution_window", + ): + assert getattr(spec, field) is None + + +def test_factor_spec_discloses_the_asymmetric_ridge_gate(): + """The scarcity-driven >= 10 ridge floor is a PINNED choice a reader must see.""" + desc = ValleyRidgeVwapRatioFactor().spec.description + assert "14:50" in desc + assert "RAW" in desc.upper() + assert str(VALLEY_RIDGE_MIN_RIDGE_BARS) in desc + assert str(VALLEY_RIDGE_MIN_VALLEY_BARS) in desc + assert "ridge" in desc.lower() and "valley" in desc.lower() + + +def test_factor_subclass_window_tracks_name(): + f = ValleyRidgeVwapRatioFactor(lookback_days=10) + assert f.name == "valley_ridge_vwap_ratio_10" + assert f.spec.factor_id == "valley_ridge_vwap_ratio_10" + + +def test_factor_compute_selects_preaggregated_column(): + f = ValleyRidgeVwapRatioFactor() + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp("2021-07-01"), _SYM)], names=["date", "symbol"] + ) + panel = pd.DataFrame({f.name: [0.98]}, index=idx) + out = f.compute(panel) + assert out.loc[(pd.Timestamp("2021-07-01"), _SYM)] == 0.98 + assert out.name == f.name + + +def test_factor_compute_missing_column_raises(): + f = ValleyRidgeVwapRatioFactor() + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp("2021-07-01"), _SYM)], names=["date", "symbol"] + ) + with pytest.raises(ValueError, match="pre-aggregated"): + f.compute(pd.DataFrame({"other": [1.0]}, index=idx)) + + +def test_factor_bad_params_raise(): + with pytest.raises(ValueError): + ValleyRidgeVwapRatioFactor(lookback_days=0)