Trading-Bot/scripts/research/risk_validation.py

import os
import sys
from pathlib import Path
from dotenv import load_dotenv
from datetime import datetime, timedelta
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

# --------------------------------------------------
# Path setup
# --------------------------------------------------
sys.path.insert(0, str(Path(__file__).parent.parent.parent))

from src.utils.logger import log
from src.data.storage import StorageManager
from src.core.engine import Engine
from src.strategies import MovingAverageCrossover
from src.risk.sizing.percent_risk import PercentRiskSizer
from src.risk.stops.atr_stop import ATRStop


# --------------------------------------------------
# CONFIG
# --------------------------------------------------
SYMBOL = "BTC/USDT"
TIMEFRAME = "1h"
DAYS_BACK = 180

INITIAL_CAPITAL = 10_000
RISK_PER_TRADE = 0.01  # 1%

ATR_PERIOD = 14
ATR_MULTIPLIER = 2.0

COMMISSION = 0.001
SLIPPAGE = 0.0005


def setup_environment():
    """Carga variables de entorno"""
    env_path = Path(__file__).parent.parent.parent / 'config' / 'secrets.env'
    load_dotenv(dotenv_path=env_path)

# --------------------------------------------------
# Load data
# --------------------------------------------------
def load_data():
    # 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"),
    )

    end_date = datetime.now()
    start_date = end_date - timedelta(days=DAYS_BACK)

    data = storage.load_ohlcv(
        symbol=SYMBOL,
        timeframe=TIMEFRAME,
        start_date=start_date,
        end_date=end_date,
        use_cache=False,
    )

    storage.close()

    if data.empty:
        raise RuntimeError("No data loaded")

    return data


# --------------------------------------------------
# Main
# --------------------------------------------------
def run():
    log.info("=" * 70)
    log.info("📐 RISK VALIDATION – PercentRiskSizer + ATRStop")
    log.info("=" * 70)

    data = load_data()

    strategy = MovingAverageCrossover(
        fast_period=10,
        slow_period=30,
        ma_type="ema",
        use_adx=False,
    )

    engine = Engine(
        strategy=strategy,
        initial_capital=INITIAL_CAPITAL,
        commission=COMMISSION,
        slippage=SLIPPAGE,
        position_sizer=PercentRiskSizer(RISK_PER_TRADE),
        stop_loss=ATRStop(
            atr_period=ATR_PERIOD,
            multiplier=ATR_MULTIPLIER,
        ),
    )

    results = engine.run(data)
    trades = results["trades"]

    # --------------------------------------------------
    # Compute real risk per trade
    # --------------------------------------------------
    risks = []

    for trade in trades:
        if trade.exit_reason != "Stop Loss":
            continue

        risk_amount = abs(
            trade.entry_price - trade.stop_price_at_entry
        ) * trade.size

        risk_pct = risk_amount / trade.capital_at_entry
        risks.append(risk_pct)

    risks = np.array(risks)

    if len(risks) == 0:
        log.warning("No stop-loss trades found")
        return

    # --------------------------------------------------
    # Print summary
    # --------------------------------------------------
    print()
    print("=" * 70)
    print("📊 RISK PER TRADE SUMMARY")
    print("=" * 70)
    print(f"Trades analysed : {len(risks)}")
    print(f"Target risk     : {RISK_PER_TRADE*100:.2f}%")
    print(f"Mean risk       : {risks.mean()*100:.2f}%")
    print(f"Std deviation   : {risks.std()*100:.2f}%")
    print(f"Min risk        : {risks.min()*100:.2f}%")
    print(f"Max risk        : {risks.max()*100:.2f}%")
    print("=" * 70)

    output_dir = Path(__file__).parent / "output"
    output_dir.mkdir(parents=True, exist_ok=True)

    # --------------------------------------------------
    # Plot risk distribution
    # --------------------------------------------------
    plt.figure(figsize=(10, 5))
    plt.hist(risks * 100, bins=25, edgecolor="black", alpha=0.7)
    plt.axvline(RISK_PER_TRADE * 100, color="red", linestyle="--", label="Target")
    plt.title("Risk distribution per trade (%)")
    plt.xlabel("Risk (%)")
    plt.ylabel("Trades")
    plt.legend()
    plt.grid(alpha=0.3)
    plt.tight_layout()
    plt.savefig(output_dir / "risk_distribution.png")
    plt.close()

    # --------------------------------------------------
    # Plot risk over time
    # --------------------------------------------------
    plt.figure(figsize=(12, 5))
    plt.plot(risks * 100, marker="o", linewidth=1)
    plt.axhline(RISK_PER_TRADE * 100, color="red", linestyle="--", label="Target")
    plt.title("Risk per trade over time")
    plt.xlabel("Trade #")
    plt.ylabel("Risk (%)")
    plt.legend()
    plt.grid(alpha=0.3)
    plt.tight_layout()
    plt.savefig(output_dir / "risk_over_time.png")
    plt.close()

    log.success("Risk validation completed ✔")


if __name__ == "__main__":
    run()