""" VYNDR Bayesian Distribution Engine Shared by NBA and MLB. Per-stat-type weights. Similar game confidence modifier. Skewness parameter. Data sufficiency smooth degradation curve. """ import numpy as np import logging logger = logging.getLogger('vyndr') # INITIAL ESTIMATES — recalculate after 500+ resolved grades per stat type # using grid search on historical Brier scores. Store optimized weights in global_calibration. BAYESIAN_WEIGHTS = { 'strikeouts': {'prior': 0.40, 'recent': 0.40, 'context': 0.20}, 'hits': {'prior': 0.30, 'recent': 0.45, 'context': 0.25}, 'rbi': {'prior': 0.25, 'recent': 0.45, 'context': 0.30}, 'home_runs': {'prior': 0.30, 'recent': 0.35, 'context': 0.35}, 'total_bases': {'prior': 0.30, 'recent': 0.40, 'context': 0.30}, 'walks': {'prior': 0.35, 'recent': 0.40, 'context': 0.25}, 'points': {'prior': 0.35, 'recent': 0.45, 'context': 0.20}, 'rebounds': {'prior': 0.40, 'recent': 0.40, 'context': 0.20}, 'assists': {'prior': 0.30, 'recent': 0.50, 'context': 0.20}, 'threes': {'prior': 0.35, 'recent': 0.45, 'context': 0.20}, 'pts_reb_ast': {'prior': 0.35, 'recent': 0.45, 'context': 0.20}, 'default': {'prior': 0.35, 'recent': 0.45, 'context': 0.20} } # Grade scale — LOCKED GRADE_THRESHOLDS = { 'A+': (0.85, 1.00), 'A': (0.78, 0.84), 'A-': (0.72, 0.77), 'B+': (0.66, 0.71), 'B': (0.60, 0.65), 'B-': (0.55, 0.59), 'C+': (0.50, 0.54), 'C': (0.45, 0.49), 'C-': (0.40, 0.44), 'D': (0.30, 0.39), 'F': (0.00, 0.29) } ABSTENTION_RULES = { 'confidence_range': (0.40, 0.55), 'similar_games_below': 3, 'data_quality_limited': True } MIN_DATA_THRESHOLDS = { 'mlb_pitcher': {'min_starts': 3, 'min_pitches': 200}, 'mlb_batter': {'min_pa': 50, 'min_games': 12}, 'nba_player': {'min_games': 8, 'min_minutes_per_game': 15} } CALIBRATION_DISCLAIMER = ( "Model in calibration period. Confidence levels are estimated, not validated. " "Track record begins building now." ) SHADOW_MODE = True # Set to False after 2 weeks of verified accuracy def norm_cdf(x, mean, std): """ Standard normal CDF using error function. Args: x: Value to evaluate. mean: Distribution mean. std: Distribution standard deviation. Returns: Cumulative probability P(X <= x). """ if std <= 0: return 1.0 if x <= mean else 0.0 z = (x - mean) / std return 0.5 * (1 + float(np.erf(z / np.sqrt(2)))) def similar_game_confidence_modifier(count): """ Adjust confidence based on historical similar game depth. Args: count: Number of similar games found. Returns: Float adjustment to confidence (positive = boost, negative = penalty). """ if count >= 10: return 0.05 elif count >= 5: return 0.02 elif count <= 1: return -0.03 return 0.0 def calculate_bayesian_projection(prior_mean, prior_std, recent_mean, recent_std, context_adjustment, line, over_under, stat_type='default', similar_game_count=0): """ Produce a posterior distribution for a stat projection. Uses per-stat-type Bayesian weights to blend prior (season baseline), recent (last N games), and context (matchup/park/weather adjustments). Args: prior_mean: Season average for the stat. prior_std: Season standard deviation. recent_mean: Recent game average (last N games). recent_std: Recent game standard deviation. context_adjustment: Aggregate contextual adjustment value. line: Prop line to evaluate against. over_under: 'over' or 'under'. stat_type: Stat type key for weight lookup (default='default'). similar_game_count: Number of similar historical games found. Returns: Dict with projected_value, projected_std, prob_clear_line, confidence, similar_game_modifier, bayesian_weights_used, and distribution details. """ weights = BAYESIAN_WEIGHTS.get(stat_type, BAYESIAN_WEIGHTS['default']) w_prior = weights['prior'] w_recent = weights['recent'] w_context = weights['context'] posterior_mean = ( prior_mean * w_prior + recent_mean * w_recent + (prior_mean + context_adjustment) * w_context ) posterior_std = np.sqrt( (prior_std ** 2 * w_prior + recent_std ** 2 * w_recent) / (w_prior + w_recent) ) # Ensure std is positive posterior_std = max(posterior_std, 0.01) if over_under == 'over': prob = 1 - norm_cdf(line, posterior_mean, posterior_std) else: prob = norm_cdf(line, posterior_mean, posterior_std) # Similar game confidence modifier sim_modifier = similar_game_confidence_modifier(similar_game_count) prob = max(0.0, min(1.0, prob + sim_modifier)) return { 'projected_value': round(float(posterior_mean), 1), 'projected_std': round(float(posterior_std), 2), 'prob_clear_line': round(float(prob), 3), 'confidence': round(float(prob), 3), 'similar_game_modifier': sim_modifier, 'bayesian_weights_used': weights, 'distribution': { 'mean': float(posterior_mean), 'std': float(posterior_std), 'p10': round(float(posterior_mean - 1.28 * posterior_std), 1), 'p90': round(float(posterior_mean + 1.28 * posterior_std), 1) } } def calculate_skewness(game_log_values): """ Measure skew of a player's performance distribution. Positive skew = occasional blowup games (favors alt line overs). Negative skew = consistent, capped upside (favors standard line overs). Args: game_log_values: List of numeric stat values from game log. Returns: Float skewness value. Returns 0.0 if insufficient data (<10 games). """ if len(game_log_values) < 10: return 0.0 try: from scipy.stats import skew return round(float(skew(game_log_values)), 2) except ImportError: # Manual skewness calculation as fallback arr = np.array(game_log_values, dtype=float) n = len(arr) mean = np.mean(arr) std = np.std(arr, ddof=1) if std == 0: return 0.0 return round(float((n / ((n - 1) * (n - 2))) * np.sum(((arr - mean) / std) ** 3)), 2) def apply_data_sufficiency_modifier(confidence, games_played, min_games): """ Smooth confidence degradation near minimum threshold. No hard cliff at min_games — gradual ramp from 70% to 100% of confidence. Full confidence at 2x minimum games. Args: confidence: Raw confidence score. games_played: Number of games the player has played this season. min_games: Minimum games required for full confidence. Returns: Adjusted confidence score. """ if games_played < min_games: return min(confidence, 0.54) # Below minimum = C+ cap ramp = min(1.0, 0.70 + 0.30 * ((games_played - min_games) / max(min_games, 1))) return confidence * ramp def should_abstain(confidence, similar_game_count, data_quality): """ Determine if the model should abstain from grading. A C grade that misses damages credibility more than no grade at all. Args: confidence: Calculated confidence score. similar_game_count: Number of similar historical games found. data_quality: 'full', 'limited', or 'minimal'. Returns: True if model should abstain, False if grade should be published. """ low, high = ABSTENTION_RULES['confidence_range'] if low <= confidence <= high and similar_game_count < ABSTENTION_RULES['similar_games_below']: return True if data_quality == 'limited' and confidence < 0.55: return True return False def score_to_grade(score, global_offset=0.0): """ Map a confidence score to a letter grade. Args: score: Raw confidence score (0.0 to 1.0). global_offset: Calibration adjustment from grade_outcomes analysis. Applied BEFORE grade mapping. Starts at 0.0, updated monthly after 100+ resolved grades. Returns: Grade string (A+ through F). """ adjusted_score = max(0.0, min(1.0, score + global_offset)) for grade, (low, high) in GRADE_THRESHOLDS.items(): if low <= adjusted_score <= high: return grade return 'F' def calculate_global_offset(resolved_outcomes, min_resolved=100): """ Calculate global calibration offset from resolved grade outcomes. Clamped to ±0.15 to prevent overcorrection. Args: resolved_outcomes: List of dicts with 'confidence' and 'hit' keys. min_resolved: Minimum resolved grades before calculating offset. Returns: Float offset value, clamped between -0.15 and 0.15. """ if len(resolved_outcomes) < min_resolved: return 0.0 grade_accuracy = {} for grade_name, (low, high) in GRADE_THRESHOLDS.items(): grade_outcomes = [o for o in resolved_outcomes if low <= o['confidence'] <= high] if len(grade_outcomes) >= 10: hit_rate = sum(1 for o in grade_outcomes if o['hit']) / len(grade_outcomes) expected_midpoint = (low + high) / 2 grade_accuracy[grade_name] = hit_rate - expected_midpoint if not grade_accuracy: return 0.0 avg_drift = sum(grade_accuracy.values()) / len(grade_accuracy) return max(-0.15, min(0.15, avg_drift)) def calculate_brier_score(resolved_grades): """ Brier score = mean((predicted_probability - actual_outcome)^2). Lower is better. 0.0 = perfect. 0.25 = coin flip. Args: resolved_grades: List of dicts with 'confidence' and 'hit' keys. Returns: Float Brier score, or None if no data. """ if not resolved_grades: return None total = sum( (g['confidence'] - (1.0 if g['hit'] else 0.0)) ** 2 for g in resolved_grades ) return round(total / len(resolved_grades), 4) def get_disclaimer(resolved_count): """ Return calibration disclaimer if model is still in calibration period. Args: resolved_count: Number of resolved grades for the sport. Returns: Disclaimer string, or None if past calibration period. """ if resolved_count < 100: return CALIBRATION_DISCLAIMER return None