Blame for src/risk_engine.py

"""Phase 7 — transparent risk fusion engine.

Fuses four inputs into a heuristic risk score and a LOW/MEDIUM/HIGH/CRITICAL
level:

- ``near_miss_probability`` — calibrated P(class_1) from the Phase 4
  near-miss/conflict model (kept semantically separate from the accident
  probability: different model, different event, never merged into one
  probability).
- ``accident_probability`` — calibrated accident probability from the Phase 6
  detector. Raw detector confidence is NEVER accepted as a probability here.
- ``severity``             — caller-supplied relative severity in [0, 1].
  Neither source dataset carries severity labels; the meaning of this number
  is owned by the caller. This engine does not measure or interpret severity.
- ``event_age_hours``      — caller-supplied event age; converted to a
  recency factor via an exponential half-life decay from config.

Missing inputs (``None``) are excluded and the weights of the PRESENT inputs
are renormalized, so an assessment is always computable from whatever
evidence exists — but an assessment with no inputs at all raises ValueError
instead of silently returning a confident-looking LOW.

The score is a HEURISTIC engineering fusion, not a probability and not a
scientifically validated quantity. All weights/thresholds come from
``src/config.py``; nothing is hardcoded here.
"""

from __future__ import annotations

import math
from typing import Any

try:  # src/ root import (risk_engine.py sits next to config.py)
    import config as _config
except ImportError:  # package-style import
    from . import config as _config

__all__ = ["assess_risk", "compute_risk_level", "recency_factor", "RiskInputError"]

#: Config keys this module reads, exposed so callers can build overrides.
CONFIG_KEYS = ("risk_weights", "risk_level_thresholds", "risk_score_scale", "recency_half_life_hours")

#: Fusion input keys, in canonical output order.
INPUT_KEYS = ("near_miss", "accident", "severity", "recency")


class RiskInputError(ValueError):
    """Raised for missing-evidence or invalid risk-fusion inputs."""


# ---------------------------------------------------------------------------
# Validation helpers — validate at the boundary, fail fast, never guess
# ---------------------------------------------------------------------------

def _validate_unit_interval(name: str, value: Any) -> float | None:
    """Validate a probability-like input; ``None`` means genuinely missing."""
    if value is None:
        return None
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise RiskInputError(f"{name} must be a number or None, got {type(value).__name__}")
    numeric = float(value)
    if not math.isfinite(numeric):
        raise RiskInputError(f"{name} must be finite, got {numeric}")
    if not 0.0 <= numeric <= 1.0:
        raise RiskInputError(f"{name} must be within [0, 1], got {numeric}")
    return numeric


def _validate_age_hours(value: Any) -> float | None:
    """Validate a caller-supplied event age in hours; ``None`` = missing."""
    if value is None:
        return None
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise RiskInputError(f"event_age_hours must be a number or None, got {type(value).__name__}")
    numeric = float(value)
    if not math.isfinite(numeric):
        raise RiskInputError(f"event_age_hours must be finite, got {numeric}")
    if numeric < 0.0:
        raise RiskInputError(f"event_age_hours must be >= 0, got {numeric}")
    return numeric


def _validate_weights(weights: dict[str, Any]) -> dict[str, float]:
    missing = [key for key in INPUT_KEYS if key not in weights]
    if missing:
        raise RiskInputError(f"risk_weights is missing keys: {missing}")
    validated: dict[str, float] = {}
    for key in INPUT_KEYS:
        raw = weights[key]
        if isinstance(raw, bool) or not isinstance(raw, (int, float)) or not math.isfinite(float(raw)):
            raise RiskInputError(f"risk_weights[{key!r}] must be a finite number, got {raw!r}")
        if float(raw) <= 0.0:
            raise RiskInputError(f"risk_weights[{key!r}] must be > 0, got {raw}")
        validated[key] = float(raw)
    return validated


