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

"""#oblast-E — headline sufficiency gate (money-path audit, 2026-08-02).

Live-prod repro that motivated this gate: Серов 2к/45м², n=3 scraped listings →
headline 42 391 ₽/м² (−36% vs the city ДКП corridor, 54 126 ₽/м²); a neighbouring
street in the same town swung ±66% on 1-2 different random listings. Каменск-
Уральский returned a LITERAL 0 ₽ for a room/area combo with no local ДКП match
either, with no honest refusal surfaced. Первоуральск (0 listings) already fell
back to the (pre-existing) ДКП deals-headline fallback correctly — this gate
routes the THIN (1..HEADLINE_LISTINGS_MIN_N-1 listings) case into that SAME,
already-tested path instead of trusting a 1-4-lot median as the headline.

Two layers:
  1. `_price_from_inputs` unit tests (no DB, no estimate_quality overhead) —
     boundary behaviour of the gate itself.
  2. `estimate_quality` integration tests — proves the money-path invariants
     that matter to a caller: literal 0 never leaks as a "confident" price,
     display `analogs` cards never outnumber what `n_analogs` claims, and the
     explanation text describes what actually happened (not a stock "аналогов
     не найдено" when some WERE found, just too few).
"""

from __future__ import annotations

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

import anyio

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

from app.services import estimator
from app.services.estimator import HEADLINE_LISTINGS_MIN_N, _price_from_inputs
from app.services.geocoder import GeocodeResult

# ─────────────────────────────────────────────────────────────────────────────
# Layer 1 — `_price_from_inputs` direct unit tests
# ─────────────────────────────────────────────────────────────────────────────


def _geo() -> GeocodeResult:
    return GeocodeResult(
        lat=59.604,
        lon=60.577,
        full_address="Свердловская обл., Серов, ул. Ленина, 5",
        provider="nominatim",
    )


def _lot(ppm2: float, address: str = "ул. Ленина, 5", source: str = "avito") -> dict[str, Any]:
    return {"price_per_m2": ppm2, "address": address, "source": source}


def _lots(prices: list[float]) -> list[dict[str, Any]]:
    return [_lot(p, address=f"ул. Ленина, {i + 5}") for i, p in enumerate(prices)]


def _call(
    *,
    listings: list[dict[str, Any]],
    area_m2: float = 45.0,
    rooms: int | None = 2,
    dkp_raw: dict[str, Any] | None = None,
    anchor_comps: list[dict[str, Any]] | None = None,
    anchor_tier_fetched: str | None = None,
) -> estimator.PricingResult:
    def ratio_resolver(_appm2: float | None) -> tuple[float | None, str | None]:
        return None, None

    return _price_from_inputs(
        listings=listings,
        area_m2=area_m2,
        rooms=rooms,
        repair_state=None,
        floor=5,
        total_floors=9,
        target_year=None,
        analog_tier="W",
        fallback_used=False,
        area_widened=False,
        anchor_comps=anchor_comps or [],
        anchor_tier_fetched=anchor_tier_fetched,
        dkp_raw=dkp_raw,
        imv_anchor=None,
        imv_eval=None,
        yandex_val_present=False,
        cian_val_present=False,
        ratio_resolver=ratio_resolver,
        quarter_index_lookup=lambda q: None,
        quarter_indexes_lookup=lambda qs: {},
        target_house_cadnum=None,
        dadata_coarse=False,
        geo=_geo(),
        dadata_qc_geo=None,
    )


def test_threshold_is_five_not_lower() -> None:
    """The chosen sufficiency floor — see estimator.py module docstring (#oblast-E)
    for the data-driven justification (n=3 live-repro'd −36%, n=5 matches the
    existing MIN_ANALOGS_TIER_0 "enough to trust" convention)."""
    assert HEADLINE_LISTINGS_MIN_N == 5


def test_four_listings_below_threshold_suppressed_no_fallback() -> None:
    """n=4 (< 5), no ДКП signal → headline suppressed to the honest zero state,
    NOT the naive median of 4 listings."""
    pr = _call(listings=_lots([200_000.0, 210_000.0, 220_000.0, 230_000.0]))
    assert pr.median_ppm2 == 0.0
    assert pr.median_price == 0
    assert pr.n_analogs == 0
    assert pr.range_low == 0
    assert pr.range_high == 0


def test_five_listings_at_threshold_not_suppressed() -> None:
    """n=5 (== threshold) → the real listings median is trusted as the headline."""
    pr = _call(listings=_lots([200_000.0, 205_000.0, 210_000.0, 215_000.0, 220_000.0]))
    assert pr.median_ppm2 == 210_000.0
    assert pr.n_analogs == 5
    assert pr.median_price == round(210_000.0 * 45.0)


def test_one_listing_below_threshold_suppressed() -> None:
    """n=1 — the sharpest form of the Серов bug (a single random lot deciding
    the whole headline) — must be suppressed exactly like n=4."""
    pr = _call(listings=_lots([200_000.0]))
    assert pr.median_ppm2 == 0.0
    assert pr.n_analogs == 0


