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

import pandas as pd
import yaml
import yfinance as yf

ROOT = 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."""
    path = path or PORTFOLIO_FILE
    with open(path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)


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