gendesign/backend/app/services/forecasting/confidence_engine.py
bot-backend 14f3ef2019
All checks were successful
Deploy / build-worker (push) Successful in 2m47s
Deploy / deploy (push) Successful in 1m20s
Deploy / changes (push) Successful in 9s
Deploy / build-frontend (push) Has been skipped
Deploy / build-backend (push) Successful in 1m52s
fix(week-review): backend-аудит v2 — 82 фиксов (#1660)
Co-authored-by: bot-backend <bot-backend@gendsgn.local>
Co-committed-by: bot-backend <bot-backend@gendsgn.local>
2026-06-17 17:13:38 +00:00

491 lines
27 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.

"""§15 движок ОТЧЁТНОЙ уверенности v2 — weakest-link + СТРУКТУРНАЯ причина.
#990 (955-A4, Site Finder v2 / «GG-форсайт» ТЗ §15), EPIC 11 «Отчёт». Это ЧИСТЫЙ
агрегатор уверенности: он сводит per-component confidence под-сервисов (#950/#952/
#985/#986…) + СЫРЫЕ счётчики качества данных (число сделок, число ЖК-аналогов,
покрытие domrf↔objective, глубина истории, шок-окно) в ОДИН отчётный уровень
High/Medium/Low + RU-причину, которая ЯВНО НАЗЫВАЕТ, ЧТО утянуло уровень вниз с
РЕАЛЬНЫМИ числами («Low потому что 7 сделок за 6 мес / только 1 ЖК-аналог /
покрытие domrf↔objective 2.5%»). Наполняет слот `ReportConfidence` отчёта #987.
ДЕТЕРМИНИРОВАННЫЙ, БЕЗ LLM, СОВЕТУЮЩИЙ. Никакого SQL/сети/print/вычислений §9.x —
движок ЧИСТЫЙ: берёт уже-посчитанные входы (их кормит сборщик #988) и только
агрегирует + формулирует. Поэтому он ЮНИТ-ТЕСТИРУЕТСЯ БЕЗ БД.
ЛОГИКА (ТЗ §15):
• Отчётная уверенность = WEAKEST-LINK (MIN ранга) вкладывающих факторов — самый
тонкий сигнал честно роняет общий уровень (зеркало future_supply._min_confidence
и product_scoring._min_confidence). Один 'low'-фактор → отчёт 'low'.
• Причина — СТРУКТУРНАЯ: называет КОНКРЕТНЫЙ фактор(ы)-виновник с его ЧИСЛОМ
(сделки / ЖК-аналоги / покрытие % / месяцы истории / шок-окно / advisory-cap),
формат «Low потому что N сделок … / только M ЖК … / покрытие X%».
• Весь форсайт-стек СОВЕТУЮЩИЙ (движок не провалидирован до бэктеста #951) →
advisory-cap: при advisory=True уровень НИКОГДА не 'high' (потолок 'medium'),
и причина это ПРОГОВАРИВАЕТ. Честность важнее оптимистичной метки.
ПОРОГИ (align с per-service gate'ами, которые читает движок):
• deal_count — зеркало market_metrics._confidence (n_lots/n_sold) + §9.6 _MIN_OBS:
мало сделок → скоростные метрики статистически ненадёжны.
• analog_count (ЖК-аналоги, = market_metrics.obj_count) — high≥3 / medium≥2 / 1 → low
(точная копия _CONF_HIGH_MIN_OBJ=3 / _CONF_MEDIUM_MIN_OBJ=2; «1 ЖК» — ТЗ §15-пример).
• domrf_coverage — главный риск проекта (domrf↔objective ~2.5%, см. market_metrics
docstring): низкое покрытие → скрытый/будущий слой §9.3 недооценён.
• history_months — зеркало §9.6 _CONF_HIGH_MIN_OBS=24 (≥2 года) / _MIN_OBS=8: короткий
ряд → связь rate↔sales / тренды не установлены.
• confounded — шок-окно (is_confounded_window, PR2): ряд пересекает структурный
разрыв → оценки смещены (НИКОГДА не 'high').
GRACEFUL (дух всего форсайт-стека): все входы None/пусто → 'low' + «недостаточно
данных» (тонкий анализ всё равно даёт честный вердикт, не crash). `as_dict()`
JSON-safe и ложится в слот ReportConfidence (#987): {level, rationale, factors}.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Literal
Confidence = Literal["high", "medium", "low"]
# ── Порядок уверенности для weakest-link MIN (хуже = ниже). Зеркало vocab
# future_supply._CONFIDENCE_RANK / product_scoring._CONFIDENCE_RANK. ───────────────
_CONFIDENCE_RANK: dict[Confidence, int] = {"low": 0, "medium": 1, "high": 2}
_RANK_TO_CONFIDENCE: dict[int, Confidence] = {0: "low", 1: "medium", 2: "high"}
# ── Имена факторов (named — переименование не рассинхронит причину/тесты молча) ──
_F_DEAL_COUNT: str = "deal_count"
_F_ANALOG_COUNT: str = "analog_count"
_F_DOMRF_COVERAGE: str = "domrf_coverage"
_F_HISTORY_MONTHS: str = "history_months"
_F_CONFOUNDED_WINDOW: str = "confounded_window"
_F_ADVISORY_CAP: str = "advisory_cap"
_F_COMPONENT: str = "component" # один вкладывающий per-service confidence
# ── Пороги счётчиков (align с per-service gate'ами, прочитанными в #949/#951) ──────
# deal_count: число сделок (продаж) за окно. high — длинная плотная выборка,
# medium — рабочий минимум, low — статистически ненадёжно (зеркало духа
# market_metrics: n_sold>0 обязателен; §9.6 _MIN_OBS=8 — пол для регрессии).
_DEAL_COUNT_HIGH: int = 50
_DEAL_COUNT_LOW: int = 15
# analog_count: число ЖК-аналогов (= market_metrics.obj_count). ТОЧНАЯ копия
# _CONF_HIGH_MIN_OBJ=3 / _CONF_MEDIUM_MIN_OBJ=2: 1 ЖК → low (ТЗ §15-пример «только 1 ЖК»).
_ANALOG_COUNT_HIGH: int = 3
_ANALOG_COUNT_LOW: int = 2 # < этого (т.е. ≤1 ЖК) → low
# domrf_coverage: доля domrf↔objective ∈ [0,1] (главный sparse-риск проекта ~2.5%).
# high — покрытие плотное; low — слой §9.3 (скрытое/будущее) недооценён. medium-порог
# созвучен supply_layers._L2_MEDIUM_MIN_COVERAGE=0.6 (доверяем при покрытии большинства).
_DOMRF_COVERAGE_HIGH: float = 0.6
_DOMRF_COVERAGE_LOW: float = 0.2
# history_months: глубина ряда (мес). Зеркало §9.6 _CONF_HIGH_MIN_OBS=24 (≥2 года) /
# _MIN_OBS=8 (пол): короткий ряд → тренды/чувствительность не установлены.
_HISTORY_MONTHS_HIGH: int = 24
_HISTORY_MONTHS_LOW: int = 12
# Потолок при advisory=True: весь стек советующий (не провалидирован) → НИКОГДА 'high'.
_ADVISORY_CEILING: Confidence = "medium"
@dataclass(frozen=True)
class ConfidenceFactor:
"""Один вклад в отчётную уверенность — фактор-сигнал с числом, уровнем и RU-нотой.
`name` — машинное имя (_F_*); `value` — сырое число/флаг, которым фактор оценён
(число сделок / ЖК / покрытие / месяцы / bool шок-окна — или None); `level` —
его собственный вклад high/medium/low; `note` — короткая RU-фраза с числом
(«7 сделок за 6 мес — мало»), из которой собирается структурная причина §15.
"""
name: str
value: Any # число/доля/флаг, обосновавшие level (или None)
level: Confidence
note: str # RU, с реальным числом
def as_dict(self) -> dict[str, Any]:
return {
"name": self.name,
"value": self.value,
"level": self.level,
"note": self.note,
}
@dataclass(frozen=True)
class ReportConfidenceResult:
"""§15 результат движка отчётной уверенности — уровень + СТРУКТУРНАЯ причина.
`level` — weakest-link MIN факторов, capped в 'medium' при advisory. `rationale` —
RU-предложение «Low потому что …», НАЗЫВАЮЩЕЕ тянущие вниз факторы с их числами.
`factors` — все учтённые ConfidenceFactor (explainability §16). `advisory_capped` —
True, если потолок advisory реально понизил уровень (был бы 'high'). `as_dict()`
ложится в слот ReportConfidence #987: {level, rationale, factors}.
"""
level: Confidence
rationale: str
factors: list[ConfidenceFactor] = field(default_factory=list)
advisory_capped: bool = False
def as_dict(self) -> dict[str, Any]:
"""JSON-safe вид под слот ReportConfidence (#987): {level, rationale, factors}.
`factors` — плоский dict {name: {value, level, note}} (стабильно, удобно
экспортёрам/чату; advisory_capped дублируем внутрь для полноты explainability).
Повторяющиеся имена (несколько `component`) суффиксуем `_2/_3…`, чтобы ни один
фактор не потерялся в плоском dict (ключи иначе бы перезаписали друг друга).
"""
factors: dict[str, Any] = {}
for f in self.factors:
key = f.name
if key in factors:
i = 2
while f"{key}_{i}" in factors:
i += 1
key = f"{key}_{i}"
factors[key] = {"value": f.value, "level": f.level, "note": f.note}
factors["advisory_capped"] = self.advisory_capped
return {
"level": self.level,
"rationale": self.rationale,
"factors": factors,
}
# ──────────────────────────────────────────────────────────────────────────────
# Pure-хелперы — без БД, полностью юнит-тестируемо. Каждый graceful: None/пусто →
# безопасная деградация ('low' / нейтраль), НИКОГДА не бросает.
# ──────────────────────────────────────────────────────────────────────────────
def _level_from_value(value: float | None, *, high_at: float, low_below: float) -> Confidence:
"""Сырое число → high/medium/low по двум именованным порогам. PURE.
value ≥ high_at → 'high'; value < low_below → 'low'; между → 'medium'. None →
'low' (нет сигнала ≠ хороший сигнал). Монотонно (выше число → не хуже уровень).
"""
if value is None:
return "low"
if value >= high_at:
return "high"
if value < low_below:
return "low"
return "medium"
def _factor_from_count(
name: str,
count: float | None,
*,
high_at: float,
low_below: float,
unit: str,
suffix: str = "",
) -> ConfidenceFactor:
"""Сырой счётчик → ConfidenceFactor (уровень по порогам + RU-нота с числом). PURE.
Маппит `count` в high/medium/low через _level_from_value(high_at/low_below) и
собирает RU-ноту с РЕАЛЬНЫМ числом и оценкой («7 сделок за 6 мес — мало» /
«3 ЖК-аналога — достаточно»). `unit` — что считаем («сделок» / «ЖК-аналога»);
`suffix` — хвост контекста («за 6 мес»). count None → low + нота «нет данных».
Args:
name: машинное имя фактора (_F_*).
count: сырое число (или None).
high_at: порог 'high' (count ≥ → high).
low_below: порог 'low' (count < → low).
unit: единица в RU-ноте (родительный падеж: «сделок», «ЖК-аналога»).
suffix: необязательный хвост ноты (например «за 6 мес»).
Returns:
ConfidenceFactor с level/value/note.
"""
level = _level_from_value(count, high_at=high_at, low_below=low_below)
tail = f" {suffix}" if suffix else ""
if count is None:
note = f"нет данных о количестве ({unit})"
else:
shown = int(count) if float(count).is_integer() else round(float(count), 1)
note = f"{shown} {unit}{tail}{_QUALITY_WORD[level]}"
return ConfidenceFactor(name=name, value=count, level=level, note=note)
# Оценочное слово на уровень (для RU-ноты фактора). Зеркало духа per-service фраз.
_QUALITY_WORD: dict[Confidence, str] = {
"high": "достаточно",
"medium": "умеренно",
"low": "мало",
}
def _coverage_factor(coverage: float | None) -> ConfidenceFactor:
"""domrf↔objective покрытие ∈ [0,1] → ConfidenceFactor с % в ноте. PURE.
Главный sparse-риск проекта (~2.5%). Нота показывает покрытие В ПРОЦЕНТАХ
(структурный §15-пример «покрытие domrf↔objective 2.5%»). None → low.
"""
level = _level_from_value(coverage, high_at=_DOMRF_COVERAGE_HIGH, low_below=_DOMRF_COVERAGE_LOW)
if coverage is None:
note = "покрытие domrf↔objective неизвестно"
else:
pct = round(float(coverage) * 100.0, 1)
note = f"покрытие domrf↔objective {pct}% — {_QUALITY_WORD[level]}"
return ConfidenceFactor(name=_F_DOMRF_COVERAGE, value=coverage, level=level, note=note)
def _confounded_factor(confounded: bool) -> ConfidenceFactor:
"""Шок-окно → ConfidenceFactor. PURE.
confounded=True (ряд пересекает структурный разрыв, PR2) → 'low' (оценки смещены,
НИКОГДА не 'high'). False → 'high' (окно чистое — фактор не тянет вниз).
"""
if confounded:
return ConfidenceFactor(
name=_F_CONFOUNDED_WINDOW,
value=True,
level="low",
note="окно ряда пересекает шок-период (оценки смещены)",
)
return ConfidenceFactor(
name=_F_CONFOUNDED_WINDOW,
value=False,
level="high",
note="окно ряда вне шок-периодов",
)
def _component_factor(level: Confidence) -> ConfidenceFactor:
"""Один вкладывающий per-service confidence → ConfidenceFactor. PURE.
Per-service уже свернул свои тонкие сигналы в high/medium/low; берём как факт,
нота поясняет вклад. value=None (число — внутри самого под-сервиса).
"""
return ConfidenceFactor(
name=_F_COMPONENT,
value=None,
level=level,
note=f"вкладывающий сервис: {level}",
)
def _aggregate(factors: list[ConfidenceFactor]) -> Confidence:
"""Weakest-link агрегат: MIN ранга факторов (худший тянет вниз). PURE.
Зеркало future_supply._min_confidence / product_scoring._min_confidence: один
'low'-фактор → общий 'low'. Пустой список → 'low' (нет ни одного сигнала).
"""
ranks = [_CONFIDENCE_RANK[f.level] for f in factors]
if not ranks:
return "low"
return _RANK_TO_CONFIDENCE[min(ranks)]
def _cap(level: Confidence, ceiling: Confidence) -> Confidence:
"""Опустить level до потолка ceiling (MIN ранга). PURE.
advisory-cap: при ceiling='medium' любое 'high' становится 'medium'. Зеркало
advisory-духа recommendation/product_scoring (confidence ≤ medium в advisory).
"""
return _RANK_TO_CONFIDENCE[min(_CONFIDENCE_RANK[level], _CONFIDENCE_RANK[ceiling])]
def _dragging_factors(factors: list[ConfidenceFactor], level: Confidence) -> list[ConfidenceFactor]:
"""Факторы, чей уровень == итоговому (т.е. ИМЕННО они тянут вниз). PURE.
Для 'low'-итога — все low-факторы; для 'medium' — все medium-факторы. Это
«виновники» структурной причины §15. Component-факторы (без своего числа)
отодвигаем в конец — приоритет адресным счётчикам с числами в ноте.
"""
same = [f for f in factors if f.level == level]
same.sort(key=lambda f: f.name == _F_COMPONENT) # named-счётчики раньше component
return same
def _build_rationale(
level: Confidence,
factors: list[ConfidenceFactor],
*,
advisory_capped: bool,
) -> str:
"""Структурная §15-причина: «Low потому что <факторы с числами>». PURE.
Берёт тянущие-вниз факторы (_dragging_factors) и НАЗЫВАЕТ их ноты (с реальными
числами) после «<Level> потому что …». Если уровень понижен именно advisory-cap'ом
(был бы 'high') — проговаривает, что движок советующий. Пустые/нейтральные факторы
→ «недостаточно данных» (graceful).
Args:
level: итоговый (уже capped) уровень.
factors: все учтённые факторы.
advisory_capped: True, если потолок advisory реально понизил уровень.
Returns:
RU-предложение-причина.
"""
label = _LEVEL_LABEL[level]
drag = _dragging_factors(factors, level)
notes = [f.note for f in drag]
if advisory_capped and level == _ADVISORY_CEILING:
# Уровень упёрся в advisory-потолок (не данные) — это и есть главная причина.
# _F_ADVISORY_CAP-фактор уже проговорён в base — исключаем его ноту из «также»,
# иначе advisory-cap-сообщение дублируется (частый all-high случай).
other = [f.note for f in drag if f.name != _F_ADVISORY_CAP]
base = f"{label}: прогноз советующий (не провалидирован) — уровень ограничен «medium»"
if other:
base += "; также " + _join_notes(other)
return base + "."
if not notes:
return f"{label}: недостаточно данных для оценки уверенности."
return f"{label} потому что {_join_notes(notes)}."
# RU-метка уровня для причины (заглавная — начало предложения).
_LEVEL_LABEL: dict[Confidence, str] = {"high": "High", "medium": "Medium", "low": "Low"}
def _join_notes(notes: list[str], *, limit: int = 3) -> str:
"""Склеить ноты факторов в RU-перечисление через « / » (не более limit). PURE.
Ограничиваем limit (по умолчанию 3), чтобы причина оставалась фразой, а не
простынёй; хвост сворачиваем в «… и др.». Дубли убираем, порядок сохраняем.
"""
seen: list[str] = []
for n in notes:
if n not in seen:
seen.append(n)
head = seen[:limit]
joined = " / ".join(head)
if len(seen) > limit:
# Без точки в маркере — её добавит вызывающий концом предложения (не «др..»).
joined += " и др"
return joined
# ──────────────────────────────────────────────────────────────────────────────
# Публичный API — PURE-сборка (входы уже посчитаны сборщиком #988, без БД).
# ──────────────────────────────────────────────────────────────────────────────
def compute_report_confidence(
*,
component_confidences: list[Confidence] | None = None,
deal_count: int | None = None,
analog_count: int | None = None,
domrf_coverage: float | None = None,
history_months: int | None = None,
confounded: bool = False,
advisory: bool = True,
) -> ReportConfidenceResult:
"""§15 свести входы в ОДНУ отчётную уверенность + структурную причину. PURE.
СОВЕТУЮЩИЙ агрегатор: НЕ ходит в БД и НЕ зовёт сервисы — берёт уже-посчитанные
счётчики качества данных + per-service confidence (их кормит сборщик #988),
строит факторы, агрегирует weakest-link (MIN), при advisory опускает потолок до
'medium' и собирает причину «Low потому что …» с реальными числами виновников.
Шаги:
1. Из сырых счётчиков (deal_count / analog_count / domrf_coverage /
history_months / confounded) → ConfidenceFactor'ы с RU-нотами (с числами).
2. Свернуть component_confidences под-сервисов в факторы (как есть).
3. level = weakest-link MIN всех факторов (худший тянет вниз).
4. advisory=True → cap level в 'medium' (НИКОГДА не 'high'); пометить
advisory_capped, если потолок реально понизил.
5. Структурная §15-причина, называющая тянущие-вниз факторы с числами.
Graceful: все входы None/пусто → 'low' + «недостаточно данных». Никогда не бросает.
Args:
component_confidences: per-service confidence (#950/#952/#985/#986…), None/[]→
нет вкладывающих компонентов.
deal_count: число сделок за окно (None → нет данных, тянет в low).
analog_count: число ЖК-аналогов в выборке (= market_metrics.obj_count).
domrf_coverage: доля domrf↔objective ∈ [0,1] (главный sparse-риск проекта).
history_months: глубина ряда (мес).
confounded: True, если окно ряда пересекает шок-период (PR2).
advisory: весь стек советующий → cap 'medium' (по умолчанию True; почти всегда).
Returns:
ReportConfidenceResult (level / rationale / factors / advisory_capped);
`as_dict()` ложится в слот ReportConfidence отчёта #987.
"""
factors: list[ConfidenceFactor] = []
# ── 1. Сырые счётчики качества данных → факторы (только заданные) ──────────
if deal_count is not None:
factors.append(
_factor_from_count(
_F_DEAL_COUNT,
deal_count,
high_at=_DEAL_COUNT_HIGH,
low_below=_DEAL_COUNT_LOW,
unit="сделок",
)
)
if analog_count is not None:
factors.append(
_factor_from_count(
_F_ANALOG_COUNT,
analog_count,
high_at=_ANALOG_COUNT_HIGH,
low_below=_ANALOG_COUNT_LOW,
unit="ЖК-аналога",
)
)
if domrf_coverage is not None:
factors.append(_coverage_factor(domrf_coverage))
if history_months is not None:
factors.append(
_factor_from_count(
_F_HISTORY_MONTHS,
history_months,
high_at=_HISTORY_MONTHS_HIGH,
low_below=_HISTORY_MONTHS_LOW,
unit="мес истории",
)
)
# Шок-окно учитываем ТОЛЬКО когда оно есть (True): чистое окно не должно
# искусственно тянуть к 'high', если других сигналов нет (см. graceful ниже).
if confounded:
factors.append(_confounded_factor(True))
# ── 2. Вкладывающие per-service confidence → факторы ──────────────────────
for c in component_confidences or []:
if c in _CONFIDENCE_RANK:
factors.append(_component_factor(c))
# ── 3. weakest-link агрегат (худший тянет вниз) ───────────────────────────
raw_level = _aggregate(factors)
# ── 4. advisory-cap (НИКОГДА не 'high', пока стек советующий) ──────────────
if advisory:
capped_level = _cap(raw_level, _ADVISORY_CEILING)
advisory_capped = capped_level != raw_level
if advisory_capped:
# Зафиксировать причину-cap как явный фактор (explainability §16).
factors.append(
ConfidenceFactor(
name=_F_ADVISORY_CAP,
value=True,
level=_ADVISORY_CEILING,
note="прогноз советующий — уровень ограничен «medium»",
)
)
level = capped_level
else:
level = raw_level
advisory_capped = False
# ── 5. структурная §15-причина ────────────────────────────────────────────
rationale = _build_rationale(level, factors, advisory_capped=advisory_capped)
return ReportConfidenceResult(
level=level,
rationale=rationale,
factors=factors,
advisory_capped=advisory_capped,
)