Vortex-Flux / src /Analytics /vortex_params.py
jaimalklyde's picture
Claude Opus 5.5
feat(exercice): separation de toute l'application par annee civile
631cd58
Raw History Blame Contribute Delete
16.3 kB
"""
PARAMETRES VORTEX - ACCESSEUR PARTAGE
======================================
Source de verite unique pour les parametres transverses stockes dans
l'onglet Google Sheets `Parametres_Vortex` (objectif annuel, horodatages
de synchronisation, etc.).
Le schema existant (`Timestamp, Variable, Marge_Globale, Raison`) est
utilise par jasmine_optimizer.py (ecriture de la marge) et lu par
client_simulator.py (filtre sur Variable == 'Marge_Globale_Optimizer').
Ce module ajoute une 5e colonne `Valeur`, de maniere additive uniquement :
aucune ligne existante n'est modifiee, aucune migration n'est necessaire.
"""
import pandas as pd
import json
from datetime import datetime
HEADER = ["Timestamp", "Variable", "Marge_Globale", "Raison", "Valeur"]
OBJECTIF_ANNUEL_VARIABLE = "Objectif_Annuel_Profit"
DERNIER_SYNC_FORECASTING_VARIABLE = "Dernier_Sync_Forecasting"
MARGE_OPTIMIZER_VARIABLE = "Marge_Globale_Optimizer"
# ==============================================================================
# RÈGLES MÉTIER DU SIMULATEUR — pilotables depuis le module "Règles Métier"
# (modules/regles_metier.py) sans redéploiement. Chaque écriture via
# set_regle_metier() ajoute une nouvelle ligne horodatée dans Parametres_Vortex
# (jamais de modification en place) : l'historique complet des valeurs
# appliquées dans le temps est donc toujours disponible pour audit/comparatif.
# ==============================================================================
REGLES_METIER = {
"Regle_Whitelist_Decouvert_Privilege": {
"label": "Clients éligibles — Offre Découvert Privilège",
"type": "liste",
"default": ["CLI-2026-0001", "CLI-2026-0002", "CLI-2026-0005", "CLI-2026-0014", "CLI-2026-0016"],
"description": "ID_Client autorisés à souscrire l'Offre Découvert Privilège dans le simulateur public.",
},
"Regle_Seuil_Petit_Montant": {
"label": "Seuil petit montant (taux plancher forcé)",
"type": "float", "default": 60000.0, "unite": "XOF",
"description": "En dessous de ce capital demandé (hors Découvert Privilège), le taux hebdo est forcé au plancher dérogatoire.",
},
"Regle_Seuil_Gros_Montant_Risque": {
"label": "Seuil gros montant à risque",
"type": "float", "default": 80000.0, "unite": "XOF",
"description": "Au-dessus de ce capital demandé, si le taux brut IA dépasse le seuil de risque, la marge stratégique est annulée.",
},
"Regle_Taux_Brut_Seuil_Annulation_Marge": {
"label": "Seuil taux brut IA (annulation marge)",
"type": "float", "default": 3.0, "unite": "%/semaine",
"description": "Au-delà de ce taux de risque brut (modèle), combiné au seuil de gros montant, la marge stratégique est annulée.",
},
"Regle_Taux_Plafond_Decouvert_Court": {
"label": "Plafond taux — Découvert Privilège (< 6 mois)",
"type": "float", "default": 5.0, "unite": "%/semaine",
"description": "Plafond du taux hebdomadaire pour l'Offre Découvert Privilège sur une durée inférieure à 6 mois.",
},
"Regle_Taux_Plafond_Decouvert_Long": {
"label": "Plafond taux — Découvert Privilège (≥ 6 mois)",
"type": "float", "default": 3.0, "unite": "%/semaine",
"description": "Taux préférentiel garanti (plafond) pour l'Offre Découvert Privilège sur une durée de 6 mois ou plus.",
},
"Regle_Taux_Force_Petit_Montant": {
"label": "Taux plancher forcé — petit montant",
"type": "float", "default": 5.0, "unite": "%/semaine",
"description": "Taux hebdomadaire plancher forcé pour les demandes sous le seuil de petit montant (hors Découvert Privilège).",
},
"Regle_Taux_Plancher_Classique": {
"label": "Taux plancher — offres classiques",
"type": "float", "default": 2.0, "unite": "%/semaine",
"description": "Borne basse du taux hebdomadaire pour les offres autres que Découvert Privilège, au-dessus du seuil petit montant.",
},
"Regle_Taux_Plafond_Classique": {
"label": "Taux plafond — offres classiques",
"type": "float", "default": 7.0, "unite": "%/semaine",
"description": "Borne haute du taux hebdomadaire pour les offres autres que Découvert Privilège, au-dessus du seuil petit montant.",
},
"Regle_Seuil_Jours_Retard_Eligible": {
"label": "Seuil de retard éligible à la majoration",
"type": "float", "default": 3.0, "unite": "jours",
"description": "Nombre de jours de retard à partir duquel (inclus) un remboursement pardonné (pénalité à 0) compte comme une occurrence facturable dans une future simulation.",
},
"Regle_Penalite_Retard_Seuil_Moyen": {
"label": "Majoration retard — seuil « montant moyen »",
"type": "float", "default": 50000.0, "unite": "XOF",
"description": "Montant demandé à partir duquel (inclus) s'applique le taux « montant moyen » de la majoration pour retard pardonné, plutôt que le taux « petit montant ».",
},
"Regle_Penalite_Retard_Seuil_Grand": {
"label": "Majoration retard — seuil « montant élevé »",
"type": "float", "default": 150000.0, "unite": "XOF",
"description": "Montant demandé strictement au-dessus duquel s'applique le taux « montant élevé » de la majoration pour retard pardonné, plutôt que le taux « montant moyen ».",
},
"Regle_Penalite_Retard_Pct_Petit": {
"label": "Majoration retard — taux petit montant",
"type": "float", "default": 5.0, "unite": "% (forfait)",
"description": "Majoration forfaitaire appliquée quand le montant demandé est inférieur au seuil « montant moyen ».",
},
"Regle_Penalite_Retard_Pct_Moyen": {
"label": "Majoration retard — taux montant moyen",
"type": "float", "default": 2.5, "unite": "% (forfait)",
"description": "Majoration forfaitaire appliquée quand le montant demandé est entre le seuil « montant moyen » et le seuil « montant élevé » (bornes incluses).",
},
"Regle_Penalite_Retard_Pct_Grand": {
"label": "Majoration retard — taux montant élevé",
"type": "float", "default": 2.0, "unite": "% (forfait)",
"description": "Majoration forfaitaire appliquée quand le montant demandé dépasse le seuil « montant élevé ».",
},
"Regle_Duree_Minimale_Semaines": {
"label": "Durée minimale d'engagement",
"type": "float", "default": 2.0, "unite": "semaines",
"description": "En dessous de cette durée réelle, le taux hebdomadaire normal est remplacé par un forfait fixe (voir Forfait — durée minimale non respectée), quel que soit le type de remboursement.",
},
"Regle_Forfait_Duree_Minimale_Pct": {
"label": "Forfait — durée minimale non respectée",
"type": "float", "default": 10.0, "unite": "% du capital (forfait)",
"description": "Forfait fixe (% du capital financé) facturé quand la durée réelle est inférieure à la durée minimale d'engagement, à la place du taux hebdomadaire calculé.",
},
"Regle_Plafond_Marge_Optimizer": {
"label": "Plafond de marge stratégique — Optimizer",
"type": "float", "default": 1.0, "unite": "%/semaine",
"description": "Marge maximale recommandée par les scénarios de l'Optimizer; la marge brute requise reste affichée séparément.",
},
}
def _get_worksheet(client, sheet_name):
sh = client.open(sheet_name)
try:
return sh.worksheet("Parametres_Vortex")
except Exception:
ws = sh.add_worksheet("Parametres_Vortex", 100, len(HEADER))
ws.append_row(HEADER)
return ws
def _ensure_valeur_column(ws):
"""Ajoute l'en-tete 'Valeur' en 5e colonne si elle n'existe pas encore (idempotent)."""
try:
row1 = ws.row_values(1)
except Exception:
return
if len(row1) < 5 or row1[4] != "Valeur":
ws.update_cell(1, 5, "Valeur")
def get_parametre(client, sheet_name, variable, default=None):
"""Renvoie la derniere valeur enregistree pour `variable` (colonne Valeur), ou `default`."""
try:
ws = _get_worksheet(client, sheet_name)
df = pd.DataFrame(ws.get_all_records())
if df.empty or "Variable" not in df.columns or "Valeur" not in df.columns:
return default
rows = df[df["Variable"] == variable]
if rows.empty:
return default
valeur = rows["Valeur"].iloc[-1]
if valeur is None or str(valeur).strip() == "":
return default
return valeur
except Exception:
return default
def set_parametre(client, sheet_name, variable, valeur, raison=""):
"""Ajoute une nouvelle ligne journalisant `variable` = `valeur` dans Parametres_Vortex."""
try:
ws = _get_worksheet(client, sheet_name)
_ensure_valeur_column(ws)
date_now = datetime.now().strftime("%d/%m/%Y %H:%M:%S")
ws.append_row([date_now, variable, "", raison, str(valeur)])
return True
except Exception:
return False
# Objectifs de benefice par exercice tant qu'aucune valeur n'est enregistree
# dans Parametres_Vortex pour l'annee (fixes par le gerant).
OBJECTIFS_PAR_DEFAUT = {2026: 2_200_000.0, 2027: 3_500_000.0}
def variable_objectif(annee: int) -> str:
"""Nom de la variable Parametres_Vortex de l'objectif d'un exercice."""
return f"{OBJECTIF_ANNUEL_VARIABLE}_{int(annee)}"
def objectif_depuis_df(df: pd.DataFrame, annee: int, default=None) -> float:
"""Objectif de benefice de l'exercice `annee`, par ordre de priorite :
1. derniere valeur de la variable Objectif_Annuel_Profit_<annee> ;
2. ancienne variable sans annee (Objectif_Annuel_Profit), rattachee a
l'annee de son Timestamp (valeurs saisies avant la separation par annee) ;
3. OBJECTIFS_PAR_DEFAUT[annee] ;
4. objectif connu de l'exercice precedent le plus proche (annee future) ;
5. `default`, sinon 0."""
annee = int(annee)
def _dernier(lignes):
if lignes.empty or "Valeur" not in lignes.columns:
return None
for v in reversed(lignes["Valeur"].tolist()):
try:
if str(v).strip() != "":
return float(str(v).replace(",", ".").replace(" ", ""))
except (TypeError, ValueError):
continue
return None
if isinstance(df, pd.DataFrame) and not df.empty and "Variable" in df.columns:
valeur = _dernier(df[df["Variable"] == variable_objectif(annee)])
if valeur is not None:
return valeur
anciens = df[df["Variable"] == OBJECTIF_ANNUEL_VARIABLE]
if not anciens.empty and "Timestamp" in anciens.columns:
from core.exercice import annee_de
valeur = _dernier(anciens[annee_de(anciens["Timestamp"]).eq(annee).fillna(False).to_numpy()])
if valeur is not None:
return valeur
if annee in OBJECTIFS_PAR_DEFAUT:
return OBJECTIFS_PAR_DEFAUT[annee]
if annee > min(OBJECTIFS_PAR_DEFAUT):
return objectif_depuis_df(df, annee - 1, default)
return float(default) if default is not None else 0.0
def get_objectif_annuel(client, sheet_name, default=None, annee=None):
"""Objectif de benefice de l'exercice `annee` (exercice courant par defaut)."""
from core.exercice import exercice_courant
annee = annee or exercice_courant()
try:
df = pd.DataFrame(_get_worksheet(client, sheet_name).get_all_records())
except Exception:
df = pd.DataFrame()
return objectif_depuis_df(df, annee, default)
def set_objectif_annuel(client, sheet_name, nouvelle_valeur, raison="Ajustement manuel", annee=None):
"""Enregistre l'objectif de l'exercice `annee` (exercice courant par defaut)."""
from core.exercice import exercice_courant
annee = annee or exercice_courant()
return set_parametre(client, sheet_name, variable_objectif(annee), nouvelle_valeur,
f"{raison} (exercice {annee})")
# ==============================================================================
# ACCÈS AUX RÈGLES MÉTIER (lecture groupée + écriture historisée)
# ==============================================================================
def charger_df_parametres(client, sheet_name) -> pd.DataFrame:
"""Lit Parametres_Vortex EN UNE SEULE FOIS. A reutiliser pour en extraire
plusieurs valeurs (regles metier, marge strategique...) sans multiplier les
appels Google Sheets."""
try:
ws = _get_worksheet(client, sheet_name)
return pd.DataFrame(ws.get_all_records())
except Exception:
return pd.DataFrame()
def _parse_valeur_regle(brut, type_):
if brut is None:
return None
if type_ == "liste":
try:
valeur = json.loads(brut) if isinstance(brut, str) else brut
except (TypeError, ValueError):
return None
return valeur if isinstance(valeur, list) else None
try:
return float(brut)
except (TypeError, ValueError):
return None
def _serialiser_valeur_regle(valeur, type_):
if type_ == "liste":
return json.dumps(list(valeur))
return str(float(valeur))
def get_regles_actuelles_depuis_df(df_param: pd.DataFrame) -> dict:
"""Valeur courante (derniere ligne) de chaque regle metier definie dans
REGLES_METIER, a partir d'un DataFrame Parametres_Vortex deja charge.
Renvoie la valeur par defaut codee en dur pour toute regle jamais modifiee."""
resultat = {}
for variable, meta in REGLES_METIER.items():
brut = None
if not df_param.empty and "Variable" in df_param.columns and "Valeur" in df_param.columns:
rows = df_param[df_param["Variable"] == variable]
if not rows.empty:
v = rows["Valeur"].iloc[-1]
if v is not None and str(v).strip() != "":
brut = v
valeur = _parse_valeur_regle(brut, meta["type"]) if brut is not None else None
resultat[variable] = valeur if valeur not in (None, []) else meta["default"]
return resultat
def get_regles_actuelles(client, sheet_name) -> dict:
"""Comme get_regles_actuelles_depuis_df, mais lit Parametres_Vortex elle-meme
(a utiliser quand on n'a pas deja un df_param sous la main)."""
return get_regles_actuelles_depuis_df(charger_df_parametres(client, sheet_name))
def set_regle_metier(client, sheet_name, variable, valeur, raison=""):
"""Historise une nouvelle valeur pour une regle metier : ajoute une ligne,
ne modifie jamais les lignes precedentes (historique + comparatif intacts)."""
meta = REGLES_METIER[variable]
return set_parametre(client, sheet_name, variable, _serialiser_valeur_regle(valeur, meta["type"]), raison=raison)
def get_historique_regle(client, sheet_name, variable) -> pd.DataFrame:
"""Historique complet (une ligne par modification, ordre chronologique) d'une
regle metier. Pour les regles numeriques, ajoute Valeur_Num et Diff_Pct
(variation en % par rapport a la valeur precedente)."""
meta = REGLES_METIER[variable]
df = charger_df_parametres(client, sheet_name)
if df.empty or "Variable" not in df.columns or "Valeur" not in df.columns:
return pd.DataFrame(columns=["Timestamp", "Valeur", "Raison"])
hist = df[df["Variable"] == variable][["Timestamp", "Valeur", "Raison"]].reset_index(drop=True)
if meta["type"] == "float":
hist["Valeur_Num"] = hist["Valeur"].apply(lambda v: _parse_valeur_regle(v, "float"))
hist["Diff_Pct"] = hist["Valeur_Num"].pct_change() * 100
return hist
def get_marge_strategique_depuis_df(df_param: pd.DataFrame) -> float:
"""Derniere marge strategique ecrite par jasmine_optimizer.py (colonne
Marge_Globale, schema historique — distinct du systeme Valeur/regles)."""
if df_param.empty or "Marge_Globale" not in df_param.columns or "Variable" not in df_param.columns:
return 0.0
rows = df_param[df_param["Variable"] == MARGE_OPTIMIZER_VARIABLE]
if rows.empty:
return 0.0
try:
return float(rows["Marge_Globale"].iloc[-1])
except (TypeError, ValueError):
return 0.0