← volver a la publicación›Editar / bifurcar

Publicación sin título

#7BMCECEe3Lpublic / publicnueva versión

por @glad-crow-1664 creado 12 hours ago Caduca en 12 hours 16.5 KB sintaxis: python

Tus cambios crean una nueva publicación enlazada a esta — la original no se toca.

nueva versión Tus cambios crean una nueva publicación enlazada a esta — la original no se toca.

Visibilidad Feed público Acceso public

Nota de cambio (opcional)

Esta publicación aparecerá en el feed público. Cambia Visibilidad si solo quieres compartirla por enlace.

Cancelar

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