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

"""Hermetic unit tests for _price_from_inputs (#1966).

Calls the pure synchronous pricing function directly with stub callables and
hand-built inputs — no DB, no network, no mocks. Verifies that the extraction
preserved the pricing logic identically to the original block in estimate_quality.

NOTE: importing app.services.estimator pulls app.core.config.Settings which
requires DATABASE_URL. Set it BEFORE importing app modules.
"""

import os

import pytest

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

from app.services import estimator
from app.services.geocoder import GeocodeResult

# ── helpers ──────────────────────────────────────────────────────────────────


def _geo(coarse: bool = False) -> GeocodeResult:
    """Minimal GeocodeResult for test injection."""
    full_address = "Екатеринбург" if coarse else "ул. Тестовая, 1"
    return GeocodeResult(
        lat=56.838,
        lon=60.597,
        full_address=full_address,
        provider="nominatim",
        confidence="approximate",
    )


def _lot(ppm2: float, address: str = "ул. Тестовая, 1", source: str = "avito") -> dict:
    return {"price_per_m2": ppm2, "address": address, "source": source}


def _lots(ppm2: float, n: int = 7) -> list[dict]:
    """n unique-address lots all at the same ppm2."""
    return [_lot(ppm2, address=f"ул. Тестовая, {i + 1}") for i in range(n)]


def _dkp_raw(
    low: int = 80_000,
    median: int = 120_000,
    high: int = 150_000,
    count: int = 20,
    period_months: int = 12,
) -> dict:
    return {
        "low_ppm2": low,
        "median_ppm2": median,
        "high_ppm2": high,
        "count": count,
        "period_months": period_months,
    }


def _anchor_comp(ppm2: float, area: float = 50.0, rooms: int = 2) -> dict:
    return {"price_per_m2": ppm2, "area_m2": area, "rooms": rooms}


# Stub callables — returned in each test via closure.
def _ratio_stub(
    ratio: float | None,
    basis: str | None = "per_rooms",
) -> "tuple[float | None, str | None]":
    return ratio, basis if ratio is not None else None


def _qi_stub_none(q: str) -> "tuple[float, int] | None":
    return None


def _qis_stub_empty(qs: list[str]) -> dict[str, float]:
    return {}


def _call(
    *,
    listings: list[dict] | None = None,
    area_m2: float = 50.0,
    rooms: int | None = 2,
    repair_state: str | None = None,
    floor: int | None = 5,
    total_floors: int | None = 10,
    target_year: int | None = None,
    analog_tier: str = "W",
    fallback_used: bool = False,
    area_widened: bool = False,
    anchor_comps: list[dict] | None = None,
    anchor_tier_fetched: str | None = None,
    dkp_raw: dict | None = None,
    imv_anchor: dict | None = None,
    imv_eval=None,
    yandex_val_present: bool = False,
    cian_val_present: bool = False,
    ratio: float | None = None,
    quarter_index_lookup=None,
    quarter_indexes_lookup=None,
    target_house_cadnum: str | None = None,
    dadata_coarse: bool = False,
    geo: GeocodeResult | None = None,
    dadata_qc_geo: int | None = None,
) -> estimator.PricingResult:
    if listings is None:
        listings = _lots(100_000)
    if anchor_comps is None:
        anchor_comps = []
    if geo is None:
        geo = _geo(coarse=dadata_coarse)
    if quarter_index_lookup is None:
        quarter_index_lookup = _qi_stub_none
    if quarter_indexes_lookup is None:
        quarter_indexes_lookup = _qis_stub_empty

    _ratio = ratio
    _basis = "per_rooms" if ratio is not None else None

    def ratio_resolver(appm2: float | None) -> tuple[float | None, str | None]:
        return _ratio, _basis if _ratio is not None else None

    return estimator._price_from_inputs(
        listings=listings,
        area_m2=area_m2,
        rooms=rooms,
        repair_state=repair_state,
        floor=floor,
        total_floors=total_floors,
        target_year=target_year,
        analog_tier=analog_tier,
        fallback_used=fallback_used,
        area_widened=area_widened,
        anchor_comps=anchor_comps,
        anchor_tier_fetched=anchor_tier_fetched,
        dkp_raw=dkp_raw,
        imv_anchor=imv_anchor,
        imv_eval=imv_eval,
        yandex_val_present=yandex_val_present,
        cian_val_present=cian_val_present,
        ratio_resolver=ratio_resolver,
        quarter_index_lookup=quarter_index_lookup,
        quarter_indexes_lookup=quarter_indexes_lookup,
        target_house_cadnum=target_house_cadnum,
        dadata_coarse=dadata_coarse,
        geo=geo,
        dadata_qc_geo=dadata_qc_geo,
    )


