Light1YT
d6ebdaf670
Add assemble_report() — PURE deterministic mapper populating SiteFinderReport (#987) from an analyze dict + advisory forecast-stack outputs, computing report confidence via compute_report_confidence (#990). No DB/LLM/SQL; accepts dataclass instances or their as_dict() dicts; graceful-partial (any None → partial section, never crashes); exec_summary synthesized from strongest signals (RU templates, no LLM); advisory always True. 49 unit tests.
900 lines
45 KiB
Python
900 lines
45 KiB
Python
"""§13 сборщик итогового советующего отчёта Site Finder v2 — ЧИСТОЕ маппирование.
|
||
|
||
#988 (955-A2, Site Finder v2 / «GG-форсайт» ТЗ §13), EPIC 11 «Отчёт». Это **СБОРЩИК**
|
||
(assembler): он НАПОЛНЯЕТ структурный контейнер `SiteFinderReport` (#987) уже-
|
||
посчитанными advisory-выводами форсайт-стека (#980…#986 / §9.1-9.3) + dict'ом
|
||
`analyze` (вывод эндпоинта analyze_parcel) и считает отчётную уверенность через
|
||
`compute_report_confidence` (#990). Своей §9.x-математики НЕ пересобирает и в БД НЕ
|
||
ходит — берёт уже-готовые входы (их считает будущий вызывающий: #961 API / #957 чат)
|
||
и РАСКЛАДЫВАЕТ их по восьми типизированным секциям §13. Поэтому он ЮНИТ-ТЕСТИРУЕТСЯ
|
||
БЕЗ БД. Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM (шаблонные RU-строки, не генерация).
|
||
|
||
ADVISORY (зеркало всего стека #980…#986 / #987 / #990): отчёт лишь СОБИРАЕТ advisory-
|
||
выводы — `advisory` ВСЕГДА True, прокидывается во все под-вызовы (в т.ч. cap-потолок
|
||
#990 'medium'). Не основание для инвест-решения.
|
||
|
||
ВОСЕМЬ СЕКЦИЙ §13 — ИЗ ЧЕГО собирается каждая:
|
||
1. exec_summary — СИНТЕЗ заголовка/вердикта/ключевых чисел из сильнейших сигналов
|
||
(deficit_index, рекомендованный класс/mix, overall-скор, уверенность). Шаблонные
|
||
RU-строки (НЕ LLM).
|
||
2. market_now — из `analyze` dict (абсорбция/цены/конкуренты/POI) + `market_metrics`
|
||
(§9.2) + `supply_layers` (§9.3). Defensive `.get()` (analyze loosely-typed).
|
||
3. future_market — `forecasts` (per-горизонт #952), `future_supply` (§9.3), будущие
|
||
конкуренты, сводка `scenarios` (#984).
|
||
4. product_tz — `recommendation_overlay` (#983: класс §10.2 / mix / commercial §10.4
|
||
/ USP §10.5 + §16-причины).
|
||
5. scenarios — `scenarios` (три ScenarioForecast #984, by_scenario).
|
||
6. scoring — `product_scores` (#985, 10 скоров) + `special_indices` (#986, 6) +
|
||
overall.
|
||
7. confidence — `compute_report_confidence(...)` (#990) из извлечённых сигналов
|
||
качества данных (deal_count/analog_count/domrf_coverage/history_months/confounded +
|
||
component-confidences). ReportConfidenceResult.as_dict() ложится в слот.
|
||
8. meta — cad_num/district/segment/horizons/advisory/schema_version.
|
||
|
||
GRACEFUL (дух всего стека): любой вход None → его секция частично заполнена / пуста —
|
||
отчёт ВАЛИДЕН как частичный (#987: все поля Optional). НИКОГДА не бросает: каждый вход
|
||
нормализуется через `_as_dict_or` (dataclass с `.as_dict()` ИЛИ уже-dict ИЛИ None),
|
||
чтобы вызывающий мог передать как объекты, так и их `.as_dict()`-словари.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from collections.abc import Sequence
|
||
from typing import Any
|
||
|
||
from app.services.forecasting.confidence_engine import (
|
||
Confidence,
|
||
compute_report_confidence,
|
||
)
|
||
from app.services.forecasting.report import (
|
||
ReportConfidence,
|
||
ReportConfidenceLevel,
|
||
ReportExecSummary,
|
||
ReportFutureMarket,
|
||
ReportMarketNow,
|
||
ReportMeta,
|
||
ReportProductTz,
|
||
ReportScenarios,
|
||
ReportScoring,
|
||
SiteFinderReport,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── Named-константы ───────────────────────────────────────────────────────────
|
||
|
||
# Горизонты прогноза по умолчанию (мес) — зеркало #952/#984 _DEFAULT_HORIZONS.
|
||
_DEFAULT_HORIZONS: tuple[int, ...] = (6, 12, 18, 24)
|
||
|
||
# Весь отчёт советующий (зеркало advisory-cap всего стека) — прокидывается в #990 cap.
|
||
_ADVISORY: bool = True
|
||
|
||
# Ранг уверенности (для выбора «худшего» компонента при синтезе exec_summary).
|
||
_CONFIDENCE_RANK: dict[Confidence, int] = {"low": 0, "medium": 1, "high": 2}
|
||
|
||
# RU-метки уровня уверенности для exec_summary-вердикта.
|
||
_LEVEL_RU: dict[ReportConfidenceLevel, str] = {
|
||
"high": "высокая",
|
||
"medium": "средняя",
|
||
"low": "низкая",
|
||
}
|
||
|
||
# Порог deficit_index, выше которого сигнал «строить» считаем сильным (зеркало духа
|
||
# special_indices._VOID_THRESHOLD: умеренно-сильный дефицит на лог-шкале #980 [−1,+1]).
|
||
_STRONG_DEFICIT_THRESHOLD: float = 0.25
|
||
|
||
# Сегментный горизонт по умолчанию для извлечения сигналов из forecasts (мес). Берём
|
||
# 12 — типовой средне-срочный продуктовый горизонт (зеркало #982/#983/#986 default).
|
||
_PRIMARY_HORIZON_MONTHS: int = 12
|
||
|
||
|
||
def _as_dict_or(x: Any) -> dict[str, Any] | None:
|
||
"""Нормализовать вход в JSON-safe dict: dataclass-с-`as_dict()` / dict / None. PURE.
|
||
|
||
Вызывающий (#961 API / #957 чат) может передать КАК объект под-сервиса (например
|
||
`MarketMetrics`), ТАК и его уже-посчитанный `.as_dict()`-словарь. Приводим к единой
|
||
форме (плоский JSON-safe dict):
|
||
• None → None (graceful — секция останется пустой).
|
||
• объект с методом `as_dict` → `x.as_dict()` (под-сервис сам отдаёт JSON-safe вид).
|
||
• уже-dict → как есть.
|
||
Любой иной тип (мусор) → None (НЕ бросаем — отчёт валиден как частичный).
|
||
|
||
Args:
|
||
x: dataclass-инстанс под-сервиса, его `.as_dict()`-словарь, или None.
|
||
|
||
Returns:
|
||
JSON-safe dict или None.
|
||
"""
|
||
if x is None:
|
||
return None
|
||
as_dict = getattr(x, "as_dict", None)
|
||
if callable(as_dict):
|
||
result = as_dict()
|
||
return result if isinstance(result, dict) else None
|
||
if isinstance(x, dict):
|
||
return x
|
||
return None
|
||
|
||
|
||
def _as_dict_list(items: Sequence[Any] | None) -> list[dict[str, Any]]:
|
||
"""Нормализовать последовательность входов в список JSON-safe dict'ов. PURE.
|
||
|
||
Каждый элемент через `_as_dict_or`; None-элементы (нормализовавшиеся в None)
|
||
отбрасываем. None/пустой вход → [] (graceful). Сохраняет порядок.
|
||
"""
|
||
if not items:
|
||
return []
|
||
out: list[dict[str, Any]] = []
|
||
for item in items:
|
||
normalized = _as_dict_or(item)
|
||
if normalized is not None:
|
||
out.append(normalized)
|
||
return out
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Извлечение сигналов качества данных для §15-уверенности (#990). Каждый pure,
|
||
# graceful: вход None/тонкий → None (#990 трактует None как «нет сигнала»).
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _deal_count(analyze: dict[str, Any], market_metrics: dict[str, Any] | None) -> int | None:
|
||
"""Число сделок (продаж) за окно — для deal_count #990. PURE.
|
||
|
||
Предпочитаем `market_metrics.n_sold` (§9.2 — прямой счётчик проданных лотов выборки);
|
||
fallback на `analyze.market_pulse.competitors_total` НЕ берём (это число ЖК, не
|
||
сделок — пошло бы в analog_count). Нет сигнала → None (#990 → тянет в low).
|
||
"""
|
||
if market_metrics is not None:
|
||
n_sold = market_metrics.get("n_sold")
|
||
if isinstance(n_sold, int):
|
||
return n_sold
|
||
return None
|
||
|
||
|
||
def _analog_count(analyze: dict[str, Any], market_metrics: dict[str, Any] | None) -> int | None:
|
||
"""Число ЖК-аналогов в выборке — для analog_count #990. PURE.
|
||
|
||
Предпочитаем `market_metrics.obj_count` (§9.2 — число отдельных ЖК выборки, ТОЧНО
|
||
то, что #990 ждёт). Fallback — число конкурентов из `analyze` (market_pulse
|
||
competitors_total ИЛИ длина списка competitors): это окружение участка, тоже мера
|
||
«сколько объектов-аналогов рядом». Нет сигнала → None.
|
||
"""
|
||
if market_metrics is not None:
|
||
obj_count = market_metrics.get("obj_count")
|
||
if isinstance(obj_count, int):
|
||
return obj_count
|
||
pulse = analyze.get("market_pulse")
|
||
if isinstance(pulse, dict):
|
||
total = pulse.get("competitors_total")
|
||
if isinstance(total, int):
|
||
return total
|
||
competitors = analyze.get("competitors")
|
||
if isinstance(competitors, list):
|
||
return len(competitors)
|
||
return None
|
||
|
||
|
||
def _domrf_coverage(analyze: dict[str, Any], supply_layers: dict[str, Any] | None) -> float | None:
|
||
"""Покрытие domrf↔objective ∈ [0,1] — для domrf_coverage #990. PURE.
|
||
|
||
Главный sparse-риск проекта (~2.5%). Источники по приоритету:
|
||
• `supply_layers.domrf_coverage` (если §9.3-слой его несёт явно).
|
||
• `analyze.market_data_coverage_pct` (доля конкурентов с ценой из objective, %) —
|
||
прокси покрытия objective-данными; делим на 100 → доля.
|
||
Нет сигнала → None (#990 → тянет в low: слой §9.3 недооценён).
|
||
"""
|
||
if supply_layers is not None:
|
||
coverage = supply_layers.get("domrf_coverage")
|
||
if isinstance(coverage, (int, float)) and not isinstance(coverage, bool):
|
||
return _coverage_to_fraction(float(coverage))
|
||
pct = analyze.get("market_data_coverage_pct")
|
||
if isinstance(pct, (int, float)) and not isinstance(pct, bool):
|
||
return _coverage_to_fraction(float(pct))
|
||
return None
|
||
|
||
|
||
def _coverage_to_fraction(value: float) -> float:
|
||
"""Привести покрытие к доле ∈ [0,1]: значения >1 трактуем как проценты. PURE.
|
||
|
||
supply_layers может отдать долю (0.025), analyze — проценты (2.5). Эвристика: >1 →
|
||
делим на 100. Зажимаем в [0,1] (защита от грязных данных).
|
||
"""
|
||
fraction = value / 100.0 if value > 1.0 else value
|
||
return max(0.0, min(1.0, fraction))
|
||
|
||
|
||
def _history_months(
|
||
market_metrics: dict[str, Any] | None, forecasts: Sequence[dict[str, Any]]
|
||
) -> int | None:
|
||
"""Глубина ряда (мес) — для history_months #990. PURE.
|
||
|
||
`market_metrics.window_months` (§9.2 окно наблюдения продаж) — единственный явный
|
||
сигнал глубины истории в доступных входах. Нет → None (#990 → короткий ряд тянет
|
||
вниз). forecasts оставлены в сигнатуре для будущего расширения (на сейчас ряд
|
||
глубины в них не выносится).
|
||
"""
|
||
if market_metrics is not None:
|
||
window = market_metrics.get("window_months")
|
||
if isinstance(window, int) and window > 0:
|
||
return window
|
||
return None
|
||
|
||
|
||
def _confounded(forecasts: Sequence[dict[str, Any]]) -> bool:
|
||
"""Пересекает ли окно прогноза шок-период — для confounded #990. PURE.
|
||
|
||
Любой per-горизонт forecast несёт флаг `confounded`/`is_confounded_window` (PR2)?
|
||
Если хоть один True → отчётное окно считаем confounded (оценки смещены, #990 →
|
||
НИКОГДА не 'high'). Нет флага нигде → False (чистое окно).
|
||
"""
|
||
for f in forecasts:
|
||
if f.get("confounded") is True or f.get("is_confounded_window") is True:
|
||
return True
|
||
return False
|
||
|
||
|
||
def _component_confidences(
|
||
market_metrics: dict[str, Any] | None,
|
||
future_supply: dict[str, Any] | None,
|
||
forecasts: Sequence[dict[str, Any]],
|
||
product_scores: dict[str, Any] | None,
|
||
special_indices: dict[str, Any] | None,
|
||
) -> list[Confidence]:
|
||
"""Собрать per-service confidence вкладывающих под-сервисов — для #990. PURE.
|
||
|
||
#990 свернёт их weakest-link (MIN) вместе с сырыми счётчиками. Берём confidence-
|
||
метку каждого доступного под-вывода (market_metrics §9.2 / future_supply §9.3 /
|
||
каждый per-горизонт forecast #952 / product_scores #985 / special_indices #986).
|
||
Только whitelisted 'high|medium|low' (мусор/None отбрасываем).
|
||
"""
|
||
out: list[Confidence] = []
|
||
for source in (market_metrics, future_supply, product_scores, special_indices):
|
||
conf = _confidence_of(source)
|
||
if conf is not None:
|
||
out.append(conf)
|
||
for f in forecasts:
|
||
conf = _confidence_of(f)
|
||
if conf is not None:
|
||
out.append(conf)
|
||
return out
|
||
|
||
|
||
def _confidence_of(source: dict[str, Any] | None) -> Confidence | None:
|
||
"""Достать whitelisted confidence-метку из под-вывода (или None). PURE."""
|
||
if not isinstance(source, dict):
|
||
return None
|
||
conf = source.get("confidence")
|
||
if conf in _CONFIDENCE_RANK:
|
||
return conf # type: ignore[return-value] # сужено проверкой членства
|
||
return None
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Извлечение опорных чисел для exec_summary-синтеза. Каждый pure, graceful → None.
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _primary_deficit_index(forecasts: Sequence[dict[str, Any]]) -> float | None:
|
||
"""deficit_index основного горизонта (или первого доступного). PURE.
|
||
|
||
Предпочитаем _PRIMARY_HORIZON_MONTHS (12 мес — продуктовый горизонт); если его нет —
|
||
первый forecast с не-None deficit_index. Нет → None.
|
||
"""
|
||
primary = next(
|
||
(f for f in forecasts if f.get("horizon_months") == _PRIMARY_HORIZON_MONTHS),
|
||
None,
|
||
)
|
||
if primary is not None and primary.get("deficit_index") is not None:
|
||
di = primary["deficit_index"]
|
||
return float(di) if isinstance(di, (int, float)) and not isinstance(di, bool) else None
|
||
for f in forecasts:
|
||
di = f.get("deficit_index")
|
||
if isinstance(di, (int, float)) and not isinstance(di, bool):
|
||
return float(di)
|
||
return None
|
||
|
||
|
||
def _recommended_class(product_tz: dict[str, Any] | None) -> str | None:
|
||
"""Рекомендованный класс из product_tz-вывода (overlay class_reco / obj_class). PURE."""
|
||
if not isinstance(product_tz, dict):
|
||
return None
|
||
obj_class = product_tz.get("obj_class")
|
||
if isinstance(obj_class, str) and obj_class:
|
||
return obj_class
|
||
class_reco = product_tz.get("class_reco")
|
||
if isinstance(class_reco, dict):
|
||
reco_class = class_reco.get("obj_class")
|
||
if isinstance(reco_class, str) and reco_class:
|
||
return reco_class
|
||
return None
|
||
|
||
|
||
def _overall_score(product_scores: dict[str, Any] | None) -> float | None:
|
||
"""overall-скор из карты #985 (ProductScoreCard.as_dict()['overall']). PURE."""
|
||
if not isinstance(product_scores, dict):
|
||
return None
|
||
overall = product_scores.get("overall")
|
||
if isinstance(overall, (int, float)) and not isinstance(overall, bool):
|
||
return float(overall)
|
||
return None
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Per-section builders — каждый PURE, graceful (входы None → пустая/частичная секция).
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _build_market_now(
|
||
analyze: dict[str, Any],
|
||
market_metrics: dict[str, Any] | None,
|
||
supply_layers: dict[str, Any] | None,
|
||
) -> ReportMarketNow:
|
||
"""§13.2 market_now — текущий рынок из analyze + §9.2/§9.3. PURE, defensive.
|
||
|
||
`market_metrics` (§9.2) / `supply_layers` (§9.3) кладём как есть (их as_dict уже
|
||
JSON-safe). `competitors` берём из analyze (defensive: список dict'ов конкурентов).
|
||
`summary` синтезируем из доступных абсорбции/цены/числа конкурентов. analyze loosely-
|
||
typed → всё через `.get()`.
|
||
"""
|
||
competitors = analyze.get("competitors")
|
||
competitor_list = (
|
||
[c for c in competitors if isinstance(c, dict)] if isinstance(competitors, list) else []
|
||
)
|
||
summary = _market_now_summary(analyze, market_metrics)
|
||
return ReportMarketNow(
|
||
market_metrics=market_metrics,
|
||
competitors=competitor_list,
|
||
supply_layers=supply_layers,
|
||
summary=summary,
|
||
)
|
||
|
||
|
||
def _market_now_summary(
|
||
analyze: dict[str, Any], market_metrics: dict[str, Any] | None
|
||
) -> str | None:
|
||
"""Короткий RU-текст про текущий рынок из доступных сигналов. PURE, шаблонный.
|
||
|
||
Собирает фразу из абсорбции (§9.2 unit_velocity), средней цены (analyze) и числа
|
||
конкурентов. Ни одного сигнала → None (graceful — секция без summary валидна).
|
||
"""
|
||
parts: list[str] = []
|
||
if market_metrics is not None:
|
||
velocity = market_metrics.get("unit_velocity")
|
||
if isinstance(velocity, (int, float)) and not isinstance(velocity, bool):
|
||
parts.append(f"абсорбция ~{round(float(velocity), 1)} ед./мес")
|
||
avg_price = analyze.get("market_avg_price_per_m2")
|
||
if isinstance(avg_price, (int, float)) and not isinstance(avg_price, bool):
|
||
parts.append(f"средняя цена ~{round(float(avg_price)):,} ₽/м²".replace(",", " "))
|
||
n_comp = _analog_count(analyze, market_metrics)
|
||
if n_comp is not None:
|
||
parts.append(f"{n_comp} ЖК-конкурентов рядом")
|
||
if not parts:
|
||
return None
|
||
return "Текущий рынок: " + ", ".join(parts) + "."
|
||
|
||
|
||
def _build_future_market(
|
||
forecasts: Sequence[dict[str, Any]],
|
||
future_supply: dict[str, Any] | None,
|
||
future_competitors: list[dict[str, Any]],
|
||
scenarios_summary: dict[str, Any] | None,
|
||
) -> ReportFutureMarket:
|
||
"""§13.3 future_market — прогноз из forecasts/future_supply/scenarios. PURE.
|
||
|
||
`forecasts_by_horizon` — список per-горизонт #952; `future_supply` (§9.3) как есть;
|
||
`future_competitors` — будущие конкуренты (из forecasts или явный аргумент);
|
||
`scenarios_summary` — сводка #984. `summary` синтезируем из дефицита основного
|
||
горизонта. Всё пустое → пустая секция (graceful).
|
||
"""
|
||
summary = _future_market_summary(forecasts)
|
||
return ReportFutureMarket(
|
||
forecasts_by_horizon=list(forecasts),
|
||
future_supply=future_supply,
|
||
future_competitors=future_competitors,
|
||
scenarios_summary=scenarios_summary,
|
||
summary=summary,
|
||
)
|
||
|
||
|
||
def _future_market_summary(forecasts: Sequence[dict[str, Any]]) -> str | None:
|
||
"""Короткий RU-текст про будущий рынок (дефицит/затоварка осн. горизонта). PURE."""
|
||
di = _primary_deficit_index(forecasts)
|
||
if di is None:
|
||
return None
|
||
if di >= _STRONG_DEFICIT_THRESHOLD:
|
||
verdict = "выраженный дефицит предложения (сигнал «строить»)"
|
||
elif di > 0:
|
||
verdict = "умеренный дефицит предложения"
|
||
elif di == 0:
|
||
verdict = "баланс спроса и предложения"
|
||
else:
|
||
verdict = "затоварка (предложения больше спроса)"
|
||
return f"Прогноз: {verdict}, индекс дефицита {_signed(di)} на горизонте."
|
||
|
||
|
||
def _build_product_tz(product_tz: dict[str, Any] | None) -> ReportProductTz:
|
||
"""§13.4 product_tz — рекомендация продукта из overlay #983. PURE, defensive.
|
||
|
||
Маппит overlay-форму (#983 build_forecast_overlay as-dict): класс §10.2 (class_reco
|
||
/ obj_class), квартирография (mix / ranked_segments), commercial §10.4, USP §10.5,
|
||
§16-причины (reasons / class_reco.reason). overlay None → пустая секция (graceful).
|
||
"""
|
||
if not isinstance(product_tz, dict):
|
||
return ReportProductTz()
|
||
obj_class = _recommended_class(product_tz)
|
||
mix = _extract_mix(product_tz)
|
||
commercial = (
|
||
product_tz.get("commercial") if isinstance(product_tz.get("commercial"), dict) else None
|
||
)
|
||
usp = _extract_list(product_tz, "usp")
|
||
reasons = _extract_reasons(product_tz)
|
||
summary = _product_tz_summary(obj_class, usp)
|
||
return ReportProductTz(
|
||
obj_class=obj_class,
|
||
mix=mix,
|
||
commercial=commercial,
|
||
usp=usp,
|
||
reasons=reasons,
|
||
summary=summary,
|
||
)
|
||
|
||
|
||
def _extract_mix(product_tz: dict[str, Any]) -> list[dict[str, Any]]:
|
||
"""Квартирография — явный `mix` ИЛИ ranked_segments overlay (формат+дефицит). PURE.
|
||
|
||
Overlay #983 не несёт готового `mix`, но его `ranked_segments` (DESC по дефициту) —
|
||
продуктовый ответ «какой формат строить». Берём `mix` если задан, иначе проецируем
|
||
ranked_segments в компактные {bucket, obj_class, deficit_index}. Нет ни того, ни
|
||
другого → [].
|
||
"""
|
||
explicit = product_tz.get("mix")
|
||
if isinstance(explicit, list):
|
||
return [m for m in explicit if isinstance(m, dict)]
|
||
ranked = product_tz.get("ranked_segments")
|
||
if isinstance(ranked, list):
|
||
return [
|
||
{
|
||
"bucket": seg.get("bucket"),
|
||
"obj_class": seg.get("obj_class"),
|
||
"deficit_index": seg.get("deficit_index"),
|
||
}
|
||
for seg in ranked
|
||
if isinstance(seg, dict)
|
||
]
|
||
return []
|
||
|
||
|
||
def _extract_list(source: dict[str, Any], key: str) -> list[dict[str, Any]]:
|
||
"""Достать список dict'ов по ключу (мусор/None → []). PURE."""
|
||
value = source.get(key)
|
||
if isinstance(value, list):
|
||
return [item for item in value if isinstance(item, dict)]
|
||
return []
|
||
|
||
|
||
def _extract_reasons(product_tz: dict[str, Any]) -> list[dict[str, Any]]:
|
||
"""§16-причины — явные `reasons` ИЛИ §16-reason из class_reco/usp overlay. PURE.
|
||
|
||
overlay #983 кладёт §16-reason ВНУТРЬ class_reco / каждого usp / ranked_segment.
|
||
Если явного `reasons` нет — поднимаем reason из class_reco (рекомендации класса) на
|
||
уровень секции (главная §16-причина продукта). Нет → [].
|
||
"""
|
||
explicit = product_tz.get("reasons")
|
||
if isinstance(explicit, list):
|
||
return [r for r in explicit if isinstance(r, dict)]
|
||
class_reco = product_tz.get("class_reco")
|
||
if isinstance(class_reco, dict):
|
||
reason = class_reco.get("reason")
|
||
if isinstance(reason, dict):
|
||
return [reason]
|
||
return []
|
||
|
||
|
||
def _product_tz_summary(obj_class: str | None, usp: list[dict[str, Any]]) -> str | None:
|
||
"""Короткий RU-текст про рекомендованный продукт. PURE, шаблонный."""
|
||
if obj_class is None and not usp:
|
||
return None
|
||
parts: list[str] = []
|
||
if obj_class is not None:
|
||
parts.append(f"рекомендован класс «{obj_class}»")
|
||
if usp:
|
||
parts.append(f"{len(usp)} USP-ниш по дефициту форматов")
|
||
return "Продукт: " + ", ".join(parts) + "."
|
||
|
||
|
||
def _build_scenarios(scenarios: Sequence[dict[str, Any]] | None) -> ReportScenarios:
|
||
"""§13.5 scenarios — карта {scenario: сводка} из трёх ScenarioForecast #984. PURE.
|
||
|
||
`scenarios` — список из трёх по `as_dict()` (conservative/base/aggressive). Свора-
|
||
чиваем в `by_scenario` (карта по имени сценария) — стабильно для экспортёров/чата.
|
||
None/пусто → пустая секция (graceful).
|
||
"""
|
||
if not scenarios:
|
||
return ReportScenarios()
|
||
by_scenario: dict[str, Any] = {}
|
||
for sc in scenarios:
|
||
if not isinstance(sc, dict):
|
||
continue
|
||
name = sc.get("scenario")
|
||
if isinstance(name, str) and name:
|
||
by_scenario[name] = sc
|
||
if not by_scenario:
|
||
return ReportScenarios()
|
||
summary = _scenarios_summary(by_scenario)
|
||
return ReportScenarios(by_scenario=by_scenario, summary=summary)
|
||
|
||
|
||
def _scenarios_summary(by_scenario: dict[str, Any]) -> str | None:
|
||
"""Короткий RU-текст про разброс сценариев. PURE, шаблонный."""
|
||
names = [n for n in ("conservative", "base", "aggressive") if n in by_scenario]
|
||
if not names:
|
||
return None
|
||
return f"Сценарии ({len(by_scenario)}): {', '.join(names)} — разброс по конверту ставки."
|
||
|
||
|
||
def _build_scoring(
|
||
product_scores: dict[str, Any] | None, special_indices: dict[str, Any] | None
|
||
) -> ReportScoring:
|
||
"""§13.6 scoring — 10 продуктовых скоров #985 + 6 спец-индексов #986 + overall. PURE.
|
||
|
||
`product_scores` (#985) / `special_indices` (#986) кладём как есть; `overall`
|
||
дублируем из карты #985 (удобство экспортёров). Всё None → пустая секция (graceful).
|
||
"""
|
||
return ReportScoring(
|
||
product_scores=product_scores,
|
||
special_indices=special_indices,
|
||
overall=_overall_score(product_scores),
|
||
)
|
||
|
||
|
||
def _build_confidence(
|
||
*,
|
||
analyze: dict[str, Any],
|
||
market_metrics: dict[str, Any] | None,
|
||
supply_layers: dict[str, Any] | None,
|
||
future_supply: dict[str, Any] | None,
|
||
forecasts: Sequence[dict[str, Any]],
|
||
product_scores: dict[str, Any] | None,
|
||
special_indices: dict[str, Any] | None,
|
||
advisory: bool,
|
||
) -> ReportConfidence:
|
||
"""§13.7 confidence — отчётная уверенность через #990 из извлечённых сигналов. PURE.
|
||
|
||
Кормит `compute_report_confidence` (#990) сырыми счётчиками качества данных
|
||
(deal_count/analog_count/domrf_coverage/history_months/confounded) + per-service
|
||
confidence вкладывающих под-сервисов. Результат (level/rationale/factors) ложится в
|
||
слот ReportConfidence. Все сигналы тонкие → #990 честно вернёт 'low' + «недостаточно
|
||
данных» (НИКОГДА не бросает).
|
||
"""
|
||
result = compute_report_confidence(
|
||
component_confidences=_component_confidences(
|
||
market_metrics, future_supply, forecasts, product_scores, special_indices
|
||
),
|
||
deal_count=_deal_count(analyze, market_metrics),
|
||
analog_count=_analog_count(analyze, market_metrics),
|
||
domrf_coverage=_domrf_coverage(analyze, supply_layers),
|
||
history_months=_history_months(market_metrics, forecasts),
|
||
confounded=_confounded(forecasts),
|
||
advisory=advisory,
|
||
)
|
||
payload = result.as_dict()
|
||
return ReportConfidence(
|
||
level=payload["level"],
|
||
rationale=payload["rationale"],
|
||
factors=payload["factors"],
|
||
)
|
||
|
||
|
||
def _build_exec_summary(
|
||
*,
|
||
forecasts: Sequence[dict[str, Any]],
|
||
product_tz: dict[str, Any] | None,
|
||
product_scores: dict[str, Any] | None,
|
||
confidence: ReportConfidence,
|
||
) -> ReportExecSummary:
|
||
"""§13.1 exec_summary — СИНТЕЗ заголовка/вердикта/чисел из сильнейших сигналов. PURE.
|
||
|
||
ДЕТЕРМИНИРОВАННЫЙ шаблонный синтез (НЕ LLM):
|
||
• headline — одна фраза «что строить здесь» из рекомендованного класса +
|
||
силы дефицита (или честный «данных недостаточно»).
|
||
• verdict — абзац: дефицит осн. горизонта + overall-скор + уровень уверенности.
|
||
• key_numbers — плоские опорные числа (deficit_index / overall_score / confidence).
|
||
• overall_confidence — уровень из §15-секции (#990).
|
||
|
||
Любого сигнала нет → соответствующая часть опускается; всё пусто → честный headline
|
||
«недостаточно данных» (секция всё равно валидна).
|
||
"""
|
||
di = _primary_deficit_index(forecasts)
|
||
obj_class = _recommended_class(product_tz)
|
||
overall = _overall_score(product_scores)
|
||
level: ReportConfidenceLevel | None = confidence.level
|
||
|
||
headline = _exec_headline(obj_class, di)
|
||
verdict = _exec_verdict(di, overall, level)
|
||
key_numbers = _exec_key_numbers(di, overall, level)
|
||
|
||
return ReportExecSummary(
|
||
headline=headline,
|
||
verdict=verdict,
|
||
key_numbers=key_numbers,
|
||
overall_confidence=level,
|
||
)
|
||
|
||
|
||
def _exec_headline(obj_class: str | None, deficit_index: float | None) -> str:
|
||
"""Одна фраза-вердикт «что строить здесь». PURE, шаблонный.
|
||
|
||
Класс задан + сильный дефицит → «Строить <класс>: рынок недонасыщен». Класс задан,
|
||
дефицита нет/слаб → «Рассмотреть <класс>». Нет класса, но дефицит → сигнал по
|
||
дефициту. Ничего → честный «данных недостаточно».
|
||
"""
|
||
strong = deficit_index is not None and deficit_index >= _STRONG_DEFICIT_THRESHOLD
|
||
if obj_class is not None:
|
||
if strong:
|
||
return f"Строить «{obj_class}»: рынок недонасыщен на целевом горизонте."
|
||
return f"Рассмотреть «{obj_class}» как целевой класс продукта."
|
||
if deficit_index is not None:
|
||
if strong:
|
||
return "Локация недонасыщена — есть сигнал «строить» (класс не уточнён)."
|
||
if deficit_index < 0:
|
||
return "Локация затоварена — выход рискован (класс не уточнён)."
|
||
return "Рынок локации сбалансирован (класс не уточнён)."
|
||
return "Данных недостаточно для однозначного вердикта по продукту."
|
||
|
||
|
||
def _exec_verdict(
|
||
deficit_index: float | None,
|
||
overall: float | None,
|
||
level: ReportConfidenceLevel | None,
|
||
) -> str | None:
|
||
"""Развёрнутый абзац-резюме из дефицита, overall-скора и уверенности. PURE.
|
||
|
||
Шаблонная RU-фраза; ни одного сигнала → None (graceful — вердикт опционален).
|
||
"""
|
||
parts: list[str] = []
|
||
if deficit_index is not None:
|
||
parts.append(f"индекс дефицита {_signed(deficit_index)} на целевом горизонте")
|
||
if overall is not None:
|
||
parts.append(f"итоговый продуктовый скор {round(overall, 2)}")
|
||
if level is not None:
|
||
parts.append(f"уверенность {_LEVEL_RU.get(level, level)}")
|
||
if not parts:
|
||
return None
|
||
return (
|
||
"Сводный советующий вердикт: "
|
||
+ "; ".join(parts)
|
||
+ ". Оценка advisory — не основание для инвест-решения."
|
||
)
|
||
|
||
|
||
def _exec_key_numbers(
|
||
deficit_index: float | None,
|
||
overall: float | None,
|
||
level: ReportConfidenceLevel | None,
|
||
) -> dict[str, Any]:
|
||
"""Плоские опорные числа exec_summary (JSON-safe). PURE.
|
||
|
||
Кладём только заданные сигналы (None опускаем) — частичный набор валиден.
|
||
"""
|
||
numbers: dict[str, Any] = {}
|
||
if deficit_index is not None:
|
||
numbers["deficit_index"] = round(deficit_index, 3)
|
||
if overall is not None:
|
||
numbers["overall_score"] = round(overall, 3)
|
||
if level is not None:
|
||
numbers["confidence"] = level
|
||
return numbers
|
||
|
||
|
||
def _build_meta(
|
||
*,
|
||
cad_num: str | None,
|
||
district: str | None,
|
||
segment: dict[str, Any] | None,
|
||
horizons: Sequence[int],
|
||
advisory: bool,
|
||
) -> ReportMeta:
|
||
"""§13.8 meta — контекст отчёта. PURE.
|
||
|
||
`generated_advisory`/`schema_version` берут дефолты ReportMeta (#987: ВСЕГДА True /
|
||
_SCHEMA_VERSION). `generated_at` НЕ ставим (PURE-маппер не дёргает текущее время —
|
||
детерминизм тестов; вызывающий/экспортёр при желании проставит).
|
||
"""
|
||
return ReportMeta(
|
||
cad_num=cad_num,
|
||
district=district,
|
||
segment=dict(segment) if segment else {},
|
||
horizons=list(horizons),
|
||
generated_advisory=advisory,
|
||
)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Микро-хелперы форматирования (RU-микрокопия). PURE.
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _signed(value: float) -> str:
|
||
"""'+0.34' / '−0.18' — знак для RU-микрокопии индекса (минус типографский). PURE."""
|
||
return f"+{value:.2f}" if value >= 0 else f"−{abs(value):.2f}"
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Публичный API — PURE-сборка отчёта (все входы уже посчитаны вызывающим, без БД).
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def assemble_report(
|
||
analyze: dict[str, Any],
|
||
*,
|
||
market_metrics: Any = None,
|
||
supply_layers: Any = None,
|
||
forecasts: Sequence[Any] | None = None,
|
||
future_supply: Any = None,
|
||
scenarios: Sequence[Any] | None = None,
|
||
recommendation_overlay: Any = None,
|
||
product_scores: Any = None,
|
||
special_indices: Any = None,
|
||
segment: dict[str, Any] | None = None,
|
||
cad_num: str | None = None,
|
||
district: str | None = None,
|
||
horizons: Sequence[int] = _DEFAULT_HORIZONS,
|
||
advisory: bool = _ADVISORY,
|
||
) -> SiteFinderReport:
|
||
"""§13 собрать `SiteFinderReport` (#987) из analyze + advisory-выводов форсайт-стека.
|
||
|
||
ЧИСТОЕ МАППИРОВАНИЕ (assembler #988): НЕ ходит в БД, НЕ зовёт §9.x-сервисы — берёт
|
||
уже-посчитанные входы (их считает вызывающий: #961 API / #957 чат) и РАСКЛАДЫВАЕТ
|
||
их по восьми типизированным секциям §13, попутно считая отчётную уверенность через
|
||
`compute_report_confidence` (#990). ДЕТЕРМИНИРОВАННО, БЕЗ LLM (шаблонные RU-строки).
|
||
ЮНИТ-ТЕСТИРУЕТСЯ БЕЗ БД.
|
||
|
||
Входы принимаются КАК dataclass-инстансы под-сервисов, ТАК и их `.as_dict()`-словари
|
||
(нормализуются через `_as_dict_or`) — вызывающий волен передать любое.
|
||
|
||
Маппинг секций:
|
||
• market_now ← analyze (конкуренты/цены/абсорбция) + market_metrics + supply_layers.
|
||
• future_market ← forecasts (per-горизонт) + future_supply + будущие конкуренты +
|
||
сводка scenarios.
|
||
• product_tz ← recommendation_overlay (класс/mix/commercial/USP + §16-причины).
|
||
• scenarios ← scenarios (три by_scenario).
|
||
• scoring ← product_scores (10) + special_indices (6) + overall.
|
||
• confidence ← compute_report_confidence(...) (#990) из извлечённых сигналов
|
||
качества данных + per-service confidence.
|
||
• exec_summary ← СИНТЕЗ из сильнейших сигналов (дефицит/класс/overall/уверенность).
|
||
• meta ← cad_num/district/segment/horizons/advisory/schema_version.
|
||
|
||
GRACEFUL: любой вход None → его секция частична/пуста — отчёт валиден как частичный
|
||
(#987). НИКОГДА не бросает. `advisory` прокидывается всюду (в т.ч. cap-потолок #990).
|
||
|
||
Args:
|
||
analyze: dict вывода analyze_parcel (loosely-typed; читаем defensive `.get()`).
|
||
market_metrics: §9.2 MarketMetrics (или as_dict / None).
|
||
supply_layers: §9.3 supply-слои (или as_dict / None).
|
||
forecasts: per-горизонт DemandSupplyForecast #952 (или as_dict-список / None).
|
||
future_supply: §9.3 FutureSupplyPressure (или as_dict / None).
|
||
scenarios: три ScenarioForecast #984 (или as_dict-список / None).
|
||
recommendation_overlay: §10 overlay #983 (или as_dict / None).
|
||
product_scores: §14.2 ProductScoreCard #985 (или as_dict / None).
|
||
special_indices: §25 SpecialIndices #986 (или as_dict / None).
|
||
segment: целевой сегмент (оси §9.x) для meta.
|
||
cad_num: кадастровый номер участка для meta.
|
||
district: район для meta.
|
||
horizons: горизонты прогноза (мес) для meta (по умолчанию _DEFAULT_HORIZONS).
|
||
advisory: советующий-флаг (по умолчанию True; прокидывается в #990 cap).
|
||
|
||
Returns:
|
||
SiteFinderReport — все восемь секций заполнены настолько, насколько хватило
|
||
входов (частичный отчёт валиден); `advisory` True; `as_dict()` JSON-сериализуем.
|
||
"""
|
||
# ── Нормализация входов (dataclass | as_dict | None → JSON-safe dict | None) ──
|
||
market_metrics_d = _as_dict_or(market_metrics)
|
||
supply_layers_d = _as_dict_or(supply_layers)
|
||
future_supply_d = _as_dict_or(future_supply)
|
||
overlay_d = _as_dict_or(recommendation_overlay)
|
||
product_scores_d = _as_dict_or(product_scores)
|
||
special_indices_d = _as_dict_or(special_indices)
|
||
forecasts_l = _as_dict_list(forecasts)
|
||
scenarios_l = _as_dict_list(scenarios)
|
||
|
||
# Будущие конкуренты — из forecasts (§9.7 top-N на горизонте, если они там есть).
|
||
future_competitors = _future_competitors(forecasts_l)
|
||
scenarios_summary = _scenarios_overall_summary(scenarios_l)
|
||
|
||
# ── Секции (каждая graceful: None-входы → частичная/пустая) ─────────────────
|
||
market_now = _build_market_now(analyze, market_metrics_d, supply_layers_d)
|
||
future_market = _build_future_market(
|
||
forecasts_l, future_supply_d, future_competitors, scenarios_summary
|
||
)
|
||
product_tz = _build_product_tz(overlay_d)
|
||
scenarios_section = _build_scenarios(scenarios_l)
|
||
scoring = _build_scoring(product_scores_d, special_indices_d)
|
||
confidence = _build_confidence(
|
||
analyze=analyze,
|
||
market_metrics=market_metrics_d,
|
||
supply_layers=supply_layers_d,
|
||
future_supply=future_supply_d,
|
||
forecasts=forecasts_l,
|
||
product_scores=product_scores_d,
|
||
special_indices=special_indices_d,
|
||
advisory=advisory,
|
||
)
|
||
exec_summary = _build_exec_summary(
|
||
forecasts=forecasts_l,
|
||
product_tz=overlay_d,
|
||
product_scores=product_scores_d,
|
||
confidence=confidence,
|
||
)
|
||
meta = _build_meta(
|
||
cad_num=cad_num,
|
||
district=district,
|
||
segment=segment,
|
||
horizons=horizons,
|
||
advisory=advisory,
|
||
)
|
||
|
||
logger.info(
|
||
"assemble_report: cad_num=%s district=%s horizons=%s confidence=%s "
|
||
"forecasts=%d scenarios=%d advisory=%s",
|
||
cad_num,
|
||
district,
|
||
list(horizons),
|
||
confidence.level,
|
||
len(forecasts_l),
|
||
len(scenarios_l),
|
||
advisory,
|
||
)
|
||
|
||
return SiteFinderReport(
|
||
exec_summary=exec_summary,
|
||
market_now=market_now,
|
||
future_market=future_market,
|
||
product_tz=product_tz,
|
||
scenarios=scenarios_section,
|
||
scoring=scoring,
|
||
confidence=confidence,
|
||
meta=meta,
|
||
advisory=advisory,
|
||
)
|
||
|
||
|
||
def _future_competitors(forecasts: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""Будущие конкуренты основного горизонта из forecasts (§9.7 top-N). PURE.
|
||
|
||
Берём `future_competitors` основного горизонта (_PRIMARY_HORIZON_MONTHS) или первого
|
||
forecast, где список не пуст. Нет → [] (graceful).
|
||
"""
|
||
primary = next(
|
||
(f for f in forecasts if f.get("horizon_months") == _PRIMARY_HORIZON_MONTHS),
|
||
None,
|
||
)
|
||
if primary is not None:
|
||
comps = primary.get("future_competitors")
|
||
if isinstance(comps, list) and comps:
|
||
return [c for c in comps if isinstance(c, dict)]
|
||
for f in forecasts:
|
||
comps = f.get("future_competitors")
|
||
if isinstance(comps, list) and comps:
|
||
return [c for c in comps if isinstance(c, dict)]
|
||
return []
|
||
|
||
|
||
def _scenarios_overall_summary(scenarios: Sequence[dict[str, Any]]) -> dict[str, Any] | None:
|
||
"""Компактная сводка сценариев для future_market (дефицит осн. горизонта). PURE.
|
||
|
||
Из каждого ScenarioForecast (#984) достаём deficit_index основного горизонта →
|
||
{scenario: deficit_index}. Так future_market.scenarios_summary несёт разброс
|
||
дефицита по конверту ставки, не дублируя полную секцию scenarios. Нет → None.
|
||
"""
|
||
summary: dict[str, Any] = {}
|
||
for sc in scenarios:
|
||
name = sc.get("scenario")
|
||
sc_forecasts = sc.get("forecasts")
|
||
if not isinstance(name, str) or not isinstance(sc_forecasts, list):
|
||
continue
|
||
di = _primary_deficit_index([f for f in sc_forecasts if isinstance(f, dict)])
|
||
if di is not None:
|
||
summary[name] = round(di, 3)
|
||
return summary or None
|