backend/scripts/backtest_rate_sensitivity.py

@@ -0,0 +1,914 @@
+"""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())

backend/tests/scripts/test_backtest_rate_sensitivity.py

@@ -0,0 +1,477 @@
+"""Unit tests for the read-only §9.6 rate-sensitivity backtest (Forgejo #978).
+
+Covers the PURE backtest logic on SYNTHETIC series (no live DB):
+  - _time_ordered_split  — train/test boundary, clamping, edge sizes
+  - _rate_first_diff     — Δ key_rate, None propagation
+  - _shift_for_lag       — lag alignment (leading None, length preserved)
+  - align_series         — inner-join by year-month
+  - evaluate_oos         — inject sales=f(rate@lag) → high OOS hit-rate;
+                           inject noise → hit-rate ≈ 0.5; point-in-time honesty
+  - backtest_tier        — thin-tier skip; happy path
+  - verdict / tier_lift  — promotion criterion, coin-flip baseline, lag stability
+
+DB is MOCKED (a fake session) only to assert the Source B SQL SHAPE — that it
+uses CAST(:x AS type) and never the psycopg3-incompatible :x::type form.
+
+NOTE: importing scripts.backtest_rate_sensitivity is cheap (the engine import
+is deferred), but evaluate_oos/backtest_tier call into
+app.services.forecasting.* which pulls app.core.config.Settings. Set a dummy
+DATABASE_URL BEFORE importing so that fail-fast doesn't trip (same pattern as
+tests/services/forecasting/test_rate_sensitivity.py).
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import math
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from scripts import backtest_rate_sensitivity as bt
+
+# --------------------------------------------------------------------------- #
+# Synthetic-series helpers
+# --------------------------------------------------------------------------- #
+
+
+def _months(n: int, *, start: dt.date | None = None) -> list[dt.date]:
+    """n consecutive month-firsts, ascending, starting at `start` (default 2019-01)."""
+    start = start or dt.date(2019, 1, 1)
+    out: list[dt.date] = []
+    y, m = start.year, start.month
+    for _ in range(n):
+        out.append(dt.date(y, m, 1))
+        m += 1
+        if m == 13:
+            m = 1
+            y += 1
+    return out
+
+
+def _aperiodic_rate_levels(n: int, *, seed: int = 13) -> list[float]:
+    """Rising key_rate levels with APERIODIC (LCG) jitter → low Δ autocorrelation.
+
+    Mirrors the engine test's regressor: a periodic (sin) jitter would give Δ a
+    sign-flipping autocorrelation so the injected lag competes with false lags.
+    An LCG jitter keeps lags weakly correlated → the true lag wins cleanly.
+    """
+    lvl = 10.0
+    state = seed
+    out: list[float] = []
+    for _ in range(n):
+        state = (state * 1103515245 + 12345) % 2147483648
+        lvl += 0.3 + (state / 2147483648.0 - 0.5) * 0.4
+        out.append(lvl)
+    return out
+
+
+def _units_from_rate(
+    rate_levels: list[float],
+    *,
+    lag: int,
+    beta: float,
+    base: float = 1000.0,
+) -> list[int]:
+    """Sold-units series s.t. log_diff(units)[t] ≈ beta·Δrate[t-lag] (injected link).
+
+    ln(u_t) = ln(u_{t-1}) + beta·Δrate[t-lag]; rounded to int (units are a
+    count). Small step so rounding doesn't kill the relationship. Mirrors the
+    engine test's _synth_sales_units.
+    """
+    rate_deltas = [0.0] + [rate_levels[i] - rate_levels[i - 1] for i in range(1, len(rate_levels))]
+    ln_u = math.log(base)
+    units: list[int] = [round(math.exp(ln_u))]
+    for t in range(1, len(rate_levels)):
+        src = rate_deltas[t - lag] if t - lag >= 0 else 0.0
+        ln_u += beta * src
+        units.append(max(1, round(math.exp(ln_u))))
+    return units
+
+
+# --------------------------------------------------------------------------- #
+# _time_ordered_split
+# --------------------------------------------------------------------------- #
+
+
+class TestTimeOrderedSplit:
+    def test_basic_fraction(self) -> None:
+        assert bt._time_ordered_split(100, 0.7) == 70
+        assert bt._time_ordered_split(30, 0.7) == 21
+
+    def test_keeps_one_month_each_side(self) -> None:
+        # frac=1.0 would put everything in train → clamp to n-1 so test has ≥1.
+        assert bt._time_ordered_split(10, 1.0) == 9
+        # frac=0.0 would empty train → clamp to ≥1.
+        assert bt._time_ordered_split(10, 0.0) == 1
+
+    def test_degenerate_sizes(self) -> None:
+        assert bt._time_ordered_split(0, 0.7) == 0
+        assert bt._time_ordered_split(1, 0.7) == 1  # nothing to split
+
+    def test_is_time_ordered_not_parity(self) -> None:
+        # The split is a single boundary index (past→train, future→test), NOT a
+        # parity/random partition: train is a contiguous prefix.
+        n_train = bt._time_ordered_split(20, 0.7)
+        assert n_train == 14  # contiguous prefix [0:14], test [14:20]
+
+
+# --------------------------------------------------------------------------- #
+# _rate_first_diff / _shift_for_lag / align_series
+# --------------------------------------------------------------------------- #
+
+
+class TestRateFirstDiff:
+    def test_first_diff(self) -> None:
+        assert bt._rate_first_diff([10.0, 12.0, 11.0]) == [None, 2.0, -1.0]
+
+    def test_none_breaks_pair(self) -> None:
+        assert bt._rate_first_diff([1.0, None, 3.0]) == [None, None, None]
+
+    def test_empty_and_single(self) -> None:
+        assert bt._rate_first_diff([]) == [None]
+        assert bt._rate_first_diff([5.0]) == [None]
+
+
+class TestShiftForLag:
+    def test_lag_zero_is_identity(self) -> None:
+        assert bt._shift_for_lag([1.0, 2.0, 3.0], 0) == [1.0, 2.0, 3.0]
+
+    def test_lag_shifts_right_and_truncates(self) -> None:
+        # y[t] ← x[t-2]: two leading None, length preserved.
+        assert bt._shift_for_lag([1.0, 2.0, 3.0, 4.0], 2) == [None, None, 1.0, 2.0]
+
+    def test_no_future_leak(self) -> None:
+        # Element at index t must equal the ORIGINAL element at t-lag (never t+k).
+        x = [10.0, 20.0, 30.0, 40.0, 50.0]
+        lag = 1
+        shifted = bt._shift_for_lag(x, lag)
+        for t in range(lag, len(x)):
+            assert shifted[t] == x[t - lag]
+
+
+class TestAlignSeries:
+    def test_inner_join_by_month(self) -> None:
+        ms = _months(4)
+        sales = {ms[0]: 100, ms[1]: 110, ms[2]: 120, ms[3]: 130}
+        # rate missing ms[0]; has an extra month not in sales.
+        rate = {ms[1]: 7.0, ms[2]: 7.5, ms[3]: 8.0, dt.date(2030, 1, 1): 9.0}
+        months, units, rates = bt.align_series(sales, rate)
+        assert months == [ms[1], ms[2], ms[3]]  # intersection only, ascending
+        assert units == [110, 120, 130]
+        assert rates == [7.0, 7.5, 8.0]
+
+    def test_empty_intersection(self) -> None:
+        months, units, rates = bt.align_series({_months(1)[0]: 1}, {dt.date(2030, 1, 1): 2.0})
+        assert months == [] and units == [] and rates == []
+
+
+# --------------------------------------------------------------------------- #
+# evaluate_oos — the core OOS metric
+# --------------------------------------------------------------------------- #
+
+
+class TestEvaluateOos:
+    def test_injected_signal_high_oos_hit_rate(self) -> None:
+        # sales react to rate at lag 2 with a clean negative β → the TRAIN fit
+        # should generalise: nearly every TEST month's predicted sign matches.
+        n = 48
+        rate = _aperiodic_rate_levels(n)
+        units = _units_from_rate(rate, lag=2, beta=-0.05)
+        delta_sales = _delta_ln(units)
+        rate_deltas = bt._rate_first_diff(rate)
+
+        res = bt.evaluate_oos(delta_sales, rate_deltas, holdout_frac=0.7)
+        assert res["train_lag"] == 2
+        assert res["train_beta"] is not None and res["train_beta"] < 0
+        assert res["oos_hit_rate"] is not None
+        # A real injected signal → directional hit-rate clearly beats a coin flip.
+        assert res["oos_hit_rate"] >= 0.8
+        # In-sample R² is high by construction (reported, not trusted).
+        assert res["in_sample_r2"] is not None and res["in_sample_r2"] > 0.9
+        # Lag stable: full-sample refit finds the same lag.
+        assert res["full_sample_lag"] == 2
+        assert res["lag_stable"] is True
+
+    def test_pure_noise_hit_rate_near_coin_flip(self) -> None:
+        # No rate→sales link: sales are an independent aperiodic walk. Either no
+        # gated lag is found on TRAIN (→ None), or any spurious fit predicts
+        # direction no better than a coin flip on held-out months.
+        n = 60
+        rate = _aperiodic_rate_levels(n, seed=1)
+        noise = _aperiodic_rate_levels(n, seed=999)  # uncorrelated second series
+        units = [max(1, round(1000.0 * math.exp(0.01 * (v - 10.0)))) for v in noise]
+        delta_sales = _delta_ln(units)
+        rate_deltas = bt._rate_first_diff(rate)
+
+        res = bt.evaluate_oos(delta_sales, rate_deltas, holdout_frac=0.7)
+        hr = res["oos_hit_rate"]
+        # Honest outcome: no signal → either ungated (None) or ~coin-flip.
+        assert hr is None or hr <= 0.7
+
+    def test_too_few_months_returns_empty(self) -> None:
+        # 1 month → can't split → empty result (all metrics None, not a crash).
+        res = bt.evaluate_oos([None], [None], holdout_frac=0.7)
+        assert res["train_lag"] is None
+        assert res["oos_hit_rate"] is None
+        assert res["n_train"] == 1 and res["n_test"] == 0
+
+    def test_no_gated_lag_on_train_returns_empty(self) -> None:
+        # Positive rate→sales link (β>0) → engine gate (slope<0) rejects every
+        # lag on TRAIN → nothing to validate → empty (None) result, no crash.
+        n = 40
+        rate = _aperiodic_rate_levels(n)
+        units = _units_from_rate(rate, lag=1, beta=+0.05)  # wrong sign
+        delta_sales = _delta_ln(units)
+        rate_deltas = bt._rate_first_diff(rate)
+        res = bt.evaluate_oos(delta_sales, rate_deltas, holdout_frac=0.7)
+        assert res["train_lag"] is None
+        assert res["oos_hit_rate"] is None
+
+    def test_point_in_time_no_future_leak(self) -> None:
+        # Build a signal, then confirm the TEST prediction at the FIRST test
+        # month uses only rate data at or before it. We reconstruct the expected
+        # prediction from the public _shift_for_lag and check evaluate_oos's MAE
+        # is finite (a future leak would mismatch lengths / shift indices).
+        n = 36
+        rate = _aperiodic_rate_levels(n)
+        units = _units_from_rate(rate, lag=3, beta=-0.04)
+        delta_sales = _delta_ln(units)
+        rate_deltas = bt._rate_first_diff(rate)
+        res = bt.evaluate_oos(delta_sales, rate_deltas, holdout_frac=0.7)
+        assert res["oos_signed_mae"] is not None
+        assert math.isfinite(res["oos_signed_mae"])
+        # First scored test month index = n_train; predictor must be Δrate[t-lag].
+        lag = res["train_lag"]
+        assert lag is not None
+        shifted = bt._shift_for_lag(rate_deltas, lag)
+        # The shifted regressor at the first test index is at or before it.
+        assert shifted[res["n_train"]] is None or isinstance(shifted[res["n_train"]], float)
+
+
+# --------------------------------------------------------------------------- #
+# backtest_tier — thin-tier skip + happy path
+# --------------------------------------------------------------------------- #
+
+
+class TestBacktestTier:
+    def test_thin_tier_skipped_not_dropped(self) -> None:
+        # Fewer than _MIN_BACKTEST_MONTHS aligned months → skipped with a reason,
+        # all metrics None (NOT a silent drop, NOT a crash).
+        ms = _months(5)
+        rate = _aperiodic_rate_levels(5)
+        sales = {ms[i]: 100 + i for i in range(5)}
+        rate_by = {ms[i]: rate[i] for i in range(5)}
+        res = bt.backtest_tier(sales, rate_by, tier="комфорт", min_months=18)
+        assert res.skipped is not None
+        assert "aligned months" in res.skipped
+        assert res.oos_hit_rate is None
+        assert res.n_aligned == 5
+
+    def test_happy_path_builds_metrics(self) -> None:
+        n = 48
+        ms = _months(n)
+        rate = _aperiodic_rate_levels(n)
+        units = _units_from_rate(rate, lag=2, beta=-0.05)
+        sales = {ms[i]: units[i] for i in range(n)}
+        rate_by = {ms[i]: rate[i] for i in range(n)}
+        res = bt.backtest_tier(sales, rate_by, tier=bt._EKB_WIDE, holdout_frac=0.7)
+        assert res.skipped is None
+        assert res.tier == bt._EKB_WIDE
+        assert res.train_lag == 2
+        assert res.oos_hit_rate is not None and res.oos_hit_rate >= 0.8
+        assert res.n_aligned == n
+
+    def test_alignment_drops_unmatched_months(self) -> None:
+        # Sales and rate only overlap on a thin window → aligned count reflects
+        # the INTERSECTION, which here is below the min → skipped.
+        ms = _months(40)
+        rate = _aperiodic_rate_levels(40)
+        sales = {ms[i]: 100 + i for i in range(40)}
+        # rate only for the last 10 months → intersection = 10 < 18.
+        rate_by = {ms[i]: rate[i] for i in range(30, 40)}
+        res = bt.backtest_tier(sales, rate_by, tier="бизнес", min_months=18)
+        assert res.n_aligned == 10
+        assert res.skipped is not None
+
+
+# --------------------------------------------------------------------------- #
+# verdict / tier_lift
+# --------------------------------------------------------------------------- #
+
+
+def _tier(
+    *,
+    tier: str = bt._EKB_WIDE,
+    n_aligned: int = 40,
+    n_train: int = 28,
+    n_test: int = 12,
+    train_lag: int | None = 2,
+    train_beta: float | None = -0.05,
+    in_sample_r2: float | None = 0.95,
+    oos_hit_rate: float | None = 0.75,
+    oos_signed_mae: float | None = 0.02,
+    full_sample_lag: int | None = 2,
+    lag_stable: bool = True,
+    skipped: str | None = None,
+) -> bt.TierResult:
+    return bt.TierResult(
+        tier=tier,
+        n_aligned=n_aligned,
+        n_train=n_train,
+        n_test=n_test,
+        train_lag=train_lag,
+        train_beta=train_beta,
+        in_sample_r2=in_sample_r2,
+        oos_hit_rate=oos_hit_rate,
+        oos_signed_mae=oos_signed_mae,
+        full_sample_lag=full_sample_lag,
+        lag_stable=lag_stable,
+        skipped=skipped,
+    )
+
+
+class TestVerdict:
+    def test_promote_when_beats_coin_and_lag_stable(self) -> None:
+        vd = bt.verdict(_tier(oos_hit_rate=0.75, lag_stable=True))
+        assert vd["promote"] is True
+        assert "OOS predictive value" in vd["reason"]
+
+    def test_keep_advisory_when_at_coin_flip(self) -> None:
+        vd = bt.verdict(_tier(oos_hit_rate=0.52, lag_stable=True))  # ≤ 0.5+margin
+        assert vd["promote"] is False
+        assert "keep advisory" in vd["reason"]
+
+    def test_keep_advisory_when_lag_unstable(self) -> None:
+        vd = bt.verdict(_tier(oos_hit_rate=0.9, lag_stable=False, full_sample_lag=6))
+        assert vd["promote"] is False
+        assert "lag unstable" in vd["reason"]
+
+    def test_keep_advisory_when_skipped(self) -> None:
+        vd = bt.verdict(_tier(skipped="only 5 aligned months (< 18)"))
+        assert vd["promote"] is False
+        assert "keep advisory" in vd["reason"]
+
+    def test_keep_advisory_when_no_hit_rate(self) -> None:
+        vd = bt.verdict(_tier(oos_hit_rate=None))
+        assert vd["promote"] is False
+
+    def test_thin_warning_set_for_small_test_window(self) -> None:
+        vd = bt.verdict(_tier(oos_hit_rate=0.9, n_test=3, lag_stable=True))
+        assert vd["promote"] is True
+        assert vd["thin_warning"] is not None
+        assert "small" in vd["thin_warning"]
+
+
+class TestTierLift:
+    def test_positive_lift_beats_ekb(self) -> None:
+        ekb = _tier(oos_hit_rate=0.6)
+        cls = _tier(tier="комфорт", oos_hit_rate=0.75)
+        assert bt.tier_lift(ekb, cls) is not None
+        assert math.isclose(bt.tier_lift(ekb, cls), 0.15)
+
+    def test_none_when_either_missing(self) -> None:
+        ekb = _tier(oos_hit_rate=None)
+        cls = _tier(oos_hit_rate=0.75)
+        assert bt.tier_lift(ekb, cls) is None
+        assert bt.tier_lift(_tier(oos_hit_rate=0.6), _tier(oos_hit_rate=None)) is None
+
+
+# --------------------------------------------------------------------------- #
+# _parse_classes
+# --------------------------------------------------------------------------- #
+
+
+class TestParseClasses:
+    def test_all_means_autodiscover(self) -> None:
+        assert bt._parse_classes("all") is None
+        assert bt._parse_classes("ALL") is None
+        assert bt._parse_classes(None) is None
+
+    def test_empty_means_ekb_only(self) -> None:
+        assert bt._parse_classes("") == []
+        assert bt._parse_classes("   ") == []
+
+    def test_csv_lowercased_and_trimmed(self) -> None:
+        assert bt._parse_classes("Комфорт, Бизнес ,премиум") == ["комфорт", "бизнес", "премиум"]
+
+
+# --------------------------------------------------------------------------- #
+# DB layer SQL SHAPE — mocked session, asserts CAST not :: and read-only
+# --------------------------------------------------------------------------- #
+
+
+class _CaptureResult:
+    """Stands in for a SQLAlchemy Result — returns canned rows from .all()."""
+
+    def __init__(self, rows: list) -> None:
+        self._rows = rows
+
+    def all(self) -> list:
+        return self._rows
+
+
+class _CaptureSession:
+    """Fake Session capturing (sql_text, params) and returning canned rows."""
+
+    def __init__(self, rows: list) -> None:
+        self.rows = rows
+        self.calls: list[tuple[str, dict]] = []
+
+    def execute(self, stmt: object, params: dict | None = None) -> _CaptureResult:
+        self.calls.append((str(stmt), dict(params or {})))
+        return _CaptureResult(self.rows)
+
+
+class TestSourceBSqlShape:
+    def test_units_sql_uses_cast_not_double_colon(self) -> None:
+        sql = str(bt._SOURCE_B_UNITS_SQL)
+        assert "CAST(:premise_kind AS text)" in sql
+        assert "CAST(:since AS date)" in sql
+        # psycopg3-incompatible :name::type must NOT appear.
+        assert "::" not in sql
+
+    def test_units_sql_is_select_only(self) -> None:
+        sql = str(bt._SOURCE_B_UNITS_SQL).strip().lower()
+        assert sql.startswith("select")
+        for forbidden in ("insert", "update", "delete", "drop", "alter", "create"):
+            assert forbidden not in sql
+
+    def test_classes_sql_uses_cast_not_double_colon(self) -> None:
+        sql = str(bt._SOURCE_B_CLASSES_SQL)
+        assert "CAST(:premise_kind AS text)" in sql
+        assert "::" not in sql
+
+    def test_load_sales_by_month_binds_and_shapes(self) -> None:
+        ms = _months(3)
+        sess = _CaptureSession([(ms[0], 10), (ms[1], 20), (None, 99)])
+        out = bt.load_sales_by_month(
+            sess,  # type: ignore[arg-type]
+            since="2019-01-01",
+            obj_class="комфорт",
+            district=None,
+        )
+        # None-month row dropped; rows mapped to {month: units}.
+        assert out == {ms[0]: 10, ms[1]: 20}
+        # Bound params include the class filter and premise kind (parametrised,
+        # not interpolated) — confirms no SQL-injection-prone string building.
+        _sql, params = sess.calls[0]
+        assert params["cls"] == "комфорт"
+        assert params["premise_kind"] == bt._PREMISE_KIND
+        assert params["since"] == "2019-01-01"
+
+    def test_load_classes_maps_rows(self) -> None:
+        sess = _CaptureSession([("комфорт",), ("бизнес",), (None,)])
+        out = bt.load_classes(sess, since="2019-01-01")  # type: ignore[arg-type]
+        assert out == ["комфорт", "бизнес"]
+
+
+# --------------------------------------------------------------------------- #
+# Local Δln helper (mirror sales_series.log_diff for building synthetic inputs)
+# --------------------------------------------------------------------------- #
+
+
+def _delta_ln(series: list[int]) -> list[float | None]:
+    """Δln for synthetic inputs — uses the production log_diff via the engine."""
+    _bl, _ols, log_diff = bt._import_engine()
+    return log_diff(series)