gendesign/tradein-mvp/backend/tests/test_estimator_source_metrics_2043.py

"""Unit tests for #2043 (BE-1) sample-confidence metrics.

Covers, without DB/network:
  - estimator._cv_from_ppm2      — coefficient of variation ₽/м² (std/mean)
  - estimator._source_counts     — per-source analog counts
  - estimator._price_from_inputs — surfaces cv on radius- and anchor-paths

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_estimator_pure_units.py).
"""

import os

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import math

import pytest

from app.services import estimator
from app.services.geocoder import GeocodeResult

# --------------------------------------------------------------------------- #
# _cv_from_ppm2
# --------------------------------------------------------------------------- #


def test_cv_from_ppm2_fewer_than_two_returns_none() -> None:
    assert estimator._cv_from_ppm2([]) is None
    assert estimator._cv_from_ppm2([100_000]) is None
    # A single usable value (the None/0 are dropped) → still < 2 → None.
    assert estimator._cv_from_ppm2([100_000, None, 0]) is None


def test_cv_from_ppm2_uniform_is_zero() -> None:
    # Zero variance → cv == 0.0 (non-None, valid "tight" sample).
    result = estimator._cv_from_ppm2([100_000, 100_000, 100_000])
    assert result == 0.0


def test_cv_from_ppm2_known_value() -> None:
    # values [90k, 100k, 110k]: mean=100k, population var = (100M+0+100M)/3
    # std = sqrt(200_000_000/3) ≈ 8164.9658 → cv ≈ 0.0816497
    result = estimator._cv_from_ppm2([90_000, 100_000, 110_000])
    assert result is not None
    assert math.isclose(result, 8164.9658 / 100_000, rel_tol=1e-4)


def test_cv_from_ppm2_drops_none_and_nonpositive() -> None:
    # None and 0 are skipped; result computed on [100k, 120k] only.
    with_noise = estimator._cv_from_ppm2([100_000, None, 0, 120_000])
    clean = estimator._cv_from_ppm2([100_000, 120_000])
    assert with_noise == clean
    assert clean is not None and clean > 0


# --------------------------------------------------------------------------- #
# _source_counts
# --------------------------------------------------------------------------- #


def test_source_counts_basic() -> None:
    counts = estimator._source_counts(["avito", "cian", "avito", "avito", "cian"])
    assert counts == {"avito": 3, "cian": 2}


def test_source_counts_skips_none_and_empty() -> None:
    counts = estimator._source_counts(["avito", None, "", "cian", None])
    assert counts == {"avito": 1, "cian": 1}


def test_source_counts_empty_input_is_empty_dict() -> None:
    assert estimator._source_counts([]) == {}
    assert estimator._source_counts([None, None]) == {}


def test_source_counts_keys_sorted() -> None:
    # Deterministic ordering for stable JSON output.
    counts = estimator._source_counts(["yandex", "avito", "cian"])
    assert list(counts.keys()) == ["avito", "cian", "yandex"]


# --------------------------------------------------------------------------- #
# _canonical_sources — #2087 (M1) single source of truth for sources_used
# --------------------------------------------------------------------------- #


def test_canonical_sources_listing_plus_valuation() -> None:
    # Listing sources come from analogs; valuation flags are unioned on top.
    result = estimator._canonical_sources(
        ["avito", "cian", "avito"], {"avito_imv", "yandex_valuation"}
    )
    assert result == ["avito", "avito_imv", "cian", "yandex_valuation"]


def test_canonical_sources_deterministic_same_inputs_same_output() -> None:
    # Same set of analogs + flags → identical, sorted sources_used (POST == GET).
    analogs = ["yandex", "avito", "domklik", "avito"]
    flags = {"cian_valuation", "avito_imv"}
    a = estimator._canonical_sources(analogs, flags)
    b = estimator._canonical_sources(list(reversed(analogs)), set(flags))
    assert a == b
    assert a == ["avito", "avito_imv", "cian_valuation", "domklik", "yandex"]


def test_canonical_sources_skips_none_and_empty_analogs() -> None:
    result = estimator._canonical_sources(["avito", None, "", "cian"], set())
    assert result == ["avito", "cian"]


def test_canonical_sources_filters_non_canonical_valuation_flags() -> None:
    # Passing a raw persisted column (listing sorts + quarter_index) as the
    # valuation arg must NOT leak listing/service labels — only the 3 canonical
    # valuation sources survive. Listing part comes exclusively from analogs.
    raw_column = ["avito", "domklik", "quarter_index", "avito_imv", "yandex"]
    result = estimator._canonical_sources(["cian"], raw_column)
    # From analogs: cian. From column: only avito_imv is a canonical valuation.
    assert result == ["avito_imv", "cian"]
    assert "quarter_index" not in result
    assert "domklik" not in result  # listing label in the column is ignored


