Light1YT
1f490b7915
Add scripts/backtest_rate_sensitivity.py — STRICTLY READ-ONLY out-of-sample validation gate for the §9.6 engine (#1009). Monthly sold-units from Source B (objective_lots) EKB-wide + per class, aligned to monthly key_rate, time-ordered holdout (fit oldest 70%, test newest), point-in-time OOS directional hit-rate + in-sample-vs-OOS honesty block + per-tier lift + coin-flip+lag-stability verdict. Deterministic, no LLM, no DB writes. CLI --since/--holdout-frac/--classes/--json. 36 unit tests on synthetic. VERDICT (run vs prod 2026-06-03): EKB-wide OOS hit-rate 0.148 (< 0.5 coin-flip), lag unstable → §9.6 engine NOT OOS-validated → stays ADVISORY. Likely confounded by Source B survivorship bias (recent-month inflation → spurious trend); clean re-validation needs Source A depth (13mo now, deepens weekly) or survivorship correction. The harness re-validates when data improves.
914 lines
35 KiB
Python
914 lines
35 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.
|
||
|
||
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. This is the SAME caveat the
|
||
§9.6 module documents; it is why Source B is used here (Source A
|
||
``objective_corpus_room_month`` has only ~13 months — too thin to
|
||
regress).
|
||
(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.
|
||
(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
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import logging
|
||
from dataclasses import dataclass
|
||
from datetime import date
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
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"
|
||
|
||
|
||
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
|
||
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,
|
||
"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 _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 backtest_tier(
|
||
sales_by_month: dict[date, int],
|
||
rate_by_month: dict[date, float],
|
||
*,
|
||
tier: str,
|
||
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
|
||
Δln(sales) (reused ``log_diff``) and Δrate (first diff), then delegates to
|
||
``evaluate_oos``. 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.
|
||
"""
|
||
_best_lag, _ols, log_diff = _import_engine()
|
||
|
||
months, units, rates = align_series(sales_by_month, rate_by_month)
|
||
n_aligned = len(months)
|
||
if n_aligned < min_months:
|
||
return TierResult(
|
||
tier=tier,
|
||
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})",
|
||
)
|
||
|
||
delta_sales = log_diff(units)
|
||
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,
|
||
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
|
||
"""
|
||
)
|
||
|
||
# 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 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_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 run_backtest(
|
||
db: Session,
|
||
*,
|
||
since: str,
|
||
holdout_frac: float,
|
||
classes: list[str] | None,
|
||
district: str | None,
|
||
) -> dict[str, Any]:
|
||
"""Drive the full read-only backtest and return a results dict. No writes.
|
||
|
||
Loads the monthly key_rate once, then the EKB-wide and per-class Source B
|
||
sold-units series, backtests each tier (``backtest_tier``), and assembles
|
||
the verdict + per-tier OOS lifts.
|
||
|
||
``classes`` None → auto-discover every class present in Source B; an empty
|
||
list → EKB-wide only. ``district`` optionally narrows ALL tiers.
|
||
"""
|
||
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(db, since=since)
|
||
logger.info("auto-discovered classes: %s", classes)
|
||
|
||
# EKB-wide tier (no class filter).
|
||
ekb_sales = load_sales_by_month(db, since=since, obj_class=None, district=district)
|
||
ekb = backtest_tier(ekb_sales, rate_by_month, tier=_EKB_WIDE, holdout_frac=holdout_frac)
|
||
logger.info(
|
||
"EKB-wide: aligned=%d train=%d test=%d lag=%s hit_rate=%s",
|
||
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_by_month(db, since=since, obj_class=cls, district=district)
|
||
res = backtest_tier(cls_sales, rate_by_month, tier=cls, holdout_frac=holdout_frac)
|
||
tiers.append(res)
|
||
lifts[cls] = tier_lift(ekb, res)
|
||
logger.info(
|
||
"tier=%s aligned=%d test=%d hit_rate=%s lift=%s skipped=%s",
|
||
cls,
|
||
res.n_aligned,
|
||
res.n_test,
|
||
res.oos_hit_rate,
|
||
lifts[cls],
|
||
res.skipped,
|
||
)
|
||
|
||
vd = verdict(ekb)
|
||
return {
|
||
"params": {
|
||
"since": since,
|
||
"holdout_frac": holdout_frac,
|
||
"district": district,
|
||
"classes": classes,
|
||
"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 _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)
|
||
|
||
|
||
def render_table(results: dict[str, Any]) -> str:
|
||
"""Render the 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"]
|
||
|
||
lines: list[str] = []
|
||
lines.append("=" * 78)
|
||
lines.append("BACKTEST: §9.6 rate-sensitivity engine — out-of-sample 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 B (objective_lots) monthly sold-units vs monthly key_rate.")
|
||
lines.append("")
|
||
|
||
header = (
|
||
f" {'tier':<12} {'aligned':>7} {'train':>6} {'test':>5} {'lag':>4} "
|
||
f"{'beta':>9} {'inR2':>7} {'OOS_hit':>8} {'OOS_MAE':>8} {'lift':>7} {'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)")
|
||
|
||
# Verdict.
|
||
lines.append("")
|
||
lines.append("VERDICT:")
|
||
lines.append(f" {vd['reason']}")
|
||
if vd.get("thin_warning"):
|
||
lines.append(f" !! {vd['thin_warning']}")
|
||
|
||
lines.append("")
|
||
lines.append("Caveats: Source B survivorship undercounts OLD months; short Δ-series + holdout")
|
||
lines.append("leaves a small test window; this validates the β/lag CORE, not the full engine.")
|
||
lines.append("=" * 78)
|
||
return "\n".join(lines)
|
||
|
||
|
||
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}"
|
||
return (
|
||
f" {t.tier:<12} {t.n_aligned:>7} {t.n_train:>6} {t.n_test:>5} "
|
||
f"{_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_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(
|
||
"--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 main(argv: list[str] | None = None) -> int:
|
||
"""CLI entry point. Returns 0 when the EKB-wide tier was backtested, 1 if skipped."""
|
||
args = _parse_args(argv)
|
||
classes = _parse_classes(args.classes)
|
||
logger.info(
|
||
"backtest start: since=%s holdout_frac=%.2f classes=%s district=%s",
|
||
args.since,
|
||
args.holdout_frac,
|
||
"auto" if classes is None else classes,
|
||
args.district,
|
||
)
|
||
|
||
db = _session()
|
||
try:
|
||
results = run_backtest(
|
||
db,
|
||
since=args.since,
|
||
holdout_frac=args.holdout_frac,
|
||
classes=classes,
|
||
district=args.district,
|
||
)
|
||
finally:
|
||
db.close()
|
||
|
||
if args.json:
|
||
payload = {k: v for k, v in results.items() if k not in ("ekb_result", "tier_results")}
|
||
print(json.dumps(payload, ensure_ascii=False, indent=2, default=str))
|
||
else:
|
||
print(render_table(results))
|
||
|
||
ekb: TierResult = results["ekb_result"]
|
||
return 0 if ekb.skipped is None else 1
|
||
|
||
|
||
if __name__ == "__main__": # pragma: no cover
|
||
raise SystemExit(main())
|