""" Budget service for variance analysis and comparisons. """ from datetime import date, datetime from typing import List, Dict, Optional from sqlalchemy.orm import Session from sqlalchemy import func, and_ from app.models.budget import Budget, BudgetLine, BudgetLineType, VarianceAnalysis, BudgetPeriod from app.models.folio import FolioItem, Folio from app.models.reservation import Reservation, ReservationStatus from app.models.room import Room from app.models.stats import DailyStats from app.models.expense import Expense, ExpenseStatus class BudgetService: """Service for budget calculations and variance analysis.""" def __init__(self, db: Session): self.db = db def calculate_actual_revenue_by_department( self, establishment_id: int, start_date: date, end_date: date ) -> Dict[str, float]: """ Calculate actual revenue by department for a period. Returns: Dict[department_code, total_amount] """ # Query folio items for the period items = self.db.query( FolioItem.department_code, func.sum(FolioItem.total_ttc).label('total') ).join( FolioItem.folio ).join( Folio.reservation ).join( Reservation.room ).filter( Room.establishment_id == establishment_id, FolioItem.created_at >= start_date, FolioItem.created_at <= end_date, FolioItem.is_voided == False ).group_by(FolioItem.department_code).all() return {item.department_code: float(item.total) for item in items} def calculate_actual_expenses_by_department( self, establishment_id: int, start_date: date, end_date: date ) -> Dict[str, float]: """ Calculate actual expenses by department for a period. Only includes APPROVED and PAID expenses. Returns: Dict[department_code, total_amount] """ expenses = self.db.query( Expense.department_code, func.sum(Expense.amount_ttc).label('total') ).filter( Expense.establishment_id == establishment_id, Expense.expense_date >= start_date, Expense.expense_date <= end_date, Expense.status.in_([ExpenseStatus.APPROVED, ExpenseStatus.PAID]) ).group_by(Expense.department_code).all() return {exp.department_code: float(exp.total) for exp in expenses} def calculate_budget_vs_actual( self, budget_id: int ) -> Dict[str, Dict]: """ Calculate variance between budget and actual for each department. Returns detailed variance analysis. """ budget = self.db.query(Budget).filter(Budget.id == budget_id).first() if not budget: return {} # Get actual revenue amounts from FolioItems actual_revenue_by_dept = self.calculate_actual_revenue_by_department( budget.establishment_id, budget.start_date, budget.end_date ) # Get actual expense amounts from Expenses table actual_expenses_by_dept = self.calculate_actual_expenses_by_department( budget.establishment_id, budget.start_date, budget.end_date ) # Calculate variances — composite key keeps revenue and expense lines separate per dept variances = {} for line in budget.lines: dept = line.department_code budget_amount = line.amount # Get actual amounts based on line type if line.line_type == BudgetLineType.REVENUE: actual_amount = actual_revenue_by_dept.get(dept, 0.0) else: actual_amount = actual_expenses_by_dept.get(dept, 0.0) variance = actual_amount - budget_amount variance_pct = (variance / budget_amount * 100) if budget_amount != 0 else 0 # Favorable if revenue is higher than budget or expense is lower is_favorable = ( (line.line_type == BudgetLineType.REVENUE and variance > 0) or (line.line_type == BudgetLineType.EXPENSE and variance < 0) ) # Use composite key so revenue and expense lines for same dept don't overwrite each other key = f"{dept}__{line.line_type.value}" variances[key] = { 'department': dept, 'budget': budget_amount, 'actual': actual_amount, 'variance': variance, 'variance_percent': variance_pct, 'is_favorable': is_favorable, 'line_type': line.line_type } return variances def calculate_year_over_year( self, establishment_id: int, current_year: int, current_start: date, current_end: date ) -> Dict[str, Dict]: """ Calculate year-over-year comparison. Compares current period with same period last year. """ # Calculate prior year dates prior_year = current_year - 1 days_diff = (current_end - current_start).days prior_start = date(prior_year, current_start.month, current_start.day) prior_end = date(prior_year, current_end.month, current_end.day) # Get current year data current_by_dept = self.calculate_actual_revenue_by_department( establishment_id, current_start, current_end ) # Get prior year data prior_by_dept = self.calculate_actual_revenue_by_department( establishment_id, prior_start, prior_end ) # Calculate YoY variances yoy_comparison = {} all_depts = set(current_by_dept.keys()) | set(prior_by_dept.keys()) for dept in all_depts: current = current_by_dept.get(dept, 0.0) prior = prior_by_dept.get(dept, 0.0) variance = current - prior variance_pct = (variance / prior * 100) if prior != 0 else 0 yoy_comparison[dept] = { 'department': dept, 'current_year': current_year, 'prior_year': prior_year, 'current_amount': current, 'prior_amount': prior, 'variance': variance, 'variance_percent': variance_pct, 'is_favorable': variance > 0 } return yoy_comparison def generate_variance_analysis( self, budget_id: int ) -> List[VarianceAnalysis]: """ Generate and save variance analysis records. Creates VarianceAnalysis records for each department. """ budget = self.db.query(Budget).filter(Budget.id == budget_id).first() if not budget: return [] # Calculate variances variances = self.calculate_budget_vs_actual(budget_id) # Calculate YoY if needed yoy_data = self.calculate_year_over_year( budget.establishment_id, budget.year, budget.start_date, budget.end_date ) # Delete existing variance analyses for this budget self.db.query(VarianceAnalysis).filter( VarianceAnalysis.budget_id == budget_id ).delete() # Create new variance analyses analyses = [] for _key, var_data in variances.items(): dept_code = var_data['department'] yoy = yoy_data.get(dept_code, {}) analysis = VarianceAnalysis( establishment_id=budget.establishment_id, budget_id=budget_id, analysis_date=date.today(), period_type=budget.period_type, department_code=dept_code, budget_amount=var_data['budget'], actual_amount=var_data['actual'], variance_amount=var_data['variance'], variance_percent=var_data['variance_percent'], prior_year_amount=yoy.get('prior_amount'), yoy_variance=yoy.get('variance'), yoy_variance_percent=yoy.get('variance_percent'), is_favorable=var_data['is_favorable'] ) self.db.add(analysis) analyses.append(analysis) self.db.commit() return analyses def get_budget_summary( self, establishment_id: int, year: Optional[int] = None ) -> List[Dict]: """ Get summary of all budgets for an establishment. Args: establishment_id: ID of establishment year: Optional year filter Returns: List of budget summaries with key metrics """ query = self.db.query(Budget).filter( Budget.establishment_id == establishment_id ) if year: query = query.filter(Budget.year == year) budgets = query.order_by(Budget.year.desc(), Budget.start_date).all() summaries = [] for budget in budgets: summaries.append({ 'id': budget.id, 'name': budget.name, 'year': budget.year, 'period_type': budget.period_type, 'status': budget.status, 'start_date': budget.start_date, 'end_date': budget.end_date, 'total_revenue': budget.total_revenue, 'total_expenses': budget.total_expenses, 'net_income': budget.net_income }) return summaries @staticmethod def validate_budget_dates( period_type: BudgetPeriod, year: int, month: Optional[int], quarter: Optional[int], start_date: date, end_date: date ) -> tuple[bool, Optional[str]]: """ Validate that budget dates match the period type. Returns: (is_valid, error_message) """ if start_date >= end_date: return False, "Start date must be before end date" if period_type == BudgetPeriod.MONTHLY: if not month: return False, "Month is required for monthly budget" if month < 1 or month > 12: return False, "Month must be between 1 and 12" elif period_type == BudgetPeriod.QUARTERLY: if not quarter: return False, "Quarter is required for quarterly budget" if quarter < 1 or quarter > 4: return False, "Quarter must be between 1 and 4" return True, None