gendesign/backend/scripts/backtest_rate_sensitivity.py

"""Backtest harness — out-of-sample validation of the §9.6 rate-sensitivity engine.

Forgejo issue #978 (#951-B validation, Site Finder v2 EPIC 7). **STRICTLY
READ-ONLY**: this script issues only SELECT queries against prod. It never
INSERTs, UPDATEs, or runs DDL.

WHAT IT MEASURES
----------------
The §9.6 engine (#1009 ``rate_sensitivity.py``) regresses a segment's monthly
Δln(sold-units) on Δ(key_rate) at several lags, picks the lag with the most
negative gate-passing slope (β), and emits an ADVISORY phrase. That engine is
the gate that decides whether the advisory forecast can be PROMOTED. This
harness is the out-of-sample (OOS) check that gate must clear: does β fit on a
TRAIN window actually predict the SIGN of Δln(sales) on a held-out TEST window?

We mirror the trade-in ``backtest_estimator.py`` discipline (read-only,
holdout, in-sample-vs-OOS honesty block):

  1. Build the monthly sold-units series from Source B (``objective_lots``,
     EKB-wide and per ``class``) and the monthly ``key_rate`` (last-known per
     month, reusing ``forecasting.macro_series.get_monthly_macro``).
  2. Align by year-month → Δln(sales) (reuse ``sales_series.log_diff``) and
     Δkey_rate (first difference).
  3. TIME-ORDERED holdout (NOT random / even-odd — this is a time series): fit
     on the first ``holdout_frac`` of months, evaluate on the rest. Fit reuses
     ``rate_sensitivity.best_lag`` (→ β + winning lag on TRAIN only).
  4. Predict each TEST month's Δln(sales) sign (and magnitude) from
     β·Δkey_rate@lag, strictly point-in-time (macro as-of each test month, the
     lagged regressor never reaches past the test month).
  5. Report the OOS directional hit-rate (fraction of test months where the
     predicted sign matches the actual sign), signed MAE, n_train / n_test, and
     the winning lag — alongside the in-sample R².

IN-SAMPLE vs OUT-OF-SAMPLE HONESTY (copy trade-in discipline)
------------------------------------------------------------
We report BOTH the in-sample R² (high by construction — the slope is fit to
minimise residuals on the very points it is scored on) AND the OOS directional
hit-rate (the only trustworthy number). Stated plainly: the in-sample R² is NOT
evidence the engine predicts — it is arithmetic. Only the held-out directional
hit-rate, computed point-in-time on months the fit never saw, tells us whether
β·Δrate@lag carries real predictive value worth promoting from advisory.

PER-TIER
--------
We backtest EKB-wide and each ``class``. A class-specific β is only worth
promoting if it beats the EKB-wide prior OUT-OF-SAMPLE (directional hit-rate
lift). A tier with fewer than ``_MIN_BACKTEST_MONTHS`` aligned train+test
months is SKIPPED with a printed note (no silent drop).

VERDICT
-------
We print whether the EKB-wide OOS directional hit-rate beats a 0.5 coin-flip
baseline (by at least ``_VERDICT_HITRATE_MARGIN``) AND the winning lag is the
same on TRAIN and on a full-sample refit (lag stability) →
"engine has OOS predictive value (candidate to promote from advisory)" vs
"insufficient OOS signal — keep advisory". We stay honest: 102 raw months still
collapse to few test points after Δ (loses 1), the survivorship-thinned early
months, and the holdout split — if the OOS test set is tiny we say so rather
than over-claim.

TWO CLEANER CROSS-CHECKS (#978b)
--------------------------------
The Source B OOS verdict was ``no signal`` (hit-rate < coin-flip). Source B is
survivorship-CONFOUNDED — its monthly counts trend ~3x upward 2019→2025 purely
because only currently-listed lots are visible, so the engine's negative verdict
could be an ARTIFACT rather than a true ``no signal``. We add two controls:

  • ``--source A`` — build the series from ``objective_corpus_room_month``
    (Objective's corp_sum monthly deal AGGREGATE). This counts deals PER MONTH
    regardless of current listing → survivorship-FREE. It is only ~13 months
    deep on prod (≈2025-05→2026-05), statistically thin, but a clean cross-check
    (the verdict carries an explicit thin-data caveat for it).
  • ``--detrend`` — before differencing, fit a linear time trend
    ``ln(units) ~ a + b·month_index`` and subtract it, regressing the Δ of the
    RESIDUALS vs Δrate. A spurious monotone survivorship trend lands almost
    entirely in ``b`` and is removed, so it can no longer drive the regression.

Read both alongside Source B raw: if Source B DETRENDED still shows no OOS
signal AND survivorship-free Source A agrees (thin caveat aside), the engine's
negative verdict is a real ``no signal``, not a survivorship artifact.

CAVEATS (read before trusting the numbers)
------------------------------------------
  (a) SURVIVORSHIP — Source B (``objective_lots``) is the last UPSERT snapshot
      per lot: only currently-listed lots are visible, so sold-and-delisted
      lots are undercounted in OLDER months. The monthly sold-units series is
      therefore biased low on the early window (a ~3x spurious upward trend).
      ``--detrend`` and ``--source A`` are the controls for this.
  (b) SHORT Δ-SERIES — even ~102 months of lots collapse after first-difference
      + the survivorship-thinned head + the holdout split. The OOS test window
      can be small; the hit-rate's confidence is correspondingly weak. The
      verdict notes this explicitly. Source A (~13 months) is thinner still and
      will usually be SKIPPED below ``_MIN_BACKTEST_MONTHS`` — never faked.
  (c) ADVISORY MECHANISM ONLY — this exercises the β / lag CORE of the engine.
      It does NOT replay the full module (shrinkage to the EKB prior, the
      confounded-window flag, the Z-bucket phrase). It answers one question:
      does the fitted slope predict direction out-of-sample?

USAGE
-----
    DATABASE_URL=postgresql+psycopg://... \
        python -m scripts.backtest_rate_sensitivity --since 2019-01-01

    # per-class, machine-readable:
    python -m scripts.backtest_rate_sensitivity --classes комфорт,бизнес --json

    # the #978b cross-checks: survivorship-free Source A + a detrend control:
    python -m scripts.backtest_rate_sensitivity --source both --detrend
"""

from __future__ import annotations

import argparse
import json
import logging
import math
from dataclasses import dataclass
from datetime import date
from pathlib import Path
from typing import Any

import numpy as np
from sqlalchemy import text
from sqlalchemy.orm import Session

# Reuse the §9.6 engine's PURE math (β / lag selection) and the Y-axis Δln
# helper, verbatim — the backtest must score the SAME functions production runs,
# not a re-implementation. Deferred import (see _import_engine) so `--help` and
# the pure-logic unit tests don't pull app.core.config.Settings, which
# fail-fasts when DATABASE_URL is unset.

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger("backtest_rate_sensitivity")

