caudal/programa/cartera/vista.py
2026-10-19 12:42:00 -05:00

409 lines
22 KiB
Python

"""
vista.py — Lo que muestra la aplicación: resumen, activos abiertos y cerrados,
evolución y avisos. Se apoya en calculo.calcular() y añade lo que el programa
viejo no tenía (inversiones cerradas con su resultado, evolución, avisos).
"""
from __future__ import annotations
import datetime as dt
from . import actividades
from .calculo import (COBROS, CON_TRANSACCION, DESINVIERTE, ENTRADAS_CANTIDAD, SALIDAS_CANTIDAD,
Precios, _precio_actual, _f, calcular, flujo, invierte, por_valor, xirr)
from . import deuda as motor_deuda
from . import inmuebles as motor_inmuebles
from .datos import TIPOS_DEUDA, TIPOS_LIQUIDEZ, Datos
TOLERANCIA_CANTIDAD = 1e-9
DIAS_PRECIO_VIEJO = 7
DIAS_SALDO_VIEJO = 45
DIAS_VALOR_VIEJO = 35
TIPOS_CON_PRECIO_AUTOMATICO = {"Fondo", "PlanPensiones", "Cripto", "Metal", "ETF", "Accion"}
def _fecha(s: str) -> dt.date:
return dt.date.fromisoformat(s)
def _es(iso: str) -> str:
return _fecha(iso).strftime("%d/%m/%Y")
def _txn(m: dict, solo_valor: set = frozenset()) -> dict:
return {"fecha": _fecha(m["fecha"]), "activo": m["activo"], "tipo": m["tipo"],
"cantidad": _f(m.get("cantidad")), "precio": _f(m.get("precio")),
"importe": _f(m.get("importe")), "por_valor": m["activo"] in solo_valor}
def cantidad_en(movs: list[dict], activo: str, fecha: dt.date) -> float:
"""Cantidad que se tenía de un activo al final de un día."""
total = 0.0
for m in movs:
if m["activo"] != activo or _fecha(m["fecha"]) > fecha:
continue
if m["tipo"] in ENTRADAS_CANTIDAD:
total += _f(m.get("cantidad"))
elif m["tipo"] in SALIDAS_CANTIDAD:
total -= _f(m.get("cantidad"))
return total
def _flujos(movs: list[dict], solo_valor: set = frozenset()) -> tuple[list[tuple], float, float]:
flujos, invertido, recuperado = [], 0.0, 0.0
for m in movs:
if m["tipo"] not in CON_TRANSACCION:
continue
t = _txn(m, solo_valor)
importe = flujo(t)
if importe <= 0:
continue
if invierte(t):
flujos.append((t["fecha"], -importe))
invertido += importe
elif t["tipo"] in DESINVIERTE:
flujos.append((t["fecha"], importe))
invertido -= importe
elif t["tipo"] in COBROS:
flujos.append((t["fecha"], importe))
recuperado += importe
return flujos, invertido, recuperado
def _realizado(movs: list[dict], solo_valor: bool = False) -> tuple[float, float, float]:
"""Ganancia realizada por ventas y reembolsos, con coste medio.
Un traspaso NO realiza ganancia: la plusvalía viaja al fondo de destino
(allí el coste es el fiscal de origen). Devuelve (ganancia realizada,
importe traspasado fuera, coste que se llevó el traspaso)."""
if solo_valor:
# Sin participaciones: lo realizado es el dinero que ha salido menos el que entró
_, invertido, recuperado = _flujos(movs, {m["activo"] for m in movs})
return recuperado - invertido, 0.0, 0.0
cantidad = coste = realizada = traspasado = coste_traspasado = 0.0
for m in movs:
c, t = _f(m.get("cantidad")), m["tipo"]
importe = _f(m.get("importe")) or c * _f(m.get("precio"))
if t == "Compra":
cantidad += c
coste += importe
elif t == "Gasto":
coste += _f(m.get("importe"))
elif t in DESINVIERTE:
coste -= _f(m.get("importe")) # devolución de un gasto: menos coste
elif t in ("Mineria", "Recompensa"):
cantidad += c
elif t == "Traspaso_Entrada":
cantidad += c
coste += m["coste_traspaso"] if (m.get("coste_traspaso") or 0) > 0 else importe
elif t == "Comision":
cantidad -= c
elif t == "Dividendo":
realizada += _f(m.get("importe"))
elif t in ("Venta", "Reembolso", "Traspaso_Salida") and cantidad > 0:
parte = coste * min(c / cantidad, 1.0)
coste -= parte
cantidad -= c
if t == "Traspaso_Salida":
traspasado += importe
coste_traspasado += parte
else:
realizada += importe - parte
return realizada, traspasado, coste_traspasado
def _evolucion(activos: list[dict], movs: list[dict], precios: Precios, hoy: dt.date,
solo_valor: set = frozenset()) -> list[dict]:
"""Valor de la cartera y dinero aportado neto a fin de cada mes."""
if not movs:
return []
inicio = min(_fecha(m["fecha"]) for m in movs)
fechas = []
y, mes = inicio.year, inicio.month
while True:
siguiente = dt.date(y + (mes == 12), mes % 12 + 1, 1)
fin_mes = siguiente - dt.timedelta(days=1)
if fin_mes >= hoy:
break
fechas.append(fin_mes)
y, mes = siguiente.year, siguiente.month
fechas.append(hoy)
movs_de: dict[str, list[dict]] = {}
for m in movs:
movs_de.setdefault(m["activo"], []).append(m)
puntos = []
for f in fechas:
valor = 0.0
for a in activos:
ms = [m for m in movs_de.get(a["id"], []) if _fecha(m["fecha"]) <= f]
if not ms:
continue
if a.get("estado") == "cerrada" and a.get("fecha_cierre") and f >= _fecha(a["fecha_cierre"]):
# Cerrada ya no vale nada (como en el resumen): si no, lo llevado por valor
# seguía sumando su último valor apuntado y el dinero contaba dos veces
continue
if a["id"] in solo_valor:
reg = precios.ultimo(a["id"], f)
valor += (reg or {}).get("valor_total") or (reg or {}).get("precio") or 0.0
continue
q = cantidad_en(ms, a["id"], f)
if abs(q) <= TOLERANCIA_CANTIDAD:
continue
p = _precio_actual(a, ms, precios, f)
if not p:
# Sin precio guardado tan atrás: el de la última operación con precio
p = next((m["precio"] for m in reversed(ms) if m.get("precio")), None)
if p:
valor += q * p
_, inv, rec = _flujos([m for m in movs if _fecha(m["fecha"]) <= f], solo_valor)
puntos.append({"fecha": f.isoformat(), "valor": valor, "aportado": inv - rec})
return puntos
def evolucion_activo(datos: Datos, aid: str, hoy: dt.date) -> list[dict]:
"""Lo mismo que la evolución de la cartera, pero de una sola inversión: su
valor y lo que se ha puesto en ella a fin de cada mes (para su ficha)."""
a = next((x for x in datos.vivos("activos") if x["id"] == aid), None)
if a is None:
return []
movs = sorted((m for m in datos.vivos("movimientos") if m["activo"] == aid and not m.get("impagado")),
key=lambda m: (m["fecha"], m.get("orden") or 0, m["id"]))
return _evolucion([a], movs, Precios(datos.vivos("precios")), hoy, {aid} if por_valor(a) else frozenset())
def _fila_deuda(a: dict, movs: list[dict], precios: Precios, hoy: dt.date) -> dict:
"""Préstamo recibido: lo que queda por pagar y sus cuotas (ver deuda.py)."""
e = motor_deuda.estado(a, movs, precios.de(a["id"]), hoy)
ultimo_saldo = next((x for x in reversed(e["eventos"]) if x["tipo"] == "saldo" and _fecha(x["fecha"]) <= hoy), None)
return {"id": a["id"], "nombre": a.get("nombre", ""), "tipo": a["tipo"], "estado": a.get("estado", "abierta"),
"tin": (a.get("deuda") or {}).get("tin"),
**{k: e[k] for k in ("pendiente", "cuota", "proxima", "cuotas_restantes", "fin", "intereses_pagados",
"intereses_año", "intereses_pendientes", "amortizado_anticipado", "cancelado")},
"fecha_saldo": ultimo_saldo["fecha"] if ultimo_saldo else None,
"diferencia_saldo": ultimo_saldo["diferencia"] if ultimo_saldo else None}
def resumen(datos: Datos, hoy: dt.date) -> dict:
base = calcular(datos, hoy)
motor, hoja = base["motor"], base["hoja"]
precios = Precios(datos.vivos("precios"))
todos = sorted(datos.vivos("activos"), key=lambda a: (a.get("orden") or 0, a["id"]))
todos_movs = sorted(datos.vivos("movimientos"), key=lambda m: (m["fecha"], m.get("orden") or 0, m["id"]))
# Un cobro marcado como impagado no es dinero recibido: queda fuera de las cifras y se avisa aparte
impagados = [m for m in todos_movs if m.get("impagado")]
movs = [m for m in todos_movs if not m.get("impagado")]
movs_de: dict[str, list[dict]] = {}
for m in movs:
movs_de.setdefault(m["activo"], []).append(m)
solo_valor = {a["id"] for a in todos if por_valor(a)}
abiertos, cerrados, liquidez, deudas, avisos, bienes = [], [], [], [], [], []
for a in todos:
aid = a["id"]
if motor_inmuebles.es_propio(a):
# Tu casa: cuenta en el patrimonio, no en la rentabilidad de las inversiones
if a.get("estado", "abierta") == "cerrada":
continue
ms = movs_de.get(aid, [])
reg = precios.ultimo(aid, hoy)
compra = motor_inmuebles.alquiler(ms, hoy)["compra"]
valor = (reg or {}).get("valor_total") or (reg or {}).get("precio") or compra
f = a.get("inmueble") or {}
bienes.append({"id": aid, "nombre": a.get("nombre", ""), "tipo": a["tipo"], "clase": f.get("clase"),
"referencia": f.get("referencia", ""), "valor": valor, "compra": compra,
"revalorizacion": valor - compra, "fecha_valor": reg["fecha"] if reg else None,
"fuente_valor": reg.get("fuente") if reg else None, "protegida": bool(f.get("protegida"))})
continue
if a["tipo"] in TIPOS_DEUDA:
fila = _fila_deuda(a, movs_de.get(aid, []), precios, hoy)
if fila["estado"] == "abierta":
deudas.append(fila)
if fila["cancelado"]:
avisos.append({"tipo": "cerrar", "activo": aid,
"texto": f"{fila['nombre']}: ya está pagado. ¿Darlo por terminado?"})
continue
if a["tipo"] in TIPOS_LIQUIDEZ:
ms = movs_de.get(aid, [])
reg = precios.ultimo(aid, hoy)
saldo = (reg or {}).get("precio") or 0.0
fila = {"id": aid, "nombre": a.get("nombre", ""), "tipo": a["tipo"], "estado": a.get("estado", "abierta"),
"saldo": saldo, "fecha_saldo": reg["fecha"] if reg else None,
"intereses": sum(_f(m.get("importe")) for m in ms if m["tipo"] == "Dividendo"),
"intereses_año": sum(_f(m.get("importe")) for m in ms
if m["tipo"] == "Dividendo" and m["fecha"][:4] == str(hoy.year))}
liquidez.append(fila)
if not reg:
avisos.append({"tipo": "saldo", "activo": aid,
"texto": f"{fila['nombre']}: aún no has puesto el saldo."})
elif (hoy - _fecha(reg["fecha"])).days > DIAS_SALDO_VIEJO:
avisos.append({"tipo": "saldo", "activo": aid,
"texto": f"{fila['nombre']}: el saldo es del {_es(reg['fecha'])}. ¿Lo actualizas?"})
continue
ms = movs_de.get(aid, [])
flujos, invertido, recuperado = _flujos(ms, solo_valor)
primera = ms[0]["fecha"] if ms else a.get("fecha_inicio")
fila = {
"id": aid, "nombre": a.get("nombre", ""), "tipo": a["tipo"],
"codigo": a.get("codigo", ""), "unidad": a.get("unidad", ""),
"estado": a.get("estado", "abierta"), "visible": a.get("visible", True),
"fecha_inicio": primera, "fecha_cierre": a.get("fecha_cierre"),
"invertido": invertido, "recuperado": recuperado, "movimientos": len(ms),
"dividendos": sum(_f(m.get("importe")) for m in ms if m["tipo"] == "Dividendo"),
"recompensas": sum(_f(m.get("cantidad")) for m in ms if m["tipo"] == "Recompensa"),
"recompensas_valor": sum(_f(m.get("valor_recibido")) for m in ms if m["tipo"] == "Recompensa"),
"dividendos_año": sum(_f(m.get("importe")) for m in ms
if m["tipo"] == "Dividendo" and m["fecha"][:4] == str(hoy.year)),
}
if fila["estado"] == "cerrada":
cierre = _fecha(a["fecha_cierre"])
realizada, traspasado, coste_traspasado = _realizado(ms, aid in solo_valor)
base = invertido - coste_traspasado
fila["ganancia"] = realizada
fila["traspasado"] = traspasado
fila["rentabilidad"] = realizada / base if base > 1e-9 else None
fila["tir"] = xirr(flujos) if flujos else None
fila["dias"] = (cierre - _fecha(primera)).days if primera else None
cerrados.append(fila)
continue
m, h = motor["activos"].get(aid), hoja["activos"].get(aid)
if m is None: # oculto
continue
reg = precios.ultimo(aid, hoy)
fila.update({
"cantidad": m["cantidad"], "precio": m["precio"],
"fecha_precio": reg["fecha"] if reg else None,
"fuente_precio": reg.get("fuente") if reg else None,
"valor": m["valor"], "coste": h["coste"],
"ganancia": h["ganancia"], "rentabilidad": h["ganancia_pct"] if h["coste"] else None,
"ganancia_total": m["ganancia"],
"dif_fiscal": h["dif_fiscal"],
"tir": m["irr"] if m["valor"] > 0 else None,
"twr": m["twr"] if m["valor"] > 0 else None,
"ytd": h["rent_ytd"], "mes": h["rent_mes"], "peso": h["peso"],
})
if a["tipo"] == "RentaFija":
rf = a.get("renta_fija") or {}
fila.update({"clase": rf.get("clase"), "vencimiento": rf.get("vencimiento"), "interes": rf.get("interes"),
"nominal": rf.get("nominal"), "emisor": rf.get("emisor", ""), "pagos": rf.get("pagos", 0)})
if a["tipo"] == "Inmueble":
f = a.get("inmueble") or {}
fila.update(motor_inmuebles.alquiler(ms, hoy))
fila.update({"clase": f.get("clase"), "uso": f.get("uso"), "referencia": f.get("referencia", ""),
"protegida": bool(f.get("protegida")), "valor_auto": bool(f.get("auto"))})
abiertos.append(fila)
# Avisos
if m["cantidad"] is not None and abs(m["cantidad"]) <= 1e-6 and ms:
avisos.append({"tipo": "cerrar", "activo": aid,
"texto": f"{fila['nombre']}: ya no tienes posición. ¿Dar la inversión por cerrada?"})
if a["tipo"] == "Prestamo" and (a.get("prestamo") or {}).get("vencimiento") \
and _fecha(a["prestamo"]["vencimiento"]) <= hoy and abs(m["cantidad"]) > 1e-6:
avisos.append({"tipo": "cerrar", "activo": aid,
"texto": f"{fila['nombre']}: venció el {_es(a['prestamo']['vencimiento'])}. ¿Registrar el cobro?"})
if a["tipo"] == "RentaFija":
rf = a.get("renta_fija") or {}
if rf.get("vencimiento") and _fecha(rf["vencimiento"]) <= hoy and m["valor"] > 0.005:
avisos.append({"tipo": "vencido", "activo": aid,
"texto": f"{fila['nombre']}: venció el {_es(rf['vencimiento'])}. Apunta lo que te devolvieron."})
elif a["tipo"] == "Inmueble":
# Un inmueble no cambia de valor cada mes: con el índice del INE no hace falta nada, y a mano basta una vez al año
if not (a.get("inmueble") or {}).get("auto") and m["valor"] > 0 and (not reg or (hoy - _fecha(reg["fecha"])).days > 365):
avisos.append({"tipo": "precio_manual", "activo": aid,
"texto": f"{fila['nombre']}: su valor es de hace más de un año. ¿Le pones uno nuevo, o lo dejas estimar con el índice del INE?"})
elif por_valor(a) and m["valor"] > 0 and (not reg or (hoy - _fecha(reg["fecha"])).days > DIAS_VALOR_VIEJO):
avisos.append({"tipo": "precio_manual", "activo": aid,
"texto": f"{fila['nombre']}: actualiza el valor de la posición"
+ (f" (el último es del {_es(reg['fecha'])})." if reg else ".")})
elif a["tipo"] in TIPOS_CON_PRECIO_AUTOMATICO and m["valor"] > 0 and \
(not reg or (hoy - _fecha(reg["fecha"])).days > DIAS_PRECIO_VIEJO):
avisos.append({"tipo": "precio", "activo": aid,
"texto": f"{fila['nombre']}: el precio es de hace más de {DIAS_PRECIO_VIEJO} días."})
if a["tipo"] in ("Tierra", "CuentaTrading") and m["valor"] > 0 and reg and \
(hoy - _fecha(reg["fecha"])).days > 90:
avisos.append({"tipo": "precio_manual", "activo": aid,
"texto": f"{fila['nombre']}: la valoración manual tiene más de 3 meses."})
if impagados:
nombres = {a["id"]: a.get("nombre", "") for a in todos}
debido = sum(_f(m.get("importe")) for m in impagados)
avisos.append({"tipo": "impagado", "activo": impagados[0]["activo"],
"texto": f"Te deben {debido:,.2f} € de {len(impagados)} cobro(s) sin pagar: ".replace(",", " ").replace(".", ",").replace(" ", ".")
+ ", ".join(sorted({nombres.get(m['activo'], '') for m in impagados})) + "."})
for m in movs:
if m.get("pendiente"):
avisos.append({"tipo": "pendiente", "movimiento": m["id"], "activo": m["activo"],
"texto": f"Operación automática del {_es(m['fecha'])} sin completar: {m.get('notas') or m['tipo']}."})
elif m.get("estimado"):
avisos.append({"tipo": "estimado", "movimiento": m["id"], "activo": m["activo"],
"texto": f"Aportación del {_es(m['fecha'])} con participaciones estimadas: confírmalas con el extracto."
if m["tipo"] == "Compra" else
f"Cobro automático del {_es(m['fecha'])} ({m.get('notas', '').split(' (')[0].lower() or 'estimado'}): confírmalo con el extracto."})
valor_total = sum(f["valor"] for f in abiertos)
coste_abiertos = sum(f["coste"] for f in abiertos)
realizada = sum(f["ganancia"] for f in cerrados)
por_tipo: dict[str, float] = {}
for f in abiertos:
if f["valor"] > 0:
por_tipo[f["tipo"]] = por_tipo.get(f["tipo"], 0.0) + f["valor"]
visibles = [a for a in todos if a.get("visible", True) and a["tipo"] not in TIPOS_DEUDA and not motor_inmuebles.es_propio(a)]
deuda_total = sum(f["pendiente"] for f in deudas)
# La hipoteca de cada inmueble: lo que queda por pagar y sus intereses, que son un gasto del alquiler
deuda_de = {f["id"]: f for f in deudas}
activo_de = {a["id"]: a for a in todos}
desde = (hoy - dt.timedelta(days=365)).isoformat()
for f in [x for x in abiertos if x["tipo"] == "Inmueble"] + bienes:
did = (activo_de[f["id"]].get("inmueble") or {}).get("hipoteca")
d = deuda_de.get(did)
if not d:
continue
cuadro = motor_deuda.cuadro(activo_de[did], movs_de.get(did, []), precios.de(did))
intereses = round(sum(x["intereses"] for x in cuadro["filas"] if desde < x["fecha"] <= hoy.isoformat()), 2)
f.update({"hipoteca": did, "hipoteca_nombre": d["nombre"], "hipoteca_pendiente": d["pendiente"],
"intereses_12m": intereses, "neto": round(f["valor"] - d["pendiente"], 2)})
if "alquiler_12m" in f:
f["gastos_12m"] = round(f["gastos_12m"] + intereses, 2)
f["rentabilidad_neta"] = round((f["alquiler_12m"] - f["gastos_12m"]) / f["compra"], 4) if f["compra"] > 0 else None
patrimonio = valor_total + sum(f["saldo"] for f in liquidez) + sum(f["valor"] for f in bienes)
suyos = [f for f in abiertos if f["tipo"] == "Inmueble"]
compra = sum(f["compra"] for f in suyos)
cobrado, gastado = sum(f["alquiler_12m"] for f in suyos), sum(f["gastos_12m"] for f in suyos)
inmuebles = {"numero": len(suyos), "valor": sum(f["valor"] for f in suyos), "compra": compra,
"alquiler_12m": cobrado, "gastos_12m": gastado,
"rentabilidad_bruta": cobrado / compra if compra > 0 else None,
"rentabilidad_neta": (cobrado - gastado) / compra if compra > 0 else None} if suyos else None
return {
"hoy": hoy.isoformat(),
"global": {
"valor": valor_total,
"coste": coste_abiertos,
"ganancia_latente": valor_total - coste_abiertos,
"rentabilidad_latente": (valor_total - coste_abiertos) / coste_abiertos if coste_abiertos else None,
"ganancia_realizada": realizada,
"dividendos_año": sum(f["dividendos_año"] for f in abiertos + cerrados)
+ sum(f["intereses_año"] for f in liquidez),
"dividendos": sum(f["dividendos"] for f in abiertos + cerrados)
+ sum(f["intereses"] for f in liquidez),
"liquidez": sum(f["saldo"] for f in liquidez),
"bienes": sum(f["valor"] for f in bienes),
"patrimonio": patrimonio,
"deuda": deuda_total,
"patrimonio_neto": patrimonio - deuda_total,
"intereses_deuda_año": sum(f["intereses_año"] for f in deudas),
"tir": motor["global"]["irr"],
"ytd": motor["global"]["ytd"],
"mes": motor["global"]["mes"],
"roi": motor["global"]["roi"],
},
"abiertos": abiertos,
"cerrados": cerrados,
"liquidez": sorted(liquidez, key=lambda f: -f["saldo"]),
"deudas": sorted(deudas, key=lambda f: -f["pendiente"]),
"bienes": bienes,
"inmuebles": inmuebles,
"por_tipo": por_tipo,
"evolucion": _evolucion(visibles, movs, precios, hoy, solo_valor),
"avisos": avisos,
"actividades": actividades.resumen(datos, hoy),
}