185 lines
8.2 KiB
Python
185 lines
8.2 KiB
Python
"""
|
|
traza_btc.py — De dónde viene cada moneda (UTXO) que tienes hoy.
|
|
|
|
Una «moneda» es una salida de transacción tuya que todavía no has gastado.
|
|
Para trazarla se va hacia atrás:
|
|
|
|
• Si la transacción que la creó es una recompensa de bloque → minado.
|
|
• Si en esa transacción no gastaste nada tuyo → entró de fuera (una compra,
|
|
una retirada de un exchange, un cobro…): es un ORIGEN.
|
|
• Si gastaste monedas tuyas → se sigue por cada una de ellas.
|
|
|
|
Cuando una transacción junta varias monedas (una consolidación), la moneda
|
|
resultante viene «un poco de cada una». Se reparte a prorrata: cada entrada
|
|
aporta la parte proporcional a lo que valía. Es el criterio habitual y el
|
|
único que no se inventa nada; la comisión de red se la reparten igual.
|
|
|
|
Si en la transacción hay entradas que no son tuyas (coinjoin, payjoin), esa
|
|
parte figura como «de terceros en la misma transacción».
|
|
|
|
Nada de esto cambia cifras de la cartera ni tiene efecto fiscal (el bitcoin
|
|
es un bien homogéneo y el coste se calcula por FIFO global): es para saber
|
|
la procedencia de cada moneda.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
SATOSHI = 100_000_000
|
|
MAX_NODOS = 400 # nodos del árbol que se enseñan (el reparto se calcula entero)
|
|
|
|
|
|
def _salida(tx: dict, indice: int) -> dict | None:
|
|
return next((s for s in tx["salidas"] if s["indice"] == indice), None)
|
|
|
|
|
|
def monedas(txs: dict, propios: dict) -> list[dict]:
|
|
"""Las salidas tuyas sin gastar."""
|
|
gastadas = {(e["txid"], e["indice"]) for tx in txs.values() if not tx["coinbase"]
|
|
for e in tx["entradas"]}
|
|
salida = []
|
|
for txid, tx in txs.items():
|
|
for s in tx["salidas"]:
|
|
p = propios.get(s["script"])
|
|
if p and (txid, s["indice"]) not in gastadas:
|
|
salida.append({"txid": txid, "indice": s["indice"], "sats": s["satoshis"],
|
|
"monedero": p["monedero"], "activo": p["activo"],
|
|
"altura": tx.get("altura", 0), "fecha": tx.get("fecha")})
|
|
salida.sort(key=lambda m: (m["fecha"] or "9999", m["txid"], m["indice"]))
|
|
return salida
|
|
|
|
|
|
def _entradas_propias(tx: dict, txs: dict, propios: dict) -> tuple[list, int]:
|
|
"""[(txid, indice, sats)] de las entradas tuyas, y lo que ponen los demás."""
|
|
propias, suyas_total = [], 0
|
|
for e in tx["entradas"]:
|
|
previa = txs.get(e["txid"])
|
|
s = _salida(previa, e["indice"]) if previa else None
|
|
if s and s["script"] in propios:
|
|
propias.append((e["txid"], e["indice"], s["satoshis"]))
|
|
suyas_total += s["satoshis"]
|
|
total_salidas = sum(s["satoshis"] for s in tx["salidas"])
|
|
# Lo de terceros no se conoce directamente: es lo que sale de más sobre lo
|
|
# tuyo (sin contar su parte de comisión, que no se puede saber)
|
|
ajeno = max(total_salidas - suyas_total, 0) if len(propias) < len(tx["entradas"]) else 0
|
|
return propias, ajeno
|
|
|
|
|
|
class Trazador:
|
|
def __init__(self, txs: dict, propios: dict):
|
|
self.txs, self.propios = txs, propios
|
|
self._memo: dict[tuple, dict] = {}
|
|
|
|
def origenes(self, txid: str, indice: int) -> dict[str, float]:
|
|
"""Fracción del valor de esa salida que viene de cada origen (clave: txid del origen,
|
|
o «ajeno:<txid>» para la parte de terceros)."""
|
|
clave = (txid, indice)
|
|
if clave in self._memo:
|
|
return self._memo[clave]
|
|
self._memo[clave] = {} # por si hubiera un ciclo (no debería)
|
|
tx = self.txs.get(txid)
|
|
if tx is None:
|
|
r = {f"desconocido:{txid}": 1.0}
|
|
elif tx["coinbase"]:
|
|
r = {txid: 1.0}
|
|
else:
|
|
propias, ajeno = _entradas_propias(tx, self.txs, self.propios)
|
|
if not propias:
|
|
r = {txid: 1.0}
|
|
else:
|
|
total = sum(v for _, _, v in propias) + ajeno
|
|
r = {}
|
|
for t, i, v in propias:
|
|
for o, f in self.origenes(t, i).items():
|
|
r[o] = r.get(o, 0.0) + f * v / total
|
|
if ajeno:
|
|
r[f"ajeno:{txid}"] = r.get(f"ajeno:{txid}", 0.0) + ajeno / total
|
|
self._memo[clave] = r
|
|
return r
|
|
|
|
def arbol(self, txid: str, indice: int, sats: float) -> dict:
|
|
"""El camino hacia atrás, para enseñarlo (se corta a MAX_NODOS)."""
|
|
cuenta = [0]
|
|
|
|
def nodo(txid, indice, atribuido, visto):
|
|
cuenta[0] += 1
|
|
tx = self.txs.get(txid)
|
|
s = _salida(tx, indice) if tx else None
|
|
p = self.propios.get(s["script"]) if s else None
|
|
n = {"txid": txid, "indice": indice, "fecha": tx.get("fecha") if tx else None,
|
|
"sats": s["satoshis"] if s else None, "atribuido": round(atribuido),
|
|
"monedero": p["monedero"] if p else None, "activo": p["activo"] if p else None,
|
|
"hijos": [], "cortado": False}
|
|
if tx is None:
|
|
n["clase"] = "desconocido"
|
|
return n
|
|
if tx["coinbase"]:
|
|
n["clase"] = "mineria"
|
|
return n
|
|
propias, ajeno = _entradas_propias(tx, self.txs, self.propios)
|
|
if not propias:
|
|
n["clase"] = "origen"
|
|
return n
|
|
n["clase"] = "paso"
|
|
total = sum(v for _, _, v in propias) + ajeno
|
|
for t, i, v in sorted(propias, key=lambda x: -x[2]):
|
|
if cuenta[0] >= MAX_NODOS or (t, i) in visto:
|
|
n["cortado"] = True
|
|
break
|
|
n["hijos"].append(nodo(t, i, atribuido * v / total, visto | {(t, i)}))
|
|
if ajeno:
|
|
n["ajeno"] = round(atribuido * ajeno / total)
|
|
return n
|
|
|
|
return nodo(txid, indice, sats, frozenset({(txid, indice)}))
|
|
|
|
|
|
def decisiones(config: dict, movimientos: list[dict]) -> dict[str, dict]:
|
|
"""Lo que decidiste de cada transacción de entrada: {txid: {como, mov, desde}}."""
|
|
por_id = {m["id"]: m for m in movimientos}
|
|
salida: dict[str, dict] = {}
|
|
for m in movimientos:
|
|
e = m.get("extracto")
|
|
if isinstance(e, str) and e.startswith("btc:") and not m.get("importacion"):
|
|
salida.setdefault(e[4:], {"como": m["tipo"].lower(), "mov": m})
|
|
for pid, r in (config.get("resueltas") or {}).items():
|
|
txid, clase = pid.split(":")[:2]
|
|
if clase not in ("entrada", "mineria"):
|
|
continue
|
|
movs = [por_id[i] for i in r.get("movs") or [] if i in por_id]
|
|
if r.get("enlace") in por_id:
|
|
movs.append(por_id[r["enlace"]])
|
|
d = {"como": r.get("como"), "mov": None, "desde": None}
|
|
for m in movs:
|
|
if m["tipo"] == "Traspaso_Salida":
|
|
d["desde"] = m["activo"]
|
|
elif m["tipo"] == "Traslado":
|
|
# Llegó de otro lugar tuyo (la retirada de un exchange)
|
|
d["desde_lugar"] = m.get("lugar") or ""
|
|
d["mov"] = m
|
|
else:
|
|
d["mov"] = m
|
|
salida[txid] = d
|
|
return salida
|
|
|
|
|
|
def describir(clave: str, txs: dict, propios: dict, decididas: dict) -> dict:
|
|
"""Qué es un origen, para enseñarlo."""
|
|
if clave.startswith("ajeno:"):
|
|
txid = clave[6:]
|
|
tx = txs.get(txid) or {}
|
|
return {"clave": clave, "txid": txid, "clase": "ajeno", "fecha": tx.get("fecha"),
|
|
"texto": "De terceros en la misma transacción (coinjoin o payjoin)"}
|
|
if clave.startswith("desconocido:"):
|
|
return {"clave": clave, "txid": clave[12:], "clase": "desconocido", "fecha": None,
|
|
"texto": "Transacción que no he podido leer"}
|
|
tx = txs.get(clave) or {}
|
|
recibido = sum(s["satoshis"] for s in tx.get("salidas") or [] if s["script"] in propios)
|
|
d = decididas.get(clave) or {}
|
|
m = d.get("mov")
|
|
clase = "mineria" if tx.get("coinbase") else "entrada"
|
|
return {"clave": clave, "txid": clave, "clase": clase, "fecha": tx.get("fecha"),
|
|
"recibido": recibido, "como": d.get("como"), "desde": d.get("desde"),
|
|
"desde_lugar": d.get("desde_lugar"),
|
|
"tipo": m["tipo"] if m else None, "importe": m.get("importe") if m else None,
|
|
"cantidad": m.get("cantidad") if m else None, "notas": (m.get("notas") or "") if m else "",
|
|
"texto": ""}
|