rendered paste body# core.pyimport numpy as npfrom scipy.stats import poissonfrom league_modules import module_registry, MODULE_EXECUTION_ORDERfrom rule_engine import RuleEngine, Contextclass FootballPredictor: def __init__(self, **kwargs): self.params = kwargs defaults = { 'home_control_rate': 0.50, 'away_press_rate': 0.25, 'home_defense_loss_rate': 0.15, 'away_counter_attacks': 2.0, 'home_stadium_condition': 'normal', 'is_old_team_return': False, 'is_top_club': False, 'var_controversy': False, 'big_ball_rate_home': 0.50, 'big_ball_rate_away': 0.50, 'home_team': 'Home', 'away_team': 'Away', 'home_shots_avg': 0, 'away_shots_avg': 0, 'home_possession': 0.50, 'away_possession': 0.50, 'odds_home_handicap': 0, 'odds_draw_handicap': 0, 'odds_away_handicap': 0, } for key, val in defaults.items(): if key not in self.params: self.params[key] = val self.league_config = kwargs.get('league_config', {}) self.modules_cfg = self.league_config.get('modules', {}) self.top_clubs = self.league_config.get('top_clubs', []) self.odds_h = self.params['odds_home'] self.odds_d = self.params['odds_draw'] self.odds_a = self.params['odds_away'] self.round = self.params['season_round'] self.context = Context() self._init_context_from_config() self.result = {} def _init_context_from_config(self): cfg = self.league_config.get('modules', {}) self.context.set('rdi_threshold', cfg.get('rdi', {}).get('threshold', 0.30)) self.context.set('uat_threshold', cfg.get('uat', {}).get('threshold', 0.75)) self.context.set('obd_threshold', cfg.get('odds_deviation', {}).get('threshold', 0.15)) self.context.set('ds_threshold', cfg.get('dynamic_pressure', {}).get('threshold', 420)) self.context.set('acf_threshold', cfg.get('adversity', {}).get('threshold', 1.00)) self.context.set('q_attack_boost', 1.0) self.context.set('fpa_boost', 1.0) self.context.set('kpa_boost', 1.0) self.context.set('tcm_boost', 1.0) self.context.set('dta_boost', 1.0) self.context.set('pwl_boost', 1.0) self.context.set('splitter_boost', 1.0) self.context.set('wfl_triggered', False) self.context.set('orc_triggered', False) self.context.set('btp_triggered', False) self.context.set('hfa_triggered', False) self.context.set('obd_triggered', False) self.context.set('drc_triggered', False) self.context.set('rdi_triggered', False) self.context.set('uat_triggered', False) self.context.set('score_corrections', {}) def _calc_match_probabilities(self, xg_h, xg_a, max_goals=6): ph = [poisson.pmf(k, xg_h) for k in range(max_goals+1)] pa = [poisson.pmf(k, xg_a) for k in range(max_goals+1)] score_probs = {} win_h = draw = win_a = 0.0 for i in range(max_goals+1): for j in range(max_goals+1): p = ph[i] * pa[j] score_probs[(i,j)] = p if i > j: win_h += p elif i == j: draw += p else: win_a += p total = win_h + draw + win_a if total > 0: win_h /= total draw /= total win_a /= total best_score = max(score_probs, key=score_probs.get) best_prob = score_probs[best_score] sorted_scores = sorted(score_probs.items(), key=lambda x: -x[1]) return { 'win_h': win_h, 'draw': draw, 'win_a': win_a, 'best_score': best_score, 'best_prob': best_prob, 'score_probs': score_probs, 'sorted_scores': sorted_scores } def _generate_score_matrix(self, sorted_scores): core = sorted_scores[0][0] if sorted_scores else (0,0) defend = [] for (i,j), p in sorted_scores: if i+j <= 2: defend.append((i,j)) if len(defend) >= 3: break if not defend: defend = [(0,0), (1,0), (1,1)] open_attack_cfg = self.league_config.get('open_attack', {}) threshold = open_attack_cfg.get('threshold', 0.50) big_ball_h = self.params.get('big_ball_rate_home', 0.50) big_ball_a = self.params.get('big_ball_rate_away', 0.50) open_attack = (big_ball_h > threshold and big_ball_a > threshold) total_xg = self.result.get('xg_h_after_A', 0) + self.result.get('xg_a_after_A', 0) if not open_attack and total_xg > 2.8: open_attack = True self.result['_debug']['open_attack_forced'] = True attack = [] if open_attack: for (i,j), p in sorted_scores: if i+j >= 3: attack.append((i,j)) if len(attack) >= 3: break return {'core': core, 'defend': defend, 'attack': attack, 'open_attack': open_attack} def _generate_strategy(self): win_h = self.result['prob_win_h'] draw = self.result['prob_draw'] win_a = self.result['prob_win_a'] strategy = {} if win_h > 0.5: strategy['Safe'] = f"Home Win ({self.odds_h:.2f})" elif draw > 0.4: strategy['Safe'] = f"Draw ({self.odds_d:.2f})" elif win_a > 0.5: strategy['Safe'] = f"Away Win ({self.odds_a:.2f})" else: strategy['Safe'] = "Double chance: Draw/Away" if self.odds_h < 1.30: strategy['Balanced'] = "Asian Handicap -1 / -1.5" elif 1.90 <= self.odds_h <= 2.05 and self.result.get('ewc_trigger', False): strategy['Balanced'] = "Home Win + Under 2.5" elif 2.00 <= self.odds_h <= 2.30: strategy['Balanced'] = "Home Win or Draw" else: strategy['Balanced'] = "Home Win / Draw" total_xg = self.result['xg_h_after_A'] + self.result['xg_a_after_A'] if total_xg < 2.0: strategy['Goals'] = "1, 2" elif total_xg < 3.0: strategy['Goals'] = "2, 3" else: strategy['Goals'] = "3, 4" half1 = self._calc_match_probabilities(self.result['xg_h_half1'], self.result['xg_a_half1']) half2 = self._calc_match_probabilities(self.result['xg_h_half2'], self.result['xg_a_half2']) h1_state = 'W' if half1['win_h'] > 0.4 else ('D' if half1['draw'] > 0.4 else 'L') h2_state = 'W' if half2['win_h'] > 0.4 else ('D' if half2['draw'] > 0.4 else 'L') strategy['HT/FT'] = f"{h1_state}/{h2_state}" self.result['strategy'] = strategy def _generate_tags(self): tags = [] if self.result.get('ewc_trigger', False): tags.append('Economic Win') if self.result.get('dsr_iron_draw', False): tags.append('Draw Alert') if self.odds_h < 1.30: tags.append('Deep Odds') if self.params['home_stadium_condition'] != 'normal': tags.append('Home Discount') if self.result.get('vtd_trap', False): tags.append('Trap Warning') if self.result.get('bpr_triggered', False): tags.append('Counter Risk') if self.result.get('open_attack', False): tags.append('Attack Mode') if self.context.get('hfa_triggered'): tags.append('HFA Boost') if self.context.get('btp_triggered'): tags.append('Clutch Alert') if self.context.get('obd_triggered'): tags.append('Odds Deviation') if self.context.get('orc_triggered'): tags.append('ORC Alert') if self.context.get('drc_triggered'): tags.append('DRC Alert') if self.context.get('wfl_triggered'): tags.append('WFL Signal') if not tags: tags.append('Regular') self.result['tags'] = tags def _apply_injury_adjustment(self, xg_h, xg_a): home_injuries = self.params.get('home_injuries', 0) away_injuries = self.params.get('away_injuries', 0) home_key_missing = self.params.get('home_key_missing', 0) away_key_missing = self.params.get('away_key_missing', 0) injury_cfg = self.league_config.get('injury', {}) home_penalty = (home_injuries * injury_cfg.get('home_injury_penalty', 0.04) + home_key_missing * injury_cfg.get('home_key_penalty', 0.08)) away_penalty = (away_injuries * injury_cfg.get('away_injury_penalty', 0.04) + away_key_missing * injury_cfg.get('away_key_penalty', 0.08)) min_xg = injury_cfg.get('min_xg_after_injury', 0.25) xg_h = max(min_xg, xg_h - home_penalty) xg_a = max(min_xg, xg_a - away_penalty) if home_penalty > 0 or away_penalty > 0: self.result['_debug']['injury_penalty_h'] = round(home_penalty, 3) self.result['_debug']['injury_penalty_a'] = round(away_penalty, 3) self.result['_debug']['xg_h_before_injury'] = round(xg_h + home_penalty, 3) self.result['_debug']['xg_a_before_injury'] = round(xg_a + away_penalty, 3) return xg_h, xg_a def _apply_handicap_correction(self, score_probs, sorted_scores): handicap_line = self.params.get('handicap_line', 0) odds_home_h = self.params.get('odds_home_handicap', 1.90) odds_away_h = self.params.get('odds_away_handicap', 1.90) if handicap_line == 0 or odds_home_h == 0 or odds_away_h == 0: return score_probs, sorted_scores prob_home_cover = 1 / odds_home_h if odds_home_h > 0 else 0 prob_away_cover = 1 / odds_away_h if odds_away_h > 0 else 0 total = prob_home_cover + prob_away_cover if total > 0: prob_home_cover /= total prob_away_cover /= total if handicap_line < 0: target_goals = abs(handicap_line) if prob_home_cover > 0.55: for (i, j), p in sorted_scores[:15]: if i - j >= int(target_goals) + 1: score_probs[(i, j)] *= 1.15 self.result['handicap_signal'] = 'Home Cover (Big Win Expected)' else: for (i, j), p in sorted_scores[:15]: if 1 <= i - j <= 2: score_probs[(i, j)] *= 1.10 self.result['handicap_signal'] = 'Home Win but Lose Cover (Small Win)' elif handicap_line > 0: if prob_away_cover > 0.55: for (i, j), p in sorted_scores[:15]: if j - i >= int(handicap_line) + 1: score_probs[(i, j)] *= 1.15 self.result['handicap_signal'] = 'Away Cover (Big Win Expected)' else: for (i, j), p in sorted_scores[:15]: if 1 <= j - i <= 2: score_probs[(i, j)] *= 1.10 self.result['handicap_signal'] = 'Away Win but Lose Cover (Small Win)' new_sorted = sorted(score_probs.items(), key=lambda x: -x[1]) return score_probs, new_sorted # ==================== AB 线通用评分（修复版） ==================== def _apply_ab_scoring(self, xg_h, xg_a): """ 通用 AB 线评分调度器（从配置读取权重，不写死任何联赛逻辑） 基于 A 线（基本面）和 B 线（盘口）的交叉验证修正 xG """ ab_cfg = self.league_config.get('ab_scoring', {}) if not ab_cfg.get('active', False): return xg_h, xg_a # 1. 计算 A 线（基本面）评分 a_cfg = ab_cfg.get('a_line', {}) home_power = self.params.get('home_power', 50) away_power = self.params.get('away_power', 50) # 基础战力值归一化 (50->0.5, 100->1.0) a_score_h = 0.5 + (home_power - 50) / 100 a_score_a = 0.5 + (away_power - 50) / 100 # 伤停修正（从配置读取惩罚系数） injuries_penalty = a_cfg.get('injuries_penalty', 0.05) home_inj = self.params.get('home_injuries', 0) * injuries_penalty away_inj = self.params.get('away_injuries', 0) * injuries_penalty a_score_h = max(0.1, a_score_h - home_inj) a_score_a = max(0.1, a_score_a - away_inj) # 战意修正（从配置读取） mot_boost = a_cfg.get('motivation_boost', 1.1) if self.params.get('motivation_home', 0.5) > 0.6: a_score_h = a_score_h * mot_boost if self.params.get('motivation_away', 0.5) > 0.6: a_score_a = a_score_a * mot_boost # 2. 计算 B 线（盘口）评分 b_cfg = ab_cfg.get('b_line', {}) # 理论胜率（基于 ELO 或 power） elo_h = self.params.get('elo_home', 1500) elo_a = self.params.get('elo_away', 1500) if elo_h == 1500 and elo_a == 1500: # 如果没有 ELO，用 power 映射（50→1300, 100→1700） elo_h = 1300 + (home_power - 50) * 8 elo_a = 1300 + (away_power - 50) * 8 theo_prob_h = 1 / (1 + 10 ** ((elo_a - elo_h) / 400)) theo_prob_a = 1 - theo_prob_h # 市场隐含概率（从赔率提取） implied_h = 1 / self.odds_h if self.odds_h > 0 else 0.5 implied_a = 1 / self.odds_a if self.odds_a > 0 else 0.5 # B 线评分 = 1 - 偏差（偏差越小，评分越高） b_score_h = 1 - min(1, abs(implied_h - theo_prob_h) * 2) b_score_a = 1 - min(1, abs(implied_a - theo_prob_a) * 2) # 3. 交叉验证与置信度计算 val_cfg = ab_cfg.get('validation', {}) resonance_threshold = val_cfg.get('resonance_threshold', 0.1) divergence_threshold = val_cfg.get('divergence_threshold', 0.3) resonance_boost = val_cfg.get('resonance_boost', 1.2) divergence_penalty = val_cfg.get('divergence_penalty', 0.7) # 计算 A/B 线对主胜和客胜评分的平均一致性 diff_h = abs(a_score_h - b_score_h) diff_a = abs(a_score_a - b_score_a) avg_diff = (diff_h + diff_a) / 2 if avg_diff <= resonance_threshold: confidence_boost = resonance_boost self.result['ab_validation'] = 'resonance' elif avg_diff >= divergence_threshold: confidence_boost = divergence_penalty self.result['ab_validation'] = 'divergence' else: confidence_boost = 1.0 self.result['ab_validation'] = 'neutral' # 4. 将修正系数应用到 xG（修复版） if confidence_boost != 1.0: if confidence_boost > 1.0: # 共振：双方信心都提升 if implied_h > theo_prob_h: # 市场更看好主队 → 主队xG提升，客队适度提升 xg_h = xg_h * confidence_boost xg_a = xg_a * (1 + (confidence_boost - 1) * 0.5) else: # 市场更看好客队 → 客队xG提升，主队适度提升 xg_a = xg_a * confidence_boost xg_h = xg_h * (1 + (confidence_boost - 1) * 0.5) else: # 背离：双方信心都降低（用同一个惩罚系数，而不是一方被放大） # 原来的 xg / confidence_boost（即 xg / 0.7）会放大 xG，这是 bug # 修正为双方都乘以 confidence_boost（即都降低），只是降低幅度不同 if implied_h > theo_prob_h: # 市场更看好主队，但模型与市场背离 → 主队xG降低，客队降得更多 xg_h = xg_h * confidence_boost # 客队 × (confidence_boost ^ 1.5)，比主队降得更多 xg_a = xg_a * (confidence_boost * confidence_boost) else: # 市场更看好客队，但模型与市场背离 → 客队xG降低，主队降得更多 xg_a = xg_a * confidence_boost # 主队 × (confidence_boost ^ 1.5)，比客队降得更多 xg_h = xg_h * (confidence_boost * confidence_boost) self.result['ab_confidence_boost'] = confidence_boost self.result['ab_scoring_active'] = True return xg_h, xg_a # ==================== 核心 predict 方法 ==================== def predict(self): self.result['_debug'] = {} xg_h = self.params['base_xg_home'] xg_a = self.params['base_xg_away'] self.result['_debug']['xg_h_initial'] = round(xg_h, 3) self.result['_debug']['xg_a_initial'] = round(xg_a, 3) # ----- 从配置读取 xG 调整 ----- xg_adj = self.league_config.get('xg_adjustments', {}) boost = xg_adj.get('default_boost', 1.10) boost_cfg = xg_adj.get('boost_conditions', {}) odds_h = self.odds_h odds_a = self.odds_a league = self.params.get('league', '') if odds_h < 1.40: boost += boost_cfg.get('deep_odds', 0.12) if odds_a > 4.00: boost += boost_cfg.get('big_away_odds', 0.10) if self.params.get('big_ball_rate_home', 0.5) > 0.55 and self.params.get('big_ball_rate_away', 0.5) > 0.55: boost += boost_cfg.get('both_attacking', 0.12) if self.params.get('relegation_zone', False): boost += boost_cfg.get('relegation_zone', 0.10) if self.params.get('first_leg_lead', 0) >= 2: boost += boost_cfg.get('second_leg_conservative', -0.20) if self.params.get('forward_missing', 0) >= 2: boost += boost_cfg.get('forward_missing_penalty', -0.08) xg_h = xg_h * xg_adj.get('home_boost_factor', 1.0) xg_a = xg_a * xg_adj.get('away_penalty_factor', 1.0) if self.params.get('relegation_zone', False) and xg_adj.get('relegation_home_boost', 0): xg_h = xg_h + xg_adj.get('relegation_home_boost', 0) xg_h = max(xg_h, xg_adj.get('relegation_min_xg', 1.0)) boost = max(xg_adj.get('boost_min', 0.85), min(xg_adj.get('boost_max', 1.60), boost)) xg_h = xg_h * boost xg_a = xg_a * boost self.result['_debug']['offensive_boost'] = round(boost, 3) self.result['_debug']['boost_reasons'] = [] if odds_h < 1.40: self.result['_debug']['boost_reasons'].append('deep_odds') if odds_a > 4.00: self.result['_debug']['boost_reasons'].append('big_away_odds') if self.params.get('big_ball_rate_home', 0.5) > 0.55 and self.params.get('big_ball_rate_away', 0.5) > 0.55: self.result['_debug']['boost_reasons'].append('both_attacking') if self.params.get('first_leg_lead', 0) >= 2: self.result['_debug']['boost_reasons'].append('second_leg_conservative') # 西甲深盘额外加成 if league == 'LaLiga' and self.params.get('is_top_club', False) and odds_h < 1.40: la_liga_extra = xg_adj.get('la_liga_deep_odds_extra', 0.08) boost += la_liga_extra self.result['_debug']['la_liga_extra_boost'] = la_liga_extra xg_max = xg_adj.get('xg_max', 5.0) xg_h = min(xg_h, xg_max) xg_a = min(xg_a, xg_max) # 西甲豪门主场保底 la_liga_floor = xg_adj.get('la_liga_home_floor', 2.2) if league == 'LaLiga' and self.params.get('is_top_club', False) and odds_h < 1.40: xg_h = max(xg_h, la_liga_floor) self.result['_debug']['la_liga_home_floor_applied'] = True self.result['_debug']['la_liga_home_floor_value'] = la_liga_floor # 伤病调整 xg_h, xg_a = self._apply_injury_adjustment(xg_h, xg_a) # ===== 🆕 AB 线评分（通用交叉验证） ===== xg_h, xg_a = self._apply_ab_scoring(xg_h, xg_a) self.result['_debug']['xg_h_after_injury'] = round(xg_h, 3) self.result['_debug']['xg_a_after_injury'] = round(xg_a, 3) # 通用主场保底（巴西杯等） if self.league_config.get('home_floor', {}).get('active', False): floor_cfg = self.league_config.get('home_floor', {}) if self.params.get('is_top_club', False) and self.params.get('home_win_rate', 0) > floor_cfg.get('win_rate_threshold', 0.70): xg_h = max(xg_h, floor_cfg.get('min_xg', 1.5)) self.result['_debug']['home_floor_applied'] = True # ===== 模块执行（含沙特新模块） ===== pre_modules = ['wfl', 'rdi', 'uat'] for module_name in pre_modules: if module_name in module_registry: module_func = module_registry[module_name] xg_h, xg_a = module_func(self, xg_h, xg_a) self.context.set('current_xg_h', xg_h) self.context.set('current_xg_a', xg_a) RuleEngine.apply_rules(self.context) for module_name in MODULE_EXECUTION_ORDER: if module_name in module_registry and module_name not in pre_modules: module_cfg = self.modules_cfg.get(module_name, {}) if module_cfg.get('active', True): module_func = module_registry[module_name] #print(f"▶️ 执行模块: {module_name}") xg_h, xg_a = module_func(self, xg_h, xg_a) self.context.set('current_xg_h', xg_h) self.context.set('current_xg_a', xg_a) RuleEngine.apply_rules(self.context) # ===== 🆕 全局 xG 上限保护（防止极端强弱对话爆炸） ===== xg_global_cap = xg_adj.get('xg_global_cap', 5.0) if xg_h > xg_global_cap: self.result['_debug']['xg_h_capped'] = round(xg_h, 3) xg_h = xg_global_cap if xg_a > xg_global_cap: self.result['_debug']['xg_a_capped'] = round(xg_a, 3) xg_a = xg_global_cap self.result['xg_h_after_A'] = xg_h self.result['xg_a_after_A'] = xg_a self.result['_debug']['xg_h_final'] = round(xg_h, 3) self.result['_debug']['xg_a_final'] = round(xg_a, 3) # 半全场划分 tds_config = self.modules_cfg.get('tds', {}) if tds_config.get('active', False): early = tds_config.get('early', [0.40, 0.60]) mid = tds_config.get('mid', [0.45, 0.55]) deep = tds_config.get('deep', [0.35, 0.65]) if self.round <= 3: half_h, half_a = early[0], early[1] elif self.odds_h < 1.30: half_h, half_a = deep[0], deep[1] else: half_h, half_a = mid[0], mid[1] else: half_h, half_a = 0.45, 0.55 h_h1, h_h2 = xg_h * half_h, xg_h * half_a a_h1, a_h2 = xg_a * half_h, xg_a * half_a if self.params.get('var_controversy', False): a_h2 *= 1.05 self.result['xg_h_half1'] = h_h1 self.result['xg_h_half2'] = h_h2 self.result['xg_a_half1'] = a_h1 self.result['xg_a_half2'] = a_h2 # 概率计算 probs = self._calc_match_probabilities(xg_h, xg_a) self.result['core_prob'] = probs['best_prob'] score_corrections = self.context.get('score_corrections', {}) for score_key, weight in score_corrections.items(): if score_key in probs['score_probs']: probs['score_probs'][score_key] *= weight probs['sorted_scores'] = sorted(probs['score_probs'].items(), key=lambda x: -x[1]) win_h, draw, win_a = probs['win_h'], probs['draw'], probs['win_a'] if self.result.get('dsr_iron_draw', False): draw *= 1.20 if self.context.get('q_attack_boost', 1.0) != 1.0: win_h *= self.context.get('q_attack_boost', 1.0) if self.context.get('fpa_boost', 1.0) != 1.0: win_h *= self.context.get('fpa_boost', 1.0) total = win_h + draw + win_a if total > 0: win_h /= total draw /= total win_a /= total draw_cap = self.modules_cfg.get('dsr', {}).get('draw_cap', 0.35) if draw > draw_cap: draw = draw_cap total = win_h + win_a if total > 0: win_h = win_h / total * (1 - draw_cap) win_a = win_a / total * (1 - draw_cap) self.result['prob_win_h'] = win_h self.result['prob_draw'] = draw self.result['prob_win_a'] = win_a if self.params.get('handicap_line', 0) != 0: probs['score_probs'], probs['sorted_scores'] = self._apply_handicap_correction( probs['score_probs'], probs['sorted_scores'] ) matrix = self._generate_score_matrix(probs['sorted_scores']) self.result['score_core'] = matrix['core'] self.result['score_defend'] = matrix['defend'] self.result['score_attack'] = matrix['attack'] self.result['open_attack'] = matrix['open_attack'] self._generate_strategy() self._generate_tags() return self.result