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

"""#2827 — confidence↔reliability consistency (live-prod smoke, 2026-08-11).

`confidence` (`_compute_confidence` — unique-address/IQR metric) and
`reliability` (`estimate_quality` — n_analogs-bucket + #oblast-F relaxation-
cascade metric, see #2823) were computed fully independently. The cascade's
room-adjacency/freshness/novostroyki steps never set `fallback_used`, so
`_compute_confidence` stayed blind to them and could keep scoring a wide,
tight-IQR (post-relaxation) sample as "high", while `reliability` — which DOES
see the relaxations — honestly read "low".

Live prod repro (customer address, rooms=1, 23.1 m², radius=2000):
    n_analogs: 39
    confidence: high
    reliability: low
    relaxations: ['снят фильтр по году постройки', 'учтены студии', 'площадь ±25%']

A client could see a "высокая уверенность" badge next to a "точность снижена"
banner on the SAME estimate — exactly the class of contradiction cleaned up
from the PDF/counters earlier in this cycle (#2824-adjacent work), now leaking
through the two headline confidence signals themselves.

Fix: `_cap_confidence_by_reliability` (estimator.py), applied ONCE in
`estimate_quality` right after both `confidence` and `reliability` are final
— NOT spread across the #oblast-F cascade steps. Rule:
    reliability == 'very_low' → confidence forced to 'low'
    reliability == 'low'      → confidence capped at 'medium'
    reliability == 'ok'       → confidence untouched (common case, unaffected)

Two layers:
  1. `_cap_confidence_by_reliability` direct unit tests — the 3 rules in
     isolation, no DB/estimate_quality overhead.
  2. `estimate_quality` integration tests — the prod repro shape (confidence
     downgraded + confidence_explanation restructured to LEAD with the
     accuracy-reduced summary) and the byte-identical-when-unrelaxed control.
"""

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.estimator import _cap_confidence_by_reliability
from app.services.geocoder import GeocodeResult

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


def test_very_low_reliability_forces_confidence_low() -> None:
    """reliability == 'very_low' → confidence forced to 'low', regardless of
    what _compute_confidence originally scored."""
    for original in ("high", "medium", "low"):
        assert _cap_confidence_by_reliability(original, "very_low") == "low", (
            f"original={original!r} must be forced to 'low' under very_low reliability"
        )


def test_low_reliability_caps_confidence_at_medium() -> None:
    """reliability == 'low' → confidence capped at 'medium': 'high' is pulled
    down to 'medium'; 'medium'/'low' pass through unchanged (already <= cap)."""
    assert _cap_confidence_by_reliability("high", "low") == "medium"
    assert _cap_confidence_by_reliability("medium", "low") == "medium"
    assert _cap_confidence_by_reliability("low", "low") == "low"


def test_ok_reliability_leaves_confidence_untouched() -> None:
    """reliability == 'ok' — the common, unrelaxed case — must NOT change
    confidence at all (byte-identical to pre-#2827 behaviour)."""
    for original in ("high", "medium", "low"):
        assert _cap_confidence_by_reliability(original, "ok") == original


def test_cap_never_raises_confidence() -> None:
    """Sanity: the cap only lowers/holds — never raises 'low' to something
    higher under any reliability value."""
    for reliability in ("ok", "low", "very_low"):
        assert _cap_confidence_by_reliability("low", reliability) == "low"


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


def _geo() -> GeocodeResult:
    return GeocodeResult(
        lat=56.838,
        lon=60.595,
        full_address="Свердловская обл., Екатеринбург, ул. Академика Парина, 46/5",
        provider="nominatim",
    )