def test_thin_sample_with_sufficient_deals_uses_deals_headline() -> None:
    """n=3 listings (thin) + a usable ДКП corridor → headline comes from the
    deal corridor median, NOT the 3-listing median (live Серов repro: 3
    listings gave 42 391 vs the honest ДКП-based ~54 126)."""
    dkp_raw = {
        "count": 54,
        "low_ppm2": 44_000,
        "median_ppm2": 65_957,
        "high_ppm2": 89_000,
        "period_months": 12,
    }
    pr = _call(
        listings=_lots([42_391.0, 26_818.0, 75_058.0]),
        dkp_raw=dkp_raw,
    )
    assert pr.median_ppm2 == 65_957.0, (
        f"headline={pr.median_ppm2} must equal the ДКП corridor median, not the "
        "noisy 3-listing median (42 391 area)"
    )
    assert pr.n_analogs == 0, "honest: 0 scraped-listing analogs back this headline"
    assert pr.confidence == "low"


def test_thin_sample_with_insufficient_deals_stays_zero() -> None:
    """n=3 listings (thin) + a ДКП corridor that is ITSELF too thin
    (< DEALS_HEADLINE_FALLBACK_MIN_N) → neither source is trusted; honest zero,
    not a fabricated number from either side."""
    dkp_raw = {
        "count": 1,
        "low_ppm2": 40_000,
        "median_ppm2": 65_957,
        "high_ppm2": 80_000,
        "period_months": 12,
    }
    pr = _call(listings=_lots([42_391.0, 26_818.0, 75_058.0]), dkp_raw=dkp_raw)
    assert pr.median_ppm2 == 0.0
    assert pr.median_price == 0
    assert pr.n_analogs == 0


def test_thin_sample_explanation_is_honest_about_count() -> None:
    """The explanation for a thin-but-nonzero sample must say HOW MANY listings
    were found (not the generic 'ничего не найдено' text used for a genuine
    zero-listing case) — #4 in the task: explanation must match reality."""
    pr = _call(listings=_lots([200_000.0, 210_000.0]))  # n=2
    assert pr.explanation is not None
    assert "2" in pr.explanation
    assert "недостаточно" in pr.explanation.lower()
    # Must NOT reuse the "nothing found at all" copy — 2 listings WERE found.
    assert "не найдено аналогов" not in pr.explanation.lower()


def test_thin_sample_deals_fallback_explanation_does_not_claim_zero_listings() -> None:
    """#4: once the ДКП fallback fires for a thin (not zero) sample, the
    explanation must not falsely claim 'рядом нет объявлений' — some WERE
    found, just not enough to trust."""
    dkp_raw = {
        "count": 20,
        "low_ppm2": 40_000,
        "median_ppm2": 60_000,
        "high_ppm2": 80_000,
        "period_months": 12,
    }
    pr = _call(listings=_lots([200_000.0, 210_000.0, 220_000.0]), dkp_raw=dkp_raw)
    assert pr.explanation is not None
    assert "рядом нет актуальных объявлений" not in pr.explanation.lower()
    assert "сделкам росреестра" in pr.explanation.lower()


def test_thin_sample_listings_clean_preserved_for_anchor_ghost_guard() -> None:
    """Regression guard: the gate must suppress the AGGREGATE (median/n_analogs)
    without clearing `listings_clean` itself — the same-building anchor's own
    ghost-anchor guard (#1871) reads `listings_clean` truthiness to tell
    "genuinely zero nearby listings" from "some nearby, just too few to trust
    as headline", and conflating the two was caught regressing
    test_estimator_split_corridor_1871.py during this change."""
    pr = _call(listings=_lots([200_000.0, 210_000.0, 220_000.0]))
    assert pr.n_analogs == 0
    assert len(pr.listings_clean) == 3
    assert pr.listings_headline_thin_n == 3


def test_sufficient_sample_listings_headline_thin_n_is_zero() -> None:
    """Sanity/control: once n reaches the threshold, the thin-marker stays 0 —
    downstream (estimate_quality) must not treat a healthy sample as thin."""
    pr = _call(listings=_lots([200_000.0, 205_000.0, 210_000.0, 215_000.0, 220_000.0]))
    assert pr.listings_headline_thin_n == 0


# ─────────────────────────────────────────────────────────────────────────────
# Layer 2 — `estimate_quality` integration tests (full stub-patched I/O path)
# ─────────────────────────────────────────────────────────────────────────────


