814 lines
34 KiB
Python
814 lines
34 KiB
Python
"""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; (#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 A/B SQL SHAPE — that it
|
|
uses CAST(:x AS type) and never the psycopg3-incompatible :x::type form, hits the
|
|
right table, and aggregates per the spec.
|
|
|
|
NOTE: importing scripts.backtest_rate_sensitivity is cheap (the engine import
|
|
is deferred), but evaluate_oos/backtest_tier call into
|
|
app.services.forecasting.* which pulls app.core.config.Settings. Set a dummy
|
|
DATABASE_URL BEFORE importing so that fail-fast doesn't trip (same pattern as
|
|
tests/services/forecasting/test_rate_sensitivity.py).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import math
|
|
import os
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
|
|
|
from scripts import backtest_rate_sensitivity as bt
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Synthetic-series helpers
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _months(n: int, *, start: dt.date | None = None) -> list[dt.date]:
|
|
"""n consecutive month-firsts, ascending, starting at `start` (default 2019-01)."""
|
|
start = start or dt.date(2019, 1, 1)
|
|
out: list[dt.date] = []
|
|
y, m = start.year, start.month
|
|
for _ in range(n):
|
|
out.append(dt.date(y, m, 1))
|
|
m += 1
|
|
if m == 13:
|
|
m = 1
|
|
y += 1
|
|
return out
|
|
|
|
|
|
def _aperiodic_rate_levels(n: int, *, seed: int = 13) -> list[float]:
|
|
"""Rising key_rate levels with APERIODIC (LCG) jitter → low Δ autocorrelation.
|
|
|
|
Mirrors the engine test's regressor: a periodic (sin) jitter would give Δ a
|
|
sign-flipping autocorrelation so the injected lag competes with false lags.
|
|
An LCG jitter keeps lags weakly correlated → the true lag wins cleanly.
|
|
"""
|
|
lvl = 10.0
|
|
state = seed
|
|
out: list[float] = []
|
|
for _ in range(n):
|
|
state = (state * 1103515245 + 12345) % 2147483648
|
|
lvl += 0.3 + (state / 2147483648.0 - 0.5) * 0.4
|
|
out.append(lvl)
|
|
return out
|
|
|
|
|
|
def _units_from_rate(
|
|
rate_levels: list[float],
|
|
*,
|
|
lag: int,
|
|
beta: float,
|
|
base: float = 1000.0,
|
|
) -> list[int]:
|
|
"""Sold-units series s.t. log_diff(units)[t] ≈ beta·Δrate[t-lag] (injected link).
|
|
|
|
ln(u_t) = ln(u_{t-1}) + beta·Δrate[t-lag]; rounded to int (units are a
|
|
count). Small step so rounding doesn't kill the relationship. Mirrors the
|
|
engine test's _synth_sales_units.
|
|
"""
|
|
rate_deltas = [0.0] + [rate_levels[i] - rate_levels[i - 1] for i in range(1, len(rate_levels))]
|
|
ln_u = math.log(base)
|
|
units: list[int] = [round(math.exp(ln_u))]
|
|
for t in range(1, len(rate_levels)):
|
|
src = rate_deltas[t - lag] if t - lag >= 0 else 0.0
|
|
ln_u += beta * src
|
|
units.append(max(1, round(math.exp(ln_u))))
|
|
return units
|
|
|
|
|
|
def _zero_drift_rate_levels(n: int, *, seed: int = 7) -> list[float]:
|
|
"""key_rate levels that OSCILLATE around a constant → Δrate has ~zero mean.
|
|
|
|
Used for the detrend test: a monotone rate would give the injected signal a
|
|
nonzero average slope that the linear detrend partly absorbs, leaving a
|
|
constant Δ-offset the intercept-free OOS predictor can't model. With ~zero
|
|
mean Δrate the detrend removes ONLY the spurious units trend, so the
|
|
differenced residual cleanly reconstructs beta·Δrate[t-lag]. LCG jitter (not
|
|
sin) keeps successive Δ weakly correlated so the true lag wins.
|
|
"""
|
|
state = seed
|
|
out: list[float] = []
|
|
for _ in range(n):
|
|
state = (state * 1103515245 + 12345) % 2147483648
|
|
# Center on 10.0, symmetric jitter → no drift in the levels.
|
|
out.append(10.0 + (state / 2147483648.0 - 0.5) * 3.0)
|
|
return out
|
|
|
|
|
|
def _units_from_rate_with_trend(
|
|
rate_levels: list[float],
|
|
*,
|
|
lag: int,
|
|
beta: float,
|
|
trend_per_month: float,
|
|
base: float = 1000.0,
|
|
) -> list[int]:
|
|
"""Units carrying BOTH an injected rate signal AND a spurious log-linear trend.
|
|
|
|
ln(u_t) = ln(base) + trend·t + Σ_{k≤t} beta·Δrate[k-lag]. The ``trend·t`` term
|
|
is the survivorship-style monotone drift #978b's --detrend control removes; the
|
|
Σ term is the real rate→sales signal. Detrending should subtract ~trend·t and
|
|
leave the rate-driven residual whose Δ reconstructs beta·Δrate[t-lag].
|
|
"""
|
|
rate_deltas = [0.0] + [rate_levels[i] - rate_levels[i - 1] for i in range(1, len(rate_levels))]
|
|
signal_cum = 0.0
|
|
units: list[int] = []
|
|
for t in range(len(rate_levels)):
|
|
if t > 0:
|
|
src = rate_deltas[t - lag] if t - lag >= 0 else 0.0
|
|
signal_cum += beta * src
|
|
ln_u = math.log(base) + trend_per_month * t + signal_cum
|
|
units.append(max(1, round(math.exp(ln_u))))
|
|
return units
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# _time_ordered_split
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestTimeOrderedSplit:
|
|
def test_basic_fraction(self) -> None:
|
|
assert bt._time_ordered_split(100, 0.7) == 70
|
|
assert bt._time_ordered_split(30, 0.7) == 21
|
|
|
|
def test_keeps_one_month_each_side(self) -> None:
|
|
# frac=1.0 would put everything in train → clamp to n-1 so test has ≥1.
|
|
assert bt._time_ordered_split(10, 1.0) == 9
|
|
# frac=0.0 would empty train → clamp to ≥1.
|
|
assert bt._time_ordered_split(10, 0.0) == 1
|
|
|
|
def test_degenerate_sizes(self) -> None:
|
|
assert bt._time_ordered_split(0, 0.7) == 0
|
|
assert bt._time_ordered_split(1, 0.7) == 1 # nothing to split
|
|
|
|
def test_is_time_ordered_not_parity(self) -> None:
|
|
# The split is a single boundary index (past→train, future→test), NOT a
|
|
# parity/random partition: train is a contiguous prefix.
|
|
n_train = bt._time_ordered_split(20, 0.7)
|
|
assert n_train == 14 # contiguous prefix [0:14], test [14:20]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# _rate_first_diff / _shift_for_lag / align_series
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestRateFirstDiff:
|
|
def test_first_diff(self) -> None:
|
|
assert bt._rate_first_diff([10.0, 12.0, 11.0]) == [None, 2.0, -1.0]
|
|
|
|
def test_none_breaks_pair(self) -> None:
|
|
assert bt._rate_first_diff([1.0, None, 3.0]) == [None, None, None]
|
|
|
|
def test_empty_and_single(self) -> None:
|
|
assert bt._rate_first_diff([]) == [None]
|
|
assert bt._rate_first_diff([5.0]) == [None]
|
|
|
|
|
|
class TestShiftForLag:
|
|
def test_lag_zero_is_identity(self) -> None:
|
|
assert bt._shift_for_lag([1.0, 2.0, 3.0], 0) == [1.0, 2.0, 3.0]
|
|
|
|
def test_lag_shifts_right_and_truncates(self) -> None:
|
|
# y[t] ← x[t-2]: two leading None, length preserved.
|
|
assert bt._shift_for_lag([1.0, 2.0, 3.0, 4.0], 2) == [None, None, 1.0, 2.0]
|
|
|
|
def test_no_future_leak(self) -> None:
|
|
# Element at index t must equal the ORIGINAL element at t-lag (never t+k).
|
|
x = [10.0, 20.0, 30.0, 40.0, 50.0]
|
|
lag = 1
|
|
shifted = bt._shift_for_lag(x, lag)
|
|
for t in range(lag, len(x)):
|
|
assert shifted[t] == x[t - lag]
|
|
|
|
|
|
class TestDetrendLog:
|
|
def test_removes_known_linear_trend(self) -> None:
|
|
# units = exp(a + b·t): a PURE log-linear trend → residuals must be ~0.
|
|
a, b = 6.0, 0.05
|
|
units = [round(math.exp(a + b * t)) for t in range(24)]
|
|
resid = bt._detrend_log(units)
|
|
assert all(r is not None for r in resid)
|
|
# Rounding to int adds tiny noise, but residuals collapse near zero.
|
|
assert max(abs(r) for r in resid) < 0.01 # type: ignore[arg-type, type-var]
|
|
|
|
def test_residuals_isolate_signal_over_trend(self) -> None:
|
|
# Trend + a single oscillation: after detrend the trend is gone and the
|
|
# residual variance is dominated by the oscillation, not the drift.
|
|
n = 30
|
|
base_units = [math.exp(6.0 + 0.08 * t + 0.3 * math.sin(t)) for t in range(n)]
|
|
units = [max(1, round(u)) for u in base_units]
|
|
resid = bt._detrend_log(units)
|
|
finite = [r for r in resid if r is not None]
|
|
# Detrended series is NOT monotone (the drift dominated the raw logs).
|
|
diffs = [finite[i] - finite[i - 1] for i in range(1, len(finite))]
|
|
assert any(d > 0 for d in diffs) and any(d < 0 for d in diffs)
|
|
|
|
def test_none_and_nonpositive_map_to_none(self) -> None:
|
|
vals = [100, None, 0, -5, 120, 130, 140]
|
|
resid = bt._detrend_log(vals)
|
|
assert len(resid) == len(vals)
|
|
assert resid[1] is None # None in
|
|
assert resid[2] is None # 0 → ln undefined
|
|
assert resid[3] is None # negative → ln undefined
|
|
# The finite positions stay finite.
|
|
assert resid[0] is not None and resid[4] is not None
|
|
|
|
def test_short_series_passthrough_is_logs(self) -> None:
|
|
# <3 finite points → can't fit a line → passthrough of ln(values).
|
|
vals = [10, 20]
|
|
resid = bt._detrend_log(vals)
|
|
assert resid[0] is not None and math.isclose(resid[0], math.log(10))
|
|
assert resid[1] is not None and math.isclose(resid[1], math.log(20))
|
|
|
|
def test_short_after_filtering_passthrough(self) -> None:
|
|
# Only 2 finite points after dropping None/≤0 → passthrough of logs.
|
|
vals = [None, 50, 0, 60]
|
|
resid = bt._detrend_log(vals)
|
|
assert resid[0] is None and resid[2] is None
|
|
assert resid[1] is not None and math.isclose(resid[1], math.log(50))
|
|
assert resid[3] is not None and math.isclose(resid[3], math.log(60))
|
|
|
|
def test_length_preserved(self) -> None:
|
|
vals = [100 + i for i in range(10)]
|
|
assert len(bt._detrend_log(vals)) == 10
|
|
|
|
|
|
class TestAlignSeries:
|
|
def test_inner_join_by_month(self) -> None:
|
|
ms = _months(4)
|
|
sales = {ms[0]: 100, ms[1]: 110, ms[2]: 120, ms[3]: 130}
|
|
# rate missing ms[0]; has an extra month not in sales.
|
|
rate = {ms[1]: 7.0, ms[2]: 7.5, ms[3]: 8.0, dt.date(2030, 1, 1): 9.0}
|
|
months, units, rates = bt.align_series(sales, rate)
|
|
assert months == [ms[1], ms[2], ms[3]] # intersection only, ascending
|
|
assert units == [110, 120, 130]
|
|
assert rates == [7.0, 7.5, 8.0]
|
|
|
|
def test_empty_intersection(self) -> None:
|
|
months, units, rates = bt.align_series({_months(1)[0]: 1}, {dt.date(2030, 1, 1): 2.0})
|
|
assert months == [] and units == [] and rates == []
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# evaluate_oos — the core OOS metric
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestEvaluateOos:
|
|
def test_injected_signal_high_oos_hit_rate(self) -> None:
|
|
# sales react to rate at lag 2 with a clean negative β → the TRAIN fit
|
|
# should generalise: nearly every TEST month's predicted sign matches.
|
|
n = 48
|
|
rate = _aperiodic_rate_levels(n)
|
|
units = _units_from_rate(rate, lag=2, beta=-0.05)
|
|
delta_sales = _delta_ln(units)
|
|
rate_deltas = bt._rate_first_diff(rate)
|
|
|
|
res = bt.evaluate_oos(delta_sales, rate_deltas, holdout_frac=0.7)
|
|
assert res["train_lag"] == 2
|
|
assert res["train_beta"] is not None and res["train_beta"] < 0
|
|
assert res["oos_hit_rate"] is not None
|
|
# A real injected signal → directional hit-rate clearly beats a coin flip.
|
|
assert res["oos_hit_rate"] >= 0.8
|
|
# In-sample R² is high by construction (reported, not trusted).
|
|
assert res["in_sample_r2"] is not None and res["in_sample_r2"] > 0.9
|
|
# Lag stable: full-sample refit finds the same lag.
|
|
assert res["full_sample_lag"] == 2
|
|
assert res["lag_stable"] is True
|
|
|
|
def test_pure_noise_hit_rate_near_coin_flip(self) -> None:
|
|
# No rate→sales link: sales are an independent aperiodic walk. Either no
|
|
# gated lag is found on TRAIN (→ None), or any spurious fit predicts
|
|
# direction no better than a coin flip on held-out months.
|
|
n = 60
|
|
rate = _aperiodic_rate_levels(n, seed=1)
|
|
noise = _aperiodic_rate_levels(n, seed=999) # uncorrelated second series
|
|
units = [max(1, round(1000.0 * math.exp(0.01 * (v - 10.0)))) for v in noise]
|
|
delta_sales = _delta_ln(units)
|
|
rate_deltas = bt._rate_first_diff(rate)
|
|
|
|
res = bt.evaluate_oos(delta_sales, rate_deltas, holdout_frac=0.7)
|
|
hr = res["oos_hit_rate"]
|
|
# Honest outcome: no signal → either ungated (None) or ~coin-flip.
|
|
assert hr is None or hr <= 0.7
|
|
|
|
def test_too_few_months_returns_empty(self) -> None:
|
|
# 1 month → can't split → empty result (all metrics None, not a crash).
|
|
res = bt.evaluate_oos([None], [None], holdout_frac=0.7)
|
|
assert res["train_lag"] is None
|
|
assert res["oos_hit_rate"] is None
|
|
assert res["n_train"] == 1 and res["n_test"] == 0
|
|
|
|
def test_no_gated_lag_on_train_returns_empty(self) -> None:
|
|
# Positive rate→sales link (β>0) → engine gate (slope<0) rejects every
|
|
# lag on TRAIN → nothing to validate → empty (None) result, no crash.
|
|
n = 40
|
|
rate = _aperiodic_rate_levels(n)
|
|
units = _units_from_rate(rate, lag=1, beta=+0.05) # wrong sign
|
|
delta_sales = _delta_ln(units)
|
|
rate_deltas = bt._rate_first_diff(rate)
|
|
res = bt.evaluate_oos(delta_sales, rate_deltas, holdout_frac=0.7)
|
|
assert res["train_lag"] is None
|
|
assert res["oos_hit_rate"] is None
|
|
|
|
def test_point_in_time_no_future_leak(self) -> None:
|
|
# Build a signal, then confirm the TEST prediction at the FIRST test
|
|
# month uses only rate data at or before it. We reconstruct the expected
|
|
# prediction from the public _shift_for_lag and check evaluate_oos's MAE
|
|
# is finite (a future leak would mismatch lengths / shift indices).
|
|
n = 36
|
|
rate = _aperiodic_rate_levels(n)
|
|
units = _units_from_rate(rate, lag=3, beta=-0.04)
|
|
delta_sales = _delta_ln(units)
|
|
rate_deltas = bt._rate_first_diff(rate)
|
|
res = bt.evaluate_oos(delta_sales, rate_deltas, holdout_frac=0.7)
|
|
assert res["oos_signed_mae"] is not None
|
|
assert math.isfinite(res["oos_signed_mae"])
|
|
# First scored test month index = n_train; predictor must be Δrate[t-lag].
|
|
lag = res["train_lag"]
|
|
assert lag is not None
|
|
shifted = bt._shift_for_lag(rate_deltas, lag)
|
|
# The shifted regressor at the first test index is at or before it.
|
|
assert shifted[res["n_train"]] is None or isinstance(shifted[res["n_train"]], float)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# backtest_tier — thin-tier skip + happy path
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestBacktestTier:
|
|
def test_thin_tier_skipped_not_dropped(self) -> None:
|
|
# Fewer than _MIN_BACKTEST_MONTHS aligned months → skipped with a reason,
|
|
# all metrics None (NOT a silent drop, NOT a crash).
|
|
ms = _months(5)
|
|
rate = _aperiodic_rate_levels(5)
|
|
sales = {ms[i]: 100 + i for i in range(5)}
|
|
rate_by = {ms[i]: rate[i] for i in range(5)}
|
|
res = bt.backtest_tier(sales, rate_by, tier="комфорт", min_months=18)
|
|
assert res.skipped is not None
|
|
assert "aligned months" in res.skipped
|
|
assert res.oos_hit_rate is None
|
|
assert res.n_aligned == 5
|
|
|
|
def test_happy_path_builds_metrics(self) -> None:
|
|
n = 48
|
|
ms = _months(n)
|
|
rate = _aperiodic_rate_levels(n)
|
|
units = _units_from_rate(rate, lag=2, beta=-0.05)
|
|
sales = {ms[i]: units[i] for i in range(n)}
|
|
rate_by = {ms[i]: rate[i] for i in range(n)}
|
|
res = bt.backtest_tier(sales, rate_by, tier=bt._EKB_WIDE, holdout_frac=0.7)
|
|
assert res.skipped is None
|
|
assert res.tier == bt._EKB_WIDE
|
|
assert res.train_lag == 2
|
|
assert res.oos_hit_rate is not None and res.oos_hit_rate >= 0.8
|
|
assert res.n_aligned == n
|
|
|
|
def test_alignment_drops_unmatched_months(self) -> None:
|
|
# Sales and rate only overlap on a thin window → aligned count reflects
|
|
# the INTERSECTION, which here is below the min → skipped.
|
|
ms = _months(40)
|
|
rate = _aperiodic_rate_levels(40)
|
|
sales = {ms[i]: 100 + i for i in range(40)}
|
|
# rate only for the last 10 months → intersection = 10 < 18.
|
|
rate_by = {ms[i]: rate[i] for i in range(30, 40)}
|
|
res = bt.backtest_tier(sales, rate_by, tier="бизнес", min_months=18)
|
|
assert res.n_aligned == 10
|
|
assert res.skipped is not None
|
|
|
|
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
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _tier(
|
|
*,
|
|
tier: str = bt._EKB_WIDE,
|
|
source: str = bt._SOURCE_B,
|
|
detrended: bool = False,
|
|
n_aligned: int = 40,
|
|
n_train: int = 28,
|
|
n_test: int = 12,
|
|
train_lag: int | None = 2,
|
|
train_beta: float | None = -0.05,
|
|
in_sample_r2: float | None = 0.95,
|
|
oos_hit_rate: float | None = 0.75,
|
|
oos_signed_mae: float | None = 0.02,
|
|
full_sample_lag: int | None = 2,
|
|
lag_stable: bool = True,
|
|
skipped: str | None = None,
|
|
) -> bt.TierResult:
|
|
return bt.TierResult(
|
|
tier=tier,
|
|
source=source,
|
|
detrended=detrended,
|
|
n_aligned=n_aligned,
|
|
n_train=n_train,
|
|
n_test=n_test,
|
|
train_lag=train_lag,
|
|
train_beta=train_beta,
|
|
in_sample_r2=in_sample_r2,
|
|
oos_hit_rate=oos_hit_rate,
|
|
oos_signed_mae=oos_signed_mae,
|
|
full_sample_lag=full_sample_lag,
|
|
lag_stable=lag_stable,
|
|
skipped=skipped,
|
|
)
|
|
|
|
|
|
class TestVerdict:
|
|
def test_promote_when_beats_coin_and_lag_stable(self) -> None:
|
|
vd = bt.verdict(_tier(oos_hit_rate=0.75, lag_stable=True))
|
|
assert vd["promote"] is True
|
|
assert "OOS predictive value" in vd["reason"]
|
|
|
|
def test_keep_advisory_when_at_coin_flip(self) -> None:
|
|
vd = bt.verdict(_tier(oos_hit_rate=0.52, lag_stable=True)) # ≤ 0.5+margin
|
|
assert vd["promote"] is False
|
|
assert "keep advisory" in vd["reason"]
|
|
|
|
def test_keep_advisory_when_lag_unstable(self) -> None:
|
|
vd = bt.verdict(_tier(oos_hit_rate=0.9, lag_stable=False, full_sample_lag=6))
|
|
assert vd["promote"] is False
|
|
assert "lag unstable" in vd["reason"]
|
|
|
|
def test_keep_advisory_when_skipped(self) -> None:
|
|
vd = bt.verdict(_tier(skipped="only 5 aligned months (< 18)"))
|
|
assert vd["promote"] is False
|
|
assert "keep advisory" in vd["reason"]
|
|
|
|
def test_keep_advisory_when_no_hit_rate(self) -> None:
|
|
vd = bt.verdict(_tier(oos_hit_rate=None))
|
|
assert vd["promote"] is False
|
|
|
|
def test_thin_warning_set_for_small_test_window(self) -> None:
|
|
vd = bt.verdict(_tier(oos_hit_rate=0.9, n_test=3, lag_stable=True))
|
|
assert vd["promote"] is True
|
|
assert vd["thin_warning"] is not None
|
|
assert "small" in vd["thin_warning"]
|
|
|
|
|
|
class TestTierLift:
|
|
def test_positive_lift_beats_ekb(self) -> None:
|
|
ekb = _tier(oos_hit_rate=0.6)
|
|
cls = _tier(tier="комфорт", oos_hit_rate=0.75)
|
|
assert bt.tier_lift(ekb, cls) is not None
|
|
assert math.isclose(bt.tier_lift(ekb, cls), 0.15)
|
|
|
|
def test_none_when_either_missing(self) -> None:
|
|
ekb = _tier(oos_hit_rate=None)
|
|
cls = _tier(oos_hit_rate=0.75)
|
|
assert bt.tier_lift(ekb, cls) is None
|
|
assert bt.tier_lift(_tier(oos_hit_rate=0.6), _tier(oos_hit_rate=None)) is None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# _parse_classes
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestParseClasses:
|
|
def test_all_means_autodiscover(self) -> None:
|
|
assert bt._parse_classes("all") is None
|
|
assert bt._parse_classes("ALL") is None
|
|
assert bt._parse_classes(None) is None
|
|
|
|
def test_empty_means_ekb_only(self) -> None:
|
|
assert bt._parse_classes("") == []
|
|
assert bt._parse_classes(" ") == []
|
|
|
|
def test_csv_lowercased_and_trimmed(self) -> None:
|
|
assert bt._parse_classes("Комфорт, Бизнес ,премиум") == ["комфорт", "бизнес", "премиум"]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# _parse_source / _plan_variants (#978b)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestParseSource:
|
|
def test_both_and_default(self) -> None:
|
|
assert bt._parse_source("both") == [bt._SOURCE_B, bt._SOURCE_A]
|
|
assert bt._parse_source(None) == [bt._SOURCE_B, bt._SOURCE_A]
|
|
assert bt._parse_source("") == [bt._SOURCE_B, bt._SOURCE_A]
|
|
|
|
def test_single_source_case_insensitive(self) -> None:
|
|
assert bt._parse_source("B") == [bt._SOURCE_B]
|
|
assert bt._parse_source("b") == [bt._SOURCE_B]
|
|
assert bt._parse_source("A") == [bt._SOURCE_A]
|
|
assert bt._parse_source(" a ") == [bt._SOURCE_A]
|
|
|
|
def test_unknown_raises(self) -> None:
|
|
import pytest
|
|
|
|
with pytest.raises(ValueError):
|
|
bt._parse_source("C")
|
|
|
|
|
|
class TestPlanVariants:
|
|
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
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class _CaptureResult:
|
|
"""Stands in for a SQLAlchemy Result — returns canned rows from .all()."""
|
|
|
|
def __init__(self, rows: list) -> None:
|
|
self._rows = rows
|
|
|
|
def all(self) -> list:
|
|
return self._rows
|
|
|
|
|
|
class _CaptureSession:
|
|
"""Fake Session capturing (sql_text, params) and returning canned rows."""
|
|
|
|
def __init__(self, rows: list) -> None:
|
|
self.rows = rows
|
|
self.calls: list[tuple[str, dict]] = []
|
|
|
|
def execute(self, stmt: object, params: dict | None = None) -> _CaptureResult:
|
|
self.calls.append((str(stmt), dict(params or {})))
|
|
return _CaptureResult(self.rows)
|
|
|
|
|
|
class TestSourceBSqlShape:
|
|
def test_units_sql_uses_cast_not_double_colon(self) -> None:
|
|
sql = str(bt._SOURCE_B_UNITS_SQL)
|
|
assert "CAST(:premise_kind AS text)" in sql
|
|
assert "CAST(:since AS date)" in sql
|
|
# psycopg3-incompatible :name::type must NOT appear.
|
|
assert "::" not in sql
|
|
|
|
def test_units_sql_is_select_only(self) -> None:
|
|
sql = str(bt._SOURCE_B_UNITS_SQL).strip().lower()
|
|
assert sql.startswith("select")
|
|
for forbidden in ("insert", "update", "delete", "drop", "alter", "create"):
|
|
assert forbidden not in sql
|
|
|
|
def test_classes_sql_uses_cast_not_double_colon(self) -> None:
|
|
sql = str(bt._SOURCE_B_CLASSES_SQL)
|
|
assert "CAST(:premise_kind AS text)" in sql
|
|
assert "::" not in sql
|
|
|
|
def test_load_sales_by_month_binds_and_shapes(self) -> None:
|
|
ms = _months(3)
|
|
sess = _CaptureSession([(ms[0], 10), (ms[1], 20), (None, 99)])
|
|
out = bt.load_sales_by_month(
|
|
sess, # type: ignore[arg-type]
|
|
since="2019-01-01",
|
|
obj_class="комфорт",
|
|
district=None,
|
|
)
|
|
# None-month row dropped; rows mapped to {month: units}.
|
|
assert out == {ms[0]: 10, ms[1]: 20}
|
|
# Bound params include the class filter and premise kind (parametrised,
|
|
# not interpolated) — confirms no SQL-injection-prone string building.
|
|
_sql, params = sess.calls[0]
|
|
assert params["cls"] == "комфорт"
|
|
assert params["premise_kind"] == bt._PREMISE_KIND
|
|
assert params["since"] == "2019-01-01"
|
|
|
|
def test_load_classes_maps_rows(self) -> None:
|
|
sess = _CaptureSession([("комфорт",), ("бизнес",), (None,)])
|
|
out = bt.load_classes(sess, since="2019-01-01") # type: ignore[arg-type]
|
|
assert out == ["комфорт", "бизнес"]
|
|
|
|
|
|
class TestSourceASqlShape:
|
|
def test_units_sql_hits_corpus_room_month_table(self) -> None:
|
|
sql = str(bt._SOURCE_A_UNITS_SQL)
|
|
assert "objective_corpus_room_month" in sql
|
|
# Survivorship-free aggregate: SUM(deals_total_count) GROUP BY the month.
|
|
assert "SUM(crm.deals_total_count)" in sql
|
|
assert "GROUP BY 1" in sql
|
|
# report_month truncated to a month-first DATE.
|
|
assert "date_trunc('month', crm.report_month)" in sql
|
|
|
|
def test_units_sql_uses_cast_not_double_colon(self) -> None:
|
|
sql = str(bt._SOURCE_A_UNITS_SQL)
|
|
assert "CAST(:since AS date)" in sql
|
|
# Optional class filter folds case (capitalised in this table).
|
|
assert "LOWER(CAST(:cls AS text))" in sql
|
|
# psycopg3-incompatible :name::type must NOT appear.
|
|
assert "::" not in sql
|
|
|
|
def test_units_sql_is_select_only(self) -> None:
|
|
sql = str(bt._SOURCE_A_UNITS_SQL).strip().lower()
|
|
assert sql.startswith("select")
|
|
for forbidden in ("insert", "update", "delete", "drop", "alter", "create"):
|
|
assert forbidden not in sql
|
|
|
|
def test_classes_sql_uses_cast_not_double_colon(self) -> None:
|
|
sql = str(bt._SOURCE_A_CLASSES_SQL)
|
|
assert "objective_corpus_room_month" in sql
|
|
assert "CAST(:since AS date)" in sql
|
|
assert "::" not in sql
|
|
|
|
def test_load_sales_source_a_binds_and_shapes(self) -> None:
|
|
ms = _months(3)
|
|
sess = _CaptureSession([(ms[0], 100), (ms[1], 200), (None, 99)])
|
|
out = bt.load_sales_by_month_source_a(
|
|
sess, # type: ignore[arg-type]
|
|
since="2025-05-01",
|
|
obj_class="комфорт",
|
|
)
|
|
# None-month row dropped; rows mapped to {month: units}.
|
|
assert out == {ms[0]: 100, ms[1]: 200}
|
|
_sql, params = sess.calls[0]
|
|
# Parametrised — no premise_kind / district for Source A.
|
|
assert params["cls"] == "комфорт"
|
|
assert params["since"] == "2025-05-01"
|
|
assert "premise_kind" not in params
|
|
assert "district" not in params
|
|
|
|
def test_load_classes_source_a_maps_rows(self) -> None:
|
|
sess = _CaptureSession([("комфорт",), ("бизнес",), (None,)])
|
|
out = bt.load_classes_source_a(sess, since="2025-05-01") # type: ignore[arg-type]
|
|
assert out == ["комфорт", "бизнес"]
|
|
|
|
|
|
class TestSourceDispatch:
|
|
def test_load_sales_dispatch_routes_by_source(self) -> None:
|
|
ms = _months(2)
|
|
sess_b = _CaptureSession([(ms[0], 10)])
|
|
bt._load_sales(
|
|
sess_b, # type: ignore[arg-type]
|
|
source=bt._SOURCE_B,
|
|
since="2019-01-01",
|
|
obj_class=None,
|
|
district=None,
|
|
)
|
|
# Source B SQL carries the premise_kind bind.
|
|
_sql_b, params_b = sess_b.calls[0]
|
|
assert params_b["premise_kind"] == bt._PREMISE_KIND
|
|
|
|
sess_a = _CaptureSession([(ms[0], 99)])
|
|
bt._load_sales(
|
|
sess_a, # type: ignore[arg-type]
|
|
source=bt._SOURCE_A,
|
|
since="2025-05-01",
|
|
obj_class=None,
|
|
district=None,
|
|
)
|
|
# Source A SQL hits the corpus_room_month table, no premise_kind.
|
|
sql_a, params_a = sess_a.calls[0]
|
|
assert "objective_corpus_room_month" in sql_a
|
|
assert "premise_kind" not in params_a
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Local Δln helper (mirror sales_series.log_diff for building synthetic inputs)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _delta_ln(series: list[int]) -> list[float | None]:
|
|
"""Δln for synthetic inputs — uses the production log_diff via the engine."""
|
|
_bl, _ols, log_diff = bt._import_engine()
|
|
return log_diff(series)
|