Monte Carlo Simulation with Ollama: Portfolio Stress Testing and VAR Calculation

# Install Ollama (macOS/Linux)
curl -fsSL https://ollama.ai/install.sh | sh

# Pull a suitable model for financial analysis
ollama pull llama2

# Install Python dependencies
pip install numpy pandas matplotlib scipy requests

import requests
import json
import numpy as np
import pandas as pd
from datetime import datetime, timedelta

class OllamaFinancialAnalyzer:
    def __init__(self, model_name="llama2"):
        self.model_name = model_name
        self.ollama_url = "http://localhost:11434/api/generate"
    
    def query_ollama(self, prompt):
        """Send query to Ollama and return response"""
        payload = {
            "model": self.model_name,
            "prompt": prompt,
            "stream": False
        }
        
        response = requests.post(self.ollama_url, json=payload)
        if response.status_code == 200:
            return response.json()["response"]
        else:
            raise Exception(f"Ollama request failed: {response.status_code}")
    
    def estimate_market_regime(self, market_data):
        """Use Ollama to identify current market conditions"""
        recent_returns = market_data['returns'].tail(30).tolist()
        volatility = np.std(recent_returns) * np.sqrt(252)
        
        prompt = f"""
        Analyze these market conditions and identify the current regime:
        - Recent 30-day returns: {recent_returns}
        - Annualized volatility: {volatility:.2%}
        
        Classify as: BULL_MARKET, BEAR_MARKET, HIGH_VOLATILITY, or STABLE
        Provide reasoning and suggested risk adjustments.
        """
        
        return self.query_ollama(prompt)

class MonteCarloPortfolioSimulator:
    def __init__(self, assets, weights, ollama_analyzer=None):
        self.assets = assets
        self.weights = np.array(weights)
        self.ollama_analyzer = ollama_analyzer
        self.results = {}
    
    def estimate_parameters(self, price_data, lookback_days=252):
        """Estimate expected returns and covariance matrix"""
        returns = price_data.pct_change().dropna()
        
        # Basic parameter estimation
        expected_returns = returns.mean() * 252  # Annualized
        cov_matrix = returns.cov() * 252  # Annualized
        
        # Enhance with Ollama insights if available
        if self.ollama_analyzer:
            market_regime = self.ollama_analyzer.estimate_market_regime(
                {'returns': returns.sum(axis=1)}
            )
            
            # Adjust parameters based on AI insights
            if "HIGH_VOLATILITY" in market_regime:
                cov_matrix *= 1.5  # Increase volatility in high-vol regime
            elif "BEAR_MARKET" in market_regime:
                expected_returns *= 0.7  # Reduce expected returns
        
        return expected_returns, cov_matrix
    
    def simulate_portfolio_paths(self, expected_returns, cov_matrix, 
                                num_simulations=10000, time_horizon=252):
        """Generate Monte Carlo price paths"""
        num_assets = len(self.assets)
        dt = 1/252  # Daily time step
        
        # Generate random shocks
        random_shocks = np.random.multivariate_normal(
            mean=np.zeros(num_assets),
            cov=cov_matrix,
            size=(num_simulations, time_horizon)
        )
        
        # Initialize price paths
        price_paths = np.zeros((num_simulations, time_horizon + 1, num_assets))
        price_paths[:, 0, :] = 100  # Starting price
        
        # Generate price paths using geometric Brownian motion
        for t in range(1, time_horizon + 1):
            price_paths[:, t, :] = price_paths[:, t-1, :] * np.exp(
                (expected_returns - 0.5 * np.diag(cov_matrix)) * dt + 
                np.sqrt(dt) * random_shocks[:, t-1, :]
            )
        
        return price_paths
    
    def calculate_portfolio_values(self, price_paths):
        """Calculate portfolio values from price paths"""
        initial_value = 100  # Starting portfolio value
        
        # Calculate portfolio values for each simulation
        portfolio_values = np.zeros((price_paths.shape[0], price_paths.shape[1]))
        
        for sim in range(price_paths.shape[0]):
            portfolio_values[sim, :] = np.sum(
                price_paths[sim, :, :] * self.weights, axis=1
            )
        
        return portfolio_values