# ── Tests ────────────────────────────────────────────────────────────────────


def test_radius_only_median_and_expected_sold(monkeypatch: pytest.MonkeyPatch) -> None:
    """Pure radius path: 7 uniform lots → correct median, n_analogs, expected_sold."""
    # #2002: this asserts the ratio mechanism (expected_sold == headline × ratio).
    # Hold the orthogonal hedonic correction OFF (OFF ⇒ exact legacy expected_sold).
    monkeypatch.setattr(estimator.settings, "estimate_hedonic_correction_enabled", False)
    pr = _call(listings=_lots(100_000, n=7), ratio=0.95)

    assert pr.median_price == int(100_000 * 50.0)  # 5_000_000
    assert pr.median_ppm2 == 100_000.0
    assert pr.n_analogs == 7
    assert pr.anchor_tier is None
    assert pr.dkp_corridor is None
    assert pr.asking_to_sold_ratio == 0.95
    assert pr.expected_sold_price == round(5_000_000 * 0.95)  # 4_750_000
    assert pr.expected_sold_per_m2 == round(100_000 * 0.95)  # 95_000
    assert "avito" in pr.sources_used_pre
    assert len(pr.listings_clean) == 7


def test_same_building_anchor_tier_a_mutates_headline() -> None:
    """Tier A same-building anchor replaces radius median with higher price.

    5 radius lots at 100k ppm2 (5M total). 5 anchor comps at 200k ppm2 (10M total).
    After anchor fires: median_price >> radius median, n_analogs == anchor count.
    """
    comps = [_anchor_comp(200_000) for _ in range(5)]
    pr = _call(
        listings=_lots(100_000, n=5),
        anchor_comps=comps,
        anchor_tier_fetched="A",
        ratio=None,
    )

    # Anchor must have fired (not suppressed).
    assert pr.anchor_tier == "A"
    # Headline is anchor-derived — must be above radius median (5_000_000).
    assert pr.median_price > 5_000_000
    # n_analogs resets to anchor population.
    assert pr.n_analogs == 5
    # anchor_comps_used is the injected comps list.
    assert len(pr.anchor_comps_used) == 5
    # No ratio → expected_sold is None.
    assert pr.expected_sold_price is None


def test_tier_c_corridor_gate_suppresses_anchor() -> None:
    """Tier C anchor ppm2 >> corridor_high × mult → anchor suppressed.

    anchor_tier remains "C" in the result (gate sets anchor=None but doesn't
    reset anchor_tier); headline stays at the radius median.
    """
    # 5 comps at 300k ppm2; corridor_high=150k; gate threshold=150k×1.5=225k.
    # 300k > 225k → suppressed.
    comps = [_anchor_comp(300_000) for _ in range(5)]
    radius_median_price = int(100_000 * 50.0)
    pr = _call(
        listings=_lots(100_000, n=5),
        anchor_comps=comps,
        anchor_tier_fetched="C",
        dkp_raw=_dkp_raw(high=150_000, count=15),
        ratio=None,
    )

    # Tier C gate sets anchor=None but leaves anchor_tier="C".
    assert pr.anchor_tier == "C"
    # Headline was NOT mutated by the suppressed anchor — stays at radius median.
    assert pr.median_price == radius_median_price
    # anchor_comps_used stays empty (anchor didn't fire).
    assert pr.anchor_comps_used == []


