"""Backtest harness — out-of-sample validation of the §9.6 rate-sensitivity engine. Forgejo issue #978 (#951-B validation, Site Finder v2 EPIC 7). **STRICTLY READ-ONLY**: this script issues only SELECT queries against prod. It never INSERTs, UPDATEs, or runs DDL. WHAT IT MEASURES ---------------- The §9.6 engine (#1009 ``rate_sensitivity.py``) regresses a segment's monthly Δln(sold-units) on Δ(key_rate) at several lags, picks the lag with the most negative gate-passing slope (β), and emits an ADVISORY phrase. That engine is the gate that decides whether the advisory forecast can be PROMOTED. This harness is the out-of-sample (OOS) check that gate must clear: does β fit on a TRAIN window actually predict the SIGN of Δln(sales) on a held-out TEST window? We mirror the trade-in ``backtest_estimator.py`` discipline (read-only, holdout, in-sample-vs-OOS honesty block): 1. Build the monthly sold-units series from Source B (``objective_lots``, EKB-wide and per ``class``) and the monthly ``key_rate`` (last-known per month, reusing ``forecasting.macro_series.get_monthly_macro``). 2. Align by year-month → Δln(sales) (reuse ``sales_series.log_diff``) and Δkey_rate (first difference). 3. TIME-ORDERED holdout (NOT random / even-odd — this is a time series): fit on the first ``holdout_frac`` of months, evaluate on the rest. Fit reuses ``rate_sensitivity.best_lag`` (→ β + winning lag on TRAIN only). 4. Predict each TEST month's Δln(sales) sign (and magnitude) from β·Δkey_rate@lag, strictly point-in-time (macro as-of each test month, the lagged regressor never reaches past the test month). 5. Report the OOS directional hit-rate (fraction of test months where the predicted sign matches the actual sign), signed MAE, n_train / n_test, and the winning lag — alongside the in-sample R². IN-SAMPLE vs OUT-OF-SAMPLE HONESTY (copy trade-in discipline) ------------------------------------------------------------ We report BOTH the in-sample R² (high by construction — the slope is fit to minimise residuals on the very points it is scored on) AND the OOS directional hit-rate (the only trustworthy number). Stated plainly: the in-sample R² is NOT evidence the engine predicts — it is arithmetic. Only the held-out directional hit-rate, computed point-in-time on months the fit never saw, tells us whether β·Δrate@lag carries real predictive value worth promoting from advisory. PER-TIER -------- We backtest EKB-wide and each ``class``. A class-specific β is only worth promoting if it beats the EKB-wide prior OUT-OF-SAMPLE (directional hit-rate lift). A tier with fewer than ``_MIN_BACKTEST_MONTHS`` aligned train+test months is SKIPPED with a printed note (no silent drop). VERDICT ------- We print whether the EKB-wide OOS directional hit-rate beats a 0.5 coin-flip baseline (by at least ``_VERDICT_HITRATE_MARGIN``) AND the winning lag is the same on TRAIN and on a full-sample refit (lag stability) → "engine has OOS predictive value (candidate to promote from advisory)" vs "insufficient OOS signal — keep advisory". We stay honest: 102 raw months still collapse to few test points after Δ (loses 1), the survivorship-thinned early months, and the holdout split — if the OOS test set is tiny we say so rather than over-claim. TWO CLEANER CROSS-CHECKS (#978b) -------------------------------- The Source B OOS verdict was ``no signal`` (hit-rate < coin-flip). Source B is survivorship-CONFOUNDED — its monthly counts trend ~3x upward 2019→2025 purely because only currently-listed lots are visible, so the engine's negative verdict could be an ARTIFACT rather than a true ``no signal``. We add two controls: • ``--source A`` — build the series from ``objective_corpus_room_month`` (Objective's corp_sum monthly deal AGGREGATE). This counts deals PER MONTH regardless of current listing → survivorship-FREE. It is only ~13 months deep on prod (≈2025-05→2026-05), statistically thin, but a clean cross-check (the verdict carries an explicit thin-data caveat for it). • ``--detrend`` — before differencing, fit a linear time trend ``ln(units) ~ a + b·month_index`` and subtract it, regressing the Δ of the RESIDUALS vs Δrate. A spurious monotone survivorship trend lands almost entirely in ``b`` and is removed, so it can no longer drive the regression. The trend is fit on the TRAIN months ONLY and projected point-in-time onto the test months (#978 Part A: fitting it on train+test together leaks future info into the test residuals and inflates the detrended OOS hit-rate). Read both alongside Source B raw: if Source B DETRENDED still shows no OOS signal AND survivorship-free Source A agrees (thin caveat aside), the engine's negative verdict is a real ``no signal``, not a survivorship artifact. CAVEATS (read before trusting the numbers) ------------------------------------------ (a) SURVIVORSHIP — Source B (``objective_lots``) is the last UPSERT snapshot per lot: only currently-listed lots are visible, so sold-and-delisted lots are undercounted in OLDER months. The monthly sold-units series is therefore biased low on the early window (a ~3x spurious upward trend). ``--detrend`` and ``--source A`` are the controls for this. (b) SHORT Δ-SERIES — even ~102 months of lots collapse after first-difference + the survivorship-thinned head + the holdout split. The OOS test window can be small; the hit-rate's confidence is correspondingly weak. The verdict notes this explicitly. Source A (~13 months) is thinner still and will usually be SKIPPED below ``_MIN_BACKTEST_MONTHS`` — never faked. (c) ADVISORY MECHANISM ONLY — this exercises the β / lag CORE of the engine. It does NOT replay the full module (shrinkage to the EKB prior, the confounded-window flag, the Z-bucket phrase). It answers one question: does the fitted slope predict direction out-of-sample? CANDIDATE-METHOD VARIANTS (#978 Almon / #979 deseasonalize) ----------------------------------------------------------- Two forecast modules were shipped advisory-only and NOT wired into prod, pending a backtest proving they recover OOS signal the production ``best_lag``-on-raw path missed. This harness evaluates BOTH as opt-in variants, each isolated against the always-on raw reference (we never explode into all combinations): • ``--deseasonalize`` (#979 ``normalize.py``) — divide month-of-year seasonal factors out of the units series BEFORE ``log_diff`` (cleaner regressand), then score with the SAME ``best_lag`` engine. Factors are fit on TRAIN months ONLY and applied point-in-time onto the test months (a test-month observation must NEVER influence the factors — same leakage discipline as ``--detrend``). • ``--almon`` (#978 ``regression.py``) — replace ``best_lag``'s single winning lag with an Almon polynomial distributed-lag estimator (a smooth β curve over lags 0..6, reported via its long-run multiplier). A NEW OOS evaluator (``evaluate_oos_almon``) fits on TRAIN only and predicts each test month ``Σ_j β_j·Δrate[t−j]`` strictly point-in-time. A NEGATIVE result is a valid, honest outcome — you cannot extract signal that isn't there. The failure mode this harness guards against is a leaky evaluator that FALSELY shows signal, so both methods' TRAIN/TEST boundaries are pinned and unit-tested on synthetic series with known answers. USAGE ----- DATABASE_URL=postgresql+psycopg://... \ python -m scripts.backtest_rate_sensitivity --since 2019-01-01 # per-class, machine-readable: python -m scripts.backtest_rate_sensitivity --classes комфорт,бизнес --json # the #978b cross-checks: survivorship-free Source A + a detrend control: python -m scripts.backtest_rate_sensitivity --source both --detrend # evaluate the candidate methods OOS (#978 Almon-ADL + #979 deseasonalize): python -m scripts.backtest_rate_sensitivity --source B --almon --deseasonalize """ from __future__ import annotations import argparse import json import logging import math from dataclasses import dataclass from datetime import date from pathlib import Path from typing import Any import numpy as np from sqlalchemy import text from sqlalchemy.orm import Session # Reuse the §9.6 engine's PURE math (β / lag selection) and the Y-axis Δln # helper, verbatim — the backtest must score the SAME functions production runs, # not a re-implementation. Deferred import (see _import_engine) so `--help` and # the pure-logic unit tests don't pull app.core.config.Settings, which # fail-fasts when DATABASE_URL is unset. logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s", ) logger = logging.getLogger("backtest_rate_sensitivity") # ── Named constants ─────────────────────────────────────────────────────────── # Default lower bound of the backtest window. 2019-01 matches the solid daily # key_rate history (macro_indicator key_rate spans 2019-01→2026-06) and lets # Source B contribute its full ~102 distinct months. _DEFAULT_SINCE: str = "2019-01-01" # Fraction of the (time-ordered) aligned months used to FIT; the remainder is # the held-out TEST window. 0.7 ≈ "fit on the older 70%, judge on the newer # 30%" — enough train depth for best_lag's gate while leaving a test window. _HOLDOUT_FRAC: float = 0.7 # Minimum aligned (train+test) months a tier must have before we backtest it. # Below this the time-ordered split leaves too few test points for the # directional hit-rate to mean anything → skip the tier with a printed note # (never a silent drop). 18 ≈ "≥1 year to fit + a handful to test". _MIN_BACKTEST_MONTHS: int = 18 # How far the EKB-wide OOS directional hit-rate must clear the 0.5 coin-flip # baseline before the verdict calls the engine predictive. A small margin so a # tiny test window can't flip the verdict on one lucky month. _VERDICT_HITRATE_MARGIN: float = 0.05 # One-sided significance level for the verdict's binomial gate. The hit-rate must # be statistically distinguishable from a fair coin (p=0.5) at this α before the # verdict promotes — i.e. P(X ≥ hits | n_test, 0.5) < _VERDICT_ALPHA. This is a # HARD gate on top of the effect-size margin above: it prevents promoting on THIN # data where a high hit-rate is consistent with chance (the #978 near-miss: # 6/10 = 0.60 has one-sided p ≈ 0.377, indistinguishable from a coin flip). _VERDICT_ALPHA: float = 0.05 # Source B premise filter — residential квартиры, the only segment §9.6 scores # (mirrors sales_series._DEFAULT_PREMISE_KIND). _PREMISE_KIND: str = "квартира" # Sentinel for the EKB-wide (all-classes) tier in tables / JSON. _EKB_WIDE: str = "EKB-wide" # Series-source labels. # B = objective_lots.registration_date COUNT(*) — long but survivorship-CONFOUNDED. # A = objective_corpus_room_month SUM(deals) — short (~13 mo) but survivorship-FREE. _SOURCE_B: str = "B" _SOURCE_A: str = "A" _SOURCE_BOTH: str = "both" _SOURCES: tuple[str, ...] = (_SOURCE_B, _SOURCE_A) # Minimum finite (>0) points _detrend_log needs to fit a line. Below this we # can't separate trend from level, so we pass the log values through unchanged # (the difference step then behaves exactly like the raw log_diff path). _DETREND_MIN_POINTS: int = 3 # Estimator selector — which OOS evaluator backtest_tier dispatches to. # best_lag = single-winning-lag OLS (the production §9.6 core, evaluate_oos). # almon = #978 Almon polynomial distributed-lag (evaluate_oos_almon). # Default is best_lag so existing variants/tests are byte-identical (back-compat). _ESTIMATOR_BEST_LAG: str = "best_lag" _ESTIMATOR_ALMON: str = "almon" def _import_engine() -> tuple[Any, Any, Any]: """Lazy import of the §9.6 engine's pure funcs + Δln helper. Returns ``(best_lag, ols_slope_r2, log_diff)``. Deferred so ``--help`` and the pure-metric unit tests don't import app.core.config.Settings (which fail-fasts without DATABASE_URL). Supports both ``python -m scripts.backtest_rate_sensitivity`` and stand-alone runs. """ try: from app.services.forecasting.rate_sensitivity import best_lag, ols_slope_r2 from app.services.forecasting.sales_series import log_diff except ImportError: # pragma: no cover — fallback for adhoc invocation import sys sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from app.services.forecasting.rate_sensitivity import best_lag, ols_slope_r2 from app.services.forecasting.sales_series import log_diff return best_lag, ols_slope_r2, log_diff def _import_lags() -> tuple[int, ...]: """Lazy import of the engine's lag grid (_LAGS) — same deferral rationale.""" try: from app.services.forecasting.rate_sensitivity import _LAGS except ImportError: # pragma: no cover — fallback for adhoc invocation import sys sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from app.services.forecasting.rate_sensitivity import _LAGS return _LAGS def _import_regression() -> tuple[Any, int]: """Lazy import of the #978 Almon distributed-lag estimator + its lag window. Returns ``(fit_almon_dl, _MAX_LAG)``. Deferred for the SAME reason as ``_import_engine`` — app.services.forecasting.regression pulls app.core.config.Settings (fail-fasts without DATABASE_URL), and the pure-logic unit tests must import this module cheaply. ``_MAX_LAG`` (the regression module's distributed-lag window upper bound, default 6) is read so the Almon evaluator's point-in-time prediction iterates the SAME lag span the estimator fitted. """ try: from app.services.forecasting.regression import _MAX_LAG, fit_almon_dl except ImportError: # pragma: no cover — fallback for adhoc invocation import sys sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from app.services.forecasting.regression import _MAX_LAG, fit_almon_dl return fit_almon_dl, _MAX_LAG def _import_normalize() -> tuple[Any, Any]: """Lazy import of the #979 seasonal deseasonalization helpers. Returns ``(seasonal_factors, deseasonalize_values)`` from app.services.forecasting.normalize. Deferred for the SAME reason as ``_import_engine`` (the normalize module pulls SalesSeries → Settings). """ try: from app.services.forecasting.normalize import ( deseasonalize_values, seasonal_factors, ) except ImportError: # pragma: no cover — fallback for adhoc invocation import sys sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from app.services.forecasting.normalize import ( deseasonalize_values, seasonal_factors, ) return seasonal_factors, deseasonalize_values def _session() -> Session: """Lazy SessionLocal factory — see _import_engine for why it's deferred.""" try: from app.core.db import SessionLocal except ImportError: # pragma: no cover — fallback for adhoc invocation import sys sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from app.core.db import SessionLocal return SessionLocal() # --------------------------------------------------------------------------- # # Data carriers # --------------------------------------------------------------------------- # @dataclass(frozen=True) class TierResult: """OOS backtest result for one tier (EKB-wide or a single class). Every numeric field is None when the tier was skipped (too few months) — ``skipped`` carries the reason. ``oos_hit_rate`` is the only trustworthy accuracy number; ``in_sample_r2`` is high by construction (see honesty block) and reported only for contrast. """ tier: str source: str detrended: bool n_aligned: int n_train: int n_test: int train_lag: int | None train_beta: float | None in_sample_r2: float | None oos_hit_rate: float | None oos_signed_mae: float | None full_sample_lag: int | None lag_stable: bool skipped: str | None deseasonalized: bool = False estimator: str = _ESTIMATOR_BEST_LAG def as_dict(self) -> dict[str, Any]: return { "tier": self.tier, "source": self.source, "detrended": self.detrended, "deseasonalized": self.deseasonalized, "estimator": self.estimator, "n_aligned": self.n_aligned, "n_train": self.n_train, "n_test": self.n_test, "train_lag": self.train_lag, "train_beta": _round_or_none(self.train_beta, 4), "in_sample_r2": _round_or_none(self.in_sample_r2, 4), "oos_hit_rate": _round_or_none(self.oos_hit_rate, 4), "oos_signed_mae": _round_or_none(self.oos_signed_mae, 4), "full_sample_lag": self.full_sample_lag, "lag_stable": self.lag_stable, "skipped": self.skipped, } def _round_or_none(value: float | None, digits: int) -> float | None: return round(value, digits) if value is not None else None # --------------------------------------------------------------------------- # # Pure logic — NO DB. Unit-tested in tests/scripts/test_backtest_rate_sensitivity # on synthetic series (inject sales=f(rate@lag) → high OOS hit-rate; inject # noise → hit-rate ≈ 0.5; time-ordered split correctness; thin-tier skip). # --------------------------------------------------------------------------- # def _binom_sf_ge(k: int, n: int, p: float = 0.5) -> float: """Exact one-sided binomial survival: P(X ≥ k) for X ~ Binomial(n, p). PURE. Computed EXACTLY via ``math.comb`` (stdlib only — NO scipy/statsmodels, to mirror the §9.6 engine's no-heavy-deps discipline): ``Σ_{i=k..n} C(n, i) · p^i · (1−p)^(n−i)``. This is the probability the observed directional hit count (or anything more extreme) arises by chance from a fair coin — small ⇒ the hit-rate is real signal, not luck. It is the verdict's significance gate (see ``_VERDICT_ALPHA``). Guards (return 1.0 = "no evidence against the null"): • ``n <= 0`` → 1.0 (no trials, nothing to distinguish); • ``k`` clamped to ``[0, n]``; • ``k <= 0`` → 1.0 (P(X ≥ 0) = 1 trivially). """ if n <= 0: return 1.0 k = max(0, min(k, n)) if k <= 0: return 1.0 q = 1.0 - p total = 0.0 for i in range(k, n + 1): total += math.comb(n, i) * (p**i) * (q ** (n - i)) # Clamp tiny floating-point overshoot — a probability can't exceed 1.0. return min(1.0, total) def _rate_first_diff(rate_levels: list[float | None]) -> list[float | None]: """First difference of the key_rate level series: out[t] = r_t − r_{t-1}. Δ in percentage points. out[0] = None (no prior point); None if either of the two points is None. Mirrors rate_sensitivity._delta (kept local so the backtest's pure logic has no DB-importing dependency at module load). PURE. """ out: list[float | None] = [None] for i in range(1, len(rate_levels)): cur = rate_levels[i] prev = rate_levels[i - 1] if cur is None or prev is None: out.append(None) else: out.append(float(cur) - float(prev)) return out def _detrend_log( values: list[float | int | None], *, fit_n: int | None = None ) -> list[float | None]: """Linear-detrend the LOG of a units series → log-residuals. PURE (no DB). The survivorship control for #978b. We: 1. Map each unit count to ``ln(units)``; None or ≤0 → None (ln undefined, same rule as ``sales_series.log_diff``). 2. Fit ``ln(units) ~ a + b·month_index`` by least squares (numpy ``polyfit`` deg-1) over the FINITE points only, ``month_index`` = the original position 0..n-1 so gaps don't shift the trend. 3. Return the residuals ``ln(units) − (a + b·month_index)`` at each finite index (None where the input was None/≤0). Output length = input length. A spurious monotone survivorship trend lands almost entirely in ``b`` and is subtracted out, so the downstream first-difference + regression can't be driven by it. The caller differences these residuals (they are already in log space) instead of calling ``log_diff`` again. LOOK-AHEAD LEAKAGE GUARD (#978 reopen, Part A): when ``fit_n`` is given the trend ``(a, b)`` is estimated ONLY on the finite points among the first ``fit_n`` months (the TRAIN slice), then PROJECTED point-in-time onto every month (test residual = ``ln(units_t) − (a + b·t)`` with TRAIN-fitted a,b). This is mandatory in the backtest: fitting the trend on train+test together lets the held-out TEST observations shape ``(a, b)``, so the test residuals embed future information and the detrended OOS hit-rate is inflated. With ``fit_n=None`` (default) the trend is fit on the whole finite series — only safe for a non-holdout, full-sample descriptive detrend, NEVER for OOS scoring. Below ``_DETREND_MIN_POINTS`` finite points (counted within the fit window when ``fit_n`` is set) a line is not identifiable, so we PASS THROUGH the log values unchanged (residual == log value); differencing them then reproduces the raw ``log_diff`` path exactly. PURE. """ logs: list[float | None] = [] for v in values: if v is None: logs.append(None) continue vf = float(v) logs.append(math.log(vf) if vf > 0 else None) # The trend is fit only on finite points whose index is inside the fit window # (TRAIN slice [0:fit_n] when fit_n is given; the whole series otherwise). fit_upper = len(logs) if fit_n is None else max(0, fit_n) fit_idx = [i for i, lv in enumerate(logs) if lv is not None and i < fit_upper] if len(fit_idx) < _DETREND_MIN_POINTS: return logs # not enough TRAIN points to fit a trend → passthrough of logs xs = np.array([float(i) for i in fit_idx], dtype=float) ys = np.array([float(logs[i]) for i in fit_idx], dtype=float) # type: ignore[arg-type] # Degenerate x-variance (all same index — impossible for ≥3 distinct idx but # guard anyway) → no trend to remove, passthrough. if float(np.ptp(xs)) == 0.0: return logs slope, intercept = np.polyfit(xs, ys, 1) # Project the TRAIN-fitted (a, b) onto EVERY month, incl. the held-out test # months — strictly point-in-time, no test observation entered the fit. out: list[float | None] = [] for i, lv in enumerate(logs): if lv is None: out.append(None) else: out.append(float(lv) - (float(slope) * float(i) + float(intercept))) return out def _time_ordered_split(n: int, holdout_frac: float) -> int: """Index where TEST begins for a time-ordered holdout of ``n`` months. Returns ``n_train`` — the count of the OLDEST months used to FIT; months ``[n_train:]`` are the held-out TEST window. NOT random / even-odd: a time series must train on the past and test on the future. ``holdout_frac`` is the TRAIN fraction; clamped so both halves keep ≥1 month when n ≥ 2. PURE. """ if n <= 0: return 0 n_train = round(n * holdout_frac) # Keep at least one month on each side once we have ≥2 months to split. n_train = max(1, min(n_train, n - 1)) if n >= 2 else n return n_train def _shift_for_lag(x: list[float | None], lag: int) -> list[float | None]: """Right-shift the regressor by ``lag`` (y[t] ← x[t−lag]); leading = None. Same alignment best_lag uses internally: the rate change LEADS sales by ``lag`` months. Output is truncated to len(x) so it stays aligned to y. PURE. """ shifted: list[float | None] = [None] * lag + list(x) return shifted[: len(x)] def evaluate_oos( delta_sales: list[float | None], rate_deltas: list[float | None], *, holdout_frac: float = _HOLDOUT_FRAC, ) -> dict[str, Any]: """Time-ordered OOS backtest of the §9.6 β/lag fit. PURE (no DB). Steps: 1. Split the aligned months time-ordered: fit on the oldest ``holdout_frac``, test on the newest remainder. 2. Reuse ``best_lag`` on the TRAIN slice only → winning lag + β (TRAIN). best_lag applies the engine's gate (n≥min, R²≥min, slope<0); if no lag passes on TRAIN there is nothing to validate → empty result. 3. For each TEST month predict Δln(sales) = β·Δrate[t−lag], strictly point-in-time: the lagged regressor is built on the FULL aligned rate series, so a test month at index t reads Δrate at t−lag which is at or before t — never the future. Score the SIGN vs the actual Δln(sales). 4. Also refit ``best_lag`` on the FULL aligned series → full-sample lag (for the verdict's lag-stability check). Returns a dict with: n_aligned, n_train, n_test, train_lag, train_beta, in_sample_r2 (R² of the winning lag ON TRAIN — high by construction), oos_hit_rate (fraction of finite test months with matching sign; None if no scorable test month), oos_signed_mae (mean |predicted − actual| on test), full_sample_lag, lag_stable. """ best_lag, _ols_slope_r2, _log_diff = _import_engine() n = len(delta_sales) n_train = _time_ordered_split(n, holdout_frac) n_test = n - n_train empty: dict[str, Any] = { "n_aligned": n, "n_train": n_train, "n_test": n_test, "train_lag": None, "train_beta": None, "in_sample_r2": None, "oos_hit_rate": None, "oos_signed_mae": None, "full_sample_lag": None, "lag_stable": False, } if n_train < 2 or n_test < 1: return empty train_sales = delta_sales[:n_train] train_rate = rate_deltas[:n_train] train_fit = best_lag(train_sales, train_rate) if train_fit is None: # No gated lag on TRAIN → the engine has nothing to validate here. return empty lag = int(train_fit["lag"]) beta = float(train_fit["slope"]) in_sample_r2 = float(train_fit["r2"]) # Point-in-time prediction on TEST. The lagged regressor is built over the # FULL aligned series, then sliced to the test window: a test month at # absolute index t reads Δrate[t−lag] (≤ t), so no future leaks in. shifted_full = _shift_for_lag(rate_deltas, lag) hits = 0 scored = 0 abs_err_sum = 0.0 for t in range(n_train, n): actual = delta_sales[t] x = shifted_full[t] if actual is None or x is None: continue predicted = beta * float(x) scored += 1 abs_err_sum += abs(predicted - float(actual)) # Directional hit: same sign. A flat actual (0.0) can't be directionally # predicted, so it never counts as a hit (only nonzero signs match). if (predicted > 0 and actual > 0) or (predicted < 0 and actual < 0): hits += 1 oos_hit_rate = (hits / scored) if scored > 0 else None oos_signed_mae = (abs_err_sum / scored) if scored > 0 else None full_fit = best_lag(delta_sales, rate_deltas) full_lag = int(full_fit["lag"]) if full_fit is not None else None lag_stable = full_lag is not None and full_lag == lag return { "n_aligned": n, "n_train": n_train, "n_test": scored, # report the number actually SCORED, not the raw span "train_lag": lag, "train_beta": beta, "in_sample_r2": in_sample_r2, "oos_hit_rate": oos_hit_rate, "oos_signed_mae": oos_signed_mae, "full_sample_lag": full_lag, "lag_stable": lag_stable, } def evaluate_oos_almon( delta_sales: list[float | None], rate_deltas: list[float | None], *, holdout_frac: float = _HOLDOUT_FRAC, ) -> dict[str, Any]: """Time-ordered OOS backtest of the #978 Almon distributed-lag fit. PURE (no DB). Mirrors ``evaluate_oos``'s contract and return-dict shape exactly so ``backtest_tier`` can wrap it identically — but the estimator is the Almon polynomial distributed-lag model (``regression.fit_almon_dl``) instead of the single-winning-lag OLS (``rate_sensitivity.best_lag``). The Almon model fits ONE smooth coefficient curve β_0..β_K over the whole 0..max_lag window, so a TEST month's prediction sums the contributions of ALL lags, not just one. Steps: 1. Split the aligned months time-ordered (same boundary ``evaluate_oos`` uses): fit on the oldest ``holdout_frac``, test on the newest remainder. 2. ``fit_almon_dl(rate_deltas[:n_train], delta_sales[:n_train])`` on the TRAIN slice only — NB the regression module's signature is ``fit_almon_dl(x, y)`` with ``x`` the regressor (Δrate) and ``y`` the regressand (Δln(sales)), so Δrate is the FIRST arg. None (math-infeasible / too thin) → empty result (nothing to validate). The fit's ``best_lag`` (peak-|β_j| lag) is the "train_lag"; its ``long_run_coef`` (Σ_j β_j) is reported as "train_beta" (the long-run multiplier); its ``r2`` is the in-sample R². 3. For each TEST month at absolute index t, predict ``Σ_{j=0..max_lag} per_lag[j]·Δrate[t−j]`` STRICTLY point-in-time: each lagged regressor is built over the FULL aligned series via ``_shift_for_lag(rate_deltas, j)`` then read at index t, so a test month only ever reads Δrate at indices ≤ t (never the future). Score the SIGN vs the actual Δln(sales); skip a month when the actual OR any required lag value is None (don't fabricate a partial lag profile). 4. Refit ``fit_almon_dl`` on the FULL aligned series → full-sample peak lag (for the verdict's lag-stability check). Returns the SAME dict keys as ``evaluate_oos`` (n_aligned, n_train, n_test, train_lag, train_beta [=long-run Σβ here], in_sample_r2, oos_hit_rate, oos_signed_mae, full_sample_lag, lag_stable). The in-sample R² is high by construction (honesty block applies identically); only oos_hit_rate is trustworthy. """ fit_almon_dl, max_lag = _import_regression() n = len(delta_sales) n_train = _time_ordered_split(n, holdout_frac) n_test = n - n_train empty: dict[str, Any] = { "n_aligned": n, "n_train": n_train, "n_test": n_test, "train_lag": None, "train_beta": None, "in_sample_r2": None, "oos_hit_rate": None, "oos_signed_mae": None, "full_sample_lag": None, "lag_stable": False, } if n_train < 2 or n_test < 1: return empty train_sales = delta_sales[:n_train] train_rate = rate_deltas[:n_train] # fit_almon_dl(x, y): x = Δrate (regressor) FIRST, y = Δln(sales) (regressand). fit = fit_almon_dl(train_rate, train_sales, max_lag=max_lag) if fit is None: # Almon fit math-infeasible on TRAIN → nothing to validate here. return empty per_lag = fit["per_lag_coef"] # tuple β_0..β_max_lag from the Almon form train_lag = int(fit["best_lag"]) train_beta = float(fit["long_run_coef"]) # long-run Σβ reported as the "beta" field in_sample_r2 = float(fit["r2"]) if fit["r2"] is not None else None # Build one point-in-time-safe shifted view per lag over the FULL aligned # series. shifted_by_lag[j][t] == rate_deltas[t-j] (None where t-j < 0), so a # test month at index t reads only Δrate at indices ≤ t — never the future. shifted_by_lag = [_shift_for_lag(rate_deltas, j) for j in range(max_lag + 1)] hits = 0 scored = 0 abs_err_sum = 0.0 for t in range(n_train, n): actual = delta_sales[t] if actual is None: continue # The Almon prediction needs the FULL lag profile at t; a single missing # lag means we can't form Σ per_lag[j]·Δrate[t-j] → skip (no fabrication). lag_vals: list[float] = [] complete = True for j in range(max_lag + 1): xv = shifted_by_lag[j][t] if xv is None: complete = False break lag_vals.append(float(xv)) if not complete: continue predicted = sum(per_lag[j] * lag_vals[j] for j in range(max_lag + 1)) scored += 1 abs_err_sum += abs(predicted - float(actual)) # Directional hit: same nonzero sign (a flat 0.0 actual can't be # directionally predicted, so it never counts as a hit). if (predicted > 0 and actual > 0) or (predicted < 0 and actual < 0): hits += 1 oos_hit_rate = (hits / scored) if scored > 0 else None oos_signed_mae = (abs_err_sum / scored) if scored > 0 else None # Full-sample refit (x=Δrate first, y=Δln(sales)) for the lag-stability check. full_fit = fit_almon_dl(rate_deltas, delta_sales, max_lag=max_lag) full_lag = int(full_fit["best_lag"]) if full_fit is not None else None lag_stable = full_lag is not None and full_lag == train_lag return { "n_aligned": n, "n_train": n_train, "n_test": scored, # report the number actually SCORED, not the raw span "train_lag": train_lag, "train_beta": train_beta, "in_sample_r2": in_sample_r2, "oos_hit_rate": oos_hit_rate, "oos_signed_mae": oos_signed_mae, "full_sample_lag": full_lag, "lag_stable": lag_stable, } def align_series( sales_by_month: dict[date, int], rate_by_month: dict[date, float], ) -> tuple[list[date], list[int], list[float]]: """Inner-join the sold-units and key_rate monthly series by year-month. PURE. Returns ``(months, units, rates)`` ASC over the months present in BOTH series (a month with no rate or no sales row is dropped so Δ pairs stay meaningful). units/rates are aligned to months by index. """ common = sorted(set(sales_by_month) & set(rate_by_month)) months = list(common) units = [int(sales_by_month[m]) for m in common] rates = [float(rate_by_month[m]) for m in common] return months, units, rates def _delta_sales_series( units: list[int], *, detrend: bool, fit_n: int | None = None ) -> list[float | None]: """Build the Δ(log-units) regressand for one tier. PURE (deferred import). Two variants, both ending in a Δ of log-space values ``evaluate_oos`` scores: • ``detrend=False`` — the production path: ``log_diff(units)`` = first difference of ``ln(units)``. • ``detrend=True`` — the #978b control: first linear-detrend ``ln(units)`` (``_detrend_log``), THEN first-difference the residuals. We difference the residuals DIRECTLY (they are already in log space) rather than ``log_diff`` (which would re-take logs of residuals that may be ≤0). ``fit_n`` (the TRAIN month count) is forwarded to ``_detrend_log`` so the detrend trend is estimated on TRAIN months only and projected point-in-time onto the test months — the #978 Part A look-ahead-leakage fix. It is ignored on the non-detrend path (``log_diff`` is already point-in-time: a first difference reads only t and t−1). """ if not detrend: _bl, _ols, log_diff = _import_engine() return log_diff(units) return _rate_first_diff(_detrend_log(units, fit_n=fit_n)) def _deseasonalize_units(months: list[date], units: list[int], *, fit_n: int) -> list[float | None]: """Deseasonalize a units series → Δln(deseasonalized units). PURE (deferred import). The #979 control. Mirrors ``_delta_sales_series``'s train-only discipline: 1. Fit month-of-year seasonal factors on the TRAIN months ONLY (``seasonal_factors(months[:fit_n], units[:fit_n])``). A test-month observation must NEVER influence the factors (look-ahead leakage rule). 2. Deseasonalize the FULL series with the TRAIN-fit factors (``deseasonalize_values(months, units, factors)`` → raw_t / factor[m]). 3. ``log_diff`` the deseasonalized units → Δln regressand (the same Y-axis ``evaluate_oos`` scores). ``log_diff`` itself is point-in-time (a first difference reads only t and t−1), so the ONLY leakage risk is the factor fit — pinned to the TRAIN slice here. The deseasonalized values are floats (raw / factor); ``log_diff`` is float-math throughout, so no int-rounding is applied (unlike the prod ``normalize_demand`` wrapper, which rounds only to honour SalesSeries.units's int contract — irrelevant for the regressand). """ seasonal_factors, deseasonalize_values = _import_normalize() _bl, _ols, log_diff = _import_engine() train_months = months[:fit_n] train_units = units[:fit_n] adjustment = seasonal_factors(train_months, train_units) deseasonalized = deseasonalize_values(months, units, adjustment.factors) return log_diff(deseasonalized) def backtest_tier( sales_by_month: dict[date, int], rate_by_month: dict[date, float], *, tier: str, source: str = _SOURCE_B, detrend: bool = False, deseasonalize: bool = False, estimator: str = _ESTIMATOR_BEST_LAG, holdout_frac: float = _HOLDOUT_FRAC, min_months: int = _MIN_BACKTEST_MONTHS, ) -> TierResult: """Build Δ-series for one tier, run the OOS backtest, wrap as TierResult. Aligns the tier's monthly sold-units to the monthly key_rate, computes the regressand and Δrate (first diff), then delegates to the OOS evaluator named by ``estimator`` (``evaluate_oos`` for ``best_lag``, ``evaluate_oos_almon`` for ``almon``). The regressand is one of three PREPROCESSING variants, all fit point-in-time on TRAIN months only (no look-ahead leakage): • plain ``log_diff(units)`` (default), • Δ of linear-detrended ``ln(units)`` when ``detrend`` (#978b control), or • ``log_diff`` of month-of-year deseasonalized units when ``deseasonalize`` (#979 control — see ``_deseasonalize_units``). ``detrend`` and ``deseasonalize`` are mutually exclusive preprocessing paths; if both are set ``deseasonalize`` takes precedence (the planner never emits that combination). ``source``/``detrend``/``deseasonalize``/``estimator`` are recorded on the result for labelling. Tiers with fewer than ``min_months`` aligned months are SKIPPED (TierResult with ``skipped`` set, all metrics None) — no silent drop. PURE aside from the deferred engine import. """ months, units, rates = align_series(sales_by_month, rate_by_month) n_aligned = len(months) if n_aligned < min_months: return TierResult( tier=tier, source=source, detrended=detrend, deseasonalized=deseasonalize, estimator=estimator, n_aligned=n_aligned, n_train=0, n_test=0, train_lag=None, train_beta=None, in_sample_r2=None, oos_hit_rate=None, oos_signed_mae=None, full_sample_lag=None, lag_stable=False, skipped=f"only {n_aligned} aligned months (< {min_months})", ) # All preprocessing that ESTIMATES anything (the detrend trend, the seasonal # factors) must be fit on TRAIN months ONLY, then projected point-in-time # onto the test months — otherwise the held-out TEST data shapes the # preprocessing and the OOS hit-rate is inflated by look-ahead leakage (#978 # Part A for detrend; the #979 leakage rule for deseasonalize). We compute the # SAME train boundary the evaluator will use (len(delta_sales) == len(units) # == n_aligned, so the split index matches) and pass it as the fit window. n_train = _time_ordered_split(n_aligned, holdout_frac) if deseasonalize: delta_sales = _deseasonalize_units(months, units, fit_n=n_train) else: delta_sales = _delta_sales_series(units, detrend=detrend, fit_n=n_train) rate_deltas = _rate_first_diff([float(r) for r in rates]) if estimator == _ESTIMATOR_ALMON: res = evaluate_oos_almon(delta_sales, rate_deltas, holdout_frac=holdout_frac) else: res = evaluate_oos(delta_sales, rate_deltas, holdout_frac=holdout_frac) return TierResult( tier=tier, source=source, detrended=detrend, deseasonalized=deseasonalize, estimator=estimator, n_aligned=res["n_aligned"], n_train=res["n_train"], n_test=res["n_test"], train_lag=res["train_lag"], train_beta=res["train_beta"], in_sample_r2=res["in_sample_r2"], oos_hit_rate=res["oos_hit_rate"], oos_signed_mae=res["oos_signed_mae"], full_sample_lag=res["full_sample_lag"], lag_stable=res["lag_stable"], skipped=None, ) def verdict( ekb: TierResult, *, margin: float = _VERDICT_HITRATE_MARGIN, ) -> dict[str, Any]: """Decide whether the EKB-wide tier shows OOS predictive value. PURE. The engine is a promotion CANDIDATE when, on the EKB-wide tier, ALL hold: • a gated lag was found and scored on a non-empty test window, AND • the OOS directional hit-rate beats the 0.5 coin-flip baseline by at least ``margin`` (minimum EFFECT SIZE), AND • the hit-rate is STATISTICALLY SIGNIFICANT vs a fair coin: the exact one-sided binomial ``P(X ≥ hits | n_test, 0.5) < _VERDICT_ALPHA`` (a HARD gate — a high rate on a tiny window is consistent with chance and must NOT promote; this is the #978 near-miss: 6/10 = 0.60 has p ≈ 0.377), AND • the winning lag is the same on TRAIN and on the full-sample refit (lag stability — a lag that jumps between windows is not a signal). ``hits`` (the integer directional-hit count the binomial needs) is recovered as ``round(oos_hit_rate * n_test)``: both the hit-rate and n_test derive from the SAME integer division in ``evaluate_oos`` (``hits / scored`` with ``n_test == scored``), so this round-trips EXACTLY without threading a new field through the deep-reviewed evaluator return shape. Returns ``{"promote": bool, "reason": str, "thin_warning": str | None}``. Honest: if the OOS test window is tiny the reason says so, and significance is now a HARD gate, not just an advisory caveat. """ if ekb.skipped is not None: return { "promote": False, "reason": f"insufficient OOS signal — keep advisory ({ekb.skipped})", "thin_warning": None, } if ekb.oos_hit_rate is None or ekb.n_test < 1: return { "promote": False, "reason": ( "insufficient OOS signal — keep advisory " "(no gated lag on TRAIN or empty test window)" ), "thin_warning": None, } beats_coin = ekb.oos_hit_rate >= 0.5 + margin # Recover the integer hit count exactly (see docstring) and test significance. hits = round(ekb.oos_hit_rate * ekb.n_test) p_value = _binom_sf_ge(hits, ekb.n_test, 0.5) significant = p_value < _VERDICT_ALPHA thin_warning: str | None = None if ekb.n_test < min(_MIN_BACKTEST_MONTHS // 2, 6): thin_warning = ( f"OOS test window is small (n_test={ekb.n_test}); the hit-rate's " "confidence is weak — treat the verdict as indicative, not proof." ) if beats_coin and ekb.lag_stable and significant: reason = ( f"engine has OOS predictive value (candidate to promote from " f"advisory): EKB-wide OOS hit-rate={ekb.oos_hit_rate:.2f} over " f"n_test={ekb.n_test} > 0.5+{margin:.2f}, lag stable " f"(lag={ekb.train_lag}), and significant " f"(one-sided binomial p={p_value:.3f} < {_VERDICT_ALPHA:.2f})" ) return {"promote": True, "reason": reason, "thin_warning": thin_warning} bits: list[str] = [] if not beats_coin: bits.append(f"hit-rate={ekb.oos_hit_rate:.2f} ≤ 0.5+{margin:.2f}") if not ekb.lag_stable: bits.append(f"lag unstable (train={ekb.train_lag}, full={ekb.full_sample_lag})") # Significance reported whenever the effect size cleared the bar but the # window is too thin to rule out chance — the #978 transparency requirement. if beats_coin and not significant: bits.append( f"hit-rate={ekb.oos_hit_rate:.2f} over n_test={ekb.n_test} is not " f"significant (one-sided binomial p={p_value:.2f} ≥ {_VERDICT_ALPHA:.2f})" ) reason = "insufficient OOS signal — keep advisory (" + "; ".join(bits) + ")" return {"promote": False, "reason": reason, "thin_warning": thin_warning} def tier_lift(ekb: TierResult, tier: TierResult) -> float | None: """OOS directional hit-rate lift of a class tier over EKB-wide. PURE. Returns ``tier.oos_hit_rate − ekb.oos_hit_rate`` (positive = the class-specific β beats the EKB-wide prior OOS). None if either side has no scorable OOS hit-rate. """ if ekb.oos_hit_rate is None or tier.oos_hit_rate is None: return None return tier.oos_hit_rate - ekb.oos_hit_rate # --------------------------------------------------------------------------- # # DB layer — READ-ONLY SELECTs only. # --------------------------------------------------------------------------- # # Source B monthly sold-units by registration month (the one solid long series). # COUNT(*) of lots whose deal registered in each month, EKB-wide or filtered to # one class / district. Parameterised; psycopg3 CAST(:x AS type), NEVER :x::type. # Survivorship caveat (only currently-listed lots) is documented in the module # docstring — it biases OLDER months low. _SOURCE_B_UNITS_SQL = text( """ SELECT CAST(date_trunc('month', ol.registration_date) AS date) AS month, COUNT(*) AS units FROM objective_lots ol WHERE ol.premise_kind = CAST(:premise_kind AS text) AND ol.registration_date IS NOT NULL AND ol.registration_date >= CAST(:since AS date) AND (CAST(:cls AS text) IS NULL OR LOWER(ol.class) = LOWER(CAST(:cls AS text))) AND (CAST(:district AS text) IS NULL OR ol.district = CAST(:district AS text)) GROUP BY 1 ORDER BY 1 """ ) # Source A monthly deal AGGREGATE — survivorship-FREE. Objective's corp_sum API # reports deals registered per month per (corpus × room_bucket) regardless of # whether the lot is still listed, so it does NOT undercount old months the way # Source B does. room_bucket is aggregated away by the SUM/GROUP BY 1 unless a # class filter is given (class is stored capitalised here — "Комфорт" — so we # fold case to match the lowercase --classes input). report_month is already a # month-first DATE; date_trunc is belt-and-braces. Only ~13 months deep on prod. # Parameterised; psycopg3 CAST(:x AS type), NEVER :x::type. _SOURCE_A_UNITS_SQL = text( """ SELECT CAST(date_trunc('month', crm.report_month) AS date) AS month, SUM(crm.deals_total_count) AS units FROM objective_corpus_room_month crm WHERE crm.report_month >= CAST(:since AS date) AND (CAST(:cls AS text) IS NULL OR LOWER(crm.class) = LOWER(CAST(:cls AS text))) GROUP BY 1 ORDER BY 1 """ ) # Distinct classes present in Source A over the window (for --classes all on A). _SOURCE_A_CLASSES_SQL = text( """ SELECT DISTINCT LOWER(crm.class) AS cls FROM objective_corpus_room_month crm WHERE crm.report_month >= CAST(:since AS date) AND crm.class IS NOT NULL ORDER BY 1 """ ) # Distinct classes present in Source B over the window (for --classes all). _SOURCE_B_CLASSES_SQL = text( """ SELECT DISTINCT LOWER(ol.class) AS cls FROM objective_lots ol WHERE ol.premise_kind = CAST(:premise_kind AS text) AND ol.registration_date IS NOT NULL AND ol.registration_date >= CAST(:since AS date) AND ol.class IS NOT NULL ORDER BY 1 """ ) def load_sales_by_month( db: Session, *, since: str, obj_class: str | None, district: str | None, ) -> dict[date, int]: """Run the Source B monthly sold-units SELECT → {month1st: units}. READ-ONLY. ``obj_class``/``district`` None → no filter (EKB-wide). Months with no rows simply do not appear (the caller aligns on the intersection with the rate series, so absent months drop out naturally). """ rows = db.execute( _SOURCE_B_UNITS_SQL, { "premise_kind": _PREMISE_KIND, "since": since, "cls": obj_class, "district": district, }, ).all() out: dict[date, int] = {} for r in rows: if r[0] is None: continue out[r[0]] = int(r[1] or 0) return out def load_classes(db: Session, *, since: str) -> list[str]: """Run the Source B distinct-classes SELECT → lowercase class list. READ-ONLY.""" rows = db.execute( _SOURCE_B_CLASSES_SQL, {"premise_kind": _PREMISE_KIND, "since": since}, ).all() return [r[0] for r in rows if r[0] is not None] def load_sales_by_month_source_a( db: Session, *, since: str, obj_class: str | None, ) -> dict[date, int]: """Run the Source A monthly deal-aggregate SELECT → {month1st: units}. READ-ONLY. ``obj_class`` None → no class filter (room_bucket aggregated away). Survivorship-FREE (deals counted regardless of current listing). Months with no rows simply do not appear. No district filter — corp_sum aggregates are not district-resolved the way the lots snapshot is. """ rows = db.execute( _SOURCE_A_UNITS_SQL, {"since": since, "cls": obj_class}, ).all() out: dict[date, int] = {} for r in rows: if r[0] is None: continue out[r[0]] = int(r[1] or 0) return out def load_classes_source_a(db: Session, *, since: str) -> list[str]: """Run the Source A distinct-classes SELECT → lowercase class list. READ-ONLY.""" rows = db.execute( _SOURCE_A_CLASSES_SQL, {"since": since}, ).all() return [r[0] for r in rows if r[0] is not None] def load_rate_by_month(db: Session, *, since: str) -> dict[date, float]: """Monthly last-known key_rate → {month1st: rate}. READ-ONLY. Reuses ``forecasting.macro_series.get_monthly_macro`` (DISTINCT ON month-end last-known + LOCF carry-forward) rather than re-deriving the resample — the backtest must read the SAME monthly key_rate the engine does. ``months_back`` is computed from ``since`` so the macro grid covers the whole backtest window. Months with a None carried rate are dropped. """ from app.services.forecasting.macro_series import ( _month_start, get_monthly_macro, ) since_date = date.fromisoformat(since) today = date.today() months_back = (today.year - since_date.year) * 12 + (today.month - since_date.month) macro = get_monthly_macro(db, months_back=max(0, months_back)) floor = _month_start(since_date) out: dict[date, float] = {} for m in macro: if m.key_rate is None or m.month < floor: continue out[m.month] = float(m.key_rate) return out # --------------------------------------------------------------------------- # # Orchestration (READ-ONLY) + rendering # --------------------------------------------------------------------------- # def _load_sales( db: Session, *, source: str, since: str, obj_class: str | None, district: str | None, ) -> dict[date, int]: """Dispatch the monthly sold-units load to the right source. READ-ONLY. Source B uses ``objective_lots`` (premise+district filters). Source A uses ``objective_corpus_room_month`` (survivorship-free aggregate; no district filter — corp_sum aggregates are not district-resolved, so ``district`` is ignored for A and the caller is responsible for warning if it was set). """ if source == _SOURCE_A: return load_sales_by_month_source_a(db, since=since, obj_class=obj_class) return load_sales_by_month(db, since=since, obj_class=obj_class, district=district) def _load_classes_for(db: Session, *, source: str, since: str) -> list[str]: """Dispatch class auto-discovery to the right source. READ-ONLY.""" if source == _SOURCE_A: return load_classes_source_a(db, since=since) return load_classes(db, since=since) def run_backtest( db: Session, *, since: str, holdout_frac: float, classes: list[str] | None, district: str | None, source: str = _SOURCE_B, detrend: bool = False, deseasonalize: bool = False, estimator: str = _ESTIMATOR_BEST_LAG, rate_by_month: dict[date, float] | None = None, ) -> dict[str, Any]: """Drive ONE source/variant of the read-only backtest → results dict. No writes. Loads the monthly key_rate (or reuses ``rate_by_month`` when the caller has already loaded it once for several variants), then the EKB-wide and per-class sold-units series for ``source``, backtests each tier (``backtest_tier``, with the ``detrend`` / ``deseasonalize`` preprocessing and the ``estimator`` applied), and assembles the per-source verdict + per-tier OOS lifts. ``classes`` None → auto-discover every class present in the chosen source; an empty list → EKB-wide only. ``district`` narrows ALL tiers for Source B only (ignored for Source A). ``deseasonalize`` (#979) and ``estimator`` ("almon" selects the #978 Almon distributed-lag evaluator) are recorded on the run dict + params alongside ``detrend`` so labels/JSON carry the full variant. """ if rate_by_month is None: rate_by_month = load_rate_by_month(db, since=since) logger.info("loaded key_rate months: %d (since=%s)", len(rate_by_month), since) if classes is None: classes = _load_classes_for(db, source=source, since=since) logger.info("source=%s auto-discovered classes: %s", source, classes) a_district_ignored = source == _SOURCE_A and district is not None eff_district = None if source == _SOURCE_A else district # EKB-wide tier (no class filter). ekb_sales = _load_sales(db, source=source, since=since, obj_class=None, district=eff_district) ekb = backtest_tier( ekb_sales, rate_by_month, tier=_EKB_WIDE, source=source, detrend=detrend, deseasonalize=deseasonalize, estimator=estimator, holdout_frac=holdout_frac, ) logger.info( "source=%s detrend=%s deseasonalize=%s estimator=%s EKB-wide: " "aligned=%d train=%d test=%d lag=%s hit_rate=%s", source, detrend, deseasonalize, estimator, ekb.n_aligned, ekb.n_train, ekb.n_test, ekb.train_lag, ekb.oos_hit_rate, ) tiers: list[TierResult] = [] lifts: dict[str, float | None] = {} for cls in classes: cls_sales = _load_sales( db, source=source, since=since, obj_class=cls, district=eff_district ) res = backtest_tier( cls_sales, rate_by_month, tier=cls, source=source, detrend=detrend, deseasonalize=deseasonalize, estimator=estimator, holdout_frac=holdout_frac, ) tiers.append(res) lifts[cls] = tier_lift(ekb, res) logger.info( "source=%s tier=%s aligned=%d test=%d hit_rate=%s lift=%s skipped=%s", source, cls, res.n_aligned, res.n_test, res.oos_hit_rate, lifts[cls], res.skipped, ) vd = verdict(ekb) return { "source": source, "detrended": detrend, "deseasonalized": deseasonalize, "estimator": estimator, "a_district_ignored": a_district_ignored, "params": { "since": since, "holdout_frac": holdout_frac, "district": district, "classes": classes, "source": source, "detrended": detrend, "deseasonalized": deseasonalize, "estimator": estimator, "min_backtest_months": _MIN_BACKTEST_MONTHS, "lags": list(_import_lags()), }, "ekb_wide": ekb.as_dict(), "tiers": [t.as_dict() for t in tiers], "lifts": {k: _round_or_none(v, 4) for k, v in lifts.items()}, "verdict": vd, "ekb_result": ekb, # carried for the renderer (stripped from JSON) "tier_results": tiers, } def _variant_label( source: str, detrend: bool, *, deseasonalize: bool = False, estimator: str = _ESTIMATOR_BEST_LAG, ) -> str: """Human label for a variant run. PURE. The method descriptor is mutually-exclusive (the planner never combines them), so the label names the ONE active method: • ``estimator == 'almon'`` → ``f"{source} Almon-ADL"`` (the #978 estimator), • ``deseasonalize`` → ``f"{source} deseasonalized"`` (#979 preprocessing), • ``detrend`` → ``f"{source} detrended"`` (#978b control), • otherwise (best_lag, raw) → ``f"{source} raw"`` (the always-on reference). """ if estimator == _ESTIMATOR_ALMON: return f"{source} Almon-ADL" if deseasonalize: return f"{source} deseasonalized" if detrend: return f"{source} detrended" return f"{source} raw" def _variant_label_for_run(run: dict[str, Any]) -> str: """Label a run dict via ``_variant_label`` (reads source/detrend/deseason/estim).""" return _variant_label( run["source"], run["detrended"], deseasonalize=run.get("deseasonalized", False), estimator=run.get("estimator", _ESTIMATOR_BEST_LAG), ) def _plan_variants( sources: list[str], detrend: bool, *, deseasonalize: bool = False, almon: bool = False, ) -> list[tuple[str, bool, bool, str]]: """Which variants to run, in report order. PURE. Each entry is ``(source, detrend, deseasonalize, estimator)``. For each requested source we ALWAYS run the RAW reference (best_lag on raw units), then ADD only the explicitly-requested method variants — we do NOT explode into all combinations: • ``--detrend`` → ``(src, True, False, best_lag)`` (#978b survivorship control), • ``--deseasonalize``→ ``(src, False, True, best_lag)`` (#979 seasonal preprocessing), • ``--almon`` → ``(src, False, False, almon)`` (#978 Almon distributed-lag). The method flags are independent: passing several adds several variants per source (raw + each requested method), each isolating ONE method against the raw reference for the verdict's side-by-side comparison. """ variants: list[tuple[str, bool, bool, str]] = [] for src in sources: variants.append((src, False, False, _ESTIMATOR_BEST_LAG)) if detrend: variants.append((src, True, False, _ESTIMATOR_BEST_LAG)) if deseasonalize: variants.append((src, False, True, _ESTIMATOR_BEST_LAG)) if almon: variants.append((src, False, False, _ESTIMATOR_ALMON)) return variants def cross_source_verdict( runs: list[dict[str, Any]], *, margin: float = _VERDICT_HITRATE_MARGIN, min_months: int = _MIN_BACKTEST_MONTHS, ) -> dict[str, Any]: """Compare the EKB-wide OOS verdict across every method variant. PURE. The §9.6 question, generalised across the variants present: does ANY method/source recover OOS directional signal the raw best_lag baseline missed? Variants can be the survivorship controls (B detrended, A) AND the two new candidate methods — ``deseasonalized`` (#979 month-of-year seasonal preprocessing) and ``Almon-ADL`` (#978 polynomial distributed-lag estimator). We line up each variant's EKB-wide OOS hit-rate vs the 0.5 coin-flip baseline (decision rule UNCHANGED: a variant "beats" only if scorable AND ``oos_hit_rate >= 0.5 + margin`` AND ``lag_stable``) and synthesise: • If NO variant clears coin-flip+margin → the negative verdict is corroborated as a real ``no signal``, not an artifact and not something the new methods rescue — neither deseasonalizing the input nor the smoother Almon estimator extracts signal that isn't there. Source A's thin-data caveat is attached when an A row drove the comparison. • If a control OR a candidate method DOES clear the bar while raw best_lag did not → flag that variant: the raw verdict may be a survivorship artifact, or the method recovered signal worth promoting from advisory. PURE — operates on already-computed run dicts. Returns a dict with a ``lines`` list (rendered as-is) plus structured fields for JSON. """ rows: list[dict[str, Any]] = [] signal_variants: list[str] = [] thin_variants: list[str] = [] for run in runs: ekb: TierResult = run["ekb_result"] label = _variant_label_for_run(run) hr = ekb.oos_hit_rate scorable = ekb.skipped is None and hr is not None and ekb.n_test >= 1 beats_margin = bool(scorable and hr is not None and hr >= 0.5 + margin) # Significance gate (mirrors verdict()): recover the integer hit count # exactly from the rate × window (same integer division in evaluate_oos) # and require the one-sided binomial p < _VERDICT_ALPHA. A high hit-rate # on a thin window is NOT a signal — the #978 6/10 near-miss. hits = round(hr * ekb.n_test) if scorable and hr is not None else 0 p_value = _binom_sf_ge(hits, ekb.n_test, 0.5) if scorable else 1.0 significant = p_value < _VERDICT_ALPHA beats = bool(beats_margin and ekb.lag_stable and significant) thin = scorable and ekb.n_test < min(min_months // 2, 6) if beats: signal_variants.append(label) if run["source"] == _SOURCE_A and (thin or not scorable): thin_variants.append(label) rows.append( { "variant": label, "source": run["source"], "detrended": run["detrended"], "deseasonalized": run.get("deseasonalized", False), "estimator": run.get("estimator", _ESTIMATOR_BEST_LAG), "scorable": scorable, "oos_hit_rate": _round_or_none(hr, 4), "n_test": ekb.n_test, "lag_stable": ekb.lag_stable, "binom_p": _round_or_none(p_value, 4) if scorable else None, "significant": significant, "beats_coin": beats, "skipped": ekb.skipped, } ) # Width the label column to the widest variant label present (Almon-ADL / # deseasonalized are longer than the original "B detrended"). label_w = max((len(r["variant"]) for r in rows), default=13) label_w = max(label_w, 13) lines: list[str] = [] lines.append( "CROSS-SOURCE / CROSS-METHOD VERDICT (raw best_lag vs controls " "[detrended, A] vs candidate methods [deseasonalized #979, Almon-ADL #978]):" ) for r in rows: if not r["scorable"]: why = r["skipped"] or "no gated lag / empty test window" lines.append(f" {r['variant']:<{label_w}} → not scorable ({why})") else: # Spell out WHY a variant does/doesn't count as signal — including the # binomial p so a "high rate but thin" row is transparent (#978). if r["beats_coin"]: tag = ( f"SIGNAL > coin-flip (binomial p={_fmt_rate(r['binom_p'])} " f"< {_VERDICT_ALPHA:.2f})" ) elif not r["significant"]: tag = ( f"no signal (hit-rate not significant: binomial " f"p={_fmt_rate(r['binom_p'])} ≥ {_VERDICT_ALPHA:.2f})" ) else: tag = "no signal (≤ coin-flip / lag unstable)" lines.append( f" {r['variant']:<{label_w}} → OOS_hit={_fmt_rate(r['oos_hit_rate'])} " f"(n_test={r['n_test']}, lag_stable={'yes' if r['lag_stable'] else 'no'}) " f"→ {tag}" ) if signal_variants: conclusion = ( "CONCLUSION: OOS signal above coin-flip appears in: " + ", ".join(signal_variants) + ". The §9.6 negative verdict on raw best_lag may be a SURVIVORSHIP " "ARTIFACT, or a candidate method (deseasonalize / Almon-ADL) recovers " "directional signal worth promoting from advisory — inspect the flagged " "variant(s) above." ) promote_any = True else: conclusion = ( "CONCLUSION: NO variant — neither the survivorship controls (detrended B, " "survivorship-free A) NOR the candidate methods (#979 deseasonalize, #978 " "Almon-ADL) — beats coin-flip+margin out-of-sample. The §9.6 negative " "verdict is a REAL 'no signal', NOT a survivorship artifact, and the new " "methods do not recover signal that isn't there. Keep advisory." ) promote_any = False lines.append(" " + conclusion) thin_caveat: str | None = None if thin_variants: thin_caveat = ( "Source A is statistically THIN (~13 months on prod). Treat any A row " "as an indicative cross-check only, never as proof — variant(s): " + ", ".join(thin_variants) + "." ) lines.append(f" !! {thin_caveat}") return { "rows": rows, "signal_variants": signal_variants, "promote_any": promote_any, "conclusion": conclusion, "thin_caveat": thin_caveat, "lines": lines, } def run_all( db: Session, *, since: str, holdout_frac: float, classes: list[str] | None, district: str | None, sources: list[str], detrend: bool, deseasonalize: bool = False, almon: bool = False, ) -> dict[str, Any]: """Run every requested variant + the cross-source/method verdict. No writes. Loads the monthly key_rate ONCE and reuses it across variants. ``sources`` is a subset of (B, A); the always-on RAW reference runs per source, plus one variant per requested method: ``detrend`` (#978b control), ``deseasonalize`` (#979 preprocessing), ``almon`` (#978 distributed-lag estimator). Returns ``{"variants": [run, ...], "cross_verdict": {...}}``. """ rate_by_month = load_rate_by_month(db, since=since) logger.info("loaded key_rate months: %d (since=%s)", len(rate_by_month), since) variants = _plan_variants(sources, detrend, deseasonalize=deseasonalize, almon=almon) runs: list[dict[str, Any]] = [] for src, dt_flag, deseason_flag, estimator in variants: runs.append( run_backtest( db, since=since, holdout_frac=holdout_frac, classes=classes, district=district, source=src, detrend=dt_flag, deseasonalize=deseason_flag, estimator=estimator, rate_by_month=rate_by_month, ) ) cross = cross_source_verdict(runs) return {"variants": runs, "cross_verdict": cross} def _fmt_rate(v: float | None) -> str: return " n/a" if v is None else f"{v:.3f}" def _fmt_lag(v: int | None) -> str: return "n/a" if v is None else str(v) _SOURCE_BLURB: dict[str, str] = { _SOURCE_B: "Source B (objective_lots.registration_date COUNT) — survivorship-CONFOUNDED.", _SOURCE_A: "Source A (objective_corpus_room_month SUM deals) — survivorship-FREE, ~13 mo.", } def render_table(results: dict[str, Any]) -> str: """Render ONE variant's backtest results as a plain-text stdout report.""" params = results["params"] ekb: TierResult = results["ekb_result"] tiers: list[TierResult] = results["tier_results"] lifts: dict[str, Any] = results["lifts"] vd = results["verdict"] source = results["source"] detrended = results["detrended"] deseasonalized = results.get("deseasonalized", False) estimator = results.get("estimator", _ESTIMATOR_BEST_LAG) # Short method tag for the title line (one active method per variant). if estimator == _ESTIMATOR_ALMON: method_tag = " · Almon-ADL" elif deseasonalized: method_tag = " · deseasonalized" elif detrended: method_tag = " · detrended" else: method_tag = "" lines: list[str] = [] lines.append("=" * 78) lines.append(f"BACKTEST [source {source}{method_tag}]: §9.6 rate-sensitivity OOS validation") lines.append("=" * 78) lines.append( f"since={params['since']} holdout_frac={params['holdout_frac']} " f"district={params['district'] or '(all)'} lags={params['lags']}" ) lines.append(_SOURCE_BLURB.get(source, source)) if detrended: lines.append( "DETRENDED: ln(units) linearly detrended (residuals) BEFORE differencing — " "removes a spurious monotone (survivorship) trend so it can't drive β. " "Trend fit on TRAIN months only, projected point-in-time onto test (no leakage)." ) if deseasonalized: lines.append( "DESEASONALIZED (#979): month-of-year seasonal factors divided out of units " "BEFORE log_diff — strips the calendar pattern so a seasonal peak isn't read as " "a rate effect. Factors fit on TRAIN months only, applied point-in-time (no leakage)." ) if estimator == _ESTIMATOR_ALMON: lines.append( "ALMON-ADL (#978): Almon polynomial distributed-lag estimator (smooth β over lags " "0..6) replaces single-lag best_lag. 'beta' column = LONG-RUN Σβ multiplier; 'lag' " "= peak-|β_j| lag. Fit on TRAIN months only; test prediction sums all lags " "point-in-time (no leakage)." ) if results.get("a_district_ignored"): lines.append( "NOTE: --district was IGNORED for Source A (corp_sum aggregates are not " "district-resolved)." ) lines.append("") header = ( f" {'tier':<12} {'src':>3} {'detr':>5} {'aligned':>7} {'train':>6} {'test':>5} " f"{'lag':>4} {'beta':>9} {'inR2':>7} {'OOS_hit':>8} {'OOS_MAE':>8} {'lift':>7} " f"{'stable':>7}" ) lines.append(header) lines.append(" " + "-" * (len(header) - 2)) lines.append(_fmt_tier_row(ekb, lift=None)) for t in tiers: lines.append(_fmt_tier_row(t, lift=lifts.get(t.tier))) # Skips called out explicitly (no silent drop). skipped = [t for t in tiers if t.skipped is not None] if skipped: lines.append("") lines.append("SKIPPED tiers (too few aligned months):") for t in skipped: lines.append(f" {t.tier:<12} {t.skipped}") # In-sample-vs-OOS honesty block (copy trade-in discipline). lines.append("") lines.append("HONESTY — in-sample vs out-of-sample:") lines.append(" inR2 is the TRAIN-window R² — high BY CONSTRUCTION (β is fit to minimise those") lines.append( " residuals). It is NOT evidence the engine predicts. The only trustworthy number" ) lines.append( " is OOS_hit: the directional hit-rate on held-out future months the fit never saw," ) lines.append( " scored strictly point-in-time. A coin flip scores ~0.50; the engine must beat it." ) # Per-tier interpretation. lines.append("") lines.append("PER-TIER — does a class-specific β beat the EKB-wide prior OOS?") any_lift = False for t in tiers: lift = lifts.get(t.tier) if lift is None: continue any_lift = True verdict_word = "beats EKB-wide" if lift > 0 else "no lift over EKB-wide" lines.append( f" {t.tier:<12} OOS_hit={_fmt_rate(t.oos_hit_rate)} " f"lift={lift:+.3f} → {verdict_word}" ) if not any_lift: lines.append(" (no class tier had a scorable OOS hit-rate to compare)") # Per-variant verdict. lines.append("") lines.append("VERDICT (this variant):") lines.append(f" {vd['reason']}") if vd.get("thin_warning"): lines.append(f" !! {vd['thin_warning']}") lines.append("") if source == _SOURCE_A: lines.append("Caveats: Source A is survivorship-FREE but THIN (~13 mo) — usually too short") lines.append("to clear _MIN_BACKTEST_MONTHS; an indicative cross-check, not proof.") else: lines.append("Caveats: Source B survivorship undercounts OLD months (use --detrend / -A") lines.append("as controls); short Δ-series + holdout → small test window. β/lag CORE only.") lines.append("=" * 78) return "\n".join(lines) def render_all(payload: dict[str, Any]) -> str: """Render every variant table then the cross-source verdict block.""" blocks: list[str] = [render_table(run) for run in payload["variants"]] cross = payload["cross_verdict"] blocks.append("=" * 78) blocks.append("\n".join(cross["lines"])) blocks.append("=" * 78) return "\n\n".join(blocks) def _fmt_tier_row(t: TierResult, *, lift: float | None) -> str: """Format one tier row for the table.""" lift_s = " -" if lift is None else f"{lift:+.3f}" detr_s = "yes" if t.detrended else "no" return ( f" {t.tier:<12} {t.source:>3} {detr_s:>5} {t.n_aligned:>7} {t.n_train:>6} " f"{t.n_test:>5} {_fmt_lag(t.train_lag):>4} {_fmt_beta(t.train_beta):>9} " f"{_fmt_rate(t.in_sample_r2):>7} {_fmt_rate(t.oos_hit_rate):>8} " f"{_fmt_rate(t.oos_signed_mae):>8} {lift_s:>7} " f"{('yes' if t.lag_stable else 'no'):>7}" ) def _fmt_beta(v: float | None) -> str: return " n/a" if v is None else f"{v:+.4f}" # --------------------------------------------------------------------------- # # Entry point # --------------------------------------------------------------------------- # def _parse_classes(raw: str | None) -> list[str] | None: """Parse --classes: None/'all' → None (auto-discover); CSV → lowercase list. An empty string → [] (EKB-wide only). PURE. """ if raw is None: return None raw = raw.strip() if raw == "": return [] if raw.lower() == "all": return None return [c.strip().lower() for c in raw.split(",") if c.strip()] def _parse_source(raw: str | None) -> list[str]: """Parse --source: B/A → that one source; both/None → [B, A]. PURE. Case-insensitive. Returns the ordered list of sources to run (B before A so the report leads with the long series). Unknown value → ValueError. """ if raw is None: return list(_SOURCES) val = raw.strip().lower() if val in ("both", ""): return list(_SOURCES) if val == _SOURCE_B.lower(): return [_SOURCE_B] if val == _SOURCE_A.lower(): return [_SOURCE_A] raise ValueError(f"--source must be one of B, A, both (got {raw!r})") def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: """argparse setup, factored out for testability.""" p = argparse.ArgumentParser( description=( "READ-ONLY out-of-sample validation of the §9.6 rate-sensitivity engine (Forgejo #978)." ), ) p.add_argument( "--since", default=_DEFAULT_SINCE, help=f"Lower bound (ISO date) of the backtest window (default {_DEFAULT_SINCE}).", ) p.add_argument( "--source", default=_SOURCE_BOTH, help="Series source: 'B' (objective_lots, survivorship-confounded), 'A' " "(objective_corpus_room_month, survivorship-free, ~13 mo), or 'both' " f"(default '{_SOURCE_BOTH}').", ) p.add_argument( "--detrend", action="store_true", help="Also run a DETRENDED variant of each source: linearly detrend " "ln(units) before differencing (removes a spurious monotone " "survivorship trend so it can't drive the regression).", ) p.add_argument( "--deseasonalize", action="store_true", help="Also run a DESEASONALIZED variant of each source (#979): divide out " "month-of-year seasonal factors before log_diff (strips the calendar " "pattern so a seasonal peak isn't read as a rate effect). Factors fit " "on TRAIN months only, applied point-in-time.", ) p.add_argument( "--almon", action="store_true", help="Also run an ALMON-ADL variant of each source (#978): replace the " "single-lag best_lag estimator with an Almon polynomial distributed-lag " "fit (smooth β over lags 0..6, long-run multiplier). Fit on TRAIN " "months only; test prediction sums all lags point-in-time.", ) p.add_argument( "--holdout-frac", type=float, default=_HOLDOUT_FRAC, help=f"TRAIN fraction of the time-ordered split (default {_HOLDOUT_FRAC}). " "Fit on the oldest fraction, test on the newest remainder.", ) p.add_argument( "--classes", default="all", help="Comma-separated classes to backtest, or 'all' to auto-discover " "(default 'all'). Empty → EKB-wide only.", ) p.add_argument( "--district", default=None, help="Optional district filter applied to ALL tiers (default: all districts).", ) p.add_argument( "--json", action="store_true", help="Emit machine-readable JSON instead of the text table.", ) return p.parse_args(argv) def _json_payload(payload: dict[str, Any]) -> dict[str, Any]: """Strip the renderer-only carriers from a run_all payload for JSON output.""" variants = [ {k: v for k, v in run.items() if k not in ("ekb_result", "tier_results")} for run in payload["variants"] ] cross = {k: v for k, v in payload["cross_verdict"].items() if k != "lines"} return {"variants": variants, "cross_verdict": cross} def main(argv: list[str] | None = None) -> int: """CLI entry point. Returns 0 when at least one variant's EKB-wide tier was scorable (backtested, not skipped); 1 if every requested variant was skipped (too thin) — e.g. ``--source A`` alone on prod today (~13 months < _MIN_BACKTEST_MONTHS). """ args = _parse_args(argv) classes = _parse_classes(args.classes) sources = _parse_source(args.source) logger.info( "backtest start: since=%s holdout_frac=%.2f classes=%s district=%s sources=%s " "detrend=%s deseasonalize=%s almon=%s", args.since, args.holdout_frac, "auto" if classes is None else classes, args.district, sources, args.detrend, args.deseasonalize, args.almon, ) db = _session() try: payload = run_all( db, since=args.since, holdout_frac=args.holdout_frac, classes=classes, district=args.district, sources=sources, detrend=args.detrend, deseasonalize=args.deseasonalize, almon=args.almon, ) finally: db.close() if args.json: print(json.dumps(_json_payload(payload), ensure_ascii=False, indent=2, default=str)) else: print(render_all(payload)) any_scorable = any(run["ekb_result"].skipped is None for run in payload["variants"]) return 0 if any_scorable else 1 if __name__ == "__main__": # pragma: no cover raise SystemExit(main())