rendered paste bodyimport jsonimport osimport mathimport requestsimport timeimport shutilfrom datetime import datetimefrom collections import defaultdict# ========== 配置 ==========RAW_MATCHES_FILE = "1_data_fetcher/output/raw_matches.json"# AI 降级补全开关（排名缺失时是否启用 AI 补全）AI_COMPLETE_MISSING_RANK = True # True=启用AI补全，False=直接跳过PREVIEW_FILE = "1_data_fetcher/output/preview_data.json"OUTPUT_DIR = "5_predictor/output"HISTORY_DIR = os.path.join(OUTPUT_DIR, "history")DAILY_DIR = os.path.join(OUTPUT_DIR, "daily")SUMMARY_DIR = os.path.join(OUTPUT_DIR, "summary")CONFIG_DIR = "5_predictor/config"LEAGUE_PARAMS_FILE = os.path.join(CONFIG_DIR, "league_params.json")for d in [OUTPUT_DIR, HISTORY_DIR, DAILY_DIR, SUMMARY_DIR, CONFIG_DIR]: os.makedirs(d, exist_ok=True)# DeepSeek API 配置DEEPSEEK_API_KEY = "sk-&"DEEPSEEK_URL = "https://api.deepseek.com/v1/chat/completions"# ========== 常量 ==========SCORE_LABELS = [ "1:0", "2:0", "2:1", "3:0", "3:1", "3:2", "4:0", "4:1", "4:2", "5:0", "5:1", "5:2", "胜其它", "0:0", "1:1", "2:2", "3:3", "平其它", "0:1", "0:2", "1:2", "0:3", "1:3", "2:3", "0:4", "1:4", "2:4", "0:5", "1:5", "2:5", "负其它"]TOTAL_GOALS_LABELS = ["0球", "1球", "2球", "3球", "4球", "5球", "6球", "7+球"]HALF_FULL_LABELS = ["胜胜", "胜平", "胜负", "平胜", "平平", "平负", "负胜", "负平", "负负"]# ========== 默认联赛参数 ==========DEFAULT_LEAGUE_PARAMS = { "avg_goals": 2.6, "draw_base": 0.24, "draw_range": 0.10, "home_advantage": 1.15, "handicap_factor": 7.5, "half_home": 0.43, "half_draw": 0.33, "half_away": 0.24}# ========== 加载联赛参数 ==========def load_league_params(): if os.path.exists(LEAGUE_PARAMS_FILE): with open(LEAGUE_PARAMS_FILE, 'r', encoding='utf-8') as f: return json.load(f) return {}def save_league_params(params): with open(LEAGUE_PARAMS_FILE, 'w', encoding='utf-8') as f: json.dump(params, f, ensure_ascii=False, indent=2)LEAGUE_PARAMS = load_league_params()# ========== AI 生成联赛参数 ==========def ai_generate_league_params(league_name): if not DEEPSEEK_API_KEY or DEEPSEEK_API_KEY == "sk-你的API密钥": return DEFAULT_LEAGUE_PARAMS.copy() prompt = f"""你是一位足球数据分析专家，请为以下联赛生成推演参数。联赛名称：{league_name}请根据你对这个联赛的了解，提供以下参数（用JSON格式）：- avg_goals: 场均进球数（2.0-3.5）- draw_base: 平局概率基准值（0.15-0.35）- draw_range: 平局概率波动范围（0.05-0.15）- home_advantage: 主场优势系数（1.0-1.4）- handicap_factor: 让球调整系数（6.0-9.0）- half_home: 半场主胜概率（0.35-0.50）- half_draw: 半场平局概率（0.25-0.40）- half_away: 半场客胜概率（0.20-0.35）只输出JSON，不要有其他内容。""" try: response = requests.post(DEEPSEEK_URL, headers={"Authorization": f"Bearer {DEEPSEEK_API_KEY}", "Content-Type": "application/json"}, json={"model": "deepseek-chat", "messages": [{"role": "system", "content": "你是一个足球数据分析专家，只输出JSON格式的参数。"}, {"role": "user", "content": prompt}], "temperature": 0.3, "response_format": {"type": "json_object"}}, timeout=30) result = response.json() params = json.loads(result['choices'][0]['message']['content']) for key in DEFAULT_LEAGUE_PARAMS: if key not in params: params[key] = DEFAULT_LEAGUE_PARAMS[key] return params except: return DEFAULT_LEAGUE_PARAMS.copy()def get_league_params(league): if not league: return DEFAULT_LEAGUE_PARAMS.copy() league_key = league.strip() if league_key in LEAGUE_PARAMS: return LEAGUE_PARAMS[league_key] for key in LEAGUE_PARAMS: if key in league_key or league_key in key: return LEAGUE_PARAMS[key] params = ai_generate_league_params(league_key) LEAGUE_PARAMS[league_key] = params save_league_params(LEAGUE_PARAMS) return params# ========== 实力分计算 ==========def calculate_strength(team_data, is_home, league_params): rank = team_data.get('rank', 15) wins = team_data.get('season_wins', 0) draws = team_data.get('season_draws', 0) losses = team_data.get('season_losses', 0) total = wins + draws + losses win_rate = wins / total if total > 0 else 0 goals_for = team_data.get('goals_for', 0) goals_against = team_data.get('goals_against', 0) rank_score = max(0, 100 - (rank - 1) * 2.5) win_rate_score = win_rate * 30 home_adv = league_params.get('home_advantage', 1.10) if is_home: home_wins = team_data.get('home_wins', 0) home_draws = team_data.get('home_draws', 0) home_losses = team_data.get('home_losses', 0) home_total = home_wins + home_draws + home_losses home_rate = home_wins / home_total if home_total > 0 else 0 home_bonus = home_rate * 12 * home_adv else: away_wins = team_data.get('away_wins', 0) away_draws = team_data.get('away_draws', 0) away_losses = team_data.get('away_losses', 0) away_total = away_wins + away_draws + away_losses away_rate = away_wins / away_total if away_total > 0 else 0 home_bonus = away_rate * 10 / home_adv goal_diff = goals_for - goals_against goal_score = max(0, min(goal_diff / 2, 1)) * 15 return round(rank_score + win_rate_score + home_bonus + goal_score, 1)def expected_goals(home_strength, away_strength, league_params): avg_goals = league_params.get('avg_goals', 2.6) diff = (home_strength - away_strength) / 100 home_xg = avg_goals / 2 * (1 + diff * 0.6) away_xg = avg_goals / 2 * (1 - diff * 0.6) return max(0.4, home_xg), max(0.4, away_xg)def poisson_prob(xg, k): if k < 0: return 0 return (math.exp(-xg) * (xg ** k)) / math.factorial(k)# ========== 各玩法分析函数 ==========def analyze_spf(home_strength, away_strength, spf_odds, league_params): """分析SPF（胜平负）赔率偏差，增加反向推荐""" if not spf_odds or len(spf_odds) < 3: return None try: home_odds = float(spf_odds[0]) draw_odds = float(spf_odds[1]) away_odds = float(spf_odds[2]) except: return None # 计算隐含概率 total = 1/home_odds + 1/draw_odds + 1/away_odds market_home = (1/home_odds) / total market_draw = (1/draw_odds) / total market_away = (1/away_odds) / total # 实力估算概率（加入联赛平局系数） strength_total = home_strength + away_strength if strength_total == 0: return None raw_home = home_strength / strength_total raw_away = away_strength / strength_total # 平局概率 = 1 - 主客概率之和，但加入联赛基准平局系数 draw_base = league_params.get('draw_base', 0.28) raw_draw = max(0, draw_base - abs(raw_home - raw_away) * 0.3) # 重新归一化 total_raw = raw_home + raw_draw + raw_away if total_raw == 0: return None model_home = raw_home / total_raw model_draw = raw_draw / total_raw model_away = raw_away / total_raw # 偏差 = 市场 - 模型 bias_home = market_home - model_home bias_draw = market_draw - model_draw bias_away = market_away - model_away result = { 'bias': {'home': bias_home, 'draw': bias_draw, 'away': bias_away}, 'market': {'home': market_home, 'draw': market_draw, 'away': market_away}, 'model': {'home': model_home, 'draw': model_draw, 'away': model_away}, } # ---------- 新增反向推荐逻辑 ---------- # 检查每个选项是否存在显著负偏差（<-8%），若有则反向推荐对立选项 reverse_recommend = None for key, bias in result['bias'].items(): if bias < -0.08: # 负偏差超过阈值 # 确定反向选项（取偏差最大的对立选项） if key == 'home': # 比较客胜和平局的偏差，选较大的 if result['bias']['away'] > result['bias']['draw']: opposite = 'away' else: opposite = 'draw' elif key == 'away': if result['bias']['home'] > result['bias']['draw']: opposite = 'home' else: opposite = 'draw' else: # draw if result['bias']['home'] > result['bias']['away']: opposite = 'home' else: opposite = 'away' reverse_recommend = { 'original': key, 'original_bias': bias, 'recommend': opposite, 'recommend_bias': -bias # 反向偏差 = 正值 } break # 只取第一个明显负偏差 if reverse_recommend: result['reverse'] = reverse_recommend return resultdef analyze_rqspf(home_strength, away_strength, rqspf_odds, handicap, league_params): """分析让球胜平负赔率偏差，增加反向推荐""" if not rqspf_odds or len(rqspf_odds) < 3: return None try: home_odds = float(rqspf_odds[0]) draw_odds = float(rqspf_odds[1]) away_odds = float(rqspf_odds[2]) except: return None # 计算隐含概率 total = 1/home_odds + 1/draw_odds + 1/away_odds market_home = (1/home_odds) / total market_draw = (1/draw_odds) / total market_away = (1/away_odds) / total # ---- 根据让球数调整实力分 ---- adjusted_home = home_strength adjusted_away = away_strength # handicap 格式如 "-1"、"+1"、"-2" 等 if handicap: try: handicap_val = int(str(handicap).replace('+', '')) # 让球数越大，主队实力折算越高（因为要赢更多球） # 每让1球，相当于主队实力增加 15 分 adjusted_home = home_strength + (handicap_val * 15) adjusted_away = away_strength - (handicap_val * 15) except: pass # 如果转换失败，使用原始实力分 # 确保实力分不为负数 adjusted_home = max(adjusted_home, 10) adjusted_away = max(adjusted_away, 10) strength_total = adjusted_home + adjusted_away if strength_total == 0: return None raw_home = adjusted_home / strength_total raw_away = adjusted_away / strength_total draw_base = league_params.get('draw_base', 0.28) raw_draw = max(0, draw_base - abs(raw_home - raw_away) * 0.3) total_raw = raw_home + raw_draw + raw_away if total_raw == 0: return None model_home = raw_home / total_raw model_draw = raw_draw / total_raw model_away = raw_away / total_raw bias_home = market_home - model_home bias_draw = market_draw - model_draw bias_away = market_away - model_away result = { 'bias': {'home': bias_home, 'draw': bias_draw, 'away': bias_away}, 'market': {'home': market_home, 'draw': market_draw, 'away': market_away}, 'model': {'home': model_home, 'draw': model_draw, 'away': model_away}, } # ---------- 新增反向推荐逻辑（同SPF） ---------- reverse_recommend = None for key, bias in result['bias'].items(): if bias < -0.08: if key == 'home': if result['bias']['away'] > result['bias']['draw']: opposite = 'away' else: opposite = 'draw' elif key == 'away': if result['bias']['home'] > result['bias']['draw']: opposite = 'home' else: opposite = 'draw' else: if result['bias']['home'] > result['bias']['away']: opposite = 'home' else: opposite = 'away' reverse_recommend = { 'original': key, 'original_bias': bias, 'recommend': opposite, 'recommend_bias': -bias } break if reverse_recommend: result['reverse'] = reverse_recommend return resultdef analyze_total_goals(home_xg, away_xg, total_odds): if len(total_odds) < 8: return None total_xg = home_xg + away_xg model_probs = [] for k in range(8): if k < 7: prob = poisson_prob(total_xg, k) else: prob = 1 - sum(poisson_prob(total_xg, i) for i in range(7)) model_probs.append(prob) imp_probs = [1/float(odds) for odds in total_odds] total_imp = sum(imp_probs) imp_probs = [p/total_imp for p in imp_probs] biases = [round(model_probs[i] - imp_probs[i], 3) for i in range(8)] max_bias = max(biases) max_idx = biases.index(max_bias) return {'model': model_probs, 'implied': imp_probs, 'bias': biases, 'max_bias': max_bias, 'max_index': max_idx}def analyze_score(home_xg, away_xg, score_odds): if len(score_odds) < 31: return None max_goals = 9 score_probs = {} for h in range(max_goals + 1): for a in range(max_goals + 1): prob = poisson_prob(home_xg, h) * poisson_prob(away_xg, a) score_probs[f"{h}:{a}"] = prob model_probs = [] for label in SCORE_LABELS: if ':' in label and '胜' not in label and '平' not in label and '负' not in label: h, a = map(int, label.split(':')) prob = score_probs.get(f"{h}:{a}", 0) elif label == '胜其它': prob = sum(score_probs.get(f"{h}:{a}", 0) for h in range(max_goals + 1) for a in range(max_goals + 1) if h > a and f"{h}:{a}" not in SCORE_LABELS[:13]) elif label == '平其它': prob = sum(score_probs.get(f"{h}:{a}", 0) for h in range(max_goals + 1) for a in range(max_goals + 1) if h == a and h > 3 and f"{h}:{a}" not in SCORE_LABELS[13:18]) elif label == '负其它': prob = sum(score_probs.get(f"{h}:{a}", 0) for h in range(max_goals + 1) for a in range(max_goals + 1) if h < a and f"{h}:{a}" not in SCORE_LABELS[18:]) else: prob = 0 model_probs.append(prob) total_model = sum(model_probs) if total_model > 0: model_probs = [p/total_model for p in model_probs] imp_probs = [] for odds in score_odds[:31]: try: imp_probs.append(1 / float(odds)) except: imp_probs.append(0) total_imp = sum(imp_probs) if total_imp > 0: imp_probs = [p/total_imp for p in imp_probs] biases = [] for i in range(min(len(model_probs), len(imp_probs))): biases.append(round(model_probs[i] - imp_probs[i], 3)) max_bias = max(biases) if biases else 0 max_idx = biases.index(max_bias) if biases else -1 # 在 return 之前添加 print(f"===== 比分调试 =====") print(f"home_xg: {home_xg}, away_xg: {away_xg}") print(f"score_odds 长度: {len(score_odds)}") if len(score_odds) > 0: print(f"score_odds 前5: {score_odds[:5]}") print(f"model_probs 前5: {model_probs[:5] if model_probs else '空'}") print(f"imp_probs 前5: {imp_probs[:5] if imp_probs else '空'}") print(f"biases 前5: {biases[:5] if biases else '空'}") print(f"max_bias: {max_bias}, recommend: {SCORE_LABELS[max_idx] if max_idx >= 0 else 'N/A'}") print(f"==================") return { 'max_bias': max_bias, 'max_index': max_idx, 'recommend': SCORE_LABELS[max_idx] if max_idx >= 0 else None, 'bias': biases }def analyze_half_full(home_strength, away_strength, half_full_odds, league_params): if len(half_full_odds) < 9: return None diff = home_strength - away_strength half_home = league_params.get('half_home', 0.43) + (diff / 100) * 0.08 half_draw = league_params.get('half_draw', 0.33) - abs(diff) / 100 * 0.05 half_away = 1 - half_home - half_draw full_home = 1 / (1 + 10 ** (-diff / 100 * 1.2)) full_away = 1 - full_home full_draw = 0.15 + 0.1 * (1 - abs(diff) / 100) full_draw = max(0.10, min(0.35, full_draw)) full_home *= (1 - full_draw) full_away *= (1 - full_draw) model_probs = [ half_home * full_home * 0.8, half_home * full_draw * 0.3, half_home * full_away * 0.2, half_draw * full_home * 0.5, half_draw * full_draw * 0.8, half_draw * full_away * 0.5, half_away * full_home * 0.2, half_away * full_draw * 0.3, half_away * full_away * 0.8 ] total_model = sum(model_probs) if total_model > 0: model_probs = [p/total_model for p in model_probs] imp_probs = [1/float(odds) for odds in half_full_odds] total_imp = sum(imp_probs) if total_imp > 0: imp_probs = [p/total_imp for p in imp_probs] biases = [round(imp_probs[i] - model_probs[i], 3) for i in range(9)] max_bias = max(biases) max_idx = biases.index(max_bias) return { 'max_bias': max_bias, 'max_index': max_idx, 'recommend': HALF_FULL_LABELS[max_idx], 'bias': biases }def generate_detailed_analysis(match, signals, comparison=None): """生成详细推演分析矩阵 返回：字符串，包含推演路径、剧本排序、投注建议 """ if not signals: return "无有效信号，建议观望" # ---- 分离正负偏差 ---- positive = [s for s in signals if s['bias'] > 0.08] negative = [s for s in signals if s['bias'] < -0.03] # ---- 新增：将负偏差转化为反向推荐 ---- reverse_signals = [] for s in negative: if s['option'] == 'home': reverse_option = 'away' elif s['option'] == 'away': reverse_option = 'home' elif s['option'] == 'draw': reverse_option = 'home' # 平局负偏差时，默认推荐主胜（可优化） else: continue reverse_signals.append({ 'play': s['play'], 'option': reverse_option, 'bias': -s['bias'], 'signal_type': '反向挖掘' }) # ---- 合并正偏差 + 反向推荐 ---- all_signals = positive + reverse_signals if not all_signals: neg_list = ", ".join([f"{s['play']} {s['option']}（{s['bias']:+.0%}）" for s in negative]) return f"⚠️ 仅负偏差信号：{neg_list}。建议回避，考虑反向选项或放弃。" # 按偏差绝对值排序 all_signals.sort(key=lambda x: abs(x['bias']), reverse=True) # ---- 构建推演路径 ---- lines = [] lines.append("**推演分析矩阵**") lines.append("") # 1. 信号汇总 lines.append("**正偏差（被低估）**：") for s in all_signals: tag = " [反向挖掘]" if s.get('signal_type') == '反向挖掘' else "" lines.append(f" + {s['play']} {s['option']}（偏差 {s['bias']:+.0%}）{tag}") if negative: lines.append("**负偏差（被高估，回避）**：") for s in negative: lines.append(f" - {s['play']} {s['option']}（偏差 {s['bias']:+.0%}）") lines.append("") # 2. 推演剧本生成 core = all_signals[0] core_play = core['play'] core_option = core['option'] core_bias = core['bias'] # 确定方向 if "主" in core_option or "home" in core_option: direction = "主队" elif "客" in core_option or "away" in core_option: direction = "客队" else: direction = "平局" aux_signals = all_signals[1:] scripts = [] # 剧本1：核心方向 + 辅助信号组合 script1 = f"**核心剧本：{direction}胜**" if core_play == "半全场": script1 += f"（{core_option}）" elif core_play in ["SPF", "RQSPF"]: script1 += f"（{core_option}）" if aux_signals: aux_desc = [] for a in aux_signals[:3]: if a['play'] in ["比分", "总进球", "半全场"]: aux_desc.append(f"{a['play']} {a['option']}（+{abs(a['bias']):.0%}）") if aux_desc: script1 += "，配合 " + "、".join(aux_desc) # 赔率变化分析（仅当 comparison 不为 None 且有数据时） if comparison and isinstance(comparison, dict) and comparison.get('changes'): changes = comparison['changes'] for key, change in changes.items(): if isinstance(change, dict) and 'direction' in change: if (direction == "主队" and "spf_home" in key and change['direction'] == "降低") or \ (direction == "客队" and "spf_away" in key and change['direction'] == "降低"): script1 += "；赔率变化确认该方向" elif (direction == "主队" and "spf_home" in key and change['direction'] == "升高") or \ (direction == "客队" and "spf_away" in key and change['direction'] == "升高"): script1 += "；但赔率变化与该方向相悖，需谨慎" script1 += "。" scripts.append(script1) # 剧本2：备选方向（平局或主/客不败） draw_signals = [s for s in all_signals if "draw" in s['option'] or "平" in s['option']] if draw_signals and draw_signals[0]['bias'] > 0.03: script2 = f"**备选剧本：平局**（{draw_signals[0]['play']} 偏差 {draw_signals[0]['bias']:+.0%}）" scripts.append(script2) elif negative: neg_option = negative[0]['option'] if "home" in neg_option or "主胜" in neg_option: script2 = "**备选剧本：客队不败**（主胜被高估，客队或平局有机会）" scripts.append(script2) elif "away" in neg_option or "客胜" in neg_option: script2 = "**备选剧本：主队不败**（客胜被高估，主队或平局有机会）" scripts.append(script2) # 剧本3：比分关注 score_signal = next((s for s in all_signals if s['play'] == "比分"), None) if score_signal and score_signal['bias'] > 0.05: script3 = f"**高赔比分关注**：{score_signal['option']}（偏差 {score_signal['bias']:+.0%}）" scripts.append(script3) lines.append("**推演剧本（按可能性排序）**：") for i, script in enumerate(scripts, 1): lines.append(f" {i}. {script}") lines.append("") # 3. 最终投注建议（差异化信心阈值） play = core['play'] if play in ['SPF', 'RQSPF']: threshold_high = 0.12 elif play in ['总进球', '半全场']: threshold_high = 0.15 else: # 比分等 threshold_high = 0.20 if core_bias >= threshold_high: confidence = "高" emoji = "✅" else: confidence = "低" emoji = "⚠️" # 胜负平分类 if "主" in core_option or "home" in core_option: bet_type = "主胜" elif "客" in core_option or "away" in core_option: bet_type = "客胜" else: bet_type = "平局" lines.append(f"**📊 投注建议**：") tag = " [反向挖掘]" if core.get('signal_type') == '反向挖掘' else "" lines.append(f" 核心推荐：{emoji} {core_play} {bet_type}（偏差 {core_bias:+.0%}，信心 {confidence}）{tag}") if len(all_signals) > 1: aux_parts = [] for s in all_signals[1:3]: aux_tag = " [反向]" if s.get('signal_type') == '反向挖掘' else "" aux_parts.append(f"{s['play']} {s['option']}{aux_tag}") if aux_parts: lines.append(f" 辅助关注：{'、'.join(aux_parts)}") if negative: neg_plays = [f"{s['play']} {s['option']}" for s in negative[:2]] lines.append(f" ⚠️ 回避：{'、'.join(neg_plays)}（被高估）") return "\n".join(lines)# ========== AI 软信息分析 ==========def ai_analyze_match(match, home_strength, away_strength, home_rank, away_rank, bias_signals): if not DEEPSEEK_API_KEY or DEEPSEEK_API_KEY == "sk-你的API密钥": return {"修正建议": "坚持信号", "信心等级": 3, "说明": "未配置API Key，使用硬数据"} bias_summary = "\n".join([f" {s['play']}: {s['option']} 偏差{s['bias']:+.0%}" for s in bias_signals]) if bias_signals else "无明显偏差" prompt = f"""你是一位专业的足球比赛分析师，请分析以下比赛。主队：{match['home_team']}（排名第{home_rank}，实力分{home_strength:.1f}）客队：{match['away_team']}（排名第{away_rank}，实力分{away_strength:.1f}）联赛：{match['league']}硬数据信号：{bias_summary}请分析战意、伤病、人气陷阱、状态异常，输出JSON：{{"战意": "...", "伤病影响": "...", "人气陷阱": "...", "状态异常": "...", "修正建议": "坚持信号/降低信心/放弃", "信心等级": 1-5, "说明": "..."}}""" try: response = requests.post(DEEPSEEK_URL, headers={"Authorization": f"Bearer {DEEPSEEK_API_KEY}", "Content-Type": "application/json"}, json={"model": "deepseek-chat", "messages": [{"role": "user", "content": prompt}], "temperature": 0.3, "response_format": {"type": "json_object"}}, timeout=30) result = response.json() return json.loads(result['choices'][0]['message']['content']) except: return {"修正建议": "坚持信号", "信心等级": 3, "说明": "AI分析失败"} def ai_complete_missing_data(match, home_team, away_team, league): """ 当排名缺失时，调用AI补全球队实力信息 返回：{'home_rank': int, 'away_rank': int, 'home_strength': float, 'away_strength': float, 'confidence': str} """ if not DEEPSEEK_API_KEY or DEEPSEEK_API_KEY == "sk-你的API密钥": return None prompt = f"""请根据你对足球的了解，补全以下比赛的数据：联赛：{league}主队：{home_team}客队：{away_team}请估算：1. 两队在当前联赛中的大概排名（1-20之间的整数）2. 两队的综合实力分（60-130之间的浮点数，主队和客队都要）请以JSON格式输出，不要有其他内容：{{"home_rank": 排名, "away_rank": 排名, "home_strength": 实力分, "away_strength": 实力分, "confidence": "高/中/低", "reason": "简要理由"}}""" try: response = requests.post( DEEPSEEK_URL, headers={"Authorization": f"Bearer {DEEPSEEK_API_KEY}", "Content-Type": "application/json"}, json={ "model": "deepseek-chat", "messages": [{"role": "user", "content": prompt}], "temperature": 0.5, "response_format": {"type": "json_object"} }, timeout=30 ) result = response.json() data = json.loads(result['choices'][0]['message']['content']) return { 'home_rank': int(data.get('home_rank', 15)), 'away_rank': int(data.get('away_rank', 15)), 'home_strength': float(data.get('home_strength', 70)), 'away_strength': float(data.get('away_strength', 70)), 'confidence': data.get('confidence', '低'), 'reason': data.get('reason', 'AI估算') } except Exception as e: print(f" ⚠️ AI补全失败: {e}") return None # ========== 🆕 历史记录加载与对比分析 ==========def load_history_for_match(match_id): """加载同一场比赛的历史推演记录""" history_file = os.path.join(HISTORY_DIR, f"{match_id}.json") if not os.path.exists(history_file): return None try: with open(history_file, 'r', encoding='utf-8') as f: return json.load(f) except: return None def signals_are_identical(current_signals, prev_signals): """ 比较当前信号和历史信号的偏差是否完全相同 返回：True 表示完全相同，False 表示有变化 """ if not prev_signals: return False def key(s): return (s.get('play', ''), s.get('option', ''), round(s.get('bias', 0), 3)) current_keys = sorted([key(s) for s in current_signals]) prev_keys = sorted([key(s) for s in prev_signals]) return current_keys == prev_keysdef save_history_for_match(match_id, record): """保存推演记录到历史文件（追加）""" history_file = os.path.join(HISTORY_DIR, f"{match_id}.json") existing = load_history_for_match(match_id) if existing: # 追加新记录（保留最近50条） if isinstance(existing, list): existing.append(record) if len(existing) > 50: existing = existing[-50:] else: existing = [existing, record] else: existing = [record] with open(history_file, 'w', encoding='utf-8') as f: json.dump(existing, f, ensure_ascii=False, indent=2)def compare_odds_changes(current, previous): """ 对比当前与历史赔率的变化 返回：变化分析字典 """ if not previous: return None changes = {} timestamp = datetime.now().isoformat() # 对比 SPF prev_spf = previous.get('odds', {}).get('spf', []) curr_spf = current.get('odds', {}).get('spf', []) if prev_spf and curr_spf and len(prev_spf) >= 3 and len(curr_spf) >= 3: for idx, label in enumerate(['home', 'draw', 'away']): try: prev_val = float(prev_spf[idx]) curr_val = float(curr_spf[idx]) diff = curr_val - prev_val if abs(diff) > 0.01: direction = "升高" if diff > 0 else "降低" changes[f"spf_{label}"] = { "from": prev_val, "to": curr_val, "diff": round(diff, 2), "direction": direction, "interpretation": "市场看衰" if diff > 0 else "市场看好" } except: pass # 对比 RQSPF prev_rq = previous.get('odds', {}).get('rqspf', []) curr_rq = current.get('odds', {}).get('rqspf', []) if prev_rq and curr_rq and len(prev_rq) >= 3 and len(curr_rq) >= 3: for idx, label in enumerate(['home', 'draw', 'away']): try: prev_val = float(prev_rq[idx]) curr_val = float(curr_rq[idx]) diff = curr_val - prev_val if abs(diff) > 0.01: direction = "升高" if diff > 0 else "降低" changes[f"rqspf_{label}"] = { "from": prev_val, "to": curr_val, "diff": round(diff, 2), "direction": direction, "interpretation": "市场看衰" if diff > 0 else "市场看好" } except: pass # 对比让球数 prev_handicap = previous.get('handicap', '0') curr_handicap = current.get('handicap', '0') if prev_handicap != curr_handicap: changes['handicap'] = { "from": prev_handicap, "to": curr_handicap, "diff": f"{prev_handicap} → {curr_handicap}", "interpretation": f"让球数变化，可能反映实力预期调整" } return changes if changes else Nonedef analyze_market_impact(changes, bias_signals): """ 分析赔率变化对偏差信号的影响 返回：综合判断 """ if not changes: return None analysis = { "summary": [], "confidence_adjustment": 0, # -1 到 +1 "recommendation": "坚持原信号" } for key, change in changes.items(): # 解析变化方向与偏差方向是否一致 if key.startswith('spf_home') or key.startswith('rqspf_home'): # 主胜赔率变化 if change['direction'] == '降低': # 赔率降低 → 市场更看好主队 # 检查偏差信号中是否有主胜 for signal in bias_signals: if signal.get('option') == 'home' or signal.get('option') == '主胜': if signal.get('bias', 0) > 0: analysis['summary'].append(f"主胜赔率{change['direction']}，与偏差信号(+{signal['bias']:.0%})方向一致，信号增强") analysis['confidence_adjustment'] += 0.2 else: analysis['summary'].append(f"主胜赔率{change['direction']}，但偏差为负，存在矛盾，需谨慎") analysis['confidence_adjustment'] -= 0.2 elif change['direction'] == '升高': for signal in bias_signals: if signal.get('option') == 'home' or signal.get('option') == '主胜': if signal.get('bias', 0) > 0: analysis['summary'].append(f"主胜赔率{change['direction']}，与偏差信号(+{signal['bias']:.0%})方向相反，可能信号被市场消化") analysis['confidence_adjustment'] -= 0.1 else: analysis['summary'].append(f"主胜赔率{change['direction']}，偏差为负，双重看衰") analysis['confidence_adjustment'] -= 0.1 elif key.startswith('spf_away') or key.startswith('rqspf_away'): if change['direction'] == '降低': for signal in bias_signals: if signal.get('option') == 'away' or signal.get('option') == '客胜': if signal.get('bias', 0) > 0: analysis['summary'].append(f"客胜赔率{change['direction']}，与偏差信号(+{signal['bias']:.0%})方向一致，信号增强") analysis['confidence_adjustment'] += 0.2 # 综合判断 if analysis['confidence_adjustment'] >= 0.3: analysis['recommendation'] = "信号增强，可加大关注" elif analysis['confidence_adjustment'] <= -0.3: analysis['recommendation'] = "信号减弱，建议降低信心或放弃" else: analysis['recommendation'] = "无明显变化，维持原判断" return analysis# ========== 主函数 ==========def main(): print("=" * 60) print(f"足球预测推演系统 (动态追踪版) - {datetime.now().strftime('%Y-%m-%d %H:%M')}") print("=" * 60) print() if not os.path.exists(RAW_MATCHES_FILE): print("❌ raw_matches.json 不存在") return if not os.path.exists(PREVIEW_FILE): print("❌ preview_data.json 不存在") return with open(RAW_MATCHES_FILE, 'r', encoding='utf-8') as f: matches = json.load(f) with open(PREVIEW_FILE, 'r', encoding='utf-8') as f: previews = json.load(f) match_results = [] ai_usage_count = 0 new_leagues = [] comparison_count = 0 for match in matches: mid = match.get('match_id') league = match.get('league', '默认') league_params = get_league_params(league) if league not in LEAGUE_PARAMS and league not in new_leagues: new_leagues.append(league) preview = previews.get(mid, {}).get('preview', {}) home_preview = preview.get('home', {}) away_preview = preview.get('away', {}) home_rank = home_preview.get('rank') away_rank = away_preview.get('rank') home_strength = None away_strength = None ai_completed = False # ---- 数据完整性检查：如果排名缺失，尝试AI补全 ---- if home_rank is None or away_rank is None: if AI_COMPLETE_MISSING_RANK: print(f" 🔄 {match.get('match_num', '')} {match.get('home_team', '')} vs {match.get('away_team', '')} 排名缺失，尝试AI补全...") ai_data = ai_complete_missing_data( match, match.get('home_team', ''), match.get('away_team', ''), league ) if ai_data: home_rank = ai_data['home_rank'] away_rank = ai_data['away_rank'] home_strength = ai_data['home_strength'] away_strength = ai_data['away_strength'] ai_completed = True print(f" ✅ AI补全完成: 主{home_rank}位({home_strength:.1f})，客{away_rank}位({away_strength:.1f})，置信度:{ai_data.get('confidence', '未知')}") else: print(f" ⏭️ AI补全失败，跳过该场") continue else: print(f" ⏭️ {match.get('match_num', '')} {match.get('home_team', '')} vs {match.get('away_team', '')} 排名缺失（AI补全已关闭），跳过") continue else: # 排名存在，正常计算 home_rank = int(home_rank) away_rank = int(away_rank) # 正常计算实力分 home_strength = calculate_strength(home_preview, True, league_params) away_strength = calculate_strength(away_preview, False, league_params) ai_completed = False # 确保实力分有值 if home_strength is None or away_strength is None: print(f" ⚠️ 实力分缺失，跳过") continue # ---- 计算预期进球（新增这一行） ---- home_xg, away_xg = expected_goals(home_strength, away_strength, league_params) result = { 'match_num': match['match_num'], 'ai_completed': ai_completed, 'home_team': match['home_team'], 'away_team': match['away_team'], 'league': league, 'strength': {'home': home_strength, 'away': away_strength}, 'rank': {'home': home_rank, 'away': away_rank}, 'expected_goals': {'home': round(home_xg, 2), 'away': round(away_xg, 2)}, 'odds': { 'spf': match.get('spf', []), 'rqspf': match.get('rqspf', []), 'total_goals': match.get('total_goals', []), 'score': match.get('score', []), 'half_full': match.get('half_full', []) }, 'handicap': match.get('handicap', '0'), 'recommendations': [], 'trial_recommendations': [], 'ai_analysis': None, 'comparison': None } # ---- 各玩法分析 ---- bias_signals = [] trial_signals = [] skip_ai = False spf_result = analyze_spf(home_strength, away_strength, match.get('spf', []), league_params) if spf_result: if 'reverse' in spf_result: rev = spf_result['reverse'] bias_signals.append({ 'play': 'SPF', 'option': rev['recommend'], 'bias': rev['recommend_bias'], 'signal_type': '反向挖掘', 'match_id': mid }) else: max_key = max(spf_result['bias'], key=lambda k: abs(spf_result['bias'][k])) if abs(spf_result['bias'][max_key]) > 0.08 and spf_result['bias'][max_key] > 0: bias_signals.append({ 'play': 'SPF', 'option': max_key, 'bias': spf_result['bias'][max_key], 'signal_type': '正向偏差', 'match_id': mid }) rqspf_result = analyze_rqspf(home_strength, away_strength, match.get('rqspf', []), match.get('handicap', '0'), league_params) if rqspf_result: if 'reverse' in rqspf_result: rev = rqspf_result['reverse'] bias_signals.append({ 'play': 'RQSPF', 'option': rev['recommend'], 'bias': rev['recommend_bias'], 'signal_type': '反向挖掘', 'match_id': mid }) else: max_key = max(rqspf_result['bias'], key=lambda k: abs(rqspf_result['bias'][k])) if abs(rqspf_result['bias'][max_key]) > 0.08 and rqspf_result['bias'][max_key] > 0: bias_signals.append({ 'play': 'RQSPF', 'option': max_key, 'bias': rqspf_result['bias'][max_key], 'signal_type': '正向偏差', 'match_id': mid }) tg_result = analyze_total_goals(home_xg, away_xg, match.get('total_goals', [])) if tg_result and tg_result['max_bias'] > 0.15: bias_signals.append({ 'play': '总进球', 'option': TOTAL_GOALS_LABELS[tg_result['max_index']], 'bias': tg_result['max_bias'], 'signal_type': '正向偏差', 'match_id': mid }) # ---- 平局推荐约束：只在实力接近时推荐 ---- strength_diff = abs(home_strength - away_strength) before_count = len(bias_signals) bias_signals = [ s for s in bias_signals if not (s.get('option') in ['draw', '平局'] and strength_diff >= 20) ] if before_count > len(bias_signals): print(f" ⚠️ 平局信号被过滤（实力分差{strength_diff:.1f} ≥ 20）") # ---- 试推演：半全场和比分（仅观察，不进入正式推荐） ---- trial_signals = [] # 半全场试推演 try: hf_result = analyze_half_full(home_strength, away_strength, match.get('half_full', []), league_params) if hf_result and hf_result.get('max_bias') is not None: if abs(hf_result['max_bias']) > 0.05: # 试推演阈值放宽到5% trial_signals.append({ 'play': '半全场', 'option': hf_result['recommend'], 'bias': hf_result['max_bias'], 'signal_type': '试推演', 'match_id': mid }) except Exception as e: print(f" ⚠️ 半全场试推演失败: {e}") # 比分试推演 try: score_result = analyze_score(home_xg, away_xg, match.get('score', [])) if score_result and score_result.get('max_bias') is not None: if abs(score_result['max_bias']) > 0.03: # 试推演阈值放宽到3% trial_signals.append({ 'play': '比分', 'option': score_result['recommend'], 'bias': score_result['max_bias'], 'signal_type': '试推演', 'match_id': mid }) except Exception as e: print(f" ⚠️ 比分试推演失败: {e}") # ---- 信心赋值（先赋值，再过滤） ---- for s in bias_signals: s['confidence'] = '高' if abs(s['bias']) > 0.12 else '低' # ---- 只保留高信心信号 ---- bias_signals = [s for s in bias_signals if s['confidence'] == '高'] # ---- 🆕 加载历史记录并对比 ---- if mid: history = load_history_for_match(mid) if history and len(history) > 0: prev_record = history[-1] prev_signals = prev_record.get('recommendations', []) # ---- 检查当前信号与历史是否完全相同 ---- if bias_signals and prev_signals and signals_are_identical(bias_signals, prev_signals): # 完全一样：直接沿用历史，跳过AI分析 bias_signals = prev_signals.copy() for s in bias_signals: s['match_id'] = mid if 'signal_type' not in s: s['signal_type'] = '历史沿用（无变化）' print(f" ⏭️ 偏差与历史完全相同，直接沿用 ({len(bias_signals)} 条)，跳过AI") skip_ai = True result['comparison'] = { 'previous_time': prev_record.get('generated_at', '未知'), 'changes': None, 'market_analysis': {'recommendation': '无变化，沿用历史'} } else: # 有变化或新比赛：正常走对比 changes = compare_odds_changes(result, prev_record) if changes: comparison_count += 1 market_analysis = analyze_market_impact(changes, bias_signals) result['comparison'] = { 'previous_time': prev_record.get('generated_at', '未知'), 'changes': changes, 'market_analysis': market_analysis } else: result['comparison'] = { 'previous_time': prev_record.get('generated_at', '未知'), 'changes': None, 'market_analysis': {'recommendation': '无赔率变化'} } # ---- 当前无信号但有历史推荐，则沿用 ---- if not bias_signals and prev_signals: bias_signals = prev_signals.copy() for s in bias_signals: s['match_id'] = mid if 'signal_type' not in s: s['signal_type'] = '历史沿用' print(f" 📋 沿用历史推荐 ({len(bias_signals)} 条)") skip_ai = True else: # 没有历史记录 skip_ai = False else: skip_ai = False # ---- AI 分析（如果 skip_ai 为 True，则跳过） ---- if not skip_ai: need_ai = any(abs(s['bias']) > 0.08 for s in bias_signals) if need_ai: ai_output = ai_analyze_match(match, home_strength, away_strength, home_rank, away_rank, bias_signals) result['ai_analysis'] = ai_output ai_usage_count += 1 if ai_output.get('修正建议') == '放弃': bias_signals = [] elif ai_output.get('修正建议') == '降低信心': for s in bias_signals: s['confidence'] = '低' else: # 直接沿用历史，无需AI分析 pass result['recommendations'] = bias_signals result['trial_recommendations'] = trial_signals # ---- 保存历史记录 ---- if mid: history_record = { 'generated_at': datetime.now().isoformat(), 'match_num': match['match_num'], 'odds': result['odds'], 'handicap': result['handicap'], 'strength': result['strength'], 'recommendations': result['recommendations'] } save_history_for_match(mid, history_record) # ---- 生成详细推演分析 ---- detailed = generate_detailed_analysis(match, bias_signals, result.get('comparison')) result['detailed_analysis'] = detailed match_results.append(result) # ---- 终端输出 ---- print(f"【{result['match_num']}】 {result['home_team']} vs {result['away_team']} ({result['league']})") print(f" 实力: {home_strength} - {away_strength} (排名:{home_rank} vs {away_rank})") print(f" 预期进球: {home_xg:.2f} - {away_xg:.2f}") if result.get('comparison'): changes = result['comparison'].get('changes') if changes and isinstance(changes, dict) and len(changes) > 0: print(f" 📊 赔率变化: 与 {result['comparison']['previous_time'][:16]} 对比") for key, change in changes.items(): if isinstance(change, dict) and 'direction' in change: print(f" {key}: {change['from']} → {change['to']} ({change['direction']})") else: print(f" 📊 赔率变化: 无变化（与 {result['comparison']['previous_time'][:16]} 对比）") if result['comparison'].get('market_analysis'): ma = result['comparison']['market_analysis'] print(f" 📈 市场分析: {ma.get('recommendation', '无变化')}") for s in ma.get('summary', [])[:3]: print(f" {s}") if bias_signals: for s in bias_signals: print(f" → {s['play']}: {s['option']} (偏差{s['bias']:+.0%}, 信心:{s.get('confidence', '中')})") print(" " + "-"*40) for line in detailed.split('\n'): if line.strip(): print(f" {line}") print(" " + "-"*40) # ---- 试推演输出（仅观察） ---- if trial_signals: print(f" 📝 试推演（仅供观察，不投注）：") for t in trial_signals: print(f" {t['play']}: {t['option']} (偏差{t['bias']:+.0%})") if result.get('ai_analysis'): print(f" 💬 {result['ai_analysis'].get('说明', '')[:60]}") print() time.sleep(0.1) if new_leagues: print(f"📌 发现 {len(new_leagues)} 个新联赛，已自动生成参数") # ---- 保存报告（带时间戳，不覆盖） ---- timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') report = { 'generated_at': datetime.now().isoformat(), 'total_matches': len(matches), 'ai_analyzed': ai_usage_count, 'compared_matches': comparison_count, 'new_leagues': new_leagues, 'matches': match_results } json_file = os.path.join(DAILY_DIR, f"report_{timestamp}.json") with open(json_file, 'w', encoding='utf-8') as f: json.dump(report, f, ensure_ascii=False, indent=2) print(f"✅ 完整报告已保存: {json_file}") # ---- 文本摘要 ---- txt_file = os.path.join(SUMMARY_DIR, f"summary_{timestamp}.txt") with open(txt_file, 'w', encoding='utf-8') as f: f.write(f"足球预测推演报告 (动态追踪版) - {datetime.now().strftime('%Y-%m-%d %H:%M')}\n") f.write("=" * 60 + "\n\n") f.write(f"总场次: {len(matches)}, AI分析: {ai_usage_count}, 赔率对比: {comparison_count}\n\n") for r in match_results: f.write(f"【{r['match_num']}】 {r['home_team']} vs {r['away_team']} ({r['league']})\n") if r.get('comparison'): ma = r['comparison'].get('market_analysis', {}) f.write(f" 市场分析: {ma.get('recommendation', '无变化')}\n") if r['recommendations']: for s in r['recommendations']: f.write(f" → {s['play']}: {s['option']} (偏差{s['bias']:+.0%})\n") if r.get('trial_recommendations'): f.write(" 📝 试推演（仅观察）：\n") for t in r['trial_recommendations']: f.write(f" {t['play']}: {t['option']} (偏差{t['bias']:+.0%})\n") else: f.write(f" → 无推荐\n") if r.get('ai_analysis'): f.write(f" 💬 {r['ai_analysis'].get('说明', '')}\n") # ✅ 新增：写入详细推演分析 if r.get('detailed_analysis'): f.write(r['detailed_analysis'] + "\n") f.write("\n") print(f"✅ 文本摘要已保存: {txt_file}") print(f"📊 共 {len(matches)} 场比赛，AI分析 {ai_usage_count} 场，赔率对比 {comparison_count} 场")if __name__ == '__main__': main()