class VARCalculator:
    def __init__(self, confidence_levels=[0.95, 0.99]):
        self.confidence_levels = confidence_levels
    
    def calculate_var(self, portfolio_values, initial_value=100):
        """Calculate Value at Risk using Monte Carlo results"""
        # Calculate portfolio returns
        final_values = portfolio_values[:, -1]  # End values
        portfolio_returns = (final_values - initial_value) / initial_value
        
        var_results = {}
        
        for confidence_level in self.confidence_levels:
            # Calculate VAR as percentile of loss distribution
            var_percentile = 1 - confidence_level
            var_value = np.percentile(portfolio_returns, var_percentile * 100)
            
            var_results[f"VAR_{confidence_level}"] = {
                "value": var_value,
                "dollar_amount": var_value * initial_value,
                "percentage": var_value * 100
            }
        
        return var_results
    
    def calculate_expected_shortfall(self, portfolio_values, confidence_level=0.95):
        """Calculate Expected Shortfall (Conditional VAR)"""
        initial_value = 100
        final_values = portfolio_values[:, -1]
        portfolio_returns = (final_values - initial_value) / initial_value
        
        # Calculate VAR threshold
        var_threshold = np.percentile(portfolio_returns, (1 - confidence_level) * 100)
        
        # Calculate Expected Shortfall as average of losses beyond VAR
        tail_losses = portfolio_returns[portfolio_returns <= var_threshold]
        expected_shortfall = np.mean(tail_losses)
        
        return {
            "expected_shortfall": expected_shortfall,
            "dollar_amount": expected_shortfall * initial_value,
            "percentage": expected_shortfall * 100
        }

class PortfolioStressTester:
    def __init__(self, simulator, ollama_analyzer):
        self.simulator = simulator
        self.ollama_analyzer = ollama_analyzer
    
    def generate_stress_scenarios(self):
        """Generate stress test scenarios using Ollama"""
        stress_prompt = """
        Generate 5 realistic financial stress scenarios for portfolio testing:
        1. Market crash scenario (specify magnitude and duration)
        2. Interest rate shock (direction and magnitude)
        3. Currency crisis (affected regions and severity)
        4. Credit crisis (sector impacts and timeline)
        5. Geopolitical event (type and market effects)
        
        For each scenario, provide:
        - Expected return adjustments (percentage)
        - Volatility multipliers
        - Correlation changes
        - Duration (trading days)
        """
        
        scenarios_text = self.ollama_analyzer.query_ollama(stress_prompt)
        return self.parse_stress_scenarios(scenarios_text)
    
    def parse_stress_scenarios(self, scenarios_text):
        """Parse Ollama response into structured scenarios"""
        # This is a simplified parser - in practice, you'd use more sophisticated NLP
        scenarios = []
        
        # Parse scenarios from text (implementation depends on Ollama output format)
        # For demonstration, return predefined scenarios
        scenarios = [
            {
                "name": "Market Crash",
                "return_adjustment": -0.30,
                "volatility_multiplier": 2.0,
                "correlation_adjustment": 0.2,
                "duration": 60
            },
            {
                "name": "Interest Rate Shock",
                "return_adjustment": -0.15,
                "volatility_multiplier": 1.5,
                "correlation_adjustment": 0.1,
                "duration": 120
            },
            {
                "name": "Currency Crisis",
                "return_adjustment": -0.20,
                "volatility_multiplier": 1.8,
                "correlation_adjustment": 0.3,
                "duration": 90
            }
        ]
        
        return scenarios
    
    def run_stress_test(self, expected_returns, cov_matrix, scenario):
        """Run stress test for a specific scenario"""
        # Adjust parameters based on scenario
        stressed_returns = expected_returns * (1 + scenario["return_adjustment"])
        stressed_cov = cov_matrix * scenario["volatility_multiplier"]
        
        # Add correlation stress
        correlation_matrix = np.corrcoef(stressed_cov)
        correlation_matrix = np.clip(
            correlation_matrix + scenario["correlation_adjustment"],
            -1, 1
        )
        
        # Reconstruct covariance matrix
        vol_vector = np.sqrt(np.diag(stressed_cov))
        stressed_cov = np.outer(vol_vector, vol_vector) * correlation_matrix
        
        # Run simulation with stressed parameters
        price_paths = self.simulator.simulate_portfolio_paths(
            stressed_returns, stressed_cov, 
            time_horizon=scenario["duration"]
        )
        
        portfolio_values = self.simulator.calculate_portfolio_values(price_paths)
        
        return {
            "scenario": scenario["name"],
            "portfolio_values": portfolio_values,
            "final_returns": (portfolio_values[:, -1] - 100) / 100
        }

import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta

class CompletePortfolioAnalyzer:
    def __init__(self, assets, weights, price_data):
        self.assets = assets
        self.weights = weights
        self.price_data = price_data
        self.ollama_analyzer = OllamaFinancialAnalyzer()
        self.simulator = MonteCarloPortfolioSimulator(assets, weights, self.ollama_analyzer)
        self.var_calculator = VARCalculator()
        self.stress_tester = PortfolioStressTester(self.simulator, self.ollama_analyzer)
    
    def run_complete_analysis(self, num_simulations=10000):
        """Run complete portfolio risk analysis"""
        print("Starting Portfolio Risk Analysis...")
        
        # Step 1: Estimate parameters
        expected_returns, cov_matrix = self.simulator.estimate_parameters(self.price_data)
        print(f"Expected Returns: {expected_returns}")
        print(f"Volatility: {np.sqrt(np.diag(cov_matrix))}")
        
        # Step 2: Run Monte Carlo simulation
        price_paths = self.simulator.simulate_portfolio_paths(
            expected_returns, cov_matrix, num_simulations
        )
        portfolio_values = self.simulator.calculate_portfolio_values(price_paths)
        
        # Step 3: Calculate VAR
        var_results = self.var_calculator.calculate_var(portfolio_values)
        es_results = self.var_calculator.calculate_expected_shortfall(portfolio_values)
        
        # Step 4: Run stress tests
        stress_scenarios = self.stress_tester.generate_stress_scenarios()
        stress_results = []
        
        for scenario in stress_scenarios:
            result = self.stress_tester.run_stress_test(
                expected_returns, cov_matrix, scenario
            )
            stress_results.append(result)
        
        # Step 5: Generate report
        self.generate_risk_report(var_results, es_results, stress_results, portfolio_values)
        
        return {
            "var_results": var_results,
            "expected_shortfall": es_results,
            "stress_results": stress_results,
            "portfolio_values": portfolio_values
        }
    
    def generate_risk_report(self, var_results, es_results, stress_results, portfolio_values):
        """Generate comprehensive risk report"""
        print("\n" + "="*50)
        print("PORTFOLIO RISK ANALYSIS REPORT")
        print("="*50)
        
        # VAR Results
        print("\nVALUE AT RISK (VAR) ANALYSIS:")
        for var_type, result in var_results.items():
            confidence = var_type.split("_")[1]
            print(f"{confidence}% VAR: {result['percentage']:.2f}% (${result['dollar_amount']:.2f})")
        
        # Expected Shortfall
        print(f"\nEXPECTED SHORTFALL (95%):")
        print(f"ES: {es_results['percentage']:.2f}% (${es_results['dollar_amount']:.2f})")
        
        # Stress Test Results
        print("\nSTRESS TEST RESULTS:")
        for result in stress_results:
            worst_case = np.percentile(result['final_returns'], 5)
            print(f"{result['scenario']}: {worst_case:.2%} (5th percentile)")
        
        # Generate visualizations
        self.create_visualizations(var_results, portfolio_values, stress_results)
    
    def create_visualizations(self, var_results, portfolio_values, stress_results):
        """Create risk visualization charts"""
        fig, axes = plt.subplots(2, 2, figsize=(15, 10))
        
        # Portfolio value distribution
        final_values = portfolio_values[:, -1]
        axes[0, 0].hist(final_values, bins=50, alpha=0.7, edgecolor='black')
        axes[0, 0].axvline(100, color='red', linestyle='--', label='Initial Value')
        axes[0, 0].set_title('Portfolio Value Distribution')
        axes[0, 0].set_xlabel('Portfolio Value')
        axes[0, 0].set_ylabel('Frequency')
        axes[0, 0].legend()
        
        # VAR visualization
        returns = (final_values - 100) / 100
        axes[0, 1].hist(returns, bins=50, alpha=0.7, edgecolor='black')
        
        # Add VAR lines
        var_95 = var_results['VAR_0.95']['value']
        var_99 = var_results['VAR_0.99']['value']
        axes[0, 1].axvline(var_95, color='orange', linestyle='--', label='95% VAR')
        axes[0, 1].axvline(var_99, color='red', linestyle='--', label='99% VAR')
        axes[0, 1].set_title('Return Distribution with VAR')
        axes[0, 1].set_xlabel('Portfolio Returns')
        axes[0, 1].set_ylabel('Frequency')
        axes[0, 1].legend()
        
        # Portfolio paths sample
        sample_paths = portfolio_values[:100, :]  # Show first 100 paths
        for i in range(sample_paths.shape[0]):
            axes[1, 0].plot(sample_paths[i, :], alpha=0.1, color='blue')
        
        axes[1, 0].set_title('Sample Portfolio Paths')
        axes[1, 0].set_xlabel('Time (Days)')
        axes[1, 0].set_ylabel('Portfolio Value')
        
        # Stress test comparison
        scenario_names = [result['scenario'] for result in stress_results]
        worst_cases = [np.percentile(result['final_returns'], 5) for result in stress_results]
        
        axes[1, 1].bar(scenario_names, worst_cases, color=['red', 'orange', 'yellow'])
        axes[1, 1].set_title('Stress Test Results (5th Percentile)')
        axes[1, 1].set_ylabel('Portfolio Return')
        axes[1, 1].tick_params(axis='x', rotation=45)
        
        plt.tight_layout()
        plt.savefig('portfolio_risk_analysis.png', dpi=300, bbox_inches='tight')
        plt.show()

