diff --git a/backend/scripts/backtest_rate_sensitivity.py b/backend/scripts/backtest_rate_sensitivity.py index ea62aed1..6791a4f3 100644 --- a/backend/scripts/backtest_rate_sensitivity.py +++ b/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 diff --git a/backend/tests/scripts/test_backtest_rate_sensitivity.py b/backend/tests/scripts/test_backtest_rate_sensitivity.py index 02ea9131..0050f700 100644 --- a/backend/tests/scripts/test_backtest_rate_sensitivity.py +++ b/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) # --------------------------------------------------------------------------- #