Funk-lab/tools/finlab/finlab/tracker.py

"""Suivi de portefeuille multi-comptes : valorisation, P&L, secteurs, concentration."""
from __future__ import annotations

import pandas as pd

from . import data


def _sector(symbol: str) -> str:
    try:
        return data.info(symbol).get("sector") or "—"
    except Exception:
        return "—"


def _positions_df(positions: list[dict], base: str, with_sector: bool = True) -> pd.DataFrame:
    """Tableau valorisé d'une liste de positions (sans les poids — dépendent du total)."""
    rows = []
    for p in positions:
        sym = p["ticker"]
        price, cur = data.last_price(sym)
        value = data.to_currency(price * p["shares"], cur, base)
        cost = data.to_currency(p["avg_price"] * p["shares"], p["avg_currency"], base)
        pnl = value - cost
        rows.append({
            "ticker": sym,
            "secteur": _sector(sym) if with_sector else "—",
            "qté": p["shares"],
            "cours": round(price, 2),
            "dev": cur,
            f"valeur_{base}": round(value, 2),
            f"PRU_{base}": round(cost, 2),
            f"P&L_{base}": round(pnl, 2),
            "P&L_%": round(100 * pnl / cost, 2) if cost else 0.0,
        })
    return pd.DataFrame(rows)


def _agg(df: pd.DataFrame, cash: float, base: str) -> dict:
    """Agrégats à partir d'un tableau de positions + liquidités (en devise de base)."""
    invested = df[f"valeur_{base}"].sum() if len(df) else 0.0
    total = invested + cash
    if len(df) and total:
        df["poids_%"] = (df[f"valeur_{base}"] / total * 100).round(2)
        by_sector = (df.groupby("secteur")[f"valeur_{base}"].sum().sort_values(ascending=False)
                     / total * 100).round(2)
        pnl_total = df[f"P&L_{base}"].sum()
        cost_total = df[f"PRU_{base}"].sum()
        top_weight = df.nlargest(3, "poids_%")[["ticker", "poids_%"]]
    else:
        by_sector = pd.Series(dtype=float)
        pnl_total = cost_total = 0.0
        top_weight = pd.DataFrame(columns=["ticker", "poids_%"])
    return {
        "base": base,
        "cash": round(cash, 2),
        "invested": round(invested, 2),
        "total": round(total, 2),
        "pnl_total": round(pnl_total, 2),
        "pnl_pct": round(100 * pnl_total / cost_total, 2) if cost_total else 0.0,
        "by_sector": by_sector,
        "top_weight": top_weight,
    }


def build_account(acc: dict, base: str, with_sector: bool = True) -> tuple[pd.DataFrame, dict]:
    """(tableau, agrégats) d'UN compte. Poids relatifs au total de ce compte."""
    df = _positions_df(acc["positions"], base, with_sector)
    cash = data.to_currency(acc["cash"]["amount"], acc["cash"]["currency"], base)
    agg = _agg(df, cash, base)
    df = df.sort_values(f"valeur_{base}", ascending=False) if len(df) else df
    return df, agg


def build_all(with_sector: bool = True) -> tuple[list[dict], dict]:
    """Tous les comptes (par compte) + l'agrégat global (poids relatifs au total global)."""
    base = data.base_currency()
    accounts = data.load_accounts()
    per = []
    all_positions, total_cash = [], 0.0
    for acc in accounts:
        df, agg = build_account(acc, base, with_sector)
        per.append({"name": acc["name"], "type": acc.get("type", "courtier"), "df": df, "agg": agg})
        all_positions += acc["positions"]
        total_cash += data.to_currency(acc["cash"]["amount"], acc["cash"]["currency"], base)
    gdf = _positions_df(all_positions, base, with_sector)
    gagg = _agg(gdf, total_cash, base)
    return per, gagg


def build(with_sector: bool = True) -> tuple[pd.DataFrame, dict]:
    """Vue GLOBALE (tous comptes fusionnés). Compat : digest/report/CLI."""
    base = data.base_currency()
    all_positions, total_cash = [], 0.0
    for acc in data.load_accounts():
        all_positions += acc["positions"]
        total_cash += data.to_currency(acc["cash"]["amount"], acc["cash"]["currency"], base)
    df = _positions_df(all_positions, base, with_sector)
    agg = _agg(df, total_cash, base)
    df = df.sort_values(f"valeur_{base}", ascending=False) if len(df) else df
    return df, agg


def report() -> str:
    per, agg = build_all()
    b = agg["base"]
    out = ["═" * 64, "  PORTEFEUILLE (tous comptes)", "═" * 64]
    for acc in per:
        a = acc["agg"]
        sign = "+" if a["pnl_total"] >= 0 else ""
        out.append(f"\n▸ {acc['name']} ({acc['type']}) — {a['total']:,.2f} {b} "
                   f"| P&L {sign}{a['pnl_total']:,.2f} {b} ({sign}{a['pnl_pct']}%) | cash {a['cash']:,.2f} {b}")
        if len(acc["df"]):
            out.append(acc["df"].to_string(index=False))
    sign = "+" if agg["pnl_total"] >= 0 else ""
    out += ["", "─" * 64,
            f"TOTAL GLOBAL : {agg['total']:>12,.2f} {b}  "
            f"(investi {agg['invested']:,.2f} + cash {agg['cash']:,.2f})",
            f"P&L GLOBAL   : {sign}{agg['pnl_total']:>11,.2f} {b}  ({sign}{agg['pnl_pct']}%)"]
    out.append("\nExposition sectorielle (globale) :")
    for sec, w in agg["by_sector"].items():
        out.append(f"  {sec:<24} {w:>6.2f}%  {'█' * int(w / 2)}")
    return "\n".join(out)


if __name__ == "__main__":
    print(report())