def _make_listing(*, price_per_m2: float, address: str, area_m2: float = 23.1) -> dict[str, Any]:
    return {
        "source": "avito",
        "source_url": f"https://avito.ru/offer/{address}",
        "address": address,
        "lat": 56.838,
        "lon": 60.595,
        "rooms": 1,
        "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 _tight_price_pool(n: int, base_ppm2: float = 195_000.0) -> list[dict[str, Any]]:
    """n listings, distinct addresses, tightly clustered price/m² (~±2%) —
    engineered to clear _compute_confidence's 'high' bar (unique_addr_count>=7
    AND iqr_pct<0.15) once assembled, exactly like the live prod repro (n=39,
    confidence='high' pre-fix)."""
    return [
        _make_listing(
            price_per_m2=base_ppm2 + (i - n / 2) * 200,
            address=f"ул. Академика Парина, {i + 1}",
        )
        for i in range(n)
    ]


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

    return TradeInEstimateInput(
        address="ЕКБ, ул. Академика Парина, 46/5",
        area_m2=23.1,
        rooms=1,
        floor=5,
        total_floors=9,
        year_built=2010,
        city_hint="Екатеринбург",
        radius_m=2000,
    )


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

    db = MagicMock()
    geo = _geo()

    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",
                side_effect=fetch_analogs_side_effect,
            ),
            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_prod_repro_high_confidence_low_reliability_gets_capped() -> None:
    """Live prod repro shape: cohort (year_built) filter gets dropped by the
    Tier-0 cascade (empty cohort-call → fallback), landing on a 39-listing,
    tight-price, 39-unique-address pool — a healthy sample that
    `_compute_confidence` would score 'high' on its own. Because a relaxation
    WAS applied to get there, `reliability` reads 'low' — the cap must pull
    `confidence` down to 'medium' (not leave it at the contradictory 'high'),
    and `confidence_explanation` must LEAD with the accuracy-reduced summary."""
    pool = _tight_price_pool(39)

    def _fetch_analogs_stub(*_args: Any, **kwargs: Any) -> tuple[list[dict[str, Any]], bool, str]:
        # Tier 0 (with cohort filter) → empty, forces the "drop cohort" fallback.
        if kwargs.get("cohort_year_min") is not None:
            return [], False, "W"
        # Every subsequent (no-cohort) call → the full healthy pool.
        return list(pool), False, "W"

    est = _run_estimate(payload=_payload_with_year(), fetch_analogs_side_effect=_fetch_analogs_stub)

    assert est.n_analogs == 39
    assert est.reliability == "low"
    assert "снят фильтр по году постройки" in est.relaxations
    assert est.confidence == "medium", (
        f"confidence={est.confidence!r} must be capped to 'medium' under reliability='low' "
        "(was 'high' pre-#2827 — the exact prod contradiction)"
    )
    explanation = est.confidence_explanation
    assert explanation is not None
    assert explanation.startswith(
        "Оценка построена с расширенными параметрами подбора — точность снижена."
    ), f"explanation must LEAD with the accuracy-reduced summary, got: {explanation!r}"
    assert "Найдено 39 аналогов" in explanation, "original detail must be preserved"
    assert "Применены послабления подбора: снят фильтр по году постройки." in explanation


def test_e2e_unrelaxed_estimate_is_byte_identical_control() -> None:
    """#3 (task spec): no relaxations, reliability=='ok' → confidence and
    confidence_explanation must be UNCHANGED by #2827 — this is the common
    path most estimates take, and it must not regress."""
    pool = _tight_price_pool(39)

    def _fetch_analogs_stub(*_args: Any, **kwargs: Any) -> tuple[list[dict[str, Any]], bool, str]:
        # No cohort filter this time (payload has no year_built) — Tier 0 is
        # skipped outright (cohort_range is None), so no relaxation fires.
        return list(pool), False, "W"

    from app.schemas.trade_in import TradeInEstimateInput

    payload = TradeInEstimateInput(
        address="ЕКБ, ул. Академика Парина, 46/5",
        area_m2=23.1,
        rooms=1,
        floor=5,
        total_floors=9,
        city_hint="Екатеринбург",
    )
    est = _run_estimate(payload=payload, fetch_analogs_side_effect=_fetch_analogs_stub)

    assert est.n_analogs == 39
    assert est.relaxations == []
    assert est.reliability == "ok"
    assert est.confidence == "high", "unrelaxed healthy sample must keep its real confidence"
    assert est.confidence_explanation is not None
    assert not est.confidence_explanation.startswith("Оценка построена с расширенными параметрами")
    assert "Применены послабления подбора" not in est.confidence_explanation
    assert est.confidence_explanation.startswith("Найдено 39 аналогов"), (
        "unrelaxed explanation must keep its original (pre-#2827) leading sentence, got: "
        f"{est.confidence_explanation!r}"
    )