bot-backend
494155654e
Some checks failed
CI / changes (push) Successful in 8s
CI / changes (pull_request) Successful in 6s
CI / backend-tests (push) Has been skipped
CI / frontend-tests (push) Has been skipped
CI / openapi-codegen-check (push) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Successful in 1m59s
CI / backend-tests (pull_request) Failing after 8m52s
_count_full_years treated units=0 as a valid observation, so a series where fill_month_grid zero-filled every month still accumulated 3 full years and passed the _MIN_FULL_YEARS guard. Zero-filled months carry no seasonal signal, so they must be skipped in the year counter — the same way None values already were. Fix: skip v==0 alongside v is None in _count_full_years. Add four tests: zero-filled 36-month series → n_full_years=0/applied=False; partial-coverage years (only 6 non-zero months/year) → not counted as full; real non-zero series still passes guard; normalize_demand on zero-filled SalesSeries returns series unchanged.
299 lines
15 KiB
Python
299 lines
15 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())
|
||
|
||
# ── fix #1638: zero-filled series must fail the min-years guard ───────────
|
||
|
||
def test_zero_filled_36m_fails_min_years_guard(self) -> None:
|
||
"""Ряд 36 мес, все units=0 (zero-fill от fill_month_grid) → applied=False.
|
||
|
||
До fix #1638 _count_full_years считал нулевые значения за наблюдения
|
||
и возвращал n_full_years=3 → guard пропускал → overall_mean==0 спасал
|
||
от деления, но прохождение guard'а было семантически неверным.
|
||
После fix: 0 не считается наблюдением → n_full_years=0 → applied=False.
|
||
"""
|
||
months = _months(36)
|
||
adj = seasonal_factors(months, [0] * 36)
|
||
assert (
|
||
adj.n_full_years == 0
|
||
), f"expected 0 full years on all-zero series, got {adj.n_full_years}"
|
||
assert adj.applied is False
|
||
assert all(f == 1.0 for f in adj.factors.values())
|
||
|
||
def test_mixed_zero_and_real_partial_years_fails_guard(self) -> None:
|
||
"""2 года месяцев, но только часть месяцев ненулевые — не 2 полных года.
|
||
|
||
Сценарий: новый ЖК, продажи только в нескольких месяцах каждого года
|
||
(остальные zero-fill). _count_full_years должен не считать это «полным годом».
|
||
"""
|
||
months = _months(24) # 2 календарных года
|
||
# Только январь-июнь каждого года ненулевые → нет полного покрытия 12 мес.
|
||
units = [10 if d.month <= 6 else 0 for d in months]
|
||
adj = seasonal_factors(months, units)
|
||
assert (
|
||
adj.n_full_years == 0
|
||
), f"partial-coverage years should not count as full, got {adj.n_full_years}"
|
||
assert adj.applied is False
|
||
|
||
def test_real_nonzero_series_passes_guard_and_applies(self) -> None:
|
||
"""Ряд с реальными ненулевыми данными за 2+ лет проходит guard и применяется.
|
||
|
||
Страховка: fix #1638 не должен ломать штатный путь с настоящими данными.
|
||
"""
|
||
months = _months(36)
|
||
units = _seasonal_units(months) # baseline×паттерн, все > 0
|
||
adj = seasonal_factors(months, units)
|
||
assert adj.n_full_years == 3
|
||
assert adj.applied is True
|
||
|
||
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)
|
||
|
||
def test_zero_filled_series_returned_unchanged(self) -> None:
|
||
"""fix #1638: zero-filled SalesSeries (все units=0) возвращается без изменений.
|
||
|
||
До fix: guard считал 3 «полных года» на нулях → adjustment.applied=True
|
||
(пройдя через overall_mean==0 check) или мог пройти guard и вернуть
|
||
идентичный ряд через нейтральные факторы. После fix: n_full_years=0 →
|
||
applied=False → функция возвращает тот же объект (short-circuit).
|
||
"""
|
||
months = _months(36)
|
||
series = _make_series(months, [0] * 36)
|
||
result = normalize_demand(series)
|
||
assert result is series, "zero-filled series must be returned as-is (no adjustment)"
|
||
assert result.units == [0] * 36
|