Calculating Penalty Risk Scores Based on State Grace Periods #

import hashlib
import logging
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal, ROUND_HALF_UP, getcontext
from enum import Enum
from typing import Any, Optional

# Fixed-point context set once at import — NOT inside request handlers,
# where thread-local leakage would cause silent rounding drift in batch runs.
getcontext().prec = 12
getcontext().rounding = ROUND_HALF_UP

logger = logging.getLogger("compliance.penalty_risk")
logger.setLevel(logging.INFO)


class EnforcementVelocity(Enum):
    """How aggressively a portal escalates to suspension/dissolution."""
    LOW = Decimal("0.25")
    MEDIUM = Decimal("0.50")
    HIGH = Decimal("0.85")
    CRITICAL = Decimal("1.00")


@dataclass(frozen=True)
class JurisdictionConfig:
    state_code: str
    grace_days: int
    base_penalty: Decimal
    monthly_interest_rate: Decimal
    enforcement_velocity: EnforcementVelocity
    suspension_days: int


@dataclass(frozen=True)
class PenaltyRiskResult:
    entity_id: str
    state_code: str
    days_past_due: int
    projected_penalty: Decimal
    penalty_risk: int          # the 0-100 signal the router consumes
    input_hash: str
    output_hash: str
    timestamp: datetime


class TTLCache:
    """Per-jurisdiction config cache so a statutory update can be invalidated."""

    def __init__(self, ttl_seconds: int = 3600) -> None:
        self._cache: dict[str, tuple[Any, float]] = {}
        self._ttl = ttl_seconds

    def get(self, key: str) -> Optional[Any]:
        hit = self._cache.get(key)
        if hit is None:
            return None
        value, ts = hit
        if time.time() - ts < self._ttl:
            return value
        self._cache.pop(key, None)
        return None

    def set(self, key: str, value: Any) -> None:
        self._cache[key] = (value, time.time())

    def invalidate(self, key: Optional[str] = None) -> None:
        if key:
            self._cache.pop(key, None)
        else:
            self._cache.clear()


class PenaltyRiskCalculator:
    # Weights for the four sub-signals; sum to 1.0 by contract.
    W_TEMPORAL = Decimal("0.35")
    W_FINANCIAL = Decimal("0.25")
    W_CRITICALITY = Decimal("0.20")
    W_ENFORCEMENT = Decimal("0.20")
    PENALTY_CAP = Decimal("10000.00")   # normalization ceiling for $ exposure

    def __init__(self, cache: Optional[TTLCache] = None) -> None:
        self.cache = cache or TTLCache()
        # Conservative fallback: an unknown state is treated as high-risk,
        # never low-risk, so missing metadata can never hide exposure.
        self._default_config = JurisdictionConfig(
            state_code="UNKNOWN",
            grace_days=0,
            base_penalty=Decimal("250.00"),
            monthly_interest_rate=Decimal("0.015"),
            enforcement_velocity=EnforcementVelocity.HIGH,
            suspension_days=60,
        )

    def _get_config(self, state_code: str) -> JurisdictionConfig:
        cached = self.cache.get(state_code)
        if cached is not None:
            return cached
        try:
            config = self._fetch_from_registry(state_code)
        except Exception as exc:  # registry down or state unknown
            logger.warning(
                "jurisdiction_metadata_unavailable_applying_fallback",
                extra={"state_code": state_code, "error": str(exc)},
            )
            config = self._default_config
        self.cache.set(state_code, config)
        return config

    def _fetch_from_registry(self, state_code: str) -> JurisdictionConfig:
        # Wire this to the compliance metadata registry in production.
        raise NotImplementedError("Registry integration required")

    def calculate(
        self,
        entity_id: str,
        state_code: str,
        due_date: datetime,
        criticality_weight: Decimal,
    ) -> PenaltyRiskResult:
        config = self._get_config(state_code)
        now = datetime.now(timezone.utc)
        # Truncate to whole calendar days; portals apply penalties at midnight,
        # not on fractional hours. due_date MUST be timezone-aware.
        days_past_due = max(0, (now - due_date).days)

        # Temporal exposure, normalized against the suspension cliff.
        if config.suspension_days > 0:
            t_norm = min(Decimal("1.0"),
                         Decimal(days_past_due) / Decimal(config.suspension_days))
        else:
            t_norm = Decimal("1.0")

        # Financial magnitude: flat until grace expires, then compounding.
        penalty = config.base_penalty
        if days_past_due > config.grace_days:
            months_late = (Decimal(days_past_due - config.grace_days)
                           / Decimal(30)).quantize(Decimal("0.01"))
            penalty += penalty * config.monthly_interest_rate * months_late
        f_norm = min(Decimal("1.0"), penalty / self.PENALTY_CAP)

        raw = (
            t_norm * self.W_TEMPORAL
            + f_norm * self.W_FINANCIAL
            + criticality_weight * self.W_CRITICALITY
            + config.enforcement_velocity.value * self.W_ENFORCEMENT
        )
        score = int((raw * Decimal("100")).quantize(Decimal("1"),
                                                     rounding=ROUND_HALF_UP))
        penalty_risk = min(100, max(0, score))

        # Cryptographic audit chain: input and output hashes make every
        # score replayable and tamper-evident during compliance review.
        input_payload = f"{entity_id}|{state_code}|{days_past_due}|{criticality_weight}"
        input_hash = hashlib.sha256(input_payload.encode()).hexdigest()
        output_payload = f"{penalty_risk}|{penalty}|{now.isoformat()}"
        output_hash = hashlib.sha256(output_payload.encode()).hexdigest()

        logger.info(
            "penalty_risk_scored",
            extra={
                "entity_id": entity_id,
                "state_code": state_code,
                "days_past_due": days_past_due,
                "penalty_risk": penalty_risk,
                "audit_hash": output_hash,
            },
        )

        return PenaltyRiskResult(
            entity_id=entity_id,
            state_code=state_code,
            days_past_due=days_past_due,
            projected_penalty=penalty,
            penalty_risk=penalty_risk,
            input_hash=input_hash,
            output_hash=output_hash,
            timestamp=now,
        )