Light1YT
5530f472b3
The OOS verdict flagged a variant 'candidate to promote' on hit-rate >= 0.5+margin + lag_stable alone. On thin data this over-claims: Source A Almon-ADL scored 6/10 (0.60) lag-stable and was flagged as signal, but P(X>=6|10,0.5)~=0.377 -- a coin flip. Live ground-truth confirmed no signal (full-sample R2~=0.003, wrong sign). Add exact stdlib-only one-sided binomial _binom_sf_ge + _VERDICT_ALPHA=0.05 and require P(X>=hits|n_test,0.5) < alpha in both verdict() and cross_source_verdict() on top of the effect-size margin. hits recovered exactly as round(hit_rate*n_test) (n_test==scored invariant; no evaluator shape change). Verdict text now states n_test + the binomial p on pass and fail. Evaluator/estimator math and the read-only SELECT discipline untouched. Refs #978.
1038 lines
46 KiB
Python
1038 lines
46 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
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# _binom_sf_ge — exact one-sided binomial survival (verdict significance gate)
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
class TestBinomSfGe:
|
||
def test_known_values(self) -> None:
|
||
# The #978 near-miss: 6/10 heads is NOT distinguishable from a fair coin.
|
||
assert math.isclose(bt._binom_sf_ge(6, 10, 0.5), 0.376953125, abs_tol=1e-9)
|
||
# A clearly-significant tail.
|
||
assert math.isclose(bt._binom_sf_ge(9, 10, 0.5), 0.0107421875, abs_tol=1e-9)
|
||
# 5/5 perfect over a tiny window is just barely significant (p < 0.05).
|
||
assert math.isclose(bt._binom_sf_ge(5, 5, 0.5), 0.03125, abs_tol=1e-12)
|
||
|
||
def test_k_zero_or_below_is_one(self) -> None:
|
||
# P(X ≥ 0) = 1 trivially; negative k clamps to 0 → 1.0.
|
||
assert bt._binom_sf_ge(0, 10, 0.5) == 1.0
|
||
assert bt._binom_sf_ge(-3, 10, 0.5) == 1.0
|
||
|
||
def test_n_zero_returns_one(self) -> None:
|
||
# No trials → no evidence against the null → 1.0 (never promotes).
|
||
assert bt._binom_sf_ge(3, 0, 0.5) == 1.0
|
||
assert bt._binom_sf_ge(0, 0, 0.5) == 1.0
|
||
|
||
def test_k_clamped_to_n(self) -> None:
|
||
# k > n clamps to n → P(X ≥ n) = p^n (only the all-success term).
|
||
assert math.isclose(bt._binom_sf_ge(20, 10, 0.5), 0.5**10, abs_tol=1e-12)
|
||
# k == n → exactly the all-success probability.
|
||
assert math.isclose(bt._binom_sf_ge(4, 4, 0.5), 0.0625, abs_tol=1e-12)
|
||
|
||
def test_full_distribution_sums_to_one(self) -> None:
|
||
# P(X ≥ 0) over all i must be 1 for any n (sanity on the comb sum).
|
||
for n in (1, 3, 7, 12, 35):
|
||
assert math.isclose(bt._binom_sf_ge(0, n, 0.5), 1.0, abs_tol=1e-9)
|
||
|
||
def test_non_half_p(self) -> None:
|
||
# Works for p ≠ 0.5: P(X ≥ 1 | n=2, p=0.1) = 1 − (0.9)^2 = 0.19.
|
||
assert math.isclose(bt._binom_sf_ge(1, 2, 0.1), 0.19, abs_tol=1e-12)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# _time_ordered_split
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
class TestTimeOrderedSplit:
|
||
def test_basic_fraction(self) -> None:
|
||
assert bt._time_ordered_split(100, 0.7) == 70
|
||
assert bt._time_ordered_split(30, 0.7) == 21
|
||
|
||
def test_keeps_one_month_each_side(self) -> None:
|
||
# frac=1.0 would put everything in train → clamp to n-1 so test has ≥1.
|
||
assert bt._time_ordered_split(10, 1.0) == 9
|
||
# frac=0.0 would empty train → clamp to ≥1.
|
||
assert bt._time_ordered_split(10, 0.0) == 1
|
||
|
||
def test_degenerate_sizes(self) -> None:
|
||
assert bt._time_ordered_split(0, 0.7) == 0
|
||
assert bt._time_ordered_split(1, 0.7) == 1 # nothing to split
|
||
|
||
def test_is_time_ordered_not_parity(self) -> None:
|
||
# The split is a single boundary index (past→train, future→test), NOT a
|
||
# parity/random partition: train is a contiguous prefix.
|
||
n_train = bt._time_ordered_split(20, 0.7)
|
||
assert n_train == 14 # contiguous prefix [0:14], test [14:20]
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# _rate_first_diff / _shift_for_lag / align_series
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
class TestRateFirstDiff:
|
||
def test_first_diff(self) -> None:
|
||
assert bt._rate_first_diff([10.0, 12.0, 11.0]) == [None, 2.0, -1.0]
|
||
|
||
def test_none_breaks_pair(self) -> None:
|
||
assert bt._rate_first_diff([1.0, None, 3.0]) == [None, None, None]
|
||
|
||
def test_empty_and_single(self) -> None:
|
||
assert bt._rate_first_diff([]) == [None]
|
||
assert bt._rate_first_diff([5.0]) == [None]
|
||
|
||
|
||
class TestShiftForLag:
|
||
def test_lag_zero_is_identity(self) -> None:
|
||
assert bt._shift_for_lag([1.0, 2.0, 3.0], 0) == [1.0, 2.0, 3.0]
|
||
|
||
def test_lag_shifts_right_and_truncates(self) -> None:
|
||
# y[t] ← x[t-2]: two leading None, length preserved.
|
||
assert bt._shift_for_lag([1.0, 2.0, 3.0, 4.0], 2) == [None, None, 1.0, 2.0]
|
||
|
||
def test_no_future_leak(self) -> None:
|
||
# Element at index t must equal the ORIGINAL element at t-lag (never t+k).
|
||
x = [10.0, 20.0, 30.0, 40.0, 50.0]
|
||
lag = 1
|
||
shifted = bt._shift_for_lag(x, lag)
|
||
for t in range(lag, len(x)):
|
||
assert shifted[t] == x[t - lag]
|
||
|
||
|
||
class TestDetrendLog:
|
||
def test_removes_known_linear_trend(self) -> None:
|
||
# units = exp(a + b·t): a PURE log-linear trend → residuals must be ~0.
|
||
a, b = 6.0, 0.05
|
||
units = [round(math.exp(a + b * t)) for t in range(24)]
|
||
resid = bt._detrend_log(units)
|
||
assert all(r is not None for r in resid)
|
||
# Rounding to int adds tiny noise, but residuals collapse near zero.
|
||
assert max(abs(r) for r in resid) < 0.01 # type: ignore[arg-type, type-var]
|
||
|
||
def test_residuals_isolate_signal_over_trend(self) -> None:
|
||
# Trend + a single oscillation: after detrend the trend is gone and the
|
||
# residual variance is dominated by the oscillation, not the drift.
|
||
n = 30
|
||
base_units = [math.exp(6.0 + 0.08 * t + 0.3 * math.sin(t)) for t in range(n)]
|
||
units = [max(1, round(u)) for u in base_units]
|
||
resid = bt._detrend_log(units)
|
||
finite = [r for r in resid if r is not None]
|
||
# Detrended series is NOT monotone (the drift dominated the raw logs).
|
||
diffs = [finite[i] - finite[i - 1] for i in range(1, len(finite))]
|
||
assert any(d > 0 for d in diffs) and any(d < 0 for d in diffs)
|
||
|
||
def test_none_and_nonpositive_map_to_none(self) -> None:
|
||
vals = [100, None, 0, -5, 120, 130, 140]
|
||
resid = bt._detrend_log(vals)
|
||
assert len(resid) == len(vals)
|
||
assert resid[1] is None # None in
|
||
assert resid[2] is None # 0 → ln undefined
|
||
assert resid[3] is None # negative → ln undefined
|
||
# The finite positions stay finite.
|
||
assert resid[0] is not None and resid[4] is not None
|
||
|
||
def test_short_series_passthrough_is_logs(self) -> None:
|
||
# <3 finite points → can't fit a line → passthrough of ln(values).
|
||
vals = [10, 20]
|
||
resid = bt._detrend_log(vals)
|
||
assert resid[0] is not None and math.isclose(resid[0], math.log(10))
|
||
assert resid[1] is not None and math.isclose(resid[1], math.log(20))
|
||
|
||
def test_short_after_filtering_passthrough(self) -> None:
|
||
# Only 2 finite points after dropping None/≤0 → passthrough of logs.
|
||
vals = [None, 50, 0, 60]
|
||
resid = bt._detrend_log(vals)
|
||
assert resid[0] is None and resid[2] is None
|
||
assert resid[1] is not None and math.isclose(resid[1], math.log(50))
|
||
assert resid[3] is not None and math.isclose(resid[3], math.log(60))
|
||
|
||
def test_length_preserved(self) -> None:
|
||
vals = [100 + i for i in range(10)]
|
||
assert len(bt._detrend_log(vals)) == 10
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Look-ahead leakage fix (#978 Part A) — detrend trend fit on TRAIN months only,
|
||
# projected point-in-time onto test (never fit on train+test together).
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
class TestDetrendNoLeakage:
|
||
def test_train_only_fit_matches_manual_polyfit_on_train_slice(self) -> None:
|
||
# With fit_n given, the trend (a, b) must be the polyfit of ONLY the
|
||
# finite points in [0:fit_n] — the test months must not enter the fit.
|
||
n, fit_n = 30, 20
|
||
units = [max(1, round(math.exp(6.0 + 0.05 * t))) for t in range(n)]
|
||
logs = [math.log(u) for u in units]
|
||
# Manual train-only line.
|
||
xs = list(range(fit_n))
|
||
ys = logs[:fit_n]
|
||
b, a = bt.np.polyfit(bt.np.array(xs, dtype=float), bt.np.array(ys, dtype=float), 1)
|
||
resid = bt._detrend_log(units, fit_n=fit_n)
|
||
# Every residual equals ln(u_t) − (a + b·t) with the TRAIN-fitted line,
|
||
# INCLUDING the test months (the line is projected forward, not refit).
|
||
for t in range(n):
|
||
assert resid[t] is not None
|
||
assert math.isclose(resid[t], logs[t] - (a + b * t), abs_tol=1e-9) # type: ignore[arg-type]
|
||
|
||
def test_test_points_do_not_shape_the_trend(self) -> None:
|
||
# A BROKEN trend: gentle slope on train, steep slope on test. A full-sample
|
||
# (leaky) fit is pulled UP by the steep test tail; a train-only fit is not.
|
||
# So the residual at the LAST month must differ between the two — proving
|
||
# the test observations leak into the leaky fit but not the train-only one.
|
||
n, fit_n = 24, 16
|
||
units: list[int] = []
|
||
for t in range(n):
|
||
slope = 0.02 if t < fit_n else 0.20 # trend break at fit_n
|
||
base = 0.02 * min(t, fit_n)
|
||
extra = 0.20 * max(0, t - fit_n)
|
||
units.append(max(1, round(math.exp(6.0 + base + extra)) if t else round(math.exp(6.0))))
|
||
_ = slope
|
||
leaky = bt._detrend_log(units) # fit_n=None → fit on train+test (leaks)
|
||
safe = bt._detrend_log(units, fit_n=fit_n)
|
||
# Last test month residual differs → the steep tail moved the leaky line
|
||
# but not the train-only line.
|
||
assert leaky[-1] is not None and safe[-1] is not None
|
||
assert abs(leaky[-1] - safe[-1]) > 0.05 # type: ignore[operator]
|
||
|
||
def test_fit_n_gates_passthrough_on_train_point_count(self) -> None:
|
||
# Plenty of finite points overall, but only 2 (< _DETREND_MIN_POINTS) fall
|
||
# inside the TRAIN window → a line is not identifiable on TRAIN → passthrough
|
||
# of the logs (residual == ln(value)), exactly like the raw log_diff path.
|
||
units = [10, 20] + [30 + i for i in range(10)] # 12 finite, fit_n=2
|
||
resid = bt._detrend_log(units, fit_n=2)
|
||
assert resid[0] is not None and math.isclose(resid[0], math.log(10))
|
||
assert resid[1] is not None and math.isclose(resid[1], math.log(20))
|
||
# Passthrough applies to ALL positions (no trend was removed anywhere).
|
||
assert resid[2] is not None and math.isclose(resid[2], math.log(30))
|
||
|
||
def test_backtest_tier_detrend_fits_train_only(self) -> None:
|
||
# End-to-end: backtest_tier must pass n_train as fit_n. We assert the
|
||
# detrended regressand it builds equals the one from a TRAIN-only detrend,
|
||
# and is NOT equal to the leaky full-sample detrend (when they differ).
|
||
n = 40
|
||
ms = _months(n)
|
||
# Trend-confounded units with a real lag-2 signal (#978b-style series).
|
||
rate = _zero_drift_rate_levels(n, seed=5)
|
||
units = _units_from_rate_with_trend(rate, lag=2, beta=-0.06, trend_per_month=0.09)
|
||
sales = {ms[i]: units[i] for i in range(n)}
|
||
rate_by = {ms[i]: rate[i] for i in range(n)}
|
||
|
||
# What backtest_tier should build internally (train-only fit).
|
||
n_train = bt._time_ordered_split(n, 0.7)
|
||
expected = bt._delta_sales_series(units, detrend=True, fit_n=n_train)
|
||
leaky = bt._delta_sales_series(units, detrend=True, fit_n=None)
|
||
|
||
# Run the tier and reconstruct its regressand path via the same helper to
|
||
# confirm n_train is threaded through (the public API has no hook, so we
|
||
# assert the train-only and full-sample series genuinely differ — i.e. the
|
||
# fix is observable — and that the tier still produces a scored result).
|
||
res = bt.backtest_tier(sales, rate_by, tier=bt._EKB_WIDE, detrend=True, holdout_frac=0.7)
|
||
assert res.skipped is None
|
||
assert res.detrended is True
|
||
# The two regressands must differ somewhere in the test region (leakage is
|
||
# observable), so the train-only fix is a real behavioural change.
|
||
assert any(
|
||
e is not None and lk is not None and abs(e - lk) > 1e-9
|
||
for e, lk in zip(expected[n_train:], leaky[n_train:], strict=False)
|
||
)
|
||
|
||
def test_no_leakage_oos_hit_rate_not_above_leaky(self) -> None:
|
||
# The core claim: look-ahead leakage INFLATES the detrended OOS hit-rate.
|
||
# On a trend-confounded series, the train-only (correct) detrend must give
|
||
# an OOS hit-rate ≤ the leaky full-sample detrend. We compare evaluate_oos
|
||
# on both regressands over the SAME aligned series.
|
||
n = 48
|
||
rate = _zero_drift_rate_levels(n, seed=11)
|
||
units = _units_from_rate_with_trend(rate, lag=2, beta=-0.05, trend_per_month=0.07)
|
||
rate_deltas = bt._rate_first_diff(rate)
|
||
n_train = bt._time_ordered_split(n, 0.7)
|
||
|
||
safe_sales = bt._delta_sales_series(units, detrend=True, fit_n=n_train)
|
||
leaky_sales = bt._delta_sales_series(units, detrend=True, fit_n=None)
|
||
safe = bt.evaluate_oos(safe_sales, rate_deltas, holdout_frac=0.7)
|
||
leaky = bt.evaluate_oos(leaky_sales, rate_deltas, holdout_frac=0.7)
|
||
|
||
# Both should find a gated lag here; if either is None the inequality is
|
||
# vacuously fine (no inflation possible). When both score, leakage may only
|
||
# help (or tie) the leaky run — it must never make the corrected run higher.
|
||
if safe["oos_hit_rate"] is not None and leaky["oos_hit_rate"] is not None:
|
||
assert safe["oos_hit_rate"] <= leaky["oos_hit_rate"] + 1e-9
|
||
|
||
|
||
class TestAlignSeries:
|
||
def test_inner_join_by_month(self) -> None:
|
||
ms = _months(4)
|
||
sales = {ms[0]: 100, ms[1]: 110, ms[2]: 120, ms[3]: 130}
|
||
# rate missing ms[0]; has an extra month not in sales.
|
||
rate = {ms[1]: 7.0, ms[2]: 7.5, ms[3]: 8.0, dt.date(2030, 1, 1): 9.0}
|
||
months, units, rates = bt.align_series(sales, rate)
|
||
assert months == [ms[1], ms[2], ms[3]] # intersection only, ascending
|
||
assert units == [110, 120, 130]
|
||
assert rates == [7.0, 7.5, 8.0]
|
||
|
||
def test_empty_intersection(self) -> None:
|
||
months, units, rates = bt.align_series({_months(1)[0]: 1}, {dt.date(2030, 1, 1): 2.0})
|
||
assert months == [] and units == [] and rates == []
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# evaluate_oos — the core OOS metric
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
class TestEvaluateOos:
|
||
def test_injected_signal_high_oos_hit_rate(self) -> None:
|
||
# sales react to rate at lag 2 with a clean negative β → the TRAIN fit
|
||
# should generalise: nearly every TEST month's predicted sign matches.
|
||
n = 48
|
||
rate = _aperiodic_rate_levels(n)
|
||
units = _units_from_rate(rate, lag=2, beta=-0.05)
|
||
delta_sales = _delta_ln(units)
|
||
rate_deltas = bt._rate_first_diff(rate)
|
||
|
||
res = bt.evaluate_oos(delta_sales, rate_deltas, holdout_frac=0.7)
|
||
assert res["train_lag"] == 2
|
||
assert res["train_beta"] is not None and res["train_beta"] < 0
|
||
assert res["oos_hit_rate"] is not None
|
||
# A real injected signal → directional hit-rate clearly beats a coin flip.
|
||
assert res["oos_hit_rate"] >= 0.8
|
||
# In-sample R² is high by construction (reported, not trusted).
|
||
assert res["in_sample_r2"] is not None and res["in_sample_r2"] > 0.9
|
||
# Lag stable: full-sample refit finds the same lag.
|
||
assert res["full_sample_lag"] == 2
|
||
assert res["lag_stable"] is True
|
||
|
||
def test_pure_noise_hit_rate_near_coin_flip(self) -> None:
|
||
# No rate→sales link: sales are an independent aperiodic walk. Either no
|
||
# gated lag is found on TRAIN (→ None), or any spurious fit predicts
|
||
# direction no better than a coin flip on held-out months.
|
||
n = 60
|
||
rate = _aperiodic_rate_levels(n, seed=1)
|
||
noise = _aperiodic_rate_levels(n, seed=999) # uncorrelated second series
|
||
units = [max(1, round(1000.0 * math.exp(0.01 * (v - 10.0)))) for v in noise]
|
||
delta_sales = _delta_ln(units)
|
||
rate_deltas = bt._rate_first_diff(rate)
|
||
|
||
res = bt.evaluate_oos(delta_sales, rate_deltas, holdout_frac=0.7)
|
||
hr = res["oos_hit_rate"]
|
||
# Honest outcome: no signal → either ungated (None) or ~coin-flip.
|
||
assert hr is None or hr <= 0.7
|
||
|
||
def test_too_few_months_returns_empty(self) -> None:
|
||
# 1 month → can't split → empty result (all metrics None, not a crash).
|
||
res = bt.evaluate_oos([None], [None], holdout_frac=0.7)
|
||
assert res["train_lag"] is None
|
||
assert res["oos_hit_rate"] is None
|
||
assert res["n_train"] == 1 and res["n_test"] == 0
|
||
|
||
def test_no_gated_lag_on_train_returns_empty(self) -> None:
|
||
# Positive rate→sales link (β>0) → engine gate (slope<0) rejects every
|
||
# lag on TRAIN → nothing to validate → empty (None) result, no crash.
|
||
n = 40
|
||
rate = _aperiodic_rate_levels(n)
|
||
units = _units_from_rate(rate, lag=1, beta=+0.05) # wrong sign
|
||
delta_sales = _delta_ln(units)
|
||
rate_deltas = bt._rate_first_diff(rate)
|
||
res = bt.evaluate_oos(delta_sales, rate_deltas, holdout_frac=0.7)
|
||
assert res["train_lag"] is None
|
||
assert res["oos_hit_rate"] is None
|
||
|
||
def test_point_in_time_no_future_leak(self) -> None:
|
||
# Build a signal, then confirm the TEST prediction at the FIRST test
|
||
# month uses only rate data at or before it. We reconstruct the expected
|
||
# prediction from the public _shift_for_lag and check evaluate_oos's MAE
|
||
# is finite (a future leak would mismatch lengths / shift indices).
|
||
n = 36
|
||
rate = _aperiodic_rate_levels(n)
|
||
units = _units_from_rate(rate, lag=3, beta=-0.04)
|
||
delta_sales = _delta_ln(units)
|
||
rate_deltas = bt._rate_first_diff(rate)
|
||
res = bt.evaluate_oos(delta_sales, rate_deltas, holdout_frac=0.7)
|
||
assert res["oos_signed_mae"] is not None
|
||
assert math.isfinite(res["oos_signed_mae"])
|
||
# First scored test month index = n_train; predictor must be Δrate[t-lag].
|
||
lag = res["train_lag"]
|
||
assert lag is not None
|
||
shifted = bt._shift_for_lag(rate_deltas, lag)
|
||
# The shifted regressor at the first test index is at or before it.
|
||
assert shifted[res["n_train"]] is None or isinstance(shifted[res["n_train"]], float)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 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_and_significant(self) -> None:
|
||
# hit-rate clears 0.5+margin, lag stable, AND a wide-enough window makes it
|
||
# statistically significant: hits=round(0.71·35)=25, P(X≥25|35)≈0.008<0.05.
|
||
vd = bt.verdict(_tier(oos_hit_rate=0.71, n_test=35, n_train=80, lag_stable=True))
|
||
assert vd["promote"] is True
|
||
assert "OOS predictive value" in vd["reason"]
|
||
# The promote message exposes the significance p (#978 transparency).
|
||
assert "binomial p=" in vd["reason"]
|
||
assert "n_test=35" 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_does_not_promote_six_of_ten_not_significant(self) -> None:
|
||
# REGRESSION GUARD — the exact #978 near-miss. Source A Almon-ADL scored
|
||
# oos_hit_rate=0.60 with n_test=10 (6/10) and lag_stable. The OLD rule
|
||
# (hit-rate ≥ 0.5+margin AND lag_stable) over-claimed "candidate to
|
||
# promote". But P(X≥6|10, 0.5)≈0.377 ≥ 0.05 — indistinguishable from a
|
||
# coin flip. The significance gate MUST keep it advisory.
|
||
vd = bt.verdict(_tier(oos_hit_rate=0.60, n_test=10, n_train=23, lag_stable=True))
|
||
assert vd["promote"] is False
|
||
assert "keep advisory" in vd["reason"]
|
||
assert "not significant" in vd["reason"]
|
||
# The message names n_test and the binomial p so the WHY is transparent.
|
||
assert "n_test=10" in vd["reason"]
|
||
assert "p=0.38" in vd["reason"]
|
||
|
||
def test_small_n_perfect_score_does_not_promote(self) -> None:
|
||
# A tiny window at 100% still can't promote: P(X≥4|4, 0.5)=0.0625 ≥ 0.05.
|
||
# Proves a perfect-but-thin run is not enough to clear significance.
|
||
vd = bt.verdict(_tier(oos_hit_rate=1.0, n_test=4, n_train=10, lag_stable=True))
|
||
assert vd["promote"] is False
|
||
assert "not significant" in vd["reason"]
|
||
assert "n_test=4" in vd["reason"]
|
||
|
||
def test_thin_warning_set_but_significant_still_promotes(self) -> None:
|
||
# A small window (n_test=5 < 6) sets the thin_warning, but 5/5 is the
|
||
# smallest perfect window that IS significant: P(X≥5|5, 0.5)=0.03125<0.05.
|
||
# So it promotes AND carries the thin caveat — the caveat is advisory,
|
||
# significance is the hard gate.
|
||
vd = bt.verdict(_tier(oos_hit_rate=1.0, n_test=5, n_train=13, lag_stable=True))
|
||
assert vd["promote"] is True
|
||
assert vd["thin_warning"] is not None
|
||
assert "small" in vd["thin_warning"]
|
||
|
||
def test_thin_window_high_rate_blocked_by_significance(self) -> None:
|
||
# The original "thin window" scenario (hit-rate=0.9, n_test=3): under the
|
||
# stricter rule it does NOT promote — hits=round(2.7)=3, P(X≥3|3)=0.125.
|
||
vd = bt.verdict(_tier(oos_hit_rate=0.9, n_test=3, lag_stable=True))
|
||
assert vd["promote"] is False
|
||
assert "not significant" in vd["reason"]
|
||
|
||
|
||
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"] == []
|
||
|
||
def test_six_of_ten_not_significant_no_signal(self) -> None:
|
||
# REGRESSION GUARD (#978) — the same near-miss in the cross-source path:
|
||
# a Source A row at oos_hit_rate=0.60, n_test=10, lag_stable=True must NOT
|
||
# count as signal (P(X≥6|10)≈0.377 ≥ 0.05). The gate applies in BOTH the
|
||
# per-variant verdict() and cross_source_verdict().
|
||
runs = [
|
||
_run(
|
||
bt._SOURCE_A,
|
||
False,
|
||
_tier(source=bt._SOURCE_A, oos_hit_rate=0.60, n_test=10, n_train=23),
|
||
),
|
||
]
|
||
cv = bt.cross_source_verdict(runs)
|
||
assert cv["promote_any"] is False
|
||
assert cv["signal_variants"] == []
|
||
# The rendered line spells out the failed-significance reason + the p.
|
||
row = cv["rows"][0]
|
||
assert row["significant"] is False
|
||
assert row["beats_coin"] is False
|
||
joined = "\n".join(cv["lines"])
|
||
assert "not significant" in joined
|
||
|
||
def test_significant_wide_window_counts_as_signal(self) -> None:
|
||
# A genuinely-significant detrended variant (hit-rate=0.71 over n_test=35,
|
||
# lag stable) DOES count as signal: P(X≥25|35)≈0.008 < 0.05.
|
||
runs = [
|
||
_run(
|
||
bt._SOURCE_B,
|
||
True,
|
||
_tier(detrended=True, oos_hit_rate=0.71, n_test=35, n_train=80),
|
||
),
|
||
]
|
||
cv = bt.cross_source_verdict(runs)
|
||
assert cv["promote_any"] is True
|
||
assert "B detrended" in cv["signal_variants"]
|
||
assert cv["rows"][0]["significant"] is True
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 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)
|