tests/test_advanced_paths.py

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]