""" Question Effectiveness Tracker for monitoring and calculating question effectiveness metrics """ import asyncio import statistics from datetime import datetime, timedelta, timezone from typing import Dict, List, Optional from sqlalchemy.orm import Session from sqlalchemy import func from models.question_metrics import QuestionMetrics from models.conversation_entry import ConversationEntry from modules.monitoring import get_monitor class QuestionEffectivenessTracker: """ Tracks question effectiveness metrics and provides scoring for question selection. This class records question usage asynchronously, calculates effectiveness scores based on variance, completion rate, and time to answer, and provides caching for frequently accessed metrics. """ def __init__(self, db_session: Session, cache: Optional[Dict] = None, config: Optional[Dict] = None): """ Initialize the Question Effectiveness Tracker. Args: db_session: SQLAlchemy database session cache: Optional cache dictionary for storing metrics (defaults to empty dict) config: Optional configuration dictionary with keys: - min_samples: Minimum number of samples before calculating effectiveness (default: 10) - cache_ttl: Cache time-to-live in seconds (default: 3600 = 1 hour) - variance_weight: Weight for variance in effectiveness formula (default: 0.6) - completion_weight: Weight for completion rate (default: 0.3) - time_weight: Weight for time to answer (default: 0.1) - time_normalizer: Normalizer for time to answer in seconds (default: 300) """ self.db_session = db_session self.cache = cache if cache is not None else {} self.monitor = get_monitor() # Configuration with defaults config = config or {} self.min_samples = config.get('min_samples', 10) self.cache_ttl = config.get('cache_ttl', 3600) # 1 hour in seconds self.variance_weight = config.get('variance_weight', 0.6) self.completion_weight = config.get('completion_weight', 0.3) self.time_weight = config.get('time_weight', 0.1) self.time_normalizer = config.get('time_normalizer', 300) # 5 minutes in seconds async def record_usage( self, question_id: int, score: float, time_to_answer: int, completed: bool ) -> None: """ Record question usage asynchronously and update metrics. This method updates the question metrics in the database by: - Incrementing usage count - Updating score statistics for variance calculation - Updating completion rate - Updating average time to answer Args: question_id: ID of the question used score: Score received for the answer (0.0 to 1.0) time_to_answer: Time taken to answer in seconds completed: Whether the question was completed (not skipped) """ try: # Run database update in thread pool to avoid blocking await asyncio.get_event_loop().run_in_executor( None, self._update_metrics_sync, question_id, score, time_to_answer, completed ) # Invalidate cache for this question cache_key = f"effectiveness_{question_id}" if cache_key in self.cache: del self.cache[cache_key] except Exception as e: # Log error but don't raise - we don't want to block interview flow print(f"Error recording usage for question {question_id}: {e}") def _update_metrics_sync( self, question_id: int, score: float, time_to_answer: int, completed: bool ) -> None: """ Synchronous method to update metrics in database. This is called by record_usage in a thread pool executor. """ try: # Get or create metrics record metrics = self.db_session.query(QuestionMetrics).filter( QuestionMetrics.question_id == question_id ).first() old_effectiveness = metrics.effectiveness_score if metrics else 0.0 if not metrics: metrics = QuestionMetrics(question_id=question_id) self.db_session.add(metrics) # Get all scores for this question to calculate variance scores = self.db_session.query(ConversationEntry.score).filter( ConversationEntry.question_id == question_id, ConversationEntry.score.isnot(None) ).all() scores = [s[0] for s in scores] + [score] # Include current score # Calculate variance if we have enough samples if len(scores) >= 2: metrics.score_variance = statistics.variance(scores) else: metrics.score_variance = 0.0 # Update usage count metrics.usage_count += 1 # Update completion rate total_attempts = self.db_session.query(func.count(ConversationEntry.id)).filter( ConversationEntry.question_id == question_id ).scalar() or 0 completed_attempts = self.db_session.query(func.count(ConversationEntry.id)).filter( ConversationEntry.question_id == question_id, ConversationEntry.answer_text.isnot(None), ConversationEntry.answer_text != '' ).scalar() or 0 if completed: completed_attempts += 1 total_attempts += 1 metrics.completion_rate = completed_attempts / total_attempts if total_attempts > 0 else 1.0 # Calculate effectiveness score new_effectiveness = self.calculate_effectiveness(question_id) metrics.effectiveness_score = new_effectiveness # Commit changes self.db_session.commit() # Log effectiveness update to monitoring self.monitor.log_effectiveness_updated( question_id=question_id, old_score=old_effectiveness, new_score=new_effectiveness, usage_count=metrics.usage_count ) except Exception as e: self.db_session.rollback() print(f"Error in _update_metrics_sync for question {question_id}: {e}") raise def calculate_effectiveness(self, question_id: int) -> float: """ Calculate effectiveness metric for a question. Formula: effectiveness = (score_variance * 0.6) + (completion_rate * 0.3) + (avg_time / 300 * 0.1) Returns 0.0 if insufficient samples (< min_samples). Args: question_id: ID of the question Returns: Effectiveness score (0.0 to ~1.0, though can exceed 1.0 in edge cases) """ try: # Get metrics from database metrics = self.db_session.query(QuestionMetrics).filter( QuestionMetrics.question_id == question_id ).first() # Return 0.0 if no metrics or insufficient samples if not metrics or metrics.usage_count < self.min_samples: return 0.0 # Get average time to answer from conversation entries avg_time_result = self.db_session.query( func.avg(ConversationEntry.time_to_answer) ).filter( ConversationEntry.question_id == question_id, ConversationEntry.time_to_answer.isnot(None) ).scalar() avg_time = avg_time_result if avg_time_result is not None else 0 # Handle edge cases variance = metrics.score_variance if metrics.score_variance is not None else 0.0 completion = metrics.completion_rate if metrics.completion_rate is not None else 1.0 # Normalize time (cap at normalizer value to prevent excessive weight) time_component = min(avg_time / self.time_normalizer, 1.0) if self.time_normalizer > 0 else 0.0 # Calculate effectiveness using weighted formula effectiveness = ( variance * self.variance_weight + completion * self.completion_weight + time_component * self.time_weight ) return effectiveness except Exception as e: print(f"Error calculating effectiveness for question {question_id}: {e}") return 0.0 def get_effectiveness_scores(self, question_ids: List[int]) -> Dict[int, float]: """ Get effectiveness scores for multiple questions with caching. Uses cache when available (TTL 1 hour), queries database otherwise. Args: question_ids: List of question IDs Returns: Dictionary mapping question_id to effectiveness score """ results = {} questions_to_query = [] current_time = datetime.now(timezone.utc) # Check cache first for question_id in question_ids: cache_key = f"effectiveness_{question_id}" if cache_key in self.cache: cached_data = self.cache[cache_key] cache_time = cached_data.get('timestamp') # Check if cache is still valid (within TTL) if cache_time and (current_time - cache_time).total_seconds() < self.cache_ttl: results[question_id] = cached_data['score'] # Log cache hit self.monitor.log_cache_hit(cache_key) else: # Cache expired, need to refresh self.monitor.log_cache_miss(cache_key) questions_to_query.append(question_id) else: # Not in cache self.monitor.log_cache_miss(cache_key) questions_to_query.append(question_id) # Query database for non-cached questions if questions_to_query: for question_id in questions_to_query: try: score = self.calculate_effectiveness(question_id) results[question_id] = score # Update cache cache_key = f"effectiveness_{question_id}" self.cache[cache_key] = { 'score': score, 'timestamp': current_time } except Exception as e: print(f"Error getting effectiveness for question {question_id}: {e}") results[question_id] = 0.0 return results