feat: user personalized MCDA with custom weights

Author: rebecamurillo

src/sum_impact_assessment/api/routes/jobs.py

@@ -70,6 +70,25 @@ def trigger_job(
                     }
                 }
             },
+            "mcda_qualitative_user_personalized": {
+                "summary": "Qualitative MCDA Analysis for user-personalized perspective",
+                "description": "Run MCDA analysis with user-personalized stakeholder weights, from qualitative data from expert surveys",
+                "value": {
+                    "params": {
+                        "perspective": "user_personalized",
+                        "goals_weights": {
+                            "Improve Accessibility": 0.12,
+                            "Improve Mobility Service": 0.16,
+                            "Improve Multimodality": 0.13,
+                            "Noise Hinderance": 0.07,
+                            "Improve Public Transport": 0.12,
+                            "Reduction of Congestion": 0.15,
+                            "Reduction of Emission": 0.14,
+                            "Improve Safety": 0.11
+                        }
+                    }
+                }
+            },
             "mcda_qualitative_regulatory_perspective": {
                 "summary": "Qualitative MCDA Analysis for regulatory perspective",
                 "description": "Run MCDA analysis with regulatory stakeholder weights, from qualitative data from expert surveys",
@@ -97,6 +116,26 @@ def trigger_job(
                     }
                 }
             },
