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

"""Tests for the asking→sold correction (#648 Stage 3).

Two layers:

1. `_get_asking_sold_ratio` lookup helper (DB mocked):
   - bucket = min(max(rooms or 0, 0), 4) — clamping + None handling
   - per-rooms row hit returns (ratio, basis)
   - per-rooms miss falls back to the global rooms_bucket=-1 row
   - empty / missing table → (None, None), never raises (graceful)
   - the in-process TTL cache memoises per bucket

2. The apply-logic in `estimate_quality` (every I/O stubbed, no DB, no network —
   same isolation harness as test_estimator_repair_coef.py):
   - when a ratio exists: expected_sold_price_rub ≈ round(median_price_rub * ratio)
     and expected_sold_per_m2 ≈ round(median_price_per_m2 * ratio)
   - the headline median_price_rub / ranges / per_m2 stay UNCHANGED (additive)
   - graceful path: lookup returns (None, None) → all expected_sold_* are None,
     no crash, headline still returned
"""

from __future__ import annotations

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

import anyio
import pytest

# Settings requires DATABASE_URL at init time. Set dummy DSN before any app import.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")


# ── Layer 1: _get_asking_sold_ratio lookup helper ───────────────────────────


class _FakeRow:
    """Stand-in for a SQLAlchemy Row (attribute access .ratio / .basis)."""

    def __init__(self, ratio: float | None, basis: str | None) -> None:
        self.ratio = ratio
        self.basis = basis


def _clear_ratio_cache() -> None:
    from app.services import estimator

    estimator._asking_sold_ratio_cache.clear()


def _db_returning(rows: list[_FakeRow | None]) -> MagicMock:
    """MagicMock db whose successive .execute(...).fetchone() yield `rows`."""
    db = MagicMock()
    db.execute.return_value.fetchone.side_effect = rows
    return db


def test_bucket_clamping() -> None:
    """rooms None→0, negative→0, >4→4; 0..4 pass through."""
    from app.services.estimator import _get_asking_sold_ratio

    cases = {None: 0, -3: 0, 0: 0, 1: 1, 2: 2, 3: 3, 4: 4, 5: 4, 99: 4}
    for rooms, expected_bucket in cases.items():
        _clear_ratio_cache()
        db = _db_returning([_FakeRow(0.8, "per_rooms")])
        _get_asking_sold_ratio(db, rooms)
        # First positional bind param is {"b": bucket}.
        bind = db.execute.call_args_list[0].args[1]
        assert (
            bind["b"] == expected_bucket
        ), f"rooms={rooms} → bucket {bind['b']} != {expected_bucket}"


def test_per_rooms_hit_returns_ratio_basis() -> None:
    """Per-rooms row present → returned directly, no fallback query."""
    from app.services.estimator import _get_asking_sold_ratio

    _clear_ratio_cache()
    db = _db_returning([_FakeRow(0.74, "per_rooms")])
    ratio, basis = _get_asking_sold_ratio(db, 1)
    assert ratio == 0.74
    assert basis == "per_rooms"
    assert db.execute.call_count == 1  # no global fallback needed


def test_global_fallback_when_per_rooms_missing() -> None:
    """Per-rooms miss (None) → second query for rooms_bucket=-1 global row."""
    from app.services.estimator import _get_asking_sold_ratio

    _clear_ratio_cache()
    db = _db_returning([None, _FakeRow(0.79, "global_fallback")])
    ratio, basis = _get_asking_sold_ratio(db, 3)
    assert ratio == 0.79
    assert basis == "global_fallback"
    assert db.execute.call_count == 2


def test_empty_table_returns_none_none() -> None:
    """Both queries miss (empty table) → (None, None), no raise."""
    from app.services.estimator import _get_asking_sold_ratio

    _clear_ratio_cache()
    db = _db_returning([None, None])
    assert _get_asking_sold_ratio(db, 2) == (None, None)


def test_missing_table_is_graceful() -> None:
    """db.execute raises (relation does not exist) → (None, None), swallowed."""
    from app.services.estimator import _get_asking_sold_ratio

    _clear_ratio_cache()
    db = MagicMock()
    db.execute.side_effect = RuntimeError("relation asking_to_sold_ratios does not exist")
    assert _get_asking_sold_ratio(db, 1) == (None, None)


