diff --git a/config/phase_f_volume_peak_count.yaml b/config/phase_f_volume_peak_count.yaml new file mode 100644 index 0000000..3ded8d8 --- /dev/null +++ b/config/phase_f_volume_peak_count.yaml @@ -0,0 +1,128 @@ +# PR-F — Fourth real factor evaluation: volume-peak-minute count. +# +# Reproduces the Kaiyuan market-microstructure series #27 (开源证券《高频成交量的峰、岭、 +# 谷信息——市场微观结构研究系列（27）》, reportId 4957417) flagship "量峰分钟数因子" as a +# first-class VolumePeakCountFactor 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). +# +# The subject factor works DIRECTLY on the 1min cache (no coarser resample): PIT-truncate +# each day at 14:50, classify every visible minute against its SAME-SLOT strictly-prior +# 20-day baseline (eruptive if vol > μ + σ else mild), mark the eruptive minutes whose +# both 1-minute same-session neighbours are mild as PEAKS (ridge / session-boundary / +# unclassifiable-neighbour minutes are not peaks), and count the peak minutes over the +# trailing 20 VALID trading days (a day is valid iff it has >= 100 classifiable bars; +# NaN below 10 valid days). Pre-registered sign = +1 (report IC positive: full-market +# RankIC +10.62% / RankICIR 4.36, CSI500 sub-domain long-short +14.96%/yr). NOTE the +# report is a MONTHLY, market-cap + industry neutral series on Wind data -- our eval cell +# is CSI500 daily with industry + size neutral, so those numbers are a LOOSE reference +# only. Raw minute volume (cached as-is) has split-day magnitude jumps that pollute the +# 20-day σ; 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: 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. The evaluator runs TWICE (see qt/eval_volume_peak_count.py): once with NO book +# (Incremental NOT_ASSESSED) and once with the confirmed book to measure whether the +# volume-peak-count factor adds alpha BEYOND value / low-vol. + +project: + name: quantitative_trading_pr_f_volume_peak_count + 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: volume_peak_count_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 (volume_peak_count_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_volume_prv.py b/data/clean/intraday_volume_prv.py new file mode 100644 index 0000000..779c5ca --- /dev/null +++ b/data/clean/intraday_volume_prv.py @@ -0,0 +1,295 @@ +"""Volume-peak-count factor (PR-F). + +Reproduces the Kaiyuan market-microstructure series #27 (开源证券《高频成交量的峰、岭、 +谷信息——市场微观结构研究系列（27）》, 2025-07-20, reportId 4957417) flagship +"volume-peak-minute-count" factor (§2) as a daily PIT-safe column derived DIRECTLY +from the 1min cache (no coarser resampling — the peak/ridge/valley taxonomy lives at +the 1-minute grain). Kept in the DATA-clean layer (like +:func:`data.clean.intraday_amp_anomaly.compute_amp_marginal_anomaly_vol`) so the +``factors`` layer only SELECTS the pre-aggregated column and never fetches or sees a +forward return. + +The report is under-specified about the PIT boundary; the interpretations below are +deliberate, DISCLOSED choices PINNED in the task card (task_card_pr_f_*.md §1), not +tuned knobs. They are reproduced on the factor spec so a reader can see exactly what +was assumed: + + 1. PIT truncation (standing authorization): each day keeps only the 1min bars with + ``available_time <= (that bar's trade_date + decision_time)`` (default 14:50) — + history days AND the signal day are truncated identically, so the same-slot + cross-day baseline is measured on a consistent window (≈ the morning session plus + the afternoon up to the cutoff). + 2. Same-slot baseline (PINNED as STRICTLY PRIOR — the report does not say whether + the current day is included, and strictly-prior is the more PIT-stable reading): + for minute slot ``s`` and day ``t`` the baseline ``μ_s`` / ``σ_s`` (ddof=1) is + the symbol's SAME-SLOT volume over the trailing ``baseline_days`` (=20) trading + days STRICTLY BEFORE ``t``; fewer than ``baseline_min_obs`` (=10) same-slot + observations in that window -> the ``(t, s)`` bar is NOT classifiable. + 3. classify: ``vol > μ_s + k*σ_s`` (k = 1) -> ERUPTIVE, else MILD (a "valley"). + 4. a slot is a PEAK iff it is eruptive AND both its 1-minute neighbours in the SAME + continuous session exist and are MILD. An eruptive bar whose neighbour is also + eruptive is a RIDGE (not a peak); a session-boundary bar (each session's first / + last visible bar, so a neighbour is missing across the lunch break or the cutoff) + is likewise NOT a peak (its isolation is unprovable — PINNED disclosure). A + neighbour that is not classifiable (baseline too thin) also blocks the peak. + "Same continuous session" and "the 1-minute neighbour" are enforced by requiring + the adjacent bar to be EXACTLY 60s away (the lunch break 11:30->13:01 and any + missing minute fail this, so they are correctly not neighbours). + +Factor value: ``volume_peak_count_20`` = the total count of peak minutes over the +symbol's most recent ``lookback_days`` (=20) VALID trading days INCLUDING ``d`` (a +simple count — after truncation the per-day slot count is uniform, so counts are +cross-sectionally comparable without normalization). A day is VALID iff it has at +least ``min_classifiable`` (=100) classifiable bars; 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 (more volume peaks = more informed-trading participation = +higher future returns; the report's full-market RankIC is +10.62% / RankICIR 4.36 and +its CSI500 sub-domain long-short is +14.96%/yr — the CSI500 line is the direct anchor +for our eval cell). NOTE the report is a monthly, market-cap + industry neutral series +on Wind data; our eval cell is CSI500 daily with industry + size neutral, so the +report numbers are a LOOSE reference only (disclosed, never mislabeled). Raw minute +volume (cached as-is) has magnitude jumps across split days that pollute the 20-day σ; +the report (Wind) does not adjust for this either, so we disclose it and do NOT correct +it. The value at ``(d, s)`` 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 + +# Factor DEFINITION constants (pinned interpretations of the report; NOT tuned knobs). +VOLUME_PRV_LOOKBACK_DAYS = 20 # trailing VALID trading-day count window (N), includes d +VOLUME_PRV_BASELINE_DAYS = 20 # strictly-prior same-slot baseline window (trading days) +VOLUME_PRV_BASELINE_MIN_OBS = 10 # min same-slot obs for a classifiable bar +VOLUME_PRV_SIGMA_K = 1.0 # eruptive threshold multiplier k in vol > μ + k*σ +VOLUME_PRV_MIN_VALID_DAYS = 10 # min valid days in the trailing window for a finite value +VOLUME_PRV_MIN_CLASSIFIABLE = 100 # a day is valid iff it has >= this many classifiable bars + +# The "strictly-next minute" test in seconds: two bars are same-session 1-minute +# neighbours iff their bar_end gap is EXACTLY 60s. The lunch break (11:30 -> 13:01), +# the session close and any missing minute all differ, so they are not neighbours. +_ONE_MINUTE_SECONDS = 60.0 + + +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 _peak_count_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, +) -> tuple[list[pd.Timestamp], list[float]]: + """Daily volume-peak-count values for ONE symbol from its PIT-visible bars. + + ``g`` holds columns ``trade_date`` / ``slot`` / ``bar_end`` / ``volume`` for a + SINGLE symbol (a fresh ``RangeIndex``). The same-slot STRICTLY-PRIOR baseline is + computed on a day x slot pivot (rolling over the day axis then ``shift(1)``), so no + per-``(day, slot)`` python loop is needed and the current day never enters its own + baseline. Peak detection is done back in long form with a within-day 60s-gap + neighbour test (which enforces "same continuous session" and "the 1-minute + neighbour" at once). No cross-symbol leakage — ``g`` is one symbol's slice — and no + cross-day leakage — the baseline is strictly prior and the neighbour test never + crosses a day boundary (grouped by ``trade_date``). + """ + # 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). + v = g.pivot(index="trade_date", columns="slot", values="volume") + v = v.sort_index().sort_index(axis=1) + + # Strictly-prior same-slot baseline: rolling over the DAY axis (per slot column), + # requiring >= baseline_min_obs actual same-slot observations, THEN shift(1) so + # day t uses days t-baseline_days .. t-1 only (never day t itself). + roll = v.rolling(baseline_days, min_periods=baseline_min_obs) + mu = roll.mean().shift(1) + sigma = roll.std().shift(1) # ddof=1 (pandas rolling default) + thr = mu + sigma_k * sigma + + thr_long = thr.reset_index().melt( + id_vars="trade_date", var_name="slot", value_name="thr" + ) + # melt can hand back the slot column labels as object dtype; keep it int so the + # merge below matches g's int ``slot`` (an object/int mismatch would silently miss). + thr_long["slot"] = thr_long["slot"].astype(int) + work = g.merge(thr_long, on=["trade_date", "slot"], how="left") + work = work.sort_values(["trade_date", "bar_end"], kind="mergesort").reset_index( + drop=True + ) + + vol = work["volume"].to_numpy(dtype=float) + thr_arr = work["thr"].to_numpy(dtype=float) + # Classifiable iff the strictly-prior baseline is finite (>= baseline_min_obs obs); + # eruptive iff strictly above μ + k*σ; mild is any other classifiable bar. + classifiable = np.isfinite(thr_arr) + eruptive = classifiable & (vol > thr_arr) + mild = classifiable & ~eruptive + work["classifiable"] = classifiable + # 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) + + by_day = work.groupby("trade_date", sort=False) + prev_end = by_day["bar_end"].shift(1) + next_end = by_day["bar_end"].shift(-1) + gap_prev = (work["bar_end"] - prev_end).dt.total_seconds().to_numpy() + gap_next = (next_end - work["bar_end"]).dt.total_seconds().to_numpy() + prev_mild = by_day["mild_i"].shift(1).fillna(0).to_numpy() > 0 + next_mild = by_day["mild_i"].shift(-1).fillna(0).to_numpy() > 0 + # A peak is an eruptive bar whose BOTH 1-minute neighbours (exactly 60s away, so + # same session) exist and are mild. Missing / lunch-break / cutoff neighbours fail + # the 60s gap; eruptive neighbours (ridge) and unclassifiable ones fail the mild + # test — all correctly block the peak. + peak = ( + eruptive + & (gap_prev == _ONE_MINUTE_SECONDS) + & prev_mild + & (gap_next == _ONE_MINUTE_SECONDS) + & next_mild + ) + work["peak"] = peak + + peak_count = work.groupby("trade_date")["peak"].sum() + classifiable_count = work.groupby("trade_date")["classifiable"].sum() + valid_days = classifiable_count.index[classifiable_count >= min_classifiable] + if len(valid_days) == 0: + return [], [] + + # Count over the trailing lookback_days VALID days (including d); NaN until + # min_valid_days valid days have accumulated. Values are emitted only on valid days. + pc_valid = peak_count.loc[valid_days].astype(float).sort_index() + factor_valid = pc_valid.rolling(lookback_days, min_periods=min_valid_days).sum() + days = [pd.Timestamp(d).normalize() for d in factor_valid.index] + return days, list(factor_valid.to_numpy(dtype=float)) + + +def compute_volume_peak_count( + bars: pd.DataFrame, + *, + lookback_days: int = VOLUME_PRV_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, + decision_time: str = DEFAULT_DECISION_TIME, + name: str = "volume_peak_count", +) -> pd.Series: + """PIT-safe daily "volume-peak-minute-count" factor from 1min ``bars``. + + Takes normalized 1min ``bars``, PIT-truncates each day at ``decision_time``, + classifies every visible minute against its SAME-SLOT strictly-prior baseline + (eruptive vs mild), marks the eruptive minutes whose both 1-minute same-session + neighbours are mild as PEAKS, and returns the trailing-``lookback_days``-VALID-day + peak count. See the module docstring for the LOCKED definition. + + 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 for the count (definition). + baseline_days: strictly-prior same-slot baseline window in trading days. + baseline_min_obs: minimum same-slot observations for a classifiable bar. + sigma_k: eruptive threshold multiplier ``k`` in ``vol > μ + k*σ``. + min_valid_days: minimum valid days in the trailing window for a finite value. + min_classifiable: a day is VALID iff it has at least this many classifiable 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 len(bars) == 0: + return _empty_series(name) + + work = bars.reset_index()[ + [SYMBOL_LEVEL, "bar_end", "available_time", "volume"] + ].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]. + cutoff = work["trade_date"] + pd.Timedelta(decision_time) + visible = work.loc[work["available_time"] <= cutoff].copy() + if visible.empty: + return _empty_series(name) + + # Guard bad volume BEFORE anything else: a non-finite / negative volume is invalid + # data that would poison the same-slot μ/σ. (Split-day magnitude jumps are NOT + # corrected — disclosed, matching the report's Wind treatment.) + vol = visible["volume"].to_numpy(dtype=float) + visible = visible.loc[np.isfinite(vol) & (vol >= 0.0)].copy() + if visible.empty: + return _empty_series(name) + + # slot = minute-of-day (minutes since midnight); bars are minute-aligned so this is + # exact and aligns the SAME time-of-day across days for the same-slot baseline. + visible["slot"] = ( + (visible["bar_end"] - visible["trade_date"]) // pd.Timedelta(minutes=1) + ).astype(int) + + index_tuples: list[tuple] = [] + values: list[float] = [] + for sym, g in visible.groupby(SYMBOL_LEVEL, sort=True): + days, vals = _peak_count_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, + ) + 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__ = [ + "VOLUME_PRV_BASELINE_DAYS", + "VOLUME_PRV_BASELINE_MIN_OBS", + "VOLUME_PRV_LOOKBACK_DAYS", + "VOLUME_PRV_MIN_CLASSIFIABLE", + "VOLUME_PRV_MIN_VALID_DAYS", + "VOLUME_PRV_SIGMA_K", + "compute_volume_peak_count", +] diff --git a/factors/compute/intraday_derived.py b/factors/compute/intraday_derived.py index 9ef82df..ee29c23 100644 --- a/factors/compute/intraday_derived.py +++ b/factors/compute/intraday_derived.py @@ -2,15 +2,17 @@ The members today are :class:`JumpAmountCorrFactor` (PR-C, the Kaiyuan report §6 "price-jump turnover correlation" factor), :class:`MinuteIdealAmplitudeFactor` -(PR-D, the Kaiyuan report §30 "minute ideal amplitude" factor) and +(PR-D, the Kaiyuan report §30 "minute ideal amplitude" factor), :class:`AmpMarginalAnomalyVolFactor` (PR-E, the Changjiang high-frequency-factor -series #19 "amplitude marginal-anomaly relative-volatility" factor). Like the value / -MMP factors, the heavy computation runs UPSTREAM (``data.clean.intraday_aggregate`` -/ ``data.clean.intraday_amplitude`` / ``data.clean.intraday_amp_anomaly`` aggregate -1min bars into a daily ``MultiIndex(date, symbol)`` column) and the Factor here simply -SELECTS its column off the panel the runner already enriched. Keeping the minute -aggregation in the data-clean layer preserves the layering: ``factors`` never fetches -and never sees a forward return. +series #19 "amplitude marginal-anomaly relative-volatility" factor) and +:class:`VolumePeakCountFactor` (PR-F, the Kaiyuan microstructure series #27 +"volume-peak-minute-count" factor). Like the value / MMP factors, the heavy +computation runs UPSTREAM (``data.clean.intraday_aggregate`` / +``data.clean.intraday_amplitude`` / ``data.clean.intraday_amp_anomaly`` / +``data.clean.intraday_volume_prv`` aggregate 1min bars into a daily +``MultiIndex(date, symbol)`` column) and the Factor here simply SELECTS its column off +the panel the runner already enriched. Keeping the minute aggregation in the data-clean +layer preserves the layering: ``factors`` never fetches and never sees a forward return. WHY ``is_intraday=False`` FOR A MINUTE-DERIVED FACTOR (deliberate, documented): ``FactorSpec.is_intraday`` flags an intraday-EXECUTION contract — the minute @@ -45,6 +47,14 @@ IDEAL_AMP_LOOKBACK_DAYS, IDEAL_AMP_MIN_MINUTES, ) +from data.clean.intraday_volume_prv import ( + VOLUME_PRV_BASELINE_DAYS, + VOLUME_PRV_BASELINE_MIN_OBS, + VOLUME_PRV_LOOKBACK_DAYS, + VOLUME_PRV_MIN_CLASSIFIABLE, + VOLUME_PRV_MIN_VALID_DAYS, + VOLUME_PRV_SIGMA_K, +) from factors.base import Factor from factors.spec import FactorSpec @@ -311,8 +321,108 @@ def compute(self, panel: pd.DataFrame) -> pd.Series: return panel[self.name].rename(self.name) +class VolumePeakCountFactor(Factor): + """Volume-peak-minute-count factor (daily signal, minute-derived). + + ``compute`` reads the pre-aggregated daily column the runner placed on the panel + (produced by ``data.clean.intraday_volume_prv.compute_volume_peak_count``); it does + NO minute work of its own, mirroring :class:`JumpAmountCorrFactor` / + :class:`MinuteIdealAmplitudeFactor` / :class:`AmpMarginalAnomalyVolFactor` and the + value / financial factors that surface an enriched column. + + Args: + lookback_days: trailing VALID trading-day count window; part of the factor + DEFINITION (a pinned interpretation of the report), not a tuned knob. It + only names the column so a non-default window cannot silently mislabel it. + """ + + name: str = f"volume_peak_count_{VOLUME_PRV_LOOKBACK_DAYS}" + + def __init__(self, lookback_days: int = VOLUME_PRV_LOOKBACK_DAYS) -> None: + if not isinstance(lookback_days, int) or lookback_days < 1: + raise ValueError( + f"volume-peak-count lookback_days must be a positive integer; got " + f"{lookback_days!r}." + ) + self._lookback_days = lookback_days + self.name = f"volume_peak_count_{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 IC is POSITIVE (full-market RankIC +10.62% / + RankICIR 4.36; CSI500 sub-domain long-short +14.96%/yr) — more volume peaks + (informed-trading participation) 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 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). 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 >= ``VOLUME_PRV_MIN_VALID_DAYS`` valid days accumulate in the + trailing window, and a day is valid only once its same-slot baselines fill in), + not a fixed leading count — the honest NaN rate is reported by data_coverage. + + The description spells out the pinned interpretations of the under-specified + report (PIT truncation, strictly-prior same-slot baseline, μ+σ eruptive + threshold, mild-neighbour peak rule, valid-day gate) so a reader sees exactly + what was assumed. + """ + return FactorSpec( + factor_id=self.name, + version="1.0", + description=( + f"Volume-peak-minute count (Kaiyuan microstructure series #27). PINNED " + f"interpretations of an under-specified report: (1) 1min bars " + f"PIT-truncated at 14:50 per bar; (2) same-slot baseline = μ/σ (ddof=1) " + f"of the STRICTLY-PRIOR {VOLUME_PRV_BASELINE_DAYS} trading days' " + f"same-slot volume, needing >= {VOLUME_PRV_BASELINE_MIN_OBS} obs else " + f"unclassifiable; (3) a minute is ERUPTIVE if vol > μ + " + f"{VOLUME_PRV_SIGMA_K:g}σ else MILD; (4) a PEAK is an eruptive minute " + f"whose both 1-minute same-session neighbours exist and are mild " + f"(ridge / session-boundary / unclassifiable-neighbour minutes are not " + f"peaks); (5) factor = peak-minute count over the trailing " + f"{self._lookback_days} VALID days (>= {VOLUME_PRV_MIN_CLASSIFIABLE} " + f"classifiable bars) including d, NaN below " + f"{VOLUME_PRV_MIN_VALID_DAYS} valid days. Derived from 1min bars but a " + f"DAILY signal traded close-to-close." + ), + expected_ic_sign=1, + is_intraday=False, + forward_return_horizon=1, + return_basis="close_to_close", + # The 1min bar field the upstream aggregation is derived from. Declared for + # honest provenance disclosure (data_coverage lists it); the daily panel + # surfaces the pre-aggregated column itself. + input_fields=("volume",), + family="microstructure", + min_history_bars=0, + ) + + def compute(self, panel: pd.DataFrame) -> pd.Series: + """Select the pre-aggregated daily volume-peak-count column off ``panel``. + + The runner runs ``compute_volume_peak_count`` 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"VolumePeakCountFactor needs the pre-aggregated '{self.name}' column " + f"on the panel (produced upstream by compute_volume_peak_count and " + f"joined by the runner); panel has {list(panel.columns)}." + ) + return panel[self.name].rename(self.name) + + __all__ = [ "AmpMarginalAnomalyVolFactor", "JumpAmountCorrFactor", "MinuteIdealAmplitudeFactor", + "VolumePeakCountFactor", ] diff --git a/qt/cli.py b/qt/cli.py index 4bb9fb0..7bf36a7 100644 --- a/qt/cli.py +++ b/qt/cli.py @@ -278,6 +278,31 @@ def _cmd_run_eval_amp_marginal_anomaly_vol(args: argparse.Namespace) -> int: return 0 +def _cmd_run_eval_volume_peak_count(args: argparse.Namespace) -> int: + """Run the two real volume-peak-count factor evaluations (cache-only) + reports.""" + from qt.eval_volume_peak_count import run_eval_volume_peak_count + + try: + result = run_eval_volume_peak_count(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-volume-peak-count: 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 @@ -430,6 +455,13 @@ def build_parser() -> argparse.ArgumentParser: p_amav.add_argument("--config", required=True, help="Path to the YAML config.") p_amav.set_defaults(func=_cmd_run_eval_amp_marginal_anomaly_vol) + p_vpc = sub.add_parser( + "run-eval-volume-peak-count", + help="Run the volume-peak-count factor evaluation (CSI500, cache-only).", + ) + p_vpc.add_argument("--config", required=True, help="Path to the YAML config.") + p_vpc.set_defaults(func=_cmd_run_eval_volume_peak_count) + 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_volume_peak_count.py b/qt/eval_volume_peak_count.py new file mode 100644 index 0000000..95dc47d --- /dev/null +++ b/qt/eval_volume_peak_count.py @@ -0,0 +1,503 @@ +"""run-eval-volume-peak-count: the fourth real factor evaluation (PR-F). + +Reproduces the Kaiyuan market-microstructure series #27 (开源证券《高频成交量的峰、岭、 +谷信息——市场微观结构研究系列（27）》, reportId 4957417) flagship "volume-peak-minute-count" +factor as a first-class +:class:`~factors.compute.intraday_derived.VolumePeakCountFactor` 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-D / PR-E used. + +The factor is a DAILY signal derived DIRECTLY from the 1min cache (no coarser resample; +see ``data.clean.intraday_volume_prv.compute_volume_peak_count``): PIT-truncate each day +at 14:50, classify every visible minute against its SAME-SLOT strictly-prior 20-day +baseline (eruptive if ``vol > μ + σ`` else mild), mark the eruptive minutes whose both +1-minute same-session neighbours are mild as PEAKS, and count the peak minutes 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). + +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 volume-peak-count +factor adds alpha BEYOND value / low-vol. +""" + +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_volume_prv import ( + VOLUME_PRV_BASELINE_DAYS, + VOLUME_PRV_BASELINE_MIN_OBS, + VOLUME_PRV_LOOKBACK_DAYS, + VOLUME_PRV_MIN_CLASSIFIABLE, + VOLUME_PRV_MIN_VALID_DAYS, + VOLUME_PRV_SIGMA_K, + compute_volume_peak_count, +) +from data.clean.schema import CORE_COLUMNS, DATE_LEVEL +from factors.compute.intraday_derived import VolumePeakCountFactor +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_volume_peak_count" +_REPORT_STEM = "eval_volume_peak_count" + + +# --------------------------------------------------------------------------- # +# Minute loading (cache-only, per-symbol -> memory-bounded) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _VolumePeakMinuteLoad: + """Diagnostics from the cache-only per-symbol minute read + aggregation.""" + + factor: pd.Series # MultiIndex(date, symbol) raw volume-peak-count + 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_volume_peak_count_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, +) -> _VolumePeakMinuteLoad: + """Compute the raw volume-peak-count 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 same-slot baseline / + peak taxonomy runs entirely inside ``compute_volume_peak_count`` on 1min bars. + """ + 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_volume_peak_count( + 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, + 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-volume-peak-count blocked: no requested symbol produced a cached " + f"volume-peak-count value over [{cfg.data.start}, {cfg.data.end}]. The " + "minute cache is required (this runner never warms it); check coverage." + ) + factor = pd.concat(series).sort_index() + logger.info( + "minute aggregation (cache-only): %d/%d symbols with a value, %d raw 1min " + "rows read, %d factor rows, stk_mins_live_calls=0", + len(covered), len(symbols), raw_rows, len(factor), + ) + return _VolumePeakMinuteLoad( + 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-volume-peak-count 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 VolumePeakCountEvalResult: + """Immutable summary of one run-eval-volume-peak-count 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-volume-peak-count 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-volume-peak-count needs data.cache.enabled=true (it reads the " + "persistent tushare cache and never warms live)." + ) + if cfg.universe.type != "index": + raise ValueError( + "run-eval-volume-peak-count needs universe.type='index' (PIT membership, " + f"e.g. 000905.SH for CSI500); got {cfg.universe.type!r}." + ) + if not cfg.processing.neutralize.enabled: + raise ValueError( + "run-eval-volume-peak-count expects processing.neutralize.enabled=true " + "(industry + size neutralization, matching the report's neutral column and " + "the EvalConfig declaration)." + ) + + +def run_eval_volume_peak_count(config_path: str) -> VolumePeakCountEvalResult: + """Run the two real volume-peak-count 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 = VolumePeakCountFactor(lookback_days=VOLUME_PRV_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) + + # Volume-peak-count factor: cache-only per-symbol aggregation -> raw -> process. + load = _load_volume_peak_count_panel( + cfg, symbols, spec, logger, + lookback_days=VOLUME_PRV_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, + ) + # 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 VolumePeakCountEvalResult( + 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__ = [ + "VolumePeakCountEvalResult", + "evaluate_two_runs", + "extract_metrics", + "run_eval_volume_peak_count", +] diff --git a/tests/test_eval_volume_peak_count_runner.py b/tests/test_eval_volume_peak_count_runner.py new file mode 100644 index 0000000..7a7e8de --- /dev/null +++ b/tests/test_eval_volume_peak_count_runner.py @@ -0,0 +1,258 @@ +"""PR-F 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 VolumePeakCountFactor +from qt.config import ( + AlphaCfg, + BacktestCfg, + CacheCfg, + CostCfg, + DataCfg, + FactorCfg, + OutputCfg, + PortfolioCfg, + RootConfig, + UniverseCfg, +) +from qt.eval_volume_peak_count import ( + _load_volume_peak_count_panel, + evaluate_two_runs, + extract_metrics, +) + + +# --------------------------------------------------------------------------- # +# Minute cache-only loader +# --------------------------------------------------------------------------- # +def _stored_rows(sym, day, vols): + """Build STORED_COLUMNS-shaped 1min rows for one session of CONSECUTIVE minutes. + + Bar i sits at ``09:31 + i`` minutes (all inside the 14:50 PIT window); consecutive + minutes are 60s apart so interior bars have both 1-minute neighbours. OHLC are dummy + constants (the factor reads only ``volume``); ``amount`` mirrors ``volume``. + """ + base = pd.Timestamp(day) + pd.Timedelta("09:31:00") + rows = [] + for i, v in enumerate(vols): + 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(v), + "freq": "1min", + } + ) + return pd.DataFrame(rows) + + +def _min_config(root, start, end): + return RootConfig( + data=DataCfg( + source="tushare", + start=start, + end=end, + external_secret_file="/nonexistent.json", + cache=CacheCfg(enabled=True, root_dir=str(root)), + ), + universe=UniverseCfg(type="static", symbols=["A.SZ"]), + factors=[FactorCfg(name="momentum_20")], + alpha=AlphaCfg(), + portfolio=PortfolioCfg(top_n=1), + backtest=BacktestCfg(), + cost=CostCfg(), + output=OutputCfg(), + ) + + +# Three constant background days (baseline) then a test day with two clean peaks (the +# 200-volume minutes at 09:32 and 09:35, both with mild neighbours). +_DAYS = ("2021-07-01", "2021-07-02", "2021-07-03") +_TEST_DAY = "2021-07-04" +_BG_VOLS = [100.0] * 6 +_PEAK_VOLS = [100.0, 200.0, 100.0, 100.0, 200.0, 100.0] + +# Small gates so a 4-day cache produces a value (baseline needs >= 2 prior obs). +_LOAD_KW = dict( + lookback_days=20, baseline_days=20, baseline_min_obs=2, + sigma_k=1.0, min_valid_days=1, min_classifiable=1, +) + + +def _seed_symbol(store, sym): + for day in _DAYS: + store.upsert(INTRADAY_ENDPOINT, sym, "1min", _stored_rows(sym, day, _BG_VOLS), KEY_COLS) + store.upsert( + INTRADAY_ENDPOINT, sym, "1min", _stored_rows(sym, _TEST_DAY, _PEAK_VOLS), 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 = VolumePeakCountFactor().spec + logger = logging.getLogger("test.vpc.loader") + load = _load_volume_peak_count_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 two-peak count (=2) on both covered symbols + d = pd.Timestamp(_TEST_DAY) + assert load.factor.loc[(d, "AAA.SZ")] == pytest.approx(2.0) + assert load.factor.loc[(d, "BBB.SZ")] == pytest.approx(2.0) + assert set(load.factor.index.get_level_values("symbol")) == {"AAA.SZ", "BBB.SZ"} + + +def test_minute_loader_blocks_when_nothing_cached(tmp_path): + cfg = _min_config(tmp_path / "empty", "2021-07-01", "2021-07-04") + spec = VolumePeakCountFactor().spec + with pytest.raises(ValueError, match="no requested symbol produced"): + _load_volume_peak_count_panel( + cfg, ["ZZZ.SZ"], spec, logging.getLogger("test.vpc.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="volume_peak_count_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 = VolumePeakCountFactor().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, VolumePeakCountFactor().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_volume_peak_count_factor.py b/tests/test_volume_peak_count_factor.py new file mode 100644 index 0000000..cdf0430 --- /dev/null +++ b/tests/test_volume_peak_count_factor.py @@ -0,0 +1,396 @@ +"""PR-F: volume-peak-minute-count factor. + +The factor PIT-truncates 1min bars at 14:50, classifies each visible minute against its +SAME-SLOT strictly-prior baseline (eruptive if ``vol > μ + σ``, else mild), marks the +eruptive minutes whose both 1-minute same-session neighbours are mild as PEAKS, and +returns the peak-minute count over the trailing ``N`` VALID trading days. Sign is +pre-registered +1 (more volume peaks -> higher forward return). + +Hand cases build a constant (or two-level) BACKGROUND of prior days so the same-slot +baseline μ/σ is exact, then a test day whose eruptive / mild pattern is engineered so +the peak count is a known small integer. Because the baseline is STRICTLY PRIOR, the +only VALID day (>= min_classifiable classifiable bars) is the test day, so the factor +value on the test day equals that day's peak count (single-valid-day window). +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from data.clean.intraday_volume_prv import ( + VOLUME_PRV_BASELINE_DAYS, + VOLUME_PRV_BASELINE_MIN_OBS, + VOLUME_PRV_LOOKBACK_DAYS, + VOLUME_PRV_MIN_CLASSIFIABLE, + VOLUME_PRV_MIN_VALID_DAYS, + VOLUME_PRV_SIGMA_K, + compute_volume_peak_count, +) +from data.clean.intraday_schema import empty_intraday_bars, normalize_intraday_bars +from factors.compute.intraday_derived import VolumePeakCountFactor +from factors.spec import FactorSpec + +_SYM = "000001.SZ" + + +def _bars(rows): + """rows = [(time, symbol, volume), ...] -> normalized 1min bars. + + OHLC are dummy constants (the factor reads only ``volume`` for the slot baseline); + ``amount`` mirrors volume harmlessly. ``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[2]) for r in rows], + } + ) + return normalize_intraday_bars(df, freq="1min") + + +def _session(day, vols, sym=_SYM, start="09:31:00"): + """One session of CONSECUTIVE 1-minute bars (start, start+1m, ...) with ``vols``. + + Consecutive minutes are exactly 60s apart, so within a session every interior bar + has both 1-minute neighbours (the 60s-gap neighbour test). + """ + base = pd.Timestamp(day) + pd.Timedelta(start) + return [(base + pd.Timedelta(minutes=i), sym, v) for i, v in enumerate(vols)] + + +def _background(n_days, bg_vols, sym=_SYM, start_day="2021-07-01", start="09:31:00"): + """``n_days`` prior days each with the SAME slots/volumes -> a clean slot baseline. + + Uses consecutive CALENDAR days (the factor treats each date-with-bars as one + trading-day row; it consults no exchange calendar), so day ``n_days`` is the test + day and has exactly ``n_days`` strictly-prior same-slot observations. + """ + rows = [] + for i in range(n_days): + day = (pd.Timestamp(start_day) + pd.Timedelta(days=i)).strftime("%Y-%m-%d") + rows += _session(day, bg_vols, sym=sym, start=start) + return rows + + +def _test_day(n_days, start_day="2021-07-01"): + return pd.Timestamp(start_day) + pd.Timedelta(days=n_days) + + +# Small gates so the single test day is the only valid day and its window is length 1. +_KW = dict( + baseline_days=VOLUME_PRV_BASELINE_DAYS, + baseline_min_obs=VOLUME_PRV_BASELINE_MIN_OBS, + sigma_k=VOLUME_PRV_SIGMA_K, + lookback_days=VOLUME_PRV_LOOKBACK_DAYS, + min_valid_days=1, + min_classifiable=1, +) + + +# --------------------------------------------------------------------------- # +# Hand-computed peak counts +# --------------------------------------------------------------------------- # +def test_two_clean_peaks_hand_value(): + # background: 10 days, 6 slots all volume 100 -> baseline μ=100, σ=0, thr=100. + # test day: eruptive (200) at 09:32 and 09:35, both with mild (100) neighbours. + rows = _background(10, [100.0] * 6) + rows += _session("2021-07-11", [100.0, 200.0, 100.0, 100.0, 200.0, 100.0]) + out = compute_volume_peak_count(_bars(rows), **_KW) + assert out.loc[(_test_day(10), _SYM)] == pytest.approx(2.0) + assert out.name == "volume_peak_count" + + +def test_peak_ridge_and_boundary_hand_value(): + # 7 slots. test day eruptions at 09:32, 09:34, 09:35, 09:37. + # 09:32 -> peak (mild neighbours 09:31, 09:33) + # 09:34 -> RIDGE (its next neighbour 09:35 is eruptive) -> not a peak + # 09:35 -> RIDGE (its prev neighbour 09:34 is eruptive) -> not a peak + # 09:37 -> last bar of the session (no next neighbour) -> not a peak + # => exactly 1 peak. + rows = _background(10, [100.0] * 7) + rows += _session( + "2021-07-11", [100.0, 200.0, 100.0, 200.0, 200.0, 100.0, 200.0] + ) + out = compute_volume_peak_count(_bars(rows), **_KW) + assert out.loc[(_test_day(10), _SYM)] == pytest.approx(1.0) + + +def test_threshold_is_mu_plus_sigma_not_mu(): + # Two-level background so σ > 0: every slot alternates 80 / 120 across 10 days -> + # μ=100, σ=std([80,120]*5, ddof=1)=21.08, thr=121.08. A test-day slot at 115 is + # ABOVE μ but BELOW μ+σ -> MILD (a μ-only rule would wrongly call it eruptive and + # turn the real peak into a ridge, giving 0). The 200 bar is the only eruption. + bg_pattern = [80.0, 120.0] * 5 # 10 prior days, same at every slot + rows = [] + for i, vol in enumerate(bg_pattern): + day = (pd.Timestamp("2021-07-01") + pd.Timedelta(days=i)).strftime("%Y-%m-%d") + rows += _session(day, [vol] * 5) + rows += _session("2021-07-11", [100.0, 115.0, 200.0, 100.0, 100.0]) + out = compute_volume_peak_count(_bars(rows), **_KW) + # 09:33 (200) eruptive; neighbours 09:32 (115, mild) and 09:34 (100, mild) -> 1 peak + assert out.loc[(_test_day(10), _SYM)] == pytest.approx(1.0) + + +def test_boundary_eruptions_at_session_ends_are_not_peaks(): + # eruptions ONLY at the first and last slot of the session -> each lacks a + # neighbour on one side -> 0 peaks. + rows = _background(10, [100.0] * 5) + rows += _session("2021-07-11", [200.0, 100.0, 100.0, 100.0, 200.0]) + out = compute_volume_peak_count(_bars(rows), **_KW) + assert out.loc[(_test_day(10), _SYM)] == pytest.approx(0.0) + + +# --------------------------------------------------------------------------- # +# Same-slot baseline is STRICTLY PRIOR (no lookahead) + PIT truncation +# --------------------------------------------------------------------------- # +def test_future_day_does_not_change_earlier_factor(): + # The baseline is strictly prior and the count is trailing, so appending a WILD + # future day must not change the test day's value. + base = _background(10, [100.0] * 6) + base += _session("2021-07-11", [100.0, 200.0, 100.0, 100.0, 200.0, 100.0]) + a = compute_volume_peak_count(_bars(base), **_KW) + future = base + _session("2021-07-12", [9_999.0, 9_999.0, 1.0, 9_999.0, 1.0, 9_999.0]) + b = compute_volume_peak_count(_bars(future), **_KW) + key = (_test_day(10), _SYM) + assert np.isfinite(a.loc[key]) + assert a.loc[key] == pytest.approx(b.loc[key]) + + +def test_current_day_volume_not_in_its_own_baseline(): + # A slot's own current-day volume must not enter its baseline. With a σ=0 + # background (thr=100), any test-day value > 100 is eruptive regardless of how big + # it is; changing 200 -> 5000 at the eruption slot keeps it eruptive (baseline + # unchanged) and its mild neighbours unchanged -> same peak count. A baseline that + # wrongly folded in the current day would shift with the perturbation. + rows_a = _background(10, [100.0] * 6) + _session( + "2021-07-11", [100.0, 200.0, 100.0, 100.0, 200.0, 100.0] + ) + rows_b = _background(10, [100.0] * 6) + _session( + "2021-07-11", [100.0, 5_000.0, 100.0, 100.0, 5_000.0, 100.0] + ) + a = compute_volume_peak_count(_bars(rows_a), **_KW) + b = compute_volume_peak_count(_bars(rows_b), **_KW) + key = (_test_day(10), _SYM) + assert a.loc[key] == pytest.approx(2.0) + assert b.loc[key] == pytest.approx(2.0) + + +def test_perturbing_post_1450_bars_does_not_change_factor(): + # A bar with bar_end >= 14:50 is available only at >= 14:51 (> cutoff) -> excluded; + # its volume must not affect the factor. + rows = _background(10, [100.0] * 6) + _session( + "2021-07-11", [100.0, 200.0, 100.0, 100.0, 200.0, 100.0] + ) + a = compute_volume_peak_count(_bars(rows), **_KW) + late = rows + [ + (pd.Timestamp("2021-07-11 14:50"), _SYM, 9_999.0), + (pd.Timestamp("2021-07-11 14:55"), _SYM, 9_999.0), + ] + b = compute_volume_peak_count(_bars(late), **_KW) + key = (_test_day(10), _SYM) + assert a.loc[key] == pytest.approx(b.loc[key]) + + +# --------------------------------------------------------------------------- # +# Cross-session non-adjacency (11:30 and 13:01 are NOT neighbours) +# --------------------------------------------------------------------------- # +def test_cross_session_11_30_and_13_01_are_not_neighbours(): + # slots 11:29, 11:30, 13:01, 13:02, 13:03 (morning end + afternoon start). + # test day eruptions at 11:30 and 13:02. + # 11:30 -> last morning bar; its "next" 13:01 is 91 min away (lunch) -> NOT a + # neighbour -> 11:30 is not a peak (would be a peak if 13:01 counted). + # 13:02 -> interior afternoon bar with mild 13:01/13:03 neighbours -> a PEAK. + # => exactly 1 peak (the positive control proves peaks ARE counted). + slots = ["11:29:00", "11:30:00", "13:01:00", "13:02:00", "13:03:00"] + + def day(d, vols): + base = pd.Timestamp(d) + return [(base + pd.Timedelta(s), _SYM, v) for s, v in zip(slots, vols)] + + rows = [] + for i in range(10): + d = (pd.Timestamp("2021-07-01") + pd.Timedelta(days=i)).strftime("%Y-%m-%d") + rows += day(d, [100.0] * 5) + rows += day("2021-07-11", [100.0, 200.0, 100.0, 200.0, 100.0]) + out = compute_volume_peak_count(_bars(rows), **_KW) + assert out.loc[(_test_day(10), _SYM)] == pytest.approx(1.0) + + +def test_missing_minute_breaks_neighbour_adjacency(): + # If the bar one minute before the eruption is missing, the eruption has no + # 1-minute prior neighbour (the gap to the surviving earlier bar is 120s) -> the + # eruption is not a peak. Drop the 09:32 bar; eruption at 09:33 then lacks its + # prior neighbour -> 0 peaks. + rows = _background(10, [100.0] * 6) + full = _session("2021-07-11", [100.0, 100.0, 200.0, 100.0, 100.0, 100.0]) + rows += [r for r in full if r[0] != pd.Timestamp("2021-07-11 09:32")] + out = compute_volume_peak_count(_bars(rows), **_KW) + assert out.loc[(_test_day(10), _SYM)] == pytest.approx(0.0) + + +# --------------------------------------------------------------------------- # +# Per-symbol isolation +# --------------------------------------------------------------------------- # +def test_per_symbol_isolation(): + rows = _background(10, [100.0] * 6, sym="AAA.SZ") + rows += _session("2021-07-11", [100.0, 200.0, 100.0, 100.0, 200.0, 100.0], sym="AAA.SZ") + rows += _background(10, [100.0] * 6, sym="BBB.SZ") + rows += _session("2021-07-11", [100.0, 200.0, 100.0, 100.0, 100.0, 100.0], sym="BBB.SZ") + out = compute_volume_peak_count(_bars(rows), **_KW) + d = _test_day(10) + assert out.loc[(d, "AAA.SZ")] == pytest.approx(2.0) # two peaks + assert out.loc[(d, "BBB.SZ")] == pytest.approx(1.0) # one peak + assert out.loc[(d, "AAA.SZ")] != out.loc[(d, "BBB.SZ")] + + +# --------------------------------------------------------------------------- # +# Gates: baseline insufficient, valid-day floor +# --------------------------------------------------------------------------- # +def test_baseline_insufficient_yields_no_value(): + # Only 9 prior days -> fewer than baseline_min_obs (=10) same-slot obs -> the test + # day has 0 classifiable bars -> not a valid day -> no value at all. + rows = _background(9, [100.0] * 6) + rows += _session("2021-07-10", [100.0, 200.0, 100.0, 100.0, 200.0, 100.0]) + out = compute_volume_peak_count(_bars(rows), **_KW) + assert out.dropna().empty + + +def test_min_classifiable_gate_invalidates_thin_days(): + # A day with fewer than min_classifiable classifiable bars is not valid. Here every + # day has only 3 slots; with min_classifiable=5 the test day is invalid -> no value. + rows = _background(10, [100.0] * 3) + rows += _session("2021-07-11", [100.0, 200.0, 100.0]) + out = compute_volume_peak_count( + _bars(rows), + baseline_days=20, baseline_min_obs=10, sigma_k=1.0, + lookback_days=20, min_valid_days=1, min_classifiable=5, + ) + assert out.dropna().empty + + +def test_valid_day_floor_returns_nan_until_enough_valid_days(): + # 10 background days + 3 valid test days, each with one big eruption (1 peak). With + # min_valid_days=3: the first two valid days are NaN (only 1 / 2 valid days in the + # trailing window), the third sums all three peak counts (=3). + rows = _background(10, [100.0] * 6) + for k in range(3): + d = (pd.Timestamp("2021-07-11") + pd.Timedelta(days=k)).strftime("%Y-%m-%d") + rows += _session(d, [100.0, 900.0, 100.0, 100.0, 100.0, 100.0]) + out = compute_volume_peak_count( + _bars(rows), + baseline_days=20, baseline_min_obs=10, sigma_k=1.0, + lookback_days=20, min_valid_days=3, min_classifiable=1, + ) + d0 = pd.Timestamp("2021-07-11") + d1 = pd.Timestamp("2021-07-12") + d2 = pd.Timestamp("2021-07-13") + assert np.isnan(out.loc[(d0, _SYM)]) + assert np.isnan(out.loc[(d1, _SYM)]) + assert out.loc[(d2, _SYM)] == pytest.approx(3.0) + + +# --------------------------------------------------------------------------- # +# Guards / purity +# --------------------------------------------------------------------------- # +def test_empty_bars_yield_empty_schema_series(): + out = compute_volume_peak_count(empty_intraday_bars()) + assert out.empty + assert list(out.index.names) == ["date", "symbol"] + assert out.name == "volume_peak_count" + + +def test_input_bars_not_mutated(): + bars = _bars(_background(10, [100.0] * 6) + _session( + "2021-07-11", [100.0, 200.0, 100.0, 100.0, 200.0, 100.0] + )) + before = bars.copy(deep=True) + compute_volume_peak_count(bars, **_KW) + pd.testing.assert_frame_equal(bars, before) + + +def test_bad_params_raise(): + bars = _bars(_session("2021-07-01", [100.0, 100.0, 100.0])) + with pytest.raises(ValueError): + compute_volume_peak_count(bars, lookback_days=0) + with pytest.raises(ValueError): + compute_volume_peak_count(bars, baseline_days=1) + with pytest.raises(ValueError): + compute_volume_peak_count(bars, baseline_min_obs=1) + with pytest.raises(ValueError): + compute_volume_peak_count(bars, sigma_k=-0.5) + with pytest.raises(ValueError): + compute_volume_peak_count(bars, min_valid_days=0) + with pytest.raises(ValueError): + compute_volume_peak_count(bars, min_classifiable=0) + + +# --------------------------------------------------------------------------- # +# Spec + Factor subclass +# --------------------------------------------------------------------------- # +def test_factor_spec_is_valid_and_daily(): + spec = VolumePeakCountFactor().spec + assert isinstance(spec, FactorSpec) + assert spec.factor_id == "volume_peak_count_20" + assert spec.is_intraday is False + assert spec.expected_ic_sign == 1 # POSITIVE (report IC positive) + assert spec.return_basis == "close_to_close" + assert spec.forward_return_horizon == 1 + assert spec.input_fields == ("volume",) + for field in ( + "decision_cutoff", "data_lag", "session_open", + "execution_model", "execution_window", + ): + assert getattr(spec, field) is None + + +def test_factor_defaults_match_module_constants(): + assert VOLUME_PRV_LOOKBACK_DAYS == 20 + assert VOLUME_PRV_BASELINE_DAYS == 20 + assert VOLUME_PRV_BASELINE_MIN_OBS == 10 + assert VOLUME_PRV_SIGMA_K == 1.0 + assert VOLUME_PRV_MIN_VALID_DAYS == 10 + assert VOLUME_PRV_MIN_CLASSIFIABLE == 100 + f = VolumePeakCountFactor() + assert f.lookback_days == VOLUME_PRV_LOOKBACK_DAYS + assert f.name == "volume_peak_count_20" + + +def test_factor_subclass_window_tracks_name(): + f = VolumePeakCountFactor(lookback_days=10) + assert f.name == "volume_peak_count_10" + assert f.spec.factor_id == "volume_peak_count_10" + + +def test_factor_compute_selects_preaggregated_column(): + f = VolumePeakCountFactor() + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp("2021-07-01"), _SYM)], names=["date", "symbol"] + ) + panel = pd.DataFrame({f.name: [7.0]}, index=idx) + out = f.compute(panel) + assert out.loc[(pd.Timestamp("2021-07-01"), _SYM)] == 7.0 + assert out.name == f.name + + +def test_factor_compute_missing_column_raises(): + f = VolumePeakCountFactor() + 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): + VolumePeakCountFactor(lookback_days=0)