Anushka-stack-queues/FinBench

1"""2Task 2 (Medium): Risk Assessment & Recommendations3====================================================4Agent analyzes a client's financial situation, identifies risks,5assigns a risk score, and provides actionable recommendations.6"""7 8from __future__ import annotations9 10 11 12from typing import Dict, List, Set13from ..models import ClientProfile, Reward, _SCORE_EPSILON as STRICT_SCORE_EPSILON14 15 16 17# ── Risk factors that should be identified based on client profile ────────────18def expected_risk_flags(client: ClientProfile) -> Dict[str, str]:19    """Return a dict of {flag_key: description} that SHOULD be identified."""20    flags: Dict[str, str] = {}21 22    # Debt risk23    if client.debt_to_income_ratio > 0.36:24        flags["high_debt"] = "debt-to-income ratio exceeds 36%"25 26    # Emergency fund27    if not client.has_emergency_fund:28        flags["no_emergency_fund"] = "no emergency fund"29 30    # Insurance gap31    if not client.has_insurance and client.dependents > 0:32        flags["insurance_gap"] = "has dependents but no insurance"33 34    # Under-saving (simplified: saves < 15% of income)35    monthly_income = client.annual_income / 1236    savings = monthly_income - client.monthly_expenses37    savings_rate = savings / monthly_income if monthly_income > 0 else 038    if savings_rate < 0.15:39        flags["low_savings_rate"] = f"savings rate ~{savings_rate*100:.0f}% below 15% target"40 41    # Retirement horizon risk42    years_to_65 = max(0, 65 - client.age)43    if years_to_65 < 10 and client.risk_tolerance == "aggressive":44        flags["near_retirement_aggressive"] = "near retirement with aggressive allocation"45 46    # Concentration risk47    if client.existing_portfolio:48        for asset, pct in client.existing_portfolio.items():49            if pct > 60:50                flags[f"concentration_risk_{asset}"] = f"over-concentrated in {asset} ({pct}%)"51 52    # High income + no tax strategy53    if client.annual_income > 150_000 and client.tax_bracket >= 0.32:54        flags["tax_optimization_needed"] = "high income with no apparent tax strategy"55 56    return flags57 58 59def expected_risk_score_range(client: ClientProfile) -> tuple[float, float]:60    """Return acceptable normalized (min, max) risk score based on client profile."""61    flags = expected_risk_flags(client)62    num_flags = len(flags)63 64    # Base from risk tolerance65    base = {"conservative": 0.3, "moderate": 0.5, "aggressive": 0.6}[client.risk_tolerance]66    adjustment = num_flags * 0.0867 68    ideal = min(1.0 - STRICT_SCORE_EPSILON, base + adjustment)69    return max(STRICT_SCORE_EPSILON, ideal - 0.2), min(1.0 - STRICT_SCORE_EPSILON, ideal + 0.2)70 71 72# FIX #7: updated signature to accept FinbenchAction directly instead of73#         RiskAssessmentAction (which doesn't exist). Reads the same fields.74def grade(action, client: ClientProfile) -> Reward:75    """76    Score a risk assessment 0.0–1.0.77    Partial credit for:78      - Identifying expected risk flags        (0.40)79      - Accurate risk score                    (0.20)80      - Quality & specificity of recommendations (0.25)81      - Priority recommendation clarity        (0.15)82    """83    components: Dict[str, float] = {}84    penalties: Dict[str, float] = {}85 86    expected_flags = expected_risk_flags(client)87    expected_keywords = {88        "high_debt": ["debt", "dti", "debt-to-income"],89        "no_emergency_fund": ["emergency", "emergency fund", "liquid"],90        "insurance_gap": ["insurance", "life insurance", "coverage"],91        "low_savings_rate": ["saving", "savings rate", "save more"],92        "near_retirement_aggressive": ["retirement", "conservative", "near retirement"],93        "concentration_risk": ["concentration", "concentrated", "diversif"],94        "tax_optimization_needed": ["tax", "tax-advantaged", "401k", "ira"],95    }96 97    # 1. Risk flag identification (0.40)98    identified_text = " ".join(action.identified_risks).lower()99    flags_caught = 0100    for flag_key in expected_flags:101        keywords = (102            expected_keywords["concentration_risk"]103            if flag_key.startswith("concentration_risk")104            else expected_keywords[flag_key]105        )106        if any(kw in identified_text for kw in keywords):107            flags_caught += 1108 109    if expected_flags:110        flag_score = (flags_caught / len(expected_flags)) * 0.40111    else:112        flag_score = 0.40  # no flags = low risk = full marks if agent finds nothing major113 114    components["risk_flags_identified"] = round(flag_score, 4)115 116    # Penalize false positives (more than 3 extra risks not expected)117    if len(action.identified_risks) > len(expected_flags) + 3:118        penalties["false_positive_risks"] = 0.05119 120    # 2. Risk score accuracy (0.20)121    score_min, score_max = expected_risk_score_range(client)122    if score_min <= action.risk_score <= score_max:123        components["risk_score_accuracy"] = 0.20124    else:125        error = min(126            abs(action.risk_score - score_min),127            abs(action.risk_score - score_max)128        )129        components["risk_score_accuracy"] = max(0, 0.20 - error * 0.03)130 131    # 3. Recommendation quality (0.25)132    rec_text = " ".join(action.recommendations).lower()133    useful_terms = [134        "emergency fund", "insurance", "diversif", "rebalance", "tax",135        "401k", "ira", "debt", "save", "budget", "inflation", "income"136    ]137    hits = sum(1 for term in useful_terms if term in rec_text)138    rec_score = min(0.25, hits * 0.04)139    # Bonus for having 3+ specific recommendations140    if len(action.recommendations) >= 3:141        rec_score = min(0.25, rec_score + 0.05)142    components["recommendation_quality"] = round(rec_score, 4)143 144    # 4. Priority recommendation clarity (0.15)145    priority = action.priority_recommendation.lower()146    clarity_score = 0.0147    if len(priority) >= 20:148        clarity_score += 0.08149    # Should match the most critical flag150    critical_flags_order = [151        "insurance_gap", "no_emergency_fund", "high_debt",152        "concentration_risk", "low_savings_rate"153    ]154    for flag in critical_flags_order:155        if flag in expected_flags:156            keywords = expected_keywords.get(flag, [])157            if any(kw in priority for kw in keywords):158                clarity_score += 0.07159            break160    components["priority_clarity"] = round(min(0.15, clarity_score), 4)161 162    total = sum(components.values()) - sum(penalties.values())163    total = max(STRICT_SCORE_EPSILON, min(1.0 - STRICT_SCORE_EPSILON, total))164 165    return Reward(166        total=round(total, 3),167        components=components,168        penalties=penalties,169        explanation=(170            f"Expected flags: {list(expected_flags.keys())} | "171            f"Caught: {flags_caught}/{len(expected_flags)} | "172            f"Risk score range: {score_min:.2f}–{score_max:.2f} | "173            f"Agent score: {action.risk_score}"174        ),175    )