# ── Named constants ───────────────────────────────────────────────────────────

# Default lower bound of the backtest window. 2019-01 matches the solid daily
# key_rate history (macro_indicator key_rate spans 2019-01→2026-06) and lets
# Source B contribute its full ~102 distinct months.
_DEFAULT_SINCE: str = "2019-01-01"

# Fraction of the (time-ordered) aligned months used to FIT; the remainder is
# the held-out TEST window. 0.7 ≈ "fit on the older 70%, judge on the newer
# 30%" — enough train depth for best_lag's gate while leaving a test window.
_HOLDOUT_FRAC: float = 0.7

# Minimum aligned (train+test) months a tier must have before we backtest it.
# Below this the time-ordered split leaves too few test points for the
# directional hit-rate to mean anything → skip the tier with a printed note
# (never a silent drop). 18 ≈ "≥1 year to fit + a handful to test".
_MIN_BACKTEST_MONTHS: int = 18

# How far the EKB-wide OOS directional hit-rate must clear the 0.5 coin-flip
# baseline before the verdict calls the engine predictive. A small margin so a
# tiny test window can't flip the verdict on one lucky month.
_VERDICT_HITRATE_MARGIN: float = 0.05

# Source B premise filter — residential квартиры, the only segment §9.6 scores
# (mirrors sales_series._DEFAULT_PREMISE_KIND).
_PREMISE_KIND: str = "квартира"

# Sentinel for the EKB-wide (all-classes) tier in tables / JSON.
_EKB_WIDE: str = "EKB-wide"

# Series-source labels.
#   B = objective_lots.registration_date COUNT(*)  — long but survivorship-CONFOUNDED.
#   A = objective_corpus_room_month SUM(deals)     — short (~13 mo) but survivorship-FREE.
_SOURCE_B: str = "B"
_SOURCE_A: str = "A"
_SOURCE_BOTH: str = "both"
_SOURCES: tuple[str, ...] = (_SOURCE_B, _SOURCE_A)

# Minimum finite (>0) points _detrend_log needs to fit a line. Below this we
# can't separate trend from level, so we pass the log values through unchanged
# (the difference step then behaves exactly like the raw log_diff path).
_DETREND_MIN_POINTS: int = 3


def _import_engine() -> tuple[Any, Any, Any]:
    """Lazy import of the §9.6 engine's pure funcs + Δln helper.

    Returns ``(best_lag, ols_slope_r2, log_diff)``. Deferred so ``--help`` and
    the pure-metric unit tests don't import app.core.config.Settings (which
    fail-fasts without DATABASE_URL). Supports both
    ``python -m scripts.backtest_rate_sensitivity`` and stand-alone runs.
    """
    try:
        from app.services.forecasting.rate_sensitivity import best_lag, ols_slope_r2
        from app.services.forecasting.sales_series import log_diff
    except ImportError:  # pragma: no cover — fallback for adhoc invocation
        import sys

        sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
        from app.services.forecasting.rate_sensitivity import best_lag, ols_slope_r2
        from app.services.forecasting.sales_series import log_diff
    return best_lag, ols_slope_r2, log_diff


def _import_lags() -> tuple[int, ...]:
    """Lazy import of the engine's lag grid (_LAGS) — same deferral rationale."""
    try:
        from app.services.forecasting.rate_sensitivity import _LAGS
    except ImportError:  # pragma: no cover — fallback for adhoc invocation
        import sys

        sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
        from app.services.forecasting.rate_sensitivity import _LAGS
    return _LAGS


def _session() -> Session:
    """Lazy SessionLocal factory — see _import_engine for why it's deferred."""
    try:
        from app.core.db import SessionLocal
    except ImportError:  # pragma: no cover — fallback for adhoc invocation
        import sys

        sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
        from app.core.db import SessionLocal
    return SessionLocal()


# --------------------------------------------------------------------------- #
# Data carriers
# --------------------------------------------------------------------------- #


@dataclass(frozen=True)
class TierResult:
    """OOS backtest result for one tier (EKB-wide or a single class).

    Every numeric field is None when the tier was skipped (too few months) —
    ``skipped`` carries the reason. ``oos_hit_rate`` is the only trustworthy
    accuracy number; ``in_sample_r2`` is high by construction (see honesty
    block) and reported only for contrast.
    """

    tier: str
    source: str
    detrended: bool
    n_aligned: int
    n_train: int
    n_test: int
    train_lag: int | None
    train_beta: float | None
    in_sample_r2: float | None
    oos_hit_rate: float | None
    oos_signed_mae: float | None
    full_sample_lag: int | None
    lag_stable: bool
    skipped: str | None

    def as_dict(self) -> dict[str, Any]:
        return {
            "tier": self.tier,
            "source": self.source,
            "detrended": self.detrended,
            "n_aligned": self.n_aligned,
            "n_train": self.n_train,
            "n_test": self.n_test,
            "train_lag": self.train_lag,
            "train_beta": _round_or_none(self.train_beta, 4),
            "in_sample_r2": _round_or_none(self.in_sample_r2, 4),
            "oos_hit_rate": _round_or_none(self.oos_hit_rate, 4),
            "oos_signed_mae": _round_or_none(self.oos_signed_mae, 4),
            "full_sample_lag": self.full_sample_lag,
            "lag_stable": self.lag_stable,
            "skipped": self.skipped,
        }


def _round_or_none(value: float | None, digits: int) -> float | None:
    return round(value, digits) if value is not None else None


# --------------------------------------------------------------------------- #
# Pure logic — NO DB. Unit-tested in tests/scripts/test_backtest_rate_sensitivity
# on synthetic series (inject sales=f(rate@lag) → high OOS hit-rate; inject
# noise → hit-rate ≈ 0.5; time-ordered split correctness; thin-tier skip).
# --------------------------------------------------------------------------- #


def _rate_first_diff(rate_levels: list[float | None]) -> list[float | None]:
    """First difference of the key_rate level series: out[t] = r_t − r_{t-1}.

    Δ in percentage points. out[0] = None (no prior point); None if either of
    the two points is None. Mirrors rate_sensitivity._delta (kept local so the
    backtest's pure logic has no DB-importing dependency at module load). PURE.
    """
    out: list[float | None] = [None]
    for i in range(1, len(rate_levels)):
        cur = rate_levels[i]
        prev = rate_levels[i - 1]
        if cur is None or prev is None:
            out.append(None)
        else:
            out.append(float(cur) - float(prev))
    return out


