feat: escalation threshold derived from costs; /metrics exposes decision economics

livemedcard/api.py

@@ -36,6 +36,7 @@ from fastapi import Body, Depends, FastAPI, HTTPException, Request, Response
 from fastapi.responses import FileResponse
 from pydantic import BaseModel
 
+from . import economics
 from .bundle import from_bundle, to_bundle
 from .config import (
     EXTRACTOR,
@@ -316,6 +317,24 @@ def metrics(state: _State = Depends(_get_state)):
         log = card.audit_log()
         n = len(log)
         escalated = sum(1 for a in log if a["escalate"])
+
+        # Економіка рішень: у що обходиться поточна політика L3 і що саме купує
+        # ескалація. `no_escalation` — контрфактичний прогін тієї самої партії
+        # документів без жодної передачі людині: різниця і є цінністю шару L3.
+        cost_actual = sum(
+            economics.expected_cost(
+                a["extraction_confidence"], a.get("hazard", "medication"),
+                escalated=a["escalate"], high_risk_signal=a.get("high_risk", False),
+            )
+            for a in log
+        )
+        cost_no_escalation = sum(
+            economics.expected_cost(
+                a["extraction_confidence"], a.get("hazard", "medication"),
+                escalated=False, high_risk_signal=a.get("high_risk", False),
+            )
+            for a in log
+        )
         return {
             "documents": card.document_count,
             "observations": len(card.observations),
@@ -325,6 +344,12 @@ def metrics(state: _State = Depends(_get_state)):
             "auto": n - escalated,
             "escalation_rate": round(escalated / n, 3) if n else 0.0,
             "active_signals": len(card.current_signals()),
+            "economics": {
+                **economics.summary(),
+                "expected_cost": round(cost_actual, 2),
+                "expected_cost_without_escalation": round(cost_no_escalation, 2),
+                "escalation_saves": round(cost_no_escalation - cost_actual, 2),
+            },
         }
 
 

livemedcard/config.py

@@ -11,8 +11,24 @@ from pathlib import Path
 DATA_DIR = Path(__file__).parent / "data"
 STATIC_DIR = Path(__file__).parent / "static"
 
-# Поріг впевненості витягу L2b, нижче якого документ іде на перегляд людиною (L3).
-LOW_CONFIDENCE = float(os.getenv("LMC_LOW_CONFIDENCE", "0.6"))
+# ── Економіка рішень (виводить поріг ескалації; див. economics.py) ────────────
+# Поріг L3 більше не константа: він рахується як c* = 1 - C_human/(π·C_harm).
+# Числа нижче — робочі ПРИПУЩЕННЯ, не дані; перед клінічним використанням
+# замінити на citable (AHRQ HCUP для вартості ADE-госпіталізації тощо).
+COST_HUMAN_REVIEW = float(os.getenv("LMC_COST_HUMAN_REVIEW", "5"))          # перевірка людиною
+COST_HARM_MEDICATION = float(os.getenv("LMC_COST_HARM_MEDICATION", "15000"))  # ADE-госпіталізація
+COST_HARM_MEASUREMENT = float(os.getenv("LMC_COST_HARM_MEASUREMENT", "500"))  # пропущений тренд
+HAZARD_PRIOR_MEDICATION = float(os.getenv("LMC_HAZARD_PRIOR_MEDICATION", "0.01"))
+HAZARD_PRIOR_MEASUREMENT = float(os.getenv("LMC_HAZARD_PRIOR_MEASUREMENT", "0.05"))
+# Ймовірність, що знайдена (детерміновано!) небезпека таки нашкодить, якщо документ
+# не проглянула людина: родина може не зрозуміти сигналу або не дійти до лікаря.
+P_UNADDRESSED = float(os.getenv("LMC_P_UNADDRESSED", "0.2"))
+
+# Явний адміністративний override порога (env `LMC_LOW_CONFIDENCE`). Порожньо →
+# поріг виводиться з вартостей. Лишається для тестів і для випадку, коли оператор
+# свідомо задає поріг вручну.
+_low_conf_env = os.getenv("LMC_LOW_CONFIDENCE", "")
+LOW_CONFIDENCE_OVERRIDE: float | None = float(_low_conf_env) if _low_conf_env else None
 
 # Впевненість, яку StubExtractor приписує кожному словниковому збігу.
 # Реальний екстрактор (MedGemma) має ставити модельну впевненість замість цієї.

livemedcard/dto.py

@@ -92,3 +92,10 @@ class AuditEntry(BaseModel):
     n_signals: int
     route_to: Literal["auto", "L4_human"]
     escalate: bool
+    # Чим ризикував документ — від цього залежав поріг ескалації (economics).
+    # Дефолт зберігає сумісність зі старими знімками.
+    hazard: str = "medication"
+    # Чи на момент рішення був детермінований сигнал major/critical. Це ДРУГЕ,
+    # незалежне джерело вартості: небезпека вже знайдена, і питання лише в тому,
+    # чи її хтось прогляне.
+    high_risk: bool = False

livemedcard/economics.py

@@ -0,0 +1,138 @@
+"""Економіка рішень TRACE: поріг ескалації як відношення вартостей.
+
+Мотивація. `LOW_CONFIDENCE = 0.6` було магічним числом: його ніхто не виводив, і
+захистити його перед регулятором (EU AI Act ст.14 — обґрунтований human oversight)
+нічим. Тут воно виводиться.
+
+**Ключова властивість TRACE.** Помилка безпеки факторизується:
+
+    P(пропуск) = (1 - R) · 1  +  R · P(правило не спрацювало)
+                                     └─────────── = 0 ────────┘
+
+Другий доданок строго нульовий: правила L1/L2a вичерпні над своєю БД і нічого не
+«вирішують». Отже весь ризик безпеки зведено до одного вимірюваного числа —
+recall витягу R. Саме тому впевненість витягу (а не якась «впевненість моделі у
+висновку») — єдина величина, за якою має ухвалюватись рішення про ескалацію.
+
+**Правило рішення.** Ескалювати до людини варто тоді, коли очікувана шкода від
+автоматичної обробки перевищує вартість перевірки:
+
+    (1 - c) · π · C_harm  >  C_human
+    ⟹  c* = 1 - C_human / (π · C_harm)
+
+де c — впевненість витягу, π — частка документів цього класу, у яких узагалі є
+факт, чий пропуск шкодить, C_harm — вартість такої шкоди, C_human — вартість
+перегляду людиною.
+
+Наслідок, який варто проговорити вголос: при асиметрії «$5 проти $15 000» поріг
+для документів із ліками виходить ≈0.97, тобто **майже будь-який сумнів має йти до
+людини**. Для документів лише з показниками (пропуск тренду не є гострою подією)
+поріг помітно нижчий, і автоматична обробка економічно виправдана. Тобто поріг —
+не один на всіх, а функція від того, чим документ ризикує.
+
+⚠ Числа вартостей нижче — **робочі припущення, не дані**. Перед будь-яким
+клінічним чи інвесторським використанням їх треба замінити на citable: вартість
+ADE-госпіталізації (AHRQ HCUP), базовий ризик для 65+ на 5+ ліках, реальну
+вартість хвилини перевірки. Модель від цього не змінюється — змінюються входи.
+"""
+from __future__ import annotations
+
+from typing import Literal
+
+from .config import (
+    COST_HARM_MEDICATION,
+    COST_HARM_MEASUREMENT,
+    COST_HUMAN_REVIEW,
+    HAZARD_PRIOR_MEDICATION,
+    HAZARD_PRIOR_MEASUREMENT,
+    LOW_CONFIDENCE_OVERRIDE,
+    P_UNADDRESSED,
+)
+
+# Чим ризикує документ. `medication` — з нього витягнуто препарати, тож пропуск
+# факту може коштувати взаємодії або передозування (гостра подія). `measurement` —
+# лише показники: пропуск тренду шкодить, але не гостро. `none` — фактів немає.
+Hazard = Literal["medication", "measurement", "none"]
+
+_PARAMS: dict[str, tuple[float, float]] = {           # hazard → (π, C_harm)
+    "medication": (HAZARD_PRIOR_MEDICATION, COST_HARM_MEDICATION),
+    "measurement": (HAZARD_PRIOR_MEASUREMENT, COST_HARM_MEASUREMENT),
+    "none": (HAZARD_PRIOR_MEASUREMENT, COST_HARM_MEASUREMENT),
+}
+
+# Поріг не має вироджуватись у крайнощі: 1.0 означав би «ніколи не довіряти», 0.0 —
+# «ніколи не перевіряти». Обидва роблять шар L3 беззмістовним.
+_FLOOR, _CEIL = 0.05, 0.99
+
+
+def escalation_threshold(hazard: Hazard = "medication") -> float:
+    """Поріг впевненості витягу, нижче якого документ іде до людини.
+
+    Виводиться з вартостей: ``c* = 1 - C_human / (π · C_harm)``. Env
+    ``LMC_LOW_CONFIDENCE`` лишається явним override — для тестів і для випадку,
+    коли оператор свідомо задає поріг адміністративно, а не економічно.
+    """
+    if LOW_CONFIDENCE_OVERRIDE is not None:
+        return LOW_CONFIDENCE_OVERRIDE
+    prior, harm = _PARAMS[hazard]
+    expected_harm = prior * harm
+    if expected_harm <= 0:
+        return _FLOOR
+    return min(_CEIL, max(_FLOOR, 1.0 - COST_HUMAN_REVIEW / expected_harm))
+
+
+def expected_cost(
+    confidence: float,
+    hazard: Hazard = "medication",
+    *,
+    escalated: bool = False,
+    high_risk_signal: bool = False,
+) -> float:
+    """Очікувана вартість одного рішення (умовні одиниці вартості).
+
+    Ескалація коштує рівно ``C_human``: людина перевіряє документ, і залишковий
+    ризик вважаємо нульовим (припущення на користь людини — воно ж робить оцінку
+    консервативною щодо цінності автоматизації).
+
+    В автоматичної обробки **два різні джерела шкоди**, і плутати їх не можна:
+
+    1. **Ризик витягу** — факт міг не витягнутись: ``(1-c)·π·C_harm``. Тут π —
+       апріорна частка документів, у яких взагалі є що пропускати.
+    2. **Непроглянута підтверджена небезпека** — детермінований сигнал уже
+       спрацював (взаємодія, перевищення дози), тобто небезпека не гіпотетична, а
+       наявна. Тоді π не потрібне: ризик у тому, що родина не відреагує на знайдене
+       без лікаря — ``P_unaddressed · C_harm``.
+
+    Без другого доданка модель вважала ескалацію на реальній взаємодії марною
+    витратою ($5 проти нуля) — і виходило, що шар L3 приносить збиток. Це був
+    дефект моделі, а не політики.
+    """
+    if escalated:
+        return COST_HUMAN_REVIEW
+    prior, harm = _PARAMS[hazard]
+    cost = max(0.0, 1.0 - confidence) * prior * harm
+    if high_risk_signal:
+        cost += P_UNADDRESSED * _PARAMS["medication"][1]
+    return cost
+
+
+def hazard_of(*, has_medications: bool, has_observations: bool) -> Hazard:
+    """Чим ризикує документ — за тим, що з нього фактично витягнуто."""
+    if has_medications:
+        return "medication"
+    return "measurement" if has_observations else "none"
+
+
+def summary() -> dict:
+    """Параметри та виведені пороги — для `/metrics` і для захисту рішення."""
+    return {
+        "cost_human_review": COST_HUMAN_REVIEW,
+        "cost_harm_medication": COST_HARM_MEDICATION,
+        "cost_harm_measurement": COST_HARM_MEASUREMENT,
+        "hazard_prior_medication": HAZARD_PRIOR_MEDICATION,
+        "hazard_prior_measurement": HAZARD_PRIOR_MEASUREMENT,
+        "p_unaddressed": P_UNADDRESSED,
+        "threshold_medication": round(escalation_threshold("medication"), 3),
+        "threshold_measurement": round(escalation_threshold("measurement"), 3),
+        "threshold_source": "override" if LOW_CONFIDENCE_OVERRIDE is not None else "derived",
+    }

livemedcard/layers/l3_router.py

@@ -13,10 +13,10 @@ from __future__ import annotations
 
 from typing import Optional
 
-from ..config import LOW_CONFIDENCE
 from ..dto import Decision, Severity, Signal, severity_rank
+from ..economics import Hazard, escalation_threshold
 
-__all__ = ["route", "LOW_CONFIDENCE"]
+__all__ = ["route"]
 
 
 def route(
@@ -24,15 +24,22 @@ def route(
     extraction_confidence: float,
     doc_kind: str,
     lang: str = "uk",
+    hazard: Hazard = "medication",
 ) -> Decision:
-    """Приймає рішення про маршрут документа: "auto" чи "L4_human"."""
+    """Приймає рішення про маршрут документа: "auto" чи "L4_human".
+
+    Поріг впевненості не константа: він виводиться з вартостей (`economics`) і
+    залежить від того, чим ризикує документ. Пропуск препарату коштує дорожче за
+    пропущений тренд, тож і планка довіри для нього вища.
+    """
     top: Optional[Signal] = max(
         signals, key=lambda s: severity_rank(s.severity), default=None
     )
     high_risk = top is not None and severity_rank(top.severity) >= severity_rank(
         _MAJOR
     )
-    low_confidence = extraction_confidence < LOW_CONFIDENCE
+    threshold = escalation_threshold(hazard)
+    low_confidence = extraction_confidence < threshold
     escalate = high_risk or low_confidence
 
     en = lang == "en"
@@ -47,10 +54,12 @@ def route(
         )
     if low_confidence:
         reasons.append(
-            f"Low extraction confidence ({extraction_confidence:.2f}) — "
-            f"human review required"
+            f"Extraction confidence {extraction_confidence:.2f} < {threshold:.2f} — "
+            f"the threshold follows from the cost of a missed fact, not a tuned "
+            f"hyperparameter — human review required"
             if en else
-            f"Низька впевненість витягу ({extraction_confidence:.2f}) — "
+            f"Впевненість витягу {extraction_confidence:.2f} < {threshold:.2f} — "
+            f"поріг виведено з вартості пропущеного факту, а не підібрано — "
             f"потрібна перевірка людиною"
         )
     if not escalate:

livemedcard/pipeline.py

@@ -12,7 +12,15 @@ from __future__ import annotations
 from datetime import datetime, timezone
 from typing import Optional
 
-from .dto import AuditEntry, Document, ExtractionResult, Signal, TraceReport
+from . import economics
+from .dto import (
+    AuditEntry,
+    Document,
+    ExtractionResult,
+    Signal,
+    TraceReport,
+    severity_rank,
+)
 from .fhir import (
     Condition,
     DocumentReference,
@@ -120,8 +128,13 @@ class LiveMedCard:
         _assert_provenance(ext)
         self._absorb(ext)
         signals = self.current_signals()
-        decision = l3_router.route(signals, ext.mean_confidence, doc.kind)
-        self._record_audit(doc, ext, signals, decision)
+        # Чим ризикує саме цей документ — визначає планку довіри (economics).
+        hazard = economics.hazard_of(
+            has_medications=bool(ext.medications),
+            has_observations=bool(ext.observations),
+        )
+        decision = l3_router.route(signals, ext.mean_confidence, doc.kind, hazard=hazard)
+        self._record_audit(doc, ext, signals, decision, hazard)
         return TraceReport(
             document_id=doc.id, extraction=ext, signals=signals, decision=decision
         )
@@ -130,7 +143,10 @@ class LiveMedCard:
         return [self.ingest(d) for d in docs]
 
     # ── аудит рішень L3 ───────────────────────────────────────────────────────
-    def _record_audit(self, doc, ext, signals, decision) -> None:
+    def _record_audit(self, doc, ext, signals, decision, hazard: str = "medication") -> None:
+        high_risk = any(
+            severity_rank(s.severity) >= severity_rank("major") for s in signals
+        )
         self._audit.append(AuditEntry(
             ts=datetime.now(timezone.utc),
             document_id=doc.id,
@@ -139,6 +155,8 @@ class LiveMedCard:
             n_signals=len(signals),
             route_to=decision.route_to,
             escalate=decision.escalate,
+            hazard=hazard,
+            high_risk=high_risk,
         ))
 
     def audit_log(self) -> list[dict]: