Light1YT
3ecfc2d7dc
REOPENED — normalize.py was never created; only rate-regime discount existed. New backend/app/services/forecasting/normalize.py with normalize_demand(series): multiplicative month-of-year deseasonalization of the raw monthly demand SalesSeries (§9.4). Pure/deterministic; min-data guard (<2 full years / empty month / overall_mean<=0 → factor 1.0, no divide-by-zero, no thin-data noise). Exposes seasonal factors for explainability. Synthetic unit test: seasonality removed (month means equalised), flat unchanged, thin/empty/all-zero safe. DoD (module + doc + test) MET. Production wiring into rate_sensitivity._align_sales_deltas DEFERRED (documented TODO): deseasonalizing the short rate-driven series perturbs the recovered β/lag on current data — needs a points-per-month gate / joint seasonal+rate estimation + backtest before wiring. Forecast stack is advisory regardless. Refs #979
241 lines
11 KiB
Python
241 lines
11 KiB
Python
"""Unit-тесты сезонной (month-of-year) нормализации спроса (#979, ТЗ §9.4).
|
||
|
||
Синтетика, без живой БД. Гейт DoD:
|
||
• flat baseline × известный month-of-year паттерн (лето +30%, зима −30%) за 2-3
|
||
года → после normalize_demand средние по календарным месяцам ~равны (сезонность
|
||
снята, within tolerance);
|
||
• плоский ряд → не меняется;
|
||
• тонкие данные (< 2 полных лет) → возвращаются ~без изменений (факторы = 1.0, без
|
||
blowup / деления на ноль);
|
||
• пустой ряд → пустой (без crash);
|
||
• factors / applied / n_full_years семантика + ортогональность прочих полей.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import statistics
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
from datetime import date
|
||
|
||
from app.services.forecasting.normalize import (
|
||
SeasonalAdjustment,
|
||
deseasonalize_values,
|
||
normalize_demand,
|
||
seasonal_factors,
|
||
)
|
||
from app.services.forecasting.sales_series import SalesSeries
|
||
|
||
# ── synthetic helpers ─────────────────────────────────────────────────────────
|
||
|
||
# Известный month-of-year паттерн: лето (июнь-авг) +30%, зима (дек-фев) −30%,
|
||
# остальное ≈ базовый уровень. Сумма множителей вокруг 1.0 → overall_mean ≈ baseline.
|
||
_SEASONAL_PATTERN: dict[int, float] = {
|
||
1: 0.70, # январь −30%
|
||
2: 0.70, # февраль −30%
|
||
3: 1.00,
|
||
4: 1.00,
|
||
5: 1.00,
|
||
6: 1.30, # июнь +30%
|
||
7: 1.30, # июль +30%
|
||
8: 1.30, # август +30%
|
||
9: 1.00,
|
||
10: 1.00,
|
||
11: 1.00,
|
||
12: 0.70, # декабрь −30%
|
||
}
|
||
|
||
_BASELINE_UNITS: int = 100
|
||
|
||
|
||
def _months(n_months: int, start: date = date(2021, 1, 1)) -> list[date]:
|
||
"""n_months подряд идущих 1-х чисел месяца, начиная со start."""
|
||
out: list[date] = []
|
||
y, m = start.year, start.month
|
||
for _ in range(n_months):
|
||
out.append(date(y, m, 1))
|
||
m += 1
|
||
if m > 12:
|
||
m = 1
|
||
y += 1
|
||
return out
|
||
|
||
|
||
def _seasonal_units(months: list[date], baseline: int = _BASELINE_UNITS) -> list[int]:
|
||
"""baseline × month-of-year паттерн, округлённый к int (сырой сезонный спрос)."""
|
||
return [round(baseline * _SEASONAL_PATTERN[d.month]) for d in months]
|
||
|
||
|
||
def _make_series(months: list[date], units: list[int]) -> SalesSeries:
|
||
"""SalesSeries-обёртка для теста (прочие поля — нейтральные заглушки)."""
|
||
return SalesSeries(
|
||
months=months,
|
||
units=units,
|
||
area_m2=[None] * len(months),
|
||
avg_price_per_m2=[None] * len(months),
|
||
n_months=len(months),
|
||
source="corpus_room_month",
|
||
segment={"obj_class": None, "room_bucket": None, "district": None, "price_bucket": None},
|
||
confidence="high",
|
||
)
|
||
|
||
|
||
def _month_of_year_means(months: list[date], units: list[float | None]) -> dict[int, float]:
|
||
"""Среднее значение по каждому календарному месяцу (1..12) — для assert «снято»."""
|
||
buckets: dict[int, list[float]] = {}
|
||
for d, v in zip(months, units, strict=False):
|
||
if v is None:
|
||
continue
|
||
buckets.setdefault(d.month, []).append(float(v))
|
||
return {m: statistics.mean(vs) for m, vs in buckets.items() if vs}
|
||
|
||
|
||
# ── seasonal_factors (pure) ───────────────────────────────────────────────────
|
||
|
||
|
||
class TestSeasonalFactors:
|
||
def test_recovers_known_pattern_factors(self) -> None:
|
||
# 3 полных года сезонного ряда → факторы должны восстановить паттерн.
|
||
months = _months(36)
|
||
units = _seasonal_units(months)
|
||
adj = seasonal_factors(months, units)
|
||
assert isinstance(adj, SeasonalAdjustment)
|
||
assert adj.applied is True
|
||
assert adj.n_full_years == 3
|
||
# Фактор каждого месяца ≈ его множитель паттерна (overall_mean ≈ baseline).
|
||
for m, expected in _SEASONAL_PATTERN.items():
|
||
assert abs(adj.factors[m] - expected) < 0.05, (m, adj.factors[m], expected)
|
||
|
||
def test_thin_data_under_two_years_neutral(self) -> None:
|
||
# 18 мес < 2 полных лет → все факторы 1.0, applied=False (шум не ловим).
|
||
months = _months(18)
|
||
units = _seasonal_units(months)
|
||
adj = seasonal_factors(months, units)
|
||
assert adj.applied is False
|
||
assert adj.n_full_years == 1
|
||
assert all(f == 1.0 for f in adj.factors.values())
|
||
|
||
def test_flat_series_factors_all_one(self) -> None:
|
||
# Плоский ряд (без сезонности) → каждый month_mean == overall_mean → 1.0.
|
||
months = _months(36)
|
||
units = [_BASELINE_UNITS] * 36
|
||
adj = seasonal_factors(months, units)
|
||
# applied может быть False (все ровно 1.0) — сезонности нет.
|
||
assert all(abs(f - 1.0) < 1e-9 for f in adj.factors.values())
|
||
|
||
def test_empty_series_neutral(self) -> None:
|
||
adj = seasonal_factors([], [])
|
||
assert adj.applied is False
|
||
assert adj.n_full_years == 0
|
||
assert all(f == 1.0 for f in adj.factors.values())
|
||
|
||
def test_all_zero_series_neutral_no_div_zero(self) -> None:
|
||
# overall_mean == 0 → нейтраль, без деления на ноль.
|
||
months = _months(36)
|
||
adj = seasonal_factors(months, [0] * 36)
|
||
assert adj.applied is False
|
||
assert all(f == 1.0 for f in adj.factors.values())
|
||
|
||
def test_month_with_no_observations_factor_one(self) -> None:
|
||
# Guard «месяц без наблюдений → фактор 1.0» (defensive): март всегда пуст
|
||
# (None). Чтобы пройти year-guard и реально дойти до per-month ветки,
|
||
# ослабляем порог (min_full_years=0). Остальные месяцы получают свои факторы,
|
||
# март — нейтраль (нет базы оценки), без KeyError / деления на пустое среднее.
|
||
months = _months(36)
|
||
units = _seasonal_units(months)
|
||
vals: list[float | int | None] = [
|
||
(None if d.month == 3 else u) for d, u in zip(months, units, strict=False)
|
||
]
|
||
adj = seasonal_factors(months, vals, min_full_years=0)
|
||
assert adj.factors[3] == 1.0 # март без наблюдений → нейтраль
|
||
# Соседний июль наблюдается → его фактор НЕ нейтральный (паттерн +30%).
|
||
assert abs(adj.factors[7] - 1.3) < 0.1
|
||
|
||
|
||
# ── deseasonalize_values (pure) ───────────────────────────────────────────────
|
||
|
||
|
||
class TestDeseasonalizeValues:
|
||
def test_divides_by_factor(self) -> None:
|
||
months = [date(2021, 1, 1), date(2021, 7, 1)]
|
||
values: list[float | int | None] = [70, 130]
|
||
factors = {1: 0.7, 7: 1.3}
|
||
out = deseasonalize_values(months, values, factors)
|
||
assert out[0] is not None and abs(out[0] - 100.0) < 1e-9
|
||
assert out[1] is not None and abs(out[1] - 100.0) < 1e-9
|
||
|
||
def test_none_preserved(self) -> None:
|
||
months = [date(2021, 1, 1), date(2021, 2, 1)]
|
||
out = deseasonalize_values(months, [None, 70], {1: 0.7, 2: 0.7})
|
||
assert out[0] is None
|
||
assert out[1] is not None and abs(out[1] - 100.0) < 1e-9
|
||
|
||
def test_nonpositive_factor_falls_back_to_one(self) -> None:
|
||
# Фактор ≤ 0 / отсутствующий месяц → нейтраль 1.0 (без деления на 0/neg).
|
||
months = [date(2021, 1, 1), date(2021, 2, 1)]
|
||
out = deseasonalize_values(months, [50, 60], {1: 0.0, 2: -1.0})
|
||
assert out == [50.0, 60.0]
|
||
|
||
|
||
# ── normalize_demand (public API, SalesSeries shape) ──────────────────────────
|
||
|
||
|
||
class TestNormalizeDemand:
|
||
def test_seasonality_removed_month_means_equalised(self) -> None:
|
||
# DoD GATE: сезонный ряд за 3 года → после normalize month-of-year средние
|
||
# ~равны (within tolerance), т.е. сезонность снята.
|
||
months = _months(36)
|
||
units = _seasonal_units(months)
|
||
raw_means = _month_of_year_means(months, [float(u) for u in units])
|
||
# До нормализации: лето заметно выше зимы.
|
||
assert max(raw_means.values()) / min(raw_means.values()) > 1.5
|
||
|
||
result = normalize_demand(_make_series(months, units))
|
||
assert isinstance(result, SalesSeries)
|
||
adj_means = _month_of_year_means(months, [float(u) for u in result.units])
|
||
# После: разброс месячных средних схлопнут (≤ ~5% спред — остаток округления).
|
||
spread = max(adj_means.values()) / min(adj_means.values())
|
||
assert spread < 1.05, (spread, adj_means)
|
||
|
||
def test_flat_series_unchanged(self) -> None:
|
||
months = _months(36)
|
||
units = [_BASELINE_UNITS] * 36
|
||
result = normalize_demand(_make_series(months, units))
|
||
assert result.units == units # плоский ряд не тронут
|
||
|
||
def test_thin_data_returned_unchanged(self) -> None:
|
||
# < 2 полных лет → факторы 1.0 → ряд возвращается без изменений (тот же объект).
|
||
months = _months(18)
|
||
units = _seasonal_units(months)
|
||
series = _make_series(months, units)
|
||
result = normalize_demand(series)
|
||
assert result is series # short-circuit: applied=False → возвращаем как есть
|
||
assert result.units == units
|
||
|
||
def test_empty_series_empty_no_crash(self) -> None:
|
||
empty = _make_series([], [])
|
||
result = normalize_demand(empty)
|
||
assert result.months == []
|
||
assert result.units == []
|
||
|
||
def test_other_fields_preserved(self) -> None:
|
||
months = _months(36)
|
||
units = _seasonal_units(months)
|
||
series = _make_series(months, units)
|
||
result = normalize_demand(series)
|
||
# Дессзонивание трогает только units; прочее — без изменений.
|
||
assert result.months == series.months
|
||
assert result.area_m2 == series.area_m2
|
||
assert result.avg_price_per_m2 == series.avg_price_per_m2
|
||
assert result.source == series.source
|
||
assert result.segment == series.segment
|
||
assert result.confidence == series.confidence
|
||
|
||
def test_deseasonalized_units_nonnegative_int(self) -> None:
|
||
# Контракт SalesSeries.units = list[int] ≥ 0 сохраняется после нормализации.
|
||
months = _months(36)
|
||
units = _seasonal_units(months)
|
||
result = normalize_demand(_make_series(months, units))
|
||
assert all(isinstance(u, int) and u >= 0 for u in result.units)
|