def test_low_conf_gate_suppresses_anchor() -> None:
    """Low-confidence anchor is suppressed; anchor_tier reset to None.

    4 comps with wide spread → high cv → fsd > 0.20 → confidence='low' → suppressed.
    """
    # 4 comps at [100k, 200k, 300k, 400k] → cv≈0.45, fsd≈0.20 → "low".
    comps = [_anchor_comp(p) for p in [100_000, 200_000, 300_000, 400_000]]
    radius_median_price = int(100_000 * 50.0)
    pr = _call(
        listings=_lots(100_000, n=5),
        anchor_comps=comps,
        anchor_tier_fetched="A",  # starts as A, gate resets to None
        ratio=None,
    )

    # Gate resets anchor_tier to None on suppression.
    assert pr.anchor_tier is None
    # Headline stays at radius median.
    assert pr.median_price == radius_median_price
    assert pr.anchor_comps_used == []


def test_imv_blend_raises_median_when_anchor_tier_none() -> None:
    """IMV blend pushes radius median up when IMV >> median × threshold.

    radius median=5M, IMV recommended=7M, area=50, weight=0.5, threshold=1.15.
    IMV/radius = 7M/5M = 1.4 > 1.15 → blend: new_median = round(5M×0.5 + 7M×0.5).
    """
    imv_anchor = {
        "recommended_price": 7_000_000,
        "lower_price": 6_000_000,
        "higher_price": 8_000_000,
        "market_count": 50,
    }
    pr = _call(
        listings=_lots(100_000, n=5),
        imv_anchor=imv_anchor,
        ratio=None,
    )

    expected_median = round(5_000_000 * 0.5 + 7_000_000 * 0.5)  # 6_000_000
    assert pr.median_price == expected_median
    assert pr.range_high == 8_000_000  # from anchor_higher
    assert pr.avito_imv_summary is not None
    assert pr.avito_imv_summary.recommended_price == 7_000_000
    assert "avito_imv" in pr.sources_used_pre


def test_quarter_index_guard2_skip_when_all_analogs_in_target_quarter() -> None:
    """Guard-2: when all analogs are in the target quarter, index is NOT applied.

    same_quarter_ratio=1.0 > skip_ratio=0.6 → Guard-2 fires → median unchanged.
    """
    target_cadnum = "66:41:0204016:350"
    target_quarter = "66:41:0204016"
    # Analogs are all in the SAME quarter as the target.
    lots = [
        {
            "price_per_m2": 100_000,
            "address": f"ул. Тестовая, {i + 1}",
            "source": "avito",
            "building_cadastral_number": f"{target_quarter}:{100 + i}",
        }
        for i in range(5)
    ]

    qi_called: list[str] = []

    def qi_lookup(q: str) -> tuple[float, int] | None:
        qi_called.append(q)
        return (1.5, 100) if q == target_quarter else None  # high index — would change price

    pr = _call(
        listings=lots,
        target_house_cadnum=target_cadnum,
        quarter_index_lookup=qi_lookup,
        quarter_indexes_lookup=_qis_stub_empty,
        ratio=None,
    )

    # Guard-2 fired: median must remain unchanged (5M, not ×1.5).
    assert pr.median_price == int(100_000 * 50.0)
    # quarter_index was looked up but did NOT add "quarter_index" to sources.
    assert "quarter_index" not in pr.sources_used_pre


