Spaces:
Running
Running
Download src/Analytics/vortex_params.py from klydekushy/Vortex-Flux: direct link, hf CLI and curl.
- Browser
- Download file 16.3 kB
-
https://huggingface.co/spaces/klydekushy/Vortex-Flux/resolve/main/src/Analytics/vortex_params.py
- Command line
-
hf download hf://spaces/klydekushy/Vortex-Flux/src/Analytics/vortex_params.py
-
curl -L -o vortex_params.py https://huggingface.co/spaces/klydekushy/Vortex-Flux/resolve/main/src/Analytics/vortex_params.py
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 | |