caudal/programa/cartera/calculo.py

"""
calculo.py — Motor de cálculo de la cartera.

Estas cifras vienen de una hoja de cálculo que estuvo años en uso, y se
comprobaron una a una contra ella antes de sustituirla: `pruebas/cartera_ficticia/`
con `pruebas/cifras_esperadas.json` son esa vara de medir y siguen ahí. Alguna
rareza heredada se conserva a propósito, señalada donde aparece; se corregirán
de una en una y de forma explícita, nunca sin querer.

Tipos añadidos después (no alteran las cifras de referencia):
  Gasto   → euros sin cantidad: flujo de dinero invertido (TIR, ROI, ganancia
            y coste fiscal); no participa en el TWR.
  Devolucion → lo contrario de un Gasto: euros que vuelven porque se anula
            parte de un gasto ya apuntado (una devolución del hosting). NO es
            un cobro: en vez de sumar a lo recuperado, RESTA de lo invertido,
            así que baja el coste del activo. La TIR y la ganancia total salen
            igual que si fuera un cobro —el euro entra el mismo día—, pero el
            coste medio baja y el ROI sube, que es lo que lo distingue de un
            Dividendo. Tampoco entra en el TWR.
  Mineria → cantidad recibida sin pagarla: suma cantidad (baja el coste medio),
            no es flujo de dinero y no marca precio en el TWR.
  Recompensa → como Mineria (bonificaciones del bróker, regalos): cantidad
            gratis, todo su valor es ganancia.
  Las cuentas (liquidez) no entran aquí: no tienen coste ni rentabilidad, solo
  saldo. Se ven aparte, en vista.resumen().
  Dividendo → dinero cobrado sin vender: cuenta como dinero recuperado (TIR, ROI,
            ganancia total); no cambia cantidad ni coste; no entra en el TWR.

calcular(datos, fecha) devuelve dos bloques —los nombres vienen de la hoja de
cálculo original y se conservan porque son los de la vara de medir:
  "motor": cómo ha ido la inversión (cantidad, precio, valor, ganancia por
           flujos de dinero, TWR, CAGR, TIR, ROI, y los mismos totales).
  "hoja":  cómo está la posición hoy (coste fiscal, precio medio, ganancia
           sobre ese coste, diferencia fiscal, peso, año y mes por activo).
"""

from __future__ import annotations

import bisect
import datetime as dt

from .datos import NO_INVERSION, Datos

ENTRADAS = {"Compra", "Traspaso_Entrada"}
SALIDAS = {"Venta", "Traspaso_Salida", "Reembolso"}          # flujos de dinero
SALIDAS_CANTIDAD = SALIDAS | {"Comision"}                    # restan participaciones
SIN_DESEMBOLSO = {"Mineria", "Recompensa"}                  # entra cantidad gratis
ENTRADAS_CANTIDAD = ENTRADAS | SIN_DESEMBOLSO                # suman participaciones
COBROS = SALIDAS | {"Dividendo"}                             # dinero que vuelve
DESINVIERTE = {"Devolucion"}                                 # dinero que vuelve restando coste
CON_TRANSACCION = ENTRADAS | SALIDAS | SIN_DESEMBOLSO | {"Gasto", "Dividendo"} | DESINVIERTE


def por_valor(a: dict) -> bool:
    """Activo del que solo se sabe el valor total (algunos planes de aseguradoras): se
    apuntan las aportaciones en € y, de vez en cuando, el valor de la posición."""
    return a.get("valoracion") == "total"


def flujo(t: dict) -> float:
    """Importe en € que mueve una transacción (positivo; el signo lo da el tipo)."""
    if t.get("por_valor"):
        return t["importe"]
    if t["tipo"] in ("Gasto", "Dividendo") or t["tipo"] in DESINVIERTE:
        return t["importe"]
    if t["tipo"] in SIN_DESEMBOLSO:
        return 0.0
    return t["cantidad"] * t["precio"]


def invierte(t: dict) -> bool:
    return t["tipo"] in ENTRADAS or t["tipo"] == "Gasto"


def _f(v) -> float:
    """Número o 0."""
    return float(v) if v else 0.0


