import sys import os import copy # Bulletproof pathing: Force Python to look in both the current folder AND the parent folder _this_dir = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, _this_dir) sys.path.insert(0, os.path.dirname(_this_dir)) import pytest import pandas as pd import numpy as np try: import matplotlib.pyplot as plt except ImportError: plt = None from solver import multi_period_optimize from backtest import monte_carlo from core_types import PortfolioState from config import DEFAULT_CONFIG # ───────────────────────────────────────────── # Note: Matplotlib cleanup fixture # ───────────────────────────────────────────── @pytest.fixture(autouse=True) def cleanup_plots(): """ Automatically closes any matplotlib figures generated during tests (e.g., by deep Monte Carlo calls) to prevent memory leaks and state bleeding. """ yield if plt is not None: plt.close('all') # ───────────────────────────────────────────── # 1. MULTI-PERIOD CONTROL (MPC) TESTS # ───────────────────────────────────────────── def test_mpc_valid_trajectory_generation(): """ Verifies that the Multi-Period Control (MPC) engine can successfully map a multi-step execution trajectory without mathematically failing, and that it outputs a valid T=0 execution policy summing to 1.0. """ rng = np.random.default_rng(42) dates = pd.date_range("2022-01-01", periods=100, freq="B") tickers = ["AAPL", "TLT", "GLD"] # Note: Force the optimizer to want to trade by making AAPL (current 90%) # have terrible returns, and GLD (current 0%) have massive positive returns. # This guarantees the optimizer tries to dump AAPL and buy GLD, ensuring # we actively slam into the max turnover constraint. rets_aapl = rng.normal(-0.01, 0.015, size=100) # Heavy negative drift rets_tlt = rng.normal(0.00, 0.010, size=100) rets_gld = rng.normal(0.02, 0.015, size=100) # Massive positive drift returns_df = pd.DataFrame({'AAPL': rets_aapl, 'TLT': rets_tlt, 'GLD': rets_gld}, index=dates) bench_rets = pd.Series(rng.normal(0.0004, 0.01, size=100), index=dates) # Establish a highly constrained portfolio state cfg = copy.deepcopy(DEFAULT_CONFIG) cfg.update({ "single_asset_min": 0.0, "single_asset_max": 0.90, # Allow heavy concentration so it tries to buy 90% GLD "sector_map": {"AAPL": "Tech", "TLT": "Bonds", "GLD": "Commodity"}, "sector_limit": 1.0, "max_turnover": 0.50, # Strict turnover budget across the horizon "tax_enabled": False }) state = PortfolioState.empty(tickers) # Simulate a heavily skewed starting portfolio state.current_weights = np.array([0.90, 0.10, 0.0]) res = multi_period_optimize( returns_df, None, bench_rets, risk_input=5, risk_factor=3.0, state=state, cfg=cfg, model=1, spread_map={"AAPL": 0.0, "TLT": 0.0, "GLD": 0.0}, horizon=4, silent=True ) # Assert: The T=0 output must be a valid portfolio summing to 1.0 assert np.isclose(res.weights.sum(), 1.0) # Assert: Because Max Turnover is 50% across a 4-step horizon, the engine is capped # at a soft maximum of 30% of total budget per period (~15% turnover for the T=0 step). delta = np.abs(res.weights.drop('CASH', errors='ignore').reindex(tickers).fillna(0).values - state.current_weights).sum() # Parameterize the tolerance based on configuration max_turnover = cfg["max_turnover"] period_ceiling = max_turnover * 0.30 # Soft maximum of 30% of total budget per period # Because we forced it to want to trade massively, delta should be right up against the ceiling assert delta > period_ceiling - 0.05 # Verify it actually traded hard assert delta <= period_ceiling + 1e-3 # ───────────────────────────────────────────── # 2. MONTE CARLO SIMULATED PATHS TESTS # ───────────────────────────────────────────── def test_monte_carlo_path_geometry(): """ Verifies that the Monte Carlo engine correctly generates geometric random walks for the exact parameterized trading calendar, and that the risk-free rate compounds accurately for cash allocations. """ rng = np.random.default_rng(42) tickers = ["ASSET_A", "ASSET_B"] # Simulate an all-cash portfolio (0% risky assets, 100% Cash) weights = pd.Series({'CASH': 1.0}) exp_rets = pd.Series({'ASSET_A': 0.10, 'ASSET_B': 0.05}) cov_mat = pd.DataFrame([[0.04, 0.0], [0.0, 0.02]], index=tickers, columns=tickers) cfg = copy.deepcopy(DEFAULT_CONFIG) rfr = 0.05 # 5% Risk Free Rate cfg.update({ "risk_free_rate": rfr, "trading_days_per_year": 252, "monte_carlo_sims": 100, "monte_carlo_years": 1.0 }) capital = 100000.0 visual_paths, stats = monte_carlo(weights, exp_rets, cov_mat, capital, cfg, seed=42) # Assert: 1.0 years * 252 days = 252 steps assert len(stats['dates']) == 252 # Assert: Because the portfolio is 100% cash, there should be zero variance. # Every single percentile path (5th, 50th, 95th) should be identical. np.testing.assert_allclose(stats[5], stats[50], atol=1e-2) np.testing.assert_allclose(stats[50], stats[95], atol=1e-2) # Assert: The final value should match exactly with the geometric compounding of the RFR expected_final_value = capital * ((1 + (rfr / 252)) ** 252) assert np.isclose(stats[50][-1], expected_final_value, atol=1.0) def test_monte_carlo_distribution_spread(): """ Verifies that a risky portfolio correctly generates a spread of percentiles. """ tickers = ["ASSET_A"] weights = pd.Series({'ASSET_A': 1.0}) exp_rets = pd.Series({'ASSET_A': 0.10}) cov_mat = pd.DataFrame([[0.04]], index=tickers, columns=tickers) # 20% volatility cfg = copy.deepcopy(DEFAULT_CONFIG) cfg.update({ "trading_days_per_year": 252, "monte_carlo_sims": 1000, "monte_carlo_years": 1.0 }) capital = 100000.0 visual_paths, stats = monte_carlo(weights, exp_rets, cov_mat, capital, cfg, seed=42) # Assert: The 95th percentile outcome must be strictly greater than the 5th percentile outcome assert stats[95][-1] > stats[50][-1] assert stats[50][-1] > stats[5][-1]