Spaces:
Running
Running
| """ | |
| DATA_DESCRIPTION.PY Unite 1 : Analyses Descriptives | |
| ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| Fonctions publiques : | |
| enrich_dx(sheets) -> 28 scores DX_* silencieux | |
| render_unit0(sheets) -> Integrite + completion matrix | |
| render_description(sheets, sec) -> visuels U1 (6+ par feuille) | |
| """ | |
| import re | |
| import streamlit as st | |
| import pandas as pd | |
| import numpy as np | |
| import plotly.graph_objects as go | |
| import plotly.express as px | |
| from datetime import datetime | |
| _C = { | |
| "bg": "#0a0e12", | |
| "card": "#0f141a", | |
| "border": "rgba(80,100,120,0.25)", | |
| "accent": "#58a6ff", | |
| "success": "#2ecc71", | |
| "warning": "#f39c12", | |
| "critical": "#c0392b", | |
| "neutral": "#5a6a7a", | |
| "text": "#a8b8c8", | |
| "subtext": "#5a6a7a", | |
| "grid": "rgba(80,100,120,0.12)", | |
| } | |
| _PAL = ["#58a6ff","#2ecc71","#f39c12","#c0392b","#a78bfa","#38bdf8","#fb7185","#34d399"] | |
| _FONT = "JetBrains Mono, Courier New, monospace" | |
| _BASE = dict( | |
| paper_bgcolor=_C["card"], plot_bgcolor=_C["card"], | |
| font=dict(color=_C["text"], family=_FONT, size=11), | |
| title_font=dict(size=12, color=_C["text"], family=_FONT), | |
| margin=dict(t=44, b=36, l=48, r=24), | |
| legend=dict(bgcolor=_C["bg"], bordercolor=_C["border"], borderwidth=1, | |
| font=dict(size=10, family=_FONT)), | |
| xaxis=dict(gridcolor=_C["grid"], zerolinecolor=_C["grid"], | |
| tickfont=dict(family=_FONT, size=10)), | |
| yaxis=dict(gridcolor=_C["grid"], zerolinecolor=_C["grid"], | |
| tickfont=dict(family=_FONT, size=10)), | |
| colorway=_PAL, | |
| ) | |
| SHEETS = ["Clients_KYC","Garants_KYC","Prets_Master", | |
| "Prets_Update","Remboursements","Ajustements_Echeances"] | |
| _CRIT_COLS = { | |
| "Clients_KYC": ["ID_Client","Revenus_Mensuels","Date_Creation"], | |
| "Garants_KYC": ["ID_Garant","Revenus_Mensuels","Verification_AML"], | |
| "Prets_Master": ["ID_Pret","ID_Client","Montant_Capital","Taux_Hebdo","Statut"], | |
| "Prets_Update": ["ID_Pret_Updated","ID_Pret","Date_Modification"], | |
| "Remboursements": ["ID_Transaction","ID_Pret","Montant_Verse","Date_Paiement"], | |
| "Ajustements_Echeances": ["ID_Ajustement","ID_Pret"], | |
| } | |
| _WARN_COLS = { | |
| "Clients_KYC": ["Employeur","Entite_Financiere","Numero_Fiscal"], | |
| "Garants_KYC": ["Employeur","Numero_Fiscal"], | |
| "Prets_Master": ["ID_Garant","Date_Update"], | |
| "Prets_Update": ["Commentaire_Modification","ID_Garant"], | |
| "Remboursements": ["Reference_Externe","Commentaire"], | |
| "Ajustements_Echeances": [], | |
| } | |
| _MOTIF_RISK = { | |
| "URGENCE_MEDICALE":0.85,"URGENCE MEDICALE":0.85, | |
| "PERTE_EMPLOI":0.78,"DETTE_ANTERIEURE":0.72, | |
| "CONSOMMATION":0.50,"AGRICULTURE":0.45,"EQUIPEMENT":0.40, | |
| "ACHAT D'EQUIPEMENT PERSONNEL":0.40,"COMMERCE":0.38, | |
| "COMMERCE / ACHAT DE STOCK":0.38,"LANCEMENT D'ACTIVITE":0.45, | |
| "LOGEMENT / HABITAT":0.42,"INVESTISSEMENT":0.35, | |
| "EDUCATION":0.30,"REPARATIONS":0.38, | |
| } | |
| # ── Helpers ─────────────────────────────────────────────────────────────────── | |
| def _L(**kw): | |
| m = {**_BASE} | |
| for ax in ("xaxis","yaxis"): | |
| if ax in kw: | |
| m[ax] = {**m[ax], **kw.pop(ax)} | |
| m.update(kw) | |
| return m | |
| def _ef(msg, h=220): | |
| fig = go.Figure() | |
| fig.add_annotation(text=msg, xref="paper", yref="paper", x=0.5, y=0.5, | |
| showarrow=False, font=dict(size=12, color=_C["subtext"], family=_FONT)) | |
| fig.update_layout(**_L(height=h)) | |
| return fig | |
| def _age_calc(series): | |
| today = datetime.today() | |
| def _a(v): | |
| try: | |
| d = pd.to_datetime(str(v), dayfirst=True, errors="coerce") | |
| return int((today - d).days / 365.25) if not pd.isna(d) else np.nan | |
| except Exception: | |
| return np.nan | |
| return series.apply(_a) | |
| def _safe(df, col, default=0): | |
| return df[col].fillna(default) if col in df.columns else pd.Series(default, index=df.index) | |
| def _motif_color(m): | |
| return _MOTIF_RISK.get(str(m).upper().strip(), 0.50) | |
| def _sub_title(txt): | |
| st.markdown( | |
| f'<div style="font-family:{_FONT};font-size:0.58rem;letter-spacing:2px;' | |
| f'text-transform:uppercase;color:{_C["subtext"]};margin:14px 0 6px 0;">' | |
| f'{txt}</div>', | |
| unsafe_allow_html=True, | |
| ) | |
| # ══════════════════════════════════════════════════════════════════════════════ | |
| # ENRICHISSEMENT DX_* (silencieux) | |
| # ══════════════════════════════════════════════════════════════════════════════ | |
| def enrich_dx(sheets: dict) -> dict: | |
| s = {k: df.copy() for k, df in sheets.items() if isinstance(df, pd.DataFrame)} | |
| cli = s.get("Clients_KYC", pd.DataFrame()) | |
| gar = s.get("Garants_KYC", pd.DataFrame()) | |
| prt = s.get("Prets_Master", pd.DataFrame()) | |
| upd = s.get("Prets_Update", pd.DataFrame()) | |
| rem = s.get("Remboursements", pd.DataFrame()) | |
| adj = s.get("Ajustements_Echeances", pd.DataFrame()) | |
| if not cli.empty: | |
| def _soc(v): | |
| if pd.isna(v) or not str(v).strip(): return 0 | |
| return min(sum(1 for p in re.split(r"[;,|\s]+", str(v)) if len(p.strip())>4), 6) | |
| cli["DX_Social_Completion_Index"] = _safe(cli,"Reseau_sociaux","").apply(_soc) | |
| rev_tot = _safe(cli,"Revenus_Mensuels") + _safe(cli,"Autres_Revenus") | |
| cli["DX_Living_Cost_Ratio"] = (_safe(cli,"Charges_Estimees")/rev_tot.replace(0,np.nan)).round(3) | |
| sw = {"CDI":1.0,"INDEPENDANT":0.75,"CDD":0.55,"INFORMEL":0.35,"SANS_EMPLOI":0.10} | |
| w_s = _safe(cli,"Statut_Pro","").map(lambda x: sw.get(str(x).upper(),0.5)) | |
| anc = _safe(cli,"Anciennete_Revenu").clip(upper=120)/120 | |
| cli["DX_Disposable_Income_Stability"] = (w_s*0.6+anc*0.4).round(3) | |
| q_p = cli.get("Quartier",pd.Series(dtype=str)).value_counts(normalize=True) | |
| cli["DX_Geographic_Risk_Entropy"] = cli.get("Quartier",pd.Series(dtype=str)).map(q_p).round(3) | |
| def _dig(row): | |
| sc=0; em=str(row.get("Email","")); ph=str(row.get("Telephone","")) | |
| sc += 1 if re.match(r"[^@]+@[^@]+\.[^@]+",em) else 0 | |
| sc += 1 if re.match(r"[\d\+][\d\s\-]{7,}",ph.strip()) else 0 | |
| sc += 1 if len(em)<60 and len(ph)<=20 else 0 | |
| return sc | |
| cli["DX_Digital_Accessibility_Grade"] = cli.apply(_dig, axis=1) | |
| cli["_age"] = _age_calc(cli.get("Date_Naissance",pd.Series(dtype=str))) | |
| sec_avg = cli.groupby("Secteur_Activite")["Revenus_Mensuels"].transform("mean") if "Secteur_Activite" in cli.columns else pd.Series(1,index=cli.index) | |
| cli["DX_Professional_Trajectory_Index"] = (_safe(cli,"Revenus_Mensuels")/(cli["_age"].fillna(35)*sec_avg.replace(0,np.nan)/1000)).round(3) | |
| cli.drop(columns=["_age"],inplace=True,errors="ignore") | |
| s["Clients_KYC"] = cli | |
| if not gar.empty: | |
| gar["DX_Guarantee_Leverage_Score"] = (_safe(gar,"Patrimoine_Declare")/_safe(gar,"Revenus_Mensuels",1).replace(0,np.nan)).round(2) | |
| rn = (_safe(gar,"Revenus_Mensuels")-_safe(gar,"Charges_Estimees")).replace(0,np.nan) | |
| gar["DX_Family_Pressure_Index"] = (_safe(gar,"Pers_Charge")/rn).round(3) | |
| if not prt.empty and "ID_Garant" in prt.columns: | |
| exp = prt["ID_Garant"].value_counts() | |
| gar["DX_Garant_Liability_Exposure"] = _safe(gar,"ID_Garant").map(exp).fillna(0).astype(int) | |
| else: | |
| gar["DX_Garant_Liability_Exposure"] = 0 | |
| if "Profession" in gar.columns: | |
| pat = _safe(gar,"Patrimoine_Declare") | |
| gar["DX_Net_Worth_Credibility_Check"] = (pat.groupby(gar["Profession"]).transform(lambda x:(x-x.mean())/x.std() if x.std()>0 else 0)).round(2) | |
| else: | |
| gar["DX_Net_Worth_Credibility_Check"] = 0.0 | |
| fast = {"MOBILE_MONEY","VIREMENT","WAVE","ORANGE_MONEY","MTN"} | |
| def _avail(row): | |
| mt=str(row.get("Moyen_Transfert","")).upper(); ef=str(row.get("Entite_Financiere","")).strip() | |
| return (1 if any(f in mt for f in fast) else 0)+(1 if ef and ef.upper() not in("NAN","NONE","") else 0) | |
| gar["DX_Garant_Availability_Score"] = gar.apply(_avail,axis=1) | |
| s["Garants_KYC"] = gar | |
| if not prt.empty: | |
| prt["DX_Interest_Yield_Factor"] = (_safe(prt,"Cout_Credit")/_safe(prt,"Montant_Capital",1).replace(0,np.nan)).round(4) | |
| prt["DX_Contract_Complexity_Index"] = (_safe(prt,"Nb_Versements")/_safe(prt,"Duree_Semaines",1).replace(0,np.nan)).round(3) | |
| if not cli.empty and "ID_Client" in cli.columns: | |
| rmap = cli.set_index("ID_Client")["Revenus_Mensuels"].to_dict() | |
| prt["_cr"] = _safe(prt,"ID_Client").map(rmap) | |
| prt["DX_Debt_to_Income_Pressure"] = (_safe(prt,"Montant_Versement")/prt["_cr"].replace(0,np.nan)).round(3) | |
| prt.drop(columns=["_cr"],inplace=True,errors="ignore") | |
| else: | |
| prt["DX_Debt_to_Income_Pressure"] = np.nan | |
| taux = _safe(prt,"Taux_Hebdo").astype(float) | |
| mu,sg = taux.mean(),taux.std() | |
| prt["DX_Yield_to_Frequency_Spread"] = ((taux-mu)/sg).round(2) if sg>0 else 0.0 | |
| prt["DX_Motif_Risk_Weight"] = _safe(prt,"Motif","").map(_motif_color) | |
| has_g = prt.get("ID_Garant",pd.Series(dtype=str)).notna().astype(float) if "ID_Garant" in prt.columns else pd.Series(0.0,index=prt.index) | |
| cap_n = _safe(prt,"Montant_Capital"); cap_mx = cap_n.max() | |
| prt["DX_Collateral_Coverage_Ratio"] = (has_g*cap_n/cap_mx).round(3) if cap_mx>0 else 0.0 | |
| s["Prets_Master"] = prt | |
| if not upd.empty: | |
| if not prt.empty and "ID_Pret" in prt.columns: | |
| mt_map = prt.set_index("ID_Pret")["Montant_Total"].to_dict() | |
| dc_map = prt.set_index("ID_Pret")["Date_Creation"].to_dict() | |
| upd["DX_Contract_Volatility_Delta"] = (_safe(upd,"Montant_Total")-_safe(upd,"ID_Pret").map(mt_map).fillna(0)).round(0) | |
| d0 = pd.to_datetime(_safe(upd,"ID_Pret").map(dc_map),errors="coerce") | |
| d1 = pd.to_datetime(upd.get("Date_Modification"),errors="coerce") | |
| upd["DX_Update_Velocity"] = (d1-d0).dt.days | |
| else: | |
| upd["DX_Contract_Volatility_Delta"] = np.nan | |
| upd["DX_Update_Velocity"] = np.nan | |
| upd["DX_Principal_Drift_Ratio"] = (_safe(upd,"Montant_Capital")/_safe(upd,"Duree_Semaines",1).replace(0,np.nan)).round(2) | |
| crisis={"crise","urgence","maladie","chomage","difficile","impaye","retard","deces","accident","perte","probleme"} | |
| def _grav(txt): | |
| if pd.isna(txt): return 0.0 | |
| return round(len(set(re.findall(r"\w+",str(txt).lower()))&crisis)/len(crisis),3) | |
| upd["DX_Renegotiation_Gravity_Score"] = upd.get("Commentaire_Modification",pd.Series(dtype=str)).apply(_grav) | |
| nb_v = upd.groupby("ID_Pret")["Version"].transform("max").fillna(1) if "ID_Pret" in upd.columns else pd.Series(1,index=upd.index) | |
| if not prt.empty and "Duree_Semaines" in prt.columns: | |
| dm = prt.set_index("ID_Pret")["Duree_Semaines"].to_dict() | |
| dur = _safe(upd,"ID_Pret").map(dm).fillna(1) | |
| else: | |
| dur = pd.Series(1,index=upd.index) | |
| upd["DX_Structural_Fragility_Index"] = (nb_v/dur.replace(0,np.nan)).round(4) | |
| s["Prets_Update"] = upd | |
| if not rem.empty: | |
| tot_int = _safe(rem,"Montant_Interets").sum() | |
| rem["DX_Profitability_Realization_Rate"] = (_safe(rem,"Montant_Interets")/max(tot_int,1)).round(4) | |
| dp = pd.to_datetime(rem.get("Date_Echeance_Prevue"),errors="coerce") | |
| dr = pd.to_datetime(rem.get("Date_Paiement"),errors="coerce") | |
| rem["DX_Collection_Efficiency_Delta"] = (dr-dp).dt.days | |
| rem["DX_Payment_Reliability_Variance"] = rem.groupby("ID_Client")["Jours_Retard"].transform(lambda x:x.std(ddof=0)).fillna(0).round(2) if "ID_Client" in rem.columns else 0.0 | |
| rem["DX_Overpayment_Tendency"] = (_safe(rem,"Montant_Verse")>_safe(rem,"Solde_Avant")).astype(int) | |
| rem["DX_Channel_Risk_Correlation"] = rem.groupby("Moyen_Paiement")["Jours_Retard"].transform("mean").round(2) if "Moyen_Paiement" in rem.columns else 0.0 | |
| def _hr(ts): | |
| try: return pd.to_datetime(str(ts),errors="coerce").hour | |
| except: return np.nan | |
| rem["DX_Temporal_Payment_Behavior"] = rem.get("Timestamp",pd.Series(dtype=str)).apply(_hr) | |
| s["Remboursements"] = rem | |
| if not adj.empty: | |
| moy_ech = _safe(prt,"Montant_Versement").mean() if not prt.empty else 1 | |
| adj["DX_Adjustment_Impact_Intensity"] = (_safe(adj,"Montant_Additionnel")/max(moy_ech,1)).round(3) | |
| adj["DX_Recurrence_Alert_Trigger"] = adj.groupby("ID_Pret").cumcount()+1 if "ID_Pret" in adj.columns else 1 | |
| if not prt.empty and "ID_Pret" in prt.columns and "Statut" in prt.columns: | |
| adj_ids = set(adj.get("ID_Pret",pd.Series(dtype=str)).dropna()) | |
| sp = prt[prt["ID_Pret"].isin(adj_ids)] | |
| adj["DX_Recovery_Probability_After_Adj"] = round(float((sp["Statut"].str.upper()=="TERMINE").mean()),3) | |
| else: | |
| adj["DX_Recovery_Probability_After_Adj"] = np.nan | |
| if not prt.empty and "Date_Fin" in prt.columns: | |
| fm = prt.set_index("ID_Pret")["Date_Fin"].to_dict() | |
| df_=pd.to_datetime(_safe(adj,"ID_Pret").map(fm),errors="coerce") | |
| da=pd.to_datetime(adj.get("Date_Creation"),errors="coerce") | |
| adj["DX_Adjustment_Timing_Sensitivity"] = (df_-da).dt.days | |
| else: | |
| adj["DX_Adjustment_Timing_Sensitivity"] = np.nan | |
| s["Ajustements_Echeances"] = adj | |
| return s | |
| # ══════════════════════════════════════════════════════════════════════════════ | |
| # UNITE 0 INTEGRITE | |
| # ══════════════════════════════════════════════════════════════════════════════ | |
| def _detect_issues(df, sheet): | |
| recs = [] | |
| if df.empty or not len(df.columns): return pd.DataFrame(columns=["FEUILLE","LIGNE","COLONNE","SEVERITE","TYPE","VALEUR"]) | |
| id_col = df.columns[0] | |
| for idx in df[df[id_col].duplicated(keep=False)].index: | |
| recs.append({"FEUILLE":sheet,"LIGNE":idx+2,"COLONNE":id_col,"SEVERITE":"CRITIQUE","TYPE":"DOUBLON_ID","VALEUR":str(df.loc[idx,id_col])}) | |
| for col in _CRIT_COLS.get(sheet,[]): | |
| if col not in df.columns: continue | |
| for idx in df[df[col].isnull()].index: | |
| recs.append({"FEUILLE":sheet,"LIGNE":idx+2,"COLONNE":col,"SEVERITE":"CRITIQUE","TYPE":"NULL_CRITIQUE","VALEUR":"NULL"}) | |
| for col in _WARN_COLS.get(sheet,[]): | |
| if col not in df.columns: continue | |
| for idx in df[df[col].isnull()].index: | |
| recs.append({"FEUILLE":sheet,"LIGNE":idx+2,"COLONNE":col,"SEVERITE":"AVERTISSEMENT","TYPE":"NULL_WARNING","VALEUR":"NULL"}) | |
| for col in ["Revenus_Mensuels","Montant_Capital","Taux_Hebdo","Montant_Verse","Jours_Retard"]: | |
| if col not in df.columns: continue | |
| for idx,val in df[col].items(): | |
| if pd.isna(val): continue | |
| if not isinstance(val,(int,float,np.integer,np.floating)): | |
| recs.append({"FEUILLE":sheet,"LIGNE":idx+2,"COLONNE":col,"SEVERITE":"ANOMALIE","TYPE":"TYPE_INCORRECT","VALEUR":str(val)}) | |
| return pd.DataFrame(recs,columns=["FEUILLE","LIGNE","COLONNE","SEVERITE","TYPE","VALEUR"]) | |
| def render_unit0(sheets: dict): | |
| total = sum(len(df) for df in sheets.values() if isinstance(df, pd.DataFrame)) | |
| empty = sum(1 for df in sheets.values() if isinstance(df, pd.DataFrame) if df.empty) | |
| iss = [] | |
| for name,df in sheets.items(): | |
| if not isinstance(df, pd.DataFrame): continue | |
| if not df.empty: iss.append(_detect_issues(df,name)) | |
| else: iss.append(pd.DataFrame([{"FEUILLE":name,"LIGNE":"-","COLONNE":"-","SEVERITE":"INFO","TYPE":"TABLE_VIDE","VALEUR":"0 ligne"}])) | |
| df_i = pd.concat(iss,ignore_index=True) | |
| nc = (df_i["SEVERITE"]=="CRITIQUE").sum() | |
| nw = (df_i["SEVERITE"]=="AVERTISSEMENT").sum() | |
| na = (df_i["SEVERITE"]=="ANOMALIE").sum() | |
| # KPI cards | |
| c1,c2,c3,c4,c5 = st.columns(5) | |
| for col,label,val,color in [ | |
| (c1,"LIGNES",total,_C["text"]), | |
| (c2,"CRITIQUES",nc,_C["critical"] if nc>0 else _C["success"]), | |
| (c3,"WARNINGS",nw,_C["warning"] if nw>0 else _C["success"]), | |
| (c4,"ANOMALIES",na,_C["warning"] if na>0 else _C["success"]), | |
| (c5,"VIDES",empty,_C["subtext"]), | |
| ]: | |
| with col: | |
| st.markdown( | |
| f'<div style="background:{_C["card"]};border:1px solid {_C["border"]};' | |
| f'border-left:3px solid {color};padding:10px 14px;font-family:{_FONT};">' | |
| f'<div style="font-size:0.50rem;letter-spacing:2px;color:{_C["subtext"]};' | |
| f'text-transform:uppercase;">{label}</div>' | |
| f'<div style="font-size:1.4rem;font-weight:700;color:{color};">{val}</div>' | |
| f'</div>', unsafe_allow_html=True) | |
| st.markdown("<br>", unsafe_allow_html=True) | |
| # A. Journal | |
| _sub_title("A · JOURNAL D'INTEGRITE CHIRURGICALE") | |
| sev_order = {"CRITIQUE":0,"ANOMALIE":1,"AVERTISSEMENT":2,"INFO":3} | |
| df_i["_s"] = df_i["SEVERITE"].map(sev_order).fillna(9) | |
| st.dataframe(df_i.sort_values("_s").drop(columns="_s"),use_container_width=True,hide_index=True) | |
| st.divider() | |
| # B. Matrice de completion globale — heatmap Plotly (une seule vue toutes feuilles) | |
| _sub_title("B · MATRICE DE COMPLETION (toutes feuilles)") | |
| # Construire une table de completion : lignes = colonnes des feuilles, colonnes = feuilles | |
| # On utilise uniquement les colonnes metier (pas DX_/EX_) | |
| pct_data = {} | |
| for sh_name, df in sheets.items(): | |
| if not isinstance(df, pd.DataFrame): continue | |
| if df.empty: | |
| pct_data[sh_name] = {} | |
| continue | |
| raw = df.loc[:, ~pd.Index(df.columns.astype(str)).str.startswith(("DX_","EX_"))] | |
| pct_data[sh_name] = ((1 - raw.isnull().mean()) * 100).round(1).to_dict() | |
| # Union de toutes les colonnes | |
| all_cols = [] | |
| for d in pct_data.values(): | |
| for c in d.keys(): | |
| if c not in all_cols: | |
| all_cols.append(c) | |
| sh_names = [sh for sh in SHEETS if isinstance(sheets.get(sh), __import__('pandas').DataFrame) and not sheets[sh].empty] | |
| # Matrice Z : rows=colonnes metier, cols=feuilles | |
| z_mat = [] | |
| for col in all_cols: | |
| row = [pct_data.get(sh,{}).get(col, np.nan) for sh in sh_names] | |
| z_mat.append(row) | |
| if z_mat and sh_names: | |
| # Texte dans chaque cellule | |
| text_mat = [ | |
| [f"{v:.0f}%" if not np.isnan(v) else "N/A" for v in row] | |
| for row in z_mat | |
| ] | |
| fig = go.Figure(go.Heatmap( | |
| z=z_mat, | |
| x=sh_names, | |
| y=all_cols, | |
| text=text_mat, | |
| texttemplate="%{text}", | |
| colorscale=[ | |
| [0.0, "#2d1010"], | |
| [0.3, "#7b2d2d"], | |
| [0.5, "#c0392b"], | |
| [0.7, "#f39c12"], | |
| [0.85, "#1a3a5c"], | |
| [1.0, "#2ecc71"], | |
| ], | |
| zmin=0, zmax=100, | |
| colorbar=dict( | |
| title="Completion %", | |
| tickvals=[0,25,50,75,100], | |
| ticktext=["0%","25%","50%","75%","100%"], | |
| tickfont=dict(size=9,family=_FONT), | |
| ), | |
| hovertemplate="<b>%{y}</b><br>%{x}<br>Completion : %{z:.1f}%<extra></extra>", | |
| xgap=2, ygap=1, | |
| )) | |
| fig.update_layout(**_L( | |
| height=max(400, len(all_cols)*20), | |
| xaxis=dict(side="top", tickangle=0, tickfont=dict(size=10,family=_FONT)), | |
| yaxis=dict(automargin=True, tickfont=dict(size=9,family=_FONT)), | |
| margin=dict(t=60,b=20,l=180,r=20), | |
| )) | |
| st.plotly_chart(fig, use_container_width=True) | |
| # Synthese par feuille (jauge compacte) | |
| st.markdown("<br>", unsafe_allow_html=True) | |
| _sub_title("SYNTHESE PAR FEUILLE") | |
| gauge_cols = st.columns(len(sh_names)) | |
| for i, sh_name in enumerate(sh_names): | |
| df = sheets[sh_name] | |
| raw = df.loc[:, ~pd.Index(df.columns.astype(str)).str.startswith(("DX_","EX_"))] | |
| overall = ((1 - raw.isnull().mean()).mean()*100) | |
| null_c = (raw.isnull().mean() > 0).sum() | |
| color = _C["success"] if overall>=90 else _C["accent"] if overall>=70 else _C["warning"] if overall>=50 else _C["critical"] | |
| with gauge_cols[i]: | |
| st.markdown( | |
| f'<div style="background:{_C["card"]};border:1px solid {_C["border"]};' | |
| f'border-top:3px solid {color};padding:8px 10px;font-family:{_FONT};text-align:center;">' | |
| f'<div style="font-size:0.48rem;letter-spacing:2px;color:{_C["subtext"]};' | |
| f'text-transform:uppercase;margin-bottom:4px;">{sh_name.replace("_"," ")}</div>' | |
| f'<div style="font-size:1.3rem;font-weight:700;color:{color};">{overall:.0f}%</div>' | |
| f'<div style="font-size:0.50rem;color:{_C["subtext"]};">{null_c} col. incompletes</div>' | |
| f'</div>', unsafe_allow_html=True) | |
| st.divider() | |
| # C. Registre brut | |
| with st.expander("REGISTRE BRUT · ACCES ARCHIVES", expanded=False): | |
| df_keys = [k for k, v in sheets.items() if isinstance(v, pd.DataFrame)] | |
| sel = st.selectbox("Feuille", df_keys, key="u0_raw_sel") | |
| df = sheets[sel] | |
| if not isinstance(df, pd.DataFrame) or df.empty: | |
| st.warning("Table vide.") | |
| else: | |
| show_enrich = st.checkbox("Afficher colonnes DX_*/EX_*", value=False, key="u0_raw_enrich") | |
| d = df if show_enrich else df.loc[:, ~pd.Index(df.columns.astype(str)).str.startswith(("DX_","EX_"))] | |
| st.caption(f"{len(d)} lignes · {len(d.columns)} colonnes") | |
| st.dataframe(d, use_container_width=True) | |
| # D. Lineage | |
| with st.expander("DATA LINEAGE · RELATIONS INTER-FEUILLES", expanded=False): | |
| st.markdown(f""" | |
| <div style="font-family:{_FONT};font-size:0.65rem;color:{_C['text']};line-height:1.9;"> | |
| <b style="color:{_C['accent']};">Prets_Update</b> est la continuité historique de | |
| <b style="color:{_C['accent']};">Prets_Master</b>.<br> | |
| Chaque ligne représente une version modifiée d'un contrat.<br><br> | |
| <span style="color:{_C['subtext']};"> | |
| Clients_KYC (11) ── Prets_Master via ID_Client<br> | |
| Garants_KYC (2) ── Prets_Master via ID_Garant (11/14 nulls = prêts sans garant, normal)<br> | |
| Prets_Master (14) ── Prets_Update (6 lignes, 5 prêts modifiés)<br> | |
| Prets_Master (14) ── Remboursements via ID_Pret<br> | |
| Prets_Master (14) ── Ajustements via ID_Pret (table vide actuellement) | |
| </span></div> | |
| """, unsafe_allow_html=True) | |
| # ══════════════════════════════════════════════════════════════════════════════ | |
| # VISUELS U1 (6+ par feuille) | |
| # ══════════════════════════════════════════════════════════════════════════════ | |
| def _render_clients(df, _sheets): | |
| if df.empty: st.info("Table Clients_KYC vide."); return | |
| c1, c2 = st.columns(2) | |
| # 1. Pyramide demographique | |
| with c1: | |
| df2 = df.copy(); df2["Age"] = _age_calc(df2.get("Date_Naissance",pd.Series(dtype=str))); df2 = df2.dropna(subset=["Age"]) | |
| if not df2.empty: | |
| bins=[0,25,35,45,55,65,120]; labs=["<25","25-34","35-44","45-54","55-64","65+"] | |
| df2["T"] = pd.cut(df2["Age"],bins=bins,labels=labs,right=False) | |
| fig = go.Figure() | |
| for i,g in enumerate(df2["Genre"].dropna().unique()): | |
| sub=df2[df2["Genre"]==g]; cnt=sub["T"].value_counts().reindex(labs,fill_value=0); sgn=-1 if i==0 else 1 | |
| fig.add_trace(go.Bar(name=str(g),y=labs,x=cnt.values*sgn,orientation="h",marker_color=_PAL[i], | |
| customdata=np.abs(cnt.values),hovertemplate=f"<b>{g}</b> %{{y}} : %{{customdata}}<extra></extra>")) | |
| fig.update_layout(**_L(title="PYRAMIDE DEMOGRAPHIQUE",barmode="relative",height=300)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 2. Revenus vs anciennete (taille = LCR) | |
| with c2: | |
| fig = go.Figure() | |
| for i,st_v in enumerate(df.get("Statut_Pro",pd.Series()).dropna().unique()): | |
| sub = df[df["Statut_Pro"]==st_v] | |
| lcr = sub.get("DX_Living_Cost_Ratio",pd.Series(0.5,index=sub.index)).fillna(0.5) | |
| fig.add_trace(go.Scatter(x=sub["Revenus_Mensuels"],y=sub["Anciennete_Emploi"],mode="markers",name=str(st_v), | |
| marker=dict(size=(lcr.clip(0,1)*14+6).tolist(),color=_PAL[i%len(_PAL)],line=dict(width=1,color=_C["bg"]),sizemode="diameter"), | |
| text=sub.get("Nom_Complet",sub.index).astype(str),customdata=lcr.round(2), | |
| hovertemplate="<b>%{text}</b><br>%{x:,} FCFA %{y} mois<br>LCR : %{customdata}<extra></extra>")) | |
| fig.update_layout(**_L(title="REVENUS x ANCIENNETE (taille = LCR)",height=300,xaxis_title="Revenus",yaxis_title="Anciennete (mois)")) | |
| st.plotly_chart(fig,use_container_width=True) | |
| c3, c4 = st.columns(2) | |
| # 3. Secteurs (couleur = Trajectory Index) | |
| with c3: | |
| if "Secteur_Activite" in df.columns: | |
| grp = df.groupby("Secteur_Activite").agg(n=("Secteur_Activite","count")).sort_values("n") | |
| pti = df.groupby("Secteur_Activite")["DX_Professional_Trajectory_Index"].mean() if "DX_Professional_Trajectory_Index" in df.columns else pd.Series(1.0,index=grp.index) | |
| bc = [_C["success"] if pti.get(idx,1)>1.2 else _C["warning"] if pti.get(idx,1)<0.8 else _C["accent"] for idx in grp.index] | |
| fig = go.Figure(go.Bar(x=grp["n"],y=grp.index.astype(str),orientation="h",marker_color=bc,text=grp["n"],textposition="outside")) | |
| fig.update_layout(**_L(title="SECTEURS (couleur = Trajectory Index)",height=max(260,len(grp)*32),showlegend=False)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 4. Origine des fonds | |
| with c4: | |
| if "Origine_Fonds" in df.columns: | |
| cnt = df["Origine_Fonds"].value_counts() | |
| fig = go.Figure(go.Pie(labels=cnt.index.astype(str),values=cnt.values,hole=0.55,marker_colors=_PAL)) | |
| fig.update_layout(**_L(title="ORIGINE DES FONDS",height=280)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| c5, c6 = st.columns(2) | |
| # 5. Box revenus par Statut_Pro | |
| with c5: | |
| if "Statut_Pro" in df.columns and "Revenus_Mensuels" in df.columns: | |
| fig = go.Figure() | |
| for i,sp in enumerate(df["Statut_Pro"].dropna().unique()): | |
| sub = df[df["Statut_Pro"]==sp]["Revenus_Mensuels"].dropna() | |
| fig.add_trace(go.Box(y=sub,name=str(sp),marker_color=_PAL[i%len(_PAL)],boxpoints="all",jitter=0.35,pointpos=0)) | |
| fig.update_layout(**_L(title="DISTRIBUTION REVENUS PAR STATUT PRO",height=300,showlegend=False,yaxis_title="Revenus (FCFA)")) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 6. Scatter patrimoine vs revenus (couleur = Statut_Logement) | |
| with c6: | |
| if "Patrimoine_Declare" in df.columns: | |
| lgt_map = {"Locataire":_C["warning"],"Proprietaire":_C["success"],"Heberge":_C["accent"],"Hébergé":_C["accent"]} | |
| fig = go.Figure() | |
| lgt_col = df.get("Statut_Logement",pd.Series("Autre",index=df.index)).fillna("Autre") | |
| for lval in lgt_col.unique(): | |
| mask = lgt_col==lval; sub = df[mask] | |
| fig.add_trace(go.Scatter(x=sub["Revenus_Mensuels"],y=sub["Patrimoine_Declare"],mode="markers",name=str(lval), | |
| marker=dict(size=11,color=lgt_map.get(str(lval),_C["neutral"]),line=dict(width=1,color=_C["bg"])), | |
| text=sub.get("Nom_Complet",sub.index).astype(str), | |
| hovertemplate="<b>%{text}</b><br>Rev : %{x:,}<br>Patr : %{y:,}<extra></extra>")) | |
| fig.update_layout(**_L(title="PATRIMOINE vs REVENUS (couleur = logement)",height=300,xaxis_title="Revenus",yaxis_title="Patrimoine")) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 7. Bar distribution Statut_Logement + Etat_Civil (grouped) | |
| if "Statut_Logement" in df.columns and "Etat_Civil" in df.columns: | |
| cross = df.groupby(["Etat_Civil","Statut_Logement"]).size().reset_index(name="n") | |
| fig = go.Figure() | |
| for i,ec in enumerate(cross["Etat_Civil"].unique()): | |
| sub = cross[cross["Etat_Civil"]==ec] | |
| fig.add_trace(go.Bar(name=str(ec),x=sub["Statut_Logement"],y=sub["n"],marker_color=_PAL[i%len(_PAL)],text=sub["n"],textposition="outside")) | |
| fig.update_layout(**_L(title="ETAT CIVIL x STATUT LOGEMENT",barmode="group",height=260)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| def _render_garants(df, _sheets): | |
| if df.empty: st.info("Table Garants_KYC vide."); return | |
| c1, c2 = st.columns(2) | |
| # 1. Scatter capacite (leverage) | |
| with c1: | |
| fig = go.Figure() | |
| for i,pr in enumerate(df.get("Profession",pd.Series()).dropna().unique()): | |
| sub = df[df["Profession"]==pr] | |
| lev = sub.get("DX_Guarantee_Leverage_Score",pd.Series(1,index=sub.index)).fillna(1).clip(0,50) | |
| mx = lev.max(); sz = (lev/mx*14+8).tolist() if mx>0 else [10]*len(sub) | |
| fig.add_trace(go.Scatter(x=sub["Revenus_Mensuels"],y=sub["Patrimoine_Declare"],mode="markers+text",name=str(pr), | |
| text=sub.get("Nom_Complet",sub.index).astype(str),textposition="top center", | |
| marker=dict(size=sz,color=_PAL[i%len(_PAL)],line=dict(width=1.5,color=_C["bg"])), | |
| customdata=lev.round(1),hovertemplate="<b>%{text}</b><br>%{x:,} %{y:,}<br>Leverage : %{customdata}x<extra></extra>")) | |
| fig.update_layout(**_L(title="MATRICE CAPACITE (taille = Leverage)",height=300,xaxis_title="Revenus",yaxis_title="Patrimoine")) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 2. Pie AML | |
| with c2: | |
| if "Verification_AML" in df.columns: | |
| cnt = df["Verification_AML"].value_counts() | |
| aml = {"VERIFIE":_C["success"],"EN_COURS":_C["warning"],"REJETE":_C["critical"]} | |
| fig = go.Figure(go.Pie(labels=cnt.index.astype(str),values=cnt.values, | |
| marker_colors=[aml.get(str(l).upper(),_C["neutral"]) for l in cnt.index])) | |
| fig.update_layout(**_L(title="VERIFICATION AML",height=300)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| c3, c4 = st.columns(2) | |
| # 3. Bar entites financieres | |
| with c3: | |
| if "Entite_Financiere" in df.columns: | |
| cnt = df["Entite_Financiere"].value_counts() | |
| fig = go.Figure(go.Bar(x=cnt.values,y=cnt.index.astype(str),orientation="h", | |
| marker_color=_PAL[2],text=cnt.values,textposition="outside")) | |
| fig.update_layout(**_L(title="ENTITES FINANCIERES",height=max(220,len(cnt)*36),showlegend=False)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 4. Scatter checking x revenus (couleur = NW Credibility) | |
| with c4: | |
| if "Checking_frequency" in df.columns: | |
| cred = df.get("DX_Net_Worth_Credibility_Check",pd.Series(0.0,index=df.index)).fillna(0) | |
| fig = go.Figure(go.Scatter(x=df["Checking_frequency"],y=df["Revenus_Mensuels"],mode="markers", | |
| marker=dict(size=12,color=cred,colorscale=[[0,"#c0392b"],[0.5,"#58a6ff"],[1,"#2ecc71"]], | |
| showscale=True,colorbar=dict(title="NW Cred.",tickfont=dict(size=9))), | |
| text=df.get("Nom_Complet",df.index).astype(str), | |
| hovertemplate="<b>%{text}</b><br>Freq : %{x}<br>Rev : %{y:,}<extra></extra>")) | |
| fig.update_layout(**_L(title="CHECKING x REVENUS (couleur = NW Credibility)",height=300,showlegend=False)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| c5, c6 = st.columns(2) | |
| # 5. Bar revenus vs charges (comparatif) | |
| with c5: | |
| fig = go.Figure() | |
| noms = df.get("Nom_Complet", df.index.astype(str)) | |
| fig.add_trace(go.Bar(name="Revenus",x=noms.tolist(),y=_safe(df,"Revenus_Mensuels").tolist(),marker_color=_C["success"])) | |
| fig.add_trace(go.Bar(name="Charges",x=noms.tolist(),y=_safe(df,"Charges_Estimees").tolist(),marker_color=_C["critical"])) | |
| fig.update_layout(**_L(title="REVENUS vs CHARGES",barmode="group",height=280,yaxis_title="FCFA")) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 6. Bar pression familiale (Pers_Charge / revenu net) | |
| with c6: | |
| disp = (_safe(df,"Revenus_Mensuels") - _safe(df,"Charges_Estimees")).round(0) | |
| noms = df.get("Nom_Complet",df.index.astype(str)) | |
| fig = go.Figure() | |
| fig.add_trace(go.Bar(name="Revenu disponible",x=noms.tolist(),y=disp.tolist(), | |
| marker_color=[_C["success"] if v>0 else _C["critical"] for v in disp],text=disp.round(0).tolist(),textposition="outside")) | |
| fp = df.get("DX_Family_Pressure_Index",pd.Series(0,index=df.index)).fillna(0) | |
| for i,(nom,val) in enumerate(zip(noms,fp)): | |
| fig.add_annotation(x=str(nom),y=0,text=f"FP:{val:.2f}",showarrow=False,yshift=-18, | |
| font=dict(size=9,color=_C["subtext"],family=_FONT)) | |
| fig.update_layout(**_L(title="REVENU DISPONIBLE (FP = Family Pressure Index)",height=280)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| def _render_prets(df, _sheets): | |
| if df.empty: st.info("Table Prets_Master vide."); return | |
| c1, c2 = st.columns(2) | |
| # 1. Architecture portefeuille | |
| with c1: | |
| cross = df.groupby(["Offre","Type_Pret"]).size().reset_index(name="n") | |
| fig = go.Figure() | |
| for i,tp in enumerate(cross["Type_Pret"].unique()): | |
| sub = cross[cross["Type_Pret"]==tp] | |
| fig.add_trace(go.Bar(name=str(tp),x=sub["Offre"],y=sub["n"],marker_color=_PAL[i%len(_PAL)])) | |
| fig.update_layout(**_L(title="ARCHITECTURE PORTEFEUILLE",barmode="stack",height=290)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 2. Spectre capital (ligne = moy ponderee risque) | |
| with c2: | |
| mrw = df.get("DX_Motif_Risk_Weight",pd.Series(0.5,index=df.index)).fillna(0.5) | |
| avg = (df["Montant_Capital"].fillna(0)*mrw).sum()/mrw.sum() | |
| fig = go.Figure(go.Histogram(x=df["Montant_Capital"].dropna(),nbinsx=12,marker_color=_PAL[0])) | |
| fig.add_vline(x=avg,line_dash="dash",line_color=_C["warning"], | |
| annotation_text=f"Moy. ponderee : {avg:,.0f}",annotation_font_size=9,annotation_font_color=_C["warning"]) | |
| fig.update_layout(**_L(title="SPECTRE CAPITAL (ligne = moy. ponderee risque)",height=290,showlegend=False)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| c3, c4 = st.columns(2) | |
| # 3. Scatter taux x duree (taille = DtI) | |
| with c3: | |
| fig = go.Figure() | |
| for i,of in enumerate(df.get("Offre",pd.Series()).dropna().unique()): | |
| sub = df[df["Offre"]==of] | |
| dti = sub.get("DX_Debt_to_Income_Pressure",pd.Series(0.3,index=sub.index)).fillna(0.3).clip(0,1) | |
| fig.add_trace(go.Scatter(x=sub["Taux_Hebdo"],y=sub["Duree_Semaines"],mode="markers",name=str(of), | |
| marker=dict(size=(dti*20+6).tolist(),color=_PAL[i%len(_PAL)],line=dict(width=1,color=_C["bg"])), | |
| text=sub["ID_Pret"].astype(str),customdata=dti.round(3), | |
| hovertemplate="<b>%{text}</b><br>Taux %{x:.2%} %{y} sem.<br>DtI : %{customdata}<extra></extra>")) | |
| fig.update_layout(**_L(title="TAUX x DUREE (taille = DtI)",height=290, | |
| xaxis=dict(tickformat=".1%"),xaxis_title="Taux Hebdo",yaxis_title="Duree (sem.)")) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 4. Treemap motifs (couleur = Motif Risk Weight) | |
| with c4: | |
| cnt = df["Motif"].value_counts().reset_index(); cnt.columns=["Motif","n"] | |
| cnt["risk"] = cnt["Motif"].map(_motif_color) | |
| fig = px.treemap(cnt,path=["Motif"],values="n",color="risk", | |
| color_continuous_scale=[[0,"#2ecc71"],[0.5,"#f39c12"],[1,"#c0392b"]],range_color=[0,1]) | |
| fig.update_traces(hovertemplate="<b>%{label}</b><br>Nb : %{value}<br>Risque : %{color:.2f}<extra></extra>",textfont_color="#a8b8c8") | |
| fig.update_layout(**_L(title="MOTIFS (couleur = Motif Risk Weight)",height=290)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| c5, c6 = st.columns(2) | |
| # 5. Sante du front | |
| with c5: | |
| cross = df.groupby(["Offre","Statut"]).size().reset_index(name="n") | |
| sc = {"ACTIF":_C["success"],"TERMINE":_C["accent"],"UPDATED":_C["warning"],"EN_RETARD":_C["warning"],"DEFAUT":_C["critical"]} | |
| fig = go.Figure() | |
| for sv in cross["Statut"].unique(): | |
| sub = cross[cross["Statut"]==sv] | |
| fig.add_trace(go.Bar(name=str(sv),x=sub["Offre"],y=sub["n"],marker_color=sc.get(str(sv).upper(),_C["neutral"]))) | |
| fig.update_layout(**_L(title="SANTE DU FRONT",barmode="stack",height=260)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 6. Box taux_endettement par offre | |
| with c6: | |
| if "Taux_Endettement" in df.columns: | |
| fig = go.Figure() | |
| for i,of in enumerate(df["Offre"].dropna().unique()): | |
| sub = df[df["Offre"]==of]["Taux_Endettement"].dropna() | |
| fig.add_trace(go.Box(y=sub,name=str(of),marker_color=_PAL[i%len(_PAL)],boxpoints="all",jitter=0.3)) | |
| fig.add_hline(y=40,line_dash="dot",line_color=_C["critical"], | |
| annotation_text="Seuil critique 40%",annotation_font_size=9,annotation_font_color=_C["critical"]) | |
| fig.update_layout(**_L(title="TAUX ENDETTEMENT PAR OFFRE",height=260,showlegend=False,yaxis_title="Taux (%)")) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 7. Scatter capital vs cout_credit | |
| if "Cout_Credit" in df.columns: | |
| fig = go.Figure() | |
| for i,of in enumerate(df.get("Offre",pd.Series()).dropna().unique()): | |
| sub = df[df["Offre"]==of] | |
| yf = sub.get("DX_Interest_Yield_Factor",pd.Series(0,index=sub.index)).fillna(0) | |
| fig.add_trace(go.Scatter(x=sub["Montant_Capital"],y=sub["Cout_Credit"],mode="markers",name=str(of), | |
| marker=dict(size=10,color=_PAL[i%len(_PAL)],line=dict(width=1,color=_C["bg"])), | |
| text=sub["ID_Pret"].astype(str),customdata=yf.round(3), | |
| hovertemplate="<b>%{text}</b><br>Capital : %{x:,}<br>Cout : %{y:,}<br>Yield : %{customdata}<extra></extra>")) | |
| fig.update_layout(**_L(title="CAPITAL vs COUT CREDIT (rendement par point)",height=260,xaxis_title="Capital",yaxis_title="Cout Credit")) | |
| st.plotly_chart(fig,use_container_width=True) | |
| def _render_prets_update(df, _sheets): | |
| if df.empty: st.info("Table Prets_Update vide."); return | |
| prt = _sheets.get("Prets_Master",pd.DataFrame()) | |
| c1, c2 = st.columns(2) | |
| # 1. Temporalite des revisions | |
| with c1: | |
| d = df.copy(); d["Date_Modification"] = pd.to_datetime(d.get("Date_Modification"),errors="coerce") | |
| d["Mois"] = d["Date_Modification"].dt.to_period("M").astype(str) | |
| cnt = d["Mois"].value_counts().sort_index() | |
| gm = d.groupby("Mois")["DX_Renegotiation_Gravity_Score"].mean() if "DX_Renegotiation_Gravity_Score" in d.columns else pd.Series(0.0,index=cnt.index) | |
| bc = [_C["critical"] if gm.get(m,0)>0.20 else _C["warning"] if gm.get(m,0)>0.05 else _C["accent"] for m in cnt.index] | |
| fig = go.Figure(go.Bar(x=cnt.index,y=cnt.values,marker_color=bc,text=cnt.values,textposition="outside")) | |
| fig.update_layout(**_L(title="TEMPORALITE DES REVISIONS (couleur = gravite)",height=280,showlegend=False)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 2. Versions par pret | |
| with c2: | |
| cnt2 = df.groupby("ID_Pret")["ID_Pret_Updated"].count().sort_values(ascending=False) | |
| bc2 = [_C["accent"] if v==1 else _C["warning"] if v==2 else _C["critical"] for v in cnt2.values] | |
| fig = go.Figure(go.Bar(x=cnt2.index.astype(str),y=cnt2.values,marker_color=bc2,text=cnt2.values,textposition="outside")) | |
| fig.update_layout(**_L(title="MUTATIONS PAR PRET",height=280,showlegend=False)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| c3, c4 = st.columns(2) | |
| # 3. Scatter delta capital vs velocity | |
| with c3: | |
| if "DX_Contract_Volatility_Delta" in df.columns and "DX_Update_Velocity" in df.columns: | |
| fig = go.Figure(go.Scatter(x=df["DX_Update_Velocity"],y=df["DX_Contract_Volatility_Delta"],mode="markers", | |
| marker=dict(size=12,color=[_C["critical"] if v>0 else _C["success"] for v in df["DX_Contract_Volatility_Delta"]], | |
| line=dict(width=1.5,color=_C["bg"])), | |
| text=df["ID_Pret"].astype(str) if "ID_Pret" in df.columns else df.index.astype(str), | |
| hovertemplate="<b>%{text}</b><br>Velocity : %{x}j<br>Delta : %{y:+,.0f}<extra></extra>")) | |
| fig.add_hline(y=0,line_dash="dot",line_color=_C["neutral"],line_width=1) | |
| fig.update_layout(**_L(title="DELTA CONTRAT vs VELOCITY (rouge = montant hausse)",height=280, | |
| xaxis_title="Velocity (jours depuis creation)",yaxis_title="Delta Montant Total",showlegend=False)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 4. Bar gravite reneg par pret | |
| with c4: | |
| if "DX_Renegotiation_Gravity_Score" in df.columns: | |
| fig = go.Figure(go.Bar(x=df["ID_Pret_Updated"].astype(str),y=df["DX_Renegotiation_Gravity_Score"], | |
| marker_color=[_C["critical"] if v>0.1 else _C["accent"] for v in df["DX_Renegotiation_Gravity_Score"]], | |
| text=df["DX_Renegotiation_Gravity_Score"].round(3),textposition="outside")) | |
| fig.update_layout(**_L(title="GRAVITE RENEGOCIATION PAR MODIFICATION",height=280,showlegend=False)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| c5, c6 = st.columns(2) | |
| # 5. Evolution taux_hebdo par pret (comparaison master vs update) | |
| with c5: | |
| if not prt.empty and "ID_Pret" in prt.columns and "Taux_Hebdo" in prt.columns and "Taux_Hebdo" in df.columns: | |
| taux_orig = prt.set_index("ID_Pret")["Taux_Hebdo"].to_dict() | |
| fig = go.Figure() | |
| for pid in df["ID_Pret"].dropna().unique(): | |
| sub = df[df["ID_Pret"]==pid].sort_values("Version") | |
| orig= taux_orig.get(pid, np.nan) | |
| if not np.isnan(orig): | |
| vals = [orig] + sub["Taux_Hebdo"].tolist() | |
| vers = ["V1"] + [f"V{v}" for v in sub["Version"]] | |
| col = _C["critical"] if vals[-1]>orig else _C["success"] | |
| fig.add_trace(go.Scatter(x=vers,y=vals,mode="lines+markers",name=str(pid), | |
| line=dict(color=col,width=1.5),marker=dict(size=6))) | |
| fig.update_layout(**_L(title="EVOLUTION TAUX HEBDO PAR VERSION",height=280,yaxis=dict(tickformat=".2%"))) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 6. Distribution fragilite structurelle | |
| with c6: | |
| if "DX_Structural_Fragility_Index" in df.columns: | |
| fig = go.Figure(go.Bar(x=df["ID_Pret"].astype(str) if "ID_Pret" in df.columns else df.index.astype(str), | |
| y=df["DX_Structural_Fragility_Index"], | |
| marker_color=[_C["critical"] if v>0.15 else _C["warning"] if v>0.08 else _C["accent"] for v in df["DX_Structural_Fragility_Index"]], | |
| text=df["DX_Structural_Fragility_Index"].round(4),textposition="outside")) | |
| fig.update_layout(**_L(title="FRAGILITE STRUCTURELLE PAR PRET",height=280,showlegend=False)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| def _render_remboursements(df, _sheets): | |
| if df.empty: st.info("Table Remboursements vide."); return | |
| # 1. Discipline de paiement | |
| cross = df.groupby(["Moyen_Paiement","Statut_Paiement"]).size().reset_index(name="n") | |
| sp_c = {"EN_RETARD":_C["critical"],"PONCTUEL":_C["success"],"ANTICIPE":_C["accent"],"PAYE":_C["success"],"PARTIEL":_C["warning"]} | |
| fig = go.Figure() | |
| for sp in cross["Statut_Paiement"].unique(): | |
| sub = cross[cross["Statut_Paiement"]==sp] | |
| fig.add_trace(go.Bar(name=str(sp),x=sub["Moyen_Paiement"],y=sub["n"],marker_color=sp_c.get(str(sp).upper(),_C["neutral"]))) | |
| if "DX_Channel_Risk_Correlation" in df.columns: | |
| for canal,val in df.groupby("Moyen_Paiement")["DX_Channel_Risk_Correlation"].mean().items(): | |
| total_c = cross[cross["Moyen_Paiement"]==canal]["n"].sum() | |
| fig.add_annotation(x=str(canal),y=total_c,text=f"retard moy. {val:.1f}j",showarrow=False,yshift=10, | |
| font=dict(size=9,color=_C["warning"],family=_FONT)) | |
| fig.update_layout(**_L(title="DISCIPLINE DE PAIEMENT",barmode="stack",height=270)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| c1, c2 = st.columns(2) | |
| # 2. Histogramme delais | |
| with c1: | |
| fig = go.Figure(go.Histogram(x=df["Jours_Retard"].dropna(),nbinsx=14,marker_color=_C["warning"])) | |
| if "DX_Payment_Reliability_Variance" in df.columns: | |
| vm = df["DX_Payment_Reliability_Variance"].mean() | |
| fig.add_vline(x=vm,line_dash="dot",line_color=_C["critical"], | |
| annotation_text=f"Variance moy. : {vm:.1f}j",annotation_font_size=9,annotation_font_color=_C["critical"]) | |
| fig.update_layout(**_L(title="SPECTRE DES DELAIS (ligne = Payment Variance)",height=260,showlegend=False)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 3. Flux tresorerie | |
| with c2: | |
| fig = go.Figure(go.Histogram(x=df["Montant_Verse"].dropna(),nbinsx=12,marker_color=_C["success"])) | |
| if "DX_Overpayment_Tendency" in df.columns: | |
| n_ov = int(df["DX_Overpayment_Tendency"].sum()) | |
| fig.add_annotation(xref="paper",yref="paper",x=0.97,y=0.95,text=f"Paiements anticipes : {n_ov}", | |
| showarrow=False,font=dict(size=10,color=_C["accent"],family=_FONT),align="right") | |
| fig.update_layout(**_L(title="FLUX TRESORERIE",height=260,showlegend=False)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| c3, c4 = st.columns(2) | |
| # 4. Scatter montant_verse vs solde_avant | |
| with c3: | |
| if "Solde_Avant" in df.columns: | |
| sp_c2 = {"EN_RETARD":_C["critical"],"PONCTUEL":_C["success"],"ANTICIPE":_C["accent"]} | |
| fig = go.Figure() | |
| for sp in df["Statut_Paiement"].dropna().unique(): | |
| mask = df["Statut_Paiement"]==sp; sub = df[mask] | |
| fig.add_trace(go.Scatter(x=sub["Solde_Avant"],y=sub["Montant_Verse"],mode="markers",name=str(sp), | |
| marker=dict(size=10,color=sp_c2.get(str(sp).upper(),_C["neutral"]),line=dict(width=1,color=_C["bg"])), | |
| hovertemplate=f"<b>{sp}</b><br>Solde : %{{x:,}}<br>Verse : %{{y:,}}<extra></extra>")) | |
| # Ligne diagonale ideale (verse = solde -> solde final 0) | |
| mx = max(df["Solde_Avant"].max(),df["Montant_Verse"].max())*1.05 | |
| fig.add_trace(go.Scatter(x=[0,mx],y=[0,mx],mode="lines",line=dict(color=_C["subtext"],dash="dot",width=1),showlegend=False,hoverinfo="skip")) | |
| fig.update_layout(**_L(title="MONTANT VERSE vs SOLDE AVANT",height=260,xaxis_title="Solde Avant",yaxis_title="Montant Verse")) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 5. Performance par canal (% paiements ponctuels) | |
| with c4: | |
| if "Moyen_Paiement" in df.columns and "Statut_Paiement" in df.columns: | |
| good = df["Statut_Paiement"].str.upper().isin(["PAYE","PONCTUEL","ANTICIPE"]) | |
| perf = df.groupby("Moyen_Paiement").apply(lambda g: good.reindex(g.index).sum()/len(g)*100).round(1) | |
| fig = go.Figure(go.Bar(x=perf.index.astype(str),y=perf.values, | |
| marker_color=[_C["success"] if v>=80 else _C["warning"] if v>=50 else _C["critical"] for v in perf.values], | |
| text=[f"{v:.1f}%" for v in perf.values],textposition="outside")) | |
| fig.update_layout(**_L(title="TAUX PONCTUALITE PAR CANAL",height=260,showlegend=False,yaxis_title="%")) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 6. Timeline chronologique des paiements | |
| if "Date_Paiement" in df.columns: | |
| d2 = df.copy(); d2["Date_Paiement"] = pd.to_datetime(d2["Date_Paiement"],errors="coerce"); d2 = d2.dropna(subset=["Date_Paiement"]) | |
| if not d2.empty: | |
| sp_c3 = {"EN_RETARD":_C["critical"],"PONCTUEL":_C["success"],"ANTICIPE":_C["accent"],"PAYE":_C["success"]} | |
| fig = go.Figure() | |
| for sp in d2["Statut_Paiement"].dropna().unique(): | |
| mask = d2["Statut_Paiement"]==sp; sub = d2[mask] | |
| fig.add_trace(go.Scatter(x=sub["Date_Paiement"],y=sub["Montant_Verse"],mode="markers",name=str(sp), | |
| marker=dict(size=10,color=sp_c3.get(str(sp).upper(),_C["neutral"]),line=dict(width=1.5,color=_C["bg"])), | |
| text=sub["ID_Pret"].astype(str),hovertemplate="<b>%{text}</b><br>%{x|%d/%m/%Y}<br>%{y:,}<extra></extra>")) | |
| fig.update_layout(**_L(title="TIMELINE DES PAIEMENTS",height=260,xaxis_title="Date",yaxis_title="Montant Verse")) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # 7. Box distribution jours_retard | |
| if "Jours_Retard" in df.columns: | |
| fig = go.Figure(go.Box(y=df["Jours_Retard"].dropna(),marker_color=_C["warning"], | |
| boxpoints="all",jitter=0.4,pointpos=0, | |
| hovertemplate="Jours retard : %{y}<extra></extra>")) | |
| fig.add_hline(y=0,line_dash="dot",line_color=_C["neutral"],line_width=1) | |
| fig.update_layout(**_L(title="DISTRIBUTION GLOBALE JOURS RETARD (negatif = anticipe)",height=250,showlegend=False)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| def _render_ajustements(df, _sheets): | |
| if df.empty: | |
| st.markdown(f""" | |
| <div style="background:{_C['card']};border:1px solid {_C['border']}; | |
| border-left:3px solid {_C['accent']};padding:16px 20px;font-family:{_FONT};"> | |
| <div style="font-size:0.50rem;letter-spacing:2px;color:{_C['subtext']}; | |
| text-transform:uppercase;margin-bottom:6px;">STATUS OPERATIONNEL</div> | |
| <div style="font-size:1rem;color:{_C['subtext']};"> | |
| TABLE VIDE EN ATTENTE D'ALIMENTATION</div> | |
| <div style="font-size:0.58rem;color:{_C['subtext']};margin-top:4px;"> | |
| Visuels actifs des le premier ajustement enregistre.</div> | |
| </div>""", unsafe_allow_html=True) | |
| return | |
| c1, c2 = st.columns(2) | |
| with c1: | |
| cnt = df["Raison"].value_counts() | |
| fig = go.Figure(go.Pie(labels=cnt.index.astype(str),values=cnt.values,hole=0.5,marker_colors=_PAL)) | |
| fig.update_layout(**_L(title="ANALYSE DES RUPTURES",height=300)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| with c2: | |
| ats = df.get("DX_Adjustment_Timing_Sensitivity",pd.Series(0,index=df.index)).fillna(0) | |
| fig = go.Figure(go.Scatter(x=df["Montant_Additionnel"].fillna(0),y=ats,mode="markers", | |
| marker=dict(size=10,color=ats,colorscale=[[0,"#c0392b"],[1,"#2ecc71"]],showscale=True))) | |
| fig.update_layout(**_L(title="IMPACT x TIMING",height=300,showlegend=False)) | |
| st.plotly_chart(fig,use_container_width=True) | |
| # ── Routeur public ──────────────────────────────────────────────────────────── | |
| _MAP = { | |
| "CLIENTS": ("Clients_KYC", _render_clients), | |
| "GARANTS": ("Garants_KYC", _render_garants), | |
| "PRETS": ("Prets_Master", _render_prets), | |
| "PRETS_UPDATE": ("Prets_Update", _render_prets_update), | |
| "PRETS UPDATE": ("Prets_Update", _render_prets_update), | |
| "REMBOURSEMENTS": ("Remboursements", _render_remboursements), | |
| "AJUSTEMENTS": ("Ajustements_Echeances", _render_ajustements), | |
| } | |
| def render_description(sheets: dict, section: str): | |
| section_up = section.upper().replace("_"," ").strip() | |
| for key in [section, section_up, section.upper()]: | |
| if key in _MAP: | |
| tab_key, fn = _MAP[key] | |
| fn(sheets.get(tab_key, pd.DataFrame()), sheets) | |
| return | |
| st.warning(f"Section inconnue : {section}") |