Backtest Drawdown Analysis

---




name: backtest-drawdown-analysis
compatibility: opencode
completeness: 95
content-types:
- code
- guidance
- config
- do-dont
description: '"Implements maximum drawdown, recovery time, and value-at-risk analysis
  for risk management and algorithmic trading execution."'
license: MIT
maturity: stable
metadata:
  domain: trading
  output-format: code
  related-skills: backtest-position-sizing, exchange-order-execution-api, fundamentals-risk-management-basics,
    risk-position-sizing risk-position-sizing
  role: implementation
  scope: implementation
  triggers: backtest drawdown analysis, backtest-drawdown-analysis, maximum, recovery,
    value-at-risk
  archetypes:
  - tactical
  anti_triggers:
  - brainstorming
  - vague ideation
  - no risk management
  response_profile:
    verbosity: low
    directive_strength: high
    abstraction_level: operational
version: "1.0.0"




---




**Role:** Risk Analysis Specialist — implements comprehensive drawdown and risk metrics to evaluate strategy drawdown behavior, recovery characteristics, and tail risk exposure.

**Philosophy:** Drawdown-Centric Risk Assessment — strategies should be evaluated by their worst-case behavior, not just expected returns; maximum drawdown and recovery time reveal true stress tolerance.

## Key Principles

1. **Maximum Drawdown Focus**: Maximum drawdown is the single most important risk metric for trading strategies, representing the largest peak-to-trough decline.

2. **Recovery Time Analysis**: A strategy's value is heavily influenced by how quickly it recovers from drawdowns; long recovery periods indicate poor strategy resilience.

3. **Distribution Analysis**: Single-point metrics (max DD) are insufficient; analyze the full distribution of drawdowns to understand frequency and severity patterns.

4. **Value-at-Risk Integration**: Combine drawdown analysis with VaR and CVaR for comprehensive tail risk assessment across different confidence levels.

5. **Stress Testing**: Use historical crisis periods to test drawdown behavior; strategies should be validated under extreme market conditions.

## Implementation Guidelines

### Structure
- Core logic: `skills/backtesting/drawdown_analysis.py`
- Risk metrics module: `skills/backtesting/risk_metrics.py`
- Tests: `skills/tests/test_drawdown_analysis.py`

### Patterns to Follow
- Implement drawdown calculation as a class with multiple analysis methods
- Support both price-based and return-based drawdown analysis
- Include recovery time calculation and statistics
- Provide VaR and CVaR calculations at multiple confidence levels
- Use vectorized operations for efficient large-scale calculations

## Adherence Checklist
Before completing your task, verify:
- [ ] **Maximum Drawdown**: Is maximum drawdown correctly calculated as peak-to-trough percentage decline?
- [ ] **Recovery Time**: Are drawdown recovery times calculated and statistics provided?
- [ ] **VaR/CVaR**: Are Value-at-Risk and Conditional Value-at-Risk calculated at multiple confidence levels?
- [ ] **Drawdown Distribution**: Is the full distribution of drawdowns analyzed, not just the maximum?
- [ ] **Stress Testing**: Are drawdowns evaluated during historical crisis periods?

## Code Examples

### Maximum Drawdown Analysis System