def _fecha(s: str) -> dt.date:
    return dt.date.fromisoformat(s)


# ── Precios ──────────────────────────────────────────────────────────────────

class Precios:
    """Serie de precios por activo; consulta del último precio ≤ fecha."""

    def __init__(self, registros: list[dict]):
        self._series: dict[str, tuple[list[dt.date], list[dict]]] = {}
        por_activo: dict[str, list[dict]] = {}
        for r in registros:
            por_activo.setdefault(r["activo"], []).append(r)
        for aid, rs in por_activo.items():
            rs.sort(key=lambda r: r["fecha"])
            self._series[aid] = ([_fecha(r["fecha"]) for r in rs], rs)

    def ultimo(self, activo: str, fecha: dt.date) -> dict | None:
        serie = self._series.get(activo)
        if not serie:
            return None
        i = bisect.bisect_right(serie[0], fecha)
        return serie[1][i - 1] if i else None

    def de(self, activo: str) -> list[dict]:
        """Todos los registros de un activo, por fecha."""
        serie = self._series.get(activo)
        return list(serie[1]) if serie else []


# ── Matemática financiera ───────────────────────────────────────────────────

def coste_restante(ms: list[dict]) -> tuple[float, float, float]:
    """(cantidad, coste fiscal, coste a valor de mercado) de lo que sigue en
    cartera, con los movimientos de UN activo en orden.

    Coste medio ponderado móvil: cada venta se lleva su parte proporcional del
    coste, así que al venderlo todo el coste vuelve a cero y una compra
    posterior empieza de nuevo. (Antes se promediaban todas las entradas de la
    historia, también las ya vendidas.) La comisión resta cantidad sin restar
    coste: el coste se reatribuye a lo que queda."""
    cantidad = fiscal = mercado = 0.0
    for m in ms:
        t, c, importe = m["tipo"], _f(m.get("cantidad")), _f(m.get("importe"))
        if t == "Compra":
            cantidad += c
            fiscal += importe
            mercado += importe
        elif t == "Gasto":
            fiscal += importe
            mercado += importe
        elif t in DESINVIERTE:
            fiscal -= importe
            mercado -= importe
        elif t in SIN_DESEMBOLSO:
            cantidad += c
        elif t == "Traspaso_Entrada":
            cantidad += c
            fiscal += m["coste_traspaso"] if (m.get("coste_traspaso") or 0) > 0 else importe
            mercado += importe
        elif t == "Comision":
            cantidad -= c
        elif t in SALIDAS:
            if cantidad > 0:
                parte = min(c / cantidad, 1.0)
                fiscal -= fiscal * parte
                mercado -= mercado * parte
            cantidad -= c
    return cantidad, fiscal, mercado


def _xnpv(rate: float, flujos: list[tuple]) -> float:
    if rate <= -1:
        return float("inf")
    t0 = flujos[0][0]
    return sum(cf / (1 + rate) ** ((d - t0).days / 365.0) for d, cf in flujos)


def xirr(flujos: list[tuple]) -> float:
    """TIR anualizada de flujos (fecha, importe). Bisección en [-0.999, 100]."""
    if len(flujos) < 2:
        return 0.0
    if not (any(cf > 0 for _, cf in flujos) and any(cf < 0 for _, cf in flujos)):
        return 0.0
    flujos = sorted(flujos, key=lambda x: x[0])
    lo, hi = -0.999, 100.0
    f_lo, f_hi = _xnpv(lo, flujos), _xnpv(hi, flujos)
    if f_lo * f_hi > 0:
        return 0.0
    for _ in range(200):
        mid = (lo + hi) / 2
        f_mid = _xnpv(mid, flujos)
        if abs(f_mid) < 1e-7 or (hi - lo) < 1e-9:
            return mid
        if f_lo * f_mid < 0:
            hi, f_hi = mid, f_mid
        else:
            lo, f_lo = mid, f_mid
    return (lo + hi) / 2