def _validate_thresholds(thresholds: dict[str, Any]) -> dict[str, float]:
    missing = [name for name in ("medium", "high", "critical") if name not in thresholds]
    if missing:
        raise RiskInputError(f"risk_level_thresholds is missing keys: {missing}")
    validated = {name: float(thresholds[name]) for name in ("medium", "high", "critical")}
    ordered = [validated["medium"], validated["high"], validated["critical"]]
    if any(not math.isfinite(v) for v in ordered):
        raise RiskInputError("risk_level_thresholds must be finite")
    if not 0 < ordered[0] < ordered[1] < ordered[2]:
        raise RiskInputError(
            "risk_level_thresholds must satisfy 0 < medium < high < critical, "
            f"got medium={ordered[0]}, high={ordered[1]}, critical={ordered[2]}"
        )
    return validated


def _validate_scale(scale: Any) -> float:
    if isinstance(scale, bool) or not isinstance(scale, (int, float)) or not math.isfinite(float(scale)):
        raise RiskInputError(f"risk_score_scale must be a finite number, got {scale!r}")
    if float(scale) <= 0.0:
        raise RiskInputError(f"risk_score_scale must be > 0, got {scale}")
    return float(scale)


def _resolve_config(overrides: dict[str, Any] | None) -> dict[str, Any]:
    """Merge caller overrides onto the src/config.py defaults."""
    resolved = {
        "risk_weights": dict(_config.RISK_WEIGHTS),
        "risk_level_thresholds": dict(_config.RISK_LEVEL_THRESHOLDS),
        "risk_score_scale": _config.RISK_SCORE_SCALE,
        "recency_half_life_hours": _config.RECENCY_HALF_LIFE_HOURS,
    }
    if overrides:
        unknown = [k for k in overrides if k not in CONFIG_KEYS]
        if unknown:
            raise RiskInputError(f"Unknown config override keys: {unknown}; expected {CONFIG_KEYS}")
        for key, value in overrides.items():
            if key in ("risk_weights", "risk_level_thresholds"):
                if not isinstance(value, dict):
                    raise RiskInputError(f"config override {key!r} must be a dict, got {type(value).__name__}")
                resolved[key].update(value)
            else:
                resolved[key] = value
    resolved["risk_weights"] = _validate_weights(resolved["risk_weights"])
    resolved["risk_level_thresholds"] = _validate_thresholds(resolved["risk_level_thresholds"])
    resolved["risk_score_scale"] = _validate_scale(resolved["risk_score_scale"])
    half_life = resolved["recency_half_life_hours"]
    if (
        isinstance(half_life, bool)
        or not isinstance(half_life, (int, float))
        or not math.isfinite(float(half_life))
        or float(half_life) <= 0.0
    ):
        raise RiskInputError(f"recency_half_life_hours must be a finite number > 0, got {half_life!r}")
    resolved["recency_half_life_hours"] = float(half_life)
    return resolved


# ---------------------------------------------------------------------------
# Fusion math
# ---------------------------------------------------------------------------

def recency_factor(event_age_hours: float | None, half_life_hours: float) -> float | None:
    """Exponential recency decay: ``0.5 ** (age / half_life)``.

    Age 0 -> 1.0, one half-life -> 0.5, and sufficiently large ages decay to
    0.0 (float underflow, which is the mathematically correct limit).
    """
    if event_age_hours is None:
        return None
    return 0.5 ** (float(event_age_hours) / float(half_life_hours))


def compute_risk_level(score: float, thresholds: dict[str, float]) -> str:
    """Map a fused score to its level (thresholds are inclusive lower bounds)."""
    if score >= thresholds["critical"]:
        return "CRITICAL"
    if score >= thresholds["high"]:
        return "HIGH"
    if score >= thresholds["medium"]:
        return "MEDIUM"
    return "LOW"


