530 lines
25 KiB
Python
530 lines
25 KiB
Python
"""
|
||
extractos.py — Lee extractos de bancos y brokers y propone las operaciones.
|
||
|
||
Los lectores de cada formato se apuntan con registrar() (CSV). Da igual que el separador
|
||
sea coma, punto y coma o tabulador, que el orden de las columnas cambie o que
|
||
traiga columnas de más. Cada lector:
|
||
• reconoce compras, ventas, traspasos, dividendos, intereses y bonificaciones;
|
||
• ignora lo que no es inversión (pagos con tarjeta, traspasos de efectivo...);
|
||
• no duplica: cada línea lleva una referencia única que se guarda en la
|
||
operación, y además compara con lo que ya tienes (fecha aproximada, mismo
|
||
activo, mismo importe) por si lo apuntaste a mano o automáticamente.
|
||
|
||
Nada se guarda aquí: se devuelven propuestas para que el usuario acepte.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import datetime as dt
|
||
import hashlib
|
||
import io
|
||
import re
|
||
import unicodedata
|
||
|
||
from . import lugares as motor_lugares
|
||
from . import spot_btc
|
||
from .datos import Datos
|
||
|
||
TOLERANCIA_DIAS = 3
|
||
TOLERANCIA_IMPORTE = 0.02 # € (o 1 %, lo que sea mayor)
|
||
SEPARADORES = ["\t", ",", ";"]
|
||
BOM = ""
|
||
|
||
# Cada formato de extracto que se sabe leer se apunta aquí con registrar(): las
|
||
# columnas que lo delatan, el nombre de la entidad y la función que lo lee.
|
||
# El núcleo no trae ninguno; los de entidades concretas viven en cartera/extras.
|
||
COLUMNAS: dict[str, set] = {}
|
||
NOMBRES: dict[str, str] = {}
|
||
LECTORES: dict = {}
|
||
|
||
|
||
def registrar(clave: str, nombre: str, columnas: set, lector) -> None:
|
||
COLUMNAS[clave], NOMBRES[clave], LECTORES[clave] = set(columnas), nombre, lector
|
||
|
||
|
||
ISIN = re.compile(r"^[A-Z]{2}[A-Z0-9]{9}\d$")
|
||
|
||
|
||
# ── utilidades ───────────────────────────────────────────────────────────────
|
||
|
||
def sin_tildes(texto: str) -> str:
|
||
limpio = unicodedata.normalize("NFD", (texto or "").lower())
|
||
return "".join(c for c in limpio if unicodedata.category(c) != "Mn").strip()
|
||
|
||
|
||
def _num(texto: str) -> float | None:
|
||
"""Número de una celda, tolerando «5.800,00 € (Bruto)» y «227.430483€»."""
|
||
texto = re.sub(r"[^0-9,.\-]", "", (texto or "").strip())
|
||
if not texto or texto in ("-", ".", ","):
|
||
return None
|
||
if "," in texto and "." in texto: # 1,234.56 o 1.234,56
|
||
texto = texto.replace(",", "") if texto.rfind(".") > texto.rfind(",") else texto.replace(".", "").replace(",", ".")
|
||
elif "," in texto:
|
||
texto = texto.replace(",", ".")
|
||
try:
|
||
return float(texto)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _fecha(texto: str) -> dt.date | None:
|
||
texto = (texto or "").strip()
|
||
if not texto:
|
||
return None
|
||
try:
|
||
return dt.date.fromisoformat(texto[:10])
|
||
except ValueError:
|
||
pass
|
||
for formato in ("%d/%m/%Y", "%d-%m-%Y", "%d/%m/%y", "%d.%m.%Y"):
|
||
try:
|
||
return dt.datetime.strptime(texto.split()[0], formato).date()
|
||
except ValueError:
|
||
continue
|
||
return None
|
||
|
||
|
||
def _arreglar(campos: list[str], esperados: int) -> list[str]:
|
||
"""Rehace las celdas partidas cuando el separador es la coma y los decimales
|
||
también: «0,1 €» llega como «0» y «1»."""
|
||
campos = list(campos)
|
||
i = 0
|
||
while len(campos) > esperados and i < len(campos) - 1:
|
||
a, b = campos[i].strip(), campos[i + 1].strip()
|
||
if re.fullmatch(r"-?\d+", a) and re.match(r"^\d{1,2}(\D|$)", b):
|
||
campos[i:i + 2] = [f"{a},{b}"]
|
||
continue
|
||
i += 1
|
||
return campos
|
||
|
||
|
||
def _cabecera(texto: str):
|
||
"""(lector, separador, columnas) o None."""
|
||
limpio = (texto or "").lstrip(BOM)
|
||
primera = limpio.splitlines()[0] if limpio.strip() else ""
|
||
for sep in SEPARADORES:
|
||
columnas = [sin_tildes(c.strip().strip('"')) for c in next(csv.reader([primera], delimiter=sep))]
|
||
for lector, necesarias in COLUMNAS.items():
|
||
if necesarias <= set(columnas):
|
||
return lector, sep, columnas
|
||
return None
|
||
|
||
|
||
def detectar(texto: str) -> str | None:
|
||
encontrada = _cabecera(texto)
|
||
return encontrada[0] if encontrada else None
|
||
|
||
|
||
def _filas(texto: str):
|
||
encontrada = _cabecera(texto)
|
||
if encontrada is None:
|
||
return
|
||
_, sep, cabecera = encontrada
|
||
lector = csv.reader(io.StringIO((texto or "").lstrip(BOM)), delimiter=sep)
|
||
next(lector, None)
|
||
for valores in lector:
|
||
if not any(v.strip() for v in valores):
|
||
continue
|
||
if len(valores) > len(cabecera):
|
||
valores = _arreglar(valores, len(cabecera))
|
||
yield {c: (valores[i].strip() if i < len(valores) else "") for i, c in enumerate(_unicas(cabecera))}
|
||
|
||
|
||
def _unicas(cabecera: list[str]) -> list[str]:
|
||
"""Columnas repetidas («Cur.» en algunos formatos): la 2.ª se llama «cur.#2»."""
|
||
vistas: dict[str, int] = {}
|
||
salida = []
|
||
for c in cabecera:
|
||
vistas[c] = vistas.get(c, 0) + 1
|
||
salida.append(c if vistas[c] == 1 else f"{c}#{vistas[c]}")
|
||
return salida
|
||
|
||
|
||
# ── Un CSV cualquiera: quien lo sube dice qué es cada columna ────────────────
|
||
#
|
||
# Para las entidades que no tienen lector propio. No se adivina nada que no se
|
||
# vea: las columnas las asigna el usuario (se le propone un reparto por el
|
||
# nombre de cada una) y lo que no se entiende se deja fuera, a la vista.
|
||
|
||
PAPELES = { # papel → (cómo se le llama, palabras que lo delatan en una cabecera)
|
||
"fecha": ("Fecha", ("fecha", "date", "datum", "dia")),
|
||
"tipo": ("Qué se hizo (compra, venta…)", ("tipo", "operacion", "concepto", "type", "transaccion", "movimiento", "descripcion")),
|
||
"producto": ("Nombre de la inversión", ("producto", "nombre", "activo", "fondo", "name", "instrumento", "symbol", "ticker")),
|
||
"isin": ("ISIN", ("isin",)),
|
||
"cantidad": ("Cantidad o participaciones", ("cantidad", "participaciones", "titulos", "unidades", "shares", "quantity", "units")),
|
||
"precio": ("Precio por unidad", ("precio", "price", "liquidativo", "cotizacion")),
|
||
"importe": ("Importe en euros", ("importe", "amount", "total", "efectivo", "neto")),
|
||
"comisiones": ("Comisiones", ("comision", "fee", "gastos")),
|
||
}
|
||
OBLIGATORIOS = ("fecha", "importe")
|
||
# Qué operación es, por lo que diga la columna «tipo»
|
||
PALABRAS_TIPO = (
|
||
("Dividendo", ("dividend", "cupon", "coupon", "interes", "interest", "renta")),
|
||
("Venta", ("venta", "sell", "sale", "reembolso", "rescate", "redemption", "vendido")),
|
||
("Compra", ("compra", "buy", "purchase", "suscripcion", "aportacion", "subscription", "comprado")),
|
||
)
|
||
CLASES_GENERICO = ("Fondo", "PlanPensiones", "ETF", "Accion", "Cripto", "Metal")
|
||
|
||
|
||
def columnas(texto: str) -> dict:
|
||
"""Lo que hace falta para que el usuario reparta las columnas de un CSV
|
||
desconocido: la cabecera, unas filas de muestra y un reparto propuesto."""
|
||
cabecera, filas = _tabla(texto)
|
||
if not cabecera:
|
||
raise ValueError("El fichero no parece un CSV: no encuentro una primera línea con los nombres de las columnas.")
|
||
propuesta, usadas = {}, set()
|
||
for papel, (_, palabras) in PAPELES.items():
|
||
for c in cabecera:
|
||
if c not in usadas and any(palabra in c for palabra in palabras):
|
||
propuesta[papel] = c
|
||
usadas.add(c)
|
||
break
|
||
return {"columnas": cabecera, "muestra": [[f.get(c, "") for c in cabecera] for f in filas[:5]], "filas": len(filas),
|
||
"papeles": [[k, nombre, k in OBLIGATORIOS] for k, (nombre, _) in PAPELES.items()],
|
||
"clases": list(CLASES_GENERICO), "propuesta": propuesta}
|
||
|
||
|
||
def _tabla(texto: str) -> tuple[list[str], list[dict]]:
|
||
"""(cabecera, filas) de un CSV cualquiera, con el separador que más columnas saque."""
|
||
limpio = (texto or "").lstrip(BOM)
|
||
primera = limpio.splitlines()[0] if limpio.strip() else ""
|
||
sep = max(SEPARADORES, key=lambda x: len(next(csv.reader([primera], delimiter=x), [])))
|
||
cabecera = _unicas([sin_tildes(c.strip().strip('"')) for c in next(csv.reader([primera], delimiter=sep), [])])
|
||
if len(cabecera) < 2:
|
||
return [], []
|
||
lector = csv.reader(io.StringIO(limpio), delimiter=sep)
|
||
next(lector, None)
|
||
filas = []
|
||
for valores in lector:
|
||
if not any(v.strip() for v in valores):
|
||
continue
|
||
if len(valores) > len(cabecera):
|
||
valores = _arreglar(valores, len(cabecera))
|
||
filas.append({c: (valores[i].strip() if i < len(valores) else "") for i, c in enumerate(cabecera)})
|
||
return cabecera, filas
|
||
|
||
|
||
def _tipo_generico(texto: str, cantidad: float | None, importe: float | None) -> str | None:
|
||
"""Compra, Venta o Dividendo. Primero por lo que diga la columna; si no hay
|
||
o no se entiende, por los signos: lo que se compra sale del bolsillo."""
|
||
limpio = sin_tildes(texto or "")
|
||
for tipo, palabras in PALABRAS_TIPO:
|
||
if any(palabra in limpio for palabra in palabras):
|
||
return tipo
|
||
if limpio:
|
||
return None # dice algo que no entiendo: mejor no inventar
|
||
if cantidad:
|
||
return "Compra" if cantidad > 0 else "Venta"
|
||
if importe:
|
||
return "Compra" if importe < 0 else "Venta"
|
||
return None
|
||
|
||
|
||
def leer_generico(texto: str, mapa: dict) -> tuple[list[dict], list[dict]]:
|
||
"""Lee un CSV con el reparto de columnas que ha dado el usuario:
|
||
mapa = {papel: nombre de columna, "clase": tipo de inversión de lo nuevo}."""
|
||
cabecera, filas = _tabla(texto)
|
||
mapa = {k: v for k, v in (mapa or {}).items() if v}
|
||
faltan = [PAPELES[k][0] for k in OBLIGATORIOS if mapa.get(k) not in cabecera]
|
||
if faltan:
|
||
raise ValueError("Falta decir qué columna es: " + ", ".join(faltan) + ".")
|
||
if not (mapa.get("producto") in cabecera or mapa.get("isin") in cabecera):
|
||
raise ValueError("Falta decir qué columna dice de qué inversión es cada línea: el nombre o el ISIN.")
|
||
clase = mapa.get("clase") if mapa.get("clase") in CLASES_GENERICO else "Fondo"
|
||
de = lambda fila, papel: fila.get(mapa.get(papel, ""), "") # noqa: E731
|
||
utiles, ignoradas = [], []
|
||
for n, fila in enumerate(filas, 2):
|
||
fecha = _fecha(de(fila, "fecha"))
|
||
cantidad, precio, importe = _num(de(fila, "cantidad")), _num(de(fila, "precio")), _num(de(fila, "importe"))
|
||
isin = de(fila, "isin").upper()
|
||
producto = de(fila, "producto") or isin
|
||
tipo = _tipo_generico(de(fila, "tipo"), cantidad, importe)
|
||
motivo = ("sin fecha" if fecha is None else "sin importe" if importe is None and not (cantidad and precio)
|
||
else "no sé qué operación es" if tipo is None else "no dice de qué inversión es" if not producto else "")
|
||
if motivo:
|
||
ignoradas.append({"fecha": de(fila, "fecha"), "que": de(fila, "tipo") or motivo,
|
||
"descripcion": f"línea {n}: {motivo}"[:80]})
|
||
continue
|
||
if tipo == "Venta" and clase in ("Fondo", "PlanPensiones"):
|
||
tipo = "Reembolso"
|
||
if importe is None:
|
||
importe = round(abs(cantidad) * precio, 2)
|
||
linea = _linea(
|
||
ref=_ref("csv", fecha.isoformat(), producto, tipo, cantidad, importe),
|
||
fecha=fecha.isoformat(), tipo=tipo, producto=producto, isin=isin if ISIN.match(isin) else "", clase=clase,
|
||
cantidad=abs(cantidad) if cantidad and tipo != "Dividendo" else None,
|
||
precio=abs(precio) if precio and tipo != "Dividendo" else None,
|
||
importe=abs(importe), comisiones=abs(_num(de(fila, "comisiones")) or 0) or None,
|
||
descripcion=de(fila, "tipo")[:80])
|
||
if linea["cantidad"] and not linea["precio"]:
|
||
linea["precio"] = round(linea["importe"] / linea["cantidad"], 6)
|
||
utiles.append(linea)
|
||
return utiles, ignoradas
|
||
|
||
|
||
GENERICO = "csv" # la clave de «lector» cuando las columnas las ha repartido el usuario
|
||
|
||
|
||
def _ref(prefijo: str, *partes) -> str:
|
||
semilla = "|".join(str(p) for p in partes)
|
||
return f"{prefijo}:" + hashlib.sha256(semilla.encode()).hexdigest()[:16]
|
||
|
||
|
||
def _linea(**kw) -> dict:
|
||
base = {"ref": "", "fecha": "", "tipo": "", "isin": "", "producto": "", "clase": "Accion",
|
||
"cantidad": None, "precio": None, "importe": None, "comisiones": None, "retencion": None,
|
||
"descripcion": "", "intereses_cuenta": False, "estimado": False, "valor_recibido": None,
|
||
"exchange": "", "en_historial": False, "codigo": "", "unidad": "",
|
||
"comision_btc": None, "sentido": ""}
|
||
base.update(kw)
|
||
return base
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
# Exportación «Trade Table» con las operaciones de todos los exchanges. Solo
|
||
# interesa el bitcoin, que va al activo «Bitcoin» de siempre: cada exchange es
|
||
# un LUGAR de ese activo (ver lugares.py), no otro activo. Así una compra que
|
||
# ya tenías apuntada se reconoce (mismo activo, fecha e importe) y solo se le
|
||
# pone el lugar.
|
||
# Una retirada es un «Traslado Exchange → ?» y un depósito un «Traslado ? →
|
||
# Exchange»: el extremo que falta lo completa el historial de bitcoin cuando ve
|
||
# llegar (o salir) esa cantidad en uno de tus monederos. La comisión de la
|
||
# retirada va en el mismo traslado.
|
||
|
||
|
||
# ── Propuestas ───────────────────────────────────────────────────────────────
|
||
|
||
def _parecido(m: dict, linea: dict) -> bool:
|
||
"""¿Esta operación ya apuntada se corresponde con la línea del extracto?"""
|
||
if m["tipo"] != linea["tipo"]:
|
||
return False
|
||
if abs((dt.date.fromisoformat(m["fecha"]) - dt.date.fromisoformat(linea["fecha"])).days) > TOLERANCIA_DIAS:
|
||
return False
|
||
a, b = m.get("importe"), linea.get("importe")
|
||
if a is not None and b is not None:
|
||
return abs(a - b) <= max(TOLERANCIA_IMPORTE, abs(b) * 0.01)
|
||
a, b = m.get("cantidad"), linea.get("cantidad")
|
||
if a is not None and b is not None:
|
||
return abs(a - b) <= max(1e-8, abs(b) * 0.005)
|
||
return False
|
||
|
||
|
||
def _buscar_activo(linea: dict, activos: list[dict]) -> dict | None:
|
||
if linea["isin"]:
|
||
return next((a for a in activos if (a.get("codigo") or "").upper() == linea["isin"]), None)
|
||
nombre = sin_tildes(linea["producto"])
|
||
if not nombre:
|
||
return None
|
||
for a in activos:
|
||
candidatos = [sin_tildes(a.get("nombre", ""))] + [sin_tildes(x) for x in (a.get("alias") or "").split(",")]
|
||
if any(c and (c == nombre or c in nombre or nombre in c) for c in candidatos):
|
||
return a
|
||
return None
|
||
|
||
|
||
NO_IMPORTAR = "no"
|
||
|
||
|
||
def _lugar_exchange(exchange: str, activo: dict | None, destinos: dict) -> tuple[str, bool]:
|
||
"""(id del lugar o "" si hay que crearlo, no_importar). Lo que eligió el
|
||
usuario para ese exchange o, si no eligió nada, el lugar con su nombre."""
|
||
elegido = destinos.get(exchange)
|
||
if elegido == NO_IMPORTAR:
|
||
return "", True
|
||
if activo is not None:
|
||
if elegido and elegido in motor_lugares.por_id(activo):
|
||
return elegido, False
|
||
ya = motor_lugares.buscar_exchange(activo, exchange)
|
||
if ya:
|
||
return ya["id"], False
|
||
return "", False
|
||
|
||
|
||
COMISION_MAXIMA_APUNTADA = 0.05 # tu importe puede llevar hasta un 5 % de comisiones que el extracto no suma
|
||
|
||
|
||
def _parecido_btc(m: dict, linea: dict) -> bool:
|
||
"""Como _parecido, pero sabiendo que tú sueles apuntar el coste TOTAL, con
|
||
comisiones: en una compra tu importe puede ser algo MAYOR que el del
|
||
extracto (y en una venta, algo menor: lo que te quedó limpio). La cantidad
|
||
también puede diferir un poco si la comisión se cobró en bitcoin."""
|
||
if m["tipo"] != linea["tipo"]:
|
||
return False
|
||
if abs((dt.date.fromisoformat(m["fecha"]) - dt.date.fromisoformat(linea["fecha"])).days) > TOLERANCIA_DIAS:
|
||
return False
|
||
a, b = m.get("importe"), linea.get("importe")
|
||
if a is not None and b is not None:
|
||
dif = a - b if m["tipo"] == "Compra" else b - a # lo que suman las comisiones
|
||
if -max(TOLERANCIA_IMPORTE, abs(b) * 0.02) <= dif <= max(TOLERANCIA_IMPORTE, abs(b) * COMISION_MAXIMA_APUNTADA):
|
||
return True
|
||
a, b = m.get("cantidad"), linea.get("cantidad")
|
||
return a is not None and b is not None and abs(a - b) <= max(1e-8, abs(b) * 0.015)
|
||
|
||
|
||
def analizar(texto: str, datos: Datos, cuenta_intereses: str | None = None,
|
||
destinos: dict | None = None, mapa: dict | None = None) -> dict:
|
||
"""destinos (solo extractos con exchanges): {exchange: id del lugar | "no" | "" (lugar nuevo)}.
|
||
mapa: el reparto de columnas, cuando el formato no es de ninguna entidad conocida (ver leer_generico)."""
|
||
lector = detectar(texto)
|
||
if lector is None and not mapa:
|
||
raise ValueError("No reconozco el formato de este extracto: di qué es cada columna para leerlo.")
|
||
destinos = destinos if isinstance(destinos, dict) else {}
|
||
if lector is None:
|
||
lector = GENERICO
|
||
lineas, ignoradas = leer_generico(texto, mapa)
|
||
else:
|
||
lineas, ignoradas = LECTORES[lector](texto)
|
||
nombre_lector = NOMBRES.get(lector, "tu entidad")
|
||
total_lineas = len(lineas) + len(ignoradas)
|
||
activos = datos.vivos("activos")
|
||
nombres = {a["id"]: a.get("nombre", a["id"]) for a in activos}
|
||
movs = datos.vivos("movimientos")
|
||
por_ref = {m["extracto"]: m for m in movs if m.get("extracto")}
|
||
cuentas = [a for a in activos if a["tipo"] == "Cuenta"]
|
||
cuenta = next((a for a in cuentas if a["id"] == cuenta_intereses), None)
|
||
if cuenta is None:
|
||
# Sin elección previa, solo se propone la cuenta que lleva el nombre de la entidad
|
||
entidad = sin_tildes(nombre_lector)
|
||
cuenta = next((a for a in cuentas if entidad in sin_tildes(a.get("nombre", ""))), None)
|
||
|
||
activo_btc = motor_lugares.activo_bitcoin(activos, movs)
|
||
propuestas, nuevos, exchanges, usados = [], {}, {}, set()
|
||
for linea in lineas:
|
||
lugar_ex = ""
|
||
if linea["exchange"]:
|
||
lugar_ex, no_importar = _lugar_exchange(linea["exchange"], activo_btc, destinos)
|
||
exchanges.setdefault(linea["exchange"], {
|
||
"nombre": linea["exchange"], "lineas": 0,
|
||
"destino": NO_IMPORTAR if no_importar else lugar_ex})
|
||
exchanges[linea["exchange"]]["lineas"] += 1
|
||
if no_importar:
|
||
ignoradas.append({"fecha": linea["fecha"], "que": "no importar",
|
||
"descripcion": f"{linea['exchange']}: elegiste no importarlo"})
|
||
continue
|
||
p = {"ref": linea["ref"], "fecha": linea["fecha"], "tipo": linea["tipo"],
|
||
"descripcion": linea["descripcion"], "estado": "nueva", "movimiento": None,
|
||
"activo_nuevo": None, "nombre_activo": "", "cantidad": linea["cantidad"],
|
||
"importe": linea["importe"] if linea["importe"] is not None else linea["valor_recibido"],
|
||
"coincide_con": None, "guardada_en": None}
|
||
|
||
if linea["intereses_cuenta"]:
|
||
if cuenta is None:
|
||
p["estado"] = "falta_cuenta"
|
||
p["nombre_activo"] = f"Cuenta de {nombre_lector}"
|
||
propuestas.append(p)
|
||
continue
|
||
activo = cuenta
|
||
else:
|
||
activo = activo_btc if linea["exchange"] else _buscar_activo(linea, activos)
|
||
if activo is None:
|
||
clave = linea["isin"] or sin_tildes(linea["producto"])
|
||
nuevos.setdefault(clave, {"isin": linea["isin"], "nombre": linea["producto"],
|
||
"tipo": linea["clase"], "codigo": linea["codigo"],
|
||
"unidad": linea["unidad"]})
|
||
p["activo_nuevo"] = nuevos[clave]
|
||
p["nombre_activo"] = p["nombre_activo"] or (activo["nombre"] if activo else linea["producto"])
|
||
|
||
movimiento = {"activo": activo["id"] if activo else None, "tipo": linea["tipo"],
|
||
"fecha": linea["fecha"], "cantidad": linea["cantidad"], "precio": linea["precio"],
|
||
"importe": linea["importe"], "comisiones": linea["comisiones"],
|
||
"notas": linea["descripcion"][:120], "extracto": linea["ref"]}
|
||
if linea["tipo"] == "Dividendo":
|
||
movimiento.update({"cantidad": None, "precio": None})
|
||
if linea["estimado"]:
|
||
movimiento["estimado"] = True
|
||
if linea["valor_recibido"] is not None:
|
||
movimiento["valor_recibido"] = linea["valor_recibido"]
|
||
if linea["tipo"] in ("Mineria", "Recompensa", "Comision"):
|
||
movimiento.update({"precio": None, "importe": None})
|
||
|
||
if linea["exchange"]:
|
||
# El exchange es un lugar del activo; si aún no existe, se crea al guardar
|
||
lado = "lugar_destino" if linea["sentido"] == "deposito" else "lugar"
|
||
movimiento[lado] = lugar_ex
|
||
if not lugar_ex:
|
||
movimiento["lugar_nuevo"] = {"nombre": linea["exchange"], "clase": "exchange", "en": lado}
|
||
p["lugar"] = linea["exchange"]
|
||
if linea["tipo"] == "Traslado":
|
||
movimiento.update({"precio": None, "importe": None, "comision": linea["comision_btc"],
|
||
"lugar": movimiento.get("lugar", ""),
|
||
"lugar_destino": movimiento.get("lugar_destino", "")})
|
||
p["desde"] = linea["exchange"] if linea["sentido"] == "retirada" else "?"
|
||
p["hacia"] = linea["exchange"] if linea["sentido"] == "deposito" else "?"
|
||
if linea["retencion"]:
|
||
movimiento["retencion"] = linea["retencion"]
|
||
p["movimiento"] = movimiento
|
||
|
||
# Algunos extractos no traen participaciones (p. ej. algunos fondos)
|
||
if linea["tipo"] in ("Compra", "Venta") and linea["exchange"] and not linea["importe"]:
|
||
p["estado"] = "sin_precio" # día sin precio de referencia: apúntala a mano
|
||
propuestas.append(p)
|
||
continue
|
||
if linea["tipo"] != "Dividendo" and not linea["cantidad"] and not linea["precio"]:
|
||
p["estado"] = "incompleta"
|
||
propuestas.append(p)
|
||
continue
|
||
|
||
if linea["ref"] in por_ref:
|
||
anterior = por_ref[linea["ref"]]
|
||
p["estado"] = "ya_estaba"
|
||
p["coincide_con"] = anterior["id"]
|
||
p["guardada_en"] = nombres.get(anterior["activo"], anterior["activo"])
|
||
if linea["intereses_cuenta"] and anterior["activo"] != activo["id"]:
|
||
p["estado"] = "mover" # se guardaron en otra cuenta
|
||
p["movimiento"]["id"] = anterior["id"]
|
||
elif activo is not None:
|
||
parecido = _parecido_btc if linea["exchange"] else _parecido
|
||
iguales = [m for m in movs if m["activo"] == activo["id"] and parecido(m, linea)
|
||
and m["id"] not in usados]
|
||
if iguales:
|
||
m = iguales[0]
|
||
usados.add(m["id"])
|
||
p["coincide_con"] = m["id"]
|
||
if linea["exchange"] and m.get("importe") is not None and linea["importe"] is not None:
|
||
# Lo tuyo no se toca; solo se enseña la diferencia (tus comisiones, casi siempre)
|
||
p["importe_tuyo"] = m["importe"]
|
||
p["dif_importe"] = round(m["importe"] - linea["importe"], 2)
|
||
p["estado"] = "afinar" if (m.get("estimado") or m.get("pendiente")) else "ya_estaba"
|
||
p["guardada_en"] = nombres.get(m["activo"], m["activo"])
|
||
if p["estado"] == "afinar":
|
||
p["movimiento"]["id"] = m["id"]
|
||
elif linea["exchange"] and not m.get("lugar") and m["tipo"] != "Traslado":
|
||
# Ya la tenías apuntada (a mano): solo se le pone el lugar
|
||
p["estado"] = "poner_lugar"
|
||
propuestas.append(p)
|
||
|
||
resumen = {
|
||
"lector": NOMBRES.get(lector, "CSV con tus columnas"), "lineas": total_lineas,
|
||
"nuevas": sum(1 for p in propuestas if p["estado"] == "nueva"),
|
||
"ya_estaban": sum(1 for p in propuestas if p["estado"] == "ya_estaba"),
|
||
"afinar": sum(1 for p in propuestas if p["estado"] == "afinar"),
|
||
"mover": sum(1 for p in propuestas if p["estado"] == "mover"),
|
||
"falta_cuenta": sum(1 for p in propuestas if p["estado"] == "falta_cuenta"),
|
||
"incompletas": sum(1 for p in propuestas if p["estado"] == "incompleta"),
|
||
"ignoradas": len(ignoradas),
|
||
"poner_lugar": sum(1 for p in propuestas if p["estado"] == "poner_lugar"),
|
||
"traslados": sum(1 for p in propuestas if p["tipo"] == "Traslado" and p["estado"] == "nueva"),
|
||
"sin_precio": sum(1 for p in propuestas if p["estado"] == "sin_precio"),
|
||
"estimadas": sum(1 for p in propuestas if (p["movimiento"] or {}).get("estimado")
|
||
and p["estado"] == "nueva"),
|
||
}
|
||
return {"resumen": resumen, "propuestas": propuestas, "nuevos_activos": list(nuevos.values()),
|
||
"lector": lector, "cuenta": cuenta["id"] if cuenta else None,
|
||
"hay_intereses": any(l["intereses_cuenta"] for l in lineas),
|
||
"cuentas": [{"id": a["id"], "nombre": a["nombre"]} for a in cuentas],
|
||
"ignoradas": ignoradas[:20],
|
||
"exchanges": sorted(exchanges.values(), key=lambda e: e["nombre"].lower()),
|
||
"activo_btc": {"id": activo_btc["id"], "nombre": activo_btc["nombre"],
|
||
"lugares": motor_lugares.de(activo_btc)} if activo_btc else None}
|
||
|
||
|
||
# Los formatos de entidades concretas, si esta instalación los trae
|
||
try:
|
||
from .extras import extractos as _extras # noqa: E402,F401
|
||
except ImportError:
|
||
pass
|