def _pares_traspaso(txns: list[dict]) -> dict:
    """Empareja cada Traspaso_Entrada con el Traspaso_Salida del mismo día y
    valor (tolerancia 1 % o 1 €). Clave (destino, fecha, valor) → origen."""
    salidas: dict = {}
    for t in txns:
        if t["tipo"] == "Traspaso_Salida":
            salidas.setdefault(t["fecha"], []).append(t)
    pares = {}
    for t in txns:
        if t["tipo"] != "Traspaso_Entrada":
            continue
        val = t["cantidad"] * t["precio"]
        candidatos = salidas.get(t["fecha"], [])
        if not candidatos:
            continue
        mejor = min(candidatos, key=lambda s: abs(s["cantidad"] * s["precio"] - val))
        if abs(mejor["cantidad"] * mejor["precio"] - val) <= max(val * 0.01, 1.0):
            pares[(t["activo"], t["fecha"], round(val, 2))] = mejor["activo"]
    return pares


def _cadena(activo: str, por_activo: dict, pares: dict, visitados: frozenset = frozenset()) -> list[dict]:
    """Movimientos del activo más la historia de los fondos de los que viene
    por traspaso (hasta el día del traspaso)."""
    if activo in visitados:
        return []
    visitados = visitados | {activo}
    propios = por_activo.get(activo, [])
    cadena = list(propios)
    for t in propios:
        if t["tipo"] != "Traspaso_Entrada":
            continue
        origen = pares.get((activo, t["fecha"], round(t["cantidad"] * t["precio"], 2)))
        if not origen or origen in visitados:
            continue
        for s in _cadena(origen, por_activo, pares, visitados):
            if s["fecha"] < t["fecha"] or (s["fecha"] == t["fecha"]
                                           and s["tipo"] == "Traspaso_Salida"
                                           and s["activo"] == origen):
                cadena.append(s)
    return sorted(cadena, key=lambda x: (x["fecha"], 0 if x["tipo"] == "Traspaso_Salida" else 1))


def _twr_cagr(cadena: list[dict], precio_actual: float, hoy: dt.date) -> tuple[float, float]:
    if not cadena:
        return 0.0, 0.0
    tenencia, producto = 0.0, 1.0
    nav_prev, tenencia_prev, fondo_prev = None, 0.0, None
    cadena = [t for t in cadena if t["tipo"] not in ("Gasto", "Dividendo") and t["tipo"] not in DESINVIERTE]
    if not cadena:
        return 0.0, 0.0
    for t in cadena:
        fondo = t["activo"]
        if t["tipo"] in SIN_DESEMBOLSO:
            # Entra cantidad sin precio: no hay revalorización que medir
            if fondo_prev is not None and fondo_prev != fondo:
                tenencia = 0.0
                nav_prev = None
            tenencia += t["cantidad"]
            tenencia_prev, fondo_prev = tenencia, fondo
            continue
        nav = t["precio"] or precio_actual or 1.0
        if fondo_prev == fondo and nav_prev and tenencia_prev > 0 and nav_prev > 0:
            producto *= (tenencia_prev * nav) / (tenencia_prev * nav_prev)
        if fondo_prev is not None and fondo_prev != fondo:
            tenencia = 0.0
        if t["tipo"] in ENTRADAS:
            tenencia += t["cantidad"]
        elif t["tipo"] in SALIDAS:
            tenencia -= t["cantidad"]
        nav_prev, tenencia_prev, fondo_prev = nav, tenencia, fondo
    if tenencia_prev > 0 and nav_prev and nav_prev > 0 and precio_actual > 0:
        producto *= (tenencia_prev * precio_actual) / (tenencia_prev * nav_prev)
    twr = producto - 1.0
    dias = (hoy - cadena[0]["fecha"]).days
    cagr = (1 + twr) ** (365.0 / dias) - 1 if dias > 0 and (1 + twr) > 0 else 0.0
    return twr, cagr


def _tir_roi(txns: list[dict], valor: float, hoy: dt.date) -> tuple[float, float]:
    flujos, invertido, recuperado = [], 0.0, 0.0
    for t in txns:
        importe = flujo(t)
        if importe <= 0:
            continue
        if invierte(t):
            flujos.append((t["fecha"], -importe))
            invertido += importe
        elif t["tipo"] in DESINVIERTE:
            # Mismo flujo que un cobro para la TIR, pero resta de lo puesto:
            # por eso baja el coste y sube el ROI (el denominador es menor)
            flujos.append((t["fecha"], importe))
            invertido -= importe
        elif t["tipo"] in COBROS:
            flujos.append((t["fecha"], importe))
            recuperado += importe
    if valor > 0:
        flujos.append((hoy, valor))
    tir = xirr(flujos)
    roi = (valor + recuperado - invertido) / invertido if invertido - recuperado > 0 else 0.0
    return tir, roi


