42. Strategy Multi-Dimensional Stress Testing System¶

services/stress-testing-center/
├── src/
│   ├── main.py
│   ├── scenario/
│   │   └── stress_scenarios.py
│   ├── simulator/
│   │   ├── liquidity_simulator.py
│   │   ├── slippage_simulator.py
│   │   └── network_delay_simulator.py
│   ├── executor/
│   │   └── stress_test_runner.py
│   ├── evaluator/
│   │   └── stress_test_evaluator.py
│   ├── reporter/
│   │   └── stress_reporter.py
│   ├── api/
│   │   └── stress_api.py
│   ├── config.py
│   └── requirements.txt
├── Dockerfile
└── tests/

import numpy as np

class StressScenarios:
    @staticmethod
    def flash_crash(price_series, crash_magnitude=0.3, recovery_time=0.1):
        """Simulate a flash crash scenario"""
        crash_point = int(len(price_series) * 0.5)
        crash_duration = int(len(price_series) * recovery_time)

        # Sudden price drop
        price_series[crash_point:crash_point + crash_duration] *= (1 - crash_magnitude)

        # Gradual recovery
        recovery_factor = np.linspace(1 - crash_magnitude, 1, crash_duration)
        price_series[crash_point:crash_point + crash_duration] *= recovery_factor

        return price_series

    @staticmethod
    def liquidity_crisis(volume_series, crisis_magnitude=0.8):
        """Simulate liquidity crisis with volume reduction"""
        crisis_start = int(len(volume_series) * 0.4)
        volume_series[crisis_start:] *= (1 - crisis_magnitude)
        return volume_series

    @staticmethod
    def volatility_spike(price_series, volatility_multiplier=5.0):
        """Simulate extreme volatility increase"""
        returns = np.diff(np.log(price_series))
        stressed_returns = returns * volatility_multiplier
        stressed_prices = np.exp(np.cumsum(stressed_returns))
        return np.concatenate([[price_series[0]], stressed_prices])

    @staticmethod
    def black_swan_event(price_series, event_magnitude=0.5):
        """Simulate unexpected extreme market event"""
        event_point = int(len(price_series) * 0.6)
        price_series[event_point:] *= (1 - event_magnitude)
        return price_series

class LiquiditySimulator:
    def simulate_liquidity_dryup(self, order_book, reduction_factor=0.2):
        """Reduce order book liquidity"""
        modified_book = order_book.copy()

        for level in modified_book['bids']:
            level['quantity'] *= reduction_factor
        for level in modified_book['asks']:
            level['quantity'] *= reduction_factor

        return modified_book

    def simulate_bid_ask_spread_widening(self, order_book, spread_multiplier=3.0):
        """Widen bid-ask spreads"""
        modified_book = order_book.copy()

        # Widen spreads by moving orders away from mid-price
        mid_price = (modified_book['bids'][0]['price'] + 
                    modified_book['asks'][0]['price']) / 2

        for level in modified_book['bids']:
            level['price'] = mid_price - (mid_price - level['price']) * spread_multiplier
        for level in modified_book['asks']:
            level['price'] = mid_price + (level['price'] - mid_price) * spread_multiplier

        return modified_book

class SlippageSimulator:
    def simulate_slippage(self, execution_price, order_size, 
                         base_slippage=0.001, stress_multiplier=5.0):
        """Simulate increased slippage under stress"""
        # Base slippage increases with order size
        size_impact = min(order_size / 1000000, 1.0)  # Normalize to 1M units

        # Stress multiplier increases slippage
        total_slippage = base_slippage * size_impact * stress_multiplier

        # Random direction for slippage
        direction = np.random.choice([-1, 1])

        return execution_price * (1 + direction * total_slippage)

    def simulate_market_impact(self, order_book, order_size, 
                             impact_factor=0.0001):
        """Simulate market impact of large orders"""
        impact = order_size * impact_factor
        return impact

import asyncio
import random

class NetworkDelaySimulator:
    def __init__(self):
        self.base_delay_ms = 10
        self.stress_delay_ms = 500

    async def simulate_delay(self, stress_level=0.0):
        """Simulate network delay under stress"""
        delay_ms = self.base_delay_ms + (self.stress_delay_ms * stress_level)
        await asyncio.sleep(delay_ms / 1000)

    def simulate_partial_fill(self, order_size, fill_ratio=0.7):
        """Simulate partial order fills under stress"""
        return order_size * fill_ratio

    def simulate_order_rejection(self, rejection_probability=0.1):
        """Simulate order rejections under stress"""
        return random.random() < rejection_probability

