tests/test_reproducibility.py

import os
# Pin thread counts BEFORE any numeric library import to ensure deterministic
# BLAS operations in scikit-learn's ElasticNetCV and XGBoost.
os.environ.setdefault("OMP_NUM_THREADS", "1")
os.environ.setdefault("OPENBLAS_NUM_THREADS", "1")
os.environ.setdefault("MKL_NUM_THREADS", "1")

import copy
import numpy as np
import pandas as pd
import pytest
from core_types import PortfolioState
from solver import build_and_optimize
from config import DEFAULT_CONFIG

def test_xgboost_determinism_across_5_runs():
    """
    Runs 5 complete ML-Stacking (Model 5) optimizations using the exact same inputs
    and ensures the weights and returns are absolutely identical.
    This protects against XGBoost thread-ordering drift.
    """
    rng = np.random.default_rng(42)
    dates = pd.date_range("2020-01-01", periods=252*3, freq="B")
    tickers = ["AAPL", "TLT", "JPM"]
    
    returns_df = pd.DataFrame(
        rng.normal(0.0004, 0.015, size=(len(dates), len(tickers))), 
        index=dates, columns=tickers
    )
    bench_rets = pd.Series(rng.normal(0.0003, 0.012, size=len(dates)), index=dates)

    cfg = copy.deepcopy(DEFAULT_CONFIG)
    cfg.update({
        "garch_enabled": False, 
        "cvar_enabled": False,
        "tax_enabled": False,
        "dynamic_risk": False,
        "hmm_regime": False,
        "sector_map": {t: "Other" for t in tickers},
        "sector_limit": 1.0,
        "single_asset_min": 0.0,
        "single_asset_max": 0.60,
    })

    runs = 5
    results = []

    for i in range(runs):
        opt_res = build_and_optimize(
            returns_df=returns_df,
            benchmark_rets=bench_rets,
            risk_input=5,
            risk_factor=3.0,
            state=PortfolioState.empty(tickers),
            cfg=copy.deepcopy(cfg),
            model=5,  # Global Pooled Panel Machine Learning
            allocation_engine=1,
            ff_df=None,
            spread_map={t: 0.0005 for t in tickers},
            silent=True,
        )
        results.append((opt_res.weights.values, opt_res.expected_returns.values, opt_res.covariance_matrix.values))

    # Assert exact match across all 5 runs
    base_w, base_r, base_c = results[0]
    for i in range(1, runs):
        curr_w, curr_r, curr_c = results[i]
        
        # Max acceptable drift is extremely tight (float rounding tolerance only)
        assert np.allclose(base_w, curr_w, atol=1e-8, rtol=1e-8), f"Weight drift detected on run {i}"
        assert np.allclose(base_r, curr_r, atol=1e-8, rtol=1e-8), f"Return drift detected on run {i}"
        assert np.allclose(base_c, curr_c, atol=1e-8, rtol=1e-8), f"Covariance drift detected on run {i}"