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 math
from pathlib import Path

import pandas as pd
import yaml
from fastapi import FastAPI, Request
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}


@app.get("/api/portfolio")
def api_portfolio():
    df, agg = tracker.build(with_sector=True)
    base = agg["base"]
    positions = [
        {
            "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["poids_%"],
        }
        for r in df.to_dict(orient="records")
    ]
    return _clean({
        "base": base, "total": agg["total"], "cash": agg["cash"],
        "invested": agg["invested"], "pnl_total": agg["pnl_total"], "pnl_pct": agg["pnl_pct"],
        "by_sector": agg["by_sector"].to_dict(),
        "positions": positions,
    })


@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)})


# ── 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)