saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
104 lines
4.0 KiB
Python
104 lines
4.0 KiB
Python
"""Layer 5: significance scoring + severity classification.
|
|
|
|
This is the documented, unit-tested formula referenced in ARCHITECTURE.md.
|
|
Deterministic on purpose - severity must be explainable and reproducible
|
|
without a model call. Phase 7's LLM analysis narrates *why* a change
|
|
matters; it does not decide *how much* it matters.
|
|
|
|
significance = base_weight
|
|
* source_trust_score (0.3 - 1.0)
|
|
* min(1.0, independent_sources / 2) # corroboration, caps at 2 sources
|
|
* focus_match_multiplier (1.3 if it matches the user's stated focus, else 1.0)
|
|
* recency_multiplier (1.0 if new, 0.5 if a repeat of a recent change)
|
|
|
|
confidence = clamp(
|
|
0.5 * extraction_confidence + 0.4 * source_trust_score + 0.1
|
|
+ (0.15 if independent_sources >= 2 else 0.0),
|
|
0.0, 1.0
|
|
)
|
|
|
|
severity = bucket(significance * confidence), with a hard floor:
|
|
CRITICAL requires confidence >= CRITICAL_MIN_CONFIDENCE regardless of score -
|
|
an uncorroborated single-source signal can never be labeled Critical.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from app.models.enums import ChangeType, SeverityLevel
|
|
|
|
BASE_WEIGHTS: dict[ChangeType, float] = {
|
|
ChangeType.LEADERSHIP_CHANGE: 0.9,
|
|
ChangeType.FILING_NEW: 0.85,
|
|
ChangeType.PRICE_CHANGE: 0.6,
|
|
ChangeType.NEW_DOCUMENT: 0.5,
|
|
ChangeType.CONTENT_MODIFIED: 0.3, # further scaled by diff_ratio - see compute_significance
|
|
ChangeType.REMOVED_DOCUMENT: 0.3,
|
|
}
|
|
|
|
# (score_threshold, severity) - first match wins, checked highest first.
|
|
# Calibrated against compute_significance/compute_confidence's actual output
|
|
# range (a single-source signal is already discounted ~2x by the
|
|
# corroboration multiplier) so CRITICAL_MIN_CONFIDENCE below is reachable:
|
|
# with significance capped at 1.0, a score of 0.6 needs confidence >= 0.6,
|
|
# which leaves room for the 0.7 confidence floor to actually bite and
|
|
# downgrade a subset of would-be-CRITICAL cases to HIGH.
|
|
SEVERITY_THRESHOLDS: list[tuple[float, SeverityLevel]] = [
|
|
(0.6, SeverityLevel.CRITICAL),
|
|
(0.4, SeverityLevel.HIGH),
|
|
(0.2, SeverityLevel.MEDIUM),
|
|
(0.0, SeverityLevel.LOW),
|
|
]
|
|
|
|
CRITICAL_MIN_CONFIDENCE = 0.7
|
|
|
|
|
|
def compute_significance(
|
|
*,
|
|
change_type: ChangeType,
|
|
source_trust_score: float,
|
|
independent_source_count: int = 1,
|
|
focus_match: bool = False,
|
|
is_repeat: bool = False,
|
|
diff_ratio: float | None = None,
|
|
) -> float:
|
|
base = BASE_WEIGHTS[change_type]
|
|
if change_type is ChangeType.CONTENT_MODIFIED and diff_ratio is not None:
|
|
# A one-line wording tweak and a full page rewrite are both
|
|
# "content_modified" but shouldn't score the same.
|
|
base = base + diff_ratio * 0.5
|
|
|
|
trust_multiplier = _clamp(source_trust_score, 0.3, 1.0)
|
|
corroboration_multiplier = min(1.0, independent_source_count / 2)
|
|
focus_multiplier = 1.3 if focus_match else 1.0
|
|
recency_multiplier = 0.5 if is_repeat else 1.0
|
|
|
|
significance = (
|
|
base * trust_multiplier * corroboration_multiplier * focus_multiplier * recency_multiplier
|
|
)
|
|
return round(_clamp(significance, 0.0, 1.0), 4)
|
|
|
|
|
|
def compute_confidence(
|
|
*,
|
|
extraction_confidence: float,
|
|
source_trust_score: float,
|
|
independent_source_count: int = 1,
|
|
) -> float:
|
|
corroboration_bonus = 0.15 if independent_source_count >= 2 else 0.0
|
|
confidence = 0.5 * extraction_confidence + 0.4 * source_trust_score + 0.1 + corroboration_bonus
|
|
return round(_clamp(confidence, 0.0, 1.0), 4)
|
|
|
|
|
|
def classify_severity(significance: float, confidence: float) -> SeverityLevel:
|
|
score = significance * confidence
|
|
for threshold, severity in SEVERITY_THRESHOLDS:
|
|
if score >= threshold:
|
|
if severity is SeverityLevel.CRITICAL and confidence < CRITICAL_MIN_CONFIDENCE:
|
|
return SeverityLevel.HIGH
|
|
return severity
|
|
return SeverityLevel.LOW # pragma: no cover - thresholds bottom out at 0.0
|
|
|
|
|
|
def _clamp(value: float, lo: float, hi: float) -> float:
|
|
return max(lo, min(hi, value))
|