"""Tests for the new Phase 1-6 ML and optimization modules.""" import numpy as np import pandas as pd import pytest # ============================================================================= # Phase 1: Quantile Ensemble + Survival Model # ============================================================================= class TestQuantileEnsemble: def test_fit_predict(self): from src.models.quantile_model import QuantileEnsemble np.random.seed(42) X = pd.DataFrame(np.random.randn(200, 8)) X.columns = [f"f{i}" for i in range(8)] y = pd.Series(np.random.randn(200) * 1.5 + 6.5) model = QuantileEnsemble(n_estimators=50) model.fit(X, y) preds = model.predict(X) assert "P10" in preds assert "P50" in preds assert "P90" in preds assert len(preds["P10"]) == len(y) assert len(preds["P50"]) == len(y) assert len(preds["P90"]) == len(y) point_preds = model.predict_points(X) assert len(point_preds) == len(y) def test_custom_quantiles(self): from src.models.quantile_model import QuantileEnsemble np.random.seed(42) X = pd.DataFrame(np.random.randn(100, 5)) y = pd.Series(np.random.randn(100) + 6.5) model = QuantileEnsemble(quantiles=(0.05, 0.50, 0.95)) model.fit(X, y) preds = model.predict(X) assert "P5" in preds assert "P50" in preds assert "P95" in preds def test_risk_functions(self): from src.models.quantile_model import QuantileEnsemble np.random.seed(42) X = pd.DataFrame(np.random.randn(100, 5)) y = pd.Series(np.random.randn(100) * 1.5 + 6.5) model = QuantileEnsemble(n_estimators=50) model.fit(X, y) downside = model.predict_downside_risk(X, threshold=5.5) assert len(downside) == len(y) assert np.all((downside >= 0) & (downside <= 1)) upside = model.predict_upside(X, threshold=7.0) assert len(upside) == len(y) assert np.all((upside >= 0) & (upside <= 1)) var_safe = model.value_at_risk_safe(X, confidence=0.90) assert len(var_safe) == len(y) def test_save_load(self, tmp_path): from src.models.quantile_model import QuantileEnsemble np.random.seed(42) X = pd.DataFrame(np.random.randn(50, 5)) y = pd.Series(np.random.randn(50) + 6.5) model = QuantileEnsemble(n_estimators=30, model_dir=str(tmp_path)) model.fit(X, y) model.save("quantile_test.pkl") loaded = QuantileEnsemble.load("quantile_test.pkl", model_dir=str(tmp_path)) preds_orig = model.predict(X) preds_loaded = loaded.predict(X) np.testing.assert_array_almost_equal(preds_orig["P50"], preds_loaded["P50"]) class TestMinutesSurvivalModel: def test_fit_predict_scipy(self): from src.models.survival_model import MinutesSurvivalModel np.random.seed(42) n = 200 X = pd.DataFrame({ "minutes_last_3": np.random.uniform(0, 90, n), "games_last_5": np.random.randint(1, 6, n), "rest_days": np.random.uniform(2, 10, n), "fatigue_rolling_3": np.random.uniform(0, 90, n), "age": np.random.uniform(18, 38, n), "noise": np.random.randn(n), }) durations = np.clip(np.random.normal(65, 20, n), 1, 90) events = (np.random.rand(n) < 0.6).astype(int) model = MinutesSurvivalModel(force_scipy=True) model.fit(X, durations, events) expected, lower, upper = model.predict_distribution(X) assert len(expected) == n assert len(lower) == n assert len(upper) == n assert np.all(expected >= 0) assert np.all(expected <= 90) assert np.all(lower <= expected) assert np.all(expected <= upper) def test_starter_probability(self): from src.models.survival_model import MinutesSurvivalModel np.random.seed(42) n = 100 X = pd.DataFrame({ "minutes_last_3": np.random.uniform(30, 90, n), "games_last_5": np.random.randint(1, 6, n), "rest_days": np.random.uniform(3, 10, n), }) durations = np.clip(np.random.normal(70, 15, n), 1, 90) events = (np.random.rand(n) < 0.7).astype(int) model = MinutesSurvivalModel(force_scipy=True) model.fit(X, durations, events) starter_probs = model.predict_starter_probability(X, min_minutes=60) assert len(starter_probs) == n assert np.all((starter_probs >= 0) & (starter_probs <= 1)) full_probs = model.predict_full_match_probability(X) assert len(full_probs) == n assert np.all((full_probs >= 0) & (full_probs <= 1)) expected_minutes = model.predict_expected_minutes(X) assert len(expected_minutes) == n assert np.all((expected_minutes >= 0) & (expected_minutes <= 90)) def test_feature_selection(self): from src.models.survival_model import MinutesSurvivalModel np.random.seed(42) n = 80 X = pd.DataFrame({ "minutes_played": np.random.uniform(0, 90, n), "games_started": np.random.randint(0, 5, n), "fatigue_level": np.random.uniform(0, 1, n), "rest_between_games": np.random.uniform(2, 7, n), "player_age": np.random.uniform(20, 35, n), "irrelevant_cat": ["A"] * 40 + ["B"] * 40, }) durations = np.clip(np.random.normal(60, 20, n), 1, 90) events = np.random.binomial(1, 0.6, n) model = MinutesSurvivalModel(force_scipy=True) model.fit(X, durations, events) expected = model.predict_expected_minutes(X) assert len(expected) == n # ============================================================================= # Phase 4: Bayesian Pooling + Conformal Predictor # ============================================================================= class TestBayesianPlayerModel: def test_fit_predict_scipy(self): from src.models.bayesian_pooling import BayesianPlayerModel np.random.seed(42) n = 150 X = pd.DataFrame({ "role": ["P"] * 15 + ["D"] * 45 + ["C"] * 45 + ["A"] * 45, "feature1": np.random.randn(n), "feature2": np.random.randn(n), }) y = pd.Series(np.random.randn(n) * 1.0 + 6.5) model = BayesianPlayerModel() model.fit(X, y) preds = model.predict(X) assert len(preds) == n def test_predict_with_uncertainty(self): from src.models.bayesian_pooling import BayesianPlayerModel np.random.seed(42) n = 100 X = pd.DataFrame({ "role": ["D"] * 40 + ["A"] * 60, "feature": np.random.randn(n), }) y = pd.Series(np.random.randn(n) + 6.5) model = BayesianPlayerModel() model.fit(X, y) mean, std = model.predict_with_uncertainty(X) assert len(mean) == n assert len(std) == n assert np.all(std >= 0) def test_reliability_scores(self): from src.models.bayesian_pooling import BayesianPlayerModel np.random.seed(42) n = 120 X = pd.DataFrame({ "role": ["D"] * 60 + ["C"] * 60, "feature": np.random.randn(n), }) y = pd.Series(np.random.randn(n) + 6.5) model = BayesianPlayerModel() model.fit(X, y) reliability = model.get_player_reliability(X) assert len(reliability) == n assert np.all((reliability > 0) & (reliability <= 1)) def test_rookie_estimates(self): from src.models.bayesian_pooling import BayesianPlayerModel np.random.seed(42) X = pd.DataFrame({ "role": ["P"] * 10 + ["D"] * 30 + ["C"] * 30, "feature": np.random.randn(70), }) y = pd.Series(np.random.randn(70) + 6.5) model = BayesianPlayerModel() model.fit(X, y) rookies = model.get_rookie_estimates(X, min_observations=20) assert isinstance(rookies, pd.DataFrame) for col in ["player_estimate", "role_mean"]: assert col in rookies.columns def test_posterior_predictive(self): from src.models.bayesian_pooling import BayesianPlayerModel np.random.seed(42) X = pd.DataFrame({ "role": ["D"] * 30 + ["A"] * 20, "feature": np.random.randn(50), }) y = pd.Series(np.random.randn(50) + 6.5) model = BayesianPlayerModel() model.fit(X, y) samples = model.posterior_predictive(X.iloc[:10], n_samples=100) assert samples.shape == (100, 10) class TestConformalPredictor: def test_calibrate_and_predict(self): from src.models.conformal_predictor import ConformalPredictor from sklearn.linear_model import Ridge np.random.seed(42) n = 200 X = np.random.randn(n, 5) y = X[:, 0] * 2 + X[:, 1] * 1.5 + np.random.randn(n) * 0.5 X_train, X_cal, X_test = ( pd.DataFrame(X[:80]), pd.DataFrame(X[80:150]), pd.DataFrame(X[150:]), ) y_train, y_cal, y_test = y[:80], y[80:150], y[150:] base = Ridge(alpha=1.0) base.fit(X_train, y_train) cp = ConformalPredictor(base, alpha=0.10) cp.calibrate(X_cal, y_cal) y_pred, y_lower, y_upper = cp.predict_with_band(X_test) assert len(y_pred) == len(y_test) assert len(y_lower) == len(y_test) assert len(y_upper) == len(y_test) assert np.all(y_lower <= y_upper) def test_coverage(self): from src.models.conformal_predictor import ConformalPredictor from sklearn.linear_model import Ridge np.random.seed(42) n = 300 X = np.random.randn(n, 5) y = X[:, 0] * 2 + X[:, 1] * 1.5 + np.random.randn(n) * 0.5 X_train = pd.DataFrame(X[:100]) X_cal = pd.DataFrame(X[100:200]) X_test = pd.DataFrame(X[200:]) y_train, y_cal, y_test = y[:100], y[100:200], y[200:] base = Ridge(alpha=1.0) base.fit(X_train, y_train) cp = ConformalPredictor(base, alpha=0.20) cp.calibrate(X_cal, y_cal) cov = cp.coverage(X_test, y_test) assert 0.5 < cov < 1.0 def test_is_inside_band(self): from src.models.conformal_predictor import ConformalPredictor from sklearn.linear_model import LinearRegression np.random.seed(42) X = pd.DataFrame(np.random.randn(100, 3)) y = X[0] * 1.5 + np.random.randn(100) * 0.3 base = LinearRegression() base.fit(X, y) cp = ConformalPredictor(base, alpha=0.10) cp.calibrate(X, y) inside = cp.is_inside_band(X, y) assert len(inside) == len(y) assert np.all((inside == 0) | (inside == 1)) def test_update_calibration(self): from src.models.conformal_predictor import ConformalPredictor from sklearn.linear_model import Ridge np.random.seed(42) X = pd.DataFrame(np.random.randn(200, 3)) y = X[0] * 2 + np.random.randn(200) * 0.5 base = Ridge() base.fit(X, y) cp = ConformalPredictor(base, alpha=0.10) X_cal1, X_cal2 = X.iloc[:100], X.iloc[100:] y_cal1, y_cal2 = y[:100], y[100:] cp.calibrate(X_cal1, y_cal1) width_before = cp.predict_interval_width(X) cp.update(X_cal2, y_cal2) width_after = cp.predict_interval_width(X) assert isinstance(width_before, float) assert isinstance(width_after, float) # ============================================================================= # Phase 2: Bandit, Opponent Bidding, Budget Optimizer # ============================================================================= class TestBanditAuctionSolver: def test_initialization(self): from src.optimization.bandit_auction import BanditAuctionSolver solver = BanditAuctionSolver() assert solver is not None stats = solver.get_arm_stats() assert len(stats) > 0 def test_select_bid_and_update(self): from src.optimization.bandit_auction import BanditAuctionSolver from src.optimization.auction_solver import AuctionConfig, PlayerValuation config = AuctionConfig(total_budget=500) solver = BanditAuctionSolver(config=config) player = PlayerValuation( name="TestPlayer", team="TeamA", role="C", projected_points=7.5, market_value=15, ceiling_price=50, ) auction_state = { "budget_remaining": 400, "total_budget": 500, "slots_remaining": {"P": 1, "D": 3, "C": 4, "A": 3}, "role_quotas": {"P": 3, "D": 8, "C": 8, "A": 6}, "slot_quotas": {"P": 3, "D": 8, "C": 8, "A": 6}, "round_number": 3, "total_rounds": 10, "opponent_budgets": [400, 450, 350], "players_remaining_in_role": {"P": 5, "D": 10, "C": 8, "A": 6}, "player_pool": [player], } arm_idx, bid_amount = solver.select_bid(player, auction_state) assert 0 <= arm_idx < len(solver.bid_fractions) assert bid_amount >= 0 solver.update(arm_idx, reward=0.8, player_role="C") stats = solver.get_arm_stats() assert stats[arm_idx]["trials"] >= 1 def test_recommend_bid_summary(self): from src.optimization.bandit_auction import BanditAuctionSolver from src.optimization.auction_solver import AuctionConfig, PlayerValuation solver = BanditAuctionSolver() player = PlayerValuation( "TestPlayer", "TeamA", "A", 8.0, 20, 60, ) auction_state = { "budget_remaining": 300, "total_budget": 500, "slots_remaining": {"P": 2, "D": 5, "C": 5, "A": 2}, "role_quotas": {"P": 3, "D": 8, "C": 8, "A": 6}, "slot_quotas": {"P": 3, "D": 8, "C": 8, "A": 6}, "round_number": 1, "total_rounds": 10, "opponent_budgets": [400], "players_remaining_in_role": {"P": 6, "D": 15, "C": 15, "A": 10}, "player_pool": [player], } summary = solver.recommend_bid_summary(player, auction_state) assert "recommended_bid" in summary class TestOpponentBidModel: def test_fit_predict_heuristic(self): from src.optimization.opponent_bidding_model import OpponentBidModel np.random.seed(42) model = OpponentBidModel() players_df = pd.DataFrame({ "name": [f"Player_{i}" for i in range(20)], "role": np.random.choice(["P", "D", "C", "A"], 20), "projected_points": np.random.uniform(5, 9, 20), }) opponent_state = { "budget_remaining": 400, "total_budget": 500, "slots_remaining": {"P": 2, "D": 6, "C": 6, "A": 4}, "role_quotas": {"P": 3, "D": 8, "C": 8, "A": 6}, } bids = model.predict_opponent_bids(players_df, opponent_state) assert len(bids) == 20 assert np.all(bids >= 0) def test_p_acquire(self): from src.optimization.opponent_bidding_model import OpponentBidModel np.random.seed(42) model = OpponentBidModel() players_df = pd.DataFrame({ "name": ["Player_A", "Player_B", "Player_C"], "role": ["A", "D", "C"], "projected_points": [8.5, 7.0, 6.5], }) opponent_state = { "budget_remaining": 350, "total_budget": 500, "slots_remaining": {"P": 2, "D": 5, "C": 5, "A": 2}, "role_quotas": {"P": 3, "D": 8, "C": 8, "A": 6}, } my_bids = pd.Series([40, 25, 15], index=players_df.index) probs = model.predict_p_acquire(players_df, my_bids, opponent_state) assert len(probs) == 3 assert np.all((probs >= 0) & (probs <= 1)) def test_simulate_live_round(self): from src.optimization.opponent_bidding_model import OpponentBidModel np.random.seed(42) model = OpponentBidModel() players_df = pd.DataFrame({ "name": [f"Player_{i}" for i in range(5)], "role": ["P"] + ["D"] * 2 + ["C"] + ["A"], "projected_points": np.random.uniform(5, 9, 5), }) opponent_state = { "budget_remaining": 350, "total_budget": 500, "slots_remaining": {"P": 2, "D": 6, "C": 6, "A": 4}, "role_quotas": {"P": 3, "D": 8, "C": 8, "A": 6}, } result = model.simulate_live_round(players_df, 250, opponent_state, n_sims=50) assert "expected_cost" in result assert "value_matrix" in result assert result["expected_cost"] >= 0 class TestBudgetOptimizer: def test_optimize_fallback(self): from src.optimization.budget_optimizer import BudgetOptimizer np.random.seed(42) optimizer = BudgetOptimizer(total_budget=500) players = [] for i in range(60): role = np.random.choice(["P", "D", "C", "A"]) players.append({ "name": f"Player_{i}", "role": role, "projected_points": np.random.uniform(5, 9), "market_value": np.random.randint(3, 40), }) player_pool = pd.DataFrame(players) allocation = optimizer.optimize(player_pool, n_calls=10) assert "P" in allocation assert "D" in allocation assert "C" in allocation assert "A" in allocation total = sum(allocation.values()) assert abs(total - 500) < 10 def test_optimize_adaptive(self): from src.optimization.budget_optimizer import BudgetOptimizer np.random.seed(42) optimizer = BudgetOptimizer(total_budget=500) players = [] for i in range(50): players.append({ "name": f"Player_{i}", "role": np.random.choice(["P", "D", "C", "A"]), "projected_points": np.random.uniform(5, 9), "market_value": np.random.randint(3, 40), }) player_pool = pd.DataFrame(players) allocation = optimizer.optimize_adaptive( player_pool, remaining_slots={"P": 2, "D": 4, "C": 5, "A": 3}, spent_per_role={"P": 10, "D": 60, "C": 40, "A": 30}, n_calls=10, ) assert isinstance(allocation, dict) for role in ["P", "D", "C", "A"]: assert allocation[role] >= 0 def test_role_value_curves(self): from src.optimization.budget_optimizer import BudgetOptimizer np.random.seed(42) optimizer = BudgetOptimizer(total_budget=500) players = [] for i in range(80): role = np.random.choice(["P", "D", "C", "A"]) players.append({ "name": f"Player_{i}", "role": role, "projected_points": np.random.uniform(5, 9), "market_value": np.random.randint(3, 40), }) player_pool = pd.DataFrame(players) curves = optimizer.get_role_value_curves(player_pool) for role in ["P", "D", "C", "A"]: assert role in curves assert len(curves[role]) > 0 def test_sensitivity_analysis(self): from src.optimization.budget_optimizer import BudgetOptimizer np.random.seed(42) optimizer = BudgetOptimizer(total_budget=500) players = [] for i in range(60): players.append({ "name": f"Player_{i}", "role": np.random.choice(["P", "D", "C", "A"]), "projected_points": np.random.uniform(5, 9), "market_value": np.random.randint(3, 40), }) player_pool = pd.DataFrame(players) analysis = optimizer.sensitivity_analysis(player_pool, n_calls=5) assert isinstance(analysis, dict) # ============================================================================= # Phase 5: GAT Chemistry + Hawkes Form # ============================================================================= class TestPlayerChemistryGAT: def test_build_graph(self): from src.models.gat_model import PlayerChemistryGAT np.random.seed(42) data = pd.DataFrame([ {"player": "A", "teammate": "B", "passes_to": 10, "assists_to": 2, "crosses_to": 3, "matchday": 1}, {"player": "A", "teammate": "C", "passes_to": 15, "assists_to": 1, "crosses_to": 5, "matchday": 1}, {"player": "B", "teammate": "A", "passes_to": 8, "assists_to": 0, "crosses_to": 2, "matchday": 1}, {"player": "B", "teammate": "C", "passes_to": 12, "assists_to": 3, "crosses_to": 4, "matchday": 2}, ]) model = PlayerChemistryGAT() model.build_graph(data) assert len(model.interaction_edges) >= 0 def test_extract_interaction_features(self): from src.models.gat_model import PlayerChemistryGAT np.random.seed(42) data = pd.DataFrame([ {"player": "A", "teammate": "B", "passes_to": 10, "assists_to": 2, "crosses_to": 3, "matchday": 1}, {"player": "A", "teammate": "C", "passes_to": 5, "assists_to": 0, "crosses_to": 1, "matchday": 1}, {"player": "B", "teammate": "C", "passes_to": 3, "assists_to": 1, "crosses_to": 0, "matchday": 2}, ]) model = PlayerChemistryGAT() model.build_graph(data) features = model.extract_interaction_features("A", ["B", "C"]) assert "interaction_outgoing_sum" in features assert "interaction_incoming_sum" in features assert "interaction_synergy" in features def test_compute_interaction_bonus(self): from src.models.gat_model import PlayerChemistryGAT data = pd.DataFrame([ {"player": "A", "teammate": "B", "passes_to": 20, "assists_to": 4, "crosses_to": 8, "matchday": 1}, {"player": "A", "teammate": "C", "passes_to": 5, "assists_to": 0, "crosses_to": 1, "matchday": 1}, ]) model = PlayerChemistryGAT() model.build_graph(data) bonus_high = model.compute_interaction_bonus("A", "B") bonus_low = model.compute_interaction_bonus("A", "C") assert bonus_high > bonus_low assert 0 <= bonus_high <= 1 assert 0 <= bonus_low <= 1 def test_fit_predict_sklearn(self): from src.models.gat_model import PlayerChemistryGAT np.random.seed(42) n = 100 data = [] n_real = np.random.RandomState(42) for i in range(n): for j in range(n): if i < j and n_real.random() < 0.05: data.append({ "player": f"P{i}", "teammate": f"P{j}", "passes_to": n_real.randint(0, 10), "assists_to": n_real.randint(0, 3), "crosses_to": n_real.randint(0, 5), "matchday": n_real.randint(1, 20), }) X = pd.DataFrame({ "player": [f"P{i}" for i in range(n)], "team": ["T1"] * (n // 2) + ["T2"] * (n - n // 2), "feature1": np.random.randn(n), "feature2": np.random.randn(n), }) y = pd.Series(np.random.randn(n) * 1.5 + 6.5) model = PlayerChemistryGAT() model.build_graph(pd.DataFrame(data)) model.fit(X, y) preds = model.predict(X) assert len(preds) == n def test_get_redundancy_penalty(self): from src.models.gat_model import PlayerChemistryGAT model = PlayerChemistryGAT() players = ["Player_A", "Player_B", "Player_C"] penalties = model.get_redundancy_penalty(players) assert isinstance(penalties, dict) class TestPlayerFormModel: def test_fit_predict(self): from src.models.hawkes_form import PlayerFormModel np.random.seed(42) n = 150 base_dates = pd.date_range("2023-08-20", periods=38, freq="7D") X = pd.DataFrame({ "player": [f"P{i % 10}" for i in range(n)], "match_date": np.random.choice(base_dates, n), "minutes": np.random.uniform(0, 90, n), "feature1": np.random.randn(n), }) y = pd.Series(np.random.randn(n) * 1.5 + 6.5) model = PlayerFormModel() model.fit(X, y) preds = model.predict(X) assert len(preds) == n def test_form_status(self): from src.models.hawkes_form import PlayerFormModel np.random.seed(42) base_dates = pd.date_range("2023-08-20", periods=20, freq="7D") X = pd.DataFrame({ "player": ["TestPlayer"] * 20, "match_date": base_dates, "minutes": np.random.uniform(30, 90, 20), "feature1": np.random.randn(20), }) y = pd.Series(np.random.randn(20) + 6.5) model = PlayerFormModel() model.fit(X, y) form_status = model.get_form_status(X) assert len(form_status) == 20 for status in form_status: assert status in ("HOT", "COLD", "NEUTRAL") def test_detect_streak(self): from src.models.hawkes_form import PlayerFormModel np.random.seed(42) base_dates = pd.date_range("2023-08-20", periods=15, freq="7D") X = pd.DataFrame({ "player": ["StreakyP"] * 15, "match_date": base_dates, "minutes": np.random.uniform(50, 90, 15), "feature1": np.random.randn(15), }) y = pd.Series(np.random.randn(15) * 1.5 + 7.5) model = PlayerFormModel() model.fit(X, y) is_streak, length, direction = model.detect_streak(X) assert isinstance(is_streak, bool) assert isinstance(length, int) assert direction in ("HOT_STREAK", "COLD_STREAK", "NO_STREAK") def test_momentum_projection(self): from src.models.hawkes_form import PlayerFormModel np.random.seed(42) base_dates = pd.date_range("2023-08-20", periods=10, freq="7D") X = pd.DataFrame({ "player": ["Player1"] * 10, "match_date": base_dates, "minutes": np.random.uniform(40, 90, 10), "feature1": np.random.randn(10), }) y = pd.Series(np.random.randn(10) * 1.5 + 6.5) model = PlayerFormModel() model.fit(X, y) momentum = model.predict_momentum(X.iloc[:5], X.iloc[:5], n_future=3) assert momentum.shape == (3, 5) # ============================================================================= # Phase 3: RL Auction Agent + Set Transformer # ============================================================================= class TestAuctionEnv: def test_reset_and_step(self): from src.optimization.rl_auction_agent import AuctionEnv from src.optimization.auction_solver import AuctionConfig np.random.seed(42) config = AuctionConfig(total_budget=500) config.n_gk = 2 config.n_def = 5 config.n_mid = 5 config.n_fwd = 3 players = [] for i in range(40): players.append({ "name": f"Player_{i}", "role": np.random.choice(["P", "D", "C", "A"]), "projected_points": np.random.uniform(5, 9), "market_value": np.random.randint(3, 40), "team": f"Team_{np.random.randint(1, 21)}", "ceiling_price": np.random.uniform(20, 80), }) player_pool = pd.DataFrame(players) env = AuctionEnv(player_pool, n_opponents=3, config=config) obs = env.reset() assert isinstance(obs, np.ndarray) assert len(obs) > 0 obs, reward, done, info = env.step(5) assert isinstance(obs, np.ndarray) assert isinstance(reward, float) assert isinstance(done, bool) def test_valid_actions(self): from src.optimization.rl_auction_agent import AuctionEnv from src.optimization.auction_solver import AuctionConfig np.random.seed(42) config = AuctionConfig(total_budget=500) config.n_gk = 1 config.n_def = 2 config.n_mid = 2 config.n_fwd = 1 players = [] for i in range(15): players.append({ "name": f"Player_{i}", "role": np.random.choice(["P", "D", "C", "A"]), "projected_points": np.random.uniform(5, 9), "market_value": np.random.randint(3, 40), "team": f"Team_{np.random.randint(1, 21)}", "ceiling_price": np.random.uniform(20, 80), }) player_pool = pd.DataFrame(players) env = AuctionEnv(player_pool, n_opponents=2, config=config) env.reset() valid = env.get_valid_actions() assert len(valid) > 0 env_no_budget = AuctionEnv(player_pool, n_opponents=2, config=config) env_no_budget.reset() env_no_budget.budget_remaining = 0 valid_no_money = env_no_budget.get_valid_actions() assert valid_no_money[0] == 0 class TestRLAuctionPolicy: def test_act_and_remember(self): from src.optimization.rl_auction_agent import RLAuctionPolicy policy = RLAuctionPolicy(state_dim=14, action_dim=11) state = np.random.randn(14).astype(np.float32) action = policy.act(state, epsilon=1.0) assert 0 <= action < 11 next_state = np.random.randn(14).astype(np.float32) policy.remember(state, action, reward=1.5, next_state=next_state, done=False) assert len(policy.replay_buffer) == 1 def test_replay_and_target_update(self): from src.optimization.rl_auction_agent import RLAuctionPolicy policy = RLAuctionPolicy(state_dim=14, action_dim=11, hidden_dim=64) for _ in range(256): s = np.random.randn(14).astype(np.float32) a = np.random.randint(0, 11) r = np.random.randn() ns = np.random.randn(14).astype(np.float32) d = np.random.rand() < 0.3 policy.remember(s, a, r, ns, d) loss = policy.replay(batch_size=64) assert loss is not None assert isinstance(loss, float) def test_save_load(self, tmp_path): from src.optimization.rl_auction_agent import RLAuctionPolicy policy = RLAuctionPolicy(state_dim=14, action_dim=11, hidden_dim=32) save_path = str(tmp_path / "test_policy.pkl") policy.save(save_path) loaded = RLAuctionPolicy.load(save_path) assert loaded.state_dim == 14 assert loaded.action_dim == 11 def test_training_loop(self): from src.optimization.rl_auction_agent import ( RLAuctionPolicy, AuctionEnv, RLAuctionTrainer, ) from src.optimization.auction_solver import AuctionConfig np.random.seed(42) config = AuctionConfig(total_budget=500) config.n_gk = 1 config.n_def = 3 config.n_mid = 3 config.n_fwd = 2 players = [] for i in range(20): players.append({ "name": f"Player_{i}", "role": np.random.choice(["P", "D", "C", "A"]), "projected_points": np.random.uniform(5, 9), "market_value": np.random.randint(3, 40), "team": f"Team_{np.random.randint(1, 10)}", "ceiling_price": np.random.uniform(20, 80), }) player_pool = pd.DataFrame(players) trainer = RLAuctionTrainer(player_pool, n_opponents=2, config=config) agent = trainer.train(n_episodes=20, verbose=False) assert agent is not None assert hasattr(agent, "q_network") class TestRLAuctionTrainer: def test_evaluate_vs_baselines(self): from src.optimization.rl_auction_agent import ( RLAuctionPolicy, AuctionEnv, RLAuctionTrainer, ) from src.optimization.auction_solver import AuctionConfig np.random.seed(42) config = AuctionConfig(total_budget=500) config.n_gk = 1 config.n_def = 2 config.n_mid = 2 config.n_fwd = 1 players = [] for i in range(15): players.append({ "name": f"Player_{i}", "role": np.random.choice(["P", "D", "C", "A"]), "projected_points": np.random.uniform(5, 9), "market_value": np.random.randint(3, 40), "team": f"Team_{np.random.randint(1, 10)}", "ceiling_price": np.random.uniform(20, 80), }) player_pool = pd.DataFrame(players) agent = RLAuctionPolicy(state_dim=14, action_dim=11, hidden_dim=32) trainer = RLAuctionTrainer(player_pool, n_opponents=2, config=config) results = trainer.evaluate_vs_baselines(agent, player_pool, n_sims=5) assert "rl_total_value" in results or "greedy_total_value" in results class TestSetTransformer: def test_fit_predict_sklearn(self): from src.models.set_transformer import SetTransformer np.random.seed(42) n_teams = 20 n_players_per_team = 25 teams_data = [] team_values = [] for t in range(n_teams): df = pd.DataFrame({ "name": [f"Team{t}_Player_{i}" for i in range(n_players_per_team)], "role": np.random.choice(["P", "D", "C", "A"], n_players_per_team), "feature1": np.random.randn(n_players_per_team), "feature2": np.random.randn(n_players_per_team), "projected_points": np.random.uniform(5, 9, n_players_per_team), }) teams_data.append(df) team_values.append(np.sum(df["projected_points"]) + np.random.randn() * 10) model = SetTransformer(use_torch=False) model.fit(teams_data, team_values) pred = model.predict(teams_data[0]) assert isinstance(pred, float) def test_value_added_and_removed(self): from src.models.set_transformer import SetTransformer np.random.seed(42) n_teams = 15 n_players = 25 teams_data = [] team_values = [] for t in range(n_teams): df = pd.DataFrame({ "name": [f"T{t}_P{i}" for i in range(n_players)], "role": np.random.choice(["P", "D", "C", "A"], n_players), "feature1": np.random.randn(n_players), "projected_points": np.random.uniform(5, 9, n_players), }) teams_data.append(df) team_values.append(np.sum(df["projected_points"]) + np.random.randn() * 5) model = SetTransformer(use_torch=False) model.fit(teams_data, team_values) new_player = pd.DataFrame([{ "name": "NewPlayer", "role": "A", "feature1": 1.5, "projected_points": 8.5, }]) va = model.value_added(teams_data[0], new_player) vr = model.value_removed(teams_data[0], "T0_P0") assert isinstance(va, float) assert isinstance(vr, float) def test_optimal_replacement(self): from src.models.set_transformer import SetTransformer np.random.seed(42) n_teams = 10 n_players = 20 teams_data = [] team_values = [] for t in range(n_teams): df = pd.DataFrame({ "name": [f"T{t}_P{i}" for i in range(n_players)], "role": np.random.choice(["P", "D", "C", "A"], n_players), "feature1": np.random.randn(n_players), "projected_points": np.random.uniform(5, 9, n_players), }) teams_data.append(df) team_values.append(np.sum(df["projected_points"]) + np.random.randn() * 5) model = SetTransformer(use_torch=False) model.fit(teams_data, team_values) pool = pd.DataFrame([ {"name": f"Free_{i}", "role": np.random.choice(["P", "D", "C", "A"]), "feature1": np.random.randn(), "projected_points": np.random.uniform(5, 9)} for i in range(10) ]) rankings = model.optimal_replacement(teams_data[0], pool, to_replace=["T0_P0"]) assert isinstance(rankings, dict) assert "T0_P0" in rankings def test_redundancy_score(self): from src.models.set_transformer import SetTransformer np.random.seed(42) n_teams = 10 n_players = 20 teams_data = [] team_values = [] for t in range(n_teams): df = pd.DataFrame({ "name": [f"T{t}_P{i}" for i in range(n_players)], "role": np.random.choice(["P", "D", "C", "A"], n_players), "feature1": np.random.randn(n_players), "projected_points": np.random.uniform(5, 9, n_players), }) teams_data.append(df) team_values.append(np.sum(df["projected_points"]) + np.random.randn() * 5) model = SetTransformer(use_torch=False) model.fit(teams_data, team_values) score = model.get_redundancy_score(teams_data[0]) assert 0 <= score <= 1 # ============================================================================= # Phase 6: Causal Forest # ============================================================================= class TestTransferCausalModel: def test_fit_predict_sklearn(self): from src.models.causal_forest import TransferCausalModel np.random.seed(42) n = 200 X = pd.DataFrame({ "role": np.random.choice(["P", "D", "C", "A"], n), "feature1": np.random.randn(n), "feature2": np.random.randn(n), "team_strength": np.random.uniform(0.5, 1.5, n), }) treatment = pd.DataFrame({ "role": np.random.choice(["P", "D", "C", "A"], n), "projected_points": np.random.uniform(5, 9, n), "days_since_last_transfer": np.random.randint(1, 30, n), }) true_effect = treatment["projected_points"] * 0.5 + np.random.randn(n) * 0.3 outcome = pd.Series(true_effect + np.random.randn(n) * 1.0) model = TransferCausalModel() model.fit(X, treatment, outcome) result = model.predict_effect(X, treatment) assert "ate" in result assert "cate_lower" in result def test_predict_individual_effect(self): from src.models.causal_forest import TransferCausalModel np.random.seed(42) n = 150 X = pd.DataFrame({ "role": np.random.choice(["P", "D", "C", "A"], n), "feature1": np.random.randn(n), "team_strength": np.random.uniform(0.5, 1.5, n), }) treatment = pd.DataFrame({ "role": np.random.choice(["P", "D", "C", "A"], n), "projected_points": np.random.uniform(5, 9, n), "days_since_last_transfer": np.random.randint(1, 30, n), }) outcome = pd.Series(np.random.randn(n) + 6.5) model = TransferCausalModel() model.fit(X, treatment, outcome) roster = pd.DataFrame({ "name": ["P1", "D1", "D2", "D3", "M1", "M2", "M3", "M4", "F1", "F2"], "role": ["P"] + ["D"] * 3 + ["C"] * 4 + ["A"] * 2, "projected_points": np.random.uniform(5, 9, 10), "team": ["TeamA"] * 10, }) candidate = pd.DataFrame([{ "name": "NewPlayer", "role": "C", "projected_points": 7.5, "team": "Available", }]) result = model.predict_individual_effect(roster, candidate, role_to_replace="M1") assert "effect" in result assert "confidence" in result def test_rank_transfers(self): from src.models.causal_forest import TransferCausalModel np.random.seed(42) n = 120 X = pd.DataFrame({ "role": np.random.choice(["P", "D", "C", "A"], n), "feature1": np.random.randn(n), "team_strength": np.random.uniform(0.5, 1.5, n), }) treatment = pd.DataFrame({ "role": np.random.choice(["P", "D", "C", "A"], n), "projected_points": np.random.uniform(5, 9, n), "days_since_last_transfer": np.random.randint(1, 30, n), }) outcome = pd.Series(np.random.randn(n) + 6.5) model = TransferCausalModel() model.fit(X, treatment, outcome) roster = pd.DataFrame({ "name": ["P1"] + [f"D{i}" for i in range(5)] + [f"M{i}" for i in range(5)] + [f"F{i}" for i in range(4)], "role": ["P"] + ["D"] * 5 + ["C"] * 5 + ["A"] * 4, "projected_points": np.random.uniform(5, 9, 15), "team": ["T1"] * 15, }) pool = pd.DataFrame([ {"name": f"Free_{i}", "role": np.random.choice(["P", "D", "C", "A"]), "projected_points": np.random.uniform(5, 9), "team": "Free"} for i in range(20) ]) ranked = model.rank_transfers(roster, pool, n_recommendations=5) assert isinstance(ranked, pd.DataFrame) assert len(ranked) <= 5 def test_auction_effect_analyzer(self): from src.models.causal_forest import TransferCausalModel, AuctionEffectAnalyzer np.random.seed(42) n = 100 X = pd.DataFrame({ "role": np.random.choice(["P", "D", "C", "A"], n), "feature1": np.random.randn(n), "team_strength": np.random.uniform(0.5, 1.5, n), }) treatment = pd.DataFrame({ "role": np.random.choice(["P", "D", "C", "A"], n), "projected_points": np.random.uniform(5, 9, n), "days_since_last_transfer": np.random.randint(1, 30, n), }) outcome = pd.Series(np.random.randn(n) + 6.5) model = TransferCausalModel() model.fit(X, treatment, outcome) roster = pd.DataFrame({ "name": [f"P{i}" for i in range(15)], "role": np.random.choice(["P", "D", "C", "A"], 15), "projected_points": np.random.uniform(5, 9, 15), "team": ["T1"] * 15, }) analyzer = AuctionEffectAnalyzer(model, roster) assert analyzer is not None player = pd.Series({ "name": "Target", "role": "C", "projected_points": 7.8, "team": "Available", }) result = analyzer.recommend_bid_adjustment(player, 25, roster) assert "adjusted_bid" in result assert isinstance(result["adjusted_bid"], float)