dam/Trading-Bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Archivos añadidos/actuaizados para comentar el bot con ChatGPT

Browse Source
This commit is contained in:
DaM
2026-01-28 09:42:16 +01:00
parent 9ab1f7fadc
commit 1add69eb56
5 changed files with 353 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -13,4 +13,5 @@ __pycache__/
# Logs # Logs
logs/ logs/
# Archivos temporales # Resultados
backtest_results/

219
src/backtest/optimizer.py
View File

@@ -2,6 +2,7 @@
""" """
Optimizador de parámetros para estrategias Optimizador de parámetros para estrategias
""" """
import os
import pandas as pd import pandas as pd
from typing import Dict, List, Any, Type from typing import Dict, List, Any, Type
from itertools import product from itertools import product
@@ -18,18 +19,24 @@ class ParameterOptimizer:
strategy_class: Type[Strategy], strategy_class: Type[Strategy],
data: pd.DataFrame, data: pd.DataFrame,
initial_capital: float = 10000, initial_capital: float = 10000,
commission: float = 0.001): commission: float = 0.001,
slippage: float = 0.0005,
position_size: float = 0.95):
""" """
Args: Args:
strategy_class: Clase de estrategia (no instancia) strategy_class: Clase de estrategia (no instancia)
data: Datos para backtest data: Datos para backtest
initial_capital: Capital inicial initial_capital: Capital inicial
commission: Comisión por trade commission: Comisión por trade
slippage: Slippage simulado
position_size: Fracción del capital por trade
""" """
self.strategy_class = strategy_class self.strategy_class = strategy_class
self.data = data self.data = data
self.initial_capital = initial_capital self.initial_capital = initial_capital
self.commission = commission self.commission = commission
self.slippage = slippage
self.position_size = position_size
self.results: List[Dict] = [] self.results: List[Dict] = []
@@ -46,21 +53,29 @@ class ParameterOptimizer:
} }
Returns: Returns:
DataFrame con resultados ordenados por retorno DataFrame con resultados (sin ordenar)
""" """
# Limpiar resultados previos
self.results = []
# Generar todas las combinaciones posibles # Generar todas las combinaciones posibles
param_names = list(param_grid.keys()) param_names = list(param_grid.keys())
param_values = list(param_grid.values()) param_values = list(param_grid.values())
combinations = list(product(*param_values)) combinations = list(product(*param_values))
total_tests = len(combinations) total_tests = len(combinations)
log.info(f"Iniciando optimización: {total_tests} combinaciones a probar") log.info(f"🔧 Iniciando optimización: {total_tests} combinaciones")
# Probar cada combinación # Probar cada combinación
successful = 0
failed = 0
for i, values in enumerate(combinations, 1): for i, values in enumerate(combinations, 1):
params = dict(zip(param_names, values)) params = dict(zip(param_names, values))
log.debug(f"[{i}/{total_tests}] Probando: {params}") # Mostrar progreso cada 10 tests o en el primero
if i % 10 == 0 or i == 1 or i == total_tests:
log.info(f" [{i}/{total_tests}] Probando: {params}")
try: try:
# Crear estrategia con estos parámetros # Crear estrategia con estos parámetros
@@ -70,7 +85,9 @@ class ParameterOptimizer:
engine = BacktestEngine( engine = BacktestEngine(
strategy=strategy, strategy=strategy,
initial_capital=self.initial_capital, initial_capital=self.initial_capital,
commission=self.commission commission=self.commission,
slippage=self.slippage,
position_size=self.position_size
) )
results = engine.run(self.data) results = engine.run(self.data)
@@ -84,30 +101,57 @@ class ParameterOptimizer:
'total_trades': results['total_trades'], 'total_trades': results['total_trades'],
'win_rate_pct': results['win_rate_pct'], 'win_rate_pct': results['win_rate_pct'],
'profit_factor': results['profit_factor'], 'profit_factor': results['profit_factor'],
'final_equity': results['final_equity'],
} }
self.results.append(result_entry) self.results.append(result_entry)
successful += 1
except Exception as e: except Exception as e:
log.error(f"Error con parámetros {params}: {e}") log.error(f"❌ Error con {params}: {e}")
failed += 1
continue continue
# Convertir a DataFrame # Convertir a DataFrame
if not self.results:
log.error("❌ No se obtuvieron resultados válidos")
return pd.DataFrame()
df_results = pd.DataFrame(self.results) df_results = pd.DataFrame(self.results)
# Ordenar por retorno (mejor primero) log.success(f"✅ Optimización completa: {successful} exitosos, {failed} fallidos")
df_results = df_results.sort_values('total_return_pct', ascending=False)
log.success(f"Optimización completa: {len(df_results)} resultados válidos") # Mostrar top 5 por Sharpe Ratio
self._print_top_results(df_results, param_names, metric='sharpe_ratio', top_n=5)
return df_results return df_results
def get_best_params(self, metric: str = 'total_return_pct') -> Dict: def _print_top_results(self, df: pd.DataFrame, param_names: List[str],
metric: str = 'sharpe_ratio', top_n: int = 5):
"""
Muestra los mejores resultados según una métrica
"""
# Ordenar por la métrica (descendente para la mayoría)
ascending = True if metric in ['max_drawdown_pct'] else False
df_sorted = df.sort_values(metric, ascending=ascending)
log.info(f"\n🏆 TOP {top_n} POR {metric.upper()}:")
for i, (idx, row) in enumerate(df_sorted.head(top_n).iterrows(), 1):
param_str = ", ".join([f"{k}={row[k]}" for k in param_names])
log.info(f" #{i}: {param_str}")
log.info(f" → {metric}: {row[metric]:.2f}, "
f"Return: {row['total_return_pct']:.2f}%, "
f"Trades: {int(row['total_trades'])}, "
f"Win Rate: {row['win_rate_pct']:.1f}%")
def get_best_params(self, metric: str = 'sharpe_ratio') -> Dict:
""" """
Obtiene los mejores parámetros según una métrica Obtiene los mejores parámetros según una métrica
Args: Args:
metric: Métrica a optimizar ('total_return_pct', 'sharpe_ratio', etc) metric: Métrica a optimizar
('sharpe_ratio', 'total_return_pct', 'profit_factor', etc)
Returns: Returns:
Diccionario con los mejores parámetros Diccionario con los mejores parámetros
@@ -117,9 +161,8 @@ class ParameterOptimizer:
df_results = pd.DataFrame(self.results) df_results = pd.DataFrame(self.results)
# Ordenar por la métrica elegida # Para drawdown queremos el MENOR (menos negativo = más cercano a 0)
if metric in ['max_drawdown_pct']: if metric in ['max_drawdown_pct']:
# Para drawdown, queremos el MENOR (menos negativo)
best_idx = df_results[metric].idxmax() best_idx = df_results[metric].idxmax()
else: else:
# Para otras métricas, queremos el MAYOR # Para otras métricas, queremos el MAYOR
@@ -127,69 +170,97 @@ class ParameterOptimizer:
best_result = df_results.loc[best_idx] best_result = df_results.loc[best_idx]
# Extraer solo los parámetros (no las métricas) # Extraer solo parámetros (no métricas)
param_names = [col for col in df_results.columns metric_cols = ['total_return_pct', 'sharpe_ratio', 'max_drawdown_pct',
if col not in ['total_return_pct', 'sharpe_ratio', 'total_trades', 'win_rate_pct', 'profit_factor', 'final_equity']
'max_drawdown_pct', 'total_trades', param_names = [col for col in df_results.columns if col not in metric_cols]
'win_rate_pct', 'profit_factor']]
best_params = {param: best_result[param] for param in param_names} best_params = {param: best_result[param] for param in param_names}
log.info(f"Mejores parámetros según {metric}: {best_params}") log.info(f"\n🎯 MEJORES PARÁMETROS SEGÚN {metric.upper()}:")
log.info(f" {metric}: {best_result[metric]:.2f}") log.info(f" Parámetros: {best_params}")
log.info(f" {metric}: {best_result[metric]:.2f}")
log.info(f" Retorno: {best_result['total_return_pct']:.2f}%")
log.info(f" Sharpe: {best_result['sharpe_ratio']:.2f}")
log.info(f" Max DD: {best_result['max_drawdown_pct']:.2f}%")
log.info(f" Trades: {int(best_result['total_trades'])}")
return best_params return best_params
# ============================================================================ def get_top_n_params(self, metric: str = 'sharpe_ratio', n: int = 5) -> List[Dict]:
# Ejemplo de Uso """
# ============================================================================ Obtiene los N mejores conjuntos de parámetros
""" Args:
from src.data.storage import StorageManager metric: Métrica a optimizar
from src.strategies.moving_average import MovingAverageCrossover n: Número de mejores configuraciones
from src.backtest.optimizer import ParameterOptimizer
Returns:
# Cargar datos Lista de diccionarios con parámetros
storage = StorageManager(...) """
data = storage.load_ohlcv('BTC/USDT', '1h') if not self.results:
raise ValueError("No hay resultados. Ejecuta optimize() primero.")
# Crear optimizador
optimizer = ParameterOptimizer( df_results = pd.DataFrame(self.results)
strategy_class=MovingAverageCrossover, # Clase, no instancia
data=data, # Ordenar
initial_capital=10000, ascending = True if metric in ['max_drawdown_pct'] else False
commission=0.001 df_sorted = df_results.sort_values(metric, ascending=ascending)
)
# Extraer parámetros
# Definir parámetros a probar metric_cols = ['total_return_pct', 'sharpe_ratio', 'max_drawdown_pct',
param_grid = { 'total_trades', 'win_rate_pct', 'profit_factor', 'final_equity']
'fast_period': [5, 10, 15, 20, 25], param_names = [col for col in df_results.columns if col not in metric_cols]
'slow_period': [30, 50, 70, 100, 150],
'ma_type': ['sma', 'ema'] top_params = []
} for idx, row in df_sorted.head(n).iterrows():
params = {param: row[param] for param in param_names}
# Ejecutar optimización (probará 5 × 5 × 2 = 50 combinaciones) top_params.append(params)
results_df = optimizer.optimize(param_grid)
return top_params
# Ver mejores resultados
print(results_df.head(10)) def save_results(self, filename: str = 'backtest_results/optimization_results.csv'):
"""
# Obtener mejores parámetros Guarda resultados en CSV
best_params = optimizer.get_best_params(metric='sharpe_ratio')
print(f"Mejores parámetros: {best_params}") Args:
filename: Nombre del archivo (puede incluir ruta)
# Usar los mejores parámetros """
best_strategy = MovingAverageCrossover(**best_params) if not self.results:
``` log.warning("⚠️ No hay resultados para guardar")
return
---
# Crear directorio si no existe
## 📊 Output esperado: os.makedirs(os.path.dirname(filename), exist_ok=True)
```
fast_period slow_period ma_type total_return_pct sharpe_ratio max_drawdown_pct df = pd.DataFrame(self.results)
0 15 50 ema 45.20 2.10 -12.30
1 10 30 sma 42.80 1.95 -15.20 # Ordenar por Sharpe Ratio antes de guardar
2 20 70 ema 38.50 1.85 -14.10 df = df.sort_values('sharpe_ratio', ascending=False)
3 5 30 sma 35.20 1.75 -18.50
... df.to_csv(filename, index=False)
""" log.success(f"💾 Resultados guardados en: {filename}")
log.info(f" Total filas: {len(df)}")
def compare_metrics(self, metrics: List[str] = None):
"""
Compara mejores parámetros según diferentes métricas
Args:
metrics: Lista de métricas a comparar
"""
if not self.results:
raise ValueError("No hay resultados. Ejecuta optimize() primero.")
if metrics is None:
metrics = ['sharpe_ratio', 'total_return_pct', 'profit_factor', 'max_drawdown_pct']
log.info("\n📊 COMPARACIÓN DE MÉTRICAS:")
log.info("="*70)
for metric in metrics:
try:
best_params = self.get_best_params(metric)
log.info("") # Línea en blanco para separar
except Exception as e:
log.error(f"Error con métrica {metric}: {e}")

20
src/backtest/visualizer.py
View File

@@ -262,18 +262,18 @@ class BacktestVisualizer:
MÉTRICAS DE PERFORMANCE MÉTRICAS DE PERFORMANCE
━━━━━━━━━━━━━━━━━━━━━━━━━ ━━━━━━━━━━━━━━━━━━━━━━━━━
💰 Capital Inicial: ${self.results['initial_capital']:,.0f} + Capital Inicial: ${self.results['initial_capital']:,.0f}
💰 Capital Final: ${self.results['final_equity']:,.0f} + Capital Final: ${self.results['final_equity']:,.0f}
📈 Retorno Total: {self.results['total_return_pct']:.2f}% ↑ Retorno Total: {self.results['total_return_pct']:.2f}%
📊 Total Trades: {self.results['total_trades']} · Total Trades: {self.results['total_trades']}
✅ Trades Ganadores: {self.results['winning_trades']} ✓ Trades Ganadores: {self.results['winning_trades']}
❌ Trades Perdedores: {self.results['losing_trades']} × Trades Perdedores: {self.results['losing_trades']}
🎯 Win Rate: {self.results['win_rate_pct']:.2f}% ◎ Win Rate: {self.results['win_rate_pct']:.2f}%
📉 Max Drawdown: {self.results['max_drawdown_pct']:.2f}% ↓ Max Drawdown: {self.results['max_drawdown_pct']:.2f}%
📈 Sharpe Ratio: {self.results['sharpe_ratio']:.2f} · Sharpe Ratio: {self.results['sharpe_ratio']:.2f}
💵 Profit Factor: {self.results['profit_factor']:.2f} $ Profit Factor: {self.results['profit_factor']:.2f}
""" """
ax4.text(0.1, 0.5, metrics_text, fontsize=11, ax4.text(0.1, 0.5, metrics_text, fontsize=11,

100
tests/test_optimizer.py Normal file
View File

@@ -0,0 +1,100 @@
# test_optimizer.py
"""
Script para probar el optimizador de parámetros
"""
import os
import sys
from dotenv import load_dotenv
from pathlib import Path
from datetime import datetime, timedelta
# Añadir raíz del proyecto al path
sys.path.insert(0, str(Path(__file__).parent.parent))
from src.utils.logger import log
from src.data.storage import StorageManager
from src.strategies import MovingAverageCrossover
from src.backtest.optimizer import ParameterOptimizer
def setup_environment():
"""Carga variables de entorno"""
env_path = Path(__file__).parent.parent / 'config' / 'secrets.env'
load_dotenv(dotenv_path=env_path)
def test_optimizer():
"""
Prueba el optimizador con Moving Average Crossover
"""
log.info("="*70)
log.info("🔧 TEST: OPTIMIZADOR DE PARÁMETROS")
log.info("="*70)
# Setup
setup_environment()
# Cargar datos
storage = StorageManager(
db_host=os.getenv('DB_HOST'),
db_port=int(os.getenv('DB_PORT', 5432)),
db_name=os.getenv('DB_NAME'),
db_user=os.getenv('DB_USER'),
db_password=os.getenv('DB_PASSWORD'),
)
log.info("\n📥 Cargando datos...")
end_date = datetime.now()
start_date = end_date - timedelta(days=60)
data = storage.load_ohlcv(
symbol='BTC/USDT',
timeframe='1h',
start_date=start_date,
end_date=end_date,
use_cache=False
)
log.success(f"✓ Datos cargados: {len(data)} velas")
# Crear optimizador
log.info("\n🔧 Creando optimizador...")
optimizer = ParameterOptimizer(
strategy_class=MovingAverageCrossover,
data=data,
initial_capital=10000,
commission=0.001
)
# Definir parámetros a probar (pequeño para empezar)
param_grid = {
'fast_period': [5, 10, 15],
'slow_period': [30, 50],
'ma_type': ['sma', 'ema']
}
log.info(f"\n📊 Grid de parámetros:")
log.info(f" fast_period: {param_grid['fast_period']}")
log.info(f" slow_period: {param_grid['slow_period']}")
log.info(f" ma_type: {param_grid['ma_type']}")
log.info(f" Total combinaciones: {3 * 2 * 2} = 12")
# Optimizar
log.info("\n🚀 Iniciando optimización...")
results_df = optimizer.optimize(param_grid)
# Mostrar resultados
log.info("\n📊 RESULTADOS COMPLETOS:")
print(results_df.to_string(index=False))
# Mejores parámetros
log.info("\n🏆 ANÁLISIS:")
best_params = optimizer.get_best_params(metric='sharpe_ratio')
# Guardar resultados (OPCIONAL - ya se guarda por defecto)
optimizer.save_results()
storage.close()
log.success("\n✅ TEST COMPLETADO")
if __name__ == "__main__":
test_optimizer()

96
tests/test_visualizer.py Normal file
View File

@@ -0,0 +1,96 @@
# test_visualizer.py
"""
Script para probar las visualizaciones
"""
import os
import sys
from dotenv import load_dotenv
from pathlib import Path
from datetime import datetime, timedelta
# Añadir raíz del proyecto al path
sys.path.insert(0, str(Path(__file__).parent.parent))
from src.utils.logger import log
from src.data.storage import StorageManager
from src.strategies import MovingAverageCrossover
from src.backtest import BacktestEngine
from src.backtest.visualizer import BacktestVisualizer
def setup_environment():
"""Carga variables de entorno"""
env_path = Path(__file__).parent.parent / 'config' / 'secrets.env'
load_dotenv(dotenv_path=env_path)
def test_visualizer():
"""
Prueba las visualizaciones con un backtest
"""
log.info("="*70)
log.info("📊 TEST: VISUALIZACIONES")
log.info("="*70)
# Setup
setup_environment()
# Cargar datos
storage = StorageManager(
db_host=os.getenv('DB_HOST'),
db_port=int(os.getenv('DB_PORT', 5432)),
db_name=os.getenv('DB_NAME'),
db_user=os.getenv('DB_USER'),
db_password=os.getenv('DB_PASSWORD'),
)
log.info("\n📥 Cargando datos...")
end_date = datetime.now()
start_date = end_date - timedelta(days=60)
data = storage.load_ohlcv(
symbol='BTC/USDT',
timeframe='1h',
start_date=start_date,
end_date=end_date,
use_cache=False
)
log.success(f"✓ Datos cargados: {len(data)} velas")
# Ejecutar backtest
log.info("\n🧪 Ejecutando backtest...")
strategy = MovingAverageCrossover(fast_period=15, slow_period=50, ma_type='sma')
engine = BacktestEngine(
strategy=strategy,
initial_capital=10000,
commission=0.001,
position_size=0.95
)
results = engine.run(data)
log.info(f" Retorno: {results['total_return_pct']:.2f}%")
log.info(f" Trades: {results['total_trades']}")
# Crear visualizador
log.info("\n📊 Generando visualizaciones...")
viz = BacktestVisualizer(results, data)
# Generar todos los gráficos
viz.generate_all_plots('backtest_results')
log.info("\n💡 Los gráficos se guardaron en: backtest_results/")
log.info(" Archivos generados:")
log.info(" - equity_curve.png")
log.info(" - drawdown.png")
log.info(" - returns_distribution.png")
log.info(" - trades_chart.png")
log.info(" - dashboard.png")
storage.close()
log.success("\n✅ TEST COMPLETADO")
log.info("\n👀 Abre la carpeta 'backtest_results/' para ver los gráficos")
if __name__ == "__main__":
test_visualizer()