caudal/programa/cartera/fuentes.py

"""
fuentes.py — Internet: buscar activos y actualizar sus precios.

Aquí está la BÚSQUEDA (dar de alta un activo escribiendo su nombre o su ISIN) y
el paso de «actualizar los precios de la cartera». De dónde sale cada precio y
cómo se lee cada web es cosa de `precios.py`.

Solo se envían identificadores públicos (ISIN, ticker, nombre buscado): nunca
cantidades ni importes.
"""

from __future__ import annotations

import datetime as dt
from concurrent.futures import ThreadPoolExecutor

from . import precios

TIPOS_AUTOMATICOS = {"Fondo", "PlanPensiones", "Cripto", "Metal", "ETF", "Accion"}
ISIN = precios.ISIN
UA = precios.UA

comprobar = precios.comprobar          # «Comprobar las fuentes» (Más → …)


def _sesion():
    import requests
    return requests.Session()


# ── Búsqueda ─────────────────────────────────────────────────────────────────

def _morningstar(q: str, sesion) -> list[dict]:
    params = {"page": "1", "pageSize": "8", "outputType": "json", "version": "1",
              "languageId": "es-ES", "currencyId": "EUR", "universeIds": "FOESP$$ALL",
              "securityDataPoints": "SecId|Name|Isin|ClosePrice|ClosePriceDate"}
    if ISIN.match(q.upper()):
        params["filters"] = f"ISIN:IN:{q.upper()}"
    else:
        params["term"] = q
    r = sesion.get("https://lt.morningstar.com/api/rest.svc/klr5zyak8x/security/screener",
                   params=params, headers={**UA, "Referer": "https://www.morningstar.es/"}, timeout=12)
    r.raise_for_status()
    return [{"nombre": f.get("Name", ""), "codigo": f.get("Isin", ""), "tipo": "Fondo",
             "unidad": "participaciones", "precio": f.get("ClosePrice"),
             "fecha_precio": (f.get("ClosePriceDate") or "")[:10], "moneda": "EUR", "fuente": "Morningstar"}
            for f in r.json().get("rows", []) if f.get("Isin")]


def _yahoo(q: str, sesion) -> list[dict]:
    # Yahoo responde 429 a cabeceras de navegador completas; con esta, no
    for host in ("query2", "query1"):
        r = sesion.get(f"https://{host}.finance.yahoo.com/v1/finance/search",
                       params={"q": q, "quotesCount": "8", "newsCount": "0"},
                       headers=precios.UA_SIMPLE, timeout=12)
        if r.status_code != 429:
            break
    r.raise_for_status()
    tipos = {"ETF": "ETF", "EQUITY": "Accion", "MUTUALFUND": "Fondo"}
    salida = []
    for f in r.json().get("quotes", []):
        tipo = tipos.get(f.get("quoteType"))
        if not tipo or not f.get("symbol"):
            continue
        salida.append({"nombre": f.get("longname") or f.get("shortname") or f["symbol"],
                       "codigo": f["symbol"], "tipo": tipo,
                       "unidad": "participaciones" if tipo != "Accion" else "acciones",
                       "precio": None, "fecha_precio": None, "moneda": None,
                       "fuente": f"Yahoo · {f.get('exchDisp', '')}".strip(" ·")})
    return salida


def _coingecko(q: str, sesion) -> list[dict]:
    r = sesion.get("https://api.coingecko.com/api/v3/search", params={"query": q}, headers=UA, timeout=12)
    r.raise_for_status()
    return [{"nombre": f"{c.get('name')} ({c.get('symbol')})", "codigo": c["id"], "tipo": "Cripto",
             "unidad": c.get("symbol", ""), "precio": None, "fecha_precio": None, "moneda": None,
             "fuente": "CoinGecko"}
            for c in r.json().get("coins", [])[:5] if c.get("id")]


