ramseshk
b0fab62a87
Phase 1 - Quick Wins: - QuantileEnsemble: P10/P50/P90 predictions for risk-aware bidding - MinutesSurvivalModel: Weibull AFT for minutes distribution modeling Phase 2 - Adaptive Auction: - BanditAuctionSolver: Thompson Sampling for live auction bids - OpponentBidModel: Predict competitor bids via LightGBM - BudgetOptimizer: Bayesian optimization for role-level allocation Phase 3 - Deep Learning: - RLAuctionPolicy: Double DQN agent for auction strategy - SetTransformer: Team composition valuation via set-based ML Phase 4 - Probabilistic: - BayesianPlayerModel: Hierarchical pooling for rookie uncertainty - ConformalPredictor: Calibrated prediction intervals Phase 5 - Chemistry & Form: - PlayerChemistryGAT: Graph attention network for player synergies - PlayerFormModel: Hawkes process for form momentum Phase 6 - Causal: - TransferCausalModel: Causal forest for transfer effects - AuctionEffectAnalyzer: Bid adjustment from causal analysis 81 tests passing
1217 lines
42 KiB
Python
1217 lines
42 KiB
Python
"""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)
|