def test_quarter_index_applied_when_analogs_in_different_quarter() -> None:
    """Quarter-index IS applied when analogs are in a different quarter from target.

    target_qi=1.2, avg_analog_qi=1.0 → factor=1.2 → median_price×1.2.
    Guard-2 skips (same_quarter_ratio=0.0 < 0.6).
    """
    target_cadnum = "66:41:0204016:350"
    target_quarter = "66:41:0204016"
    analog_quarter = "66:41:0999999"
    lots = [
        {
            "price_per_m2": 100_000,
            "address": f"ул. Иная, {i + 1}",
            "source": "avito",
            "building_cadastral_number": f"{analog_quarter}:{i + 1}",
        }
        for i in range(5)
    ]

    def qi_lookup(q: str) -> tuple[float, int] | None:
        return (1.2, 100) if q == target_quarter else None

    def qis_lookup(qs: list[str]) -> dict[str, float]:
        return {q: 1.0 for q in qs if q == analog_quarter}

    pr = _call(
        listings=lots,
        target_house_cadnum=target_cadnum,
        quarter_index_lookup=qi_lookup,
        quarter_indexes_lookup=qis_lookup,
        ratio=None,
    )

    # factor=1.2/1.0=1.2; original radius=5M → adjusted=6M (±rounding via _apply_quarter_index)
    assert pr.median_price > int(100_000 * 50.0)  # index pushed price up
    assert "quarter_index" in pr.sources_used_pre


def test_corridor_soft_clamp_headline_above_cap() -> None:
    """Headline above corridor_high × (1+slack) is clamped down.

    radius lots at 250k ppm2. corridor_high=150k, slack=0.40 →
    cap=150k×1.40=210k. 250k > 210k → clamped to 210k.
    """
    pr = _call(
        listings=_lots(250_000, n=7),
        dkp_raw=_dkp_raw(low=80_000, median=120_000, high=150_000, count=15),
        ratio=None,
    )

    # cap = 150_000 × 1.40 = 210_000
    # Clamped: new ppm2 == 210_000, new_price = round(210_000 × 50)
    assert pr.median_ppm2 == pytest.approx(210_000.0, rel=1e-4)
    assert pr.median_price == round(210_000 * 50.0)
    # DKP corridor present in result.
    assert pr.dkp_corridor is not None
    assert pr.dkp_corridor.high_ppm2 == 150_000