def _detrend_log(values: list[float | int | None]) -> list[float | None]:
    """Linear-detrend the LOG of a units series → log-residuals. PURE (no DB).

    The survivorship control for #978b. We:
      1. Map each unit count to ``ln(units)``; None or ≤0 → None (ln undefined,
         same rule as ``sales_series.log_diff``).
      2. Fit ``ln(units) ~ a + b·month_index`` by least squares (numpy
         ``polyfit`` deg-1) over the FINITE points only, ``month_index`` = the
         original position 0..n-1 so gaps don't shift the trend.
      3. Return the residuals ``ln(units) − (a + b·month_index)`` at each finite
         index (None where the input was None/≤0). Output length = input length.

    A spurious monotone survivorship trend lands almost entirely in ``b`` and is
    subtracted out, so the downstream first-difference + regression can't be
    driven by it. The caller differences these residuals (they are already in
    log space) instead of calling ``log_diff`` again.

    Below ``_DETREND_MIN_POINTS`` finite points a line is not identifiable, so we
    PASS THROUGH the log values unchanged (residual == log value); differencing
    them then reproduces the raw ``log_diff`` path exactly. PURE.
    """
    logs: list[float | None] = []
    for v in values:
        if v is None:
            logs.append(None)
            continue
        vf = float(v)
        logs.append(math.log(vf) if vf > 0 else None)

    finite_idx = [i for i, lv in enumerate(logs) if lv is not None]
    if len(finite_idx) < _DETREND_MIN_POINTS:
        return logs  # not enough points to fit a trend → passthrough of logs

    xs = np.array([float(i) for i in finite_idx], dtype=float)
    ys = np.array([float(logs[i]) for i in finite_idx], dtype=float)  # type: ignore[arg-type]
    # Degenerate x-variance (all same index — impossible for ≥3 distinct idx but
    # guard anyway) → no trend to remove, passthrough.
    if float(np.ptp(xs)) == 0.0:
        return logs
    slope, intercept = np.polyfit(xs, ys, 1)

    out: list[float | None] = []
    for i, lv in enumerate(logs):
        if lv is None:
            out.append(None)
        else:
            out.append(float(lv) - (float(slope) * float(i) + float(intercept)))
    return out


def _time_ordered_split(n: int, holdout_frac: float) -> int:
    """Index where TEST begins for a time-ordered holdout of ``n`` months.

    Returns ``n_train`` — the count of the OLDEST months used to FIT; months
    ``[n_train:]`` are the held-out TEST window. NOT random / even-odd: a time
    series must train on the past and test on the future. ``holdout_frac`` is
    the TRAIN fraction; clamped so both halves keep ≥1 month when n ≥ 2. PURE.
    """
    if n <= 0:
        return 0
    n_train = round(n * holdout_frac)
    # Keep at least one month on each side once we have ≥2 months to split.
    n_train = max(1, min(n_train, n - 1)) if n >= 2 else n
    return n_train


def _shift_for_lag(x: list[float | None], lag: int) -> list[float | None]:
    """Right-shift the regressor by ``lag`` (y[t] ← x[t−lag]); leading = None.

    Same alignment best_lag uses internally: the rate change LEADS sales by
    ``lag`` months. Output is truncated to len(x) so it stays aligned to y.
    PURE.
    """
    shifted: list[float | None] = [None] * lag + list(x)
    return shifted[: len(x)]


def evaluate_oos(
    delta_sales: list[float | None],
    rate_deltas: list[float | None],
    *,
    holdout_frac: float = _HOLDOUT_FRAC,
) -> dict[str, Any]:
    """Time-ordered OOS backtest of the §9.6 β/lag fit. PURE (no DB).

    Steps:
      1. Split the aligned months time-ordered: fit on the oldest
         ``holdout_frac``, test on the newest remainder.
      2. Reuse ``best_lag`` on the TRAIN slice only → winning lag + β (TRAIN).
         best_lag applies the engine's gate (n≥min, R²≥min, slope<0); if no lag
         passes on TRAIN there is nothing to validate → empty result.
      3. For each TEST month predict Δln(sales) = β·Δrate[t−lag], strictly
         point-in-time: the lagged regressor is built on the FULL aligned rate
         series, so a test month at index t reads Δrate at t−lag which is at or
         before t — never the future. Score the SIGN vs the actual Δln(sales).
      4. Also refit ``best_lag`` on the FULL aligned series → full-sample lag
         (for the verdict's lag-stability check).

    Returns a dict with: n_aligned, n_train, n_test, train_lag, train_beta,
    in_sample_r2 (R² of the winning lag ON TRAIN — high by construction),
    oos_hit_rate (fraction of finite test months with matching sign; None if no
    scorable test month), oos_signed_mae (mean |predicted − actual| on test),
    full_sample_lag, lag_stable.
    """
    best_lag, _ols_slope_r2, _log_diff = _import_engine()

    n = len(delta_sales)
    n_train = _time_ordered_split(n, holdout_frac)
    n_test = n - n_train

    empty: dict[str, Any] = {
        "n_aligned": n,
        "n_train": n_train,
        "n_test": n_test,
        "train_lag": None,
        "train_beta": None,
        "in_sample_r2": None,
        "oos_hit_rate": None,
        "oos_signed_mae": None,
        "full_sample_lag": None,
        "lag_stable": False,
    }
    if n_train < 2 or n_test < 1:
        return empty

    train_sales = delta_sales[:n_train]
    train_rate = rate_deltas[:n_train]
    train_fit = best_lag(train_sales, train_rate)
    if train_fit is None:
        # No gated lag on TRAIN → the engine has nothing to validate here.
        return empty

    lag = int(train_fit["lag"])
    beta = float(train_fit["slope"])
    in_sample_r2 = float(train_fit["r2"])

    # Point-in-time prediction on TEST. The lagged regressor is built over the
    # FULL aligned series, then sliced to the test window: a test month at
    # absolute index t reads Δrate[t−lag] (≤ t), so no future leaks in.
    shifted_full = _shift_for_lag(rate_deltas, lag)
    hits = 0
    scored = 0
    abs_err_sum = 0.0
    for t in range(n_train, n):
        actual = delta_sales[t]
        x = shifted_full[t]
        if actual is None or x is None:
            continue
        predicted = beta * float(x)
        scored += 1
        abs_err_sum += abs(predicted - float(actual))
        # Directional hit: same sign. A flat actual (0.0) can't be directionally
        # predicted, so it never counts as a hit (only nonzero signs match).
        if (predicted > 0 and actual > 0) or (predicted < 0 and actual < 0):
            hits += 1

    oos_hit_rate = (hits / scored) if scored > 0 else None
    oos_signed_mae = (abs_err_sum / scored) if scored > 0 else None

    full_fit = best_lag(delta_sales, rate_deltas)
    full_lag = int(full_fit["lag"]) if full_fit is not None else None
    lag_stable = full_lag is not None and full_lag == lag

    return {
        "n_aligned": n,
        "n_train": n_train,
        "n_test": scored,  # report the number actually SCORED, not the raw span
        "train_lag": lag,
        "train_beta": beta,
        "in_sample_r2": in_sample_r2,
        "oos_hit_rate": oos_hit_rate,
        "oos_signed_mae": oos_signed_mae,
        "full_sample_lag": full_lag,
        "lag_stable": lag_stable,
    }


