api/src/domain/quality_metrics.py

@@ -0,0 +1,139 @@
+"""Матрица причин и ряд по месяцам (§14.3 `05-domain-rules.md`; design `quality-metrics` §6, §7).
+
+Чистая функция от фактов исхода: сессию не читает и в БД не пишет. Причина документа (`primary_cause`) вычислена один
+раз при записи факта (приоритет §14.3 п.1-6: сервис > наш разбор > документ клиента > без участия > как есть), здесь
+она только группируется — атрибуция не пересчитывается.
+
+Метрики берут по документу последний факт исхода (`outcome_at`): повторное подтверждение ОСВ и
+`*_AFTER_CONFIRM` пишут новый факт поверх прежнего. Исходы `DELETED` / `SUPERSEDED` («до подтверждения») и
+`DELETED_AFTER_CONFIRM` / `SUPERSEDED_AFTER_CONFIRM` («после подтверждения») — отдельные столбцы, в доли не входят.
+Знаменатель долей — документы с исходом `AUTO_CONFIRMED` / `CONFIRMED`. «В работе» (документ без исхода) и технические
+отказы (считаются по файлам) сюда не передаются — их считает вызывающий сервис отдельными запросами.
+"""
+
+import datetime
+import uuid
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+from src.models.enums import QualityCause, QualityOutcome
+
+__all__ = [
+    "MatrixResult",
+    "MonthPoint",
+    "QualityOutcomeFactRow",
+    "compute_matrix",
+    "compute_monthly_series",
+]
+
+_IN_DENOMINATOR = frozenset({QualityOutcome.AUTO_CONFIRMED, QualityOutcome.CONFIRMED})
+_BEFORE_CONFIRM = frozenset({QualityOutcome.DELETED, QualityOutcome.SUPERSEDED})
+_AFTER_CONFIRM = frozenset({QualityOutcome.DELETED_AFTER_CONFIRM, QualityOutcome.SUPERSEDED_AFTER_CONFIRM})
+
+
+@dataclass(frozen=True, slots=True)
+class QualityOutcomeFactRow:
+    """Минимум полей `quality_outcome_fact`, нужных агрегации (вызывающий сервис строит его из SELECT)."""
+
+    document_id: uuid.UUID
+    outcome: QualityOutcome
+    outcome_at: datetime.datetime
+    primary_cause: QualityCause | None
+
+
+@dataclass(frozen=True, slots=True)
+class MatrixResult:
+    """Столбцы матрицы причин (§14.3).
+
+    `counts` — документы по причинам, всегда со всеми ключами `QualityCause` (нулевые тоже); `denominator` — сумма
+    `counts` (документы с исходом `AUTO_CONFIRMED` / `CONFIRMED`); `shares` — `counts / denominator`, при нулевом
+    знаменателе все доли 0.0. `removed_before_confirm` / `removed_after_confirm` — отдельные столбцы, в доли не входят.
+    """
+
+    counts: dict[QualityCause, int]
+    shares: dict[QualityCause, float]
+    denominator: int
+    removed_before_confirm: int
+    removed_after_confirm: int
+
+
+@dataclass(frozen=True, slots=True)
+class MonthPoint:
+    """Точка ряда: `month` — первое число месяца `outcome_at`; `share` — доля причины от знаменателя месяца."""
+
+    month: datetime.date
+    count: int
+    denominator: int
+    share: float
+
+
+def _latest_per_document(outcomes: Sequence[QualityOutcomeFactRow]) -> list[QualityOutcomeFactRow]:
+    """Последний по `outcome_at` факт каждого документа (design §4.2); при равенстве времени — поздний во входе."""
+    latest: dict[uuid.UUID, QualityOutcomeFactRow] = {}
+    for row in outcomes:
+        current = latest.get(row.document_id)
+        if current is None or row.outcome_at >= current.outcome_at:
+            latest[row.document_id] = row
+    return list(latest.values())
+
+
+def _cause(row: QualityOutcomeFactRow) -> QualityCause:
+    if row.primary_cause is None:
+        raise ValueError(f"у исхода {row.outcome} документа {row.document_id} нет primary_cause (инвариант факта)")
+    return row.primary_cause
+
+
+def _share(count: int, denominator: int) -> float:
+    return count / denominator if denominator else 0.0
+
+
+def compute_matrix(outcomes: Sequence[QualityOutcomeFactRow]) -> MatrixResult:
+    """§14.3: матрица причин по фактам исхода (группировка по сохранённой `primary_cause`, без пересчёта)."""
+    counts = dict.fromkeys(QualityCause, 0)
+    before = after = 0
+    for row in _latest_per_document(outcomes):
+        if row.outcome in _IN_DENOMINATOR:
+            counts[_cause(row)] += 1
+        elif row.outcome in _BEFORE_CONFIRM:
+            before += 1
+        elif row.outcome in _AFTER_CONFIRM:
+            after += 1
+    denominator = sum(counts.values())
+    return MatrixResult(
+        counts=counts,
+        shares={cause: _share(count, denominator) for cause, count in counts.items()},
+        denominator=denominator,
+        removed_before_confirm=before,
+        removed_after_confirm=after,
+    )
+
+
+def _next_month(month: datetime.date) -> datetime.date:
+    return datetime.date(month.year + month.month // 12, month.month % 12 + 1, 1)
+
+
+def compute_monthly_series(outcomes: Sequence[QualityOutcomeFactRow], cause: QualityCause) -> list[MonthPoint]:
+    """§14.3 / design §7: ряд по месяцам `outcome_at` для выбранной причины.
+
+    В ряд входят только документы со знаменателем (`AUTO_CONFIRMED` / `CONFIRMED`, последний факт документа).
+    Месяцы между первым и последним месяцем данных без документов заполняются нулями (непрерывная ось).
+    """
+    count_by_month: dict[datetime.date, int] = {}
+    total_by_month: dict[datetime.date, int] = {}
+    for row in _latest_per_document(outcomes):
+        if row.outcome not in _IN_DENOMINATOR:
+            continue
+        month = datetime.date(row.outcome_at.year, row.outcome_at.month, 1)
+        total_by_month[month] = total_by_month.get(month, 0) + 1
+        if _cause(row) is cause:
+            count_by_month[month] = count_by_month.get(month, 0) + 1
+    if not total_by_month:
+        return []
+    points: list[MonthPoint] = []
+    month, last = min(total_by_month), max(total_by_month)
+    while month <= last:
+        total = total_by_month.get(month, 0)
+        count = count_by_month.get(month, 0)
+        points.append(MonthPoint(month=month, count=count, denominator=total, share=_share(count, total)))
+        month = _next_month(month)
+    return points

api/tests/test_domain_quality_metrics.py

@@ -0,0 +1,111 @@
+"""Матрица причин и ряд по месяцам (§14.3; design quality-metrics §6, §7) — чистая функция, без БД."""
+
+import datetime
+import uuid
+
+import pytest
+
+from src.domain.quality_metrics import QualityOutcomeFactRow, compute_matrix, compute_monthly_series
+from src.models.enums import QualityCause, QualityOutcome
+
+TWO = 2
+HALF = 0.5
+
+
+def make_outcome_row(
+    *,
+    primary_cause: QualityCause | None = QualityCause.AS_IS,
+    outcome: QualityOutcome = QualityOutcome.CONFIRMED,
+    outcome_at: datetime.datetime | None = None,
+    document_id: uuid.UUID | None = None,
+) -> QualityOutcomeFactRow:
+    return QualityOutcomeFactRow(
+        document_id=document_id or uuid.uuid4(),
+        outcome=outcome,
+        outcome_at=outcome_at or datetime.datetime(2026, 10, 15, tzinfo=datetime.UTC),
+        primary_cause=primary_cause,
+    )
+
+
+def test_compute_matrix_priority_service_over_our_error():
+    # атрибуция уже в факте — матрица только группирует
+    matrix = compute_matrix(
+        [
+            make_outcome_row(primary_cause=QualityCause.SERVICE_ERROR),
+            make_outcome_row(primary_cause=QualityCause.OUR_ERROR),
+        ]
+    )
+    assert matrix.counts[QualityCause.SERVICE_ERROR] == 1
+    assert matrix.counts[QualityCause.OUR_ERROR] == 1
+    assert matrix.counts[QualityCause.AS_IS] == 0
+
+
+def test_compute_matrix_denominator_excludes_removed_documents():
+    outcomes = [
+        make_outcome_row(primary_cause=QualityCause.WITHOUT_PARTICIPATION, outcome=QualityOutcome.AUTO_CONFIRMED),
+        make_outcome_row(primary_cause=QualityCause.AS_IS),
+        make_outcome_row(primary_cause=None, outcome=QualityOutcome.DELETED),
+        make_outcome_row(primary_cause=None, outcome=QualityOutcome.SUPERSEDED),
+        make_outcome_row(primary_cause=None, outcome=QualityOutcome.SUPERSEDED_AFTER_CONFIRM),
+    ]
+    matrix = compute_matrix(outcomes)
+    assert matrix.denominator == TWO
+    assert matrix.removed_before_confirm == TWO
+    assert matrix.removed_after_confirm == 1
+    assert matrix.shares[QualityCause.AS_IS] == HALF
+    assert sum(matrix.shares.values()) == pytest.approx(1.0)
+
+
+def test_compute_matrix_empty_has_no_division_by_zero():
+    matrix = compute_matrix([])
+    assert matrix.denominator == 0
+    assert set(matrix.counts) == set(QualityCause)
+    assert all(share == 0.0 for share in matrix.shares.values())
+
+
+def test_compute_matrix_takes_latest_fact_per_document():
+    doc = uuid.uuid4()
+    first = make_outcome_row(document_id=doc, primary_cause=QualityCause.OUR_ERROR)
+    later = make_outcome_row(
+        document_id=doc,
+        primary_cause=None,
+        outcome=QualityOutcome.DELETED_AFTER_CONFIRM,
+        outcome_at=datetime.datetime(2026, 10, 20, tzinfo=datetime.UTC),
+    )
+    for ordered in ([first, later], [later, first]):
+        matrix = compute_matrix(ordered)
+        assert matrix.denominator == 0
+        assert matrix.removed_after_confirm == 1
+
+
+def test_compute_matrix_confirmed_without_cause_is_invariant_violation():
+    with pytest.raises(ValueError):
+        compute_matrix([make_outcome_row(primary_cause=None)])
+
+
+def test_monthly_series_fills_gaps_and_ignores_removed():
+    def at(month: int) -> datetime.datetime:
+        return datetime.datetime(2026, month, 5, tzinfo=datetime.UTC)
+
+    outcomes = [
+        make_outcome_row(primary_cause=QualityCause.OUR_ERROR, outcome_at=at(8)),
+        make_outcome_row(primary_cause=QualityCause.AS_IS, outcome_at=at(8)),
+        make_outcome_row(primary_cause=QualityCause.OUR_ERROR, outcome_at=at(10)),
+        make_outcome_row(primary_cause=None, outcome=QualityOutcome.DELETED, outcome_at=at(12)),
+    ]
+    series = compute_monthly_series(outcomes, QualityCause.OUR_ERROR)
+    assert [(p.month.month, p.count, p.denominator, p.share) for p in series] == [
+        (8, 1, 2, 0.5),
+        (9, 0, 0, 0.0),
+        (10, 1, 1, 1.0),
+    ]
+
+
+def test_monthly_series_empty_and_year_rollover():
+    assert compute_monthly_series([], QualityCause.AS_IS) == []
+    outcomes = [
+        make_outcome_row(outcome_at=datetime.datetime(2026, 12, 31, tzinfo=datetime.UTC)),
+        make_outcome_row(outcome_at=datetime.datetime(2027, 1, 1, tzinfo=datetime.UTC)),
+    ]
+    months = [p.month for p in compute_monthly_series(outcomes, QualityCause.AS_IS)]
+    assert months == [datetime.date(2026, 12, 1), datetime.date(2027, 1, 1)]