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

"""Tests for Fix 4: clamp expected_sold <= asking (ratio cap 1.0).

Smoke-диагноз: 3к/27.0М -> expected_sold 31.4М > asking.
Причина: _get_asking_sold_ratio для high-price tier возвращает ratio > 1.0
(product artefact) -> expected_sold = headline x ratio > headline.

Проверяем:
1. ratio=1.15 + флаг on -> expected_sold == headline (не выше); лог clamp.
2. ratio=0.8 -> expected_sold = headline x 0.8 (норм, без clamp).
3. ratio None -> expected_sold не выводится (graceful).
4. Флаг off + ratio > 1 -> expected_sold > headline (старое поведение).
"""

from __future__ import annotations

import logging
import os
from datetime import UTC, datetime
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

import anyio
import pytest

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")


def _make_listing(*, price_per_m2: float, area_m2: float = 50.0) -> dict[str, Any]:
    return {
        "source": "cian",
        "source_url": "https://cian.ru/offer/1",
        "address": "ЕКБ, ул. Малышева, 30",
        "lat": 56.838,
        "lon": 60.595,
        "rooms": 3,
        "area_m2": area_m2,
        "floor": 10,
        "total_floors": 20,
        "price_rub": price_per_m2 * area_m2,
        "price_per_m2": price_per_m2,
        "listing_date": datetime(2026, 5, 1),
        "days_on_market": 5,
        "photo_urls": [],
        "scraped_at": datetime(2026, 5, 20, tzinfo=UTC),
        "distance_m": 100.0,
        "relevance_score": 0.1,
    }


# #oblast-E: 5 analogs (>= HEADLINE_LISTINGS_MIN_N) so the new headline
# sufficiency gate doesn't suppress the median before this test's target
# mechanic (expected-sold clamp) runs. Median stays 270_000.
_ANALOGS: list[dict[str, Any]] = [
    _make_listing(price_per_m2=250_000.0),
    _make_listing(price_per_m2=260_000.0),
    _make_listing(price_per_m2=270_000.0),
    _make_listing(price_per_m2=280_000.0),
    _make_listing(price_per_m2=290_000.0),
]


def _make_geo():
    from app.services.geocoder import GeocodeResult

    return GeocodeResult(
        lat=56.838,
        lon=60.595,
        full_address="Свердловская обл., Екатеринбург, ул. Малышева, 30",
        provider="nominatim",
    )


def _make_payload(rooms: int = 3):
    from app.schemas.trade_in import TradeInEstimateInput

    return TradeInEstimateInput(
        address="ЕКБ, ул. Малышева, 30", area_m2=50.0, rooms=rooms, floor=10, total_floors=20
    )


def _run_estimate(
    ratio_tuple: tuple[float | None, str | None],
) -> Any:
    from app.services.estimator import estimate_quality

    db = MagicMock()
    payload = _make_payload()

    async def _run() -> Any:
        with (
            # #2002: these tests assert the clamp/ratio math exactly. Hold the
            # orthogonal hedonic year+area correction OFF (OFF ⇒ legacy expected_sold).
            patch("app.core.config.settings.estimate_hedonic_correction_enabled", new=False),
            patch("app.services.estimator.geocode", new=AsyncMock(return_value=_make_geo())),
            patch("app.services.estimator.dadata_clean_address", new=AsyncMock(return_value=None)),
            patch("app.services.estimator.match_house_readonly", return_value=None),
            patch(
                "app.services.estimator.get_house_metadata",
                new=AsyncMock(return_value=None),
            ),
            patch(
                "app.services.estimator._fetch_analogs",
                return_value=(list(_ANALOGS), False, "S"),
            ),
            patch("app.services.estimator._fetch_deals", return_value=[]),
            patch(
                "app.services.estimator._get_or_fetch_imv_cached",
                new=AsyncMock(return_value=None),
            ),
            patch(
                "app.services.estimator._get_or_fetch_yandex_valuation_cached",
                new=AsyncMock(return_value=None),
            ),
            patch(
                "app.services.estimator.estimate_via_cian_valuation",
                new=AsyncMock(return_value=None),
            ),
            patch("app.services.estimator._fetch_dkp_corridor", return_value=None),
            patch("app.services.estimator._get_asking_sold_ratio", return_value=ratio_tuple),
        ):
            return await estimate_quality(payload, db)

    return anyio.run(_run)


