189 lines
8.6 KiB
Python
189 lines
8.6 KiB
Python
"""
|
||
Motor de cálculo y ficheros de datos. Solo usan la cartera FICTICIA (sin datos reales).
|
||
|
||
python3 -m unittest discover -s pruebas -v
|
||
|
||
La cartera ficticia (`pruebas/cartera_ficticia/`) reúne todos los casos raros de
|
||
una cartera de verdad: traspasos entre fondos, ventas parciales, reembolsos,
|
||
comisiones, préstamos, cuentas de trading... y `cifras_esperadas.json` guarda lo
|
||
que el programa debe dar con esos datos a una fecha fija. Si una cifra cambia
|
||
sin querer al tocar el motor, aquí salta.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import datetime as dt
|
||
import json
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
PROYECTO = Path(__file__).resolve().parent.parent
|
||
sys.path.insert(0, str(PROYECTO))
|
||
|
||
from cartera.calculo import calcular, xirr # noqa: E402
|
||
from cartera.datos import Datos, ErrorDatos, escribir, leer # noqa: E402
|
||
|
||
FICTICIA = PROYECTO / "pruebas" / "cartera_ficticia"
|
||
CIFRAS = PROYECTO / "pruebas" / "cifras_esperadas.json"
|
||
FECHA = dt.date(2026, 9, 16)
|
||
TOL_REL, TOL_ABS = 1e-9, 1e-6
|
||
|
||
|
||
def _aplanar(obj, prefijo=""):
|
||
if isinstance(obj, dict):
|
||
for k, v in obj.items():
|
||
yield from _aplanar(v, f"{prefijo}.{k}" if prefijo else k)
|
||
else:
|
||
yield prefijo, obj
|
||
|
||
|
||
def comparar(esperado: dict, obtenido: dict) -> list[str]:
|
||
"""Qué cifras no coinciden (nunca imprime importes: solo el nombre y la diferencia)."""
|
||
a = dict(_aplanar({s: esperado[s] for s in ("motor", "hoja") if esperado.get(s)}))
|
||
b = dict(_aplanar({s: obtenido[s] for s in ("motor", "hoja") if esperado.get(s)}))
|
||
problemas = []
|
||
for k in sorted(set(a) | set(b)):
|
||
if k not in a or k not in b:
|
||
problemas.append(f"falta en {'lo obtenido' if k in a else 'lo esperado'}: {k}")
|
||
continue
|
||
va, vb = a[k], b[k]
|
||
if all(isinstance(v, (int, float)) and not isinstance(v, bool) for v in (va, vb)):
|
||
escala = max(abs(va), abs(vb))
|
||
if abs(va - vb) > max(TOL_ABS, TOL_REL * escala):
|
||
problemas.append(f"distinta: {k} (dif. relativa {abs(va - vb) / escala:.1e})")
|
||
elif va != vb:
|
||
problemas.append(f"distinta: {k}")
|
||
return problemas
|
||
|
||
|
||
def _activo(**kw):
|
||
base = {"id": "F1", "nombre": "Fondo", "tipo": "Fondo", "codigo": "", "unidad": "p",
|
||
"fecha_inicio": "2024-01-01", "visible": True, "orden": 1}
|
||
base.update(kw)
|
||
return base
|
||
|
||
|
||
def _mov(id_, fecha, tipo, cantidad, precio, activo="F1", **kw):
|
||
base = {"id": id_, "fecha": fecha, "activo": activo, "tipo": tipo, "cantidad": cantidad,
|
||
"precio": precio, "importe": cantidad * precio, "orden": 1}
|
||
base.update(kw)
|
||
return base
|
||
|
||
|
||
class TestDatos(unittest.TestCase):
|
||
def test_ida_y_vuelta_y_permisos(self):
|
||
datos = Datos([_activo()], [_mov("m1", "2024-01-01", "Compra", 10, 1.5)],
|
||
[{"fecha": "2024-01-01", "activo": "F1", "precio": 1.5, "fuente": "x"}])
|
||
with tempfile.TemporaryDirectory() as t:
|
||
escribir(Path(t), datos)
|
||
self.assertEqual(leer(Path(t)), datos)
|
||
for f in Path(t).iterdir():
|
||
self.assertEqual(f.stat().st_mode & 0o777, 0o600, f.name)
|
||
self.assertEqual([], [f for f in Path(t).iterdir() if f.name.startswith(".")])
|
||
|
||
def test_rechaza_datos_mal_formados(self):
|
||
casos = [
|
||
("movimientos", '{"id": "m1", "fecha": "01/02/2024", "activo": "F1", "tipo": "Compra"}', "fecha"),
|
||
("movimientos", '{"id": "m1", "fecha": "2024-02-01", "activo": "F1", "tipo": "Regalo"}', "tipo"),
|
||
("movimientos", '{"id": "m1", "fecha": "2024-02-01", "activo": "F1", "tipo": "Compra", "cantidad": "3"}', "número"),
|
||
("activos", '{"id": "X", "tipo": "Fondo"}\n{"id": "X", "tipo": "Fondo"}', "repetida"),
|
||
("precios", '{"activo": "F1", "fecha": "2024-01-01", "precio": 1', "JSON"),
|
||
]
|
||
for fichero, contenido, pista in casos:
|
||
with self.subTest(pista), tempfile.TemporaryDirectory() as t:
|
||
(Path(t) / f"{fichero}.jsonl").write_text(contenido + "\n", encoding="utf-8")
|
||
with self.assertRaises(ErrorDatos) as ctx:
|
||
leer(Path(t))
|
||
self.assertRegex(str(ctx.exception), r"\.jsonl:\d+: ") # indica fichero:línea
|
||
self.assertIn(pista, str(ctx.exception))
|
||
|
||
def test_borrados_no_cuentan(self):
|
||
datos = Datos([_activo()],
|
||
[_mov("m1", "2024-01-01", "Compra", 10, 1.0),
|
||
_mov("m2", "2024-06-01", "Compra", 5, 1.0, borrado=True)],
|
||
[{"fecha": "2024-06-01", "activo": "F1", "precio": 2.0}])
|
||
r = calcular(datos, dt.date(2024, 12, 31))
|
||
self.assertEqual(r["motor"]["activos"]["F1"]["cantidad"], 10)
|
||
|
||
|
||
class TestCalculo(unittest.TestCase):
|
||
def test_xirr_un_ano_al_diez_por_ciento(self):
|
||
tir = xirr([(dt.date(2023, 1, 1), -1000.0), (dt.date(2024, 1, 1), 1100.0)])
|
||
self.assertAlmostEqual(tir, 0.10, places=6)
|
||
|
||
def test_xirr_sin_flujos_de_ambos_signos(self):
|
||
self.assertEqual(xirr([(dt.date(2023, 1, 1), -1000.0), (dt.date(2024, 1, 1), -5.0)]), 0.0)
|
||
|
||
def test_prestamo_devengo_lineal_y_tope_al_vencer(self):
|
||
a = _activo(id="PR", tipo="Prestamo", fecha_inicio="2025-01-01",
|
||
prestamo={"importe_final": 1100.0, "vencimiento": "2026-01-01"})
|
||
datos = Datos([a], [_mov("m1", "2025-01-01", "Compra", 1, 1000.0, activo="PR")], [])
|
||
mitad = calcular(datos, dt.date(2025, 7, 2))["motor"]["activos"]["PR"]["precio"]
|
||
self.assertAlmostEqual(mitad, 1000 + 100 * 182 / 365)
|
||
vencido = calcular(datos, dt.date(2026, 9, 1))["motor"]["activos"]["PR"]["precio"]
|
||
self.assertEqual(vencido, 1100.0)
|
||
|
||
def test_comision_resta_cantidad_pero_no_es_flujo(self):
|
||
datos = Datos([_activo()],
|
||
[_mov("m1", "2024-01-01", "Compra", 100, 10.0),
|
||
_mov("m2", "2024-12-31", "Comision", 1, 12.0)],
|
||
[{"fecha": "2025-01-01", "activo": "F1", "precio": 12.0}])
|
||
r = calcular(datos, dt.date(2025, 1, 1))
|
||
self.assertEqual(r["motor"]["activos"]["F1"]["cantidad"], 99)
|
||
self.assertAlmostEqual(r["hoja"]["activos"]["F1"]["coste"], 1000.0) # coste se reatribuye
|
||
|
||
def test_tras_venderlo_todo_la_compra_nueva_empieza_de_cero(self):
|
||
# Se vendió todo y se volvió a comprar: el coste es solo el de la compra nueva
|
||
datos = Datos([_activo()],
|
||
[_mov("m1", "2024-01-01", "Compra", 10, 10.0),
|
||
_mov("m2", "2024-03-01", "Compra", 10, 30.0),
|
||
_mov("m3", "2024-06-01", "Venta", 20, 25.0),
|
||
_mov("m4", "2025-02-01", "Compra", 5, 40.0)],
|
||
[{"fecha": "2025-03-01", "activo": "F1", "precio": 44.0}])
|
||
h = calcular(datos, dt.date(2025, 3, 1))["hoja"]["activos"]["F1"]
|
||
self.assertAlmostEqual(h["coste"], 200.0)
|
||
self.assertAlmostEqual(h["precio_medio"], 40.0)
|
||
self.assertAlmostEqual(h["ganancia"], 20.0)
|
||
|
||
def test_venta_parcial_se_lleva_su_parte_del_coste(self):
|
||
# Coste medio móvil: la venta saca coste al medio de ese momento
|
||
datos = Datos([_activo()],
|
||
[_mov("m1", "2024-01-01", "Compra", 10, 10.0),
|
||
_mov("m2", "2024-06-01", "Venta", 5, 15.0),
|
||
_mov("m3", "2025-01-01", "Compra", 5, 20.0)],
|
||
[{"fecha": "2025-03-01", "activo": "F1", "precio": 20.0}])
|
||
h = calcular(datos, dt.date(2025, 3, 1))["hoja"]["activos"]["F1"]
|
||
self.assertAlmostEqual(h["coste"], 150.0) # 5 × 10 que quedaban + 100 nuevos
|
||
self.assertAlmostEqual(h["precio_medio"], 15.0)
|
||
|
||
|
||
class TestCifrasDeReferencia(unittest.TestCase):
|
||
"""La cartera ficticia tiene que seguir dando las mismas cifras que siempre."""
|
||
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
cls.datos = leer(FICTICIA)
|
||
cls.esperado = json.loads(CIFRAS.read_text(encoding="utf-8"))
|
||
|
||
def test_reproduce_las_cifras_esperadas(self):
|
||
fecha = dt.date.fromisoformat(self.esperado["fecha_calculo"])
|
||
self.assertEqual(comparar(self.esperado, calcular(self.datos, fecha)), [])
|
||
|
||
def test_pasa_por_disco_sin_cambiar_nada(self):
|
||
with tempfile.TemporaryDirectory() as t:
|
||
escribir(Path(t), self.datos)
|
||
self.assertEqual(leer(Path(t)), self.datos)
|
||
|
||
def test_el_prestamo_tiene_sus_campos_propios(self):
|
||
pr1 = next(a for a in self.datos.activos if a["id"] == "PR1")
|
||
self.assertEqual(pr1["prestamo"], {"importe_final": 5500.0, "vencimiento": "2026-08-01"})
|
||
|
||
def test_ids_unicos(self):
|
||
ids = [m["id"] for m in self.datos.movimientos]
|
||
self.assertEqual(len(ids), len(set(ids)))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|