# ── Precio actual por tipo de activo ─────────────────────────────────────────

def _precio_actual(activo: dict, movs: list[dict], precios: Precios, hoy: dt.date) -> float | None:
    aid, tipo = activo["id"], activo["tipo"]
    reg = precios.ultimo(aid, hoy)

    if tipo == "CuentaTrading":
        # Participaciones = movimientos con cantidad (las comisiones no cuentan)
        partes = 0.0
        for m in movs:
            if not m.get("cantidad"):
                continue
            if m["tipo"] in ENTRADAS:
                partes += m["cantidad"]
            elif m["tipo"] in SALIDAS:
                partes -= m["cantidad"]
        if partes <= 0 or reg is None:
            return None
        if reg.get("valor_total"):
            return reg["valor_total"] / partes if reg["valor_total"] > 0 else None
        v = reg.get("precio")
        if not v or v <= 0:
            return None
        return v / partes if v > 10 else v      # > 10 = equity (heredado del viejo)

    if tipo == "Prestamo":
        devengado = _devengo_prestamo(activo, movs, hoy)
        if devengado is not None:
            return devengado
        # Sin datos para calcularlo: el último precio guardado (como la hoja vieja)

    v = reg.get("precio") if reg else None
    if v and v > 0:
        return v
    if tipo == "Tierra":
        # Sin precio: coste medio de sus movimientos
        cant = sum(_f(m.get("cantidad")) for m in movs)
        coste = sum(_f(m.get("cantidad")) * _f(m.get("precio")) for m in movs)
        return coste / cant if cant > 0 else None
    return None


def _devengo_prestamo(activo: dict, movs: list[dict], hoy: dt.date) -> float | None:
    """Valor del préstamo interpolando linealmente entre principal e importe final."""
    p = activo.get("prestamo") or {}
    inicio = activo.get("fecha_inicio")
    if p.get("importe_final") is None or not p.get("vencimiento") or not inicio:
        return None
    principal = sum(_f(m.get("importe")) or _f(m.get("cantidad"))
                    for m in movs if m["tipo"] == "Compra")
    if principal <= 0:
        return None
    inicio, fin = _fecha(inicio), _fecha(p["vencimiento"])
    total = (fin - inicio).days
    if total <= 0:
        return None
    transcurrido = max(0, min((hoy - inicio).days, total))
    return principal + (p["importe_final"] - principal) * transcurrido / total


def _refs_por_valor(aid: str, precios: Precios, ene1: dt.date, mes1: dt.date) -> tuple:
    def valor(fecha):
        reg = precios.ultimo(aid, fecha)
        return (reg or {}).get("valor_total") or (reg or {}).get("precio") if reg else None
    return valor(ene1), valor(mes1)


def _precio_referencia(activo: dict, precios: Precios, fecha: dt.date, partes: float) -> float | None:
    """Precio de referencia (01/ENE, 01/MES): último ≤ fecha."""
    reg = precios.ultimo(activo["id"], fecha)
    if reg is None:
        return None
    if reg.get("valor_total"):
        return reg["valor_total"] / partes if partes > 0 else None
    return reg.get("precio")


# ── Cálculo principal ────────────────────────────────────────────────────────

