323 lines
16 KiB
Python
323 lines
16 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 .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["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 _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"]))
|
|
movs = sorted(datos.vivos("movimientos"), key=lambda m: (m["fecha"], m.get("orden") or 0, m["id"]))
|
|
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 = [], [], [], [], []
|
|
for a in todos:
|
|
aid = a["id"]
|
|
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"],
|
|
})
|
|
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 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."})
|
|
|
|
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."})
|
|
|
|
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]
|
|
deuda_total = sum(f["pendiente"] for f in deudas)
|
|
patrimonio = valor_total + sum(f["saldo"] for f in liquidez)
|
|
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),
|
|
"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"]),
|
|
"por_tipo": por_tipo,
|
|
"evolucion": _evolucion(visibles, movs, precios, hoy, solo_valor),
|
|
"avisos": avisos,
|
|
"actividades": actividades.resumen(datos, hoy),
|
|
}
|