def align_series(
    sales_by_month: dict[date, int],
    rate_by_month: dict[date, float],
) -> tuple[list[date], list[int], list[float]]:
    """Inner-join the sold-units and key_rate monthly series by year-month. PURE.

    Returns ``(months, units, rates)`` ASC over the months present in BOTH
    series (a month with no rate or no sales row is dropped so Δ pairs stay
    meaningful). units/rates are aligned to months by index.
    """
    common = sorted(set(sales_by_month) & set(rate_by_month))
    months = list(common)
    units = [int(sales_by_month[m]) for m in common]
    rates = [float(rate_by_month[m]) for m in common]
    return months, units, rates


def _delta_sales_series(units: list[int], *, detrend: bool) -> list[float | None]:
    """Build the Δ(log-units) regressand for one tier. PURE (deferred import).

    Two variants, both ending in a Δ of log-space values ``evaluate_oos`` scores:
      • ``detrend=False`` — the production path: ``log_diff(units)`` = first
        difference of ``ln(units)``.
      • ``detrend=True``  — the #978b control: first linear-detrend ``ln(units)``
        (``_detrend_log``), THEN first-difference the residuals. We difference
        the residuals DIRECTLY (they are already in log space) rather than
        ``log_diff`` (which would re-take logs of residuals that may be ≤0).
    """
    if not detrend:
        _bl, _ols, log_diff = _import_engine()
        return log_diff(units)
    return _rate_first_diff(_detrend_log(units))


def backtest_tier(
    sales_by_month: dict[date, int],
    rate_by_month: dict[date, float],
    *,
    tier: str,
    source: str = _SOURCE_B,
    detrend: bool = False,
    holdout_frac: float = _HOLDOUT_FRAC,
    min_months: int = _MIN_BACKTEST_MONTHS,
) -> TierResult:
    """Build Δ-series for one tier, run the OOS backtest, wrap as TierResult.

    Aligns the tier's monthly sold-units to the monthly key_rate, computes the
    regressand (``log_diff`` raw, or Δ of linear-detrended ``ln`` when
    ``detrend`` — see ``_delta_sales_series``) and Δrate (first diff), then
    delegates to ``evaluate_oos``. ``source`` (B/A) and ``detrend`` are recorded
    on the result for labelling, not used in the math here. Tiers with fewer than
    ``min_months`` aligned months are SKIPPED (TierResult with ``skipped`` set,
    all metrics None) — no silent drop. PURE aside from the deferred engine
    import.
    """
    months, units, rates = align_series(sales_by_month, rate_by_month)
    n_aligned = len(months)
    if n_aligned < min_months:
        return TierResult(
            tier=tier,
            source=source,
            detrended=detrend,
            n_aligned=n_aligned,
            n_train=0,
            n_test=0,
            train_lag=None,
            train_beta=None,
            in_sample_r2=None,
            oos_hit_rate=None,
            oos_signed_mae=None,
            full_sample_lag=None,
            lag_stable=False,
            skipped=f"only {n_aligned} aligned months (< {min_months})",
        )

    delta_sales = _delta_sales_series(units, detrend=detrend)
    rate_deltas = _rate_first_diff([float(r) for r in rates])
    res = evaluate_oos(delta_sales, rate_deltas, holdout_frac=holdout_frac)

    return TierResult(
        tier=tier,
        source=source,
        detrended=detrend,
        n_aligned=res["n_aligned"],
        n_train=res["n_train"],
        n_test=res["n_test"],
        train_lag=res["train_lag"],
        train_beta=res["train_beta"],
        in_sample_r2=res["in_sample_r2"],
        oos_hit_rate=res["oos_hit_rate"],
        oos_signed_mae=res["oos_signed_mae"],
        full_sample_lag=res["full_sample_lag"],
        lag_stable=res["lag_stable"],
        skipped=None,
    )


def verdict(
    ekb: TierResult,
    *,
    margin: float = _VERDICT_HITRATE_MARGIN,
) -> dict[str, Any]:
    """Decide whether the EKB-wide tier shows OOS predictive value. PURE.

    The engine is a promotion CANDIDATE when, on the EKB-wide tier:
      • a gated lag was found and scored on a non-empty test window, AND
      • the OOS directional hit-rate beats the 0.5 coin-flip baseline by at
        least ``margin``, AND
      • the winning lag is the same on TRAIN and on the full-sample refit
        (lag stability — a lag that jumps between windows is not a signal).

    Returns ``{"promote": bool, "reason": str, "thin_warning": str | None}``.
    Honest: if the OOS test window is tiny the reason says so even when the
    hit-rate happens to clear the bar.
    """
    if ekb.skipped is not None:
        return {
            "promote": False,
            "reason": f"insufficient OOS signal — keep advisory ({ekb.skipped})",
            "thin_warning": None,
        }
    if ekb.oos_hit_rate is None or ekb.n_test < 1:
        return {
            "promote": False,
            "reason": (
                "insufficient OOS signal — keep advisory "
                "(no gated lag on TRAIN or empty test window)"
            ),
            "thin_warning": None,
        }

    beats_coin = ekb.oos_hit_rate >= 0.5 + margin
    thin_warning: str | None = None
    if ekb.n_test < min(_MIN_BACKTEST_MONTHS // 2, 6):
        thin_warning = (
            f"OOS test window is small (n_test={ekb.n_test}); the hit-rate's "
            "confidence is weak — treat the verdict as indicative, not proof."
        )

    if beats_coin and ekb.lag_stable:
        reason = (
            f"engine has OOS predictive value (candidate to promote from "
            f"advisory): EKB-wide OOS hit-rate={ekb.oos_hit_rate:.2f} > "
            f"0.5+{margin:.2f} and lag stable (lag={ekb.train_lag})"
        )
        return {"promote": True, "reason": reason, "thin_warning": thin_warning}

    bits: list[str] = []
    if not beats_coin:
        bits.append(f"hit-rate={ekb.oos_hit_rate:.2f} ≤ 0.5+{margin:.2f}")
    if not ekb.lag_stable:
        bits.append(f"lag unstable (train={ekb.train_lag}, full={ekb.full_sample_lag})")
    reason = "insufficient OOS signal — keep advisory (" + "; ".join(bits) + ")"
    return {"promote": False, "reason": reason, "thin_warning": thin_warning}


def tier_lift(ekb: TierResult, tier: TierResult) -> float | None:
    """OOS directional hit-rate lift of a class tier over EKB-wide. PURE.

    Returns ``tier.oos_hit_rate − ekb.oos_hit_rate`` (positive = the
    class-specific β beats the EKB-wide prior OOS). None if either side has no
    scorable OOS hit-rate.
    """
    if ekb.oos_hit_rate is None or tier.oos_hit_rate is None:
        return None
    return tier.oos_hit_rate - ekb.oos_hit_rate


# --------------------------------------------------------------------------- #
# DB layer — READ-ONLY SELECTs only.
# --------------------------------------------------------------------------- #

# Source B monthly sold-units by registration month (the one solid long series).
# COUNT(*) of lots whose deal registered in each month, EKB-wide or filtered to
# one class / district. Parameterised; psycopg3 CAST(:x AS type), NEVER :x::type.
# Survivorship caveat (only currently-listed lots) is documented in the module
# docstring — it biases OLDER months low.
_SOURCE_B_UNITS_SQL = text(
    """
    SELECT
        CAST(date_trunc('month', ol.registration_date) AS date) AS month,
        COUNT(*)                                                 AS units
      FROM objective_lots ol
     WHERE ol.premise_kind = CAST(:premise_kind AS text)
       AND ol.registration_date IS NOT NULL
       AND ol.registration_date >= CAST(:since AS date)
       AND (CAST(:cls AS text) IS NULL OR LOWER(ol.class) = LOWER(CAST(:cls AS text)))
       AND (CAST(:district AS text) IS NULL OR ol.district = CAST(:district AS text))
     GROUP BY 1
     ORDER BY 1
    """
)

# Source A monthly deal AGGREGATE — survivorship-FREE. Objective's corp_sum API
# reports deals registered per month per (corpus × room_bucket) regardless of
# whether the lot is still listed, so it does NOT undercount old months the way
# Source B does. room_bucket is aggregated away by the SUM/GROUP BY 1 unless a
# class filter is given (class is stored capitalised here — "Комфорт" — so we
# fold case to match the lowercase --classes input). report_month is already a
# month-first DATE; date_trunc is belt-and-braces. Only ~13 months deep on prod.
# Parameterised; psycopg3 CAST(:x AS type), NEVER :x::type.
_SOURCE_A_UNITS_SQL = text(
    """
    SELECT
        CAST(date_trunc('month', crm.report_month) AS date) AS month,
        SUM(crm.deals_total_count)                          AS units
      FROM objective_corpus_room_month crm
     WHERE crm.report_month >= CAST(:since AS date)
       AND (CAST(:cls AS text) IS NULL OR LOWER(crm.class) = LOWER(CAST(:cls AS text)))
     GROUP BY 1
     ORDER BY 1
    """
)

# Distinct classes present in Source A over the window (for --classes all on A).
_SOURCE_A_CLASSES_SQL = text(
    """
    SELECT DISTINCT LOWER(crm.class) AS cls
      FROM objective_corpus_room_month crm
     WHERE crm.report_month >= CAST(:since AS date)
       AND crm.class IS NOT NULL
     ORDER BY 1
    """
)

# Distinct classes present in Source B over the window (for --classes all).
_SOURCE_B_CLASSES_SQL = text(
    """
    SELECT DISTINCT LOWER(ol.class) AS cls
      FROM objective_lots ol
     WHERE ol.premise_kind = CAST(:premise_kind AS text)
       AND ol.registration_date IS NOT NULL
       AND ol.registration_date >= CAST(:since AS date)
       AND ol.class IS NOT NULL
     ORDER BY 1
    """
)


def load_sales_by_month(
    db: Session,
    *,
    since: str,
    obj_class: str | None,
    district: str | None,
) -> dict[date, int]:
    """Run the Source B monthly sold-units SELECT → {month1st: units}.

    READ-ONLY. ``obj_class``/``district`` None → no filter (EKB-wide). Months
    with no rows simply do not appear (the caller aligns on the intersection
    with the rate series, so absent months drop out naturally).
    """
    rows = db.execute(
        _SOURCE_B_UNITS_SQL,
        {
            "premise_kind": _PREMISE_KIND,
            "since": since,
            "cls": obj_class,
            "district": district,
        },
    ).all()
    out: dict[date, int] = {}
    for r in rows:
        if r[0] is None:
            continue
        out[r[0]] = int(r[1] or 0)
    return out


def load_classes(db: Session, *, since: str) -> list[str]:
    """Run the Source B distinct-classes SELECT → lowercase class list. READ-ONLY."""
    rows = db.execute(
        _SOURCE_B_CLASSES_SQL,
        {"premise_kind": _PREMISE_KIND, "since": since},
    ).all()
    return [r[0] for r in rows if r[0] is not None]


def load_sales_by_month_source_a(
    db: Session,
    *,
    since: str,
    obj_class: str | None,
) -> dict[date, int]:
    """Run the Source A monthly deal-aggregate SELECT → {month1st: units}.

    READ-ONLY. ``obj_class`` None → no class filter (room_bucket aggregated
    away). Survivorship-FREE (deals counted regardless of current listing).
    Months with no rows simply do not appear. No district filter — corp_sum
    aggregates are not district-resolved the way the lots snapshot is.
    """
    rows = db.execute(
        _SOURCE_A_UNITS_SQL,
        {"since": since, "cls": obj_class},
    ).all()
    out: dict[date, int] = {}
    for r in rows:
        if r[0] is None:
            continue
        out[r[0]] = int(r[1] or 0)
    return out


def load_classes_source_a(db: Session, *, since: str) -> list[str]:
    """Run the Source A distinct-classes SELECT → lowercase class list. READ-ONLY."""
    rows = db.execute(
        _SOURCE_A_CLASSES_SQL,
        {"since": since},
    ).all()
    return [r[0] for r in rows if r[0] is not None]


def load_rate_by_month(db: Session, *, since: str) -> dict[date, float]:
    """Monthly last-known key_rate → {month1st: rate}. READ-ONLY.

    Reuses ``forecasting.macro_series.get_monthly_macro`` (DISTINCT ON
    month-end last-known + LOCF carry-forward) rather than re-deriving the
    resample — the backtest must read the SAME monthly key_rate the engine
    does. ``months_back`` is computed from ``since`` so the macro grid covers
    the whole backtest window. Months with a None carried rate are dropped.
    """
    from app.services.forecasting.macro_series import (
        _month_start,
        get_monthly_macro,
    )

    since_date = date.fromisoformat(since)
    today = date.today()
    months_back = (today.year - since_date.year) * 12 + (today.month - since_date.month)
    macro = get_monthly_macro(db, months_back=max(0, months_back))
    floor = _month_start(since_date)
    out: dict[date, float] = {}
    for m in macro:
        if m.key_rate is None or m.month < floor:
            continue
        out[m.month] = float(m.key_rate)
    return out


# --------------------------------------------------------------------------- #
# Orchestration (READ-ONLY) + rendering
# --------------------------------------------------------------------------- #


def _load_sales(
    db: Session,
    *,
    source: str,
    since: str,
    obj_class: str | None,
    district: str | None,
) -> dict[date, int]:
    """Dispatch the monthly sold-units load to the right source. READ-ONLY.

    Source B uses ``objective_lots`` (premise+district filters). Source A uses
    ``objective_corpus_room_month`` (survivorship-free aggregate; no district
    filter — corp_sum aggregates are not district-resolved, so ``district`` is
    ignored for A and the caller is responsible for warning if it was set).
    """
    if source == _SOURCE_A:
        return load_sales_by_month_source_a(db, since=since, obj_class=obj_class)
    return load_sales_by_month(db, since=since, obj_class=obj_class, district=district)


def _load_classes_for(db: Session, *, source: str, since: str) -> list[str]:
    """Dispatch class auto-discovery to the right source. READ-ONLY."""
    if source == _SOURCE_A:
        return load_classes_source_a(db, since=since)
    return load_classes(db, since=since)


def run_backtest(
    db: Session,
    *,
    since: str,
    holdout_frac: float,
    classes: list[str] | None,
    district: str | None,
    source: str = _SOURCE_B,
    detrend: bool = False,
    rate_by_month: dict[date, float] | None = None,
) -> dict[str, Any]:
    """Drive ONE source/variant of the read-only backtest → results dict. No writes.

    Loads the monthly key_rate (or reuses ``rate_by_month`` when the caller has
    already loaded it once for several variants), then the EKB-wide and per-class
    sold-units series for ``source``, backtests each tier (``backtest_tier``,
    with ``detrend`` applied), and assembles the per-source verdict + per-tier
    OOS lifts.

    ``classes`` None → auto-discover every class present in the chosen source; an
    empty list → EKB-wide only. ``district`` narrows ALL tiers for Source B only
    (ignored for Source A).
    """
    if rate_by_month is None:
        rate_by_month = load_rate_by_month(db, since=since)
        logger.info("loaded key_rate months: %d (since=%s)", len(rate_by_month), since)

    if classes is None:
        classes = _load_classes_for(db, source=source, since=since)
        logger.info("source=%s auto-discovered classes: %s", source, classes)

    a_district_ignored = source == _SOURCE_A and district is not None
    eff_district = None if source == _SOURCE_A else district

    # EKB-wide tier (no class filter).
    ekb_sales = _load_sales(db, source=source, since=since, obj_class=None, district=eff_district)
    ekb = backtest_tier(
        ekb_sales,
        rate_by_month,
        tier=_EKB_WIDE,
        source=source,
        detrend=detrend,
        holdout_frac=holdout_frac,
    )
    logger.info(
        "source=%s detrend=%s EKB-wide: aligned=%d train=%d test=%d lag=%s hit_rate=%s",
        source,
        detrend,
        ekb.n_aligned,
        ekb.n_train,
        ekb.n_test,
        ekb.train_lag,
        ekb.oos_hit_rate,
    )

    tiers: list[TierResult] = []
    lifts: dict[str, float | None] = {}
    for cls in classes:
        cls_sales = _load_sales(
            db, source=source, since=since, obj_class=cls, district=eff_district
        )
        res = backtest_tier(
            cls_sales,
            rate_by_month,
            tier=cls,
            source=source,
            detrend=detrend,
            holdout_frac=holdout_frac,
        )
        tiers.append(res)
        lifts[cls] = tier_lift(ekb, res)
        logger.info(
            "source=%s tier=%s aligned=%d test=%d hit_rate=%s lift=%s skipped=%s",
            source,
            cls,
            res.n_aligned,
            res.n_test,
            res.oos_hit_rate,
            lifts[cls],
            res.skipped,
        )

    vd = verdict(ekb)
    return {
        "source": source,
        "detrended": detrend,
        "a_district_ignored": a_district_ignored,
        "params": {
            "since": since,
            "holdout_frac": holdout_frac,
            "district": district,
            "classes": classes,
            "source": source,
            "detrended": detrend,
            "min_backtest_months": _MIN_BACKTEST_MONTHS,
            "lags": list(_import_lags()),
        },
        "ekb_wide": ekb.as_dict(),
        "tiers": [t.as_dict() for t in tiers],
        "lifts": {k: _round_or_none(v, 4) for k, v in lifts.items()},
        "verdict": vd,
        "ekb_result": ekb,  # carried for the renderer (stripped from JSON)
        "tier_results": tiers,
    }


def _variant_label(source: str, detrend: bool) -> str:
    """Human label for a (source, detrend) run, e.g. 'B raw' / 'B detrended' / 'A raw'."""
    return f"{source} {'detrended' if detrend else 'raw'}"


def _plan_variants(sources: list[str], detrend: bool) -> list[tuple[str, bool]]:
    """Which (source, detrend) variants to run, in report order. PURE.

    For each requested source we always run the RAW variant (the reference). When
    ``--detrend`` is set we ALSO run the detrended variant of that source, so a
    single invocation can show ``B raw`` next to ``B detrended`` (the survivorship
    control) for the verdict's side-by-side comparison.
    """
    variants: list[tuple[str, bool]] = []
    for src in sources:
        variants.append((src, False))
        if detrend:
            variants.append((src, True))
    return variants


def cross_source_verdict(
    runs: list[dict[str, Any]],
    *,
    margin: float = _VERDICT_HITRATE_MARGIN,
    min_months: int = _MIN_BACKTEST_MONTHS,
) -> dict[str, Any]:
    """Compare the EKB-wide OOS verdict across variants (B raw / B detrended / A).

    The #978b question: is Source B's negative OOS verdict a SURVIVORSHIP
    ARTIFACT or a real ``no signal``? We line up each variant's EKB-wide
    OOS hit-rate vs the 0.5 coin-flip baseline and synthesise a conclusion:

      • If NO variant (B raw, B detrended, or survivorship-free A) clears
        coin-flip+margin → the negative verdict is corroborated as a real
        ``no signal``, not an artifact (the detrend + survivorship-free controls
        agree). Source A's thin-data caveat is attached when A drove a verdict.
      • If the detrended or survivorship-free variant DOES clear the bar while
        raw B did not → the raw verdict may have been a survivorship artifact;
        flag the variant that shows signal.

    PURE — operates on already-computed run dicts. Returns a dict with a
    ``lines`` list (rendered as-is) plus structured fields for JSON.
    """
    rows: list[dict[str, Any]] = []
    signal_variants: list[str] = []
    thin_variants: list[str] = []
    for run in runs:
        ekb: TierResult = run["ekb_result"]
        label = _variant_label(run["source"], run["detrended"])
        hr = ekb.oos_hit_rate
        scorable = ekb.skipped is None and hr is not None and ekb.n_test >= 1
        beats = bool(scorable and hr is not None and hr >= 0.5 + margin and ekb.lag_stable)
        thin = scorable and ekb.n_test < min(min_months // 2, 6)
        if beats:
            signal_variants.append(label)
        if run["source"] == _SOURCE_A and (thin or not scorable):
            thin_variants.append(label)
        rows.append(
            {
                "variant": label,
                "source": run["source"],
                "detrended": run["detrended"],
                "scorable": scorable,
                "oos_hit_rate": _round_or_none(hr, 4),
                "n_test": ekb.n_test,
                "lag_stable": ekb.lag_stable,
                "beats_coin": beats,
                "skipped": ekb.skipped,
            }
        )

    lines: list[str] = []
    lines.append("CROSS-SOURCE VERDICT (B raw vs B detrended vs A — #978b):")
    for r in rows:
        if not r["scorable"]:
            why = r["skipped"] or "no gated lag / empty test window"
            lines.append(f"  {r['variant']:<13} → not scorable ({why})")
        else:
            tag = "SIGNAL > coin-flip" if r["beats_coin"] else "no signal (≤ coin-flip)"
            lines.append(
                f"  {r['variant']:<13} → OOS_hit={_fmt_rate(r['oos_hit_rate'])} "
                f"(n_test={r['n_test']}, lag_stable={'yes' if r['lag_stable'] else 'no'}) "
                f"→ {tag}"
            )

    if signal_variants:
        conclusion = (
            "CONCLUSION: OOS signal above coin-flip appears in: "
            + ", ".join(signal_variants)
            + ". The §9.6 negative verdict on raw Source B may be a SURVIVORSHIP "
            "ARTIFACT — the control(s) above recover directional signal."
        )
        promote_any = True
    else:
        conclusion = (
            "CONCLUSION: NO variant (raw B, detrended B, or survivorship-free A) "
            "beats coin-flip+margin out-of-sample. The §9.6 negative verdict is a "
            "REAL 'no signal', NOT a survivorship artifact — detrending B and the "
            "survivorship-free Source A both agree. Keep advisory."
        )
        promote_any = False
    lines.append("  " + conclusion)

    thin_caveat: str | None = None
    if thin_variants:
        thin_caveat = (
            "Source A is statistically THIN (~13 months on prod). Treat any A row "
            "as an indicative cross-check only, never as proof — variant(s): "
            + ", ".join(thin_variants)
            + "."
        )
        lines.append(f"  !! {thin_caveat}")

    return {
        "rows": rows,
        "signal_variants": signal_variants,
        "promote_any": promote_any,
        "conclusion": conclusion,
        "thin_caveat": thin_caveat,
        "lines": lines,
    }


def run_all(
    db: Session,
    *,
    since: str,
    holdout_frac: float,
    classes: list[str] | None,
    district: str | None,
    sources: list[str],
    detrend: bool,
) -> dict[str, Any]:
    """Run every requested (source, detrend) variant + the cross-source verdict.

    Loads the monthly key_rate ONCE and reuses it across variants. ``sources`` is
    a subset of (B, A); ``detrend`` adds the detrended variant of each. No
    writes. Returns ``{"variants": [run, ...], "cross_verdict": {...}}``.
    """
    rate_by_month = load_rate_by_month(db, since=since)
    logger.info("loaded key_rate months: %d (since=%s)", len(rate_by_month), since)

    variants = _plan_variants(sources, detrend)
    runs: list[dict[str, Any]] = []
    for src, dt_flag in variants:
        runs.append(
            run_backtest(
                db,
                since=since,
                holdout_frac=holdout_frac,
                classes=classes,
                district=district,
                source=src,
                detrend=dt_flag,
                rate_by_month=rate_by_month,
            )
        )
    cross = cross_source_verdict(runs)
    return {"variants": runs, "cross_verdict": cross}


def _fmt_rate(v: float | None) -> str:
    return "  n/a" if v is None else f"{v:.3f}"


def _fmt_lag(v: int | None) -> str:
    return "n/a" if v is None else str(v)


_SOURCE_BLURB: dict[str, str] = {
    _SOURCE_B: "Source B (objective_lots.registration_date COUNT) — survivorship-CONFOUNDED.",
    _SOURCE_A: "Source A (objective_corpus_room_month SUM deals) — survivorship-FREE, ~13 mo.",
}


def render_table(results: dict[str, Any]) -> str:
    """Render ONE variant's backtest results as a plain-text stdout report."""
    params = results["params"]
    ekb: TierResult = results["ekb_result"]
    tiers: list[TierResult] = results["tier_results"]
    lifts: dict[str, Any] = results["lifts"]
    vd = results["verdict"]
    source = results["source"]
    detrended = results["detrended"]

    lines: list[str] = []
    lines.append("=" * 78)
    lines.append(
        f"BACKTEST [source {source}{' · detrended' if detrended else ''}]: "
        "§9.6 rate-sensitivity OOS validation"
    )
    lines.append("=" * 78)
    lines.append(
        f"since={params['since']}  holdout_frac={params['holdout_frac']}  "
        f"district={params['district'] or '(all)'}  lags={params['lags']}"
    )
    lines.append(_SOURCE_BLURB.get(source, source))
    if detrended:
        lines.append(
            "DETRENDED: ln(units) linearly detrended (residuals) BEFORE differencing — "
            "removes a spurious monotone (survivorship) trend so it can't drive β."
        )
    if results.get("a_district_ignored"):
        lines.append(
            "NOTE: --district was IGNORED for Source A (corp_sum aggregates are not "
            "district-resolved)."
        )
    lines.append("")

    header = (
        f"  {'tier':<12} {'src':>3} {'detr':>5} {'aligned':>7} {'train':>6} {'test':>5} "
        f"{'lag':>4} {'beta':>9} {'inR2':>7} {'OOS_hit':>8} {'OOS_MAE':>8} {'lift':>7} "
        f"{'stable':>7}"
    )
    lines.append(header)
    lines.append("  " + "-" * (len(header) - 2))
    lines.append(_fmt_tier_row(ekb, lift=None))
    for t in tiers:
        lines.append(_fmt_tier_row(t, lift=lifts.get(t.tier)))

    # Skips called out explicitly (no silent drop).
    skipped = [t for t in tiers if t.skipped is not None]
    if skipped:
        lines.append("")
        lines.append("SKIPPED tiers (too few aligned months):")
        for t in skipped:
            lines.append(f"  {t.tier:<12} {t.skipped}")

    # In-sample-vs-OOS honesty block (copy trade-in discipline).
    lines.append("")
    lines.append("HONESTY — in-sample vs out-of-sample:")
    lines.append("  inR2 is the TRAIN-window R² — high BY CONSTRUCTION (β is fit to minimise those")
    lines.append(
        "  residuals). It is NOT evidence the engine predicts. The only trustworthy number"
    )
    lines.append(
        "  is OOS_hit: the directional hit-rate on held-out future months the fit never saw,"
    )
    lines.append(
        "  scored strictly point-in-time. A coin flip scores ~0.50; the engine must beat it."
    )

    # Per-tier interpretation.
    lines.append("")
    lines.append("PER-TIER — does a class-specific β beat the EKB-wide prior OOS?")
    any_lift = False
    for t in tiers:
        lift = lifts.get(t.tier)
        if lift is None:
            continue
        any_lift = True
        verdict_word = "beats EKB-wide" if lift > 0 else "no lift over EKB-wide"
        lines.append(
            f"  {t.tier:<12} OOS_hit={_fmt_rate(t.oos_hit_rate)}  "
            f"lift={lift:+.3f}  → {verdict_word}"
        )
    if not any_lift:
        lines.append("  (no class tier had a scorable OOS hit-rate to compare)")

    # Per-variant verdict.
    lines.append("")
    lines.append("VERDICT (this variant):")
    lines.append(f"  {vd['reason']}")
    if vd.get("thin_warning"):
        lines.append(f"  !! {vd['thin_warning']}")

    lines.append("")
    if source == _SOURCE_A:
        lines.append("Caveats: Source A is survivorship-FREE but THIN (~13 mo) — usually too short")
        lines.append("to clear _MIN_BACKTEST_MONTHS; an indicative cross-check, not proof.")
    else:
        lines.append("Caveats: Source B survivorship undercounts OLD months (use --detrend / -A")
        lines.append("as controls); short Δ-series + holdout → small test window. β/lag CORE only.")
    lines.append("=" * 78)
    return "\n".join(lines)


def render_all(payload: dict[str, Any]) -> str:
    """Render every variant table then the cross-source verdict block."""
    blocks: list[str] = [render_table(run) for run in payload["variants"]]
    cross = payload["cross_verdict"]
    blocks.append("=" * 78)
    blocks.append("\n".join(cross["lines"]))
    blocks.append("=" * 78)
    return "\n\n".join(blocks)


def _fmt_tier_row(t: TierResult, *, lift: float | None) -> str:
    """Format one tier row for the table."""
    lift_s = "      -" if lift is None else f"{lift:+.3f}"
    detr_s = "yes" if t.detrended else "no"
    return (
        f"  {t.tier:<12} {t.source:>3} {detr_s:>5} {t.n_aligned:>7} {t.n_train:>6} "
        f"{t.n_test:>5} {_fmt_lag(t.train_lag):>4} {_fmt_beta(t.train_beta):>9} "
        f"{_fmt_rate(t.in_sample_r2):>7} {_fmt_rate(t.oos_hit_rate):>8} "
        f"{_fmt_rate(t.oos_signed_mae):>8} {lift_s:>7} "
        f"{('yes' if t.lag_stable else 'no'):>7}"
    )


def _fmt_beta(v: float | None) -> str:
    return "      n/a" if v is None else f"{v:+.4f}"


# --------------------------------------------------------------------------- #
# Entry point
# --------------------------------------------------------------------------- #


def _parse_classes(raw: str | None) -> list[str] | None:
    """Parse --classes: None/'all' → None (auto-discover); CSV → lowercase list.

    An empty string → [] (EKB-wide only). PURE.
    """
    if raw is None:
        return None
    raw = raw.strip()
    if raw == "":
        return []
    if raw.lower() == "all":
        return None
    return [c.strip().lower() for c in raw.split(",") if c.strip()]


def _parse_source(raw: str | None) -> list[str]:
    """Parse --source: B/A → that one source; both/None → [B, A]. PURE.

    Case-insensitive. Returns the ordered list of sources to run (B before A so
    the report leads with the long series). Unknown value → ValueError.
    """
    if raw is None:
        return list(_SOURCES)
    val = raw.strip().lower()
    if val in ("both", ""):
        return list(_SOURCES)
    if val == _SOURCE_B.lower():
        return [_SOURCE_B]
    if val == _SOURCE_A.lower():
        return [_SOURCE_A]
    raise ValueError(f"--source must be one of B, A, both (got {raw!r})")


def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    """argparse setup, factored out for testability."""
    p = argparse.ArgumentParser(
        description=(
            "READ-ONLY out-of-sample validation of the §9.6 rate-sensitivity engine (Forgejo #978)."
        ),
    )
    p.add_argument(
        "--since",
        default=_DEFAULT_SINCE,
        help=f"Lower bound (ISO date) of the backtest window (default {_DEFAULT_SINCE}).",
    )
    p.add_argument(
        "--source",
        default=_SOURCE_BOTH,
        help="Series source: 'B' (objective_lots, survivorship-confounded), 'A' "
        "(objective_corpus_room_month, survivorship-free, ~13 mo), or 'both' "
        f"(default '{_SOURCE_BOTH}').",
    )
    p.add_argument(
        "--detrend",
        action="store_true",
        help="Also run a DETRENDED variant of each source: linearly detrend "
        "ln(units) before differencing (removes a spurious monotone "
        "survivorship trend so it can't drive the regression).",
    )
    p.add_argument(
        "--holdout-frac",
        type=float,
        default=_HOLDOUT_FRAC,
        help=f"TRAIN fraction of the time-ordered split (default {_HOLDOUT_FRAC}). "
        "Fit on the oldest fraction, test on the newest remainder.",
    )
    p.add_argument(
        "--classes",
        default="all",
        help="Comma-separated classes to backtest, or 'all' to auto-discover "
        "(default 'all'). Empty → EKB-wide only.",
    )
    p.add_argument(
        "--district",
        default=None,
        help="Optional district filter applied to ALL tiers (default: all districts).",
    )
    p.add_argument(
        "--json",
        action="store_true",
        help="Emit machine-readable JSON instead of the text table.",
    )
    return p.parse_args(argv)


def _json_payload(payload: dict[str, Any]) -> dict[str, Any]:
    """Strip the renderer-only carriers from a run_all payload for JSON output."""
    variants = [
        {k: v for k, v in run.items() if k not in ("ekb_result", "tier_results")}
        for run in payload["variants"]
    ]
    cross = {k: v for k, v in payload["cross_verdict"].items() if k != "lines"}
    return {"variants": variants, "cross_verdict": cross}


def main(argv: list[str] | None = None) -> int:
    """CLI entry point.

    Returns 0 when at least one variant's EKB-wide tier was scorable (backtested,
    not skipped); 1 if every requested variant was skipped (too thin) — e.g.
    ``--source A`` alone on prod today (~13 months < _MIN_BACKTEST_MONTHS).
    """
    args = _parse_args(argv)
    classes = _parse_classes(args.classes)
    sources = _parse_source(args.source)
    logger.info(
        "backtest start: since=%s holdout_frac=%.2f classes=%s district=%s sources=%s detrend=%s",
        args.since,
        args.holdout_frac,
        "auto" if classes is None else classes,
        args.district,
        sources,
        args.detrend,
    )

    db = _session()
    try:
        payload = run_all(
            db,
            since=args.since,
            holdout_frac=args.holdout_frac,
            classes=classes,
            district=args.district,
            sources=sources,
            detrend=args.detrend,
        )
    finally:
        db.close()

    if args.json:
        print(json.dumps(_json_payload(payload), ensure_ascii=False, indent=2, default=str))
    else:
        print(render_all(payload))

    any_scorable = any(run["ekb_result"].skipped is None for run in payload["variants"])
    return 0 if any_scorable else 1


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main())