def test_expected_sold_from_ratio_and_none_when_ratio_none(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """expected_sold = headline × ratio; when ratio is None, all expected_sold fields None."""
    # #2002: ratio-mechanism test — hold the orthogonal hedonic correction OFF
    # so expected_sold == headline × ratio exactly (OFF ⇒ legacy behavior).
    monkeypatch.setattr(estimator.settings, "estimate_hedonic_correction_enabled", False)
    # Case A: ratio=0.90 → expected_sold fields filled.
    pr_ratio = _call(listings=_lots(100_000, n=5), ratio=0.90)
    assert pr_ratio.asking_to_sold_ratio == 0.90
    assert pr_ratio.expected_sold_price is not None
    assert pr_ratio.expected_sold_price == round(pr_ratio.median_price * 0.90)
    assert pr_ratio.expected_sold_per_m2 is not None
    assert pr_ratio.expected_sold_range_low is not None
    assert pr_ratio.expected_sold_range_high is not None

    # Case B: ratio=None → all expected_sold fields None.
    pr_none = _call(listings=_lots(100_000, n=5), ratio=None)
    assert pr_none.asking_to_sold_ratio is None
    assert pr_none.ratio_basis is None
    assert pr_none.expected_sold_price is None
    assert pr_none.expected_sold_per_m2 is None
    assert pr_none.expected_sold_range_low is None
    assert pr_none.expected_sold_range_high is None


def test_coarse_geo_downgrades_confidence_to_low() -> None:
    """dadata_coarse=True with qc_geo=2 → confidence='low' with settlement label."""
    pr = _call(
        listings=_lots(100_000, n=7),
        dadata_coarse=True,
        dadata_qc_geo=2,
        ratio=None,
        # No anchor, no IMV → radius path → anchor_tier is None (not "A" → downgrade applies)
        geo=_geo(coarse=False),  # geo itself not coarse; using dadata_coarse signal
    )

    assert pr.confidence == "low"
    assert "населённого пункта" in pr.explanation


# ── #2141: честный asking_to_sold_ratio (дескриптор бейджа «−N% к рынку») ──────


def test_honest_ratio_invariant_with_hedonic_uplift(monkeypatch: pytest.MonkeyPatch) -> None:
    """#2141: hedonic-uplift (#2002) поднимает expected_sold выше median×raw_ratio →
    сохранённый asking_to_sold_ratio пересчитывается в ЧЕСТНЫЙ expected_sold/median,
    но сам expected_sold (выкуп) НЕ падает до наивного median×raw. Это ровно тот
    live-дефект, где бейдж показывал «−23%» при фактических «−5%»."""
    monkeypatch.setattr(estimator.settings, "estimate_hedonic_correction_enabled", True)
    raw_ratio = 0.74
    # Малая площадь + новостройка → hedonic factor у верхней границы (≈1.28).
    pr = _call(
        listings=_lots(200_000, n=7),
        area_m2=25.0,
        rooms=1,
        target_year=2024,
        ratio=raw_ratio,
    )
    assert pr.expected_sold_price is not None
    assert pr.asking_to_sold_ratio is not None
    assert pr.median_price > 0
    # Выкуп НЕ упал до наивного median×raw_ratio — hedonic-uplift сохранён.
    assert pr.expected_sold_price > round(pr.median_price * raw_ratio)
    # Инвариант: сохранённый ratio == фактическому expected_sold/median (бейдж честный).
    assert abs(pr.asking_to_sold_ratio - pr.expected_sold_price / pr.median_price) < 1e-3
    # Пересчёт действительно сработал (ratio уже не сырой табличный).
    assert pr.asking_to_sold_ratio > raw_ratio + 1e-3


def test_honest_ratio_invariant_under_corridor_clamp(monkeypatch: pytest.MonkeyPatch) -> None:
    """#2141: даже когда corridor-clamp прижимает headline ВНИЗ к ДКП-коридору,
    сохранённый asking_to_sold_ratio == expected_sold/median на ФИНАЛЬНЫХ значениях."""
    monkeypatch.setattr(estimator.settings, "estimate_hedonic_correction_enabled", True)
    pr = _call(
        listings=_lots(250_000, n=7),
        area_m2=50.0,
        rooms=2,
        target_year=2024,
        dkp_raw=_dkp_raw(low=80_000, median=120_000, high=150_000, count=15),
        ratio=0.80,
    )
    assert pr.expected_sold_price is not None
    assert pr.asking_to_sold_ratio is not None
    # corridor-clamp прижал headline (250k > cap=150k×1.40=210k).
    assert pr.median_ppm2 < 250_000
    # Инвариант держится на финальных значениях (после clamp + hedonic).
    assert abs(pr.asking_to_sold_ratio - pr.expected_sold_price / pr.median_price) < 1e-3


def test_honest_ratio_byte_identical_without_shift(monkeypatch: pytest.MonkeyPatch) -> None:
    """#2141 regression: hedonic OFF + без клампа → expected == median×ratio, и
    сохранённый ratio остаётся СЫРЫМ табличным байт-в-байт (дескриптор не пересчитан)."""
    monkeypatch.setattr(estimator.settings, "estimate_hedonic_correction_enabled", False)
    pr = _call(listings=_lots(100_000, n=7), area_m2=50.0, rooms=2, ratio=0.90)
    assert pr.asking_to_sold_ratio == 0.90  # байт-в-байт сырой табличный ratio
    assert pr.expected_sold_price == round(pr.median_price * 0.90)
    assert abs(pr.asking_to_sold_ratio - pr.expected_sold_price / pr.median_price) < 1e-3