Deepvest/ProfilingAI

1# src/analysis/goal_simulator.py2 3import numpy as np4import pandas as pd5from typing import Dict, List, Optional6from dataclasses import dataclass7from datetime import datetime8from ..core.profiling import PersonalProfile9 10@dataclass11class GoalSimulationResult:12    """Résultat de la simulation d'un objectif"""13    success_probability: float14    required_monthly_investment: float15    risk_assessment: Dict[str, float]16    recommended_adjustments: List[str]17    timeline: pd.DataFrame18    stress_test_results: Dict[str, float]19 20@dataclass21class InvestmentGoal:22    """Définition d'un objectif d'investissement"""23    name: str24    target_amount: float25    timeframe: int  # en années26    priority: int27    current_progress: float28    required_return: float29 30class GoalSimulator:31    """Simulateur d'objectifs financiers"""32    33    def __init__(self, market_analyzer, risk_manager):34        self.market_analyzer = market_analyzer35        self.risk_manager = risk_manager36        self.monte_carlo_sims = 100037        38    async def simulate_goal_achievement(self,39                                      goal: InvestmentGoal,40                                      current_portfolio: Dict,41                                      personal_profile: PersonalProfile) -> GoalSimulationResult:42        """Simule la réalisation d'un objectif financier"""43        try:44            # Générer les scénarios de marché45            scenarios = self._generate_market_scenarios(46                timeframe=goal.timeframe,47                current_portfolio=current_portfolio,48                profile=personal_profile49            )50            51            # Calculer la probabilité de succès52            success_prob = self._calculate_success_probability(53                scenarios=scenarios,54                goal=goal55            )56            57            # Calculer l'investissement mensuel requis58            required_investment = self._calculate_required_investment(59                goal=goal,60                scenarios=scenarios,61                success_target=0.8  # 80% de probabilité de succès62            )63            64            # Analyse de risque65            risk_assessment = self._assess_goal_risks(66                goal=goal,67                scenarios=scenarios,68                personal_profile=personal_profile69            )70            71            # Recommandations d'ajustement72            adjustments = self._generate_recommendations(73                goal=goal,74                success_prob=success_prob,75                risk_assessment=risk_assessment,76                personal_profile=personal_profile77            )78            79            return GoalSimulationResult(80                success_probability=success_prob,81                required_monthly_investment=required_investment,82                risk_assessment=risk_assessment,83                recommended_adjustments=adjustments,84                timeline=self._generate_timeline(scenarios, goal),85                stress_test_results=self._run_stress_tests(scenarios, goal)86            )87            88        except Exception as e:89            print(f"Erreur dans la simulation de l'objectif: {e}")90            raise91 92    def _generate_market_scenarios(self,93                                 timeframe: int,94                                 current_portfolio: Dict,95                                 profile: PersonalProfile) -> np.ndarray:96        """Génère des scénarios de marché via Monte Carlo"""97        try:98            # Paramètres de simulation99            n_months = timeframe * 12100            n_sims = self.monte_carlo_sims101            102            # Générer les rendements mensuels103            monthly_returns = np.random.normal(104                loc=0.005,  # 6% annuel105                scale=0.02,  # Volatilité mensuelle106                size=(n_sims, n_months)107            )108            109            # Ajuster les rendements selon le profil de risque110            adjusted_returns = self._adjust_returns_for_risk_profile(111                monthly_returns, profile112            )113            114            # Générer les trajectoires de portefeuille115            initial_value = sum(current_portfolio.values())116            scenarios = np.zeros((n_sims, n_months))117            118            for i in range(n_sims):119                scenarios[i] = initial_value * np.cumprod(1 + adjusted_returns[i])120                121            return scenarios122            123        except Exception as e:124            print(f"Erreur dans la génération des scénarios: {e}")125            raise126 127    def _calculate_success_probability(self,128                                    scenarios: np.ndarray,129                                    goal: InvestmentGoal) -> float:130        """Calcule la probabilité d'atteindre l'objectif"""131        try:132            # Valeurs finales des scénarios133            final_values = scenarios[:, -1]134            135            # Nombre de scénarios atteignant l'objectif136            success_count = np.sum(final_values >= goal.target_amount)137            138            return success_count / len(scenarios)139            140        except Exception as e:141            print(f"Erreur dans le calcul de la probabilité de succès: {e}")142            return 0.0143 144    def _calculate_required_investment(self,145                                    goal: InvestmentGoal,146                                    scenarios: np.ndarray,147                                    success_target: float) -> float:148        """Calcule l'investissement mensuel nécessaire"""149        try:150            current_gap = goal.target_amount - goal.current_progress151            months_remaining = goal.timeframe * 12152            153            # Calculer le taux de rendement moyen des scénarios154            avg_return = np.mean(np.diff(scenarios) / scenarios[:, :-1])155            156            # Formule PMT modifiée pour tenir compte du risque157            r = (1 + avg_return) ** (1/12) - 1  # Taux mensuel158            required_monthly = (current_gap * r) / ((1 + r) ** months_remaining - 1)159            160            # Ajouter une marge de sécurité161            safety_margin = 1.1  # 10% de marge162            return required_monthly * safety_margin163            164        except Exception as e:165            print(f"Erreur dans le calcul de l'investissement requis: {e}")166            return 0.0167 168    def _assess_goal_risks(self,169                          goal: InvestmentGoal,170                          scenarios: np.ndarray,171                          personal_profile: PersonalProfile) -> Dict[str, float]:172        """Évalue les risques liés à l'objectif"""173        try:174            return {175                'shortfall_risk': self._calculate_shortfall_risk(scenarios, goal),176                'time_risk': self._calculate_time_risk(scenarios, goal),177                'investment_risk': self._calculate_investment_risk(goal, personal_profile),178                'liquidity_risk': self._calculate_liquidity_risk(goal, personal_profile)179            }180        except Exception as e:181            print(f"Erreur dans l'évaluation des risques: {e}")182            return {}183 184    def _generate_recommendations(self,185                                goal: InvestmentGoal,186                                success_prob: float,187                                risk_assessment: Dict[str, float],188                                personal_profile: PersonalProfile) -> List[str]:189        """Génère des recommandations d'ajustement"""190        recommendations = []191        192        # Analyse du taux de succès193        if success_prob < 0.6:194            recommendations.append(195                "La probabilité d'atteinte de l'objectif est faible. Considérez : "196                "1. Augmenter l'épargne mensuelle "197                "2. Allonger l'horizon d'investissement "198                "3. Ajuster l'objectif à la baisse"199            )200            201        # Analyse des risques202        for risk_type, risk_level in risk_assessment.items():203            if risk_level > 0.7:204                recommendations.append(205                    f"Risque {risk_type} élevé. Suggestion : "206                    f"{self._get_risk_mitigation_strategy(risk_type)}"207                )208                209        return recommendations210 211    def _get_risk_mitigation_strategy(self, risk_type: str) -> str:212        """Retourne une stratégie de mitigation pour chaque type de risque"""213        strategies = {214            'shortfall_risk': "Augmenter l'épargne ou réduire l'objectif",215            'time_risk': "Prévoir une période tampon supplémentaire",216            'investment_risk': "Diversifier davantage le portefeuille",217            'liquidity_risk': "Maintenir une réserve de liquidité suffisante"218        }219        return strategies.get(risk_type, "Revoir la stratégie globale")