def assess_risk(
    near_miss_probability: float | None = None,
    accident_probability: float | None = None,
    severity: float | None = None,
    event_age_hours: float | None = None,
    config: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Fuse the four inputs into a transparent heuristic risk assessment.

    All inputs are optional; ``None`` means the input is genuinely missing and
    it is excluded, with the weights of the present inputs renormalized. At
    least one input must be present.

    Returns a JSON-serializable dict (see the module docstring for the input
    semantics and the caveats).
    """
    cfg = _resolve_config(config)
    weights = cfg["risk_weights"]
    thresholds = cfg["risk_level_thresholds"]
    scale = cfg["risk_score_scale"]

    values: dict[str, float | None] = {
        "near_miss": _validate_unit_interval("near_miss_probability", near_miss_probability),
        "accident": _validate_unit_interval("accident_probability", accident_probability),
        "severity": _validate_unit_interval("severity", severity),
        "recency": recency_factor(
            _validate_age_hours(event_age_hours), cfg["recency_half_life_hours"]
        ),
    }
    present = {key: value for key, value in values.items() if value is not None}
    if not present:
        raise RiskInputError(
            "No risk inputs provided — refused to invent a risk level. "
            f"Provide at least one of: {', '.join(INPUT_KEYS)}."
        )

    weight_total = sum(weights[key] for key in present)
    contributions = {key: weights[key] * value for key, value in present.items()}
    fused = sum(contributions.values())
    risk_score = scale * fused / weight_total
    risk_level = compute_risk_level(risk_score, thresholds)

    echo = {
        "near_miss": near_miss_probability,
        "accident": accident_probability,
        "severity": severity,
        "event_age_hours": event_age_hours,
    }
    missing = [key for key in INPUT_KEYS if key not in present]
    explanation = _build_explanation(
        echo=echo,
        present=present,
        missing=missing,
        weights=weights,
        contributions=contributions,
        weight_total=weight_total,
        risk_score=risk_score,
        thresholds=thresholds,
        half_life=cfg["recency_half_life_hours"],
    )

    return {
        "near_miss_probability": near_miss_probability,
        "accident_probability": accident_probability,
        "severity": severity,
        "event_age_hours": event_age_hours,
        "risk_score": round(risk_score, 4),
        "risk_level": risk_level,
        "present_inputs": sorted(present),
        "missing_inputs": missing,
        "weights_applied": {key: round(weights[key] / weight_total, 6) for key in present},
        "contributions": {key: round(scale * contributions[key] / weight_total, 4) for key in present},
        "recency_factor": None if values["recency"] is None else round(values["recency"], 6),
        "is_probability": False,
        "explanation": explanation,
    }


def _fmt(value: float | None) -> str:
    return "missing" if value is None else f"{value}"


def _build_explanation(
    echo: dict[str, float | None],
    present: dict[str, float],
    missing: list[str],
    weights: dict[str, float],
    contributions: dict[str, float],
    weight_total: float,
    risk_score: float,
    thresholds: dict[str, float],
    half_life: float,
) -> list[str]:
    """Human-readable, formula-transparent rationale for the assessment."""
    lines = [
        (
            "risk_score is a heuristic fusion score, NOT a probability and NOT "
            "scientifically validated; it must not be used to control traffic signals."
        ),
        (
            "Formula: risk_score = scale * sum(weight_i * value_i for present inputs) "
            "/ sum(weight_i for present inputs)."
        ),
        (
            "near_miss_probability and accident_probability are kept semantically "
            "separate: they come from different models measuring different events "
            "and are never merged into a single probability."
        ),
        (
            "accident_probability is the calibrated accident probability from the "
            "Phase 6 detector; raw detector confidence is never treated as a probability."
        ),
        (
            "near-miss label semantics are unconfirmed in the source data (class_1 is "
            "only statistically consistent with a conflict/near-miss); the value is "
            "echoed from the Phase 4 artifact without added interpretation."
        ),
        (
            "severity is a caller-supplied relative value in [0, 1]; this pipeline "
            "does not measure severity and does not define its meaning."
        ),
    ]
    for key in INPUT_KEYS:
        label = {
            "near_miss": "near_miss_probability",
            "accident": "accident_probability",
            "severity": "severity",
            "recency": f"recency factor from event_age_hours (half-life {half_life} h)",
        }[key]
        if key in present:
            display = echo["event_age_hours"] if key == "recency" else echo[key]
            lines.append(
                f"Input {label} = {_fmt(display)}"
                f" -> contribution {contributions[key]:.4f} (weight {weights[key]})."
            )
        else:
            lines.append(f"Input {label} is missing -> excluded from the fusion.")
    if missing:
        lines.append(
            "Missing inputs are excluded and weights are renormalized over the "
            f"present inputs (weight total used: {weight_total:.4f}); absence of an "
            "input is treated as absence of evidence, not as evidence of absence."
        )
    lines.append(
        f"Fused score {risk_score:.4f} vs thresholds (inclusive lower bounds) "
        f"medium={thresholds['medium']}, high={thresholds['high']}, "
        f"critical={thresholds['critical']}."
    )
    return lines