esetheyers
7e18576f86
Taxes for the whole of Spain (common territory, Navarre, Álava, Bizkaia and Gipuzkoa); flats, garages and land valued on their own; Treasury bills, bonds and deposits priced on BME; broker accounts; a guided start; a page per investment with its chart and the tax of selling today; net worth at the top; undo when deleting; Woodpecker CI for the tests and the Linux build.
101 lines
5 KiB
Python
101 lines
5 KiB
Python
"""
|
|
bme.py — El precio de mercado de bonos, obligaciones y Letras en BME.
|
|
|
|
BME publica en su web los precios de la sesión de renta fija (AIAF y SENAF)
|
|
a través de un canal SignalR. Aquí se usa su modo «long polling», que son
|
|
peticiones HTTP normales, y el protocolo MessagePack (msgpack_min.py):
|
|
|
|
1. POST …/negotiate?negotiateVersion=1 → connectionToken
|
|
2. POST …?id=TOKEN {"protocol":"messagepack","version":1}\\x1e
|
|
3. POST …?id=TOKEN la invocación GetSessionPrices (Segment, Group, Search)
|
|
4. GET …?id=TOKEN → «{}\\x1e» del saludo y la respuesta, en marcos MessagePack
|
|
|
|
Cada fila de la respuesta es una lista sin nombres de campo; su orden se
|
|
dedujo de las columnas de la web (ver CAMPOS). Los precios van en % del
|
|
nominal y sin el cupón corrido. Solo se pregunta por los ISIN de la cartera.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import json
|
|
import re
|
|
|
|
from . import msgpack_min as mp
|
|
|
|
URL = "https://apiweb.bolsasymercados.es/Market/v1/SignalR/FI/Prices"
|
|
ORIGEN = "https://www.bolsasymercados.es"
|
|
# (segmento, grupo, nombre): deuda pública, corporativa y extranjera de AIAF, SENAF y MARF
|
|
MERCADOS = [("SD", "DE", "AIAF deuda pública"), ("MV", "", "SENAF"), ("SD", "CP,DT", "AIAF deuda corporativa"),
|
|
("SD", "EX", "AIAF deuda extranjera"), ("AF", "", "MARF")]
|
|
CAMPOS = ["fecha", "segmento", "grupo", "tipo", "nombre_tipo", "nombre", "isin", "ticker",
|
|
"ordenes_compra", "importe_compra", "tir_compra", "precio_compra", "precio_venta", "tir_venta",
|
|
"importe_venta", "ordenes_venta", "ultimo", "tir_ultimo", "volumen"]
|
|
ISIN = re.compile(r"^[A-Z]{2}[A-Z0-9]{9}[0-9]$")
|
|
|
|
|
|
class Canal:
|
|
"""Una conexión al canal de precios de BME."""
|
|
|
|
def __init__(self, sesion):
|
|
self.s = sesion
|
|
cab = {"Origin": ORIGEN, "Content-Length": "0"}
|
|
r = self.s.post(f"{URL}/negotiate?negotiateVersion=1", data=b"", headers=cab, timeout=30)
|
|
r.raise_for_status()
|
|
self.url = f"{URL}?id={r.json()['connectionToken']}"
|
|
self.s.post(self.url, data=json.dumps({"protocol": "messagepack", "version": 1}) + "\x1e",
|
|
headers={"Origin": ORIGEN}, timeout=30)
|
|
self.n = 0
|
|
|
|
def precios(self, segmento: str, grupo: str, buscar: str) -> list[dict]:
|
|
self.n += 1
|
|
ident = str(self.n)
|
|
inv = [1, {}, ident, "GetSessionPrices",
|
|
[{"Language": "es", "Segment": segmento, "Group": grupo, "Search": buscar, "Page": 0, "PageSize": 0}], []]
|
|
self.s.post(self.url, data=mp.marco(mp.empaquetar(inv)),
|
|
headers={"Origin": ORIGEN, "Content-Type": "application/octet-stream"}, timeout=30)
|
|
for _ in range(6):
|
|
crudo = self.s.get(self.url, headers={"Origin": ORIGEN}, timeout=60).content
|
|
if crudo.startswith(b"{"): # la respuesta al saludo
|
|
crudo = crudo[crudo.index(b"\x1e") + 1:]
|
|
for m in mp.marcos(crudo):
|
|
if isinstance(m, list) and m and m[0] == 3 and m[2] == ident:
|
|
if m[3] == 1: # error del servidor
|
|
raise RuntimeError(str(m[4]))
|
|
datos = (m[4] if len(m) > 4 and isinstance(m[4], dict) else {}).get("Data") or []
|
|
return [dict(zip(CAMPOS, fila)) for fila in datos if isinstance(fila, list)]
|
|
if isinstance(m, list) and m and m[0] == 7:
|
|
raise RuntimeError("BME cerró la conexión")
|
|
raise RuntimeError("BME no contestó")
|
|
|
|
|
|
def precio(fila: dict) -> tuple[float, str] | None:
|
|
"""(precio en % del nominal, de dónde sale): el último negociado o, si no
|
|
hubo operaciones, el punto medio entre la mejor compra y la mejor venta."""
|
|
if (fila.get("ultimo") or 0) > 0:
|
|
return float(fila["ultimo"]), "último precio negociado"
|
|
compra, venta = fila.get("precio_compra") or 0, fila.get("precio_venta") or 0
|
|
if compra > 0 and venta > 0:
|
|
return round((compra + venta) / 2, 4), "punto medio entre compra y venta"
|
|
return None
|
|
|
|
|
|
def consultar(isins: list[str], sesion) -> dict[str, dict]:
|
|
"""El precio de cada ISIN en el primer mercado de BME que lo tenga:
|
|
{isin: {fecha, precio, origen, mercado, nombre}}. Los que no cotizan, fuera."""
|
|
buscados = [i.strip().upper() for i in isins if ISIN.match((i or "").strip().upper())]
|
|
if not buscados:
|
|
return {}
|
|
canal = Canal(sesion)
|
|
salida = {}
|
|
for isin in buscados:
|
|
for seg, grupo, nombre in MERCADOS:
|
|
filas = [f for f in canal.precios(seg, grupo, isin) if f.get("isin") == isin]
|
|
p = next((precio(f) for f in filas if precio(f)), None)
|
|
if p:
|
|
fila = filas[0]
|
|
fecha = str(fila.get("fecha") or dt.date.today().strftime("%Y%m%d"))
|
|
salida[isin] = {"fecha": f"{fecha[:4]}-{fecha[4:6]}-{fecha[6:8]}", "precio": p[0], "origen": p[1],
|
|
"mercado": nombre, "nombre": fila.get("nombre", "")}
|
|
break
|
|
return salida
|