def calcular(datos: Datos, hoy: dt.date) -> dict:
    activos = sorted((a for a in datos.vivos("activos")
                      if a.get("visible", True) and a["tipo"] not in NO_INVERSION),
                     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"]))
    # Nota: los movimientos con fecha futura cuentan (rareza heredada, a propósito).
    precios = Precios([p for p in datos.vivos("precios")])

    movs_de: dict[str, list[dict]] = {}
    for m in movs:
        movs_de.setdefault(m["activo"], []).append(m)

    # Transacciones con flujo de dinero (todas, también de activos ocultos)
    solo_valor = {a["id"] for a in datos.vivos("activos") if por_valor(a)}
    txns = [{"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}
            for m in movs if m["tipo"] in CON_TRANSACCION]
    txns_de: dict[str, list[dict]] = {}
    for t in txns:
        txns_de.setdefault(t["activo"], []).append(t)
    pares = _pares_traspaso(txns)

    ene1 = dt.date(hoy.year, 1, 1)
    mes1 = dt.date(hoy.year, hoy.month, 1)

    motor_activos, hoja_activos = {}, {}
    flujos_globales, inv_total, rec_total = [], 0.0, 0.0
    twr_val = []   # (twr, valor) para el TWR global ponderado

    for a in activos:
        aid, ms = a["id"], movs_de.get(a["id"], [])
        solo = por_valor(a)
        if solo:
            reg = precios.ultimo(aid, hoy)
            valor = (reg or {}).get("valor_total") or (reg or {}).get("precio") or 0.0
            cantidad, precio, p = None, None, 0.0
        else:
            cantidad = sum(_f(m.get("cantidad")) * (1 if m["tipo"] in ENTRADAS_CANTIDAD else -1)
                           for m in ms if m["tipo"] in ENTRADAS_CANTIDAD | SALIDAS_CANTIDAD)
            precio = _precio_actual(a, ms, precios, hoy)
            p = precio or 0.0
            valor = cantidad * p

        twr, cagr = (0.0, 0.0) if solo else _twr_cagr(_cadena(aid, txns_de, pares), p, hoy)
        propios = txns_de.get(aid, [])
        tir, roi = _tir_roi(propios, valor, hoy)

        inv = rec = 0.0
        for t in propios:
            importe = flujo(t)
            if importe <= 0:
                continue
            if invierte(t):
                flujos_globales.append((t["fecha"], -importe))
                inv += importe
            elif t["tipo"] in DESINVIERTE:
                flujos_globales.append((t["fecha"], importe))
                inv -= importe                       # una devolución no es un cobro: es menos puesto
            else:
                flujos_globales.append((t["fecha"], importe))
                rec += importe
        inv_total += inv
        rec_total += rec

        vivo = valor > 0
        twr_val.append((twr if vivo else 0.0, valor))
        motor_activos[aid] = {
            "nombre": a.get("nombre", ""), "tipo": a["tipo"],
            "cantidad": cantidad, "precio": p, "valor": valor,
            "ganancia": (valor + rec - inv) if vivo or inv > 0 else 0.0,
            "twr": twr if vivo else 0.0, "cagr": cagr if vivo else 0.0,
            "irr": tir if vivo else 0.0, "roi": roi if vivo else None,
        }

        if solo:
            # Sin participaciones: el coste es el dinero neto que sigue dentro
            coste = max(inv - rec, 0.0)
            hoja_activos[aid] = {
                "cantidad": None, "coste": coste, "precio_medio": None, "precio": None, "valor": valor,
                "ganancia": valor - coste, "ganancia_pct": (valor - coste) / coste if coste else 0.0,
                "dif_fiscal": None, "_refs": _refs_por_valor(aid, precios, ene1, mes1), "_aportado": (inv, rec),
            }
            continue

        # ── Hoja: coste fiscal y derivados (fórmulas de Posiciones) ──
        _, coste, mercado = coste_restante(ms)
        if abs(cantidad) <= 1e-9:
            coste = mercado = 0.0
        precio_hoja = precio if precio is not None else 0.0
        dif = mercado - coste
        ganancia_hoja = valor - coste
        hoja_activos[aid] = {
            "cantidad": cantidad,
            "coste": coste,
            "precio_medio": coste / cantidad if cantidad else 0.0,
            "precio": precio_hoja,
            "valor": valor,
            "ganancia": ganancia_hoja,
            "ganancia_pct": ganancia_hoja / coste if coste else 0.0,
            "dif_fiscal": dif if dif is not None and abs(dif) >= 0.01 else None,
        }
        partes = cantidad
        ref_ene = _precio_referencia(a, precios, ene1, partes)
        ref_mes = _precio_referencia(a, precios, mes1, partes)
        hoja_activos[aid]["_refs"] = (ref_ene, ref_mes)

    # ── Globales del motor ──
    valor_total = sum(v for _, v in twr_val)
    twr_global = sum(t * v / valor_total for t, v in twr_val) if valor_total > 0 else 0.0
    if txns:
        dias = (hoy - min(t["fecha"] for t in txns)).days
        cagr_global = (1 + twr_global) ** (365.0 / dias) - 1 if dias > 0 and (1 + twr_global) > 0 else 0.0
    else:
        cagr_global = 0.0
    if valor_total > 0:
        flujos_globales.append((hoy, valor_total))
    tir_global = xirr(flujos_globales) if flujos_globales else 0.0
    roi_global = (valor_total + rec_total - inv_total) / inv_total if inv_total > 0 else 0.0

    ytd_w = mes_w = base = 0.0
    for a in activos:
        m = motor_activos[a["id"]]
        h = hoja_activos[a["id"]]
        if m["valor"] <= 0:
            continue
        ref_ene, ref_mes = h["_refs"]
        if por_valor(a):
            # Sin precio unitario: se descuenta lo aportado en el periodo
            for ref, fecha_ref, acumulador in ((ref_ene, ene1, "ytd"), (ref_mes, mes1, "mes")):
                if not ref or ref <= 0:
                    continue
                neto = sum(flujo(t) * (-1 if invierte(t) else 1)
                           for t in txns_de.get(a["id"], []) if t["fecha"] >= fecha_ref)
                variacion = (m["valor"] + neto) / ref - 1
                if acumulador == "ytd":
                    ytd_w += variacion * m["valor"]
                else:
                    mes_w += variacion * m["valor"]
            base += m["valor"]
            continue
        if ref_ene and ref_ene > 0:
            ytd_w += (m["precio"] / ref_ene - 1) * m["valor"]
        if ref_mes and ref_mes > 0:
            mes_w += (m["precio"] / ref_mes - 1) * m["valor"]
        base += m["valor"]

    # ── Hoja: pesos, YTD y mes por activo (fórmulas de Rentabilidad) ──
    total_hoja = sum(h["valor"] for h in hoja_activos.values())
    for aid, h in hoja_activos.items():
        ref_ene, ref_mes = h.pop("_refs")
        aportado = h.pop("_aportado", None)
        h["peso"] = h["valor"] / total_hoja if total_hoja else 0.0
        cero = h["valor"] == 0
        h["rent_valor"] = h["valor"]
        h["rent_ganancia"] = None if cero else h["ganancia"]
        h["rent_ganancia_pct"] = None if cero else h["ganancia_pct"]

        def variacion(referencia, desde):
            """Cuánto ha variado desde esa fecha, descontando lo aportado."""
            if cero or not referencia:
                return None
            if aportado is None:                       # con participaciones: por precio
                return h["precio"] / referencia - 1 if h["precio"] else None
            neto = sum(flujo(t) * (-1 if invierte(t) else 1)
                       for t in txns_de.get(aid, []) if t["fecha"] >= desde)
            return (h["valor"] + neto) / referencia - 1

        h["rent_ytd"] = variacion(ref_ene, ene1)
        h["rent_mes"] = variacion(ref_mes, mes1)
        h["rent_peso"] = h["valor"] / total_hoja if total_hoja else 0.0

    coste_total = sum(h["coste"] for h in hoja_activos.values())
    return {
        "fecha_calculo": hoy.isoformat(),
        "motor": {
            "activos": motor_activos,
            "global": {
                "twr": twr_global, "cagr": cagr_global, "irr": tir_global, "roi": roi_global,
                "ytd": ytd_w / base if base > 0 else 0.0,
                "mes": mes_w / base if base > 0 else 0.0,
                "valor_total": valor_total,
            },
        },
        "hoja": {
            "activos": hoja_activos,
            "global": {
                "valor_total": total_hoja,
                "coste_total": coste_total,
                "ganancia": total_hoja - coste_total,
                "ganancia_pct": (total_hoja - coste_total) / coste_total if coste_total else 0.0,
            },
        },
    }