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

"""Import d'un relevé de compte Revolut (« trading account statement » CSV) → positions.

Le relevé liste TOUTES les transactions (BUY/SELL/CASH/DIVIDEND). On reconstruit les positions
actuelles par méthode du **coût moyen** : pour chaque titre, qté nette = achats − ventes ;
PRU = coût moyen des achats (une vente ne modifie pas le PRU, elle réduit la quantité).

→ **Idempotent** : relancer avec un relevé à jour (export sur toute la période) reconstruit le
portefeuille exact. C'est la procédure de mise à jour « à chaque trade ».

Met à jour **uniquement** le compte ciblé (par défaut « Revolut ») dans portfolio.yaml ; les
autres comptes (ex. BNP Paribas), les liquidités et les commentaires sont **préservés**
(round-trip ruamel).

Usage :
  python -m finlab.revolut <account-statement.csv>            # aperçu (ne touche rien)
  python -m finlab.revolut <account-statement.csv> --write    # met à jour portfolio.yaml
  python -m finlab.revolut <csv> --account Revolut --no-verify
"""
from __future__ import annotations

import csv
import re
import sys
from collections import defaultdict
from pathlib import Path

from . import data

# Tickers Revolut non standards → symboles Yahoo Finance (places EU).
REVOLUT_TO_YAHOO = {
    "ASME": "ASML.AS",   # ASML (Amsterdam)
    "TOTB": "TTE.PA",    # TotalEnergies (Paris)
    "BNP": "BNP.PA",     # BNP Paribas
    "AXA": "CS.PA",      # AXA
    "FTE": "ORA.PA",     # Orange
    "AIR1": "AI.PA",     # Air Liquide
}


def _num(s: str) -> float:
    return float(re.sub(r"[^0-9.\-]", "", s or "0"))


def _ccy(s: str) -> str | None:
    m = re.match(r"\s*([A-Z]{3})", s or "")
    return m.group(1) if m else None


def parse_statement(path: str | Path) -> list[dict]:
    """Transactions BUY/SELL d'un relevé de compte Revolut, triées par date."""
    txs = []
    with open(path, encoding="utf-8") as f:
        for r in csv.DictReader(f):
            tk = (r.get("Ticker") or "").strip()
            t = (r.get("Type") or "")
            side = "BUY" if t.startswith("BUY") else ("SELL" if t.startswith("SELL") else None)
            if not tk or side is None:
                continue
            price_field = r.get("Price per share") or ""
            txs.append({
                "date": r.get("Date", ""),
                "rev": tk,
                "side": side,
                "qty": float(r["Quantity"]),
                "price": _num(price_field),
                "currency": _ccy(price_field) or r.get("Currency"),
            })
    txs.sort(key=lambda x: x["date"])
    return txs


def reconstruct(txs: list[dict], eps: float = 1e-6) -> list[dict]:
    """Positions nettes (coût moyen). Renvoie [{rev, qty, pru, currency}] (qté > 0)."""
    st = defaultdict(lambda: {"qty": 0.0, "cost": 0.0, "currency": None})
    for tx in txs:
        s = st[tx["rev"]]
        s["currency"] = tx["currency"]
        if tx["side"] == "BUY":
            s["qty"] += tx["qty"]
            s["cost"] += tx["qty"] * tx["price"]
        else:  # SELL — coût moyen : PRU inchangé, on réduit la quantité
            if s["qty"] > eps:
                avg = s["cost"] / s["qty"]
                s["qty"] = max(s["qty"] - tx["qty"], 0.0)
                s["cost"] = avg * s["qty"]
            else:
                s["qty"] -= tx["qty"]  # vente sans achat connu → négatif (historique incomplet)
    out = []
    for rev, s in st.items():
        if s["qty"] > eps:
            out.append({"rev": rev, "qty": s["qty"], "pru": s["cost"] / s["qty"],
                        "currency": s["currency"]})
    return out


def to_positions(holdings: list[dict], verify: bool = True) -> tuple[list[dict], list[str]]:
    """Convertit en positions portfolio.yaml (mapping Yahoo + vérif cours). Renvoie (positions, warnings)."""
    positions, warns = [], []
    for h in sorted(holdings, key=lambda x: x["rev"]):
        ya = REVOLUT_TO_YAHOO.get(h["rev"], h["rev"])
        if verify:
            try:
                fi = data._ticker(ya).fast_info
                lp, lc = fi.last_price, fi.currency
                if lp is None:
                    warns.append(f"{h['rev']}→{ya}: cours indisponible")
                elif lc != h["currency"]:
                    warns.append(f"{h['rev']}→{ya}: devise {lc}≠{h['currency']} (mapping à vérifier)")
            except Exception as e:
                warns.append(f"{h['rev']}→{ya}: {e}")
        positions.append({
            "ticker": ya,
            "shares": round(h["qty"], 8),
            "avg_price": round(h["pru"], 4),
            "avg_currency": h["currency"],
        })
    return positions, warns


def update_portfolio(positions: list[dict], account: str = "Revolut",
                     path: Path | None = None) -> None:
    """Remplace les positions du compte `account` dans portfolio.yaml (préserve le reste)."""
    from ruamel.yaml import YAML
    from ruamel.yaml.comments import CommentedMap
    path = path or data.PORTFOLIO_FILE
    yaml = YAML()
    yaml.preserve_quotes = True
    yaml.width = 4096  # pas de retour à la ligne au milieu des maps inline
    yaml.indent(mapping=2, sequence=4, offset=2)  # style: « accounts:\n  - name: ... »
    doc = yaml.load(open(path, encoding="utf-8"))
    accounts = doc.get("accounts")
    if accounts is None:
        raise SystemExit("portfolio.yaml n'est pas au format multi-comptes (clé 'accounts').")
    target = next((a for a in accounts if a.get("name") == account), None)
    if target is None:
        raise SystemExit(f"compte '{account}' introuvable dans portfolio.yaml.")
    seq = []
    for p in positions:
        cm = CommentedMap(p)
        cm.fa.set_flow_style()  # rendu inline { ticker: ..., shares: ... }
        seq.append(cm)
    target["positions"] = seq
    with open(path, "w", encoding="utf-8") as f:
        yaml.dump(doc, f)


def main() -> None:
    args = sys.argv[1:]
    if not args:
        print(__doc__)
        return
    csv_path = args[0]
    write = "--write" in args
    verify = "--no-verify" not in args
    account = "Revolut"
    if "--account" in args:
        account = args[args.index("--account") + 1]

    txs = parse_statement(csv_path)
    holdings = reconstruct(txs)
    positions, warns = to_positions(holdings, verify=verify)

    print(f"=== {len(positions)} positions reconstruites (compte « {account} ») ===")
    for p in positions:
        print(f"  {p['ticker']:9} {p['shares']:>14.8f} @ {p['avg_price']:>10.4f} {p['avg_currency']}")
    for w in warns:
        print(f"  ⚠ {w}")
    if write:
        update_portfolio(positions, account=account)
        print(f"\n✅ portfolio.yaml mis à jour (compte « {account} »). Autres comptes/cash préservés.")
    else:
        print("\n(aperçu — relance avec --write pour écrire dans portfolio.yaml)")


if __name__ == "__main__":
    main()