esetheyers
4b158e662d
Link another Caudal (phone, USB) with a code or a QR instead of the password, choosing the server's root certificate as a file; read Bitcoin straight from Bitcoin Core or Knots, without an Electrum server; Android app (beta) and a Windows build script. Fixed: Save did nothing for expenses, refunds and dividends entered by hand.
1363 lines
71 KiB
Python
1363 lines
71 KiB
Python
"""
|
|
operaciones.py — Todo lo que cambia los datos: altas, ediciones, borrados,
|
|
traspasos, cierres, precios manuales y operaciones automáticas.
|
|
|
|
Cartera abre una carpeta de datos con un cerrojo (no puede haber dos
|
|
aplicaciones escribiendo a la vez) y guarda en disco tras cada cambio.
|
|
Los errores de validación se lanzan como ErrorOperacion con un mensaje
|
|
pensado para mostrárselo al usuario tal cual.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import calendar
|
|
import copy
|
|
import datetime as dt
|
|
import threading
|
|
import uuid
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
|
|
from . import adjuntos
|
|
from . import inmuebles as motor_inmuebles
|
|
from . import rentafija as motor_rentafija
|
|
from . import tierra as motor_tierra
|
|
from . import lugares as motor_lugares
|
|
|
|
from .calculo import ENTRADAS, SALIDAS, por_valor
|
|
from . import deuda as motor_deuda
|
|
from .datos import (MONEDAS, TIPOS_ACTIVO, TIPOS_MOVIMIENTO, Datos, ErrorDatos,
|
|
escribir, leer, pendiente_de_migrar)
|
|
from .vista import cantidad_en
|
|
|
|
try:
|
|
import fcntl
|
|
except ImportError: # Windows
|
|
fcntl = None
|
|
try:
|
|
import msvcrt
|
|
except ImportError: # todo lo que no es Windows
|
|
msvcrt = None
|
|
|
|
|
|
def _echar_cerrojo(f) -> bool:
|
|
"""Reserva el fichero para este proceso. OSError si ya lo tiene otro;
|
|
False si el sistema no ofrece con qué (se abre igual, sin esa protección)."""
|
|
if fcntl is not None:
|
|
fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
return True
|
|
if msvcrt is not None:
|
|
f.seek(0)
|
|
msvcrt.locking(f.fileno(), msvcrt.LK_NBLCK, 1) # el primer byte, aunque el fichero esté vacío
|
|
return True
|
|
return False
|
|
|
|
PREFIJOS = {"Fondo": "F", "PlanPensiones": "P", "Cripto": "C", "Metal": "M", "ETF": "E",
|
|
"Accion": "A", "CuentaTrading": "CT", "Prestamo": "PR", "Tierra": "T", "Inmueble": "IN", "RentaFija": "RF",
|
|
"Cuenta": "CC", "Deuda": "D"}
|
|
NECESITAN_CANTIDAD = ENTRADAS | SALIDAS | {"Comision", "Mineria"}
|
|
DIAS_MAX_VL_ESTIMADO = 5
|
|
|
|
|
|
class ErrorOperacion(ValueError):
|
|
pass
|
|
|
|
|
|
class CarteraOcupada(RuntimeError):
|
|
pass
|
|
|
|
|
|
def ahora() -> str:
|
|
return dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
|
|
|
|
|
|
def _fecha(v, campo: str = "fecha") -> dt.date:
|
|
try:
|
|
return dt.date.fromisoformat(str(v))
|
|
except ValueError:
|
|
raise ErrorOperacion(f"La {campo} no es válida.") from None
|
|
|
|
|
|
def _numero(v, campo: str, positivo: bool = False, opcional: bool = True):
|
|
if v is None or v == "":
|
|
if opcional:
|
|
return None
|
|
raise ErrorOperacion(f"Falta {campo}.")
|
|
if isinstance(v, bool):
|
|
raise ErrorOperacion(f"{campo} no es un número.")
|
|
try:
|
|
n = float(str(v).replace(",", ".")) if isinstance(v, str) else float(v)
|
|
except ValueError:
|
|
raise ErrorOperacion(f"{campo} no es un número.") from None
|
|
if n != n or n in (float("inf"), float("-inf")):
|
|
raise ErrorOperacion(f"{campo} no es un número.")
|
|
if positivo and n <= 0:
|
|
raise ErrorOperacion(f"{campo} debe ser mayor que cero.")
|
|
if n < 0:
|
|
raise ErrorOperacion(f"{campo} no puede ser negativo.")
|
|
return n
|
|
|
|
|
|
def _f_num(v) -> float:
|
|
try:
|
|
return float(v or 0)
|
|
except (TypeError, ValueError):
|
|
return 0.0
|
|
|
|
|
|
def _texto(v, maximo: int = 200) -> str:
|
|
return str(v or "").strip()[:maximo]
|
|
|
|
|
|
class Cartera:
|
|
def __init__(self, carpeta: Path):
|
|
self.carpeta = Path(carpeta)
|
|
self.carpeta.mkdir(parents=True, exist_ok=True)
|
|
self._lock = threading.RLock()
|
|
self._diferir = False # dentro de _bloque(): se guarda al final
|
|
self._cerrojo = None
|
|
if fcntl is not None or msvcrt is not None:
|
|
self._cerrojo = open(self.carpeta / ".cerrojo", "a", encoding="utf-8") # «a»: abrirlo no lo vacía ni molesta a quien lo tenga
|
|
try:
|
|
_echar_cerrojo(self._cerrojo)
|
|
except OSError:
|
|
self._cerrojo.close()
|
|
self._cerrojo = None
|
|
raise CarteraOcupada("La cartera ya está abierta en otra ventana o programa.") from None
|
|
try:
|
|
self.datos = leer(self.carpeta)
|
|
if pendiente_de_migrar(self.carpeta):
|
|
escribir(self.carpeta, self.datos) # formato al día (antes guarda una copia_vN)
|
|
except BaseException:
|
|
self.cerrar() # que un error al leer no deje la cartera bloqueada
|
|
raise
|
|
|
|
def reemplazar(self, datos: Datos) -> None:
|
|
"""Sustituye TODO por esos datos (restaurar una copia de seguridad).
|
|
Se validan al escribirlos; si no valen, se queda lo que había."""
|
|
with self._lock:
|
|
antes = self.datos
|
|
self.datos = datos
|
|
try:
|
|
escribir(self.carpeta, self.datos)
|
|
except ErrorDatos as e:
|
|
self.datos = antes
|
|
raise ErrorOperacion(f"No se ha restaurado: {e}") from None
|
|
|
|
def cerrar(self) -> None:
|
|
if self._cerrojo:
|
|
self._cerrojo.close()
|
|
self._cerrojo = None
|
|
|
|
# ── utilidades ──
|
|
@contextmanager
|
|
def _bloque(self):
|
|
"""Varios cambios que se guardan juntos o no se guarda ninguno: si algo
|
|
falla a medias, los datos en memoria vuelven a como estaban."""
|
|
if self._diferir: # dentro de otro bloque: guarda el de fuera
|
|
yield
|
|
return
|
|
copia = copy.deepcopy(self.datos)
|
|
self._diferir = True
|
|
try:
|
|
yield
|
|
except BaseException:
|
|
self.datos = copia
|
|
raise
|
|
finally:
|
|
self._diferir = False
|
|
self._guardar()
|
|
|
|
def _guardar(self) -> None:
|
|
if self._diferir:
|
|
return
|
|
self._revalorar_inmuebles()
|
|
try:
|
|
escribir(self.carpeta, self.datos)
|
|
except ErrorDatos as e:
|
|
# Nunca debería pasar (se valida antes); se recarga lo que hay en disco
|
|
self.datos = leer(self.carpeta)
|
|
raise ErrorOperacion(f"No se ha guardado: {e}") from None
|
|
|
|
def _revalorar_inmuebles(self, hoy: dt.date | None = None) -> None:
|
|
"""Pone al día los valores que la cartera apunta sola en cada inmueble: el
|
|
de la compra (fuente «compra», para que valga algo desde el primer día) y
|
|
los estimados con el índice del INE (fuente «IPV») si lo tiene activado, o
|
|
en las fincas rústicas con el precio de la tierra del MAPA (fuente «MAPA»).
|
|
Quita los que sobran; los valores apuntados a mano no se tocan."""
|
|
hoy = hoy or dt.date.today()
|
|
gestionadas = (motor_inmuebles.FUENTE, motor_inmuebles.FUENTE_COMPRA, motor_rentafija.FUENTE, motor_tierra.FUENTE)
|
|
activos = [a for a in self.datos.vivos("activos") if a["tipo"] in ("Inmueble", "RentaFija")]
|
|
if not activos and not any(r.get("fuente") in gestionadas for r in self.datos.vivos("precios")):
|
|
return
|
|
serie = motor_inmuebles.serie(self.carpeta.parent)
|
|
movs = self.datos.vivos("movimientos")
|
|
quieren: dict[tuple, tuple] = {}
|
|
for a in activos:
|
|
suyos = [m for m in movs if m["activo"] == a["id"]]
|
|
precios = [r for r in self.datos.vivos("precios") if r["activo"] == a["id"]]
|
|
manuales = {r["fecha"] for r in precios if r.get("fuente") not in gestionadas}
|
|
if a["tipo"] == "RentaFija":
|
|
# Sus cupones y la devolución al vencer, según la ficha (id fijo: lo borrado no vuelve)
|
|
if a.get("estado") != "cerrada":
|
|
existentes = {m["id"] for m in self.datos.movimientos}
|
|
for nuevo in motor_rentafija.calendario(a, suyos, hoy):
|
|
if nuevo["id"] not in existentes:
|
|
nuevo.update({"orden": self._siguiente_orden("movimientos"), "modificado": ahora()})
|
|
self.datos.movimientos.append(nuevo)
|
|
suyos.append(nuevo)
|
|
# Depósitos, Letras y bonos: el valor de cada fin de mes, sin apuntarlo
|
|
for e in motor_rentafija.estimaciones(a, suyos, hoy):
|
|
if e["fecha"] not in manuales:
|
|
quieren[(a["id"], e["fecha"])] = (e["valor_total"], motor_rentafija.FUENTE)
|
|
continue
|
|
if motor_tierra.es_rustico(a):
|
|
# Fincas rústicas: con el precio de la tierra de su cultivo, cada año
|
|
inicio = motor_tierra.punto_de_partida(a, suyos)
|
|
if inicio and inicio[0] not in manuales:
|
|
quieren[(a["id"], inicio[0])] = (inicio[1], motor_inmuebles.FUENTE_COMPRA)
|
|
for e in motor_tierra.estimaciones(a, suyos, precios, hoy):
|
|
if e["fecha"] not in manuales:
|
|
quieren[(a["id"], e["fecha"])] = (e["valor_total"], motor_tierra.FUENTE)
|
|
continue
|
|
inicio = motor_inmuebles.punto_de_partida(a, suyos)
|
|
if inicio and inicio[0] not in manuales:
|
|
quieren[(a["id"], inicio[0])] = (inicio[1], motor_inmuebles.FUENTE_COMPRA)
|
|
for e in motor_inmuebles.estimaciones(a, suyos, precios, serie, hoy):
|
|
if e["fecha"] not in manuales:
|
|
quieren[(a["id"], e["fecha"])] = (e["valor_total"], motor_inmuebles.FUENTE)
|
|
for r in self.datos.vivos("precios"):
|
|
if r.get("fuente") not in gestionadas:
|
|
continue
|
|
clave = (r["activo"], r["fecha"])
|
|
if clave not in quieren:
|
|
r.update({"borrado": True, "modificado": ahora()})
|
|
elif (r.get("valor_total"), r.get("fuente")) == quieren[clave]:
|
|
del quieren[clave]
|
|
for (aid, fecha), (valor, fuente) in quieren.items():
|
|
self._poner_precio(aid, dt.date.fromisoformat(fecha), {"valor_total": valor}, fuente, guardar=False)
|
|
|
|
def activo(self, aid: str) -> dict:
|
|
for a in self.datos.vivos("activos"):
|
|
if a["id"] == aid:
|
|
return a
|
|
raise ErrorOperacion("Ese activo no existe.")
|
|
|
|
def _movimiento(self, mid: str) -> dict:
|
|
for m in self.datos.vivos("movimientos"):
|
|
if m["id"] == mid:
|
|
return m
|
|
raise ErrorOperacion("Esa operación no existe.")
|
|
|
|
def _siguiente_orden(self, nombre: str) -> int:
|
|
return max((r.get("orden") or 0 for r in getattr(self.datos, nombre)), default=0) + 1
|
|
|
|
# ── activos ──
|
|
def _ficha_rustica(self, ficha: dict, f: dict, aid: str | None) -> None:
|
|
"""Superficie, cultivo y de dónde parte el valor de una finca rústica. El
|
|
cultivo guarda su historia: si cambia, se apunta desde la fecha que diga
|
|
«cultivo_desde» (hoy si no la dice); el mismo día, sustituye al anterior."""
|
|
ha = _numero(f.get("superficie"), "la superficie", positivo=True)
|
|
if ha is not None:
|
|
ficha["superficie"] = ha
|
|
ficha["origen"] = f.get("origen") if f.get("origen") in motor_tierra.ORIGENES else "compra"
|
|
previa = []
|
|
if aid:
|
|
viejo = self.activo(aid)
|
|
previa = list(((viejo.get("inmueble") or {}).get("cultivos")) or []) if viejo["tipo"] == "Inmueble" else []
|
|
cultivo = f.get("cultivo")
|
|
if cultivo and cultivo not in motor_tierra.cultivos():
|
|
raise ErrorOperacion("Elige el cultivo de la lista.")
|
|
if cultivo and (not previa or previa[-1].get("cultivo") != cultivo):
|
|
desde = _fecha(f["cultivo_desde"], "fecha del cambio de cultivo").isoformat() if previa and f.get("cultivo_desde") else ""
|
|
if previa and not desde:
|
|
desde = dt.date.today().isoformat()
|
|
previa = [x for x in previa if (x.get("desde") or "") < desde] if desde else []
|
|
previa.append({"desde": desde, "cultivo": cultivo})
|
|
if previa:
|
|
ficha["cultivos"] = previa
|
|
if ficha["auto"] and not ficha.get("cultivos"):
|
|
raise ErrorOperacion("Para estimar el valor de una finca hace falta su cultivo.")
|
|
if ficha["origen"] == "media" and not ficha.get("superficie"):
|
|
raise ErrorOperacion("Para partir de la media hace falta la superficie.")
|
|
|
|
def guardar_activo(self, c: dict) -> dict:
|
|
with self._lock:
|
|
tipo = c.get("tipo")
|
|
if tipo not in TIPOS_ACTIVO:
|
|
raise ErrorOperacion("Elige un tipo de activo.")
|
|
nombre = _texto(c.get("nombre"), 120)
|
|
if not nombre:
|
|
raise ErrorOperacion("Pon un nombre al activo.")
|
|
campos = {
|
|
"nombre": nombre, "tipo": tipo, "codigo": _texto(c.get("codigo"), 60),
|
|
"unidad": _texto(c.get("unidad"), 30) or "participaciones",
|
|
"visible": bool(c.get("visible", True)),
|
|
"alias": _texto(c.get("alias"), 120),
|
|
"valoracion": "total" if c.get("valoracion") == "total" else "unitaria",
|
|
}
|
|
if c.get("fecha_inicio"):
|
|
campos["fecha_inicio"] = _fecha(c["fecha_inicio"], "fecha de inicio").isoformat()
|
|
if tipo == "Prestamo":
|
|
campos["prestamo"] = {
|
|
"importe_final": _numero(c.get("importe_final"), "el importe a cobrar", positivo=True, opcional=False),
|
|
"vencimiento": _fecha(c.get("vencimiento"), "fecha de vencimiento").isoformat(),
|
|
}
|
|
if not campos.get("fecha_inicio"):
|
|
raise ErrorOperacion("Un préstamo necesita fecha de inicio.")
|
|
if tipo == "Inmueble":
|
|
f = c.get("inmueble") or {}
|
|
clase = f.get("clase") if f.get("clase") in motor_inmuebles.CLASES else "vivienda"
|
|
uso = f.get("uso") if f.get("uso") in motor_inmuebles.USOS else "alquilado"
|
|
ficha = {"clase": clase, "uso": uso,
|
|
"comunidad": f.get("comunidad") if f.get("comunidad") in motor_inmuebles.comunidades() else "nacional",
|
|
"auto": bool(f.get("auto")), "protegida": bool(f.get("protegida")),
|
|
"referencia": _texto(f.get("referencia"), 120),
|
|
"catastral_revisado": bool(f.get("catastral_revisado"))}
|
|
if f.get("contrato") in ("habitual", "tensionado", "programa", "programa_tensionado", "despoblacion", "temporada"):
|
|
ficha["contrato"] = f["contrato"]
|
|
reduccion = _numero(f.get("reduccion"), "la reducción")
|
|
if reduccion is not None:
|
|
if not 0 <= reduccion <= 100:
|
|
raise ErrorOperacion("La reducción va de 0 a 100 %.")
|
|
ficha["reduccion"] = reduccion
|
|
for k, nombre in (("precio_compra", "el precio de compra"), ("valor_catastral", "el valor catastral"),
|
|
("porcentaje_construccion", "el porcentaje de construcción"),
|
|
("precio_maximo", "el precio máximo de venta")):
|
|
v = _numero(f.get(k), nombre, positivo=True)
|
|
if v is not None:
|
|
ficha[k] = v
|
|
if f.get("hipoteca"):
|
|
deuda = next((d for d in self.datos.vivos("activos") if d["id"] == f["hipoteca"]), None)
|
|
if deuda is None or deuda["tipo"] != "Deuda":
|
|
raise ErrorOperacion("Esa hipoteca no está entre tus préstamos recibidos.")
|
|
ficha["hipoteca"] = deuda["id"]
|
|
if ficha.get("porcentaje_construccion", 0) > 100:
|
|
raise ErrorOperacion("El porcentaje de construcción va de 0 a 100.")
|
|
if ficha["protegida"] and not ficha.get("precio_maximo"):
|
|
raise ErrorOperacion("Una vivienda protegida necesita su precio máximo de venta.")
|
|
renta = _numero(f.get("renta_mensual"), "la renta mensual", positivo=True)
|
|
if renta:
|
|
try:
|
|
dia = int(f.get("dia_cobro") or 1)
|
|
except (TypeError, ValueError):
|
|
dia = 0
|
|
if not 1 <= dia <= 28:
|
|
raise ErrorOperacion("El día de cobro va del 1 al 28.")
|
|
ficha.update({"renta_mensual": renta, "dia_cobro": dia,
|
|
"renta_desde": _fecha(f.get("renta_desde") or dt.date.today().isoformat(), "fecha desde la que cobras").isoformat()})
|
|
if clase == "terreno_rustico":
|
|
self._ficha_rustica(ficha, f, c.get("id"))
|
|
campos.update({"inmueble": ficha, "valoracion": "total", "codigo": "", "unidad": "inmueble"})
|
|
if tipo == "RentaFija":
|
|
f = c.get("renta_fija") or {}
|
|
if f.get("clase") not in motor_rentafija.CLASES:
|
|
raise ErrorOperacion("Elige si es un depósito, una Letra o un bono.")
|
|
try:
|
|
pagos = int(f.get("pagos") or 0)
|
|
except (TypeError, ValueError):
|
|
pagos = -1
|
|
if pagos not in motor_rentafija.PAGOS:
|
|
raise ErrorOperacion("Elige cada cuánto paga intereses.")
|
|
ficha = {"clase": f["clase"], "pagos": 0 if f["clase"] == "letra" else pagos,
|
|
"vencimiento": _fecha(f.get("vencimiento"), "fecha de vencimiento").isoformat(),
|
|
"emisor": _texto(f.get("emisor"), 120)}
|
|
for k, nombre in (("nominal", "el nominal"), ("interes", "el tipo de interés")):
|
|
v = _numero(f.get(k), nombre, positivo=True)
|
|
if v is not None:
|
|
ficha[k] = v
|
|
if ficha.get("interes", 0) > 30:
|
|
raise ErrorOperacion("El tipo de interés va de 0 a 30 %.")
|
|
campos.update({"renta_fija": ficha, "valoracion": "total", "unidad": "€ nominal"})
|
|
if tipo == "Deuda":
|
|
campos.update({"unidad": "€", "codigo": "", "valoracion": "unitaria"})
|
|
campos.setdefault("fecha_inicio", dt.date.today().isoformat())
|
|
cuotas = _numero(c.get("cuotas"), "el número de cuotas que quedan", positivo=True, opcional=False)
|
|
if cuotas != int(cuotas):
|
|
raise ErrorOperacion("Las cuotas que quedan deben ser un número entero.")
|
|
primera = _fecha(c.get("primera_cuota"), "fecha de la próxima cuota")
|
|
if primera <= _fecha(campos["fecha_inicio"]):
|
|
raise ErrorOperacion("La próxima cuota debe ser posterior a la fecha de los datos.")
|
|
campos["deuda"] = {
|
|
"capital": _numero(c.get("capital"), "el capital pendiente", positivo=True, opcional=False),
|
|
"tin": _numero(c.get("tin"), "el tipo de interés (TIN)", opcional=False),
|
|
"cuotas": int(cuotas),
|
|
"primera_cuota": primera.isoformat(),
|
|
}
|
|
if campos["deuda"]["tin"] < 0 or campos["deuda"]["tin"] > 30:
|
|
raise ErrorOperacion("El TIN debe estar entre 0 y 30 %.")
|
|
if c.get("id"):
|
|
a = self.activo(c["id"])
|
|
if a["tipo"] != tipo and any(m["activo"] == a["id"] for m in self.datos.vivos("movimientos")):
|
|
raise ErrorOperacion("No se puede cambiar el tipo de un activo que ya tiene operaciones.")
|
|
a.update(campos)
|
|
a["modificado"] = ahora()
|
|
else:
|
|
prefijo = PREFIJOS[tipo]
|
|
usados = {a["id"] for a in self.datos.activos}
|
|
n = 1
|
|
while f"{prefijo}{n}" in usados:
|
|
n += 1
|
|
# «uid»: lo que distingue a esta inversión de otra que, en otra copia de los
|
|
# datos, haya recibido el mismo identificador correlativo (ver datos.choques_de_activos)
|
|
a = {"id": f"{prefijo}{n}", "uid": uuid.uuid4().hex[:12], **campos, "estado": "abierta",
|
|
"orden": self._siguiente_orden("activos"), "modificado": ahora()}
|
|
self.datos.activos.append(a)
|
|
if tipo == "Inmueble":
|
|
self._cobro_del_alquiler(a)
|
|
self._guardar()
|
|
return a
|
|
|
|
def _cobro_del_alquiler(self, a: dict) -> None:
|
|
"""La renta mensual de un inmueble alquilado es una operación automática
|
|
de cobro; si se quita la renta o deja de estar alquilado, se para (lo ya
|
|
cobrado se queda)."""
|
|
f = a.get("inmueble") or {}
|
|
rid = f"alquiler-{a['id']}"
|
|
r = next((x for x in self.datos.recurrentes if x["id"] == rid), None)
|
|
if f.get("renta_mensual") and f.get("uso") == "alquilado":
|
|
campos = {"activo": a["id"], "tipo": "Dividendo", "moneda": "EUR", "dia": f["dia_cobro"], "importe": f["renta_mensual"],
|
|
"desde": f["renta_desde"], "hasta": None, "pausado": False, "notas": "Alquiler (automático)", "actividad": ""}
|
|
if r is None:
|
|
self.datos.recurrentes.append({"id": rid, **campos, "orden": self._siguiente_orden("recurrentes"), "modificado": ahora()})
|
|
elif any(r.get(k) != v for k, v in campos.items()) or r.get("borrado"):
|
|
r.pop("borrado", None)
|
|
r.update({**campos, "modificado": ahora()})
|
|
elif r is not None and not r.get("borrado") and not r.get("hasta"):
|
|
r.update({"hasta": dt.date.today().isoformat(), "modificado": ahora()})
|
|
|
|
def alta_con_operacion(self, activo: dict, movimiento: dict, precio: dict | None = None) -> dict:
|
|
"""Da de alta un activo y apunta su primera operación, todo o nada."""
|
|
with self._lock:
|
|
a = self.guardar_activo({k: activo.get(k) for k in ("nombre", "tipo", "codigo", "unidad", "alias")}
|
|
| {"fecha_inicio": movimiento.get("fecha")})
|
|
try:
|
|
m = self.guardar_movimiento({**movimiento, "activo": a["id"]})
|
|
except ErrorOperacion:
|
|
a.update({"borrado": True, "modificado": ahora()})
|
|
self._guardar()
|
|
raise
|
|
if precio and precio.get("precio") and precio.get("fecha"):
|
|
try:
|
|
self._poner_precio(a["id"], _fecha(precio["fecha"]), {"precio": float(precio["precio"])},
|
|
precio.get("fuente") or "busqueda")
|
|
except (ErrorOperacion, TypeError, ValueError):
|
|
pass
|
|
return {"activo": a, "movimiento": m}
|
|
|
|
def cerrar_activo(self, aid: str, fecha, importe) -> dict:
|
|
"""Da la inversión por terminada. Si aún quedan participaciones, registra
|
|
la venta (o el cobro, en préstamos) de todo lo que queda por el importe
|
|
indicado. La inversión sigue contando en la rentabilidad global."""
|
|
with self._lock:
|
|
a = self.activo(aid)
|
|
if a.get("estado") == "cerrada":
|
|
raise ErrorOperacion("Esa inversión ya está cerrada.")
|
|
f = _fecha(fecha, "fecha de cierre")
|
|
if a["tipo"] == "Deuda":
|
|
e = self.estado_deuda(aid, f)
|
|
if not e["cancelado"]:
|
|
raise ErrorOperacion(f"Aún quedan {e['pendiente']:.2f} € por devolver: apunta la "
|
|
"amortización que lo cancela.")
|
|
a.update({"estado": "cerrada", "fecha_cierre": f.isoformat(), "modificado": ahora()})
|
|
self._guardar()
|
|
return a
|
|
importe = _numero(importe, "el importe recibido", opcional=True) or 0.0
|
|
movs = self.datos.vivos("movimientos")
|
|
if any(m["activo"] == aid and _fecha(m["fecha"]) > f for m in movs):
|
|
raise ErrorOperacion("Hay operaciones posteriores a esa fecha de cierre.")
|
|
q = cantidad_en(movs, aid, f)
|
|
if abs(q) <= 1e-6: # restos de redondeo
|
|
q = 0.0
|
|
if q < 0:
|
|
raise ErrorOperacion("La cantidad de este activo es negativa: revisa sus operaciones antes de cerrarlo.")
|
|
if q > 0:
|
|
self.datos.movimientos.append({
|
|
"id": "m" + uuid.uuid4().hex[:15], "fecha": f.isoformat(), "activo": aid,
|
|
"tipo": "Reembolso" if a["tipo"] == "Prestamo" else "Venta",
|
|
"cantidad": q, "precio": importe / q, "importe": importe,
|
|
"notas": "Cierre de la inversión", "orden": self._siguiente_orden("movimientos"),
|
|
"modificado": ahora()})
|
|
elif importe > 0:
|
|
raise ErrorOperacion("Ya no quedan participaciones: el dinero recibido debería estar "
|
|
"apuntado como venta. Cierra con importe 0 o añade la venta.")
|
|
a.update({"estado": "cerrada", "fecha_cierre": f.isoformat(), "modificado": ahora()})
|
|
self._guardar()
|
|
return a
|
|
|
|
# ── préstamos recibidos ──
|
|
def _datos_deuda(self, aid: str) -> tuple[dict, list[dict], list[dict]]:
|
|
a = self.activo(aid)
|
|
if a["tipo"] != "Deuda":
|
|
raise ErrorOperacion("Eso no es un préstamo recibido.")
|
|
movs = [m for m in self.datos.vivos("movimientos") if m["activo"] == aid]
|
|
saldos = [r for r in self.datos.vivos("precios") if r["activo"] == aid]
|
|
return a, movs, saldos
|
|
|
|
def estado_deuda(self, aid: str, hoy: dt.date | None = None) -> dict:
|
|
a, movs, saldos = self._datos_deuda(aid)
|
|
return motor_deuda.estado(a, movs, saldos, hoy or dt.date.today())
|
|
|
|
def simular_amortizacion(self, aid: str, fecha, importe) -> dict:
|
|
a, movs, saldos = self._datos_deuda(aid)
|
|
f = _fecha(fecha)
|
|
importe = _numero(importe, "el importe", positivo=True, opcional=False)
|
|
pendiente = motor_deuda.estado(a, movs, saldos, f)["pendiente"]
|
|
if importe > pendiente + 0.005:
|
|
raise ErrorOperacion(f"En esa fecha solo quedan {pendiente:.2f} € por devolver.")
|
|
return motor_deuda.simular(a, movs, saldos, f, importe)
|
|
|
|
def reabrir_activo(self, aid: str) -> dict:
|
|
with self._lock:
|
|
a = self.activo(aid)
|
|
a.update({"estado": "abierta", "fecha_cierre": None, "modificado": ahora()})
|
|
self._guardar()
|
|
return a
|
|
|
|
# ── movimientos ──
|
|
def guardar_movimiento(self, c: dict) -> dict:
|
|
with self._lock:
|
|
tipo = c.get("tipo")
|
|
if tipo not in TIPOS_MOVIMIENTO:
|
|
raise ErrorOperacion("Elige el tipo de operación.")
|
|
if tipo in ("Traspaso_Entrada", "Traspaso_Salida") and not c.get("id") and not c.get("extracto"):
|
|
raise ErrorOperacion("Los traspasos se registran con la opción «Traspaso».")
|
|
if tipo == "Traslado":
|
|
return self.guardar_traslado(c)
|
|
a = self.activo(c.get("activo"))
|
|
fecha = _fecha(c.get("fecha"))
|
|
if fecha > dt.date.today() + dt.timedelta(days=1):
|
|
raise ErrorOperacion("La fecha está en el futuro.")
|
|
if (a["tipo"] == "Deuda") != (tipo == "Amortizacion"):
|
|
raise ErrorOperacion("En un préstamo recibido solo se apuntan amortizaciones anticipadas "
|
|
"(las cuotas se calculan solas)." if a["tipo"] == "Deuda" else
|
|
"Las amortizaciones son solo para préstamos recibidos.")
|
|
cantidad = _numero(c.get("cantidad"), "La cantidad")
|
|
precio = _numero(c.get("precio"), "El precio")
|
|
importe = _numero(c.get("importe"), "El importe")
|
|
|
|
if por_valor(a) and tipo in ("Compra", "Venta", "Reembolso"):
|
|
# Solo se sigue el valor: basta con el dinero que entra o sale
|
|
if not importe:
|
|
raise ErrorOperacion("Indica el importe en euros.")
|
|
# En renta fija, la cantidad es el nominal (opcional): permite vender solo una parte
|
|
cantidad = cantidad if a["tipo"] == "RentaFija" and cantidad and cantidad > 0 else None
|
|
precio = None
|
|
elif tipo == "Gasto":
|
|
if not importe:
|
|
raise ErrorOperacion("Indica el importe del gasto.")
|
|
cantidad = precio = None
|
|
elif tipo == "Devolucion":
|
|
if not importe:
|
|
raise ErrorOperacion("Indica el importe que te han devuelto.")
|
|
cantidad = precio = None
|
|
elif tipo == "Amortizacion":
|
|
if not importe:
|
|
raise ErrorOperacion("Indica el importe que amortizas.")
|
|
if c.get("modalidad") not in motor_deuda.MODALIDADES:
|
|
raise ErrorOperacion("Elige si reduces el plazo o la cuota.")
|
|
if fecha < _fecha(a["fecha_inicio"]):
|
|
raise ErrorOperacion("La fecha es anterior a los datos del préstamo.")
|
|
otros = [m for m in self.datos.vivos("movimientos") if m["activo"] == a["id"] and m["id"] != c.get("id")]
|
|
saldos = [r for r in self.datos.vivos("precios") if r["activo"] == a["id"]]
|
|
pendiente = motor_deuda.estado(a, otros, saldos, fecha)["pendiente"]
|
|
if importe > pendiente + 0.005:
|
|
raise ErrorOperacion(f"En esa fecha solo quedaban {pendiente:.2f} € por devolver.")
|
|
cantidad = precio = None
|
|
elif tipo == "Dividendo":
|
|
if not importe:
|
|
raise ErrorOperacion("Indica el importe cobrado (neto, lo que te ingresaron).")
|
|
cantidad = precio = None
|
|
elif tipo in ("Mineria", "Recompensa"):
|
|
if not cantidad:
|
|
raise ErrorOperacion("Indica la cantidad recibida.")
|
|
precio = importe = None
|
|
else:
|
|
# Con dos de los tres datos se calcula el tercero
|
|
if cantidad and precio and importe is None:
|
|
importe = cantidad * precio
|
|
elif cantidad and importe is not None and precio is None:
|
|
precio = importe / cantidad
|
|
elif precio and importe is not None and cantidad is None:
|
|
cantidad = importe / precio
|
|
if not cantidad:
|
|
raise ErrorOperacion("Indica la cantidad (o el importe y el precio).")
|
|
if precio is None and tipo != "Comision":
|
|
raise ErrorOperacion("Indica el precio (o el importe).")
|
|
# En los que se siguen por valor no hay participaciones que comprobar
|
|
if tipo in SALIDAS | {"Comision"} and cantidad is not None and not por_valor(a):
|
|
disponible = cantidad_en([m for m in self.datos.vivos("movimientos") if m["id"] != c.get("id")],
|
|
a["id"], fecha)
|
|
if cantidad > disponible + 1e-9:
|
|
raise ErrorOperacion("No tenías tanta cantidad de ese activo en esa fecha.")
|
|
|
|
campos = {"fecha": fecha.isoformat(), "activo": a["id"], "tipo": tipo,
|
|
"cantidad": cantidad, "precio": precio, "importe": importe,
|
|
"comisiones": _numero(c.get("comisiones"), "Las comisiones"),
|
|
"notas": _texto(c.get("notas")), "estimado": bool(c.get("estimado")), "pendiente": False}
|
|
if tipo == "Amortizacion":
|
|
campos["modalidad"] = c["modalidad"]
|
|
if c.get("actividad") is not None:
|
|
campos["actividad"] = _texto(c.get("actividad"), 40)
|
|
if tipo in ("Gasto", "Devolucion") and c.get("amortizable") is not None:
|
|
campos["amortizable"] = bool(c.get("amortizable"))
|
|
if tipo == "Dividendo" and c.get("impagado") is not None:
|
|
campos["impagado"] = bool(c.get("impagado"))
|
|
if tipo in ("Gasto", "Devolucion") and c.get("clase_gasto"):
|
|
from .fiscal.alquiler import CLASES_GASTO
|
|
if c["clase_gasto"] not in CLASES_GASTO:
|
|
raise ErrorOperacion("Esa clase de gasto no existe.")
|
|
campos["clase_gasto"] = c["clase_gasto"]
|
|
if c.get("lugar") is not None:
|
|
campos["lugar"] = self._lugar_valido(a, c.get("lugar"))
|
|
if c.get("extracto"):
|
|
campos["extracto"] = _texto(c.get("extracto"), 80) # línea del extracto de la que viene
|
|
if c.get("importacion"):
|
|
campos["importacion"] = _texto(c.get("importacion"), 40) # para poder deshacerla entera
|
|
if c.get("moneda") == "USD" and c.get("importe_original"):
|
|
campos.update({"moneda": "USD",
|
|
"importe_original": _numero(c.get("importe_original"), "El importe en dólares")})
|
|
if c.get("retencion") is not None:
|
|
campos["retencion"] = _numero(c.get("retencion"), "La retención")
|
|
if tipo == "Recompensa":
|
|
campos["valor_recibido"] = _numero(c.get("valor_recibido"), "El valor al recibirlo")
|
|
if tipo == "Traspaso_Entrada":
|
|
campos["coste_traspaso"] = _numero(c.get("coste_traspaso"), "El coste fiscal")
|
|
if c.get("id"):
|
|
m = self._movimiento(c["id"])
|
|
m.update(campos)
|
|
m["modificado"] = ahora()
|
|
else:
|
|
m = {"id": "m" + uuid.uuid4().hex[:15], **campos,
|
|
"orden": self._siguiente_orden("movimientos"), "modificado": ahora()}
|
|
self.datos.movimientos.append(m)
|
|
self._guardar()
|
|
return m
|
|
|
|
def borrar_movimiento(self, mid: str) -> dict:
|
|
"""Borra una operación (y la comisión unida a un traslado). Devuelve los
|
|
ids borrados, para poder deshacerlo (restaurar_movimientos)."""
|
|
with self._lock:
|
|
m = self._movimiento(mid)
|
|
borrados = [m] + self._comisiones_de(m)
|
|
with self._bloque():
|
|
for x in borrados:
|
|
x.update({"borrado": True, "modificado": ahora()})
|
|
return {"ids": [x["id"] for x in borrados]}
|
|
|
|
def restaurar_movimientos(self, ids) -> dict:
|
|
"""Deshace un borrado: las operaciones vuelven tal como estaban."""
|
|
if not isinstance(ids, list) or not ids or len(ids) > 5000:
|
|
raise ErrorOperacion("No hay nada que deshacer.")
|
|
with self._lock:
|
|
borrados = {m["id"]: m for m in self.datos.movimientos if m.get("borrado")}
|
|
if any(i not in borrados for i in ids):
|
|
raise ErrorOperacion("Ya no se puede deshacer: alguna de esas operaciones no está entre las borradas.")
|
|
with self._bloque():
|
|
for i in ids:
|
|
borrados[i].update({"borrado": False, "modificado": ahora()})
|
|
return {"restauradas": len(ids)}
|
|
|
|
def _comisiones_de(self, m: dict) -> list[dict]:
|
|
"""La comisión que va unida a un traslado (se borra con él)."""
|
|
if m.get("tipo") != "Traslado":
|
|
return []
|
|
return [x for x in self.datos.vivos("movimientos") if x.get("traslado") == m["id"]]
|
|
|
|
# ── varias operaciones a la vez ──
|
|
def _seleccion(self, ids) -> list[dict]:
|
|
if not isinstance(ids, list) or not ids:
|
|
raise ErrorOperacion("No has seleccionado ninguna operación.")
|
|
if len(ids) > 5000:
|
|
raise ErrorOperacion("Demasiadas operaciones de una vez (máximo 5000).")
|
|
porid = {m["id"]: m for m in self.datos.vivos("movimientos")}
|
|
if any(i not in porid for i in ids):
|
|
raise ErrorOperacion("Alguna de las operaciones ya no existe: recarga la página.")
|
|
return [porid[i] for i in dict.fromkeys(ids)]
|
|
|
|
def borrar_movimientos(self, ids) -> dict:
|
|
with self._lock:
|
|
movs = self._seleccion(ids)
|
|
ids = []
|
|
with self._bloque():
|
|
for m in list(movs):
|
|
for x in [m] + self._comisiones_de(m):
|
|
x.update({"borrado": True, "modificado": ahora()})
|
|
ids.append(x["id"])
|
|
return {"borradas": len(movs), "ids": ids}
|
|
|
|
# ── lugares y traslados (ver lugares.py) ──
|
|
def _lugar_valido(self, activo: dict, lid) -> str:
|
|
lid = _texto(lid, 40)
|
|
if lid and lid not in motor_lugares.por_id(activo):
|
|
raise ErrorOperacion("Ese lugar no es de este activo.")
|
|
return lid
|
|
|
|
def guardar_lugar(self, aid: str, c: dict) -> dict:
|
|
"""Crea o renombra un lugar del activo. Devuelve el lugar."""
|
|
with self._lock:
|
|
a = self.activo(aid)
|
|
nombre = _texto(c.get("nombre"), 60)
|
|
if not nombre:
|
|
raise ErrorOperacion("Pon un nombre al lugar.")
|
|
clase = c.get("clase") or "otro"
|
|
if clase not in motor_lugares.NOMBRES_CLASE:
|
|
raise ErrorOperacion("Elige qué es el lugar (monedero, exchange, nodo Lightning u otro).")
|
|
lista = motor_lugares.de(a)
|
|
otro = motor_lugares.buscar(a, nombre)
|
|
if otro and otro["id"] != c.get("id"):
|
|
raise ErrorOperacion(f"Ya hay un lugar llamado «{otro['nombre']}».")
|
|
if c.get("id"):
|
|
lugar = next((x for x in lista if x["id"] == c["id"]), None)
|
|
if lugar is None:
|
|
raise ErrorOperacion("Ese lugar ya no está.")
|
|
lugar.update({"nombre": nombre, "clase": clase})
|
|
else:
|
|
lugar = {"id": motor_lugares.nuevo_id(nombre, [x["id"] for x in lista]),
|
|
"nombre": nombre, "clase": clase}
|
|
lista.append(lugar)
|
|
a.update({"lugares": lista, "modificado": ahora()})
|
|
self._guardar()
|
|
return dict(lugar)
|
|
|
|
def asegurar_lugar(self, aid: str, nombre: str, clase: str = "otro") -> dict:
|
|
"""El lugar con ese nombre; si no existe, se crea."""
|
|
with self._lock:
|
|
ya = motor_lugares.buscar(self.activo(aid), nombre)
|
|
return dict(ya) if ya else self.guardar_lugar(aid, {"nombre": nombre, "clase": clase})
|
|
|
|
def borrar_lugar(self, aid: str, lid: str) -> None:
|
|
with self._lock:
|
|
a = self.activo(aid)
|
|
usado = [m for m in self.datos.vivos("movimientos")
|
|
if m["activo"] == aid and lid in (m.get("lugar"), m.get("lugar_destino"))]
|
|
if usado:
|
|
raise ErrorOperacion(f"Ese lugar tiene {len(usado)} operación(es): cámbialas antes de quitarlo.")
|
|
a.update({"lugares": [x for x in motor_lugares.de(a) if x["id"] != lid], "modificado": ahora()})
|
|
self._guardar()
|
|
|
|
def unir_lugares(self, aid: str, desde: str, hacia: str) -> dict:
|
|
"""Junta el lugar «desde» dentro de «hacia» (p. ej. «BTC en Binance» y
|
|
«Binance», que son el mismo sitio): todas sus operaciones pasan a
|
|
«hacia» y «desde» desaparece. Un traslado entre los dos ya no mueve
|
|
nada y se quita, pero su comisión (dinero que sí se perdió) se queda."""
|
|
with self._lock:
|
|
a = self.activo(aid)
|
|
ls = motor_lugares.por_id(a)
|
|
if desde not in ls or hacia not in ls:
|
|
raise ErrorOperacion("Ese lugar ya no está.")
|
|
if desde == hacia:
|
|
raise ErrorOperacion("Elige otro lugar.")
|
|
movidas = quitados = 0
|
|
with self._bloque():
|
|
for m in self.datos.vivos("movimientos"):
|
|
if m["activo"] != aid:
|
|
continue
|
|
cambia = False
|
|
for campo in ("lugar", "lugar_destino"):
|
|
if m.get(campo) == desde:
|
|
m[campo] = hacia
|
|
cambia = True
|
|
if not cambia:
|
|
continue
|
|
m["modificado"] = ahora()
|
|
movidas += 1
|
|
if m["tipo"] == "Traslado" and m.get("lugar") == m.get("lugar_destino"):
|
|
for x in self._comisiones_de(m):
|
|
x.pop("traslado", None)
|
|
x["modificado"] = ahora()
|
|
m["borrado"] = True
|
|
quitados += 1
|
|
a.update({"lugares": [x for x in motor_lugares.de(a) if x["id"] != desde], "modificado": ahora()})
|
|
return {"movidas": movidas, "traslados_quitados": quitados}
|
|
|
|
def poner_lugar(self, ids, lid: str) -> dict:
|
|
"""Pone el lugar a varias operaciones (sin cambiar ninguna cifra)."""
|
|
with self._lock:
|
|
movs = [m for m in self._seleccion(ids) if m["tipo"] != "Traslado"]
|
|
with self._bloque():
|
|
for m in movs:
|
|
valido = self._lugar_valido(self.activo(m["activo"]), lid)
|
|
if valido:
|
|
m["lugar"] = valido
|
|
else:
|
|
m.pop("lugar", None)
|
|
m["modificado"] = ahora()
|
|
return {"cambiadas": len(movs)}
|
|
|
|
def guardar_traslado(self, c: dict) -> dict:
|
|
"""Mover cantidad entre dos lugares tuyos del mismo activo: una línea
|
|
«Traslado» y, si hubo comisión, un «Comision» unido a él en el origen.
|
|
c: activo, fecha, cantidad (lo que llega), comision (cantidad perdida,
|
|
opcional), lugar (origen), lugar_destino, notas, extracto, importacion,
|
|
txid, id (para editar). Un extremo puede quedar "" (sin saber aún)."""
|
|
with self._lock:
|
|
a = self.activo(c.get("activo"))
|
|
fecha = _fecha(c.get("fecha"))
|
|
cantidad = _numero(c.get("cantidad"), "La cantidad que llega", positivo=True, opcional=False)
|
|
comision = _numero(c.get("comision"), "La comisión") or 0.0
|
|
if comision < 0:
|
|
raise ErrorOperacion("La comisión no puede ser negativa.")
|
|
origen = self._lugar_valido(a, c.get("lugar"))
|
|
destino = self._lugar_valido(a, c.get("lugar_destino"))
|
|
if origen and origen == destino:
|
|
raise ErrorOperacion("El origen y el destino deben ser distintos.")
|
|
if not origen and not destino:
|
|
raise ErrorOperacion("Elige al menos de dónde sale o a dónde llega.")
|
|
anterior = self._movimiento(c["id"]) if c.get("id") else None
|
|
if anterior is not None and anterior["tipo"] != "Traslado":
|
|
raise ErrorOperacion("Esa operación no es un traslado.")
|
|
campos = {"fecha": fecha.isoformat(), "activo": a["id"], "tipo": "Traslado",
|
|
"cantidad": cantidad, "precio": None, "importe": None,
|
|
"lugar": origen, "lugar_destino": destino, "notas": _texto(c.get("notas")),
|
|
"modificado": ahora()}
|
|
for k in ("extracto", "importacion", "txid"):
|
|
if c.get(k):
|
|
campos[k] = _texto(c.get(k), 80)
|
|
with self._bloque():
|
|
if anterior is None:
|
|
t = {"id": "m" + uuid.uuid4().hex[:15], **campos, "orden": self._siguiente_orden("movimientos")}
|
|
self.datos.movimientos.append(t)
|
|
else:
|
|
t = anterior
|
|
t.update(campos)
|
|
comisiones = self._comisiones_de(t)
|
|
if comision > 0:
|
|
if comision > cantidad_en(self.datos.vivos("movimientos"), a["id"], fecha) + 1e-9:
|
|
raise ErrorOperacion("No tenías tanta cantidad para pagar esa comisión.")
|
|
com = comisiones[0] if comisiones else {"id": "m" + uuid.uuid4().hex[:15],
|
|
"orden": self._siguiente_orden("movimientos")}
|
|
com.update({"fecha": fecha.isoformat(), "activo": a["id"], "tipo": "Comision",
|
|
"cantidad": comision, "precio": None, "importe": None,
|
|
"lugar": origen, "traslado": t["id"],
|
|
"notas": "Comisión del traslado", "modificado": ahora()})
|
|
for k in ("extracto", "importacion"):
|
|
if c.get(k):
|
|
com[k] = _texto(c.get(k), 80) + (":comision" if k == "extracto" else "")
|
|
if not comisiones:
|
|
self.datos.movimientos.append(com)
|
|
comisiones = comisiones[1:]
|
|
for x in comisiones:
|
|
x.update({"borrado": True, "modificado": ahora()})
|
|
return t
|
|
|
|
def completar_traslado(self, mid: str, lugar: str | None = None, lugar_destino: str | None = None,
|
|
txid: str | None = None, importacion: str | None = None) -> dict:
|
|
"""Rellena el extremo que faltaba (lo hace el historial al ver llegar o
|
|
salir esa cantidad de un monedero) y anota el txid. No cambia cifras."""
|
|
with self._lock:
|
|
t = self._movimiento(mid)
|
|
if t["tipo"] != "Traslado":
|
|
raise ErrorOperacion("Esa operación no es un traslado.")
|
|
a = self.activo(t["activo"])
|
|
if lugar is not None and not t.get("lugar"):
|
|
t["lugar"] = self._lugar_valido(a, lugar)
|
|
for x in self._comisiones_de(t):
|
|
x["lugar"] = t["lugar"]
|
|
if lugar_destino is not None and not t.get("lugar_destino"):
|
|
t["lugar_destino"] = self._lugar_valido(a, lugar_destino)
|
|
if t.get("lugar") and t.get("lugar") == t.get("lugar_destino"):
|
|
raise ErrorOperacion("El origen y el destino del traslado serían el mismo.")
|
|
if txid:
|
|
t["txid"] = _texto(txid, 64)
|
|
if importacion:
|
|
t["completado_en"] = _texto(importacion, 40)
|
|
t["modificado"] = ahora()
|
|
self._guardar()
|
|
return t
|
|
|
|
def vaciar_extremo(self, mid: str, cual: str) -> None:
|
|
"""Deshace completar_traslado (lo usa «deshacer» del historial)."""
|
|
with self._lock:
|
|
t = self._movimiento(mid)
|
|
if cual in ("lugar", "lugar_destino"):
|
|
t[cual] = ""
|
|
if cual == "lugar":
|
|
for x in self._comisiones_de(t):
|
|
x["lugar"] = ""
|
|
t.pop("txid", None)
|
|
t.pop("completado_en", None)
|
|
t["modificado"] = ahora()
|
|
self._guardar()
|
|
|
|
def convertir_en_lugar(self, aid: str, destino: str, clase: str = "otro") -> dict:
|
|
"""Un activo que en realidad es un sitio donde tienes el mismo bitcoin
|
|
(«BTC en mi nodo») pasa a ser un lugar de «destino». Sus operaciones se
|
|
mueven allí con ese lugar, y sus traspasos con «destino» pasan a ser
|
|
traslados. Devuelve el lugar creado."""
|
|
with self._lock:
|
|
a, d = self.activo(aid), self.activo(destino)
|
|
if a["id"] == d["id"]:
|
|
raise ErrorOperacion("Elige otro activo.")
|
|
if a["tipo"] != d["tipo"] or (a.get("codigo") or "").lower() != (d.get("codigo") or "").lower():
|
|
raise ErrorOperacion("Solo se puede unir a un activo de la misma moneda.")
|
|
movs = [m for m in self.datos.vivos("movimientos") if m["activo"] == a["id"]]
|
|
todos = self.datos.vivos("movimientos")
|
|
pares, sueltos = [], []
|
|
for m in movs:
|
|
if m["tipo"] not in ("Traspaso_Entrada", "Traspaso_Salida"):
|
|
continue
|
|
buscado = "Traspaso_Salida" if m["tipo"] == "Traspaso_Entrada" else "Traspaso_Entrada"
|
|
otro = next((x for x in todos if x["activo"] == d["id"] and x["tipo"] == buscado
|
|
and x["fecha"] == m["fecha"] and abs((x.get("orden") or 0) - (m.get("orden") or 0)) == 1),
|
|
None)
|
|
(pares if otro else sueltos).append((m, otro))
|
|
if sueltos:
|
|
raise ErrorOperacion(f"«{a['nombre']}» tiene {len(sueltos)} traspaso(s) con otros activos: "
|
|
"deshazlos antes de unirlo.")
|
|
with self._bloque():
|
|
lugar = self.asegurar_lugar(d["id"], a["nombre"], clase)
|
|
for m in movs:
|
|
if m["tipo"] in ("Traspaso_Entrada", "Traspaso_Salida"):
|
|
continue
|
|
m.update({"activo": d["id"], "lugar": m.get("lugar") if m.get("lugar") in
|
|
motor_lugares.por_id(d) else lugar["id"], "modificado": ahora()})
|
|
for m, otro in pares:
|
|
sal, ent = (otro, m) if m["tipo"] == "Traspaso_Entrada" else (m, otro)
|
|
q_sal, q_ent = _f_num(sal.get("cantidad")), _f_num(ent.get("cantidad"))
|
|
desde = lugar["id"] if sal is m else (sal.get("lugar") or "")
|
|
hacia = lugar["id"] if ent is m else (ent.get("lugar") or "")
|
|
self.guardar_traslado({"activo": d["id"], "fecha": sal["fecha"], "cantidad": q_ent,
|
|
"comision": round(q_sal - q_ent, 8) if q_sal > q_ent else 0,
|
|
"lugar": desde, "lugar_destino": hacia,
|
|
"notas": sal.get("notas") or ent.get("notas") or ""})
|
|
for x in (sal, ent):
|
|
x.update({"borrado": True, "modificado": ahora()})
|
|
a.update({"borrado": True, "modificado": ahora()})
|
|
return lugar
|
|
|
|
def borrar_activos_vacios(self, ids) -> int:
|
|
"""Borra activos que no tienen ninguna operación viva (p. ej. los que
|
|
creó una importación que se ha deshecho). Los que tengan alguna, no."""
|
|
with self._lock:
|
|
con_movs = {m["activo"] for m in self.datos.vivos("movimientos")}
|
|
borrados = 0
|
|
with self._bloque():
|
|
for a in self.datos.vivos("activos"):
|
|
if a["id"] in ids and a["id"] not in con_movs:
|
|
a.update({"borrado": True, "modificado": ahora()})
|
|
borrados += 1
|
|
return borrados
|
|
|
|
def confirmar_movimientos(self, ids) -> dict:
|
|
"""Confirma las estimadas; las pendientes necesitan antes sus cifras."""
|
|
with self._lock:
|
|
movs = self._seleccion(ids)
|
|
listas = [m for m in movs if m.get("estimado") and not m.get("pendiente")]
|
|
with self._bloque():
|
|
for m in listas:
|
|
m.update({"estimado": False, "modificado": ahora()})
|
|
return {"confirmadas": len(listas),
|
|
"pendientes": sum(1 for m in movs if m.get("pendiente"))}
|
|
|
|
def poner_importe(self, mid: str, importe) -> dict:
|
|
"""Lo que pagaste (o cobraste) de verdad por una compra o venta ya
|
|
apuntada: cambia el importe y el precio por unidad, no la cantidad, y
|
|
la operación deja de ser estimada."""
|
|
with self._lock:
|
|
m = self._movimiento(mid)
|
|
if m["tipo"] not in ("Compra", "Venta"):
|
|
raise ErrorOperacion("Solo las compras y las ventas tienen importe pagado o cobrado.")
|
|
importe = _numero(importe, "el importe", positivo=True, opcional=False)
|
|
cantidad = _f_num(m.get("cantidad"))
|
|
if cantidad <= 0:
|
|
raise ErrorOperacion("Esa operación no tiene cantidad: edítala entera.")
|
|
m.update({"importe": round(importe, 2), "precio": importe / cantidad,
|
|
"estimado": False, "modificado": ahora()})
|
|
self._guardar()
|
|
return m
|
|
|
|
def revisar_prima(self, ids, ok: bool = True) -> dict:
|
|
"""Da por buena (o deja de dar por buena) la cifra de unas operaciones
|
|
cuyo precio se aleja del mercado de ese día (ver prima.py)."""
|
|
with self._lock:
|
|
movs = self._seleccion(ids)
|
|
with self._bloque():
|
|
for m in movs:
|
|
if ok:
|
|
m["prima_ok"] = True
|
|
else:
|
|
m.pop("prima_ok", None)
|
|
m["modificado"] = ahora()
|
|
return {"cambiadas": len(movs)}
|
|
|
|
def enlazar_txid(self, mid: str, txid: str | None) -> dict:
|
|
"""Anota (o quita, con None) la transacción de bitcoin de una operación
|
|
ya apuntada. No cambia ninguna cifra."""
|
|
with self._lock:
|
|
m = self._movimiento(mid)
|
|
if txid:
|
|
m["txid"] = _texto(txid, 64)
|
|
else:
|
|
m.pop("txid", None)
|
|
m["modificado"] = ahora()
|
|
self._guardar()
|
|
return m
|
|
|
|
def anotar_movimientos(self, ids, campos: dict) -> None:
|
|
"""Marcas que no cambian cifras (extracto, importación, estimado)."""
|
|
permitidos = {"extracto", "importacion", "estimado"}
|
|
with self._lock:
|
|
movs = self._seleccion(ids)
|
|
with self._bloque():
|
|
for m in movs:
|
|
m.update({k: v for k, v in campos.items() if k in permitidos})
|
|
m["modificado"] = ahora()
|
|
|
|
def mover_movimientos(self, ids, activo: str) -> dict:
|
|
"""Pasa las operaciones a otro activo, con las mismas comprobaciones que
|
|
al editarlas una a una (tipo de operación, cantidad disponible...)."""
|
|
with self._lock:
|
|
destino = self.activo(activo)
|
|
movs = [m for m in self._seleccion(ids) if m["activo"] != destino["id"]]
|
|
# En orden: las entradas de un día antes que las salidas, para que
|
|
# una venta encuentre ya en el activo nuevo lo que se compró antes.
|
|
movs.sort(key=lambda m: (m["fecha"], m["tipo"] in SALIDAS | {"Comision"}, m.get("orden") or 0))
|
|
with self._bloque():
|
|
for m in movs:
|
|
try:
|
|
self.guardar_movimiento({**m, "activo": destino["id"]})
|
|
except ErrorOperacion as e:
|
|
raise ErrorOperacion(f"No se ha movido ninguna. La del {m['fecha']} "
|
|
f"({m['tipo']}): {e}") from None
|
|
return {"movidas": len(movs), "activo": destino["id"]}
|
|
|
|
# ── justificantes ──
|
|
def adjuntar(self, mid: str, nombre: str, contenido: bytes) -> dict:
|
|
try:
|
|
nombre = adjuntos.limpiar_nombre(nombre)
|
|
tipo = adjuntos.tipo_de(nombre)
|
|
with self._lock:
|
|
m = self._movimiento(mid)
|
|
lista = list(m.get("adjuntos") or [])
|
|
if len(lista) >= adjuntos.MAX_POR_OPERACION:
|
|
raise ErrorOperacion(f"Como mucho {adjuntos.MAX_POR_OPERACION} justificantes por operación.")
|
|
huella = adjuntos.guardar(self.carpeta, contenido)
|
|
if any(a["id"] == huella for a in lista):
|
|
raise ErrorOperacion("Ese fichero ya está adjunto a esta operación.")
|
|
lista.append({"id": huella, "nombre": nombre, "tipo": tipo, "bytes": len(contenido)})
|
|
m.update({"adjuntos": lista, "modificado": ahora()})
|
|
self._guardar()
|
|
return m
|
|
except adjuntos.ErrorAdjunto as e:
|
|
raise ErrorOperacion(str(e)) from None
|
|
|
|
def quitar_adjunto(self, mid: str, huella: str) -> dict:
|
|
"""Quita la referencia; el fichero se queda (ver adjuntos.py)."""
|
|
with self._lock:
|
|
m = self._movimiento(mid)
|
|
lista = [a for a in m.get("adjuntos") or [] if a["id"] != huella]
|
|
if len(lista) == len(m.get("adjuntos") or []):
|
|
raise ErrorOperacion("Esa operación no tiene ese justificante.")
|
|
m.update({"adjuntos": lista, "modificado": ahora()})
|
|
self._guardar()
|
|
return m
|
|
|
|
def leer_adjunto(self, mid: str, huella: str) -> tuple[dict, bytes]:
|
|
with self._lock:
|
|
m = self._movimiento(mid)
|
|
ref = next((a for a in m.get("adjuntos") or [] if a["id"] == huella), None)
|
|
if ref is None:
|
|
raise ErrorOperacion("Esa operación no tiene ese justificante.")
|
|
try:
|
|
return ref, adjuntos.leer(self.carpeta, huella)
|
|
except adjuntos.ErrorAdjunto as e:
|
|
raise ErrorOperacion(str(e)) from None
|
|
|
|
def confirmar_movimiento(self, mid: str) -> dict:
|
|
"""Marca como confirmada una aportación con participaciones estimadas."""
|
|
with self._lock:
|
|
m = self._movimiento(mid)
|
|
if m.get("pendiente"):
|
|
raise ErrorOperacion("Completa primero la cantidad y el precio.")
|
|
m.update({"estimado": False, "modificado": ahora()})
|
|
self._guardar()
|
|
return m
|
|
|
|
def traspaso(self, c: dict) -> list[dict]:
|
|
"""Traspaso entre fondos o planes, o entre dos sitios donde tienes la
|
|
misma cripto (de un exchange a un nodo Lightning, por ejemplo): salida
|
|
del origen y entrada en el destino el mismo día y por el mismo importe,
|
|
conservando el coste fiscal. «lugar_origen» y «lugar_destino»
|
|
(opcionales) dicen de qué lugar sale y a cuál llega."""
|
|
with self._lock:
|
|
origen, destino = self.activo(c.get("origen")), self.activo(c.get("destino"))
|
|
if origen["id"] == destino["id"]:
|
|
raise ErrorOperacion("El origen y el destino deben ser distintos.")
|
|
lugar_sal = self._lugar_valido(origen, c.get("lugar_origen"))
|
|
lugar_ent = self._lugar_valido(destino, c.get("lugar_destino"))
|
|
cripto = "Cripto" in (origen["tipo"], destino["tipo"])
|
|
if cripto and (origen["tipo"] != destino["tipo"]
|
|
or (origen.get("codigo") or "").lower() != (destino.get("codigo") or "").lower()):
|
|
raise ErrorOperacion("Un traspaso de cripto tiene que ser entre dos activos de la misma "
|
|
"moneda (por ejemplo, de «BTC en un exchange» a «BTC en mi monedero»).")
|
|
fecha = _fecha(c.get("fecha"))
|
|
q_sal = _numero(c.get("cantidad_salida"), "Las participaciones que salen", positivo=True, opcional=False)
|
|
q_ent = _numero(c.get("cantidad_entrada"), "Las participaciones que entran", positivo=True,
|
|
opcional=cripto) or q_sal
|
|
if cripto and q_ent > q_sal + 1e-12:
|
|
raise ErrorOperacion("No puede llegar más de lo que sale (la diferencia es la comisión).")
|
|
importe = _numero(c.get("importe"), "El importe traspasado", positivo=True, opcional=cripto)
|
|
estimado = False
|
|
if importe is None:
|
|
# En cripto, si no se sabe, el valor de ese día
|
|
importe = self._valor_de_referencia(origen, fecha, q_sal)
|
|
estimado = True
|
|
coste = _numero(c.get("coste_fiscal"), "El coste fiscal")
|
|
if coste is None and cripto:
|
|
# Lo que te costó lo que sale, al coste medio del origen
|
|
from .historial_btc import _coste_medio
|
|
medio = _coste_medio(self.datos.vivos("movimientos"), origen["id"], fecha.isoformat())
|
|
coste = round(medio * q_sal, 2) if medio else None
|
|
if q_sal > cantidad_en(self.datos.vivos("movimientos"), origen["id"], fecha) + 1e-9:
|
|
raise ErrorOperacion("No tenías tantas participaciones en el origen en esa fecha.")
|
|
nota = _texto(c.get("notas"))
|
|
orden = self._siguiente_orden("movimientos")
|
|
sal = {"id": "m" + uuid.uuid4().hex[:15], "fecha": fecha.isoformat(), "activo": origen["id"],
|
|
"tipo": "Traspaso_Salida", "cantidad": q_sal, "precio": importe / q_sal,
|
|
"importe": importe, "notas": nota or f"Traspaso a {destino['nombre']}",
|
|
"orden": orden, "modificado": ahora()}
|
|
if estimado:
|
|
sal["estimado"] = True
|
|
ent = {"id": "m" + uuid.uuid4().hex[:15], "fecha": fecha.isoformat(), "activo": destino["id"],
|
|
"tipo": "Traspaso_Entrada", "cantidad": q_ent, "precio": importe / q_ent,
|
|
"importe": importe, "coste_traspaso": coste,
|
|
"notas": nota or f"Traspaso desde {origen['nombre']}",
|
|
"orden": orden + 1, "modificado": ahora()}
|
|
if estimado:
|
|
ent["estimado"] = True
|
|
if lugar_sal:
|
|
sal["lugar"] = lugar_sal
|
|
if lugar_ent:
|
|
ent["lugar"] = lugar_ent
|
|
self.datos.movimientos += [sal, ent]
|
|
self._guardar()
|
|
return [sal, ent]
|
|
|
|
def _valor_de_referencia(self, activo: dict, fecha: dt.date, cantidad: float) -> float:
|
|
from . import spot_btc
|
|
otros = [r for r in self.datos.vivos("precios") if r["activo"] == activo["id"]]
|
|
precio = None
|
|
if (activo.get("codigo") or "").lower() == "bitcoin":
|
|
precio, _ = spot_btc.precio(fecha.isoformat(), otros)
|
|
if precio is None:
|
|
previos = [r for r in otros if r.get("precio") and r["fecha"] <= fecha.isoformat()]
|
|
precio = float(max(previos, key=lambda r: r["fecha"])["precio"]) if previos else None
|
|
if not precio:
|
|
raise ErrorOperacion("No sé a qué precio estaba ese día: pon el importe en euros.")
|
|
return round(precio * cantidad, 2)
|
|
|
|
# ── precios ──
|
|
def precio_manual(self, aid: str, fecha, valor, es_valor_total: bool = False) -> dict:
|
|
with self._lock:
|
|
a = self.activo(aid)
|
|
f = _fecha(fecha)
|
|
v = _numero(valor, "El valor", positivo=True, opcional=False)
|
|
campo = "valor_total" if es_valor_total or a["tipo"] == "CuentaTrading" or por_valor(a) else "precio"
|
|
if a["tipo"] in ("Cuenta", "Deuda"):
|
|
campo = "precio" # el saldo de la cuenta, o lo que queda por pagar según el banco
|
|
return self._poner_precio(aid, f, {campo: v}, "manual")
|
|
|
|
def _poner_precio(self, aid: str, fecha: dt.date, valores: dict, fuente: str, guardar: bool = True) -> dict:
|
|
clave = fecha.isoformat()
|
|
for r in self.datos.precios:
|
|
if r["activo"] == aid and r["fecha"] == clave:
|
|
for campo in ("precio", "valor_total", "borrado"):
|
|
r.pop(campo, None)
|
|
r.update(valores)
|
|
r.update({"fuente": fuente, "modificado": ahora()})
|
|
break
|
|
else:
|
|
r = {"fecha": clave, "activo": aid, **valores, "fuente": fuente, "modificado": ahora()}
|
|
self.datos.precios.append(r)
|
|
if guardar:
|
|
self._guardar()
|
|
return r
|
|
|
|
def aplicar_valores(self, resultados: list[dict]) -> None:
|
|
"""Guarda valores totales descargados: [{activo, fecha, valor_total, fuente}]."""
|
|
with self._lock:
|
|
for r in resultados:
|
|
self._poner_precio(r["activo"], _fecha(r["fecha"]), {"valor_total": r["valor_total"]}, r["fuente"], guardar=False)
|
|
self._guardar()
|
|
|
|
def aplicar_precios(self, resultados: list[dict]) -> None:
|
|
"""Guarda los precios descargados: [{activo, fecha, precio, fuente}]."""
|
|
with self._lock:
|
|
for r in resultados:
|
|
self._poner_precio(r["activo"], _fecha(r["fecha"]), {"precio": r["precio"]},
|
|
r["fuente"], guardar=False)
|
|
self._guardar()
|
|
|
|
# ── operaciones automáticas ──
|
|
def guardar_recurrente(self, c: dict) -> dict:
|
|
with self._lock:
|
|
a = self.activo(c.get("activo"))
|
|
tipo = c.get("tipo")
|
|
if tipo not in ("Gasto", "Compra", "Dividendo"):
|
|
raise ErrorOperacion("El tipo debe ser gasto, aportación o cobro.")
|
|
moneda = c.get("moneda") or "EUR"
|
|
if moneda not in MONEDAS:
|
|
raise ErrorOperacion("Moneda no admitida.")
|
|
try:
|
|
dia = int(c.get("dia"))
|
|
except (TypeError, ValueError):
|
|
raise ErrorOperacion("Indica el día del mes.") from None
|
|
if not 1 <= dia <= 28:
|
|
raise ErrorOperacion("El día debe estar entre 1 y 28.")
|
|
campos = {"activo": a["id"], "tipo": tipo, "moneda": moneda, "dia": dia,
|
|
"importe": _numero(c.get("importe"), "El importe", positivo=True, opcional=False),
|
|
"desde": _fecha(c.get("desde"), "fecha de inicio").isoformat(),
|
|
"hasta": _fecha(c["hasta"], "fecha final").isoformat() if c.get("hasta") else None,
|
|
"pausado": bool(c.get("pausado")), "notas": _texto(c.get("notas")),
|
|
"actividad": _texto(c.get("actividad"), 40)}
|
|
if c.get("id"):
|
|
r = next((r for r in self.datos.vivos("recurrentes") if r["id"] == c["id"]), None)
|
|
if r is None:
|
|
raise ErrorOperacion("Esa operación automática no existe.")
|
|
r.update(campos)
|
|
r["modificado"] = ahora()
|
|
else:
|
|
r = {"id": "r" + uuid.uuid4().hex[:11], **campos,
|
|
"orden": self._siguiente_orden("recurrentes"), "modificado": ahora()}
|
|
self.datos.recurrentes.append(r)
|
|
self._guardar()
|
|
return r
|
|
|
|
def borrar_recurrente(self, rid: str) -> None:
|
|
with self._lock:
|
|
for r in self.datos.vivos("recurrentes"):
|
|
if r["id"] == rid:
|
|
r.update({"borrado": True, "modificado": ahora()})
|
|
self._guardar()
|
|
return
|
|
raise ErrorOperacion("Esa operación automática no existe.")
|
|
|
|
def etiquetar(self, ids: list, actividad: str) -> dict:
|
|
"""Pone (o quita, con actividad vacía) la etiqueta a varias operaciones."""
|
|
nombre = _texto(actividad, 40)
|
|
with self._lock:
|
|
porid = {m["id"]: m for m in self.datos.vivos("movimientos")}
|
|
tocados = 0
|
|
for mid in list(ids or [])[:5000]:
|
|
m = porid.get(mid)
|
|
if m is None or (m.get("actividad") or "") == nombre:
|
|
continue
|
|
m["actividad"] = nombre
|
|
m["modificado"] = ahora()
|
|
tocados += 1
|
|
if tocados:
|
|
self._guardar()
|
|
return {"marcadas": tocados, "actividad": nombre}
|
|
|
|
def marcar_amortizable(self, ids: list, si: bool) -> dict:
|
|
"""Marca (o desmarca) gastos y devoluciones como la compra de un aparato
|
|
de la actividad. Solo lo mira el cálculo fiscal (fiscal/actividad.py)."""
|
|
with self._lock:
|
|
porid = {m["id"]: m for m in self.datos.vivos("movimientos")}
|
|
tocados = 0
|
|
for mid in list(ids or [])[:5000]:
|
|
m = porid.get(mid)
|
|
if m is None or m["tipo"] not in ("Gasto", "Devolucion") or bool(m.get("amortizable")) == bool(si):
|
|
continue
|
|
m["amortizable"] = bool(si)
|
|
m["modificado"] = ahora()
|
|
tocados += 1
|
|
if tocados:
|
|
self._guardar()
|
|
return {"marcadas": tocados}
|
|
|
|
def renombrar_actividad(self, viejo: str, nuevo: str) -> dict:
|
|
"""Cambia el nombre en todas las operaciones y automáticas que lo llevan."""
|
|
from .actividades import clave
|
|
origen, destino = clave(viejo), _texto(nuevo, 40)
|
|
if not destino:
|
|
raise ErrorOperacion("Ponle un nombre a la actividad.")
|
|
with self._lock:
|
|
tocados = 0
|
|
for registros in (self.datos.vivos("movimientos"), self.datos.vivos("recurrentes")):
|
|
for r in registros:
|
|
if clave(r.get("actividad") or "") == origen:
|
|
r["actividad"] = destino
|
|
r["modificado"] = ahora()
|
|
tocados += 1
|
|
if tocados:
|
|
self._guardar()
|
|
return {"cambiadas": tocados, "actividad": destino}
|
|
|
|
def generar_recurrentes(self, hoy: dt.date, cambio=None) -> dict:
|
|
"""Crea las operaciones automáticas vencidas hasta hoy y completa las
|
|
pendientes. cambio(fecha, "USD") → € por dólar, o None si no hay red.
|
|
|
|
Id determinista (recurrente + fecha): generar en el USB y en el
|
|
ordenador produce la misma operación y la fusión no la duplica."""
|
|
from .calculo import Precios
|
|
creadas = completadas = 0
|
|
with self._lock:
|
|
precios = Precios(self.datos.vivos("precios"))
|
|
existentes = {m["id"]: m for m in self.datos.movimientos} # incluye borrados
|
|
for r in self.datos.vivos("recurrentes"):
|
|
if r.get("pausado"):
|
|
continue
|
|
inicio = _fecha(r["desde"])
|
|
fin = min(hoy, _fecha(r["hasta"])) if r.get("hasta") else hoy
|
|
y, mes = inicio.year, inicio.month
|
|
while True:
|
|
f = dt.date(y, mes, min(r["dia"], calendar.monthrange(y, mes)[1]))
|
|
if f > fin:
|
|
break
|
|
if f >= inicio:
|
|
mid = f"r-{r['id']}-{f.isoformat()}"
|
|
m = existentes.get(mid)
|
|
if m is None:
|
|
m = {"id": mid, "fecha": f.isoformat(), "activo": r["activo"], "tipo": r["tipo"],
|
|
"cantidad": None, "precio": None, "importe": None,
|
|
"notas": r.get("notas") or "Automática", "recurrente": r["id"],
|
|
"importe_original": r["importe"], "moneda": r.get("moneda", "EUR"),
|
|
"pendiente": True, "orden": self._siguiente_orden("movimientos"),
|
|
"actividad": r.get("actividad") or "", "modificado": ahora()}
|
|
self.datos.movimientos.append(m)
|
|
existentes[mid] = m
|
|
creadas += 1
|
|
if m.get("borrado"):
|
|
pass
|
|
elif m.get("pendiente"):
|
|
if self._completar(m, precios, cambio):
|
|
completadas += 1
|
|
elif m.get("estimado") and m["tipo"] == "Compra":
|
|
# Llegó el valor liquidativo exacto de ese día: se afina la estimación
|
|
reg = precios.ultimo(m["activo"], f)
|
|
if reg and reg["fecha"] == m["fecha"] and reg.get("precio") \
|
|
and reg["precio"] != m.get("precio") and m.get("importe"):
|
|
m.update({"precio": reg["precio"], "cantidad": m["importe"] / reg["precio"],
|
|
"modificado": ahora()})
|
|
completadas += 1
|
|
y, mes = (y + 1, 1) if mes == 12 else (y, mes + 1)
|
|
if creadas or completadas:
|
|
self._guardar()
|
|
pendientes = sum(1 for m in self.datos.vivos("movimientos") if m.get("pendiente"))
|
|
return {"creadas": creadas, "completadas": completadas, "pendientes": pendientes}
|
|
|
|
def _completar(self, m: dict, precios, cambio) -> bool:
|
|
f = _fecha(m["fecha"])
|
|
importe = m["importe_original"]
|
|
if m.get("moneda", "EUR") != "EUR":
|
|
tasa = cambio(f, m["moneda"]) if cambio else None
|
|
if not tasa:
|
|
return False
|
|
importe = importe * tasa
|
|
if m["tipo"] in ("Gasto", "Dividendo"):
|
|
m.update({"importe": importe, "pendiente": False, "modificado": ahora()})
|
|
return True
|
|
activo = next((a for a in self.datos.vivos("activos") if a["id"] == m["activo"]), None)
|
|
if activo is not None and por_valor(activo):
|
|
m.update({"importe": importe, "pendiente": False, "modificado": ahora()})
|
|
return True
|
|
# Aportación: participaciones con el valor liquidativo de ese día (estimado)
|
|
reg = precios.ultimo(m["activo"], f)
|
|
if not reg or not reg.get("precio") or (f - _fecha(reg["fecha"])).days > DIAS_MAX_VL_ESTIMADO:
|
|
m["importe"] = importe
|
|
return False
|
|
m.update({"importe": importe, "precio": reg["precio"], "cantidad": importe / reg["precio"],
|
|
"pendiente": False, "estimado": True, "modificado": ahora()})
|
|
return True
|