def test_cache_memoises_per_bucket() -> None:
    """Second call for same bucket within TTL → no extra DB query."""
    from app.services.estimator import _get_asking_sold_ratio

    _clear_ratio_cache()
    db = _db_returning([_FakeRow(0.8, "per_rooms")])
    first = _get_asking_sold_ratio(db, 2)
    second = _get_asking_sold_ratio(db, 2)
    assert first == second == (0.8, "per_rooms")
    assert db.execute.call_count == 1  # second served from cache


# ── Layer 2: apply-logic inside estimate_quality (all I/O stubbed) ──────────


def _make_listing(*, price_per_m2: float, area_m2: float = 40.0) -> dict[str, Any]:
    price_rub = price_per_m2 * area_m2
    return {
        "source": "cian",
        "source_url": "https://cian.ru/offer/1",
        "address": "ЕКБ, ул. Учителей, 18",
        "lat": 56.838,
        "lon": 60.595,
        "rooms": 1,
        "area_m2": area_m2,
        "floor": 4,
        "total_floors": 16,
        "price_rub": price_rub,
        "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": 100.0,
        "relevance_score": 0.1,
    }


# Three fixed analogs → deterministic median ppm2 = 150_000 (< 5 ⇒ no outlier drop).
_ANALOGS: list[dict[str, Any]] = [
    _make_listing(price_per_m2=140_000.0),
    _make_listing(price_per_m2=150_000.0),
    _make_listing(price_per_m2=160_000.0),
]


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

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


def _make_payload():
    from app.schemas.trade_in import TradeInEstimateInput

    return TradeInEstimateInput(
        address="ЕКБ, ул. Учителей, 18",
        area_m2=40.0,
        rooms=1,
        floor=4,
        total_floors=16,
    )


