Light1YT
cc66f51863
The OOS verdict flagged a variant 'candidate to promote' on hit-rate >= 0.5+margin + lag_stable alone. On thin data this over-claims: Source A Almon-ADL scored 6/10 (0.60) lag-stable and was flagged as signal, but P(X>=6|10,0.5)~=0.377 -- a coin flip. Live ground-truth confirmed no signal (full-sample R2~=0.003, wrong sign). Add exact stdlib-only one-sided binomial _binom_sf_ge + _VERDICT_ALPHA=0.05 and require P(X>=hits|n_test,0.5) < alpha in both verdict() and cross_source_verdict() on top of the effect-size margin. hits recovered exactly as round(hit_rate*n_test) (n_test==scored invariant; no evaluator shape change). Verdict text now states n_test + the binomial p on pass and fail. Evaluator/estimator math and the read-only SELECT discipline untouched. Refs #978.
1919 lines
79 KiB
Python
1919 lines
79 KiB
Python
"""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())
|