class StressTestRunner:
    def __init__(self, backtest_engine, risk_manager):
        self.backtest_engine = backtest_engine
        self.risk_manager = risk_manager
        self.scenarios = StressScenarios()
        self.liquidity_sim = LiquiditySimulator()
        self.slippage_sim = SlippageSimulator()
        self.delay_sim = NetworkDelaySimulator()

    async def run_stress_test(self, strategy_id, stress_scenario, 
                            stress_parameters):
        """Run comprehensive stress test"""

        # Get baseline performance
        baseline_results = await self.backtest_engine.run_backtest(strategy_id)

        # Apply stress conditions
        stressed_market_data = self.apply_stress_conditions(
            baseline_results['market_data'], 
            stress_scenario, 
            stress_parameters
        )

        # Run strategy under stress
        stressed_results = await self.backtest_engine.run_backtest(
            strategy_id, 
            market_data=stressed_market_data
        )

        return {
            'baseline': baseline_results,
            'stressed': stressed_results,
            'scenario': stress_scenario,
            'parameters': stress_parameters
        }

    def apply_stress_conditions(self, market_data, scenario, parameters):
        """Apply various stress conditions to market data"""
        stressed_data = market_data.copy()

        if scenario == 'flash_crash':
            stressed_data['prices'] = self.scenarios.flash_crash(
                market_data['prices'], 
                parameters.get('crash_magnitude', 0.3)
            )

        if scenario == 'liquidity_crisis':
            stressed_data['volumes'] = self.scenarios.liquidity_crisis(
                market_data['volumes'], 
                parameters.get('crisis_magnitude', 0.8)
            )

        return stressed_data

class StressTestEvaluator:
    def evaluate_stress_performance(self, baseline_results, stressed_results):
        """Evaluate strategy performance degradation under stress"""

        baseline_metrics = self.calculate_metrics(baseline_results)
        stressed_metrics = self.calculate_metrics(stressed_results)

        evaluation = {
            'performance_degradation': {
                'return_drop': (baseline_metrics['total_return'] - 
                              stressed_metrics['total_return']) / 
                              abs(baseline_metrics['total_return']),
                'sharpe_drop': baseline_metrics['sharpe_ratio'] - 
                              stressed_metrics['sharpe_ratio'],
                'max_drawdown_increase': stressed_metrics['max_drawdown'] - 
                                        baseline_metrics['max_drawdown']
            },
            'risk_metrics': {
                'var_increase': stressed_metrics['var_95'] - 
                               baseline_metrics['var_95'],
                'volatility_increase': stressed_metrics['volatility'] - 
                                      baseline_metrics['volatility']
            },
            'execution_metrics': {
                'slippage_increase': stressed_metrics['avg_slippage'] - 
                                    baseline_metrics['avg_slippage'],
                'fill_rate_decrease': baseline_metrics['fill_rate'] - 
                                     stressed_metrics['fill_rate']
            }
        }

        return evaluation

    def calculate_metrics(self, results):
        """Calculate comprehensive performance metrics"""
        returns = results['returns']

        return {
            'total_return': np.sum(returns),
            'sharpe_ratio': np.mean(returns) / np.std(returns),
            'max_drawdown': self.calculate_max_drawdown(returns),
            'var_95': np.percentile(returns, 5),
            'volatility': np.std(returns),
            'avg_slippage': results.get('avg_slippage', 0),
            'fill_rate': results.get('fill_rate', 1.0)
        }

    def calculate_max_drawdown(self, returns):
        """Calculate maximum drawdown"""
        cumulative = np.cumprod(1 + returns)
        running_max = np.maximum.accumulate(cumulative)
        drawdown = (cumulative - running_max) / running_max
        return np.min(drawdown)

from fpdf import FPDF
import matplotlib.pyplot as plt
import seaborn as sns

class StressReporter:
    def generate_stress_report(self, stress_results, output_path):
        """Generate comprehensive stress testing report"""

        pdf = FPDF()
        pdf.add_page()

        # Title
        pdf.set_font("Arial", 'B', 16)
        pdf.cell(0, 10, "Strategy Stress Testing Report", ln=True, align='C')
        pdf.ln(10)

        # Executive Summary
        pdf.set_font("Arial", 'B', 14)
        pdf.cell(0, 10, "Executive Summary", ln=True)
        pdf.set_font("Arial", '', 12)

        summary = self.generate_executive_summary(stress_results)
        for line in summary:
            pdf.cell(0, 8, line, ln=True)

        pdf.ln(10)

        # Detailed Results
        pdf.set_font("Arial", 'B', 14)
        pdf.cell(0, 10, "Detailed Stress Test Results", ln=True)
        pdf.set_font("Arial", '', 12)

        for scenario, results in stress_results.items():
            pdf.cell(0, 8, f"Scenario: {scenario}", ln=True)
            pdf.cell(0, 8, f"Performance Drop: {results['performance_degradation']['return_drop']:.2%}", ln=True)
            pdf.cell(0, 8, f"Max Drawdown Increase: {results['performance_degradation']['max_drawdown_increase']:.2%}", ln=True)
            pdf.ln(5)

        # Add charts
        self.add_performance_charts(pdf, stress_results)

        pdf.output(output_path)
        return output_path

    def generate_executive_summary(self, stress_results):
        """Generate executive summary of stress test results"""
        summary = []

        total_scenarios = len(stress_results)
        avg_performance_drop = np.mean([
            results['performance_degradation']['return_drop'] 
            for results in stress_results.values()
        ])

        summary.append(f"Total Stress Scenarios Tested: {total_scenarios}")
        summary.append(f"Average Performance Drop: {avg_performance_drop:.2%}")
        summary.append("")

        # Identify worst-case scenario
        worst_scenario = min(stress_results.items(), 
                           key=lambda x: x[1]['performance_degradation']['return_drop'])
        summary.append(f"Worst-Case Scenario: {worst_scenario[0]}")
        summary.append(f"Performance Drop: {worst_scenario[1]['performance_degradation']['return_drop']:.2%}")

        return summary

    def add_performance_charts(self, pdf, stress_results):
        """Add performance comparison charts to report"""
        # Create performance comparison chart
        scenarios = list(stress_results.keys())
        performance_drops = [results['performance_degradation']['return_drop'] 
                           for results in stress_results.values()]

        plt.figure(figsize=(10, 6))
        plt.bar(scenarios, performance_drops)
        plt.title('Strategy Performance Under Stress Scenarios')
        plt.ylabel('Performance Drop (%)')
        plt.xticks(rotation=45)
        plt.tight_layout()

        chart_path = 'stress_performance_chart.png'
        plt.savefig(chart_path)
        plt.close()

        # Add chart to PDF
        pdf.image(chart_path, x=10, y=pdf.get_y(), w=190)

