diff --git a/config/phase_l_valley_price_quantile.yaml b/config/phase_l_valley_price_quantile.yaml new file mode 100644 index 0000000..8cf335e --- /dev/null +++ b/config/phase_l_valley_price_quantile.yaml @@ -0,0 +1,198 @@ +# PR-L — Tenth real factor evaluation: VALLEY WEIGHTED-PRICE-QUANTILE. +# +# Reproduces the SIXTH factor of the Kaiyuan market-microstructure series #27 (开源证券 +# 《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, reportId 4957417, §5) — the +# "量谷加权价格分位点因子" — as a first-class ValleyPriceQuantileFactor and runs it through the +# FROZEN StandardFactorEvaluator on REAL cached A-share data. CACHE-ONLY: the minute read is +# provably live-call-free; daily / universe / covariate endpoints go through the read-through +# cache (warm -> 0 gap fetches). +# +# WHY THIS FACTOR, NOW: the five prior reproductions from this report covered a COUNT (PR-F +# volume_peak_count, weak), a TIMING moment (PR-H peak_interval_kurtosis, null), two price +# RATIOS (PR-I valley_relative_vwap and PR-J valley_ridge_vwap_ratio, both PASSING) and a +# RETURN (PR-K ridge_minute_return, sign transferred but Reject). This factor is a price +# POSITION — where inside the day's range the valley VWAP sits — so it separates two +# explanations of PR-I/PR-J's success: the ratio FORM, or price-level INFORMATION generally. +# +# SAME MACHINE as PR-F / PR-H / PR-I / PR-J / PR-K. The minute classification is REUSED from +# data/clean/intraday_volume_prv.py (not re-implemented, and not modified by this PR): +# PIT-truncate each day at 14:50, classify every visible minute against its SAME-SLOT +# strictly-prior 20-day baseline (ERUPTIVE if vol > mu + sigma, else a VALLEY 量谷). Because the +# classification is shared verbatim, a different verdict here cannot be an artefact of a +# differently implemented taxonomy. +# +# PINNED choices (all disclosed on the factor spec; the report specifies only the shape): +# (1) THE PRICE RANGE reads three prices — the visible day's high, the visible day's low, +# and the PREVIOUS trading day's close — as hi = max(visible high, prev_close) and +# lo = min(visible low, prev_close). prev_close is pinned to the LAST VISIBLE (<=14:50) +# RAW CLOSE of the previous trading day. DISCLOSED DEVIATION from the report, which uses +# the true previous daily close. NOTE this is NOT a leakage fix — the previous day's +# 15:00 close is already past at a 14:50 decision on day d — it is a +# SINGLE-VISIBILITY-DEFINITION choice, so every price entering the factor comes from one +# source under one cutoff instead of mixing the minute cache with a daily-close feed; +# (2) the daily quantile (valley VWAP - lo)/(hi - lo) is NOT CLIPPED. A VWAP outside the +# range means the range construction is wrong, and clipping to [0,1] would hide exactly +# the defect worth seeing; +# (3) THE REVERSAL NEUTRALIZATION IS TAKEN AT T-1. The report residualizes this factor +# against a 20-day reversal because the raw quantile is strongly exposed to it. The naive +# -(close_d/close_{d-20} - 1) reads day d's CLOSE, which is 15:00 information and WOULD +# be a lookahead at our 14:50 decision, so we use rev20 = -(close_{d-1}/close_{d-21} - 1) +# on FRONT-ADJUSTED daily closes taken from the panel the runner already loads — NO new +# data source. Front-adjusted closes are REQUIRED here (unlike the within-day quantities) +# because the ratio spans 20 trading days. Day d's close is therefore never an input to +# day d's factor value, and a dedicated test perturbs it to prove so; +# (4) RAW minute prices for the range and the VWAP, with ONE DISCLOSED IMPERFECTION that +# does NOT apply to PR-I/PR-J/PR-K: those factors could argue the adjustment factor +# cancels because both legs sat inside ONE trading day, but prev_close crosses the +# overnight boundary, so on an ex-dividend / split date the previous raw close sits on +# the OLD (higher) price scale while the day's own bars sit on the new one. The bias is +# ONE-SIDED, not symmetric: prev_close pulls hi UP (measured on real cache, 42 +# top-widening vs 4 bottom-widening among true ex-dates), inflating the denominator +# hi - lo while the valley VWAP and lo stay on the new scale — which compresses that +# day's quantile toward the LOW END of an artificially wide range, NOT toward the +# middle. MEASURED on the 30-name prototype panel: 15.6% of valid days show ANY +# prev_close widening, but that is overwhelmingly ordinary overnight gapping, which is +# the INTENDED behaviour (the report's range includes the previous close precisely so a +# gap counts); only 0.73% of valid days are true ex-dates, and only ~0.21% are BOTH a +# true ex-date AND actually range-distorted, before the 20-day mean dilutes it further. +# DISCLOSED rather than silently corrected, because correcting it would mix an adjusted +# price into an otherwise raw, single-visibility construction; +# (5) a PRICE guard (finite, high >= low > 0) admits a bar to the range while a TRADE guard +# (finite positive volume AND amount) admits it to the VWAP — a zero-volume minute still +# quotes a real price. Both run at the summation step only, so PR-F's same-slot baseline +# — and therefore all five merged factors — stay bit-identical; +# (6) a day is VALID iff it has >= 100 classifiable bars AND >= 20 TRADABLE valley bars AND +# positive valley volume AND an available prev_close AND hi > lo, so a symbol's FIRST +# visible day never produces a value; NaN below 10 valid days; +# (7) the residualization is PER DATE on the covered cross-section (PR-G's pattern): fewer +# than 10 paired symbols, a missing rev20, or a degenerate (zero-variance) rev20 all +# yield NaN, never a zero-filled or unresidualized value passed off as neutralized. The +# runner REPORTS the before/after exposure and the realized cross-section sizes; +# (8) DEVIATION FROM THE REPORT, disclosed: everything spans the PIT-VISIBLE window +# 09:31-14:50 only, while the report uses the FULL day. Reading the closing auction would +# be lookahead at our 14:50 decision time. +# +# Pre-registered sign = +1 (report full-market RankIC +6.34% / RankICIR 4.32, long leg +# 13.1%/yr, long-short 20.22%/yr, IR 3.29, max drawdown 10.18%, monthly win rate 80.4%; +# CSI500 sub-domain long-short 11.71% / IR 1.76 — the closest comparable to our eval cell, +# quoted because the report actually gives one for this factor). Semantics per the report: a +# HIGH position of the calm-minute price within the day's range means the informed, unhurried +# part of the day traded near the top of the range -> higher future return. NOTE the report is +# a MONTHLY, market-cap + industry neutral FULL-MARKET series on Wind data — our eval cell is +# CSI500 daily with industry + size neutral, so those numbers are a LOOSE reference only and +# are NOT written in as expected values. Raw minute volume (cached as-is) has split-day +# magnitude jumps that pollute the 20-day sigma; the report (Wind) does not adjust for this +# either, so it is disclosed and NOT corrected. +# +# SIGN = +1 means this factor is NOT affected by the frozen layer's aligned-spread cost-sign +# defect (which computes sign * (gross - cost) and therefore ADDS costs back only when the sign +# is negative). Unlike PR-K, the aligned_spread_* fields may be read directly here. +# net_long_short_by_cost remains the primary tradability read, for sibling comparability. +# +# Eval CELL is IDENTICAL to PR-C .. PR-K: universe = CSI500 (000905.SH), PIT membership; window +# = 2021-07-01 .. 2026-06-30 (the project's minute-coverage window); daily rebalance; OOS split +# 2024-01-01; book = value_ep/value_bp/volatility_20; fee 0.001. The ONLY substantive difference +# is the subject factor, which is minute-derived and computed by the runner from the intraday +# cache — it is NOT a config-listed daily factor. Because the cell matches its siblings exactly, +# this run is directly comparable to them. The evaluator runs TWICE (see +# qt/eval_valley_price_quantile.py): once with NO book (Incremental NOT_ASSESSED) and once with +# the confirmed book — which matters here because a within-range price POSITION is a plausible +# cousin of a value signal, and the with-book run is what says whether this is a new bet. + +project: + name: quantitative_trading_pr_l_valley_price_quantile + 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_price_quantile_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_price_quantile_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. This is SEPARATE from, and applied AFTER, the + # factor's own reversal neutralization (which is part of the factor definition). + 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_quantile.py b/data/clean/intraday_valley_quantile.py new file mode 100644 index 0000000..c12434e --- /dev/null +++ b/data/clean/intraday_valley_quantile.py @@ -0,0 +1,566 @@ +"""VALLEY WEIGHTED-PRICE-QUANTILE factor (PR-L). + +Reproduces the SIXTH factor of the Kaiyuan market-microstructure series #27 (开源证券 +《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, 2025-07-20, reportId 4957417, +§5): "将『日内最高价、日内最低价、昨日收盘价』三者的最高值与最低值作为区间的价格高低位，并计算 +每日量谷成交量加权价格的相对分位点，将 20 日的分位点均值作为量谷加权价格分位点因子…考虑到因子 +计算中正向暴露 20 日反转因子，我们进一步对该因子做反转中性化处理". + +Same MACHINE as PR-F / PR-H / PR-I / PR-J / PR-K, a new STATISTIC FAMILY. The minute +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. A VALLEY (量谷) is exactly +that module's ``valley`` column. Where PR-F counted peaks, PR-H measured their timing, +PR-I / PR-J took price RATIOS and PR-K a RETURN, this module measures a price POSITION: +WHERE in the day's price range the valley VWAP sits. That is the scientific point of the +run — PR-I and PR-J both passed, and a position statistic tests whether that success came +from the ratio FORM or from price-level INFORMATION generally. + +Definition, per symbol and per panel date ``d`` (a DAILY signal traded close-to-close, +``is_intraday=False`` like all eight precedents): + + 1. PIT truncation (standing authorization): each day keeps only the 1min bars with + ``available_time <= (that bar's trade_date + decision_time)`` (default 14:50). + 2. PRICE RANGE from three prices — the visible day's high, the visible day's low, and + the PREVIOUS trading day's close: ``hi = max(visible high, prev_close)`` and + ``lo = min(visible low, prev_close)``. + 3. VALLEY VWAP = ``Σamount(valley bars) / Σvolume(valley bars)`` (PR-I's identity). + 4. DAILY QUANTILE ``q_day = (valley VWAP - lo) / (hi - lo)``, deliberately NOT clipped. + 5. TRAILING MEAN ``qbar`` over the most recent ``lookback_days`` (=20) VALID days + including ``d``; fewer than ``min_valid_days`` (=10) valid days -> NaN. + 6. REVERSAL NEUTRALIZATION: per panel date, the cross-sectional OLS residual of + ``qbar`` on a 20-day reversal factor. + +PINNED choices (deliberate and DISCLOSED, reproduced on the factor spec): + + 1. PREV_CLOSE IS THE LAST VISIBLE (<=14:50) RAW CLOSE OF THE PREVIOUS TRADING DAY, not + that day's true 15:00 daily close. DEVIATION FROM THE REPORT, which uses the real + previous close. The previous day's 15:00 close is NOT itself a lookahead at a 14:50 + decision on day d — it is already past — so this is not a leakage fix; it is a + SINGLE-VISIBILITY-DEFINITION choice, so that every price entering the factor (the + range, the VWAP, the previous close) comes from one source under one cutoff rather + than mixing a minute-cache window with a daily-close feed. Disclosed, never silently + equated to the report's construction. + 2. NO CLIPPING of ``q_day``. If the valley VWAP falls outside ``[lo, hi]`` the range + construction is wrong, and clipping to [0,1] would hide exactly the defect worth + seeing. Under a correct range a traded price cannot escape it, so an out-of-range + value is a signal to investigate, not a value to sanitize. + 3. THE REVERSAL IS TAKEN AT T-1: ``rev20 = -(close_{d-1}/close_{d-21} - 1)`` on + FRONT-ADJUSTED (qfq) daily closes. The report's naive 20-day reversal uses + ``close_d``, which is 15:00 information and WOULD be a lookahead at our 14:50 + decision time. This is the direct application of the standing authorization that + implicitly-leaking fields be taken from pre-14:50 data. The closes come from the + panel the runner already loads — NO new data source. Front-adjusted closes are + REQUIRED here (unlike the within-day quantities below) because the ratio spans 20 + trading days and would otherwise break across any split or dividend. + 4. RAW (UNADJUSTED) MINUTE PRICES for the range and the VWAP, with ONE DISCLOSED + IMPERFECTION. PR-I/PR-J/PR-K could argue the adjustment factor cancels because both + of their legs sat inside ONE trading day. That argument does NOT fully hold here: + ``prev_close`` crosses the overnight boundary, so on an EX-DIVIDEND / SPLIT date the + previous raw close sits on the OLD (higher) price scale while the day's own bars sit + on the new one. The DIRECTION is therefore one-sided, not symmetric: ``prev_close`` + pulls ``hi`` UP (measured on real cache: 42 top-widening vs 4 bottom-widening among + true ex-dates), inflating the denominator ``hi - lo`` while the valley VWAP and + ``lo`` stay on the new scale — which compresses that day's ``q_day`` toward the LOW + END of an artificially wide range, NOT toward the middle. + MEASURED on the 30-name prototype panel: 15.6% of valid days show ANY range + widening from ``prev_close``, but that is overwhelmingly ordinary overnight gapping, + which is the INTENDED behaviour (the report's range includes the previous close + precisely so a gap counts); only 0.73% of valid days are true ex-dates, and only + ~0.21% are BOTH a true ex-date AND actually range-distorted — before the 20-day mean + dilutes it further. DISCLOSED here rather than silently corrected, because + correcting it would mix an adjusted price into an otherwise raw, single-visibility + construction. + 5. THE RANGE USES A PRICE GUARD, THE VWAP USES A TRADE GUARD. A bar enters the + high/low range if its prices are finite with ``high >= low > 0`` (a zero-volume + minute still quotes a real price); it enters the VWAP only if BOTH ``volume`` and + ``amount`` are finite and strictly positive (PR-I's positive-trade guard). Both + guards run at the summation step only, never before classification, so PR-F's + same-slot baseline — and therefore all five merged factors — stay bit-identical. + 6. DAY VALIDITY needs all of: >= ``min_classifiable`` (=100) classifiable bars (PR-F's + gate, unchanged), >= ``min_valley_bars`` (=20) TRADABLE valley bars (PR-I's floor), + strictly positive valley volume, an available ``prev_close``, and ``hi > lo``. A + symbol's FIRST visible day therefore never produces a value (no previous day). + 7. THE RESIDUALIZATION IS PER DATE ON THE COVERED CROSS-SECTION, mirroring PR-G's + per-date cross-sectional post-processing. A date with fewer than + ``min_cross_section`` (=10) symbols carrying BOTH ``qbar`` and ``rev20`` is all-NaN; + a symbol missing ``rev20`` is dropped from the regression and its residual is NaN + (never a silent zero fill); a degenerate cross-section (zero-variance ``rev20``, + which cannot identify a slope) is likewise NaN rather than an unresidualized + ``qbar`` passed off as neutralized. + 8. EVERYTHING SPANS THE PIT-VISIBLE WINDOW 09:31-14:50 ONLY, while the report uses the + full session. Reading the closing auction would be lookahead at our decision time. + Disclosed, not silently equated. + +Pre-registered sign = +1. The report's full-market RankIC is +6.34% / RankICIR 4.32 +(long leg 13.1%/yr, long-short 20.22%/yr, IR 3.29, max drawdown 10.18%, monthly win rate +80.4%); its CSI500 sub-domain long-short is 11.71% / IR 1.76, the closest comparable to +our eval cell. Semantics per the report: a HIGH relative position of the calm-minute +price within the day's range means the informed, unhurried part of the day traded near +the top of the range -> higher future return. 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 and closes at dates <= d-1, so a factor +value never sees a future bar (invariant #1). This module is DATA-layer only: it does not +fetch, does not touch factors / alpha / portfolio / runtime, and never sees a token. + +The heavy per-symbol work (steps 1-5) is split from the cross-sectional residualization +(step 6) on purpose — the same split PR-G uses: the runner streams one symbol at a time +through :func:`compute_valley_price_quantile_stats` (memory-bounded), assembles the +full-universe ``qbar`` panel, computes ``rev20`` once from the daily panel, then calls +:func:`residualize_on_reversal` ONCE. :func:`compute_valley_price_quantile` chains the +three for a single-call path. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + +from data.clean.intraday_aggregate import ( + DAILY_INDEX_NAMES, + DATE_LEVEL, + 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_QUANTILE_LOOKBACK_DAYS = 20 # trailing VALID trading-day mean window, includes d +VALLEY_QUANTILE_MIN_VALLEY_BARS = 20 # min TRADABLE valley bars for a valid day (PR-I's) +VALLEY_QUANTILE_REVERSAL_DAYS = 20 # span of the reversal factor neutralized against +VALLEY_QUANTILE_MIN_CROSS_SECTION = 10 # min paired cross-section for a finite residual + +# The extra 1min columns this family needs on top of PR-F's (volume): the traded value +# for the VWAP, and the prices that form the day's range / the previous close. +_EXTRA_COLUMNS = ("amount", "high", "low", "close") + + +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_price_quantile_by_day( + work: pd.DataFrame, + *, + min_valley_bars: int = VALLEY_QUANTILE_MIN_VALLEY_BARS, + min_classifiable: int = VOLUME_PRV_MIN_CLASSIFIABLE, +) -> pd.Series: + """Daily valley-VWAP price quantile 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`, built from bars + prepared with ``extra_columns=("amount", "high", "low", "close")`` so the traded + value and the prices ride alongside the ``valley`` / ``classifiable`` masks. + + Implements pinned choices 1-6 of the module docstring: the range is + ``[min(visible low, prev_close), max(visible high, prev_close)]`` where ``prev_close`` + is the previous trading day's LAST VISIBLE usable close; the valley VWAP is + ``Σamount/Σvolume`` over guarded valley bars; the quantile is NOT clipped. + + Returns: + Series indexed by ``trade_date`` (ascending) holding ``q_day`` for the days that + clear every validity gate — invalid days are ABSENT, not NaN, so they do not + occupy a slot in the caller's trailing window (the rule PR-F..PR-K use). + """ + # Chronological within the symbol so "the day's last visible close" is well defined. + ordered = work.sort_values("bar_end", kind="mergesort") + + vol = ordered["volume"].to_numpy(dtype=float) + amt = ordered["amount"].to_numpy(dtype=float) + high = ordered["high"].to_numpy(dtype=float) + low = ordered["low"].to_numpy(dtype=float) + close = ordered["close"].to_numpy(dtype=float) + + # PINNED §5: a TRADE guard for the VWAP, a PRICE guard for the range. A zero-volume + # minute still quotes a real price, so it may set the day's high/low while + # contributing nothing to the volume-weighted price. + tradable = np.isfinite(vol) & (vol > 0.0) & np.isfinite(amt) & (amt > 0.0) + priced = np.isfinite(high) & np.isfinite(low) & (low > 0.0) & (high >= low) + valley = ordered["valley"].to_numpy(dtype=bool) & tradable + + per_bar = pd.DataFrame( + { + "trade_date": ordered["trade_date"].to_numpy(), + "valley_amt": np.where(valley, amt, 0.0), + "valley_vol": np.where(valley, vol, 0.0), + "valley_bars": valley.astype(np.int64), + "classifiable_bars": ordered["classifiable"] + .to_numpy(dtype=bool) + .astype(np.int64), + # +/-inf so a day with no priced bar reduces to a non-finite range and fails + # the hi > lo gate rather than fabricating one. + "day_high": np.where(priced, high, -np.inf), + "day_low": np.where(priced, low, np.inf), + } + ) + grouped = per_bar.groupby("trade_date", sort=True) + agg = grouped[ + ["valley_amt", "valley_vol", "valley_bars", "classifiable_bars"] + ].sum() + agg["day_high"] = grouped["day_high"].max() + agg["day_low"] = grouped["day_low"].min() + + # PINNED §1: prev_close = the previous trading day's LAST VISIBLE usable close. Taken + # from the SAME visible (<=14:50) bars as everything else — a post-cutoff bar on the + # previous day can never become it. + usable_close = np.isfinite(close) & (close > 0.0) + closes = pd.DataFrame( + { + "trade_date": ordered["trade_date"].to_numpy()[usable_close], + "close": close[usable_close], + } + ) + last_close = closes.groupby("trade_date", sort=True)["close"].last() + # shift(1) over the symbol's own trading days: day d takes day d-1's last visible + # close, and the FIRST day has none (NaN) -> that day is invalid. + prev_close = last_close.reindex(agg.index).shift(1) + + pc = prev_close.to_numpy(dtype=float) + hi = np.maximum(agg["day_high"].to_numpy(dtype=float), pc) + lo = np.minimum(agg["day_low"].to_numpy(dtype=float), pc) + + valid = ( + (agg["classifiable_bars"].to_numpy() >= min_classifiable) + & (agg["valley_bars"].to_numpy() >= min_valley_bars) + & (agg["valley_vol"].to_numpy(dtype=float) > 0.0) + & np.isfinite(pc) + & np.isfinite(hi) + & np.isfinite(lo) + & (hi > lo) + ) + if not valid.any(): + return pd.Series( + [], index=pd.DatetimeIndex([], name="trade_date"), dtype=float + ) + + valley_vwap = ( + agg["valley_amt"].to_numpy(dtype=float)[valid] + / agg["valley_vol"].to_numpy(dtype=float)[valid] + ) + # PINNED §2: NOT clipped — an out-of-range value must be visible, not sanitized. + q = (valley_vwap - lo[valid]) / (hi[valid] - lo[valid]) + return pd.Series(q, index=agg.index[valid], dtype=float) + + +def _quantile_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, +) -> tuple[list[pd.Timestamp], list[float]]: + """Trailing-mean daily quantile 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 price quantile, 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, the previous close is the + previous day's). + """ + work = peak_mask_for_symbol( + g, + baseline_days=baseline_days, + baseline_min_obs=baseline_min_obs, + sigma_k=sigma_k, + ) + q = valley_price_quantile_by_day( + work, min_valley_bars=min_valley_bars, min_classifiable=min_classifiable + ) + if q.empty: + return [], [] + rolled = q.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)) + + +def compute_valley_price_quantile_stats( + bars: pd.DataFrame, + *, + lookback_days: int = VALLEY_QUANTILE_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_QUANTILE_MIN_VALLEY_BARS, + decision_time: str = DEFAULT_DECISION_TIME, + name: str = "valley_price_quantile_raw", +) -> pd.Series: + """RAW trailing-mean price quantile panel (steps 1-5; NO neutralization yet). + + 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 quantile of the valley VWAP within the prev-close-extended price range, and + returns the trailing-``lookback_days``-VALID-day mean of that quantile. + + The REVERSAL NEUTRALIZATION (step 6) is deliberately NOT applied here — it needs the + full-universe panel plus daily closes and is done by :func:`residualize_on_reversal`. + The returned values are therefore the RAW ``qbar``, which is exactly what the + prototype's before/after neutralization comparison needs. + + 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. + decision_time: per-bar PIT cutoff time-of-day (default 14:50:00). + name: the returned Series name. + + Returns: + ``MultiIndex(date, symbol)`` Series (midnight-normalized dates) of the raw + trailing-mean quantile, 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 len(bars) == 0: + return _empty_series(name) + + visible = prepare_visible_minute_bars( + bars, decision_time=decision_time, extra_columns=_EXTRA_COLUMNS + ) + if visible.empty: + return _empty_series(name) + + index_tuples: list[tuple] = [] + values: list[float] = [] + for sym, g in visible.groupby(SYMBOL_LEVEL, sort=True): + days, vals = _quantile_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, + ) + 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() + + +def reversal_20( + closes: pd.DataFrame | pd.Series, + *, + days: int = VALLEY_QUANTILE_REVERSAL_DAYS, + close_col: str = "close", + name: str = "rev20", +) -> pd.Series: + """T-1-based ``days``-day reversal factor from FRONT-ADJUSTED daily closes. + + ``rev20(d) = -(close_{d-1} / close_{d-(days+1)} - 1)`` — PINNED §3 of the module + docstring. The report's naive form uses ``close_d``, which is 15:00 information and + would be a LOOKAHEAD at our 14:50 decision time; taking the whole ratio one day + earlier makes every input strictly visible when the factor is formed. + + The closes MUST be front-adjusted: the ratio spans ``days`` trading days and a raw + series would break across any split or dividend inside the window. + + Args: + closes: ``MultiIndex(date, symbol)`` daily panel (DataFrame with ``close_col``, + or a Series of closes). Typically ``panel[["close"]]`` from the runner — + NO new data source. + days: reversal span in trading days (definition, not a tuned knob). + close_col: the close column name when ``closes`` is a DataFrame. + name: the returned Series name. + + Returns: + ``MultiIndex(date, symbol)`` Series over the SAME rows as the input, NaN where + the T-1 window is incomplete or a close is non-positive / non-finite (honest + missing — never fabricated). Pure: never mutates ``closes``. + """ + if days < 1: + raise ValueError(f"days must be >= 1; got {days!r}.") + if isinstance(closes, pd.DataFrame): + if close_col not in closes.columns: + raise ValueError( + f"reversal_20 needs a '{close_col}' column; got {list(closes.columns)}." + ) + series = closes[close_col] + else: + series = closes + if series.empty: + return _empty_series(name) + + s = series.astype(float).sort_index() + # Non-positive / non-finite closes cannot form a ratio; blank them BEFORE shifting so + # a bad print poisons only the windows that actually touch it. + clean = s.where(np.isfinite(s.to_numpy(dtype=float)) & (s.to_numpy(dtype=float) > 0.0)) + by_symbol = clean.groupby(level=SYMBOL_LEVEL, sort=False) + prior = by_symbol.shift(1) # close_{d-1} + base = by_symbol.shift(days + 1) # close_{d-(days+1)} + rev = -(prior / base - 1.0) + return rev.rename(name) + + +def residualize_on_reversal( + qbar: pd.Series, + rev: pd.Series, + *, + min_cross_section: int = VALLEY_QUANTILE_MIN_CROSS_SECTION, + name: str = "valley_price_quantile", +) -> pd.Series: + """Per-date cross-sectional OLS residual of ``qbar`` on ``rev`` (step 6). + + The report neutralizes this factor against the 20-day reversal because the raw + quantile is positively exposed to it. For each panel date the cross-section is the + symbols carrying BOTH a finite ``qbar`` and a finite ``rev``; the residual is + ``qbar - (a + b*rev)`` from an intercept OLS fit on that cross-section. Mirrors PR-G's + per-date cross-sectional post-processing, and follows the project's neutralization + convention that an under-determined cross-section yields NaN rather than a fabricated + value. + + PINNED §7: a date with fewer than ``min_cross_section`` paired symbols is all-NaN; a + symbol missing ``rev`` is dropped from the fit and its residual is NaN (never a silent + zero fill); a degenerate cross-section (zero-variance ``rev``, which cannot identify a + slope) is likewise NaN rather than an unresidualized ``qbar`` passed off as + neutralized. + + Args: + qbar: ``MultiIndex(date, symbol)`` raw trailing-mean quantile panel. + rev: ``MultiIndex(date, symbol)`` reversal panel (from :func:`reversal_20`). + min_cross_section: minimum paired cross-section for a finite residual. + name: the returned Series name (the factor-panel column name). + + Returns: + A Series over EXACTLY ``qbar``'s rows, in ``qbar``'s order, holding the residual + (or NaN). Pure: never mutates its inputs. + """ + if min_cross_section < 2: + # Need >= 2 cross-section members before an intercept + slope fit is determined. + raise ValueError( + f"min_cross_section must be >= 2; got {min_cross_section!r}." + ) + if qbar is None or qbar.empty: + return _empty_series(name) + + y_all = qbar.to_numpy(dtype=float) + # Align the reversal onto qbar's rows WITHOUT reordering them (the output must line + # up with the caller's panel row-for-row). + x_all = rev.reindex(qbar.index).to_numpy(dtype=float) + out = np.full(y_all.shape, np.nan, dtype=float) + + dates = qbar.index.get_level_values(DATE_LEVEL).to_numpy() + paired = np.isfinite(y_all) & np.isfinite(x_all) + for date in pd.unique(dates): + on_date = dates == date + fit_rows = on_date & paired + n = int(fit_rows.sum()) + if n < min_cross_section: + continue # honest missing: the whole date stays NaN + x = x_all[fit_rows] + y = y_all[fit_rows] + x_mean = x.mean() + sxx = float(((x - x_mean) ** 2).sum()) + if not np.isfinite(sxx) or sxx <= 0.0: + continue # degenerate: no slope is identified -> NaN, not a raw qbar + y_mean = y.mean() + slope = float(((x - x_mean) * (y - y_mean)).sum()) / sxx + out[fit_rows] = y - (y_mean + slope * (x - x_mean)) + + return pd.Series(out, index=qbar.index, name=name) + + +def compute_valley_price_quantile( + bars: pd.DataFrame, + closes: pd.DataFrame | pd.Series, + *, + lookback_days: int = VALLEY_QUANTILE_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_QUANTILE_MIN_VALLEY_BARS, + min_cross_section: int = VALLEY_QUANTILE_MIN_CROSS_SECTION, + reversal_days: int = VALLEY_QUANTILE_REVERSAL_DAYS, + decision_time: str = DEFAULT_DECISION_TIME, + name: str = "valley_price_quantile", +) -> pd.Series: + """Single-call factor: raw quantile stats (1-5) then reversal neutralization (6). + + Convenience that chains :func:`compute_valley_price_quantile_stats`, + :func:`reversal_20` and :func:`residualize_on_reversal`. The runner does NOT use this + (it needs the per-symbol stats loop for memory-boundedness, and it reports the + before/after-neutralization diagnostics) but tests and any all-in-memory caller can. + + ``closes`` must be the FRONT-ADJUSTED daily close panel — see PINNED §3. Pure: never + mutates its inputs. + """ + stats = compute_valley_price_quantile_stats( + 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, + decision_time=decision_time, + ) + if stats.empty: + return _empty_series(name) + rev = reversal_20(closes, days=reversal_days) + return residualize_on_reversal( + stats, rev, min_cross_section=min_cross_section, name=name + ) + + +__all__ = [ + "VALLEY_QUANTILE_LOOKBACK_DAYS", + "VALLEY_QUANTILE_MIN_CROSS_SECTION", + "VALLEY_QUANTILE_MIN_VALLEY_BARS", + "VALLEY_QUANTILE_REVERSAL_DAYS", + "compute_valley_price_quantile", + "compute_valley_price_quantile_stats", + "residualize_on_reversal", + "reversal_20", + "valley_price_quantile_by_day", +] diff --git a/factors/compute/intraday_derived.py b/factors/compute/intraday_derived.py index b3c5148..3493115 100644 --- a/factors/compute/intraday_derived.py +++ b/factors/compute/intraday_derived.py @@ -62,6 +62,12 @@ RIDGE_RETURN_LOOKBACK_DAYS, RIDGE_RETURN_MIN_RIDGE_BARS, ) +from data.clean.intraday_valley_quantile import ( + VALLEY_QUANTILE_LOOKBACK_DAYS, + VALLEY_QUANTILE_MIN_CROSS_SECTION, + VALLEY_QUANTILE_MIN_VALLEY_BARS, + VALLEY_QUANTILE_REVERSAL_DAYS, +) from data.clean.intraday_valley_ridge_vwap import ( VALLEY_RIDGE_LOOKBACK_DAYS, VALLEY_RIDGE_MIN_RIDGE_BARS, @@ -1023,6 +1029,152 @@ def compute(self, panel: pd.DataFrame) -> pd.Series: return panel[self.name].rename(self.name) +class ValleyPriceQuantileFactor(Factor): + """Valley weighted-price-quantile factor (daily signal, minute-derived). + + ``compute`` reads the pre-aggregated daily column the runner placed on the panel + (produced by + ``data.clean.intraday_valley_quantile.compute_valley_price_quantile_stats`` and then + reversal-neutralized by ``residualize_on_reversal``); it does NO minute work of its + own, mirroring the eight precedents in this module. + + 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_price_quantile_{VALLEY_QUANTILE_LOOKBACK_DAYS}" + + def __init__(self, lookback_days: int = VALLEY_QUANTILE_LOOKBACK_DAYS) -> None: + if not isinstance(lookback_days, int) or lookback_days < 1: + raise ValueError( + f"valley-price-quantile lookback_days must be a positive integer; got " + f"{lookback_days!r}." + ) + self._lookback_days = lookback_days + self.name = f"valley_price_quantile_{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 (report full-market RankIC +6.34%, RankICIR 4.32, long leg + 13.1%/yr, long-short 20.22%/yr, IR 3.29, max drawdown 10.18%, monthly win rate + 80.4%; CSI500 sub-domain long-short 11.71% / IR 1.76 — the closest comparable to + our eval cell). Semantics per the report: a HIGH position of the calm-minute + (valley) price within the day's range means the informed, unhurried part of the + day traded near the top of the range -> higher future return. 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: minute INPUT but a DAILY signal traded + close-to-close. min_history_bars=0: the warm-up is DATA-dependent, not a fixed + leading count — the honest NaN rate is reported by data_coverage. + + The description spells out the two subtleties that make this factor harder than + its five siblings: the PREV_CLOSE-extended range read at OUR 14:50 visibility + rather than the report's true daily close, and the reversal neutralization taken + at T-1 so day d's 15:00 close is never an input to day d's value. + """ + return FactorSpec( + factor_id=self.name, + version="1.0", + description=( + f"Valley weighted-price-quantile (Kaiyuan microstructure series #27, " + f"SIXTH factor 量谷加权价格分位点). SAME minute classification as PR-F " + f"volume_peak_count / PR-H peak_interval_kurtosis / PR-I " + f"valley_relative_vwap / PR-J valley_ridge_vwap_ratio / PR-K " + f"ridge_minute_return (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, and a VALLEY (量谷) is a " + f"classifiable NON-eruptive minute. NEW STATISTIC vs PR-F..PR-K: a price " + f"POSITION rather than a count, a timing moment, a price RATIO or a " + f"return — each valid day scores WHERE the valley VWAP sits inside the " + f"day's price range, and the factor averages that over the trailing " + f"{self._lookback_days} VALID days before a cross-sectional reversal " + f"neutralization. PINNED choices: (1) the range is [min(visible low, " + f"prev_close), max(visible high, prev_close)] where prev_close is the " + f"LAST VISIBLE (<=14:50) RAW CLOSE of the PREVIOUS trading day — a " + f"DISCLOSED DEVIATION from the report, which uses the true previous " + f"daily close; the previous 15:00 close is not itself a lookahead at a " + f"14:50 decision, so this is a SINGLE-VISIBILITY-DEFINITION choice (every " + f"price in the factor comes from one source under one cutoff), not a " + f"leakage fix; (2) the daily quantile (valley VWAP - lo)/(hi - lo) is NOT " + f"clipped — a VWAP outside the range means the range is wrong and must be " + f"visible rather than sanitized; (3) THE REVERSAL NEUTRALIZATION IS TAKEN " + f"AT T-1: rev20 = -(close_(d-1)/close_(d-{VALLEY_QUANTILE_REVERSAL_DAYS + 1}) " + f"- 1) on FRONT-ADJUSTED daily closes from the panel (NO new data source). " + f"The report's naive form uses close_d, which is 15:00 information and " + f"WOULD be a lookahead at our 14:50 decision — day d's close is therefore " + f"never an input to day d's factor value; front-adjusted closes are " + f"required because the ratio spans {VALLEY_QUANTILE_REVERSAL_DAYS} " + f"trading days; (4) RAW minute prices for the range and the VWAP, with a " + f"DISCLOSED imperfection: prev_close crosses the overnight boundary, so on " + f"an ex-dividend / split date the previous raw close sits on the OLD " + f"(higher) scale while the day's own bars sit on the new one. The bias is " + f"ONE-SIDED: prev_close pulls hi UP (measured 42 top-widening vs 4 " + f"bottom-widening among true ex-dates), inflating hi - lo while the valley " + f"VWAP and lo stay on the new scale, which compresses that day's quantile " + f"toward the LOW END of an artificially wide range — NOT toward the middle. " + f"Measured: 15.6% of valid days show any prev_close widening (overwhelmingly " + f"ordinary overnight gaps, which is INTENDED), only 0.73% are true ex-dates " + f"and only ~0.21% are both a true ex-date AND range-distorted, before the " + f"{self._lookback_days}-day mean dilutes it; disclosed rather than silently " + f"corrected; (5) a PRICE guard (finite, high >= low > 0) admits a bar to " + f"the range while a TRADE guard (finite positive volume AND amount) " + f"admits it to the VWAP, both applied at the summation step only so PR-F's " + f"baseline stays bit-identical; (6) a day is VALID iff it has >= " + f"{VOLUME_PRV_MIN_CLASSIFIABLE} classifiable bars AND >= " + f"{VALLEY_QUANTILE_MIN_VALLEY_BARS} TRADABLE valley bars AND positive " + f"valley volume AND an available prev_close AND hi > lo — so a symbol's " + f"FIRST visible day never produces a value; (7) the residualization is " + f"PER DATE on the covered cross-section, with < " + f"{VALLEY_QUANTILE_MIN_CROSS_SECTION} paired symbols, a missing rev20, or " + f"a degenerate (zero-variance) rev20 all yielding NaN rather than a " + f"zero-filled or unresidualized value passed off as neutralized; (8) " + f"DEVIATION FROM THE REPORT, disclosed: everything spans the PIT-VISIBLE " + f"window 09:31-14:50 only, not the full session. 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, plus the daily + # close the T-1 reversal neutralization regresses against. Declared for honest + # provenance disclosure (data_coverage lists them); the daily panel surfaces + # the pre-aggregated column itself. + input_fields=("volume", "amount", "high", "low", "close"), + family="microstructure", + min_history_bars=0, + ) + + def compute(self, panel: pd.DataFrame) -> pd.Series: + """Select the pre-aggregated daily valley-price-quantile column off ``panel``. + + The runner runs ``compute_valley_price_quantile_stats`` per symbol on the minute + cache and ``residualize_on_reversal`` once on the assembled panel upstream, then + 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"ValleyPriceQuantileFactor needs the pre-aggregated '{self.name}' " + f"column on the panel (produced upstream by " + f"compute_valley_price_quantile_stats + residualize_on_reversal and " + f"joined by the runner); panel has {list(panel.columns)}." + ) + return panel[self.name].rename(self.name) + + __all__ = [ "AmpMarginalAnomalyVolFactor", "IntradayAmpCutFactor", @@ -1030,6 +1182,7 @@ def compute(self, panel: pd.DataFrame) -> pd.Series: "MinuteIdealAmplitudeFactor", "PeakIntervalKurtosisFactor", "RidgeMinuteReturnFactor", + "ValleyPriceQuantileFactor", "ValleyRelativeVwapFactor", "ValleyRidgeVwapRatioFactor", "VolumePeakCountFactor", diff --git a/qt/cli.py b/qt/cli.py index 55a2f0d..eea9e36 100644 --- a/qt/cli.py +++ b/qt/cli.py @@ -436,6 +436,56 @@ def _cmd_run_eval_ridge_minute_return(args: argparse.Namespace) -> int: return 0 +def _cmd_run_eval_valley_price_quantile(args: argparse.Namespace) -> int: + """Run the two real valley-price-quantile evaluations (cache-only) + reports.""" + from qt.eval_valley_price_quantile import run_eval_valley_price_quantile + + try: + result = run_eval_valley_price_quantile(args.config) + except (ConfigError, ValueError, FileNotFoundError) as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 1 + nb, wb = result.no_book_metrics, result.with_book_metrics + net = ( + " ".join( + f"{m:g}x={v:+.6f}" for m, v in sorted(nb["net_long_short_by_cost"].items()) + ) + or "n/a" + ) + cov = result.neutralization + print( + f"OK run-eval-valley-price-quantile: 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" + # The reversal neutralization is the structural novelty of this factor; a + # neutralization that silently ate the panel shows up here as a number. + f"neutralization (T-1 rev20): raw_rows={cov.raw_rows} " + f"rev_paired={cov.rev_rows} residual_rows={cov.residual_rows} " + f"dates={cov.dates_residualized}/{cov.dates_total} " + f"cross_section min/med/max={cov.cross_section_min}/" + f"{cov.cross_section_median:.1f}/{cov.cross_section_max} " + f"mean_spearman(raw,rev20)={cov.raw_rev_spearman_mean:+.4f}\n" + f"no-book: {nb['deployment']} (predictive={nb['predictive']}) " + f"ic_mean={nb['ic_mean']:.4f} ic_ir={nb['ic_ir']:.3f} N_eff={nb['effective_samples']:.1f}\n" + f"with-book: {wb['deployment']} (incremental={wb['incremental']}) " + f"incr_ic_ir={wb['incremental_ic_ir']:.3f}\n" + # sign=+1, so the frozen layer's aligned-spread cost-sign defect does NOT apply + # to this factor; net spreads are still printed for sibling comparability. + f"net long-short by cost: {net}\n" + # The PR-K review's regularity, first tested here on a positive-sign factor. + f"ic_rank={_fmt_metric(nb['ic_mean'])} " + f"ic_pearson={_fmt_metric(nb['ic_pearson_mean'])} " + f"monotonicity={_fmt_metric(nb['monotonicity_spearman'])}\n" + f"turnover={_fmt_metric(nb['long_short_turnover'])} " + f"rank_autocorr_lag1={_fmt_metric(nb['rank_autocorr_lag1'])} " + f"half_life={_fmt_metric(nb['half_life_periods'], '.2f')}\n" + f"reports: {result.reports.no_book_md} | {result.reports.with_book_md}\n" + f"dashboards: {result.reports.no_book_dashboard} | " + f"{result.reports.with_book_dashboard}" + ) + return 0 + + def _cmd_run_eval_valley_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 @@ -658,6 +708,13 @@ def build_parser() -> argparse.ArgumentParser: p_rmr.add_argument("--config", required=True, help="Path to the YAML config.") p_rmr.set_defaults(func=_cmd_run_eval_ridge_minute_return) + p_vpq = sub.add_parser( + "run-eval-valley-price-quantile", + help="Run the valley-price-quantile factor evaluation (CSI500, cache-only).", + ) + p_vpq.add_argument("--config", required=True, help="Path to the YAML config.") + p_vpq.set_defaults(func=_cmd_run_eval_valley_price_quantile) + 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_price_quantile.py b/qt/eval_valley_price_quantile.py new file mode 100644 index 0000000..99cd152 --- /dev/null +++ b/qt/eval_valley_price_quantile.py @@ -0,0 +1,683 @@ +"""run-eval-valley-price-quantile: the tenth real factor evaluation (PR-L). + +Reproduces the SIXTH factor of the Kaiyuan market-microstructure series #27 (开源证券 +《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, reportId 4957417, §5) — the +"量谷加权价格分位点" factor — as a first-class +:class:`~factors.compute.intraday_derived.ValleyPriceQuantileFactor` 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-K used. + +WHY THIS FACTOR, NOW: the five prior reproductions from this report covered a COUNT (PR-F +volume_peak_count, weak), a TIMING moment (PR-H peak_interval_kurtosis, null), two price +RATIOS (PR-I valley_relative_vwap and PR-J valley_ridge_vwap_ratio, both passing) and a +RETURN (PR-K ridge_minute_return, sign transferred but Reject). This factor is a price +POSITION — where in the day's range the valley VWAP sits — so it tests whether PR-I/PR-J +succeeded because of the ratio FORM or because of price-level INFORMATION generally. + +SAME MACHINE as PR-F .. PR-K: the minute classification is REUSED from +``data/clean/intraday_volume_prv.py`` (not re-implemented, and not modified by this PR), +so a different verdict here cannot be an artefact of a differently implemented taxonomy. + +TWO STRUCTURAL DIFFERENCES from its siblings, both of which this runner is built around: + + 1. THE PRICE RANGE READS THE PREVIOUS DAY'S CLOSE. Pinned to the last VISIBLE (<=14:50) + raw close of the previous trading day, so every price in the factor comes from ONE + source under ONE visibility rule. That is a DISCLOSED DEVIATION from the report, + which uses the true previous daily close. + 2. THE FACTOR IS REVERSAL-NEUTRALIZED, which makes it CROSS-SECTIONAL. Like PR-G, the + runner streams one symbol at a time through ``compute_valley_price_quantile_stats`` + (memory-bounded), assembles the full-universe raw panel, and then residualizes ONCE + against the 20-day reversal. The reversal is taken at T-1 — + ``rev20 = -(close_{d-1}/close_{d-21} - 1)`` — because the report's naive ``close_d`` + form is 15:00 information and WOULD be a lookahead at our 14:50 decision. Those + closes come from the FRONT-ADJUSTED daily panel this runner already loads; NO new + data source is introduced. + +The runner REPORTS the before/after-neutralization IC exposure and the day-validity / +cross-section coverage, so a neutralization that silently ate the signal (or a +cross-section too thin to regress) shows up as a number rather than being hidden. + +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 price quantile adds +alpha BEYOND value / low-vol. That check matters more than usual here: a within-range +price POSITION is a plausible cousin of both a value signal and a reversal signal, and +although the factor is already reversal-neutralized, the with-book run is what says +whether it is a genuinely new bet. + +⚠️ SIGN = +1, so this factor is NOT affected by the frozen layer's aligned-spread cost +defect (which mis-signs costs only for negative-sign factors). The ``aligned_spread_*`` +fields may be read directly here — unlike PR-K — though ``net_long_short_by_cost`` remains +the primary tradability read for comparability with its siblings. +""" + +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_quantile import ( + VALLEY_QUANTILE_LOOKBACK_DAYS, + VALLEY_QUANTILE_MIN_CROSS_SECTION, + VALLEY_QUANTILE_MIN_VALLEY_BARS, + VALLEY_QUANTILE_REVERSAL_DAYS, + compute_valley_price_quantile_stats, + residualize_on_reversal, + reversal_20, +) +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 ValleyPriceQuantileFactor +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_price_quantile" +_REPORT_STEM = "eval_valley_price_quantile" + + +# --------------------------------------------------------------------------- # +# Neutralization coverage (the factor-specific diagnostic) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class NeutralizationCoverage: + """What the reversal neutralization actually did, measured rather than assumed. + + A neutralization can fail quietly in two ways: the reversal can be unavailable for + most of the panel (so most residuals are NaN), or the cross-section can be too thin + to regress on many dates. Both are counted here and logged, so a coverage regression + is a number in the run record instead of an unexplained drop in sample size. + """ + + raw_rows: int # finite RAW qbar values + rev_rows: int # finite rev20 values on those rows + residual_rows: int # finite residuals (the shipped factor) + dates_total: int + dates_residualized: int # dates that cleared min_cross_section AND were non-degenerate + cross_section_min: int + cross_section_median: float + cross_section_max: int + raw_rev_spearman_mean: float # mean per-date exposure of the RAW factor to rev20 + + +def summarize_neutralization( + raw: pd.Series, + rev: pd.Series, + residual: pd.Series, + *, + min_cross_section: int, +) -> NeutralizationCoverage: + """Reduce the raw / reversal / residual panels to the coverage diagnostic.""" + raw_finite = raw.dropna() + rev_on_raw = rev.reindex(raw.index) + paired = pd.DataFrame({"f": raw, "r": rev_on_raw}).dropna() + + sizes: list[int] = [] + exposures: list[float] = [] + for _, g in paired.groupby(level=DATE_LEVEL, sort=True): + sizes.append(len(g)) + if len(g) >= min_cross_section: + f = g["f"].to_numpy(dtype=float) + r = g["r"].to_numpy(dtype=float) + fr = pd.Series(f).rank().to_numpy() + rr = pd.Series(r).rank().to_numpy() + if fr.std() > 0.0 and rr.std() > 0.0: + exposures.append(float(np.corrcoef(fr, rr)[0, 1])) + + resid_finite = residual.dropna() + dates_resid = int( + resid_finite.index.get_level_values(DATE_LEVEL).unique().size + ) + return NeutralizationCoverage( + raw_rows=int(len(raw_finite)), + rev_rows=int(len(paired)), + residual_rows=int(len(resid_finite)), + dates_total=int(raw.index.get_level_values(DATE_LEVEL).unique().size), + dates_residualized=dates_resid, + cross_section_min=int(min(sizes)) if sizes else 0, + cross_section_median=float(np.median(sizes)) if sizes else float("nan"), + cross_section_max=int(max(sizes)) if sizes else 0, + raw_rev_spearman_mean=( + float(np.mean(exposures)) if exposures else float("nan") + ), + ) + + +# --------------------------------------------------------------------------- # +# Minute loading (cache-only, per-symbol stats -> ONE cross-sectional residualization) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _ValleyQuantileMinuteLoad: + """Diagnostics from the cache-only per-symbol read + the neutralization.""" + + factor: pd.Series # MultiIndex(date, symbol) RESIDUALIZED factor (the shipped one) + raw: pd.Series # MultiIndex(date, symbol) RAW trailing-mean quantile (pre-neutral) + requested: int + covered: tuple[str, ...] # symbols that produced >= 1 finite RAW 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) + coverage: NeutralizationCoverage + + +def _load_valley_price_quantile_panel( + cfg: RootConfig, + symbols: list[str], + spec: FactorSpec, + panel: pd.DataFrame, + 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_cross_section: int, + reversal_days: int, +) -> _ValleyQuantileMinuteLoad: + """Compute the factor from the minute cache: per-symbol stats then ONE residualization. + + Memory-bounded: one symbol's minute history is read, aggregated to its daily raw + quantile 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 REVERSAL comes from ``panel['close']``, which :func:`qt.pipeline._load_panel` has + already FRONT-ADJUSTED — required, because the ratio spans 20 trading days. Taking it + at T-1 is what keeps day d's 15:00 close out of day d's factor value. + """ + root = cfg.data.cache.root_dir + store = IntradayParquetStore(root) + start = pd.Timestamp(cfg.data.start).normalize() + end = pd.Timestamp(cfg.data.end).normalize() + pd.Timedelta("23:59:59") + + series: list[pd.Series] = [] + covered: list[str] = [] + empty: list[str] = [] + raw_rows = 0 + for i, sym in enumerate(symbols): + part = store.read_range(INTRADAY_ENDPOINT, sym, RAW_INTRADAY_FREQ, start, end) + if part.empty: + empty.append(sym) + continue + raw_rows += len(part) + bars = normalize_intraday_bars( + part.rename(columns={"bar_end": "time"})[READ_COLUMNS], + freq=RAW_INTRADAY_FREQ, + ) + s = compute_valley_price_quantile_stats( + 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, + ) + if s.notna().any(): + series.append(s) + covered.append(sym) + else: + empty.append(sym) + if (i + 1) % 100 == 0: + logger.info( + "minute aggregation: %d/%d symbols processed (%d with a value)", + i + 1, len(symbols), len(covered), + ) + + if not series: + raise ValueError( + "run-eval-valley-price-quantile blocked: no requested symbol produced a " + f"cached valley-price-quantile value over [{cfg.data.start}, {cfg.data.end}]. " + "The minute cache is required (this runner never warms it); check coverage." + ) + raw = pd.concat(series).sort_index() + + # ONE cross-sectional reversal neutralization over the assembled universe panel. + # The closes are the panel's FRONT-ADJUSTED closes -- no new data source. + rev = reversal_20(panel[["close"]], days=reversal_days) + factor = residualize_on_reversal( + raw, rev, min_cross_section=min_cross_section, name=spec.factor_id + ) + coverage = summarize_neutralization( + raw, rev, factor, min_cross_section=min_cross_section + ) + logger.info( + "minute aggregation (cache-only): %d/%d symbols with a value, %d raw 1min rows " + "read, %d raw factor rows, stk_mins_live_calls=0", + len(covered), len(symbols), raw_rows, len(raw), + ) + logger.info( + "reversal neutralization (T-1 basis, %d-day span): raw=%d rev-paired=%d " + "residual=%d rows; dates %d/%d residualized; cross-section min/median/max=" + "%d/%.1f/%d; mean per-date Spearman(raw, rev20)=%.6f", + reversal_days, coverage.raw_rows, coverage.rev_rows, coverage.residual_rows, + coverage.dates_residualized, coverage.dates_total, + coverage.cross_section_min, coverage.cross_section_median, + coverage.cross_section_max, coverage.raw_rev_spearman_mean, + ) + return _ValleyQuantileMinuteLoad( + factor=factor, + raw=raw, + requested=len(symbols), + covered=tuple(covered), + empty_symbols=tuple(empty), + raw_rows=raw_rows, + live_calls=0, + coverage=coverage, + ) + + +# --------------------------------------------------------------------------- # +# Evaluation core (network-free seam: given panels, run the two evaluations) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _RunReports: + """The two evaluation reports (no-book / with-book) + their file paths.""" + + no_book: FactorEvalReport + with_book: FactorEvalReport + no_book_md: Path + no_book_json: Path + with_book_md: Path + with_book_json: Path + no_book_dashboard: Path + with_book_dashboard: Path + + +def evaluate_two_runs( + factor_panel: pd.Series | pd.DataFrame, + spec: FactorSpec, + eval_cfg: EvalConfig, + price_panel: pd.DataFrame, + book: pd.DataFrame, + *, + universe_symbols: tuple[str, ...], + fee_rate: float, + report_dir: Path, + stem: str = _REPORT_STEM, +) -> _RunReports: + """Run the StandardFactorEvaluator TWICE (no book / with book) and write reports. + + This is the network-free seam: given the PROCESSED factor panel, the qfq price panel + (for forward returns), and the PROCESSED known-factor book, it does the two + ``evaluate`` calls and writes ``{stem}_no_book`` / ``{stem}_with_book`` as Markdown + + JSON. Run 1 omits ``known_factors`` (Incremental NOT_ASSESSED); run 2 supplies the + book (Incremental measured). + """ + evaluator = StandardFactorEvaluator() + report_dir.mkdir(parents=True, exist_ok=True) + + ctx_no_book = EvalContext( + price_panel=price_panel, + universe_symbols=universe_symbols, + fee_rate=fee_rate, + ) + # evaluate_with_ir yields the SAME report as evaluate() plus the IR the research-style + # dashboard needs (per-period IC + quantile return series). + report_no_book, ir_no_book = evaluator.evaluate_with_ir( + factor_panel, spec, eval_cfg, ctx_no_book + ) + + ctx_with_book = EvalContext( + price_panel=price_panel, + universe_symbols=universe_symbols, + fee_rate=fee_rate, + known_factors=book, + ) + report_with_book, ir_with_book = evaluator.evaluate_with_ir( + factor_panel, spec, eval_cfg, ctx_with_book + ) + + nb_md, nb_json = _write_report(report_no_book, report_dir, f"{stem}_no_book") + wb_md, wb_json = _write_report(report_with_book, report_dir, f"{stem}_with_book") + nb_png = render_factor_dashboard( + report_no_book, ir_no_book, report_dir / f"{stem}_no_book_dashboard.png" + ) + wb_png = render_factor_dashboard( + report_with_book, ir_with_book, report_dir / f"{stem}_with_book_dashboard.png" + ) + return _RunReports( + no_book=report_no_book, + with_book=report_with_book, + no_book_md=nb_md, + no_book_json=nb_json, + with_book_md=wb_md, + with_book_json=wb_json, + no_book_dashboard=nb_png, + with_book_dashboard=wb_png, + ) + + +def _write_report(report: FactorEvalReport, report_dir: Path, stem: str) -> tuple[Path, Path]: + """Write ``report`` as deterministic Markdown + JSON; return the two paths.""" + md_path = report_dir / f"{stem}.md" + json_path = report_dir / f"{stem}.json" + md_path.write_text(report.render(), encoding="utf-8") + json_path.write_text(report.to_json(), encoding="utf-8") + return md_path, json_path + + +# --------------------------------------------------------------------------- # +# Metric extraction (for the CLI line + the handoff) +# --------------------------------------------------------------------------- # +def _section_payload(report: FactorEvalReport, name: str) -> dict: + section = report.by_name().get(name) + return dict(getattr(section, "payload", {}) or {}) + + +def extract_metrics(report: FactorEvalReport) -> dict: + """Pull the headline verdict + gated metrics out of a finished report. + + Surfaces the same comparison quantities PR-K extracted (turnover, net long-short by + cost scenario, lag-1 rank autocorrelation + half-life, cross-section size) so PR-L + sits in one table with PR-I / PR-J / PR-K, PLUS ``ic_pearson_mean`` alongside the rank + ``ic_mean``: the PR-K review found that a divergence between the two predicts the + monotonicity gate failing, and this run is the first test of that regularity on a + POSITIVE-sign factor (the four prior cases were all sign=-1). + + Unlike PR-K, ``aligned_spread_*`` is NOT unreliable here: the frozen layer's + cost-sign defect only mis-signs negative-sign factors, and this factor's pre-registered + sign is +1. ``net_long_short_by_cost`` is still surfaced as the primary read, for + sibling comparability. + """ + verdict = report.require_verdict() + pred = _section_payload(report, "predictive_power") + coverage = _section_payload(report, "data_coverage") + incr = _section_payload(report, "purity") + ret_risk = _section_payload(report, "return_risk") + stability = _section_payload(report, "stability_cost") + autocorr = dict(stability.get("factor_rank_autocorr_by_lag", {}) or {}) + return { + "deployment": verdict.verdict, + "predictive": verdict.predictive.verdict, + "incremental": verdict.incremental.verdict, + "tradable": verdict.tradable.verdict, + "ic_mean": pred.get("ic_mean"), + "ic_pearson_mean": pred.get("ic_pearson_mean"), + "ic_ir": pred.get("ic_ir"), + "ic_ir_ci_low": pred.get("ic_ir_ci_low"), + "ic_ir_ci_high": pred.get("ic_ir_ci_high"), + "ic_ir_ci_n_eff": pred.get("ic_ir_ci_n_eff"), + "ic_win_rate": pred.get("ic_win_rate"), + "ic_nw_t": pred.get("ic_nw_t"), + "settled_rebalances": coverage.get("settled_rebalances"), + "effective_samples": coverage.get("effective_samples"), + "span_days": coverage.get("span_days"), + "cross_section_size_mean": coverage.get("cross_section_size_mean"), + "cross_section_size_median": coverage.get("cross_section_size_median"), + "incremental_ic_ir": incr.get("incremental_ic_ir"), + "incremental_ic_ir_ci_low": incr.get("incremental_ic_ir_ci_low"), + "incremental_ic_ir_ci_high": incr.get("incremental_ic_ir_ci_high"), + "incremental_ic_ir_ci_n_eff": incr.get("incremental_ic_ir_ci_n_eff"), + "incremental_ic_mean": incr.get("incremental_ic_mean"), + "monotonicity_spearman": ret_risk.get("monotonicity_spearman"), + "gross_long_short_mean": ret_risk.get("gross_long_short_mean"), + "net_long_short_by_cost": dict(ret_risk.get("net_long_short_by_cost", {}) or {}), + # sign=+1 -> the aligned spreads are NOT mis-signed for this factor. + "aligned_spread_by_cost": dict(ret_risk.get("aligned_spread_by_cost", {}) or {}), + "long_short_turnover": stability.get("turnover_mean_long_short_legs"), + "rank_autocorr_lag1": autocorr.get(1), + "half_life_periods": stability.get("half_life_periods"), + } + + +# --------------------------------------------------------------------------- # +# EvalConfig construction (HONEST provenance of what the runner actually did) +# --------------------------------------------------------------------------- # +def _build_eval_config(cfg: RootConfig) -> EvalConfig: + """Build the per-run EvalConfig, declaring EXACTLY what the pipeline applied. + + The declarations must not overstate: this codebase's winsorize step is a P0 no-op, so + ``winsorize`` is declared None (nothing was clipped) even though the config may toggle + it. z-score + industry/size neutralization ARE applied, so they are declared. + ``oos_split`` (from the config's ``oos`` block) makes the OOS section run so the + Predictive axis can be assessed. ``is_exploratory=True``: this is a reproduction on a + shorter window / narrower neutralization than the report, not a return claim (it caps + the deployment label at Watch). + """ + if cfg.oos is None: + raise ValueError( + "run-eval-valley-price-quantile 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 ValleyPriceQuantileEvalResult: + """Immutable summary of one run-eval-valley-price-quantile 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 + neutralization: NeutralizationCoverage + 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-price-quantile 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-price-quantile 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-price-quantile needs universe.type='index' (PIT membership, " + f"e.g. 000905.SH for CSI500); got {cfg.universe.type!r}." + ) + if not cfg.processing.neutralize.enabled: + raise ValueError( + "run-eval-valley-price-quantile expects processing.neutralize.enabled=true " + "(industry + size neutralization, matching the report's neutral column and " + "the EvalConfig declaration)." + ) + + +def run_eval_valley_price_quantile(config_path: str) -> ValleyPriceQuantileEvalResult: + """Run the two real valley-price-quantile 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 = ValleyPriceQuantileFactor(lookback_days=VALLEY_QUANTILE_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-price-quantile: cache-only per-symbol stats + ONE reversal neutralization. + load = _load_valley_price_quantile_panel( + cfg, symbols, spec, panel, logger, + lookback_days=VALLEY_QUANTILE_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_QUANTILE_MIN_VALLEY_BARS, + min_cross_section=VALLEY_QUANTILE_MIN_CROSS_SECTION, + reversal_days=VALLEY_QUANTILE_REVERSAL_DAYS, + ) + # 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"], + ) + logger.info( + "net long-short by cost: %s", no_book_metrics["net_long_short_by_cost"] + ) + # The PR-K review's regularity, tested here on a POSITIVE-sign factor for the first + # time: a divergence between the rank IC and the Pearson IC predicted the monotonicity + # gate failing on all four prior (negative-sign) factors. + logger.info( + "ic_mean(rank)=%s vs ic_pearson_mean=%s; monotonicity_spearman=%s", + no_book_metrics["ic_mean"], no_book_metrics["ic_pearson_mean"], + no_book_metrics["monotonicity_spearman"], + ) + + return ValleyPriceQuantileEvalResult( + 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, + neutralization=load.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__ = [ + "NeutralizationCoverage", + "ValleyPriceQuantileEvalResult", + "evaluate_two_runs", + "extract_metrics", + "run_eval_valley_price_quantile", + "summarize_neutralization", +] diff --git a/tests/test_eval_valley_price_quantile_runner.py b/tests/test_eval_valley_price_quantile_runner.py new file mode 100644 index 0000000..fe25272 --- /dev/null +++ b/tests/test_eval_valley_price_quantile_runner.py @@ -0,0 +1,509 @@ +"""PR-L runner: cache-only minute loader + reversal neutralization + the two evaluations. + +Network-free throughout: the minute "cache" is a real ``IntradayParquetStore`` seeded in a +tmp_path, and the two-run evaluation seam is exercised on synthetic processed panels. + +The structural novelty this file has to pin down, beyond PR-I..PR-K's loader contract, is +that the loader RESIDUALIZES: it streams per-symbol raw quantiles, then applies ONE +cross-sectional reversal neutralization against the T-1 reversal built from the daily +panel's FRONT-ADJUSTED closes. Tests cover that the shipped factor is the residual (not +the raw quantile), that the daily closes actually reach it, and that the neutralization +coverage is measured rather than assumed. +""" + +from __future__ import annotations + +import logging + +import numpy as np +import pandas as pd +import pytest + +from analytics.eval import MANDATORY_SECTIONS, EvalConfig, 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_quantile import ( + VALLEY_QUANTILE_MIN_VALLEY_BARS, + reversal_20, +) +from factors.compute.intraday_derived import ValleyPriceQuantileFactor +from qt.config import ( + AlphaCfg, + BacktestCfg, + CacheCfg, + CostCfg, + DataCfg, + FactorCfg, + OutputCfg, + PortfolioCfg, + RootConfig, + UniverseCfg, +) +from qt.eval_valley_price_quantile import ( + _load_valley_price_quantile_panel, + evaluate_two_runs, + extract_metrics, + summarize_neutralization, +) + + +# --------------------------------------------------------------------------- # +# Minute cache fixtures +# --------------------------------------------------------------------------- # +_N_SLOTS = 16 +_ERUPT = (3, 7) +_HIGH_SLOT, _LOW_SLOT = 5, 9 +_BG_DAYS = ("2021-07-01", "2021-07-02", "2021-07-05") +_TEST_DAY = "2021-07-06" +# Range on the test day: intraday [80, 120], prev_close 100 (inside) -> hi=120, lo=80. +_HI, _LO = 120.0, 80.0 +_PREV_CLOSE = 100.0 +# Per-symbol valley VWAP -> q = (vwap - 80) / 40. +_SYM_VWAP = {"AAA.SZ": 90.0, "BBB.SZ": 100.0, "CCC.SZ": 110.0} +_SYM_Q = {s: (v - _LO) / (_HI - _LO) for s, v in _SYM_VWAP.items()} + + +def _stored_rows(sym, day, *, vols, amts, highs, lows, closes): + """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). ``amount`` is + set INDEPENDENTLY of ``volume`` (the per-bar price is amount/volume) and high / low / + close are explicit, because this factor reads the day's range and the previous day's + last visible close alongside the valley VWAP. + """ + base = pd.Timestamp(day) + pd.Timedelta("09:31:00") + rows = [] + for i in range(len(vols)): + be = base + pd.Timedelta(minutes=i) + rows.append( + { + "symbol": sym, + "bar_end": be, + "source_trade_time": be, + "open": float(closes[i]), + "high": float(highs[i]), + "low": float(lows[i]), + "close": float(closes[i]), + "volume": float(vols[i]), + "amount": float(amts[i]), + "freq": "1min", + } + ) + return pd.DataFrame(rows) + + +def _background_day(sym, day): + """Flat volume-100 session at price 100 -> baseline mu=100, sigma=0; hi == lo.""" + n = _N_SLOTS + return _stored_rows( + sym, day, + vols=[100.0] * n, amts=[100.0 * _PREV_CLOSE] * n, + highs=[_PREV_CLOSE] * n, lows=[_PREV_CLOSE] * n, closes=[_PREV_CLOSE] * n, + ) + + +def _test_day(sym): + """Engineered day whose valley VWAP is ``_SYM_VWAP[sym]`` inside the range [80, 120].""" + n = _N_SLOTS + vwap = _SYM_VWAP[sym] + vols = [100.0] * n + amts = [100.0 * vwap] * n + for s in _ERUPT: + vols[s] = 200.0 + amts[s] = 200.0 * 500.0 # eruptive bars traded elsewhere; excluded from the VWAP + highs = [110.0] * n + lows = [95.0] * n + highs[_HIGH_SLOT] = _HI + lows[_LOW_SLOT] = _LO + return _stored_rows( + sym, _TEST_DAY, + vols=vols, amts=amts, highs=highs, lows=lows, closes=[100.0] * n, + ) + + +def _seed_symbol(store, sym): + for day in _BG_DAYS: + store.upsert(INTRADAY_ENDPOINT, sym, "1min", _background_day(sym, day), KEY_COLS) + store.upsert(INTRADAY_ENDPOINT, sym, "1min", _test_day(sym), KEY_COLS) + + +def _min_config(root, start, end): + return RootConfig( + data=DataCfg( + source="tushare", + start=start, + end=end, + external_secret_file="/nonexistent.json", + cache=CacheCfg(enabled=True, root_dir=str(root)), + ), + universe=UniverseCfg(type="static", symbols=["A.SZ"]), + factors=[FactorCfg(name="momentum_20")], + alpha=AlphaCfg(), + portfolio=PortfolioCfg(top_n=1), + backtest=BacktestCfg(), + cost=CostCfg(), + output=OutputCfg(), + ) + + +def _daily_panel(symbols, *, n_days=25, end=_TEST_DAY, seed=5): + """Daily FRONT-ADJUSTED close panel spanning enough history for a T-1 20-day reversal. + + The panel is INDEPENDENT of the minute cache (it is the runner's own daily panel) and + must reach back >= 22 business days before the signal day so ``rev20`` is finite there. + """ + dates = pd.bdate_range(end=pd.Timestamp(end), periods=n_days) + rng = np.random.default_rng(seed) + idx = pd.MultiIndex.from_product([dates, symbols], names=["date", "symbol"]) + close = 100.0 * np.exp( + np.cumsum(rng.normal(0, 0.02, size=(len(dates), len(symbols))), axis=0) + ) + return pd.DataFrame({"close": close.reshape(-1)}, index=idx) + + +# Small gates so a 4-day cache produces a value. lookback_days=1 so the test day's value IS +# its own daily quantile; the flat background days have hi == lo, hence no valid day, so +# they cannot accumulate in. +_LOAD_KW = dict( + lookback_days=1, baseline_days=20, baseline_min_obs=2, + sigma_k=1.0, min_valid_days=1, min_classifiable=1, min_valley_bars=1, + min_cross_section=3, reversal_days=20, +) + + +# --------------------------------------------------------------------------- # +# Cache-only loader +# --------------------------------------------------------------------------- # +def test_minute_loader_is_cache_only_and_discloses_empty(tmp_path): + root = tmp_path / "cache" + store = IntradayParquetStore(str(root)) + for sym in _SYM_VWAP: + _seed_symbol(store, sym) + + syms = sorted(_SYM_VWAP) + cfg = _min_config(root, "2021-07-01", _TEST_DAY) + spec = ValleyPriceQuantileFactor().spec + load = _load_valley_price_quantile_panel( + cfg, syms + ["ZZZ.SZ"], spec, _daily_panel(syms + ["ZZZ.SZ"]), + logging.getLogger("test.vpq.loader"), **_LOAD_KW, + ) + assert load.live_calls == 0 # store read has no fetch closure + assert set(load.covered) == set(syms) # all three produced a raw value + assert load.empty_symbols == ("ZZZ.SZ",) # uncovered disclosed, never fetched + assert load.factor.name == spec.factor_id + + # The RAW panel carries the hand-computed daily quantiles. + d = pd.Timestamp(_TEST_DAY) + for sym, q in _SYM_Q.items(): + assert load.raw.loc[(d, sym)] == pytest.approx(q) + + # A flat background day (hi == lo) is invalid -> no value emitted. + assert (pd.Timestamp(_BG_DAYS[-1]), "AAA.SZ") not in load.raw.index + + +def test_loader_ships_the_residual_not_the_raw_quantile(tmp_path): + """The factor the loader returns is the REVERSAL-NEUTRALIZED panel. + + Guards the wiring bug where the neutralization is computed and then discarded. The + residuals of a 3-name cross-section also sum to ~0 by construction of an intercept + OLS fit, which the raw quantiles (0.25 / 0.50 / 0.75) do not. + """ + root = tmp_path / "cache" + store = IntradayParquetStore(str(root)) + for sym in _SYM_VWAP: + _seed_symbol(store, sym) + syms = sorted(_SYM_VWAP) + + load = _load_valley_price_quantile_panel( + _min_config(root, "2021-07-01", _TEST_DAY), syms, + ValleyPriceQuantileFactor().spec, _daily_panel(syms), + logging.getLogger("test.vpq.resid"), **_LOAD_KW, + ) + d = pd.Timestamp(_TEST_DAY) + shipped = np.array([float(load.factor.loc[(d, s)]) for s in syms]) + raw = np.array([float(load.raw.loc[(d, s)]) for s in syms]) + assert np.isfinite(shipped).all() + assert not np.allclose(shipped, raw) + assert shipped.sum() == pytest.approx(0.0, abs=1e-12) # intercept OLS residual + + +def test_loader_actually_consumes_the_daily_closes(tmp_path): + """A different daily close panel must produce a different factor. + + Without this, the reversal could be silently absent and every test above would still + pass on the raw quantiles. + """ + root = tmp_path / "cache" + store = IntradayParquetStore(str(root)) + for sym in _SYM_VWAP: + _seed_symbol(store, sym) + syms = sorted(_SYM_VWAP) + cfg = _min_config(root, "2021-07-01", _TEST_DAY) + spec = ValleyPriceQuantileFactor().spec + + a = _load_valley_price_quantile_panel( + cfg, syms, spec, _daily_panel(syms, seed=5), + logging.getLogger("test.vpq.px.a"), **_LOAD_KW, + ) + b = _load_valley_price_quantile_panel( + cfg, syms, spec, _daily_panel(syms, seed=99), + logging.getLogger("test.vpq.px.b"), **_LOAD_KW, + ) + d = pd.Timestamp(_TEST_DAY) + va = [float(a.factor.loc[(d, s)]) for s in syms] + vb = [float(b.factor.loc[(d, s)]) for s in syms] + assert not np.allclose(va, vb) + # ... while the RAW quantiles, which never see a daily close, are identical. + assert [float(a.raw.loc[(d, s)]) for s in syms] == [ + float(b.raw.loc[(d, s)]) for s in syms + ] + + +def test_loader_reports_the_neutralization_coverage(tmp_path): + root = tmp_path / "cache" + store = IntradayParquetStore(str(root)) + for sym in _SYM_VWAP: + _seed_symbol(store, sym) + syms = sorted(_SYM_VWAP) + + load = _load_valley_price_quantile_panel( + _min_config(root, "2021-07-01", _TEST_DAY), syms, + ValleyPriceQuantileFactor().spec, _daily_panel(syms), + logging.getLogger("test.vpq.cov"), **_LOAD_KW, + ) + cov = load.coverage + assert cov.raw_rows == 3 # one valid day x three symbols + assert cov.rev_rows == 3 # all three had a finite T-1 reversal + assert cov.residual_rows == 3 # ... and all three were residualized + assert cov.dates_total == 1 + assert cov.dates_residualized == 1 + assert cov.cross_section_min == 3 and cov.cross_section_max == 3 + assert np.isfinite(cov.raw_rev_spearman_mean) + + +def test_loader_blocks_when_nothing_cached(tmp_path): + cfg = _min_config(tmp_path / "empty", "2021-07-01", _TEST_DAY) + with pytest.raises(ValueError, match="no requested symbol produced"): + _load_valley_price_quantile_panel( + cfg, ["ZZZ.SZ"], ValleyPriceQuantileFactor().spec, + _daily_panel(["ZZZ.SZ"]), logging.getLogger("test.vpq.block"), **_LOAD_KW, + ) + + +def test_thin_cross_section_leaves_the_factor_nan(tmp_path): + """Below ``min_cross_section`` the date cannot be residualized -> NaN, never raw.""" + root = tmp_path / "cache" + store = IntradayParquetStore(str(root)) + for sym in _SYM_VWAP: + _seed_symbol(store, sym) + syms = sorted(_SYM_VWAP) + + load = _load_valley_price_quantile_panel( + _min_config(root, "2021-07-01", _TEST_DAY), syms, + ValleyPriceQuantileFactor().spec, _daily_panel(syms), + logging.getLogger("test.vpq.thin"), + **dict(_LOAD_KW, min_cross_section=4), + ) + assert not np.isfinite(load.factor.to_numpy(dtype=float)).any() + assert load.raw.notna().any() # the raw panel is unaffected + assert load.coverage.residual_rows == 0 + + +# --------------------------------------------------------------------------- # +# summarize_neutralization +# --------------------------------------------------------------------------- # +def _panel(dates, syms, values, name): + idx = pd.MultiIndex.from_product([dates, syms], names=["date", "symbol"]) + return pd.Series(np.asarray(values, dtype=float).reshape(-1), index=idx, name=name) + + +def test_summarize_neutralization_counts_missing_reversal_rows(): + dates = pd.bdate_range("2023-01-02", periods=2) + syms = ["A", "B", "C"] + raw = _panel(dates, syms, [[0.1, 0.2, 0.3], [0.4, 0.5, 0.6]], "q") + rev = _panel(dates, syms, [[1.0, 2.0, 3.0], [1.0, np.nan, np.nan]], "r") + resid = _panel(dates, syms, [[0.0, 0.0, 0.0], [np.nan] * 3], "f") + cov = summarize_neutralization(raw, rev, resid, min_cross_section=3) + assert cov.raw_rows == 6 + assert cov.rev_rows == 4 # only four rows had BOTH + assert cov.residual_rows == 3 + assert cov.dates_total == 2 + assert cov.dates_residualized == 1 + assert cov.cross_section_min == 1 and cov.cross_section_max == 3 + + +def test_summarize_neutralization_handles_an_all_missing_reversal(): + dates = pd.bdate_range("2023-01-02", periods=1) + syms = ["A", "B"] + raw = _panel(dates, syms, [[0.1, 0.2]], "q") + rev = _panel(dates, syms, [[np.nan, np.nan]], "r") + resid = _panel(dates, syms, [[np.nan, np.nan]], "f") + cov = summarize_neutralization(raw, rev, resid, min_cross_section=3) + assert cov.rev_rows == 0 + assert cov.residual_rows == 0 + assert cov.dates_residualized == 0 + assert not np.isfinite(cov.raw_rev_spearman_mean) # no exposure is measurable + + +def test_reversal_from_a_daily_panel_is_the_t_minus_1_ratio(): + """The runner's reversal input is the panel's close column, read at T-1.""" + dates = pd.bdate_range("2023-01-02", periods=25) + panel = pd.DataFrame( + {"close": np.tile(np.arange(100.0, 125.0), 1)}, + index=pd.MultiIndex.from_product([dates, ["A"]], names=["date", "symbol"]), + ) + rev = reversal_20(panel[["close"]], days=20) + d = dates[22] + assert float(rev.loc[(d, "A")]) == pytest.approx(-((121.0 / 101.0) - 1.0)) + + +# --------------------------------------------------------------------------- # +# 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"]) + + factor = pd.Series( + rng.standard_normal(len(idx)), index=idx, name="valley_price_quantile_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, + ) + 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 = ValleyPriceQuantileFactor().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) + assert all(isinstance(s, (Section, Skipped)) for s in by.values()) + assert report.verdict is not None + + assert isinstance(reports.no_book.by_name()["purity"], Skipped) + assert reports.no_book.verdict.incremental.verdict == AXIS_NOT_ASSESSED + + 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 + + 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 + + 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" + + m = extract_metrics(reports.no_book) + assert m["deployment"] in {"Adopt", "Watch", "Reject", "INSUFFICIENT-DATA"} + assert "effective_samples" in m and "ic_ir" in m + + +def test_extract_metrics_surfaces_the_pr_l_comparison_quantities(tmp_path): + """PR-L is compared head-on with PR-I / PR-J / PR-K on the same cell. + + Beyond the shared tradability quantities, this run needs ``ic_pearson_mean`` next to + the rank ``ic_mean``: the PR-K review found their divergence predicted the + monotonicity gate failing, and PR-L is the first test of that on a sign=+1 factor. + """ + factor, book, price = _synthetic_panels() + reports = evaluate_two_runs( + factor, ValleyPriceQuantileFactor().spec, _eval_cfg(), price, book, + universe_symbols=tuple(sorted(set(book.index.get_level_values("symbol")))), + fee_rate=0.001, + report_dir=tmp_path, + ) + m = extract_metrics(reports.no_book) + for key in ( + "long_short_turnover", "rank_autocorr_lag1", "half_life_periods", + "cross_section_size_mean", "cross_section_size_median", + "monotonicity_spearman", "gross_long_short_mean", + "ic_mean", "ic_pearson_mean", + ): + assert key in m, key + assert m[key] is not None, key + net = m["net_long_short_by_cost"] + assert isinstance(net, dict) and net + assert {1.0, 2.0, 4.0} <= {float(k) for k in net} + assert len({round(float(v), 12) for v in net.values()}) > 1 + # sign=+1 -> the aligned spreads are NOT mis-signed for this factor and are surfaced. + assert isinstance(m["aligned_spread_by_cost"], dict) + + +def test_spec_sign_is_positive_so_aligned_spread_is_not_mis_signed(): + """The frozen layer's cost-sign defect only bites at sign=-1; PR-L is +1.""" + assert ValleyPriceQuantileFactor().spec.expected_ic_sign == 1 + + +def test_no_book_run_never_reaches_adopt(tmp_path): + factor, book, price = _synthetic_panels(seed=3) + reports = evaluate_two_runs( + factor, ValleyPriceQuantileFactor().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 + + +def test_min_valley_bars_default_matches_the_pr_i_floor(): + """PR-L reuses PR-I's valley floor so the two runs' coverage is comparable.""" + from data.clean.intraday_valley_vwap import VALLEY_VWAP_MIN_VALLEY_BARS + + assert VALLEY_QUANTILE_MIN_VALLEY_BARS == VALLEY_VWAP_MIN_VALLEY_BARS diff --git a/tests/test_valley_price_quantile_factor.py b/tests/test_valley_price_quantile_factor.py new file mode 100644 index 0000000..61997b1 --- /dev/null +++ b/tests/test_valley_price_quantile_factor.py @@ -0,0 +1,901 @@ +"""PR-L: VALLEY WEIGHTED-PRICE-QUANTILE factor. + +Same volume classification as PR-F/PR-H/PR-I/PR-J/PR-K (REUSED, not re-implemented), +a new STATISTIC family: a price POSITION (where in the day's range the valley VWAP +sits) rather than a count / timing moment / price RATIO / return. Sign is pre-registered ++1. + +TWO cross-sectional/lookahead subtleties get dedicated tests here, because they are what +makes this factor harder than its five siblings: + + 1. PREV_CLOSE extends the price range. The range is + ``[min(intraday low, prev_close), max(intraday high, prev_close)]`` where prev_close + is the last VISIBLE (<=14:50) raw close of the PREVIOUS trading day -- ONE visibility + definition for the whole factor. Hand cases cover prev_close inside the intraday + range, above it, and below it. + 2. REVERSAL NEUTRALIZATION at T-1. The report residualizes against a 20-day reversal + factor. The naive ``-(close_d/close_{d-20} - 1)`` reads day d's CLOSE, which is 15:00 + information and a lookahead at our 14:50 decision. We use + ``rev20 = -(close_{d-1}/close_{d-21} - 1)`` on FRONT-ADJUSTED daily closes. + ``test_reversal_uses_t_minus_1_close_not_day_d`` perturbs ONE SYMBOL's day-d close + and asserts the factor is UNCHANGED -- the single most important test in this PR. + The single-symbol shape is load-bearing: a UNIFORM bump of every cross-section + member's ``close_d`` sends the naive regressor to an AFFINE reparametrization of + itself, and intercept-OLS residuals are invariant under that, so a uniform bump + passes under BOTH bases and proves nothing. Two companions keep it honest: + ``test_perturbing_the_t_minus_1_close_does_move_the_factor`` shows the closes are + genuinely an input, and + ``test_uniform_close_d_bump_cannot_distinguish_the_two_reversal_bases`` records why + the uniform shape is inadmissible. Verified by monkeypatching ``reversal_20`` to the + naive ``shift(0)/shift(days)`` form: the critical test FAILS there and passes on the + real implementation. + +Hand cases build a constant BACKGROUND of 10 prior days so the same-slot baseline is +exact (mu=100, sigma=0 -> the eruptive threshold is exactly 100) and, because +``VOLUME_PRV_BASELINE_MIN_OBS`` is 10, so that the engineered test day is the ONLY valid +day -- the trailing mean therefore equals that day's quantile in closed form. +""" + +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_quantile import ( + VALLEY_QUANTILE_LOOKBACK_DAYS, + VALLEY_QUANTILE_MIN_CROSS_SECTION, + VALLEY_QUANTILE_MIN_VALLEY_BARS, + VALLEY_QUANTILE_REVERSAL_DAYS, + compute_valley_price_quantile, + compute_valley_price_quantile_stats, + residualize_on_reversal, + reversal_20, + valley_price_quantile_by_day, +) +from data.clean.intraday_volume_prv import ( + VOLUME_PRV_BASELINE_DAYS, + VOLUME_PRV_BASELINE_MIN_OBS, + VOLUME_PRV_MIN_CLASSIFIABLE, + VOLUME_PRV_MIN_VALID_DAYS, + VOLUME_PRV_SIGMA_K, + peak_mask_for_symbol, + prepare_visible_minute_bars, +) +from factors.compute.intraday_derived import ValleyPriceQuantileFactor +from factors.spec import FactorSpec + +_SYM = "000001.SZ" +_TEST_DAY = pd.Timestamp("2021-07-11") +_BG_DAYS = 10 # == VOLUME_PRV_BASELINE_MIN_OBS -> only the engineered day is classifiable + + +# --------------------------------------------------------------------------- # +# Bar builders +# --------------------------------------------------------------------------- # +def _bars(rows): + """rows = [(time, symbol, volume, amount, high, low, close), ...] -> 1min bars. + + ``amount`` is independent of ``volume`` (the per-bar price is ``amount/volume``) and + high / low / close are set per bar, because this factor reads the day's price RANGE + (max high, min low) and the previous day's last visible CLOSE alongside the valley + VWAP. ``normalize_intraday_bars`` sets ``available_time = bar_end + 1min``, so the + 14:50 PIT cutoff excludes any bar with ``bar_end >= 14:50``. + """ + return normalize_intraday_bars( + pd.DataFrame( + { + "time": pd.to_datetime([r[0] for r in rows]), + "symbol": [r[1] for r in rows], + "open": [float(r[6]) for r in rows], + "high": [float(r[4]) for r in rows], + "low": [float(r[5]) for r in rows], + "close": [float(r[6]) for r in rows], + "volume": [float(r[2]) for r in rows], + "amount": [float(r[3]) for r in rows], + } + ), + freq="1min", + ) + + +def _session(day, vols, amts, highs, lows, closes, sym=_SYM, start="09:31:00"): + """One session of CONSECUTIVE 1-minute bars carrying the given per-bar fields.""" + base = pd.Timestamp(day) + pd.Timedelta(start) + return [ + (base + pd.Timedelta(minutes=i), sym, v, a, h, lo, c) + for i, (v, a, h, lo, c) in enumerate(zip(vols, amts, highs, lows, closes)) + ] + + +def _background(n_days, n_slots, sym=_SYM, start_day="2021-07-01", last_close=10.0): + """``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. The LAST + bar of the LAST background day carries ``last_close``: that bar is exactly the + ``prev_close`` the test day's price range extends with. + """ + rows = [] + for i in range(n_days): + day = (pd.Timestamp(start_day) + pd.Timedelta(days=i)).strftime("%Y-%m-%d") + closes = [10.0] * n_slots + if i == n_days - 1: + closes[-1] = last_close + rows += _session( + day, + [100.0] * n_slots, + [1000.0] * n_slots, + [10.0] * n_slots, + [10.0] * n_slots, + closes, + sym=sym, + ) + return rows + + +# Gates small enough that the single engineered test day is the only VALID day; the +# valley-bar and valid-day floors get their own dedicated tests below. +_KW = dict( + baseline_days=VOLUME_PRV_BASELINE_DAYS, + baseline_min_obs=VOLUME_PRV_BASELINE_MIN_OBS, + sigma_k=VOLUME_PRV_SIGMA_K, + lookback_days=VALLEY_QUANTILE_LOOKBACK_DAYS, + min_valid_days=1, + min_classifiable=1, + min_valley_bars=1, +) + + +# --------------------------------------------------------------------------- # +# The engineered test day: valley VWAP = 11.0, intraday high 12.0, intraday low 8.0 +# --------------------------------------------------------------------------- # +# 12 slots, eruptions at slots 3 and 7 (volume 200, amount 4000 -> price 20). +# valley bars: 10 x volume 100 (total 1000); five carry amount 1000 (price 10), five +# amount 1200 (price 12) -> valley VWAP = 11000 / 1000 = 11.0 exactly. +# price range: every bar has high 11.5 / low 9.5 EXCEPT slot 5 (high 12.0) and slot 9 +# (low 8.0) -- so max(high) = 12.0 and min(low) = 8.0 come from DIFFERENT bars and a +# first-bar / last-bar bug cannot pass. +_N = 12 +_ERUPT = (3, 7) +_HIGH_SLOT, _LOW_SLOT = 5, 9 +_VALLEY_VWAP = 11.0 +_INTRADAY_HIGH, _INTRADAY_LOW = 12.0, 8.0 + + +def _test_day_arrays(): + 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 + highs = [11.5] * _N + lows = [9.5] * _N + highs[_HIGH_SLOT] = _INTRADAY_HIGH + lows[_LOW_SLOT] = _INTRADAY_LOW + closes = [10.0] * _N + return vols, amts, highs, lows, closes + + +def _rows_with_prev_close(prev_close, n_slots=_N): + """Background (ending on ``prev_close``) + the engineered test day.""" + vols, amts, highs, lows, closes = _test_day_arrays() + return _background(_BG_DAYS, n_slots, last_close=prev_close) + _session( + "2021-07-11", vols, amts, highs, lows, closes + ) + + +# --------------------------------------------------------------------------- # +# Hand-computed quantiles (3 cases: prev_close inside / above / below the range) +# --------------------------------------------------------------------------- # +def test_hand_value_prev_close_inside_intraday_range(): + """prev_close 10.0 is inside [8, 12] -> range unchanged; q = (11-8)/(12-8) = 0.75.""" + out = compute_valley_price_quantile_stats(_bars(_rows_with_prev_close(10.0)), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(0.75) + + +def test_hand_value_prev_close_above_intraday_high_extends_the_top(): + """prev_close 16.0 > intraday high 12.0 -> hi becomes 16; q = (11-8)/(16-8) = 0.375. + + A higher previous close pushes the SAME valley VWAP to a LOWER position in the range + -- the whole point of including prev_close in the range. + """ + out = compute_valley_price_quantile_stats(_bars(_rows_with_prev_close(16.0)), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(0.375) + + +def test_hand_value_prev_close_below_intraday_low_extends_the_bottom(): + """prev_close 4.0 < intraday low 8.0 -> lo becomes 4; q = (11-4)/(12-4) = 0.875. + + Opposite direction from the case above, so a sign / min-max inversion bug in the + range construction cannot pass both. + """ + out = compute_valley_price_quantile_stats(_bars(_rows_with_prev_close(4.0)), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(0.875) + + +def test_quantile_is_in_the_unit_interval_in_normal_conditions(): + """A valley VWAP inside the range gives q in [0, 1] (the no-clip sanity check).""" + for prev_close in (4.0, 10.0, 16.0): + out = compute_valley_price_quantile_stats( + _bars(_rows_with_prev_close(prev_close)), **_KW + ) + q = float(out.loc[(_TEST_DAY, _SYM)]) + assert 0.0 <= q <= 1.0 + + +def test_quantile_is_not_clipped_when_the_vwap_escapes_the_range(): + """A VWAP outside [lo, hi] is EXPOSED as q outside [0,1], never clipped. + + The pinned choice: an out-of-range quantile means the range construction is wrong, + and silently clipping it to [0,1] would hide exactly the defect worth seeing. Here + the bars' high/low are deliberately set BELOW the traded price so the VWAP escapes. + """ + vols, amts, highs, lows, closes = _test_day_arrays() + highs = [10.2] * _N # max high 10.2 < valley VWAP 11.0 + lows = [9.5] * _N + rows = _background(_BG_DAYS, _N, last_close=10.0) + _session( + "2021-07-11", vols, amts, highs, lows, closes + ) + out = compute_valley_price_quantile_stats(_bars(rows), **_KW) + q = float(out.loc[(_TEST_DAY, _SYM)]) + # hi = max(10.2, 10.0) = 10.2, lo = min(9.5, 10.0) = 9.5 -> (11-9.5)/0.7 > 1 + assert q == pytest.approx((11.0 - 9.5) / (10.2 - 9.5)) + assert q > 1.0 + + +def test_valley_vwap_uses_the_amount_over_volume_identity(): + """The valley leg is Σamount/Σvolume, verified the LONG way from per-bar prices.""" + bars = _bars(_rows_with_prev_close(10.0)) + visible = prepare_visible_minute_bars( + bars, extra_columns=("amount", "high", "low", "close") + ) + 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) + is_valley = day["valley"].to_numpy(dtype=bool) + long_way = float((a[is_valley] / v[is_valley] * v[is_valley]).sum() / v[is_valley].sum()) + assert long_way == pytest.approx(_VALLEY_VWAP) + assert float(a[is_valley].sum() / v[is_valley].sum()) == pytest.approx(_VALLEY_VWAP) + + +# --------------------------------------------------------------------------- # +# Day validity gates +# --------------------------------------------------------------------------- # +def _work_frame(days, *, high, low, close, n_bars=4): + """Hand-built ``peak_mask_for_symbol``-shaped frame with EVERY bar a classifiable valley. + + Bypasses the same-slot baseline warm-up on purpose: the only thing under test is the + prev_close rule, so every day here is classifiable by construction and a day can only + be invalid because of the range / previous-close gates. + """ + rows = [] + for day in days: + base = pd.Timestamp(day) + pd.Timedelta("09:31:00") + for i in range(n_bars): + rows.append( + { + "bar_end": base + pd.Timedelta(minutes=i), + "trade_date": pd.Timestamp(day), + "volume": 100.0, + "amount": 1000.0, # per-bar price 10.0 -> valley VWAP 10.0 + "high": high, + "low": low, + "close": close, + "valley": True, + "classifiable": True, + } + ) + return pd.DataFrame(rows) + + +def test_first_day_of_the_window_has_no_prev_close_and_is_invalid(): + """A symbol's FIRST visible day cannot have a prev_close -> no value that day. + + Every day in this frame is classifiable and has a real high/low range, so the ONLY + reason the first day is absent is the missing previous close -- and the later days, + which do have one, are present. + """ + days = ["2021-07-01", "2021-07-02", "2021-07-05"] + work = _work_frame(days, high=12.0, low=8.0, close=10.0) + q = valley_price_quantile_by_day(work, min_valley_bars=1, min_classifiable=1) + assert pd.Timestamp(days[0]) not in q.index + assert pd.Timestamp(days[1]) in q.index + assert pd.Timestamp(days[2]) in q.index + # prev_close 10.0 sits inside [8, 12], so the range is unchanged: (10-8)/(12-8). + assert float(q.loc[pd.Timestamp(days[1])]) == pytest.approx(0.5) + + +def test_flat_day_with_hi_equal_lo_is_invalid(): + """hi <= lo (no visible price movement at all) -> the day is invalid, not 0/0.""" + vols, amts, _, _, closes = _test_day_arrays() + flat = [10.0] * _N + rows = _background(_BG_DAYS, _N, last_close=10.0) + _session( + "2021-07-11", vols, amts, flat, flat, closes + ) + stats = compute_valley_price_quantile_stats(_bars(rows), **_KW) + assert (_TEST_DAY, _SYM) not in stats.index + + +def test_min_valley_bars_gate_rejects_a_day_with_too_few_valleys(): + """The >= min_valley_bars floor counts TRADABLE valley bars (post-guard).""" + kw = dict(_KW, min_valley_bars=11) # the engineered day has exactly 10 + stats = compute_valley_price_quantile_stats( + _bars(_rows_with_prev_close(10.0)), **kw + ) + assert (_TEST_DAY, _SYM) not in stats.index + kw_ok = dict(_KW, min_valley_bars=10) + stats_ok = compute_valley_price_quantile_stats( + _bars(_rows_with_prev_close(10.0)), **kw_ok + ) + assert (_TEST_DAY, _SYM) in stats_ok.index + + +def test_min_classifiable_gate_rejects_a_thin_day(): + """PR-F's >= min_classifiable floor is enforced unchanged.""" + kw = dict(_KW, min_classifiable=_N + 1) + stats = compute_valley_price_quantile_stats( + _bars(_rows_with_prev_close(10.0)), **kw + ) + assert (_TEST_DAY, _SYM) not in stats.index + + +def test_zero_volume_valley_bar_contributes_nothing_but_is_still_classified(): + """The positive-trade guard drops a no-trade bar from the VWAP, not from the taxonomy.""" + vols, amts, highs, lows, closes = _test_day_arrays() + # Slot 0 is a valley; zero its trade so it cannot enter the VWAP sums. + vols[0], amts[0] = 0.0, 0.0 + rows = _background(_BG_DAYS, _N, last_close=10.0) + _session( + "2021-07-11", vols, amts, highs, lows, closes + ) + stats = compute_valley_price_quantile_stats(_bars(rows), **dict(_KW, min_valley_bars=9)) + # Remaining valleys: four at price 10 (amount 1000) and five at price 12 -> VWAP + # = (4*1000 + 5*1200) / 900 = 10000/900 + expected_vwap = (4 * 1000.0 + 5 * 1200.0) / 900.0 + assert float(stats.loc[(_TEST_DAY, _SYM)]) == pytest.approx( + (expected_vwap - _INTRADAY_LOW) / (_INTRADAY_HIGH - _INTRADAY_LOW) + ) + # ... and the day still fails a floor of 10 tradable valley bars. + thin = compute_valley_price_quantile_stats(_bars(rows), **dict(_KW, min_valley_bars=10)) + assert (_TEST_DAY, _SYM) not in thin.index + + +# --------------------------------------------------------------------------- # +# Trailing mean + valid-day floor +# --------------------------------------------------------------------------- # +def _multi_valid_day_rows(n_extra_days, quantile_targets): + """Background + ``n_extra_days`` engineered days whose quantiles are known. + + Each engineered day reuses the standard construction but scales the two valley + amount tiers so the day's valley VWAP lands on a chosen value; the range is always + [8, 12] (prev_close 10 stays inside), so q = (vwap - 8) / 4. + """ + rows = _background(_BG_DAYS, _N, last_close=10.0) + for i, vwap in enumerate(quantile_targets[:n_extra_days]): + day = (pd.Timestamp("2021-07-11") + pd.Timedelta(days=i)).strftime("%Y-%m-%d") + vols, _, highs, lows, closes = _test_day_arrays() + amts = [0.0] * _N + valley_slots = [s for s in range(_N) if s not in _ERUPT] + for s in valley_slots: + amts[s] = 100.0 * vwap # every valley bar priced at `vwap` -> VWAP = vwap + for s in _ERUPT: + amts[s] = 4000.0 + rows += _session(day, vols, amts, highs, lows, closes) + return rows + + +def test_trailing_mean_averages_the_daily_quantiles(): + """The factor is the MEAN of q_day over the trailing valid days.""" + vwaps = [9.0, 10.0, 11.0] # -> q = 0.25, 0.50, 0.75 + rows = _multi_valid_day_rows(3, vwaps) + stats = compute_valley_price_quantile_stats(_bars(rows), **dict(_KW, min_valid_days=3)) + third_day = pd.Timestamp("2021-07-13") + assert float(stats.loc[(third_day, _SYM)]) == pytest.approx((0.25 + 0.50 + 0.75) / 3) + + +def test_below_min_valid_days_is_nan(): + """Fewer than min_valid_days valid days in the window -> NaN (honest missing).""" + rows = _multi_valid_day_rows(2, [9.0, 10.0]) + stats = compute_valley_price_quantile_stats(_bars(rows), **dict(_KW, min_valid_days=3)) + assert not np.isfinite(stats.to_numpy(dtype=float)).any() + + +def test_lookback_window_drops_days_beyond_the_horizon(): + """Only the most recent ``lookback_days`` VALID days enter the mean.""" + vwaps = [8.0, 12.0, 12.0] # q = 0.0, 1.0, 1.0 + rows = _multi_valid_day_rows(3, vwaps) + stats = compute_valley_price_quantile_stats( + _bars(rows), **dict(_KW, lookback_days=2, min_valid_days=2) + ) + third_day = pd.Timestamp("2021-07-13") + assert float(stats.loc[(third_day, _SYM)]) == pytest.approx(1.0) + + +# --------------------------------------------------------------------------- # +# Reversal factor: the T-1 basis (THE lookahead subtlety of this PR) +# --------------------------------------------------------------------------- # +def _closes_frame(dates, symbols, values): + """MultiIndex(date, symbol) daily close frame.""" + idx = pd.MultiIndex.from_product([dates, symbols], names=["date", "symbol"]) + return pd.DataFrame({"close": np.asarray(values, dtype=float).reshape(-1)}, index=idx) + + +def test_reversal_20_is_minus_the_t_minus_1_twenty_day_return(): + """rev20(d) = -(close_{d-1}/close_{d-21} - 1), hand-checked on a known series.""" + dates = pd.bdate_range("2021-01-04", periods=25) + # close_i = 100 + i so every 20-day span is a known ratio. + closes = _closes_frame(dates, [_SYM], [[100.0 + i] for i in range(len(dates))]) + rev = reversal_20(closes, days=VALLEY_QUANTILE_REVERSAL_DAYS) + d = dates[21] # first date with both d-1 and d-21 present + expected = -((100.0 + 20) / (100.0 + 0) - 1.0) + assert float(rev.loc[(d, _SYM)]) == pytest.approx(expected) + # Dates without a full 21-observation history are NaN, never fabricated. + assert not np.isfinite(float(rev.loc[(dates[20], _SYM)])) + + +def test_reversal_20_ignores_day_d_close_entirely(): + """Perturbing close_d must not move rev20 AT d -- it is 15:00 information.""" + dates = pd.bdate_range("2021-01-04", periods=25) + base = [[100.0 + i] for i in range(len(dates))] + rev_a = reversal_20(_closes_frame(dates, [_SYM], base)) + bumped = [list(r) for r in base] + bumped[22][0] = 9999.0 # day d itself + rev_b = reversal_20(_closes_frame(dates, [_SYM], bumped)) + d = dates[22] + assert float(rev_a.loc[(d, _SYM)]) == float(rev_b.loc[(d, _SYM)]) + + +# --------------------------------------------------------------------------- # +# Reversal neutralization (cross-sectional OLS residual) +# --------------------------------------------------------------------------- # +def _panel_series(date, symbols, values, name): + idx = pd.MultiIndex.from_arrays( + [[pd.Timestamp(date)] * len(symbols), list(symbols)], names=["date", "symbol"] + ) + return pd.Series(np.asarray(values, dtype=float), index=idx, name=name) + + +def test_residualization_hand_computed_three_symbol_cross_section(): + """OLS residual of qbar on rev20, hand-computed. + + qbar = [0.2, 0.6, 0.7], rev20 = [-0.1, 0.0, 0.1] + x_bar = 0, y_bar = 0.5 + Sxy = (-0.1)(-0.3) + 0*(0.1) + (0.1)(0.2) = 0.05 ; Sxx = 0.01 + 0 + 0.01 = 0.02 + slope = 2.5, intercept = 0.5 + fitted = [0.25, 0.5, 0.75] -> residuals = [-0.05, +0.10, -0.05] (sum 0) + """ + syms = ["A", "B", "C"] + qbar = _panel_series("2023-05-10", syms, [0.2, 0.6, 0.7], "q") + rev = _panel_series("2023-05-10", syms, [-0.1, 0.0, 0.1], "rev") + out = residualize_on_reversal(qbar, rev, min_cross_section=3, name="f") + assert float(out.loc[(pd.Timestamp("2023-05-10"), "A")]) == pytest.approx(-0.05) + assert float(out.loc[(pd.Timestamp("2023-05-10"), "B")]) == pytest.approx(0.10) + assert float(out.loc[(pd.Timestamp("2023-05-10"), "C")]) == pytest.approx(-0.05) + assert float(out.sum()) == pytest.approx(0.0, abs=1e-12) + + +def test_residualization_removes_the_reversal_exposure(): + """The residual is ORTHOGONAL to rev20 -- the point of the neutralization.""" + rng = np.random.default_rng(20260720) + syms = [f"S{i}" for i in range(30)] + rev_vals = rng.normal(size=30) + qbar_vals = 0.4 + 1.7 * rev_vals + 0.05 * rng.normal(size=30) # strong exposure + qbar = _panel_series("2023-05-10", syms, qbar_vals, "q") + rev = _panel_series("2023-05-10", syms, rev_vals, "rev") + assert abs(np.corrcoef(qbar_vals, rev_vals)[0, 1]) > 0.9 # before + out = residualize_on_reversal(qbar, rev, min_cross_section=10, name="f") + resid = out.to_numpy(dtype=float) + assert abs(np.corrcoef(resid, rev_vals)[0, 1]) < 1e-10 # after + + +def test_residualization_drops_symbols_without_a_reversal_value(): + """A symbol whose rev20 is missing gets a NaN residual -- never a silent 0 fill.""" + syms = ["A", "B", "C", "D"] + qbar = _panel_series("2023-05-10", syms, [0.2, 0.6, 0.7, 0.9], "q") + rev = _panel_series("2023-05-10", syms, [-0.1, 0.0, 0.1, np.nan], "rev") + out = residualize_on_reversal(qbar, rev, min_cross_section=3, name="f") + assert not np.isfinite(float(out.loc[(pd.Timestamp("2023-05-10"), "D")])) + # The other three are the hand-computed regression above: D did not perturb it. + assert float(out.loc[(pd.Timestamp("2023-05-10"), "B")]) == pytest.approx(0.10) + + +def test_residualization_is_invariant_to_the_reversal_sign_convention(): + """Residualizing on ``rev`` and on ``-rev`` gives the SAME factor. + + Matters because the report's "正向暴露 20 日反转因子" does not say whether its 反转因子 + is the past return or its negation, and our prototype measures the exposure with the + opposite label to the report's. An intercept OLS residual spans the same column space + either way (span{1, x} == span{1, -x}), so the shipped factor cannot depend on which + convention was meant -- the ambiguity is real but harmless, and this test proves it. + """ + syms = [f"S{i}" for i in range(12)] + rng = np.random.default_rng(11) + qbar = _panel_series("2023-05-10", syms, rng.normal(size=12), "q") + rev = _panel_series("2023-05-10", syms, rng.normal(size=12), "rev") + a = residualize_on_reversal(qbar, rev, min_cross_section=10, name="f") + b = residualize_on_reversal(qbar, -rev, min_cross_section=10, name="f") + np.testing.assert_allclose(a.to_numpy(dtype=float), b.to_numpy(dtype=float), atol=1e-12) + + +def test_residualization_below_min_cross_section_is_all_nan(): + """A thin cross-section -> the whole date is NaN (honest missing).""" + syms = ["A", "B", "C"] + qbar = _panel_series("2023-05-10", syms, [0.2, 0.6, 0.7], "q") + rev = _panel_series("2023-05-10", syms, [-0.1, 0.0, 0.1], "rev") + out = residualize_on_reversal(qbar, rev, min_cross_section=4, name="f") + assert not np.isfinite(out.to_numpy(dtype=float)).any() + + +def test_residualization_degenerate_reversal_cross_section_is_nan(): + """Zero-variance rev20 cannot identify a slope -> NaN, not an unresidualized qbar.""" + syms = ["A", "B", "C"] + qbar = _panel_series("2023-05-10", syms, [0.2, 0.6, 0.7], "q") + rev = _panel_series("2023-05-10", syms, [0.3, 0.3, 0.3], "rev") + out = residualize_on_reversal(qbar, rev, min_cross_section=3, name="f") + assert not np.isfinite(out.to_numpy(dtype=float)).any() + + +def test_residualization_is_per_date_and_never_pools_dates(): + """Each date is regressed on ITS OWN cross-section; a second date cannot leak in.""" + syms = ["A", "B", "C"] + q1 = _panel_series("2023-05-10", syms, [0.2, 0.6, 0.7], "q") + r1 = _panel_series("2023-05-10", syms, [-0.1, 0.0, 0.1], "rev") + q2 = _panel_series("2023-05-11", syms, [5.0, 9.0, 40.0], "q") + r2 = _panel_series("2023-05-11", syms, [3.0, 1.0, -8.0], "rev") + both = residualize_on_reversal( + pd.concat([q1, q2]), pd.concat([r1, r2]), min_cross_section=3, name="f" + ) + alone = residualize_on_reversal(q1, r1, min_cross_section=3, name="f") + for s in syms: + key = (pd.Timestamp("2023-05-10"), s) + assert float(both.loc[key]) == pytest.approx(float(alone.loc[key])) + + +# --------------------------------------------------------------------------- # +# END-TO-END: the T-1 reversal basis is what the FACTOR uses (the critical test) +# --------------------------------------------------------------------------- # +def _end_to_end_inputs(n_syms=3, n_days=30): + """Minute bars for ``n_syms`` symbols + a daily qfq close panel over the same dates. + + Each symbol gets the standard background plus engineered days with per-symbol + quantiles, so every symbol has a finite qbar on the later dates and the + cross-section is large enough to regress. + """ + rows = [] + vwap_cycle = [8.5, 9.5, 10.5, 11.5] + for k in range(n_syms): + sym = f"S{k}.SZ" + rows += _background(_BG_DAYS, _N, sym=sym, last_close=10.0) + for i in range(n_days): + day = (pd.Timestamp("2021-07-11") + pd.Timedelta(days=i)).strftime("%Y-%m-%d") + vols, _, highs, lows, closes = _test_day_arrays() + vwap = vwap_cycle[(i + k) % len(vwap_cycle)] + amts = [0.0] * _N + for s in range(_N): + amts[s] = 4000.0 if s in _ERUPT else 100.0 * vwap + rows += _session(day, vols, amts, highs, lows, closes, sym=sym) + bars = _bars(rows) + + dates = sorted(pd.unique(bars.index.get_level_values("time").normalize())) + syms = [f"S{k}.SZ" for k in range(n_syms)] + rng = np.random.default_rng(7) + vals = 100.0 * np.exp(np.cumsum(rng.normal(0, 0.02, size=(len(dates), n_syms)), axis=0)) + closes = pd.DataFrame( + {"close": vals.reshape(-1)}, + index=pd.MultiIndex.from_product([dates, syms], names=["date", "symbol"]), + ) + return bars, closes, dates, syms + + +_E2E_KW = dict( + baseline_days=VOLUME_PRV_BASELINE_DAYS, + baseline_min_obs=VOLUME_PRV_BASELINE_MIN_OBS, + sigma_k=VOLUME_PRV_SIGMA_K, + lookback_days=5, + min_valid_days=3, + min_classifiable=1, + min_valley_bars=1, + min_cross_section=3, +) + + +def test_reversal_uses_t_minus_1_close_not_day_d(): + """THE critical lookahead test: perturbing day d's CLOSE leaves the factor at d fixed. + + The report's reversal is ``-(close_d/close_{d-20} - 1)``, which reads the 15:00 close + on the very day we decide at 14:50. We use the T-1 basis instead, so day d's close is + NOT an input to the factor value at d. If someone "simplifies" the implementation back + to the report's naive form, this test fails. + + THE PERTURBATION MUST BE SINGLE-SYMBOL, and that is not a stylistic choice. Bumping + EVERY cross-section member's ``close_d`` by the same factor ``k`` sends the naive + regressor ``1 - close_d/close_{d-20}`` to ``k*rev - (k-1)`` -- an AFFINE + reparametrization of x. Intercept-OLS residuals are invariant under exactly that + (``span{1, x} == span{1, a*x + b}``), which is the very property + ``test_residualization_is_invariant_to_the_reversal_sign_convention`` relies on. So a + uniform bump leaves the residuals bit-identical under BOTH the T-1 and the naive form + and can never tell them apart. Perturbing ONE symbol moves that symbol's regressor + relative to the rest of the cross-section, changes the fitted line, and therefore + moves every residual on the date -- under the naive form only. Verified by + monkeypatching ``reversal_20`` to the naive ``shift(0)/shift(days)`` form: this test + FAILS there and passes here. + """ + bars, closes, dates, syms = _end_to_end_inputs() + base = compute_valley_price_quantile(bars, closes, **_E2E_KW) + + d = dates[-3] # a date with a finite factor value and a full history behind it + bumped = closes.copy() + # SINGLE symbol, so the change is NOT an affine map of the whole regressor vector. + bumped.loc[(d, syms[0]), "close"] = float(bumped.loc[(d, syms[0]), "close"]) * 1.5 + after = compute_valley_price_quantile(bars, bumped, **_E2E_KW) + + on_d_before = base[base.index.get_level_values("date") == d] + on_d_after = after[after.index.get_level_values("date") == d] + assert on_d_before.notna().any(), "test is vacuous unless d carries a finite value" + pd.testing.assert_series_equal(on_d_before, on_d_after) + + +def test_uniform_close_d_bump_cannot_distinguish_the_two_reversal_bases(): + """Documents WHY the test above must perturb a single symbol. + + A uniform bump is an affine reparametrization of the OLS regressor, so it leaves the + residuals unchanged even under the naive close_d form. Asserting that here keeps the + reasoning from being quietly lost: if someone later "simplifies" the critical test + back to a uniform bump, this test explains what they broke. + """ + syms = [f"S{i}" for i in range(12)] + idx = pd.MultiIndex.from_arrays( + [[pd.Timestamp("2023-05-10")] * 12, syms], names=["date", "symbol"] + ) + rng = np.random.default_rng(4) + qbar = pd.Series(rng.normal(size=12), index=idx) + rev = pd.Series(rng.normal(size=12), index=idx) + + base = residualize_on_reversal(qbar, rev, min_cross_section=10, name="f") + # k=1.5 uniform bump of close_d maps rev -> 1.5*rev - 0.5 for EVERY symbol. + affine = residualize_on_reversal(qbar, 1.5 * rev - 0.5, min_cross_section=10, name="f") + np.testing.assert_allclose( + base.to_numpy(dtype=float), affine.to_numpy(dtype=float), atol=1e-12 + ) + + # Moving ONE symbol is not affine on the vector, so it DOES move the residuals. + single = rev.copy() + single.iloc[0] = single.iloc[0] * 1.5 - 0.5 + moved = residualize_on_reversal(qbar, single, min_cross_section=10, name="f") + assert not np.allclose(base.to_numpy(dtype=float), moved.to_numpy(dtype=float)) + + +def test_perturbing_the_t_minus_1_close_does_move_the_factor(): + """TEETH for the test above: the closes are genuinely an input, just lagged by one day. + + Without this, the T-1 test could pass simply by the reversal never being applied. + """ + bars, closes, dates, syms = _end_to_end_inputs() + base = compute_valley_price_quantile(bars, closes, **_E2E_KW) + + d = dates[-3] + d_prev = dates[-4] + bumped = closes.copy() + bumped.loc[(d_prev, syms[0]), "close"] = float( + bumped.loc[(d_prev, syms[0]), "close"] + ) * 1.5 + after = compute_valley_price_quantile(bars, bumped, **_E2E_KW) + + on_d_before = base[base.index.get_level_values("date") == d] + on_d_after = after[after.index.get_level_values("date") == d] + assert not on_d_before.equals(on_d_after), ( + "perturbing close_{d-1} must change rev20 at d and hence the residualized factor" + ) + + +def test_factor_is_residualized_not_raw_qbar(): + """The returned factor differs from the raw trailing quantile mean. + + Guards against a wiring bug where the reversal neutralization is computed but the + RAW qbar is returned. + """ + bars, closes, dates, syms = _end_to_end_inputs() + raw = compute_valley_price_quantile_stats( + bars, + **{k: v for k, v in _E2E_KW.items() if k != "min_cross_section"}, + ) + final = compute_valley_price_quantile(bars, closes, **_E2E_KW) + common = raw.index.intersection(final.index) + assert len(common) > 0 + assert not np.allclose( + raw.loc[common].to_numpy(dtype=float), + final.loc[common].to_numpy(dtype=float), + equal_nan=True, + ) + + +# --------------------------------------------------------------------------- # +# PIT / leakage / isolation +# --------------------------------------------------------------------------- # +def test_bars_after_the_cutoff_are_invisible_and_the_test_has_teeth(): + """Post-14:50 bars cannot move the value; PRE-cutoff bars can (the teeth).""" + vols, amts, highs, lows, closes = _test_day_arrays() + rows = _background(_BG_DAYS, _N, last_close=10.0) + _session( + "2021-07-11", vols, amts, highs, lows, closes + ) + base = compute_valley_price_quantile_stats(_bars(rows), **_KW) + + # A wild bar at 14:55 (available 14:56) is AFTER the cutoff -> invisible. + late = rows + [ + (pd.Timestamp("2021-07-11 14:55:00"), _SYM, 100.0, 500000.0, 9999.0, 0.01, 5000.0) + ] + after = compute_valley_price_quantile_stats(_bars(late), **_KW) + assert float(after.loc[(_TEST_DAY, _SYM)]) == float(base.loc[(_TEST_DAY, _SYM)]) + + # The SAME bar placed at 10:00 (visible) DOES move it -> the test is not vacuous. + early = rows + [ + (pd.Timestamp("2021-07-11 10:00:00"), _SYM, 100.0, 500000.0, 9999.0, 0.01, 5000.0) + ] + moved = compute_valley_price_quantile_stats(_bars(early), **_KW) + assert float(moved.loc[(_TEST_DAY, _SYM)]) != float(base.loc[(_TEST_DAY, _SYM)]) + + +def test_prev_close_comes_from_the_visible_window_not_the_real_daily_close(): + """prev_close is the last bar <= 14:50 of d-1, NOT d-1's 15:00 close. + + A late bar on the PREVIOUS day (after 14:50) must not become the prev_close -- that + is the single-visibility-definition pin. + """ + rows = _rows_with_prev_close(10.0) + base = compute_valley_price_quantile_stats(_bars(rows), **_KW) + prev_day = (pd.Timestamp("2021-07-11") - pd.Timedelta(days=1)).strftime("%Y-%m-%d") + with_late_close = rows + [ + (pd.Timestamp(f"{prev_day} 14:56:00"), _SYM, 100.0, 1000.0, 50.0, 50.0, 50.0) + ] + after = compute_valley_price_quantile_stats(_bars(with_late_close), **_KW) + assert float(after.loc[(_TEST_DAY, _SYM)]) == float(base.loc[(_TEST_DAY, _SYM)]) + + +def test_cross_symbol_isolation(): + """A second symbol's bars never touch the first symbol's per-symbol statistics.""" + rows_a = _rows_with_prev_close(10.0) + alone = compute_valley_price_quantile_stats(_bars(rows_a), **_KW) + + other = [] + for r in _rows_with_prev_close(16.0): + other.append((r[0], "600000.SH", r[2] * 7.0, r[3] * 3.0, r[4], r[5], r[6])) + together = compute_valley_price_quantile_stats(_bars(rows_a + other), **_KW) + assert float(together.loc[(_TEST_DAY, _SYM)]) == float(alone.loc[(_TEST_DAY, _SYM)]) + + +def test_prev_close_does_not_cross_symbols(): + """Symbol B's previous-day close cannot become symbol A's prev_close.""" + rows_a = _rows_with_prev_close(4.0) + alone = compute_valley_price_quantile_stats(_bars(rows_a), **_KW) + other = [ + (r[0], "600000.SH", r[2], r[3], r[4], r[5], r[6] * 25.0) + for r in _rows_with_prev_close(4.0) + ] + together = compute_valley_price_quantile_stats(_bars(rows_a + other), **_KW) + assert float(together.loc[(_TEST_DAY, _SYM)]) == float(alone.loc[(_TEST_DAY, _SYM)]) + + +def test_empty_bars_return_empty_schema_shaped_output(): + out = compute_valley_price_quantile_stats(empty_intraday_bars(), **_KW) + assert out.empty + assert list(out.index.names) == ["date", "symbol"] + + +def test_inputs_are_never_mutated(): + bars = _bars(_rows_with_prev_close(10.0)) + before = bars.copy(deep=True) + compute_valley_price_quantile_stats(bars, **_KW) + pd.testing.assert_frame_equal(bars, before) + + +def test_parameter_validation(): + bars = _bars(_rows_with_prev_close(10.0)) + with pytest.raises(ValueError, match="lookback_days"): + compute_valley_price_quantile_stats(bars, **dict(_KW, lookback_days=0)) + with pytest.raises(ValueError, match="min_valley_bars"): + compute_valley_price_quantile_stats(bars, **dict(_KW, min_valley_bars=0)) + with pytest.raises(ValueError, match="min_cross_section"): + residualize_on_reversal( + _panel_series("2023-05-10", ["A"], [0.1], "q"), + _panel_series("2023-05-10", ["A"], [0.1], "r"), + min_cross_section=1, + name="f", + ) + + +# --------------------------------------------------------------------------- # +# Factor class / spec +# --------------------------------------------------------------------------- # +def test_factor_selects_the_pre_aggregated_column(): + f = ValleyPriceQuantileFactor() + assert f.name == f"valley_price_quantile_{VALLEY_QUANTILE_LOOKBACK_DAYS}" + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp("2023-01-03"), "A")], names=["date", "symbol"] + ) + panel = pd.DataFrame({f.name: [0.25], "close": [10.0]}, index=idx) + out = f.compute(panel) + assert float(out.iloc[0]) == pytest.approx(0.25) + assert out.name == f.name + + +def test_factor_raises_readably_without_the_column(): + f = ValleyPriceQuantileFactor() + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp("2023-01-03"), "A")], names=["date", "symbol"] + ) + with pytest.raises(ValueError, match="pre-aggregated"): + f.compute(pd.DataFrame({"close": [10.0]}, index=idx)) + + +def test_spec_declares_the_pre_registered_sign_and_the_pinned_deviations(): + spec = ValleyPriceQuantileFactor().spec + assert isinstance(spec, FactorSpec) + assert spec.expected_ic_sign == 1 + assert spec.is_intraday is False + assert spec.return_basis == "close_to_close" + text = spec.description + # The two subtleties this PR exists to get right must be ON the spec. + assert "prev_close" in text or "previous" in text.lower() + assert "14:50" in text + assert "reversal" in text.lower() + assert "d-1" in text or "T-1" in text or "t-1" in text.lower() + + +def test_factor_rejects_a_bad_lookback(): + with pytest.raises(ValueError, match="lookback_days"): + ValleyPriceQuantileFactor(lookback_days=0) + + +# --------------------------------------------------------------------------- # +# The five already-merged factors must not drift +# --------------------------------------------------------------------------- # +def test_prior_factor_modules_are_untouched_by_this_pr(): + """PR-L must not have changed the shared classification or the five prior factors. + + A cheap structural guard: the reused entry points still exist with the same + behaviour on a trivial input, so an accidental edit to the shared module shows up + here as well as in the five factors' own test files. + """ + from data.clean.intraday_ridge_return import compute_ridge_minute_return + from data.clean.intraday_valley_ridge_vwap import compute_valley_ridge_vwap_ratio + from data.clean.intraday_valley_vwap import compute_valley_relative_vwap + + empty = empty_intraday_bars() + for fn in ( + compute_valley_relative_vwap, + compute_valley_ridge_vwap_ratio, + compute_ridge_minute_return, + ): + out = fn(empty) + assert out.empty + assert list(out.index.names) == ["date", "symbol"] + + # The shared taxonomy constants are the ones PR-F pinned (unchanged by PR-L). + assert VOLUME_PRV_SIGMA_K == 1.0 + assert VOLUME_PRV_BASELINE_DAYS == 20 + assert VOLUME_PRV_BASELINE_MIN_OBS == 10 + assert VOLUME_PRV_MIN_CLASSIFIABLE == 100 + assert VOLUME_PRV_MIN_VALID_DAYS == 10 + + +def test_definition_constants_are_the_pinned_values(): + assert VALLEY_QUANTILE_LOOKBACK_DAYS == 20 + assert VALLEY_QUANTILE_MIN_VALLEY_BARS == 20 + assert VALLEY_QUANTILE_REVERSAL_DAYS == 20 + assert VALLEY_QUANTILE_MIN_CROSS_SECTION == 10 + + +def test_valley_price_quantile_by_day_is_reusable_on_a_prepared_frame(): + """The per-day seam is callable directly (what the trailing mean is built on).""" + bars = _bars(_rows_with_prev_close(16.0)) + visible = prepare_visible_minute_bars( + bars, extra_columns=("amount", "high", "low", "close") + ) + work = peak_mask_for_symbol(visible.reset_index(drop=True)) + q = valley_price_quantile_by_day(work, min_valley_bars=1, min_classifiable=1) + assert float(q.loc[_TEST_DAY]) == pytest.approx(0.375)