print("🔥 ROUTES.PY CHARGE DEPUIS LE VOLUME DOCKER 🔥") from functools import wraps from flask import ( Blueprint, render_template, request, redirect, url_for, session, flash, Response, jsonify, current_app, ) import io import csv from datetime import datetime from werkzeug.security import check_password_hash, generate_password_hash from sqlalchemy import text from app import db, csrf from app import models # noqa: F401 (assure l'enregistrement des modèles ORM) main = Blueprint("main", __name__) # --------------------------------------------------------------------------- # Décorateur d'authentification # --------------------------------------------------------------------------- def login_required(f): @wraps(f) def decorated(*args, **kwargs): if "user_id" not in session: return redirect(url_for("main.index")) return f(*args, **kwargs) return decorated # --------------------------------------------------------------------------- # Route pour la page d'accueil (connexion) # --------------------------------------------------------------------------- @main.route("/", methods=["GET", "POST"]) def index(): if "user_id" in session: return redirect(url_for("main.menu")) if "login_attempts" not in session: session["login_attempts"] = 0 if request.method == "POST": user_login = request.form.get("login") user_pass = request.form.get("pass") query = text( "SELECT id, pass AS user_pass, nom, prenom FROM login " "WHERE login = :login AND actif = 1" ) result = db.session.execute(query, {"login": user_login}).fetchone() if result and check_password_hash(result.user_pass, user_pass): # Régénération de session : protection contre la fixation de session session.clear() session["user_id"] = result.id session["user_name"] = f"{result.prenom} {result.nom}" session["login_attempts"] = 0 return redirect(url_for("main.menu")) else: session["login_attempts"] += 1 essais = session["login_attempts"] if essais >= 3: session["login_attempts"] = 0 flash( "Trop de tentatives. Réessayez dans quelques instants.", "danger", ) else: flash( f"Identifiant ou mot de passe incorrect. " f"Il vous reste {3 - essais} essai(s).", "danger", ) return render_template("index.html") # --------------------------------------------------------------------------- # Déconnexion # --------------------------------------------------------------------------- @main.route("/logout") def logout(): session.clear() return redirect(url_for("main.index")) # --------------------------------------------------------------------------- # Menu principal # --------------------------------------------------------------------------- @main.route("/menu") @login_required def menu(): return render_template("menu.html") # --------------------------------------------------------------------------- # Gestion des utilisateurs # --------------------------------------------------------------------------- @main.route("/gestion-utilisateurs", methods=["GET", "POST"]) @login_required def gestion_utilisateurs(): if request.method == "POST": action = request.form.get("action") user_id = request.form.get("id") user_actif = request.form.get("actif", "1") user_prenom = request.form.get("prenom") user_nom = request.form.get("nom") user_login = request.form.get("login") user_mail = request.form.get("mail") user_pass = request.form.get("pass") if action == "creer": if user_pass: hashed_pass = generate_password_hash(user_pass) query = text( "INSERT INTO login (login, nom, prenom, mail, pass, actif) " "VALUES (:login, :nom, :prenom, :mail, :pass, :actif)" ) db.session.execute( query, { "login": user_login, "nom": user_nom, "prenom": user_prenom, "mail": user_mail, "pass": hashed_pass, "actif": user_actif, }, ) db.session.commit() flash("Utilisateur créé avec succès.", "success") else: flash( "Le mot de passe est obligatoire pour créer un utilisateur.", "danger", ) elif action == "modifier" and user_id: if user_pass: hashed_pass = generate_password_hash(user_pass) query = text( "UPDATE login SET login = :login, nom = :nom, prenom = :prenom, " "mail = :mail, pass = :pass, actif = :actif WHERE id = :id" ) db.session.execute( query, { "id": user_id, "login": user_login, "nom": user_nom, "prenom": user_prenom, "mail": user_mail, "pass": hashed_pass, "actif": user_actif, }, ) else: query = text( "UPDATE login SET login = :login, nom = :nom, prenom = :prenom, " "mail = :mail, actif = :actif WHERE id = :id" ) db.session.execute( query, { "id": user_id, "login": user_login, "nom": user_nom, "prenom": user_prenom, "mail": user_mail, "actif": user_actif, }, ) db.session.commit() flash("Utilisateur modifié avec succès.", "success") return redirect(url_for("main.gestion_utilisateurs")) users_query = text( "SELECT id, login, nom, prenom, mail, actif FROM login ORDER BY id ASC" ) result = db.session.execute(users_query) users = [ { "id": row.id, "login": row.login if row.login else "", "nom": row.nom if row.nom else "", "prenom": row.prenom if row.prenom else "", "mail": row.mail if row.mail else "", "actif": row.actif if row.actif is not None else 1, } for row in result ] return render_template("gestion_utilisateurs.html", users=users) # --------------------------------------------------------------------------- # Gestion des actions # --------------------------------------------------------------------------- @main.route("/gestion-actions", methods=["GET", "POST"]) @login_required def gestion_actions(): if request.method == "POST": action_type = request.form.get("action") action_id = request.form.get("id") isin = request.form.get("isin") or None ticker = request.form.get("ticker") company_name = request.form.get("company_name") exchange = request.form.get("exchange") pays = request.form.get("pays") currency = request.form.get("currency") if action_type == "creer": try: query = text( "INSERT INTO actions (isin, ticker, company_name, exchange, " "pays, currency, updated_at) " "VALUES (:isin, :ticker, :company_name, :exchange, :pays, " ":currency, NOW())" ) db.session.execute( query, { "isin": isin, "ticker": ticker, "company_name": company_name, "exchange": exchange, "pays": pays, "currency": currency, }, ) db.session.commit() flash("Action créée avec succès", "success") except Exception: db.session.rollback() flash("Erreur lors de la création de l'action.", "danger") elif action_type == "modifier" and action_id: try: query = text( "UPDATE actions SET isin = :isin, ticker = :ticker, " "company_name = :company_name, exchange = :exchange, " "pays = :pays, currency = :currency, updated_at = NOW() " "WHERE id = :id" ) db.session.execute( query, { "id": action_id, "isin": isin, "ticker": ticker, "company_name": company_name, "exchange": exchange, "pays": pays, "currency": currency, }, ) db.session.commit() flash("Modification effectuée", "success") except Exception: db.session.rollback() flash("Erreur lors de la modification.", "danger") return redirect(url_for("main.gestion_actions")) actions_query = text( "SELECT id, isin, ticker, company_name, exchange, pays, currency, " "updated_at FROM actions ORDER BY company_name ASC" ) result = db.session.execute(actions_query) actions_list = [ { "id": row.id, "isin": row.isin if row.isin else "", "ticker": row.ticker if row.ticker else "", "company_name": row.company_name if row.company_name else "Sans nom", "exchange": row.exchange if row.exchange else "", "pays": row.pays if row.pays else "", "currency": row.currency if row.currency else "", "updated_at": row.updated_at.strftime("%Y-%m-%d %H:%M:%S") if row.updated_at else "", } for row in result ] return render_template("gestion_actions.html", actions=actions_list) # --------------------------------------------------------------------------- # Gestion des sociétés # --------------------------------------------------------------------------- @main.route("/gestion-societes", methods=["GET", "POST"]) @login_required def gestion_societes(): if request.method == "POST": action = request.form.get("action") soc_id = request.form.get("id") nom = request.form.get("nom") forme = request.form.get("forme") capital = request.form.get("capital") siren = request.form.get("siren") siret = request.form.get("siret") rcs = request.form.get("rcs") tva = request.form.get("tva") naf = request.form.get("naf") tel = request.form.get("tel") web = request.form.get("web") mail = request.form.get("mail") if action == "creer": try: query = text( "INSERT INTO societe (nom, forme, capital, siren, siret, rcs, " "tva, naf, tel, web, mail) " "VALUES (:nom, :forme, :capital, :siren, :siret, :rcs, :tva, " ":naf, :tel, :web, :mail)" ) db.session.execute( query, { "nom": nom, "forme": forme, "capital": capital, "siren": siren, "siret": siret, "rcs": rcs, "tva": tva, "naf": naf, "tel": tel, "web": web, "mail": mail, }, ) db.session.commit() flash("Société créée avec succès.", "success") except Exception as e: db.session.rollback() current_app.logger.error("Erreur création société : %s", e, exc_info=True) flash("Erreur lors de la création de la société.", "danger") elif action == "modifier" and soc_id: try: query = text( "UPDATE societe SET nom = :nom, forme = :forme, " "capital = :capital, siren = :siren, siret = :siret, " "rcs = :rcs, tva = :tva, naf = :naf, tel = :tel, web = :web, " "mail = :mail WHERE id = :id" ) db.session.execute( query, { "id": soc_id, "nom": nom, "forme": forme, "capital": capital, "siren": siren, "siret": siret, "rcs": rcs, "tva": tva, "naf": naf, "tel": tel, "web": web, "mail": mail, }, ) db.session.commit() flash("Société modifiée avec succès.", "success") except Exception as e: db.session.rollback() current_app.logger.error("Erreur modification société : %s", e, exc_info=True) flash("Erreur lors de la modification de la société.", "danger") return redirect(url_for("main.gestion_societes")) societes_query = text( "SELECT id, nom, forme, capital, siren, siret, rcs, tva, naf, tel, web, " "mail FROM societe ORDER BY id ASC" ) result = db.session.execute(societes_query) societes = [ { "id": row.id, "nom": row.nom if row.nom else "", "forme": row.forme if row.forme else "", "capital": row.capital if row.capital is not None else 0, "siren": row.siren if row.siren is not None else "", "siret": row.siret if row.siret is not None else "", "rcs": row.rcs if row.rcs else "", "tva": row.tva if row.tva else "", "naf": row.naf if row.naf else "", "tel": row.tel if row.tel else "", "web": row.web if row.web else "", "mail": row.mail if row.mail else "", } for row in result ] return render_template("gestion_societes.html", societes=societes) # --------------------------------------------------------------------------- # Import / Export des actions CSV # --------------------------------------------------------------------------- @main.route("/gestion-import-export-actions-csv", methods=["GET"]) @login_required def gestion_import_export_actions_csv(): return render_template("gestion_import_export_actions_csv.html") @main.route("/export-actions-csv", methods=["GET"]) @login_required def export_actions_csv(): try: query = text( "SELECT id, isin, ticker, company_name, exchange, pays, currency, " "updated_at FROM actions" ) result = db.session.execute(query) output = io.StringIO() writer = csv.writer(output, delimiter=";", quoting=csv.QUOTE_MINIMAL) writer.writerow( ["id", "isin", "ticker", "company_name", "exchange", "pays", "currency", "updated_at"] ) for row in result: writer.writerow( [ row.id, row.isin or "", row.ticker or "", row.company_name or "", row.exchange or "", row.pays or "", row.currency or "", row.updated_at.strftime("%Y-%m-%d %H:%M:%S") if row.updated_at else "", ] ) output.seek(0) return Response( output.getvalue(), mimetype="text/csv", headers={"Content-Disposition": "attachment;filename=actions_export.csv"}, ) except Exception as e: current_app.logger.error("Erreur export CSV : %s", e, exc_info=True) flash("Erreur lors de l'exportation du fichier CSV.", "danger") return redirect(url_for("main.gestion_import_export_actions_csv")) @main.route("/import-actions-csv", methods=["POST"]) @login_required def import_actions_csv(): if "file" not in request.files: flash("Aucun fichier sélectionné.", "danger") return redirect(url_for("main.gestion_import_export_actions_csv")) file = request.files["file"] if file.filename == "": flash("Aucun fichier sélectionné.", "danger") return redirect(url_for("main.gestion_import_export_actions_csv")) if not file.filename.endswith(".csv"): flash("Veuillez fournir un fichier au format .csv valide.", "warning") return redirect(url_for("main.gestion_import_export_actions_csv")) try: stream = io.TextIOWrapper(file.stream, encoding="utf-8") csv_reader = csv.reader(stream, delimiter=";") header = next(csv_reader, None) if not header: flash("Le fichier CSV est vide.", "danger") return redirect(url_for("main.gestion_import_export_actions_csv")) count = 0 for row in csv_reader: if len(row) >= 6: isin = row[0].strip() if row[0] != "" else None ticker = row[1].strip() if row[1] != "" else None company_name = row[2].strip() if row[2] != "" else None exchange = row[3].strip() if row[3] != "" else "" pays = row[4].strip() if row[4] != "" else "" currency = row[5].strip() if row[5] != "" else "" if not isin or not ticker: continue query = text( "INSERT INTO actions (isin, ticker, company_name, exchange, " "pays, currency, updated_at) " "VALUES (:isin, :ticker, :company_name, :exchange, :pays, " ":currency, NOW()) " "ON DUPLICATE KEY UPDATE ticker = :ticker, " "company_name = :company_name, exchange = :exchange, " "pays = :pays, currency = :currency, updated_at = NOW()" ) db.session.execute( query, { "isin": isin, "ticker": ticker, "company_name": company_name, "exchange": exchange, "pays": pays, "currency": currency, }, ) count += 1 db.session.commit() flash(f"Importation réussie : {count} actions traitées.", "success") except Exception as e: db.session.rollback() current_app.logger.error("Erreur import CSV : %s", e, exc_info=True) flash(f"Erreur lors de l'importation du fichier : {str(e)}", "danger") return redirect(url_for("main.gestion_import_export_actions_csv")) # --------------------------------------------------------------------------- # Import historique CSV (cotation d'une action) # --------------------------------------------------------------------------- @main.route("/gestion-import-historique-csv", methods=["GET"]) @login_required def gestion_import_historique_csv(): return render_template("gestion_import_historique_csv.html") def _convertir_vol(valeur): """Convertit une valeur de volume du CSV en entier. Suffixes gérés : 'K' (x1000), 'M' (x1 000 000). Sinon entier brut. """ if valeur is None: return None s = str(valeur).strip() if s == "" or s.upper() == "N/A" or s.upper() == "NA": return None try: if s.upper().endswith("K"): return int(float(s[:-1]) * 1000) if s.upper().endswith("M"): return int(float(s[:-1]) * 1000000) return int(float(s)) except (TypeError, ValueError): return None def _to_decimal(valeur): """Convertit une valeur CSV en float (Decimal), None si vide/invalide. Retire le suffixe '%' (ex: '0.7%' -> 0.7) et les séparateurs de milliers.""" if valeur is None: return None s = str(valeur).strip() if s == "" or s.upper() in ("N/A", "NA"): return None # Suppression d'un éventuel suffixe '%' (et espaces autour) if s.endswith("%"): s = s[:-1].strip() # Suppression d'un éventuel séparateur de milliers (virgule ou espace) s = s.replace(" ", "").replace(",", "") try: return float(s) except (TypeError, ValueError): return None @main.route("/import-historique-csv", methods=["POST"]) @login_required def import_historique_csv(): isin = (request.form.get("isin") or "").strip().upper() # Vérification que l'ISIN existe bien dans le référentiel actions if not isin: flash("Veuillez saisir un code ISIN.", "danger") return redirect(url_for("main.gestion_import_historique_csv")) action = db.session.execute( text("SELECT company_name FROM actions WHERE isin = :isin"), {"isin": isin} ).fetchone() if not action: flash(f"L'ISIN {isin} n'existe pas dans le référentiel actions.", "danger") return redirect(url_for("main.gestion_import_historique_csv")) if "file" not in request.files: flash("Aucun fichier sélectionné.", "danger") return redirect(url_for("main.gestion_import_historique_csv")) file = request.files["file"] if file.filename == "": flash("Aucun fichier sélectionné.", "danger") return redirect(url_for("main.gestion_import_historique_csv")) if not file.filename.lower().endswith(".csv"): flash("Veuillez fournir un fichier au format .csv valide.", "warning") return redirect(url_for("main.gestion_import_historique_csv")) try: # Lecture du fichier brut avec fallback d'encodage (utf-8 puis latin-1) raw_bytes = file.stream.read() # Suppression d'un éventuel BOM UTF-8 (\xef\xbb\xbf) en début de fichier if raw_bytes.startswith(b"\xef\xbb\xbf"): raw_bytes = raw_bytes[3:] try: content = raw_bytes.decode("utf-8") except UnicodeDecodeError: content = raw_bytes.decode("latin-1") stream = io.StringIO(content) # Auto-détection du séparateur (, ; ou tabulation) sur la 1re ligne premiere_ligne = stream.readline() stream.seek(0) for delim in (",", ";", "\t"): nb = len(premiere_ligne.split(delim)) if nb > 1: separateur = delim break else: separateur = "," csv_reader = csv.reader(stream, delimiter=separateur) header = next(csv_reader, None) if not header: flash("Le fichier CSV est vide.", "danger") return redirect(url_for("main.gestion_import_historique_csv")) # Normalisation des en-têtes : strip + suppression des quotes + lower def _norm_header(h): h = h.strip() if len(h) >= 2 and h[0] in ('"', "'") and h[-1] == h[0]: h = h[1:-1] return h.strip().lower() header_norm = [_norm_header(h) for h in header] # Mapping alias -> nom canonique (couvre les variantes EODHD, Yahoo, etc.) alias = { "date": "date", "price": "price", "close": "price", "last": "price", "adj_close": "price", "adjusted_close": "price", "adj. close": "price", "open": "open", "hight": "hight", "high": "hight", "low": "low", "vol": "vol", "vol.": "vol", "volume": "vol", "change": "change", "change %": "change", "change_pct": "change", "pct_change": "change", "change_percent": "change", "chg": "change", } # Construction du dictionnaire nom_canonique -> index idx = {} for i, h in enumerate(header_norm): canon = alias.get(h) if canon and canon not in idx: idx[canon] = i def col(row, name): """Récupère la valeur d'une colonne par son nom canonique.""" if name not in idx: return None i = idx[name] return row[i] if i < len(row) else None # Vérification : la colonne 'date' est obligatoire if "date" not in idx: flash( f"Pas OK - Colonne 'date' introuvable dans l'en-tête CSV. " f"En-tête détecté : {','.join(header)}", "danger", ) return redirect(url_for("main.gestion_import_historique_csv")) # Requête d'insertion (INSERT IGNORE grâce à l'unique key isin+date) insert = text( "INSERT IGNORE INTO historique " "(isin, date, price, `open`, hight, low, vol, `change`) " "VALUES (:isin, :date, :price, :open, :hight, :low, :vol, :change)" ) count_insere = 0 count_ignore = 0 count_erreurs = 0 exemples_erreurs = [] for row in csv_reader: date_raw = col(row, "date") if not date_raw: count_erreurs += 1 if len(exemples_erreurs) < 5: exemples_erreurs.append(f"Date vide | ligne: {row}") continue # Normalisation de la date -> objet date (pour insertion YYYY-MM-DD) # L'ordre compte : %Y/%d/%m (AAAA/JJ/MM) AVANT %Y/%m/%d pour éviter # l'ambiguïté (un jour >12 invaliderait %Y/%m/%d et tomberait juste sur %Y/%d/%m). date_str = date_raw.strip()[:10] date_val = None for fmt in ("%Y-%m-%d", "%Y/%d/%m", "%Y/%m/%d", "%m/%d/%Y", "%d/%m/%Y", "%d-%m-%Y"): try: date_val = datetime.strptime(date_str, fmt).date() break except ValueError: continue if date_val is None: count_erreurs += 1 if len(exemples_erreurs) < 10: exemples_erreurs.append( f"Date invalide '{date_raw}' (nb colonnes ligne={len(row)}) | ligne: {row}" ) continue price = _to_decimal(col(row, "price")) open_ = _to_decimal(col(row, "open")) hight = _to_decimal(col(row, "hight")) low = _to_decimal(col(row, "low")) vol = _convertir_vol(col(row, "vol")) change = _to_decimal(col(row, "change")) result = db.session.execute( insert, { "isin": isin, "date": date_val, "price": price, "open": open_, "hight": hight, "low": low, "vol": vol, "change": change, }, ) # rowcount == 1 si réellement inséré, 0 si ignoré (doublon unique key) if result.rowcount == 1: count_insere += 1 else: count_ignore += 1 db.session.commit() detail = "" if exemples_erreurs: detail = " | Exemples d'erreurs: " + " ; ".join(exemples_erreurs) flash( f"OK - Importation réussie pour {action.company_name} ({isin}) : " f"{count_insere} ligne(s) insérée(s), " f"{count_ignore} ligne(s) déjà existante(s) ignorée(s), " f"{count_erreurs} ligne(s) en erreur. " f"[Séparateur: '{separateur}' | En-tête: {','.join(header)}]" f"{detail}", "success", ) except Exception as e: db.session.rollback() current_app.logger.error("Erreur import historique CSV : %s", e, exc_info=True) flash(f"Pas OK - Erreur lors de l'importation du fichier : {str(e)}", "danger") return redirect(url_for("main.gestion_import_historique_csv")) # --------------------------------------------------------------------------- # Analyse technique # --------------------------------------------------------------------------- @main.route("/analyse", methods=["GET"]) @login_required def analyse(): """Page d'analyse technique. Si un ISIN est passé en paramètre (?isin=...), on l'analyse et on affiche la fiche complète ; sinon on affiche la page de recherche.""" isin = request.args.get("isin", "").strip().upper() if not isin: return render_template("analyse.html", resultat=None) from app.services.analysis import analyse_action print(f"[analyse] ISIN demandé : {isin}", flush=True) resultat = analyse_action(isin) if resultat is None: # ISIN absent du référentiel actions return render_template( "analyse.html", resultat={ "isin": isin, "action": None, "erreur": f"Aucune donnée trouvée pour l'ISIN {isin}. " f"Vérifiez le code ou importez d'abord l'historique.", }, isin_recherche=isin, ) if resultat.get("erreur"): # ISIN présent mais historique insuffisant print(f"[analyse] {isin} : {resultat['erreur']}", flush=True) else: print( f"[analyse] {isin} : cours={resultat.get('cours')} " f"score={resultat.get('score')} tendance={resultat.get('tendance')}", flush=True, ) return render_template("analyse.html", resultat=resultat, isin_recherche=isin) @main.route("/analyse/", methods=["GET"]) @login_required def analyse_isin(isin): """Page d'analyse technique complète pour un ISIN donné.""" from app.services.analysis import analyse_action isin = isin.strip().upper() resultat = analyse_action(isin) return render_template("analyse.html", resultat=resultat, isin_recherche=isin) @main.route("/api/analyse/", methods=["GET"]) @login_required def api_analyse(isin): """API JSON retournant l'analyse technique complète d'un ISIN.""" from app.services.analysis import analyse_action isin = isin.strip().upper() resultat = analyse_action(isin) if resultat is None: return jsonify({"erreur": f"ISIN {isin} introuvable dans le référentiel actions."}), 404 if resultat.get("erreur"): return jsonify(resultat), 200 return jsonify(resultat) # --------------------------------------------------------------------------- # Gestion des ordres # --------------------------------------------------------------------------- @main.route("/gestion_ordres", methods=["GET", "POST"]) @login_required def gestion_ordres(): selected_societe_id = ( request.form.get("societe_id") or request.args.get("societe_id") or session.get("selected_societe_id") ) selected_annee = ( request.form.get("annee") or request.args.get("annee") or session.get("selected_annee", "Tout") ) session["selected_annee"] = selected_annee societes = [] selected_societe = "" groupes_dict = {} total_plus_value_globale = 0.0 with db.engine.connect() as connection: result_soc = connection.execute( text("SELECT id, nom FROM societe ORDER BY nom ASC") ) societes = [dict(row._mapping) for row in result_soc] if societes: if not selected_societe_id: selected_societe_id = societes[0]["id"] else: try: selected_societe_id = int(selected_societe_id) except (TypeError, ValueError): selected_societe_id = societes[0]["id"] session["selected_societe_id"] = selected_societe_id for soc in societes: if soc["id"] == selected_societe_id: selected_societe = soc["nom"] break # --- CALCUL DE LA PLUS-VALUE TOTALE --- annee_str = str(selected_annee).strip().lower() if not annee_str or annee_str == "tout": query_pv = text( "SELECT SUM(plus_value) AS total_pv FROM ordre " "WHERE id_societe = :soc_id AND plus_value IS NOT NULL" ) res_pv = connection.execute( query_pv, {"soc_id": selected_societe_id} ).fetchone() else: query_pv = text( "SELECT SUM(plus_value) AS total_pv FROM ordre " "WHERE id_societe = :soc_id AND plus_value IS NOT NULL " "AND LEFT(CAST(date_plus_value AS CHAR), 4) = :annee" ) res_pv = connection.execute( query_pv, {"soc_id": selected_societe_id, "annee": annee_str} ).fetchone() if res_pv and res_pv.total_pv is not None: total_plus_value_globale = float(res_pv.total_pv) # --- CHARGEMENT DES ORDRES --- query = text( "SELECT o.id AS ordre_id, o.isin, o.plus_value, o.date_plus_value, " "op.id AS pied_id, op.date_op, op.quantite, op.prix_brut, " "op.frais_broker, op.frais_etat, op.frais_autre, " "op.sens, op.position, op.etat, " "a.ticker, a.company_name, a.currency " "FROM ordre o " "JOIN ordre_pied op ON o.id = op.id_entete " "JOIN actions a ON o.isin = a.isin " "WHERE o.id_societe = :soc_id " "ORDER BY o.id DESC, op.date_op DESC" ) result_ordres = connection.execute( query, {"soc_id": selected_societe_id} ) for row in result_ordres: r = dict(row._mapping) ordre_id = r["ordre_id"] isin = r["isin"] quantite = float(r["quantite"]) if r["quantite"] is not None else 0.0 prix_brut = float(r["prix_brut"]) if r["prix_brut"] is not None else 0.0 frais_broker = ( float(r["frais_broker"]) if r["frais_broker"] is not None else 0.0 ) frais_etat = ( float(r["frais_etat"]) if r["frais_etat"] is not None else 0.0 ) frais_autre = ( float(r["frais_autre"]) if r["frais_autre"] is not None else 0.0 ) frais_total = frais_broker + frais_etat + frais_autre sens_ordre = str(r["sens"]).strip().lower() type_ordre = str(r["position"]).strip().lower() if r["position"] else "long" # Prix Net unitaire if sens_ordre == "vente": prix_net = prix_brut - frais_total else: prix_net = prix_brut + frais_total # Engagements ligne eng_brut = quantite * prix_brut eng_net = quantite * prix_net pied_data = { "id": r["pied_id"], "date_op": r["date_op"], "quantite": quantite, "prix_brut": prix_brut, "frais": frais_total, "frais_broker": frais_broker, "frais_etat": frais_etat, "frais_autre": frais_autre, "prix_net": prix_net, "eng_brut": eng_brut, "eng_net": eng_net, "ordre": r["sens"], "type_ordre": r["position"], "etat": r["etat"], } if ordre_id not in groupes_dict: groupes_dict[ordre_id] = { "id": ordre_id, "isin": isin, "company_name": r["company_name"], "ticker": r["ticker"], "pieds": [], "total_quantite": 0.0, "total_prix_brut": 0.0, "total_frais": 0.0, "total_prix_net": 0.0, "total_eng_brut": 0.0, "total_eng_net": 0.0, "plus_value": r["plus_value"], "date_plusvalue": r["date_plus_value"], } groupes_dict[ordre_id]["pieds"].append(pied_data) groupes_dict[ordre_id]["total_prix_brut"] += prix_brut groupes_dict[ordre_id]["total_frais"] += frais_total groupes_dict[ordre_id]["total_prix_net"] += prix_net # Règles de totaux if sens_ordre == "achat": groupes_dict[ordre_id]["total_quantite"] += quantite groupes_dict[ordre_id]["total_eng_brut"] -= eng_brut groupes_dict[ordre_id]["total_eng_net"] -= eng_net elif sens_ordre == "vente": groupes_dict[ordre_id]["total_quantite"] -= quantite groupes_dict[ordre_id]["total_eng_brut"] += eng_brut groupes_dict[ordre_id]["total_eng_net"] += eng_net # --- MISE À JOUR AUTO DE LA PLUS-VALUE (ordres soldés) --- try: with db.engine.begin() as conn: for ordre_id, groupe in groupes_dict.items(): if abs(groupe["total_quantite"]) < 1e-9: total_eng_net = groupe["total_eng_net"] max_date = max( (p["date_op"] for p in groupe["pieds"] if p["date_op"] is not None), default=None, ) conn.execute( text( "UPDATE ordre_pied SET etat = :nouvel_etat " "WHERE id_entete = :ordre_id" ), {"nouvel_etat": "soldé", "ordre_id": ordre_id}, ) conn.execute( text( "UPDATE ordre SET plus_value = :plus_value, " "date_plus_value = :date_pv WHERE id = :ordre_id" ), { "plus_value": total_eng_net, "date_pv": max_date, "ordre_id": ordre_id, }, ) groupe["plus_value"] = total_eng_net groupe["date_plusvalue"] = max_date for pied in groupe["pieds"]: pied["etat"] = "soldé" except Exception as e: current_app.logger.error("Erreur MAJ plus-value : %s", e, exc_info=True) groupes_ordres = list(groupes_dict.values()) annees_combo = ["Tout"] + [str(an) for an in range(2020, 2041)] return render_template( "gestion_ordres.html", societes=societes, selected_societe_id=selected_societe_id, selected_societe=selected_societe, groupes_ordres=groupes_ordres, total_plus_value_globale=total_plus_value_globale, annees_combo=annees_combo, selected_annee=selected_annee, ) # --------------------------------------------------------------------------- # Création d'ordre (la saisie se fait via la modale de gestion_ordres ; # cette route est conservée pour compatibilité et redirige vers la liste) # --------------------------------------------------------------------------- @main.route("/creer-ordre", methods=["GET", "POST"]) @login_required def creer_ordre(): return redirect(url_for("main.gestion_ordres")) # --------------------------------------------------------------------------- # API : vérifier l'existence d'un ISIN (JSON, exemptée de CSRF) # --------------------------------------------------------------------------- @main.route("/api/verifier_isin", methods=["POST"]) @csrf.exempt def verifier_isin(): data = request.get_json(silent=True) or {} isin = (data.get("isin") or "").strip() query = text( "SELECT company_name, ticker, currency FROM actions WHERE isin = :isin" ) action = db.session.execute(query, {"isin": isin}).fetchone() if action: return jsonify( { "exists": True, "company_name": action.company_name, "ticker": action.ticker, "currency": action.currency, } ) return jsonify({"exists": False}) # --------------------------------------------------------------------------- # Traitement de la création d'un ordre (appelé par la modale) # --------------------------------------------------------------------------- @main.route("/creer-ordre-traitement", methods=["POST"]) @login_required def creer_ordre_traitement(): isin = request.form.get("isin") date_op = request.form.get("date_op") selected_annee = request.form.get("annee", "tout") try: quantite = float(request.form.get("quantite") or 0) prix_brut = float(request.form.get("prix_brut") or 0) frais_broker = float(request.form.get("frais_broker") or 0) frais_etat = float(request.form.get("frais_etat") or 0) frais_autre = float(request.form.get("frais_autre") or 0) except ValueError: flash( "Veuillez saisir des valeurs numériques valides pour les quantités, " "prix et frais.", "danger", ) return redirect(url_for("main.gestion_ordres", annee=selected_annee)) sens = request.form.get("ordre") position = request.form.get("type_ordre") etat = request.form.get("etat") id_societe = session.get("selected_societe_id") if not id_societe: flash( "Veuillez sélectionner une société active avant de créer un ordre.", "danger", ) return redirect(url_for("main.gestion_ordres", annee=selected_annee)) try: query_ordre = text( "INSERT INTO ordre (id_societe, isin) VALUES (:id_societe, :isin)" ) db.session.execute(query_ordre, {"id_societe": id_societe, "isin": isin}) result_id = db.session.execute(text("SELECT LAST_INSERT_ID()")).scalar() if not result_id: result_id = db.session.execute( text( "SELECT id FROM ordre WHERE isin = :isin " "ORDER BY id DESC LIMIT 1" ), {"isin": isin}, ).scalar() query_pied = text( "INSERT INTO ordre_pied (id_entete, date_op, quantite, prix_brut, " "frais_broker, frais_etat, frais_autre, sens, position, etat) " "VALUES (:id_entete, :date_op, :quantite, :prix_brut, :frais_broker, " ":frais_etat, :frais_autre, :sens, :position, :etat)" ) db.session.execute( query_pied, { "id_entete": result_id, "date_op": date_op, "quantite": quantite, "prix_brut": prix_brut, "frais_broker": frais_broker, "frais_etat": frais_etat, "frais_autre": frais_autre, "sens": sens, "position": position, "etat": etat, }, ) db.session.commit() flash("Ordre créé avec succès !", "success") except Exception as e: db.session.rollback() current_app.logger.error("Erreur création ordre : %s", e, exc_info=True) flash("Erreur lors de la création de l'ordre.", "danger") return redirect(url_for("main.gestion_ordres", annee=selected_annee)) # --------------------------------------------------------------------------- # Ajout d'une ligne à un ordre existant (POST depuis la modale) # --------------------------------------------------------------------------- @main.route("/ordre//ajouter-ligne", methods=["GET", "POST"]) @login_required def ajouter_ligne_ordre(ordre_id): selected_annee = request.args.get("annee", "tout") if request.method == "POST": selected_annee = request.form.get("annee", "tout") try: quantite = float(request.form.get("quantite") or 0) prix_brut = float(request.form.get("prix_brut") or 0) frais_broker = float(request.form.get("frais_broker") or 0) frais_etat = float(request.form.get("frais_etat") or 0) frais_autre = float(request.form.get("frais_autre") or 0) except ValueError: flash("Veuillez saisir des valeurs numériques valides.", "danger") return redirect( url_for( "main.ajouter_ligne_ordre", ordre_id=ordre_id, annee=selected_annee, ) ) try: query_pied = text( "INSERT INTO ordre_pied (id_entete, date_op, quantite, prix_brut, " "frais_broker, frais_etat, frais_autre, sens, position, etat) " "VALUES (:id_entete, :date_op, :quantite, :prix_brut, :frais_broker, " ":frais_etat, :frais_autre, :sens, :position, :etat)" ) db.session.execute( query_pied, { "id_entete": ordre_id, "date_op": request.form.get("date_op"), "quantite": quantite, "prix_brut": prix_brut, "frais_broker": frais_broker, "frais_etat": frais_etat, "frais_autre": frais_autre, "sens": request.form.get("ordre"), "position": request.form.get("type_ordre"), "etat": request.form.get("etat"), }, ) db.session.commit() flash("Ligne ajoutée avec succès à l'ordre !", "success") return redirect(url_for("main.gestion_ordres", annee=selected_annee)) except Exception as e: db.session.rollback() current_app.logger.error("Erreur ajout ligne : %s", e, exc_info=True) flash("Erreur lors de l'ajout de la ligne.", "danger") return redirect(url_for("main.gestion_ordres", annee=selected_annee)) # --------------------------------------------------------------------------- # Modification d'une ligne d'ordre (POST depuis la modale) # --------------------------------------------------------------------------- @main.route("/modifier-ligne-ordre/", methods=["GET", "POST"]) @login_required def modifier_ligne_ordre(pied_id): selected_annee = request.args.get("annee", "tout") if request.method == "POST": selected_annee = request.form.get("annee", "tout") try: quantite = float(request.form.get("quantite") or 0) prix_brut = float(request.form.get("prix_brut") or 0) frais_broker = float(request.form.get("frais_broker") or 0) frais_etat = float(request.form.get("frais_etat") or 0) frais_autre = float(request.form.get("frais_autre") or 0) except ValueError: flash("Veuillez saisir des valeurs numériques valides.", "danger") return redirect( url_for( "main.modifier_ligne_ordre", pied_id=pied_id, annee=selected_annee, ) ) try: query_update = text( "UPDATE ordre_pied SET date_op = :date_op, quantite = :quantite, " "prix_brut = :prix_brut, frais_broker = :frais_broker, " "frais_etat = :frais_etat, frais_autre = :frais_autre, " "sens = :sens, position = :position, etat = :etat " "WHERE id = :pied_id" ) db.session.execute( query_update, { "date_op": request.form.get("date_op"), "quantite": quantite, "prix_brut": prix_brut, "frais_broker": frais_broker, "frais_etat": frais_etat, "frais_autre": frais_autre, "sens": request.form.get("ordre"), "position": request.form.get("type_ordre"), "etat": request.form.get("etat"), "pied_id": pied_id, }, ) db.session.commit() flash("Ligne modifiée avec succès !", "success") return redirect(url_for("main.gestion_ordres", annee=selected_annee)) except Exception as e: db.session.rollback() current_app.logger.error("Erreur modification ligne : %s", e, exc_info=True) flash("Erreur lors de la modification de la ligne.", "danger") return redirect(url_for("main.gestion_ordres", annee=selected_annee)) # --------------------------------------------------------------------------- # Gestion des comptes # --------------------------------------------------------------------------- @main.route("/gestion_comptes", methods=["GET", "POST"]) @login_required def gestion_comptes(): from app.models import Societe, Compte societes = Societe.query.order_by(Societe.nom.asc()).all() # La société sélectionnée vient du formulaire (changement manuel), # ou d'un paramètre d'URL (redirection après création/modification), # ou de la session (mémorisation entre navigations). selected_societe_id = ( request.form.get("societe_id") or request.args.get("societe_id") or session.get("selected_compte_societe_id") ) if not selected_societe_id and societes: selected_societe_id = societes[0].id else: try: selected_societe_id = int(selected_societe_id) if selected_societe_id else None except ValueError: selected_societe_id = societes[0].id if societes else None # Mémorisation en session : la société active ne change que manuellement if selected_societe_id is not None: session["selected_compte_societe_id"] = selected_societe_id selected_societe_nom = "" ecritures = [] total_credit = 0.0 total_debit = 0.0 # Agrégats financiers (voir règles de calcul dans la documentation) tcredit = 0.0 # Depot : somme des crédits des écritures manuelles tdebit = 0.0 # Retrait : somme des débits des écritures manuelles engagement = 0.0 # ENGA. : positions actives (achat +, vente -) pv = 0.0 # PV : positions soldées (achat -, vente +) dispo = 0.0 # DISPO = TOTAL CREDIT - TOTAL DEBIT (toutes écritures) solde = 0.0 # SOLDE = DISPO + ENGA. if selected_societe_id: societe_obj = Societe.query.get(selected_societe_id) if societe_obj: selected_societe_nom = societe_obj.nom # Tri par date décroissante (du plus récent au plus ancien) ecritures = ( Compte.query.filter_by(id_societe=selected_societe_id) .order_by(Compte.date_operation.desc(), Compte.id.desc()) .all() ) # Totaux bruts du tableau (toutes écritures confondues) total_credit = sum(float(e.credit or 0) for e in ecritures) total_debit = sum(float(e.debit or 0) for e in ecritures) # --- Dépôts / Retraits : uniquement les écritures manuelles --- for e in ecritures: if e.source == "manuel": tcredit += float(e.credit or 0) tdebit += float(e.debit or 0) # --- DISPO = TOTAL CREDIT - TOTAL DEBIT (toutes écritures) --- dispo = total_credit - total_debit # --- Engagement & PV : depuis les lignes d'ordres (ordre_pied) --- # Règles : # etat='actif' : achat -> engagement += qté*(prix_brut+frais) # vente -> engagement -= qté*(prix_brut-frais) # etat='soldé' : achat -> PV -= qté*(prix_brut+frais) # vente -> PV += qté*(prix_brut-frais) # etat='annulé' : ignoré (filtré en SQL) # Filtres : société en cours + lignes non annulées query_ordres = text( "SELECT op.sens, op.etat, op.quantite, op.prix_brut, " "op.frais_broker, op.frais_etat, op.frais_autre " "FROM ordre o " "JOIN ordre_pied op ON op.id_entete = o.id " "WHERE o.id_societe = :soc_id " "AND op.etat <> 'annulé'" ) for row in db.session.execute(query_ordres, {"soc_id": selected_societe_id}): quantite = float(row.quantite or 0) prix_brut = float(row.prix_brut or 0) frais_broker = float(row.frais_broker or 0) frais_etat = float(row.frais_etat or 0) frais_autre = float(row.frais_autre or 0) # Achat : on paie le prix brut + les frais montant_achat = quantite * (prix_brut + frais_broker + frais_etat + frais_autre) # Vente : on encaisse le prix brut - les frais montant_vente = quantite * (prix_brut - frais_broker - frais_etat - frais_autre) sens = (row.sens or "").strip().lower() etat = (row.etat or "").strip().lower() if etat == "actif": if sens == "achat": engagement += montant_achat elif sens == "vente": engagement -= montant_vente elif etat == "soldé" or etat == "solde": if sens == "achat": pv -= montant_achat elif sens == "vente": pv += montant_vente # etat == 'annulé' : ne rien faire (déjà filtré en SQL) solde = dispo + engagement return render_template( "gestion_comptes.html", societes=societes, selected_societe_id=selected_societe_id, selected_societe=selected_societe_nom, ecritures=ecritures, total_credit=total_credit, total_debit=total_debit, tcredit=tcredit, tdebit=tdebit, engagement=engagement, pv=pv, dispo=dispo, solde=solde, ) # --------------------------------------------------------------------------- # Génération automatique des écritures de compte depuis les ordres # --------------------------------------------------------------------------- @main.route("/generer_ecritures_ordres", methods=["POST"]) @login_required def generer_ecritures_ordres(): """Génère les écritures comptables (table `compte`) à partir des ordres et de leurs lignes (`ordre_pied`) pour la société actuellement sélectionnée. Règles : - achat : debit = quantite * (prix_brut + frais_broker + frais_etat + frais_autre), credit = 0 - vente : credit = quantite * (prix_brut + frais_broker + frais_etat + frais_autre), debit = 0 - libelle= "Achat/Vente actions {isin} {company_name}" - source = 'ordre' (non modifiable depuis l'IHM) Idempotent : supprime d'abord les écritures source='ordre' de la société. """ id_societe = request.form.get("id_societe") or session.get( "selected_compte_societe_id" ) if not id_societe: flash("Aucune société sélectionnée pour la génération.", "danger") return redirect(url_for("main.gestion_comptes")) try: id_societe_int = int(id_societe) except (TypeError, ValueError): flash("Identifiant de société invalide.", "danger") return redirect(url_for("main.gestion_comptes")) try: # 1. Suppression des anciennes écritures générées pour cette société db.session.execute( text( "DELETE FROM compte WHERE id_societe = :soc_id AND source = 'ordre'" ), {"soc_id": id_societe_int}, ) # 2. Parcours des ordres de la société, puis de leurs lignes # Filtres : société en cours + lignes non annulées query = text( "SELECT o.id, o.isin, a.company_name, " "op.date_op, op.quantite, op.prix_brut, " "op.frais_broker, op.frais_etat, op.frais_autre, op.sens " "FROM ordre o " "JOIN ordre_pied op ON op.id_entete = o.id " "JOIN actions a ON o.isin = a.isin " "WHERE o.id_societe = :soc_id " "AND op.etat <> 'annulé' " "ORDER BY o.id, op.date_op" ) lignes = db.session.execute(query, {"soc_id": id_societe_int}).fetchall() insert = text( "INSERT INTO compte (id_societe, date_operation, debit, credit, " "libelle, source) " "VALUES (:id_societe, :date_op, :debit, :credit, :libelle, 'ordre')" ) count = 0 from markupsafe import escape for row in lignes: quantite = float(row.quantite or 0) prix_brut = float(row.prix_brut or 0) frais_broker = float(row.frais_broker or 0) frais_etat = float(row.frais_etat or 0) frais_autre = float(row.frais_autre or 0) sens = (row.sens or "").strip().lower() isin = row.isin or "" # Nom de l'action échappé puis mis en gras (affichage |safe côté template) company = f"{escape(row.company_name or '')}" if sens == "achat": # Achat : on paie le prix brut + les frais montant = quantite * (prix_brut + frais_broker + frais_etat + frais_autre) debit = montant credit = 0.0 libelle = f"Achat actions {isin} {company}".strip() elif sens == "vente": # Vente : on encaisse le prix brut - les frais montant = quantite * (prix_brut - frais_broker - frais_etat - frais_autre) debit = 0.0 credit = montant libelle = f"Vente actions {isin} {company}".strip() else: # Sens inattendu : on ignore la ligne continue db.session.execute( insert, { "id_societe": id_societe_int, "date_op": row.date_op, "debit": debit, "credit": credit, "libelle": libelle, }, ) count += 1 db.session.commit() flash( f"Génération terminée : {count} écriture(s) créée(s) pour " f"la société.", "success", ) except Exception as e: db.session.rollback() current_app.logger.error("Erreur génération écritures : %s", e, exc_info=True) flash("Erreur lors de la génération des écritures.", "danger") return redirect(url_for("main.gestion_comptes", societe_id=id_societe_int)) # --------------------------------------------------------------------------- # Création d'une écriture comptable # --------------------------------------------------------------------------- @main.route("/creer_compte_traitement", methods=["POST"]) @login_required def creer_compte_traitement(): id_societe = request.form.get("id_societe") date_operation = request.form.get("date_operation") libelle = request.form.get("libelle", "").strip() debit = request.form.get("debit", "0") credit = request.form.get("credit", "0") if not id_societe or not date_operation: flash("Veuillez remplir tous les champs obligatoires.", "danger") return redirect(url_for("main.gestion_comptes", societe_id=id_societe or "")) try: debit_val = float(debit) if debit else 0.0 credit_val = float(credit) if credit else 0.0 id_societe_int = int(id_societe) except ValueError: flash("Veuillez saisir des valeurs numériques valides.", "danger") return redirect(url_for("main.gestion_comptes", societe_id=id_societe or "")) try: db.session.execute( text( "INSERT INTO compte (id_societe, date_operation, libelle, " "debit, credit) VALUES (:id_societe, :date_op, :libelle, :debit, :credit)" ), { "id_societe": id_societe_int, "date_op": date_operation, "libelle": libelle or None, "debit": debit_val, "credit": credit_val, }, ) db.session.commit() flash("Écriture créée avec succès !", "success") except Exception as e: db.session.rollback() current_app.logger.error("Erreur création écriture : %s", e, exc_info=True) flash("Erreur lors de la création de l'écriture.", "danger") # On reste sur la même société après validation return redirect(url_for("main.gestion_comptes", societe_id=id_societe_int)) # --------------------------------------------------------------------------- # Modification d'une écriture comptable # --------------------------------------------------------------------------- @main.route("/modifier_compte/", methods=["POST"]) @login_required def modifier_compte(id): date_operation = request.form.get("date_operation") libelle = request.form.get("libelle", "").strip() debit = request.form.get("debit", "0") credit = request.form.get("credit", "0") # Récupération de la société de l'écriture avant modification (pour la redirection) id_societe = request.form.get("id_societe") if not id_societe: row = db.session.execute( text("SELECT id_societe FROM compte WHERE id = :id"), {"id": id} ).fetchone() id_societe = str(row.id_societe) if row else "" if not date_operation: flash("Veuillez remplir tous les champs obligatoires.", "danger") return redirect(url_for("main.gestion_comptes", societe_id=id_societe)) try: debit_val = float(debit) if debit else 0.0 credit_val = float(credit) if credit else 0.0 except ValueError: flash("Veuillez saisir des valeurs numériques valides.", "danger") return redirect(url_for("main.gestion_comptes", societe_id=id_societe)) try: db.session.execute( text( "UPDATE compte SET date_operation = :date_op, " "libelle = :libelle, debit = :debit, credit = :credit WHERE id = :id" ), { "date_op": date_operation, "libelle": libelle or None, "debit": debit_val, "credit": credit_val, "id": id, }, ) db.session.commit() flash("Écriture modifiée avec succès !", "success") except Exception as e: db.session.rollback() current_app.logger.error("Erreur modification écriture : %s", e, exc_info=True) flash("Erreur lors de la modification de l'écriture.", "danger") # On reste sur la même société après validation return redirect(url_for("main.gestion_comptes", societe_id=id_societe))