```python
from dataclasses import dataclass
from typing import List, Tuple, Optional, Dict
import numpy as np
import pandas as pd
from scipy import stats
from enum import Enum


@dataclass
class DrawdownEvent:
    """Single drawdown event with full characteristics."""
    peak_date: pd.Timestamp
    peak_value: float
    trough_date: pd.Timestamp
    trough_value: float
    trough_return: float
    recovery_date: Optional[pd.Timestamp]
    recovery_value: Optional[float]
    recovery_return: Optional[float]
    duration_days: int
    recovery_days: Optional[int]
    depth_percentage: float
    severity: float  # Depth × Duration


@dataclass
class DrawdownStats:
    """Comprehensive drawdown statistics."""
    max_drawdown: float
    max_drawdown_date: pd.Timestamp
    max_drawdown_peak: float
    max_drawdown_trough: float
    average_drawdown: float
    median_drawdown: float
    drawdown_std: float
    n_drawdowns: int
    average_recovery_time: Optional[float]
    median_recovery_time: Optional[float]
    max_drawdown_recovery_days: Optional[int]
    underwater_duration_days: int
    time_in_drawdown: float


class DrawdownAnalyzer:
    """
    Comprehensive drawdown analysis system.
    Calculates drawdowns, recovery times, and provides detailed statistics.
    """
    
    def __init__(self, prices: pd.Series, returns: Optional[pd.Series] = None):
        """
        Initialize drawdown analyzer.
        
        Args:
            prices: Price series for drawdown calculation
            returns: Optional return series for additional analysis
        """
        self.prices = prices.dropna()
        self.returns = returns.dropna() if returns is not None else None
        self.running_max = self.prices.cummax()
        self.drawdowns = (self.prices - self.running_max) / self.running_max
        
    def find_drawdown_events(self) -> List[DrawdownEvent]:
        """
        Identify all drawdown events in the time series.
        
        Returns:
            List of DrawdownEvent objects with full characteristics
        """
        events = []
        in_drawdown = False
        peak_date = None
        peak_value = None
        trough_date = None
        trough_value = None
        
        for i, (date, price) in enumerate(self.prices.items()):
            current_max = self.running_max.loc[date]
            dd = self.drawdowns.loc[date]
            
            if dd < 0:  # Currently in drawdown
                if not in_drawdown:
                    # Starting new drawdown
                    peak_date = date
                    peak_value = price
                    in_drawdown = True
                
                # Update trough if deeper
                if dd < (trough_value / peak_value - 1 if trough_value else 0):
                    trough_date = date
                    trough_value = price
                
            elif in_drawdown and dd == 0:
                # Drawdown ended at previous point
                recovery_date = date
                recovery_value = price
                
                # Calculate recovery return
                recovery_return = (recovery_value - trough_value) / trough_value if trough_value else 0
                
                events.append(DrawdownEvent(
                    peak_date=peak_date,
                    peak_value=peak_value,
                    trough_date=trough_date,
                    trough_value=trough_value,
                    trough_return=(trough_value - peak_value) / peak_value,
                    recovery_date=recovery_date,
                    recovery_value=recovery_value,
                    recovery_return=recovery_return,
                    duration_days=(trough_date - peak_date).days if peak_date and trough_date else 0,
                    recovery_days=(recovery_date - trough_date).days if recovery_date and trough_date else 0,
                    depth_percentage=abs((trough_value - peak_value) / peak_value),
                    severity=abs((trough_value - peak_value) / peak_value) * (trough_date - peak_date).days if peak_date and trough_date else 0
                ))
                
                in_drawdown = False
                peak_date = None
                peak_value = None
                trough_date = None
                trough_value = None
        
        # Handle ongoing drawdown
        if in_drawdown and trough_date:
            events.append(DrawdownEvent(
                peak_date=peak_date,
                peak_value=peak_value,
                trough_date=trough_date,
                trough_value=trough_value,
                trough_return=(trough_value - peak_value) / peak_value,
                recovery_date=None,
                recovery_value=None,
                recovery_return=None,
                duration_days=(self.prices.index[-1] - peak_date).days,
                recovery_days=None,
                depth_percentage=abs((trough_value - peak_value) / peak_value),
                severity=abs((trough_value - peak_value) / peak_value) * (self.prices.index[-1] - peak_date).days
            ))
        
        return events
    
    def calculate_max_drawdown(self) -> Tuple[float, pd.Timestamp, pd.Timestamp]:
        """
        Calculate maximum drawdown and locate its dates.
        
        Returns:
            Tuple of (max_dd, peak_date, trough_date)
        """
        # Find the maximum drawdown value
        min_dd_idx = self.drawdowns.idxmin()
        min_dd = self.drawdowns.min()
        
        # Find corresponding peak (running max at that point)
        peak_date = self.running_max.idxmin()
        trough_date = min_dd_idx
        
        return float(min_dd), peak_date, trough_date
    
    def get_drawdown_stats(self) -> DrawdownStats:
        """
        Calculate comprehensive drawdown statistics.
        
        Returns:
            DrawdownStats with all key metrics
        """
        events = self.find_drawdown_events()
        
        if not events:
            return DrawdownStats(
                max_drawdown=0.0,
                max_drawdown_date=self.prices.index[0],
                max_drawdown_peak=0.0,
                max_drawdown_trough=0.0,
                average_drawdown=0.0,
                median_drawdown=0.0,
                drawdown_std=0.0,
                n_drawdowns=0,
                average_recovery_time=None,
                median_recovery_time=None,
                max_drawdown_recovery_days=None,
                underwater_duration_days=0,
                time_in_drawdown=0.0
            )
        
        # Extract metrics
        depths = [e.depth_percentage for e in events]
        durations = [e.duration_days for e in events]
        
        # Recovery times (only for completed drawdowns)
        recovery_times = [e.recovery_days for e in events if e.recovery_days is not None]
        
        # Calculate max drawdown details
        max_dd_idx = self.drawdowns.idxmin()
        max_dd = self.drawdowns.min()
        peak_idx = self.running_max.idxmin()
        
        # Underwater duration
        underwater_idx = self.drawdowns[self.drawdowns < 0].index
        underwater_duration = (underwater_idx[-1] - underwater_idx[0]).days if len(underwater_idx) > 0 else 0
        
        # Time in drawdown (percentage of total)
        time_in_dd = len(underwater_idx) / len(self.drawdowns) if len(self.drawdowns) > 0 else 0
        
        return DrawdownStats(
            max_drawdown=float(max_dd),
            max_drawdown_date=max_dd_idx,
            max_drawdown_peak=float(self.running_max.loc[peak_idx]),
            max_drawdown_trough=float(self.prices.loc[max_dd_idx]),
            average_drawdown=float(np.mean(depths)),
            median_drawdown=float(np.median(depths)),
            drawdown_std=float(np.std(depths)),
            n_drawdowns=len(events),
            average_recovery_time=float(np.mean(recovery_times)) if recovery_times else None,
            median_recovery_time=float(np.median(recovery_times)) if recovery_times else None,
            max_drawdown_recovery_days=max(durations) if durations else None,
            underwater_duration_days=underwater_duration,
            time_in_drawdown=time_in_dd
        )
    
    def calculate_recovery_ratio(self) -> float:
        """
        Calculate average recovery ratio.
        Measures how much of a drawdown is typically recovered.
        """
        events = self.find_drawdown_events()
        
        recovery_ratios = []
        for event in events:
            if event.recovery_return and event.trough_return:
                # Ratio of recovered return to original drawdown
                ratio = event.recovery_return / abs(event.trough_return) if event.trough_return != 0 else 0
                recovery_ratios.append(ratio)
        
        return float(np.mean(recovery_ratios)) if recovery_ratios else 0.0
    
    def calculate_drawdown_severity_index(self) -> float:
        """
        Calculate a composite severity index.
        Combines depth and duration of drawdowns.
        """
        stats = self.get_drawdown_stats()
        
        # Severity = max_dd × average_duration × time_in_dd
        severity = abs(stats.max_drawdown) * stats.median_drawdown * stats.time_in_drawdown
        
        return severity


# Value at Risk and Conditional Value at Risk
class RiskMetricsCalculator:
    """
    Calculate VaR and CVaR (Expected Shortfall) for risk assessment.
    """
    
    def __init__(self, returns: pd.Series, confidence_levels: List[float] = None):
        """
        Initialize risk metrics calculator.
        
        Args:
            returns: Return series
            confidence_levels: List of confidence levels for VaR (default: [0.95, 0.99])
        """
        self.returns = returns.dropna()
        self.confidence_levels = confidence_levels or [0.95, 0.99, 0.995]
    
    def calculate_var_historic(self, confidence: float = 0.95) -> float:
        """
        Calculate historic Value-at-Risk.
        The loss threshold that will not be exceeded with given confidence.
        """
        if confidence <= 0 or confidence >= 1:
            raise ValueError("Confidence must be between 0 and 1 (exclusive)")
        
        # VaR is the quantile of losses (negative of returns quantile)
        var = -np.percentile(self.returns, (1 - confidence) * 100)
        return float(var)
    
    def calculate_var_parametric(self, confidence: float = 0.95) -> float:
        """
        Calculate parametric (variance-covariance) VaR.
        Assumes returns are normally distributed.
        """
        mean_ret = self.returns.mean()
        std_ret = self.returns.std(ddof=1)
        
        # Z-score for confidence level
        z_score = stats.norm.ppf(1 - confidence)
        
        # VaR = -(mean + z * std)
        var = -(mean_ret + z_score * std_ret)
        return float(var)
    
    def calculate_var_historical_simulation(self, 
                                           confidence: float = 0.95,
                                           n_days: int = 1) -> float:
        """
        Calculate multi-day historic VaR using simulation.
        """
        var_single_day = self.calculate_var_historic(confidence)
        
        # Scale to n days (square root of time rule)
        var_n_day = var_single_day * np.sqrt(n_days)
        return float(var_n_day)
    
    def calculate_cvar(self, confidence: float = 0.95) -> float:
        """
        Calculate Conditional Value-at-Risk (Expected Shortfall).
        Expected loss given that the loss exceeds VaR.
        """
        var = self.calculate_var_historic(confidence)
        
        # CVaR is the mean of returns below the VaR threshold
        returns_below_var = self.returns[self.returns <= -var]
        
        if len(returns_below_var) == 0:
            return float(var)
        
        cvar = -returns_below_var.mean()
        return float(cvar)
    
    def calculate_all_risk_metrics(self) -> Dict:
        """
        Calculate all risk metrics at once.
        
        Returns:
            Dictionary with all risk metrics
        """
        results = {}
        
        for conf in self.confidence_levels:
            conf_key = f"{int(conf * 100)}%"
            results[f"var_historic_{conf_key}"] = self.calculate_var_historic(conf)
            results[f"var_parametric_{conf_key}"] = self.calculate_var_parametric(conf)
            results[f"var_1d_historic_{conf_key}"] = self.calculate_var_historic(conf)
            results[f"var_10d_historic_{conf_key}"] = self.calculate_var_historical_simulation(conf, 10)
            results[f"var_20d_historic_{conf_key}"] = self.calculate_var_historical_simulation(conf, 20)
            results[f"cvar_{conf_key}"] = self.calculate_cvar(conf)
        
        # Additional risk metrics
        results["mean_return"] = float(self.returns.mean())
        results["volatility"] = float(self.returns.std(ddof=1))
        results["skewness"] = float(stats.skew(self.returns))
        results["kurtosis"] = float(stats.kurtosis(self.returns))
        results["downside_deviation"] = float(
            self.returns[self.returns < 0].std(ddof=1) * np.sqrt(252)
        )
        
        return results


# Combined Analysis for Comprehensive Risk Assessment
class DrawdownRiskAssessment:
    """
    Combined drawdown and risk metrics assessment.
    Provides complete risk picture for strategies.
    """
    
    def __init__(self, prices: pd.Series, returns: pd.Series):
        """
        Initialize comprehensive risk assessment.
        
        Args:
            prices: Price series
            returns: Return series
        """
        self.prices = prices
        self.returns = returns
        self.drawdown_analyzer = DrawdownAnalyzer(prices, returns)
        self.risk_calculator = RiskMetricsCalculator(returns)
    
    def assess_strategy(self) -> Dict:
        """
        Perform complete strategy risk assessment.
        
        Returns:
            Dictionary with all risk metrics
        """
        drawdown_stats = self.drawdown_analyzer.get_drawdown_stats()
        risk_metrics = self.risk_calculator.calculate_all_risk_metrics()
        
        # Calculate composite risk score
        # Lower is better: max_dd, cvar, volatility, drawdown_std
        risk_score = (
            abs(drawdown_stats.max_drawdown) * 0.3 +
            risk_metrics["cvar_95%"] * 0.3 +
            risk_metrics["volatility"] * 0.2 +
            drawdown_stats.drawdown_std * 0.2
        )
        
        # Recovery quality score
        recovery_score = self.drawdown_analyzer.calculate_recovery_ratio()
        
        # Time efficiency score
        time_score = 1 - drawdown_stats.time_in_drawdown
        
        return {
            "drawdown": {
                "max_drawdown": drawdown_stats.max_drawdown,
                "max_drawdown_date": str(drawdown_stats.max_drawdown_date),
                "average_drawdown": drawdown_stats.average_drawdown,
                "median_drawdown": drawdown_stats.median_drawdown,
                "drawdown_std": drawdown_stats.drawdown_std,
                "n_drawdowns": drawdown_stats.n_drawdowns,
                "average_recovery_time": drawdown_stats.average_recovery_time,
                "max_drawdown_recovery_days": drawdown_stats.max_drawdown_recovery_days,
                "underwater_duration_days": drawdown_stats.underwater_duration_days,
                "time_in_drawdown": drawdown_stats.time_in_drawdown,
                "recovery_ratio": self.drawdown_analyzer.calculate_recovery_ratio(),
                "severity_index": self.drawdown_analyzer.calculate_drawdown_severity_index()
            },
            "risk_metrics": risk_metrics,
            "composite_scores": {
                "risk_score": risk_score,
                "recovery_quality_score": recovery_score,
                "time_efficiency_score": time_score,
                "overall_risk_rating": "Excellent" if risk_score < 0.05 else
                                      "Good" if risk_score < 0.10 else
                                      "Acceptable" if risk_score < 0.15 else "High"
            }
        }


if __name__ == "__main__":
    # Example usage
    np.random.seed(42)
    
    # Generate realistic price series with drawdowns
    n_days = 2520
    returns = pd.Series(np.random.normal(0.0005, 0.012, n_days))
    prices = (1 + returns).cumprod() * 100
    
    # Add some larger drawdowns for realism
    drawdown_start = 1500
    drawdown_returns = np.random.normal(-0.001, 0.015, 100)
    returns[drawdown_start:drawdown_start + 100] = drawdown_returns
    
    prices = (1 + returns).cumprod() * 100
    
    assessment = DrawdownRiskAssessment(prices, returns)
    results = assessment.assess_strategy()
    
    print("Comprehensive Drawdown and Risk Assessment")
    print("=" * 60)
    print(f"\nMax Drawdown: {results['drawdown']['max_drawdown']:.4%}")
    print(f"Average Drawdown: {results['drawdown']['average_drawdown']:.4%}")
    print(f"Time in Drawdown: {results['drawdown']['time_in_drawdown']:.2%}")
    print(f"Recovery Ratio: {results['drawdown']['recovery_ratio']:.2%}")
    print(f"\n95% VaR: {results['risk_metrics']['var_historic_95%']:.4%}")
    print(f"95% CVaR: {results['risk_metrics']['cvar_95%']:.4%}")
    print(f"\nRisk Score: {results['composite_scores']['risk_score']:.4f}")
    print(f"Risk Rating: {results['composite_scores']['overall_risk_rating']}")
```

