backend/scripts/backtest_rate_sensitivity.py

@@ -58,19 +58,39 @@ 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. This is the SAME caveat the
-      §9.6 module documents; it is why Source B is used here (Source A
-      ``objective_corpus_room_month`` has only ~13 months — too thin to
-      regress).
+      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.
+      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:
@@ -83,6 +103,9 @@ USAGE
 
     # 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
@@ -90,11 +113,13 @@ 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
 
@@ -140,6 +165,19 @@ _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.
@@ -201,6 +239,8 @@ class TierResult:
     """
 
     tier: str
+    source: str
+    detrended: bool
     n_aligned: int
     n_train: int
     n_test: int
@@ -216,6 +256,8 @@ class TierResult:
     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,
@@ -259,6 +301,56 @@ def _rate_first_diff(rate_levels: list[float | None]) -> list[float | None]:
     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.
 
@@ -403,29 +495,51 @@ def align_series(
     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
-    Δln(sales) (reused ``log_diff``) and Δrate (first diff), then delegates to
-    ``evaluate_oos``. Tiers with fewer than ``min_months`` aligned months are
-    SKIPPED (TierResult with ``skipped`` set, all metrics None) — no silent
-    drop. PURE aside from the deferred engine import.
+    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.
     """
-    _best_lag, _ols, log_diff = _import_engine()
-
     months, units, rates = align_series(sales_by_month, rate_by_month)
     n_aligned = len(months)
     if n_aligned < min_months:
         return TierResult(
             tier=tier,
+            source=source,
+            detrended=detrend,
             n_aligned=n_aligned,
             n_train=0,
             n_test=0,
@@ -439,12 +553,14 @@ def backtest_tier(
             skipped=f"only {n_aligned} aligned months (< {min_months})",
         )
 
-    delta_sales = log_diff(units)
+    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"],
@@ -555,6 +671,38 @@ _SOURCE_B_UNITS_SQL = text(
     """
 )
 
+# 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(
     """
@@ -600,7 +748,7 @@ def load_sales_by_month(
 
 
 def load_classes(db: Session, *, since: str) -> list[str]:
-    """Run the distinct-classes SELECT → lowercase class list. READ-ONLY."""
+    """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},
@@ -608,6 +756,40 @@ def load_classes(db: Session, *, since: str) -> list[str]:
     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.
 
@@ -640,6 +822,33 @@ def load_rate_by_month(db: Session, *, since: str) -> dict[date, float]:
 # --------------------------------------------------------------------------- #
 
 
+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,
     *,
@@ -647,28 +856,47 @@ def run_backtest(
     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 the full read-only backtest and return a results dict. No writes.
+    """Drive ONE source/variant of the read-only backtest → results dict. No writes.
 
-    Loads the monthly key_rate once, then the EKB-wide and per-class Source B
-    sold-units series, backtests each tier (``backtest_tier``), and assembles
-    the verdict + per-tier OOS lifts.
+    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 Source B; an empty
-    list → EKB-wide only. ``district`` optionally narrows ALL tiers.
+    ``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).
     """
-    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 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(db, since=since)
-        logger.info("auto-discovered classes: %s", classes)
+        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_by_month(db, since=since, obj_class=None, district=district)
-    ekb = backtest_tier(ekb_sales, rate_by_month, tier=_EKB_WIDE, holdout_frac=holdout_frac)
+    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(
-        "EKB-wide: aligned=%d train=%d test=%d lag=%s hit_rate=%s",
+        "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,
@@ -679,12 +907,22 @@ def run_backtest(
     tiers: list[TierResult] = []
     lifts: dict[str, float | None] = {}
     for cls in classes:
-        cls_sales = load_sales_by_month(db, since=since, obj_class=cls, district=district)
-        res = backtest_tier(cls_sales, rate_by_month, tier=cls, holdout_frac=holdout_frac)
+        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(
-            "tier=%s aligned=%d test=%d hit_rate=%s lift=%s skipped=%s",
+            "source=%s tier=%s aligned=%d test=%d hit_rate=%s lift=%s skipped=%s",
+            source,
             cls,
             res.n_aligned,
             res.n_test,
@@ -695,11 +933,16 @@ def run_backtest(
 
     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()),
         },
@@ -712,6 +955,168 @@ def run_backtest(
     }
 
 
+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}"
 
@@ -720,28 +1125,50 @@ 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 the backtest results as a plain-text stdout report."""
+    """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("BACKTEST: §9.6 rate-sensitivity engine — out-of-sample validation")
+    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 B (objective_lots) monthly sold-units vs monthly key_rate.")
+    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} {'aligned':>7} {'train':>6} {'test':>5} {'lag':>4} "
-        f"{'beta':>9} {'inR2':>7} {'OOS_hit':>8} {'OOS_MAE':>8} {'lift':>7} {'stable':>7}"
+        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))
@@ -788,26 +1215,41 @@ def render_table(results: dict[str, Any]) -> str:
     if not any_lift:
         lines.append("  (no class tier had a scorable OOS hit-rate to compare)")
 
-    # Verdict.
+    # Per-variant verdict.
     lines.append("")
-    lines.append("VERDICT:")
+    lines.append("VERDICT (this variant):")
     lines.append(f"  {vd['reason']}")
     if vd.get("thin_warning"):
         lines.append(f"  !! {vd['thin_warning']}")
 
     lines.append("")
-    lines.append("Caveats: Source B survivorship undercounts OLD months; short Δ-series + holdout")
-    lines.append("leaves a small test window; this validates the β/lag CORE, not the full engine.")
+    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.n_aligned:>7} {t.n_train:>6} {t.n_test:>5} "
-        f"{_fmt_lag(t.train_lag):>4} {_fmt_beta(t.train_beta):>9} "
+        f"  {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}"
@@ -838,6 +1280,24 @@ def _parse_classes(raw: str | None) -> list[str] | 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(
@@ -850,6 +1310,20 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         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,
@@ -876,38 +1350,57 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     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 the EKB-wide tier was backtested, 1 if skipped."""
+    """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",
+        "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:
-        results = run_backtest(
+        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:
-        payload = {k: v for k, v in results.items() if k not in ("ekb_result", "tier_results")}
-        print(json.dumps(payload, ensure_ascii=False, indent=2, default=str))
+        print(json.dumps(_json_payload(payload), ensure_ascii=False, indent=2, default=str))
     else:
-        print(render_table(results))
+        print(render_all(payload))
 
-    ekb: TierResult = results["ekb_result"]
-    return 0 if ekb.skipped is None else 1
+    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

backend/tests/scripts/test_backtest_rate_sensitivity.py

@@ -1,17 +1,23 @@
-"""Unit tests for the read-only §9.6 rate-sensitivity backtest (Forgejo #978).
+"""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
-  - backtest_tier        — thin-tier skip; happy path
+  - backtest_tier        — thin-tier skip; happy path; (#978b) detrended variant
+                           recovers an injected signal masked by a trend
   - verdict / tier_lift  — promotion criterion, coin-flip baseline, lag stability
+  - _parse_source / _plan_variants — (#978b) B/A/both selection + variant plan
+  - cross_source_verdict — (#978b) B raw vs B detrended vs A conclusion
 
-DB is MOCKED (a fake session) only to assert the Source B SQL SHAPE — that it
-uses CAST(:x AS type) and never the psycopg3-incompatible :x::type form.
+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
@@ -89,6 +95,52 @@ def _units_from_rate(
     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
+
+
 # --------------------------------------------------------------------------- #
 # _time_ordered_split
 # --------------------------------------------------------------------------- #
@@ -150,6 +202,58 @@ class TestShiftForLag:
             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
+
+
 class TestAlignSeries:
     def test_inner_join_by_month(self) -> None:
         ms = _months(4)
@@ -294,6 +398,57 @@ class TestBacktestTier:
         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
+
 
 # --------------------------------------------------------------------------- #
 # verdict / tier_lift
@@ -303,6 +458,8 @@ class TestBacktestTier:
 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,
@@ -317,6 +474,8 @@ def _tier(
 ) -> bt.TierResult:
     return bt.TierResult(
         tier=tier,
+        source=source,
+        detrended=detrended,
         n_aligned=n_aligned,
         n_train=n_train,
         n_test=n_test,
@@ -396,6 +555,102 @@ class TestParseClasses:
         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:
+    def test_raw_only_without_detrend(self) -> None:
+        assert bt._plan_variants([bt._SOURCE_B], detrend=False) == [(bt._SOURCE_B, False)]
+
+    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),
+            (bt._SOURCE_B, True),
+            (bt._SOURCE_A, False),
+            (bt._SOURCE_A, True),
+        ]
+
+
+# --------------------------------------------------------------------------- #
+# cross_source_verdict (#978b) — B raw vs B detrended vs A
+# --------------------------------------------------------------------------- #
+
+
+def _run(source: str, detrended: bool, ekb: bt.TierResult) -> dict:
+    """Minimal run dict (only the fields cross_source_verdict reads)."""
+    return {"source": source, "detrended": detrended, "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"] == []
+
+
 # --------------------------------------------------------------------------- #
 # DB layer SQL SHAPE — mocked session, asserts CAST not :: and read-only
 # --------------------------------------------------------------------------- #
@@ -466,6 +721,88 @@ class TestSourceBSqlShape:
         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)
 # --------------------------------------------------------------------------- #