Spaces:
Sleeping
Sleeping
File size: 3,930 Bytes
a8c0492 68bc1c7 a8c0492 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | from __future__ import annotations
from typing import Literal
from app.ml.archive_model import clamp
RUNTIME_STACK_VERSION = "archive-evidence-fusion-v4"
SourceHint = Literal["roi_original", "palpebral", "forniceal_palpebral"]
DEFAULT_SOURCE_THRESHOLDS: dict[SourceHint, float] = {
"roi_original": 0.495,
"palpebral": 0.65,
"forniceal_palpebral": 0.65,
}
DEFAULT_RISK_ARCHIVE_WEIGHTS: dict[SourceHint, float] = {
"roi_original": 0.55,
"palpebral": 1.0,
"forniceal_palpebral": 1.0,
}
DEFAULT_HB_ARCHIVE_WEIGHTS: dict[SourceHint, float] = {
"roi_original": 0.70,
"palpebral": 1.0,
"forniceal_palpebral": 1.0,
}
def decision_threshold_for_source(source_hint: SourceHint = "roi_original") -> float:
return float(DEFAULT_SOURCE_THRESHOLDS.get(source_hint, DEFAULT_SOURCE_THRESHOLDS["roi_original"]))
def risk_archive_weight_for_source(source_hint: SourceHint = "roi_original") -> float:
return float(DEFAULT_RISK_ARCHIVE_WEIGHTS.get(source_hint, DEFAULT_RISK_ARCHIVE_WEIGHTS["roi_original"]))
def hb_archive_weight_for_source(source_hint: SourceHint = "roi_original") -> float:
return float(DEFAULT_HB_ARCHIVE_WEIGHTS.get(source_hint, DEFAULT_HB_ARCHIVE_WEIGHTS["roi_original"]))
def build_runtime_stack_prediction(
archive_prediction: dict[str, float],
*,
efficientnet_prediction: dict[str, float] | None = None,
source_hint: SourceHint = "roi_original",
) -> dict[str, float]:
archive_risk = float(archive_prediction["anemia_risk"])
archive_hb = float(archive_prediction["predicted_hemoglobin"])
archive_uncertainty = float(archive_prediction["uncertainty"])
risk = archive_risk
predicted_hemoglobin = archive_hb
uncertainty = archive_uncertainty
if efficientnet_prediction is not None:
efficientnet_risk = float(efficientnet_prediction["anemia_risk"])
efficientnet_hb = float(efficientnet_prediction["predicted_hemoglobin"])
efficientnet_uncertainty = float(efficientnet_prediction.get("uncertainty", 0.35))
# Confidence-weighted ensemble: lower uncertainty → higher weight
archive_conf = clamp(1.0 - archive_uncertainty)
efficientnet_conf = clamp(1.0 - efficientnet_uncertainty)
total_conf = archive_conf + efficientnet_conf + 1e-9
# Apply source-specific floor weight for archive (it has calibrated features)
source_floor = risk_archive_weight_for_source(source_hint)
raw_archive_w = archive_conf / total_conf
# Blend floor weight with confidence-derived weight
archive_w = clamp(0.5 * source_floor + 0.5 * raw_archive_w, 0.25, 0.80)
efficientnet_w = 1.0 - archive_w
disagreement = abs(archive_risk - efficientnet_risk)
hemoglobin_gap = abs(archive_hb - efficientnet_hb)
# Agreement bonus: both models agree on direction → reduce uncertainty
agreement_bonus = 0.0
if (archive_risk > 0.5) == (efficientnet_risk > 0.5):
agreement_bonus = 0.04 + disagreement * 0.06
risk = (archive_w * archive_risk) + (efficientnet_w * efficientnet_risk)
hb_archive_w = hb_archive_weight_for_source(source_hint)
predicted_hemoglobin = (hb_archive_w * archive_hb) + ((1.0 - hb_archive_w) * efficientnet_hb)
uncertainty = clamp(
(archive_w * archive_uncertainty)
+ (efficientnet_w * efficientnet_uncertainty)
+ (disagreement * 0.08)
+ (min(hemoglobin_gap / 12.0, 1.0) * 0.025)
- agreement_bonus,
0.04,
0.92,
)
return {
"anemia_risk": clamp(risk, 0.0, 1.0),
"predicted_hemoglobin": predicted_hemoglobin,
"uncertainty": clamp(uncertainty, 0.04, 0.95),
"decision_threshold": decision_threshold_for_source(source_hint),
}
|