mirror of
https://github.com/Alkatrazz24/Funk-lab.git
synced 2026-07-09 10:54:43 +02:00
ALI YESILKAYA
c182bac1d2
Permet de suivre plusieurs comptes (courtiers/banques) côte à côte, avec vue par compte et agrégée. Ajoute le PEA BNP Paribas à côté de Revolut. - portfolio.yaml : nouveau format `accounts` (name/type/cash/positions par compte). Compat ascendante : ancien format (cash/positions à la racine) lu comme compte unique « Principal » - data.py : load_accounts(), base_currency(), portfolio_tickers() (union dédupliquée) - tracker.py : build_account / build_all (par compte + agrégat global) ; build() reste la vue globale (digest/report) ; report() multi-comptes - scanner/technical/alerts : helpers « portfolio » → union de tous les comptes - dashboard /api/portfolio : renvoie accounts[] (positions+agg par compte) + global - dashboard front : onglets de comptes (Tous / Revolut / BNP Paribas), KPI + positions + secteurs qui suivent le périmètre sélectionné, badge de type par ligne en vue Tous - BNP Paribas (PEA) saisi depuis la capture : AI/PAEEM/DCAM/PCEU/PSP5/BNP/ENGI/TTE .PA (tickers Yahoo vérifiés, cours conformes à la capture) Vérifié : totaux conformes à la capture BNP (6 395,49 € ; P&L -39 €), global 22 444 € ; bascule d'onglet OK au navigateur (Playwright) ; compile + JS + CLI report OK. Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
"""Évaluation des alertes : ne renvoie que les ÉVÉNEMENTS du jour (transitions).
|
|
|
|
Économe en tokens : au lieu de déverser tout l'état technique, on ne remonte
|
|
que ce qui vient de basculer (croisement MACD, cassure MM50, survente...).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pandas as pd
|
|
import yaml
|
|
|
|
from . import data, indicators as ind, scanner
|
|
|
|
# Config lue depuis FINLAB_HOME (workspace persistant) si défini — cf. data.ROOT.
|
|
ALERTS_FILE = Path(os.environ.get("FINLAB_HOME") or Path(__file__).resolve().parent.parent) / "alerts.yaml"
|
|
|
|
|
|
def _events(symbol: str) -> dict:
|
|
"""Calcule les flags d'événement (aujourd'hui vs hier) pour un titre."""
|
|
df = data.history(symbol, period="1y")
|
|
close = df["Close"]
|
|
rsi = ind.rsi(close)
|
|
macd = ind.macd(close)
|
|
m, s = macd["macd"], macd["signal"]
|
|
mm50 = ind.sma(close, 50)
|
|
mm200 = ind.sma(close, 200)
|
|
|
|
def crossed_up(a, b):
|
|
return a.iloc[-2] <= b.iloc[-2] and a.iloc[-1] > b.iloc[-1]
|
|
|
|
def crossed_down(a, b):
|
|
return a.iloc[-2] >= b.iloc[-2] and a.iloc[-1] < b.iloc[-1]
|
|
|
|
return {
|
|
"rsi": float(rsi.iloc[-1]),
|
|
"macd_cross_up": crossed_up(m, s),
|
|
"macd_cross_down": crossed_down(m, s),
|
|
"cross_above_mm50": crossed_up(close, mm50),
|
|
"cross_below_mm50": crossed_down(close, mm50),
|
|
"golden_cross": crossed_up(mm50, mm200) if mm200.notna().iloc[-1] else False,
|
|
"death_cross": crossed_down(mm50, mm200) if mm200.notna().iloc[-1] else False,
|
|
"price": round(float(close.iloc[-1]), 2),
|
|
}
|
|
|
|
|
|
def _matches(rule: dict, ev: dict) -> bool:
|
|
w = rule["when"]
|
|
if w == "rsi_below":
|
|
return ev["rsi"] < rule["value"]
|
|
if w == "rsi_above":
|
|
return ev["rsi"] > rule["value"]
|
|
return bool(ev.get(w, False))
|
|
|
|
|
|
def _watchlist(name: str) -> list[str]:
|
|
if name == "portfolio":
|
|
return data.portfolio_tickers() # union de tous les comptes
|
|
return scanner.load_theme(name) # 'all' ou un thème
|
|
|
|
|
|
def run(watch: str | None = None) -> list[dict]:
|
|
"""Renvoie la liste des alertes déclenchées : [{ticker, prix, alerte}]."""
|
|
cfg = yaml.safe_load(open(ALERTS_FILE, encoding="utf-8"))
|
|
rules = cfg["rules"]
|
|
syms = _watchlist(watch or cfg.get("watch", "all"))
|
|
|
|
hits = []
|
|
for sym in syms:
|
|
try:
|
|
ev = _events(sym)
|
|
except Exception:
|
|
continue
|
|
for rule in rules:
|
|
if _matches(rule, ev):
|
|
hits.append({"ticker": sym, "prix": ev["price"], "alerte": rule["name"]})
|
|
return hits
|
|
|
|
|
|
def render(hits: list[dict]) -> str:
|
|
if not hits:
|
|
return "Aucune alerte déclenchée aujourd'hui."
|
|
df = pd.DataFrame(hits).sort_values("alerte")
|
|
return df.to_string(index=False)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
print(render(run(sys.argv[1] if len(sys.argv) > 1 else None)))
|