ProfilingAI / src /core /life_events_manager.py
Sandrine Guétin
Version propre de DeepVest
2106f78
raw
history blame
4.65 kB
from dataclasses import dataclass
from datetime import datetime
from typing import List, Dict, Optional
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
@dataclass
class LifeEvent:
event_type: str
timestamp: datetime
impact_score: float
duration: Optional[int] # en mois
financial_impact: float
risk_adjustment: float
description: str
related_goals: List[str]
class LifeEventsManager:
"""Gestionnaire intelligent des événements de vie"""
def __init__(self):
self.events_history = []
self.impact_model = RandomForestClassifier()
self.scaler = StandardScaler()
def register_event(self, event: Dict) -> LifeEvent:
"""Enregistre un nouvel événement de vie"""
# Calcul de l'impact
impact_score = self._calculate_event_impact(event)
# Création de l'événement
life_event = LifeEvent(
event_type=event['type'],
timestamp=datetime.now(),
impact_score=impact_score,
duration=event.get('duration'),
financial_impact=self._calculate_financial_impact(event),
risk_adjustment=self._calculate_risk_adjustment(event),
description=event['description'],
related_goals=event.get('related_goals', [])
)
# Ajout à l'historique
self.events_history.append(life_event)
return life_event
def analyze_portfolio_impact(self, event: LifeEvent, portfolio: Dict) -> Dict:
"""Analyse l'impact d'un événement sur le portefeuille"""
impact = {
'reallocation_needed': False,
'risk_adjustment': 0.0,
'suggested_changes': {}
}
# Analyser l'impact sur différents aspects
risk_impact = self._analyze_risk_impact(event)
liquidity_impact = self._analyze_liquidity_impact(event)
horizon_impact = self._analyze_horizon_impact(event)
# Déterminer si un rééquilibrage est nécessaire
if abs(risk_impact) > 0.1 or abs(liquidity_impact) > 0.2:
impact['reallocation_needed'] = True
impact['risk_adjustment'] = risk_impact
impact['suggested_changes'] = self._generate_portfolio_adjustments(
portfolio, risk_impact, liquidity_impact, horizon_impact
)
return impact
def _calculate_event_impact(self, event: Dict) -> float:
"""Calcul de l'impact global d'un événement"""
# Facteurs d'impact
impact_factors = {
'financial_weight': 0.4,
'emotional_weight': 0.3,
'risk_weight': 0.3
}
# Calcul des impacts individuels
financial_impact = self._calculate_financial_impact(event)
emotional_impact = self._calculate_emotional_impact(event)
risk_impact = self._calculate_risk_impact(event)
# Impact total pondéré
total_impact = (
financial_impact * impact_factors['financial_weight'] +
emotional_impact * impact_factors['emotional_weight'] +
risk_impact * impact_factors['risk_weight']
)
return np.clip(total_impact, 0, 1)
def _generate_portfolio_adjustments(self,
portfolio: Dict,
risk_impact: float,
liquidity_impact: float,
horizon_impact: float) -> Dict[str, float]:
"""Génère les ajustements suggérés pour le portefeuille"""
adjustments = {}
# Ajustement basé sur le risque
if abs(risk_impact) > 0:
risk_adjustments = self._calculate_risk_based_adjustments(
portfolio, risk_impact
)
adjustments.update(risk_adjustments)
# Ajustement basé sur la liquidité
if abs(liquidity_impact) > 0:
liquidity_adjustments = self._calculate_liquidity_based_adjustments(
portfolio, liquidity_impact
)
adjustments.update(liquidity_adjustments)
# Normalisation des ajustements
total_adjustment = sum(abs(adj) for adj in adjustments.values())
if total_adjustment > 0:
adjustments = {
asset: adj/total_adjustment
for asset, adj in adjustments.items()
}
return adjustments