# Example usage
if __name__ == "__main__":
    # Sample data setup
    assets = ['AAPL', 'GOOGL', 'MSFT', 'AMZN']
    weights = [0.25, 0.25, 0.25, 0.25]
    
    # Generate sample price data (in practice, load from your data source)
    dates = pd.date_range('2023-01-01', '2024-12-31', freq='D')
    np.random.seed(42)
    price_data = pd.DataFrame({
        'AAPL': 100 * np.exp(np.cumsum(np.random.normal(0.0005, 0.02, len(dates)))),
        'GOOGL': 100 * np.exp(np.cumsum(np.random.normal(0.0003, 0.025, len(dates)))),
        'MSFT': 100 * np.exp(np.cumsum(np.random.normal(0.0004, 0.022, len(dates)))),
        'AMZN': 100 * np.exp(np.cumsum(np.random.normal(0.0002, 0.03, len(dates))))
    }, index=dates)
    
    # Run analysis
    analyzer = CompletePortfolioAnalyzer(assets, weights, price_data)
    results = analyzer.run_complete_analysis(num_simulations=5000)

from multiprocessing import Pool
import functools

def parallel_simulation(args):
    """Run simulation in parallel"""
    simulator, expected_returns, cov_matrix, num_sims = args
    return simulator.simulate_portfolio_paths(expected_returns, cov_matrix, num_sims)

def run_parallel_monte_carlo(simulator, expected_returns, cov_matrix, 
                           total_simulations=10000, num_processes=4):
    """Run Monte Carlo simulation in parallel"""
    sims_per_process = total_simulations // num_processes
    
    args = [(simulator, expected_returns, cov_matrix, sims_per_process) 
            for _ in range(num_processes)]
    
    with Pool(processes=num_processes) as pool:
        results = pool.map(parallel_simulation, args)
    
    # Combine results
    return np.vstack(results)

class ModelValidator:
    def __init__(self, actual_returns, simulated_returns):
        self.actual_returns = actual_returns
        self.simulated_returns = simulated_returns
    
    def validate_distribution(self):
        """Validate that simulated returns match actual distribution"""
        from scipy import stats
        
        # Kolmogorov-Smirnov test
        ks_statistic, p_value = stats.ks_2samp(
            self.actual_returns, self.simulated_returns
        )
        
        print(f"KS Test - Statistic: {ks_statistic:.4f}, P-value: {p_value:.4f}")
        
        # Compare moments
        actual_mean = np.mean(self.actual_returns)
        simulated_mean = np.mean(self.simulated_returns)
        
        actual_std = np.std(self.actual_returns)
        simulated_std = np.std(self.simulated_returns)
        
        print(f"Mean - Actual: {actual_mean:.4f}, Simulated: {simulated_mean:.4f}")
        print(f"Std - Actual: {actual_std:.4f}, Simulated: {simulated_std:.4f}")
        
        return {
            "ks_statistic": ks_statistic,
            "p_value": p_value,
            "mean_difference": abs(actual_mean - simulated_mean),
            "std_difference": abs(actual_std - simulated_std)
        }

class DynamicRiskModel:
    def __init__(self, analyzer):
        self.analyzer = analyzer
        self.risk_history = []
    
    def update_risk_metrics(self, new_market_data):
        """Update risk metrics with new market data"""
        # Re-estimate parameters with new data
        expected_returns, cov_matrix = self.analyzer.simulator.estimate_parameters(new_market_data)
        
        # Run updated simulation
        price_paths = self.analyzer.simulator.simulate_portfolio_paths(expected_returns, cov_matrix)
        portfolio_values = self.analyzer.simulator.calculate_portfolio_values(price_paths)
        
        # Calculate new VAR
        var_results = self.analyzer.var_calculator.calculate_var(portfolio_values)
        
        # Store in history
        self.risk_history.append({
            "date": datetime.now(),
            "var_95": var_results["VAR_0.95"]["value"],
            "var_99": var_results["VAR_0.99"]["value"]
        })
        
        return var_results
    
    def detect_regime_change(self, threshold=0.05):
        """Detect significant changes in risk metrics"""
        if len(self.risk_history) < 2:
            return False
        
        current_var = self.risk_history[-1]["var_95"]
        previous_var = self.risk_history[-2]["var_95"]
        
        change_magnitude = abs(current_var - previous_var) / abs(previous_var)
        
        return change_magnitude > threshold

