""" 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