"""Fusion Engine - Confidence-weighted Score Fusion""" from typing import Dict, Tuple, Optional from config import Config class FusionEngine: """Combines scores from all modules with confidence weighting""" def __init__(self): # Base weights (when no domain evidence) self.base_weights = { 'universal': Config.UNIVERSAL_WEIGHT, 'personality': Config.PERSONALITY_WEIGHT, 'text': Config.TEXT_WEIGHT } # Extended weights (when domain evidence exists) self.extended_weights = { 'universal': 0.30, # Reduced from base 'personality': 0.25, 'text': 0.25, 'domain': 0.20 # New domain component } def fuse_scores( self, universal_score: float, universal_confidence: float, personality_score: float, personality_confidence: float, text_score: float, text_confidence: float, domain_score: Optional[float] = None, domain_confidence: Optional[float] = None ) -> Tuple[float, Dict]: """ Fuse scores with confidence weighting Supports optional domain score for pluggable domain evidence Returns: (final_score, breakdown) """ # Determine which weights to use has_domain = domain_score is not None and domain_confidence is not None and domain_confidence > 0 weights = self.extended_weights if has_domain else self.base_weights # Calculate effective weights (weight * confidence) effective_weights = { 'universal': weights['universal'] * universal_confidence, 'personality': weights['personality'] * personality_confidence, 'text': weights['text'] * text_confidence } # Add domain if available if has_domain: effective_weights['domain'] = weights['domain'] * domain_confidence # Sum of effective weights (for normalization) total_effective_weight = sum(effective_weights.values()) # Prevent division by zero if total_effective_weight == 0: breakdown = { 'final_score': 0.0, 'component_scores': { 'universal': 0.0, 'personality': 0.0, 'text': 0.0 }, 'confidences': { 'universal': 0.0, 'personality': 0.0, 'text': 0.0 }, 'effective_weights': effective_weights, 'has_domain': False } if has_domain: breakdown['component_scores']['domain'] = 0.0 breakdown['confidences']['domain'] = 0.0 return 0.0, breakdown # Calculate fused score fused_score = ( effective_weights['universal'] * universal_score + effective_weights['personality'] * personality_score + effective_weights['text'] * text_score ) if has_domain: fused_score += effective_weights['domain'] * domain_score fused_score /= total_effective_weight # Prepare breakdown breakdown = { 'final_score': round(fused_score, 4), 'component_scores': { 'universal': round(universal_score, 4), 'personality': round(personality_score, 4), 'text': round(text_score, 4) }, 'confidences': { 'universal': round(universal_confidence, 4), 'personality': round(personality_confidence, 4), 'text': round(text_confidence, 4) }, 'effective_weights': { k: round(v / total_effective_weight, 4) for k, v in effective_weights.items() }, 'base_weights': weights, 'has_domain': has_domain } # Add domain info if present if has_domain: breakdown['component_scores']['domain'] = round(domain_score, 4) breakdown['confidences']['domain'] = round(domain_confidence, 4) return fused_score, breakdown def get_grade(self, final_score: float) -> str: """Convert score to letter grade""" if final_score >= 0.9: return 'A+' elif final_score >= 0.85: return 'A' elif final_score >= 0.8: return 'A-' elif final_score >= 0.75: return 'B+' elif final_score >= 0.7: return 'B' elif final_score >= 0.65: return 'B-' elif final_score >= 0.6: return 'C+' elif final_score >= 0.55: return 'C' elif final_score >= 0.5: return 'C-' else: return 'D' def get_percentile(self, final_score: float) -> int: """Estimate percentile (mock for MVP)""" # In production, this would query actual distribution return min(int(final_score * 100), 99)