src/mcda_v4.py

import pandas as pd
import numpy as np
from typing import Dict, List, Tuple

class UtilityCalculator:
    """

    Multi-Criteria Decision Analysis (MCDA) utility calculator.

    

    This class implements a weighted utility model for ranking alternatives across

    multiple criteria. It normalizes scores to a 0-100 scale, applies user-defined

    weights, and calculates final utility scores for decision making.

    

    Core Components:

    - Data Storage: Raw product scores and configuration

    - Normalization: Min-max scaling with direction handling  

    - Utility Calculation: Weighted linear aggregation

    - Caching: Performance optimization for repeated calculations

    - I/O: Excel integration and results formatting

    

    Attributes:

        categories (List[str]): Names of evaluation criteria

        maximize (Dict[str, bool]): Direction of optimization per category

        weights (Dict[str, float]): Importance weights per category (sum=1.0)

        products (Dict[str, Dict[str, float]]): Raw scores {product: {category: score}}

        thresholds (Dict[str, float]): Minimum acceptable values per category

        objectives (Dict[str, float]): Target/ideal values per category

        use_penalties (bool): Enable/disable threshold-objective penalty system

        _cache_valid (bool): Cache validity flag for performance optimization

        _cached_normalized (Dict): Cached normalized scores

        _cached_utilities (Dict): Cached utility calculations

    """
    
    def __init__(self, categories: List[str], maximize: Dict[str, bool]):
        """

        Initialize the MCDA calculator with evaluation criteria.

        

        Sets up the decision framework by defining what criteria to evaluate and

        their optimization direction. Initializes equal weights for all categories.

        

        Args:

            categories: List of category names (e.g., ['price', 'quality', 'speed'])

            maximize: Dict indicating optimization direction per category

                    {category: True} for "higher is better"

                    {category: False} for "lower is better"

        

        Raises:

            ValueError: If categories and maximize keys don't match exactly

            

        Example:

            calc = UtilityCalculator(

                categories=['price', 'quality'], 

                maximize={'price': False, 'quality': True}

            )

        """
        self.categories = categories
        self.maximize = maximize
        
        # Validate configuration consistency
        if set(categories) != set(maximize.keys()):
            raise ValueError("Categories and maximize keys must match exactly")
        
        # Initialize equal weights (will sum to 1.0)
        n_categories = len(self.categories)
        self.weights = {cat: 1.0/n_categories for cat in self.categories}
        
        # Initialize aggregation method
        self.aggregation_method = 'weighted_sum'  # Default to weighted sum
        
        # Initialize threshold and objective values for penalty system
        self.thresholds = {cat: None for cat in self.categories}  # Minimum acceptable values
        self.objectives = {cat: None for cat in self.categories}  # Target/ideal values
        self.use_penalties = False  # Global penalty system toggle

        # Initialize data storage
        self.products = {}  # {product_name: {category: raw_score}}
        
        # Initialize cache management
        self._cache_valid = False
        self._cached_normalized = None  # Stores normalized scores
        self._cached_utilities = None   # Stores final utility values

    @classmethod
    def from_excel(cls, file_path: str, config_sheet: str = 'Config', data_sheet: str = 'Data'):
        """

        Factory method to create calculator from Excel configuration.

        

        This method provides a convenient way to set up the calculator using

        Excel files for configuration management. Expects two sheets:

        - Config sheet: category definitions and optimization directions

        - Data sheet: product names and their scores

        

        Args:

            file_path: Path to Excel file containing configuration and data

            config_sheet: Name of sheet with category configuration (default: 'Config')

            data_sheet: Name of sheet with product data (default: 'Data')

            

        Returns:

            UtilityCalculator: Configured instance with loaded data

            

        Raises:

            ValueError: If required columns are missing from either sheet

            

        Expected Excel Format:

            Config Sheet: columns ['category', 'maximize']

            Data Sheet: columns ['name'] + all categories from config

            

        Example:

            calc = UtilityCalculator.from_excel('decisions.xlsx')

        """
        # Load and validate configuration sheet
        config_df = pd.read_excel(file_path, sheet_name=config_sheet)
        
        required_cols = ['category', 'maximize']
        if not all(col in config_df.columns for col in required_cols):
            raise ValueError(f"Config sheet must have columns: {required_cols}")
        
        # Extract configuration parameters
        categories = config_df['category'].tolist()
        maximize = dict(zip(config_df['category'], config_df['maximize']))
        
        # Create calculator instance
        calc = cls(categories, maximize)
        
        # Load and validate data sheet
        data_df = pd.read_excel(file_path, sheet_name=data_sheet)
        
        required_data_cols = ['name'] + categories
        missing_cols = [col for col in required_data_cols if col not in data_df.columns]
        if missing_cols:
            raise ValueError(f"Data sheet missing columns: {missing_cols}")
        
        # Load products into calculator
        products = data_df.to_dict('records')
        calc.add_products_batch(products)
        
        return calc

    def add_products_batch(self, products_data: List[Dict]):
        """

        Add multiple products to the calculator in a single operation.

        

        This method provides efficient bulk loading of product data. Each product

        dictionary should contain a 'name' key and scores for all categories.

        Invalidates cache to ensure fresh calculations.

        

        Args:

            products_data: List of dictionaries, each containing:

                          {'name': str, category1: float, category2: float, ...}

                          

        Example:

            products = [

                {'name': 'Product A', 'price': 100, 'quality': 8.5},

                {'name': 'Product B', 'price': 150, 'quality': 9.2}

            ]

            calc.add_products_batch(products)

        """
        for product in products_data:
            # Create copy to avoid mutating input data
            product_copy = product.copy()
            name = product_copy.pop('name')
            self.add_product(name, product_copy)

    def add_product(self, name: str, scores: Dict[str, float]):
        """

        Add a single product with its category scores.

        

        This is the core method for adding product data. Validates that scores

        are provided for all required categories and stores the data for analysis.

        Invalidates cache to ensure calculations reflect new data.

        

        Args:

            name: Unique identifier for the product

            scores: Dictionary mapping categories to numeric scores

                   {category: score} for all categories in self.categories

                   

        Raises:

            ValueError: If scores don't include all required categories

            

        Example:

            calc.add_product('Laptop X', {'price': 999, 'performance': 85, 'battery': 8})

        """
        # Validate completeness of scores
        if not all(cat in scores for cat in self.categories):
            raise ValueError(f"Must provide scores for: {self.categories}")
        
        # Store product data (copy to prevent external modification)
        self.products[name] = scores.copy()
        
        # Invalidate cache since data has changed
        self._cache_valid = False

    def set_weights(self, weights: Dict[str, float]):
        """

        Update the importance weights for categories.

        

        Weights represent the relative importance of each category in the final

        decision. They must sum to 1.0 to maintain the utility scale. Only

        invalidates utility cache since normalization is weight-independent.

        

        Args:

            weights: Dictionary mapping categories to weight values

                    Must include all categories and sum to 1.0

                    

        Raises:

            ValueError: If weights don't sum to 1.0 (within floating point tolerance)

            

        Example:

            calc.set_weights({'price': 0.4, 'quality': 0.4, 'support': 0.2})

        """
        # Validate weight constraints
        if not np.isclose(sum(weights.values()), 1.0):
            raise ValueError("Weights must sum to 1.0")
        
        # Update weights
        self.weights.update(weights)
        
        # Invalidate cache since weights affect utility calculations
        self._cache_valid = False



    def normalize_scores(self) -> Dict[str, Dict[str, float]]:
        """

        Normalize all product scores to a 0-100 scale with direction handling.

        

        This method implements normalization to make scores comparable across 

        different categories and scales. Routes to penalty system if enabled,

        otherwise uses standard min-max normalization.

        

        Uses caching to avoid recomputation when data hasn't changed.

        

        Returns:

            Dict[str, Dict[str, float]]: Nested dictionary structure

            {product_name: {category: normalized_score}}

            where normalized_score is in range [0, 100]

            

        Raises:

            ValueError: If no products have been added to analyze

        """
        # Return cached results if available and valid
        if self._cache_valid and self._cached_normalized:
            return self._cached_normalized
        
        # Route to penalty system if enabled
        if self.use_penalties:
            normalized = self.normalize_scores_with_penalties()
        else:
            # Use standard min-max normalization
            normalized = self._standard_normalize_scores()
        
        # Cache results and mark as valid
        self._cached_normalized = normalized
        self._cache_valid = True
        return normalized

    def _standard_normalize_scores(self) -> Dict[str, Dict[str, float]]:
        """

        Standard min-max normalization without penalty system.

        

        This is the original normalization logic extracted into a separate

        method for clarity and maintainability.

        

        Returns:

            Dict[str, Dict[str, float]]: Standard normalized scores

        """
        # Validate that we have data to normalize
        if not self.products:
            raise ValueError("No products to analyze")
        
        normalized = {}
        
        # Process each category independently
        for category in self.categories:
            # Extract all values for this category to find range
            values = [self.products[p][category] for p in self.products]
            min_val, max_val = min(values), max(values)
            range_val = max_val - min_val
            
            # Normalize each product's score for this category
            for product in self.products:
                if product not in normalized:
                    normalized[product] = {}
                
                raw_score = self.products[product][category]
                
                # Handle edge case: no variation in scores
                if range_val == 0:
                    normalized_score = 50  # Neutral score
                # Apply direction-aware normalization
                elif self.maximize[category]:
                    # Higher raw scores get higher normalized scores
                    normalized_score = ((raw_score - min_val) / range_val) * 100
                else:
                    # Lower raw scores get higher normalized scores
                    normalized_score = ((max_val - raw_score) / range_val) * 100
                
                normalized[product][category] = normalized_score
        
        return normalized



    def normalize_scores_with_penalties(self) -> Dict[str, Dict[str, float]]:
        """

        Normalize scores using threshold/objective penalty system.

        

        This method implements a three-zone penalty system for each category:

        1. Below threshold: Score set to 0 (elimination)

        2. Threshold to objective: Linear penalty scale (graduated)

        3. At/above objective: Full normalized score (no penalty)

        

        The penalty system operates on raw scores before standard normalization,

        creating a more realistic evaluation that reflects minimum requirements

        and ideal targets.

        

        Returns:

            Dict[str, Dict[str, float]]: Nested dictionary structure

            {product_name: {category: penalized_score}}

            where penalized_score incorporates threshold/objective logic

            

        Raises:

            ValueError: If penalty system is enabled but not properly configured

            

        Penalty Logic:

            For each category:

            - raw_score < threshold: penalized_score = 0

            - threshold <= raw_score < objective: linear interpolation

            - raw_score >= objective: standard normalization

            

        Example:

            Category: reliability (maximize=True, threshold=80, objective=95)

            Raw scores: [70, 85, 98] → Penalized: [0, ~33, 100]

        """
        # Validate penalty configuration
        validation_errors = self.validate_penalty_configuration()
        if validation_errors:
            raise ValueError(f"Penalty configuration errors: {validation_errors}")
        
        # Validate that we have data to normalize
        if not self.products:
            raise ValueError("No products to analyze")
        
        penalized = {}
        
        # Process each category independently
        for category in self.categories:
            threshold = self.thresholds[category]
            objective = self.objectives[category]
            maximize = self.maximize[category]
            
            # Extract all values for this category
            values = [self.products[p][category] for p in self.products]
            
            # Calculate penalized scores for each product
            for product in self.products:
                if product not in penalized:
                    penalized[product] = {}
                
                raw_score = self.products[product][category]
                
                # Apply three-zone penalty logic
                if maximize:
                    # For maximize categories: higher is better
                    if raw_score < threshold:
                        # Zone 1: Below threshold = elimination
                        penalized_score = 0.0
                    elif raw_score < objective:
                        # Zone 2: Threshold to objective = linear penalty
                        # Scale from 0 to some intermediate value (e.g., 50)
                        progress = (raw_score - threshold) / (objective - threshold)
                        penalized_score = progress * 50.0  # Scale to 0-50 range
                    else:
                        # Zone 3: At/above objective = standard normalization
                        # Find min/max among products that meet objective
                        qualified_values = [v for v in values if v >= objective]
                        if len(qualified_values) > 1:
                            min_qual = min(qualified_values)
                            max_qual = max(qualified_values)
                            range_qual = max_qual - min_qual
                            if range_qual > 0:
                                penalized_score = 50 + ((raw_score - min_qual) / range_qual) * 50
                            else:
                                penalized_score = 100.0  # All qualified scores are equal
                        else:
                            penalized_score = 100.0  # Only one or no qualified products
                else:
                    # For minimize categories: lower is better
                    if raw_score > threshold:
                        # Zone 1: Above threshold = elimination
                        penalized_score = 0.0
                    elif raw_score > objective:
                        # Zone 2: Objective to threshold = linear penalty
                        progress = (threshold - raw_score) / (threshold - objective)
                        penalized_score = progress * 50.0
                    else:
                        # Zone 3: At/below objective = standard normalization
                        qualified_values = [v for v in values if v <= objective]
                        if len(qualified_values) > 1:
                            min_qual = min(qualified_values)
                            max_qual = max(qualified_values)
                            range_qual = max_qual - min_qual
                            if range_qual > 0:
                                penalized_score = 50 + ((max_qual - raw_score) / range_qual) * 50
                            else:
                                penalized_score = 100.0
                        else:
                            penalized_score = 100.0
                
                penalized[product][category] = penalized_score
        
        return penalized



    def set_aggregation_method(self, method: str):
        """

        Set the aggregation method for utility calculation.

        

        This method allows switching between different mathematical approaches

        for combining normalized scores. Affects risk tolerance and compensation

        between criteria.

        

        Args:

            method: Aggregation approach to use

                'weighted_sum': Linear aggregation (full compensation)

                'geometric_mean': Geometric aggregation (penalizes poor performance)

                'threshold_penalty': Threshold/objective penalty system

                

        Raises:

            ValueError: If method is not supported

            

        Example:

            calc.set_aggregation_method('threshold_penalty')  # Enable threshold penalties

        """
        valid_methods = ['weighted_sum', 'geometric_mean', 'threshold_penalty']
        if method not in valid_methods:
            raise ValueError(f"Method must be one of: {valid_methods}")
        
        self.aggregation_method = method
        
        # Enable penalty system if threshold_penalty method is selected
        if method == 'threshold_penalty':
            self.use_penalties = True
        else:
            self.use_penalties = False
        
        # Invalidate cache since calculation method has changed
        self._cache_valid = False



    def set_thresholds(self, thresholds: Dict[str, float]):
        """

        Set minimum acceptable threshold values for categories.

        

        Thresholds represent the minimum acceptable raw score for each category.

        Products scoring below threshold in any category will be heavily penalized

        or eliminated from consideration (depending on penalty settings).

        

        Args:

            thresholds: Dictionary mapping categories to minimum threshold values

                    {category: threshold_value} in raw score units

                    

        Raises:

            ValueError: If thresholds don't include all categories

            

        Example:

            calc.set_thresholds({'reliability': 80, 'performance': 60, 'cost': 1000})

        """
        # Validate all categories are included
        missing_cats = set(self.categories) - set(thresholds.keys())
        if missing_cats:
            raise ValueError(f"Must provide thresholds for all categories. Missing: {missing_cats}")
        
        # Update thresholds
        self.thresholds.update(thresholds)
        
        # Invalidate cache since penalty calculations may change
        self._cache_valid = False

    def set_objectives(self, objectives: Dict[str, float]):
        """

        Set target/ideal objective values for categories.

        

        Objectives represent the ideal or target raw score for each category.

        Products meeting or exceeding objectives receive full normalized scores.

        Products between threshold and objective receive graduated penalties.

        

        Args:

            objectives: Dictionary mapping categories to target objective values

                    {category: objective_value} in raw score units

                    

        Raises:

            ValueError: If objectives don't include all categories

            

        Example:

            calc.set_objectives({'reliability': 95, 'performance': 90, 'cost': 500})

        """
        # Validate all categories are included
        missing_cats = set(self.categories) - set(objectives.keys())
        if missing_cats:
            raise ValueError(f"Must provide objectives for all categories. Missing: {missing_cats}")
        
        # Update objectives
        self.objectives.update(objectives)
        
        # Invalidate cache since penalty calculations may change
        self._cache_valid = False

    def set_penalty_system(self, enabled: bool):
        """

        Enable or disable the threshold/objective penalty system.

        

        When enabled, products are evaluated using threshold and objective values:

        - Below threshold: Severe penalty or elimination

        - Threshold to objective: Linear penalty scale

        - At/above objective: Full reward

        

        When disabled, uses standard min-max normalization without penalties.

        

        Args:

            enabled: True to enable penalty system, False to use standard normalization

            

        Example:

            calc.set_penalty_system(True)  # Enable penalties

        """
        self.use_penalties = enabled
        
        # Invalidate cache since calculation method changes
        self._cache_valid = False

    def validate_penalty_configuration(self) -> List[str]:
        """

        Validate the penalty system configuration and return any issues.

        

        Checks for common configuration problems like missing values,

        threshold > objective, or invalid relationships between values.

        

        Returns:

            List[str]: List of validation error messages (empty if valid)

            

        Example:

            errors = calc.validate_penalty_configuration()

            if errors:

                print("Configuration issues:", errors)

        """
        errors = []
        
        if self.use_penalties:
            # Check for missing threshold/objective values
            for cat in self.categories:
                if self.thresholds[cat] is None:
                    errors.append(f"Missing threshold value for category: {cat}")
                if self.objectives[cat] is None:
                    errors.append(f"Missing objective value for category: {cat}")
            
            # Check threshold <= objective relationship
            for cat in self.categories:
                threshold = self.thresholds[cat]
                objective = self.objectives[cat]
                
                if threshold is not None and objective is not None:
                    if self.maximize[cat]:
                        # For maximize: threshold should be <= objective
                        if threshold > objective:
                            errors.append(f"Category '{cat}': threshold ({threshold}) should be <= objective ({objective}) for maximize categories")
                    else:
                        # For minimize: threshold should be >= objective  
                        if threshold < objective:
                            errors.append(f"Category '{cat}': threshold ({threshold}) should be >= objective ({objective}) for minimize categories")
        
        return errors



    def calculate_utilities(self) -> Dict[str, float]:
        """

        Calculate final weighted utility scores for all products.

        

        This method implements the core MCDA calculation by combining normalized

        scores with user-defined weights. The result is a single utility value

        per product that can be used for ranking and decision making.

        

        Uses caching to avoid recomputation. Depends on normalize_scores() for

        input data, creating a calculation chain: raw → normalized → utilities.

        

        Returns:

            Dict[str, float]: Mapping of product names to utility scores

            {product_name: utility_score} where utility is roughly 0-100 scale

            

        Utility Formula:

            utility = Σ(weight[category] * normalized_score[category])

            for all categories

            

        Example Output:

            {'Product A': 78.5, 'Product B': 65.2, 'Product C': 82.1}

            

        Note:

            Higher utility scores indicate better overall performance considering

            all criteria and their relative importance weights.

        """
        # Return cached results if available and valid
        if self._cache_valid and self._cached_utilities:
            return self._cached_utilities
        
        # Get normalized scores (may trigger normalization if needed)
        normalized = self.normalize_scores()
        utilities = {}
        
        # Calculate utilities based on selected aggregation method
        for product, scores in normalized.items():
            if self.aggregation_method == 'weighted_sum':
                # Linear aggregation: full compensation between criteria
                utility = sum(self.weights[cat] * scores[cat] for cat in self.categories)

            elif self.aggregation_method == 'threshold_penalty':
                # Threshold penalty system uses weighted sum on penalized scores
                utility = sum(self.weights[cat] * scores[cat] for cat in self.categories)

            elif self.aggregation_method == 'geometric_mean':
                # Geometric aggregation: penalizes poor performance
                utility = 1.0
                for cat in self.categories:
                    # Convert normalized score to 0-1 scale for geometric mean
                    # Add small epsilon to avoid zero values that would make product zero
                    score_01 = max(scores[cat] / 100.0, 0.001)
                    utility *= score_01 ** self.weights[cat]
                
                # Convert back to 0-100 scale for consistency with weighted sum
                utility *= 100.0
                
            utilities[product] = utility
        
        # Cache results
        self._cached_utilities = utilities
        return utilities

    def rank_products(self) -> List[Tuple[str, float]]:
        """

        Rank all products by utility score in descending order.

        

        This method provides the primary output for decision making by ordering

        products from best (highest utility) to worst (lowest utility). Uses

        the calculated utilities as the ranking criterion.

        

        Returns:

            List[Tuple[str, float]]: List of (product_name, utility_score) tuples

            ordered by utility score (highest first)

            

        Example Output:

            [('Product C', 82.1), ('Product A', 78.5), ('Product B', 65.2)]

            

        Usage:

            rankings = calc.rank_products()

            best_product = rankings[0][0]  # Name of top-ranked product

            best_score = rankings[0][1]    # Utility score of best product

        """
        utilities = self.calculate_utilities()
        return sorted(utilities.items(), key=lambda x: x[1], reverse=True)

    def get_results_df(self) -> pd.DataFrame:
        """

        Generate comprehensive results as a pandas DataFrame.

        

        This method creates a detailed output table showing raw scores, normalized

        scores, and final utilities for all products. Useful for detailed analysis,

        reporting, and understanding how the calculations work.

        

        Returns:

            pd.DataFrame: Results table with columns:

                - Product: product name

                - Utility: final utility score

                - {category}_raw: original score for each category

                - {category}_norm: normalized score for each category

            Sorted by utility score (highest first)

            

        Returns empty DataFrame if no products have been added.



        Usage:

            df = calc.get_results_df()

            df.to_excel('results.xlsx', index=False)

        """
        # Handle empty case
        if not self.products:
            return pd.DataFrame()
        
        # Get calculated values
        utilities = self.calculate_utilities()
        normalized = self.normalize_scores()
        
        # Build comprehensive results
        results = []
        for product in self.products:
            # Start with product name and utility
            row = {'Product': product, 'Utility': utilities[product]}
            
            # Add raw and normalized scores for each category
            for category in self.categories:
                row[f'{category}_raw'] = self.products[product][category]
                row[f'{category}_norm'] = normalized[product][category]
            
            results.append(row)
        
        # Return as sorted DataFrame
        return pd.DataFrame(results).sort_values('Utility', ascending=False)

    def print_summary(self):
        """

        Print a concise summary of the calculator state and results.

        

        This method provides a quick overview for interactive use, showing

        the configuration and current rankings without requiring additional

        data processing or formatting.

        

        Output includes:

        - Number and names of evaluation categories

        - Number of products loaded

        - Current product rankings (if any products exist)

        """
        print(f"\nUtility Calculator Summary")
        print(f"Categories: {', '.join(self.categories)}")
        print(f"Products: {len(self.products)}")
        
        # Show rankings if we have products
        if self.products:
            rankings = self.rank_products()
            print(f"\nRankings:")
            for i, (product, utility) in enumerate(rankings, 1):
                print(f"  {i}. {product}: {utility:.1f}")