bot-backend
2ea49b637f
Behavior-preserving structural refactor: the ~870-line deterministic pricing block inside estimate_quality() is extracted into a new pure synchronous function _price_from_inputs() returning a PricingResult dataclass. Key design decisions: - All async DB fetches (imv_eval, yandex_val, cian_val) hoisted to estimate_quality BEFORE the call; passed as pre-fetched values / bool flags. - DB-dependent helpers whose arguments are computed inside the block (_get_asking_sold_ratio, _lookup_quarter_index, _lookup_quarter_indexes) injected as Callable parameters (ratio_resolver, quarter_index_lookup, quarter_indexes_lookup). - _fetch_house_imv_anchor called once in estimate_quality (was two separate conditional calls inside the block); single result passed as imv_anchor dict. - _fetch_dkp_corridor and _fetch_anchor_comps hoisted to estimate_quality with identical guards. All 2443 existing tests pass UNMODIFIED. 9 new hermetic unit tests added that call _price_from_inputs directly with stub callables.
401 lines
14 KiB
Python
401 lines
14 KiB
Python
"""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
|