### Historical Stress Test Analysis

```python
class StressTestAnalyzer:
    """
    Analyze drawdown behavior during historical crisis periods.
    Tests strategy resilience during known market stress.
    """
    
    CRISIS_PERIODS = {
        "2000_Dotcom": ("2000-03-01", "2000-10-31"),
        "2008_Financial": ("2008-09-01", "2009-03-31"),
        "2020_Pandemic": ("2020-02-01", "2020-04-30"),
        "2022_Inflation": ("2022-01-01", "2022-10-31")
    }
    
    def __init__(self, prices: pd.Series, returns: pd.Series):
        self.prices = prices
        self.returns = returns
        self.drawdown_analyzer = DrawdownAnalyzer(prices, returns)
    
    def analyze_crisis_periods(self) -> Dict:
        """
        Analyze performance during historical crisis periods.
        
        Returns:
            Dictionary with crisis period analysis
        """
        results = {}
        
        for period_name, (start_date, end_date) in self.CRISIS_PERIODS.items():
            # Extract crisis period data
            mask = (self.prices.index >= start_date) & (self.prices.index <= end_date)
            crisis_prices = self.prices[mask]
            crisis_returns = self.returns[mask]
            
            if len(crisis_prices) > 0:
                crisis_dd_analyzer = DrawdownAnalyzer(crisis_prices, crisis_returns)
                dd_stats = crisis_dd_analyzer.get_drawdown_stats()
                
                # Total return during crisis
                total_return = (1 + crisis_returns).prod() - 1
                
                # Sharpe during crisis (using risk-free proxy)
                risk_free_proxy = 0.05 / 252
                excess_returns = crisis_returns - risk_free_proxy
                sharpe_crisis = (
                    excess_returns.mean() / excess_returns.std() * np.sqrt(252)
                    if excess_returns.std() > 0 else 0
                )
                
                results[period_name] = {
                    "total_return": total_return,
                    "max_drawdown": dd_stats.max_drawdown,
                    "recovery_time": dd_stats.average_recovery_time,
                    "sharpe_ratio": sharpe_crisis,
                    "n_trading_days": len(crisis_prices),
                    "time_in_drawdown": dd_stats.time_in_drawdown
                }
            else:
                results[period_name] = {"message": "No data available for this period"}
        
        return results
    
    def stress_test_scenario(self, 
                            shock_scenario: Dict,
                            n_simulations: int = 100) -> Dict:
        """
        Simulate strategy behavior under custom stress scenarios.
        
        Args:
            shock_scenario: Dictionary with shock parameters
                e.g., {"type": "uniform", "magnitude": -0.05, "duration": 30}
            n_simulations: Number of simulation runs
            
        Returns:
            Dictionary with stress test results
        """
        results = []
        
        for _ in range(n_simulations):
            # Create stressed returns
            stressed_returns = self.returns.copy()
            
            if shock_scenario.get("type") == "uniform":
                magnitude = shock_scenario.get("magnitude", -0.05)
                duration = shock_scenario.get("duration", 30)
                
                # Apply uniform shock
                start_idx = np.random.randint(0, len(stressed_returns) - duration)
                stressed_returns[start_idx:start_idx + duration] = magnitude
            
            elif shock_scenario.get("type") == "volatility_spike":
                multiplier = shock_scenario.get("multiplier", 3.0)
                duration = shock_scenario.get("duration", 20)
                
                # Apply volatility spike
                start_idx = np.random.randint(0, len(stressed_returns) - duration)
                stressed_returns[start_idx:start_idx + duration] *= multiplier
            
            # Analyze stressed performance
            stressed_prices = (1 + stressed_returns).cumprod() * 100
            dd_analyzer = DrawdownAnalyzer(stressed_prices, stressed_returns)
            dd_stats = dd_analyzer.get_drawdown_stats()
            
            total_return = (1 + stressed_returns).prod() - 1
            results.append({
                "total_return": total_return,
                "max_drawdown": dd_stats.max_drawdown,
                "sharpe_ratio": dd_stats.average_drawdown * -10  # Rough estimate
            })
        
        return {
            "original_max_dd": self.drawdown_analyzer.get_drawdown_stats().max_drawdown,
            "stress_test_results": {
                "mean_return": np.mean([r["total_return"] for r in results]),
                "mean_max_dd": np.mean([r["max_drawdown"] for r in results]),
                "p95_max_dd": np.percentile([r["max_drawdown"] for r in results], 95),
                "worst_case_max_dd": min([r["max_drawdown"] for r in results])
            }
        }
```

