rendered paste body # ---- 数据完整性检查：如果排名缺失，尝试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()