def _make_listing(*, price_per_m2: float, address: str, area_m2: float = 45.0) -> dict[str, Any]:
    return {
        "source": "avito",
        "source_url": f"https://avito.ru/offer/{address}",
        "address": address,
        "lat": 59.604,
        "lon": 60.577,
        "rooms": 2,
        "area_m2": area_m2,
        "floor": 5,
        "total_floors": 9,
        "price_rub": price_per_m2 * area_m2,
        "price_per_m2": price_per_m2,
        "listing_date": datetime(2026, 5, 1),
        "days_on_market": 10,
        "photo_urls": [],
        "scraped_at": datetime(2026, 5, 20, tzinfo=UTC),
        "distance_m": 150.0,
        "relevance_score": 0.1,
    }


def _serov_payload() -> Any:
    from app.schemas.trade_in import TradeInEstimateInput

    return TradeInEstimateInput(
        address="Серов, ул. Ленина, 5",
        area_m2=45.0,
        rooms=2,
        floor=5,
        total_floors=9,
        city_hint="Серов",
    )


def _run_estimate(
    *,
    analogs: list[dict[str, Any]],
    dkp_raw: dict[str, Any] | None,
) -> Any:
    from app.services.estimator import estimate_quality

    db = MagicMock()
    payload = _serov_payload()

    async def _run() -> Any:
        with (
            patch("app.services.estimator.geocode", new=AsyncMock(return_value=_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, "W"),
            ),
            patch("app.services.estimator._fetch_anchor_comps", return_value=([], None)),
            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=dkp_raw),
            patch("app.services.estimator._get_asking_sold_ratio", return_value=(None, None)),
        ):
            return await estimate_quality(payload, db)

    return anyio.run(_run)


def test_e2e_thin_no_deals_never_leaks_literal_zero_as_confident_price() -> None:
    """Каменск-Уральский-style repro: thin listings, no usable ДКП corridor —
    median_price_rub must be 0 AND insufficient_data must be True TOGETHER
    (the AggregatedEstimate.insufficient_data computed_field invariant that
    stops a literal 0 ₽ reaching the user as a confident number)."""
    analogs = [
        _make_listing(price_per_m2=200_000.0, address="ул. Ленина, 5"),
        _make_listing(price_per_m2=210_000.0, address="ул. Ленина, 7"),
    ]
    est = _run_estimate(analogs=analogs, dkp_raw=None)
    assert est.median_price_rub == 0
    assert est.insufficient_data is True
    assert est.n_analogs == 0
    assert est.confidence == "low"


def test_e2e_thin_sample_display_cards_never_outnumber_n_analogs() -> None:
    """The 2 thin listings must NOT be surfaced as `analogs` display cards while
    n_analogs reports 0 — that would be the same dishonesty (confident-looking
    UI) this whole gate exists to remove."""
    analogs = [
        _make_listing(price_per_m2=200_000.0, address="ул. Ленина, 5"),
        _make_listing(price_per_m2=210_000.0, address="ул. Ленина, 7"),
    ]
    est = _run_estimate(analogs=analogs, dkp_raw=None)
    assert est.n_analogs == 0
    assert est.analogs == []


def test_e2e_serov_repro_thin_sample_routes_to_deals_headline() -> None:
    """Live Серов repro (n=3 scraped listings, wide ДКП corridor available):
    headline must come from the deal corridor, not the noisy 3-listing median,
    and the estimate must be honestly non-'insufficient' (a real number, low
    confidence, deals-sourced)."""
    analogs = [
        _make_listing(price_per_m2=42_391.0, address="ул. Льва Толстого, 8А"),
        _make_listing(price_per_m2=26_818.0, address="ул. Кирова, 4"),
        _make_listing(price_per_m2=75_058.0, address="ул. Льва Толстого, 34"),
    ]
    dkp_raw = {
        "count": 54,
        "low_ppm2": 44_000,
        "median_ppm2": 65_957,
        "high_ppm2": 89_000,
        "period_months": 12,
    }
    est = _run_estimate(analogs=analogs, dkp_raw=dkp_raw)
    assert est.median_price_per_m2 == 65_957
    assert est.insufficient_data is False
    assert est.n_analogs == 0
    assert est.confidence == "low"
    assert est.confidence_explanation is not None
    assert "сделкам росреестра" in est.confidence_explanation.lower()


def test_e2e_sufficient_five_analogs_unaffected_control() -> None:
    """Control (mirrors the Екатеринбург prod check in the PR): a sample that
    clears the threshold is priced exactly as before — headline is the real
    listings median, all 5 analogs counted."""
    analogs = [
        _make_listing(price_per_m2=195_000.0, address="ул. Ленина, 5"),
        _make_listing(price_per_m2=205_000.0, address="ул. Ленина, 7"),
        _make_listing(price_per_m2=210_000.0, address="ул. Ленина, 9"),
        _make_listing(price_per_m2=215_000.0, address="ул. Ленина, 11"),
        _make_listing(price_per_m2=225_000.0, address="ул. Ленина, 13"),
    ]
    est = _run_estimate(analogs=analogs, dkp_raw=None)
    assert est.median_price_per_m2 == 210_000
    assert est.n_analogs == 5
    assert est.insufficient_data is False