class RiskBudgetingOptimizer:
    def __init__(self, simulator, target_var):
        self.simulator = simulator
        self.target_var = target_var
    
    def optimize_weights(self, expected_returns, cov_matrix):
        """Optimize portfolio weights to achieve target VAR"""
        from scipy.optimize import minimize
        
        def objective(weights):
            # Ensure weights sum to 1
            weights = weights / np.sum(weights)
            
            # Update simulator weights
            self.simulator.weights = weights
            
            # Run simulation
            price_paths = self.simulator.simulate_portfolio_paths(
                expected_returns, cov_matrix, num_simulations=1000
            )
            portfolio_values = self.simulator.calculate_portfolio_values(price_paths)
            
            # Calculate VAR
            var_calculator = VARCalculator()
            var_results = var_calculator.calculate_var(portfolio_values)
            current_var = abs(var_results["VAR_0.95"]["value"])
            
            # Minimize difference from target VAR
            return (current_var - self.target_var) ** 2
        
        # Initial guess (equal weights)
        initial_weights = np.ones(len(self.simulator.assets)) / len(self.simulator.assets)
        
        # Constraints
        constraints = [{'type': 'eq', 'fun': lambda x: np.sum(x) - 1}]
        bounds = [(0.01, 0.8) for _ in range(len(self.simulator.assets))]
        
        # Optimize
        result = minimize(
            objective, initial_weights, 
            method='SLSQP', bounds=bounds, constraints=constraints
        )
        
        return result.x

class RealTimeRiskMonitor:
    def __init__(self, analyzer, alert_thresholds):
        self.analyzer = analyzer
        self.alert_thresholds = alert_thresholds
        self.monitoring_active = False
    
    def start_monitoring(self, update_interval=300):  # 5 minutes
        """Start real-time risk monitoring"""
        import time
        import threading
        
        self.monitoring_active = True
        
        def monitor_loop():
            while self.monitoring_active:
                try:
                    # Fetch latest market data
                    latest_data = self.fetch_latest_market_data()
                    
                    # Update risk metrics
                    risk_metrics = self.analyzer.run_complete_analysis()
                    
                    # Check for alerts
                    self.check_risk_alerts(risk_metrics)
                    
                    time.sleep(update_interval)
                    
                except Exception as e:
                    print(f"Monitoring error: {e}")
                    time.sleep(60)  # Wait before retrying
        
        monitor_thread = threading.Thread(target=monitor_loop)
        monitor_thread.daemon = True
        monitor_thread.start()
    
    def check_risk_alerts(self, risk_metrics):
        """Check if risk metrics exceed alert thresholds"""
        var_95 = abs(risk_metrics["var_results"]["VAR_0.95"]["value"])
        
        if var_95 > self.alert_thresholds["var_95"]:
            self.send_alert(f"VAR 95% exceeded threshold: {var_95:.2%}")
    
    def send_alert(self, message):
        """Send risk alert (implement your preferred notification method)"""
        print(f"RISK ALERT: {message}")
        # Implement email, Slack, or other notification system

class PortfolioIntegration:
    def __init__(self, analyzer, portfolio_api):
        self.analyzer = analyzer
        self.portfolio_api = portfolio_api
    
    def sync_portfolio_data(self):
        """Sync portfolio data from external system"""
        # Fetch current holdings
        holdings = self.portfolio_api.get_current_holdings()
        
        # Update analyzer with current positions
        self.analyzer.weights = self.calculate_weights(holdings)
        
        # Fetch latest prices
        price_data = self.portfolio_api.get_price_history()
        self.analyzer.price_data = price_data
        
        return holdings, price_data
    
    def calculate_weights(self, holdings):
        """Calculate portfolio weights from holdings"""
        total_value = sum(holding['market_value'] for holding in holdings)
        return [holding['market_value'] / total_value for holding in holdings]
    
    def generate_risk_report(self):
        """Generate and save risk report"""
        # Sync latest data
        self.sync_portfolio_data()
        
        # Run analysis
        results = self.analyzer.run_complete_analysis()
        
        # Save report
        report_path = f"risk_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
        self.save_report_to_pdf(results, report_path)
        
        return report_path