""" User Prediction Service. This module provides business logic for tracking user predictions, calculating ROI and accuracy rates. """ from datetime import datetime from typing import List, Tuple, Optional from sqlalchemy.orm import Session from app.models import User, UserPrediction, Prediction, Match class UserPredictionService: """ Service for managing user predictions and statistics. This service handles: - Tracking which predictions users have viewed - Updating prediction results when matches complete - Calculating user accuracy rates - Calculating user ROI (Return on Investment) """ def __init__(self, db: Session): """Initialize service with database session.""" self.db = db def record_prediction_view(self, user_id: int, prediction_id: int) -> UserPrediction: """ Record that a user viewed a prediction. Args: user_id: ID of the user prediction_id: ID of the prediction viewed Returns: Created or existing UserPrediction record Raises: ValueError: If user or prediction doesn't exist """ # Check if user exists user = self.db.query(User).filter(User.id == user_id).first() if not user: raise ValueError(f"User with id {user_id} not found") # Check if prediction exists prediction = self.db.query(Prediction).filter(Prediction.id == prediction_id).first() if not prediction: raise ValueError(f"Prediction with id {prediction_id} not found") # Check if already viewed (unique constraint) existing = self.db.query(UserPrediction).filter( UserPrediction.user_id == user_id, UserPrediction.prediction_id == prediction_id ).first() if existing: return existing # Create new record user_prediction = UserPrediction( user_id=user_id, prediction_id=prediction_id, viewed_at=datetime.utcnow(), was_correct=None # Will be set when match completes ) self.db.add(user_prediction) self.db.commit() self.db.refresh(user_prediction) return user_prediction def get_user_prediction_history( self, user_id: int, limit: int = 50, offset: int = 0 ) -> Tuple[List[dict], int]: """ Get a user's prediction viewing history. Args: user_id: ID of the user limit: Maximum number of records to return offset: Number of records to skip Returns: Tuple of (list of user predictions with details, total count) """ query = self.db.query(UserPrediction).filter( UserPrediction.user_id == user_id ).order_by(UserPrediction.viewed_at.desc()) total = query.count() user_predictions = query.limit(limit).offset(offset).all() # Build response with full details result = [] for up in user_predictions: pred = up.prediction match = pred.match if pred else None result.append({ 'id': up.id, 'user_id': up.user_id, 'prediction_id': up.prediction_id, 'viewed_at': up.viewed_at.isoformat() if up.viewed_at else None, 'was_correct': up.was_correct, 'prediction': { 'id': pred.id if pred else None, 'match_id': pred.match_id if pred else None, 'energy_score': pred.energy_score if pred else None, 'confidence': pred.confidence if pred else None, 'predicted_winner': pred.predicted_winner if pred else None, 'created_at': pred.created_at.isoformat() if pred and pred.created_at else None } if pred else None, 'match': { 'id': match.id if match else None, 'home_team': match.home_team if match else None, 'away_team': match.away_team if match else None, 'date': match.date.isoformat() if match and match.date else None, 'league': match.league if match else None, 'status': match.status if match else None, 'actual_winner': match.actual_winner if match else None } if match else None }) return result, total def update_prediction_result(self, prediction_id: int, actual_winner: Optional[str]) -> None: """ Update all user predictions for a given prediction with match result. This is called when a match completes to mark which predictions were correct. Args: prediction_id: ID of the prediction actual_winner: Actual winner of the match ("home", "away", "draw", or None) """ # Get prediction prediction = self.db.query(Prediction).filter(Prediction.id == prediction_id).first() if not prediction: return # Update match result match = self.db.query(Match).filter(Match.id == prediction.match_id).first() if match: match.actual_winner = actual_winner # Determine if prediction was correct was_correct = False if actual_winner: # Normalize winner comparison predicted = prediction.predicted_winner.lower() actual = actual_winner.lower() if predicted == actual: was_correct = True elif (predicted == 'home' and actual == 'home_team') or \ (predicted == 'away' and actual == 'away_team'): was_correct = True # Update all user predictions for this prediction user_predictions = self.db.query(UserPrediction).filter( UserPrediction.prediction_id == prediction_id ).all() for up in user_predictions: up.was_correct = was_correct self.db.commit() def get_user_stats(self, user_id: int) -> dict: """ Calculate user statistics including accuracy and ROI. Args: user_id: ID of the user Returns: Dictionary with statistics: - total_predictions_viewed: Total predictions viewed - correct_predictions: Number of correct predictions - incorrect_predictions: Number of incorrect predictions - accuracy_rate: Accuracy as percentage - roi: Return on Investment in EUR """ user_predictions = self.db.query(UserPrediction).filter( UserPrediction.user_id == user_id ).all() total = len(user_predictions) correct = sum(1 for up in user_predictions if up.was_correct is True) incorrect = sum(1 for up in user_predictions if up.was_correct is False) # Calculate accuracy accuracy_rate = 0.0 if correct + incorrect > 0: accuracy_rate = (correct / (correct + incorrect)) * 100 # Calculate ROI # Assumptions: Each correct prediction = +100€, Each incorrect = -50€ # This is a simplified model - can be adjusted based on actual betting rules roi = (correct * 100) - (incorrect * 50) return { 'total_predictions_viewed': total, 'correct_predictions': correct, 'incorrect_predictions': incorrect, 'accuracy_rate': round(accuracy_rate, 1), 'roi': roi }