def _run_estimate(ratio_tuple: tuple[float | None, str | None]):
    """estimate_quality with all deps stubbed; _get_asking_sold_ratio forced."""
    from app.services.estimator import estimate_quality

    db = MagicMock()
    payload = _make_payload()

    async def _run():
        with (
            patch("app.services.estimator.geocode", new=AsyncMock(return_value=_make_fake_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._get_asking_sold_ratio", return_value=ratio_tuple),
        ):
            return await estimate_quality(payload, db)

    return anyio.run(_run)


def test_expected_sold_applied_when_ratio_present(monkeypatch: pytest.MonkeyPatch) -> None:
    """ratio present → expected_sold_* ≈ asking × ratio; headline UNCHANGED."""
    # #2002: asserts the ratio mechanism (expected_sold == asking × ratio). Hold the
    # orthogonal hedonic year+area correction OFF (OFF ⇒ exact legacy expected_sold).
    from app.services import estimator as _est

    monkeypatch.setattr(_est.settings, "estimate_hedonic_correction_enabled", False)
    ratio = 0.74
    est = _run_estimate((ratio, "per_rooms"))

    # Headline asking-median is the unadjusted market median (additive invariant).
    expected_headline = int(150_000.0 * 40.0)  # 6_000_000
    assert est.median_price_rub == expected_headline
    assert est.median_price_per_m2 == 150_000

    # Parallel expected_sold point = asking × ratio (round()).
    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: expected_sold range is now a calibrated ~80% PI around the POINT
    # (point × [p10, p90] of sold/expected_sold), not the old asking-IQR × ratio band.
    from app.core.config import settings

    assert est.expected_sold_range_low_rub == round(
        est.expected_sold_price_rub * settings.estimate_pi_low_mult
    )
    assert est.expected_sold_range_high_rub == round(
        est.expected_sold_price_rub * settings.estimate_pi_high_mult
    )
    assert (
        est.expected_sold_range_low_rub
        <= est.expected_sold_price_rub
        <= est.expected_sold_range_high_rub
    )
    assert est.asking_to_sold_ratio == ratio
    assert est.ratio_basis == "per_rooms"


def test_headline_unchanged_vs_no_ratio() -> None:
    """median_price_rub / ranges / per_m2 identical with and without a ratio."""
    with_ratio = _run_estimate((0.8, "global_fallback"))
    without_ratio = _run_estimate((None, None))

    assert with_ratio.median_price_rub == without_ratio.median_price_rub
    assert with_ratio.range_low_rub == without_ratio.range_low_rub
    assert with_ratio.range_high_rub == without_ratio.range_high_rub
    assert with_ratio.median_price_per_m2 == without_ratio.median_price_per_m2


def test_graceful_when_ratio_none() -> None:
    """Lookup returns (None, None) → all expected_sold_* None, no crash."""
    est = _run_estimate((None, None))

    assert est.expected_sold_price_rub is None
    assert est.expected_sold_range_low_rub is None
    assert est.expected_sold_range_high_rub is None
    assert est.expected_sold_per_m2 is None
    assert est.asking_to_sold_ratio is None
    assert est.ratio_basis is None
    # Headline still produced normally.
    assert est.median_price_rub == int(150_000.0 * 40.0)


def test_global_fallback_basis_carried_through(monkeypatch: pytest.MonkeyPatch) -> None:
    """basis='global_fallback' propagates to the returned estimate."""
    # #2141: сохранённый asking_to_sold_ratio теперь честный дескриптор
    # (== expected_sold/median). Держим hedonic-коррекцию OFF, иначе uplift
    # сдвинул бы ratio с сырого 0.79 (сам пересчёт проверяют
    # test_honest_ratio_* в test_estimator_price_spine.py).
    from app.services import estimator as _est

    monkeypatch.setattr(_est.settings, "estimate_hedonic_correction_enabled", False)
    est = _run_estimate((0.79, "global_fallback"))
    assert est.ratio_basis == "global_fallback"
    assert est.asking_to_sold_ratio == 0.79


# ── #773: guard was `and listings_clean` → fixed to `and median_price > 0` ──


# Minimal anchor comp to produce a deterministic median_price via same-building anchor.
_ANCHOR_COMP_ONLY: list[dict] = [
    {"price_per_m2": 150_000, "area_m2": 40.0, "rooms": 1},
    {"price_per_m2": 155_000, "area_m2": 40.0, "rooms": 1},  # 4-й компл (#755 min_comps=4)
    {"price_per_m2": 160_000, "area_m2": 40.0, "rooms": 1},
    {"price_per_m2": 170_000, "area_m2": 40.0, "rooms": 1},
]


def _run_estimate_anchor_only(
    ratio_tuple: tuple[float | None, str | None],
    anchor_comps: list[dict] | None = None,
    anchor_tier: str | None = "A",
):
    """estimate_quality: пустой радиус (listings_clean=[]), якорь задаёт median_price."""
    from app.core.config import settings
    from app.services.estimator import estimate_quality

    db = MagicMock()
    payload = _make_payload()
    comps = _ANCHOR_COMP_ONLY if anchor_comps is None else anchor_comps

    async def _run():
        with (
            patch.object(settings, "estimate_same_building_anchor_enabled", True),
            patch("app.services.estimator.geocode", new=AsyncMock(return_value=_make_fake_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)),
            # Empty radius — listings_clean will be [] → median_price=0 before anchor.
            patch("app.services.estimator._fetch_analogs", return_value=([], False, None)),
            patch("app.services.estimator._fetch_deals", return_value=[]),
            patch("app.services.estimator._fetch_dkp_corridor", return_value=None),
            patch(
                "app.services.estimator._fetch_anchor_comps",
                return_value=(list(comps), anchor_tier),
            ),
            patch("app.services.estimator._fetch_house_imv_anchor", return_value=None),
            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._get_asking_sold_ratio", return_value=ratio_tuple),
        ):
            return await estimate_quality(payload, db)

    return anyio.run(_run)


def test_expected_sold_fires_on_anchor_only_no_radius_comps(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """#773: listings_clean=[], anchor sets median_price>0, ratio present
    → expected_sold_price > 0 (old guard `and listings_clean` blocked this)."""
    # #2002: asserts expected_sold == headline × ratio; hold hedonic correction OFF.
    from app.services import estimator as _est

    monkeypatch.setattr(_est.settings, "estimate_hedonic_correction_enabled", False)
    ratio = 0.82
    est = _run_estimate_anchor_only((ratio, "per_rooms"))

    # Anchor must have produced a positive headline.
    assert est.median_price_rub > 0, "anchor должен был задать median_price"

    # expected_sold must now be computed (was None before #773 fix).
    assert est.expected_sold_price_rub is not None
    assert est.expected_sold_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).
    from app.core.config import settings

    assert est.expected_sold_range_low_rub == round(
        est.expected_sold_price_rub * settings.estimate_pi_low_mult
    )
    assert est.expected_sold_range_high_rub == round(
        est.expected_sold_price_rub * settings.estimate_pi_high_mult
    )


def test_expected_sold_none_when_median_price_zero() -> None:
    """#773: ни радиуса, ни якоря → median_price=0, ratio present
    → expected_sold_* must remain None (guard `median_price > 0` blocks fabrication)."""
    ratio = 0.82
    est = _run_estimate_anchor_only((ratio, "per_rooms"), anchor_comps=[], anchor_tier=None)

    assert est.median_price_rub == 0
    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