def test_canonical_source_counts_keys_subset_of_sources_used() -> None:
    # Invariant #2087: every key of source_counts (derived from the same analog
    # sources) is present in sources_used.
    analog_sources = ["avito", "cian", "avito", "yandex", None, ""]
    sources_used = estimator._canonical_sources(analog_sources, {"avito_imv"})
    source_counts = estimator._source_counts(analog_sources)
    assert set(source_counts.keys()) <= set(sources_used)
    # And every listing source present in analogs is reflected in sources_used.
    for s in analog_sources:
        if s:
            assert s in sources_used


def test_canonical_sources_empty_is_empty_list() -> None:
    assert estimator._canonical_sources([], []) == []
    assert estimator._canonical_sources([None, ""], ["quarter_index"]) == []


# --------------------------------------------------------------------------- #
# _price_from_inputs — cv surfaced on both paths
# --------------------------------------------------------------------------- #


def _geo() -> GeocodeResult:
    return GeocodeResult(
        lat=56.838,
        lon=60.597,
        full_address="ул. Тестовая, 1",
        provider="nominatim",
        confidence="approximate",
    )


def _lot(ppm2: float, addr: str = "ул. Тестовая, 1", source: str = "avito") -> dict:
    return {"price_per_m2": ppm2, "address": addr, "source": source}


def _anchor_comp(ppm2: float, area: float = 50.0, rooms: int = 2) -> dict:
    return {"price_per_m2": ppm2, "area_m2": area, "rooms": rooms}


def _call(
    *,
    listings: list[dict],
    anchor_comps: list[dict] | None = None,
    anchor_tier_fetched: str | None = None,
) -> estimator.PricingResult:
    return estimator._price_from_inputs(
        listings=listings,
        area_m2=50.0,
        rooms=2,
        repair_state=None,
        floor=5,
        total_floors=10,
        target_year=None,
        analog_tier="W",
        fallback_used=False,
        area_widened=False,
        anchor_comps=anchor_comps or [],
        anchor_tier_fetched=anchor_tier_fetched,
        dkp_raw=None,
        imv_anchor=None,
        imv_eval=None,
        yandex_val_present=False,
        cian_val_present=False,
        ratio_resolver=lambda _appm2: (None, None),
        quarter_index_lookup=lambda _q: None,
        quarter_indexes_lookup=lambda _qs: {},
        target_house_cadnum=None,
        dadata_coarse=False,
        geo=_geo(),
        dadata_qc_geo=None,
    )


def test_radius_path_surfaces_nonempty_cv() -> None:
    # 7 tight-but-varied radius lots → cv is a small positive float (non-None).
    ppm2s = [95_000, 98_000, 100_000, 102_000, 105_000, 103_000, 97_000]
    listings = [_lot(p, addr=f"ул. Тестовая, {i}") for i, p in enumerate(ppm2s)]
    pr = _call(listings=listings)

    assert pr.anchor_tier is None  # radius path
    assert pr.cv is not None
    assert pr.cv > 0
    # Matches the direct CV of the surviving (outlier-filtered) ppm² sample.
    survivors = [lot["price_per_m2"] for lot in pr.listings_clean]
    assert math.isclose(pr.cv, estimator._cv_from_ppm2(survivors), rel_tol=1e-9)


def test_empty_listings_cv_is_none() -> None:
    pr = _call(listings=[])
    assert pr.n_analogs == 0
    assert pr.cv is None


def test_anchor_path_surfaces_anchor_cv() -> None:
    # Anchor fires (Tier A) → cv reflects the anchor comps, not the radius lots.
    comps = [_anchor_comp(p) for p in (190_000, 200_000, 210_000, 205_000, 195_000)]
    pr = _call(
        listings=[_lot(100_000, addr=f"ул. Тестовая, {i}") for i in range(5)],
        anchor_comps=comps,
        anchor_tier_fetched="A",
    )
    assert pr.anchor_tier == "A"
    assert pr.cv is not None
    assert pr.cv > 0
    # cv computed on the anchor comps ppm², independent of the 100k radius lots.
    expected = estimator._cv_from_ppm2([c["price_per_m2"] for c in comps])
    assert math.isclose(pr.cv, expected, rel_tol=1e-9)


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(pytest.main([__file__, "-q"]))