Spaces:
Running
Running
| """ | |
| 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) | |
| # ============================================================================= | |
| 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) | |
| 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 | |
| # ============================================================================= | |
| 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 | |
| # ============================================================================= | |
| 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) |