""" 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