Light1YT
9a845646fd
REOPENED 951-B §9.6.
PART A: fix look-ahead leakage in backtest_rate_sensitivity --detrend. The
ln(units) trend was fit over train+test then split, so test data shaped the
detrend and inflated the OOS hit-rate. _detrend_log now takes fit_n; backtest_tier
fits the trend on TRAIN months only (same split evaluate_oos uses) and projects
(a,b) point-in-time onto test. Default fit_n=None preserves prior behaviour.
PART B (DoD): new app/services/forecasting/regression.py — Almon polynomial
distributed-lag (deg 2) of Δln(district demand) on Δkey_rate lags 0..6 via
OLS-on-Almon-regressors (numpy lstsq) + per-lag reconstruction + manual
Newey-West HAC SEs (NO statsmodels). Output {best_lag_months, coef=long-run
multiplier, x_pct, r2, n, per_lag_coef, hac_se,...}; gate mirrors _elasticity_coef
(n<30 OR R²<0.1 OR Σβ≥0 → fallback); §9.6 phrase from the lag shape. ADVISORY,
shipped standalone (integration point documented), NOT wired — protects the live
compute_rate_sensitivity consumers.
125+31 tests (synthetic known-lag recovery, HAC computed/differs-from-OLS,
fallback gating, no-leakage detrend). ruff clean. Refs #978
1445 lines
56 KiB
Python
1445 lines
56 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?
|
||
|
||
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
|
||
"""
|
||
|
||
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
|
||
|
||
# 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
|
||
|
||
|
||
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 _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
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"tier": self.tier,
|
||
"source": self.source,
|
||
"detrended": self.detrended,
|
||
"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 _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 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 backtest_tier(
|
||
sales_by_month: dict[date, int],
|
||
rate_by_month: dict[date, float],
|
||
*,
|
||
tier: str,
|
||
source: str = _SOURCE_B,
|
||
detrend: bool = False,
|
||
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 (``log_diff`` raw, or Δ of linear-detrended ``ln`` when
|
||
``detrend`` — see ``_delta_sales_series``) and Δrate (first diff), then
|
||
delegates to ``evaluate_oos``. ``source`` (B/A) and ``detrend`` are recorded
|
||
on the result for labelling, not used in the math here. 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,
|
||
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})",
|
||
)
|
||
|
||
# Detrend (when enabled) must be fit on TRAIN months ONLY, then projected
|
||
# point-in-time onto the test months — otherwise the held-out TEST data
|
||
# shapes the trend and the OOS hit-rate is inflated by look-ahead leakage
|
||
# (#978 Part A). We compute the SAME train boundary evaluate_oos will use
|
||
# (len(delta_sales) == len(units) == n_aligned, so the split index matches)
|
||
# and pass it as the detrend fit window.
|
||
n_train = _time_ordered_split(n_aligned, holdout_frac)
|
||
delta_sales = _delta_sales_series(units, detrend=detrend, fit_n=n_train)
|
||
rate_deltas = _rate_first_diff([float(r) for r in rates])
|
||
res = evaluate_oos(delta_sales, rate_deltas, holdout_frac=holdout_frac)
|
||
|
||
return TierResult(
|
||
tier=tier,
|
||
source=source,
|
||
detrended=detrend,
|
||
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:
|
||
• 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``, 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).
|
||
|
||
Returns ``{"promote": bool, "reason": str, "thin_warning": str | None}``.
|
||
Honest: if the OOS test window is tiny the reason says so even when the
|
||
hit-rate happens to clear the bar.
|
||
"""
|
||
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
|
||
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:
|
||
reason = (
|
||
f"engine has OOS predictive value (candidate to promote from "
|
||
f"advisory): EKB-wide OOS hit-rate={ekb.oos_hit_rate:.2f} > "
|
||
f"0.5+{margin:.2f} and lag stable (lag={ekb.train_lag})"
|
||
)
|
||
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})")
|
||
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,
|
||
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 ``detrend`` 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).
|
||
"""
|
||
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,
|
||
holdout_frac=holdout_frac,
|
||
)
|
||
logger.info(
|
||
"source=%s detrend=%s EKB-wide: aligned=%d train=%d test=%d lag=%s hit_rate=%s",
|
||
source,
|
||
detrend,
|
||
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,
|
||
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,
|
||
"a_district_ignored": a_district_ignored,
|
||
"params": {
|
||
"since": since,
|
||
"holdout_frac": holdout_frac,
|
||
"district": district,
|
||
"classes": classes,
|
||
"source": source,
|
||
"detrended": detrend,
|
||
"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) -> str:
|
||
"""Human label for a (source, detrend) run, e.g. 'B raw' / 'B detrended' / 'A raw'."""
|
||
return f"{source} {'detrended' if detrend else 'raw'}"
|
||
|
||
|
||
def _plan_variants(sources: list[str], detrend: bool) -> list[tuple[str, bool]]:
|
||
"""Which (source, detrend) variants to run, in report order. PURE.
|
||
|
||
For each requested source we always run the RAW variant (the reference). When
|
||
``--detrend`` is set we ALSO run the detrended variant of that source, so a
|
||
single invocation can show ``B raw`` next to ``B detrended`` (the survivorship
|
||
control) for the verdict's side-by-side comparison.
|
||
"""
|
||
variants: list[tuple[str, bool]] = []
|
||
for src in sources:
|
||
variants.append((src, False))
|
||
if detrend:
|
||
variants.append((src, True))
|
||
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 variants (B raw / B detrended / A).
|
||
|
||
The #978b question: is Source B's negative OOS verdict a SURVIVORSHIP
|
||
ARTIFACT or a real ``no signal``? We line up each variant's EKB-wide
|
||
OOS hit-rate vs the 0.5 coin-flip baseline and synthesise a conclusion:
|
||
|
||
• If NO variant (B raw, B detrended, or survivorship-free A) clears
|
||
coin-flip+margin → the negative verdict is corroborated as a real
|
||
``no signal``, not an artifact (the detrend + survivorship-free controls
|
||
agree). Source A's thin-data caveat is attached when A drove a verdict.
|
||
• If the detrended or survivorship-free variant DOES clear the bar while
|
||
raw B did not → the raw verdict may have been a survivorship artifact;
|
||
flag the variant that shows signal.
|
||
|
||
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(run["source"], run["detrended"])
|
||
hr = ekb.oos_hit_rate
|
||
scorable = ekb.skipped is None and hr is not None and ekb.n_test >= 1
|
||
beats = bool(scorable and hr is not None and hr >= 0.5 + margin and ekb.lag_stable)
|
||
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"],
|
||
"scorable": scorable,
|
||
"oos_hit_rate": _round_or_none(hr, 4),
|
||
"n_test": ekb.n_test,
|
||
"lag_stable": ekb.lag_stable,
|
||
"beats_coin": beats,
|
||
"skipped": ekb.skipped,
|
||
}
|
||
)
|
||
|
||
lines: list[str] = []
|
||
lines.append("CROSS-SOURCE VERDICT (B raw vs B detrended vs A — #978b):")
|
||
for r in rows:
|
||
if not r["scorable"]:
|
||
why = r["skipped"] or "no gated lag / empty test window"
|
||
lines.append(f" {r['variant']:<13} → not scorable ({why})")
|
||
else:
|
||
tag = "SIGNAL > coin-flip" if r["beats_coin"] else "no signal (≤ coin-flip)"
|
||
lines.append(
|
||
f" {r['variant']:<13} → 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 Source B may be a SURVIVORSHIP "
|
||
"ARTIFACT — the control(s) above recover directional signal."
|
||
)
|
||
promote_any = True
|
||
else:
|
||
conclusion = (
|
||
"CONCLUSION: NO variant (raw B, detrended B, or survivorship-free A) "
|
||
"beats coin-flip+margin out-of-sample. The §9.6 negative verdict is a "
|
||
"REAL 'no signal', NOT a survivorship artifact — detrending B and the "
|
||
"survivorship-free Source A both agree. 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,
|
||
) -> dict[str, Any]:
|
||
"""Run every requested (source, detrend) variant + the cross-source verdict.
|
||
|
||
Loads the monthly key_rate ONCE and reuses it across variants. ``sources`` is
|
||
a subset of (B, A); ``detrend`` adds the detrended variant of each. No
|
||
writes. 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)
|
||
runs: list[dict[str, Any]] = []
|
||
for src, dt_flag in variants:
|
||
runs.append(
|
||
run_backtest(
|
||
db,
|
||
since=since,
|
||
holdout_frac=holdout_frac,
|
||
classes=classes,
|
||
district=district,
|
||
source=src,
|
||
detrend=dt_flag,
|
||
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"]
|
||
|
||
lines: list[str] = []
|
||
lines.append("=" * 78)
|
||
lines.append(
|
||
f"BACKTEST [source {source}{' · detrended' if detrended else ''}]: "
|
||
"§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 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(
|
||
"--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",
|
||
args.since,
|
||
args.holdout_frac,
|
||
"auto" if classes is None else classes,
|
||
args.district,
|
||
sources,
|
||
args.detrend,
|
||
)
|
||
|
||
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,
|
||
)
|
||
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())
|