68 lines
2.0 KiB
Python
68 lines
2.0 KiB
Python
|
"""
|
||
|
|
database.py — Accès aux données historiques pour l'analyse technique.
|
||
|
|
"""
|
||
|
|
from sqlalchemy import text
|
||
|
|
from app import db
|
||
|
|
|
||
|
|
|
||
|
|
def get_historique_df(isin):
|
||
|
|
"""Récupère l'historique complet d'un ISIN trié par date croissante.
|
||
|
|
Retourne un DataFrame pandas ou None si aucune donnée.
|
||
|
|
|
||
|
|
Colonnes : date, price, open, hight, low, vol, change
|
||
|
|
"""
|
||
|
|
import pandas as pd
|
||
|
|
|
||
|
|
rows = db.session.execute(
|
||
|
|
text(
|
||
|
|
"SELECT date, price, `open`, hight, low, vol, `change` "
|
||
|
|
"FROM historique WHERE isin = :isin ORDER BY date ASC"
|
||
|
|
),
|
||
|
|
{"isin": isin},
|
||
|
|
).fetchall()
|
||
|
|
|
||
|
|
if not rows:
|
||
|
|
return None
|
||
|
|
|
||
|
|
df = pd.DataFrame(rows, columns=["date", "price", "open", "hight", "low", "vol", "change"])
|
||
|
|
|
||
|
|
# Conversion des types
|
||
|
|
df["date"] = pd.to_datetime(df["date"])
|
||
|
|
df["price"] = pd.to_numeric(df["price"], errors="coerce")
|
||
|
|
df["open"] = pd.to_numeric(df["open"], errors="coerce")
|
||
|
|
df["hight"] = pd.to_numeric(df["hight"], errors="coerce")
|
||
|
|
df["low"] = pd.to_numeric(df["low"], errors="coerce")
|
||
|
|
df["vol"] = pd.to_numeric(df["vol"], errors="coerce").astype("Int64")
|
||
|
|
df["change"] = pd.to_numeric(df["change"], errors="coerce")
|
||
|
|
|
||
|
|
# Nettoyage : doublons de dates (on garde la dernière)
|
||
|
|
df = df.drop_duplicates(subset="date", keep="last")
|
||
|
|
|
||
|
|
# Suppression des lignes sans prix (essentielles)
|
||
|
|
df = df.dropna(subset=["price"])
|
||
|
|
|
||
|
|
# Tri par date
|
||
|
|
df = df.sort_values("date").reset_index(drop=True)
|
||
|
|
|
||
|
|
return df
|
||
|
|
|
||
|
|
|
||
|
|
def get_action_info(isin):
|
||
|
|
"""Récupère les infos de l'action (company_name, ticker, currency).
|
||
|
|
Retourne un dict ou None si l'ISIN n'existe pas dans le référentiel.
|
||
|
|
"""
|
||
|
|
row = db.session.execute(
|
||
|
|
text("SELECT isin, ticker, company_name, currency FROM actions WHERE isin = :isin"),
|
||
|
|
{"isin": isin},
|
||
|
|
).fetchone()
|
||
|
|
|
||
|
|
if not row:
|
||
|
|
return None
|
||
|
|
|
||
|
|
return {
|
||
|
|
"isin": row.isin,
|
||
|
|
"ticker": row.ticker or "",
|
||
|
|
"company_name": row.company_name or "",
|
||
|
|
"currency": row.currency or "EUR",
|
||
|
|
}
|