def _ls(q: str, sesion) -> list[dict]:
    """LS Exchange: solo busca por ISIN."""
    if not ISIN.match(q.upper()):
        return []
    x = precios._ls_instrumento(q, sesion)
    if not x:
        return []
    tipo = {"STK": "Accion", "ETF": "ETF", "FND": "Fondo"}.get(x.get("categorySymbol"), "Accion")
    p = precios.ls_exchange(q, sesion)
    return [{"nombre": " ".join((x.get("displayname") or q).split()), "codigo": q.upper(), "tipo": tipo,
             "unidad": "acciones" if tipo == "Accion" else "participaciones",
             "precio": p.precio if p else None, "fecha_precio": p.fecha.isoformat() if p else None,
             "moneda": "EUR", "fuente": "LS Exchange"}]


def buscar(q: str) -> dict:
    """Busca en paralelo; devuelve resultados y qué fuentes han fallado."""
    q = (q or "").strip()
    if len(q) < 2:
        return {"resultados": [], "errores": []}
    sesion = _sesion()
    fuentes = {"Morningstar": _morningstar, "Yahoo": _yahoo}
    if ISIN.match(q.upper()):
        fuentes["LS Exchange"] = _ls
    else:
        fuentes["CoinGecko"] = _coingecko
    resultados, errores = [], []
    with ThreadPoolExecutor(max_workers=3) as ex:
        futuros = {nombre: ex.submit(f, q, sesion) for nombre, f in fuentes.items()}
        for nombre, fut in futuros.items():
            try:
                resultados += fut.result(timeout=20)
            except Exception:
                errores.append(nombre)
    return {"resultados": sorted(resultados, key=lambda x: _relevancia(q, x)), "errores": errores}


def _relevancia(q: str, x: dict) -> tuple:
    """Primero el código exacto; luego fondos/ETF/acciones; las cripto al final
    salvo que se busque justo su símbolo (evita que el ticker de una acción case con tokens de CoinGecko)."""
    qn = q.strip().lower()
    codigo = (x.get("codigo") or "").lower()
    exacto = codigo == qn or codigo.split(".")[0] == qn or f"({qn})" in x.get("nombre", "").lower() and x["tipo"] == "Cripto"
    return (0 if codigo == qn or codigo.split(".")[0] == qn else 1 if exacto else 2,
            1 if x["tipo"] == "Cripto" else 0,
            0 if x.get("fuente", "").startswith("LS") else 1)


# ── Precios ──────────────────────────────────────────────────────────────────

def precio(activo: dict, sesion=None) -> tuple[dict | None, str]:
    """(resultado, mensajes) — resultado = {activo, fecha, precio, fuente}."""
    if activo["tipo"] not in TIPOS_AUTOMATICOS or not activo.get("codigo"):
        return None, "precio manual"
    avisos: list[str] = []
    try:
        p = precios.de_activo(activo, sesion or _sesion(), avisos)
    except Exception as e:                  # una fuente rota no debe tumbar la app
        return None, f"error: {e}"
    if p is None or p.precio <= 0:
        return None, "; ".join(avisos) or "sin precio"
    return ({"activo": activo["id"], "fecha": p.fecha.isoformat(),
             "precio": float(p.precio), "fuente": p.fuente}, "; ".join(avisos))


def actualizar(activos: list[dict]) -> list[dict]:
    """Descarga el precio de cada activo abierto con fuente automática."""
    sesion = _sesion()
    informe = []
    for a in activos:
        if a["tipo"] not in TIPOS_AUTOMATICOS or a.get("estado") == "cerrada":
            continue
        r, msg = precio(a, sesion)
        informe.append({"activo": a["id"], "nombre": a.get("nombre", ""), "ok": r is not None,
                        "resultado": r, "mensaje": "" if r else msg})
    return informe


def cambio(fecha: dt.date, moneda: str) -> float | None:
    """Euros por unidad de moneda en esa fecha (o el día hábil anterior)."""
    if moneda == "EUR":
        return 1.0
    try:
        return precios.tipo_cambio(fecha, moneda, "eur", _sesion())
    except Exception:
        return None