Light1YT
5d1260c6fb
Add confidence_engine.py — deterministic, advisory, DB-free aggregator folding
per-component confidences + raw data-quality counts (deal/analog count,
domrf↔objective coverage, history months, confounded window) into one
High/Medium/Low via weakest-link MIN, capped at medium while advisory. Builds a
structural RU rationale naming the dragging factors with real numbers ("Low
потому что 7 сделок / 1 ЖК / покрытие 2.5%"). Fills the ReportConfidence slot of
#987 SiteFinderReport. 33 unit tests.
357 lines
16 KiB
Python
357 lines
16 KiB
Python
"""Unit-тесты §15 движка отчётной уверенности (#990, 955-A4).
|
||
|
||
Чистые тесты — БЕЗ БД (движок PURE: берёт уже-посчитанные входы, ничего не считает
|
||
из §9.x и не ходит в БД):
|
||
• _level_from_value / _factor_from_count — каждый порог-бэнд (high/medium/low) +
|
||
RU-нота с РЕАЛЬНЫМ числом и оценочным словом; None → low + «нет данных».
|
||
• _coverage_factor — покрытие domrf↔objective в % в ноте (структурный §15-пример).
|
||
• _aggregate — weakest-link MIN (один low → общий low); пустой → low.
|
||
• _cap — advisory-потолок опускает high → medium.
|
||
• _build_rationale — называет тянущий-вниз фактор с его числом, «Low потому что N…»;
|
||
advisory-cap проговаривается; пусто → «недостаточно данных».
|
||
• compute_report_confidence — смешанные факторы → правильный level + структурная
|
||
причина; advisory НИКОГДА не отдаёт high; all-None → low + «недостаточно данных»;
|
||
as_dict() ложится в слот ReportConfidence (#987) и JSON-сериализуем.
|
||
|
||
Детерминированно, без LLM, без сети. DATABASE_URL до импорта app-модулей (зеркало
|
||
соседних тестов — на случай side-effect'ов импорта пакета forecasting).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
from app.services.forecasting.confidence_engine import (
|
||
_ANALOG_COUNT_HIGH,
|
||
_DEAL_COUNT_HIGH,
|
||
_DEAL_COUNT_LOW,
|
||
_HISTORY_MONTHS_HIGH,
|
||
ConfidenceFactor,
|
||
ReportConfidenceResult,
|
||
_aggregate,
|
||
_build_rationale,
|
||
_cap,
|
||
_coverage_factor,
|
||
_factor_from_count,
|
||
_level_from_value,
|
||
compute_report_confidence,
|
||
)
|
||
|
||
|
||
def _factor(name: str, level: str) -> ConfidenceFactor:
|
||
"""Лёгкий фактор фиксированного уровня (для агрегатных/причинных тестов)."""
|
||
return ConfidenceFactor(name=name, value=None, level=level, note=f"{name}:{level}") # type: ignore[arg-type]
|
||
|
||
|
||
# ── _level_from_value — пороговые бэнды ────────────────────────────────────────
|
||
|
||
|
||
class TestLevelFromValue:
|
||
def test_high_band(self) -> None:
|
||
assert _level_from_value(50, high_at=50, low_below=15) == "high"
|
||
assert _level_from_value(120, high_at=50, low_below=15) == "high"
|
||
|
||
def test_medium_band(self) -> None:
|
||
# Между low_below (вкл.) и high_at (искл.) → medium.
|
||
assert _level_from_value(15, high_at=50, low_below=15) == "medium"
|
||
assert _level_from_value(30, high_at=50, low_below=15) == "medium"
|
||
|
||
def test_low_band(self) -> None:
|
||
assert _level_from_value(14, high_at=50, low_below=15) == "low"
|
||
assert _level_from_value(0, high_at=50, low_below=15) == "low"
|
||
|
||
def test_none_is_low(self) -> None:
|
||
# Нет сигнала ≠ хороший сигнал.
|
||
assert _level_from_value(None, high_at=50, low_below=15) == "low"
|
||
|
||
|
||
# ── _factor_from_count — бэнды + RU-нота с числом ──────────────────────────────
|
||
|
||
|
||
class TestFactorFromCount:
|
||
def test_low_count_note_has_number_and_word(self) -> None:
|
||
f = _factor_from_count(
|
||
"deal_count",
|
||
7,
|
||
high_at=_DEAL_COUNT_HIGH,
|
||
low_below=_DEAL_COUNT_LOW,
|
||
unit="сделок",
|
||
suffix="за 6 мес",
|
||
)
|
||
assert f.level == "low"
|
||
assert f.value == 7
|
||
# Структурный §15-стиль: реальное число + контекст + оценка.
|
||
assert "7 сделок" in f.note
|
||
assert "за 6 мес" in f.note
|
||
assert "мало" in f.note
|
||
|
||
def test_medium_count_band(self) -> None:
|
||
f = _factor_from_count(
|
||
"deal_count", 20, high_at=_DEAL_COUNT_HIGH, low_below=_DEAL_COUNT_LOW, unit="сделок"
|
||
)
|
||
assert f.level == "medium"
|
||
assert "20 сделок" in f.note
|
||
assert "умеренно" in f.note
|
||
|
||
def test_high_count_band(self) -> None:
|
||
f = _factor_from_count(
|
||
"deal_count", 80, high_at=_DEAL_COUNT_HIGH, low_below=_DEAL_COUNT_LOW, unit="сделок"
|
||
)
|
||
assert f.level == "high"
|
||
assert "80 сделок" in f.note
|
||
assert "достаточно" in f.note
|
||
|
||
def test_one_analog_is_low_with_number(self) -> None:
|
||
# ТЗ §15-пример «только 1 ЖК» — 1 аналог обязан давать low с числом в ноте.
|
||
f = _factor_from_count(
|
||
"analog_count", 1, high_at=_ANALOG_COUNT_HIGH, low_below=2, unit="ЖК-аналога"
|
||
)
|
||
assert f.level == "low"
|
||
assert "1 ЖК-аналога" in f.note
|
||
|
||
def test_none_count_low_no_data_note(self) -> None:
|
||
f = _factor_from_count(
|
||
"deal_count", None, high_at=_DEAL_COUNT_HIGH, low_below=_DEAL_COUNT_LOW, unit="сделок"
|
||
)
|
||
assert f.level == "low"
|
||
assert f.value is None
|
||
assert "нет данных" in f.note
|
||
|
||
def test_float_count_rendered_compactly(self) -> None:
|
||
# Целое-как-float → без хвоста; дробное → 1 знак.
|
||
f_int = _factor_from_count(
|
||
"history_months", 12.0, high_at=_HISTORY_MONTHS_HIGH, low_below=12, unit="мес истории"
|
||
)
|
||
assert "12 мес истории" in f_int.note # не "12.0"
|
||
f_frac = _factor_from_count(
|
||
"history_months", 12.5, high_at=_HISTORY_MONTHS_HIGH, low_below=12, unit="мес истории"
|
||
)
|
||
assert "12.5 мес истории" in f_frac.note
|
||
|
||
|
||
# ── _coverage_factor — покрытие domrf↔objective в % ────────────────────────────
|
||
|
||
|
||
class TestCoverageFactor:
|
||
def test_low_coverage_percent_in_note(self) -> None:
|
||
# Главный sparse-риск проекта: 2.5% покрытие → low, % в ноте (структурный §15).
|
||
f = _coverage_factor(0.025)
|
||
assert f.level == "low"
|
||
assert f.value == 0.025
|
||
assert "2.5%" in f.note
|
||
assert "domrf↔objective" in f.note
|
||
|
||
def test_high_coverage(self) -> None:
|
||
f = _coverage_factor(0.75)
|
||
assert f.level == "high"
|
||
assert "75.0%" in f.note
|
||
|
||
def test_none_coverage_low(self) -> None:
|
||
f = _coverage_factor(None)
|
||
assert f.level == "low"
|
||
assert "неизвестно" in f.note
|
||
|
||
|
||
# ── _aggregate — weakest-link MIN ──────────────────────────────────────────────
|
||
|
||
|
||
class TestAggregate:
|
||
def test_one_low_drags_overall_to_low(self) -> None:
|
||
factors = [_factor("a", "high"), _factor("b", "medium"), _factor("c", "low")]
|
||
assert _aggregate(factors) == "low"
|
||
|
||
def test_all_high_stays_high(self) -> None:
|
||
assert _aggregate([_factor("a", "high"), _factor("b", "high")]) == "high"
|
||
|
||
def test_medium_floor_when_no_low(self) -> None:
|
||
assert _aggregate([_factor("a", "high"), _factor("b", "medium")]) == "medium"
|
||
|
||
def test_empty_is_low(self) -> None:
|
||
# Нет ни одного сигнала → low (graceful).
|
||
assert _aggregate([]) == "low"
|
||
|
||
|
||
# ── _cap — advisory-потолок ────────────────────────────────────────────────────
|
||
|
||
|
||
class TestCap:
|
||
def test_high_capped_to_medium(self) -> None:
|
||
assert _cap("high", "medium") == "medium"
|
||
|
||
def test_low_unchanged_below_ceiling(self) -> None:
|
||
assert _cap("low", "medium") == "low"
|
||
|
||
def test_medium_unchanged_at_ceiling(self) -> None:
|
||
assert _cap("medium", "medium") == "medium"
|
||
|
||
|
||
# ── _build_rationale — структурная §15-причина ─────────────────────────────────
|
||
|
||
|
||
class TestBuildRationale:
|
||
def test_low_names_dragging_factor_with_number(self) -> None:
|
||
# «Low потому что 7 сделок …» — называет виновника с его числом.
|
||
deal = _factor_from_count(
|
||
"deal_count",
|
||
7,
|
||
high_at=_DEAL_COUNT_HIGH,
|
||
low_below=_DEAL_COUNT_LOW,
|
||
unit="сделок",
|
||
suffix="за 6 мес",
|
||
)
|
||
good = _factor("history_months", "high")
|
||
text = _build_rationale("low", [deal, good], advisory_capped=False)
|
||
assert text.startswith("Low потому что")
|
||
assert "7 сделок" in text
|
||
# Не-тянущий (high) фактор НЕ называется.
|
||
assert "history_months" not in text
|
||
|
||
def test_low_lists_multiple_dragging_factors(self) -> None:
|
||
deal = _factor_from_count(
|
||
"deal_count", 7, high_at=_DEAL_COUNT_HIGH, low_below=_DEAL_COUNT_LOW, unit="сделок"
|
||
)
|
||
analog = _factor_from_count(
|
||
"analog_count", 1, high_at=_ANALOG_COUNT_HIGH, low_below=2, unit="ЖК-аналога"
|
||
)
|
||
cov = _coverage_factor(0.025)
|
||
text = _build_rationale("low", [deal, analog, cov], advisory_capped=False)
|
||
assert "7 сделок" in text
|
||
assert "1 ЖК-аналога" in text
|
||
assert "2.5%" in text
|
||
assert " / " in text # перечисление виновников
|
||
|
||
def test_advisory_cap_spoken_in_rationale(self) -> None:
|
||
# Уровень упёрся в advisory-потолок (не данные) → причина это проговаривает.
|
||
text = _build_rationale("medium", [_factor("x", "medium")], advisory_capped=True)
|
||
assert "советующий" in text
|
||
assert "medium" in text
|
||
|
||
def test_empty_factors_insufficient(self) -> None:
|
||
assert "недостаточно данных" in _build_rationale("low", [], advisory_capped=False)
|
||
|
||
|
||
# ── compute_report_confidence — интеграция (PURE, без БД) ──────────────────────
|
||
|
||
|
||
class TestComputeReportConfidence:
|
||
def test_mixed_factors_weakest_link_low(self) -> None:
|
||
# Сильная история, но 7 сделок и 1 ЖК → weakest-link low; причина структурная.
|
||
res = compute_report_confidence(
|
||
component_confidences=["medium", "high"],
|
||
deal_count=7,
|
||
analog_count=1,
|
||
domrf_coverage=0.4,
|
||
history_months=36,
|
||
confounded=False,
|
||
advisory=True,
|
||
)
|
||
assert isinstance(res, ReportConfidenceResult)
|
||
assert res.level == "low"
|
||
assert res.rationale.startswith("Low потому что")
|
||
assert "7 сделок" in res.rationale
|
||
assert "1 ЖК-аналога" in res.rationale
|
||
|
||
def test_advisory_never_high(self) -> None:
|
||
# Все сигналы отличные, но advisory=True → НИКОГДА не high (потолок medium).
|
||
res = compute_report_confidence(
|
||
component_confidences=["high", "high"],
|
||
deal_count=200,
|
||
analog_count=10,
|
||
domrf_coverage=0.9,
|
||
history_months=48,
|
||
confounded=False,
|
||
advisory=True,
|
||
)
|
||
assert res.level == "medium"
|
||
assert res.advisory_capped is True
|
||
assert "советующий" in res.rationale
|
||
|
||
def test_non_advisory_allows_high(self) -> None:
|
||
# advisory=False (гипотетически провалидированный стек) → high достижим.
|
||
res = compute_report_confidence(
|
||
component_confidences=["high"],
|
||
deal_count=200,
|
||
analog_count=10,
|
||
domrf_coverage=0.9,
|
||
history_months=48,
|
||
confounded=False,
|
||
advisory=False,
|
||
)
|
||
assert res.level == "high"
|
||
assert res.advisory_capped is False
|
||
|
||
def test_confounded_window_drags_down(self) -> None:
|
||
# Шок-окно → low даже при прочих хороших сигналах; названо в причине.
|
||
res = compute_report_confidence(
|
||
component_confidences=["high"],
|
||
deal_count=200,
|
||
analog_count=10,
|
||
domrf_coverage=0.9,
|
||
history_months=48,
|
||
confounded=True,
|
||
advisory=True,
|
||
)
|
||
assert res.level == "low"
|
||
assert "шок" in res.rationale
|
||
|
||
def test_all_none_is_low_insufficient(self) -> None:
|
||
# Граничный graceful: ни одного входа → low + «недостаточно данных».
|
||
res = compute_report_confidence()
|
||
assert res.level == "low"
|
||
assert "недостаточно данных" in res.rationale
|
||
assert res.factors == []
|
||
assert res.advisory_capped is False
|
||
|
||
def test_clean_inputs_medium_under_advisory(self) -> None:
|
||
# Рабочие (medium-бэнд) счётчики без low-факторов → medium под advisory-cap.
|
||
res = compute_report_confidence(
|
||
component_confidences=["medium"],
|
||
deal_count=20,
|
||
analog_count=2,
|
||
domrf_coverage=0.4,
|
||
history_months=18,
|
||
advisory=True,
|
||
)
|
||
assert res.level == "medium"
|
||
# Здесь потолок не понижал (raw уже medium) → не помечаем capped.
|
||
assert res.advisory_capped is False
|
||
|
||
def test_as_dict_fits_report_confidence_slot(self) -> None:
|
||
# as_dict() ложится в слот ReportConfidence (#987): {level, rationale, factors}.
|
||
res = compute_report_confidence(deal_count=7, analog_count=1, advisory=True)
|
||
d = res.as_dict()
|
||
assert set(d.keys()) == {"level", "rationale", "factors"}
|
||
assert d["level"] == "low"
|
||
assert isinstance(d["rationale"], str)
|
||
assert isinstance(d["factors"], dict)
|
||
# Факторы по именам + флаг advisory_capped внутри.
|
||
assert "deal_count" in d["factors"]
|
||
assert d["factors"]["deal_count"]["value"] == 7
|
||
assert "advisory_capped" in d["factors"]
|
||
# Полностью JSON-сериализуем (контракт для экспортёров/чата).
|
||
assert json.loads(json.dumps(d, ensure_ascii=False)) == d
|
||
|
||
def test_ignores_garbage_component_confidence(self) -> None:
|
||
# Мусорный component-уровень не учитывается (whitelist), не роняет искусственно.
|
||
res = compute_report_confidence(
|
||
component_confidences=["high", "bogus"], # type: ignore[list-item]
|
||
deal_count=200,
|
||
analog_count=10,
|
||
domrf_coverage=0.9,
|
||
history_months=48,
|
||
advisory=False,
|
||
)
|
||
assert res.level == "high"
|
||
|
||
def test_as_dict_keeps_all_component_factors(self) -> None:
|
||
# Несколько component-факторов не теряются в плоском dict (суффиксуются).
|
||
res = compute_report_confidence(
|
||
component_confidences=["high", "medium", "low"], advisory=False
|
||
)
|
||
d = res.as_dict()
|
||
component_keys = [k for k in d["factors"] if k.startswith("component")]
|
||
assert len(component_keys) == 3 # ни один не перезаписан
|
||
assert json.loads(json.dumps(d, ensure_ascii=False)) == d
|