OpenFin/backend/services/price_history.py

"""
Price History — fetch + cache des cours historiques via yfinance.
Cache SQLite dans la table price_history_cache.
"""
import json
import sqlite3
from datetime import datetime, timedelta, date as date_type
from typing import Optional


YF_TICKERS: dict[str, str] = {
    "EURUSD": "EURUSD=X",
    "USDJPY": "USDJPY=X",
    "XAUUSD": "GC=F",
    "SP500":  "^GSPC",
    "TLT":    "TLT",
    "GBPUSD": "GBPUSD=X",
    "EEM":    "EEM",
    "QQQ":    "QQQ",
}

PRICE_LABELS: dict[str, str] = {
    "EURUSD": "EUR/USD",
    "USDJPY": "USD/JPY",
    "XAUUSD": "Or ($/oz)",
    "SP500":  "S&P 500",
    "TLT":    "TLT ETF",
    "GBPUSD": "GBP/USD",
    "EEM":    "EEM ETF",
    "QQQ":    "QQQ ETF",
}

PERIOD_DAYS: dict[str, int] = {
    "5d": 7, "1mo": 35, "3mo": 95, "6mo": 190, "1y": 370, "2y": 740,
}


def _ensure_cache_table(conn):
    conn.execute("""
        CREATE TABLE IF NOT EXISTS price_history_cache (
            id          INTEGER PRIMARY KEY,
            instrument  TEXT NOT NULL,
            date        TEXT NOT NULL,
            close       REAL NOT NULL,
            fetched_at  TEXT DEFAULT (datetime('now')),
            UNIQUE(instrument, date)
        )
    """)
    conn.execute("""
        CREATE INDEX IF NOT EXISTS idx_price_cache_inst_date
        ON price_history_cache(instrument, date)
    """)
    conn.commit()


def _fetch_yf(instrument: str, period_days: int) -> list[dict]:
    """Fetch from yfinance. Returns [{date, close}] sorted ascending."""
    try:
        import yfinance as yf
        ticker = YF_TICKERS.get(instrument.upper())
        if not ticker:
            return []
        # yfinance period string
        if period_days <= 7:   p = "5d"
        elif period_days <= 35:  p = "1mo"
        elif period_days <= 95:  p = "3mo"
        elif period_days <= 190: p = "6mo"
        elif period_days <= 370: p = "1y"
        else:                    p = "2y"

        df = yf.download(ticker, period=p, interval="1d", progress=False, auto_adjust=True)
        if df is None or df.empty:
            return []

        # Handle MultiIndex columns (yfinance 0.2+)
        if hasattr(df.columns, 'levels'):
            df.columns = df.columns.get_level_values(0)

        close_col = next((c for c in ["Close", "Adj Close", "close"] if c in df.columns), None)
        if not close_col:
            return []

        result = []
        for idx, row in df.iterrows():
            dt = str(idx)[:10]
            v  = float(row[close_col])
            if v and v == v:  # not NaN
                result.append({"date": dt, "close": round(v, 6)})
        return result

    except Exception:
        return []


def get_price_history(
    conn,
    instrument: str,
    period: str = "1y",
    force_refresh: bool = False,
) -> list[dict]:
    """
    Retourne [{date, close}] pour l'instrument sur la période.
    Cache SQLite — rafraîchit si les données datent de plus de 6h.
    """
    _ensure_cache_table(conn)
    inst = instrument.upper()
    days = PERIOD_DAYS.get(period, 370)
    date_from = (datetime.utcnow().date() - timedelta(days=days)).isoformat()

    # Check cache freshness
    cache_ok = False
    if not force_refresh:
        newest = conn.execute(
            "SELECT fetched_at FROM price_history_cache WHERE instrument=? ORDER BY fetched_at DESC LIMIT 1",
            (inst,)
        ).fetchone()
        if newest:
            try:
                age_h = (datetime.utcnow() - datetime.fromisoformat(str(newest[0])[:19])).total_seconds() / 3600
                cache_ok = age_h < 6.0
            except Exception:
                pass

    if not cache_ok:
        rows = _fetch_yf(inst, days + 30)
        if rows:
            conn.executemany(
                "INSERT OR REPLACE INTO price_history_cache (instrument, date, close) VALUES (?,?,?)",
                [(inst, r["date"], r["close"]) for r in rows]
            )
            conn.commit()

    # Read from cache
    rows_db = conn.execute(
        "SELECT date, close FROM price_history_cache WHERE instrument=? AND date>=? ORDER BY date ASC",
        (inst, date_from)
    ).fetchall()
    return [{"date": r[0], "close": r[1]} for r in rows_db]


def calibrate_intercept(
    conn,
    instrument: str,
    model_pips_at_ref: float,
    ref_date: Optional[str] = None,
) -> Optional[float]:
    """
    Calcule l'intercept = real_price - model_pips × pip_to_price au point de référence.
    Si ref_date non fourni, utilise il y a 30 jours.
    """
    from services.instrument_models import INSTRUMENT_MODELS
    meta = INSTRUMENT_MODELS.get(instrument.upper(), {})
    pip_to_price = meta.get("pip_to_price", 0.0001)

    if ref_date is None:
        ref_date = (datetime.utcnow().date() - timedelta(days=30)).isoformat()

    # Find nearest price to ref_date
    row = conn.execute(
        """SELECT date, close FROM price_history_cache
           WHERE instrument=? AND date<=? ORDER BY date DESC LIMIT 1""",
        (instrument.upper(), ref_date)
    ).fetchone()
    if not row:
        # Try fetching
        history = get_price_history(conn, instrument, "3mo", force_refresh=True)
        row = conn.execute(
            "SELECT date, close FROM price_history_cache WHERE instrument=? AND date<=? ORDER BY date DESC LIMIT 1",
            (instrument.upper(), ref_date)
        ).fetchone()
    if not row:
        return None

    real_price = float(row[1])
    intercept  = real_price - model_pips_at_ref * pip_to_price
    return round(intercept, 6)