""" historial_btc.py — Importar el historial COMPLETO de tus monederos de bitcoin. Recorre todas las transacciones de todos los monederos desde la primera y te propone qué apuntar de cada una, sin tocar nada hasta que lo aceptas. La comprobación diaria es este mismo motor enseñando solo lo NUEVO: las transacciones que ningún monedero había visto todavía (ver diaria()). Lo que se decide en una se ve decidido en la otra. Todos los monederos se miran como UN conjunto: si el bitcoin pasa de un monedero tuyo a otro, no es una venta ni una compra, aunque los monederos estén en activos distintos de la cartera. Cada transacción se parte en «partes», una por cada cosa que hay que decidir: mineria recompensa de bloque (coinbase) que llega a un monedero tuyo entrada llega bitcoin de fuera: compra, cobro, regalo… o vuelve de algo tuyo que no está en los monederos (un nodo Lightning, un exchange) salida sale bitcoin a fuera: venta, pago… o va a algo tuyo fuera de los monederos (abrir un canal Lightning, un exchange). NO se supone venta: se pregunta. interno pasa de un monedero tuyo a otro que está en OTRO lugar (traslado) o en otro activo (traspaso) comision la comisión de red de lo que envías tú Cada monedero es un «lugar» de su activo (ver lugares.py). Mover entre lugares tuyos es un traslado: no es compra ni venta. Si el extracto de un exchange dejó apuntada una retirada «Exchange → ?» y llega esa cantidad a uno de tus monederos, se propone ENLAZARLA: se completa el destino del traslado. Un monedero puede ser COMPARTIDO: solo una parte de lo que hay en él es tuya («parte»: 0,5 = la mitad). Entonces todo lo suyo cuenta por esa parte: de 1 BTC que llega se propone 0,5, y cuadra contra la mitad de lo que ve el nodo. En un reparto (una salida a un monedero tuyo y otra al otro dueño) lo tuyo es un traslado y lo del otro ni aparece: nunca fue tuyo. Lo que ya tienes apuntado no se toca: • Si una transacción ya la guardó la comprobación diaria de antes de unificarse con esto (extracto «btc:» sin importación), no se vuelve a proponer. • Si hay una operación tuya que parece la misma (mismo activo, cantidad casi igual, fecha cercana), se propone ENLAZARLA: solo se le anota el txid, sin cambiar ninguna cifra. Todo lo que se guarda en una importación lleva su marca y se puede deshacer entera («Deshacer esta importación»). Precios: el importe que se propone es una ESTIMACIÓN con el cierre del día (spot_btc.py) y se guarda como «estimado». En una compra P2P lo que cuenta es lo que pagaste de verdad: corrígelo. """ from __future__ import annotations import datetime as dt import uuid from . import btc, cache_btc, spot_btc from .calculo import coste_restante from .electrum import ErrorElectrum, conectar from .operaciones import ErrorOperacion SATOSHI = btc.SATOSHI # Qué se puede hacer con cada clase de parte (la primera no es la propuesta: # la propuesta la decide _proponer) ACCIONES = { "mineria": ["mineria", "enlazar", "no_apuntar"], "entrada": ["compra", "cobro", "regalo", "mineria", "traslado", "traspaso", "enlazar", "no_apuntar"], "salida": ["venta", "pago", "traslado", "traspaso", "enlazar", "no_apuntar"], "interno": ["traslado", "traspaso", "enlazar", "no_apuntar"], "comision": ["comision", "enlazar", "no_apuntar"], } # Tipos de operación ya apuntada con los que puede enlazarse cada clase ENLAZABLES = { "mineria": {"Mineria", "Recompensa", "Compra"}, "entrada": {"Compra", "Mineria", "Recompensa", "Traspaso_Entrada", "Traslado"}, "salida": {"Venta", "Traspaso_Salida", "Reembolso", "Traslado"}, "interno": {"Traspaso_Salida", "Traspaso_Entrada", "Comision", "Traslado"}, "comision": {"Comision"}, } VENTANA = { # días (antes, después) de la transacción en que se busca la operación apuntada "entrada": (30, 3), # la compra suele apuntarse el día que se paga, antes de que llegue "mineria": (3, 3), "salida": (3, 30), # la venta se cobra después de enviar "interno": (3, 3), "comision": (3, 3), } TOLERANCIA_RELATIVA = 0.005 # 0,5 % de diferencia de cantidad TOLERANCIA_SATS = 1000 MEDIO_SAT = 0.25 # por debajo de esto, una cifra en satoshis es cero DECIMALES = 9 # medio satoshi cabe: la mitad de un monedero compartido es exacta class ErrorHistorial(btc.ErrorBitcoin): pass # ── Escaneo de todos los monederos a la vez ────────────────────────────────── def escanear_todo(cliente, monederos: list[dict]) -> dict: propios: dict[str, dict] = {} # script → {monedero, activo} alturas: dict[str, int] = {} saldos: dict[str, int] = {} nombres: dict[str, str] = {} usadas: dict[str, int] = {} txids: dict[str, list[str]] = {} # monedero → sus transacciones cortados: list[str] = [] for m in monederos: try: r = btc.escanear(cliente, m) except (ValueError, ErrorElectrum) as e: raise ErrorHistorial(f"«{m.get('nombre')}»: {e}") from None nombres[m["id"]] = m.get("nombre") or m["id"] saldos[m["id"]] = r["sats"] usadas[m["id"]] = r["usadas"] txids[m["id"]] = sorted({t["txid"] for t in r["historial"] if t["txid"]}) if r.get("cortada"): cortados.append(nombres[m["id"]]) for script in r["scripts"]: propios[script] = {"monedero": m["id"], "activo": m["activo"]} for t in r["historial"]: if t["txid"]: alturas[t["txid"]] = max(alturas.get(t["txid"], 0), t["altura"]) return {"propios": propios, "alturas": alturas, "saldos": saldos, "nombres": nombres, "cortados": cortados, "usadas": usadas, "txids": txids} # ── Análisis de una transacción ────────────────────────────────────────────── def analizar(tx: dict, propios: dict, anteriores: dict, alturas: dict) -> dict: """Cuánto entra y sale de cada activo, y cuánto va a (o viene de) fuera. De un monedero compartido solo cuenta tu parte (propios[script]["parte"]), así que las cifras pueden llevar medio satoshi.""" entra: dict[str, float] = {} # activo → sats que recibe (tu parte) sale: dict[str, float] = {} # activo → sats que gasta (tu parte) monederos: set[str] = set() fuera = 0 # sats a direcciones que no son tuyas scripts_fuera = [] for s in tx["salidas"]: p = propios.get(s["script"]) if p: entra[p["activo"]] = entra.get(p["activo"], 0) + s["satoshis"] * p.get("parte", 1) monederos.add(p["monedero"]) else: fuera += s["satoshis"] scripts_fuera.append(s["script"]) entradas_ajenas = incompleta = False total_entradas = 0 if not tx["coinbase"]: for e in tx["entradas"]: previa = anteriores.get(e["txid"]) if previa is None: # Una entrada solo puede ser tuya si su transacción está en tu historial if e["txid"] in alturas: incompleta = True else: entradas_ajenas = True continue salida = next((s for s in previa["salidas"] if s["indice"] == e["indice"]), None) if salida is None: # De las recompensas de bloque solo se guardan tus salidas: # si no está, era de otro minero if previa.get("recortada"): entradas_ajenas = True else: incompleta = True continue total_entradas += salida["satoshis"] p = propios.get(salida["script"]) if p: sale[p["activo"]] = sale.get(p["activo"], 0) + salida["satoshis"] * p.get("parte", 1) monederos.add(p["monedero"]) else: entradas_ajenas = True todas_mias = bool(sale) and not entradas_ajenas and not incompleta comision = total_entradas - sum(s["satoshis"] for s in tx["salidas"]) if todas_mias else None desigual = False if todas_mias and total_entradas: # Con un monedero compartido, de la comisión pagas tu parte, y «fuera» # es lo TUYO que no vuelve a ti: lo que se lleva el otro dueño era suyo mio = sum(sale.values()) comision = comision * mio / total_entradas fuera = mio - sum(entra.values()) - comision # Te llega más de lo que era tuyo (un reparto que no es a partes # iguales): se propone por el efecto neto, como una transacción mixta desigual = fuera < -MEDIO_SAT fuera = fuera if fuera > MEDIO_SAT else 0 return {"entra": entra, "sale": sale, "fuera": fuera, "scripts_fuera": scripts_fuera, "entradas_ajenas": entradas_ajenas, "incompleta": incompleta, "desigual": desigual, "comision": comision if comision is None or comision >= 0 else None, "monederos": sorted(monederos), "coinbase": tx["coinbase"]} def _pista_salida(scripts_fuera: list[str]) -> str: """Una salida única a un script de 34 bytes P2WSH suele ser abrir un canal Lightning.""" if len(scripts_fuera) == 1 and scripts_fuera[0].startswith("0020") and len(scripts_fuera[0]) == 68: return "Va a una dirección de contrato (P2WSH): ¿abriste un canal Lightning?" return "" def partes_de(txid: str, a: dict) -> list[dict]: """Las decisiones que hay que tomar sobre una transacción.""" partes = [] def parte(clase, activo, sats, **extra): partes.append({"id": f"{txid}:{clase}:{activo}", "txid": txid, "clase": clase, "activo": activo, "cantidad": round(sats / SATOSHI, DECIMALES), **extra}) if a["coinbase"]: for activo, sats in sorted(a["entra"].items()): parte("mineria", activo, sats) return partes if not a["sale"]: # Nada tuyo sale: todo lo que llega viene de fuera for activo, sats in sorted(a["entra"].items()): parte("entrada", activo, sats) return partes # Sale algo tuyo. El activo de origen es el que más pone. origen = max(a["sale"].items(), key=lambda x: (x[1], x[0]))[0] mixta = len(a["sale"]) > 1 or a["entradas_ajenas"] or a["incompleta"] or a.get("desigual") if mixta: # Transacción con entradas de varios dueños (coinjoin, payjoin) o que # no se ha podido leer entera: se propone por el efecto neto en cada activo activos = sorted(set(a["entra"]) | set(a["sale"])) for activo in activos: neto = a["entra"].get(activo, 0) - a["sale"].get(activo, 0) if neto > MEDIO_SAT: parte("entrada", activo, neto, mixta=True) elif neto < -MEDIO_SAT: parte("salida", activo, -neto, mixta=True) return partes # Todas las entradas son tuyas y de un solo activo for activo, sats in sorted(a["entra"].items()): if activo != origen and sats > 0: parte("interno", origen, sats, destino=activo) if a["fuera"] > 0: parte("salida", origen, a["fuera"], pista=_pista_salida(a["scripts_fuera"])) if a["comision"] and a["comision"] > MEDIO_SAT: parte("comision", origen, a["comision"]) return partes def cuenta(activo: str, lugar: str | None) -> str: """Clave «activo@lugar» con la que se agrupa el dinero de cada monedero.""" return f"{activo}@{lugar}" if lugar else activo def separar(clave: str) -> tuple[str, str]: activo, _, lugar = str(clave).partition("@") return activo, lugar def propios_por_lugar(propios: dict, monederos: list[dict]) -> dict: """Los scripts de cada monedero, agrupados por activo Y lugar, con la parte que es tuya si el monedero es compartido.""" lugar_de = {m["id"]: m.get("lugar") or "" for m in monederos} parte_de = {m["id"]: btc.parte_de(m) for m in monederos} return {s: {**p, "activo": cuenta(p["activo"], lugar_de.get(p["monedero"], "")), "parte": parte_de.get(p["monedero"], 1)} for s, p in propios.items()} def _acciones(p: dict) -> list[str]: if p["clase"] == "interno": otra = "traspaso" if p.get("destino") == p["activo"] else "traslado" return [a for a in ACCIONES["interno"] if a != otra] return ACCIONES[p["clase"]] # ── Lo que ya está apuntado ────────────────────────────────────────────────── def _txid_de(m: dict) -> str | None: if m.get("txid"): return m["txid"] e = m.get("extracto") return e[4:] if isinstance(e, str) and e.startswith("btc:") else None def ya_resueltas(movimientos: list[dict], config: dict) -> tuple[set, set]: """(txids resueltos enteros, ids de partes resueltas).""" txids = set() for m in movimientos: # Lo que guardó la comprobación diaria antigua (extracto «btc:» sin # importación) resuelve la transacción entera. Lo de una importación y los enlaces («txid») # resuelven solo su parte, que queda en «resueltas»: si no, al enlazar # la salida de una transacción se perdería su comisión sin decidir. e = m.get("extracto") if isinstance(e, str) and e.startswith("btc:") and not m.get("importacion"): txids.add(e[4:]) return txids, set((config.get("resueltas") or {}).keys()) def _traslado_encaja(parte: dict, m: dict) -> bool: """¿Ese traslado apuntado puede ser esta parte? Le falta justo el extremo que es este monedero (o ya lo tiene) y el otro extremo es otro lugar.""" aqui = parte.get("lugar") or "" if parte["clase"] == "entrada": return (m.get("lugar_destino") or "") in ("", aqui) and (m.get("lugar") or "") != aqui if parte["clase"] == "salida": return (m.get("lugar") or "") in ("", aqui) and (m.get("lugar_destino") or "") != aqui if parte["clase"] == "interno": return (m.get("lugar") or "") in ("", aqui) \ and (m.get("lugar_destino") or "") in ("", parte.get("destino_lugar") or "") return False def _candidatas(parte: dict, movimientos: list[dict], fecha: str) -> list[dict]: tipos = ENLAZABLES[parte["clase"]] antes, despues = VENTANA[parte["clase"]] dia = dt.date.fromisoformat(fecha) q = parte["cantidad"] salida = [] for m in movimientos: if m["activo"] != parte["activo"] or m["tipo"] not in tipos or _txid_de(m): continue if m["tipo"] == "Traslado" and not _traslado_encaja(parte, m): continue # Lo apuntado en OTRO lugar no es esto (la comisión de retirada de # un exchange no es la comisión de red de tu monedero, aunque cueste igual) if m["tipo"] != "Traslado" and m.get("lugar") and parte.get("lugar") and m["lugar"] != parte["lugar"]: continue if m.get("traslado"): continue if parte["clase"] == "interno" and m["tipo"] != "Traslado" and parte.get("destino_lugar"): continue dm = dt.date.fromisoformat(m["fecha"]) if not -antes <= (dm - dia).days <= despues: continue mq = float(m.get("cantidad") or 0) if not mq: continue dif = abs(mq - q) exacta = dif <= max(q * TOLERANCIA_RELATIVA, TOLERANCIA_SATS / SATOSHI) salida.append({"id": m["id"], "fecha": m["fecha"], "tipo": m["tipo"], "cantidad": mq, "importe": m.get("importe"), "notas": m.get("notas") or "", "lugar": m.get("lugar") or "", "lugar_destino": m.get("lugar_destino") or "", "exacta": exacta, # Un traslado que encaja va primero: es la retirada del exchange "_orden": (not (exacta and m["tipo"] == "Traslado"), dif / q if q else dif, abs((dm - dia).days))}) salida.sort(key=lambda c: c["_orden"]) return salida[:8] def _otra_cantidad(r: dict, ahora: float, por_id: dict) -> bool: """¿Esa decisión se guardó por una cantidad distinta de la que tiene hoy la parte? Pasa al marcar un monedero como compartido (o al dejar de serlo).""" if r.get("como") == "no_apuntar": return False un_sat = 1 / SATOSHI if r.get("cantidad") is not None: return abs(r["cantidad"] - ahora) > un_sat # Decisiones de antes de guardar la cantidad: se mira lo que se creó o enlazó if r.get("como") != "traspaso" and len(r.get("movs") or []) == 1: m = por_id.get(r["movs"][0]) return bool(m) and abs(float(m.get("cantidad") or 0) - ahora) > un_sat if r.get("enlace"): m = por_id.get(r["enlace"]) # un enlace no tiene por qué ser exacto return bool(m) and abs(float(m.get("cantidad") or 0) - ahora) > ahora * 0.1 return False # ── Proponer ───────────────────────────────────────────────────────────────── def _coste_medio(movimientos: list[dict], activo: str, fecha: str) -> float | None: """Coste fiscal medio por unidad hasta esa fecha (mismo criterio que la hoja).""" ms = sorted((m for m in movimientos if m["activo"] == activo and m["fecha"] <= fecha), key=lambda m: (m["fecha"], m.get("orden") or 0, m["id"])) unidades, coste, _ = coste_restante(ms) return coste / unidades if unidades > 1e-12 else None def _proponer(p: dict, usados: set) -> None: exacta = next((c for c in p["candidatas"] if c["exacta"] and c["id"] not in usados), None) if exacta: p["accion"], p["enlace"] = "enlazar", exacta["id"] usados.add(exacta["id"]) return p["enlace"] = None interno = "traslado" if p.get("destino") == p["activo"] else "traspaso" p["accion"] = {"mineria": "mineria", "entrada": "compra", "interno": interno, "comision": "comision", "salida": ""}[p["clase"]] # una salida no se supone venta: se pregunta def _vigilables(config: dict) -> list[dict]: monederos = [m for m in config.get("monederos") or [] if btc.vigilable(m)] if not monederos: raise ErrorHistorial("Todavía no has añadido ningún monedero.") return monederos def leer_nodo(config: dict, carpeta) -> dict: """Pregunta al nodo por todos los monederos (solo trae las transacciones que faltan en la caché) y guarda lo leído en este equipo.""" monederos = _vigilables(config) servidores = [btc.normalizar_servidor(s) for s in config.get("servidores") or []] if not servidores: raise ErrorHistorial("Añade primero la dirección de tu servidor Electrum.") cache = cache_btc.cargar(carpeta) cliente, servidor, fallos = conectar(servidores, btc.proxy_de(config)) try: esc = cache_btc.leer_nodo(cliente, monederos, cache, escanear_todo) finally: cliente.cerrar() cache_btc.guardar(carpeta, cache) return {"cache": cache, "servidor": servidor.get("nombre") or servidor.get("host"), "fallos": fallos, "saldos": esc["saldos"], "usadas": esc["usadas"], "txids": esc["txids"]} def historial(config: dict, datos, hoy: dt.date, precios_activo=None, carpeta=None, leer: bool = True, lectura: dict | None = None) -> dict: """Las partes pendientes de todo el historial. leer=True pregunta al nodo (solo trae lo que falta en la caché); con leer=False se usa lo último que se leyó en este equipo, al instante. «lectura» es lo que devolvió leer_nodo(), si quien llama ya lo ha leído. precios_activo(activo) -> [{fecha, precio}] guardados en la cartera (para los días que la serie empaquetada no cubre). """ monederos = _vigilables(config) if leer and lectura is None: lectura = leer_nodo(config, carpeta) cache = lectura["cache"] if lectura else cache_btc.cargar(carpeta) servidor, fallos = (lectura["servidor"], lectura["fallos"]) if lectura else (None, []) esc = cache_btc.escaneo_guardado(cache, monederos) txs = {t: cache["txs"][t] for t in esc["alturas"] if t in cache["txs"]} ilegibles = sorted(t for t in esc["alturas"] if t not in txs) propios = propios_por_lugar(esc["propios"], monederos) movimientos = datos.vivos("movimientos") por_id = {m["id"]: m for m in movimientos} txids_hechos, _ = ya_resueltas(movimientos, config) resueltas = config.get("resueltas") or {} resueltas_de: dict[str, list[str]] = {} for pid in resueltas: resueltas_de.setdefault(pid.split(":", 1)[0], []).append(pid) borradores = config.get("borradores") or {} partes, cambiadas, total, hechas = [], [], 0, 0 for txid, tx in txs.items(): if txid in txids_hechos: continue a = analizar(tx, propios, txs, esc["alturas"]) altura = tx.get("altura", 0) fecha = min(tx.get("fecha") or hoy.isoformat(), hoy.isoformat()) actuales = [p for p in partes_de(txid, a) if p["cantidad"] > 0] ids = {p["id"] for p in actuales} # Una decisión guardada que ya no encaja (p. ej. has añadido el monedero # al que fue aquella «venta»): hay que revisarla viejas = [pid for pid in resueltas_de.get(txid, []) if pid not in ids] # …o que se guardó por otra cantidad (has marcado el monedero como compartido) viejas += [p["id"] for p in actuales if p["id"] in resueltas and _otra_cantidad(resueltas[p["id"]], p["cantidad"], por_id)] if viejas: cambiadas.append({"txid": txid, "fecha": fecha, "antes": [{"id": pid, **{k: resueltas[pid].get(k) for k in ("como", "importacion")}} for pid in viejas], "ahora": sorted(ids)}) for p in actuales: p["activo"], p["lugar"] = separar(p["activo"]) if p.get("destino"): p["destino"], p["destino_lugar"] = separar(p["destino"]) total += 1 if p["id"] in resueltas: hechas += 1 continue p.update({"fecha": fecha, "altura": altura, "confirmada": altura > 0, "monederos": [esc["nombres"].get(i, i) for i in a["monederos"]], "incompleta": a["incompleta"], "borrador": borradores.get(p["id"])}) p.setdefault("mixta", False) p.setdefault("pista", "") partes.append(p) # Orden cronológico; dentro del día, lo que entra antes de lo que sale peso = {"mineria": 0, "entrada": 1, "interno": 2, "salida": 3, "comision": 4} partes.sort(key=lambda p: (p["fecha"], p["altura"] or 10**9, p["txid"], peso[p["clase"]])) usados: set[str] = set() for p in partes: p["acciones"] = _acciones(p) p["candidatas"] = _candidatas(p, movimientos, p["fecha"]) _proponer(p, usados) otros = precios_activo(p["activo"]) if precios_activo else [] precio, dia = spot_btc.precio(p["fecha"], otros) p["precio_ref"], p["dia_precio"] = precio, dia p["importe_ref"] = round(precio * p["cantidad"], 2) if precio else None for c in p["candidatas"]: c.pop("_orden", None) en_nodo: dict[str, int] = {} for m in monederos: k = cuenta(m["activo"], m.get("lugar")) en_nodo[k] = en_nodo.get(k, 0) + esc["saldos"].get(m["id"], 0) * btc.parte_de(m) return {"partes": partes, "cambiadas": cambiadas, "en_nodo": {a: s / SATOSHI for a, s in en_nodo.items()}, "servidor": servidor, "fallos": fallos, "leido": esc["fecha"], "cortados": esc["cortados"], "ilegibles": ilegibles, "transacciones": len(esc["alturas"]), "ya_apuntadas": len(set(esc["alturas"]) & txids_hechos), "progreso": {"total": total, "hechas": hechas}} # ── La comprobación diaria: solo lo nuevo ──────────────────────────────────── def dar_por_vistas(config: dict, txids) -> None: """La comprobación diaria deja de avisar de esas transacciones (si les queda algo por decidir, sigue en el historial completo).""" for m in config.get("monederos") or []: m["vistas"] = sorted(set(m.get("vistas") or []) | set(txids))[-btc.MAX_VISTAS:] def diaria(config: dict, datos, hoy: dt.date, precios_activo=None, carpeta=None) -> dict: """Lee el nodo y devuelve lo mismo que historial(), pero con las partes de las transacciones NUEVAS: las que ningún monedero había visto. Lo demás que quede sin decidir se cuenta en «atrasadas» y se revisa en el historial. De paso deja al día, en la configuración, el saldo de cada monedero. """ monederos = _vigilables(config) lectura = leer_nodo(config, carpeta) ahora = dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds") for m in monederos: m["saldo"] = lectura["saldos"].get(m["id"], 0) / SATOSHI m["direcciones_usadas"] = lectura["usadas"].get(m["id"]) m["comprobado"] = ahora if m.get("inicial"): # Un monedero recién añadido no trae su pasado: se da por visto m["vistas"] = sorted(set(m.get("vistas") or []) | set(lectura["txids"].get(m["id"]) or []))[-btc.MAX_VISTAS:] m["inicial"] = False r = historial(config, datos, hoy, precios_activo, carpeta, lectura=lectura) # Todos los monederos cuentan como uno: basta con que la haya visto uno vistas = {t for m in config.get("monederos") or [] for t in m.get("vistas") or []} nuevas = [p for p in r["partes"] if p["txid"] not in vistas] r["atrasadas"] = len(r["partes"]) - len(nuevas) r["partes"] = nuevas r["diaria"] = True return r # ── Guardar lo decidido ────────────────────────────────────────────────────── NOTAS = { "compra": "Compra", "cobro": "Cobro en bitcoin", "regalo": "Regalo recibido", "mineria": "Minado", "venta": "Venta", "pago": "Pago o regalo en bitcoin (valor de mercado)", "comision": "Comisión de red", "traspaso": "Traspaso entre tus bitcoins", "traslado": "Traslado entre tus lugares", } def _nota(p: dict, accion: str) -> str: base = str(p.get("notas") or "").strip() or NOTAS.get(accion, "") return f"{base} · del nodo {p['txid'][:12]}…"[:200] def aplicar(cartera, config: dict, decisiones: list[dict], hoy: dt.date) -> dict: """Guarda las partes decididas. Cada una: la parte tal como vino, más «accion», y según el caso «importe», «destino»/«origen» y «enlace».""" imp = {"id": "i" + dt.datetime.now().strftime("%Y%m%d%H%M%S") + uuid.uuid4().hex[:4], "fecha": dt.datetime.now().isoformat(timespec="seconds"), "creados": [], "enlazados": [], "partes": []} resueltas = config.setdefault("resueltas", {}) fallos, guardadas = [], 0 # En orden: lo que entra antes de lo que sale, para que haya saldo peso = {"mineria": 0, "entrada": 1, "interno": 2, "salida": 3, "comision": 4} decisiones = sorted((d for d in decisiones if isinstance(d, dict) and d.get("accion")), key=lambda d: (str(d.get("fecha")), str(d.get("txid")), peso.get(d.get("clase"), 9))) for p in decisiones[:5000]: accion, clase = p.get("accion"), p.get("clase") if clase not in ACCIONES or accion not in ACCIONES[clase] or not p.get("txid") or not p.get("id"): fallos.append({"fecha": p.get("fecha"), "tipo": accion, "motivo": "decisión no válida"}) continue if p["id"] in resueltas: continue try: creados, enlazado, extra = _guardar_parte(cartera, p, accion, imp["id"]) except (btc.ErrorBitcoin, ErrorOperacion) as e: fallos.append({"fecha": p.get("fecha"), "tipo": accion, "motivo": str(e)}) continue imp["creados"] += creados if enlazado: imp["enlazados"].append(enlazado) imp["partes"].append(p["id"]) resueltas[p["id"]] = {"como": accion, "importacion": imp["id"], "movs": creados, "enlace": enlazado, "cantidad": float(p["cantidad"]), **extra} (config.get("borradores") or {}).pop(p["id"], None) guardadas += 1 if imp["partes"]: config.setdefault("importaciones", []).append(imp) return {"guardadas": guardadas, "creados": len(imp["creados"]), "enlazados": len(imp["enlazados"]), "fallos": fallos, "importacion": imp["id"] if imp["partes"] else None, "partes": imp["partes"]} def _importe(p: dict) -> float | None: v = p.get("importe") if v in (None, ""): v = p.get("importe_ref") try: v = float(v) except (TypeError, ValueError): return None return round(v, 2) if v > 0 else None def _guardar_parte(cartera, p: dict, accion: str, imp_id: str) -> tuple[list[str], str | None, dict]: """(ids creados, id enlazado, datos para poder deshacerlo).""" activo, fecha, q = p["activo"], p["fecha"], float(p["cantidad"]) aqui = p.get("lugar") or "" base = {"activo": activo, "fecha": fecha, "cantidad": q, "notas": _nota(p, accion), "extracto": f"btc:{p['txid']}"[:80], "importacion": imp_id, "actividad": p.get("actividad") or "", "lugar": aqui} if accion == "no_apuntar": return [], None, {} if accion == "enlazar": mid = p.get("enlace") if not mid: raise btc.ErrorBitcoin("elige con qué operación apuntada la enlazas") m = next((x for x in cartera.datos.vivos("movimientos") if x["id"] == mid), None) if m is None: raise btc.ErrorBitcoin("la operación con la que enlazar ya no está") if m["tipo"] == "Traslado": # La retirada del exchange («Exchange → ?») llega a este monedero: se completa if p["clase"] == "entrada": extremo, valor = "lugar_destino", aqui elif p["clase"] == "salida": extremo, valor = "lugar", aqui else: extremo, valor = ("lugar", aqui) if not m.get("lugar") else ("lugar_destino", p.get("destino_lugar") or "") vacio = not m.get(extremo) cartera.completar_traslado(mid, **{extremo: valor}, txid=p["txid"], importacion=imp_id) return [], mid, {"extremo": extremo} if vacio and valor else {} cartera.enlazar_txid(mid, p["txid"]) if aqui and not m.get("lugar") and m["tipo"] != "Traslado": cartera.poner_lugar([mid], aqui) return [], mid, {"lugar_puesto": True} return [], mid, {} if accion == "traslado": otro = p.get("otro_lugar") or "" if p["clase"] == "entrada": desde, hacia = otro, aqui elif p["clase"] == "salida": desde, hacia = aqui, otro else: desde, hacia = aqui, p.get("destino_lugar") or "" if not desde or not hacia: raise btc.ErrorBitcoin("elige el otro lugar del traslado") t = cartera.guardar_traslado({"activo": activo, "fecha": fecha, "cantidad": q, "lugar": desde, "lugar_destino": hacia, "notas": _nota(p, "traslado"), "extracto": base["extracto"], "importacion": imp_id, "txid": p["txid"]}) return [t["id"]], None, {} if accion == "traspaso": otro = p.get("destino") if p["clase"] in ("salida", "interno") else p.get("origen") if not otro or otro == activo: raise btc.ErrorBitcoin("elige el otro activo del traspaso") origen, destino = (activo, otro) if p["clase"] in ("salida", "interno") else (otro, activo) movs = cartera.datos.vivos("movimientos") precio = p.get("precio_ref") importe = _importe(p) or (round(precio * q, 2) if precio else None) if not importe: raise btc.ErrorBitcoin("no sé a qué precio estaba el bitcoin ese día: pon el importe") medio = _coste_medio(movs, origen, fecha) coste = round(medio * q, 2) if medio else importe pares = cartera.traspaso({"origen": origen, "destino": destino, "fecha": fecha, "cantidad_salida": q, "cantidad_entrada": q, "importe": importe, "coste_fiscal": coste, "notas": _nota(p, "traspaso")}) ids = [m["id"] for m in pares] cartera.anotar_movimientos(ids, {"extracto": base["extracto"], "importacion": imp_id, "estimado": True}) if aqui: cartera.poner_lugar([m["id"] for m in pares if m["activo"] == activo], aqui) return ids, None, {} tipo = {"compra": "Compra", "venta": "Venta", "pago": "Venta", "cobro": "Recompensa", "regalo": "Recompensa", "mineria": "Mineria", "comision": "Comision"}[accion] mov = dict(base, tipo=tipo) if tipo in ("Compra", "Venta"): importe = _importe(p) if not importe: raise btc.ErrorBitcoin("no sé a qué precio estaba el bitcoin ese día: pon el importe en euros") mov.update(importe=importe, estimado=not p.get("importe_propio")) elif tipo == "Recompensa": mov["valor_recibido"] = _importe(p) m = cartera.guardar_movimiento(mov) return [m["id"]], None, {} def deshacer(cartera, config: dict, imp_id: str) -> dict: """Borra lo que creó una importación, quita los enlaces que hizo y vuelve a dejar sus partes pendientes.""" imps = config.get("importaciones") or [] imp = next((i for i in imps if i.get("id") == imp_id), None) if imp is None: raise ErrorHistorial("Esa importación ya no está.") vivos = {m["id"] for m in cartera.datos.vivos("movimientos")} creados = [i for i in imp.get("creados") or [] if i in vivos] if creados: cartera.borrar_movimientos(creados) enlazados = [i for i in imp.get("enlazados") or [] if i in vivos] resueltas = config.get("resueltas") or {} como_se_enlazo = {r.get("enlace"): r for pid in imp.get("partes") or [] if (r := resueltas.get(pid) or {}).get("importacion") == imp_id and r.get("enlace")} for mid in enlazados: _deshacer_enlace(cartera, mid, como_se_enlazo.get(mid) or {}) for pid in imp.get("partes") or []: if (resueltas.get(pid) or {}).get("importacion") == imp_id: resueltas.pop(pid, None) # Vuelven a quedar por decidir, pero en el historial: no son novedad dar_por_vistas(config, {pid.split(":", 1)[0] for pid in imp.get("partes") or []}) config["importaciones"] = [i for i in imps if i.get("id") != imp_id] return {"borrados": len(creados), "desenlazados": len(enlazados), "partes": len(imp.get("partes") or [])} def _deshacer_enlace(cartera, mid: str, r: dict) -> None: """Quita el txid y lo que puso el enlace (el extremo de un traslado o el lugar).""" m = next((x for x in cartera.datos.vivos("movimientos") if x["id"] == mid), None) if m is None: return if m["tipo"] == "Traslado": cartera.vaciar_extremo(mid, r.get("extremo") or "") return cartera.enlazar_txid(mid, None) if r.get("lugar_puesto"): cartera.poner_lugar([mid], "") def deshacer_partes(cartera, config: dict, pids: list) -> dict: """Deshace varias decisiones de una vez (las que han dejado de encajar).""" total = {"borrados": 0, "desenlazados": 0, "partes": 0} for pid in pids[:5000]: if pid not in (config.get("resueltas") or {}): continue r = deshacer_parte(cartera, config, str(pid)) total["borrados"] += r["borrados"] total["desenlazados"] += r["desenlazados"] total["partes"] += 1 return total def deshacer_parte(cartera, config: dict, pid: str) -> dict: """Deshace UNA decisión guardada (p. ej. la que ha dejado de encajar al añadir un monedero): borra lo que creó, quita su enlace y la deja pendiente.""" resueltas = config.get("resueltas") or {} r = resueltas.get(pid) if r is None: raise ErrorHistorial("Esa decisión ya no está guardada.") vivos = {m["id"] for m in cartera.datos.vivos("movimientos")} creados = [i for i in r.get("movs") or [] if i in vivos] if creados: cartera.borrar_movimientos(creados) if r.get("enlace") in vivos: _deshacer_enlace(cartera, r["enlace"], r) resueltas.pop(pid, None) dar_por_vistas(config, [pid.split(":", 1)[0]]) for imp in config.get("importaciones") or []: if imp.get("id") == r.get("importacion"): imp["partes"] = [x for x in imp.get("partes") or [] if x != pid] imp["creados"] = [x for x in imp.get("creados") or [] if x not in (r.get("movs") or [])] imp["enlazados"] = [x for x in imp.get("enlazados") or [] if x != r.get("enlace")] config["importaciones"] = [i for i in config.get("importaciones") or [] if i.get("partes")] return {"borrados": len(creados), "desenlazados": 1 if r.get("enlace") in vivos else 0} CAMPOS_BORRADOR = ("accion", "importe", "otro", "enlace") def poner_borradores(config: dict, cambios: dict) -> int: """Lo decidido y aún sin guardar, para seguir otro rato: {id_parte: decisión | None}.""" borradores = config.setdefault("borradores", {}) for pid, d in list(cambios.items())[:5000]: if not isinstance(pid, str) or ":" not in pid: continue if not d: borradores.pop(pid, None) elif isinstance(d, dict): borradores[pid] = {k: d.get(k) for k in CAMPOS_BORRADOR if d.get(k) not in (None, "")} return len(borradores)