diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 0bd631bf..309dc90b 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/tradein-mvp/backend/app/services/estimator.py @@ -1759,7 +1759,18 @@ def _filter_outliers(lots: list[dict[str, Any]]) -> list[dict[str, Any]]: low = q1 - 1.5 * iqr high = q3 + 1.5 * iqr - clean = [lot for lot in lots if low <= lot.get("price_per_m2", 0) <= high] + # None-safe: listings.price_per_m2 is nullable, и lot.get(..., 0) вернёт None + # (а не дефолт 0) когда ключ ПРИСУТСТВУЕТ со значением None → low <= None <= high + # бросает TypeError в Python 3. Лоты без цены судить как outlier нечем — оставляем их. + clean = [] + for lot in lots: + ppm2 = lot.get("price_per_m2") + if ppm2 is None: + clean.append(lot) # нечего сравнивать — keep + continue + if low <= ppm2 <= high: + clean.append(lot) # priced лот внутри Tukey-границ — keep + # else: priced outlier за пределами границ — drop if len(clean) < len(lots): logger.info("outlier filter: %d → %d (Q1=%d Q3=%d)", len(lots), len(clean), q1, q3) return clean diff --git a/tradein-mvp/backend/tests/test_estimator_pure_units.py b/tradein-mvp/backend/tests/test_estimator_pure_units.py new file mode 100644 index 00000000..3fbbce4e --- /dev/null +++ b/tradein-mvp/backend/tests/test_estimator_pure_units.py @@ -0,0 +1,293 @@ +"""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: + 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 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(pytest.main([__file__, "-q"]))