diff --git a/config/phase_i_valley_relative_vwap.yaml b/config/phase_i_valley_relative_vwap.yaml new file mode 100644 index 0000000..3b54c2c --- /dev/null +++ b/config/phase_i_valley_relative_vwap.yaml @@ -0,0 +1,150 @@ +# PR-I — Seventh real factor evaluation: VALLEY-RELATIVE VWAP. +# +# Reproduces the THIRD factor of the Kaiyuan market-microstructure series #27 (开源证券 +# 《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, reportId 4957417, §4) — the +# "量谷相对加权价格因子" — as a first-class ValleyRelativeVwapFactor 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). +# +# SAME MACHINE as PR-F / PR-H, DIFFERENT FAMILY. The minute classification is REUSED from +# data/clean/intraday_volume_prv.py (not re-implemented): PIT-truncate each day at 14:50, +# classify every visible minute against its SAME-SLOT strictly-prior 20-day baseline +# (ERUPTIVE if vol > mu + sigma, else a VALLEY 量谷). Where PR-F counted the PEAKS and PR-H +# measured their TIMING -- both null results -- this factor prices the VALLEYS: the daily +# ratio (valley VWAP) / (whole visible day VWAP), averaged over the trailing 20 VALID days. +# Because the classification is shared verbatim, a different verdict here cannot be an +# artefact of a differently-implemented taxonomy. +# +# PINNED choices (disclosed on the factor spec): +# (1) each VWAP uses the aggregation identity sum(p*v)/sum(v) == sum(amount)/sum(volume), +# i.e. the day's REAL volume-weighted price rather than a close approximation; +# (2) bars with non-finite or non-positive volume OR amount are dropped from BOTH sums; +# the guard runs at the summation step only, so PR-F's same-slot baseline -- and +# therefore volume_peak_count / peak_interval_kurtosis -- is bit-identical; +# (3) RAW (unadjusted) prices are correct here: the adjustment factor is constant within +# a day and cancels exactly in the ratio (same reasoning as PR-D amplitude / I5b); +# (4) DEVIATION FROM THE REPORT, disclosed and NOT silently equated: our day VWAP 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; +# (5) a day is VALID iff it has >= 100 classifiable bars AND >= 20 TRADABLE valley bars +# (counted AFTER the positive-trade guard) AND positive volume in both denominators; +# NaN below 10 valid days in the trailing window (honest missing). +# +# Pre-registered sign = +1 (report full-market RankIC +8.69% / RankICIR 4.44, long-short +# 25.35%/yr, IR 3.04, monthly win rate 79.7% -- the strongest factor in the report; CSI500 +# sub-domain long-short 9.94% / IR 1.26, the closest comparable to this cell). NOTE the +# report is a MONTHLY, market-cap + industry neutral FULL-MARKET series on Wind data -- +# our eval cell is CSI500 daily with industry + size neutral, so those numbers are a LOOSE +# reference only and are NOT written in as expected values. Raw minute volume (cached +# as-is) has split-day magnitude jumps that pollute the 20-day sigma; the report (Wind) +# does not adjust for this either, so it is disclosed and NOT corrected. +# +# Eval CELL is IDENTICAL to PR-C / PR-D / PR-E / PR-F / PR-G / PR-H: universe = CSI500 +# (000905.SH), PIT membership; window = 2021-07-01 .. 2026-06-30 (the project's +# minute-coverage window); daily rebalance; OOS split 2024-01-01; book = +# value_ep/value_bp/volatility_20; fee 0.001. The ONLY substantive difference is the +# subject factor, which is minute-derived and computed by the runner from the intraday +# cache -- it is NOT a config-listed daily factor. Because the cell matches PR-F / PR-H +# exactly, this run is directly comparable to its siblings. The evaluator runs TWICE (see +# qt/eval_valley_relative_vwap.py): once with NO book (Incremental NOT_ASSESSED) and once +# with the confirmed book -- which matters more than usual here, since a relative PRICE +# LEVEL is a plausible cousin of a value signal and the with-book run is what says whether +# this is a genuinely new bet. + +project: + name: quantitative_trading_pr_i_valley_relative_vwap + 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_relative_vwap_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_relative_vwap_20) is minute-derived and computed by the runner +# from the intraday cache -- it is NOT a config-listed daily factor. +factors: +- name: value_ep + enabled: true +- name: value_bp + enabled: true +- name: volatility_20 + enabled: true + params: + window: 20 + price_col: close +processing: + drop_missing: true + standardize: + enabled: true + method: zscore + # winsorize is a P0 no-op in this codebase; the EvalConfig declares winsorize=None + # accordingly (nothing is clipped), so the report never overstates preprocessing. + winsorize: + enabled: false + method: mad + n: 3.0 + # Industry + market-cap neutralization (SW-L1), matching the report's neutral column + # and the EvalConfig neutralization declaration. + neutralize: + enabled: true + industry_col: industry + size_col: market_cap + industry_level: L1 +alpha: + model: equal_weight + params: {} +portfolio: + # Required by the schema but unused: this runner evaluates a factor, it does not + # build/execute a portfolio. + constructor: topn_equal_weight + top_n: 50 + long_only: true + max_weight: null + turnover_cap: null +backtest: + # Required by the schema but unused (no backtest is run). The EvalConfig uses a DAILY + # rebalance (contract default), independent of this field. + initial_nav: 1.0 + rebalance: monthly + event_order: close_to_next_period + cash_return: 0.0 +cost: + fee_rate: 0.001 + slippage_rate: 0.0 + turnover_formula: l1 +analytics: + forward_return_periods: + - 1 + quantiles: 5 + benchmark: null +# OOS split (window midpoint) so the OOS section runs and the Predictive axis can be +# assessed (sign consistency across both holdout subperiods). +oos: + split_date: '2024-01-01' +output: + root_dir: artifacts + data_dir: artifacts/data + factor_dir: artifacts/factors + report_dir: artifacts/reports + log_dir: artifacts/logs + overwrite: true diff --git a/data/clean/intraday_valley_vwap.py b/data/clean/intraday_valley_vwap.py new file mode 100644 index 0000000..35c704e --- /dev/null +++ b/data/clean/intraday_valley_vwap.py @@ -0,0 +1,307 @@ +"""VALLEY-RELATIVE VWAP factor (PR-I). + +Reproduces the THIRD factor of the Kaiyuan market-microstructure series #27 (开源证券 +《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, 2025-07-20, reportId 4957417, +§4): "参考聪明钱因子的构建方式，计算每日量谷的成交量加权价格，与当日总成交量加权价格做比， +并计算 20 日均值作为量谷相对加权价格因子". + +Same MACHINE as PR-F / PR-H, different FAMILY. The volume classification is REUSED +verbatim from :mod:`data.clean.intraday_volume_prv` (``prepare_visible_minute_bars`` + +``peak_mask_for_symbol``) — same-slot strictly-prior μ+kσ eruptive/mild test, same +classifiable rule, same valid-day floor. A VALLEY (量谷) is exactly that module's +``valley`` column: a classifiable, NON-eruptive minute. Nothing about the taxonomy is +re-implemented here, so the three factors can never drift apart. Where PR-F counts peaks +and PR-H measures the TIMING of peaks, this module measures the PRICE LEVEL at the +valleys — a different statistic on a different part of the taxonomy, which is why it is +worth running after two null results on the peak-counting family. + +PINNED choices (deliberate and DISCLOSED, not tuned knobs; reproduced on the factor spec +so a reader sees exactly what was assumed): + + 1. VWAP VIA THE AGGREGATION IDENTITY. A bar's volume-weighted price is + ``p = amount / volume``, so a set's volume-weighted price + ``Σ(p_i·v_i) / Σv_i`` collapses EXACTLY to ``Σamount / Σvolume``. Both legs use + that identity: valley VWAP = Σamount(valley bars)/Σvolume(valley bars), day VWAP + = Σamount(all visible bars)/Σvolume(all visible bars). This is the day's REAL + VWAP, strictly better than approximating each bar by its close. + 2. POSITIVE-TRADE GUARD. A bar with non-finite or non-positive ``volume`` OR + ``amount`` carries no price information (``amount/volume`` is meaningless or + degenerate), so it is dropped from BOTH sums. The guard runs at the summation + step only — never before classification — because the same-slot μ/σ baseline is + PR-F's and must stay bit-identical. A zero-volume minute is therefore still + classified (and still a valley) but contributes nothing to either VWAP. + 3. RAW (UNADJUSTED) PRICES. The cached minute bars are unadjusted. Both legs are + sums over the SAME trading day, and a split/dividend adjustment factor is constant + within a day, so it cancels exactly in the ratio — no adjustment is needed (the + same reasoning PR-D's amplitude and I5b's price-limit checks rely on). + 4. THE DAY VWAP COVERS THE PIT-VISIBLE WINDOW ONLY (09:31–14:50), not the full + session. This is a NECESSARY DEVIATION from the report, which uses the whole day: + the standing 14:50 decision cutoff truncates history days and the signal day + identically, and reading the closing auction would be lookahead at our decision + time. Disclosed, not silently equated to the report's construction. + 5. VALLEY-BAR COUNT IS TAKEN AFTER THE GUARD. The ``min_valley_bars`` floor counts + the valley minutes that actually CONTRIBUTE to the numerator, so a day cannot + qualify on the strength of valley bars that traded nothing. + +Factor value: ``valley_relative_vwap_20`` = the MEAN of the daily ratio +``valley VWAP / day VWAP`` over the symbol's most recent ``lookback_days`` (=20) VALID +trading days INCLUDING ``d``. A day is VALID iff it clears three gates: at least +``min_classifiable`` (=100) classifiable bars (PR-F's gate, unchanged), at least +``min_valley_bars`` (=20) tradable valley bars, and strictly positive volume in BOTH +denominators. Fewer than ``min_valid_days`` (=10) valid days in the trailing window -> +NaN (honest missing; the runner discloses the coverage). Values are emitted only on +valid days. + +Pre-registered sign = +1. The report's full-market RankIC is +8.69% / RankICIR 4.44 +(long-short 25.35%/yr, IR 3.04, monthly win rate 79.7%) — the strongest factor in the +report; its CSI500 sub-domain long-short is 9.94% / IR 1.26, the closest comparable to +our eval cell. Semantics per the report: valley minutes are moments of subdued trading +sentiment where prices are unlikely to have over-reacted, so a HIGH relative valley +price means the calm, informed part of the day was bid up -> 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, so a factor value never sees a future +bar (invariant #1); it is a DAILY signal traded close-to-close from d+1. This module is +DATA-layer only: it does not fetch, does not touch factors / alpha / portfolio / +runtime, and never sees a token. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + +from data.clean.intraday_aggregate import DAILY_INDEX_NAMES, DEFAULT_DECISION_TIME +from data.clean.intraday_schema import SYMBOL_LEVEL, validate_intraday_bars +from data.clean.intraday_volume_prv import ( + VOLUME_PRV_BASELINE_DAYS, + VOLUME_PRV_BASELINE_MIN_OBS, + VOLUME_PRV_MIN_CLASSIFIABLE, + VOLUME_PRV_MIN_VALID_DAYS, + VOLUME_PRV_SIGMA_K, + peak_mask_for_symbol, + prepare_visible_minute_bars, +) + +# Factor DEFINITION constants (pinned interpretations of the report; NOT tuned knobs). +# The classification constants are IMPORTED from PR-F, never redefined. +VALLEY_VWAP_LOOKBACK_DAYS = 20 # trailing VALID trading-day mean window, includes d +VALLEY_VWAP_MIN_VALLEY_BARS = 20 # min TRADABLE valley bars for a valid day + +# The extra 1min column this family needs on top of PR-F's (volume): the traded value, +# which turns the bar set into a volume-weighted price via Σamount/Σvolume. +_AMOUNT = "amount" + + +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_vwap_ratio_by_day( + work: pd.DataFrame, + *, + min_valley_bars: int = VALLEY_VWAP_MIN_VALLEY_BARS, + min_classifiable: int = VOLUME_PRV_MIN_CLASSIFIABLE, +) -> pd.Series: + """Daily ``valley VWAP / day VWAP`` ratio for ONE symbol, on VALID days only. + + ``work`` is one symbol's frame as returned by + :func:`~data.clean.intraday_volume_prv.peak_mask_for_symbol`, which must have been + built from bars prepared with ``extra_columns=("amount",)`` so the traded value is + available alongside the ``valley`` / ``classifiable`` masks. + + Both VWAPs use the ``Σamount / Σvolume`` aggregation identity (PINNED §1 of the + module docstring). The POSITIVE-TRADE GUARD (§2) drops non-finite / non-positive + volume or amount bars from both sums; the day leg spans ALL visible bars that clear + the guard (eruptive minutes included — it is the WHOLE visible day's VWAP), while + the valley leg spans only the guarded valley bars. + + Returns: + Series indexed by ``trade_date`` (ascending) holding the ratio for the days that + clear all three validity gates — invalid days are ABSENT, not NaN, so they do + not occupy a slot in the caller's trailing window (the same rule PR-F/PR-H use). + """ + vol = work["volume"].to_numpy(dtype=float) + amt = work[_AMOUNT].to_numpy(dtype=float) + # Positive-trade guard: no price information in a bar that traded nothing (or whose + # fields are non-finite / negative). Applied HERE, at the summation step, never + # before classification — PR-F's same-slot baseline must stay bit-identical. + tradable = np.isfinite(vol) & (vol > 0.0) & np.isfinite(amt) & (amt > 0.0) + valley = work["valley"].to_numpy(dtype=bool) & tradable + + per_bar = pd.DataFrame( + { + "trade_date": work["trade_date"].to_numpy(), + "day_amt": np.where(tradable, amt, 0.0), + "day_vol": np.where(tradable, vol, 0.0), + "valley_amt": np.where(valley, amt, 0.0), + "valley_vol": np.where(valley, vol, 0.0), + "valley_bars": valley.astype(np.int64), + "classifiable_bars": work["classifiable"].to_numpy(dtype=bool).astype( + np.int64 + ), + } + ) + agg = per_bar.groupby("trade_date", sort=True).sum() + + # Three validity gates (§ module docstring / task card §1.4): PR-F's classifiable + # floor (unchanged), enough TRADABLE valley bars, and a positive denominator on both + # legs (a 0/0 ratio is never fabricated). + valid = ( + (agg["classifiable_bars"] >= min_classifiable) + & (agg["valley_bars"] >= min_valley_bars) + & (agg["day_vol"] > 0.0) + & (agg["valley_vol"] > 0.0) + ) + ok = agg.loc[valid] + if ok.empty: + return pd.Series( + [], index=pd.DatetimeIndex([], name="trade_date"), dtype=float + ) + valley_vwap = ok["valley_amt"] / ok["valley_vol"] + day_vwap = ok["day_amt"] / ok["day_vol"] + return (valley_vwap / day_vwap).astype(float) + + +def _valley_ratio_mean_for_symbol( + g: pd.DataFrame, + *, + baseline_days: int, + baseline_min_obs: int, + sigma_k: float, + lookback_days: int, + min_valid_days: int, + min_classifiable: int, + min_valley_bars: int, +) -> tuple[list[pd.Timestamp], list[float]]: + """Daily valley-relative-VWAP values for ONE symbol from its PIT-visible bars. + + Classifies the minutes with the REUSED :func:`peak_mask_for_symbol`, reduces each + valid day to its VWAP ratio, then takes the trailing-``lookback_days``-valid-day + mean. No cross-symbol leakage (``g`` is one symbol's slice) and no lookahead (the + baseline is strictly prior, the mean window is trailing). + """ + work = peak_mask_for_symbol( + g, + baseline_days=baseline_days, + baseline_min_obs=baseline_min_obs, + sigma_k=sigma_k, + ) + ratio = valley_vwap_ratio_by_day( + work, min_valley_bars=min_valley_bars, min_classifiable=min_classifiable + ) + if ratio.empty: + return [], [] + + # Mean over the trailing lookback_days VALID days (including d); NaN until + # min_valid_days valid days have accumulated. Emitted only on valid days. + rolled = ratio.sort_index().rolling(lookback_days, min_periods=min_valid_days).mean() + days = [pd.Timestamp(d).normalize() for d in rolled.index] + return days, list(rolled.to_numpy(dtype=float)) + + +def compute_valley_relative_vwap( + bars: pd.DataFrame, + *, + lookback_days: int = VALLEY_VWAP_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_VWAP_MIN_VALLEY_BARS, + decision_time: str = DEFAULT_DECISION_TIME, + name: str = "valley_relative_vwap", +) -> pd.Series: + """PIT-safe daily "valley-relative VWAP" factor from 1min ``bars``. + + Takes normalized 1min ``bars``, PIT-truncates each day at ``decision_time``, + classifies every visible minute with the REUSED PR-F taxonomy, computes each valid + day's ``valley VWAP / whole-visible-day VWAP`` ratio via the ``Σamount/Σvolume`` + identity, and returns the trailing-``lookback_days``-VALID-day mean of that ratio. + See the module docstring for the LOCKED definition and the five pinned choices. + + Args: + bars: normalized 1min bars (:mod:`data.clean.intraday_schema`), + ``MultiIndex(time, symbol)``. May carry one or many symbols; the grouping is + strictly per symbol (no cross-symbol leakage). + lookback_days: trailing VALID trading-day window averaged (definition). + baseline_days: strictly-prior same-slot baseline window in trading days (PR-F). + baseline_min_obs: minimum same-slot observations for a classifiable bar (PR-F). + sigma_k: eruptive threshold multiplier ``k`` in ``vol > μ + k*σ`` (PR-F). + min_valid_days: minimum valid days in the trailing window for a finite value. + min_classifiable: a day needs at least this many classifiable bars (PR-F). + min_valley_bars: a day needs at least this many TRADABLE valley bars. + decision_time: per-bar PIT cutoff time-of-day (default 14:50:00). + name: the returned Series name (the factor-panel column name). + + Returns: + ``MultiIndex(date, symbol)`` Series (midnight-normalized dates) of the daily + factor value, sorted, named ``name``. Pure: never mutates ``bars``. + """ + validate_intraday_bars(bars) + if lookback_days < 1: + raise ValueError(f"lookback_days must be >= 1; got {lookback_days!r}.") + if 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) + + # extra_columns=("amount",) is the ONLY difference from the PR-F / PR-H entry + # points: the traded value rides along on the surviving rows, and the truncation / + # volume guard / slot assignment are untouched. + visible = prepare_visible_minute_bars( + bars, decision_time=decision_time, extra_columns=(_AMOUNT,) + ) + if visible.empty: + return _empty_series(name) + + index_tuples: list[tuple] = [] + values: list[float] = [] + for sym, g in visible.groupby(SYMBOL_LEVEL, sort=True): + days, vals = _valley_ratio_mean_for_symbol( + g.reset_index(drop=True), + baseline_days=baseline_days, + baseline_min_obs=baseline_min_obs, + sigma_k=sigma_k, + lookback_days=lookback_days, + min_valid_days=min_valid_days, + min_classifiable=min_classifiable, + min_valley_bars=min_valley_bars, + ) + for day, val in zip(days, vals): + index_tuples.append((day, str(sym))) + values.append(val) + + if not index_tuples: + return _empty_series(name) + index = pd.MultiIndex.from_tuples(index_tuples, names=DAILY_INDEX_NAMES) + return pd.Series(values, index=index, name=name).sort_index() + + +__all__ = [ + "VALLEY_VWAP_LOOKBACK_DAYS", + "VALLEY_VWAP_MIN_VALLEY_BARS", + "compute_valley_relative_vwap", + "valley_vwap_ratio_by_day", +] diff --git a/data/clean/intraday_volume_prv.py b/data/clean/intraday_volume_prv.py index e7fc71b..5726a44 100644 --- a/data/clean/intraday_volume_prv.py +++ b/data/clean/intraday_volume_prv.py @@ -61,6 +61,8 @@ from __future__ import annotations +from collections.abc import Sequence + import numpy as np import pandas as pd @@ -80,6 +82,12 @@ # the session close and any missing minute all differ, so they are not neighbours. _ONE_MINUTE_SECONDS = 60.0 +# The columns every peak-family factor needs. ``prepare_visible_minute_bars`` always +# emits EXACTLY these (plus the derived ``trade_date`` / ``slot``); anything else a +# caller needs is opt-in via ``extra_columns``, so the default output stays byte-identical +# for the callers that predate the option. +_BASE_VISIBLE_COLUMNS = [SYMBOL_LEVEL, "bar_end", "available_time", "volume"] + def _empty_series(name: str) -> pd.Series: """Schema-shaped empty ``MultiIndex(date, symbol)`` factor Series.""" @@ -91,7 +99,10 @@ def _empty_series(name: str) -> pd.Series: def prepare_visible_minute_bars( - bars: pd.DataFrame, *, decision_time: str = DEFAULT_DECISION_TIME + bars: pd.DataFrame, + *, + decision_time: str = DEFAULT_DECISION_TIME, + extra_columns: Sequence[str] = (), ) -> pd.DataFrame: """PIT-truncate ``bars`` at ``decision_time`` and add ``trade_date`` / ``slot``. @@ -109,15 +120,30 @@ def prepare_visible_minute_bars( Args: bars: normalized 1min bars, ``MultiIndex(time, symbol)``; one or many symbols. decision_time: per-bar PIT cutoff time-of-day (default 14:50:00). + extra_columns: additional ``bars`` columns to carry through, appended AFTER the + base columns. The truncation and the volume guard are unaffected — the extra + values merely ride along on the surviving rows — so the default ``()`` keeps + the output byte-identical for callers that do not need them (PR-F / PR-H). + PR-I passes ``("amount",)`` to compute a volume-weighted price. Returns: A fresh ``RangeIndex`` frame with columns ``symbol`` / ``bar_end`` / - ``available_time`` / ``volume`` / ``trade_date`` / ``slot`` (possibly empty). - Pure: never mutates ``bars``. + ``available_time`` / ``volume`` / ``trade_date`` / ``slot`` plus any + ``extra_columns`` (possibly empty). Pure: never mutates ``bars``. """ - work = bars.reset_index()[ - [SYMBOL_LEVEL, "bar_end", "available_time", "volume"] - ].copy() + extra = list(extra_columns) + unknown = [c for c in extra if c not in bars.columns] + if unknown: + raise ValueError( + f"prepare_visible_minute_bars extra_columns not present on bars: {unknown}; " + f"bars has {list(bars.columns)}." + ) + clashing = [c for c in extra if c in _BASE_VISIBLE_COLUMNS] + if clashing: + raise ValueError( + f"prepare_visible_minute_bars extra_columns are already emitted: {clashing}." + ) + work = bars.reset_index()[_BASE_VISIBLE_COLUMNS + extra].copy() work["trade_date"] = work["bar_end"].dt.normalize() # PIT truncation FIRST (per-bar timestamps): each bar's cutoff is its own # trade_date + decision_time, so every day is truncated to [open, cutoff]. @@ -169,8 +195,10 @@ def peak_mask_for_symbol( Returns: ``g`` sorted by ``(trade_date, bar_end)`` on a fresh ``RangeIndex`` with the - added boolean columns ``classifiable`` (a strictly-prior baseline existed) and - ``peak``. Pure: never mutates ``g``. + added boolean columns ``classifiable`` (a strictly-prior baseline existed), + ``valley`` (classifiable and NOT eruptive — the report's 量谷, what PR-I prices) + and ``peak``. ``valley`` and ``peak`` are disjoint: a peak is eruptive by + construction. Pure: never mutates ``g``. """ # day x slot volume matrix: rows are the symbol's trading days, columns are the # minute-of-day slots; a missing (day, slot) cell is NaN (no bar that minute). @@ -204,6 +232,12 @@ def peak_mask_for_symbol( eruptive = classifiable & (vol > thr_arr) mild = classifiable & ~eruptive work["classifiable"] = classifiable + # VALLEY (量谷 in the report's peak/ridge/valley taxonomy) == MILD: a classifiable, + # non-eruptive minute. Exposed as a first-class boolean so the valley-PRICE family + # (PR-I) consumes THE SAME classification instead of re-deriving it and drifting. + # Purely additive — ``classifiable`` and ``peak`` below are computed exactly as + # before and no consumer of this frame reads columns positionally. + work["valley"] = mild # Carry mild as 0/1 so the within-day neighbour shift stays numeric (a shifted bool # column would go through an object-dtype fillna and warn). work["mild_i"] = mild.astype(np.int8) diff --git a/factors/compute/intraday_derived.py b/factors/compute/intraday_derived.py index ad675cb..160c797 100644 --- a/factors/compute/intraday_derived.py +++ b/factors/compute/intraday_derived.py @@ -58,6 +58,10 @@ PEAK_INTERVAL_LOOKBACK_DAYS, PEAK_INTERVAL_MIN_INTERVALS, ) +from data.clean.intraday_valley_vwap import ( + VALLEY_VWAP_LOOKBACK_DAYS, + VALLEY_VWAP_MIN_VALLEY_BARS, +) from data.clean.intraday_volume_prv import ( VOLUME_PRV_BASELINE_DAYS, VOLUME_PRV_BASELINE_MIN_OBS, @@ -638,11 +642,125 @@ def compute(self, panel: pd.DataFrame) -> pd.Series: return panel[self.name].rename(self.name) +class ValleyRelativeVwapFactor(Factor): + """Valley-relative VWAP factor (daily signal, minute-derived). + + ``compute`` reads the pre-aggregated daily column the runner placed on the panel + (produced by ``data.clean.intraday_valley_vwap.compute_valley_relative_vwap``); it + does NO minute work of its own, mirroring :class:`JumpAmountCorrFactor` / + :class:`MinuteIdealAmplitudeFactor` / :class:`AmpMarginalAnomalyVolFactor` / + :class:`VolumePeakCountFactor` / :class:`IntradayAmpCutFactor` / + :class:`PeakIntervalKurtosisFactor` and the value / financial factors that surface an + enriched column. + + Args: + lookback_days: trailing VALID trading-day window averaged; part of the factor + DEFINITION (reproduced from the report), not a tuned knob. It only names the + column so a non-default window cannot silently mislabel it. + """ + + name: str = f"valley_relative_vwap_{VALLEY_VWAP_LOOKBACK_DAYS}" + + def __init__(self, lookback_days: int = VALLEY_VWAP_LOOKBACK_DAYS) -> None: + if not isinstance(lookback_days, int) or lookback_days < 1: + raise ValueError( + f"valley-relative-vwap lookback_days must be a positive integer; got " + f"{lookback_days!r}." + ) + self._lookback_days = lookback_days + self.name = f"valley_relative_vwap_{lookback_days}" + + @property + def lookback_days(self) -> int: + return self._lookback_days + + @property + def spec(self) -> FactorSpec: + """Evaluation contract; a property so ``factor_id`` tracks the window. + + expected_ic_sign=+1: the report's full-market RankIC is +8.69% (RankICIR 4.44, + long-short 25.35%/yr, IR 3.04, monthly win rate 79.7% — the strongest factor in + the report), and its CSI500 sub-domain long-short is 9.94% / IR 1.26, the closest + comparable to our eval cell. Semantics per the report: valley minutes are moments + of subdued sentiment where prices are unlikely to have over-reacted, so a HIGH + relative valley price predicts HIGHER forward returns. The sign is fixed BEFORE + the run (a validated prototype must reproduce it). NOTE the report is a MONTHLY, + market-cap + industry neutral full-market series on Wind data while our eval cell + is CSI500 daily with industry + size neutral, so the report numbers are a LOOSE + reference only (disclosed, never mislabeled, never written in as an expected + value). is_intraday=False by the module docstring's reasoning: minute INPUT but a + DAILY signal traded close-to-close. min_history_bars=0: the warm-up is + DATA-dependent (a value appears once enough VALID days accumulate), not a fixed + leading count — the honest NaN rate is reported by data_coverage. + + The description spells out the DIFFERENT FAMILY vs PR-F / PR-H (the same reused + classification, but a PRICE LEVEL at the VALLEYS rather than a count or timing of + the PEAKS) plus the pinned choices, including the one place we KNOWINGLY DEVIATE + from the report: our day VWAP covers the PIT-visible window only. + """ + return FactorSpec( + factor_id=self.name, + version="1.0", + description=( + f"Valley-relative VWAP (Kaiyuan microstructure series #27, THIRD factor " + f"量谷相对加权价格). SAME minute classification as PR-F volume_peak_count " + f"and PR-H peak_interval_kurtosis (REUSED from " + f"data.clean.intraday_volume_prv, not re-implemented): 1min bars " + f"PIT-truncated at 14:50, a minute is ERUPTIVE if vol > μ + " + f"{VOLUME_PRV_SIGMA_K:g}σ of its SAME-SLOT strictly-prior " + f"{VOLUME_PRV_BASELINE_DAYS}-day baseline, else it is a VALLEY (量谷). " + f"DIFFERENT FAMILY: instead of counting or timing the PEAKS this factor " + f"prices the VALLEYS — daily ratio = (valley VWAP) / (whole visible day " + f"VWAP), averaged over the trailing {self._lookback_days} VALID days. " + f"PINNED choices: (1) each VWAP uses the aggregation identity Σ(p·v)/Σv " + f"= Σamount/Σvolume, the day's REAL volume-weighted price rather than a " + f"close approximation; (2) bars with non-finite or non-positive volume " + f"or amount are dropped from BOTH sums (guard applied at summation only, " + f"so PR-F's baseline is untouched); (3) RAW unadjusted prices are correct " + f"here because the adjustment factor is constant within a day and cancels " + f"in the ratio; (4) DEVIATION FROM THE REPORT, disclosed: our day VWAP " + f"spans the PIT-VISIBLE window 09:31-14:50 only, not the full session — " + f"reading the close would be lookahead at our 14:50 decision time; " + f"(5) a day is VALID iff it has >= {VOLUME_PRV_MIN_CLASSIFIABLE} " + f"classifiable bars AND >= {VALLEY_VWAP_MIN_VALLEY_BARS} TRADABLE valley " + f"bars (counted after the guard) AND positive volume in both " + f"denominators; NaN below {VOLUME_PRV_MIN_VALID_DAYS} valid days. Derived " + f"from 1min bars but a DAILY signal traded close-to-close." + ), + expected_ic_sign=1, + is_intraday=False, + forward_return_horizon=1, + return_basis="close_to_close", + # The 1min bar fields the upstream aggregation is derived from. Declared for + # honest provenance disclosure (data_coverage lists them); the daily panel + # surfaces the pre-aggregated column itself. + input_fields=("volume", "amount"), + family="microstructure", + min_history_bars=0, + ) + + def compute(self, panel: pd.DataFrame) -> pd.Series: + """Select the pre-aggregated daily valley-relative-VWAP column off ``panel``. + + The runner runs ``compute_valley_relative_vwap`` per symbol on the minute cache + upstream and joins the result as ``self.name``; here we only surface it, so this + factor does no temporal logic and cannot introduce lookahead. + """ + if self.name not in panel.columns: + raise ValueError( + f"ValleyRelativeVwapFactor needs the pre-aggregated '{self.name}' column " + f"on the panel (produced upstream by compute_valley_relative_vwap and " + f"joined by the runner); panel has {list(panel.columns)}." + ) + return panel[self.name].rename(self.name) + + __all__ = [ "AmpMarginalAnomalyVolFactor", "IntradayAmpCutFactor", "JumpAmountCorrFactor", "MinuteIdealAmplitudeFactor", "PeakIntervalKurtosisFactor", + "ValleyRelativeVwapFactor", "VolumePeakCountFactor", ] diff --git a/qt/cli.py b/qt/cli.py index 09f6d7a..d6ae5dd 100644 --- a/qt/cli.py +++ b/qt/cli.py @@ -353,6 +353,31 @@ def _cmd_run_eval_peak_interval_kurtosis(args: argparse.Namespace) -> int: return 0 +def _cmd_run_eval_valley_relative_vwap(args: argparse.Namespace) -> int: + """Run the two real valley-relative-VWAP evaluations (cache-only) + reports.""" + from qt.eval_valley_relative_vwap import run_eval_valley_relative_vwap + + try: + result = run_eval_valley_relative_vwap(args.config) + except (ConfigError, ValueError, FileNotFoundError) as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 1 + nb, wb = result.no_book_metrics, result.with_book_metrics + print( + f"OK run-eval-valley-relative-vwap: covered={result.covered_symbols}/" + f"{result.requested_symbols}, stk_mins_live_calls={result.minute_live_calls}, " + f"factor_rows={result.factor_rows} ({result.elapsed:.1f}s)\n" + f"no-book: {nb['deployment']} (predictive={nb['predictive']}) " + f"ic_mean={nb['ic_mean']:.4f} ic_ir={nb['ic_ir']:.3f} N_eff={nb['effective_samples']:.1f}\n" + f"with-book: {wb['deployment']} (incremental={wb['incremental']}) " + f"incr_ic_ir={wb['incremental_ic_ir']:.3f}\n" + f"reports: {result.reports.no_book_md} | {result.reports.with_book_md}\n" + f"dashboards: {result.reports.no_book_dashboard} | " + f"{result.reports.with_book_dashboard}" + ) + return 0 + + def _cmd_data_update(args: argparse.Namespace) -> int: """Warm/update the tushare caches (P4-3); never runs a backtest.""" from qt.data_updater import format_summary, run_data_update @@ -526,6 +551,13 @@ def build_parser() -> argparse.ArgumentParser: p_pik.add_argument("--config", required=True, help="Path to the YAML config.") p_pik.set_defaults(func=_cmd_run_eval_peak_interval_kurtosis) + p_vrv = sub.add_parser( + "run-eval-valley-relative-vwap", + help="Run the valley-relative-VWAP factor evaluation (CSI500, cache-only).", + ) + p_vrv.add_argument("--config", required=True, help="Path to the YAML config.") + p_vrv.set_defaults(func=_cmd_run_eval_valley_relative_vwap) + 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_relative_vwap.py b/qt/eval_valley_relative_vwap.py new file mode 100644 index 0000000..0300cc0 --- /dev/null +++ b/qt/eval_valley_relative_vwap.py @@ -0,0 +1,521 @@ +"""run-eval-valley-relative-vwap: the seventh real factor evaluation (PR-I). + +Reproduces the THIRD factor of the Kaiyuan market-microstructure series #27 (开源证券 +《高频成交量的峰、岭、谷信息——市场微观结构研究系列（27）》, reportId 4957417, §4) — the +"量谷相对加权价格" factor — as a first-class +:class:`~factors.compute.intraday_derived.ValleyRelativeVwapFactor` 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-H used. + +This is the THIRD reproduction from the same report, and the first from a DIFFERENT +FAMILY: PR-F counted volume peaks and PR-H measured their timing (both null results); +this factor prices the VALLEYS. The minute classification is REUSED verbatim, so a +different verdict here cannot be an artefact of a differently-implemented taxonomy. + +The factor is a DAILY signal derived DIRECTLY from the 1min cache (see +``data.clean.intraday_valley_vwap.compute_valley_relative_vwap``): PIT-truncate each day +at 14:50, classify every visible minute with the taxonomy REUSED from PR-F +(``data.clean.intraday_volume_prv``: same-slot strictly-prior μ+σ eruptive test; a VALLEY +is a classifiable non-eruptive minute), take each valid day's ratio of the valley VWAP to +the WHOLE VISIBLE DAY's VWAP (both via the Σamount/Σvolume identity), and average that +ratio over the trailing 20 VALID trading days. It is executed CLOSE-TO-CLOSE (daily +default), so ``is_intraday=False`` (the reasoning is documented on the factor's spec). +The eval CELL is identical to PR-C..PR-H, so this run is directly comparable to its +PR-F / PR-H siblings — same classification, different statistic and different family. + +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 valley-relative +VWAP adds alpha BEYOND value / low-vol. That check matters more than usual here: a +relative PRICE LEVEL is a plausible cousin of a value signal, so the with-book run is +the one that says whether this is a new bet. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from pathlib import Path + +import 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_vwap import ( + VALLEY_VWAP_LOOKBACK_DAYS, + VALLEY_VWAP_MIN_VALLEY_BARS, + compute_valley_relative_vwap, +) +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 ValleyRelativeVwapFactor +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_relative_vwap" +_REPORT_STEM = "eval_valley_relative_vwap" + + +# --------------------------------------------------------------------------- # +# Minute loading (cache-only, per-symbol -> memory-bounded) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _ValleyVwapMinuteLoad: + """Diagnostics from the cache-only per-symbol minute read + aggregation.""" + + factor: pd.Series # MultiIndex(date, symbol) raw valley-relative VWAP + requested: int + covered: tuple[str, ...] # symbols that produced >= 1 finite factor value + empty_symbols: tuple[str, ...] # requested but no cached minute / no value + raw_rows: int + live_calls: int # provably 0 (store read has no fetch closure) + + +def _load_valley_relative_vwap_panel( + cfg: RootConfig, + symbols: list[str], + spec: FactorSpec, + logger, + *, + lookback_days: int, + baseline_days: int, + baseline_min_obs: int, + sigma_k: float, + min_valid_days: int, + min_classifiable: int, + min_valley_bars: int, +) -> _ValleyVwapMinuteLoad: + """Compute the raw valley-relative-VWAP panel per symbol from the minute cache. + + Memory-bounded: one symbol's minute history is read, aggregated to its daily factor + series, and discarded before the next — the multi-year all-symbol minute panel is + NEVER materialized. Read-only: :meth:`IntradayParquetStore.read_range` has no fetch + closure, so ``stk_mins`` live calls are provably zero (a symbol with no cached + minute simply yields no rows and is disclosed as empty). The classification (reused + from PR-F) and the VWAP-ratio reduction run entirely inside + ``compute_valley_relative_vwap`` on 1min bars, which read ``volume`` and ``amount``. + """ + 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_relative_vwap( + 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, + name=spec.factor_id, + ) + if s.notna().any(): + series.append(s) + covered.append(sym) + else: + empty.append(sym) + if (i + 1) % 100 == 0: + logger.info( + "minute aggregation: %d/%d symbols processed (%d with a value)", + i + 1, len(symbols), len(covered), + ) + + if not series: + raise ValueError( + "run-eval-valley-relative-vwap blocked: no requested symbol produced a " + f"cached valley-relative-VWAP value over [{cfg.data.start}, " + f"{cfg.data.end}]. The minute cache is required (this runner never warms " + "it); check coverage." + ) + factor = pd.concat(series).sort_index() + logger.info( + "minute aggregation (cache-only): %d/%d symbols with a value, %d raw 1min " + "rows read, %d factor rows, stk_mins_live_calls=0", + len(covered), len(symbols), raw_rows, len(factor), + ) + return _ValleyVwapMinuteLoad( + factor=factor, + requested=len(symbols), + covered=tuple(covered), + empty_symbols=tuple(empty), + raw_rows=raw_rows, + live_calls=0, + ) + + +# --------------------------------------------------------------------------- # +# Evaluation core (network-free seam: given panels, run the two evaluations) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _RunReports: + """The two evaluation reports (no-book / with-book) + their file paths.""" + + no_book: FactorEvalReport + with_book: FactorEvalReport + no_book_md: Path + no_book_json: Path + with_book_md: Path + with_book_json: Path + no_book_dashboard: Path + with_book_dashboard: Path + + +def evaluate_two_runs( + factor_panel: pd.Series | pd.DataFrame, + spec: FactorSpec, + eval_cfg: EvalConfig, + price_panel: pd.DataFrame, + book: pd.DataFrame, + *, + universe_symbols: tuple[str, ...], + fee_rate: float, + report_dir: Path, + stem: str = _REPORT_STEM, +) -> _RunReports: + """Run the StandardFactorEvaluator TWICE (no book / with book) and write reports. + + This is the network-free seam: given the PROCESSED factor panel, the qfq price + panel (for forward returns), and the PROCESSED known-factor book, it does the two + ``evaluate`` calls and writes ``{stem}_no_book`` / ``{stem}_with_book`` as + Markdown + JSON. Run 1 omits ``known_factors`` (Incremental NOT_ASSESSED); run 2 + supplies the book (Incremental measured). + """ + evaluator = StandardFactorEvaluator() + report_dir.mkdir(parents=True, exist_ok=True) + + ctx_no_book = EvalContext( + price_panel=price_panel, + universe_symbols=universe_symbols, + fee_rate=fee_rate, + ) + # evaluate_with_ir yields the SAME report as evaluate() plus the IR the + # research-style dashboard needs (per-period IC + quantile return series). + report_no_book, ir_no_book = evaluator.evaluate_with_ir( + factor_panel, spec, eval_cfg, ctx_no_book + ) + + ctx_with_book = EvalContext( + price_panel=price_panel, + universe_symbols=universe_symbols, + fee_rate=fee_rate, + known_factors=book, + ) + report_with_book, ir_with_book = evaluator.evaluate_with_ir( + factor_panel, spec, eval_cfg, ctx_with_book + ) + + nb_md, nb_json = _write_report(report_no_book, report_dir, f"{stem}_no_book") + wb_md, wb_json = _write_report(report_with_book, report_dir, f"{stem}_with_book") + nb_png = render_factor_dashboard( + report_no_book, ir_no_book, report_dir / f"{stem}_no_book_dashboard.png" + ) + wb_png = render_factor_dashboard( + report_with_book, ir_with_book, report_dir / f"{stem}_with_book_dashboard.png" + ) + return _RunReports( + no_book=report_no_book, + with_book=report_with_book, + no_book_md=nb_md, + no_book_json=nb_json, + with_book_md=wb_md, + with_book_json=wb_json, + no_book_dashboard=nb_png, + with_book_dashboard=wb_png, + ) + + +def _write_report(report: FactorEvalReport, report_dir: Path, stem: str) -> tuple[Path, Path]: + """Write ``report`` as deterministic Markdown + JSON; return the two paths.""" + md_path = report_dir / f"{stem}.md" + json_path = report_dir / f"{stem}.json" + md_path.write_text(report.render(), encoding="utf-8") + json_path.write_text(report.to_json(), encoding="utf-8") + return md_path, json_path + + +# --------------------------------------------------------------------------- # +# Metric extraction (for the CLI line + the handoff) +# --------------------------------------------------------------------------- # +def _section_payload(report: FactorEvalReport, name: str) -> dict: + section = report.by_name().get(name) + return dict(getattr(section, "payload", {}) or {}) + + +def extract_metrics(report: FactorEvalReport) -> dict: + """Pull the headline verdict + gated metrics out of a finished report.""" + verdict = report.require_verdict() + pred = _section_payload(report, "predictive_power") + purity = _section_payload(report, "data_coverage") + incr = _section_payload(report, "purity") + return { + "deployment": verdict.verdict, + "predictive": verdict.predictive.verdict, + "incremental": verdict.incremental.verdict, + "tradable": verdict.tradable.verdict, + "ic_mean": pred.get("ic_mean"), + "ic_ir": pred.get("ic_ir"), + "ic_ir_ci_low": pred.get("ic_ir_ci_low"), + "ic_ir_ci_high": pred.get("ic_ir_ci_high"), + "ic_ir_ci_n_eff": pred.get("ic_ir_ci_n_eff"), + "ic_win_rate": pred.get("ic_win_rate"), + "ic_nw_t": pred.get("ic_nw_t"), + "settled_rebalances": purity.get("settled_rebalances"), + "effective_samples": purity.get("effective_samples"), + "span_days": purity.get("span_days"), + "incremental_ic_ir": incr.get("incremental_ic_ir"), + "incremental_ic_ir_ci_low": incr.get("incremental_ic_ir_ci_low"), + "incremental_ic_ir_ci_high": incr.get("incremental_ic_ir_ci_high"), + "incremental_ic_ir_ci_n_eff": incr.get("incremental_ic_ir_ci_n_eff"), + "incremental_ic_mean": incr.get("incremental_ic_mean"), + } + + +# --------------------------------------------------------------------------- # +# EvalConfig construction (HONEST provenance of what the runner actually did) +# --------------------------------------------------------------------------- # +def _build_eval_config(cfg: RootConfig) -> EvalConfig: + """Build the per-run EvalConfig, declaring EXACTLY what the pipeline applied. + + The declarations must not overstate: this codebase's winsorize step is a P0 no-op, + so ``winsorize`` is declared None (nothing was clipped) even though the config may + toggle it. z-score + industry/size neutralization ARE applied, so they are + declared. ``oos_split`` (from the config's ``oos`` block) makes the OOS section run + so the Predictive axis can be assessed. ``is_exploratory=True``: this is a + reproduction on a shorter window / narrower neutralization than the report, not a + return claim (it caps the deployment label at Watch). + """ + if cfg.oos is None: + raise ValueError( + "run-eval-valley-relative-vwap 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 ValleyRelativeVwapEvalResult: + """Immutable summary of one run-eval-valley-relative-vwap run.""" + + config: RootConfig + spec: FactorSpec + requested_symbols: int + covered_symbols: int + empty_symbols: int + factor_rows: int + minute_raw_rows: int + minute_live_calls: int + no_book_metrics: dict + with_book_metrics: dict + reports: _RunReports + log_path: Path + elapsed: float + + +def _check_preconditions(cfg: RootConfig) -> None: + """Fail readably if the config cannot drive a real, cache-only CSI500 eval.""" + if cfg.data.source != "tushare": + raise ValueError( + "run-eval-valley-relative-vwap 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-relative-vwap 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-relative-vwap needs universe.type='index' (PIT " + f"membership, e.g. 000905.SH for CSI500); got {cfg.universe.type!r}." + ) + if not cfg.processing.neutralize.enabled: + raise ValueError( + "run-eval-valley-relative-vwap expects processing.neutralize.enabled=true " + "(industry + size neutralization, matching the report's neutral column and " + "the EvalConfig declaration)." + ) + + +def run_eval_valley_relative_vwap(config_path: str) -> ValleyRelativeVwapEvalResult: + """Run the two real valley-relative-VWAP 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 = ValleyRelativeVwapFactor(lookback_days=VALLEY_VWAP_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-relative-VWAP factor: cache-only per-symbol aggregation -> raw -> process. + load = _load_valley_relative_vwap_panel( + cfg, symbols, spec, logger, + lookback_days=VALLEY_VWAP_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_VWAP_MIN_VALLEY_BARS, + ) + # A minute date with no daily bar cannot have a forward return; keep the factor on + # the daily trading grid so the analytics boundary has a price for every date. + factor_raw = load.factor[ + load.factor.index.get_level_values(DATE_LEVEL).isin(panel_dates) + ] + factor_processed = _process_factors(cfg, factor_raw.to_frame(spec.factor_id), panel) + factor_series = factor_processed[spec.factor_id] + + price_panel = panel[CORE_COLUMNS] + reports = evaluate_two_runs( + factor_series, + spec, + eval_cfg, + price_panel, + book_processed, + universe_symbols=tuple(symbols), + fee_rate=float(cfg.cost.fee_rate), + report_dir=Path(cfg.output.report_dir), + ) + + no_book_metrics = extract_metrics(reports.no_book) + with_book_metrics = extract_metrics(reports.with_book) + logger.info( + "verdict no-book: %s (predictive=%s); with-book: %s (incremental=%s)", + no_book_metrics["deployment"], no_book_metrics["predictive"], + with_book_metrics["deployment"], with_book_metrics["incremental"], + ) + + return ValleyRelativeVwapEvalResult( + config=cfg, + spec=spec, + requested_symbols=load.requested, + covered_symbols=len(load.covered), + empty_symbols=len(load.empty_symbols), + factor_rows=int(len(factor_series)), + minute_raw_rows=load.raw_rows, + minute_live_calls=load.live_calls, + no_book_metrics=no_book_metrics, + with_book_metrics=with_book_metrics, + reports=reports, + log_path=log_path, + elapsed=time.monotonic() - started, + ) + + +def _build_book_factors() -> list: + """Instantiate the confirmed book (value_ep / value_bp / volatility_20).""" + from factors.compute.candidates import ValueFactor, VolatilityFactor + + return [ + ValueFactor("value_ep"), + ValueFactor("value_bp"), + VolatilityFactor(window=20), + ] + + +__all__ = [ + "ValleyRelativeVwapEvalResult", + "evaluate_two_runs", + "extract_metrics", + "run_eval_valley_relative_vwap", +] diff --git a/tests/test_eval_valley_relative_vwap_runner.py b/tests/test_eval_valley_relative_vwap_runner.py new file mode 100644 index 0000000..d4b0958 --- /dev/null +++ b/tests/test_eval_valley_relative_vwap_runner.py @@ -0,0 +1,288 @@ +"""PR-I runner: cache-only minute loader + the two-run evaluation core (no network).""" + +from __future__ import annotations + +import logging + +import numpy as np +import pandas as pd +import pytest + +from analytics.eval import EvalConfig, MANDATORY_SECTIONS, Section, Skipped +from analytics.eval.verdict import AXIS_NOT_ASSESSED, AXIS_VERDICTS +from data.cache.intraday_cache import ENDPOINT as INTRADAY_ENDPOINT +from data.cache.intraday_parquet_store import KEY_COLS, IntradayParquetStore +from factors.compute.intraday_derived import ValleyRelativeVwapFactor +from qt.config import ( + AlphaCfg, + BacktestCfg, + CacheCfg, + CostCfg, + DataCfg, + FactorCfg, + OutputCfg, + PortfolioCfg, + RootConfig, + UniverseCfg, +) +from qt.eval_valley_relative_vwap import ( + _load_valley_relative_vwap_panel, + evaluate_two_runs, + extract_metrics, +) + + +# --------------------------------------------------------------------------- # +# Minute cache-only loader +# --------------------------------------------------------------------------- # +def _stored_rows(sym, day, vols, amts): + """Build STORED_COLUMNS-shaped 1min rows for one session of CONSECUTIVE minutes. + + Bar i sits at ``09:31 + i`` minutes (all inside the 14:50 PIT window). OHLC are dummy + constants; ``volume`` drives the classification and ``amount`` -- set INDEPENDENTLY -- + drives the VWAPs. + """ + base = pd.Timestamp(day) + pd.Timedelta("09:31:00") + rows = [] + for i, (v, a) in enumerate(zip(vols, amts)): + be = base + pd.Timedelta(minutes=i) + rows.append( + { + "symbol": sym, + "bar_end": be, + "source_trade_time": be, + "open": 100.0, + "high": 100.0, + "low": 100.0, + "close": 100.0, + "volume": float(v), + "amount": float(a), + "freq": "1min", + } + ) + return pd.DataFrame(rows) + + +def _min_config(root, start, end): + return RootConfig( + data=DataCfg( + source="tushare", + start=start, + end=end, + external_secret_file="/nonexistent.json", + cache=CacheCfg(enabled=True, root_dir=str(root)), + ), + universe=UniverseCfg(type="static", symbols=["A.SZ"]), + factors=[FactorCfg(name="momentum_20")], + alpha=AlphaCfg(), + portfolio=PortfolioCfg(top_n=1), + backtest=BacktestCfg(), + cost=CostCfg(), + output=OutputCfg(), + ) + + +# Three constant background days (baseline mu=100, sigma=0 -> eruptive threshold 100), +# then the hand-computed CASE A test day: 12 slots, eruptions at 3 and 7 (volume 200 / +# amount 4000 -> price 20); the ten valley bars carry volume 100 with amount 1000 (x5) +# and 1200 (x5) -> valley VWAP 11.0, day VWAP 19000/1400, ratio 77/95. +_DAYS = ("2021-07-01", "2021-07-02", "2021-07-03") +_TEST_DAY = "2021-07-04" +_N_SLOTS = 12 +_ERUPT = (3, 7) +_CASE_A_RATIO = 77.0 / 95.0 +_BG_VOLS = [100.0] * _N_SLOTS +_BG_AMTS = [1000.0] * _N_SLOTS + + +def _case_a(): + vols = [100.0] * _N_SLOTS + amts = [0.0] * _N_SLOTS + valley_slots = [s for s in range(_N_SLOTS) 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 + return vols, amts + + +# Small gates so a 4-day cache produces a value (baseline needs >= 2 prior obs; the test +# day carries 10 tradable valley bars). lookback_days=1 so the test day's value IS its own +# ratio: with these tiny gates the last background day also becomes valid, and a day where +# NOTHING erupts has ratio exactly 1.0 (every bar is a valley, so the two VWAPs coincide), +# which would otherwise average into the assertion. +_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, +) + + +def _seed_symbol(store, sym): + for day in _DAYS: + store.upsert( + INTRADAY_ENDPOINT, sym, "1min", + _stored_rows(sym, day, _BG_VOLS, _BG_AMTS), KEY_COLS, + ) + store.upsert( + INTRADAY_ENDPOINT, sym, "1min", + _stored_rows(sym, _TEST_DAY, *_case_a()), KEY_COLS, + ) + + +def test_minute_loader_is_cache_only_and_discloses_empty(tmp_path): + root = tmp_path / "cache" + store = IntradayParquetStore(str(root)) + # Two symbols with cached minute bars; a third has NO cached minute (must be + # disclosed as empty, never fetched). + for sym in ("AAA.SZ", "BBB.SZ"): + _seed_symbol(store, sym) + + cfg = _min_config(root, "2021-07-01", "2021-07-04") + spec = ValleyRelativeVwapFactor().spec + logger = logging.getLogger("test.vrv.loader") + load = _load_valley_relative_vwap_panel( + cfg, ["AAA.SZ", "BBB.SZ", "CCC.SZ"], spec, logger, **_LOAD_KW + ) + assert load.live_calls == 0 # store read has no fetch closure + assert set(load.covered) == {"AAA.SZ", "BBB.SZ"} # both produced a value + assert load.empty_symbols == ("CCC.SZ",) # uncovered disclosed, not fetched + assert load.factor.name == spec.factor_id + assert load.factor.notna().any() + # the test day carries the hand-computed valley/day VWAP ratio + d = pd.Timestamp(_TEST_DAY) + assert load.factor.loc[(d, "AAA.SZ")] == pytest.approx(_CASE_A_RATIO) + assert load.factor.loc[(d, "BBB.SZ")] == pytest.approx(_CASE_A_RATIO) + assert set(load.factor.index.get_level_values("symbol")) == {"AAA.SZ", "BBB.SZ"} + # the last background day (nothing erupts -> every bar is a valley) has ratio + # exactly 1.0: the two VWAPs are sums over the SAME bar set + bg = pd.Timestamp(_DAYS[-1]) + assert load.factor.loc[(bg, "AAA.SZ")] == pytest.approx(1.0) + + +def test_minute_loader_blocks_when_nothing_cached(tmp_path): + cfg = _min_config(tmp_path / "empty", "2021-07-01", "2021-07-04") + spec = ValleyRelativeVwapFactor().spec + with pytest.raises(ValueError, match="no requested symbol produced"): + _load_valley_relative_vwap_panel( + cfg, ["ZZZ.SZ"], spec, logging.getLogger("test.vrv.block"), **_LOAD_KW + ) + + +# --------------------------------------------------------------------------- # +# Evaluation core (two runs) — synthetic processed panels, no network +# --------------------------------------------------------------------------- # +def _synthetic_panels(n_days=90, n_symbols=15, seed=7): + rng = np.random.default_rng(seed) + dates = pd.bdate_range("2022-01-03", periods=n_days) + symbols = [f"{i:06d}.SZ" for i in range(n_symbols)] + idx = pd.MultiIndex.from_product([dates, symbols], names=["date", "symbol"]) + + # processed (z-scored-ish) subject factor + a 3-column book on the same grid + factor = pd.Series( + rng.standard_normal(len(idx)), index=idx, name="valley_relative_vwap_20" + ) + book = pd.DataFrame( + { + "value_ep": rng.standard_normal(len(idx)), + "value_bp": rng.standard_normal(len(idx)), + "volatility_20": rng.standard_normal(len(idx)), + }, + index=idx, + ) + # a qfq-style price panel with the CORE_COLUMNS the forward-return boundary needs + close = pd.Series( + 100.0 * np.exp(np.cumsum(0.001 * rng.standard_normal(len(idx)))), index=idx + ) + price = pd.DataFrame( + { + "open": close.to_numpy(), + "high": close.to_numpy() * 1.01, + "low": close.to_numpy() * 0.99, + "close": close.to_numpy(), + "volume": 1_000.0, + "amount": 100_000.0, + "adj_factor": 1.0, + }, + index=idx, + ) + return factor, book, price + + +def _eval_cfg(): + return EvalConfig( + universe="000905.SH", + universe_is_pit=True, + start="2022-01-03", + end="2022-05-10", + is_exploratory=True, + post_hoc_selected=False, + rebalance="daily", + n_quantiles=5, + oos_split="2022-03-15", + winsorize=None, + standardize="zscore", + neutralization=("industry", "size"), + industry_level="L1", + ) + + +def test_evaluate_two_runs_produces_full_reports_and_incremental_axis(tmp_path): + factor, book, price = _synthetic_panels() + spec = ValleyRelativeVwapFactor().spec + reports = evaluate_two_runs( + factor, spec, _eval_cfg(), price, book, + universe_symbols=tuple(sorted(set(book.index.get_level_values("symbol")))), + fee_rate=0.001, + report_dir=tmp_path, + ) + + for report in (reports.no_book, reports.with_book): + by = report.by_name() + assert set(by) == set(MANDATORY_SECTIONS) # all 8 present + assert all(isinstance(s, (Section, Skipped)) for s in by.values()) + assert report.verdict is not None # a verdict was produced + + # no-book: purity Skipped (no book) -> Incremental NOT_ASSESSED + assert isinstance(reports.no_book.by_name()["purity"], Skipped) + assert reports.no_book.verdict.incremental.verdict == AXIS_NOT_ASSESSED + + # with-book: purity is a real Section that populated the Incremental facts + purity = reports.with_book.by_name()["purity"] + assert isinstance(purity, Section) + assert purity.payload["known_factors_supplied"] is True + assert "incremental_ic_ir" in purity.payload + # the axis is now assessed (not NOT_ASSESSED) and is a valid axis state + incr = reports.with_book.verdict.incremental.verdict + assert incr in AXIS_VERDICTS and incr != AXIS_NOT_ASSESSED + + # reports were written to disk (md + json), both runs + for p in (reports.no_book_md, reports.no_book_json, + reports.with_book_md, reports.with_book_json): + assert p.exists() and p.stat().st_size > 0 + + # the research-style dashboard PNG is emitted for both runs (mandatory report + # artifact alongside md/json) + for p in (reports.no_book_dashboard, reports.with_book_dashboard): + assert p.exists() and p.stat().st_size > 20_000 + with open(p, "rb") as fh: + assert fh.read(8) == b"\x89PNG\r\n\x1a\n" + + # metrics extraction surfaces the gated fields + m = extract_metrics(reports.no_book) + assert m["deployment"] in {"Adopt", "Watch", "Reject", "INSUFFICIENT-DATA"} + assert "effective_samples" in m and "ic_ir" in m + + +def test_no_book_run_never_reaches_adopt(tmp_path): + # Structural guarantee (design §6): with no book + no execution facts, at most + # Watch — regardless of the signal (exploratory cap + NOT_ASSESSED axes). + factor, book, price = _synthetic_panels(seed=3) + reports = evaluate_two_runs( + factor, ValleyRelativeVwapFactor().spec, _eval_cfg(), price, book, + universe_symbols=(), + fee_rate=0.001, + report_dir=tmp_path, + ) + assert reports.no_book.verdict.verdict != "Adopt" + assert reports.no_book.verdict.tradable.verdict == AXIS_NOT_ASSESSED diff --git a/tests/test_valley_relative_vwap_factor.py b/tests/test_valley_relative_vwap_factor.py new file mode 100644 index 0000000..1066cba --- /dev/null +++ b/tests/test_valley_relative_vwap_factor.py @@ -0,0 +1,688 @@ +"""PR-I: VALLEY-RELATIVE VWAP factor. + +Same volume classification as PR-F/PR-H (REUSED, not re-implemented), different +STATISTIC and different FAMILY: a PRICE level rather than a peak count / timing shape. +A "valley" (量谷) is a classifiable, NON-eruptive minute -- PR-F's internal ``mild``, +now exposed as a first-class ``valley`` column. The factor is the trailing-20-valid-day +mean of the daily ratio ``valley VWAP / whole-visible-day VWAP``. Sign is pre-registered ++1 (a high relative valley price = little downward over-reaction in calm minutes -> +higher forward return). + +Hand cases build a constant BACKGROUND of prior days so the same-slot baseline is exact +(mu=100, sigma=0 -> the eruptive threshold is exactly 100), then a test day whose +volumes place valleys / eruptions at chosen slots and whose AMOUNTS are set +independently, so both VWAPs -- and therefore the ratio -- are known 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_vwap import ( + VALLEY_VWAP_LOOKBACK_DAYS, + VALLEY_VWAP_MIN_VALLEY_BARS, + compute_valley_relative_vwap, + valley_vwap_ratio_by_day, +) +from data.clean.intraday_volume_prv import ( + VOLUME_PRV_BASELINE_DAYS, + VOLUME_PRV_BASELINE_MIN_OBS, + VOLUME_PRV_MIN_VALID_DAYS, + VOLUME_PRV_SIGMA_K, + peak_mask_for_symbol, + prepare_visible_minute_bars, +) +from factors.compute.intraday_derived import ValleyRelativeVwapFactor +from factors.spec import FactorSpec + +_SYM = "000001.SZ" + + +def _bars(rows): + """rows = [(time, symbol, volume, amount), ...] -> normalized 1min bars. + + Unlike the PR-F / PR-H helpers, ``amount`` is set INDEPENDENTLY of ``volume`` -- + that is the whole point here: the per-bar price is ``amount / volume``. OHLC are + dummy constants (this factor reads only volume + amount). + ``normalize_intraday_bars`` sets ``available_time = bar_end + 1min``, so the 14:50 + PIT cutoff excludes any bar with ``bar_end >= 14:50``. + """ + df = pd.DataFrame( + { + "time": pd.to_datetime([r[0] for r in rows]), + "symbol": [r[1] for r in rows], + "open": 100.0, + "high": 100.0, + "low": 100.0, + "close": 100.0, + "volume": [float(r[2]) for r in rows], + "amount": [float(r[3]) for r in rows], + } + ) + return normalize_intraday_bars(df, freq="1min") + + +def _session(day, vols, amts, sym=_SYM, start="09:31:00"): + """One session of CONSECUTIVE 1-minute bars carrying ``vols`` / ``amts``.""" + base = pd.Timestamp(day) + pd.Timedelta(start) + return [ + (base + pd.Timedelta(minutes=i), sym, v, a) + for i, (v, a) in enumerate(zip(vols, amts)) + ] + + +def _background(n_days, n_slots, sym=_SYM, start_day="2021-07-01"): + """``n_days`` prior days of flat volume-100 / amount-1000 sessions. + + Same-slot baseline becomes mu=100, sigma=0 -> the eruptive threshold is exactly + 100, so on the test day a volume of 100 is a VALLEY and anything above erupts. + """ + rows = [] + for i in range(n_days): + day = (pd.Timestamp(start_day) + pd.Timedelta(days=i)).strftime("%Y-%m-%d") + rows += _session(day, [100.0] * n_slots, [1000.0] * n_slots, sym=sym) + return rows + + +_BG_DAYS = 10 +_TEST_DAY = pd.Timestamp("2021-07-11") + +# Gates small enough that the single engineered test day is the only VALID day; the +# valley-bar 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_VWAP_LOOKBACK_DAYS, + min_valid_days=1, + min_classifiable=1, + min_valley_bars=1, +) + + +# --------------------------------------------------------------------------- # +# Hand-computed VWAP ratios (2 non-trivial cases, closed-form fractions) +# --------------------------------------------------------------------------- # +# CASE A -- 12 slots, eruptions at slots 3 and 7 (volume 200, amount 4000 -> price 20). +# valley bars: 10 x volume 100; five carry amount 1000 (price 10), five amount 1200 +# (price 12) -> valley VWAP = 11000 / 1000 = 11.0 +# whole visible day: sum(amount) = 11000 + 2*4000 = 19000 +# sum(volume) = 1000 + 2*200 = 1400 -> day VWAP = 19000/1400 +# ratio = 11.0 * 1400 / 19000 = 15400/19000 = 77/95 = 0.810526... +# The eruptive minutes traded HIGHER and heavier, so the calm-minute price sits BELOW +# the day VWAP -> ratio < 1. +_CASE_A_N = 12 +_CASE_A_ERUPT = (3, 7) +_CASE_A_RATIO = 77.0 / 95.0 + + +def _case_a_day(): + vols = [100.0] * _CASE_A_N + amts = [0.0] * _CASE_A_N + valley_slots = [s for s in range(_CASE_A_N) if s not in _CASE_A_ERUPT] + for i, s in enumerate(valley_slots): + amts[s] = 1000.0 if i < 5 else 1200.0 + for s in _CASE_A_ERUPT: + vols[s] = 200.0 + amts[s] = 4000.0 + return vols, amts + + +# CASE B -- 14 slots, eruptions at slots 2, 6, 10 (volume 300, amount 2400 -> price 8). +# valley bars: 11 x volume 100; six carry amount 1000 (price 10), five amount 1500 +# (price 15) -> valley VWAP = 13500 / 1100 +# whole visible day: sum(amount) = 13500 + 3*2400 = 20700 +# sum(volume) = 1100 + 3*300 = 2000 -> day VWAP = 10.35 +# ratio = (13500/1100) / 10.35 = 27000000/22770000 = 300/253 = 1.185770... +# Here the eruptions traded LOWER, so the calm-minute price sits ABOVE the day VWAP +# -> ratio > 1. Opposite direction from case A, so a sign/inversion bug cannot pass both. +_CASE_B_N = 14 +_CASE_B_ERUPT = (2, 6, 10) +_CASE_B_RATIO = 300.0 / 253.0 + + +def _case_b_day(): + vols = [100.0] * _CASE_B_N + amts = [0.0] * _CASE_B_N + valley_slots = [s for s in range(_CASE_B_N) if s not in _CASE_B_ERUPT] + for i, s in enumerate(valley_slots): + amts[s] = 1000.0 if i < 6 else 1500.0 + for s in _CASE_B_ERUPT: + vols[s] = 300.0 + amts[s] = 2400.0 + return vols, amts + + +def test_hand_value_case_a_ratio_below_one(): + vols, amts = _case_a_day() + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", vols, amts) + out = compute_valley_relative_vwap(_bars(rows), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(_CASE_A_RATIO) + assert out.loc[(_TEST_DAY, _SYM)] < 1.0 + + +def test_hand_value_case_b_ratio_above_one(): + vols, amts = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", vols, amts) + out = compute_valley_relative_vwap(_bars(rows), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(_CASE_B_RATIO) + assert out.loc[(_TEST_DAY, _SYM)] > 1.0 + + +def test_vwap_uses_the_amount_over_volume_aggregation_identity(): + """sum(p_i * v_i) / sum(v_i) with p_i = amount_i/volume_i IS sum(amount)/sum(volume). + + The PINNED VWAP definition. Recomputed here the LONG way from per-bar prices, for + BOTH legs (valley bars and the whole visible day), and checked against the closed- + form ratio the factor returns -- so the identity is verified, not assumed. + """ + vols, amts = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", vols, amts) + bars = _bars(rows) + + visible = prepare_visible_minute_bars(bars, extra_columns=("amount",)) + work = peak_mask_for_symbol(visible.reset_index(drop=True)) + day = work[work["trade_date"] == _TEST_DAY] + + v = day["volume"].to_numpy(dtype=float) + a = day["amount"].to_numpy(dtype=float) + price = a / v # per-bar VWAP + is_valley = day["valley"].to_numpy(dtype=bool) + + # whole visible day, the long way vs the aggregation identity + day_long = float((price * v).sum() / v.sum()) + assert day_long == pytest.approx(float(a.sum() / v.sum())) + assert day_long == pytest.approx(10.35) + + # valley leg, the long way vs the aggregation identity + valley_long = float( + (price[is_valley] * v[is_valley]).sum() / v[is_valley].sum() + ) + assert valley_long == pytest.approx( + float(a[is_valley].sum() / v[is_valley].sum()) + ) + assert valley_long == pytest.approx(13500.0 / 1100.0) + + # and the factor's ratio is exactly those two legs divided + out = compute_valley_relative_vwap(bars, **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(valley_long / day_long) + + +def test_day_vwap_covers_the_whole_visible_day_not_only_valleys(): + """The denominator spans ALL visible bars, including the eruptive ones. + + Guard against the easy mistake of dividing the valley VWAP by itself (ratio == 1). + """ + vols, amts = _case_a_day() + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", vols, amts) + out = compute_valley_relative_vwap(_bars(rows), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] != pytest.approx(1.0) + # moving ONLY the eruptive bars' price must move the ratio (they are in the + # denominator and nowhere else) + vols2, amts2 = _case_a_day() + for s in _CASE_A_ERUPT: + amts2[s] = 8000.0 # price 40 instead of 20 + rows2 = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", vols2, amts2) + out2 = compute_valley_relative_vwap(_bars(rows2), **_KW) + assert out2.loc[(_TEST_DAY, _SYM)] < out.loc[(_TEST_DAY, _SYM)] + + +def test_day_with_no_eruption_has_ratio_exactly_one(): + """If NOTHING erupts, every bar is a valley and the two VWAPs are the same sum. + + An exact structural identity, useful as a scale check: the factor is centred on 1, + and any deviation is attributable to the eruptive minutes alone. + """ + n = 12 + amts = [1000.0, 1500.0, 800.0, 2000.0, 1200.0, 900.0] * 2 # prices vary a lot + rows = _background(_BG_DAYS, n) + _session("2021-07-11", [100.0] * n, amts) + out = compute_valley_relative_vwap(_bars(rows), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(1.0) + + +# --------------------------------------------------------------------------- # +# Non-positive volume / amount bars are DROPPED from the VWAP sums +# --------------------------------------------------------------------------- # +def _case_a_with_degenerate_bars(): + """Case A plus two extra slots (12, 13) that are CLASSIFIABLE VALLEYS but untradable. + + Slot 12 has volume 0 (amount 500); slot 13 has amount 0 (volume 50). Both are below + the eruptive threshold, so the reused classifier calls them valleys -- they must be + excluded by the POSITIVE-TRADE GUARD, not by the classification. + """ + n = _CASE_A_N + 2 + vols, amts = _case_a_day() + vols = vols + [0.0, 50.0] + amts = amts + [500.0, 0.0] + rows = _background(_BG_DAYS, n) + _session("2021-07-11", vols, amts) + return rows + + +def test_degenerate_bars_are_classified_as_valleys_but_excluded_from_the_vwap(): + rows = _case_a_with_degenerate_bars() + bars = _bars(rows) + + # they really ARE valleys under the reused classifier (so the guard is what drops + # them -- this is not an accident of classification) + visible = prepare_visible_minute_bars(bars, extra_columns=("amount",)) + work = peak_mask_for_symbol(visible.reset_index(drop=True)) + day = work[work["trade_date"] == _TEST_DAY].reset_index(drop=True) + assert bool(day.loc[_CASE_A_N, "valley"]) is True # volume 0 bar + assert bool(day.loc[_CASE_A_N + 1, "valley"]) is True # amount 0 bar + + # ...and the ratio is EXACTLY the clean case-A hand value: neither the numerator nor + # the denominator was polluted. + out = compute_valley_relative_vwap(bars, **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(_CASE_A_RATIO) + + +def test_negative_amount_bar_is_dropped_from_the_vwap(): + n = _CASE_A_N + 1 + vols, amts = _case_a_day() + rows = _background(_BG_DAYS, n) + _session( + "2021-07-11", vols + [50.0], amts + [-9_999.0] + ) + out = compute_valley_relative_vwap(_bars(rows), **_KW) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(_CASE_A_RATIO) + + +def test_valley_bar_count_is_taken_AFTER_the_positive_trade_guard(): + """The >= min_valley_bars gate counts bars that actually CONTRIBUTE to the VWAP. + + Case A + 2 degenerate valleys = 12 classifiable valleys but only 10 tradable ones. + A floor of 10 must pass; a floor of 11 must invalidate the day. + """ + bars = _bars(_case_a_with_degenerate_bars()) + ok = compute_valley_relative_vwap(bars, **{**_KW, "min_valley_bars": 10}) + assert ok.loc[(_TEST_DAY, _SYM)] == pytest.approx(_CASE_A_RATIO) + too_few = compute_valley_relative_vwap(bars, **{**_KW, "min_valley_bars": 11}) + assert too_few.dropna().empty + + +def test_day_with_no_tradable_volume_is_invalid(): + n = _CASE_A_N + rows = _background(_BG_DAYS, n) + _session( + "2021-07-11", [0.0] * n, [0.0] * n + ) + out = compute_valley_relative_vwap(_bars(rows), **_KW) + assert out.dropna().empty + + +def test_day_with_no_tradable_valley_is_invalid(): + # Every valley minute is untradable (volume 0); only the eruptive bars trade, so the + # numerator has no support -> honest NaN rather than a 0/0 or a day-VWAP-only number. + vols, amts = _case_a_day() + for s in range(_CASE_A_N): + if s not in _CASE_A_ERUPT: + vols[s] = 0.0 + amts[s] = 0.0 + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", vols, amts) + out = compute_valley_relative_vwap(_bars(rows), **_KW) + assert out.dropna().empty + + +# --------------------------------------------------------------------------- # +# Window mechanics: the trailing mean over VALID days +# --------------------------------------------------------------------------- # +def _two_case_days(n_slots=None): + """Background + day1 = case A pattern, day2 = case B pattern, on a shared slot grid.""" + n = n_slots or _CASE_B_N + a_vols, a_amts = _case_a_day() + b_vols, b_amts = _case_b_day() + # pad case A up to the shared grid with plain valley bars (volume 100 / amount 1000) + pad = n - _CASE_A_N + a_vols = a_vols + [100.0] * pad + a_amts = a_amts + [1000.0] * pad + rows = _background(_BG_DAYS, n) + rows += _session("2021-07-11", a_vols, a_amts) + rows += _session("2021-07-12", b_vols, b_amts) + return rows + + +def test_factor_is_the_mean_of_the_trailing_valid_day_ratios(): + rows = _two_case_days() + bars = _bars(rows) + ratios = compute_valley_relative_vwap(bars, **{**_KW, "lookback_days": 1}) + d1, d2 = pd.Timestamp("2021-07-11"), pd.Timestamp("2021-07-12") + r1 = ratios.loc[(d1, _SYM)] + r2 = ratios.loc[(d2, _SYM)] + assert r1 != pytest.approx(r2) # the two engineered days really do differ + + # lookback 2 -> day 2 is the MEAN of the two daily ratios (not the last, not a sum) + out = compute_valley_relative_vwap(bars, **{**_KW, "lookback_days": 2}) + assert out.loc[(d1, _SYM)] == pytest.approx(r1) + assert out.loc[(d2, _SYM)] == pytest.approx((r1 + r2) / 2.0) + + +def test_window_drops_days_older_than_lookback(): + rows = _two_case_days() + rows += _session("2021-07-13", *_case_b_day()) + bars = _bars(rows) + per_day = compute_valley_relative_vwap(bars, **{**_KW, "lookback_days": 1}) + d1, d2, d3 = (pd.Timestamp(f"2021-07-1{k}") for k in (1, 2, 3)) + out = compute_valley_relative_vwap(bars, **{**_KW, "lookback_days": 2}) + # day 3 pools days 2+3 only -- day 1 has aged out of a 2-day window + expected = (per_day.loc[(d2, _SYM)] + per_day.loc[(d3, _SYM)]) / 2.0 + assert out.loc[(d3, _SYM)] == pytest.approx(expected) + three = compute_valley_relative_vwap(bars, **{**_KW, "lookback_days": 3}) + assert three.loc[(d3, _SYM)] != pytest.approx(expected) + + +def test_min_valid_days_floor_returns_nan_until_enough_valid_days(): + rows = _background(_BG_DAYS, _CASE_B_N) + for k in range(3): + d = (pd.Timestamp("2021-07-11") + pd.Timedelta(days=k)).strftime("%Y-%m-%d") + rows += _session(d, *_case_b_day()) + out = compute_valley_relative_vwap(_bars(rows), **{**_KW, "min_valid_days": 3}) + assert np.isnan(out.loc[(pd.Timestamp("2021-07-11"), _SYM)]) + assert np.isnan(out.loc[(pd.Timestamp("2021-07-12"), _SYM)]) + assert np.isfinite(out.loc[(pd.Timestamp("2021-07-13"), _SYM)]) + + +def test_min_classifiable_gate_invalidates_thin_days(): + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", *_case_a_day()) + out = compute_valley_relative_vwap(_bars(rows), **{**_KW, "min_classifiable": 100}) + assert out.dropna().empty + + +def test_baseline_insufficient_yields_no_value(): + # Only 9 prior days -> fewer than baseline_min_obs (=10) same-slot observations -> + # nothing classifiable -> no valley, no valid day, no value at all. + rows = _background(9, _CASE_A_N) + _session("2021-07-10", *_case_a_day()) + out = compute_valley_relative_vwap(_bars(rows), **_KW) + assert out.dropna().empty + + +def test_default_gates_match_the_pinned_definition(): + assert VALLEY_VWAP_LOOKBACK_DAYS == 20 + assert VALLEY_VWAP_MIN_VALLEY_BARS == 20 + assert VOLUME_PRV_MIN_VALID_DAYS == 10 # the < 10 valid days -> NaN floor + + +# --------------------------------------------------------------------------- # +# PIT: no lookahead, no post-cutoff influence +# --------------------------------------------------------------------------- # +def test_perturbing_post_1450_bars_does_not_change_factor(): + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", *_case_a_day()) + a = compute_valley_relative_vwap(_bars(rows), **_KW) + late = rows + [ + (pd.Timestamp("2021-07-11 14:50"), _SYM, 9_999.0, 9_999_999.0), + (pd.Timestamp("2021-07-11 14:55"), _SYM, 1.0, 1.0), + (pd.Timestamp("2021-07-11 14:56"), _SYM, 5_000.0, 1.0), + ] + b = compute_valley_relative_vwap(_bars(late), **_KW) + key = (_TEST_DAY, _SYM) + assert a.loc[key] == pytest.approx(_CASE_A_RATIO) + assert a.loc[key] == pytest.approx(b.loc[key]) + + +def test_future_day_does_not_change_earlier_factor(): + base = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", *_case_a_day()) + a = compute_valley_relative_vwap(_bars(base), **_KW) + future = base + _session( + "2021-07-12", [9_999.0] * _CASE_A_N, [1.0] * _CASE_A_N + ) + b = compute_valley_relative_vwap(_bars(future), **_KW) + key = (_TEST_DAY, _SYM) + assert np.isfinite(a.loc[key]) + assert a.loc[key] == pytest.approx(b.loc[key]) + + +# --------------------------------------------------------------------------- # +# Per-symbol isolation +# --------------------------------------------------------------------------- # +def test_per_symbol_isolation(): + n = _CASE_B_N + a_vols, a_amts = _case_a_day() + a_vols = a_vols + [100.0] * (n - _CASE_A_N) + a_amts = a_amts + [1000.0] * (n - _CASE_A_N) + + rows = _background(_BG_DAYS, n, sym="AAA.SZ") + rows += _session("2021-07-11", a_vols, a_amts, sym="AAA.SZ") + rows += _background(_BG_DAYS, n, sym="BBB.SZ") + rows += _session("2021-07-11", *_case_b_day(), sym="BBB.SZ") + out = compute_valley_relative_vwap(_bars(rows), **_KW) + + solo_a = compute_valley_relative_vwap( + _bars( + _background(_BG_DAYS, n, sym="AAA.SZ") + + _session("2021-07-11", a_vols, a_amts, sym="AAA.SZ") + ), + **_KW, + ) + assert out.loc[(_TEST_DAY, "AAA.SZ")] == pytest.approx(solo_a.loc[(_TEST_DAY, "AAA.SZ")]) + assert out.loc[(_TEST_DAY, "BBB.SZ")] == pytest.approx(_CASE_B_RATIO) + + +# --------------------------------------------------------------------------- # +# §0 REUSE NON-DRIFT: the exposure refactor changed nothing for PR-F / PR-H +# --------------------------------------------------------------------------- # +def test_prepare_visible_minute_bars_default_output_is_unchanged(): + """No ``extra_columns`` -> EXACTLY the historical column list, in order. + + This is the lock on the additive exposure: PR-F / PR-H call this helper with no + extra columns, so their input frame must be byte-identical to before. + """ + rows = _background(3, 5) + out = prepare_visible_minute_bars(_bars(rows)) + assert list(out.columns) == [ + "symbol", "bar_end", "available_time", "volume", "trade_date", "slot" + ] + assert "amount" not in out.columns + + +def test_extra_columns_only_add_and_never_alter_the_shared_columns(): + rows = _background(3, 5) + _session("2021-07-04", *_case_a_day()) + bars = _bars(rows) + plain = prepare_visible_minute_bars(bars) + with_amount = prepare_visible_minute_bars(bars, extra_columns=("amount",)) + # extras ride with the other RAW columns, ahead of the derived trade_date / slot + assert list(with_amount.columns) == [ + "symbol", "bar_end", "available_time", "volume", "amount", "trade_date", "slot" + ] + # ...and every shared column is untouched, row for row + pd.testing.assert_frame_equal(with_amount[plain.columns], plain) + + +def test_extra_columns_rejects_unknown_and_duplicate_names(): + bars = _bars(_background(2, 3)) + with pytest.raises(ValueError, match="not present"): + prepare_visible_minute_bars(bars, extra_columns=("nope",)) + with pytest.raises(ValueError, match="already"): + prepare_visible_minute_bars(bars, extra_columns=("volume",)) + + +def test_peak_and_classifiable_are_unaffected_by_carrying_amount(): + """peak / classifiable must not depend on whether ``amount`` rides along. + + Together with the two factor-level locks below, this is the numeric statement of + "the exposure refactor is behaviour-preserving". + """ + vols, amts = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", vols, amts) + bars = _bars(rows) + plain = peak_mask_for_symbol( + prepare_visible_minute_bars(bars).reset_index(drop=True) + ) + carried = peak_mask_for_symbol( + prepare_visible_minute_bars(bars, extra_columns=("amount",)).reset_index(drop=True) + ) + for col in ("trade_date", "bar_end", "slot", "volume", "classifiable", "peak"): + pd.testing.assert_series_equal(carried[col], plain[col], check_names=False) + + +def test_valley_is_exactly_classifiable_and_not_eruptive(): + """valley == PR-F's internal ``mild`` == classifiable & ~eruptive, and never a peak.""" + vols, amts = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", vols, amts) + work = peak_mask_for_symbol( + prepare_visible_minute_bars(_bars(rows)).reset_index(drop=True) + ) + classifiable = work["classifiable"].to_numpy(dtype=bool) + valley = work["valley"].to_numpy(dtype=bool) + peak = work["peak"].to_numpy(dtype=bool) + # reconstruct "eruptive" independently from the raw threshold rule + eruptive = classifiable & ( + work["volume"].to_numpy(dtype=float) > work["thr"].to_numpy(dtype=float) + ) + np.testing.assert_array_equal(valley, classifiable & ~eruptive) + # a peak is an ERUPTIVE minute, so peak and valley are disjoint + assert not (valley & peak).any() + # the engineered day really does contain both kinds (the assertion is not vacuous) + day = work["trade_date"].to_numpy() == np.datetime64(_TEST_DAY) + assert valley[day].any() and eruptive[day].any() + + +def test_volume_peak_count_unchanged_by_the_exposure_refactor(): + """PR-F's factor value is bit-identical whether or not amount is carried.""" + from data.clean.intraday_volume_prv import compute_volume_peak_count + + vols, amts = _case_b_day() + rows = _background(_BG_DAYS, _CASE_B_N) + _session("2021-07-11", vols, amts) + kw = dict( + baseline_days=VOLUME_PRV_BASELINE_DAYS, + baseline_min_obs=VOLUME_PRV_BASELINE_MIN_OBS, + sigma_k=VOLUME_PRV_SIGMA_K, + lookback_days=1, + min_valid_days=1, + min_classifiable=1, + ) + out = compute_volume_peak_count(_bars(rows), **kw) + # three isolated eruptions with mild neighbours -> exactly three peaks + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(3.0) + + +def test_peak_interval_kurtosis_unchanged_by_the_exposure_refactor(): + """PR-H's factor value is bit-identical whether or not amount is carried.""" + from data.clean.intraday_peak_interval import compute_peak_interval_kurtosis + + # the PR-H hand case: peaks at 1, 3, 6, 10, 15, 21, 23 -> intervals [2,3,4,5,6,2] + n = 25 + vols = [100.0] * n + for p in (1, 3, 6, 10, 15, 21, 23): + vols[p] = 200.0 + rows = _background(_BG_DAYS, n) + _session("2021-07-11", vols, [1000.0] * n) + out = compute_peak_interval_kurtosis( + _bars(rows), + baseline_days=VOLUME_PRV_BASELINE_DAYS, + baseline_min_obs=VOLUME_PRV_BASELINE_MIN_OBS, + sigma_k=VOLUME_PRV_SIGMA_K, + lookback_days=20, + min_valid_days=1, + min_classifiable=1, + min_intervals=4, + ) + assert out.loc[(_TEST_DAY, _SYM)] == pytest.approx(-1.48125) + + +# --------------------------------------------------------------------------- # +# The per-day ratio helper (exposed for the reuse / diagnostics path) +# --------------------------------------------------------------------------- # +def test_valley_vwap_ratio_by_day_returns_only_valid_days(): + rows = _background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", *_case_a_day()) + work = peak_mask_for_symbol( + prepare_visible_minute_bars( + _bars(rows), extra_columns=("amount",) + ).reset_index(drop=True) + ) + ratio = valley_vwap_ratio_by_day(work, min_valley_bars=1, min_classifiable=1) + # the background days are unclassifiable -> not valid -> absent entirely + assert list(ratio.index) == [_TEST_DAY] + assert ratio.loc[_TEST_DAY] == pytest.approx(_CASE_A_RATIO) + + +# --------------------------------------------------------------------------- # +# Guards / purity +# --------------------------------------------------------------------------- # +def test_empty_bars_yield_empty_schema_series(): + out = compute_valley_relative_vwap(empty_intraday_bars()) + assert out.empty + assert list(out.index.names) == ["date", "symbol"] + assert out.name == "valley_relative_vwap" + + +def test_input_bars_not_mutated(): + bars = _bars(_background(_BG_DAYS, _CASE_A_N) + _session("2021-07-11", *_case_a_day())) + before = bars.copy(deep=True) + compute_valley_relative_vwap(bars, **_KW) + pd.testing.assert_frame_equal(bars, before) + + +def test_bad_params_raise(): + bars = _bars(_session("2021-07-01", [100.0] * 3, [1000.0] * 3)) + with pytest.raises(ValueError): + compute_valley_relative_vwap(bars, lookback_days=0) + with pytest.raises(ValueError): + compute_valley_relative_vwap(bars, baseline_days=1) + with pytest.raises(ValueError): + compute_valley_relative_vwap(bars, baseline_min_obs=1) + with pytest.raises(ValueError): + compute_valley_relative_vwap(bars, sigma_k=-0.5) + with pytest.raises(ValueError): + compute_valley_relative_vwap(bars, min_valid_days=0) + with pytest.raises(ValueError): + compute_valley_relative_vwap(bars, min_classifiable=0) + with pytest.raises(ValueError): + compute_valley_relative_vwap(bars, min_valley_bars=0) + + +# --------------------------------------------------------------------------- # +# Spec + Factor subclass +# --------------------------------------------------------------------------- # +def test_factor_spec_is_valid_and_daily(): + spec = ValleyRelativeVwapFactor().spec + assert isinstance(spec, FactorSpec) + assert spec.factor_id == "valley_relative_vwap_20" + assert spec.is_intraday is False + assert spec.expected_ic_sign == 1 # POSITIVE (report RankIC +8.69%) + assert spec.return_basis == "close_to_close" + assert spec.forward_return_horizon == 1 + assert set(spec.input_fields) == {"volume", "amount"} + for field in ( + "decision_cutoff", "data_lag", "session_open", + "execution_model", "execution_window", + ): + assert getattr(spec, field) is None + + +def test_factor_spec_discloses_pinned_interpretations(): + # The choices the report is silent on (or where we knowingly deviate) MUST be on the + # spec a reader sees: the visible-window day VWAP deviation and the raw-price note. + desc = ValleyRelativeVwapFactor().spec.description + assert "14:50" in desc + assert "RAW" in desc.upper() + assert "amount" in desc and "volume" in desc + + +def test_factor_subclass_window_tracks_name(): + f = ValleyRelativeVwapFactor(lookback_days=10) + assert f.name == "valley_relative_vwap_10" + assert f.spec.factor_id == "valley_relative_vwap_10" + + +def test_factor_compute_selects_preaggregated_column(): + f = ValleyRelativeVwapFactor() + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp("2021-07-01"), _SYM)], names=["date", "symbol"] + ) + panel = pd.DataFrame({f.name: [0.98]}, index=idx) + out = f.compute(panel) + assert out.loc[(pd.Timestamp("2021-07-01"), _SYM)] == 0.98 + assert out.name == f.name + + +def test_factor_compute_missing_column_raises(): + f = ValleyRelativeVwapFactor() + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp("2021-07-01"), _SYM)], names=["date", "symbol"] + ) + with pytest.raises(ValueError, match="pre-aggregated"): + f.compute(pd.DataFrame({"other": [1.0]}, index=idx)) + + +def test_factor_bad_params_raise(): + with pytest.raises(ValueError): + ValleyRelativeVwapFactor(lookback_days=0)