feat(tradein): калибровка repair-коэффициентов с base 1.0 (#7)

_REPAIR_COEF → tunable market-эвристика: needs_repair=0.94, standard=1.00
(baseline no-op; было 0.98 — срезало 2% с каждой standard-оценки), good=1.05,
excellent=1.10. Деривация из данных отклонена: listings.repair_state покрытие
~2%, сырые un-normalized значения, медианы confounded площадью (issue #7).

+ regression-тест (читает _REPAIR_COEF динамически — переживёт рекалибровку):
ratio excellent/needs_repair, standard/None no-op, наличие/отсутствие note.
Поправлен mislabeled comment над _IMV_HOUSE_TYPE_MAP. _IMV_REPAIR_MAP не тронут.

code-reviewer:  APPROVE no critical/minor. Tests: 5 new + 7 cian-integration.
Ruff clean.

Closes #7
This commit is contained in:
Light1YT 2026-05-28 16:15:54 +05:00
parent 1058dc2430
commit 33ee00c065
2 changed files with 199 additions and 7 deletions

View file

@ -107,9 +107,8 @@ def _target_cohort_range(year_built: int | None) -> tuple[int, int] | None:
return None
# Поправочные коэффициенты на состояние ремонта. Аналоги в выборке — микс
# состояний (≈ "стандартный/косметический"), коэффициент сдвигает медиану под
# конкретный ремонт целевой квартиры. Встреча Птицы: ремонт влияет на цену.
# Маппинг наших house_type → словарь Avito-IMV (внешний source). НЕ путать с
# _REPAIR_COEF (heuristic-множитель ниже).
_IMV_HOUSE_TYPE_MAP: dict[str | None, str | None] = {
"panel": "panel",
"brick": "brick",
@ -128,11 +127,22 @@ _IMV_REPAIR_MAP: dict[str | None, str | None] = {
None: None,
}
# Множители к медиане по состоянию ремонта. Аналоги в выборке — микс состояний;
# коэффициент сдвигает оценку под ремонт целевой квартиры (встреча Птицы: ремонт
# влияет на цену).
#
# WARNING: tunable МАРКЕТ-ЭВРИСТИКА, НЕ data-derived (issue #7). Вывести из данных пока
# нельзя: listings.repair_state покрыт только ~2%, хранит un-normalized исходные
# значения (cosmetic/euro/fine/without/designer/rough — НЕ целевой enum
# needs_repair/standard/good/excellent), а медианы по нему confounded by area
# (немонотонны). Baseline = standard = 1.00 (no-op: было 0.98, срезало каждую
# «стандартную» оценку на 2% — пофикшено). Пересмотреть, когда покрытие
# repair_state вырастет и появится нормализация.
_REPAIR_COEF: dict[str, float] = {
"needs_repair": 0.92, # требует ремонта — ниже рынка
"standard": 0.98,
"good": 1.03,
"excellent": 1.08, # евроремонт — выше рынка
"needs_repair": 0.94, # требует ремонта — ниже рынка
"standard": 1.00, # baseline
"good": 1.05,
"excellent": 1.10, # евроремонт — выше рынка
}
_REPAIR_LABEL: dict[str | None, str] = {
"needs_repair": "требует ремонта",

View file

@ -0,0 +1,182 @@
"""Regression tests for the repair-state coefficient (issue #7).
Verifies the `_REPAIR_COEF` heuristic multiplier in `estimate_quality`:
- excellent / needs_repair median ratio == coef ratio (within int rounding)
- standard (baseline 1.0) is a true no-op vs the unadjusted market median
- repair_state=None is a no-op (coef 1.0)
- coef != 1.0 emits the repair note into confidence_explanation
Coefficient values are read DYNAMICALLY from `_REPAIR_COEF` so the assertions
survive future re-calibration. The full `estimate_quality` flow is exercised with
all I/O stubbed (geocode / house-meta / _fetch_analogs / _fetch_deals / IMV /
Yandex / Cian / DB) so the test isolates the multiplier no DB, no network.
"""
from __future__ import annotations
import os
from datetime import UTC, datetime
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import anyio
# 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")
# ── Fixtures / helpers ──────────────────────────────────────────────────────
def _make_listing(*, price_per_m2: float, area_m2: float = 40.0) -> dict[str, Any]:
"""Minimal listing dict matching the keys the aggregation block reads."""
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(repair_state: str | None):
from app.schemas.trade_in import TradeInEstimateInput
return TradeInEstimateInput(
address="ЕКБ, ул. Учителей, 18",
area_m2=40.0,
rooms=1,
floor=4,
total_floors=16,
repair_state=repair_state,
)
def _run_estimate(repair_state: str | None):
"""Invoke estimate_quality with every external dependency stubbed."""
from app.services.estimator import estimate_quality
db = MagicMock()
payload = _make_payload(repair_state)
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)),
# 3-tuple: (listings, fallback_used, analog_tier). Same analogs every call so
# all fallback tiers are equivalent and the median is stable.
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)),
):
return await estimate_quality(payload, db)
return anyio.run(_run)
# ── Tests ───────────────────────────────────────────────────────────────────
def test_excellent_to_needs_repair_ratio_matches_coef() -> None:
"""excellent ÷ needs_repair median ≈ _REPAIR_COEF["excellent"] / [...]["needs_repair"]."""
from app.services.estimator import _REPAIR_COEF
excellent = _run_estimate("excellent")
needs_repair = _run_estimate("needs_repair")
expected_ratio = _REPAIR_COEF["excellent"] / _REPAIR_COEF["needs_repair"]
actual_ratio = excellent.median_price_rub / needs_repair.median_price_rub
# int() truncation on both medians ⇒ allow a small tolerance.
assert abs(actual_ratio - expected_ratio) < 0.005, (
f"excellent/needs_repair ratio {actual_ratio:.5f} != coef ratio {expected_ratio:.5f}"
)
def test_standard_is_baseline_noop() -> None:
"""repair_state='standard' → median equals the unadjusted market median (coef 1.0)."""
from app.services.estimator import _REPAIR_COEF
assert _REPAIR_COEF["standard"] == 1.0, "baseline invariant: standard must be 1.0"
standard = _run_estimate("standard")
baseline = _run_estimate(None)
# Unadjusted median: median ppm2 (150_000) × area (40.0) = 6_000_000.
expected_median = int(150_000.0 * 40.0)
assert standard.median_price_rub == expected_median
assert standard.median_price_rub == baseline.median_price_rub
def test_none_repair_state_is_noop() -> None:
"""repair_state=None → coef 1.0, no adjustment, no repair note."""
baseline = _run_estimate(None)
expected_median = int(150_000.0 * 40.0)
assert baseline.median_price_rub == expected_median
assert "скорректирована на состояние ремонта" not in (
baseline.confidence_explanation or ""
)
def test_repair_note_present_when_coef_differs() -> None:
"""coef != 1.0 (excellent) → confidence_explanation contains the repair note."""
from app.services.estimator import _REPAIR_COEF
assert _REPAIR_COEF["excellent"] != 1.0, "fixture invariant: excellent must adjust"
excellent = _run_estimate("excellent")
explanation = excellent.confidence_explanation or ""
assert "скорректирована на состояние ремонта" in explanation
# Note reports the signed percentage derived from the coefficient.
pct = round((_REPAIR_COEF["excellent"] - 1.0) * 100)
assert f"{pct:+d}%" in explanation
def test_standard_emits_no_repair_note() -> None:
"""Baseline standard (1.0) is a true no-op — no repair note appended."""
standard = _run_estimate("standard")
assert "скорректирована на состояние ремонта" not in (
standard.confidence_explanation or ""
)