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. где дефицит ПИКОВЫЙ = рекомендованное окно запуска. SOLID.
2. Product Void #981 rank_segments: сегменты с deficit_index ≥ _VOID_THRESHOLD 2. Product Void #981 rank_segments: сегменты с deficit_index ≥ _VOID_THRESHOLD
(белое пятно недообеспеченный спрос). QUICK-WIN, SOLID. (белое пятно недообеспеченный спрос). QUICK-WIN, SOLID.
3. Cannibalization §9.1 competitors: доля конкурентов В ТОМ ЖЕ классе, взвешенная 3. Cannibalization §25.3 ИСТИННОЕ пересечение рек. сегмента-кандидата с НАШИМ
relevance пересечение рек. сегмента с чужим/своим стоком. портфелем (own_portfolio #1169) по 4 осям (класс/аудитория,
DEGRADE-to-None, если нет геометрии cad_num. цена /м², квартирография, тайминг) + гео-вес близости наших
проектов к участку. Портфель пуст (own_developer_ids не задан И
нет manual-строк) FALLBACK на ПРОКСИ (доля конкурентов того же
класса, явно помеченный). DEGRADE-to-None, если нет геометрии.
4. Competitor Strength §9.1 competitors: среднее relevance_weight топ-N. DEGRADE, 4. Competitor Strength §9.1 competitors: среднее relevance_weight топ-N. DEGRADE,
если нет cad_num/геометрии. если нет cad_num/геометрии.
5. Artificial Demand НОВЫЙ read objective_lots: доля проданных лотов с ипотекой 5. Artificial Demand НОВЫЙ read objective_lots: доля проданных лотов с ипотекой
@ -38,8 +41,10 @@ value=None, method='unavailable', confidence='low', а остальные пят
from __future__ import annotations from __future__ import annotations
import logging import logging
import math
from collections.abc import Sequence from collections.abc import Sequence
from dataclasses import dataclass from dataclasses import dataclass
from datetime import date
from typing import Any, Literal from typing import Any, Literal
from sqlalchemy import text from sqlalchemy import text
@ -48,10 +53,20 @@ from sqlalchemy.orm import Session
from app.schemas.parcel import CompetitorsRequest from app.schemas.parcel import CompetitorsRequest
from app.services.forecasting.affordability import compute_affordability from app.services.forecasting.affordability import compute_affordability
from app.services.forecasting.demand_supply_forecast import compute_demand_supply_forecast 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.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.market_metrics import compute_market_metrics
from app.services.site_finder.own_portfolio import OwnProject, get_own_portfolio
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -102,6 +117,40 @@ _COMPETITOR_RADIUS_KM: float = 1.0
# конкурент, распроданный к нашему запуску, менее релевантен. 12 мес — дефолт. # конкурент, распроданный к нашему запуску, менее релевантен. 12 мес — дефолт.
_COMPETITOR_HORIZON_MONTHS: int = 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]: при каком # Saturation для нормализации пикового дефицита Launch Window в [0,1]: при каком
# deficit_index сила окна достигает 1.0. 0.5 = половина шкалы #980 уже «сильное окно». # deficit_index сила окна достигает 1.0. 0.5 = половина шкалы #980 уже «сильное окно».
# Линейная отсечка max(0, min(1, peak / saturation)). Tunable. # Линейная отсечка max(0, min(1, peak / saturation)). Tunable.
@ -339,6 +388,238 @@ def _cannibalization_index(
return _clamp01(same / total) 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: def _artificial_demand_share(n_mortgage: int | None, n_sold: int | None) -> float | None:
"""Доля проданных лотов, профинансированных ипотекой (Artificial Demand). PURE. """Доля проданных лотов, профинансированных ипотекой (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-математика выше тестируется без БД. # 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: # Метка-каведат для проксированной каннибализации (own-portfolio не сконфигурирован).
"""Index 3 — Cannibalization: доля relevance-силы конкурентов В классе. Degrade.""" _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: if not cad_num:
return _unavailable( return _unavailable(
KEY_CANNIBALIZATION, 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) competitors = _fetch_competitors(db, cad_num=cad_num)
all_relevance = [c.relevance_weight for c in competitors] 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 = [ same_class_relevance = [
c.relevance_weight c.relevance_weight
for c in competitors 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) value = _cannibalization_index(same_class_relevance, all_relevance)
if value is None: if value is None:
return _unavailable( return _unavailable(
KEY_CANNIBALIZATION, KEY_CANNIBALIZATION,
reason="нет конкурентов в радиусе — пересечения нет", reason="нет конкурентов в радиусе — пересечения нет (прокси)",
) )
return SpecialIndex( return SpecialIndex(
key=KEY_CANNIBALIZATION, key=KEY_CANNIBALIZATION,
value=value, value=value,
label=spec.obj_class, label=f"прокси · {spec.obj_class}" if spec.obj_class else "прокси",
confidence=_cap_confidence("medium"), confidence=_cap_confidence("low"), # прокси — заведомо слабее истинного
detail={ detail={
"mode": "proxy",
"proxy": True,
"proxy_reason": _PROXY_REASON,
"n_competitors": len(competitors), "n_competitors": len(competitors),
"n_same_class": len(same_class_relevance), "n_same_class": len(same_class_relevance),
"target_class": spec.obj_class, "target_class": spec.obj_class,
"radius_km": _COMPETITOR_RADIUS_KM, "radius_km": _COMPETITOR_RADIUS_KM,
}, },
method="same_class_relevance_share", method="proxy_same_class_relevance_share",
advisory=_ADVISORY, 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( def _build_artificial_demand(
db: Session, *, spec: SegmentSpec, district: str | None, premise_kind: str db: Session, *, spec: SegmentSpec, district: str | None, premise_kind: str
) -> SpecialIndex: ) -> SpecialIndex:

View file

@ -16,12 +16,19 @@ test_market_metrics.
from __future__ import annotations from __future__ import annotations
from datetime import date
from typing import Any from typing import Any
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
import pytest 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 ( from app.services.forecasting.special_indices import (
_INDEX_KEYS, _INDEX_KEYS,
_METHOD_UNAVAILABLE, _METHOD_UNAVAILABLE,
@ -31,21 +38,31 @@ from app.services.forecasting.special_indices import (
KEY_COST_OF_ERROR, KEY_COST_OF_ERROR,
KEY_LAUNCH_WINDOW, KEY_LAUNCH_WINDOW,
KEY_PRODUCT_VOID, KEY_PRODUCT_VOID,
_aggregate_overlap,
_artificial_demand_share, _artificial_demand_share,
_avg_ticket_rub, _avg_ticket_rub,
_cannibalization_index, _cannibalization_index,
_cap_confidence, _cap_confidence,
_clamp01, _clamp01,
_class_overlap,
_competitor_strength, _competitor_strength,
_cost_of_error_index, _cost_of_error_index,
_count_void, _count_void,
_geo_weight,
_haversine_km,
_min_confidence, _min_confidence,
_oversupply_risk_from_deficit, _oversupply_risk_from_deficit,
_own_portfolio_overlap,
_pick_launch_window, _pick_launch_window,
_price_bucket_to_band,
_price_overlap,
_query_artificial_demand, _query_artificial_demand,
_timing_overlap,
_unit_mix_similarity,
_void_index, _void_index,
compute_special_indices, compute_special_indices,
) )
from app.services.site_finder.own_portfolio import OwnProject
# Patch-таргеты — имена, импортированные В модуль special_indices. _DSF используется # Patch-таргеты — имена, импортированные В модуль special_indices. _DSF используется
# точечно (re-patch поверх _full_stack_patch); остальные сервисы патчатся через # точечно (re-patch поверх _full_stack_patch); остальные сервисы патчатся через
@ -257,6 +274,291 @@ class TestCannibalizationIndex:
assert v is not None and 0.0 <= v <= 1.0 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 (доля ипотеки) # Pure: Artificial Demand (доля ипотеки)
# ────────────────────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────────────────────
@ -877,3 +1179,241 @@ class TestComputeSpecialIndicesGraceful:
assert captured["obj_class"] is None assert captured["obj_class"] is None
assert captured["district"] is None assert captured["district"] is None
assert card.indices[KEY_ARTIFICIAL_DEMAND].value == pytest.approx(0.25) 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 == ["Андрей", "Яков"]