""" precios.py — De dónde sale el precio de cada activo. Cada sitio del que se descarga un precio tiene aquí su función, y todas tienen la misma forma: funcion(identificador, sesion, avisos) -> Precio | None Ninguna deja escapar una excepción: si la fuente falla, devuelve None y escribe el motivo en `avisos`. Así una web caída nunca tumba la actualización entera. `de_activo()` es la que elige la fuente según el tipo de activo, y `comprobar()` prueba todas de una vez con identificadores públicos para saber cuál está rota HOY, sin esperar a que un precio se quede viejo. PRIVACIDAD: solo salen identificadores públicos (ISIN, ticker, id de CoinGecko). Las cantidades, los importes y las ganancias no se envían a ningún sitio. POR QUÉ ESTÁ SOLO EN SU FICHERO: es la parte que más se rompe —en agosto de 2026 se cayeron a la vez Morningstar, la CNMV y la web de una gestora—, así que vive aparte y con sus propias pruebas: cuando una fuente cambia, se arregla aquí sin tocar nada más. """ from __future__ import annotations import datetime as dt import json import re import time from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass NAVEGADOR = "Mozilla/5.0 (X11; Linux x86_64; rv:120.0) Gecko/20100101 Firefox/120.0" UA = {"User-Agent": NAVEGADOR} UA_SIMPLE = {"User-Agent": "Mozilla/5.0"} # Yahoo y LS dan 429 con uno completo UA_PROPIO = {"User-Agent": "cartera-inversor/1.0"} ESPERA_MORNINGSTAR = 1.5 # segundos entre consultas, por cortesía ISIN = re.compile(r"^[A-Z]{2}[A-Z0-9]{9}\d$") # Identificador de Morningstar (0P…), que Yahoo usa para fondos y planes: 0P0000ABCD.F ID_MORNINGSTAR = re.compile(r"^0P[0-9A-Z]{8}(\.[A-Z]{1,3})?$") YAHOO_GRAFICO = "https://query1.finance.yahoo.com/v8/finance/chart/{simbolo}" FAWAZ = "https://cdn.jsdelivr.net/npm/@fawazahmed0/currency-api@{version}/v1/currencies/{moneda}.json" LS = "https://www.ls-tc.de/_rpc/json" LS_WEB = "https://www.ls-tc.de" @dataclass class Precio: """Un precio descargado. `fecha` es la del precio, no la de hoy.""" activo: str codigo: str precio: float fecha: dt.date fuente: str def _avisar(avisos: list[str] | None, texto: str) -> None: """Apunta un problema; sin lista donde apuntarlo, lo imprime.""" if avisos is None: print(f" ⚠ {texto}") else: avisos.append(texto) def _fecha(crudo, formatos=("%Y-%m-%dT%H:%M:%S", "%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y")) -> dt.date: for f in formatos: try: return dt.datetime.strptime(str(crudo)[:19], f).date() except (ValueError, TypeError): continue return dt.date.today() def _numero(texto: str) -> float | None: """«1.234,56 €» → 1234.56. Devuelve None si no es un número positivo.""" t = str(texto).replace("€", "").replace("\xa0", "").replace("%", "") t = t.replace(".", "").replace(",", ".").strip() try: v = float(t) except ValueError: return None return v if v > 0 else None def _sopa(html: str): from bs4 import BeautifulSoup, FeatureNotFound try: return BeautifulSoup(html, "lxml") except FeatureNotFound: # sin lxml instalado: el analizador que trae Python, algo más lento return BeautifulSoup(html, "html.parser") # ── Fondos y planes de pensiones ───────────────────────────────────────────── def morningstar(isin: str, sesion, avisos: list[str] | None = None) -> Precio | None: """Fuente principal de fondos y planes con ISIN. `tools.morningstar.es` dejó de resolver en agosto de 2026 (su CNAME apunta a un destino sin registro A). `lt.morningstar.com` expone la misma API con la misma clave. Se prueban en orden y vale el primero que conteste. """ hosts = ("lt.morningstar.com", "tools.morningstar.es", "tools.morningstar.co.uk") params = {"page": "1", "pageSize": "5", "sortOrder": "ClosePriceDate desc", "outputType": "json", "version": "1", "languageId": "es-ES", "currencyId": "EUR", "universeIds": "FOESP$$ALL|FOESP$$PENSION", "securityDataPoints": "SecId,ISIN,Name,ClosePrice,ClosePriceDate,CloseDate", "filters": f"ISIN:IN:{isin}"} cabeceras = {**UA, "Referer": "https://www.morningstar.es/", "Accept": "application/json"} ultimo = None for host in hosts: try: r = sesion.get(f"https://{host}/api/rest.svc/klr5zyak8x/security/screener", params=params, headers=cabeceras, timeout=15) r.raise_for_status() filas = r.json().get("rows", []) if not filas: return None # el ISIN no está, no es un fallo de la fuente fila = filas[0] # Llega como número JSON (200.8208), no como texto español valor = float(str(fila.get("ClosePrice", "0")).replace(",", ".")) if valor <= 0: return None # El campo de fecha cambió de nombre en 2026; se aceptan los dos. crudo = fila.get("ClosePriceDate") or fila.get("CloseDate") or "" return Precio("", isin, valor, _fecha(crudo), "morningstar") except Exception as e: ultimo = f"{host}: {e}" if ultimo: _avisar(avisos, f"Morningstar: {ultimo}") return None # Las fuentes de la web de una gestora concreta no van en el núcleo: quien las # tenga (cartera/extras) las apunta aquí al cargarse. FONDOS_POR_ISIN: list = [] # f(isin, sesion, nombre, avisos) → Precio | None, tras Morningstar FONDOS_POR_NOMBRE: list = [] # f(nombre, sesion, avisos) → Precio | None, para fondos y planes sin ISIN SONDAS_EXTRA: list = [] # (clave, nombre, para qué sirve, sonda) LIMPIEZAS: list = [] # funciones que olvidan lo descargado def limpiar_cache() -> None: """Olvida lo descargado en esta ejecución (la usan las pruebas).""" _cache_ls.clear() for limpieza in LIMPIEZAS: limpieza() def cnmv(isin: str, sesion, avisos: list[str] | None = None) -> Precio | None: """Último respaldo de fondos. Desde 2026 la ficha ya no trae el VL: avisa.""" try: r = sesion.get("https://www.cnmv.es/portal/Consultas/IIC/Fondo.aspx", params={"isin": isin}, headers=UA_SIMPLE, timeout=20) r.raise_for_status() sopa = _sopa(r.text) celda = sopa.find("td", string=lambda s: s and "Valor liquidativo" in s) if not celda: _avisar(avisos, "CNMV: la ficha ya no publica el valor liquidativo") return None siguiente = celda.find_next_sibling("td") if not siguiente: return None valor = _numero(siguiente.get_text(strip=True).split()[0]) if valor is None: return None fecha = dt.date.today() celda_fecha = sopa.find("td", string=lambda s: s and "Fecha valor liquidativo" in s) if celda_fecha and celda_fecha.find_next_sibling("td"): fecha = _fecha(celda_fecha.find_next_sibling("td").get_text(strip=True), ("%d/%m/%Y",)) return Precio("", isin, valor, fecha, "cnmv") except ImportError: return None except Exception as e: _avisar(avisos, f"CNMV: {e}") return None # ── Cripto, metales, acciones y ETF ────────────────────────────────────────── def coingecko(moneda: str, sesion, avisos: list[str] | None = None) -> Precio | None: try: r = sesion.get("https://api.coingecko.com/api/v3/simple/price", params={"ids": moneda, "vs_currencies": "eur", "include_last_updated_at": "true"}, headers={**UA_PROPIO, "Accept": "application/json"}, timeout=15) r.raise_for_status() datos = r.json() if moneda not in datos: return None cuando = datos[moneda].get("last_updated_at") fecha = dt.datetime.fromtimestamp(cuando).date() if cuando else dt.date.today() return Precio("", moneda, float(datos[moneda]["eur"]), fecha, "coingecko") except Exception as e: _avisar(avisos, f"CoinGecko: {e}") return None def metal(codigo: str, sesion, avisos: list[str] | None = None) -> Precio | None: """Oro (xau) y plata (xag) en euros por onza troy.""" codigo_min = codigo.lower() try: r = sesion.get(FAWAZ.format(version="latest", moneda=codigo_min), headers=UA_PROPIO, timeout=15) r.raise_for_status() datos = r.json() return Precio("", codigo, float(datos[codigo_min]["eur"]), _fecha(datos.get("date", str(dt.date.today())), ("%Y-%m-%d",)), "fawaz") except Exception as e: _avisar(avisos, f"Fawaz API ({codigo}): {e}") return None def _yahoo_grafico(simbolo: str, sesion, avisos: list[str] | None = None) -> Precio | None: """El último cierre publicado de un símbolo, de la serie diaria del último mes de Yahoo. En euros, cotice en lo que cotice.""" try: r = sesion.get(YAHOO_GRAFICO.format(simbolo=simbolo), params={"range": "1mo", "interval": "1d"}, headers=UA_SIMPLE, timeout=20) r.raise_for_status() serie = (r.json().get("chart", {}).get("result") or [None])[0] if not serie: _avisar(avisos, f"Yahoo no tiene {simbolo}") return None cierres = serie["indicators"]["quote"][0]["close"] ultimo = next(((t, c) for t, c in zip(reversed(serie["timestamp"]), reversed(cierres)) if c and c > 0), None) if ultimo is None: return None valor = float(ultimo[1]) moneda = serie.get("meta", {}).get("currency") or "EUR" if moneda != "EUR": valor = a_euros(valor, moneda, sesion) if valor is None: return None fecha = dt.datetime.fromtimestamp(ultimo[0], dt.timezone.utc).date() return Precio("", simbolo, valor, fecha, "yahoo") except Exception as e: _avisar(avisos, f"Yahoo ({simbolo}): {e}") return None def yahoo(ticker: str, sesion, avisos: list[str] | None = None) -> Precio | None: """Acciones y ETF por ticker. Convierte a euros si cotiza en otra moneda.""" return _yahoo_grafico(ticker.strip().upper(), sesion, avisos) def yahoo_fondo(codigo: str, sesion, avisos: list[str] | None = None) -> Precio | None: """Fondos y planes sin ISIN por su identificador de Morningstar (0P…), en Yahoo. Sirve de segunda fuente para los planes que solo publica la gestora: cuando su web se atasca, Yahoo a veces va unos días por delante (y al revés). Lee la serie diaria del último mes y se queda con el último cierre publicado.""" return _yahoo_grafico(codigo.upper() if "." in codigo else f"{codigo.upper()}.F", sesion, avisos) _cache_ls: dict[str, dict | None] = {} def _ls_instrumento(isin: str, sesion) -> dict | None: isin = isin.upper() if isin not in _cache_ls: r = sesion.get(f"{LS}/.lstc/instrument/search/main", params={"q": isin, "localeId": "2"}, headers=UA_SIMPLE, timeout=12) r.raise_for_status() _cache_ls[isin] = next((x for x in r.json() if (x.get("isin") or "").upper() == isin), None) return _cache_ls[isin] def _ls_compra(instrumento: dict, sesion) -> float | None: """El precio de COMPRA (bid, «Geld») de la ficha del valor: lo que te pagarían ahora por venderlo. Sale del HTML de la ficha; si cambia, no hay precio.""" enlace = instrumento.get("link") if not enlace: return None r = sesion.get(f"{LS_WEB}{enlace}", headers=UA_SIMPLE, timeout=12) r.raise_for_status() m = re.search(r'item="%s@1"\s+field="bid"[^>]*>\s*([\d.,]+)\s*<' % re.escape(str(instrumento.get("id"))), r.text) return _numero(m.group(1)) if m else None def ls_exchange(isin: str, sesion, avisos: list[str] | None = None) -> Precio | None: """Lang & Schwarz (LS Exchange), un mercado alemán que cotiza en euros. Al cotizar en euros no hay que convertir divisas ni fiarse de cierres en otra plaza, y es el precio que enseñan varios brókers. Se valora al precio de COMPRA (bid), que es lo que enseñan las aplicaciones de los brókers para lo que ya tienes: lo que cobrarías si vendieras ahora. El gráfico da el precio medio entre compra y venta, que es algo mayor; se usa para la fecha y como respaldo si la ficha no se puede leer. """ medio = _ls_medio(isin, sesion, avisos) if medio is None: return None try: compra = _ls_compra(_ls_instrumento(isin, sesion), sesion) except Exception as e: compra = None _avisar(avisos, f"LS Exchange ({isin}): no leo el precio de compra y uso el medio ({e})") # Solo si es creíble (el gráfico va unos minutos por detrás, así que puede quedar algo por encima) if compra and abs(compra / medio.precio - 1) <= 0.1: return Precio("", isin, compra, medio.fecha, "ls-exchange") return medio def _ls_medio(isin: str, sesion, avisos: list[str] | None = None) -> Precio | None: """El precio medio entre compra y venta: la última cotización del gráfico.""" try: instrumento = _ls_instrumento(isin, sesion) if not instrumento: return None r = sesion.get(f"{LS}/instrument/chart/dataForInstrument", params={"instrumentId": instrumento["id"]}, headers=UA_SIMPLE, timeout=12) r.raise_for_status() datos = r.json() puntos = [p for p in (((datos.get("series") or {}).get("intraday") or {}).get("data") or []) if isinstance(p, list) and len(p) == 2 and p[1]] if puntos: cuando, valor = puntos[-1] return Precio("", isin, float(valor), dt.datetime.fromtimestamp(cuando / 1000).date(), "ls-exchange") # Sin cotizaciones hoy (fin de semana o festivo): el cierre anterior for linea in (datos.get("info") or {}).get("plotlines") or []: if linea.get("id") == "previousDay" and linea.get("value"): dia = dt.date.today() - dt.timedelta(days=1) while dia.weekday() >= 5: dia -= dt.timedelta(days=1) return Precio("", isin, float(linea["value"]), dia, "ls-exchange") return None except Exception as e: _avisar(avisos, f"LS Exchange ({isin}): {e}") return None # ── Divisas ────────────────────────────────────────────────────────────────── def a_euros(importe: float, moneda: str, sesion) -> float | None: """Convierte un importe a euros al cambio de hoy.""" codigo = moneda.lower() if codigo == "eur": return importe try: r = sesion.get(FAWAZ.format(version="latest", moneda=codigo), timeout=10) r.raise_for_status() return importe * float(r.json()[codigo]["eur"]) except Exception: return None def tipo_cambio(fecha: dt.date, origen: str, destino: str, sesion) -> float | None: """Cambio origen→destino de esa fecha. Si ese día no cotiza (fin de semana o festivo), retrocede hasta seis días.""" for atras in range(7): dia = fecha - dt.timedelta(days=atras) try: r = sesion.get(FAWAZ.format(version=dia.isoformat(), moneda=origen.lower()), timeout=10) if r.status_code != 200: continue valor = r.json().get(origen.lower(), {}).get(destino.lower()) if valor and float(valor) > 0: return float(valor) except Exception: continue return None # ── La fuente que le toca a cada activo ────────────────────────────────────── def de_activo(activo: dict, sesion, avisos: list[str] | None = None) -> Precio | None: """El precio de un activo, probando sus fuentes en orden.""" tipo = activo.get("tipo") codigo = (activo.get("codigo") or "").strip() nombre = activo.get("nombre", "") if not codigo: return None es_isin = bool(ISIN.match(codigo.upper())) resultado = None if tipo in ("Fondo", "PlanPensiones"): if es_isin: resultado = morningstar(codigo, sesion, avisos) for fuente in FONDOS_POR_ISIN: if resultado is None: resultado = fuente(codigo, sesion, nombre, avisos) if resultado is None: resultado = cnmv(codigo, sesion, avisos) else: # Sin ISIN: la web de la gestora, por nombre, y si el código es el # identificador de Morningstar (0P…), también Yahoo. Gana la fecha # más reciente; a igualdad, la gestora. candidatos = [fuente(nombre, sesion, avisos) for fuente in FONDOS_POR_NOMBRE] if ID_MORNINGSTAR.match(codigo.upper()): candidatos.append(yahoo_fondo(codigo, sesion, avisos)) validos = [c for c in candidatos if c is not None] resultado = max(validos, key=lambda c: c.fecha) if validos else None time.sleep(ESPERA_MORNINGSTAR) elif tipo == "Cripto": resultado = coingecko(codigo, sesion, avisos) elif tipo == "Metal": resultado = metal(codigo, sesion, avisos) elif tipo in ("ETF", "Accion"): # Con ISIN, en euros desde LS Exchange; si no, por ticker if es_isin: resultado = ls_exchange(codigo, sesion, avisos) if resultado is None: _avisar(avisos, "LS Exchange no tiene ese ISIN") else: resultado = yahoo(codigo, sesion, avisos) if resultado: resultado.activo = activo.get("id", "") return resultado # ── Comprobar las fuentes ──────────────────────────────────────────────────── # Identificadores PÚBLICOS y genéricos, elegidos a propósito para que la # comprobación no diga nada de la cartera de nadie: un fondo indexado mundial # muy extendido, el ETF más común del mundo, bitcoin y el oro. SONDA_FONDO = "IE00B03HCZ61" # Vanguard Global Stock Index Fund EUR Acc SONDA_ETF = "IE00B5BMR087" # iShares Core S&P 500 SONDA_TICKER = "SPY" SONDA_CRIPTO = "bitcoin" SONDA_METAL = "xau" def _euros(valor: float) -> str: """1234.5 → «1.234,5000 €»""" return f"{valor:,.4f}".replace(",", "@").replace(".", ",").replace("@", ".") + " €" def _bien(precio: Precio, extra: str = "") -> dict: return {"ok": True, "detalle": f"{_euros(precio.precio)} del {precio.fecha.strftime('%d/%m/%Y')}{extra}"} def _sonda_morningstar(sesion) -> dict: avisos: list[str] = [] p = morningstar(SONDA_FONDO, sesion, avisos) if p: return _bien(p) return {"ok": False, "detalle": "; ".join(avisos) or f"no devuelve nada para {SONDA_FONDO}"} def _sonda_cnmv(sesion) -> dict: avisos: list[str] = [] p = cnmv(SONDA_FONDO, sesion, avisos) if p: return _bien(p) return {"ok": False, "detalle": "; ".join(avisos) or "sin valor liquidativo", "esperado": True} def _sonda_coingecko(sesion) -> dict: avisos: list[str] = [] p = coingecko(SONDA_CRIPTO, sesion, avisos) return _bien(p) if p else {"ok": False, "detalle": "; ".join(avisos) or "sin precio"} def _sonda_metal(sesion) -> dict: avisos: list[str] = [] p = metal(SONDA_METAL, sesion, avisos) return _bien(p, " · la onza") if p else {"ok": False, "detalle": "; ".join(avisos) or "sin precio"} def _sonda_yahoo(sesion) -> dict: avisos: list[str] = [] p = yahoo(SONDA_TICKER, sesion, avisos) return _bien(p) if p else {"ok": False, "detalle": "; ".join(avisos) or "sin precio"} def _sonda_ls(sesion) -> dict: avisos: list[str] = [] p = ls_exchange(SONDA_ETF, sesion, avisos) return _bien(p) if p else {"ok": False, "detalle": "; ".join(avisos) or "sin precio"} def _sonda_cambio(sesion) -> dict: ayer = dt.date.today() - dt.timedelta(days=1) valor = tipo_cambio(ayer, "usd", "eur", sesion) if valor: return {"ok": True, "detalle": f"1 $ = {valor:.4f} €".replace(".", ",")} return {"ok": False, "detalle": "no devuelve el cambio del dólar"} # (clave, nombre, para qué sirve, sonda) SONDAS = ( ("morningstar", "Morningstar", "fondos y planes con ISIN", _sonda_morningstar), ("cnmv", "CNMV", "último respaldo de fondos (retirado desde 2026)", _sonda_cnmv), ("coingecko", "CoinGecko", "criptomonedas", _sonda_coingecko), ("fawaz", "Fawaz", "oro y plata", _sonda_metal), ("ls-exchange", "LS Exchange", "acciones y ETF con ISIN", _sonda_ls), ("yahoo", "Yahoo Finance", "acciones y ETF por ticker", _sonda_yahoo), ("cambio", "Cambio de divisa", "convertir dólares a euros", _sonda_cambio), ) def comprobar(sesion=None) -> dict: """Prueba todas las fuentes a la vez y dice cuál contesta y cuál no. Consulta internet de verdad: es justo lo que se quiere saber. Usa solo los identificadores públicos de arriba, nunca los activos de la cartera. La CNMV lleva rota desde 2026 y se sabe: sale aparte, como «retirada», para que no parezca una avería nueva. """ sondas = [*SONDAS, *SONDAS_EXTRA] if sesion is None: import requests sesion = requests.Session() limpiar_cache() resultados: dict[str, dict] = {} with ThreadPoolExecutor(max_workers=5) as ejecutor: lanzadas = {clave: ejecutor.submit(sonda, sesion) for clave, _, _, sonda in sondas} for clave, nombre, para_que, _ in sondas: try: r = lanzadas[clave].result(timeout=60) except Exception as e: r = {"ok": False, "detalle": str(e)} resultados[clave] = {"fuente": clave, "nombre": nombre, "para": para_que, **r} lista = [resultados[c] for c, _, _, _ in sondas] rotas = [r for r in lista if not r["ok"] and not r.get("esperado")] return {"fuentes": lista, "rotas": [r["nombre"] for r in rotas], "fecha": dt.datetime.now().isoformat(timespec="seconds")} if __name__ == "__main__": # python3 -m cartera.precios informe = comprobar() ancho = max(len(f["nombre"]) for f in informe["fuentes"]) for f in informe["fuentes"]: marca = "OK " if f["ok"] else ("-- " if f.get("esperado") else "ROTA") print(f"{marca} {f['nombre']:<{ancho}} {f['detalle']}") print("\nTodo en orden." if not informe["rotas"] else "\nRotas: " + ", ".join(informe["rotas"])) # Las fuentes de gestoras concretas, si esta instalación las trae try: from .extras import precios as _extras # noqa: E402,F401 except ImportError: pass