src/Algorithms/twin_engine.py

"""
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)