def test_expected_sold_clamped_to_headline_when_ratio_above_1(
    caplog: pytest.LogCaptureFixture,
) -> None:
    ratio = 1.15
    with caplog.at_level(logging.INFO, logger="app.services.estimator"):
        est = _run_estimate((ratio, "per_rooms"))
    assert est.median_price_rub > 0, "headline должен быть задан аналогами"
    assert est.expected_sold_price_rub is not None
    assert est.expected_sold_price_rub <= est.median_price_rub, (
        f"expected_sold {est.expected_sold_price_rub} > asking {est.median_price_rub}"
    )
    assert est.expected_sold_per_m2 is not None
    assert est.expected_sold_per_m2 <= est.median_price_per_m2, (
        "expected_sold_per_m2 превышает median_price_per_m2"
    )
    # #1966: expected_sold range is now a calibrated ~80% PI around the point
    # (point × [p10, p90] of sold/expected_sold), so it is NO LONGER bounded by the
    # asking-IQR band — the high arm legitimately exceeds range_high.
    # The invariant we keep is the calibrated band itself + ordering low ≤ point ≤ high.
    # #3xxx: multipliers are regional now — resolve via pi_multipliers_for_region.
    from app.services.estimator import pi_multipliers_for_region

    _low_mult, _high_mult = pi_multipliers_for_region(66)
    assert est.expected_sold_range_high_rub is not None
    assert est.expected_sold_range_high_rub == round(est.expected_sold_price_rub * _high_mult)
    assert est.expected_sold_range_low_rub is not None
    assert est.expected_sold_range_low_rub == round(est.expected_sold_price_rub * _low_mult)
    assert (
        est.expected_sold_range_low_rub
        <= est.expected_sold_price_rub
        <= est.expected_sold_range_high_rub
    )
    assert est.expected_sold_price_rub == est.median_price_rub
    assert est.expected_sold_per_m2 == est.median_price_per_m2
    clamp_logs = [r for r in caplog.records if "clamped" in r.getMessage()]
    assert clamp_logs, "Ожидался log-message о clamp ratio"
    assert "1.150" in clamp_logs[0].getMessage() or "1.15" in clamp_logs[0].getMessage()


def test_expected_sold_not_clamped_when_ratio_below_1() -> None:
    ratio = 0.8
    est = _run_estimate((ratio, "per_rooms"))
    assert est.median_price_rub > 0
    assert est.expected_sold_price_rub == round(est.median_price_rub * ratio)
    assert est.expected_sold_per_m2 == round(est.median_price_per_m2 * ratio)
    # #1966: calibrated ~80% PI around the expected_sold point (not asking-IQR × ratio).
    # #3xxx: multipliers are regional now — resolve via pi_multipliers_for_region.
    from app.services.estimator import pi_multipliers_for_region

    _low_mult, _high_mult = pi_multipliers_for_region(66)
    assert est.expected_sold_range_low_rub == round(est.expected_sold_price_rub * _low_mult)
    assert est.expected_sold_range_high_rub == round(est.expected_sold_price_rub * _high_mult)
    assert est.expected_sold_price_rub < est.median_price_rub


def test_expected_sold_none_when_ratio_none() -> None:
    est = _run_estimate((None, None))
    assert est.expected_sold_price_rub is None
    assert est.expected_sold_per_m2 is None
    assert est.expected_sold_range_low_rub is None
    assert est.expected_sold_range_high_rub is None
    assert est.asking_to_sold_ratio is None
    assert est.median_price_rub > 0