diff --git a/tradein-mvp/backend/tests/test_estimator_price_spine.py b/tradein-mvp/backend/tests/test_estimator_price_spine.py new file mode 100644 index 00000000..cebee96b --- /dev/null +++ b/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