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

"""Couche d'accès aux données de marché (Yahoo Finance via yfinance).

Centralise les appels réseau, le cache devises et la lecture du portefeuille
pour que tous les outils partagent la même source.
"""
from __future__ import annotations

import functools
import os
import pickle
import time
from pathlib import Path

import pandas as pd
import yaml
import yfinance as yf

# FINLAB_HOME découple le CODE (package) de la DONNÉE (configs éditables + cache). En
# déploiement, le code vit dans l'image (/opt/app) et les configs dans le workspace persistant
# (PVC). Non défini → repli sur le dossier parent du package (dév local inchangé).
ROOT = Path(os.environ.get("FINLAB_HOME") or Path(__file__).resolve().parent.parent)
PORTFOLIO_FILE = ROOT / "portfolio.yaml"
CACHE_DIR = ROOT / ".cache"

# Durées de fraîcheur du cache disque (secondes)
TTL_HISTORY = 3600      # cours : 1h (suffisant pour des scans intraday espacés)
TTL_INFO = 7 * 86400    # fondamentaux/secteur : 1 semaine (change peu)
TTL_PRICE = 600         # dernier cours : 10 min


def _disk_cache(key: str, ttl: int, producer):
    """Renvoie l'objet caché si frais (< ttl), sinon le (re)calcule et le stocke."""
    CACHE_DIR.mkdir(exist_ok=True)
    path = CACHE_DIR / (key.replace("/", "_").replace("=", "_") + ".pkl")
    if path.exists() and (time.time() - path.stat().st_mtime) < ttl:
        try:
            with open(path, "rb") as f:
                return pickle.load(f)
        except Exception:
            pass  # cache corrompu -> on recalcule
    obj = producer()
    with open(path, "wb") as f:
        pickle.dump(obj, f)
    return obj


def load_portfolio(path: Path | None = None) -> dict:
    """Charge portfolio.yaml (brut)."""
    path = path or PORTFOLIO_FILE
    with open(path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)


def base_currency() -> str:
    return load_portfolio().get("base_currency", "EUR")


def load_accounts() -> list[dict]:
    """Comptes normalisés [{name, type, cash, positions}].

    Compat ascendante : un fichier à l'ancien format (cash/positions à la racine,
    sans `accounts`) est présenté comme un compte unique « Principal »."""
    raw = load_portfolio()
    if "accounts" in raw:
        return raw["accounts"]
    return [{
        "name": raw.get("name", "Principal"),
        "type": raw.get("type", "courtier"),
        "cash": raw.get("cash", {"amount": 0.0, "currency": raw.get("base_currency", "EUR")}),
        "positions": raw.get("positions", []),
    }]


def portfolio_tickers() -> list[str]:
    """Union dédupliquée des tickers de tous les comptes (ordre stable)."""
    seen: dict[str, None] = {}
    for acc in load_accounts():
        for p in acc["positions"]:
            seen.setdefault(p["ticker"], None)
    return list(seen)


@functools.lru_cache(maxsize=64)
def _ticker(symbol: str) -> yf.Ticker:
    return yf.Ticker(symbol)


def history(symbol: str, period: str = "1y", interval: str = "1d", fresh: bool = False) -> pd.DataFrame:
    """Historique OHLCV (caché sur disque). fresh=True force un re-fetch."""
    def fetch():
        df = _ticker(symbol).history(period=period, interval=interval, auto_adjust=True)
        if df.empty:
            raise ValueError(f"Aucune donnée pour {symbol} (period={period})")
        return df

    ttl = 0 if fresh else TTL_HISTORY
    return _disk_cache(f"hist_{symbol}_{period}_{interval}", ttl, fetch)


def last_price(symbol: str, fresh: bool = False) -> tuple[float, str]:
    """(dernier cours, devise) en devise native du titre (caché 10 min)."""
    def fetch():
        fi = _ticker(symbol).fast_info
        return float(fi.last_price), fi.currency

    return _disk_cache(f"price_{symbol}", 0 if fresh else TTL_PRICE, fetch)


def day_change(symbol: str, fresh: bool = False) -> float | None:
    """Variation du jour en % (dernier cours vs clôture précédente), caché 10 min."""
    def fetch():
        fi = _ticker(symbol).fast_info
        last, prev = fi.last_price, fi.previous_close
        return round((last - prev) / prev * 100, 2) if (last and prev) else None

    try:
        return _disk_cache(f"day_{symbol}", 0 if fresh else TTL_PRICE, fetch)
    except Exception:
        return None


def company_name(symbol: str) -> str:
    """Nom court de la société (via info, caché 1 semaine)."""
    try:
        return info(symbol).get("shortName") or info(symbol).get("longName") or symbol
    except Exception:
        return symbol


def fx_rate(base: str, quote: str) -> float:
    """Taux de change : combien de `quote` pour 1 `base` (ex: EUR->USD ~ 1.07)."""
    if base == quote:
        return 1.0

    def fetch():
        df = _ticker(f"{base}{quote}=X").history(period="5d", auto_adjust=True)
        if df.empty:
            raise ValueError(f"Pas de taux {base}->{quote}")
        return float(df["Close"].iloc[-1])

    return _disk_cache(f"fx_{base}{quote}", TTL_PRICE, fetch)


def to_currency(amount: float, frm: str, to: str) -> float:
    """Convertit un montant d'une devise vers une autre."""
    if frm == to:
        return amount
    # On passe par EUR->X pour limiter le nombre de paires interrogées.
    try:
        return amount * fx_rate(frm, to)
    except ValueError:
        return amount / fx_rate(to, frm)


def info(symbol: str) -> dict:
    """Métadonnées fondamentales (caché 1 semaine ; peut être lent / vide)."""
    return _disk_cache(f"info_{symbol}", TTL_INFO, lambda: _ticker(symbol).get_info())


def clear_cache() -> int:
    """Vide le cache disque. Renvoie le nombre de fichiers supprimés."""
    if not CACHE_DIR.exists():
        return 0
    files = list(CACHE_DIR.glob("*.pkl"))
    for f in files:
        f.unlink()
    return len(files)