lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions
gendesign/backend/tests/services/forecasting/test_special_indices.py
Light1YT cd48a095c0
Some checks failed
Deploy / build-worker (push) Successful in 2m32s
Details
Deploy / changes (push) Successful in 6s
Details
Deploy / build-frontend (push) Has been skipped
Details
Deploy / build-backend (push) Successful in 1m32s
Details
Deploy / deploy (push) Has been cancelled
Details
feat(forecasting): activate §25.3 cannibalization timing axis from launch window (#1169) 
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
2026-06-09 07:10:59 +00:00

1610 lines
73 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Тесты §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 == ["Андрей", "Яков"]
Reference in a new issue View git blame Copy permalink