151 lines
5.3 KiB
Python
151 lines
5.3 KiB
Python
"""
|
|
analysis.py — Orchestrateur de l'analyse technique complète.
|
|
|
|
La fonction analyse_action(isin) :
|
|
1. vérifie l'ISIN dans le référentiel actions
|
|
2. récupère l'historique depuis la base
|
|
3. calcule tous les indicateurs via indicators.py
|
|
4. calcule le score global
|
|
5. prépare les données pour les graphiques
|
|
6. retourne un dict JSON-sérialisable conforme au format spécifié
|
|
"""
|
|
from datetime import datetime
|
|
|
|
from app.services.database import get_historique_df, get_action_info
|
|
from app.services import indicators as ind
|
|
|
|
|
|
def analyse_action(isin):
|
|
"""Analyse technique complète d'une action par son ISIN.
|
|
Retourne un dict JSON-sérialisable ou None si l'ISIN est introuvable.
|
|
"""
|
|
# --- 1) Vérification de l'ISIN dans le référentiel ---
|
|
info = get_action_info(isin)
|
|
if not info:
|
|
return None
|
|
|
|
# --- 2) Récupération de l'historique ---
|
|
df = get_historique_df(isin)
|
|
if df is None or len(df) < 5:
|
|
return {
|
|
"isin": isin,
|
|
"action": info,
|
|
"erreur": "Données historiques insuffisantes pour l'analyse (minimum 5 cours requis).",
|
|
}
|
|
|
|
# --- 3) Calcul de tous les indicateurs ---
|
|
performances = ind.calcul_performances(df)
|
|
moyennes_mobiles = ind.calcul_moyennes_mobiles(df)
|
|
volatilite = ind.calcul_volatilite(df)
|
|
atr = ind.calcul_atr(df)
|
|
gaps = ind.calcul_gaps(df)
|
|
rsi = {
|
|
"RSI14": ind.calcul_rsi(df, 14),
|
|
"RSI7": ind.calcul_rsi(df, 7),
|
|
}
|
|
macd = ind.calcul_macd(df)
|
|
bollinger = ind.calcul_bollinger(df)
|
|
supports_resistances = ind.calcul_supports_resistances(df)
|
|
volumes = ind.calcul_volumes(df)
|
|
drawdown = ind.calcul_drawdown(df)
|
|
series = ind.calcul_series(df)
|
|
chandeliers = ind.calcul_chandeliers(df)
|
|
|
|
# --- 4) Score global ---
|
|
score_data = ind.calcul_score_global(
|
|
df, performances, moyennes_mobiles, volatilite, rsi["RSI14"], macd, drawdown, volumes
|
|
)
|
|
|
|
# --- 5) Données pour les graphiques ---
|
|
# On prend les 250 derniers jours pour les graphiques (performance)
|
|
nb_points = min(250, len(df))
|
|
df_chart = df.tail(nb_points)
|
|
|
|
graphique_cours = {
|
|
"dates": [d.strftime("%Y-%m-%d") for d in df_chart["date"]],
|
|
"close": [round(float(p), 2) for p in df_chart["price"]],
|
|
"open": [round(float(p), 2) if not _is_nan(p) else None for p in df_chart["open"]],
|
|
"high": [round(float(p), 2) if not _is_nan(p) else None for p in df_chart["hight"]],
|
|
"low": [round(float(p), 2) if not _is_nan(p) else None for p in df_chart["low"]],
|
|
}
|
|
|
|
# Moyennes mobiles pour le graphique
|
|
close = df["price"]
|
|
for p in [20, 50, 200]:
|
|
sma = close.rolling(window=p).mean()
|
|
sma_tail = sma.tail(nb_points).tolist()
|
|
graphique_cours[f"MM{p}"] = [round(float(v), 2) if v == v else None for v in sma_tail]
|
|
|
|
# Volumes pour le graphique
|
|
vol_chart = df_chart["vol"].astype(float).tolist()
|
|
vol_moy_20 = close.rolling(window=20).mean() # pas le bon, il faut volume
|
|
vol_series = df["vol"].astype(float)
|
|
vm20_chart = vol_series.rolling(window=20).mean().tail(nb_points).tolist()
|
|
graphique_volumes = {
|
|
"dates": graphique_cours["dates"],
|
|
"volumes": [int(v) if v == v else 0 for v in vol_chart],
|
|
"moyenne_20j": [round(float(v), 0) if v == v else None for v in vm20_chart],
|
|
}
|
|
|
|
# Volatilité pour le graphique (20j, fenêtre glissante)
|
|
ret = close.pct_change()
|
|
vol_20j_series = (ret.rolling(window=20).std() * np_sqrt_252() * 100).tail(nb_points).tolist()
|
|
graphique_volatilite = {
|
|
"dates": graphique_cours["dates"],
|
|
"volatilite_20j": [round(float(v), 2) if v == v else None for v in vol_20j_series],
|
|
}
|
|
|
|
# --- 6) Assemblage du résultat JSON ---
|
|
cours_actuel = round(float(df["price"].iloc[-1]), 2)
|
|
derniere_date = df["date"].iloc[-1].strftime("%Y-%m-%d")
|
|
var_jour = performances.get("variation_journaliere")
|
|
|
|
result = {
|
|
"isin": isin,
|
|
"action": info,
|
|
"date_analyse": datetime.now().strftime("%Y-%m-%d %H:%M"),
|
|
"derniere_date_cours": derniere_date,
|
|
"cours": cours_actuel,
|
|
"variation_jour": var_jour,
|
|
"tendance": score_data["tendance"],
|
|
"score": score_data["score"],
|
|
"points_positifs": score_data["points_positifs"],
|
|
"points_negatifs": score_data["points_negatifs"],
|
|
"performances": performances,
|
|
"moyennes_mobiles": moyennes_mobiles,
|
|
"volatilite": volatilite,
|
|
"atr": atr,
|
|
"gap": gaps,
|
|
"rsi": rsi,
|
|
"macd": macd,
|
|
"bollinger": bollinger,
|
|
"supports_resistances": supports_resistances,
|
|
"volumes": volumes,
|
|
"drawdown": drawdown,
|
|
"series": series,
|
|
"chandeliers": chandeliers,
|
|
"graphique_cours": graphique_cours,
|
|
"graphique_volumes": graphique_volumes,
|
|
"graphique_volatilite": graphique_volatilite,
|
|
"erreur": None,
|
|
}
|
|
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers privés
|
|
# ---------------------------------------------------------------------------
|
|
def _is_nan(val):
|
|
"""Vérifie si une valeur est NaN."""
|
|
try:
|
|
return val != val # NaN != NaN est True
|
|
except (TypeError, ValueError):
|
|
return True
|
|
|
|
|
|
def np_sqrt_252():
|
|
"""Retourne sqrt(252) sans import répété."""
|
|
import math
|
|
return math.sqrt(252)
|