bot-backend
7d9ecb117f
Held-out fit (n=2366 prod deals, 2026-06-27) of log(actual_sold/expected_sold) ~ year + ln(area) gives a multiplicative factor that corrects systematic mis-estimation by building era + unit size (underestimates newer buildings, mis-handles large units). factor = exp(b0 + b_year*(year-2000)/20 + b_larea*ln(area)), clamped [0.75, 1.30]; applied to the expected_sold point BEFORE the calibrated PI range block (so the range follows the corrected point). Headline/asking is untouched. Gated behind estimate_hedonic_correction_enabled (OFF => exact old behavior, proven by the regression gate). Frozen baseline regenerated: overall expected_sold MAPE 18.63->14.24; per-segment median_bias_pct moves toward 0 (бизнес -21.5->-9.2, комфорт -8.0->-3.6, эконом 17.2->4.3); элит mildly better, no harm. Tests that assert the exact ratio relation (expected_sold == headline x ratio) hold the orthogonal hedonic layer OFF.
202 lines
7.9 KiB
Python
202 lines
7.9 KiB
Python
"""Tests for Fix 4: clamp expected_sold <= asking (ratio cap 1.0).
|
||
|
||
Smoke-диагноз: 3к/27.0М -> expected_sold 31.4М > asking.
|
||
Причина: _get_asking_sold_ratio для high-price tier возвращает ratio > 1.0
|
||
(product artefact) -> expected_sold = headline x ratio > headline.
|
||
|
||
Проверяем:
|
||
1. ratio=1.15 + флаг on -> expected_sold == headline (не выше); лог clamp.
|
||
2. ratio=0.8 -> expected_sold = headline x 0.8 (норм, без clamp).
|
||
3. ratio None -> expected_sold не выводится (graceful).
|
||
4. Флаг off + ratio > 1 -> expected_sold > headline (старое поведение).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import os
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import anyio
|
||
import pytest
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
|
||
def _make_listing(*, price_per_m2: float, area_m2: float = 50.0) -> dict[str, Any]:
|
||
return {
|
||
"source": "cian",
|
||
"source_url": "https://cian.ru/offer/1",
|
||
"address": "ЕКБ, ул. Малышева, 30",
|
||
"lat": 56.838,
|
||
"lon": 60.595,
|
||
"rooms": 3,
|
||
"area_m2": area_m2,
|
||
"floor": 10,
|
||
"total_floors": 20,
|
||
"price_rub": price_per_m2 * area_m2,
|
||
"price_per_m2": price_per_m2,
|
||
"listing_date": datetime(2026, 5, 1),
|
||
"days_on_market": 5,
|
||
"photo_urls": [],
|
||
"scraped_at": datetime(2026, 5, 20, tzinfo=UTC),
|
||
"distance_m": 100.0,
|
||
"relevance_score": 0.1,
|
||
}
|
||
|
||
|
||
_ANALOGS: list[dict[str, Any]] = [
|
||
_make_listing(price_per_m2=260_000.0),
|
||
_make_listing(price_per_m2=270_000.0),
|
||
_make_listing(price_per_m2=280_000.0),
|
||
]
|
||
|
||
|
||
def _make_geo():
|
||
from app.services.geocoder import GeocodeResult
|
||
|
||
return GeocodeResult(
|
||
lat=56.838,
|
||
lon=60.595,
|
||
full_address="Свердловская обл., Екатеринбург, ул. Малышева, 30",
|
||
provider="nominatim",
|
||
)
|
||
|
||
|
||
def _make_payload(rooms: int = 3):
|
||
from app.schemas.trade_in import TradeInEstimateInput
|
||
|
||
return TradeInEstimateInput(
|
||
address="ЕКБ, ул. Малышева, 30", area_m2=50.0, rooms=rooms, floor=10, total_floors=20
|
||
)
|
||
|
||
|
||
def _run_estimate(
|
||
ratio_tuple: tuple[float | None, str | None],
|
||
*,
|
||
clamp_enabled: bool = True,
|
||
) -> Any:
|
||
from app.services.estimator import estimate_quality
|
||
|
||
db = MagicMock()
|
||
payload = _make_payload()
|
||
|
||
async def _run() -> Any:
|
||
with (
|
||
patch("app.core.config.settings.estimate_expected_sold_le_asking", new=clamp_enabled),
|
||
# #2002: these tests assert the clamp/ratio math exactly. Hold the
|
||
# orthogonal hedonic year+area correction OFF (OFF ⇒ legacy expected_sold).
|
||
patch("app.core.config.settings.estimate_hedonic_correction_enabled", new=False),
|
||
patch("app.services.estimator.geocode", new=AsyncMock(return_value=_make_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),
|
||
),
|
||
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),
|
||
),
|
||
patch("app.services.estimator._fetch_dkp_corridor", return_value=None),
|
||
patch("app.services.estimator._get_asking_sold_ratio", return_value=ratio_tuple),
|
||
):
|
||
return await estimate_quality(payload, db)
|
||
|
||
return anyio.run(_run)
|
||
|
||
|
||
def test_expected_sold_clamped_to_headline_when_ratio_above_1(
|
||
caplog: pytest.LogCaptureFixture,
|
||
) -> None:
|
||
ratio = 1.15
|
||
with caplog.at_level(logging.INFO, logger="app.services.estimator"):
|
||
est = _run_estimate((ratio, "per_rooms"), clamp_enabled=True)
|
||
assert est.median_price_rub > 0, "headline должен быть задан аналогами"
|
||
assert est.expected_sold_price_rub is not None
|
||
assert (
|
||
est.expected_sold_price_rub <= est.median_price_rub
|
||
), f"expected_sold {est.expected_sold_price_rub} > asking {est.median_price_rub}"
|
||
assert est.expected_sold_per_m2 is not None
|
||
assert (
|
||
est.expected_sold_per_m2 <= est.median_price_per_m2
|
||
), "expected_sold_per_m2 превышает median_price_per_m2"
|
||
# #1966: expected_sold range is now a calibrated ~80% PI around the point
|
||
# (point × [p10, p90] of sold/expected_sold), so it is NO LONGER bounded by the
|
||
# asking-IQR band — the high arm (point × 1.392) legitimately exceeds range_high.
|
||
# The invariant we keep is the calibrated band itself + ordering low ≤ point ≤ high.
|
||
from app.core.config import settings
|
||
|
||
assert est.expected_sold_range_high_rub is not None
|
||
assert est.expected_sold_range_high_rub == round(
|
||
est.expected_sold_price_rub * settings.estimate_pi_high_mult
|
||
)
|
||
assert est.expected_sold_range_low_rub is not None
|
||
assert est.expected_sold_range_low_rub == round(
|
||
est.expected_sold_price_rub * settings.estimate_pi_low_mult
|
||
)
|
||
assert (
|
||
est.expected_sold_range_low_rub
|
||
<= est.expected_sold_price_rub
|
||
<= est.expected_sold_range_high_rub
|
||
)
|
||
assert est.expected_sold_price_rub == est.median_price_rub
|
||
assert est.expected_sold_per_m2 == est.median_price_per_m2
|
||
clamp_logs = [r for r in caplog.records if "clamped" in r.getMessage()]
|
||
assert clamp_logs, "Ожидался log-message о clamp ratio"
|
||
assert "1.150" in clamp_logs[0].getMessage() or "1.15" in clamp_logs[0].getMessage()
|
||
|
||
|
||
def test_expected_sold_not_clamped_when_ratio_below_1() -> None:
|
||
ratio = 0.8
|
||
est = _run_estimate((ratio, "per_rooms"), clamp_enabled=True)
|
||
assert est.median_price_rub > 0
|
||
assert est.expected_sold_price_rub == round(est.median_price_rub * ratio)
|
||
assert est.expected_sold_per_m2 == round(est.median_price_per_m2 * ratio)
|
||
# #1966: calibrated ~80% PI around the expected_sold point (not asking-IQR × ratio).
|
||
from app.core.config import settings
|
||
|
||
assert est.expected_sold_range_low_rub == round(
|
||
est.expected_sold_price_rub * settings.estimate_pi_low_mult
|
||
)
|
||
assert est.expected_sold_range_high_rub == round(
|
||
est.expected_sold_price_rub * settings.estimate_pi_high_mult
|
||
)
|
||
assert est.expected_sold_price_rub < est.median_price_rub
|
||
|
||
|
||
def test_expected_sold_none_when_ratio_none() -> None:
|
||
est = _run_estimate((None, None), clamp_enabled=True)
|
||
assert est.expected_sold_price_rub is None
|
||
assert est.expected_sold_per_m2 is None
|
||
assert est.expected_sold_range_low_rub is None
|
||
assert est.expected_sold_range_high_rub is None
|
||
assert est.asking_to_sold_ratio is None
|
||
assert est.median_price_rub > 0
|
||
|
||
|
||
def test_expected_sold_exceeds_asking_when_clamp_flag_off() -> None:
|
||
ratio = 1.15
|
||
est = _run_estimate((ratio, "per_rooms"), clamp_enabled=False)
|
||
assert est.median_price_rub > 0
|
||
assert est.expected_sold_price_rub is not None
|
||
assert est.expected_sold_price_rub > est.median_price_rub, (
|
||
f"Ожидалось expected_sold > asking при clamp=False, "
|
||
f"но {est.expected_sold_price_rub} <= {est.median_price_rub}"
|
||
)
|
||
assert est.expected_sold_price_rub == round(est.median_price_rub * ratio)
|