Upload mcda_v4.py

src/mcda_v4.py

@@ -0,0 +1,729 @@
+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}")