src/analysis/risk.py

"""Risk scoring: converts verifier findings + ratio results into a risk
heatmap and an overall reliability score for the Advanced Analysis mode.
"""

from __future__ import annotations

from dataclasses import dataclass, field

RISK_LEVELS = ("low", "medium", "high")


@dataclass
class RiskItem:
    category: str          # e.g. "Consistency", "Liquidity", "Leverage"
    level: str             # low | medium | high
    rationale: str
    citations: list[str] = field(default_factory=list)


@dataclass
class RiskAssessment:
    items: list[RiskItem]
    reliability_score: int   # 0-100, document-set reliability

    def heatmap_rows(self) -> list[dict]:
        return [{"Category": i.category, "Risk": i.level.upper(),
                 "Rationale": i.rationale} for i in self.items]


def _ratio_risks(ratio_results: list[dict]) -> list[RiskItem]:
    items = []
    by_name = {r["name"]: r for r in ratio_results if r.get("value") is not None}

    cr = by_name.get("Current Ratio")
    if cr:
        v = cr["value"]
        level = "high" if v < 1.0 else "medium" if v < 1.2 else "low"
        items.append(RiskItem("Liquidity", level,
                              f"Current ratio {v:.2f}. {cr.get('interpretation', '')}"))
    de = by_name.get("Debt / Equity")
    if de:
        v = de["value"]
        level = "high" if v > 2.0 else "medium" if v > 1.0 else "low"
        items.append(RiskItem("Leverage", level,
                              f"Debt/Equity {v:.2f}. {de.get('interpretation', '')}"))
    ic = by_name.get("Interest Coverage")
    if ic:
        v = ic["value"]
        level = "high" if v < 1.5 else "medium" if v < 3.0 else "low"
        items.append(RiskItem("Solvency", level,
                              f"Interest coverage {v:.2f}x. {ic.get('interpretation', '')}"))
    return items


def _consistency_risk(findings: list[dict]) -> RiskItem:
    differing = [f for f in findings if f.get("status") == "differ"]
    if not findings:
        return RiskItem("Consistency", "medium",
                        "No cross-checkable claims found — consistency could not be assessed.")
    if not differing:
        return RiskItem("Consistency", "low",
                        f"All {len(findings)} cross-checked claims agree across sources.")
    high_conf = [f for f in differing if f.get("confidence", 0) >= 70]
    level = "high" if high_conf else "medium"
    topics = ", ".join(sorted({f.get("topic", "?") for f in differing}))
    return RiskItem("Consistency", level,
                    f"{len(differing)} potential inconsistency(ies) flagged for review: {topics}.")


def assess(findings: list[dict], ratio_results: list[dict] | None = None) -> RiskAssessment:
    items = [_consistency_risk(findings)]
    items.extend(_ratio_risks(ratio_results or []))

    penalty = {"low": 0, "medium": 12, "high": 30}
    reliability = max(0, 100 - sum(penalty[i.level] for i in items))
    # small bonus when many claims were actually verifiable
    verifiable = len([f for f in findings if f.get("status") in ("agree", "differ")])
    reliability = min(100, reliability + min(verifiable, 5))
    return RiskAssessment(items=items, reliability_score=reliability)