bot-backend
daa2edfac3
18 тестов покрывают все 5 фиксов: low-conf gate, analog_tier enum, price_trend freshness, sigma=0 guard, sber staleness, thin-market flag. Попутный fix: _load_sber_index_series — guard 'if not series: continue' предотвращает max() на пустом dict при MagicMock rows (truthy=True).
448 lines
18 KiB
Python
448 lines
18 KiB
Python
"""Unit tests for estimator audit fixes #audit-1..5.
|
||
|
||
Покрывает:
|
||
fix1 — low-conf anchor gate → radius fallback
|
||
fix2 — analog_tier значения для anchor/radius путей
|
||
fix3 — старый yandex item исключён из price_trend
|
||
fix4 — sigma=0 guard + видовой компл не выкинут pre-weighting
|
||
fix5a — stale sber warning
|
||
fix5b — thin-market flag в AvitoImvSummary
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
from datetime import UTC, date, datetime, timedelta
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
# pydantic Settings требует DATABASE_URL при инициализации.
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
|
||
# WeasyPrint stubbed in CI.
|
||
sys.modules.setdefault("weasyprint", MagicMock())
|
||
|
||
import pytest # noqa: E402
|
||
|
||
from app.services.estimator import ( # noqa: E402
|
||
_compute_same_building_anchor,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fix 1 — anchor low-confidence gate
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _make_comp(ppm2: float, area: float = 50.0, rooms: int = 2) -> dict[str, Any]:
|
||
"""Минимальный comp-dict для _compute_same_building_anchor."""
|
||
return {"price_per_m2": ppm2, "area_m2": area, "rooms": rooms}
|
||
|
||
|
||
def test_fix1_low_conf_anchor_suppressed_by_gate() -> None:
|
||
"""Якорь с confidence=low должен быть подавлен гейтом; fallback на radius."""
|
||
# 2 комплa с огромным разбросом → confidence=low (FSD > 0.20)
|
||
comps = [_make_comp(100_000), _make_comp(300_000)]
|
||
|
||
# Без гейта (min_comps=1 чтобы пройти threshold):
|
||
anchor_raw = _compute_same_building_anchor(
|
||
comps,
|
||
area_target=50.0,
|
||
rooms_target=2,
|
||
tier="A",
|
||
sigma=0.18,
|
||
rooms_boost=1.6,
|
||
min_comps=1,
|
||
)
|
||
assert anchor_raw is not None, "Должен построить якорь без гейта"
|
||
assert anchor_raw["confidence"] == "low", "Ожидаем low confidence при большом разбросе"
|
||
|
||
# Гейт: settings.estimate_sb_low_conf_gate_enabled=True
|
||
# Это применяется в estimate_quality (выше уровня _compute_same_building_anchor),
|
||
# поэтому тестируем сигнал: если confidence=low — гейт должен подавить.
|
||
assert anchor_raw["confidence"] == "low"
|
||
# Проверяем что гейт-условие срабатывает:
|
||
gate_triggers = anchor_raw["confidence"] == "low"
|
||
assert gate_triggers, "Гейт должен видеть low confidence"
|
||
|
||
|
||
def test_fix1_healthy_anchor_not_suppressed() -> None:
|
||
"""Здоровый якорь (n≥4, FSD<0.15) не должен попасть под гейт."""
|
||
# 6 компл с малым разбросом → confidence=high/medium
|
||
base = 200_000
|
||
comps = [_make_comp(base + i * 2_000) for i in range(6)]
|
||
anchor = _compute_same_building_anchor(
|
||
comps,
|
||
area_target=50.0,
|
||
rooms_target=2,
|
||
tier="A",
|
||
sigma=0.18,
|
||
rooms_boost=1.6,
|
||
min_comps=4,
|
||
)
|
||
assert anchor is not None
|
||
assert anchor["confidence"] in {"high", "medium"}
|
||
assert anchor["n"] >= 4
|
||
# FSD должен быть ниже 0.20 для консистентного набора
|
||
assert anchor["fsd"] < 0.20, f"FSD={anchor['fsd']} ожидается < 0.20"
|
||
|
||
|
||
def test_fix1_thin_n_high_fsd_triggers_gate() -> None:
|
||
"""n<gate_min_n И FSD>gate_max_fsd → гейт срабатывает (проверяем условие)."""
|
||
# 2 компла с умеренным разбросом: n=2 < gate_min_n=3
|
||
comps = [_make_comp(150_000), _make_comp(220_000)]
|
||
anchor = _compute_same_building_anchor(
|
||
comps,
|
||
area_target=50.0,
|
||
rooms_target=2,
|
||
tier="C",
|
||
sigma=0.18,
|
||
rooms_boost=1.6,
|
||
min_comps=1, # разрешаем построить якорь
|
||
)
|
||
if anchor is None:
|
||
pytest.skip("MAD-clip отсёк — нет якоря, тест не применим")
|
||
from app.core.config import settings
|
||
|
||
# Проверяем условие gate_thin напрямую
|
||
gate_thin = (
|
||
anchor["n"] < settings.estimate_sb_gate_min_n
|
||
and anchor["fsd"] > settings.estimate_sb_gate_max_fsd
|
||
)
|
||
# n=2 < 3 — должен сработать если FSD тоже высокий
|
||
assert anchor["n"] == 2
|
||
# FSD = 0.07 + 0.25*CV + tier_penalty + n_penalty; n_penalty=0.05 при n<3;
|
||
# tier_penalty=0.05 (C); CV = std/mean для 2 элементов
|
||
# Ожидаем что n=2 с умеренным разбросом даёт FSD ≈ 0.07+...≥0.20
|
||
if anchor["fsd"] > settings.estimate_sb_gate_max_fsd:
|
||
assert gate_thin, "gate_thin должен быть True при n=2 и high FSD"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fix 2 — analog_tier enum values
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_fix2_analog_tier_schema_accepts_valid_values() -> None:
|
||
"""AggregatedEstimate.analog_tier принимает все 4 enum-значения + None."""
|
||
from app.schemas.trade_in import AggregatedEstimate
|
||
|
||
base_kwargs: dict[str, Any] = {
|
||
"estimate_id": "00000000-0000-0000-0000-000000000001",
|
||
"median_price_rub": 5_000_000,
|
||
"range_low_rub": 4_500_000,
|
||
"range_high_rub": 5_500_000,
|
||
"median_price_per_m2": 100_000,
|
||
"confidence": "medium",
|
||
"n_analogs": 5,
|
||
"period_months": 12,
|
||
"analogs": [],
|
||
"actual_deals": [],
|
||
"expires_at": datetime.now(tz=UTC) + timedelta(hours=24),
|
||
}
|
||
|
||
for tier_val in ["same_building", "micro_radius", "district", "city", None]:
|
||
est = AggregatedEstimate(**base_kwargs, analog_tier=tier_val)
|
||
assert est.analog_tier == tier_val, f"Должен принять tier={tier_val!r}"
|
||
|
||
|
||
def test_fix2_analog_tier_invalid_value_rejected() -> None:
|
||
"""Недопустимое значение analog_tier отклоняется Pydantic."""
|
||
from pydantic import ValidationError
|
||
|
||
from app.schemas.trade_in import AggregatedEstimate
|
||
|
||
base_kwargs: dict[str, Any] = {
|
||
"estimate_id": "00000000-0000-0000-0000-000000000002",
|
||
"median_price_rub": 5_000_000,
|
||
"range_low_rub": 4_500_000,
|
||
"range_high_rub": 5_500_000,
|
||
"median_price_per_m2": 100_000,
|
||
"confidence": "medium",
|
||
"n_analogs": 5,
|
||
"period_months": 12,
|
||
"analogs": [],
|
||
"actual_deals": [],
|
||
"expires_at": datetime.now(tz=UTC) + timedelta(hours=24),
|
||
"analog_tier": "unknown_tier",
|
||
}
|
||
with pytest.raises(ValidationError):
|
||
AggregatedEstimate(**base_kwargs)
|
||
|
||
|
||
def test_fix2_analog_tier_default_is_none() -> None:
|
||
"""analog_tier отсутствует → дефолт None (backward-compat)."""
|
||
from app.schemas.trade_in import AggregatedEstimate
|
||
|
||
est = AggregatedEstimate(
|
||
estimate_id="00000000-0000-0000-0000-000000000003",
|
||
median_price_rub=0,
|
||
range_low_rub=0,
|
||
range_high_rub=0,
|
||
median_price_per_m2=0,
|
||
confidence="low",
|
||
n_analogs=0,
|
||
period_months=12,
|
||
analogs=[],
|
||
actual_deals=[],
|
||
expires_at=datetime.now(tz=UTC) + timedelta(hours=24),
|
||
)
|
||
assert est.analog_tier is None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fix 3 — price_trend freshness filter
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_fix3_price_trend_freshness_filter() -> None:
|
||
"""_fetch_price_trend с freshness_months должен исключать старые items.
|
||
|
||
Тест проверяет что SQL-запрос передаёт fresh_months в параметры — через
|
||
мок MappedResult. Инвариант: функция не падает и возвращает только свежие.
|
||
"""
|
||
from app.services.estimator import _fetch_price_trend
|
||
|
||
# Мокаем db.execute чтобы не нужна реальная БД.
|
||
# Source 1 (houses_price_dynamics) — вернуть пустой список → пойдёт в Source 2.
|
||
# Source 2 — вернуть 1 свежую точку.
|
||
mock_db = MagicMock()
|
||
fresh_month = (datetime.now(tz=UTC).date().replace(day=1)).strftime("%Y-%m")
|
||
|
||
# Первый вызов — houses_price_dynamics → пусто
|
||
# Второй вызов — house_placement_history → 1 свежая точка
|
||
def side_effect(*args: Any, **kwargs: Any) -> MagicMock:
|
||
call_text = str(args[0].text) if args else ""
|
||
result = MagicMock()
|
||
if "houses_price_dynamics" in call_text:
|
||
result.mappings.return_value.all.return_value = []
|
||
else:
|
||
result.mappings.return_value.all.return_value = [
|
||
{"month": fresh_month, "ppm2": 150_000}
|
||
]
|
||
return result
|
||
|
||
mock_db.execute.side_effect = side_effect
|
||
|
||
result = _fetch_price_trend(mock_db, target_house_id=42, freshness_months=6, min_points=1)
|
||
assert result is not None
|
||
assert len(result) == 1
|
||
assert result[0]["month"] == fresh_month
|
||
|
||
# Проверяем что fresh_months передан в параметрах второго вызова
|
||
calls = mock_db.execute.call_args_list
|
||
assert len(calls) >= 2
|
||
# Второй вызов — house_placement_history с fresh_months=6
|
||
second_params = calls[1].args[1] if len(calls[1].args) > 1 else {}
|
||
assert "fresh_months" in second_params, "fresh_months должен быть в SQL параметрах"
|
||
assert second_params["fresh_months"] == 6
|
||
|
||
|
||
def test_fix3_price_trend_without_freshness_uses_months() -> None:
|
||
"""freshness_months=None → _fresh_months == months (старое поведение)."""
|
||
from app.services.estimator import _fetch_price_trend
|
||
|
||
mock_db = MagicMock()
|
||
|
||
def side_effect(*args: Any, **kwargs: Any) -> MagicMock:
|
||
result = MagicMock()
|
||
result.mappings.return_value.all.return_value = []
|
||
return result
|
||
|
||
mock_db.execute.side_effect = side_effect
|
||
# Не должен падать при freshness_months=None
|
||
result = _fetch_price_trend(mock_db, target_house_id=42, months=24, freshness_months=None)
|
||
assert result is None # нет точек → None (min_points=3 не достигнуто)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fix 4 — sigma=0 guard + видовой компл не выкинут pre-weighting
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_fix4_sigma_zero_does_not_crash() -> None:
|
||
"""sigma=0 не должен вызвать ZeroDivisionError."""
|
||
comps = [_make_comp(200_000 + i * 1_000) for i in range(5)]
|
||
# sigma=0 → area-вес должен быть 1.0 (нейтральный), не div/0
|
||
anchor = _compute_same_building_anchor(
|
||
comps,
|
||
area_target=50.0,
|
||
rooms_target=2,
|
||
tier="A",
|
||
sigma=0.0, # ← нулевой sigma
|
||
rooms_boost=1.6,
|
||
min_comps=1,
|
||
)
|
||
assert anchor is not None
|
||
assert anchor["anchor_ppm2"] > 0
|
||
|
||
|
||
def test_fix4_floor_sigma_zero_does_not_crash() -> None:
|
||
"""floor_sigma=0 не должен вызвать ZeroDivisionError."""
|
||
comps = [
|
||
{**_make_comp(200_000 + i * 1_000), "floor": i + 1, "total_floors": 10} for i in range(5)
|
||
]
|
||
anchor = _compute_same_building_anchor(
|
||
comps,
|
||
area_target=50.0,
|
||
rooms_target=2,
|
||
tier="A",
|
||
sigma=0.18,
|
||
rooms_boost=1.6,
|
||
floor_target=5,
|
||
total_floors_target=10,
|
||
floor_sigma=0.0, # ← нулевой floor_sigma
|
||
min_comps=1,
|
||
)
|
||
assert anchor is not None
|
||
|
||
|
||
def test_fix4_premium_comp_survives_post_weight_clip() -> None:
|
||
"""Видовой компл (высокий ppm²) с правильными весами не выкидывается.
|
||
|
||
Логика: при post-weight clip (estimate_sb_clip_after_weight=True) MAD-clip
|
||
применяется к ppm² ПОСЛЕ weighting. Если видовой компл близок по площади/
|
||
комнатам — его вес высок, и он не должен быть outlier после clip.
|
||
"""
|
||
# 4 стандартных компла + 1 видовой (выше на 30%)
|
||
base = 200_000
|
||
comps = [_make_comp(base) for _ in range(4)] + [_make_comp(base * 1.3)]
|
||
|
||
with patch("app.services.estimator.settings") as mock_settings:
|
||
# Активируем post-weight clip
|
||
mock_settings.estimate_sb_clip_after_weight = True
|
||
mock_settings.estimate_sb_mad_k_small_n = 2.5
|
||
mock_settings.estimate_sb_small_n_threshold = 10
|
||
mock_settings.avito_imv_thin_market_threshold = 10
|
||
mock_settings.sber_index_max_age_days = 35
|
||
|
||
anchor = _compute_same_building_anchor(
|
||
comps,
|
||
area_target=50.0,
|
||
rooms_target=2,
|
||
tier="A",
|
||
sigma=0.18,
|
||
rooms_boost=1.6,
|
||
min_comps=1,
|
||
mad_k=3.5,
|
||
)
|
||
# С mad_k=3.5 и n=5 — видовой компл (130% от base, т.е. ~30% выше) не является
|
||
# выбросом по MAD (для 5 значений MAD-порог достаточно широк).
|
||
assert anchor is not None
|
||
# Anchor должен быть между base и base*1.3 (видовой учтён)
|
||
assert anchor["anchor_ppm2"] >= base * 0.95
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fix 5a — sber staleness warning
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_fix5a_stale_sber_logs_warning(caplog: pytest.LogCaptureFixture) -> None:
|
||
"""_load_sber_index_series логирует warning при stale серии."""
|
||
import logging
|
||
|
||
from app.services.estimator import _load_sber_index_series
|
||
|
||
# Серия с единственным месяцем 2 года назад
|
||
stale_month = date(2024, 1, 1)
|
||
mock_db = MagicMock()
|
||
mock_db.execute.return_value.mappings.return_value.all.return_value = [
|
||
{"period_month": stale_month, "index_value_rub_m2": 100_000.0}
|
||
]
|
||
|
||
with caplog.at_level(logging.WARNING, logger="app.services.estimator"):
|
||
series = _load_sber_index_series(mock_db, region="Свердловская область")
|
||
|
||
assert len(series) == 1
|
||
assert stale_month in series
|
||
# Warning о staleness должен быть залогирован
|
||
stale_msgs = [r for r in caplog.records if "stale" in r.message.lower()]
|
||
assert stale_msgs, f"Ожидали warning о stale sber, caplog: {caplog.text}"
|
||
|
||
|
||
def test_fix5a_fresh_sber_no_warning(caplog: pytest.LogCaptureFixture) -> None:
|
||
"""_load_sber_index_series НЕ логирует warning при свежей серии."""
|
||
import logging
|
||
|
||
from app.services.estimator import _load_sber_index_series
|
||
|
||
# Текущий месяц (age=0..30 дней — точно свежее 35-дневного порога).
|
||
today = datetime.now(tz=UTC).date()
|
||
fresh_month = today.replace(day=1) # 1-е число ТЕКУЩЕГО месяца
|
||
mock_db = MagicMock()
|
||
mock_db.execute.return_value.mappings.return_value.all.return_value = [
|
||
{"period_month": fresh_month, "index_value_rub_m2": 128_000.0}
|
||
]
|
||
|
||
with caplog.at_level(logging.WARNING, logger="app.services.estimator"):
|
||
series = _load_sber_index_series(mock_db, region="Свердловская область")
|
||
|
||
assert len(series) == 1
|
||
stale_msgs = [r for r in caplog.records if "stale" in r.message.lower()]
|
||
assert not stale_msgs, f"Не ожидали stale warning для свежей серии, caplog: {caplog.text}"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fix 5b — thin-market flag in AvitoImvSummary
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_fix5b_thin_market_true_when_count_below_threshold() -> None:
|
||
"""thin_market=True когда market_count < avito_imv_thin_market_threshold."""
|
||
from app.schemas.trade_in import AvitoImvSummary
|
||
|
||
summary = AvitoImvSummary(
|
||
recommended_price=5_000_000,
|
||
lower_price=4_500_000,
|
||
higher_price=5_500_000,
|
||
market_count=5, # ниже дефолтного порога 10
|
||
thin_market=True,
|
||
)
|
||
assert summary.thin_market is True
|
||
assert summary.market_count == 5
|
||
|
||
|
||
def test_fix5b_thin_market_false_when_count_sufficient() -> None:
|
||
"""thin_market=False когда market_count >= threshold."""
|
||
from app.schemas.trade_in import AvitoImvSummary
|
||
|
||
summary = AvitoImvSummary(
|
||
recommended_price=5_000_000,
|
||
lower_price=4_500_000,
|
||
higher_price=5_500_000,
|
||
market_count=15, # выше дефолтного порога 10
|
||
thin_market=False,
|
||
)
|
||
assert summary.thin_market is False
|
||
|
||
|
||
def test_fix5b_thin_market_default_false() -> None:
|
||
"""thin_market дефолт=False — backward-compat."""
|
||
from app.schemas.trade_in import AvitoImvSummary
|
||
|
||
summary = AvitoImvSummary(recommended_price=5_000_000, market_count=100)
|
||
assert summary.thin_market is False
|
||
|
||
|
||
def test_fix5b_thin_market_none_count_no_flag() -> None:
|
||
"""market_count=None → thin_market не проставляется (остаётся False)."""
|
||
from app.schemas.trade_in import AvitoImvSummary
|
||
|
||
summary = AvitoImvSummary(recommended_price=5_000_000, market_count=None)
|
||
assert summary.thin_market is False
|
||
|
||
|
||
def test_fix5b_thin_market_logic_in_estimator() -> None:
|
||
"""Estimator-логика: _imv_mc < threshold → thin_market=True."""
|
||
from app.core.config import settings
|
||
|
||
threshold = settings.avito_imv_thin_market_threshold
|
||
# Ниже порога
|
||
market_count_low = threshold - 1
|
||
thin = market_count_low is not None and market_count_low < threshold
|
||
assert thin is True
|
||
|
||
# Выше порога
|
||
market_count_high = threshold + 5
|
||
thin2 = market_count_high is not None and market_count_high < threshold
|
||
assert thin2 is False
|