feat(forecasting): §25.3 TRUE own-portfolio cannibalization overlap (#1169 PR2)
All checks were successful
CI / changes (push) Successful in 6s
CI / frontend-tests (push) Has been skipped
CI / changes (pull_request) Successful in 6s
CI / backend-tests (push) Successful in 6m26s
CI / backend-tests (pull_request) Successful in 6m21s
Deploy / changes (push) Successful in 6s
Deploy / build-worker (push) Successful in 3m10s
Deploy / deploy (push) Successful in 1m41s
CI / frontend-tests (pull_request) Has been skipped
Deploy / build-frontend (push) Has been skipped
Deploy / build-backend (push) Successful in 1m47s

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
This commit is contained in:
Light1YT 2026-06-08 16:47:18 +05:00
parent df34e55ab4
commit 347203dfda
2 changed files with 1109 additions and 14 deletions

View file

@ -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:

View file

@ -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 → 10.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 →
# 10.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 == ["Андрей", "Яков"]