Merge pull request 'refactor(tradein/estimator): extract deterministic pricing into pure _price_from_inputs (#1966 foundation)' (#1976) from refactor/tradein-estimator-price-spine into main

Reviewed-on: #1976

tradein-mvp/backend/tests/test_estimator_price_spine.py

@@ -0,0 +1,401 @@
+"""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() -> None:
+    """Pure radius path: 7 uniform lots → correct median, n_analogs, expected_sold."""
+    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() -> None:
+    """expected_sold = headline × ratio; when ratio is None, all expected_sold fields None."""
+    # 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