477 lines
19 KiB
Python
477 lines
19 KiB
Python
"""Unit tests for the read-only §9.6 rate-sensitivity backtest (Forgejo #978).
|
|
|
|
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)
|
|
- 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
|
|
- verdict / tier_lift — promotion criterion, coin-flip baseline, lag stability
|
|
|
|
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.
|
|
|
|
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
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# _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 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
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# verdict / tier_lift
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _tier(
|
|
*,
|
|
tier: str = bt._EKB_WIDE,
|
|
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,
|
|
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("Комфорт, Бизнес ,премиум") == ["комфорт", "бизнес", "премиум"]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 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 == ["комфорт", "бизнес"]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 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)
|