458 lines
20 KiB
Python
458 lines
20 KiB
Python
"""Unit-тесты §11 макро-сценариев (#984, 954-A, ADVISORY).
|
||
|
||
Чистые тесты — БЕЗ живой БД (мок #952 compute_demand_supply_forecast +
|
||
get_monthly_macro):
|
||
• pure build_rate_envelopes: три именованных path; инвариант conservative ≥ base ≥
|
||
aggressive на КАЖДОМ горизонте; кламп ставки ≥ 0; расширение конверта с
|
||
горизонтом; base None → все пути None.
|
||
• pure helpers: _horizon_widen_factor (монотонно, h≤0/выкл → 1.0), _shift_rate
|
||
(знак + кламп ≥0 + None passthrough), _first_non_none, _drop_none_rates.
|
||
• compute_scenarios через MagicMock-сессию + @patch #952 и get_monthly_macro:
|
||
ТРИ сценария; у каждого свой rate_path; advisory ВСЕГДА True; conservative
|
||
rate_path выше aggressive; rate_path проброшен в #952; graceful (нет макро →
|
||
три сценария с None-rate_path, не crash).
|
||
|
||
Детерминированно, без LLM. Мокаем #952 + get_monthly_macro + db (нет живой БД).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
from app.services.forecasting.scenarios import (
|
||
_AGGRESSIVE_RATE_DELTA_PP,
|
||
_CONSERVATIVE_RATE_DELTA_PP,
|
||
_MIN_RATE_PCT,
|
||
ScenarioForecast,
|
||
_drop_none_rates,
|
||
_first_non_none,
|
||
_horizon_widen_factor,
|
||
_round_or_none,
|
||
_shift_rate,
|
||
build_rate_envelopes,
|
||
compute_scenarios,
|
||
)
|
||
|
||
# Пути патча #952 и макро-ридера (импортированы в модуль scenarios).
|
||
_FORECAST = "app.services.forecasting.scenarios.compute_demand_supply_forecast"
|
||
_MACRO = "app.services.forecasting.scenarios.get_monthly_macro"
|
||
_HOLD = "app.services.forecasting.scenarios.hold_last_rate"
|
||
|
||
|
||
# ── pure: _horizon_widen_factor ───────────────────────────────────────────────
|
||
|
||
|
||
class TestHorizonWidenFactor:
|
||
def test_twelve_months_is_one_year(self) -> None:
|
||
# h=12 → 1 + widen×1 (один год расширения).
|
||
assert _horizon_widen_factor(12, widen_per_year=0.5) == pytest.approx(1.5)
|
||
|
||
def test_monotonic_in_horizon(self) -> None:
|
||
f6 = _horizon_widen_factor(6)
|
||
f12 = _horizon_widen_factor(12)
|
||
f24 = _horizon_widen_factor(24)
|
||
assert f6 < f12 < f24
|
||
|
||
def test_nonpositive_horizon_no_widening(self) -> None:
|
||
assert _horizon_widen_factor(0) == 1.0
|
||
assert _horizon_widen_factor(-5) == 1.0
|
||
|
||
def test_widen_off_when_zero(self) -> None:
|
||
# widen_per_year ≤ 0 → плоский конверт (множитель 1.0 на любом горизонте).
|
||
assert _horizon_widen_factor(24, widen_per_year=0.0) == 1.0
|
||
|
||
def test_at_least_one(self) -> None:
|
||
assert _horizon_widen_factor(6) >= 1.0
|
||
|
||
|
||
# ── pure: _shift_rate ─────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestShiftRate:
|
||
def test_up_shift(self) -> None:
|
||
assert _shift_rate(16.0, +2.0) == pytest.approx(18.0)
|
||
|
||
def test_down_shift(self) -> None:
|
||
assert _shift_rate(16.0, -3.0) == pytest.approx(13.0)
|
||
|
||
def test_clamps_at_min(self) -> None:
|
||
# Сдвиг вниз ниже нуля → кламп к _MIN_RATE_PCT (нет отрицательной ставки).
|
||
assert _shift_rate(2.0, -5.0) == _MIN_RATE_PCT
|
||
|
||
def test_none_passthrough(self) -> None:
|
||
assert _shift_rate(None, +2.0) is None
|
||
|
||
def test_zero_shift_is_base(self) -> None:
|
||
assert _shift_rate(16.0, 0.0) == pytest.approx(16.0)
|
||
|
||
|
||
# ── pure: build_rate_envelopes (центральный §11 конверт) ──────────────────────
|
||
|
||
|
||
class TestBuildRateEnvelopes:
|
||
def test_three_named_paths(self) -> None:
|
||
env = build_rate_envelopes(16.0, [6, 12])
|
||
assert set(env.keys()) == {"conservative", "base", "aggressive"}
|
||
|
||
def test_each_path_covers_all_horizons(self) -> None:
|
||
env = build_rate_envelopes(16.0, [6, 12, 18, 24])
|
||
for name in ("conservative", "base", "aggressive"):
|
||
assert set(env[name].keys()) == {6, 12, 18, 24}
|
||
|
||
def test_base_is_flat_hold(self) -> None:
|
||
# base = base_rate на КАЖДОМ горизонте (ставка не меняется).
|
||
env = build_rate_envelopes(16.0, [6, 12, 24])
|
||
assert env["base"] == {6: 16.0, 12: 16.0, 24: 16.0}
|
||
|
||
def test_conservative_above_base_above_aggressive(self) -> None:
|
||
# ИНВАРИАНТ §11: conservative ≥ base ≥ aggressive на любом горизонте.
|
||
env = build_rate_envelopes(16.0, [6, 12, 18, 24])
|
||
for h in (6, 12, 18, 24):
|
||
cons = env["conservative"][h]
|
||
base = env["base"][h]
|
||
aggr = env["aggressive"][h]
|
||
assert cons is not None and base is not None and aggr is not None
|
||
assert cons > base > aggr
|
||
|
||
def test_conservative_uses_up_delta_at_one_year(self) -> None:
|
||
# На h=12 widen=1.5 → conservative = base + delta×1.5.
|
||
env = build_rate_envelopes(16.0, [12])
|
||
expected = 16.0 + _CONSERVATIVE_RATE_DELTA_PP * 1.5
|
||
assert env["conservative"][12] == pytest.approx(expected)
|
||
|
||
def test_aggressive_uses_down_delta_at_one_year(self) -> None:
|
||
env = build_rate_envelopes(16.0, [12])
|
||
expected = 16.0 - _AGGRESSIVE_RATE_DELTA_PP * 1.5
|
||
assert env["aggressive"][12] == pytest.approx(expected)
|
||
|
||
def test_envelope_widens_with_horizon(self) -> None:
|
||
# Разброс conservative−aggressive РАСТЁТ с горизонтом (дальше неопределённее).
|
||
env = build_rate_envelopes(16.0, [6, 24])
|
||
spread_6 = env["conservative"][6] - env["aggressive"][6] # type: ignore[operator]
|
||
spread_24 = env["conservative"][24] - env["aggressive"][24] # type: ignore[operator]
|
||
assert spread_24 > spread_6
|
||
|
||
def test_aggressive_clamped_non_negative(self) -> None:
|
||
# Низкая база + большой down-сдвиг на дальнем горизонте → кламп ≥0.
|
||
env = build_rate_envelopes(1.0, [24])
|
||
assert env["aggressive"][24] == _MIN_RATE_PCT
|
||
# base/conservative при этом остаются положительными.
|
||
assert env["base"][24] == 1.0
|
||
assert env["conservative"][24] is not None and env["conservative"][24] > 1.0
|
||
|
||
def test_none_base_all_paths_none(self) -> None:
|
||
# Нет базовой ставки → все пути None на всех горизонтах (graceful).
|
||
env = build_rate_envelopes(None, [6, 12, 24])
|
||
for name in ("conservative", "base", "aggressive"):
|
||
assert all(v is None for v in env[name].values())
|
||
|
||
def test_custom_deltas(self) -> None:
|
||
env = build_rate_envelopes(10.0, [12], conservative_delta_pp=1.0, aggressive_delta_pp=1.0)
|
||
# widen(12)=1.5 → ±1.5 вокруг базы.
|
||
assert env["conservative"][12] == pytest.approx(11.5)
|
||
assert env["aggressive"][12] == pytest.approx(8.5)
|
||
|
||
def test_pure_no_mutation_of_input(self) -> None:
|
||
horizons = [6, 12]
|
||
build_rate_envelopes(16.0, horizons)
|
||
assert horizons == [6, 12]
|
||
|
||
|
||
# ── pure: _first_non_none ─────────────────────────────────────────────────────
|
||
|
||
|
||
class TestFirstNonNone:
|
||
def test_returns_first_present(self) -> None:
|
||
assert _first_non_none({6: 16.0, 12: 16.0}) == 16.0
|
||
|
||
def test_skips_none(self) -> None:
|
||
assert _first_non_none({6: None, 12: 17.0}) == 17.0
|
||
|
||
def test_all_none_returns_none(self) -> None:
|
||
assert _first_non_none({6: None, 12: None}) is None
|
||
|
||
def test_empty_returns_none(self) -> None:
|
||
assert _first_non_none({}) is None
|
||
|
||
|
||
# ── pure: _drop_none_rates ────────────────────────────────────────────────────
|
||
|
||
|
||
class TestDropNoneRates:
|
||
def test_keeps_present_rates(self) -> None:
|
||
assert _drop_none_rates({6: 18.0, 12: 16.0}) == {6: 18.0, 12: 16.0}
|
||
|
||
def test_drops_none_values(self) -> None:
|
||
assert _drop_none_rates({6: None, 12: 16.0}) == {12: 16.0}
|
||
|
||
def test_all_none_returns_none(self) -> None:
|
||
# Path целиком из None → None (#952 берёт свой дефолт hold_last_rate).
|
||
assert _drop_none_rates({6: None, 12: None}) is None
|
||
|
||
def test_empty_returns_none(self) -> None:
|
||
assert _drop_none_rates({}) is None
|
||
|
||
|
||
# ── pure: _round_or_none ──────────────────────────────────────────────────────
|
||
|
||
|
||
class TestRoundOrNone:
|
||
def test_rounds(self) -> None:
|
||
assert _round_or_none(0.123456, 3) == 0.123
|
||
|
||
def test_none_passthrough(self) -> None:
|
||
assert _round_or_none(None, 3) is None
|
||
|
||
|
||
# ── dataclass as_dict ─────────────────────────────────────────────────────────
|
||
|
||
|
||
def _forecast_stub(*, horizon: int, deficit_index: float | None = 0.3) -> MagicMock:
|
||
"""Одиночный DemandSupplyForecast-стаб (нужны horizon_months + as_dict)."""
|
||
f = MagicMock()
|
||
f.horizon_months = horizon
|
||
f.deficit_index = deficit_index
|
||
f.as_dict.return_value = {"horizon_months": horizon, "deficit_index": deficit_index}
|
||
return f
|
||
|
||
|
||
class TestScenarioForecastAsDict:
|
||
def test_as_dict_shape(self) -> None:
|
||
sf = ScenarioForecast(
|
||
scenario="base",
|
||
rate_path={12: 16.0, 6: 16.0},
|
||
forecasts=[_forecast_stub(horizon=6), _forecast_stub(horizon=12)],
|
||
advisory=True,
|
||
)
|
||
d = sf.as_dict()
|
||
assert d["scenario"] == "base"
|
||
assert d["advisory"] is True
|
||
assert len(d["forecasts"]) == 2
|
||
|
||
def test_rate_path_sorted_and_rounded(self) -> None:
|
||
sf = ScenarioForecast(
|
||
scenario="conservative",
|
||
rate_path={12: 18.123456, 6: 17.987654},
|
||
forecasts=[],
|
||
advisory=True,
|
||
)
|
||
d = sf.as_dict()
|
||
# Отсортирован по горизонту, округлён до 2 знаков.
|
||
assert list(d["rate_path"].keys()) == [6, 12]
|
||
assert d["rate_path"][6] == 17.99
|
||
assert d["rate_path"][12] == 18.12
|
||
|
||
def test_rate_path_none_preserved(self) -> None:
|
||
sf = ScenarioForecast(
|
||
scenario="base", rate_path={6: None, 12: None}, forecasts=[], advisory=True
|
||
)
|
||
d = sf.as_dict()
|
||
assert d["rate_path"] == {6: None, 12: None}
|
||
|
||
|
||
# ── orchestrator helpers (стаб #952 + макро) ──────────────────────────────────
|
||
|
||
_BASE_KEY_RATE = 16.0
|
||
|
||
|
||
def _macro_stub_with_rate(rate: float | None) -> list[MagicMock]:
|
||
"""Макро-ряд из одной точки с заданной key_rate (hold_last_rate возьмёт её)."""
|
||
m = MagicMock()
|
||
m.key_rate = rate
|
||
return [m]
|
||
|
||
|
||
def _forecast_side_effect() -> Any:
|
||
"""side_effect #952: вернуть по forecast-стабу на каждый горизонт; запомнить rate_path."""
|
||
|
||
def _side(db: Any, *, spec: Any, horizons: Any, rate_path: Any = None, **_: Any) -> Any:
|
||
return [_forecast_stub(horizon=h) for h in horizons]
|
||
|
||
return _side
|
||
|
||
|
||
def _run(**over: object) -> list[ScenarioForecast]:
|
||
kwargs: dict[str, object] = {
|
||
"spec": MagicMock(),
|
||
"district": "Академический",
|
||
"cad_num": "66:41:0303161:123",
|
||
}
|
||
kwargs.update(over)
|
||
return compute_scenarios(MagicMock(), **kwargs) # type: ignore[arg-type]
|
||
|
||
|
||
# ── orchestrator: три сценария ────────────────────────────────────────────────
|
||
|
||
|
||
class TestComputeScenariosShape:
|
||
def test_returns_three_scenarios(self) -> None:
|
||
with (
|
||
patch(_MACRO, return_value=_macro_stub_with_rate(_BASE_KEY_RATE)),
|
||
patch(_FORECAST, side_effect=_forecast_side_effect()),
|
||
):
|
||
res = _run(horizons=[6, 12])
|
||
assert len(res) == 3
|
||
assert [s.scenario for s in res] == ["conservative", "base", "aggressive"]
|
||
|
||
def test_each_scenario_advisory_true(self) -> None:
|
||
with (
|
||
patch(_MACRO, return_value=_macro_stub_with_rate(_BASE_KEY_RATE)),
|
||
patch(_FORECAST, side_effect=_forecast_side_effect()),
|
||
):
|
||
res = _run(horizons=[6, 12])
|
||
assert all(s.advisory is True for s in res)
|
||
|
||
def test_each_scenario_has_forecast_per_horizon(self) -> None:
|
||
with (
|
||
patch(_MACRO, return_value=_macro_stub_with_rate(_BASE_KEY_RATE)),
|
||
patch(_FORECAST, side_effect=_forecast_side_effect()),
|
||
):
|
||
res = _run(horizons=[6, 12, 18, 24])
|
||
for s in res:
|
||
assert len(s.forecasts) == 4
|
||
assert [f.horizon_months for f in s.forecasts] == [6, 12, 18, 24]
|
||
|
||
def test_default_horizons_used(self) -> None:
|
||
with (
|
||
patch(_MACRO, return_value=_macro_stub_with_rate(_BASE_KEY_RATE)),
|
||
patch(_FORECAST, side_effect=_forecast_side_effect()),
|
||
):
|
||
res = _run() # без horizons → _DEFAULT_HORIZONS (6,12,18,24)
|
||
assert all(len(s.forecasts) == 4 for s in res)
|
||
|
||
|
||
# ── orchestrator: rate_path per-scenario ──────────────────────────────────────
|
||
|
||
|
||
class TestComputeScenariosRatePath:
|
||
def test_each_scenario_carries_its_rate_path(self) -> None:
|
||
with (
|
||
patch(_MACRO, return_value=_macro_stub_with_rate(_BASE_KEY_RATE)),
|
||
patch(_FORECAST, side_effect=_forecast_side_effect()),
|
||
):
|
||
res = _run(horizons=[12])
|
||
by_name = {s.scenario: s for s in res}
|
||
# base = голая база; conservative выше; aggressive ниже.
|
||
assert by_name["base"].rate_path[12] == pytest.approx(_BASE_KEY_RATE)
|
||
assert by_name["conservative"].rate_path[12] > _BASE_KEY_RATE # type: ignore[operator]
|
||
assert by_name["aggressive"].rate_path[12] < _BASE_KEY_RATE # type: ignore[operator]
|
||
|
||
def test_conservative_rate_path_higher_than_aggressive(self) -> None:
|
||
with (
|
||
patch(_MACRO, return_value=_macro_stub_with_rate(_BASE_KEY_RATE)),
|
||
patch(_FORECAST, side_effect=_forecast_side_effect()),
|
||
):
|
||
res = _run(horizons=[6, 12, 18, 24])
|
||
by_name = {s.scenario: s for s in res}
|
||
for h in (6, 12, 18, 24):
|
||
assert by_name["conservative"].rate_path[h] > by_name["aggressive"].rate_path[h] # type: ignore[operator]
|
||
|
||
def test_rate_path_passed_to_forecast_engine(self) -> None:
|
||
# rate_path, переданный в #952, совпадает с конвертом сценария (None убраны).
|
||
captured: dict[str, Any] = {}
|
||
|
||
def _side(db: Any, *, spec: Any, horizons: Any, rate_path: Any = None, **_: Any) -> Any:
|
||
captured[rate_path[12]] = rate_path # type: ignore[index]
|
||
return [_forecast_stub(horizon=h) for h in horizons]
|
||
|
||
with (
|
||
patch(_MACRO, return_value=_macro_stub_with_rate(_BASE_KEY_RATE)),
|
||
patch(_FORECAST, side_effect=_side),
|
||
):
|
||
res = _run(horizons=[12])
|
||
by_name = {s.scenario: s for s in res}
|
||
# Каждый сценарный путь (без None) дошёл до #952.
|
||
assert by_name["base"].rate_path[12] in captured
|
||
assert by_name["conservative"].rate_path[12] in captured
|
||
assert by_name["aggressive"].rate_path[12] in captured
|
||
|
||
|
||
# ── orchestrator: graceful (нет макро / нет базовой ставки) ────────────────────
|
||
|
||
|
||
class TestComputeScenariosGraceful:
|
||
def test_no_macro_still_three_scenarios(self) -> None:
|
||
# Пустой макро-ряд → база None → конверты None, но ТРИ сценария (не crash).
|
||
with (
|
||
patch(_MACRO, return_value=[]),
|
||
patch(_FORECAST, side_effect=_forecast_side_effect()) as fc,
|
||
):
|
||
res = _run(horizons=[6, 12])
|
||
assert len(res) == 3
|
||
assert all(s.advisory is True for s in res)
|
||
# Все rate_path None при отсутствии базы.
|
||
for s in res:
|
||
assert all(v is None for v in s.rate_path.values())
|
||
# #952 всё равно вызван по разу на сценарий (с rate_path=None → его дефолт).
|
||
assert fc.call_count == 3
|
||
|
||
def test_no_key_rate_point_still_three_scenarios(self) -> None:
|
||
# Макро есть, но key_rate None во всех точках → база None.
|
||
with (
|
||
patch(_MACRO, return_value=_macro_stub_with_rate(None)),
|
||
patch(_FORECAST, side_effect=_forecast_side_effect()),
|
||
):
|
||
res = _run(horizons=[12])
|
||
assert len(res) == 3
|
||
for s in res:
|
||
assert s.rate_path[12] is None
|
||
|
||
def test_no_base_passes_none_rate_path_to_engine(self) -> None:
|
||
# Нет базы → #952 получает rate_path=None (берёт свой hold_last_rate-дефолт).
|
||
captured: list[Any] = []
|
||
|
||
def _side(db: Any, *, spec: Any, horizons: Any, rate_path: Any = None, **_: Any) -> Any:
|
||
captured.append(rate_path)
|
||
return [_forecast_stub(horizon=h) for h in horizons]
|
||
|
||
with patch(_MACRO, return_value=[]), patch(_FORECAST, side_effect=_side):
|
||
_run(horizons=[6, 12])
|
||
assert captured == [None, None, None]
|
||
|
||
def test_forecast_engine_exception_propagates_not_swallowed(self) -> None:
|
||
# #952 сам graceful; если он всё же кинул — НЕ глотаем молча (честный провал).
|
||
with (
|
||
patch(_MACRO, return_value=_macro_stub_with_rate(_BASE_KEY_RATE)),
|
||
patch(_FORECAST, side_effect=RuntimeError("engine boom")),
|
||
):
|
||
with pytest.raises(RuntimeError, match="engine boom"):
|
||
_run(horizons=[12])
|
||
|
||
|
||
# ── orchestrator: база читается через hold_last_rate ──────────────────────────
|
||
|
||
|
||
class TestComputeScenariosBaseRate:
|
||
def test_base_rate_from_last_key_rate(self) -> None:
|
||
# hold_last_rate берёт ПОСЛЕДНЮЮ непустую key_rate ряда → она и есть база.
|
||
m_old = MagicMock()
|
||
m_old.key_rate = 14.0
|
||
m_new = MagicMock()
|
||
m_new.key_rate = 18.0 # свежее (ряд ASC) → база
|
||
with (
|
||
patch(_MACRO, return_value=[m_old, m_new]),
|
||
patch(_FORECAST, side_effect=_forecast_side_effect()),
|
||
):
|
||
res = _run(horizons=[12])
|
||
by_name = {s.scenario: s for s in res}
|
||
assert by_name["base"].rate_path[12] == pytest.approx(18.0)
|
||
|
||
def test_hold_last_rate_invoked(self) -> None:
|
||
# Поток явно опирается на #952 hold_last_rate (база = его выход).
|
||
macro = _macro_stub_with_rate(_BASE_KEY_RATE)
|
||
with (
|
||
patch(_MACRO, return_value=macro),
|
||
patch(_HOLD, return_value={12: 15.5}) as hold,
|
||
patch(_FORECAST, side_effect=_forecast_side_effect()),
|
||
):
|
||
res = _run(horizons=[12])
|
||
hold.assert_called_once()
|
||
by_name = {s.scenario: s for s in res}
|
||
assert by_name["base"].rate_path[12] == pytest.approx(15.5)
|