+            "mcda_quantitative_user_personalized": {
+                "summary": "Quantitative MCDA Analysis for user-personalized perspective",
+                "description": "Run MCDA analysis with user-personalized stakeholder weights, from Quantitative data from KPI/measures impact analysis",
+                "value": {
+                    "params": {
+                        "perspective": "user_personalized",
+                        "kpi_group_type": "MCDA_GOALS",
+                        "goals_weights": {
+                            "Improve Accessibility": 0.12,
+                            "Improve Mobility Service": 0.16,
+                            "Improve Multimodality": 0.13,
+                            "Noise Hinderance": 0.07,
+                            "Improve Public Transport": 0.12,
+                            "Reduction of Congestion": 0.15,
+                            "Reduction of Emission": 0.14,
+                            "Improve Safety": 0.11
+                        }
+                    }
+                }
+            },
             "mcda_quantitative_regulatory_perspective": {
                 "summary": "Quantitative MCDA Analysis for regulatory perspective",
                 "description": "Run MCDA analysis with regulatory stakeholder weights, from quantitative data form KPI/measures impact analysis",

src/sum_impact_assessment/services/mcda_qualitative_job.py

@@ -9,7 +9,7 @@
 from ..schemas.job import JobStatusEnum
 from ..schemas.mcda import Goal, Alternative
 from ..utils.logger import get_logger
-from ..utils.data_loaders import get_goal_weights_for_perspective, load_mcda_config
+from ..utils.data_loaders import get_goal_weights_for_perspective, load_mcda_config, normalize_goal_weights
 
 # Initialize logger
 logger = get_logger(__name__)
@@ -26,19 +26,11 @@ class McdaQualitativeJob:
     4. Saves structured MCDA input and output snapshots
     """
 
-    @staticmethod
-    def _normalize_goal_weights(goal_weights: Dict[str, float]) -> Dict[str, float]:
-        """Normalize all goal weight keys to lower-case."""
-        return {
-            goal_name : weight
-            for goal_name, weight in goal_weights.items()
-        }
-
     @staticmethod
     def _normalize_goal_scores(goal_scores: Dict[str, float]) -> Dict[str, float]:
         """Normalize all goal score keys to lower-case."""
         return {
-            goal_name : score
+            goal_name: score
             for goal_name, score in goal_scores.items()
         }
 
@@ -108,7 +100,8 @@ def build_alternatives(config: Dict[str, Any]) -> list[Alternative]:
 
         alternatives = []
         for activity_key, activity_goal_scores in goals_score.items():
-            values = McdaQualitativeJob._normalize_goal_scores(activity_goal_scores)
+            values = McdaQualitativeJob._normalize_goal_scores(
+                activity_goal_scores)
             alternative = Alternative(
                 name=activity_labels.get(activity_key, activity_key),
                 values=values
@@ -136,11 +129,6 @@ def build_goals(alternatives: list[Alternative], goal_weights: Optional[Dict[str
         if not alternatives:
             return []
 
-        normalized_goal_weights = (
-            McdaQualitativeJob._normalize_goal_weights(goal_weights)
-            if goal_weights else None
-        )
-
         goal_names = list(alternatives[0].values.keys())
         if not goal_names:
             return []
@@ -150,7 +138,7 @@ def build_goals(alternatives: list[Alternative], goal_weights: Optional[Dict[str
 
         for goal_name in goal_names:
             weight = McdaQualitativeJob._get_weight_by_goal(
-                goal_name, normalized_goal_weights, default_weight)
+                goal_name, goal_weights, default_weight)
 
             min_value, max_value = McdaQualitativeJob._get_min_max_values_per_goal(
                 goal_name, alternatives)
@@ -196,7 +184,7 @@ def get_goal_weights(perspective: Optional[str]) -> Optional[Dict[str, float]]:
                 goal_weights = get_goal_weights_for_perspective(perspective)
                 logger.info(
                     f"Using qualitative goal weights for perspective: {perspective}")
-                return McdaQualitativeJob._normalize_goal_weights(goal_weights)
+                return goal_weights
             except ValueError as e:
                 logger.warning(
                     f"Failed to load perspective weights: {e}. Using equal weights.")
@@ -227,9 +215,16 @@ def run(job_id: str, db: Session, params: Optional[Dict] = None) -> None:
             )
 
             perspective = params.get("perspective") if params else None
+            analysis_name = params.get("name") if params else None
+            personalized_goal_weights = params.get(
+                "goals_weights") if params else None
             logger.debug(
                 "Qualitative MCDA analysis parameters",
-                extra={"perspective": perspective}
+                extra={
+                    "perspective": perspective,
+                    "analysis_name": analysis_name,
+                    "personalized_goal_weights": bool(personalized_goal_weights)
+                }
             )
 
             config = load_mcda_config()
@@ -238,16 +233,25 @@ def run(job_id: str, db: Session, params: Optional[Dict] = None) -> None:
                 "perspectives": config.get("perspectives", {}),
                 "business_activities": config.get("business_activities", {})
             }
-            job_repo.update_job_data(job_id=job_id, input_data=input_data_snapshot)
+            job_repo.update_job_data(
+                job_id=job_id, input_data=input_data_snapshot)
+
+            if perspective == "user_personalized" and personalized_goal_weights:
+                goal_weights = normalize_goal_weights(
+                    personalized_goal_weights)
+                logger.info("Using user-personalized qualitative goal weights")
+            else:
+                goal_weights = McdaQualitativeJob.get_goal_weights(perspective)
 
-            goal_weights = McdaQualitativeJob.get_goal_weights(perspective)
             alternatives = McdaQualitativeJob.build_alternatives(config)
             goals = McdaQualitativeJob.build_goals(alternatives, goal_weights)
 
             if not alternatives:
-                raise Exception("No business activities were found. Cannot proceed with qualitative MCDA.")
+                raise Exception(
+                    "No business activities were found. Cannot proceed with qualitative MCDA.")
             if not goals:
-                raise Exception("No goals were built from business activity scores. Cannot proceed with qualitative MCDA.")
+                raise Exception(
+                    "No goals were built from business activity scores. Cannot proceed with qualitative MCDA.")
 
             mcda_analyzer = PrometheeGaiaAnalyzer(
                 goals=goals,
@@ -256,22 +260,26 @@ def run(job_id: str, db: Session, params: Optional[Dict] = None) -> None:
 
             mcda_input_data_snapshot = {
                 "perspective": perspective,
+                "name": analysis_name,
                 "goals": [goal.model_dump() for goal in goals],
                 "alternatives": [alt.model_dump() for alt in alternatives],
                 "timestamp": datetime.utcnow().isoformat()
             }
             input_data_snapshot.update(mcda_input_data_snapshot)
-            job_repo.update_job_data(job_id=job_id, input_data=input_data_snapshot)
+            job_repo.update_job_data(
+                job_id=job_id, input_data=input_data_snapshot)
 
             mcda_output = mcda_analyzer.run_analysis(run_visualizations=False)
 
             output_data_snapshot = {
+                "name": analysis_name,
                 "kpi_impact_results": [],
                 "kpi_impact_errors": [],
                 "mcda_results": mcda_output.model_dump(),
                 "timestamp": datetime.utcnow().isoformat()
             }
-            job_repo.update_job_data(job_id=job_id, output_data=output_data_snapshot)
+            job_repo.update_job_data(
+                job_id=job_id, output_data=output_data_snapshot)
 
             top_alt_key = mcda_output.ranking[0] if mcda_output.ranking else "N/A"
             top_alt_name = mcda_output.alternative_labels.get(
@@ -291,7 +299,8 @@ def run(job_id: str, db: Session, params: Optional[Dict] = None) -> None:
                 message=success_message,
                 completed_at=datetime.utcnow()
             )
-            logger.info(f"Qualitative MCDA analysis job completed successfully: {job_id}")
+            logger.info(
+                f"Qualitative MCDA analysis job completed successfully: {job_id}")
 
         except Exception as e:
             error_message = f"McdaQualitativeJob failed: {str(e)}"

src/sum_impact_assessment/services/mcda_quantitative_job.py

@@ -11,7 +11,7 @@
 from ..schemas.mcda import Goal, Alternative
 from ..schemas.core import Measure
 from ..utils.logger import get_logger
-from ..utils.data_loaders import get_goal_weights_for_perspective
+from ..utils.data_loaders import get_goal_weights_for_perspective, normalize_goal_weights
 
 # Initialize logger
 logger = get_logger(__name__)
@@ -232,11 +232,16 @@ def run(job_id: str, db: Session, params: Optional[Dict] = None) -> None:
             # Extract optional parameters
             kpi_group_filter = params.get("kpi_group_type") if params else None
             perspective = params.get("perspective") if params else None
+            analysis_name = params.get("name") if params else None
+            personalized_goal_weights = params.get(
+                "goals_weights") if params else None
             logger.debug(
                 f"MCDA analysis parameters",
                 extra={
                     "kpi_group_filter": kpi_group_filter,
-                    "perspective": perspective
+                    "perspective": perspective,
+                    "analysis_name": analysis_name,
+                    "personalized_goal_weights": bool(personalized_goal_weights)
                 }
             )
 
@@ -269,7 +274,14 @@ def run(job_id: str, db: Session, params: Optional[Dict] = None) -> None:
 
             # Build Goals from KPI groups with perspective-based weights
             # Load weights from perspective if provided, otherwise use equal weights
-            goal_weights = McdaQuantitativeJob.get_goal_weights(perspective)
+            if perspective == "user_personalized" and personalized_goal_weights:
+                goal_weights = normalize_goal_weights(
+                    personalized_goal_weights)
+                logger.info(
+                    "Using user-personalized quantitative goal weights")
+            else:
+                goal_weights = McdaQuantitativeJob.get_goal_weights(
+                    perspective)
 
             # Build Alternatives from measures
             alternatives = McdaQuantitativeJob.build_alternatives(
@@ -289,6 +301,7 @@ def run(job_id: str, db: Session, params: Optional[Dict] = None) -> None:
             # save MCDA input data snapshot, added to previous input_data_snapshot
             mcda_input_data_snapshot = {
                 "perspective": perspective,
+                "name": analysis_name,
                 "goals": [goal.model_dump() for goal in goals],
                 "alternatives": [alt.model_dump() for alt in alternatives],
                 "timestamp": datetime.utcnow().isoformat()
@@ -313,6 +326,7 @@ def run(job_id: str, db: Session, params: Optional[Dict] = None) -> None:
             # PHASE 5: Save output data
             logger.debug("Phase 5: Saving MCDA output")
             output_data_snapshot = {
+                "name": analysis_name,
                 "kpi_impact_results": kpi_impact_results,
                 "kpi_impact_errors": error_results,
                 "mcda_results": mcda_output.model_dump(),

src/sum_impact_assessment/utils/data_loaders.py

@@ -3,7 +3,7 @@
 """
 import json
 from pathlib import Path
-from typing import Dict, Any
+from typing import Dict, Any, Optional
 
 # Path to data directory (relative to this file)
 DATA_DIR = Path(__file__).parent.parent / "data"
@@ -60,6 +60,7 @@ def load_mcda_goal_weights() -> Dict[str, Dict[str, float]]:
     config = load_mcda_config()
     return config.get("perspectives", {}).get("weights", {})
 
+
 def get_mcda_perspectives_labels() -> Dict[str, Dict[str, str]]:
     """
     Get the labels for MCDA perspectives and their goals.
@@ -90,7 +91,6 @@ def get_mcda_perspectives_labels() -> Dict[str, Dict[str, str]]:
     return config.get("perspectives", {}).get("labels", {})
 
 
-
 def get_goal_weights_for_perspective(perspective: str) -> Dict[str, float]:
     """
     Get goal weights for a specific perspective.
@@ -114,3 +114,32 @@ def get_goal_weights_for_perspective(perspective: str) -> Dict[str, float]:
         )
 
     return all_weights[perspective]
+
+
+def normalize_goal_weights(goal_weights: Optional[Dict[str, float]]) -> Optional[Dict[str, float]]:
+    """
+    Normalize goal weights so the total equals 1.0 (100%).
+
+    Args:
+        goal_weights: Dictionary mapping goal names to non-negative weights.
+
+    Returns:
+        Normalized weights dictionary, or None if goal_weights is None.
+
+    Raises:
+        ValueError: If goal_weights is empty or total weight is zero.
+    """
+    if goal_weights is None:
+        return None
+
+    if not goal_weights:
+        raise ValueError("Goal weights are empty and cannot be normalized")
+
+    total_weight = sum(goal_weights.values())
+    if total_weight == 0:
+        raise ValueError("Goal weights sum to zero and cannot be normalized")
+
+    return {
+        goal_name: weight / total_weight
+        for goal_name, weight in goal_weights.items()
+    }