esetheyers
cf7cc9a8c7
Diseño para móvil, programa de escritorio para Linux (un solo fichero) y versiones firmadas: manifiesto de huellas, firma PGP y la clave del autor (VERIFICAR.md).
536 lines
23 KiB
Python
536 lines
23 KiB
Python
"""
|
|
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
|