781 lines
37 KiB
Python
781 lines
37 KiB
Python
"""
|
|
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:<txid>» 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)
|