340 lines
16 KiB
Python
340 lines
16 KiB
Python
"""§9.7 forecast-overlay для живого квартирография-рекомендатора (#982, 953-A).
|
||
|
||
Это ТОНКИЙ СОВЕТУЮЩИЙ мост между ЖИВЫМ `analytics_queries.recommend_mix` и
|
||
форсайт-движком §9.x (#980/#981). Своей §9.x-математики НЕ пересобирает — собирает
|
||
ответ из уже-смерженных сервисов и форматирует под schemas.recommend overlay-контракт.
|
||
Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM; своего SQL НЕТ.
|
||
|
||
ADVISORY (КРИТИЧНО — честность, зеркало what_to_build.py): overlay наследует
|
||
advisory-статус §9.x (движок не провалидирован до бэктеста #951). Поэтому
|
||
`advisory` ВСЕГДА True, confidence ≤ 'medium' в demand_supply-режиме и 'low' в
|
||
demand_only. Цифры — для explainability/прототипа, НЕ основание для инвест-решения.
|
||
LIVE-ENDPOINT-SAFE: НИКОГДА не бросает — на любом тонком/сбойном входе возвращает
|
||
пустой ranked_segments + warning (вызывающий в recommend_mix ещё и оборачивает всё
|
||
в try/except, но мы и сами graceful).
|
||
|
||
ДВА РЕЖИМА (выбор по наличию геометрии участка `cad_num`):
|
||
• mode="demand_supply" (cad_num задан): полноценный §9.7-ранкер
|
||
`what_to_build.rank_segments` — спрос (§9.4ק9.5) vs предложение (§9.3) +
|
||
конкуренты участка → `deficit_index` ∈ [−1,+1] (>0 строить / <0 затоварка),
|
||
`balance_units`, confidence ≤ 'medium'. DESC-порядок наследуется от ранкера.
|
||
• mode="demand_only" (cad_num=None): БЕЗ геометрии участка supply/конкуренты
|
||
НЕизмеримы → считаем ТОЛЬКО темп спроса (§9.2 × §9.4 × §9.5) per default
|
||
room-bucket, ранжируем DESC по темпу. `deficit_index` здесь = pace/max_pace ∈
|
||
(0,1] — это ПРОКСИ относительной силы спроса, НЕ supply-based дефицит;
|
||
`balance_units=None`, confidence='low', обязательный warning что supply НЕ учтён.
|
||
НИКОГДА не фабрикуем предложение/конкурентов из воздуха.
|
||
|
||
ИМПОРТЫ §9.x — ЛОКАЛЬНЫЕ (внутри функций), чтобы избежать import-cycle: пакет
|
||
`forecasting/__init__` тянет `affordability`, который импортит
|
||
`analytics_queries._current_mortgage_rate`, а тот — часть живого стека. Top-level
|
||
здесь — только `sales_series.SegmentSpec` (лёгкий, без обратных зависимостей) +
|
||
stdlib/sqlalchemy.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Any
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.services.forecasting.sales_series import SegmentSpec
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── PURE bridge-таблицы (named, unit-тестируемы в обе стороны) ─────────────────
|
||
|
||
# Forecast room-bucket (Source-B room_area-вокабуляр what_to_build) → live
|
||
# bucket-id (ключ analytics_queries._BUCKET_PRETTY). ТОЧНЫЙ инверс _BUCKET_PRETTY;
|
||
# 5 литералов ДУБЛИРУЕМ намеренно (избегаем import-цикла analytics_queries →
|
||
# recommendation). Drift-guard в тестах сверяет это с inverse(_BUCKET_PRETTY).
|
||
_FORECAST_TO_LIVE_BUCKET: dict[str, str] = {
|
||
"Студии 15-30": "1-Студия",
|
||
"1-к 30-45": "2-1-к",
|
||
"2-к 45-60": "3-2-к",
|
||
"3-к 60-80": "4-3-к",
|
||
"80+ м²": "5-80+ м²",
|
||
}
|
||
|
||
# Инверс — live bucket-id → forecast room-bucket (для demand_only прохода по
|
||
# дефолтной сетке room-bucket'ов §9.7 и для map_room_bucket_inverse).
|
||
_LIVE_TO_FORECAST_BUCKET: dict[str, str] = {v: k for k, v in _FORECAST_TO_LIVE_BUCKET.items()}
|
||
|
||
# Live target_class (вокабуляр schemas.recommend ClassLiteral, «человеческий»
|
||
# регистр) → forecast obj_class (3 массовых класса §9.7 _DEFAULT_CLASSES). None и
|
||
# незнакомое → None (вызывающий тогда отдаёт движку дефолтную сетку классов).
|
||
_CLASS_TO_FORECAST: dict[str, str] = {
|
||
"Comfort": "комфорт",
|
||
"Comfort+": "комфорт",
|
||
"Business": "бизнес",
|
||
"Elite": "бизнес",
|
||
"Economy": "эконом",
|
||
}
|
||
|
||
|
||
def map_room_bucket(forecast_bucket: str | None) -> str | None:
|
||
"""Forecast room-bucket → live bucket-id. PURE. Неизвестное/None → None."""
|
||
if forecast_bucket is None:
|
||
return None
|
||
return _FORECAST_TO_LIVE_BUCKET.get(forecast_bucket)
|
||
|
||
|
||
def map_room_bucket_inverse(live_bucket: str | None) -> str | None:
|
||
"""Live bucket-id → forecast room-bucket (инверс map_room_bucket). PURE."""
|
||
if live_bucket is None:
|
||
return None
|
||
return _LIVE_TO_FORECAST_BUCKET.get(live_bucket)
|
||
|
||
|
||
def map_class(target_class: str | None) -> str | None:
|
||
"""Live target_class → forecast obj_class. PURE. None/неизвестное → None.
|
||
|
||
Comfort/Comfort+ сворачиваются в «комфорт», Business/Elite — в «бизнес»,
|
||
Economy — в «эконом» (3 массовых класса §9.7). None → None = «без фильтра
|
||
класса», вызывающий отдаёт движку дефолтную сетку классов.
|
||
"""
|
||
if target_class is None:
|
||
return None
|
||
return _CLASS_TO_FORECAST.get(target_class)
|
||
|
||
|
||
# ── DB-оркестратор overlay — тонкий, graceful, live-safe ──────────────────────
|
||
|
||
|
||
def build_forecast_overlay(
|
||
db: Session,
|
||
*,
|
||
district: str | None,
|
||
cad_num: str | None,
|
||
horizon_months: int,
|
||
target_class: str | None,
|
||
market_buckets: list[dict[str, Any]] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Собрать СОВЕТУЮЩИЙ §9.7 forecast-overlay для recommend_mix. LIVE-SAFE.
|
||
|
||
ADVISORY (наследует advisory-статус §9.x, см. module docstring) — overlay
|
||
исключительно explainability, НЕ основание для инвест-решения. НИКОГДА не
|
||
бросает: на тонком/сбойном входе → пустой ranked_segments + warning.
|
||
|
||
Режим выбирается по `cad_num`:
|
||
• cad_num задан → mode="demand_supply": §9.7 rank_segments (спрос vs
|
||
предложение + конкуренты участка), маппинг RankedSegment → segment-dict,
|
||
DESC-порядок сохранён.
|
||
• cad_num=None → mode="demand_only": ТОЛЬКО темп спроса (§9.2ק9.4ק9.5)
|
||
per default room-bucket, DESC по темпу, deficit_index = pace/max_pace
|
||
(ПРОКСИ, не supply-based), balance_units=None, confidence='low' +
|
||
обязательный warning что supply/конкуренты не учтены.
|
||
|
||
Args:
|
||
db: SQLAlchemy sync Session.
|
||
district: район участка (None → ЕКБ-wide).
|
||
cad_num: кадастровый номер участка; None → demand_only режим.
|
||
horizon_months: горизонт прогноза (мес) — продуктовый «на горизонте».
|
||
target_class: живой класс недвижимости (ClassLiteral) или None.
|
||
market_buckets: живые бакеты recommend_mix (пока для совместимости
|
||
сигнатуры; demand_only ходит по дефолтной §9.7 room-сетке).
|
||
|
||
Returns:
|
||
RecommendForecastOverlay-shaped dict (horizon_months/mode/advisory/
|
||
ranked_segments/warnings). advisory ВСЕГДА True.
|
||
"""
|
||
if cad_num is not None:
|
||
return _demand_supply_overlay(
|
||
db,
|
||
district=district,
|
||
cad_num=cad_num,
|
||
horizon_months=horizon_months,
|
||
target_class=target_class,
|
||
)
|
||
return _demand_only_overlay(
|
||
db,
|
||
district=district,
|
||
horizon_months=horizon_months,
|
||
target_class=target_class,
|
||
)
|
||
|
||
|
||
def _demand_supply_overlay(
|
||
db: Session,
|
||
*,
|
||
district: str | None,
|
||
cad_num: str,
|
||
horizon_months: int,
|
||
target_class: str | None,
|
||
) -> dict[str, Any]:
|
||
"""mode="demand_supply": §9.7 rank_segments → overlay-dict. Graceful → []."""
|
||
# Локальный импорт — избегаем import-cycle (forecasting/__init__ → affordability
|
||
# → analytics_queries). См. module docstring.
|
||
from app.services.forecasting.what_to_build import rank_segments
|
||
|
||
warnings: list[str] = []
|
||
mapped_class = map_class(target_class)
|
||
# None класс → ОПУСКАЕМ kwarg classes, чтобы применилась движковая дефолтная
|
||
# сетка §9.7 (rank_segments default = _DEFAULT_CLASSES); room_buckets всегда
|
||
# опускаем → дефолтная §9.7 сетка форматов. Передать None нельзя — ранкер
|
||
# итерирует по этим осям (None → TypeError).
|
||
rank_kwargs: dict[str, Any] = {}
|
||
if mapped_class is not None:
|
||
rank_kwargs["classes"] = [mapped_class]
|
||
|
||
try:
|
||
ranking = rank_segments(
|
||
db,
|
||
district=district,
|
||
cad_num=cad_num,
|
||
horizon_months=horizon_months,
|
||
**rank_kwargs,
|
||
)
|
||
except ValueError as e:
|
||
# Нет геометрии участка / невалидный вход — НЕ валим живой ответ.
|
||
logger.info(
|
||
"forecast-overlay demand_supply: rank_segments ValueError "
|
||
"(district=%s cad_num=%s horizon=%d) → пустой ранкинг: %s",
|
||
district,
|
||
cad_num,
|
||
horizon_months,
|
||
e,
|
||
)
|
||
warnings.append(
|
||
"demand_supply: ранкинг недоступен (нет геометрии участка или тонкие "
|
||
"данные) — сегменты не ранжированы."
|
||
)
|
||
return _overlay(horizon_months, "demand_supply", [], warnings)
|
||
|
||
ranked_segments: list[dict[str, Any]] = []
|
||
for seg in ranking.ranked: # уже DESC по deficit_index
|
||
live_bucket = map_room_bucket(seg.segment.get("room_bucket"))
|
||
if live_bucket is None:
|
||
# Сегмент в неизвестном room-вокабуляре — отбрасываем (не отображаем мусор).
|
||
continue
|
||
ranked_segments.append(
|
||
{
|
||
"bucket": live_bucket,
|
||
"obj_class": seg.segment.get("obj_class"),
|
||
"deficit_index": seg.deficit_index,
|
||
"balance_units": seg.balance_units,
|
||
"confidence": seg.confidence,
|
||
}
|
||
)
|
||
|
||
if not ranked_segments:
|
||
warnings.append(
|
||
"demand_supply: все ячейки сетки тонкие (предложение неизмеримо) — "
|
||
"сегменты не ранжированы."
|
||
)
|
||
|
||
logger.info(
|
||
"forecast-overlay demand_supply: district=%s cad_num=%s horizon=%d ranked=%d (ADVISORY)",
|
||
district,
|
||
cad_num,
|
||
horizon_months,
|
||
len(ranked_segments),
|
||
)
|
||
return _overlay(horizon_months, "demand_supply", ranked_segments, warnings)
|
||
|
||
|
||
def _demand_only_overlay(
|
||
db: Session,
|
||
*,
|
||
district: str | None,
|
||
horizon_months: int,
|
||
target_class: str | None,
|
||
) -> dict[str, Any]:
|
||
"""mode="demand_only" (cad_num=None): ТОЛЬКО темп спроса per room-bucket.
|
||
|
||
БЕЗ геометрии участка supply/конкуренты НЕизмеримы. Считаем темп спроса
|
||
pace = §9.2 unit_velocity × §9.4 norm.coefficient × §9.5 macro.coefficient
|
||
per дефолтный room-bucket, ранжируем DESC. deficit_index = pace/max_pace ∈
|
||
(0,1] — ПРОКСИ относительной силы спроса (НЕ supply-based дефицит).
|
||
balance_units=None, confidence='low', обязательный warning. НИКОГДА не
|
||
фабрикуем supply. Graceful → [].
|
||
"""
|
||
# Локальные импорты — избегаем import-cycle (см. module docstring).
|
||
from app.services.forecasting.demand_normalization import compute_demand_normalization
|
||
from app.services.forecasting.demand_supply_forecast import hold_last_rate
|
||
from app.services.forecasting.macro_coefficient import compute_macro_coefficient
|
||
from app.services.forecasting.macro_series import get_monthly_macro
|
||
from app.services.site_finder.market_metrics import compute_market_metrics
|
||
|
||
warnings: list[str] = [
|
||
"demand-only: без геометрии участка (cad_num) supply/конкуренты НЕ учтены"
|
||
]
|
||
mapped_class = map_class(target_class)
|
||
|
||
# Один раз на вызов: §9.2 наблюдаемый темп (base_pace) + hold-last-rate для §9.4.
|
||
metrics = compute_market_metrics(db, district=district)
|
||
base_pace = metrics.unit_velocity
|
||
macro = get_monthly_macro(db)
|
||
rate_future = hold_last_rate(macro, [horizon_months]).get(horizon_months)
|
||
|
||
if base_pace is None:
|
||
# Нет наблюдаемого темпа — ранжировать нечего (НЕ фабрикуем 0-сигнал).
|
||
warnings.append(
|
||
"demand-only: нет наблюдаемого темпа продаж (§9.2) — сегменты не ранжированы."
|
||
)
|
||
return _overlay(horizon_months, "demand_only", [], warnings)
|
||
|
||
paces: list[tuple[str, str | None, float]] = [] # (live_bucket, obj_class, pace)
|
||
for live_bucket, forecast_bucket in _LIVE_TO_FORECAST_BUCKET.items():
|
||
spec = SegmentSpec(
|
||
obj_class=mapped_class,
|
||
room_bucket=forecast_bucket,
|
||
district=district,
|
||
)
|
||
# §9.4 нормализация под будущий режим ставки (β внутри; rate_future None →
|
||
# деградирует к нейтрали внутри себя, передаём 0.0 как placeholder).
|
||
norm = compute_demand_normalization(
|
||
db, spec=spec, rate_future=rate_future if rate_future is not None else 0.0
|
||
)
|
||
# §9.5 макро-коэффициент (ортогонален β); профиль — класс + room_bucket.
|
||
profile: dict[str, Any] = {"room_bucket": forecast_bucket}
|
||
if mapped_class is not None:
|
||
profile["obj_class"] = mapped_class
|
||
macro_coef = compute_macro_coefficient(db, segment_profile=profile)
|
||
pace = base_pace * norm.coefficient * macro_coef.coefficient
|
||
paces.append((live_bucket, mapped_class, pace))
|
||
|
||
max_pace = max((p for _, _, p in paces), default=0.0)
|
||
if max_pace <= 0.0:
|
||
warnings.append("demand-only: темп спроса по всем форматам ≤0 — сегменты не ранжированы.")
|
||
return _overlay(horizon_months, "demand_only", [], warnings)
|
||
|
||
# DESC по темпу; deficit_index = pace/max_pace ∈ (0,1] (ПРОКСИ, НЕ supply-based).
|
||
paces.sort(key=lambda t: t[2], reverse=True)
|
||
ranked_segments: list[dict[str, Any]] = [
|
||
{
|
||
"bucket": live_bucket,
|
||
"obj_class": obj_class,
|
||
"deficit_index": pace / max_pace,
|
||
"balance_units": None, # supply неизмеримо без cad_num — НЕ фабрикуем
|
||
"confidence": "low",
|
||
}
|
||
for live_bucket, obj_class, pace in paces
|
||
]
|
||
|
||
logger.info(
|
||
"forecast-overlay demand_only: district=%s horizon=%d ranked=%d base_pace=%.3f "
|
||
"(ADVISORY, supply НЕ учтён)",
|
||
district,
|
||
horizon_months,
|
||
len(ranked_segments),
|
||
base_pace,
|
||
)
|
||
return _overlay(horizon_months, "demand_only", ranked_segments, warnings)
|
||
|
||
|
||
def _overlay(
|
||
horizon_months: int,
|
||
mode: str,
|
||
ranked_segments: list[dict[str, Any]],
|
||
warnings: list[str],
|
||
) -> dict[str, Any]:
|
||
"""Собрать RecommendForecastOverlay-shaped dict. advisory ВСЕГДА True. PURE."""
|
||
return {
|
||
"horizon_months": horizon_months,
|
||
"mode": mode,
|
||
"advisory": True,
|
||
"ranked_segments": ranked_segments,
|
||
"warnings": warnings,
|
||
}
|