diff --git a/tradein-mvp/backend/app/api/v1/trade_in.py b/tradein-mvp/backend/app/api/v1/trade_in.py index 17067741..dff19ea0 100644 --- a/tradein-mvp/backend/app/api/v1/trade_in.py +++ b/tradein-mvp/backend/app/api/v1/trade_in.py @@ -235,15 +235,24 @@ def get_estimate( _assert_estimate_access(row.created_by, x_authenticated_user) from app.services.estimator import ( + _cv_from_ppm2, _fetch_dkp_corridor, _fetch_house_imv_anchor, _fetch_price_trend, _qc_geo_to_precision, + _source_counts, ) analogs = [AnalogLot(**a) for a in (row.analogs or [])] actual_deals = [AnalogLot(**a) for a in (row.actual_deals or [])] + # #2043 (BE-1): CV / счётчики источников на rehydrate — best-effort из + # сохранённых analogs (top-N, усечённо: полная выборка не персистится). На + # свежей оценке (POST) считаются по полной выборке; здесь — по тому, что есть + # в строке. created_at берём из колонки (persisted). + cv = _cv_from_ppm2([a.price_per_m2 for a in analogs]) + source_counts = _source_counts([a.source for a in analogs]) + # #696: POST-only производные поля (price_trend / avito_imv / dkp_corridor / # last_scraped_at) на GET-rehydrate ранее были null → на shared-link / PDF / # ?id= restore пропадали графики тренда, IMV-якорь, коридор ДКП и метка @@ -336,6 +345,10 @@ def get_estimate( house_fias_id=row.house_fias_id, address_precision=_qc_geo_to_precision(row.dadata_qc_geo), metro_nearest=(row.dadata_metro or []), + # #2043 (BE-1): достоверность выборки — CV, счётчики источников, дата создания. + cv=cv, + source_counts=source_counts, + created_at=row.created_at, ) @@ -609,6 +622,9 @@ def estimate_history( свои оценки (created_by = X-Authenticated-User); legacy NULL-строки без владельца не попадают. Admin видит все строки, либо фильтрует по ?account=. 401 если заголовок отсутствует (mirror /me — Caddy basic_auth обязателен). + + #2043 (BE-1): в проекцию добавлен sources_used (jsonb-массив источников) — + разлочивает колонку «ИСТОЧНИКОВ N/7» в CacheView (len(sources_used) на строку). """ if not x_authenticated_user: raise HTTPException( @@ -644,7 +660,8 @@ def estimate_history( text( f""" SELECT id, address, rooms, area_m2, median_price, - confidence, n_analogs, created_at + confidence, n_analogs, created_at, + COALESCE(sources_used, '[]'::jsonb) AS sources_used FROM trade_in_estimates {where} ORDER BY created_at DESC @@ -661,7 +678,18 @@ def estimate_history( @router.get("/cache-stats") def cache_stats(db: Annotated[Session, Depends(get_db)]) -> dict[str, object]: - """Состояние данных и кэшей (#399) — для страницы «Кэш».""" + """Состояние данных и кэшей (#399) — для страницы «Кэш». + + #2043 (BE-1) KPI-расширение для CacheView: + - avg_median_price — средняя headline-медиана по НЕпустым оценкам + (median_price > 0, чтобы insufficient-data нули не занижали среднее). + NULL если непустых оценок нет. + - repeat_address_pct — доля строк-оценок, чей address встречается ≥2 раз + (индикатор потенциала кэш-хитов «повторный адрес»). Считается по + trade_in_estimates с непустым address; NULL при отсутствии адресов. + NB: это честный best-effort по persisted оценкам, а не hit-rate реального + кэша (отдельного счётчика попаданий не ведём). + """ row = ( db.execute( text( @@ -675,7 +703,17 @@ def cache_stats(db: Annotated[Session, Depends(get_db)]) -> dict[str, object]: (SELECT count(*) FROM deals) AS deals, (SELECT count(*) FROM gendesign_cad_buildings) AS cad_buildings, (SELECT count(*) FROM house_metadata) AS house_metadata, - (SELECT count(*) FROM trade_in_estimates) AS estimates_total + (SELECT count(*) FROM trade_in_estimates) AS estimates_total, + (SELECT round(avg(median_price)) + FROM trade_in_estimates WHERE median_price > 0) AS avg_median_price, + (SELECT round( + 100.0 * count(*) FILTER (WHERE addr_count > 1) + / NULLIF(count(*), 0), 1) + FROM ( + SELECT count(*) OVER (PARTITION BY address) AS addr_count + FROM trade_in_estimates + WHERE address IS NOT NULL AND address <> '' + ) t) AS repeat_address_pct """ ) ) @@ -876,11 +914,16 @@ def get_estimate_house_analytics( estimate_id: UUID, db: Annotated[Session, Depends(get_db)], x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None, + radius_m: int | None = None, ) -> HouseAnalyticsResponse: """House-level analytics from house_placement_history backfill. Resolves target house(s) — если в самом доме <8 hist rows — расширяем поиск до 300м. Возвращает: price-history by year (median ₽/м²), recent sold (12mo), KPI. + + #2044 (BE-2): optional query-param radius_m (100–5000) явно задаёт радиус + расширения выборки домов. None → текущая авто-логика (расширяем до 300 м + только если в самом доме <8 записей). radius_m возвращается в ответе. """ _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user) target = db.execute( @@ -911,27 +954,42 @@ def get_estimate_house_analytics( ).all() house_ids = [r.id for r in rows] - # 2. Expand to 300m if too few hist rows + # 2. Expand search radius. #2044 (BE-2): explicit radius_m query-param + # overrides the auto 0→300 heuristic. None → byte-identical current + # behaviour (расширяем до 300 м только при <8 in-house rows). radius_used = 0 - n_in_house = 0 - if house_ids: - n_in_house = ( - db.execute( - text("SELECT COUNT(*) FROM house_placement_history WHERE house_id = ANY(:ids)"), - {"ids": house_ids}, - ).scalar() - or 0 - ) - if n_in_house < 8 and target.lat is not None and target.lon is not None: - rows = db.execute( - text( - "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin(" - "geom::geography, ST_MakePoint(:lon, :lat)::geography, 300) LIMIT 30" - ), - {"lat": target.lat, "lon": target.lon}, - ).all() - house_ids = sorted(set(house_ids) | {r.id for r in rows}) - radius_used = 300 + if radius_m is not None: + expand_radius = max(100, min(radius_m, 5000)) + if target.lat is not None and target.lon is not None: + rows = db.execute( + text( + "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin(" + "geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius) LIMIT 30" + ), + {"lat": target.lat, "lon": target.lon, "radius": expand_radius}, + ).all() + house_ids = sorted(set(house_ids) | {r.id for r in rows}) + radius_used = expand_radius + else: + n_in_house = 0 + if house_ids: + n_in_house = ( + db.execute( + text("SELECT COUNT(*) FROM house_placement_history WHERE house_id = ANY(:ids)"), + {"ids": house_ids}, + ).scalar() + or 0 + ) + if n_in_house < 8 and target.lat is not None and target.lon is not None: + rows = db.execute( + text( + "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin(" + "geom::geography, ST_MakePoint(:lon, :lat)::geography, 300) LIMIT 30" + ), + {"lat": target.lat, "lon": target.lon}, + ).all() + house_ids = sorted(set(house_ids) | {r.id for r in rows}) + radius_used = 300 if not house_ids: return HouseAnalyticsResponse( @@ -1158,12 +1216,16 @@ def get_estimate_sell_time_sensitivity( estimate_id: UUID, db: Annotated[Session, Depends(get_db)], x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None, + radius_m: int | None = None, ) -> SellTimeSensitivityResponse: """Срок продажи в зависимости от цены к медиане дома/района. 4 бакета: -5% / медиана (±3%) / +5% / +10%. Median exposure_days + p25/p75. Filter last_price > start_price * 0.7 — отбрасываем подозрительно заниженные лоты (выбросы, ошибки парсинга). + + #2044 (BE-2): optional query-param radius_m (100–5000) явно задаёт радиус + расширения выборки домов. None → текущая авто-логика (300 м при <8 записях). """ _assert_estimate_access_by_id(db, estimate_id, x_authenticated_user) # 1. Resolve house_ids (same logic as house-analytics) @@ -1194,27 +1256,41 @@ def get_estimate_sell_time_sensitivity( ).all() house_ids = [r.id for r in rows] - # Expand to 300m if too few rows (same threshold as house-analytics) + # Expand search radius (same threshold as house-analytics). #2044 (BE-2): + # explicit radius_m overrides the auto 0→300 heuristic; None → byte-identical. radius_used = 0 - n_in_house = 0 - if house_ids: - n_in_house = ( - db.execute( - text("SELECT COUNT(*) FROM house_placement_history WHERE house_id = ANY(:ids)"), - {"ids": house_ids}, - ).scalar() - or 0 - ) - if n_in_house < 8 and target.lat is not None and target.lon is not None: - rows = db.execute( - text( - "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin(" - "geom::geography, ST_MakePoint(:lon, :lat)::geography, 300) LIMIT 30" - ), - {"lat": target.lat, "lon": target.lon}, - ).all() - house_ids = sorted(set(house_ids) | {r.id for r in rows}) - radius_used = 300 + if radius_m is not None: + expand_radius = max(100, min(radius_m, 5000)) + if target.lat is not None and target.lon is not None: + rows = db.execute( + text( + "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin(" + "geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius) LIMIT 30" + ), + {"lat": target.lat, "lon": target.lon, "radius": expand_radius}, + ).all() + house_ids = sorted(set(house_ids) | {r.id for r in rows}) + radius_used = expand_radius + else: + n_in_house = 0 + if house_ids: + n_in_house = ( + db.execute( + text("SELECT COUNT(*) FROM house_placement_history WHERE house_id = ANY(:ids)"), + {"ids": house_ids}, + ).scalar() + or 0 + ) + if n_in_house < 8 and target.lat is not None and target.lon is not None: + rows = db.execute( + text( + "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin(" + "geom::geography, ST_MakePoint(:lon, :lat)::geography, 300) LIMIT 30" + ), + {"lat": target.lat, "lon": target.lon}, + ).all() + house_ids = sorted(set(house_ids) | {r.id for r in rows}) + radius_used = 300 if not house_ids: return SellTimeSensitivityResponse( diff --git a/tradein-mvp/backend/app/schemas/trade_in.py b/tradein-mvp/backend/app/schemas/trade_in.py index 2af77c3f..62f9ae54 100644 --- a/tradein-mvp/backend/app/schemas/trade_in.py +++ b/tradein-mvp/backend/app/schemas/trade_in.py @@ -246,6 +246,19 @@ class AggregatedEstimate(BaseModel): # рекомендация не сработала (или флаг estimate_manual_review_enabled выключен). manual_review_recommended: bool = False manual_review_reasons: list[str] = Field(default_factory=list) + # ── #2043 (BE-1): метрики достоверности выборки — уже считаются, отдаём наружу ── + # cv — коэффициент вариации ₽/м² по аналогам (std/mean). Метрика разброса цен: + # <0.10 «тесно» / 0.10-0.20 «умеренно» / >0.20 «широко». Anchor-путь берёт + # CV комплов дома, radius-путь — CV радиусной выборки ₽/м². None если <2 цен. + # На GET /estimate/{id} пересчитывается из сохранённых analogs (best-effort). + # source_counts — счётчики аналогов по источнику ({'avito': 12, 'cian': 5}). На + # POST считается по ПОЛНОЙ выборке до top-N отсечки; на GET/rehydrate — по + # сохранённым top-N analogs (усечённо, best-effort). Пусто при отсутствии. + # created_at — момент создания оценки (колонка trade_in_estimates.created_at), + # для метки «отчёт от DD.MM» в UI. None если не проставлен. + cv: float | None = None + source_counts: dict[str, int] = Field(default_factory=dict) + created_at: datetime | None = None # ── Параметры оценённой квартиры — нужны, чтобы восстановить карточку # при открытии оценки по ссылке (?id=), когда формы-инпута уже нет ── area_m2: float | None = None diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 0e4d60e0..090be6af 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/tradein-mvp/backend/app/services/estimator.py @@ -1805,6 +1805,40 @@ async def _with_budget(coro: Any, budget_s: float, *, label: str) -> Any: # ── PricingResult dataclass (pure, no I/O) ────────────────────────────────── +def _cv_from_ppm2(values: list[float | int | None]) -> float | None: + """Коэффициент вариации ₽/м² (std/mean) по выборке — #2043 (BE-1). + + Метрика разброса цен аналогов: делит population std на среднее. Совпадает с + CV, который _compute_same_building_anchor уже считает для anchor-комплов + (та же формула), но применима и к радиусной выборке / rehydrate из + сохранённых analogs. Возвращает None при <2 валидных (>0) значениях или + нулевом среднем (недостаточно данных → честный None, а не 0.0). + """ + vals = [float(v) for v in values if v] + n = len(vals) + if n < 2: + return None + mean = sum(vals) / n + if mean <= 0: + return None + var = sum((v - mean) ** 2 for v in vals) / n + return math.sqrt(var) / mean + + +def _source_counts(sources: list[str | None]) -> dict[str, int]: + """Счётчики по источнику ({'avito': 12, 'cian': 5}) — #2043 (BE-1). + + Считает по ПОЛНОЙ выборке аналогов (до top-N отсечки для UI). Пустые/None + источники пропускаются. Порядок ключей стабильно отсортирован для + детерминированного ответа. + """ + counts: dict[str, int] = {} + for s in sources: + if s: + counts[s] = counts.get(s, 0) + 1 + return dict(sorted(counts.items())) + + @dataclass class PricingResult: """Return type of _price_from_inputs — все переменные, нужные estimate_quality после блока.""" @@ -1828,6 +1862,10 @@ class PricingResult: ratio_basis: str | None sources_used_pre: list[str] listings_clean: list[dict] + # #2043 (BE-1): коэффициент вариации ₽/м² по выборке, на которой построен + # headline. Anchor-путь → CV комплов (anchor["cv"]); radius-путь → CV + # радиусной ₽/м²-выборки. None если <2 цен (недостаточно данных). + cv: float | None = None def _price_from_inputs( @@ -1877,6 +1915,9 @@ def _price_from_inputs( range_high = int(q3_ppm2 * area_m2) # #2: n_analogs считается по prices_ppm2, а не len(listings_clean). n_analogs = len(prices_ppm2) + # #2043 (BE-1): CV радиусной выборки ₽/м² (переопределяется anchor CV ниже, + # если якорь сработал). None при <2 ценах. + cv = _cv_from_ppm2(list(prices_ppm2)) else: median_ppm2 = 0.0 q1_ppm2 = 0.0 @@ -1885,6 +1926,7 @@ def _price_from_inputs( range_low = 0 range_high = 0 n_analogs = 0 + cv = None # 4b. Repair coefficient repair_coef = _repair_coefficient(repair_state) @@ -2046,6 +2088,8 @@ def _price_from_inputs( ) + repair_note # #695 (QA fixup): n_analogs по anchor-популяции. n_analogs = anchor["n"] + # #2043 (BE-1): headline построен на комплах якоря → CV тоже по ним. + cv = anchor["cv"] # #1871 P1: ghost-anchor guard. if not listings_clean and confidence != "low": logger.warning( @@ -2482,6 +2526,7 @@ def _price_from_inputs( ratio_basis=ratio_basis, sources_used_pre=sources_used_pre, listings_clean=listings_clean, + cv=cv, ) @@ -2874,6 +2919,7 @@ async def estimate_quality( ratio_basis = pr.ratio_basis sources_used_pre = pr.sources_used_pre listings_clean = pr.listings_clean + cv = pr.cv # 5. Deals — ДКП-only sales (вторичка) из rosreestr_deals. # Importer фильтрует doc_type='ДКП' (PR-A 2026-05-24), ДДУ застройщиков @@ -2906,6 +2952,9 @@ async def estimate_quality( analogs_lots = [_listing_to_analog(lot) for lot in listings_clean[:10]] metadata_lots = listings_clean deals_lots = [_deal_to_analog(d) for d in deals[:10]] + # #2043 (BE-1): счётчики по источнику считаем по ПОЛНОЙ выборке (metadata_lots + # — anchor-комплы или радиусные listings_clean, ДО top-N отсечки на UI). + source_counts = _source_counts([lot.get("source") for lot in metadata_lots]) freshness_pre = _compute_freshness_minutes(metadata_lots) # DaData enrichment (PR Q1) — заполняется только если service отработал. # При DaData = None все колонки идут в DB как NULL (graceful). @@ -3184,6 +3233,10 @@ async def estimate_quality( premium_building_class=premium_building_class, manual_review_recommended=manual_review_recommended, manual_review_reasons=manual_review_reasons, + # #2043 (BE-1): достоверность выборки — CV, счётчики источников, дата. + cv=cv, + source_counts=source_counts, + created_at=now, ) diff --git a/tradein-mvp/backend/tests/test_estimate_idor.py b/tradein-mvp/backend/tests/test_estimate_idor.py index 45bbf59d..2336199e 100644 --- a/tradein-mvp/backend/tests/test_estimate_idor.py +++ b/tradein-mvp/backend/tests/test_estimate_idor.py @@ -139,6 +139,10 @@ def _stub_precision_and_pdf(): _fetch_price_trend=lambda *a, **k: None, _fetch_dkp_corridor=lambda *a, **k: None, _fetch_house_imv_anchor=lambda *a, **k: None, + # #2043 (BE-1): GET-rehydrate also recomputes cv / source_counts from the + # persisted analogs. Empty analogs in the fixture → None / {} (real behaviour). + _cv_from_ppm2=lambda *a, **k: None, + _source_counts=lambda *a, **k: {}, ) real_estimator = sys.modules.get("app.services.estimator") sys.modules["app.services.estimator"] = estimator_stub # type: ignore[assignment] diff --git a/tradein-mvp/backend/tests/test_estimator_radius_m_2044.py b/tradein-mvp/backend/tests/test_estimator_radius_m_2044.py new file mode 100644 index 00000000..f7444f65 --- /dev/null +++ b/tradein-mvp/backend/tests/test_estimator_radius_m_2044.py @@ -0,0 +1,61 @@ +"""Unit tests for #2044 (BE-2) optional radius_m input. + +The schema field + estimate_quality wiring (base_radius_m / fallback_radius_m) +landed in e23dabe4; these tests lock the contract: radius_m is an optional, +bounded field whose None value preserves the byte-identical two-tier default +radii, while an explicit value overrides both search radii. + +NOTE: importing app.services.estimator pulls app.core.config.Settings which +requires DATABASE_URL. Set it BEFORE importing app modules. +""" + +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +import pytest +from pydantic import ValidationError + +from app.schemas.trade_in import TradeInEstimateInput +from app.services import estimator + + +def test_radius_m_defaults_to_none() -> None: + payload = TradeInEstimateInput(address="ул. Тестовая, 1", area_m2=50, rooms=2) + assert payload.radius_m is None + + +def test_radius_m_accepts_valid_int() -> None: + payload = TradeInEstimateInput(address="ул. Тестовая, 1", area_m2=50, rooms=2, radius_m=1500) + assert payload.radius_m == 1500 + + +def test_radius_m_out_of_range_rejected() -> None: + with pytest.raises(ValidationError): + TradeInEstimateInput(address="ул. Тестовая, 1", area_m2=50, rooms=2, radius_m=50) + with pytest.raises(ValidationError): + TradeInEstimateInput(address="ул. Тестовая, 1", area_m2=50, rooms=2, radius_m=9000) + + +def test_radius_none_preserves_default_radii() -> None: + # estimate_quality resolves `payload.radius_m or DEFAULT_RADIUS_M`. None must + # keep the byte-identical two-tier defaults; an explicit value overrides both. + assert (None or estimator.DEFAULT_RADIUS_M) == estimator.DEFAULT_RADIUS_M + assert (None or estimator.FALLBACK_RADIUS_M) == estimator.FALLBACK_RADIUS_M + assert (500 or estimator.DEFAULT_RADIUS_M) == 500 + assert (500 or estimator.FALLBACK_RADIUS_M) == 500 + + +def test_api_radius_expand_clamp_bounds() -> None: + # The house-analytics / sell-time endpoints clamp an explicit radius into + # [100, 5000] before ST_DWithin. Mirror that guard so the contract is locked. + def _clamp(r: int) -> int: + return max(100, min(r, 5000)) + + assert _clamp(50) == 100 + assert _clamp(300) == 300 + assert _clamp(9000) == 5000 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(pytest.main([__file__, "-q"])) diff --git a/tradein-mvp/backend/tests/test_estimator_source_metrics_2043.py b/tradein-mvp/backend/tests/test_estimator_source_metrics_2043.py new file mode 100644 index 00000000..66a7054d --- /dev/null +++ b/tradein-mvp/backend/tests/test_estimator_source_metrics_2043.py @@ -0,0 +1,178 @@ +"""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"] + + +# --------------------------------------------------------------------------- # +# _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"]))