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=<user>.
     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 radius_m is not None:
-    if house_ids:
+        expand_radius = max(100, min(radius_m, 5000))
-        n_in_house = (
+        if target.lat is not None and target.lon is not None:
-            db.execute(
+            rows = db.execute(
-                text("SELECT COUNT(*) FROM house_placement_history WHERE house_id = ANY(:ids)"),
+                text(
-                {"ids": house_ids},
+                    "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
-            ).scalar()
+                    "geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius) LIMIT 30"
-            or 0
+                ),
-        )
+                {"lat": target.lat, "lon": target.lon, "radius": expand_radius},
-    if n_in_house < 8 and target.lat is not None and target.lon is not None:
+            ).all()
-        rows = db.execute(
+            house_ids = sorted(set(house_ids) | {r.id for r in rows})
-            text(
+        radius_used = expand_radius
-                "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
+    else:
-                "geom::geography, ST_MakePoint(:lon, :lat)::geography, 300) LIMIT 30"
+        n_in_house = 0
-            ),
+        if house_ids:
-            {"lat": target.lat, "lon": target.lon},
+            n_in_house = (
-        ).all()
+                db.execute(
-        house_ids = sorted(set(house_ids) | {r.id for r in rows})
+                    text("SELECT COUNT(*) FROM house_placement_history WHERE house_id = ANY(:ids)"),
-        radius_used = 300
+                    {"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 radius_m is not None:
-    if house_ids:
+        expand_radius = max(100, min(radius_m, 5000))
-        n_in_house = (
+        if target.lat is not None and target.lon is not None:
-            db.execute(
+            rows = db.execute(
-                text("SELECT COUNT(*) FROM house_placement_history WHERE house_id = ANY(:ids)"),
+                text(
-                {"ids": house_ids},
+                    "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
-            ).scalar()
+                    "geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius) LIMIT 30"
-            or 0
+                ),
-        )
+                {"lat": target.lat, "lon": target.lon, "radius": expand_radius},
-    if n_in_house < 8 and target.lat is not None and target.lon is not None:
+            ).all()
-        rows = db.execute(
+            house_ids = sorted(set(house_ids) | {r.id for r in rows})
-            text(
+        radius_used = expand_radius
-                "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin("
+    else:
-                "geom::geography, ST_MakePoint(:lon, :lat)::geography, 300) LIMIT 30"
+        n_in_house = 0
-            ),
+        if house_ids:
-            {"lat": target.lat, "lon": target.lon},
+            n_in_house = (
-        ).all()
+                db.execute(
-        house_ids = sorted(set(house_ids) | {r.id for r in rows})
+                    text("SELECT COUNT(*) FROM house_placement_history WHERE house_id = ANY(:ids)"),
-        radius_used = 300
+                    {"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(

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

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,
     )
 
 

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]

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"]))

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"]))