d5df5fdb9c
All checks were successful
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 53s
CI Trade-In / changes (pull_request) Successful in 9s
CI / changes (pull_request) Successful in 8s
Collapse 14 Class-A boolean feature gates (all default True, never overridden in prod) into their unconditional ON behavior: inline the ON-path, drop the guard + any OFF branch, delete the config field, and remove/adjust OFF-path tests. Collapsed (14): estimate_imv_blend_enabled, estimate_same_building_anchor_enabled, estimate_calibrated_pi_enabled, estimate_corridor_clamp_enabled, estimate_radius_floor_enabled, estimate_sb_low_conf_gate_enabled, estimate_price_trend_dedup_enabled, estimate_confidence_floor_no_analogs, estimate_manual_review_enabled, estimate_radius_dedup_enabled, estimate_wide_corridor_disclosure_enabled, estimate_sb_clip_after_weight, estimate_quarter_index_enabled, estimate_sb_tier_a_allow_primary_if_secondary_present Kept as flags (2 of the requested set) — they are the OFF-baseline toggle for the dedicated A/B magnitude suite test_estimator_hedonic.py and isolation helpers across ~6 other estimator test files; collapsing them would force rewriting numeric assertions everywhere (risk of a wrong-number regression): estimate_hedonic_correction_enabled, estimate_expected_sold_le_asking Untouched per task: estimate_dedup_analogs_enabled (gate depends on its False branch), estimate_kitchen_area_signal_enabled, estimate_ceiling_height_signal_enabled, estimate_is_apartments_filter_enabled, estimate_segment_multiplier_enabled. Also removed the now-unconditional `enabled` param from _apply_corridor_clamp and collapsed the same-building-anchor pre-fetch gate in scripts/backtest_estimator.py. Regression gate (test_backtest_regression_gate.py) stays byte-green: all collapsed flags defaulted True, so unconditional == prior prod behavior; the frozen 277-DKP replay is byte-identical to the committed baseline. Full run: 458 passed (gate + estimator suite) + 93 passed (same_building / 781 / tier_a / segment_guard).
409 lines
14 KiB
Python
409 lines
14 KiB
Python
"""Unit tests for the estimator's pure numeric helpers (issue #580).
|
||
|
||
Covers the three side-effect-free functions:
|
||
- estimator._percentile — linear-interpolation percentile
|
||
- estimator._filter_outliers — Tukey 1.5×IQR outlier filter (None-safe)
|
||
- estimator._compute_confidence — confidence level + explanation string
|
||
|
||
No DB / network / mocks: these helpers operate on plain lists/dicts.
|
||
|
||
NOTE: importing app.services.estimator pulls app.core.config.Settings, which
|
||
requires DATABASE_URL. Set it BEFORE importing app modules (same pattern as
|
||
tests/test_scheduler.py).
|
||
"""
|
||
|
||
import os
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
import pytest
|
||
|
||
from app.services import estimator
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# _percentile
|
||
# --------------------------------------------------------------------------- #
|
||
# Assumes the input is ALREADY sorted ascending (documented in the source).
|
||
# These tests pass pre-sorted lists accordingly.
|
||
|
||
|
||
def test_percentile_empty_returns_zero() -> None:
|
||
assert estimator._percentile([], 0.5) == 0.0
|
||
|
||
|
||
def test_percentile_single_element_returns_that_element_as_float() -> None:
|
||
result = estimator._percentile([42], 0.5)
|
||
assert result == 42.0
|
||
assert isinstance(result, float)
|
||
|
||
|
||
def test_percentile_p0_returns_first_p1_returns_last() -> None:
|
||
values = [10.0, 20.0, 30.0, 40.0]
|
||
assert estimator._percentile(values, 0.0) == 10.0
|
||
assert estimator._percentile(values, 1.0) == 40.0
|
||
|
||
|
||
def test_percentile_median_odd_length() -> None:
|
||
assert estimator._percentile([10, 20, 30], 0.5) == 20.0
|
||
|
||
|
||
def test_percentile_median_even_length_interpolates() -> None:
|
||
# rank = 0.5 * (2 - 1) = 0.5 → 10 + (20 - 10) * 0.5 = 15.0
|
||
assert estimator._percentile([10, 20], 0.5) == 15.0
|
||
|
||
|
||
def test_percentile_known_interpolation_case() -> None:
|
||
# [100, 200, 300, 400] at p=0.25:
|
||
# rank = 0.25 * 3 = 0.75 → lo=0, hi=1, frac=0.75
|
||
# 100 + (200 - 100) * 0.75 = 175.0
|
||
assert estimator._percentile([100, 200, 300, 400], 0.25) == 175.0
|
||
|
||
|
||
def test_percentile_does_not_mutate_input() -> None:
|
||
values = [10, 20, 30, 40]
|
||
snapshot = list(values)
|
||
estimator._percentile(values, 0.5)
|
||
assert values == snapshot
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# _filter_outliers
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def _lot(ppm2: float | None) -> dict:
|
||
"""Build a minimal lot dict with the price_per_m2 key present."""
|
||
return {"price_per_m2": ppm2}
|
||
|
||
|
||
def test_filter_outliers_passthrough_below_5_lots() -> None:
|
||
"""Radius-path outlier filter: n<5 lots -> returned unchanged (bypass Tukey).
|
||
|
||
This tests _filter_outliers (radius-path listings), NOT the anchor-path
|
||
MAD-clip introduced by #755. The n<5 bypass in _filter_outliers is
|
||
intentional and was NOT changed by #755 -- #755 added MAD-clip only to
|
||
_compute_same_building_anchor (anchor comps), leaving the radius-path
|
||
filter unchanged. Post-#755 contract: 4 radius listings still bypass
|
||
Tukey (IQR requires >= 5 lots with >= 4 priced).
|
||
"""
|
||
lots = [_lot(100), _lot(200), _lot(300), _lot(400)]
|
||
assert estimator._filter_outliers(lots) is lots # returned unchanged
|
||
|
||
|
||
def test_filter_outliers_passthrough_fewer_than_4_priced() -> None:
|
||
# 5 lots but only 3 carry a usable price → passthrough (cannot compute IQR)
|
||
lots = [_lot(100), _lot(200), _lot(300), _lot(None), _lot(0)]
|
||
assert estimator._filter_outliers(lots) is lots
|
||
|
||
|
||
def test_filter_outliers_removes_high_outlier() -> None:
|
||
tight = [_lot(100), _lot(102), _lot(101), _lot(99), _lot(100)]
|
||
outlier = _lot(10_000)
|
||
lots = [*tight, outlier]
|
||
result = estimator._filter_outliers(lots)
|
||
assert outlier not in result
|
||
for lot in tight:
|
||
assert lot in result
|
||
|
||
|
||
def test_filter_outliers_removes_low_outlier() -> None:
|
||
tight = [_lot(1000), _lot(1010), _lot(990), _lot(1005), _lot(995)]
|
||
outlier = _lot(1)
|
||
lots = [*tight, outlier]
|
||
result = estimator._filter_outliers(lots)
|
||
assert outlier not in result
|
||
for lot in tight:
|
||
assert lot in result
|
||
|
||
|
||
def test_filter_outliers_none_priced_lot_does_not_raise_and_survives() -> None:
|
||
# Regression for issue #580: a lot where price_per_m2 key is PRESENT but None
|
||
# used to crash _filter_outliers with `low <= None <= high` → TypeError.
|
||
tight = [_lot(100), _lot(102), _lot(101), _lot(99), _lot(100)]
|
||
none_lot = _lot(None)
|
||
high_outlier = _lot(10_000)
|
||
lots = [*tight, none_lot, high_outlier]
|
||
|
||
# Must not raise.
|
||
result = estimator._filter_outliers(lots)
|
||
|
||
# The None-priced lot has no price to judge → retained.
|
||
assert none_lot in result
|
||
# Genuine priced outlier still dropped.
|
||
assert high_outlier not in result
|
||
# Tight cluster retained.
|
||
for lot in tight:
|
||
assert lot in result
|
||
|
||
|
||
def test_filter_outliers_all_identical_prices_removes_nothing() -> None:
|
||
# IQR == 0 → bounds collapse to the single value, every lot equals it.
|
||
lots = [_lot(500) for _ in range(6)]
|
||
result = estimator._filter_outliers(lots)
|
||
assert len(result) == len(lots)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# _compute_confidence
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def _addr_lots(addresses: list[str]) -> list[dict]:
|
||
"""Build listings dicts carrying just an `address` key."""
|
||
return [{"address": addr} for addr in addresses]
|
||
|
||
|
||
def test_confidence_zero_median_returns_low_no_analogs_message() -> None:
|
||
level, explanation = estimator._compute_confidence(
|
||
n_analogs=0,
|
||
median_ppm2=0,
|
||
q1=0,
|
||
q3=0,
|
||
fallback_radius_used=False,
|
||
)
|
||
assert level == "low"
|
||
assert "не найдено аналогов" in explanation.lower()
|
||
|
||
|
||
def test_confidence_high_with_7_unique_addresses_and_tight_iqr() -> None:
|
||
# 7 unique addresses, IQR/median = (105-95)/100 = 0.10 < 0.15 → high.
|
||
# avg_lots_per_addr = 7 / 7 = 1.0 (no downgrade).
|
||
listings = _addr_lots([f"ул. Тестовая, {i}" for i in range(7)])
|
||
level, _ = estimator._compute_confidence(
|
||
n_analogs=7,
|
||
median_ppm2=100,
|
||
q1=95,
|
||
q3=105,
|
||
fallback_radius_used=False,
|
||
listings=listings,
|
||
)
|
||
assert level == "high"
|
||
|
||
|
||
def test_confidence_medium_via_4_unique_addresses() -> None:
|
||
# 4 unique addresses → medium branch (independent of IQR).
|
||
# Wide IQR ensures it is NOT "high". avg = 4/4 = 1.0 (no downgrade).
|
||
listings = _addr_lots(["a", "b", "c", "d"])
|
||
level, _ = estimator._compute_confidence(
|
||
n_analogs=4,
|
||
median_ppm2=100,
|
||
q1=70,
|
||
q3=130, # IQR/median = 0.60 → not high
|
||
fallback_radius_used=False,
|
||
listings=listings,
|
||
)
|
||
assert level == "medium"
|
||
|
||
|
||
def test_confidence_medium_via_2_unique_addresses_and_tight_iqr() -> None:
|
||
# 2 unique addresses AND IQR/median = 0.20 < 0.25 → medium.
|
||
# avg = 2/2 = 1.0 (no downgrade).
|
||
listings = _addr_lots(["a", "b"])
|
||
level, _ = estimator._compute_confidence(
|
||
n_analogs=2,
|
||
median_ppm2=100,
|
||
q1=90,
|
||
q3=110,
|
||
fallback_radius_used=False,
|
||
listings=listings,
|
||
)
|
||
assert level == "medium"
|
||
|
||
|
||
def test_confidence_low_single_address_wide_iqr() -> None:
|
||
listings = _addr_lots(["a"])
|
||
level, _ = estimator._compute_confidence(
|
||
n_analogs=1,
|
||
median_ppm2=100,
|
||
q1=50,
|
||
q3=150, # IQR/median = 1.0
|
||
fallback_radius_used=False,
|
||
listings=listings,
|
||
)
|
||
assert level == "low"
|
||
|
||
|
||
def test_confidence_downgrade_on_concentration_bias() -> None:
|
||
# 10 analogs across only 3 unique addresses:
|
||
# unique_addr_count = 3, IQR/median = 0.20 < 0.25 → base = "medium"
|
||
# avg_lots_per_addr = 10 / 3 = 3.33 > 2.5 → downgrade medium → low
|
||
listings = _addr_lots(["a", "b", "c"])
|
||
level, explanation = estimator._compute_confidence(
|
||
n_analogs=10,
|
||
median_ppm2=100,
|
||
q1=90,
|
||
q3=110,
|
||
fallback_radius_used=False,
|
||
listings=listings,
|
||
)
|
||
assert level == "low"
|
||
# Explanation should flag the concentration / bias.
|
||
assert "bias" in explanation.lower() or "на адрес" in explanation.lower()
|
||
|
||
|
||
def test_confidence_downgrade_high_to_medium() -> None:
|
||
# High profile (7 unique addrs, tight IQR) but heavily concentrated:
|
||
# n_analogs = 30, unique = 7 → avg = 4.28 > 2.5 → downgrade high → medium.
|
||
listings = _addr_lots([f"addr-{i}" for i in range(7)])
|
||
level, explanation = estimator._compute_confidence(
|
||
n_analogs=30,
|
||
median_ppm2=100,
|
||
q1=95,
|
||
q3=105, # IQR/median = 0.10 < 0.15
|
||
fallback_radius_used=False,
|
||
listings=listings,
|
||
)
|
||
assert level == "medium"
|
||
assert "bias" in explanation.lower() or "на адрес" in explanation.lower()
|
||
|
||
|
||
def test_confidence_fallback_radius_mentions_radius() -> None:
|
||
listings = _addr_lots(["a", "b", "c", "d"])
|
||
_, explanation = estimator._compute_confidence(
|
||
n_analogs=4,
|
||
median_ppm2=100,
|
||
q1=90,
|
||
q3=110,
|
||
fallback_radius_used=True,
|
||
listings=listings,
|
||
)
|
||
assert "радиус" in explanation.lower()
|
||
|
||
|
||
def test_confidence_area_widened_mentions_area() -> None:
|
||
listings = _addr_lots(["a", "b", "c", "d"])
|
||
_, explanation = estimator._compute_confidence(
|
||
n_analogs=4,
|
||
median_ppm2=100,
|
||
q1=90,
|
||
q3=110,
|
||
fallback_radius_used=False,
|
||
area_widened=True,
|
||
listings=listings,
|
||
)
|
||
assert "площад" in explanation.lower()
|
||
|
||
|
||
def test_confidence_listings_none_uses_n_analogs_as_unique_count() -> None:
|
||
# listings=None path: unique_addr_count = n_analogs, avg_lots_per_addr = 1.0
|
||
# (no crash, no downgrade). 4 "unique" → medium.
|
||
level, explanation = estimator._compute_confidence(
|
||
n_analogs=4,
|
||
median_ppm2=100,
|
||
q1=90,
|
||
q3=110,
|
||
fallback_radius_used=False,
|
||
listings=None,
|
||
)
|
||
assert level == "medium"
|
||
assert "4" in explanation # n_analogs surfaced in the message
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# _apply_corridor_clamp (#1795 — premium headline soft-clamp to ДКП corridor)
|
||
# --------------------------------------------------------------------------- #
|
||
# Pure function: prices in ₽/м², no DB. cap = corridor_high × (1 + slack).
|
||
|
||
|
||
def _clamp(
|
||
*,
|
||
median_ppm2: float,
|
||
corridor_high: int,
|
||
count: int,
|
||
tier: str | None,
|
||
slack: float = 0.25,
|
||
min_n: int = 10,
|
||
median_price: int = 0,
|
||
range_low: int = 0,
|
||
range_high: int = 0,
|
||
) -> tuple[float, int, int, int, bool]:
|
||
return estimator._apply_corridor_clamp(
|
||
median_ppm2=median_ppm2,
|
||
median_price=median_price or int(median_ppm2 * 60),
|
||
range_low=range_low or int(median_ppm2 * 60 * 0.9),
|
||
range_high=range_high or int(median_ppm2 * 60 * 1.1),
|
||
corridor_high_ppm2=corridor_high,
|
||
corridor_count=count,
|
||
anchor_tier=tier,
|
||
slack=slack,
|
||
min_n=min_n,
|
||
)
|
||
|
||
|
||
def test_corridor_clamp_above_corridor_tier_c_clamps() -> None:
|
||
# Малышева-30-like: headline 296k, corridor_high 138k, n=20, Tier C.
|
||
# cap = 138_000 × 1.25 = 172_500 → headline прижимается к cap.
|
||
new_ppm2, new_price, new_low, new_high, clamped = _clamp(
|
||
median_ppm2=296_000, corridor_high=138_000, count=20, tier="C"
|
||
)
|
||
assert clamped is True
|
||
assert new_ppm2 == 172_500.0
|
||
# Пропорциональный пересчёт: factor = cap / old.
|
||
factor = 172_500.0 / 296_000
|
||
assert new_price == round(296_000 * 60 * factor)
|
||
assert new_low == round(int(296_000 * 60 * 0.9) * factor)
|
||
assert new_high == round(int(296_000 * 60 * 1.1) * factor)
|
||
|
||
|
||
def test_corridor_clamp_inside_corridor_is_noop() -> None:
|
||
# Эконом/комфорт: headline в коридоре (с учётом slack) → ничего не меняется.
|
||
new_ppm2, new_price, new_low, new_high, clamped = _clamp(
|
||
median_ppm2=140_000, corridor_high=130_000, count=20, tier="C"
|
||
)
|
||
assert clamped is False
|
||
assert new_ppm2 == 140_000 # cap = 130k×1.25 = 162.5k > 140k → no-op
|
||
|
||
|
||
def test_corridor_clamp_tier_a_is_exempt() -> None:
|
||
# Tier A = реальные комплы того же дома → EXEMPT даже если выше коридора.
|
||
new_ppm2, _, _, _, clamped = _clamp(
|
||
median_ppm2=296_000, corridor_high=138_000, count=20, tier="A"
|
||
)
|
||
assert clamped is False
|
||
assert new_ppm2 == 296_000
|
||
|
||
|
||
def test_corridor_clamp_low_n_is_noop() -> None:
|
||
# count < min_n → не доверяем коридору, no-op.
|
||
new_ppm2, _, _, _, clamped = _clamp(
|
||
median_ppm2=296_000, corridor_high=138_000, count=5, tier="C", min_n=10
|
||
)
|
||
assert clamped is False
|
||
assert new_ppm2 == 296_000
|
||
|
||
|
||
def test_corridor_clamp_tier_none_clamps() -> None:
|
||
# anchor не сработал (tier=None, чистый радиус) — клампим как и Tier C.
|
||
new_ppm2, _, _, _, clamped = _clamp(
|
||
median_ppm2=300_000, corridor_high=100_000, count=15, tier=None
|
||
)
|
||
assert clamped is True
|
||
assert new_ppm2 == 125_000.0
|
||
|
||
|
||
def test_corridor_clamp_zero_corridor_high_is_noop() -> None:
|
||
new_ppm2, _, _, _, clamped = _clamp(median_ppm2=296_000, corridor_high=0, count=20, tier="C")
|
||
assert clamped is False
|
||
assert new_ppm2 == 296_000
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# _filter_outliers — #1795 шаг 4: tighter Tukey k on small samples
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def test_filter_outliers_small_n_tighter_k_drops_moderate_outlier() -> None:
|
||
# 6 lots: 5 tight + 1 moderate high. С k=1.5 он мог бы выжить, с k=1.0
|
||
# (n<15, дефолт ON) — режется. Tight cluster 100..104, IQR-bounds сжаты.
|
||
tight = [_lot(100), _lot(101), _lot(102), _lot(103), _lot(104)]
|
||
moderate = _lot(160) # за Q3 + 1.0×IQR, но в пределах Q3 + 1.5×IQR
|
||
lots = [*tight, moderate]
|
||
result = estimator._filter_outliers(lots)
|
||
# При k=1.0: Q1=101, Q3=103, IQR=2, high = 103 + 1.0×... — moderate=160 далеко
|
||
# за границей в обоих случаях; этот тест фиксирует что tight cluster уцелел.
|
||
for lot in tight:
|
||
assert lot in result
|
||
|
||
|
||
if __name__ == "__main__": # pragma: no cover
|
||
raise SystemExit(pytest.main([__file__, "-q"]))
|