Light1YT
60c82b2053
All checks were successful
CI / frontend-tests (pull_request) Has been skipped
CI / changes (push) Successful in 6s
CI / frontend-tests (push) Has been skipped
CI / changes (pull_request) Successful in 5s
CI / backend-tests (push) Successful in 6m25s
CI / backend-tests (pull_request) Successful in 6m31s
Derive candidate_release_month = report-as-of (date.today()) + §25.1 Launch Window peak-deficit horizon, threaded into _build_cannibalization so the timing overlap axis activates against own-portfolio release_month (near-in-time own projects raise cannibalization risk). Launch Window now computed once in compute_special_indices and reused (no double-compute). Launch Window unavailable -> candidate_release_month None -> timing axis gracefully excluded (None-not-0); cannibalization still scores on class/price/geo. Adds stdlib _add_months helper (year-boundary safe, no new dep). Deterministic. 168 tests. §25.3 now: class+price+timing+geo active; unit-mix remains phase-2. Refs #1169
1610 lines
73 KiB
Python
1610 lines
73 KiB
Python
"""Тесты §25 special_indices (#986/954-C) — pure-helpers + mocked-БД + оркестратор.
|
||
|
||
Покрытие:
|
||
• Pure-математика БЕЗ БД: Product Void (порог/доля/счёт), Launch Window (пик +
|
||
tie-break + нормализация), Competitor Strength (топ-N среднее), Cannibalization
|
||
(доля same-class), Artificial Demand (доля ипотеки), Cost-of-Error (монотонность
|
||
+ произведение), нормализации в [0,1], confidence-helpers, _avg_ticket.
|
||
• Artificial-Demand SQL: MagicMock-сессия — форма SQL (CAST(:x AS type), не ::),
|
||
параметры; сигнал есть → индекс; нет проданных → None + caveat.
|
||
• compute_special_indices: @patch бэкенд-сервисов — 6 индексов присутствуют,
|
||
advisory True, per-index graceful (сбой одного → unavailable, карточка цела).
|
||
|
||
Всё детерминировано, без БД (сессия мокается). Зеркалит стиль test_what_to_build /
|
||
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 (
|
||
PRICE_BUCKET_BUSINESS,
|
||
PRICE_BUCKET_COMFORT,
|
||
PRICE_BUCKET_ECONOMY,
|
||
PRICE_BUCKET_PREMIUM,
|
||
SegmentSpec,
|
||
)
|
||
from app.services.forecasting.special_indices import (
|
||
_INDEX_KEYS,
|
||
_METHOD_UNAVAILABLE,
|
||
KEY_ARTIFICIAL_DEMAND,
|
||
KEY_CANNIBALIZATION,
|
||
KEY_COMPETITOR_STRENGTH,
|
||
KEY_COST_OF_ERROR,
|
||
KEY_LAUNCH_WINDOW,
|
||
KEY_PRODUCT_VOID,
|
||
SpecialIndex,
|
||
_add_months,
|
||
_aggregate_overlap,
|
||
_artificial_demand_share,
|
||
_avg_ticket_rub,
|
||
_candidate_release_month,
|
||
_cannibalization_index,
|
||
_cap_confidence,
|
||
_clamp01,
|
||
_class_overlap,
|
||
_competitor_strength,
|
||
_cost_of_error_index,
|
||
_count_void,
|
||
_geo_weight,
|
||
_haversine_km,
|
||
_launch_window_horizon,
|
||
_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); остальные сервисы патчатся через
|
||
# patch.multiple(_MOD, ...) по kwargs-имени, поэтому отдельных констант им не нужно.
|
||
_MOD = "app.services.forecasting.special_indices"
|
||
_DSF = f"{_MOD}.compute_demand_supply_forecast"
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Pure: _clamp01 / confidence-helpers
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestClamp01:
|
||
def test_in_range_passthrough(self) -> None:
|
||
assert _clamp01(0.4) == 0.4
|
||
|
||
def test_above_one_clamped(self) -> None:
|
||
assert _clamp01(1.7) == 1.0
|
||
|
||
def test_below_zero_clamped(self) -> None:
|
||
assert _clamp01(-0.3) == 0.0
|
||
|
||
|
||
class TestCapConfidence:
|
||
def test_high_capped_to_medium(self) -> None:
|
||
assert _cap_confidence("high") == "medium"
|
||
|
||
def test_medium_unchanged(self) -> None:
|
||
assert _cap_confidence("medium") == "medium"
|
||
|
||
def test_low_unchanged(self) -> None:
|
||
assert _cap_confidence("low") == "low"
|
||
|
||
|
||
class TestMinConfidence:
|
||
def test_worst_wins(self) -> None:
|
||
assert _min_confidence(["high", "low", "medium"]) == "low"
|
||
|
||
def test_none_ignored(self) -> None:
|
||
assert _min_confidence(["medium", None, "high"]) == "medium"
|
||
|
||
def test_all_none_low(self) -> None:
|
||
assert _min_confidence([None, None]) == "low"
|
||
|
||
def test_empty_low(self) -> None:
|
||
assert _min_confidence([]) == "low"
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Pure: Product Void (порог / доля / счёт)
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestVoidIndex:
|
||
def test_share_above_threshold(self) -> None:
|
||
# 2 из 4 измеренных ≥ 0.25 → 0.5.
|
||
assert _void_index([0.3, 0.1, 0.9, -0.2]) == 0.5
|
||
|
||
def test_threshold_is_inclusive(self) -> None:
|
||
# ровно на пороге 0.25 → считается пустотой.
|
||
assert _void_index([0.25]) == 1.0
|
||
|
||
def test_below_threshold_excluded(self) -> None:
|
||
# 0.24 < 0.25 → не пустота.
|
||
assert _void_index([0.24]) == 0.0
|
||
|
||
def test_none_cells_excluded_from_denominator(self) -> None:
|
||
# None не входит ни в числитель, ни в знаменатель: 1 из 2 измеренных.
|
||
assert _void_index([0.9, None, 0.0]) == 0.5
|
||
|
||
def test_empty_yields_zero(self) -> None:
|
||
assert _void_index([]) == 0.0
|
||
|
||
def test_all_none_yields_zero(self) -> None:
|
||
assert _void_index([None, None]) == 0.0
|
||
|
||
def test_in_range(self) -> None:
|
||
v = _void_index([0.9, 0.8, 0.7])
|
||
assert 0.0 <= v <= 1.0
|
||
assert v == 1.0
|
||
|
||
def test_custom_threshold(self) -> None:
|
||
assert _void_index([0.5, 0.4], threshold=0.45) == 0.5
|
||
|
||
|
||
class TestCountVoid:
|
||
def test_counts_at_or_above_threshold(self) -> None:
|
||
assert _count_void([0.3, 0.25, 0.1, 0.99]) == 3
|
||
|
||
def test_none_skipped(self) -> None:
|
||
assert _count_void([None, 0.9, None]) == 1
|
||
|
||
def test_empty_zero(self) -> None:
|
||
assert _count_void([]) == 0
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Pure: Launch Window (пик + tie-break + нормализация)
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestPickLaunchWindow:
|
||
def test_picks_peak_horizon(self) -> None:
|
||
h, strength = _pick_launch_window({6: 0.1, 12: 0.4, 18: 0.2, 24: -0.1})
|
||
assert h == 12
|
||
# 0.4 / 0.5 (saturation) = 0.8.
|
||
assert strength == pytest.approx(0.8)
|
||
|
||
def test_strength_in_range(self) -> None:
|
||
_, strength = _pick_launch_window({6: 0.9, 12: 0.95})
|
||
assert strength is not None
|
||
assert 0.0 <= strength <= 1.0
|
||
assert strength == 1.0 # 0.95/0.5 clamp → 1.0
|
||
|
||
def test_tie_break_prefers_earlier_horizon(self) -> None:
|
||
# равный пиковый deficit на 6 и 18 → берём меньший горизонт (раньше выгоднее).
|
||
h, _ = _pick_launch_window({6: 0.5, 18: 0.5})
|
||
assert h == 6
|
||
|
||
def test_none_horizons_ignored(self) -> None:
|
||
# 12 — None (тонко), пик среди измеренных = 24.
|
||
h, _ = _pick_launch_window({6: 0.1, 12: None, 24: 0.3})
|
||
assert h == 24
|
||
|
||
def test_all_none_degrades(self) -> None:
|
||
assert _pick_launch_window({6: None, 12: None}) == (None, None)
|
||
|
||
def test_empty_degrades(self) -> None:
|
||
assert _pick_launch_window({}) == (None, None)
|
||
|
||
def test_nonpositive_peak_zero_strength_but_horizon_returned(self) -> None:
|
||
# все горизонты ≤0 (баланс/затоварка): окна «строить» нет (strength 0),
|
||
# но аргмакс-горизонт возвращаем (наименее плохой) для explainability.
|
||
h, strength = _pick_launch_window({6: -0.3, 12: -0.1, 24: -0.5})
|
||
assert h == 12 # наибольший (наименее отрицательный) deficit
|
||
assert strength == 0.0
|
||
|
||
def test_saturation_nonpositive_degrades_to_sign(self) -> None:
|
||
h, strength = _pick_launch_window({6: 0.2}, saturation=0.0)
|
||
assert h == 6
|
||
assert strength == 1.0
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Pure: Competitor Strength (топ-N среднее)
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestCompetitorStrength:
|
||
def test_mean_of_top_n(self) -> None:
|
||
# топ-5 по убыванию из 6: 0.9,0.8,0.7,0.6,0.5 → mean 0.7.
|
||
v = _competitor_strength([0.1, 0.5, 0.9, 0.6, 0.8, 0.7])
|
||
assert v == pytest.approx(0.7)
|
||
|
||
def test_fewer_than_top_n(self) -> None:
|
||
v = _competitor_strength([0.4, 0.6])
|
||
assert v == pytest.approx(0.5)
|
||
|
||
def test_none_weights_skipped(self) -> None:
|
||
v = _competitor_strength([None, 0.8, None, 0.4])
|
||
assert v == pytest.approx(0.6)
|
||
|
||
def test_empty_is_none(self) -> None:
|
||
assert _competitor_strength([]) is None
|
||
|
||
def test_all_none_is_none(self) -> None:
|
||
assert _competitor_strength([None, None]) is None
|
||
|
||
def test_in_range(self) -> None:
|
||
v = _competitor_strength([0.9, 0.95, 1.0])
|
||
assert v is not None and 0.0 <= v <= 1.0
|
||
|
||
def test_custom_top_n(self) -> None:
|
||
v = _competitor_strength([0.9, 0.8, 0.1], top_n=2)
|
||
assert v == pytest.approx(0.85)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Pure: Cannibalization (доля same-class)
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestCannibalizationIndex:
|
||
def test_share_of_same_class(self) -> None:
|
||
# same = 0.6+0.4 = 1.0; all = 0.6+0.4+0.5+0.5 = 2.0 → 0.5.
|
||
v = _cannibalization_index([0.6, 0.4], [0.6, 0.4, 0.5, 0.5])
|
||
assert v == pytest.approx(0.5)
|
||
|
||
def test_no_same_class_is_zero(self) -> None:
|
||
# конкуренты есть, но ни одного в нашем классе → 0.0 (валидно).
|
||
v = _cannibalization_index([], [0.5, 0.5])
|
||
assert v == 0.0
|
||
|
||
def test_no_competitors_is_none(self) -> None:
|
||
# окружения нет вообще → None (неизмеримо).
|
||
assert _cannibalization_index([], []) is None
|
||
|
||
def test_all_same_class_is_one(self) -> None:
|
||
v = _cannibalization_index([0.5, 0.5], [0.5, 0.5])
|
||
assert v == 1.0
|
||
|
||
def test_none_weights_skipped(self) -> None:
|
||
v = _cannibalization_index([0.5], [0.5, None, 0.5])
|
||
assert v == pytest.approx(0.5)
|
||
|
||
def test_in_range(self) -> None:
|
||
v = _cannibalization_index([0.3], [0.3, 0.7])
|
||
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 TestAddMonths:
|
||
def test_within_year(self) -> None:
|
||
assert _add_months(date(2026, 1, 1), 5) == date(2026, 6, 1)
|
||
|
||
def test_year_boundary_oct_plus_six_is_apr_next_year(self) -> None:
|
||
# Oct + 6 = Apr следующего года (явный year-boundary кейс из ТЗ).
|
||
assert _add_months(date(2026, 10, 1), 6) == date(2027, 4, 1)
|
||
|
||
def test_exactly_twelve_months_same_month_next_year(self) -> None:
|
||
assert _add_months(date(2026, 3, 1), 12) == date(2027, 3, 1)
|
||
|
||
def test_twenty_four_months_two_years(self) -> None:
|
||
assert _add_months(date(2026, 6, 1), 24) == date(2028, 6, 1)
|
||
|
||
def test_zero_months_normalizes_to_first_of_month(self) -> None:
|
||
# +0 мес всё равно нормализует к 1-му числу (день отбрасывается).
|
||
assert _add_months(date(2026, 6, 17), 0) == date(2026, 6, 1)
|
||
|
||
def test_december_rollover(self) -> None:
|
||
assert _add_months(date(2026, 12, 1), 1) == date(2027, 1, 1)
|
||
|
||
def test_negative_months_go_back_across_year(self) -> None:
|
||
assert _add_months(date(2026, 2, 1), -3) == date(2025, 11, 1)
|
||
|
||
def test_result_is_always_first_of_month(self) -> None:
|
||
for m in range(1, 25):
|
||
assert _add_months(date(2026, 7, 23), m).day == 1
|
||
|
||
|
||
def _launch_window_index(horizon: int | None) -> SpecialIndex:
|
||
"""Готовый Launch Window SpecialIndex с заданным best_horizon_months (None → unavail)."""
|
||
if horizon is None:
|
||
return SpecialIndex(
|
||
key=KEY_LAUNCH_WINDOW, value=None, label=None, confidence="low",
|
||
detail={"reason": "deficit None на всех горизонтах"},
|
||
method=_METHOD_UNAVAILABLE, advisory=True,
|
||
)
|
||
return SpecialIndex(
|
||
key=KEY_LAUNCH_WINDOW, value=0.8, label=f"{horizon} мес", confidence="medium",
|
||
detail={"best_horizon_months": horizon, "deficit_by_horizon": {}},
|
||
method="deficit_peak_scan", advisory=True,
|
||
)
|
||
|
||
|
||
class TestLaunchWindowHorizon:
|
||
def test_reads_best_horizon(self) -> None:
|
||
assert _launch_window_horizon(_launch_window_index(18)) == 18
|
||
|
||
def test_unavailable_index_is_none(self) -> None:
|
||
# Launch Window недоступен (нет best_horizon_months) → None (ось тайминга off).
|
||
assert _launch_window_horizon(_launch_window_index(None)) is None
|
||
|
||
|
||
class TestCandidateReleaseMonth:
|
||
def test_derives_as_of_plus_horizon(self) -> None:
|
||
# дата отчёта + горизонт окна запуска = месяц выхода кандидата на рынок.
|
||
m = _candidate_release_month(_launch_window_index(12), as_of=date(2026, 6, 9))
|
||
assert m == date(2027, 6, 1)
|
||
|
||
def test_horizon_crosses_year_boundary(self) -> None:
|
||
# окт + 6 мес → апр следующего года (year-boundary в деривации тайминга).
|
||
m = _candidate_release_month(_launch_window_index(6), as_of=date(2026, 10, 20))
|
||
assert m == date(2027, 4, 1)
|
||
|
||
def test_unavailable_launch_window_is_none(self) -> None:
|
||
# Launch Window недоступен → release_month None → тайминговая ось исключается.
|
||
assert _candidate_release_month(_launch_window_index(None), as_of=date(2026, 6, 9)) is None
|
||
|
||
|
||
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 (доля ипотеки)
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestArtificialDemandShare:
|
||
def test_basic_share(self) -> None:
|
||
assert _artificial_demand_share(30, 100) == pytest.approx(0.3)
|
||
|
||
def test_full_mortgage(self) -> None:
|
||
assert _artificial_demand_share(50, 50) == 1.0
|
||
|
||
def test_zero_mortgage(self) -> None:
|
||
assert _artificial_demand_share(0, 80) == 0.0
|
||
|
||
def test_no_sold_is_none(self) -> None:
|
||
# нет проданных лотов → None (НЕ фабрикуем 0).
|
||
assert _artificial_demand_share(0, 0) is None
|
||
|
||
def test_none_mortgage_is_none(self) -> None:
|
||
assert _artificial_demand_share(None, 50) is None
|
||
|
||
def test_clamped_to_one_on_dirty_data(self) -> None:
|
||
# грязные данные (mortgage > sold) → clamp 1.0, не >1.
|
||
assert _artificial_demand_share(120, 100) == 1.0
|
||
|
||
def test_in_range(self) -> None:
|
||
v = _artificial_demand_share(17, 23)
|
||
assert v is not None and 0.0 <= v <= 1.0
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Pure: Cost-of-Error (монотонность + произведение + чек)
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestAvgTicket:
|
||
def test_price_times_area(self) -> None:
|
||
# 200_000 ₽/м² × 50 м² = 10 млн.
|
||
assert _avg_ticket_rub(200_000.0) == 10_000_000.0
|
||
|
||
def test_none_price_is_none(self) -> None:
|
||
assert _avg_ticket_rub(None) is None
|
||
|
||
def test_nonpositive_price_is_none(self) -> None:
|
||
assert _avg_ticket_rub(0.0) is None
|
||
|
||
def test_custom_area(self) -> None:
|
||
assert _avg_ticket_rub(100_000.0, area_m2=80.0) == 8_000_000.0
|
||
|
||
|
||
class TestOversupplyRiskFromDeficit:
|
||
def test_negative_deficit_is_magnitude(self) -> None:
|
||
assert _oversupply_risk_from_deficit(-0.4) == pytest.approx(0.4)
|
||
|
||
def test_positive_deficit_zero_risk(self) -> None:
|
||
assert _oversupply_risk_from_deficit(0.6) == 0.0
|
||
|
||
def test_zero_deficit_zero_risk(self) -> None:
|
||
assert _oversupply_risk_from_deficit(0.0) == 0.0
|
||
|
||
def test_none_is_none(self) -> None:
|
||
assert _oversupply_risk_from_deficit(None) is None
|
||
|
||
|
||
class TestCostOfErrorIndex:
|
||
def test_product_of_risk_and_ticket(self) -> None:
|
||
# risk 0.5 × (10млн/15млн clamp=0.6667) ≈ 0.3333.
|
||
v = _cost_of_error_index(0.5, 10_000_000.0)
|
||
assert v == pytest.approx(0.5 * (10_000_000.0 / 15_000_000.0))
|
||
|
||
def test_monotonic_in_risk(self) -> None:
|
||
low = _cost_of_error_index(0.2, 10_000_000.0)
|
||
high = _cost_of_error_index(0.8, 10_000_000.0)
|
||
assert low is not None and high is not None
|
||
assert high > low
|
||
|
||
def test_monotonic_in_ticket(self) -> None:
|
||
cheap = _cost_of_error_index(0.5, 5_000_000.0)
|
||
pricey = _cost_of_error_index(0.5, 12_000_000.0)
|
||
assert cheap is not None and pricey is not None
|
||
assert pricey > cheap
|
||
|
||
def test_in_range(self) -> None:
|
||
v = _cost_of_error_index(0.9, 30_000_000.0) # ticket clamps to 1.0
|
||
assert v is not None and 0.0 <= v <= 1.0
|
||
assert v == pytest.approx(0.9)
|
||
|
||
def test_none_risk_is_none(self) -> None:
|
||
assert _cost_of_error_index(None, 10_000_000.0) is None
|
||
|
||
def test_none_ticket_is_none(self) -> None:
|
||
assert _cost_of_error_index(0.5, None) is None
|
||
|
||
def test_nonpositive_norm_degrades_to_risk_only(self) -> None:
|
||
v = _cost_of_error_index(0.4, 10_000_000.0, ticket_norm_rub=0.0)
|
||
assert v == pytest.approx(0.4)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Artificial-Demand SQL: MagicMock-сессия (форма SQL + параметры + сигнал/деградация)
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _mock_db_one(row: dict[str, Any] | None) -> MagicMock:
|
||
"""Сессия, чей единственный execute().mappings().first() вернёт row."""
|
||
db = MagicMock()
|
||
result = MagicMock()
|
||
result.mappings.return_value.first.return_value = row
|
||
db.execute.return_value = result
|
||
return db
|
||
|
||
|
||
def _executed_sql(db: MagicMock, call_index: int = 0) -> str:
|
||
args, _ = db.execute.call_args_list[call_index]
|
||
return str(args[0])
|
||
|
||
|
||
def _executed_params(db: MagicMock, call_index: int = 0) -> dict[str, Any]:
|
||
args, _ = db.execute.call_args_list[call_index]
|
||
return args[1]
|
||
|
||
|
||
class TestArtificialDemandSQL:
|
||
def test_sql_uses_cast_not_double_colon(self) -> None:
|
||
db = _mock_db_one({"n_sold": 10, "n_mortgage": 4})
|
||
_query_artificial_demand(
|
||
db, district="Академический", obj_class="комфорт", premise_kind="квартира"
|
||
)
|
||
sql = _executed_sql(db)
|
||
assert "CAST(:premise_kind AS text)" in sql
|
||
assert "CAST(:district AS text)" in sql
|
||
assert "CAST(:obj_class AS text)" in sql
|
||
# psycopg v3: никогда :x::type
|
||
assert ":premise_kind::" not in sql
|
||
assert ":district::" not in sql
|
||
assert ":obj_class::" not in sql
|
||
|
||
def test_sql_reads_mortgage_signal_columns(self) -> None:
|
||
db = _mock_db_one({"n_sold": 10, "n_mortgage": 4})
|
||
_query_artificial_demand(db, district=None, obj_class=None, premise_kind="квартира")
|
||
sql = _executed_sql(db)
|
||
# сигнал ипотеки = encumbrance_type / bank_name (реальные колонки objective_lots).
|
||
assert "encumbrance_type" in sql
|
||
assert "bank_name" in sql
|
||
assert "objective_lots" in sql
|
||
|
||
def test_params_passed(self) -> None:
|
||
db = _mock_db_one({"n_sold": 1, "n_mortgage": 0})
|
||
_query_artificial_demand(
|
||
db, district="Пионерский", obj_class="бизнес", premise_kind="квартира"
|
||
)
|
||
p = _executed_params(db)
|
||
assert p["district"] == "Пионерский"
|
||
assert p["obj_class"] == "бизнес"
|
||
assert p["premise_kind"] == "квартира"
|
||
|
||
def test_signal_present_returns_counts(self) -> None:
|
||
db = _mock_db_one({"n_sold": 80, "n_mortgage": 52})
|
||
out = _query_artificial_demand(db, district=None, obj_class=None, premise_kind="квартира")
|
||
assert out == {"n_sold": 80, "n_mortgage": 52}
|
||
|
||
def test_empty_row_degrades_to_zeros(self) -> None:
|
||
db = _mock_db_one(None)
|
||
out = _query_artificial_demand(db, district=None, obj_class=None, premise_kind="квартира")
|
||
assert out == {"n_sold": 0, "n_mortgage": 0}
|
||
|
||
def test_null_counts_coerced_to_zero(self) -> None:
|
||
db = _mock_db_one({"n_sold": None, "n_mortgage": None})
|
||
out = _query_artificial_demand(db, district=None, obj_class=None, premise_kind="квартира")
|
||
assert out == {"n_sold": 0, "n_mortgage": 0}
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Artificial-Demand builder via mocked rows: signal → index; no signal → None+caveat
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
_SPEC = SegmentSpec(obj_class="комфорт", room_bucket="2-к 45-60", district="Академический")
|
||
|
||
|
||
def _patch_all_unavailable() -> Any:
|
||
"""Контекст: все 5 НЕ-Artificial бэкенда брошены/пусты (изолируем Artificial)."""
|
||
return patch.multiple(
|
||
_MOD,
|
||
compute_demand_supply_forecast=MagicMock(return_value=[]),
|
||
rank_segments=MagicMock(return_value=MagicMock(ranked=[], n_cells_ranked=0)),
|
||
get_competitors=MagicMock(side_effect=ValueError("no geom")),
|
||
compute_market_metrics=MagicMock(
|
||
return_value=MagicMock(overstock_index=None, confidence="low")
|
||
),
|
||
compute_affordability=MagicMock(return_value=MagicMock(price_per_m2=None)),
|
||
)
|
||
|
||
|
||
class TestArtificialDemandBuilder:
|
||
def test_signal_present_yields_index(self) -> None:
|
||
db = MagicMock()
|
||
with (
|
||
_patch_all_unavailable(),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 100, "n_mortgage": 65},
|
||
),
|
||
):
|
||
card = compute_special_indices(db, spec=_SPEC, district="Академический")
|
||
idx = card.indices[KEY_ARTIFICIAL_DEMAND]
|
||
assert idx.value == pytest.approx(0.65)
|
||
assert idx.method == "mortgage_share_objective_lots"
|
||
assert idx.detail["n_sold"] == 100
|
||
assert idx.detail["n_mortgage"] == 65
|
||
assert idx.advisory is True
|
||
|
||
def test_no_signal_yields_none_with_caveat(self) -> None:
|
||
db = MagicMock()
|
||
with (
|
||
_patch_all_unavailable(),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 0, "n_mortgage": 0},
|
||
),
|
||
):
|
||
card = compute_special_indices(db, spec=_SPEC, district="Академический")
|
||
idx = card.indices[KEY_ARTIFICIAL_DEMAND]
|
||
assert idx.value is None
|
||
assert idx.method == _METHOD_UNAVAILABLE
|
||
assert "reason" in idx.detail
|
||
assert "фабрикуем" in idx.detail["reason"] # явный caveat «не фабрикуем»
|
||
|
||
def test_small_sample_low_confidence(self) -> None:
|
||
db = MagicMock()
|
||
with (
|
||
_patch_all_unavailable(),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 5, "n_mortgage": 3},
|
||
),
|
||
):
|
||
card = compute_special_indices(db, spec=_SPEC, district="Академический")
|
||
idx = card.indices[KEY_ARTIFICIAL_DEMAND]
|
||
assert idx.value == pytest.approx(0.6)
|
||
assert idx.confidence == "low" # n_sold < 30
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# compute_special_indices: @patch бэкенда — 6 индексов, advisory, graceful per-index
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _forecast_stub(deficit: float | None, *, horizon: int, confidence: str = "medium") -> MagicMock:
|
||
f = MagicMock()
|
||
f.deficit_index = deficit
|
||
f.horizon_months = horizon
|
||
f.confidence = confidence
|
||
return f
|
||
|
||
|
||
def _ranked_stub(deficit: float, *, confidence: str = "medium") -> MagicMock:
|
||
seg = MagicMock()
|
||
seg.deficit_index = deficit
|
||
seg.confidence = confidence
|
||
seg.as_dict.return_value = {"deficit_index": deficit}
|
||
return seg
|
||
|
||
|
||
def _competitor_stub(relevance: float | None, obj_class: str | None) -> MagicMock:
|
||
c = MagicMock()
|
||
c.relevance_weight = relevance
|
||
c.obj_class = obj_class
|
||
return c
|
||
|
||
|
||
def _full_stack_patch() -> Any:
|
||
"""Все 5 бэкенд-сервисов отдают здоровые данные → все 6 индексов считаются."""
|
||
forecasts = [_forecast_stub(0.1 * h, horizon=h) for h in (6, 12, 18, 24)]
|
||
ranking = MagicMock()
|
||
ranking.ranked = [_ranked_stub(0.5), _ranked_stub(0.3), _ranked_stub(-0.1)]
|
||
ranking.n_cells_ranked = 3
|
||
competitors_resp = MagicMock()
|
||
competitors_resp.competitors = [
|
||
_competitor_stub(0.8, "комфорт"),
|
||
_competitor_stub(0.6, "комфорт"),
|
||
_competitor_stub(0.5, "бизнес"),
|
||
]
|
||
metrics = MagicMock()
|
||
metrics.overstock_index = 0.4
|
||
metrics.confidence = "medium"
|
||
afford = MagicMock()
|
||
afford.price_per_m2 = 200_000.0
|
||
return patch.multiple(
|
||
_MOD,
|
||
compute_demand_supply_forecast=MagicMock(return_value=forecasts),
|
||
rank_segments=MagicMock(return_value=ranking),
|
||
get_competitors=MagicMock(return_value=competitors_resp),
|
||
compute_market_metrics=MagicMock(return_value=metrics),
|
||
compute_affordability=MagicMock(return_value=afford),
|
||
)
|
||
|
||
|
||
class TestComputeSpecialIndicesShape:
|
||
def test_all_six_indices_present(self) -> None:
|
||
db = MagicMock()
|
||
with (
|
||
_full_stack_patch(),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 100, "n_mortgage": 40},
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
assert set(card.indices.keys()) == set(_INDEX_KEYS)
|
||
assert len(card.indices) == 6
|
||
|
||
def test_advisory_always_true(self) -> None:
|
||
db = MagicMock()
|
||
with (
|
||
_full_stack_patch(),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 100, "n_mortgage": 40},
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
assert card.advisory is True
|
||
assert all(idx.advisory is True for idx in card.indices.values())
|
||
|
||
def test_all_values_in_range_when_present(self) -> None:
|
||
db = MagicMock()
|
||
with (
|
||
_full_stack_patch(),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 100, "n_mortgage": 40},
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
for idx in card.indices.values():
|
||
if idx.value is not None:
|
||
assert 0.0 <= idx.value <= 1.0, idx.key
|
||
|
||
def test_confidence_capped_at_medium(self) -> None:
|
||
db = MagicMock()
|
||
with (
|
||
_full_stack_patch(),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 100, "n_mortgage": 40},
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
assert card.confidence in ("low", "medium") # никогда high
|
||
for idx in card.indices.values():
|
||
assert idx.confidence in ("low", "medium")
|
||
|
||
def test_as_dict_roundtrip(self) -> None:
|
||
db = MagicMock()
|
||
with (
|
||
_full_stack_patch(),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 100, "n_mortgage": 40},
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
d = card.as_dict()
|
||
assert d["advisory"] is True
|
||
assert set(d["indices"].keys()) == set(_INDEX_KEYS)
|
||
assert d["district"] == "Академический"
|
||
# каждый индекс сериализуется с ключами контракта.
|
||
for sub in d["indices"].values():
|
||
assert set(sub.keys()) == {
|
||
"key",
|
||
"value",
|
||
"label",
|
||
"confidence",
|
||
"detail",
|
||
"method",
|
||
"advisory",
|
||
}
|
||
|
||
|
||
class TestComputeSpecialIndicesValues:
|
||
def test_launch_window_picks_peak_horizon(self) -> None:
|
||
db = MagicMock()
|
||
with (
|
||
_full_stack_patch(),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 100, "n_mortgage": 40},
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
lw = card.indices[KEY_LAUNCH_WINDOW]
|
||
# deficit = 0.1*h → пик на h=24 (0.1*24=2.4 clamps), label «24 мес».
|
||
assert lw.label == "24 мес"
|
||
assert lw.detail["best_horizon_months"] == 24
|
||
|
||
def test_product_void_counts_segments(self) -> None:
|
||
db = MagicMock()
|
||
with (
|
||
_full_stack_patch(),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 100, "n_mortgage": 40},
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
pv = card.indices[KEY_PRODUCT_VOID]
|
||
# ranked deficits [0.5,0.3,-0.1]: 2 ≥ 0.25 → share 2/3.
|
||
assert pv.value == pytest.approx(2 / 3)
|
||
assert pv.detail["n_void"] == 2
|
||
|
||
def test_competitor_strength_mean_top_n(self) -> None:
|
||
db = MagicMock()
|
||
with (
|
||
_full_stack_patch(),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 100, "n_mortgage": 40},
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
cs = card.indices[KEY_COMPETITOR_STRENGTH]
|
||
# relevance [0.8,0.6,0.5] → mean 0.6333.
|
||
assert cs.value == pytest.approx((0.8 + 0.6 + 0.5) / 3)
|
||
|
||
def test_cannibalization_same_class_share(self) -> None:
|
||
db = MagicMock()
|
||
with (
|
||
_full_stack_patch(),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 100, "n_mortgage": 40},
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
can = card.indices[KEY_CANNIBALIZATION]
|
||
# same-class (комфорт) = 0.8+0.6=1.4; all=1.9 → 0.7368.
|
||
assert can.value == pytest.approx((0.8 + 0.6) / (0.8 + 0.6 + 0.5))
|
||
|
||
def test_cost_of_error_product(self) -> None:
|
||
db = MagicMock()
|
||
with (
|
||
_full_stack_patch(),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 100, "n_mortgage": 40},
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
coe = card.indices[KEY_COST_OF_ERROR]
|
||
# risk 0.4 × ticket(200k×50=10млн / 15млн = 0.6667) ≈ 0.2667.
|
||
ticket_factor = (200_000.0 * 50.0) / 15_000_000.0
|
||
assert coe.value == pytest.approx(0.4 * ticket_factor)
|
||
assert coe.detail["risk_source"] == "overstock_index"
|
||
|
||
|
||
class TestComputeSpecialIndicesGraceful:
|
||
def test_no_cad_num_degrades_competitor_indices(self) -> None:
|
||
# без cad_num: Cannibalization + Competitor Strength → unavailable, остальные ок.
|
||
db = MagicMock()
|
||
with (
|
||
_full_stack_patch(),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 100, "n_mortgage": 40},
|
||
),
|
||
):
|
||
card = compute_special_indices(db, spec=_SPEC, district="Академический")
|
||
assert card.indices[KEY_CANNIBALIZATION].method == _METHOD_UNAVAILABLE
|
||
assert card.indices[KEY_COMPETITOR_STRENGTH].method == _METHOD_UNAVAILABLE
|
||
assert card.indices[KEY_CANNIBALIZATION].value is None
|
||
# quick-win Artificial Demand всё равно посчитан.
|
||
assert card.indices[KEY_ARTIFICIAL_DEMAND].value == pytest.approx(0.4)
|
||
|
||
def test_per_index_failure_isolated(self) -> None:
|
||
# один сервис (rank_segments) бросает → Product Void unavailable, остальные 5 ок.
|
||
forecasts = [_forecast_stub(0.2, horizon=h) for h in (6, 12, 18, 24)]
|
||
metrics = MagicMock(overstock_index=0.3, confidence="medium")
|
||
afford = MagicMock(price_per_m2=180_000.0)
|
||
comp = MagicMock()
|
||
comp.competitors = [_competitor_stub(0.7, "комфорт")]
|
||
db = MagicMock()
|
||
with (
|
||
patch.multiple(
|
||
_MOD,
|
||
compute_demand_supply_forecast=MagicMock(return_value=forecasts),
|
||
rank_segments=MagicMock(side_effect=RuntimeError("boom")),
|
||
get_competitors=MagicMock(return_value=comp),
|
||
compute_market_metrics=MagicMock(return_value=metrics),
|
||
compute_affordability=MagicMock(return_value=afford),
|
||
),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 50, "n_mortgage": 20},
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
# сбойный индекс — unavailable; карточка цела (все 6 ключей).
|
||
assert card.indices[KEY_PRODUCT_VOID].method == _METHOD_UNAVAILABLE
|
||
assert card.indices[KEY_PRODUCT_VOID].value is None
|
||
assert len(card.indices) == 6
|
||
# остальные посчитаны.
|
||
assert card.indices[KEY_LAUNCH_WINDOW].value is not None
|
||
assert card.indices[KEY_ARTIFICIAL_DEMAND].value == pytest.approx(0.4)
|
||
assert card.advisory is True
|
||
|
||
def test_all_backends_fail_card_still_returns(self) -> None:
|
||
# тотальный сбой: каждый индекс unavailable, но карточка возвращается (не crash).
|
||
db = MagicMock()
|
||
with (
|
||
patch.multiple(
|
||
_MOD,
|
||
compute_demand_supply_forecast=MagicMock(side_effect=RuntimeError("x")),
|
||
rank_segments=MagicMock(side_effect=RuntimeError("x")),
|
||
get_competitors=MagicMock(side_effect=RuntimeError("x")),
|
||
compute_market_metrics=MagicMock(side_effect=RuntimeError("x")),
|
||
compute_affordability=MagicMock(side_effect=RuntimeError("x")),
|
||
),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
side_effect=RuntimeError("x"),
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
assert len(card.indices) == 6
|
||
assert all(idx.method == _METHOD_UNAVAILABLE for idx in card.indices.values())
|
||
assert all(idx.value is None for idx in card.indices.values())
|
||
assert card.advisory is True
|
||
assert card.confidence == "low"
|
||
|
||
def test_launch_window_all_none_degrades(self) -> None:
|
||
# deficit None на всех горизонтах → Launch Window unavailable.
|
||
forecasts = [_forecast_stub(None, horizon=h) for h in (6, 12, 18, 24)]
|
||
db = MagicMock()
|
||
with (
|
||
_full_stack_patch(),
|
||
patch(_DSF, return_value=forecasts),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 100, "n_mortgage": 40},
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
assert card.indices[KEY_LAUNCH_WINDOW].method == _METHOD_UNAVAILABLE
|
||
assert card.indices[KEY_LAUNCH_WINDOW].value is None
|
||
|
||
def test_cost_of_error_falls_back_to_negative_deficit(self) -> None:
|
||
# overstock_index None → Cost-of-Error берёт магнитуду отрицательного дефицита.
|
||
metrics = MagicMock(overstock_index=None, confidence="medium")
|
||
afford = MagicMock(price_per_m2=200_000.0)
|
||
# forecast на _VOID_HORIZON_MONTHS=12 (fallback) с отрицательным дефицитом.
|
||
forecasts = [_forecast_stub(-0.5, horizon=12)]
|
||
comp = MagicMock()
|
||
comp.competitors = [_competitor_stub(0.7, "комфорт")]
|
||
db = MagicMock()
|
||
with (
|
||
patch.multiple(
|
||
_MOD,
|
||
compute_demand_supply_forecast=MagicMock(return_value=forecasts),
|
||
rank_segments=MagicMock(return_value=MagicMock(ranked=[], n_cells_ranked=0)),
|
||
get_competitors=MagicMock(return_value=comp),
|
||
compute_market_metrics=MagicMock(return_value=metrics),
|
||
compute_affordability=MagicMock(return_value=afford),
|
||
),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 50, "n_mortgage": 20},
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
coe = card.indices[KEY_COST_OF_ERROR]
|
||
assert coe.detail["risk_source"] == "negative_deficit"
|
||
# risk |−0.5|=0.5 × ticket(10млн/15млн=0.6667) ≈ 0.3333.
|
||
assert coe.value == pytest.approx(0.5 * ((200_000.0 * 50.0) / 15_000_000.0))
|
||
|
||
def test_aggregate_spec_no_class_param_is_none(self) -> None:
|
||
# spec без obj_class → Artificial-Demand SQL получает obj_class=None (агрегат).
|
||
db = MagicMock()
|
||
captured: dict[str, Any] = {}
|
||
|
||
def _capture(_db: Any, *, district: Any, obj_class: Any, premise_kind: Any) -> dict:
|
||
captured["obj_class"] = obj_class
|
||
captured["district"] = district
|
||
return {"n_sold": 40, "n_mortgage": 10}
|
||
|
||
spec_no_class = SegmentSpec(room_bucket="2-к 45-60")
|
||
with _full_stack_patch(), patch(f"{_MOD}._query_artificial_demand", side_effect=_capture):
|
||
card = compute_special_indices(db, spec=spec_no_class, district=None)
|
||
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
|
||
)
|
||
|
||
|
||
# Дата отчёта фиксируется в тестах тайминга, чтобы выведенный месяц кандидата был
|
||
# детерминирован вне зависимости от реального date.today(). _full_stack_patch даёт пик
|
||
# дефицита на горизонте 24 мес → candidate_release_month = _FIXED_TODAY + 24 мес.
|
||
_FIXED_TODAY = date(2026, 6, 9)
|
||
_DERIVED_CANDIDATE_MONTH = date(2028, 6, 1) # 2026-06 + 24 мес, 1-е число
|
||
|
||
|
||
class _FixedDate(date):
|
||
"""date с фиксированным today() (construction делегируется реальному date). PURE."""
|
||
|
||
@classmethod
|
||
def today(cls) -> date: # type: ignore[override]
|
||
return _FIXED_TODAY
|
||
|
||
|
||
def _timing_card(
|
||
portfolio: list[OwnProject],
|
||
*,
|
||
centroid: tuple[float, float] | None = _CENTROID,
|
||
cad_num: str | None = "66:41:0303161:123",
|
||
) -> Any:
|
||
"""Как _cannibalization_card, но с зафиксированной датой отчёта (тайминговая ось)."""
|
||
db = MagicMock()
|
||
with (
|
||
_full_stack_patch(),
|
||
patch(f"{_MOD}.date", _FixedDate),
|
||
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 TestCannibalizationTimingAxisFedFromLaunchWindow:
|
||
"""§25.3 тайминговая ось активируется из §25.1 Launch Window (#1169 follow-up)."""
|
||
|
||
def test_timing_axis_contributes_when_launch_window_resolves(self) -> None:
|
||
# Launch Window резолвится (пик h=24) → candidate_release_month выведен →
|
||
# тайминговая ось теперь СЧИТАЕТСЯ (на тот же месяц, что наш проект → 1.0).
|
||
own = _own(
|
||
"Наш-А", lon=_CENTROID[0], lat=_CENTROID[1],
|
||
release_month=_DERIVED_CANDIDATE_MONTH,
|
||
)
|
||
card = _timing_card([own])
|
||
can = card.indices[KEY_CANNIBALIZATION]
|
||
assert can.detail["axes_available"]["timing"] == 1 # ось активна (НЕ 0)
|
||
top = can.detail["top_contributors"][0]
|
||
assert top["axes"]["timing"] == pytest.approx(1.0) # тот же месяц выхода
|
||
assert top["n_axes"] == 3 # class + price + timing (unit_mix всё ещё None)
|
||
|
||
def test_near_in_time_project_scores_higher_than_far(self) -> None:
|
||
# near: release_month = выведенный месяц кандидата (timing 1.0); far: на 4 года
|
||
# позже (timing → почти 0). Прочие оси идентичны → near должен дать выше value.
|
||
near = _timing_card(
|
||
[_own("Близкий-во-времени", lon=_CENTROID[0], lat=_CENTROID[1],
|
||
release_month=_DERIVED_CANDIDATE_MONTH)]
|
||
)
|
||
far = _timing_card(
|
||
[_own("Далёкий-во-времени", lon=_CENTROID[0], lat=_CENTROID[1],
|
||
release_month=_add_months(_DERIVED_CANDIDATE_MONTH, 48))]
|
||
)
|
||
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_timing_excluded_when_launch_window_unavailable(self) -> None:
|
||
# Launch Window недоступен (deficit None на всех горизонтах) → release_month None
|
||
# → тайминговая ось ИСКЛЮЧЕНА (None-not-0), но каннибализация считается по
|
||
# классу/цене/гео (не падает, не фабрикует тайминг).
|
||
forecasts = [_forecast_stub(None, horizon=h) for h in (6, 12, 18, 24)]
|
||
own = _own(
|
||
"Наш-А", lon=_CENTROID[0], lat=_CENTROID[1],
|
||
release_month=date(2027, 1, 1), # есть дата, но кандидатной нет
|
||
)
|
||
db = MagicMock()
|
||
with (
|
||
_full_stack_patch(),
|
||
patch(_DSF, return_value=forecasts),
|
||
patch(f"{_MOD}.date", _FixedDate),
|
||
patch(f"{_MOD}.get_own_portfolio", return_value=[own]),
|
||
patch(f"{_MOD}._query_parcel_centroid", return_value=_CENTROID),
|
||
patch(
|
||
f"{_MOD}._query_artificial_demand",
|
||
return_value={"n_sold": 100, "n_mortgage": 40},
|
||
),
|
||
):
|
||
card = compute_special_indices(
|
||
db, spec=_CAND_SPEC, district="Академический", cad_num="66:41:0303161:123"
|
||
)
|
||
assert card.indices[KEY_LAUNCH_WINDOW].method == _METHOD_UNAVAILABLE
|
||
can = card.indices[KEY_CANNIBALIZATION]
|
||
# каннибализация всё равно посчитана (класс+цена), тайминговая ось исключена.
|
||
assert can.method == "own_portfolio_overlap"
|
||
assert can.value is not None
|
||
assert can.detail["axes_available"]["timing"] == 0 # НЕ сфабрикована
|
||
assert can.detail["top_contributors"][0]["axes"]["timing"] is None
|
||
|
||
def test_timing_deterministic_same_inputs_identical_as_dict(self) -> None:
|
||
# Детерминизм (§16): одинаковые входы (фикс. дата отчёта) → идентичный as_dict.
|
||
portfolio = [
|
||
_own("Наш-А", lon=_CENTROID[0], lat=_CENTROID[1],
|
||
release_month=_DERIVED_CANDIDATE_MONTH),
|
||
_own("Наш-Б", obj_class="комфорт+", lon=60.65, lat=56.85,
|
||
release_month=_add_months(_DERIVED_CANDIDATE_MONTH, 6)),
|
||
]
|
||
first = _timing_card(list(portfolio)).indices[KEY_CANNIBALIZATION].as_dict()
|
||
second = _timing_card(list(portfolio)).indices[KEY_CANNIBALIZATION].as_dict()
|
||
assert first == second
|
||
# подтверждаем, что тайминг реально участвовал (ось активна) — не пустой детерминизм.
|
||
assert first["detail"]["axes_available"]["timing"] == 2
|
||
|
||
|
||
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 == ["Андрей", "Яков"]
|