gendesign/backend/tests/scripts/test_backtest_rate_sensitivity.py

"""Unit tests for the read-only §9.6 rate-sensitivity backtest (Forgejo #978/#978b).

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)
  - _detrend_log         — (#978b) removes a known linear trend → flat residuals;
                           None/≤0 → None; <3 finite points → passthrough of logs
  - 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
  - evaluate_oos_almon   — (#978) Almon distributed-lag OOS evaluator: recovers a
                           known peak lag + negative long-run on a clean signal;
                           train fit IMMUNE to test-half corruption (no leakage);
                           predictor never reads a future rate index; same return
                           keys as evaluate_oos
  - _deseasonalize_units — (#979) seasonal factors fit on TRAIN months only,
                           applied point-in-time; recovers a known month pattern;
                           a TEST-window spike does NOT move the fitted factors
  - backtest_tier        — thin-tier skip; happy path; (#978b) detrended variant;
                           (#978) almon estimator path; (#979) deseasonalize path;
                           BACKWARD-COMPAT: default args == original raw best_lag
  - verdict / tier_lift  — promotion criterion, coin-flip baseline, lag stability
  - _variant_label / _plan_variants — raw/detrended/deseasonalized/Almon-ADL
                           labels + the per-flag variant plan (no all-combos)
  - cross_source_verdict — controls (detrended/A) + candidate methods
                           (deseasonalize #979, Almon-ADL #978) verdict + labels

DB is MOCKED (a fake session) only to assert the Source A/B SQL SHAPE — that it
uses CAST(:x AS type) and never the psycopg3-incompatible :x::type form, hits the
right table, and aggregates per the spec.

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


def _zero_drift_rate_levels(n: int, *, seed: int = 7) -> list[float]:
    """key_rate levels that OSCILLATE around a constant → Δrate has ~zero mean.

    Used for the detrend test: a monotone rate would give the injected signal a
    nonzero average slope that the linear detrend partly absorbs, leaving a
    constant Δ-offset the intercept-free OOS predictor can't model. With ~zero
    mean Δrate the detrend removes ONLY the spurious units trend, so the
    differenced residual cleanly reconstructs beta·Δrate[t-lag]. LCG jitter (not
    sin) keeps successive Δ weakly correlated so the true lag wins.
    """
    state = seed
    out: list[float] = []
    for _ in range(n):
        state = (state * 1103515245 + 12345) % 2147483648
        # Center on 10.0, symmetric jitter → no drift in the levels.
        out.append(10.0 + (state / 2147483648.0 - 0.5) * 3.0)
    return out


def _units_from_rate_with_trend(
    rate_levels: list[float],
    *,
    lag: int,
    beta: float,
    trend_per_month: float,
    base: float = 1000.0,
) -> list[int]:
    """Units carrying BOTH an injected rate signal AND a spurious log-linear trend.

    ln(u_t) = ln(base) + trend·t + Σ_{k≤t} beta·Δrate[k-lag]. The ``trend·t`` term
    is the survivorship-style monotone drift #978b's --detrend control removes; the
    Σ term is the real rate→sales signal. Detrending should subtract ~trend·t and
    leave the rate-driven residual whose Δ reconstructs beta·Δrate[t-lag].
    """
    rate_deltas = [0.0] + [rate_levels[i] - rate_levels[i - 1] for i in range(1, len(rate_levels))]
    signal_cum = 0.0
    units: list[int] = []
    for t in range(len(rate_levels)):
        if t > 0:
            src = rate_deltas[t - lag] if t - lag >= 0 else 0.0
            signal_cum += beta * src
        ln_u = math.log(base) + trend_per_month * t + signal_cum
        units.append(max(1, round(math.exp(ln_u))))
    return units


# --------------------------------------------------------------------------- #
# Almon distributed-lag synthetic helpers (#978) — MIRROR the proven
# construction in tests/services/forecasting/test_regression.py so the Almon
# evaluator is exercised on a signal the estimator demonstrably recovers. The
# regressor is a DIRECT LCG-jittered Δrate series (low cross-lag autocorrelation
# → the per-lag reconstruction is faithful); the regressand is a quadratic-shaped
# distributed lag the Almon deg-2 polynomial represents exactly.
# --------------------------------------------------------------------------- #


def _aperiodic_rate_deltas(n: int, *, seed: int = 13) -> list[float]:
    """Δrate series with APERIODIC (LCG) jitter → low autocorrelation across lags.

    Mirrors regression's ``_aperiodic_rate_deltas``: a periodic regressor would let
    false lags compete with the injected one; LCG jitter keeps successive Δ weakly
    correlated so the true lag shape wins. out[0] = 0.0 (finite from index 0); the
    Almon lag-matrix builder drops incomplete leading rows itself.
    """
    lvl = 10.0
    state = seed
    levels: list[float] = []
    for _ in range(n):
        state = (state * 1103515245 + 12345) % 2147483648
        lvl += 0.3 + (state / 2147483648.0 - 0.5) * 0.8
        levels.append(lvl)
    return [0.0] + [levels[i] - levels[i - 1] for i in range(1, n)]


def _hump_beta(max_lag: int, *, peak: int, scale: float = 0.06) -> list[float]:
    """A negative 'hump' lag shape peaking (in magnitude) at ``peak``. Mirror reg.

    |β_j| = scale − 0.012·(j−peak)² (floored at 0.005), all signs negative — the
    economically expected shape (rate ↑ → demand ↓, response builds then fades),
    representable by an Almon deg-2 polynomial so the fit recovers the peak.
    """
    betas: list[float] = []
    for j in range(max_lag + 1):
        mag = scale - 0.012 * (j - peak) ** 2
        betas.append(-max(0.005, mag))
    return betas


def _delta_sales_from_lag_shape(
    rate_deltas: list[float], beta: list[float], *, max_lag: int
) -> list[float | None]:
    """delta_sales[t] = Σ_j β_j·rate_deltas[t−j]; leading (t<max_lag) → None.

    The clean, noiseless distributed-lag regressand carrying the injected shape
    exactly. ``evaluate_oos_almon`` fits β on TRAIN and predicts the same Σ form,
    so on this construction the OOS directional hit-rate is ~1.0.
    """
    out: list[float | None] = [None] * max_lag
    for t in range(max_lag, len(rate_deltas)):
        out.append(sum(beta[j] * rate_deltas[t - j] for j in range(max_lag + 1)))
    return out


# --------------------------------------------------------------------------- #
# Seasonal synthetic helpers (#979) — a units series carrying a KNOWN
# month-of-year multiplicative pattern over ≥2 full years.
# --------------------------------------------------------------------------- #

# A known month-of-year seasonal pattern (multiplicative): summer peak, winter dip.
_KNOWN_SEASONAL: dict[int, float] = {
    1: 0.70,
    2: 0.80,
    3: 1.00,
    4: 1.10,
    5: 1.20,
    6: 1.30,
    7: 1.40,
    8: 1.20,
    9: 1.00,
    10: 0.90,
    11: 0.80,
    12: 0.60,
}


def _seasonal_units(
    months: list[dt.date], *, base: float = 1000.0, factor: dict[int, float] | None = None
) -> list[float]:
    """units[t] = base · factor[month_of(t)] — a clean known seasonal pattern.

    Float values (the deseasonalize path is float-math throughout: divide by
    factor then log_diff). With ≥2 full years the seasonal guard passes and
    ``seasonal_factors`` recovers ``factor`` up to the overall-mean normalisation.
    """
    fac = factor or _KNOWN_SEASONAL
    return [base * fac[m.month] for m in months]


# --------------------------------------------------------------------------- #
# _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 TestDetrendLog:
    def test_removes_known_linear_trend(self) -> None:
        # units = exp(a + b·t): a PURE log-linear trend → residuals must be ~0.
        a, b = 6.0, 0.05
        units = [round(math.exp(a + b * t)) for t in range(24)]
        resid = bt._detrend_log(units)
        assert all(r is not None for r in resid)
        # Rounding to int adds tiny noise, but residuals collapse near zero.
        assert max(abs(r) for r in resid) < 0.01  # type: ignore[arg-type, type-var]

    def test_residuals_isolate_signal_over_trend(self) -> None:
        # Trend + a single oscillation: after detrend the trend is gone and the
        # residual variance is dominated by the oscillation, not the drift.
        n = 30
        base_units = [math.exp(6.0 + 0.08 * t + 0.3 * math.sin(t)) for t in range(n)]
        units = [max(1, round(u)) for u in base_units]
        resid = bt._detrend_log(units)
        finite = [r for r in resid if r is not None]
        # Detrended series is NOT monotone (the drift dominated the raw logs).
        diffs = [finite[i] - finite[i - 1] for i in range(1, len(finite))]
        assert any(d > 0 for d in diffs) and any(d < 0 for d in diffs)

    def test_none_and_nonpositive_map_to_none(self) -> None:
        vals = [100, None, 0, -5, 120, 130, 140]
        resid = bt._detrend_log(vals)
        assert len(resid) == len(vals)
        assert resid[1] is None  # None in
        assert resid[2] is None  # 0 → ln undefined
        assert resid[3] is None  # negative → ln undefined
        # The finite positions stay finite.
        assert resid[0] is not None and resid[4] is not None

    def test_short_series_passthrough_is_logs(self) -> None:
        # <3 finite points → can't fit a line → passthrough of ln(values).
        vals = [10, 20]
        resid = bt._detrend_log(vals)
        assert resid[0] is not None and math.isclose(resid[0], math.log(10))
        assert resid[1] is not None and math.isclose(resid[1], math.log(20))

    def test_short_after_filtering_passthrough(self) -> None:
        # Only 2 finite points after dropping None/≤0 → passthrough of logs.
        vals = [None, 50, 0, 60]
        resid = bt._detrend_log(vals)
        assert resid[0] is None and resid[2] is None
        assert resid[1] is not None and math.isclose(resid[1], math.log(50))
        assert resid[3] is not None and math.isclose(resid[3], math.log(60))

    def test_length_preserved(self) -> None:
        vals = [100 + i for i in range(10)]
        assert len(bt._detrend_log(vals)) == 10


# --------------------------------------------------------------------------- #
# Look-ahead leakage fix (#978 Part A) — detrend trend fit on TRAIN months only,
# projected point-in-time onto test (never fit on train+test together).
# --------------------------------------------------------------------------- #


class TestDetrendNoLeakage:
    def test_train_only_fit_matches_manual_polyfit_on_train_slice(self) -> None:
        # With fit_n given, the trend (a, b) must be the polyfit of ONLY the
        # finite points in [0:fit_n] — the test months must not enter the fit.
        n, fit_n = 30, 20
        units = [max(1, round(math.exp(6.0 + 0.05 * t))) for t in range(n)]
        logs = [math.log(u) for u in units]
        # Manual train-only line.
        xs = list(range(fit_n))
        ys = logs[:fit_n]
        b, a = bt.np.polyfit(bt.np.array(xs, dtype=float), bt.np.array(ys, dtype=float), 1)
        resid = bt._detrend_log(units, fit_n=fit_n)
        # Every residual equals ln(u_t) − (a + b·t) with the TRAIN-fitted line,
        # INCLUDING the test months (the line is projected forward, not refit).
        for t in range(n):
            assert resid[t] is not None
            assert math.isclose(resid[t], logs[t] - (a + b * t), abs_tol=1e-9)  # type: ignore[arg-type]

    def test_test_points_do_not_shape_the_trend(self) -> None:
        # A BROKEN trend: gentle slope on train, steep slope on test. A full-sample
        # (leaky) fit is pulled UP by the steep test tail; a train-only fit is not.
        # So the residual at the LAST month must differ between the two — proving
        # the test observations leak into the leaky fit but not the train-only one.
        n, fit_n = 24, 16
        units: list[int] = []
        for t in range(n):
            slope = 0.02 if t < fit_n else 0.20  # trend break at fit_n
            base = 0.02 * min(t, fit_n)
            extra = 0.20 * max(0, t - fit_n)
            units.append(max(1, round(math.exp(6.0 + base + extra)) if t else round(math.exp(6.0))))
            _ = slope
        leaky = bt._detrend_log(units)  # fit_n=None → fit on train+test (leaks)
        safe = bt._detrend_log(units, fit_n=fit_n)
        # Last test month residual differs → the steep tail moved the leaky line
        # but not the train-only line.
        assert leaky[-1] is not None and safe[-1] is not None
        assert abs(leaky[-1] - safe[-1]) > 0.05  # type: ignore[operator]

    def test_fit_n_gates_passthrough_on_train_point_count(self) -> None:
        # Plenty of finite points overall, but only 2 (< _DETREND_MIN_POINTS) fall
        # inside the TRAIN window → a line is not identifiable on TRAIN → passthrough
        # of the logs (residual == ln(value)), exactly like the raw log_diff path.
        units = [10, 20] + [30 + i for i in range(10)]  # 12 finite, fit_n=2
        resid = bt._detrend_log(units, fit_n=2)
        assert resid[0] is not None and math.isclose(resid[0], math.log(10))
        assert resid[1] is not None and math.isclose(resid[1], math.log(20))
        # Passthrough applies to ALL positions (no trend was removed anywhere).
        assert resid[2] is not None and math.isclose(resid[2], math.log(30))

    def test_backtest_tier_detrend_fits_train_only(self) -> None:
        # End-to-end: backtest_tier must pass n_train as fit_n. We assert the
        # detrended regressand it builds equals the one from a TRAIN-only detrend,
        # and is NOT equal to the leaky full-sample detrend (when they differ).
        n = 40
        ms = _months(n)
        # Trend-confounded units with a real lag-2 signal (#978b-style series).
        rate = _zero_drift_rate_levels(n, seed=5)
        units = _units_from_rate_with_trend(rate, lag=2, beta=-0.06, trend_per_month=0.09)
        sales = {ms[i]: units[i] for i in range(n)}
        rate_by = {ms[i]: rate[i] for i in range(n)}

        # What backtest_tier should build internally (train-only fit).
        n_train = bt._time_ordered_split(n, 0.7)
        expected = bt._delta_sales_series(units, detrend=True, fit_n=n_train)
        leaky = bt._delta_sales_series(units, detrend=True, fit_n=None)

        # Run the tier and reconstruct its regressand path via the same helper to
        # confirm n_train is threaded through (the public API has no hook, so we
        # assert the train-only and full-sample series genuinely differ — i.e. the
        # fix is observable — and that the tier still produces a scored result).
        res = bt.backtest_tier(sales, rate_by, tier=bt._EKB_WIDE, detrend=True, holdout_frac=0.7)
        assert res.skipped is None
        assert res.detrended is True
        # The two regressands must differ somewhere in the test region (leakage is
        # observable), so the train-only fix is a real behavioural change.
        assert any(
            e is not None and lk is not None and abs(e - lk) > 1e-9
            for e, lk in zip(expected[n_train:], leaky[n_train:], strict=False)
        )

    def test_no_leakage_oos_hit_rate_not_above_leaky(self) -> None:
        # The core claim: look-ahead leakage INFLATES the detrended OOS hit-rate.
        # On a trend-confounded series, the train-only (correct) detrend must give
        # an OOS hit-rate ≤ the leaky full-sample detrend. We compare evaluate_oos
        # on both regressands over the SAME aligned series.
        n = 48
        rate = _zero_drift_rate_levels(n, seed=11)
        units = _units_from_rate_with_trend(rate, lag=2, beta=-0.05, trend_per_month=0.07)
        rate_deltas = bt._rate_first_diff(rate)
        n_train = bt._time_ordered_split(n, 0.7)

        safe_sales = bt._delta_sales_series(units, detrend=True, fit_n=n_train)
        leaky_sales = bt._delta_sales_series(units, detrend=True, fit_n=None)
        safe = bt.evaluate_oos(safe_sales, rate_deltas, holdout_frac=0.7)
        leaky = bt.evaluate_oos(leaky_sales, rate_deltas, holdout_frac=0.7)

        # Both should find a gated lag here; if either is None the inequality is
        # vacuously fine (no inflation possible). When both score, leakage may only
        # help (or tie) the leaky run — it must never make the corrected run higher.
        if safe["oos_hit_rate"] is not None and leaky["oos_hit_rate"] is not None:
            assert safe["oos_hit_rate"] <= leaky["oos_hit_rate"] + 1e-9


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)


# --------------------------------------------------------------------------- #
# evaluate_oos_almon (#978) — the new Almon distributed-lag OOS evaluator
# --------------------------------------------------------------------------- #


class TestEvaluateOosAlmon:
    def test_recovers_known_distributed_lag(self) -> None:
        # Clean noiseless distributed lag with a quadratic hump peaking at lag 2.
        # The Almon deg-2 fit on TRAIN must recover that peak and a negative
        # long-run multiplier, and predict direction OOS ~perfectly (clean signal).
        max_lag = 6
        n = 72
        rate_deltas = _aperiodic_rate_deltas(n)
        beta = _hump_beta(max_lag, peak=2)
        delta_sales = _delta_sales_from_lag_shape(rate_deltas, beta, max_lag=max_lag)

        res = bt.evaluate_oos_almon(delta_sales, rate_deltas, holdout_frac=0.7)
        # train_lag = the fitted peak-|β_j| lag; matches the injected peak (±0).
        assert res["train_lag"] == 2
        # "train_beta" reports the long-run Σβ multiplier — negative here.
        assert res["train_beta"] is not None and res["train_beta"] < 0
        # Clean noiseless construction → directional hit-rate clearly beats coin.
        assert res["oos_hit_rate"] is not None and res["oos_hit_rate"] > 0.5
        assert res["oos_hit_rate"] >= 0.9
        # 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: the full-sample refit finds the same peak lag.
        assert res["full_sample_lag"] == 2
        assert res["lag_stable"] is True

    def test_recovers_different_peak_lag(self) -> None:
        # Shift the injected peak to lag 4 → the Almon fit must track it.
        max_lag = 6
        n = 80
        rate_deltas = _aperiodic_rate_deltas(n, seed=29)
        beta = _hump_beta(max_lag, peak=4)
        delta_sales = _delta_sales_from_lag_shape(rate_deltas, beta, max_lag=max_lag)
        res = bt.evaluate_oos_almon(delta_sales, rate_deltas, holdout_frac=0.7)
        assert res["train_lag"] == 4
        assert res["oos_hit_rate"] is not None and res["oos_hit_rate"] >= 0.9

    def test_no_look_ahead_leakage_train_fit_immune_to_test_corruption(self) -> None:
        # Build a clean signal, then corrupt ONLY the test-half delta_sales (flip
        # sign + scale + offset). The TRAIN fit cannot see the test window, so
        # train_lag / train_beta / in_sample_r2 must be byte-identical to the
        # uncorrupted run; only the OOS score may move.
        max_lag = 6
        n = 72
        rate_deltas = _aperiodic_rate_deltas(n)
        beta = _hump_beta(max_lag, peak=2)
        clean_sales = _delta_sales_from_lag_shape(rate_deltas, beta, max_lag=max_lag)

        n_train = bt._time_ordered_split(n, 0.7)
        corrupt_sales: list[float | None] = list(clean_sales)
        for t in range(n_train, n):
            v = corrupt_sales[t]
            if v is not None:
                corrupt_sales[t] = -v * 5.0 + 0.123  # arbitrary test-only corruption

        clean = bt.evaluate_oos_almon(clean_sales, rate_deltas, holdout_frac=0.7)
        corrupt = bt.evaluate_oos_almon(corrupt_sales, rate_deltas, holdout_frac=0.7)

        # TRAIN fit identical — the corruption is entirely in the held-out window.
        assert clean["train_lag"] == corrupt["train_lag"]
        assert clean["train_beta"] is not None and corrupt["train_beta"] is not None
        assert math.isclose(clean["train_beta"], corrupt["train_beta"], rel_tol=0, abs_tol=1e-12)
        assert clean["in_sample_r2"] is not None and corrupt["in_sample_r2"] is not None
        assert math.isclose(
            clean["in_sample_r2"], corrupt["in_sample_r2"], rel_tol=0, abs_tol=1e-12
        )
        # The OOS hit-rate DID respond to the corruption (flipped signs miss) —
        # proving the test window is actually scored, not ignored.
        assert clean["oos_hit_rate"] is not None and corrupt["oos_hit_rate"] is not None
        assert corrupt["oos_hit_rate"] < clean["oos_hit_rate"]

    def test_point_in_time_predictor_never_reads_future_rate(self) -> None:
        # Structural no-future-leak assertion: the per-lag shifted views the
        # evaluator reads at a test index t are _shift_for_lag(rate_deltas, j),
        # whose element at t equals the ORIGINAL rate_deltas[t-j] (≤ t) — never an
        # index > t. We assert this for every lag j across every test month.
        max_lag = 6
        n = 60
        rate_deltas = _aperiodic_rate_deltas(n)
        beta = _hump_beta(max_lag, peak=2)
        delta_sales = _delta_sales_from_lag_shape(rate_deltas, beta, max_lag=max_lag)
        res = bt.evaluate_oos_almon(delta_sales, rate_deltas, holdout_frac=0.7)
        n_train = res["n_train"]
        for j in range(max_lag + 1):
            shifted = bt._shift_for_lag(rate_deltas, j)
            for t in range(n_train, n):
                # The value the predictor uses at (t, lag j) is rate_deltas[t-j],
                # which is at or before t (None when t-j < 0). It is NEVER t+k.
                if shifted[t] is not None:
                    assert t - j >= 0
                    assert shifted[t] == rate_deltas[t - j]

    def test_skips_test_month_with_incomplete_lag_profile(self) -> None:
        # A None in the rate series punches a hole: a test month whose full lag
        # profile can't be formed is skipped (not fabricated). With one rate hole
        # near the test boundary, the evaluator still scores the remaining months
        # and never crashes / never counts the holed month.
        max_lag = 6
        n = 72
        rate_deltas: list[float | None] = list(_aperiodic_rate_deltas(n))
        beta = _hump_beta(max_lag, peak=2)
        delta_sales = _delta_sales_from_lag_shape(
            [r if r is not None else 0.0 for r in rate_deltas], beta, max_lag=max_lag
        )
        n_train = bt._time_ordered_split(n, 0.7)
        # Punch a hole in a test-window rate delta → the months that read it via
        # any lag j become unscorable.
        hole = n_train + 2
        rate_deltas[hole] = None
        res = bt.evaluate_oos_almon(delta_sales, rate_deltas, holdout_frac=0.7)
        # Still produced a result, fewer scored months than the raw test span.
        assert res["oos_hit_rate"] is not None
        assert res["n_test"] <= n - n_train
        assert math.isfinite(res["oos_signed_mae"])

    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_almon([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_infeasible_fit_returns_empty(self) -> None:
        # Too few aligned points for the Almon fit (< _MIN_FIT_OBS usable rows) →
        # fit_almon_dl returns None → empty result, no crash.
        n = 20  # > min split but Almon needs more usable rows after max_lag drop
        rate_deltas = _aperiodic_rate_deltas(n)
        # Flat regressand → zero-variance / infeasible fit on the train slice.
        delta_sales: list[float | None] = [None] + [0.0] * (n - 1)
        res = bt.evaluate_oos_almon(delta_sales, rate_deltas, holdout_frac=0.7)
        assert res["train_lag"] is None
        assert res["oos_hit_rate"] is None

    def test_return_dict_has_same_keys_as_evaluate_oos(self) -> None:
        # backtest_tier wraps both evaluators identically → identical key sets.
        max_lag = 6
        n = 60
        rate_deltas = _aperiodic_rate_deltas(n)
        beta = _hump_beta(max_lag, peak=2)
        delta_sales = _delta_sales_from_lag_shape(rate_deltas, beta, max_lag=max_lag)
        almon = bt.evaluate_oos_almon(delta_sales, rate_deltas, holdout_frac=0.7)
        # evaluate_oos on the same arrays (best_lag) for a key-set comparison.
        bl = bt.evaluate_oos(delta_sales, rate_deltas, holdout_frac=0.7)
        assert set(almon.keys()) == set(bl.keys())


# --------------------------------------------------------------------------- #
# Deseasonalization (#979) — month-of-year factors recovered + TRAIN-only fit
# --------------------------------------------------------------------------- #


class TestDeseasonalize:
    def test_recovers_known_seasonal_pattern(self) -> None:
        # units = base · known_factor[month] over 3 full years → seasonal_factors
        # must recover the known pattern (up to overall-mean normalisation) and
        # deseasonalize_values must flatten the month-means to ~equal.
        seasonal_factors, deseasonalize_values = bt._import_normalize()
        n = 36  # 3 full years
        ms = _months(n)
        units = _seasonal_units(ms)
        adj = seasonal_factors(ms, units)
        assert adj.applied is True
        assert adj.n_full_years == 3
        # Expected normalised factor = known[m] / mean(known).
        overall = sum(_KNOWN_SEASONAL.values()) / 12.0
        for m in range(1, 13):
            expected = _KNOWN_SEASONAL[m] / overall
            assert math.isclose(adj.factors[m], expected, abs_tol=1e-9)
        # Deseasonalized month-means collapse to a single value (pattern removed).
        des = deseasonalize_values(ms, units, adj.factors)
        by_month: dict[int, list[float]] = {}
        for d, v in zip(ms, des, strict=False):
            assert v is not None
            by_month.setdefault(d.month, []).append(v)
        means = [sum(vs) / len(vs) for vs in by_month.values()]
        assert max(means) - min(means) < 1e-6

    def test_factors_fit_on_train_only_immune_to_test_spike(self) -> None:
        # Insert an EXTREME spike in a TEST-window month and assert the seasonal
        # factors fit on the TRAIN slice are UNCHANGED vs the no-spike series. The
        # train/test boundary is _time_ordered_split — exactly what
        # _deseasonalize_units slices to.
        seasonal_factors, _deseason = bt._import_normalize()
        n = 48
        ms = _months(n)
        units = _seasonal_units(ms)
        n_train = bt._time_ordered_split(n, 0.7)

        clean = seasonal_factors(ms[:n_train], units[:n_train])
        spiked_units = list(units)
        spiked_units[n - 1] = spiked_units[n - 1] * 100.0  # extreme TEST-window spike
        spiked = seasonal_factors(ms[:n_train], spiked_units[:n_train])

        for m in range(1, 13):
            assert math.isclose(clean.factors[m], spiked.factors[m], abs_tol=1e-12)
        # Sanity: a LEAKY full-series fit WOULD have moved (the spike is real) —
        # so the train-only slice is what protects us, not a no-op.
        full_clean = seasonal_factors(ms, units)
        full_spiked = seasonal_factors(ms, spiked_units)
        assert any(abs(full_clean.factors[m] - full_spiked.factors[m]) > 1e-9 for m in range(1, 13))

    def test_deseasonalize_units_helper_uses_time_ordered_boundary(self) -> None:
        # The backtest helper _deseasonalize_units must fit factors on months[:fit_n]
        # ONLY. We feed fit_n = _time_ordered_split and confirm the regressand it
        # builds equals a manual TRAIN-fit-then-full-apply-then-log_diff, and is
        # NOT equal to a leaky full-sample-fit version (when they differ).
        seasonal_factors, deseasonalize_values = bt._import_normalize()
        _bl, _ols, log_diff = bt._import_engine()
        n = 48
        ms = _months(n)
        units_f = _seasonal_units(ms)
        units = [max(1, round(v)) for v in units_f]
        n_train = bt._time_ordered_split(n, 0.7)

        # Spike a TEST-window month so train-fit and full-fit factors differ.
        units[n - 1] = units[n - 1] * 50

        got = bt._deseasonalize_units(ms, units, fit_n=n_train)

        train_factors = seasonal_factors(ms[:n_train], units[:n_train]).factors
        expected = log_diff(deseasonalize_values(ms, units, train_factors))
        full_factors = seasonal_factors(ms, units).factors
        leaky = log_diff(deseasonalize_values(ms, units, full_factors))

        # The helper matches the TRAIN-only path exactly.
        assert len(got) == len(expected)
        for g, e in zip(got, expected, strict=False):
            assert (g is None and e is None) or (
                g is not None and e is not None and math.isclose(g, e, abs_tol=1e-12)
            )
        # And the train-only vs leaky paths genuinely differ (fix is observable).
        assert any(
            g is not None and lk is not None and abs(g - lk) > 1e-9
            for g, lk in zip(got, leaky, strict=False)
        )


# --------------------------------------------------------------------------- #
# 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

    def test_records_source_and_detrended_flags(self) -> None:
        # The TierResult carries the source label and detrend flag for the table.
        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, source=bt._SOURCE_A, detrend=True)
        assert res.source == bt._SOURCE_A
        assert res.detrended is True

    def test_detrended_recovers_signal_masked_by_trend(self) -> None:
        # Units carry a strong spurious upward (survivorship-like) trend PLUS a
        # real rate signal at lag 2. After --detrend strips the trend, the
        # differenced residual must still reconstruct the negative-β lag and
        # predict direction OOS well above a coin flip. We use a ~zero-drift rate
        # so the linear detrend removes ONLY the units trend, not the signal.
        n = 54
        ms = _months(n)
        rate = _zero_drift_rate_levels(n)
        units = _units_from_rate_with_trend(rate, lag=2, beta=-0.06, trend_per_month=0.08)
        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, detrend=True, holdout_frac=0.7)
        assert res.detrended is True
        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 and res.oos_hit_rate >= 0.8

    def test_detrend_strips_trend_raw_path_does_not(self) -> None:
        # Same trended+signal series: the RAW path's TRAIN fit is dominated by the
        # spurious monotone trend (Δln has a large positive constant from the
        # trend), so the gate either rejects (slope≥0) or the OOS direction is
        # poor; the DETRENDED path recovers the lag-2 signal. This is the #978b
        # premise: detrending changes the verdict on a trend-confounded series.
        n = 54
        ms = _months(n)
        rate = _zero_drift_rate_levels(n, seed=21)
        units = _units_from_rate_with_trend(rate, lag=2, beta=-0.06, trend_per_month=0.10)
        sales = {ms[i]: units[i] for i in range(n)}
        rate_by = {ms[i]: rate[i] for i in range(n)}
        raw = bt.backtest_tier(sales, rate_by, tier=bt._EKB_WIDE, detrend=False)
        detr = bt.backtest_tier(sales, rate_by, tier=bt._EKB_WIDE, detrend=True)
        # Detrended recovers a clean negative-β lag-2 fit.
        assert detr.train_lag == 2 and detr.train_beta is not None and detr.train_beta < 0
        # Raw is degraded by the trend: either no gated lag (None) or a weaker
        # OOS hit-rate than the detrended variant.
        if raw.oos_hit_rate is not None and detr.oos_hit_rate is not None:
            assert detr.oos_hit_rate >= raw.oos_hit_rate

    def test_records_deseasonalized_and_estimator_flags(self) -> None:
        # The TierResult carries the new deseasonalize flag and estimator label.
        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, deseasonalize=True, estimator=bt._ESTIMATOR_ALMON
        )
        assert res.deseasonalized is True
        assert res.estimator == bt._ESTIMATOR_ALMON
        d = res.as_dict()
        assert d["deseasonalized"] is True
        assert d["estimator"] == bt._ESTIMATOR_ALMON

    def test_almon_estimator_path_runs(self) -> None:
        # estimator="almon" routes backtest_tier to evaluate_oos_almon. On a clean
        # distributed-lag series it recovers the peak lag and scores OOS well.
        max_lag = 6
        n = 72
        ms = _months(n)
        rate_deltas = _aperiodic_rate_deltas(n)
        # Reconstruct rate LEVELS from the deltas so align_series has a rate series;
        # the tier re-differences them → the same rate_deltas reach the evaluator.
        rate_levels = [10.0]
        for j in range(1, n):
            rate_levels.append(rate_levels[-1] + rate_deltas[j])
        beta = _hump_beta(max_lag, peak=2)
        delta_sales = _delta_sales_from_lag_shape(rate_deltas, beta, max_lag=max_lag)
        # Turn the Δln signal into a units series (cumulative exp) so the tier's
        # log_diff(units) reproduces delta_sales on the finite region.
        ln_u = math.log(1000.0)
        units: list[int] = [round(math.exp(ln_u))]
        for t in range(1, n):
            step = delta_sales[t] if delta_sales[t] is not None else 0.0
            ln_u += step
            units.append(max(1, round(math.exp(ln_u))))
        sales = {ms[i]: units[i] for i in range(n)}
        rate_by = {ms[i]: rate_levels[i] for i in range(n)}
        res = bt.backtest_tier(
            sales, rate_by, tier=bt._EKB_WIDE, estimator=bt._ESTIMATOR_ALMON, holdout_frac=0.7
        )
        assert res.skipped is None
        assert res.estimator == bt._ESTIMATOR_ALMON
        assert res.train_lag == 2
        assert res.oos_hit_rate is not None and res.oos_hit_rate >= 0.8

    def test_deseasonalize_path_runs_and_uses_train_only_fit(self) -> None:
        # deseasonalize=True routes through _deseasonalize_units; the regressand it
        # builds must equal a TRAIN-only-fit reconstruction (no leakage) and the
        # tier still produces a scored result.
        seasonal_factors, deseasonalize_values = bt._import_normalize()
        _bl, _ols, log_diff = bt._import_engine()
        n = 48
        ms = _months(n)
        # Seasonal units with a mild rate-driven drift so a lag can gate.
        rate = _aperiodic_rate_levels(n)
        rate_deltas = bt._rate_first_diff(rate)
        ln_u = math.log(1000.0)
        units: list[int] = [round(math.exp(ln_u) * _KNOWN_SEASONAL[ms[0].month])]
        for t in range(1, n):
            src = rate_deltas[t - 2] if t - 2 >= 1 and rate_deltas[t - 2] is not None else 0.0
            ln_u += -0.04 * src
            units.append(max(1, round(math.exp(ln_u) * _KNOWN_SEASONAL[ms[t].month])))
        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, deseasonalize=True, holdout_frac=0.7
        )
        assert res.skipped is None
        assert res.deseasonalized is True

        # The regressand the tier built equals a TRAIN-only-fit reconstruction.
        months_aligned, units_aligned, _rates = bt.align_series(sales, rate_by)
        n_train = bt._time_ordered_split(len(months_aligned), 0.7)
        train_factors = seasonal_factors(months_aligned[:n_train], units_aligned[:n_train]).factors
        expected = log_diff(deseasonalize_values(months_aligned, units_aligned, train_factors))
        got = bt._deseasonalize_units(months_aligned, units_aligned, fit_n=n_train)
        for g, e in zip(got, expected, strict=False):
            assert (g is None and e is None) or (
                g is not None and e is not None and math.isclose(g, e, abs_tol=1e-12)
            )

    def test_backward_compat_defaults_unchanged(self) -> None:
        # The CRITICAL back-compat check: a default backtest_tier call (no
        # deseasonalize, estimator=best_lag) must produce the SAME metric fields
        # as the pre-change raw path. We pin every metric to an explicit raw
        # best_lag run and confirm the new descriptor fields default correctly.
        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)
        # New descriptor fields default to the production raw path.
        assert res.deseasonalized is False
        assert res.estimator == bt._ESTIMATOR_BEST_LAG
        assert res.detrended is False

        # Metric fields equal a direct evaluate_oos (best_lag) on the same arrays —
        # i.e. the default path is byte-identical to the original implementation.
        n_train = bt._time_ordered_split(n, 0.7)
        delta_sales = bt._delta_sales_series(units, detrend=False, fit_n=n_train)
        rate_deltas = bt._rate_first_diff([float(r) for r in rate])
        direct = bt.evaluate_oos(delta_sales, rate_deltas, holdout_frac=0.7)
        assert res.train_lag == direct["train_lag"]
        assert res.train_beta == direct["train_beta"]
        assert res.in_sample_r2 == direct["in_sample_r2"]
        assert res.oos_hit_rate == direct["oos_hit_rate"]
        assert res.oos_signed_mae == direct["oos_signed_mae"]
        assert res.full_sample_lag == direct["full_sample_lag"]
        assert res.lag_stable == direct["lag_stable"]


# --------------------------------------------------------------------------- #
# verdict / tier_lift
# --------------------------------------------------------------------------- #


def _tier(
    *,
    tier: str = bt._EKB_WIDE,
    source: str = bt._SOURCE_B,
    detrended: bool = False,
    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,
        source=source,
        detrended=detrended,
        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("Комфорт, Бизнес ,премиум") == ["комфорт", "бизнес", "премиум"]


# --------------------------------------------------------------------------- #
# _parse_source / _plan_variants (#978b)
# --------------------------------------------------------------------------- #


class TestParseSource:
    def test_both_and_default(self) -> None:
        assert bt._parse_source("both") == [bt._SOURCE_B, bt._SOURCE_A]
        assert bt._parse_source(None) == [bt._SOURCE_B, bt._SOURCE_A]
        assert bt._parse_source("") == [bt._SOURCE_B, bt._SOURCE_A]

    def test_single_source_case_insensitive(self) -> None:
        assert bt._parse_source("B") == [bt._SOURCE_B]
        assert bt._parse_source("b") == [bt._SOURCE_B]
        assert bt._parse_source("A") == [bt._SOURCE_A]
        assert bt._parse_source(" a ") == [bt._SOURCE_A]

    def test_unknown_raises(self) -> None:
        import pytest

        with pytest.raises(ValueError):
            bt._parse_source("C")


class TestPlanVariants:
    # Each entry is (source, detrend, deseasonalize, estimator). The RAW
    # reference (best_lag on raw units) is always first per source; method flags
    # ADD one variant each (no all-combinations explosion).
    _BL = bt._ESTIMATOR_BEST_LAG
    _AL = bt._ESTIMATOR_ALMON

    def test_raw_only_without_any_flag(self) -> None:
        assert bt._plan_variants([bt._SOURCE_B], detrend=False) == [
            (bt._SOURCE_B, False, False, self._BL)
        ]

    def test_detrend_adds_detrended_variant_per_source(self) -> None:
        plan = bt._plan_variants([bt._SOURCE_B, bt._SOURCE_A], detrend=True)
        assert plan == [
            (bt._SOURCE_B, False, False, self._BL),
            (bt._SOURCE_B, True, False, self._BL),
            (bt._SOURCE_A, False, False, self._BL),
            (bt._SOURCE_A, True, False, self._BL),
        ]

    def test_deseasonalize_adds_deseasonalized_variant(self) -> None:
        plan = bt._plan_variants([bt._SOURCE_B], detrend=False, deseasonalize=True)
        assert plan == [
            (bt._SOURCE_B, False, False, self._BL),
            (bt._SOURCE_B, False, True, self._BL),
        ]

    def test_almon_adds_almon_variant(self) -> None:
        plan = bt._plan_variants([bt._SOURCE_B], detrend=False, almon=True)
        assert plan == [
            (bt._SOURCE_B, False, False, self._BL),
            (bt._SOURCE_B, False, False, self._AL),
        ]

    def test_all_flags_add_one_variant_each_per_source(self) -> None:
        # raw + detrended + deseasonalized + Almon-ADL, in that order, per source.
        plan = bt._plan_variants(
            [bt._SOURCE_B, bt._SOURCE_A], detrend=True, deseasonalize=True, almon=True
        )
        assert plan == [
            (bt._SOURCE_B, False, False, self._BL),
            (bt._SOURCE_B, True, False, self._BL),
            (bt._SOURCE_B, False, True, self._BL),
            (bt._SOURCE_B, False, False, self._AL),
            (bt._SOURCE_A, False, False, self._BL),
            (bt._SOURCE_A, True, False, self._BL),
            (bt._SOURCE_A, False, True, self._BL),
            (bt._SOURCE_A, False, False, self._AL),
        ]

    def test_no_all_combinations_explosion(self) -> None:
        # Two method flags on one source → 1 raw + 2 method variants = 3, NOT the
        # 2x2x... cross-product of preprocessing x estimator.
        plan = bt._plan_variants([bt._SOURCE_B], detrend=True, almon=True)
        assert len(plan) == 3
        assert plan == [
            (bt._SOURCE_B, False, False, self._BL),
            (bt._SOURCE_B, True, False, self._BL),
            (bt._SOURCE_B, False, False, self._AL),
        ]


class TestVariantLabel:
    def test_raw_detrended_deseasonalized_almon_labels(self) -> None:
        assert bt._variant_label(bt._SOURCE_B, False) == "B raw"
        assert bt._variant_label(bt._SOURCE_B, True) == "B detrended"
        assert bt._variant_label(bt._SOURCE_B, False, deseasonalize=True) == "B deseasonalized"
        assert (
            bt._variant_label(bt._SOURCE_A, False, estimator=bt._ESTIMATOR_ALMON) == "A Almon-ADL"
        )

    def test_estimator_takes_precedence_in_label(self) -> None:
        # The planner never combines methods, but if both were set the estimator
        # (the strongest method signal) names the variant.
        assert (
            bt._variant_label(bt._SOURCE_B, True, deseasonalize=True, estimator=bt._ESTIMATOR_ALMON)
            == "B Almon-ADL"
        )


# --------------------------------------------------------------------------- #
# cross_source_verdict (#978b) — B raw vs B detrended vs A
# --------------------------------------------------------------------------- #


def _run(
    source: str,
    detrended: bool,
    ekb: bt.TierResult,
    *,
    deseasonalized: bool = False,
    estimator: str = bt._ESTIMATOR_BEST_LAG,
) -> dict:
    """Minimal run dict (only the fields cross_source_verdict reads)."""
    return {
        "source": source,
        "detrended": detrended,
        "deseasonalized": deseasonalized,
        "estimator": estimator,
        "ekb_result": ekb,
    }


class TestCrossSourceVerdict:
    def test_no_signal_anywhere_is_real_no_signal(self) -> None:
        # B raw + B detrended both at coin-flip, A skipped (thin) → the negative
        # verdict is corroborated as REAL, not a survivorship artifact.
        runs = [
            _run(bt._SOURCE_B, False, _tier(oos_hit_rate=0.45)),
            _run(bt._SOURCE_B, True, _tier(detrended=True, oos_hit_rate=0.50)),
            _run(
                bt._SOURCE_A,
                False,
                _tier(source=bt._SOURCE_A, skipped="only 13 aligned months (< 18)"),
            ),
        ]
        cv = bt.cross_source_verdict(runs)
        assert cv["promote_any"] is False
        assert cv["signal_variants"] == []
        assert "REAL 'no signal'" in cv["conclusion"]
        # The thin Source A row gets the explicit thin-data caveat.
        assert cv["thin_caveat"] is not None
        assert "THIN" in cv["thin_caveat"]

    def test_detrended_signal_flags_possible_artifact(self) -> None:
        # Raw B no signal, but DETRENDED B clears coin-flip+margin (lag stable) →
        # the raw verdict may be a survivorship artifact; the detrended variant
        # is flagged as showing signal.
        runs = [
            _run(bt._SOURCE_B, False, _tier(oos_hit_rate=0.48)),
            _run(bt._SOURCE_B, True, _tier(detrended=True, oos_hit_rate=0.80)),
        ]
        cv = bt.cross_source_verdict(runs)
        assert cv["promote_any"] is True
        assert "B detrended" in cv["signal_variants"]
        assert "ARTIFACT" in cv["conclusion"]

    def test_unstable_lag_not_counted_as_signal(self) -> None:
        # High hit-rate but unstable lag → not a signal (mirrors verdict()).
        runs = [
            _run(
                bt._SOURCE_B,
                True,
                _tier(detrended=True, oos_hit_rate=0.9, lag_stable=False, full_sample_lag=6),
            ),
        ]
        cv = bt.cross_source_verdict(runs)
        assert cv["promote_any"] is False
        assert cv["signal_variants"] == []

    def test_candidate_methods_labelled_and_no_signal(self) -> None:
        # raw + deseasonalized + Almon-ADL all at/below coin-flip → REAL no signal,
        # the conclusion mentions the candidate methods, and each variant is
        # labelled by its method (not lumped under "raw"/"detrended").
        runs = [
            _run(bt._SOURCE_B, False, _tier(oos_hit_rate=0.48)),
            _run(bt._SOURCE_B, False, _tier(oos_hit_rate=0.50), deseasonalized=True),
            _run(
                bt._SOURCE_B,
                False,
                _tier(oos_hit_rate=0.47),
                estimator=bt._ESTIMATOR_ALMON,
            ),
        ]
        cv = bt.cross_source_verdict(runs)
        assert cv["promote_any"] is False
        labels = [r["variant"] for r in cv["rows"]]
        assert labels == ["B raw", "B deseasonalized", "B Almon-ADL"]
        # The conclusion is generalised to the candidate methods.
        assert "deseasonalize" in cv["conclusion"] and "Almon-ADL" in cv["conclusion"]
        # Row descriptors carry the method so JSON consumers can filter.
        assert cv["rows"][1]["deseasonalized"] is True
        assert cv["rows"][2]["estimator"] == bt._ESTIMATOR_ALMON

    def test_candidate_method_recovers_signal_is_flagged(self) -> None:
        # raw best_lag no signal, but the Almon-ADL variant clears coin-flip+margin
        # (lag stable) → flagged as a variant recovering signal worth inspecting.
        runs = [
            _run(bt._SOURCE_B, False, _tier(oos_hit_rate=0.49)),
            _run(
                bt._SOURCE_B,
                False,
                _tier(oos_hit_rate=0.82),
                estimator=bt._ESTIMATOR_ALMON,
            ),
        ]
        cv = bt.cross_source_verdict(runs)
        assert cv["promote_any"] is True
        assert "B Almon-ADL" in cv["signal_variants"]
        # Conclusion offers the candidate-method reading.
        assert "candidate method" in cv["conclusion"]


# --------------------------------------------------------------------------- #
# 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 == ["комфорт", "бизнес"]


class TestSourceASqlShape:
    def test_units_sql_hits_corpus_room_month_table(self) -> None:
        sql = str(bt._SOURCE_A_UNITS_SQL)
        assert "objective_corpus_room_month" in sql
        # Survivorship-free aggregate: SUM(deals_total_count) GROUP BY the month.
        assert "SUM(crm.deals_total_count)" in sql
        assert "GROUP BY 1" in sql
        # report_month truncated to a month-first DATE.
        assert "date_trunc('month', crm.report_month)" in sql

    def test_units_sql_uses_cast_not_double_colon(self) -> None:
        sql = str(bt._SOURCE_A_UNITS_SQL)
        assert "CAST(:since AS date)" in sql
        # Optional class filter folds case (capitalised in this table).
        assert "LOWER(CAST(:cls AS text))" 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_A_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_A_CLASSES_SQL)
        assert "objective_corpus_room_month" in sql
        assert "CAST(:since AS date)" in sql
        assert "::" not in sql

    def test_load_sales_source_a_binds_and_shapes(self) -> None:
        ms = _months(3)
        sess = _CaptureSession([(ms[0], 100), (ms[1], 200), (None, 99)])
        out = bt.load_sales_by_month_source_a(
            sess,  # type: ignore[arg-type]
            since="2025-05-01",
            obj_class="комфорт",
        )
        # None-month row dropped; rows mapped to {month: units}.
        assert out == {ms[0]: 100, ms[1]: 200}
        _sql, params = sess.calls[0]
        # Parametrised — no premise_kind / district for Source A.
        assert params["cls"] == "комфорт"
        assert params["since"] == "2025-05-01"
        assert "premise_kind" not in params
        assert "district" not in params

    def test_load_classes_source_a_maps_rows(self) -> None:
        sess = _CaptureSession([("комфорт",), ("бизнес",), (None,)])
        out = bt.load_classes_source_a(sess, since="2025-05-01")  # type: ignore[arg-type]
        assert out == ["комфорт", "бизнес"]


class TestSourceDispatch:
    def test_load_sales_dispatch_routes_by_source(self) -> None:
        ms = _months(2)
        sess_b = _CaptureSession([(ms[0], 10)])
        bt._load_sales(
            sess_b,  # type: ignore[arg-type]
            source=bt._SOURCE_B,
            since="2019-01-01",
            obj_class=None,
            district=None,
        )
        # Source B SQL carries the premise_kind bind.
        _sql_b, params_b = sess_b.calls[0]
        assert params_b["premise_kind"] == bt._PREMISE_KIND

        sess_a = _CaptureSession([(ms[0], 99)])
        bt._load_sales(
            sess_a,  # type: ignore[arg-type]
            source=bt._SOURCE_A,
            since="2025-05-01",
            obj_class=None,
            district=None,
        )
        # Source A SQL hits the corpus_room_month table, no premise_kind.
        sql_a, params_a = sess_a.calls[0]
        assert "objective_corpus_room_month" in sql_a
        assert "premise_kind" not in params_a


# --------------------------------------------------------------------------- #
# 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)