""" risk_engine.py ---------------- Combines: 1. Validated questionnaire scores (PHQ-9, GAD-7, PSS-10) [primary, weighted 75%] 2. Aggregated local NLP signal from conversation text [secondary, weighted 25%] 3. An optional scikit-learn ML classifier signal (see ml_classifier.py) shown alongside as a secondary, clearly-labeled research signal. into a final explainable screening result: per-domain risk scores, an overall wellness score, a confidence score, and a plain-language explanation listing exactly which inputs drove the result (explainable AI requirement). The rule-based combination (rather than a black-box score) is a deliberate design choice for this domain: every number a participant or researcher sees must be traceable to a specific, inspectable cause. """ from dataclasses import dataclass, field from typing import Dict, List, Optional from .questionnaires import QuestionnaireResult, Severity from .ml_classifier import MLRiskClassifier, FEATURE_NAMES QUESTIONNAIRE_WEIGHT = 0.75 NLP_WEIGHT = 0.25 class RiskLevel: LOW = "Low" MODERATE = "Moderate" HIGH = "High" @dataclass class DomainScore: domain: str questionnaire_component: float nlp_component: float combined_score: float # 0-100 risk_level: str explanation: List[str] = field(default_factory=list) @dataclass class ScreeningResult: depression: DomainScore anxiety: DomainScore stress: DomainScore overall_wellness_score: float # 0-100, higher = better wellbeing confidence_score: float # 0-1 ml_signal: Optional[Dict] = None explanation_summary: List[str] = field(default_factory=list) def to_dict(self) -> Dict: def domain_dict(d: DomainScore) -> Dict: return { "domain": d.domain, "combined_score": d.combined_score, "risk_level": d.risk_level, "explanation": d.explanation, } return { "depression": domain_dict(self.depression), "anxiety": domain_dict(self.anxiety), "stress": domain_dict(self.stress), "overall_wellness_score": self.overall_wellness_score, "confidence_score": self.confidence_score, "ml_signal": self.ml_signal, "explanation_summary": self.explanation_summary, } def _nlp_adjustment_for_domain( marker_count: int, sentiment: float, absolutist: float, weight_markers: float = 8.0 ) -> float: """Turn raw NLP signals into a 0-100 'distress' style component so it can be combined on the same scale as the normalized questionnaire score.""" marker_component = min(60.0, marker_count * weight_markers) sentiment_component = max(0.0, -sentiment) * 30.0 # only negative sentiment adds risk absolutist_component = min(10.0, absolutist * 5.0) return round(min(100.0, marker_component + sentiment_component + absolutist_component), 1) def _level_from_score(score: float) -> str: if score < 34: return RiskLevel.LOW elif score < 67: return RiskLevel.MODERATE return RiskLevel.HIGH class RiskEngine: def __init__(self, use_ml_signal: bool = True): self._ml_classifier = MLRiskClassifier() if use_ml_signal else None def assess( self, phq9: QuestionnaireResult, gad7: QuestionnaireResult, pss10: QuestionnaireResult, nlp_summary: Dict, items_answered: int, items_total: int, ) -> ScreeningResult: # ---- Depression ---- dep_nlp = _nlp_adjustment_for_domain( nlp_summary.get("total_depression_markers", 0), nlp_summary.get("avg_sentiment", 0.0), nlp_summary.get("avg_absolutist_count", 0.0), ) dep_combined = round( QUESTIONNAIRE_WEIGHT * phq9.normalized_score + NLP_WEIGHT * dep_nlp, 1 ) dep_explanation = [ f"PHQ-9 raw score {phq9.raw_score}/27 ({phq9.severity.value}) " f"contributes {round(QUESTIONNAIRE_WEIGHT * phq9.normalized_score, 1)} pts.", f"Conversation language contributed {round(NLP_WEIGHT * dep_nlp, 1)} pts " f"based on {nlp_summary.get('total_depression_markers', 0)} depression-related " f"phrase(s) and average sentiment {nlp_summary.get('avg_sentiment', 0.0)}.", ] if phq9.flagged_items: dep_explanation.append( "PHQ-9 item 9 (thoughts of self-harm) was endorsed above zero — " "this always triggers the safety protocol regardless of total score." ) depression = DomainScore( domain="Depression", questionnaire_component=phq9.normalized_score, nlp_component=dep_nlp, combined_score=dep_combined, risk_level=_level_from_score(dep_combined), explanation=dep_explanation, ) # ---- Anxiety ---- anx_nlp = _nlp_adjustment_for_domain( nlp_summary.get("total_anxiety_markers", 0), nlp_summary.get("avg_sentiment", 0.0), nlp_summary.get("avg_absolutist_count", 0.0), ) anx_combined = round( QUESTIONNAIRE_WEIGHT * gad7.normalized_score + NLP_WEIGHT * anx_nlp, 1 ) anx_explanation = [ f"GAD-7 raw score {gad7.raw_score}/21 ({gad7.severity.value}) " f"contributes {round(QUESTIONNAIRE_WEIGHT * gad7.normalized_score, 1)} pts.", f"Conversation language contributed {round(NLP_WEIGHT * anx_nlp, 1)} pts " f"based on {nlp_summary.get('total_anxiety_markers', 0)} anxiety-related phrase(s).", ] anxiety = DomainScore( domain="Anxiety", questionnaire_component=gad7.normalized_score, nlp_component=anx_nlp, combined_score=anx_combined, risk_level=_level_from_score(anx_combined), explanation=anx_explanation, ) # ---- Stress ---- stress_nlp = _nlp_adjustment_for_domain( nlp_summary.get("total_stress_markers", 0), nlp_summary.get("avg_sentiment", 0.0), nlp_summary.get("avg_absolutist_count", 0.0), ) stress_combined = round( QUESTIONNAIRE_WEIGHT * pss10.normalized_score + NLP_WEIGHT * stress_nlp, 1 ) stress_explanation = [ f"PSS-10 raw score {pss10.raw_score}/40 (perceived stress: {pss10.severity.value}) " f"contributes {round(QUESTIONNAIRE_WEIGHT * pss10.normalized_score, 1)} pts.", f"Conversation language contributed {round(NLP_WEIGHT * stress_nlp, 1)} pts " f"based on {nlp_summary.get('total_stress_markers', 0)} stress-related phrase(s).", ] stress = DomainScore( domain="Stress", questionnaire_component=pss10.normalized_score, nlp_component=stress_nlp, combined_score=stress_combined, risk_level=_level_from_score(stress_combined), explanation=stress_explanation, ) overall_wellness = round( 100 - (dep_combined + anx_combined + stress_combined) / 3, 1 ) # Confidence: more questionnaire completeness + more conversational # text => higher confidence. Capped at 0.95 since this is a screening # tool, never fully certain. completeness_ratio = items_answered / items_total if items_total else 0.0 text_richness = min(1.0, nlp_summary.get("message_count", 0) / 10.0) confidence = round(min(0.95, 0.6 * completeness_ratio + 0.4 * text_richness), 2) ml_signal = None if self._ml_classifier is not None: feature_vector = { "phq9_normalized": phq9.normalized_score, "gad7_normalized": gad7.normalized_score, "pss10_normalized": pss10.normalized_score, "avg_sentiment": nlp_summary.get("avg_sentiment", 0.0), "total_depression_markers": nlp_summary.get("total_depression_markers", 0), "total_anxiety_markers": nlp_summary.get("total_anxiety_markers", 0), "total_stress_markers": nlp_summary.get("total_stress_markers", 0), "avg_first_person_ratio": nlp_summary.get("avg_first_person_ratio", 0.0), "avg_absolutist_count": nlp_summary.get("avg_absolutist_count", 0.0), } ml_signal = self._ml_classifier.predict(feature_vector) summary = [ f"Depression: {depression.risk_level} ({dep_combined}/100).", f"Anxiety: {anxiety.risk_level} ({anx_combined}/100).", f"Stress: {stress.risk_level} ({stress_combined}/100).", f"Overall wellness score: {overall_wellness}/100 " f"(higher is better; based 75% on validated questionnaires, " f"25% on conversational language analysis).", f"Confidence in this estimate: {confidence} " f"(based on {items_answered}/{items_total} questionnaire items " f"answered and {nlp_summary.get('message_count', 0)} messages analyzed).", ] if ml_signal: summary.append( f"Secondary ML signal (demonstration model, not clinically " f"validated) predicts: {ml_signal['predicted_risk']}." ) return ScreeningResult( depression=depression, anxiety=anxiety, stress=stress, overall_wellness_score=overall_wellness, confidence_score=confidence, ml_signal=ml_signal, explanation_summary=summary, )