La automatización mediante Python y la librería ccxt permite a los asesores contables unificar el historial de transacciones de múltiples exchanges, estandarizar cotizaciones en euros y aplicar de forma programática el algoritmo de valoración FIFO exigido por la Agencia Tributaria (AEAT).
Arquitectura del Pipeline de Datos Fiscal
| Componente | Librería / Herramienta | Función en el Flujo Fiscal |
| Extracción API | ccxt | Conexión unificada a +100 CEXs (Binance, Kraken, Coinbase) para obtener trades, depósitos y retiradas |
| Procesamiento | pandas | Normalización de timestamps a hora local, limpieza de duplicados y conversión a EUR |
| Motor FIFO | collections.deque | Gestión de colas de lotes de adquisición para apareamiento estricto de costes |
| Exportación | openpyxl | Generación automática de libros registro preparados para inspecciones del Modelo 100/714 |
Script Base: Extracción y cálculo de pérdidas/ganancias FIFO
Python
import ccxt
import pandas as pd
from collections import deque
def calcular_ganancia_fifo(api_key, secret_key, exchange_id='binance', symbol='BTC/EUR'):
# Inicialización del cliente en modo solo lectura
exchange = getattr(ccxt, exchange_id)({
'apiKey': api_key,
'secret': secret_key,
'enableRateLimit': True,
})
# Obtención del historial de operaciones
trades = exchange.fetch_my_trades(symbol=symbol)
df = pd.DataFrame(trades)
fifo_queue = deque() # Cola de lotes de compra: {'amount': float, 'price': float}
ganancia_patrimonial_total = 0.0
for _, row in df.iterrows():
cantidad = row['amount']
precio = row['price']
if row['side'] == 'buy':
fifo_queue.append({'amount': cantidad, 'price': precio})
elif row['side'] == 'sell':
cantidad_vender = cantidad
while cantidad_vender > 0 and fifo_queue:
lote_antiguo = fifo_queue[0]
cantidad_emparejada = min(cantidad_vender, lote_antiguo['amount'])
# Cálculo del resultado en EUR
ganancia = cantidad_emparejada * (precio - lote_antiguo['price'])
ganancia_patrimonial_total += ganancia
# Actualización de inventario
lote_antiguo['amount'] -= cantidad_emparejada
cantidad_vender -= cantidad_emparejada
if lote_antiguo['amount'] == 0:
fifo_queue.popleft()
return ganancia_patrimonial_total
Buenas Prácticas de Ciberseguridad para Asesores (OpSec)
- Restricción de IP estática: Configura las claves API del cliente para que solo respondan a la dirección IP fija del servidor o VPN de la asesoría.
- Desactivación de permisos de ejecución: Verifica que la API tenga únicamente habilitado el permiso Read-Only (sin retiro ni operaciones).
- Gestión segura de credenciales: Almacena las claves en variables de entorno o gestores de secretos cifrados (
python-dotenv), evitando incrustar tokens en el código fuente.