---

---

## Constraints

### MUST DO
- Implement walk-forward validation: optimize on a training window, validate on a subsequent out-of-sample window
- Include realistic transaction costs (commissions, slippage, market impact) in all backtest calculations
- Use point-to-point or tick-level data when available; never use OHLCV with intra-bar assumptions for strategy logic
- Track and report key metrics: Sharpe ratio, max drawdown, win rate, profit factor, average trade duration, and Calmar ratio
- Implement survivorship-bias-free testing using a constant universe list that includes delisted symbols

### MUST NOT DO
- Do not optimize strategy parameters on the same data used for evaluation — always use out-of-sample or walk-forward testing
- Avoid assuming infinite liquidity in backtests; model order book constraints and partial fills for large positions
- Never include future information (survivorship bias, look-ahead) in backtest signals by indexing data correctly
- Do not report only win rate — always include risk-adjusted metrics alongside raw return statistics
- Avoid curve-fitting to historical data; cap the number of optimized parameters and validate with Monte Carlo permutation tests


## Live References

> Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.

- [Understanding Drawdown in Trading](https://www.investopedia.com/terms/d/drawdown.asp)
- [Maximum Drawdown Calculation](https://docs.quantconnect.com/tutorials/backtesting-pitfalls)
- [Drawdown Recovery Metrics](https://www.investopedia.com/articles/trading/06/maxdrawdown.asp)
- [Portfolio Risk and Drawdown Management](https://en.wikipedia.org/wiki/Downside_risk)
- [Advanced Drawdown Analysis Techniques](https://docs.quantconnect.com/tutorials/risk-management)