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.
126 lines
5.8 KiB
Python
126 lines
5.8 KiB
Python
"""
|
|
Pruebas del precio de mercado de la renta fija (cartera/bme.py y
|
|
msgpack_min.py) con un BME de mentira: no salen a internet.
|
|
|
|
python3 -m unittest pruebas.test_bme -v
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from cartera import bme, rentafija # noqa: E402
|
|
from cartera import msgpack_min as mp # noqa: E402
|
|
|
|
|
|
def fila(isin, ultimo=0.0, compra=0.0, venta=0.0, seg="SD", grupo="DE"):
|
|
return ["20261015", seg, grupo, "BO", "Bonos y Obligaciones", f"BONO {isin}", isin, "T1",
|
|
1.0, 1000.0, 3.0, compra, venta, 2.9, 1000.0, 1.0, ultimo, 3.0, 5000.0, True]
|
|
|
|
|
|
class Respuesta:
|
|
def __init__(self, contenido=b"", js=None):
|
|
self.content, self._js = contenido, js
|
|
|
|
def json(self):
|
|
return self._js
|
|
|
|
def raise_for_status(self):
|
|
pass
|
|
|
|
|
|
class BMEdeMentira:
|
|
"""Contesta como el canal SignalR de BME: saludo y una respuesta por invocación."""
|
|
|
|
def __init__(self, mercados: dict):
|
|
self.mercados, self.pendiente, self.saludo = mercados, None, True
|
|
|
|
def post(self, url, data=b"", headers=None, timeout=None):
|
|
if "negotiate" in url:
|
|
return Respuesta(js={"connectionToken": "t"})
|
|
if isinstance(data, bytes) and data:
|
|
for inv in mp.marcos(data):
|
|
args = inv[4][0]
|
|
filas = [f for f in self.mercados.get((args["Segment"], args["Group"]), []) if args["Search"] in f[6]]
|
|
self.pendiente = [3, {}, inv[2], 3, {"Data": filas, "TotalResults": len(filas)}]
|
|
return Respuesta()
|
|
|
|
def get(self, url, headers=None, timeout=None):
|
|
cuerpo = b"{}\x1e" if self.saludo else b""
|
|
self.saludo = False
|
|
if self.pendiente:
|
|
cuerpo += mp.marco(mp.empaquetar(self.pendiente))
|
|
self.pendiente = None
|
|
return Respuesta(cuerpo)
|
|
|
|
|
|
class TestMessagePack(unittest.TestCase):
|
|
def test_ida_y_vuelta(self):
|
|
dato = [1, {}, "1", "GetSessionPrices", [{"Segment": "SD", "Page": 0, "Precio": 98.5, "Vale": True, "Nada": None}], []]
|
|
self.assertEqual(mp.desempaquetar(mp.empaquetar(dato))[0], dato)
|
|
self.assertEqual(list(mp.marcos(mp.marco(mp.empaquetar("hola")) * 2)), ["hola", "hola"])
|
|
|
|
|
|
class TestConsulta(unittest.TestCase):
|
|
def test_el_ultimo_precio_o_el_medio_y_en_el_primer_mercado_que_lo_tenga(self):
|
|
sesion = BMEdeMentira({("SD", "DE"): [fila("ES0000012M77", ultimo=98.949), fila("ES0000012P90", compra=96.0, venta=98.0)],
|
|
("MV", ""): [fila("ES0000099999", ultimo=101.5, seg="MV", grupo="DS")]})
|
|
r = bme.consultar(["es0000012m77", "ES0000012P90", "ES0000099999", "ES0000000001", "no-es-un-isin"], sesion)
|
|
self.assertEqual((r["ES0000012M77"]["precio"], r["ES0000012M77"]["origen"]), (98.949, "último precio negociado"))
|
|
self.assertEqual((r["ES0000012P90"]["precio"], r["ES0000012P90"]["origen"]), (97.0, "punto medio entre compra y venta"))
|
|
self.assertEqual((r["ES0000099999"]["mercado"], r["ES0000099999"]["fecha"]), ("SENAF", "2026-10-15"))
|
|
self.assertNotIn("ES0000000001", r)
|
|
|
|
def test_sin_compra_ni_venta_no_hay_precio(self):
|
|
self.assertIsNone(bme.precio(dict(zip(bme.CAMPOS, fila("ES0000012M77")))))
|
|
|
|
|
|
class TestValorDeMercado(unittest.TestCase):
|
|
def test_nominal_por_precio_mas_el_cupon_corrido(self):
|
|
a = {"id": "RF1", "tipo": "RentaFija", "renta_fija": {"clase": "bono", "nominal": 10000, "interes": 2.5,
|
|
"vencimiento": "2027-05-31", "pagos": 1}}
|
|
movs = [{"id": "m1", "activo": "RF1", "tipo": "Compra", "fecha": "2024-06-10", "importe": 9900, "cantidad": 10000}]
|
|
# 31-may → 15-oct: 137 días de cupón corrido al 2,5 %
|
|
self.assertEqual(rentafija.valor_de_mercado(a, movs, dt.date(2026, 10, 15), 98.949),
|
|
round(10000 * 0.98949 + 10000 * 0.025 * 137 / 365, 2))
|
|
movs.append({"id": "m2", "activo": "RF1", "tipo": "Reembolso", "fecha": "2026-01-10", "importe": 5000, "cantidad": 5000})
|
|
self.assertEqual(rentafija.valor_de_mercado(a, movs, dt.date(2026, 10, 15), 100.0),
|
|
round(5000 + 5000 * 0.025 * 137 / 365, 2))
|
|
|
|
|
|
class TestAlActualizarPrecios(unittest.TestCase):
|
|
def test_guarda_el_valor_de_mercado_y_sin_cotizacion_lo_dice(self):
|
|
import tempfile
|
|
from unittest import mock
|
|
from cartera import servidor
|
|
from cartera.datos import Datos, escribir
|
|
from cartera.operaciones import Cartera
|
|
t = tempfile.TemporaryDirectory()
|
|
self.addCleanup(t.cleanup)
|
|
carpeta = Path(t.name) / "datos"
|
|
escribir(carpeta, Datos([], [], [], []))
|
|
c = Cartera(carpeta)
|
|
self.addCleanup(c.cerrar)
|
|
for isin in ("ES0000012M77", "ES0000000001"):
|
|
a = c.guardar_activo({"tipo": "RentaFija", "nombre": isin, "codigo": isin,
|
|
"renta_fija": {"clase": "bono", "nominal": 10000, "interes": 2.5, "vencimiento": "2027-05-31", "pagos": 1}})
|
|
c.guardar_movimiento({"activo": a["id"], "tipo": "Compra", "fecha": "2024-06-10", "importe": 9900, "cantidad": 10000})
|
|
app = servidor.App(c)
|
|
hoy = dt.date.today().isoformat()
|
|
with mock.patch.object(bme, "consultar", return_value={"ES0000012M77": {"fecha": hoy, "precio": 99.0, "origen": "último",
|
|
"mercado": "AIAF deuda pública", "nombre": ""}}):
|
|
informe = app._precios_renta_fija()
|
|
self.assertEqual([i["ok"] for i in informe], [True, False])
|
|
r = [x for x in c.datos.vivos("precios") if x.get("fuente") == "BME"]
|
|
self.assertEqual(len(r), 1)
|
|
self.assertGreater(r[0]["valor_total"], 9900)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|