Funk-lab/tools/finlab/dashboard/server.py

"""FinLab Dashboard — interface web (graphiques marché, portefeuille, watchlists, actions).

Backend FastAPI exposant les données finlab en JSON. **Aucun LLM, aucune clé API** : c'est de
la donnée et de l'analyse technique pure. La partie conversationnelle/agentique vit dans la
console Claude Code (bouton « Console IA » → /console).

Lancement :  uvicorn dashboard.server:app --host 0.0.0.0 --port 8800
"""
from __future__ import annotations

import datetime as dt
import math
from pathlib import Path

import pandas as pd
import yaml
from fastapi import FastAPI, File, Request, UploadFile
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles

from finlab import (
    alerts,
    data,
    indicators as ind,
    plan as plan_mod,
    scanner,
    technical,
    tracker,
)

app = FastAPI(title="FinLab Dashboard")
STATIC = Path(__file__).resolve().parent / "static"


# ── Helpers ───────────────────────────────────────────────────────────────────
def _clean(obj):
    """Rend un objet JSON-safe (NaN/inf → None, types numpy → natifs)."""
    if isinstance(obj, float):
        return None if (math.isnan(obj) or math.isinf(obj)) else obj
    if isinstance(obj, dict):
        return {k: _clean(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [_clean(v) for v in obj]
    if hasattr(obj, "item"):  # numpy scalar
        return _clean(obj.item())
    return obj


def _records(df: pd.DataFrame):
    return _clean(df.where(pd.notna(df), None).to_dict(orient="records"))


@app.exception_handler(Exception)
async def _on_error(request: Request, exc: Exception):
    # Un échec ponctuel (ticker introuvable, Yahoo bridé...) ne doit pas casser le front.
    return JSONResponse(status_code=500, content={"error": str(exc)})


# ── API ───────────────────────────────────────────────────────────────────────
@app.get("/healthz")
def healthz():
    return {"ok": True}


def _positions_of(df, base):
    return [
        {
            "ticker": r["ticker"], "secteur": r["secteur"], "qte": r["qté"],
            "cours": r["cours"], "dev": r["dev"],
            "valeur": r[f"valeur_{base}"], "pru": r[f"PRU_{base}"],
            "pnl": r[f"P&L_{base}"], "pnl_pct": r["P&L_%"], "poids": r.get("poids_%"),
        }
        for r in df.to_dict(orient="records")
    ]


@app.get("/api/portfolio")
def api_portfolio():
    per, gagg = tracker.build_all(with_sector=True)
    base = gagg["base"]
    accounts = [
        {
            "name": a["name"], "type": a["type"],
            "total": a["agg"]["total"], "cash": a["agg"]["cash"], "invested": a["agg"]["invested"],
            "pnl_total": a["agg"]["pnl_total"], "pnl_pct": a["agg"]["pnl_pct"],
            "by_sector": a["agg"]["by_sector"].to_dict(),
            "positions": _positions_of(a["df"], base),
        }
        for a in per
    ]
    return _clean({
        "base": base, "total": gagg["total"], "cash": gagg["cash"], "invested": gagg["invested"],
        "pnl_total": gagg["pnl_total"], "pnl_pct": gagg["pnl_pct"],
        "by_sector": gagg["by_sector"].to_dict(),
        "accounts": accounts,
    })


@app.get("/api/themes")
def api_themes():
    themes = yaml.safe_load(open(scanner.WATCHLISTS_FILE, encoding="utf-8"))["themes"]
    return {"themes": themes}


@app.get("/api/scan")
def api_scan(theme: str = "all", target: float = 5.0, bullish: bool = False):
    df = scanner.scan_theme(theme, target, bullish_only=bullish)
    return {"theme": theme, "target": target, "rows": _records(df)}


# Couches de la chaîne de valeur IA/datacenter, de l'électron au logiciel.
LAYER_ORDER = ["energy_power", "chips", "datacenter_infra", "cables_optical_network", "software_cloud"]


@app.get("/api/layers")
def api_layers(target: float = 5.0):
    """Watchlists regroupées par couche de la chaîne IA, avec biais/signal par titre."""
    themes = yaml.safe_load(open(scanner.WATCHLISTS_FILE, encoding="utf-8"))["themes"]
    order = LAYER_ORDER + [k for k in themes if k not in LAYER_ORDER]
    layers = []
    for key in order:
        if key not in themes:
            continue
        df = scanner.scan(themes[key], target)
        layers.append({"key": key, "tickers": _records(df)})
    return {"layers": layers}


@app.get("/api/ohlc")
def api_ohlc(ticker: str, period: str = "6mo"):
    df = data.history(ticker, period=period)
    idx = [d.strftime("%Y-%m-%d") for d in df.index]
    candles = [
        {"time": t, "open": round(float(o), 2), "high": round(float(h), 2),
         "low": round(float(l), 2), "close": round(float(c), 2)}
        for t, o, h, l, c in zip(idx, df["Open"], df["High"], df["Low"], df["Close"])
    ]
    volume = [
        {"time": t, "value": float(v), "color": "#2a9d8f88" if c >= o else "#e76f5188"}
        for t, v, o, c in zip(idx, df["Volume"], df["Open"], df["Close"])
    ]

    def line(series):
        return [
            {"time": t, "value": round(float(x), 2)}
            for t, x in zip(idx, series) if pd.notna(x)
        ]

    try:
        tech = technical.analyze(ticker, period="1y")
    except Exception:
        tech = None
    return _clean({
        "ticker": ticker, "period": period, "candles": candles, "volume": volume,
        "ma50": line(ind.sma(df["Close"], 50)), "ma200": line(ind.sma(df["Close"], 200)),
        "technical": tech,
    })


@app.get("/api/alerts")
def api_alerts(watch: str = "all"):
    return {"watch": watch, "hits": _clean(alerts.run(watch))}


@app.get("/api/plan")
def api_plan(ticker: str, capital: float = 1427.0):
    p = plan_mod.plan(ticker, capital)
    return _clean({"plan": p, "render": plan_mod.render(p)})


# ── Import de relevés (image → analysée par la Console IA) ─────────────────────
IMPORTS_DIR = data.ROOT / "imports"
_IMG_EXT = {".png", ".jpg", ".jpeg", ".webp", ".gif"}


@app.post("/api/import")
async def api_import(file: UploadFile = File(...)):
    """Enregistre une capture de relevé dans le workspace (imports/) ; la Console IA la scanne
    ensuite pour mettre à jour portfolio.yaml (vision côté abonnement, pas de clé API ici)."""
    ext = Path(file.filename or "").suffix.lower()
    if ext not in _IMG_EXT:
        return JSONResponse(status_code=400, content={"error": f"format image non supporté: {ext or '?'}"})
    IMPORTS_DIR.mkdir(parents=True, exist_ok=True)
    ts = dt.datetime.now().strftime("%Y%m%d-%H%M%S")
    name = f"releve-{ts}{ext}"
    (IMPORTS_DIR / name).write_bytes(await file.read())
    return {"saved": name, "rel": f"imports/{name}"}


@app.get("/api/imports")
def api_imports():
    if not IMPORTS_DIR.exists():
        return {"imports": []}
    files = sorted((f for f in IMPORTS_DIR.iterdir() if f.suffix.lower() in _IMG_EXT),
                   key=lambda f: f.stat().st_mtime, reverse=True)
    return {"imports": [f.name for f in files]}


# ── Statique ──────────────────────────────────────────────────────────────────
@app.get("/")
def index():
    return FileResponse(STATIC / "index.html")


app.mount("/static", StaticFiles(directory=STATIC), name="static")


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8800)