gendesign/tradein-mvp/backend/tests/test_estimator_audit_fixes.py

"""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
    SB_AREA_SIGMA,
    SB_FLOOR_SIGMA,
    SB_MAD_K,
    SB_MIN_COMPS,
    SB_ROOMS_MATCH_BOOST,
    _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 при большом разбросе"

    # Гейт low-confidence якоря применяется в 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 _c(ppm2: float, area: float | None = 50.0, rooms: int | None = 2, **kw: Any) -> dict:
    return {"price_per_m2": ppm2, "area_m2": area, "rooms": rooms, **kw}


# (tier, area_target, rooms_target, floor_target, total_floors_target, комплы)
_ANCHOR_POOLS: list[tuple[str, float, int | None, int | None, int | None, list[dict]]] = [
    # Tier A, топ-юнит дороже соседей по дому → uplift к p70.
    (
        "A",
        120.0,
        3,
        None,
        None,
        [_c(180_000, 120.0, 3), _c(200_000, 60.0), _c(210_000, 70.0), _c(220_000, 80.0)],
    ),
    # Tier A, широкий внутридомовой разброс (Хохрякова 48), без uplift.
    (
        "A",
        146.2,
        4,
        5,
        14,
        [
            _c(399_478, 153.2, 3),
            _c(472_298, 110.1, 3),
            _c(683_995, 146.2, 4),
            _c(510_000, 125.0, 3),
        ],
    ),
    # равные цены: взвешенное среднее обязано не уйти ниже минимума.
    ("A", 64.0, 3, None, None, [_c(112_500, 64.0, 3)] * 4),
    # Tier C, малая выборка: жёсткий MAD-клип срезает 600k.
    (
        "C",
        50.0,
        2,
        3,
        9,
        [_c(150_000), _c(155_000), _c(160_000), _c(158_000), _c(152_000), _c(600_000)],
    ),
    # тир вне A/C, широкий разброс → FSD > 0.20.
    ("", 50.0, 2, None, None, [_c(100_000), _c(180_000), _c(260_000), _c(340_000)]),
    # вес собран на самом дешёвом компле (площадь, комнаты, этаж), площадь не у всех.
    (
        "A",
        40.0,
        1,
        2,
        10,
        [
            _c(120_000, 40.0, 1, floor=2, total_floors=10),
            _c(130_000, 90.0, 3, floor=9, total_floors=10),
            _c(135_000, None, None),
            _c(128_000, 70.0, 2),
            _c(140_000, 85.0, 3),
        ],
    ),
]


def test_fix1_anchor_invariants_make_thin_gate_and_guardrail_dead() -> None:
    """#3590: инварианты, на которых держатся удаление gate_thin и недостижимость
    SB_GUARDRAIL_TOL. Если какой-то покраснел — ветка ожила, вернуть и покрыть.

    1. n ≥ SB_MIN_COMPS: якорь с n < 3 не выходит из функции (гейт «n < 3 И
       FSD > 0.20» удалён как недостижимый).
    2. FSD > 0.20 ⇒ confidence == "low": оставшийся гейт по confidence покрывает
       и второе условие удалённого гейта.
    3. anchor_ppm2 ≥ comp_min_ppm2: пол comp_min×(1−SB_GUARDRAIL_TOL) не срабатывает.
    """
    results = []
    for tier, area, rooms, floor, total_floors, comps in _ANCHOR_POOLS:
        a = _compute_same_building_anchor(
            comps,
            area_target=area,
            rooms_target=rooms,
            tier=tier,
            sigma=SB_AREA_SIGMA,
            rooms_boost=SB_ROOMS_MATCH_BOOST,
            floor_target=floor,
            total_floors_target=total_floors,
            floor_sigma=SB_FLOOR_SIGMA,
            min_comps=SB_MIN_COMPS,
            mad_k=SB_MAD_K,
        )
        assert a is not None, (tier, comps)
        assert a["n"] >= SB_MIN_COMPS, a
        assert a["fsd"] <= 0.20 or a["confidence"] == "low", a
        assert a["anchor_ppm2"] >= a["comp_min_ppm2"], a
        results.append((len(comps), a))

    # Пулы проходят ветки, где инвариант мог бы сломаться, а не только тривиальные.
    assert any(a["used_uplift"] for _, a in results), "нет пула с uplift к p70"
    assert any(a["n"] < n for n, a in results), "нет пула с MAD-клипом"
    assert any(a["fsd"] > 0.20 for _, a in results), "нет пула с FSD > 0.20"

    # Три компла — меньше SB_MIN_COMPS: якоря нет вовсе.
    three = [_c(150_000), _c(300_000), _c(160_000)]
    assert (
        _compute_same_building_anchor(
            three,
            area_target=50.0,
            rooms_target=2,
            tier="C",
            sigma=SB_AREA_SIGMA,
            rooms_boost=SB_ROOMS_MATCH_BOOST,
            min_comps=SB_MIN_COMPS,
        )
        is None
    )


# ---------------------------------------------------------------------------
# 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 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:
        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 — per-estimate sber staleness warning УДАЛЁН (#2846)
#
# Guard сравнивал возраст latest периода с sber_index_max_age_days=35. Такой
# возраст недостижим по построению (period_month — метка первого числа + лаг
# публикации ⇒ пол 46 суток), поэтому на проде warning писался у КАЖДОЙ оценки
# и не нёс ни бита. Пара прежних тестов зеленела только на фикстуре с «свежим»
# месяцем, которого в реальной серии не бывает. Свежесть теперь мерит одно место —
# tasks/sber_freshness_monitor, по отставанию ЗАГРУЗКИ.
# ---------------------------------------------------------------------------


def test_fix5a_no_per_estimate_staleness_warning(caplog: pytest.LogCaptureFixture) -> None:
    """Серия отдаётся как есть; календарного warning'а в горячем пути больше нет."""
    import logging

    from app.services.estimator import _load_sber_index_series

    stale_month = date(2024, 1, 1)  # два года назад — прежний guard тут кричал
    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 series == {stale_month: 100_000.0}
    stale_msgs = [r for r in caplog.records if "stale" in r.message.lower()]
    assert not stale_msgs, f"per-estimate guard вернулся, 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