from fastapi import APIRouter, HTTPException, BackgroundTasks
from typing import Dict, List, Optional
from pydantic import BaseModel

router = APIRouter()

class StressTestRequest(BaseModel):
    strategy_id: str
    scenarios: List[str]
    parameters: Dict[str, float]
    priority: str = "normal"

class StressTestResult(BaseModel):
    test_id: str
    status: str
    results: Optional[Dict] = None
    report_path: Optional[str] = None

@router.post("/stress/start", response_model=StressTestResult)
async def start_stress_test(request: StressTestRequest, 
                          background_tasks: BackgroundTasks):
    """Start a new stress test"""
    test_id = generate_test_id()

    background_tasks.add_task(
        run_stress_test_background,
        test_id,
        request.strategy_id,
        request.scenarios,
        request.parameters
    )

    return StressTestResult(
        test_id=test_id,
        status="running"
    )

@router.get("/stress/status/{test_id}")
async def get_stress_test_status(test_id: str):
    """Get stress test status and results"""
    status = await get_test_status(test_id)
    return status

@router.get("/stress/results/{test_id}")
async def get_stress_test_results(test_id: str):
    """Get detailed stress test results"""
    results = await get_test_results(test_id)
    if not results:
        raise HTTPException(status_code=404, detail="Test not found")
    return results

@router.get("/stress/report/{test_id}")
async def download_stress_report(test_id: str):
    """Download stress test report"""
    report_path = await get_report_path(test_id)
    if not report_path:
        raise HTTPException(status_code=404, detail="Report not found")
    return FileResponse(report_path)

@router.get("/stress/scenarios")
async def get_available_scenarios():
    """Get available stress test scenarios"""
    return {
        "scenarios": [
            "flash_crash",
            "liquidity_crisis", 
            "volatility_spike",
            "black_swan_event",
            "network_failure",
            "combined_stress"
        ]
    }

// StressTestView.tsx
interface StressTestConfig {
  strategyId: string;
  scenarios: string[];
  parameters: Record<string, number>;
}

interface StressTestResult {
  testId: string;
  status: 'running' | 'completed' | 'failed';
  results?: Record<string, any>;
  reportPath?: string;
}

const StressTestView: React.FC = () => {
  const [testConfig, setTestConfig] = useState<StressTestConfig>({
    strategyId: '',
    scenarios: [],
    parameters: {}
  });
  const [activeTests, setActiveTests] = useState<StressTestResult[]>([]);

  const startStressTest = async () => {
    const response = await api.post('/stress/start', testConfig);
    setActiveTests(prev => [...prev, response.data]);
  };

  return (
    <div className="stress-test-dashboard">
      {/* Test Configuration Panel */}
      <StressTestConfigPanel 
        config={testConfig}
        onConfigChange={setTestConfig}
        onStartTest={startStressTest}
      />

      {/* Active Tests Monitor */}
      <ActiveTestsMonitor tests={activeTests} />

      {/* Results Visualization */}
      <StressTestResultsView tests={activeTests} />
    </div>
  );
};

# stress-testing-config.yaml
scenarios:
  flash_crash:
    crash_magnitude: 0.3
    recovery_time: 0.1

  liquidity_crisis:
    crisis_magnitude: 0.8
    duration: 0.2

  volatility_spike:
    volatility_multiplier: 5.0

  black_swan_event:
    event_magnitude: 0.5

simulation:
  slippage_base: 0.001
  slippage_stress_multiplier: 5.0
  network_delay_base_ms: 10
  network_delay_stress_ms: 500

evaluation:
  performance_threshold: 0.2
  risk_threshold: 0.3
  report_format: "pdf"