diegobeyl/backtesting
2
1"""
2Ejemplos de Implementación - Mejoras Críticas para V2
3Estos son fragmentos de código listos para usar/adaptar
4"""
5
6# ============================================================================
7# 1. SISTEMA DE EXCEPCIONES PERSONALIZADO
8# ============================================================================
9
10# utils/exceptions.py
11"""Excepciones personalizadas para el backtesting"""
12
13class BacktestingException(Exception):
14 """Excepción base para todos los errores de backtesting"""
15 pass
16
17
18class DataLoadException(BacktestingException):
19 """Error al cargar datos de mercado"""
20
21 def __init__(self, symbol: str, reason: str):
22 self.symbol = symbol
23 self.reason = reason
24 super().__init__(f"Failed to load data for {symbol}: {reason}")
25
26
27class InsufficientDataException(DataLoadException):
28 """Datos insuficientes para ejecutar backtest"""
29
30 def __init__(self, symbol: str, required: int, actual: int):
31 self.symbol = symbol
32 self.required = required
33 self.actual = actual
34 super().__init__(
35 symbol,
36 f"Need {required} candles, got {actual}"
37 )
38
39
40class ValidationException(BacktestingException):
41 """Error en validación de datos de entrada"""
42
43 def __init__(self, field: str, message: str):
44 self.field = field
45 super().__init__(f"Validation error in {field}: {message}")
46
47
48class AlgorithmException(BacktestingException):
49 """Error en ejecución del algoritmo"""
50
51 def __init__(self, algorithm_name: str, message: str):
52 self.algorithm = algorithm_name
53 super().__init__(f"Algorithm {algorithm_name} failed: {message}")
54
55
56class TimeoutException(BacktestingException):
57 """Operación excedió timeout"""
58
59 def __init__(self, operation: str, timeout_seconds: float):
60 self.operation = operation
61 self.timeout = timeout_seconds
62 super().__init__(
63 f"{operation} exceeded timeout of {timeout_seconds}s"
64 )
65
66
67class ConnectionException(BacktestingException):
68 """Error de conexión a datos provider"""
69
70 def __init__(self, provider: str, message: str):
71 self.provider = provider
72 super().__init__(f"Connection error with {provider}: {message}")
73
74
75# ============================================================================
76# 2. VALIDADORES DE ENTRADA
77# ============================================================================
78
79# utils/validators.py
80"""Validadores para datos de entrada"""
81
82from typing import Dict, Any
83import pandas as pd
84import numpy as np
85from datetime import datetime
86from utils.exceptions import ValidationException
87
88
89class InputValidator:
90 """Validador de datos de entrada"""
91
92 REQUIRED_OHLCV_COLUMNS = {'open', 'high', 'low', 'close', 'volume'}
93 MIN_CANDLES = 10
94 MAX_CAPITAL = 1_000_000_000
95 MAX_RISK_PERCENT = 100.0
96
97 @staticmethod
98 def validate_dataframe(df: pd.DataFrame, symbol: str = "unknown") -> bool:
99 """Valida DataFrame con datos OHLCV"""
100
101 # Validar que no es None/vacío
102 if df is None:
103 raise ValidationException(
104 "dataframe",
105 f"DataFrame is None for {symbol}"
106 )
107
108 if df.empty:
109 raise ValidationException(
110 "dataframe",
111 f"DataFrame is empty for {symbol}"
112 )
113
114 # Validar cantidad mínima de candles
115 if len(df) < InputValidator.MIN_CANDLES:
116 raise ValidationException(
117 "dataframe",
118 f"Need at least {InputValidator.MIN_CANDLES} candles, got {len(df)}"
119 )
120
121 # Validar columnas requeridas
122 columns_lower = {col.lower() for col in df.columns}
123 missing = InputValidator.REQUIRED_OHLCV_COLUMNS - columns_lower
124 if missing:
125 raise ValidationException(
126 "columns",
127 f"Missing required columns: {missing}"
128 )
129
130 # Validar datos numéricos
131 numeric_cols = ['open', 'high', 'low', 'close', 'volume']
132 for col in numeric_cols:
133 col_actual = [c for c in df.columns if c.lower() == col][0]
134 if not pd.api.types.is_numeric_dtype(df[col_actual]):
135 raise ValidationException(
136 col,
137 f"Column {col} must be numeric"
138 )
139
140 # Validar NaN
141 null_cols = df.columns[df.isnull().any()].tolist()
142 if null_cols:
143 raise ValidationException(
144 "data",
145 f"Found NaN values in columns: {null_cols}"
146 )
147
148 # Validar valores positivos
149 for col in ['open', 'high', 'low', 'close']:
150 col_actual = [c for c in df.columns if c.lower() == col][0]
151 if (df[col_actual] <= 0).any():
152 raise ValidationException(
153 col,
154 f"{col} must be positive (all values > 0)"
155 )
156
157 # Validar OHLC lógicos (High >= Low, etc.)
158 h_col = [c for c in df.columns if c.lower() == 'high'][0]
159 l_col = [c for c in df.columns if c.lower() == 'low'][0]
160
161 if (df[h_col] < df[l_col]).any():
162 raise ValidationException(
163 "ohlc",
164 "High must be >= Low in all candles"
165 )
166
167 # Validar índice (fechas únicas y ordenadas)
168 if df.index.duplicated().any():
169 raise ValidationException(
170 "index",
171 "Duplicate timestamps in data"
172 )
173
174 if not df.index.is_monotonic_increasing:
175 raise ValidationException(
176 "index",
177 "Timestamps must be in ascending order"
178 )
179
180 return True
181
182 @staticmethod
183 def validate_backtest_params(
184 capital: float,
185 risk_percent: float,
186 commission_pct: float,
187 position_sizing: str
188 ) -> bool:
189 """Valida parámetros de backtest"""
190
191 # Validar capital
192 if capital <= 0:
193 raise ValidationException(
194 "capital",
195 "Capital must be positive"
196 )
197
198 if capital > InputValidator.MAX_CAPITAL:
199 raise ValidationException(
200 "capital",
201 f"Capital exceeds maximum of {InputValidator.MAX_CAPITAL}"
202 )
203
204 # Validar risk percent
205 if not (0 < risk_percent <= InputValidator.MAX_RISK_PERCENT):
206 raise ValidationException(
207 "risk_percent",
208 f"Risk percent must be between 0 and {InputValidator.MAX_RISK_PERCENT}"
209 )
210
211 # Validar comisión
212 if not (0 <= commission_pct < 1):
213 raise ValidationException(
214 "commission_pct",
215 "Commission percent must be between 0 and 1"
216 )
217
218 # Validar position sizing
219 if position_sizing not in ['risk', 'fixed']:
220 raise ValidationException(
221 "position_sizing",
222 "Position sizing must be 'risk' or 'fixed'"
223 )
224
225 return True
226
227 @staticmethod
228 def validate_symbol(symbol: str) -> bool:
229 """Valida símbolo"""
230 if not symbol:
231 raise ValidationException(
232 "symbol",
233 "Symbol cannot be empty"
234 )
235
236 if len(symbol) > 20:
237 raise ValidationException(
238 "symbol",
239 "Symbol too long (max 20 chars)"
240 )
241
242 if not symbol.replace('-', '').replace('/', '').isalnum():
243 raise ValidationException(
244 "symbol",
245 "Symbol contains invalid characters"
246 )
247
248 return True
249
250 @staticmethod
251 def validate_date_range(start_date: datetime, end_date: datetime) -> bool:
252 """Valida rango de fechas"""
253 if start_date >= end_date:
254 raise ValidationException(
255 "date_range",
256 "Start date must be before end date"
257 )
258
259 if (end_date - start_date).days > 10 * 365:
260 raise ValidationException(
261 "date_range",
262 "Date range cannot exceed 10 years"
263 )
264
265 return True
266
267
268# ============================================================================
269# 3. DECORADOR CON TIMEOUT
270# ============================================================================
271
272# utils/timeout.py
273"""Decoradores para manejar timeouts"""
274
275import asyncio
276from functools import wraps
277from typing import Callable, Any, Optional
278from utils.exceptions import TimeoutException
279from utils.logger import setup_logger
280
281logger = setup_logger(__name__)
282
283
284def async_timeout(seconds: float):
285 """
286 Decorador para async functions con timeout
287
288 Uso:
289 @async_timeout(30)
290 async def fetch_data(symbol):
291 ...
292 """
293 def decorator(func: Callable) -> Callable:
294 @wraps(func)
295 async def wrapper(*args, **kwargs) -> Any:
296 operation = f"{func.__module__}.{func.__name__}"
297 try:
298 return await asyncio.wait_for(
299 func(*args, **kwargs),
300 timeout=seconds
301 )
302 except asyncio.TimeoutError:
303 logger.error(f"Operation '{operation}' timed out after {seconds}s")
304 raise TimeoutException(operation, seconds)
305
306 return wrapper
307 return decorator
308
309
310def sync_timeout(seconds: float):
311 """
312 Decorador para funciones síncronas con timeout
313
314 Uso:
315 @sync_timeout(30)
316 def fetch_data(symbol):
317 ...
318 """
319 def decorator(func: Callable) -> Callable:
320 @wraps(func)
321 def wrapper(*args, **kwargs) -> Any:
322 operation = f"{func.__module__}.{func.__name__}"
323
324 # Para operaciones I/O, usar multiprocessing
325 import signal
326
327 def timeout_handler(signum, frame):
328 raise TimeoutException(operation, seconds)
329
330 # Solo funciona en Unix
331 try:
332 signal.signal(signal.SIGALRM, timeout_handler)
333 signal.alarm(int(seconds))
334 result = func(*args, **kwargs)
335 signal.alarm(0) # Cancelar alarm
336 return result
337 except Exception as e:
338 signal.alarm(0)
339 if isinstance(e, TimeoutException):
340 raise
341 raise
342
343 return wrapper
344 return decorator
345
346
347# ============================================================================
348# 4. CONTROL DE CONCURRENCIA
349# ============================================================================
350
351# api/backtest_queue.py
352"""Cola de backtests con control de concurrencia"""
353
354from asyncio import Semaphore, Lock
355from uuid import uuid4
356from enum import Enum
357from datetime import datetime
358from typing import Optional, Dict, List
359from dataclasses import dataclass, field
360import logging
361
362logger = logging.getLogger(__name__)
363
364
365class JobStatus(Enum):
366 """Estados de un job de backtest"""
367 QUEUED = "queued"
368 RUNNING = "running"
369 COMPLETED = "completed"
370 FAILED = "failed"
371 TIMEOUT = "timeout"
372
373
374@dataclass
375class BacktestJob:
376 """Representa un job de backtest"""
377 id: str = field(default_factory=lambda: str(uuid4())[:8])
378 status: JobStatus = JobStatus.QUEUED
379 request: Optional[Dict] = None
380 result: Optional[Dict] = None
381 error: Optional[str] = None
382 created_at: datetime = field(default_factory=datetime.now)
383 started_at: Optional[datetime] = None
384 completed_at: Optional[datetime] = None
385
386 @property
387 def duration_seconds(self) -> Optional[float]:
388 if self.completed_at and self.started_at:
389 return (self.completed_at - self.started_at).total_seconds()
390 return None
391
392 def to_dict(self) -> Dict:
393 return {
394 'id': self.id,
395 'status': self.status.value,
396 'created_at': self.created_at.isoformat(),
397 'started_at': self.started_at.isoformat() if self.started_at else None,
398 'completed_at': self.completed_at.isoformat() if self.completed_at else None,
399 'duration_seconds': self.duration_seconds,
400 'error': self.error,
401 'result': self.result
402 }
403
404
405class BacktestQueue:
406 """Gestor de cola de backtests con control de concurrencia"""
407
408 def __init__(self, max_concurrent: int = 3):
409 self.max_concurrent = max_concurrent
410 self.semaphore = Semaphore(max_concurrent)
411 self.jobs: Dict[str, BacktestJob] = {}
412 self.lock = Lock()
413
414 async def submit(self, request: Dict) -> str:
415 """Envía un job a la cola"""
416 job = BacktestJob(request=request)
417
418 async with self.lock:
419 self.jobs[job.id] = job
420
421 logger.info(f"Job {job.id} submitted for {request.get('symbol')}")
422 return job.id
423
424 async def execute(self, job_id: str, executor_func) -> BacktestJob:
425 """
426 Ejecuta un job con control de concurrencia
427
428 executor_func: función async que ejecuta el backtest
429 """
430 job = self.jobs[job_id]
431
432 async with self.semaphore: # Limita concurrencia
433 try:
434 job.status = JobStatus.RUNNING
435 job.started_at = datetime.now()
436
437 logger.info(f"Job {job_id} started")
438
439 # Ejecutar con timeout
440 try:
441 job.result = await asyncio.wait_for(
442 executor_func(job.request),
443 timeout=300 # 5 minutos máximo
444 )
445 job.status = JobStatus.COMPLETED
446 logger.info(f"Job {job_id} completed successfully")
447
448 except asyncio.TimeoutError:
449 job.status = JobStatus.TIMEOUT
450 job.error = "Backtest exceeded 5 minute timeout"
451 logger.error(f"Job {job_id} timeout")
452
453 except Exception as e:
454 job.status = JobStatus.FAILED
455 job.error = str(e)
456 logger.error(f"Job {job_id} failed: {e}", exc_info=True)
457
458 finally:
459 job.completed_at = datetime.now()
460
461 return job
462
463 async def get_status(self, job_id: str) -> Optional[BacktestJob]:
464 """Obtiene estado de un job"""
465 return self.jobs.get(job_id)
466
467 async def get_jobs(self, status: Optional[JobStatus] = None) -> List[BacktestJob]:
468 """Lista jobs, opcionalmente filtrados por estado"""
469 jobs = self.jobs.values()
470 if status:
471 jobs = [j for j in jobs if j.status == status]
472 return sorted(jobs, key=lambda j: j.created_at, reverse=True)
473
474
475# Instancia global
476backtest_queue = BacktestQueue(max_concurrent=3)
477
478
479# ============================================================================
480# 5. CACHÉ OPTIMIZADO
481# ============================================================================
482
483# utils/cache_optimized.py
484"""Caché mejorado para performance"""
485
486import hashlib
487import pickle
488import logging
489from typing import Callable, Any, Optional
490from functools import wraps
491from datetime import timedelta
492import diskcache
493from config import config
494
495logger = logging.getLogger(__name__)
496
497
498class OptimizedCache:
499 """Caché optimizado con keys más eficientes"""
500
501 def __init__(self, cache_dir: str):
502 self.cache = diskcache.Cache(cache_dir)
503 self._hits = 0
504 self._misses = 0
505
506 def _generate_smart_key(self, func_name: str, *args, **kwargs) -> str:
507 """
508 Genera key de caché sin serializar objetos grandes
509
510 Para DataFrames: usa hash de shape + columns + dtypes
511 Para escalares: usa directamente el valor
512 """
513 key_parts = [func_name]
514
515 # Procesar args posicionales
516 for i, arg in enumerate(args):
517 if arg is None:
518 key_parts.append(f"arg{i}:None")
519 elif isinstance(arg, (int, float, str, bool)):
520 key_parts.append(f"arg{i}:{arg}")
521 elif isinstance(arg, (list, tuple)):
522 # Solo usar longitud y tipos
523 key_parts.append(
524 f"arg{i}:seq_len{len(arg)}_"
525 f"types_{'_'.join(type(x).__name__ for x in arg[:3])}"
526 )
527 else:
528 # Para otros tipos: usar type name + id
529 key_parts.append(f"arg{i}:{type(arg).__name__}")
530
531 # Procesar kwargs
532 for k, v in sorted(kwargs.items()):
533 if isinstance(v, (int, float, str, bool)):
534 key_parts.append(f"{k}:{v}")
535 else:
536 key_parts.append(f"{k}:{type(v).__name__}")
537
538 # Generar hash
539 key_str = ":".join(key_parts)
540 return hashlib.md5(key_str.encode()).hexdigest()
541
542 def get(self, key: str) -> Optional[Any]:
543 """Obtiene valor de caché"""
544 try:
545 value = self.cache.get(key)
546 if value is not None:
547 self._hits += 1
548 logger.debug(f"Cache HIT: {key}")
549 return value
550 except Exception as e:
551 logger.warning(f"Cache read error: {e}")
552 return None
553
554 def set(self, key: str, value: Any, expire_hours: int = 24):
555 """Almacena en caché"""
556 try:
557 expire_seconds = expire_hours * 3600
558 self.cache.set(key, value, expire=expire_seconds)
559 logger.debug(f"Cache SET: {key} (expires in {expire_hours}h)")
560 except Exception as e:
561 logger.warning(f"Cache write error: {e}")
562
563 def cached(self, expire_hours: int = 24):
564 """
565 Decorador para cachear función
566
567 @cached(expire_hours=24)
568 def fetch_data(symbol: str):
569 return data
570 """
571 def decorator(func: Callable) -> Callable:
572 @wraps(func)
573 def wrapper(*args, **kwargs) -> Any:
574 if not config.ENABLE_CACHE:
575 return func(*args, **kwargs)
576
577 key = self._generate_smart_key(func.__name__, *args, **kwargs)
578
579 # Intentar caché
580 cached_value = self.get(key)
581 if cached_value is not None:
582 return cached_value
583
584 # Cache miss - ejecutar función
585 self._misses += 1
586 result = func(*args, **kwargs)
587
588 # Almacenar resultado
589 self.set(key, result, expire_hours)
590
591 return result
592
593 return wrapper
594 return decorator
595
596 def clear(self):
597 """Limpia todo el caché"""
598 self.cache.clear()
599 logger.info("Cache cleared")
600
601 def get_stats(self) -> Dict[str, Any]:
602 """Obtiene estadísticas de caché"""
603 total = self._hits + self._misses
604 hit_rate = (self._hits / total * 100) if total > 0 else 0
605
606 return {
607 'hits': self._hits,
608 'misses': self._misses,
609 'hit_rate': f"{hit_rate:.1f}%",
610 'total_requests': total,
611 'size_mb': self.cache.volume() / (1024 * 1024),
612 'entry_count': len(self.cache)
613 }
614
615
616# Instancia global
617optimized_cache = OptimizedCache(str(config.CACHE_DIR))
618cached = optimized_cache.cached
619
620
621# ============================================================================
622# 6. EJEMPLO DE USO EN API
623# ============================================================================
624
625# api/routes_improved.py
626"""Rutas mejoradas con manejo de errores y validación"""
627
628from fastapi import APIRouter, HTTPException
629from typing import Optional
630from backtesting_app.core.backtester import Backtester
631from utils.validators import InputValidator
632from utils.exceptions import (
633 BacktestingException, ValidationException, DataLoadException
634)
635from utils.timeout import async_timeout
636from api.backtest_queue import backtest_queue
637
638router = APIRouter()
639
640
641@router.post("/backtest")
642@async_timeout(300) # 5 minutos máximo
643async def run_backtest(request: BacktestRequest):
644 """
645 Ejecuta un backtest con validación, timeout y manejo de errores
646 """
647 try:
648 # 1. VALIDAR ENTRADA
649 InputValidator.validate_symbol(request.symbol)
650 InputValidator.validate_backtest_params(
651 request.capital,
652 request.risk_percent,
653 request.commission_pct,
654 request.position_sizing
655 )
656 InputValidator.validate_date_range(
657 request.start_date,
658 request.end_date
659 )
660
661 # 2. CARGAR DATOS
662 logger.info(f"Loading data for {request.symbol} ({request.timeframe})")
663 try:
664 df = await load_market_data(
665 request.symbol,
666 request.start_date,
667 request.end_date,
668 request.timeframe
669 )
670 except Exception as e:
671 raise DataLoadException(request.symbol, str(e))
672
673 # 3. VALIDAR DATOS
674 InputValidator.validate_dataframe(df, request.symbol)
675
676 # 4. EJECUTAR BACKTEST
677 logger.info(f"Running backtest for {request.symbol}")
678 backtester = Backtester(
679 algorithm=get_algorithm(request.algorithm),
680 initial_capital=request.capital,
681 risk_percent=request.risk_percent
682 )
683
684 # Validar parámetros del algoritmo
685 algo = backtester.algorithm
686 default_params = algo.get_default_params()
687 algo_params = request.params or default_params
688 if not algo.validate_params(algo_params):
689 raise ValidationException(
690 "algorithm_params",
691 f"Invalid params for {algo.name}"
692 )
693
694 result = backtester.run(
695 df,
696 algo_params,
697 symbol=request.symbol,
698 timeframe=request.timeframe
699 )
700
701 # 5. RETORNAR RESULTADO
702 logger.info(
703 f"Backtest completed: {result.total_trades} trades, "
704 f"P&L: {result.total_return_percent:.2f}%"
705 )
706
707 return {
708 "status": "success",
709 "result": result.to_dict()
710 }
711
712 except ValidationException as e:
713 logger.warning(f"Validation error: {e}")
714 raise HTTPException(
715 status_code=400,
716 detail=f"Validation error: {e}"
717 )
718
719 except DataLoadException as e:
720 logger.error(f"Data load error: {e}")
721 raise HTTPException(
722 status_code=503,
723 detail=f"Failed to load market data: {e}"
724 )
725
726 except BacktestingException as e:
727 logger.error(f"Backtesting error: {e}")
728 raise HTTPException(
729 status_code=400,
730 detail=f"Backtesting error: {e}"
731 )
732
733 except Exception as e:
734 logger.error(f"Unexpected error: {e}", exc_info=True)
735 raise HTTPException(
736 status_code=500,
737 detail="Internal server error"
738 )
739
740
741@router.post("/backtest/async")
742async def run_backtest_async(request: BacktestRequest):
743 """
744 Ejecuta backtest de forma asincrónica con fila de tareas
745 """
746 try:
747 # Validar input
748 InputValidator.validate_symbol(request.symbol)
749 InputValidator.validate_backtest_params(
750 request.capital,
751 request.risk_percent,
752 request.commission_pct,
753 request.position_sizing
754 )
755
756 # Enviar a cola
757 job_id = await backtest_queue.submit(request.dict())
758
759 logger.info(f"Backtest job {job_id} queued for {request.symbol}")
760
761 return {
762 "job_id": job_id,
763 "status": "queued",
764 "message": f"Backtest queued with ID {job_id}"
765 }
766
767 except ValidationException as e:
768 raise HTTPException(status_code=400, detail=str(e))
769 except Exception as e:
770 logger.error(f"Error queuing backtest: {e}")
771 raise HTTPException(status_code=500, detail=str(e))
772
773
774@router.get("/backtest/{job_id}")
775async def get_backtest_status(job_id: str):
776 """Obtiene estado de un backtest asincrónico"""
777 job = await backtest_queue.get_status(job_id)
778
779 if not job:
780 raise HTTPException(status_code=404, detail=f"Job {job_id} not found")
781
782 return job.to_dict()
783