feat(forecasting): §25.3 TRUE own-portfolio cannibalization overlap (#1169 PR2) #1171
2 changed files with 1109 additions and 14 deletions
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -16,12 +16,19 @@ test_market_metrics.
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.services.forecasting.sales_series import SegmentSpec
|
||||
from app.services.forecasting.sales_series import (
|
||||
PRICE_BUCKET_BUSINESS,
|
||||
PRICE_BUCKET_COMFORT,
|
||||
PRICE_BUCKET_ECONOMY,
|
||||
PRICE_BUCKET_PREMIUM,
|
||||
SegmentSpec,
|
||||
)
|
||||
from app.services.forecasting.special_indices import (
|
||||
_INDEX_KEYS,
|
||||
_METHOD_UNAVAILABLE,
|
||||
|
|
@ -31,21 +38,31 @@ from app.services.forecasting.special_indices import (
|
|||
KEY_COST_OF_ERROR,
|
||||
KEY_LAUNCH_WINDOW,
|
||||
KEY_PRODUCT_VOID,
|
||||
_aggregate_overlap,
|
||||
_artificial_demand_share,
|
||||
_avg_ticket_rub,
|
||||
_cannibalization_index,
|
||||
_cap_confidence,
|
||||
_clamp01,
|
||||
_class_overlap,
|
||||
_competitor_strength,
|
||||
_cost_of_error_index,
|
||||
_count_void,
|
||||
_geo_weight,
|
||||
_haversine_km,
|
||||
_min_confidence,
|
||||
_oversupply_risk_from_deficit,
|
||||
_own_portfolio_overlap,
|
||||
_pick_launch_window,
|
||||
_price_bucket_to_band,
|
||||
_price_overlap,
|
||||
_query_artificial_demand,
|
||||
_timing_overlap,
|
||||
_unit_mix_similarity,
|
||||
_void_index,
|
||||
compute_special_indices,
|
||||
)
|
||||
from app.services.site_finder.own_portfolio import OwnProject
|
||||
|
||||
# Patch-таргеты — имена, импортированные В модуль special_indices. _DSF используется
|
||||
# точечно (re-patch поверх _full_stack_patch); остальные сервисы патчатся через
|
||||
|
|
@ -257,6 +274,291 @@ class TestCannibalizationIndex:
|
|||
assert v is not None and 0.0 <= v <= 1.0
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# §25.3 TRUE own-portfolio overlap — pure-оси (класс/цена/квартирография/тайминг/гео)
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestClassOverlap:
|
||||
def test_same_class_full(self) -> None:
|
||||
assert _class_overlap("комфорт", "комфорт") == 1.0
|
||||
|
||||
def test_case_and_language_insensitive(self) -> None:
|
||||
# 'Комфорт' (Title) vs 'comfort' (EN) → один класс → 1.0 (reuse _normalize_class).
|
||||
assert _class_overlap("Комфорт", "comfort") == 1.0
|
||||
assert _class_overlap("комфорт-класс", "Комфорт") == 1.0
|
||||
|
||||
def test_adjacent_class_partial(self) -> None:
|
||||
# комфорт (1) ↔ комфорт+ (2) = 1 шаг → 0.5.
|
||||
assert _class_overlap("комфорт", "комфорт+") == 0.5
|
||||
|
||||
def test_two_steps(self) -> None:
|
||||
# комфорт (1) ↔ бизнес (3) = 2 шага → 0.2.
|
||||
assert _class_overlap("комфорт", "бизнес") == 0.2
|
||||
|
||||
def test_far_class_low(self) -> None:
|
||||
# эконом (0) ↔ премиум (5) = 5 шагов → far 0.05.
|
||||
assert _class_overlap("эконом", "премиум") == 0.05
|
||||
|
||||
def test_unknown_class_is_none(self) -> None:
|
||||
# нераспознанный класс → ось НЕДОСТУПНА (None, НЕ 0).
|
||||
assert _class_overlap("комфорт", "абракадабра") is None
|
||||
assert _class_overlap(None, "комфорт") is None
|
||||
|
||||
|
||||
class TestPriceBucketToBand:
|
||||
def test_economy_band(self) -> None:
|
||||
assert _price_bucket_to_band(PRICE_BUCKET_ECONOMY) == (0.0, 120_000.0)
|
||||
|
||||
def test_comfort_band(self) -> None:
|
||||
assert _price_bucket_to_band(PRICE_BUCKET_COMFORT) == (120_000.0, 160_000.0)
|
||||
|
||||
def test_business_band(self) -> None:
|
||||
assert _price_bucket_to_band(PRICE_BUCKET_BUSINESS) == (160_000.0, 220_000.0)
|
||||
|
||||
def test_premium_band_open_right(self) -> None:
|
||||
band = _price_bucket_to_band(PRICE_BUCKET_PREMIUM)
|
||||
assert band is not None
|
||||
assert band[0] == 220_000.0
|
||||
assert band[1] == float("inf")
|
||||
|
||||
def test_unknown_is_none(self) -> None:
|
||||
assert _price_bucket_to_band(None) is None
|
||||
assert _price_bucket_to_band("unknown") is None
|
||||
|
||||
|
||||
class TestPriceOverlap:
|
||||
def test_full_containment_of_narrow_band(self) -> None:
|
||||
# кандидат [120k,160k] (40k), наш [100k,200k] (100k): пересечение 40k /
|
||||
# min-ширина 40k = 1.0 (узкая вилка полностью внутри широкой).
|
||||
assert _price_overlap((120_000.0, 160_000.0), 100_000.0, 200_000.0) == 1.0
|
||||
|
||||
def test_partial_overlap(self) -> None:
|
||||
# кандидат [120k,160k] (40k), наш [140k,180k] (40k): пересечение [140k,160k]=20k
|
||||
# / min-ширина 40k = 0.5.
|
||||
assert _price_overlap((120_000.0, 160_000.0), 140_000.0, 180_000.0) == pytest.approx(0.5)
|
||||
|
||||
def test_no_overlap_is_zero(self) -> None:
|
||||
# непересекающиеся вилки → 0.0 (валидно: нет ценовой конкуренции).
|
||||
assert _price_overlap((120_000.0, 160_000.0), 200_000.0, 240_000.0) == 0.0
|
||||
|
||||
def test_swapped_own_bounds_handled(self) -> None:
|
||||
# наш min>max (грязь) — нормализуем порядок, результат тот же.
|
||||
assert _price_overlap((120_000.0, 160_000.0), 200_000.0, 140_000.0) == pytest.approx(0.5)
|
||||
|
||||
def test_premium_both_open_right_full(self) -> None:
|
||||
# оба премиум (right=+inf): полное пересечение → 1.0.
|
||||
assert _price_overlap((220_000.0, float("inf")), 250_000.0, float("inf")) == 1.0
|
||||
|
||||
def test_missing_candidate_band_is_none(self) -> None:
|
||||
assert _price_overlap(None, 100_000.0, 200_000.0) is None
|
||||
|
||||
def test_missing_own_bound_is_none(self) -> None:
|
||||
# граница нашего проекта None → ось НЕДОСТУПНА (None, НЕ 0).
|
||||
assert _price_overlap((120_000.0, 160_000.0), None, 200_000.0) is None
|
||||
assert _price_overlap((120_000.0, 160_000.0), 100_000.0, None) is None
|
||||
|
||||
|
||||
class TestUnitMixSimilarity:
|
||||
def test_identical_mix_is_one(self) -> None:
|
||||
mix = {"studio": 0.3, "1k": 0.4, "2k": 0.3}
|
||||
assert _unit_mix_similarity(mix, dict(mix)) == pytest.approx(1.0)
|
||||
|
||||
def test_disjoint_mix_is_zero(self) -> None:
|
||||
# непересекающиеся ключи → L1=2 → 1−0.5·2 = 0.0.
|
||||
a = {"studio": 1.0}
|
||||
b = {"3k": 1.0}
|
||||
assert _unit_mix_similarity(a, b) == pytest.approx(0.0)
|
||||
|
||||
def test_partial_similarity(self) -> None:
|
||||
# a={s:0.5,1k:0.5}, b={s:0.5,2k:0.5}: |0.5-0.5|+|0.5-0|+|0-0.5| = 1.0 →
|
||||
# 1−0.5·1.0 = 0.5.
|
||||
a = {"studio": 0.5, "1k": 0.5}
|
||||
b = {"studio": 0.5, "2k": 0.5}
|
||||
assert _unit_mix_similarity(a, b) == pytest.approx(0.5)
|
||||
|
||||
def test_unnormalized_input_normalized(self) -> None:
|
||||
# доли не в сумме 1 (счётчики) — нормируются на свою сумму перед сравнением.
|
||||
a = {"studio": 30.0, "1k": 30.0} # → 0.5/0.5
|
||||
b = {"studio": 1.0, "1k": 1.0} # → 0.5/0.5
|
||||
assert _unit_mix_similarity(a, b) == pytest.approx(1.0)
|
||||
|
||||
def test_missing_mix_is_none(self) -> None:
|
||||
assert _unit_mix_similarity(None, {"studio": 1.0}) is None
|
||||
assert _unit_mix_similarity({"studio": 1.0}, None) is None
|
||||
assert _unit_mix_similarity({}, {"studio": 1.0}) is None
|
||||
|
||||
def test_in_range(self) -> None:
|
||||
v = _unit_mix_similarity({"studio": 0.7, "1k": 0.3}, {"studio": 0.2, "1k": 0.8})
|
||||
assert v is not None and 0.0 <= v <= 1.0
|
||||
|
||||
|
||||
class TestTimingOverlap:
|
||||
def test_same_month_full(self) -> None:
|
||||
m = date(2026, 6, 1)
|
||||
assert _timing_overlap(m, m) == pytest.approx(1.0)
|
||||
|
||||
def test_half_life_is_half(self) -> None:
|
||||
# 12 мес расхождения = half_life → 0.5.
|
||||
a = date(2026, 6, 1)
|
||||
b = date(2027, 6, 1)
|
||||
assert _timing_overlap(a, b) == pytest.approx(0.5)
|
||||
|
||||
def test_decays_with_distance(self) -> None:
|
||||
near = _timing_overlap(date(2026, 6, 1), date(2026, 9, 1))
|
||||
far = _timing_overlap(date(2026, 6, 1), date(2030, 6, 1))
|
||||
assert near is not None and far is not None
|
||||
assert near > far
|
||||
|
||||
def test_symmetric(self) -> None:
|
||||
a, b = date(2026, 1, 1), date(2027, 1, 1)
|
||||
assert _timing_overlap(a, b) == _timing_overlap(b, a)
|
||||
|
||||
def test_missing_date_is_none(self) -> None:
|
||||
assert _timing_overlap(None, date(2026, 6, 1)) is None
|
||||
assert _timing_overlap(date(2026, 6, 1), None) is None
|
||||
|
||||
def test_in_range(self) -> None:
|
||||
v = _timing_overlap(date(2026, 6, 1), date(2028, 1, 1))
|
||||
assert v is not None and 0.0 <= v <= 1.0
|
||||
|
||||
|
||||
class TestGeoWeight:
|
||||
def test_zero_distance_full_weight(self) -> None:
|
||||
assert _geo_weight(0.0) == pytest.approx(1.0)
|
||||
|
||||
def test_decays_with_distance(self) -> None:
|
||||
near = _geo_weight(1.0)
|
||||
far = _geo_weight(10.0)
|
||||
assert near > far
|
||||
assert 0.0 < far < near <= 1.0
|
||||
|
||||
def test_scale_km_at_one_e_inverse(self) -> None:
|
||||
# distance == scale → exp(-1) ≈ 0.368.
|
||||
import math
|
||||
|
||||
assert _geo_weight(3.0) == pytest.approx(math.exp(-1.0))
|
||||
|
||||
def test_unknown_distance_floor_weight(self) -> None:
|
||||
# нет координат → низкий floor (НЕ 0, НЕ 1): проект сигналит, но не доминирует.
|
||||
assert _geo_weight(None) == pytest.approx(0.1)
|
||||
|
||||
def test_negative_distance_clamped(self) -> None:
|
||||
assert _geo_weight(-5.0) == pytest.approx(1.0)
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# §25.3 own-portfolio overlap — пара (среднее доступных осей) + агрегация (soft-max)
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestOwnPortfolioOverlapPair:
|
||||
def test_averages_available_axes(self) -> None:
|
||||
# доступны class=0.8, price=0.4 → среднее 0.6; signal = 0.6 × geo 0.5 = 0.3.
|
||||
pair = _own_portfolio_overlap(
|
||||
class_overlap=0.8,
|
||||
price_overlap=0.4,
|
||||
unit_mix_overlap=None,
|
||||
timing_overlap=None,
|
||||
geo_weight=0.5,
|
||||
)
|
||||
assert pair.overlap == pytest.approx(0.6)
|
||||
assert pair.n_axes == 2
|
||||
assert pair.signal == pytest.approx(0.3)
|
||||
|
||||
def test_none_axes_excluded_not_zeroed(self) -> None:
|
||||
# только class=1.0 доступен (остальные None) → overlap=1.0 (НЕ размыт нулями).
|
||||
pair = _own_portfolio_overlap(
|
||||
class_overlap=1.0,
|
||||
price_overlap=None,
|
||||
unit_mix_overlap=None,
|
||||
timing_overlap=None,
|
||||
geo_weight=1.0,
|
||||
)
|
||||
assert pair.overlap == pytest.approx(1.0)
|
||||
assert pair.n_axes == 1
|
||||
|
||||
def test_no_axes_yields_none(self) -> None:
|
||||
# ни одной оси → overlap/signal None (пара неинформативна, НЕ 0).
|
||||
pair = _own_portfolio_overlap(
|
||||
class_overlap=None,
|
||||
price_overlap=None,
|
||||
unit_mix_overlap=None,
|
||||
timing_overlap=None,
|
||||
geo_weight=1.0,
|
||||
)
|
||||
assert pair.overlap is None
|
||||
assert pair.signal is None
|
||||
assert pair.n_axes == 0
|
||||
|
||||
def test_all_four_axes(self) -> None:
|
||||
pair = _own_portfolio_overlap(
|
||||
class_overlap=1.0,
|
||||
price_overlap=0.5,
|
||||
unit_mix_overlap=0.5,
|
||||
timing_overlap=0.0,
|
||||
geo_weight=1.0,
|
||||
)
|
||||
assert pair.overlap == pytest.approx(0.5)
|
||||
assert pair.n_axes == 4
|
||||
|
||||
|
||||
class TestAggregateOverlap:
|
||||
def test_takes_max_signal_not_mean(self) -> None:
|
||||
# один сильный (0.9) + два слабых (0.1) → агрегат = 0.9 (soft-max), не среднее.
|
||||
strong = _own_portfolio_overlap(
|
||||
class_overlap=0.9, price_overlap=0.9,
|
||||
unit_mix_overlap=None, timing_overlap=None, geo_weight=1.0,
|
||||
)
|
||||
weak1 = _own_portfolio_overlap(
|
||||
class_overlap=0.1, price_overlap=0.1,
|
||||
unit_mix_overlap=None, timing_overlap=None, geo_weight=1.0,
|
||||
)
|
||||
weak2 = _own_portfolio_overlap(
|
||||
class_overlap=0.1, price_overlap=0.1,
|
||||
unit_mix_overlap=None, timing_overlap=None, geo_weight=1.0,
|
||||
)
|
||||
assert _aggregate_overlap([strong, weak1, weak2]) == pytest.approx(0.9)
|
||||
|
||||
def test_geo_weight_attenuates_distant_strong_overlap(self) -> None:
|
||||
# сильное пересечение, но далеко (geo 0.1) → signal 0.05; близкое слабое
|
||||
# (overlap 0.3, geo 1.0 → 0.3) перебивает. Агрегат = 0.3.
|
||||
distant_strong = _own_portfolio_overlap(
|
||||
class_overlap=1.0, price_overlap=1.0,
|
||||
unit_mix_overlap=None, timing_overlap=None, geo_weight=0.1,
|
||||
)
|
||||
near_weak = _own_portfolio_overlap(
|
||||
class_overlap=0.3, price_overlap=0.3,
|
||||
unit_mix_overlap=None, timing_overlap=None, geo_weight=1.0,
|
||||
)
|
||||
assert _aggregate_overlap([distant_strong, near_weak]) == pytest.approx(0.3)
|
||||
|
||||
def test_all_none_signals_is_none(self) -> None:
|
||||
empty = _own_portfolio_overlap(
|
||||
class_overlap=None, price_overlap=None,
|
||||
unit_mix_overlap=None, timing_overlap=None, geo_weight=1.0,
|
||||
)
|
||||
assert _aggregate_overlap([empty, empty]) is None
|
||||
|
||||
def test_empty_is_none(self) -> None:
|
||||
assert _aggregate_overlap([]) is None
|
||||
|
||||
|
||||
class TestHaversine:
|
||||
def test_zero_distance(self) -> None:
|
||||
assert _haversine_km(60.6, 56.8, 60.6, 56.8) == pytest.approx(0.0, abs=1e-9)
|
||||
|
||||
def test_known_distance_ekb_scale(self) -> None:
|
||||
# ~1 км по долготе на широте ЕКБ (56.8°): 0.0164° lon ≈ 1 км. Допуск широкий.
|
||||
d = _haversine_km(60.6, 56.8, 60.6164, 56.8)
|
||||
assert 0.9 < d < 1.1
|
||||
|
||||
def test_symmetric(self) -> None:
|
||||
a = _haversine_km(60.6, 56.8, 60.7, 56.9)
|
||||
b = _haversine_km(60.7, 56.9, 60.6, 56.8)
|
||||
assert a == pytest.approx(b)
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# Pure: Artificial Demand (доля ипотеки)
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
|
@ -877,3 +1179,241 @@ class TestComputeSpecialIndicesGraceful:
|
|||
assert captured["obj_class"] is None
|
||||
assert captured["district"] is None
|
||||
assert card.indices[KEY_ARTIFICIAL_DEMAND].value == pytest.approx(0.25)
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
# §25.3 Cannibalization dispatch — TRUE own-portfolio vs PROXY fallback
|
||||
# ──────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
# Центроид участка для гео-веса (lon, lat) в окрестностях ЕКБ.
|
||||
_CENTROID = (60.6000, 56.8000)
|
||||
_CAND_SPEC = SegmentSpec(
|
||||
obj_class="комфорт", room_bucket="2-к 45-60", district="Академический",
|
||||
price_bucket=PRICE_BUCKET_COMFORT,
|
||||
)
|
||||
|
||||
|
||||
def _own(
|
||||
name: str,
|
||||
*,
|
||||
source: str = "future",
|
||||
obj_class: str | None = "комфорт",
|
||||
price_min: float | None = 120_000.0,
|
||||
price_max: float | None = 160_000.0,
|
||||
unit_mix: dict[str, float] | None = None,
|
||||
release_month: date | None = None,
|
||||
lon: float | None = None,
|
||||
lat: float | None = None,
|
||||
) -> OwnProject:
|
||||
return OwnProject(
|
||||
name=name,
|
||||
source=source, # type: ignore[arg-type]
|
||||
obj_class=obj_class,
|
||||
release_month=release_month,
|
||||
price_min_per_m2=price_min,
|
||||
price_max_per_m2=price_max,
|
||||
unit_mix=unit_mix,
|
||||
district="Академический",
|
||||
lon=lon,
|
||||
lat=lat,
|
||||
)
|
||||
|
||||
|
||||
def _cannibalization_card(
|
||||
portfolio: list[OwnProject],
|
||||
*,
|
||||
centroid: tuple[float, float] | None = _CENTROID,
|
||||
cad_num: str | None = "66:41:0303161:123",
|
||||
) -> Any:
|
||||
"""Прогнать compute_special_indices с замоканными own-portfolio + центроидом."""
|
||||
db = MagicMock()
|
||||
with (
|
||||
_full_stack_patch(),
|
||||
patch(f"{_MOD}.get_own_portfolio", return_value=portfolio),
|
||||
patch(f"{_MOD}._query_parcel_centroid", return_value=centroid),
|
||||
patch(
|
||||
f"{_MOD}._query_artificial_demand",
|
||||
return_value={"n_sold": 100, "n_mortgage": 40},
|
||||
),
|
||||
):
|
||||
return compute_special_indices(
|
||||
db, spec=_CAND_SPEC, district="Академический", cad_num=cad_num
|
||||
)
|
||||
|
||||
|
||||
class TestCannibalizationTrueMode:
|
||||
def test_nonempty_portfolio_uses_own_portfolio_mode(self) -> None:
|
||||
# наш проект на участке (distance 0 → geo 1.0), класс/цена совпадают → overlap 1.0.
|
||||
portfolio = [
|
||||
_own("Наш-А", lon=_CENTROID[0], lat=_CENTROID[1]),
|
||||
_own("Наш-Б", obj_class="бизнес", price_min=200_000.0, price_max=240_000.0,
|
||||
lon=60.9, lat=57.1),
|
||||
]
|
||||
card = _cannibalization_card(portfolio)
|
||||
can = card.indices[KEY_CANNIBALIZATION]
|
||||
assert can.method == "own_portfolio_overlap"
|
||||
assert can.detail["mode"] == "own_portfolio"
|
||||
assert can.detail["proxy"] is False
|
||||
# сильнейший каннибализатор = Наш-А (class 1.0 + price 1.0)/2 × geo 1.0 = 1.0.
|
||||
assert can.value == pytest.approx(1.0)
|
||||
|
||||
def test_top_contributor_surfaced(self) -> None:
|
||||
portfolio = [
|
||||
_own("Наш-А", lon=_CENTROID[0], lat=_CENTROID[1]),
|
||||
_own("Наш-Б", obj_class="бизнес", price_min=200_000.0, price_max=240_000.0,
|
||||
lon=60.9, lat=57.1),
|
||||
]
|
||||
card = _cannibalization_card(portfolio)
|
||||
can = card.indices[KEY_CANNIBALIZATION]
|
||||
top = can.detail["top_contributors"]
|
||||
assert top[0]["name"] == "Наш-А"
|
||||
assert "Наш-А" in (can.label or "")
|
||||
# explainability: пер-ось breakdown присутствует.
|
||||
assert top[0]["axes"]["class"] == pytest.approx(1.0)
|
||||
assert top[0]["axes"]["price"] == pytest.approx(1.0)
|
||||
# квартирография/тайминг недоступны из spec → None (НЕ 0).
|
||||
assert top[0]["axes"]["unit_mix"] is None
|
||||
assert top[0]["axes"]["timing"] is None
|
||||
|
||||
def test_geo_weight_attenuates_distant_project(self) -> None:
|
||||
# тот же сильный overlap, но проект далеко → значение заметно ниже 1.0.
|
||||
near = _cannibalization_card([_own("Близкий", lon=_CENTROID[0], lat=_CENTROID[1])])
|
||||
far = _cannibalization_card([_own("Далёкий", lon=61.5, lat=57.5)])
|
||||
near_v = near.indices[KEY_CANNIBALIZATION].value
|
||||
far_v = far.indices[KEY_CANNIBALIZATION].value
|
||||
assert near_v is not None and far_v is not None
|
||||
assert near_v > far_v
|
||||
|
||||
def test_no_geometry_still_computes_via_other_axes(self) -> None:
|
||||
# нет центроида → гео-вес на floor (0.1), но класс+цена дают overlap → индекс есть.
|
||||
card = _cannibalization_card([_own("Без-гео")], centroid=None)
|
||||
can = card.indices[KEY_CANNIBALIZATION]
|
||||
assert can.method == "own_portfolio_overlap"
|
||||
assert can.detail["has_geometry"] is False
|
||||
# overlap 1.0 × geo floor 0.1 = 0.1.
|
||||
assert can.value == pytest.approx(0.1)
|
||||
|
||||
def test_unit_mix_and_timing_axes_when_supplied(self) -> None:
|
||||
# caller передаёт квартирографию/тайминг кандидата → оси становятся доступны.
|
||||
db = MagicMock()
|
||||
own_proj = _own(
|
||||
"Наш-А", lon=_CENTROID[0], lat=_CENTROID[1],
|
||||
unit_mix={"studio": 0.5, "1k": 0.5}, release_month=date(2026, 6, 1),
|
||||
)
|
||||
with (
|
||||
patch(f"{_MOD}.get_own_portfolio", return_value=[own_proj]),
|
||||
patch(f"{_MOD}._query_parcel_centroid", return_value=_CENTROID),
|
||||
):
|
||||
from app.services.forecasting.special_indices import _build_cannibalization
|
||||
|
||||
idx = _build_cannibalization(
|
||||
db, spec=_CAND_SPEC, cad_num="66:41:0303161:123",
|
||||
candidate_unit_mix={"studio": 0.5, "1k": 0.5},
|
||||
candidate_release_month=date(2026, 6, 1),
|
||||
)
|
||||
top = idx.detail["top_contributors"][0]
|
||||
assert top["axes"]["unit_mix"] == pytest.approx(1.0) # одинаковый микс
|
||||
assert top["axes"]["timing"] == pytest.approx(1.0) # одинаковый месяц
|
||||
assert top["n_axes"] == 4
|
||||
|
||||
def test_none_not_zero_when_no_comparable_axes(self) -> None:
|
||||
# наш проект без класса И без цены → ни класс, ни цена не считаются;
|
||||
# тайминг/квартирография тоже None → пара без осей → индекс unavailable (НЕ 0).
|
||||
card = _cannibalization_card(
|
||||
[_own("Пустой", obj_class=None, price_min=None, price_max=None,
|
||||
lon=_CENTROID[0], lat=_CENTROID[1])]
|
||||
)
|
||||
can = card.indices[KEY_CANNIBALIZATION]
|
||||
assert can.value is None
|
||||
assert can.method == _METHOD_UNAVAILABLE
|
||||
assert "фабрикуем" in can.detail["reason"]
|
||||
|
||||
def test_axes_available_summary_excludes_missing(self) -> None:
|
||||
# 1 проект, доступны только class+price → summary это отражает (unit_mix/timing 0).
|
||||
card = _cannibalization_card([_own("Наш-А", lon=_CENTROID[0], lat=_CENTROID[1])])
|
||||
summary = card.indices[KEY_CANNIBALIZATION].detail["axes_available"]
|
||||
assert summary["class"] == 1
|
||||
assert summary["price"] == 1
|
||||
assert summary["unit_mix"] == 0
|
||||
assert summary["timing"] == 0
|
||||
|
||||
|
||||
class TestCannibalizationThinPortfolioConfidence:
|
||||
def test_only_current_is_low_with_note(self) -> None:
|
||||
# портфель только из current (нет future-пайплайна) → honest low + нота §26.
|
||||
portfolio = [
|
||||
_own("Текущий-1", source="current", lon=_CENTROID[0], lat=_CENTROID[1]),
|
||||
_own("Текущий-2", source="current", lon=_CENTROID[0], lat=_CENTROID[1]),
|
||||
_own("Текущий-3", source="current", lon=_CENTROID[0], lat=_CENTROID[1]),
|
||||
]
|
||||
can = _cannibalization_card(portfolio).indices[KEY_CANNIBALIZATION]
|
||||
assert can.confidence == "low"
|
||||
assert can.detail["confidence_note"] is not None
|
||||
assert "future" in can.detail["confidence_note"]
|
||||
|
||||
def test_single_project_is_low_data_scarce(self) -> None:
|
||||
can = _cannibalization_card(
|
||||
[_own("Один", lon=_CENTROID[0], lat=_CENTROID[1])]
|
||||
).indices[KEY_CANNIBALIZATION]
|
||||
assert can.confidence == "low"
|
||||
assert "данных мало" in can.detail["confidence_note"]
|
||||
|
||||
def test_healthy_portfolio_medium(self) -> None:
|
||||
# ≥2 проекта И есть future → medium (всё равно ≤ cap).
|
||||
portfolio = [
|
||||
_own("Будущий-1", source="future", lon=_CENTROID[0], lat=_CENTROID[1]),
|
||||
_own("Текущий-1", source="current", lon=_CENTROID[0], lat=_CENTROID[1]),
|
||||
]
|
||||
can = _cannibalization_card(portfolio).indices[KEY_CANNIBALIZATION]
|
||||
assert can.confidence == "medium"
|
||||
assert can.detail["confidence_note"] is None
|
||||
|
||||
|
||||
class TestCannibalizationProxyFallback:
|
||||
def test_empty_portfolio_falls_back_to_proxy(self) -> None:
|
||||
# own-portfolio пуст → ПРОКСИ, явно помеченный.
|
||||
can = _cannibalization_card([]).indices[KEY_CANNIBALIZATION]
|
||||
assert can.method == "proxy_same_class_relevance_share"
|
||||
assert can.detail["mode"] == "proxy"
|
||||
assert can.detail["proxy"] is True
|
||||
# каведат §26: прокси НЕ выдаётся за истинный индекс.
|
||||
assert "прокси" in can.detail["proxy_reason"].lower()
|
||||
assert "own-portfolio" in can.detail["proxy_reason"]
|
||||
|
||||
def test_proxy_value_matches_same_class_share(self) -> None:
|
||||
# competitors из _full_stack_patch: same-class (комфорт) 0.8+0.6=1.4; all=1.9.
|
||||
can = _cannibalization_card([]).indices[KEY_CANNIBALIZATION]
|
||||
assert can.value == pytest.approx((0.8 + 0.6) / (0.8 + 0.6 + 0.5))
|
||||
|
||||
def test_proxy_label_marked(self) -> None:
|
||||
can = _cannibalization_card([]).indices[KEY_CANNIBALIZATION]
|
||||
assert "прокси" in (can.label or "")
|
||||
|
||||
def test_proxy_no_cad_num_unavailable(self) -> None:
|
||||
# пустой портфель + нет cad_num → прокси не определить → unavailable.
|
||||
can = _cannibalization_card([], cad_num=None).indices[KEY_CANNIBALIZATION]
|
||||
assert can.value is None
|
||||
assert can.method == _METHOD_UNAVAILABLE
|
||||
|
||||
|
||||
class TestCannibalizationDeterminism:
|
||||
def test_same_inputs_identical_output(self) -> None:
|
||||
portfolio = [
|
||||
_own("Наш-А", lon=_CENTROID[0], lat=_CENTROID[1]),
|
||||
_own("Наш-Б", obj_class="бизнес", price_min=200_000.0, price_max=240_000.0,
|
||||
lon=60.9, lat=57.1),
|
||||
_own("Наш-В", obj_class="комфорт+", lon=60.65, lat=56.85),
|
||||
]
|
||||
first = _cannibalization_card(list(portfolio)).indices[KEY_CANNIBALIZATION].as_dict()
|
||||
second = _cannibalization_card(list(portfolio)).indices[KEY_CANNIBALIZATION].as_dict()
|
||||
assert first == second
|
||||
|
||||
def test_top_contributor_tie_break_by_name(self) -> None:
|
||||
# два проекта с ИДЕНТИЧНЫМ сигналом → детерминированный tie-break по имени (А раньше Я).
|
||||
portfolio = [
|
||||
_own("Яков", lon=_CENTROID[0], lat=_CENTROID[1]),
|
||||
_own("Андрей", lon=_CENTROID[0], lat=_CENTROID[1]),
|
||||
]
|
||||
can = _cannibalization_card(portfolio).indices[KEY_CANNIBALIZATION]
|
||||
names = [c["name"] for c in can.detail["top_contributors"]]
|
||||
assert names == ["Андрей", "Яков"]
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue