From 347203dfda4d68404879ea598e0ea855883fc6f8 Mon Sep 17 00:00:00 2001 From: Light1YT Date: Mon, 8 Jun 2026 16:47:18 +0500 Subject: [PATCH] =?UTF-8?q?feat(forecasting):=20=C2=A725.3=20TRUE=20own-po?= =?UTF-8?q?rtfolio=20cannibalization=20overlap=20(#1169=20PR2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Upgrade the §25.3 cannibalization index from a same-class-competitor proxy to true own-portfolio overlap: score the recommended candidate segment against the developer's own portfolio (get_own_portfolio, #1169 PR1) across axes — audience/class (ordinal distance), price ₽/м² (interval overlap), unit-mix (L1 similarity), timing (half-life decay) — geo-weighted by haversine proximity to the parcel, aggregated by geo-weighted soft-max (the strongest nearby cannibalizer dominates, not a diluting mean). Empty portfolio -> labelled proxy fallback, confidence forced low, never presented as the true index. Pure scoring fns unit-tested without DB; None-not-0 on missing axes; thin/ only-current portfolio -> low confidence + §26 note; deterministic (sorted tie-breaks, no RNG). class+price+geo active now; unit-mix+timing plumbed via optional params for a follow-up that wires them from the orchestrator horizon. ruff + mypy clean; 151 special-index tests pass (964 forecasting dir, no regr). Refs #1169 --- .../services/forecasting/special_indices.py | 581 +++++++++++++++++- .../forecasting/test_special_indices.py | 542 +++++++++++++++- 2 files changed, 1109 insertions(+), 14 deletions(-) diff --git a/backend/app/services/forecasting/special_indices.py b/backend/app/services/forecasting/special_indices.py index 79c4cb90..244090b2 100644 --- a/backend/app/services/forecasting/special_indices.py +++ b/backend/app/services/forecasting/special_indices.py @@ -11,9 +11,12 @@ где дефицит ПИКОВЫЙ = рекомендованное окно запуска. SOLID. 2. Product Void ← #981 rank_segments: сегменты с deficit_index ≥ _VOID_THRESHOLD (белое пятно — недообеспеченный спрос). QUICK-WIN, SOLID. - 3. Cannibalization ← §9.1 competitors: доля конкурентов В ТОМ ЖЕ классе, взвешенная - relevance — пересечение рек. сегмента с чужим/своим стоком. - DEGRADE-to-None, если нет геометрии cad_num. + 3. Cannibalization ← §25.3 ИСТИННОЕ пересечение рек. сегмента-кандидата с НАШИМ + портфелем (own_portfolio #1169) по 4 осям (класс/аудитория, + цена ₽/м², квартирография, тайминг) + гео-вес близости наших + проектов к участку. Портфель пуст (own_developer_ids не задан И + нет manual-строк) → FALLBACK на ПРОКСИ (доля конкурентов того же + класса, явно помеченный). DEGRADE-to-None, если нет геометрии. 4. Competitor Strength ← §9.1 competitors: среднее relevance_weight топ-N. DEGRADE, если нет cad_num/геометрии. 5. Artificial Demand ← НОВЫЙ read objective_lots: доля проданных лотов с ипотекой @@ -38,8 +41,10 @@ value=None, method='unavailable', confidence='low', а остальные пят from __future__ import annotations import logging +import math from collections.abc import Sequence from dataclasses import dataclass +from datetime import date from typing import Any, Literal from sqlalchemy import text @@ -48,10 +53,20 @@ from sqlalchemy.orm import Session from app.schemas.parcel import CompetitorsRequest from app.services.forecasting.affordability import compute_affordability from app.services.forecasting.demand_supply_forecast import compute_demand_supply_forecast -from app.services.forecasting.sales_series import SegmentSpec +from app.services.forecasting.sales_series import ( + _PRICE_BAND_BUSINESS_MAX, + _PRICE_BAND_COMFORT_MAX, + _PRICE_BAND_ECONOMY_MAX, + PRICE_BUCKET_BUSINESS, + PRICE_BUCKET_COMFORT, + PRICE_BUCKET_ECONOMY, + PRICE_BUCKET_PREMIUM, + SegmentSpec, +) from app.services.forecasting.what_to_build import rank_segments -from app.services.site_finder.competitors import get_competitors +from app.services.site_finder.competitors import _CLASS_ORDER, _normalize_class, get_competitors from app.services.site_finder.market_metrics import compute_market_metrics +from app.services.site_finder.own_portfolio import OwnProject, get_own_portfolio logger = logging.getLogger(__name__) @@ -102,6 +117,40 @@ _COMPETITOR_RADIUS_KM: float = 1.0 # конкурент, распроданный к нашему запуску, менее релевантен. 12 мес — дефолт. _COMPETITOR_HORIZON_MONTHS: int = 12 +# ── §25.3 own-portfolio overlap (TRUE cannibalization, #1169 PR2) ────────────── + +# Похожесть аудитории/класса по дистанции в _CLASS_ORDER (competitors._CLASS_ORDER): +# тот же класс = полная каннибализация аудитории, соседний (1 шаг) — частичная, +# дальше — слабая. Чуть резче, чем competitors._CLASS_SIM_* (там relevance- +# корректировка), т.к. каннибализация бьёт ПО АУДИТОРИИ: 0 шагов→1.0, 1→0.5, 2→0.2, +# ≥3→0.05. Класс одной из сторон не распознан → ось НЕДОСТУПНА (None, НЕ 0). +_OVERLAP_CLASS_BY_STEPS: dict[int, float] = {0: 1.0, 1: 0.5, 2: 0.2} +_OVERLAP_CLASS_FAR: float = 0.05 + +# Период полу-затухания (мес) тайминговой близости запусков: |Δмес| = half_life → +# вклад тайминга 0.5; exp(-|Δ|/half_life). 12 мес — год между выходами ≈ половинное +# пересечение окон продаж (типовая экспозиция ЖК на рынке). Tunable. +_TIMING_HALF_LIFE_MONTHS: float = 12.0 + +# Масштаб (км) гео-затухания веса нашего проекта: weight = exp(-distance_km/scale). +# 3 км → вес 0.37, 1 км → 0.72, 10 км → 0.04. Ближний наш проект каннибализирует +# сильнее (общий локальный спрос); дальний — почти нет. Tunable. +_GEO_WEIGHT_SCALE_KM: float = 3.0 + +# Гео-вес нашего проекта без координат (тайминг/цена сравнимы, но близость неизвестна): +# НЕ исключаем проект и НЕ даём полный вес — низкий floor, чтобы он мог сигналить, но +# не доминировал над проектом с подтверждённой близостью. None-not-0 дух (есть проект, +# но геопривязка слабая → down-weight, а не выкидываем). +_GEO_WEIGHT_UNKNOWN: float = 0.1 + +# Минимум доступных осей пересечения для пары кандидат↔наш проект, чтобы считать оценку +# надёжной. < этого (напр. только класс) → пара low-confidence (сигнал есть, но тонкий). +_OVERLAP_MIN_AXES: int = 2 + +# Порог «тонкого» портфеля для honest-confidence (§26): портфель только из current +# (нет future-пайплайна) ИЛИ проектов меньше этого числа → confidence='low' + нота. +_THIN_PORTFOLIO_MAX_PROJECTS: int = 2 + # Saturation для нормализации пикового дефицита Launch Window в [0,1]: при каком # deficit_index сила окна достигает 1.0. 0.5 = половина шкалы #980 уже «сильное окно». # Линейная отсечка max(0, min(1, peak / saturation)). Tunable. @@ -339,6 +388,238 @@ def _cannibalization_index( return _clamp01(same / total) +# ── §25.3 TRUE own-portfolio overlap — pure-оси (None-not-0) + агрегация ──────── + + +def _class_overlap(candidate_class: str | None, own_class: str | None) -> float | None: + """Ось аудитория/класс: пересечение классов кандидата и нашего проекта. PURE. + + Нормализуем оба класса (_normalize_class — регистро/язык-независимо: 'Комфорт'/ + 'comfort'/'комфорт-класс' → один ключ) и берём похожесть по дистанции в _CLASS_ORDER: + 0 шагов (тот же класс) → 1.0, 1 (соседний) → 0.5, 2 → 0.2, ≥3 → 0.05. Класс ХОТЯ БЫ + одной стороны не распознан → None (ось НЕДОСТУПНА — НЕ фабрикуем 0). PURE. + """ + c = _normalize_class(candidate_class) + o = _normalize_class(own_class) + if c is None or o is None: + return None + steps = abs(_CLASS_ORDER[c] - _CLASS_ORDER[o]) + return _OVERLAP_CLASS_BY_STEPS.get(steps, _OVERLAP_CLASS_FAR) + + +def _price_bucket_to_band(price_bucket: str | None) -> tuple[float, float] | None: + """price_bucket кандидата → ценовая вилка [lo, hi] ₽/м² (зеркало _PRICE_BAND_*). PURE. + + Кандидат-сегмент несёт лишь bucket-ключ (не вилку); разворачиваем его в тот же + диапазон, что sales_series.price_bucket_of: эконом [0, 120k), комфорт [120k, 160k), + бизнес [160k, 220k), премиум [220k, +inf). Неизвестный/None bucket → None (ось цены + НЕДОСТУПНА — НЕ фабрикуем). PURE. + """ + if price_bucket == PRICE_BUCKET_ECONOMY: + return (0.0, _PRICE_BAND_ECONOMY_MAX) + if price_bucket == PRICE_BUCKET_COMFORT: + return (_PRICE_BAND_ECONOMY_MAX, _PRICE_BAND_COMFORT_MAX) + if price_bucket == PRICE_BUCKET_BUSINESS: + return (_PRICE_BAND_COMFORT_MAX, _PRICE_BAND_BUSINESS_MAX) + if price_bucket == PRICE_BUCKET_PREMIUM: + return (_PRICE_BAND_BUSINESS_MAX, math.inf) + return None + + +def _price_overlap( + candidate_band: tuple[float, float] | None, + own_lo: float | None, + own_hi: float | None, +) -> float | None: + """Ось цены: доля пересечения ценовых вилок кандидата и нашего проекта. PURE. + + Пересечение интервалов [c_lo,c_hi] и [o_lo,o_hi]; нормируем на УЖЕ узкую вилку + (min ширины) → полное накрытие более узкого диапазона = 1.0 (стороны полностью + конкурируют в этой ценовой зоне). Любая граница / вилка кандидата неизвестна или + обе вилки нулевой ширины → None (ось НЕДОСТУПНА — НЕ фабрикуем 0). Открытая правая + граница (премиум, +inf) поддержана: ширина считается от конечной нижней. PURE. + """ + if candidate_band is None or own_lo is None or own_hi is None: + return None + c_lo, c_hi = candidate_band + o_lo, o_hi = (own_lo, own_hi) if own_lo <= own_hi else (own_hi, own_lo) + lo = max(c_lo, o_lo) + hi = min(c_hi, o_hi) + overlap = max(0.0, hi - lo) + if not math.isfinite(overlap): + # Обе вилки уходят в +inf (оба премиум) — считаем полным пересечением. + return 1.0 + c_width = c_hi - c_lo + o_width = o_hi - o_lo + widths = [w for w in (c_width, o_width) if math.isfinite(w) and w > 0] + if not widths: + return None + return _clamp01(overlap / min(widths)) + + +def _unit_mix_similarity( + candidate_mix: dict[str, float] | None, + own_mix: dict[str, float] | None, +) -> float | None: + """Ось квартирографии: похожесть распределений долей. PURE. + + similarity = 1 − 0.5·Σ|p_k − q_k| по ОБЪЕДИНЕНИЮ ключей (отсутствующий ключ = доля 0). + Для нормированных распределений (Σ=1) это ∈ [0,1]: идентичные миксы → 1.0, + непересекающиеся → 0.0. Доли нормируем на свою сумму (страховка от «не-в-сумме-1» + входа), clamp [0,1]. Любой микс None/пустой → None (ось НЕДОСТУПНА — НЕ фабрикуем). + PURE, не мутирует вход. + """ + if not candidate_mix or not own_mix: + return None + cand = _normalize_shares(candidate_mix) + own = _normalize_shares(own_mix) + if cand is None or own is None: + return None + keys = set(cand) | set(own) + l1 = sum(abs(cand.get(k, 0.0) - own.get(k, 0.0)) for k in sorted(keys)) + return _clamp01(1.0 - 0.5 * l1) + + +def _normalize_shares(mix: dict[str, float]) -> dict[str, float] | None: + """Нормировать доли на их сумму → {key: share}, Σ=1. PURE. + + Отрицательные/None доли отбрасываем (грязь); сумма ≤ 0 → None (нет осмысленного + распределения). Ключи сортируются для детерминизма обхода. PURE. + """ + clean = {k: float(v) for k, v in mix.items() if v is not None and float(v) > 0} + total = sum(clean.values()) + if total <= 0: + return None + return {k: clean[k] / total for k in sorted(clean)} + + +def _timing_overlap( + candidate_month: date | None, + own_month: date | None, + *, + half_life_months: float = _TIMING_HALF_LIFE_MONTHS, +) -> float | None: + """Ось тайминга: временна́я близость окон запуска. PURE. + + exp(−|Δмесяцев| / half_life): одновременный выход → 1.0, расхождение в half_life мес + → 0.5, дальше затухает. Чем ближе наши запуски, тем сильнее пересекаются окна продаж + = выше каннибализация. Любая дата None → None (ось НЕДОСТУПНА — НЕ фабрикуем). PURE. + """ + if candidate_month is None or own_month is None: + return None + months_apart = abs(_months_between(candidate_month, own_month)) + if half_life_months <= 0: + return 1.0 if months_apart == 0 else 0.0 + # Истинный период полу-затухания: 0.5 ** (Δ/half_life) → ровно 0.5 при Δ=half_life. + return _clamp01(0.5 ** (months_apart / half_life_months)) + + +def _months_between(a: date, b: date) -> int: + """Целое число месяцев между двумя датами (знак = направление). PURE.""" + return (a.year - b.year) * 12 + (a.month - b.month) + + +def _geo_weight( + distance_km: float | None, + *, + scale_km: float = _GEO_WEIGHT_SCALE_KM, + unknown_weight: float = _GEO_WEIGHT_UNKNOWN, +) -> float: + """Гео-вес нашего проекта по близости к участку кандидата. PURE. + + exp(−distance_km / scale): на участке (0 км) → 1.0, дальше затухает (1 км ≈ 0.72, + 3 км ≈ 0.37). Расстояние неизвестно (нет координат у проекта/участка) → НИЗКИЙ floor + `unknown_weight` (проект ещё может сигналить, но не доминирует — близость не + подтверждена). Это ВЕС (не ось пересечения), поэтому всегда задан (не None). PURE. + + Args: + distance_km: расстояние от участка до нашего проекта (км) или None. + scale_km: масштаб затухания. + unknown_weight: floor-вес при неизвестном расстоянии. + + Returns: + Вес ∈ (0,1]. + """ + if distance_km is None: + return unknown_weight + if scale_km <= 0: + return 1.0 if distance_km <= 0 else 0.0 + return _clamp01(math.exp(-max(0.0, distance_km) / scale_km)) + + +@dataclass(frozen=True) +class _PairOverlap: + """Пересечение кандидата с ОДНИМ нашим проектом (промежуточный, не публичный). + + `overlap` — среднее ДОСТУПНЫХ осей ∈ [0,1] или None (ни одной оси). `n_axes` — + сколько осей реально посчитано (None-оси исключены). `geo_weight` — вес близости. + `signal` = overlap × geo_weight (гео-взвешенный вклад проекта). `axes` — пер-ось + breakdown для explainability. + """ + + overlap: float | None + n_axes: int + geo_weight: float + signal: float | None + axes: dict[str, float | None] + + +def _own_portfolio_overlap( + *, + class_overlap: float | None, + price_overlap: float | None, + unit_mix_overlap: float | None, + timing_overlap: float | None, + geo_weight: float, +) -> _PairOverlap: + """Свести оси пересечения кандидат↔наш-проект в гео-взвешенный сигнал. PURE. + + Усредняем ТОЛЬКО доступные (не-None) оси — пропущенная ось НЕ тянет пересечение к 0 + (None-not-0): нет данных по оси → её просто нет в среднем. Ни одной оси → overlap + None (пара неинформативна). signal = overlap × geo_weight: близкий и сильно + пересекающийся наш проект даёт высокий сигнал каннибализации. PURE. + + Args: + class_overlap/price_overlap/unit_mix_overlap/timing_overlap: оси ∈ [0,1] | None. + geo_weight: вес близости проекта к участку ∈ (0,1]. + + Returns: + _PairOverlap (overlap=None, signal=None, если ни одной оси не доступно). + """ + axes: dict[str, float | None] = { + "class": class_overlap, + "price": price_overlap, + "unit_mix": unit_mix_overlap, + "timing": timing_overlap, + } + available = [v for v in axes.values() if v is not None] + if not available: + return _PairOverlap(overlap=None, n_axes=0, geo_weight=geo_weight, signal=None, axes=axes) + overlap = sum(available) / len(available) + return _PairOverlap( + overlap=overlap, + n_axes=len(available), + geo_weight=geo_weight, + signal=overlap * geo_weight, + axes=axes, + ) + + +def _aggregate_overlap(pairs: Sequence[_PairOverlap]) -> float | None: + """Свести пер-проектные гео-взвешенные сигналы в индекс каннибализации. PURE. + + Берём МАКСИМУМ signal (= overlap × geo_weight) по нашим проектам, а НЕ среднее: + ОДИН близкий сильно-пересекающийся проект — уже самостоятельный риск каннибализации, + и среднее размыло бы его длинным хвостом далёких/непохожих проектов (soft-max=max — + «сильнейший каннибализатор»). Ни одной пары с сигналом → None (нечего агрегировать, + НЕ 0). Детерминированно (max по числам). PURE. + """ + signals = [p.signal for p in pairs if p.signal is not None] + if not signals: + return None + return _clamp01(max(signals)) + + def _artificial_demand_share(n_mortgage: int | None, n_sold: int | None) -> float | None: """Доля проданных лотов, профинансированных ипотекой (Artificial Demand). PURE. @@ -499,6 +780,82 @@ def _query_artificial_demand( } +# ────────────────────────────────────────────────────────────────────────────── +# §25.3 cannibalization — центроид участка (гео-вес наших проектов). psycopg v3. +# ────────────────────────────────────────────────────────────────────────────── + +# Центроид участка кандидата (lon/lat) для гео-веса наших проектов — reuse семантики +# competitors._PARCEL_CENTROID_SQL (cad_parcels_geom → fallback cad_quarters_geom). +# Параметризован, без f-string. Гео тут — ВЕС (не ось пересечения): нет геометрии → +# все гео-веса падают на _GEO_WEIGHT_UNKNOWN (overlap всё равно считается по остальным +# осям). НИКОГДА :x::type. +_PARCEL_CENTROID_SQL = text( + """ + SELECT ST_X(pt) AS lon, ST_Y(pt) AS lat + FROM ( + SELECT ST_Centroid(geom) AS pt + FROM cad_parcels_geom + WHERE cad_num = :cad_num AND geom IS NOT NULL + UNION ALL + SELECT ST_Centroid(geom) AS pt + FROM cad_quarters_geom + WHERE cad_number = :quarter AND geom IS NOT NULL + ) sub + LIMIT 1 + """ +) + +# Средний радиус Земли (км) для гаверсинуса. Распределённость наших проектов в пределах +# города — гаверсинус точен достаточно (PostGIS-geography для пары точек тут избыточен). +_EARTH_RADIUS_KM: float = 6371.0 + + +def _quarter_from_cad(cad_num: str) -> str: + """Кадастровый квартал из номера участка (66:41:0303161:123 → 66:41:0303161). PURE. + + Нестандартный формат → cad_num как есть (fallback). Зеркалит competitors._quarter_from_cad. + """ + parts = cad_num.split(":") + if len(parts) >= 3: + return ":".join(parts[:3]) + return cad_num + + +def _haversine_km(lon1: float, lat1: float, lon2: float, lat2: float) -> float: + """Гаверсинус-расстояние (км) между двумя (lon,lat). PURE. + + Детерминированно; для внутригородских пар точек точности достаточно (гео — лишь вес). + """ + rlat1, rlat2 = math.radians(lat1), math.radians(lat2) + dlat = math.radians(lat2 - lat1) + dlon = math.radians(lon2 - lon1) + h = math.sin(dlat / 2) ** 2 + math.cos(rlat1) * math.cos(rlat2) * math.sin(dlon / 2) ** 2 + return 2.0 * _EARTH_RADIUS_KM * math.asin(min(1.0, math.sqrt(h))) + + +def _query_parcel_centroid(db: Session, *, cad_num: str) -> tuple[float, float] | None: + """Центроид участка (lon, lat) для гео-веса. Graceful → None. + + Нет геометрии / сбой → None (гео-веса упадут на floor; overlap считается по остальным + осям, индекс НЕ деградирует целиком). Параметризовано (psycopg v3). Детерминированно. + """ + try: + row = ( + db.execute( + _PARCEL_CENTROID_SQL, + {"cad_num": cad_num, "quarter": _quarter_from_cad(cad_num)}, + ) + .mappings() + .first() + ) + except Exception: + logger.exception("cannibalization: centroid query failed for cad_num=%s", cad_num) + return None + if not row or row["lon"] is None or row["lat"] is None: + return None + return (float(row["lon"]), float(row["lat"])) + + # ────────────────────────────────────────────────────────────────────────────── # Per-index builders — каждый тонкий, graceful; pure-математика выше тестируется без БД. # ────────────────────────────────────────────────────────────────────────────── @@ -632,8 +989,22 @@ def _build_competitor_strength(db: Session, *, cad_num: str | None) -> SpecialIn ) -def _build_cannibalization(db: Session, *, spec: SegmentSpec, cad_num: str | None) -> SpecialIndex: - """Index 3 — Cannibalization: доля relevance-силы конкурентов В классе. Degrade.""" +# Метка-каведат для проксированной каннибализации (own-portfolio не сконфигурирован). +_PROXY_REASON: str = ( + "own-portfolio не сконфигурирован (own_developer_ids не задан И нет manual-строк) " + "→ ПРОКСИ: плотность конкурентов того же класса (НЕ истинное пересечение портфеля)" +) + + +def _build_cannibalization_proxy( + db: Session, *, spec: SegmentSpec, cad_num: str | None +) -> SpecialIndex: + """Index 3 FALLBACK — ПРОКСИ: доля relevance-силы конкурентов В классе. Degrade. + + Используется ТОЛЬКО когда наш портфель пуст (own-portfolio не сконфигурирован): + честно помечается method='proxy_*' и detail.proxy=True, чтобы прокси НИКОГДА не + выдавался за истинный own-portfolio индекс. Логика — прежний _cannibalization_index. + """ if not cad_num: return _unavailable( KEY_CANNIBALIZATION, @@ -641,34 +1012,218 @@ def _build_cannibalization(db: Session, *, spec: SegmentSpec, cad_num: str | Non ) competitors = _fetch_competitors(db, cad_num=cad_num) all_relevance = [c.relevance_weight for c in competitors] - target_class = (spec.obj_class or "").strip().lower() + target = _normalize_class(spec.obj_class) same_class_relevance = [ c.relevance_weight for c in competitors - if target_class and (c.obj_class or "").strip().lower() == target_class + if target is not None and _normalize_class(c.obj_class) == target ] value = _cannibalization_index(same_class_relevance, all_relevance) if value is None: return _unavailable( KEY_CANNIBALIZATION, - reason="нет конкурентов в радиусе — пересечения нет", + reason="нет конкурентов в радиусе — пересечения нет (прокси)", ) return SpecialIndex( key=KEY_CANNIBALIZATION, value=value, - label=spec.obj_class, - confidence=_cap_confidence("medium"), + label=f"прокси · {spec.obj_class}" if spec.obj_class else "прокси", + confidence=_cap_confidence("low"), # прокси — заведомо слабее истинного detail={ + "mode": "proxy", + "proxy": True, + "proxy_reason": _PROXY_REASON, "n_competitors": len(competitors), "n_same_class": len(same_class_relevance), "target_class": spec.obj_class, "radius_km": _COMPETITOR_RADIUS_KM, }, - method="same_class_relevance_share", + method="proxy_same_class_relevance_share", advisory=_ADVISORY, ) +def _build_cannibalization_true( + *, + spec: SegmentSpec, + portfolio: Sequence[OwnProject], + centroid: tuple[float, float] | None, + candidate_unit_mix: dict[str, float] | None, + candidate_release_month: date | None, +) -> SpecialIndex: + """Index 3 TRUE — §25.3 пересечение кандидата с НАШИМ портфелем по 4 осям + гео. + + Для каждого нашего проекта считаем оси (класс/цена/квартирография/тайминг — + None-not-0 при тонких данных) + гео-вес близости к участку, сводим в гео-взвешенный + сигнал, агрегируем МАКСИМУМОМ (сильнейший каннибализатор). Топ-проекты — в detail + (explainability §16). Confidence отражает покрытие портфеля (тонкий/только-current → + low + нота §26). Чистая агрегация над pure-осями — без БД здесь. Детерминированно. + """ + candidate_band = _price_bucket_to_band(spec.price_bucket) + pairs: list[_PairOverlap] = [] + contributions: list[tuple[float, OwnProject, _PairOverlap]] = [] + for own in portfolio: + distance_km = _own_distance_km(centroid, own) + price_overlap = _price_overlap( + candidate_band, own.price_min_per_m2, own.price_max_per_m2 + ) + pair = _own_portfolio_overlap( + class_overlap=_class_overlap(spec.obj_class, own.obj_class), + price_overlap=price_overlap, + unit_mix_overlap=_unit_mix_similarity(candidate_unit_mix, own.unit_mix), + timing_overlap=_timing_overlap(candidate_release_month, own.release_month), + geo_weight=_geo_weight(distance_km), + ) + pairs.append(pair) + if pair.signal is not None: + contributions.append((pair.signal, own, pair)) + + value = _aggregate_overlap(pairs) + if value is None: + return _unavailable( + KEY_CANNIBALIZATION, + reason=( + "ни по одному нашему проекту нет сопоставимых осей " + "(класс/цена/квартирография/тайминг) — пересечение неизмеримо (НЕ фабрикуем)" + ), + ) + + # Топ-вкладчики (сильнейшие каннибализаторы) — детерминированно по signal DESC, + # tie-break по имени проекта (стабильно, без RNG/порядка set). + contributions.sort(key=lambda t: (-t[0], t[1].name)) + top = contributions[: _COMPETITOR_TOP_N] + confidence = _cap_confidence(_portfolio_confidence(portfolio)) + return SpecialIndex( + key=KEY_CANNIBALIZATION, + value=value, + label=_top_contributor_label(top), + confidence=confidence, + detail={ + "mode": "own_portfolio", + "proxy": False, + "n_own_projects": len(portfolio), + "n_current": sum(1 for p in portfolio if p.source == "current"), + "n_future": sum(1 for p in portfolio if p.source == "future"), + "has_geometry": centroid is not None, + "axes_available": _axes_available_summary(pairs), + "top_contributors": [_contribution_detail(s, own, pair) for s, own, pair in top], + "confidence_note": _portfolio_confidence_note(portfolio), + }, + method="own_portfolio_overlap", + advisory=_ADVISORY, + ) + + +def _own_distance_km(centroid: tuple[float, float] | None, own: OwnProject) -> float | None: + """Расстояние (км) от центроида участка до нашего проекта или None. PURE. + + Нет центроида ИЛИ нет координат проекта → None (гео-вес упадёт на floor). PURE. + """ + if centroid is None or own.lon is None or own.lat is None: + return None + return _haversine_km(centroid[0], centroid[1], own.lon, own.lat) + + +def _portfolio_confidence(portfolio: Sequence[OwnProject]) -> Confidence: + """Honest-confidence по покрытию портфеля (§26). PURE. + + Тонкий портфель (< _THIN_PORTFOLIO_MAX_PROJECTS проектов) ИЛИ ТОЛЬКО current + (нет future-пайплайна — мы не знаем своих будущих запусков) → 'low' (данных мало). + Иначе 'medium' (всё равно ≤ cap: слой advisory). PURE. + """ + has_future = any(p.source == "future" for p in portfolio) + if len(portfolio) < _THIN_PORTFOLIO_MAX_PROJECTS or not has_future: + return "low" + return "medium" + + +def _portfolio_confidence_note(portfolio: Sequence[OwnProject]) -> str | None: + """RU-нота §26 о тонком покрытии портфеля или None, если покрытие достаточное. PURE.""" + n = len(portfolio) + has_future = any(p.source == "future" for p in portfolio) + if n < _THIN_PORTFOLIO_MAX_PROJECTS: + return f"данных мало: в портфеле {n} проект(ов) — оценка ориентировочная" + if not has_future: + return "только текущие проекты (нет future-пайплайна) — будущие запуски не учтены" + return None + + +def _axes_available_summary(pairs: Sequence[_PairOverlap]) -> dict[str, int]: + """Сколько пар имеют каждую ось доступной (explainability «что реально считалось»). PURE.""" + summary = {"class": 0, "price": 0, "unit_mix": 0, "timing": 0} + for pair in pairs: + for axis, val in pair.axes.items(): + if val is not None: + summary[axis] += 1 + return summary + + +def _contribution_detail( + signal: float, own: OwnProject, pair: _PairOverlap +) -> dict[str, Any]: + """Explainability-карточка одного топ-каннибализатора. PURE.""" + return { + "name": own.name, + "source": own.source, + "obj_class": own.obj_class, + "district": own.district, + "signal": _round_or_none(signal, 3), + "overlap": _round_or_none(pair.overlap, 3), + "geo_weight": _round_or_none(pair.geo_weight, 3), + "n_axes": pair.n_axes, + "axes": {k: _round_or_none(v, 3) for k, v in pair.axes.items()}, + "low_confidence_pair": pair.n_axes < _OVERLAP_MIN_AXES, + } + + +def _top_contributor_label( + top: Sequence[tuple[float, OwnProject, _PairOverlap]], +) -> str | None: + """Человекочитаемый label = сильнейший наш каннибализатор. PURE.""" + if not top: + return None + _, own, _ = top[0] + return f"пересечение с «{own.name}»" + + +def _build_cannibalization( + db: Session, + *, + spec: SegmentSpec, + cad_num: str | None, + candidate_unit_mix: dict[str, float] | None = None, + candidate_release_month: date | None = None, +) -> SpecialIndex: + """Index 3 — Cannibalization (§25.3): истинное own-portfolio-пересечение / прокси. + + Портфель НЕпуст → истинный индекс (overlap кандидата с нашими проектами по 4 осям + + гео-вес близости к участку); explainability называет режим own-portfolio и топ- + каннибализаторов. Портфель ПУСТ (own_developer_ids не задан И нет manual-строк) → + FALLBACK на ПРОКСИ (_build_cannibalization_proxy), явно помеченный method='proxy_*' + — прокси НИКОГДА не выдаётся за истинный индекс. Детерминированно. + + candidate_unit_mix / candidate_release_month — опциональные оси квартирографии/ + тайминга кандидата (из рекомендации what_to_build, если доступны caller'у). Не + переданы → эти оси НЕДОСТУПНЫ и честно ИСКЛЮЧАЮТСЯ из среднего (None-not-0). + """ + portfolio = get_own_portfolio(db) + if not portfolio: + logger.info( + "cannibalization: own-portfolio пуст → FALLBACK на прокси (cad_num=%s)", cad_num + ) + return _build_cannibalization_proxy(db, spec=spec, cad_num=cad_num) + + # Гео — лишь ВЕС: нет геометрии → веса на floor, overlap считается по остальным осям. + centroid = _query_parcel_centroid(db, cad_num=cad_num) if cad_num else None + return _build_cannibalization_true( + spec=spec, + portfolio=portfolio, + centroid=centroid, + candidate_unit_mix=candidate_unit_mix, + candidate_release_month=candidate_release_month, + ) + + def _build_artificial_demand( db: Session, *, spec: SegmentSpec, district: str | None, premise_kind: str ) -> SpecialIndex: diff --git a/backend/tests/services/forecasting/test_special_indices.py b/backend/tests/services/forecasting/test_special_indices.py index dcce5050..b555d41d 100644 --- a/backend/tests/services/forecasting/test_special_indices.py +++ b/backend/tests/services/forecasting/test_special_indices.py @@ -16,12 +16,19 @@ test_market_metrics. from __future__ import annotations +from datetime import date from typing import Any from unittest.mock import MagicMock, patch import pytest -from app.services.forecasting.sales_series import SegmentSpec +from app.services.forecasting.sales_series import ( + PRICE_BUCKET_BUSINESS, + PRICE_BUCKET_COMFORT, + PRICE_BUCKET_ECONOMY, + PRICE_BUCKET_PREMIUM, + SegmentSpec, +) from app.services.forecasting.special_indices import ( _INDEX_KEYS, _METHOD_UNAVAILABLE, @@ -31,21 +38,31 @@ from app.services.forecasting.special_indices import ( KEY_COST_OF_ERROR, KEY_LAUNCH_WINDOW, KEY_PRODUCT_VOID, + _aggregate_overlap, _artificial_demand_share, _avg_ticket_rub, _cannibalization_index, _cap_confidence, _clamp01, + _class_overlap, _competitor_strength, _cost_of_error_index, _count_void, + _geo_weight, + _haversine_km, _min_confidence, _oversupply_risk_from_deficit, + _own_portfolio_overlap, _pick_launch_window, + _price_bucket_to_band, + _price_overlap, _query_artificial_demand, + _timing_overlap, + _unit_mix_similarity, _void_index, compute_special_indices, ) +from app.services.site_finder.own_portfolio import OwnProject # Patch-таргеты — имена, импортированные В модуль special_indices. _DSF используется # точечно (re-patch поверх _full_stack_patch); остальные сервисы патчатся через @@ -257,6 +274,291 @@ class TestCannibalizationIndex: assert v is not None and 0.0 <= v <= 1.0 +# ────────────────────────────────────────────────────────────────────────────── +# §25.3 TRUE own-portfolio overlap — pure-оси (класс/цена/квартирография/тайминг/гео) +# ────────────────────────────────────────────────────────────────────────────── + + +class TestClassOverlap: + def test_same_class_full(self) -> None: + assert _class_overlap("комфорт", "комфорт") == 1.0 + + def test_case_and_language_insensitive(self) -> None: + # 'Комфорт' (Title) vs 'comfort' (EN) → один класс → 1.0 (reuse _normalize_class). + assert _class_overlap("Комфорт", "comfort") == 1.0 + assert _class_overlap("комфорт-класс", "Комфорт") == 1.0 + + def test_adjacent_class_partial(self) -> None: + # комфорт (1) ↔ комфорт+ (2) = 1 шаг → 0.5. + assert _class_overlap("комфорт", "комфорт+") == 0.5 + + def test_two_steps(self) -> None: + # комфорт (1) ↔ бизнес (3) = 2 шага → 0.2. + assert _class_overlap("комфорт", "бизнес") == 0.2 + + def test_far_class_low(self) -> None: + # эконом (0) ↔ премиум (5) = 5 шагов → far 0.05. + assert _class_overlap("эконом", "премиум") == 0.05 + + def test_unknown_class_is_none(self) -> None: + # нераспознанный класс → ось НЕДОСТУПНА (None, НЕ 0). + assert _class_overlap("комфорт", "абракадабра") is None + assert _class_overlap(None, "комфорт") is None + + +class TestPriceBucketToBand: + def test_economy_band(self) -> None: + assert _price_bucket_to_band(PRICE_BUCKET_ECONOMY) == (0.0, 120_000.0) + + def test_comfort_band(self) -> None: + assert _price_bucket_to_band(PRICE_BUCKET_COMFORT) == (120_000.0, 160_000.0) + + def test_business_band(self) -> None: + assert _price_bucket_to_band(PRICE_BUCKET_BUSINESS) == (160_000.0, 220_000.0) + + def test_premium_band_open_right(self) -> None: + band = _price_bucket_to_band(PRICE_BUCKET_PREMIUM) + assert band is not None + assert band[0] == 220_000.0 + assert band[1] == float("inf") + + def test_unknown_is_none(self) -> None: + assert _price_bucket_to_band(None) is None + assert _price_bucket_to_band("unknown") is None + + +class TestPriceOverlap: + def test_full_containment_of_narrow_band(self) -> None: + # кандидат [120k,160k] (40k), наш [100k,200k] (100k): пересечение 40k / + # min-ширина 40k = 1.0 (узкая вилка полностью внутри широкой). + assert _price_overlap((120_000.0, 160_000.0), 100_000.0, 200_000.0) == 1.0 + + def test_partial_overlap(self) -> None: + # кандидат [120k,160k] (40k), наш [140k,180k] (40k): пересечение [140k,160k]=20k + # / min-ширина 40k = 0.5. + assert _price_overlap((120_000.0, 160_000.0), 140_000.0, 180_000.0) == pytest.approx(0.5) + + def test_no_overlap_is_zero(self) -> None: + # непересекающиеся вилки → 0.0 (валидно: нет ценовой конкуренции). + assert _price_overlap((120_000.0, 160_000.0), 200_000.0, 240_000.0) == 0.0 + + def test_swapped_own_bounds_handled(self) -> None: + # наш min>max (грязь) — нормализуем порядок, результат тот же. + assert _price_overlap((120_000.0, 160_000.0), 200_000.0, 140_000.0) == pytest.approx(0.5) + + def test_premium_both_open_right_full(self) -> None: + # оба премиум (right=+inf): полное пересечение → 1.0. + assert _price_overlap((220_000.0, float("inf")), 250_000.0, float("inf")) == 1.0 + + def test_missing_candidate_band_is_none(self) -> None: + assert _price_overlap(None, 100_000.0, 200_000.0) is None + + def test_missing_own_bound_is_none(self) -> None: + # граница нашего проекта None → ось НЕДОСТУПНА (None, НЕ 0). + assert _price_overlap((120_000.0, 160_000.0), None, 200_000.0) is None + assert _price_overlap((120_000.0, 160_000.0), 100_000.0, None) is None + + +class TestUnitMixSimilarity: + def test_identical_mix_is_one(self) -> None: + mix = {"studio": 0.3, "1k": 0.4, "2k": 0.3} + assert _unit_mix_similarity(mix, dict(mix)) == pytest.approx(1.0) + + def test_disjoint_mix_is_zero(self) -> None: + # непересекающиеся ключи → L1=2 → 1−0.5·2 = 0.0. + a = {"studio": 1.0} + b = {"3k": 1.0} + assert _unit_mix_similarity(a, b) == pytest.approx(0.0) + + def test_partial_similarity(self) -> None: + # a={s:0.5,1k:0.5}, b={s:0.5,2k:0.5}: |0.5-0.5|+|0.5-0|+|0-0.5| = 1.0 → + # 1−0.5·1.0 = 0.5. + a = {"studio": 0.5, "1k": 0.5} + b = {"studio": 0.5, "2k": 0.5} + assert _unit_mix_similarity(a, b) == pytest.approx(0.5) + + def test_unnormalized_input_normalized(self) -> None: + # доли не в сумме 1 (счётчики) — нормируются на свою сумму перед сравнением. + a = {"studio": 30.0, "1k": 30.0} # → 0.5/0.5 + b = {"studio": 1.0, "1k": 1.0} # → 0.5/0.5 + assert _unit_mix_similarity(a, b) == pytest.approx(1.0) + + def test_missing_mix_is_none(self) -> None: + assert _unit_mix_similarity(None, {"studio": 1.0}) is None + assert _unit_mix_similarity({"studio": 1.0}, None) is None + assert _unit_mix_similarity({}, {"studio": 1.0}) is None + + def test_in_range(self) -> None: + v = _unit_mix_similarity({"studio": 0.7, "1k": 0.3}, {"studio": 0.2, "1k": 0.8}) + assert v is not None and 0.0 <= v <= 1.0 + + +class TestTimingOverlap: + def test_same_month_full(self) -> None: + m = date(2026, 6, 1) + assert _timing_overlap(m, m) == pytest.approx(1.0) + + def test_half_life_is_half(self) -> None: + # 12 мес расхождения = half_life → 0.5. + a = date(2026, 6, 1) + b = date(2027, 6, 1) + assert _timing_overlap(a, b) == pytest.approx(0.5) + + def test_decays_with_distance(self) -> None: + near = _timing_overlap(date(2026, 6, 1), date(2026, 9, 1)) + far = _timing_overlap(date(2026, 6, 1), date(2030, 6, 1)) + assert near is not None and far is not None + assert near > far + + def test_symmetric(self) -> None: + a, b = date(2026, 1, 1), date(2027, 1, 1) + assert _timing_overlap(a, b) == _timing_overlap(b, a) + + def test_missing_date_is_none(self) -> None: + assert _timing_overlap(None, date(2026, 6, 1)) is None + assert _timing_overlap(date(2026, 6, 1), None) is None + + def test_in_range(self) -> None: + v = _timing_overlap(date(2026, 6, 1), date(2028, 1, 1)) + assert v is not None and 0.0 <= v <= 1.0 + + +class TestGeoWeight: + def test_zero_distance_full_weight(self) -> None: + assert _geo_weight(0.0) == pytest.approx(1.0) + + def test_decays_with_distance(self) -> None: + near = _geo_weight(1.0) + far = _geo_weight(10.0) + assert near > far + assert 0.0 < far < near <= 1.0 + + def test_scale_km_at_one_e_inverse(self) -> None: + # distance == scale → exp(-1) ≈ 0.368. + import math + + assert _geo_weight(3.0) == pytest.approx(math.exp(-1.0)) + + def test_unknown_distance_floor_weight(self) -> None: + # нет координат → низкий floor (НЕ 0, НЕ 1): проект сигналит, но не доминирует. + assert _geo_weight(None) == pytest.approx(0.1) + + def test_negative_distance_clamped(self) -> None: + assert _geo_weight(-5.0) == pytest.approx(1.0) + + +# ────────────────────────────────────────────────────────────────────────────── +# §25.3 own-portfolio overlap — пара (среднее доступных осей) + агрегация (soft-max) +# ────────────────────────────────────────────────────────────────────────────── + + +class TestOwnPortfolioOverlapPair: + def test_averages_available_axes(self) -> None: + # доступны class=0.8, price=0.4 → среднее 0.6; signal = 0.6 × geo 0.5 = 0.3. + pair = _own_portfolio_overlap( + class_overlap=0.8, + price_overlap=0.4, + unit_mix_overlap=None, + timing_overlap=None, + geo_weight=0.5, + ) + assert pair.overlap == pytest.approx(0.6) + assert pair.n_axes == 2 + assert pair.signal == pytest.approx(0.3) + + def test_none_axes_excluded_not_zeroed(self) -> None: + # только class=1.0 доступен (остальные None) → overlap=1.0 (НЕ размыт нулями). + pair = _own_portfolio_overlap( + class_overlap=1.0, + price_overlap=None, + unit_mix_overlap=None, + timing_overlap=None, + geo_weight=1.0, + ) + assert pair.overlap == pytest.approx(1.0) + assert pair.n_axes == 1 + + def test_no_axes_yields_none(self) -> None: + # ни одной оси → overlap/signal None (пара неинформативна, НЕ 0). + pair = _own_portfolio_overlap( + class_overlap=None, + price_overlap=None, + unit_mix_overlap=None, + timing_overlap=None, + geo_weight=1.0, + ) + assert pair.overlap is None + assert pair.signal is None + assert pair.n_axes == 0 + + def test_all_four_axes(self) -> None: + pair = _own_portfolio_overlap( + class_overlap=1.0, + price_overlap=0.5, + unit_mix_overlap=0.5, + timing_overlap=0.0, + geo_weight=1.0, + ) + assert pair.overlap == pytest.approx(0.5) + assert pair.n_axes == 4 + + +class TestAggregateOverlap: + def test_takes_max_signal_not_mean(self) -> None: + # один сильный (0.9) + два слабых (0.1) → агрегат = 0.9 (soft-max), не среднее. + strong = _own_portfolio_overlap( + class_overlap=0.9, price_overlap=0.9, + unit_mix_overlap=None, timing_overlap=None, geo_weight=1.0, + ) + weak1 = _own_portfolio_overlap( + class_overlap=0.1, price_overlap=0.1, + unit_mix_overlap=None, timing_overlap=None, geo_weight=1.0, + ) + weak2 = _own_portfolio_overlap( + class_overlap=0.1, price_overlap=0.1, + unit_mix_overlap=None, timing_overlap=None, geo_weight=1.0, + ) + assert _aggregate_overlap([strong, weak1, weak2]) == pytest.approx(0.9) + + def test_geo_weight_attenuates_distant_strong_overlap(self) -> None: + # сильное пересечение, но далеко (geo 0.1) → signal 0.05; близкое слабое + # (overlap 0.3, geo 1.0 → 0.3) перебивает. Агрегат = 0.3. + distant_strong = _own_portfolio_overlap( + class_overlap=1.0, price_overlap=1.0, + unit_mix_overlap=None, timing_overlap=None, geo_weight=0.1, + ) + near_weak = _own_portfolio_overlap( + class_overlap=0.3, price_overlap=0.3, + unit_mix_overlap=None, timing_overlap=None, geo_weight=1.0, + ) + assert _aggregate_overlap([distant_strong, near_weak]) == pytest.approx(0.3) + + def test_all_none_signals_is_none(self) -> None: + empty = _own_portfolio_overlap( + class_overlap=None, price_overlap=None, + unit_mix_overlap=None, timing_overlap=None, geo_weight=1.0, + ) + assert _aggregate_overlap([empty, empty]) is None + + def test_empty_is_none(self) -> None: + assert _aggregate_overlap([]) is None + + +class TestHaversine: + def test_zero_distance(self) -> None: + assert _haversine_km(60.6, 56.8, 60.6, 56.8) == pytest.approx(0.0, abs=1e-9) + + def test_known_distance_ekb_scale(self) -> None: + # ~1 км по долготе на широте ЕКБ (56.8°): 0.0164° lon ≈ 1 км. Допуск широкий. + d = _haversine_km(60.6, 56.8, 60.6164, 56.8) + assert 0.9 < d < 1.1 + + def test_symmetric(self) -> None: + a = _haversine_km(60.6, 56.8, 60.7, 56.9) + b = _haversine_km(60.7, 56.9, 60.6, 56.8) + assert a == pytest.approx(b) + + # ────────────────────────────────────────────────────────────────────────────── # Pure: Artificial Demand (доля ипотеки) # ────────────────────────────────────────────────────────────────────────────── @@ -877,3 +1179,241 @@ class TestComputeSpecialIndicesGraceful: assert captured["obj_class"] is None assert captured["district"] is None assert card.indices[KEY_ARTIFICIAL_DEMAND].value == pytest.approx(0.25) + + +# ────────────────────────────────────────────────────────────────────────────── +# §25.3 Cannibalization dispatch — TRUE own-portfolio vs PROXY fallback +# ────────────────────────────────────────────────────────────────────────────── + +# Центроид участка для гео-веса (lon, lat) в окрестностях ЕКБ. +_CENTROID = (60.6000, 56.8000) +_CAND_SPEC = SegmentSpec( + obj_class="комфорт", room_bucket="2-к 45-60", district="Академический", + price_bucket=PRICE_BUCKET_COMFORT, +) + + +def _own( + name: str, + *, + source: str = "future", + obj_class: str | None = "комфорт", + price_min: float | None = 120_000.0, + price_max: float | None = 160_000.0, + unit_mix: dict[str, float] | None = None, + release_month: date | None = None, + lon: float | None = None, + lat: float | None = None, +) -> OwnProject: + return OwnProject( + name=name, + source=source, # type: ignore[arg-type] + obj_class=obj_class, + release_month=release_month, + price_min_per_m2=price_min, + price_max_per_m2=price_max, + unit_mix=unit_mix, + district="Академический", + lon=lon, + lat=lat, + ) + + +def _cannibalization_card( + portfolio: list[OwnProject], + *, + centroid: tuple[float, float] | None = _CENTROID, + cad_num: str | None = "66:41:0303161:123", +) -> Any: + """Прогнать compute_special_indices с замоканными own-portfolio + центроидом.""" + db = MagicMock() + with ( + _full_stack_patch(), + patch(f"{_MOD}.get_own_portfolio", return_value=portfolio), + patch(f"{_MOD}._query_parcel_centroid", return_value=centroid), + patch( + f"{_MOD}._query_artificial_demand", + return_value={"n_sold": 100, "n_mortgage": 40}, + ), + ): + return compute_special_indices( + db, spec=_CAND_SPEC, district="Академический", cad_num=cad_num + ) + + +class TestCannibalizationTrueMode: + def test_nonempty_portfolio_uses_own_portfolio_mode(self) -> None: + # наш проект на участке (distance 0 → geo 1.0), класс/цена совпадают → overlap 1.0. + portfolio = [ + _own("Наш-А", lon=_CENTROID[0], lat=_CENTROID[1]), + _own("Наш-Б", obj_class="бизнес", price_min=200_000.0, price_max=240_000.0, + lon=60.9, lat=57.1), + ] + card = _cannibalization_card(portfolio) + can = card.indices[KEY_CANNIBALIZATION] + assert can.method == "own_portfolio_overlap" + assert can.detail["mode"] == "own_portfolio" + assert can.detail["proxy"] is False + # сильнейший каннибализатор = Наш-А (class 1.0 + price 1.0)/2 × geo 1.0 = 1.0. + assert can.value == pytest.approx(1.0) + + def test_top_contributor_surfaced(self) -> None: + portfolio = [ + _own("Наш-А", lon=_CENTROID[0], lat=_CENTROID[1]), + _own("Наш-Б", obj_class="бизнес", price_min=200_000.0, price_max=240_000.0, + lon=60.9, lat=57.1), + ] + card = _cannibalization_card(portfolio) + can = card.indices[KEY_CANNIBALIZATION] + top = can.detail["top_contributors"] + assert top[0]["name"] == "Наш-А" + assert "Наш-А" in (can.label or "") + # explainability: пер-ось breakdown присутствует. + assert top[0]["axes"]["class"] == pytest.approx(1.0) + assert top[0]["axes"]["price"] == pytest.approx(1.0) + # квартирография/тайминг недоступны из spec → None (НЕ 0). + assert top[0]["axes"]["unit_mix"] is None + assert top[0]["axes"]["timing"] is None + + def test_geo_weight_attenuates_distant_project(self) -> None: + # тот же сильный overlap, но проект далеко → значение заметно ниже 1.0. + near = _cannibalization_card([_own("Близкий", lon=_CENTROID[0], lat=_CENTROID[1])]) + far = _cannibalization_card([_own("Далёкий", lon=61.5, lat=57.5)]) + near_v = near.indices[KEY_CANNIBALIZATION].value + far_v = far.indices[KEY_CANNIBALIZATION].value + assert near_v is not None and far_v is not None + assert near_v > far_v + + def test_no_geometry_still_computes_via_other_axes(self) -> None: + # нет центроида → гео-вес на floor (0.1), но класс+цена дают overlap → индекс есть. + card = _cannibalization_card([_own("Без-гео")], centroid=None) + can = card.indices[KEY_CANNIBALIZATION] + assert can.method == "own_portfolio_overlap" + assert can.detail["has_geometry"] is False + # overlap 1.0 × geo floor 0.1 = 0.1. + assert can.value == pytest.approx(0.1) + + def test_unit_mix_and_timing_axes_when_supplied(self) -> None: + # caller передаёт квартирографию/тайминг кандидата → оси становятся доступны. + db = MagicMock() + own_proj = _own( + "Наш-А", lon=_CENTROID[0], lat=_CENTROID[1], + unit_mix={"studio": 0.5, "1k": 0.5}, release_month=date(2026, 6, 1), + ) + with ( + patch(f"{_MOD}.get_own_portfolio", return_value=[own_proj]), + patch(f"{_MOD}._query_parcel_centroid", return_value=_CENTROID), + ): + from app.services.forecasting.special_indices import _build_cannibalization + + idx = _build_cannibalization( + db, spec=_CAND_SPEC, cad_num="66:41:0303161:123", + candidate_unit_mix={"studio": 0.5, "1k": 0.5}, + candidate_release_month=date(2026, 6, 1), + ) + top = idx.detail["top_contributors"][0] + assert top["axes"]["unit_mix"] == pytest.approx(1.0) # одинаковый микс + assert top["axes"]["timing"] == pytest.approx(1.0) # одинаковый месяц + assert top["n_axes"] == 4 + + def test_none_not_zero_when_no_comparable_axes(self) -> None: + # наш проект без класса И без цены → ни класс, ни цена не считаются; + # тайминг/квартирография тоже None → пара без осей → индекс unavailable (НЕ 0). + card = _cannibalization_card( + [_own("Пустой", obj_class=None, price_min=None, price_max=None, + lon=_CENTROID[0], lat=_CENTROID[1])] + ) + can = card.indices[KEY_CANNIBALIZATION] + assert can.value is None + assert can.method == _METHOD_UNAVAILABLE + assert "фабрикуем" in can.detail["reason"] + + def test_axes_available_summary_excludes_missing(self) -> None: + # 1 проект, доступны только class+price → summary это отражает (unit_mix/timing 0). + card = _cannibalization_card([_own("Наш-А", lon=_CENTROID[0], lat=_CENTROID[1])]) + summary = card.indices[KEY_CANNIBALIZATION].detail["axes_available"] + assert summary["class"] == 1 + assert summary["price"] == 1 + assert summary["unit_mix"] == 0 + assert summary["timing"] == 0 + + +class TestCannibalizationThinPortfolioConfidence: + def test_only_current_is_low_with_note(self) -> None: + # портфель только из current (нет future-пайплайна) → honest low + нота §26. + portfolio = [ + _own("Текущий-1", source="current", lon=_CENTROID[0], lat=_CENTROID[1]), + _own("Текущий-2", source="current", lon=_CENTROID[0], lat=_CENTROID[1]), + _own("Текущий-3", source="current", lon=_CENTROID[0], lat=_CENTROID[1]), + ] + can = _cannibalization_card(portfolio).indices[KEY_CANNIBALIZATION] + assert can.confidence == "low" + assert can.detail["confidence_note"] is not None + assert "future" in can.detail["confidence_note"] + + def test_single_project_is_low_data_scarce(self) -> None: + can = _cannibalization_card( + [_own("Один", lon=_CENTROID[0], lat=_CENTROID[1])] + ).indices[KEY_CANNIBALIZATION] + assert can.confidence == "low" + assert "данных мало" in can.detail["confidence_note"] + + def test_healthy_portfolio_medium(self) -> None: + # ≥2 проекта И есть future → medium (всё равно ≤ cap). + portfolio = [ + _own("Будущий-1", source="future", lon=_CENTROID[0], lat=_CENTROID[1]), + _own("Текущий-1", source="current", lon=_CENTROID[0], lat=_CENTROID[1]), + ] + can = _cannibalization_card(portfolio).indices[KEY_CANNIBALIZATION] + assert can.confidence == "medium" + assert can.detail["confidence_note"] is None + + +class TestCannibalizationProxyFallback: + def test_empty_portfolio_falls_back_to_proxy(self) -> None: + # own-portfolio пуст → ПРОКСИ, явно помеченный. + can = _cannibalization_card([]).indices[KEY_CANNIBALIZATION] + assert can.method == "proxy_same_class_relevance_share" + assert can.detail["mode"] == "proxy" + assert can.detail["proxy"] is True + # каведат §26: прокси НЕ выдаётся за истинный индекс. + assert "прокси" in can.detail["proxy_reason"].lower() + assert "own-portfolio" in can.detail["proxy_reason"] + + def test_proxy_value_matches_same_class_share(self) -> None: + # competitors из _full_stack_patch: same-class (комфорт) 0.8+0.6=1.4; all=1.9. + can = _cannibalization_card([]).indices[KEY_CANNIBALIZATION] + assert can.value == pytest.approx((0.8 + 0.6) / (0.8 + 0.6 + 0.5)) + + def test_proxy_label_marked(self) -> None: + can = _cannibalization_card([]).indices[KEY_CANNIBALIZATION] + assert "прокси" in (can.label or "") + + def test_proxy_no_cad_num_unavailable(self) -> None: + # пустой портфель + нет cad_num → прокси не определить → unavailable. + can = _cannibalization_card([], cad_num=None).indices[KEY_CANNIBALIZATION] + assert can.value is None + assert can.method == _METHOD_UNAVAILABLE + + +class TestCannibalizationDeterminism: + def test_same_inputs_identical_output(self) -> None: + portfolio = [ + _own("Наш-А", lon=_CENTROID[0], lat=_CENTROID[1]), + _own("Наш-Б", obj_class="бизнес", price_min=200_000.0, price_max=240_000.0, + lon=60.9, lat=57.1), + _own("Наш-В", obj_class="комфорт+", lon=60.65, lat=56.85), + ] + first = _cannibalization_card(list(portfolio)).indices[KEY_CANNIBALIZATION].as_dict() + second = _cannibalization_card(list(portfolio)).indices[KEY_CANNIBALIZATION].as_dict() + assert first == second + + def test_top_contributor_tie_break_by_name(self) -> None: + # два проекта с ИДЕНТИЧНЫМ сигналом → детерминированный tie-break по имени (А раньше Я). + portfolio = [ + _own("Яков", lon=_CENTROID[0], lat=_CENTROID[1]), + _own("Андрей", lon=_CENTROID[0], lat=_CENTROID[1]), + ] + can = _cannibalization_card(portfolio).indices[KEY_CANNIBALIZATION] + names = [c["name"] for c in can.detail["top_contributors"]] + assert names == ["Андрей", "Яков"]