Trading-Bot/tests/test_walkforwad.py

# tests/test_walkforward.py
"""
Script para probar Walk-Forward Validation
Guarda resultados en CSV para análisis posterior
"""
import os
import sys
from dotenv import load_dotenv
from pathlib import Path
import pandas as pd

# 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.core.walk_forward import WalkForwardValidator


def setup_environment():
    """Carga variables de entorno"""
    env_path = Path(__file__).parent.parent / "config" / "secrets.env"
    load_dotenv(dotenv_path=env_path)
    log.success("✓ Variables de entorno cargadas")


def test_walkforward():
    """
    Test de Walk-Forward Validation con múltiples configuraciones
    """
    log.info("=" * 70)
    log.info("🪜 TEST: WALK-FORWARD VALIDATION (MULTI CONFIG)")
    log.info("=" * 70)

    # --------------------------------------------------
    # Setup
    # --------------------------------------------------
    setup_environment()

    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...")
    data = storage.load_ohlcv(
        symbol="BTC/USDT",
        timeframe="1h",
        start_date=None,
        end_date=None,
        use_cache=False,
    )

    log.success(f"✓ Datos cargados: {len(data)} velas")

    # --------------------------------------------------
    # Grid de parámetros de la estrategia
    # --------------------------------------------------
    param_grid = {
        "fast_period": [10, 15],
        "slow_period": [30, 50],
        "ma_type": ["sma"],
        "use_adx": [True],
        "adx_threshold": [20, 25, 30],
    }

    # --------------------------------------------------
    # Configuraciones Walk-Forward a comparar
    # --------------------------------------------------
    wf_configs = [
        {"name": "WF_12_3", "train_days": 365, "test_days": 90},
        {"name": "WF_24_3", "train_days": 365 * 2, "test_days": 90},
        {"name": "WF_24_6", "train_days": 365 * 2, "test_days": 180},
    ]

    all_windows = []
    summary_rows = []

    # --------------------------------------------------
    # Ejecutar Walk-Forward por configuración
    # --------------------------------------------------
    for cfg in wf_configs:
        log.info("\n" + "=" * 70)
        log.info(f"🧪 EJECUTANDO {cfg['name']}")
        log.info("=" * 70)

        wf = WalkForwardValidator(
            strategy_class=MovingAverageCrossover,
            param_grid=param_grid,
            data=data,
            train_window=pd.Timedelta(days=cfg["train_days"]),
            test_window=pd.Timedelta(days=cfg["test_days"]),
            initial_capital=10_000,
            commission=0.001,
            slippage=0.0005,
            position_size=0.95,
            optimizer_metric="sharpe_ratio",
        )

        wf_result = wf.run()

        # -------------------------------
        # Validaciones básicas
        # -------------------------------
        assert isinstance(wf_result, dict), "wf_result debe ser dict"
        assert "windows" in wf_result, "wf_result debe contener 'windows'"

        df_windows = wf_result["windows"].copy()
        df_windows["wf_name"] = cfg["name"]

        all_windows.append(df_windows)

        # -------------------------------
        # Métricas agregadas por WF
        # -------------------------------
        summary_rows.append({
            "wf_name": cfg["name"],
            "train_days": cfg["train_days"],
            "test_days": cfg["test_days"],
            "windows": len(df_windows),
            "avg_return_pct": df_windows["return_pct"].mean(),
            "avg_sharpe": df_windows["sharpe"].mean(),
            "avg_max_dd_pct": df_windows["max_dd_pct"].mean(),
            "avg_trades": df_windows["trades"].mean(),
        })

    # --------------------------------------------------
    # Consolidar resultados
    # --------------------------------------------------
    df_all_windows = pd.concat(all_windows, ignore_index=True)
    df_summary = pd.DataFrame(summary_rows)

    # --------------------------------------------------
    # Guardar CSVs
    # --------------------------------------------------
    output_dir = Path("backtest_results/walkforward")
    output_dir.mkdir(parents=True, exist_ok=True)

    windows_path = output_dir / "walkforward_windows.csv"
    summary_path = output_dir / "walkforward_summary.csv"

    df_all_windows.to_csv(windows_path, index=False)
    df_summary.to_csv(summary_path, index=False)

    log.success(f"💾 CSV ventanas guardado: {windows_path}")
    log.success(f"💾 CSV resumen guardado:  {summary_path}")

    print("\n📊 RESUMEN WALK-FORWARD:")
    print(df_summary.to_string(index=False))

    storage.close()
    log.success("\n✅ TEST WALK-FORWARD COMPLETADO")


if __name__ == "__main__":
    test_walkforward()