""" MODULE TWIN ENGINE - JUMEAU NUMERIQUE DU PORTEFEUILLE ===================================================== Moteur de simulation pur (sans Streamlit, sans Google Sheets). Principe : 1. build_portfolio_state() -> etat A-Box courant (photo du portefeuille) 2. calibrate_behavior() -> parametres comportementaux estimes sur l'historique 3. run_simulation() -> N trajectoires Monte Carlo sur H semaines 4. compare_scenarios() -> tableau comparatif entre variantes de parametres Le moteur reproduit les regles metier de AnalyseRepayment : - echeancier issu de Dates_Versements - penalites = montant * taux_hebdo * ceil(jours_retard / 7) - paiement partiel -> reliquat reporte sur l'echeance suivante """ from dataclasses import dataclass, field, replace from datetime import datetime, date, timedelta import math import random import pandas as pd # ============================================================================= # 1. ETAT DU PORTEFEUILLE (A-BOX) # ============================================================================= @dataclass class LoanState: loan_id: str client_id: str montant_versement: float echeances: list # list[date] - echeancier complet idx_next: int # index de la prochaine echeance due solde: float # reste du (capital + interets) reliquat: float = 0.0 # impaye sur l'echeance courante source: str = "MASTER" part_interets: float = 0.0 # part des interets dans chaque versement (Cout_Credit / Montant_Total) @property def is_closed(self): return self.idx_next >= len(self.echeances) or self.solde <= 0 def _parse_echeances(raw): if not raw or (isinstance(raw, float) and math.isnan(raw)): return [] s = str(raw) sep = ';' if ';' in s else ',' out = [] for d in s.split(sep): d = d.strip() if not d: continue for fmt in ("%d/%m/%Y", "%Y-%m-%d"): try: out.append(datetime.strptime(d, fmt).date()) break except ValueError: continue return out def _safe_float(v, default=0.0): try: f = float(v) return default if math.isnan(f) else f except (TypeError, ValueError): return default def build_portfolio_state(df_prets, df_prets_update, df_remb, today=None): """Construit l'etat courant du portefeuille (prets non clotures uniquement).""" if today is None: today = date.today() df_prets_update = df_prets_update if df_prets_update is not None else pd.DataFrame() df_remb = df_remb if df_remb is not None else pd.DataFrame() def latest_update(root_id): if df_prets_update.empty or 'ID_Pret' not in df_prets_update.columns: return None sub = df_prets_update[df_prets_update['ID_Pret'].astype(str) == str(root_id)].copy() if sub.empty: return None sub['_v'] = pd.to_numeric(sub.get('Version'), errors='coerce').fillna(0) return sub.loc[sub['_v'].idxmax()] # Paiements agreges par pret paid_by_loan, complete_by_loan = {}, {} if not df_remb.empty and 'ID_Pret' in df_remb.columns: for _, p in df_remb.iterrows(): lid = str(p['ID_Pret']) paid_by_loan[lid] = paid_by_loan.get(lid, 0.0) + ( _safe_float(p.get('Montant_Principal')) + _safe_float(p.get('Montant_Interets')) ) if 'PARTIEL' not in str(p.get('Statut_Paiement', '')).upper(): complete_by_loan[lid] = complete_by_loan.get(lid, 0) + 1 states = [] for _, loan in df_prets.iterrows(): statut = str(loan.get('Statut', '')).strip() root_id = str(loan.get('ID_Pret', '')) row, source = loan, 'MASTER' if statut == 'UPDATED': latest = latest_update(root_id) if latest is None or str(latest.get('Statut', '')).strip() == 'TERMINE': continue row, source = latest, 'UPDATE' elif statut not in ('ACTIF', 'EN_RETARD', 'LITIGE'): continue echeances = _parse_echeances(row.get('Dates_Versements')) if not echeances: continue total = _safe_float(row.get('Montant_Total')) paye = paid_by_loan.get(root_id, 0.0) solde = max(0.0, total - paye) if solde <= 0: continue idx_next = min(complete_by_loan.get(root_id, 0), len(echeances) - 1) cout = _safe_float(row.get('Cout_Credit')) part_interets = min(1.0, max(0.0, cout / total)) if total > 0 else 0.0 states.append(LoanState( loan_id=root_id, client_id=str(loan.get('ID_Client', '')), montant_versement=_safe_float(row.get('Montant_Versement')), echeances=echeances, idx_next=idx_next, solde=solde, source=source, part_interets=part_interets, )) return states # ============================================================================= # 2. PARAMETRES COMPORTEMENTAUX # ============================================================================= @dataclass class BehaviorParams: p_on_time: float = 0.60 p_late: float = 0.37 p_partial: float = 0.03 p_default: float = 0.02 # NON calibre : hypothese a fixer delays: tuple = (1, 2, 3, 5, 7) # distribution empirique des jours de retard partial_ratio: float = 0.5 # part moyenne versee lors d'un paiement partiel taux_hebdo: float = 0.05 grace_days: int = 0 # jours avant application des penalites n_observations: int = 0 # taille de l'echantillon de calibration taux_ref: float = 0.05 # taux de reference utilise pour le couplage (HYPOTHESE) def normalized(self): """Renormalise les probabilites d'issue a 1.""" tot = self.p_on_time + self.p_late + self.p_partial + self.p_default if tot <= 0: return replace(self, p_on_time=1.0, p_late=0.0, p_partial=0.0, p_default=0.0) return replace(self, p_on_time=self.p_on_time / tot, p_late=self.p_late / tot, p_partial=self.p_partial / tot, p_default=self.p_default / tot) def apply_penalty_coupling(params: BehaviorParams, elasticity: float) -> BehaviorParams: """ HYPOTHESE NON CALIBREE : couplage penalite -> comportement de paiement. Aucune donnee historique ne permet de mesurer cet effet (le taux de penalite n'a jamais varie chez Vortex). C'est un pari que l'utilisateur choisit d'activer pour explorer une consequence possible, pas une valeur mesuree. elasticity > 0 : baisser la penalite AMELIORE le comportement de paiement (hypothese "geste commercial" - le client paie mieux car il se sent moins presse). elasticity < 0 : baisser la penalite DEGRADE le comportement de paiement (hypothese "risque moral" - moins de pression, moins de discipline de paiement). elasticity = 0 : aucun lien suppose (comportement identique quel que soit le taux de penalite) - c'est le reglage par defaut. Pour chaque point de pourcentage sous params.taux_ref, `elasticity` points de probabilite basculent entre "en retard" et "a l'heure". """ if elasticity == 0.0: return params delta_taux_pts = (params.taux_ref - params.taux_hebdo) * 100 # points de % # positif si penalite plus douce que la reference reduction_p_late = elasticity * delta_taux_pts / 100 new_p_late = max(0.0, min(1.0, params.p_late - reduction_p_late)) shift = params.p_late - new_p_late new_p_on_time = max(0.0, params.p_on_time + shift) return replace(params, p_late=new_p_late, p_on_time=new_p_on_time).normalized() def calibrate_behavior(df_remb, p_default=0.02, taux_hebdo=0.05): """Estime les parametres comportementaux sur l'historique des remboursements.""" if df_remb is None or df_remb.empty or 'Statut_Paiement' not in df_remb.columns: return BehaviorParams(p_default=p_default, taux_hebdo=taux_hebdo).normalized() n_on_time = n_late = n_partial = 0 for s in df_remb['Statut_Paiement'].astype(str).str.upper(): if 'PARTIEL' in s: n_partial += 1 elif 'RETARD' in s: n_late += 1 else: n_on_time += 1 n = max(1, n_on_time + n_late + n_partial) scale = 1.0 - p_default delays = [] if 'Jours_Retard' in df_remb.columns: delays = [int(d) for d in pd.to_numeric(df_remb['Jours_Retard'], errors='coerce') .dropna().tolist() if d > 0] ratio = 0.5 if 'Montant_Verse' in df_remb.columns: part = df_remb[df_remb['Statut_Paiement'].astype(str).str.upper().str.contains('PARTIEL')] if not part.empty: vals = [] for _, p in part.iterrows(): due = _safe_float(p.get('Solde_Avant')) verse = _safe_float(p.get('Montant_Verse')) if due > 0: vals.append(min(1.0, verse / due)) if vals: ratio = sum(vals) / len(vals) return BehaviorParams( p_on_time=(n_on_time / n) * scale, p_late=(n_late / n) * scale, p_partial=(n_partial / n) * scale, p_default=p_default, delays=tuple(delays) if delays else (1, 2, 3, 5, 7), partial_ratio=ratio, taux_hebdo=taux_hebdo, taux_ref=taux_hebdo, n_observations=n, ).normalized() # ============================================================================= # 3. SCENARIOS # ============================================================================= @dataclass class Scenario: name: str overrides: dict = field(default_factory=dict) def apply(self, base: BehaviorParams) -> BehaviorParams: return replace(base, **self.overrides).normalized() # ============================================================================= # 4. MOTEUR DE SIMULATION # ============================================================================= def _penalite(montant, jours_retard, taux_hebdo, grace_days): effectif = jours_retard - grace_days if effectif <= 0: return 0.0 return round(montant * taux_hebdo * math.ceil(effectif / 7), 0) def _simulate_loan(loan, p, rng, horizon_end): """Simule un pret et renvoie la liste des evenements de paiement.""" events = [] reliquat = loan.reliquat defaulted = False for idx in range(loan.idx_next, len(loan.echeances)): due = loan.echeances[idx] if due > horizon_end: break montant = loan.montant_versement + reliquat reliquat = 0.0 u = rng.random() if u < p.p_on_time: events.append((due, montant, 0.0, 'ON_TIME')) elif u < p.p_on_time + p.p_late: delay = rng.choice(p.delays) pen = _penalite(montant, delay, p.taux_hebdo, p.grace_days) events.append((due + timedelta(days=delay), montant, pen, 'LATE')) elif u < p.p_on_time + p.p_late + p.p_partial: verse = round(montant * p.partial_ratio, 0) reliquat = montant - verse events.append((due, verse, 0.0, 'PARTIAL')) else: defaulted = True events.append((due, 0.0, 0.0, 'DEFAULT')) break return events, defaulted def run_simulation(states, params, horizon_weeks=26, n_runs=200, today=None, seed=42): """Monte Carlo. Renvoie un DataFrame long : run, period, date, cash, penalites, ...""" if today is None: today = date.today() horizon_end = today + timedelta(weeks=horizon_weeks) rng = random.Random(seed) # Garde-fou : ne jamais simuler un pret deja clos (echeancier epuise ou # solde nul). build_portfolio_state() les filtre deja a la construction, # mais on le revalide ici au cas ou `states` soit construit autrement. states = [s for s in states if not s.is_closed] rows = [] for run in range(n_runs): buckets = {w: {'cash': 0.0, 'penalites': 0.0, 'benefice': 0.0, 'n_paiements': 0, 'n_defauts': 0, 'n_retards': 0} for w in range(horizon_weeks + 1)} encours_perdu = 0.0 for loan in states: events, defaulted = _simulate_loan(loan, params, rng, horizon_end) if defaulted: encours_perdu += loan.solde for ev_date, montant, pen, kind in events: w = max(0, min(horizon_weeks, (ev_date - today).days // 7)) b = buckets[w] b['cash'] += montant + pen b['penalites'] += pen # Benefice encaisse = part interets du versement + penalites. b['benefice'] += montant * loan.part_interets + pen if kind == 'DEFAULT': b['n_defauts'] += 1 else: b['n_paiements'] += 1 if kind == 'LATE': b['n_retards'] += 1 cumul = 0.0 for w in range(horizon_weeks + 1): b = buckets[w] cumul += b['cash'] rows.append({ 'run': run, 'period': w, 'date': today + timedelta(weeks=w), 'cash': b['cash'], 'cash_cumule': cumul, 'penalites': b['penalites'], 'benefice': b['benefice'], 'n_paiements': b['n_paiements'], 'n_retards': b['n_retards'], 'n_defauts': b['n_defauts'], 'encours_perdu': encours_perdu, }) return pd.DataFrame(rows) def summarize(df_runs): """Agrege les trajectoires : mediane et intervalle p10-p90 par periode.""" g = df_runs.groupby('period') out = pd.DataFrame({ 'date': g['date'].first(), 'cash_p10': g['cash_cumule'].quantile(0.10), 'cash_median': g['cash_cumule'].median(), 'cash_p90': g['cash_cumule'].quantile(0.90), 'penalites_median': g['penalites'].median(), 'retards_moyen': g['n_retards'].mean(), 'defauts_moyen': g['n_defauts'].mean(), }).reset_index() return out def compare_to_objective(resume: pd.DataFrame, objectif_annuel: float) -> pd.DataFrame: """ Enrichit le tableau comparatif produit par compare_scenarios() avec l'objectif annuel centralise : colonnes Objectif_Annuel, Ecart_vs_Objectif, Progression_Pct. ATTENTION : 'cash_median' agrege des ENCAISSEMENTS TOTAUX (capital + interets + penalites), pas un profit net. Cette comparaison est un proxy de tresorerie, pas un calcul de rentabilite equivalent a celui de l'Optimizer (qui se base sur Cout_Credit). Ne pas presenter les deux chiffres comme interchangeables. """ out = resume.copy() out['Objectif_Annuel'] = objectif_annuel out['Ecart_vs_Objectif'] = out['cash_median'] - objectif_annuel out['Progression_Pct'] = ( (out['cash_median'] / objectif_annuel * 100) if objectif_annuel > 0 else float('nan') ) return out def semaines_jusqu_a_fin_exercice(annee: int, today=None) -> int: """Semaines a simuler pour couvrir le reste de l'exercice `annee` (0 si l'exercice est deja clos).""" today = today or date.today() fin = date(annee + 1, 1, 1) if fin <= today: return 0 return max(1, min(208, math.ceil((fin - today).days / 7))) def projection_fin_exercice(details: dict, resume: pd.DataFrame, annee: int, realise: float, objectif_annuel: float, base_accordee: bool = False) -> pd.DataFrame: """Projection du benefice de l'exercice `annee`, par scenario, comparee a l'objectif — sur la MEME base que le Dashboard et l'Optimizer : - base encaissee : realise (interets + penalites deja recus) + benefice simule (part interets des versements + penalites) des semaines qui tombent dans l'exercice ; - base accordee : le portefeuille existant est deja compte au deblocage, la projection = realise (le Twin ne simule pas de nouveaux prets). Les colonnes de tresorerie (cash_*) de `resume` sont conservees a titre indicatif.""" debut, fin = pd.Timestamp(annee, 1, 1), pd.Timestamp(annee + 1, 1, 1) lignes = [] for _, r in resume.iterrows(): runs = details.get(r['scenario'], pd.DataFrame()) if base_accordee or runs.empty or 'benefice' not in runs.columns: med = p10 = p90 = 0.0 else: dates = pd.to_datetime(runs['date']) par_run = runs[(dates >= debut) & (dates < fin)].groupby('run')['benefice'].sum() par_run = par_run.reindex(runs['run'].unique(), fill_value=0.0) med, p10, p90 = par_run.median(), par_run.quantile(0.10), par_run.quantile(0.90) projete = realise + med lignes.append({ **r.to_dict(), 'Benefice_Realise': round(realise, 0), 'Benefice_Simule_Reste_Exercice': round(med, 0), 'Benefice_Projete_Fin_Exercice': round(projete, 0), 'Benefice_Projete_p10': round(realise + p10, 0), 'Benefice_Projete_p90': round(realise + p90, 0), 'Objectif_Annuel': objectif_annuel, 'Ecart_vs_Objectif': round(projete - objectif_annuel, 0), 'Progression_Pct': (projete / objectif_annuel * 100) if objectif_annuel > 0 else float('nan'), }) return pd.DataFrame(lignes) def compare_scenarios(states, base_params, scenarios, horizon_weeks=26, n_runs=200, today=None, seed=42, elasticity=0.0): """ Execute plusieurs scenarios et renvoie (details, tableau comparatif). elasticity : active le couplage HYPOTHETIQUE penalite -> comportement (voir apply_penalty_coupling). 0.0 = desactive (defaut). """ details, resume = {}, [] for sc in scenarios: p = sc.apply(base_params) p = apply_penalty_coupling(p, elasticity) runs = run_simulation(states, p, horizon_weeks, n_runs, today, seed) details[sc.name] = runs final = runs[runs['period'] == horizon_weeks] resume.append({ 'scenario': sc.name, 'cash_median': round(final['cash_cumule'].median(), 0), 'cash_p10': round(final['cash_cumule'].quantile(0.10), 0), 'cash_p90': round(final['cash_cumule'].quantile(0.90), 0), 'penalites_total': round(runs.groupby('run')['penalites'].sum().median(), 0), 'defauts_moyen': round(runs.groupby('run')['n_defauts'].sum().mean(), 2), 'encours_perdu_moyen': round(runs.groupby('run')['encours_perdu'].first().mean(), 0), 'taux_hebdo_pct': round(p.taux_hebdo * 100, 2), 'p_late_effectif_pct': round(p.p_late * 100, 1), }) return details, pd.DataFrame(resume)