python案例如何量化防守反击的效率值？

本文目录导读：

1. 防守反击效率量化模型
2. 效率量化公式总结

我来设计一个量化防守反击效率的Python案例,包含多种计算模型。

防守反击效率量化模型

基础数据准备

import pandas as pd
import numpy as np
from datetime import datetime
# 创建防守反击事件数据
data = {
    'match_id': ['M1', 'M1', 'M1', 'M2', 'M2', 'M2', 'M3', 'M3'],
    'team': ['皇马', '皇马', '皇马', '利物浦', '利物浦', '利物浦', '拜仁', '拜仁'],
    'timestamp': [10, 23, 45, 15, 38, 55, 20, 60],  # 分钟
    'start_position': [30, 25, 40, 35, 28, 45, 38, 32],  # 抢断位置（本方半场0-50，对方半场50-100）
    'duration': [8, 6, 10, 7, 5, 9, 6, 7],  # 进攻持续时间（秒）
    'passes': [3, 2, 4, 3, 2, 5, 2, 3],  # 传球次数
    'speed': [5.2, 6.1, 4.8, 5.8, 6.5, 4.2, 6.0, 5.5],  # 推进速度 m/s
    'end_position': [80, 75, 90, 85, 70, 95, 78, 88],  # 最终位置
    'outcome': ['goal', 'shot_on_target', 'shot_off_target', 'shot_on_target', 
                'tackle', 'goal', 'corner', 'shot_on_target'],  # 结果
    'xG': [0.35, 0.15, 0.08, 0.12, 0.02, 0.42, 0.05, 0.18]  # 预期进球
}
df = pd.DataFrame(data)
print("基础数据:")
print(df.head())

基础效率指标计算

class CounterAttackAnalyzer:
    def __init__(self, df):
        self.df = df.copy()
        self.result_point = {
            'goal': 1.0,           # 进球
            'shot_on_target': 0.6, # 射正
            'shot_off_target': 0.3, # 射偏
            'corner': 0.2,         # 角球
            'tackle': 0.1          # 被抢断
        }
    def basic_efficiency(self):
        """基础效率指标"""
        df = self.df.copy()
        # 1. 成功率
        df['success'] = df['outcome'].isin(['goal', 'shot_on_target']).astype(int)
        # 2. 空间推进效率
        df['space_gain'] = df['end_position'] - df['start_position']
        # 3. 速度效率
        df['speed_efficiency'] = df['speed'] / 6.0  # 规范化为最大值6m/s
        # 4. 传球效率
        df['pass_efficiency'] = df['passes'] / df['duration'] * 60  # 每分钟传球数
        # 5. 单个指标得分
        df['basic_score'] = (
            df['success'] * 0.3 +
            df['space_gain'] / 50 * 0.2 +  # 标准化空间增益
            df['speed_efficiency'] * 0.2 +
            (df['pass_efficiency'] / 30) * 0.15 +  # 标准化传球频率
            df['xG'] * 2 * 0.15  # xG贡献
        )
        return df[['match_id', 'team', 'outcome', 'space_gain', 
                  'speed_efficiency', 'pass_efficiency', 'basic_score']]
    def visualization_score(self, df):
        """可视化/综合评分模型"""
        df = df.copy()
        # 威胁等级评定
        df['threat_level'] = df['outcome'].map({
            'goal': 5, 'shot_on_target': 4, 'shot_off_target': 3,
            'corner': 2, 'tackle': 1
        })
        # 进攻速度等级
        df['speed_level'] = pd.cut(df['speed'], 
                                  bins=[0, 4, 5, 6, 10], 
                                  labels=['慢', '中', '快', '极快'])
        # 综合效率指数（0-100分）
        score = (
            df['threat_level'] * 15 +                     # 最大75分
            (df['duration'] / 10) * 10 +                  # 持续进攻10分
            df['space_gain'] / 50 * 10 +                  # 推进距离10分
            df['xG'] * 0.5 * 5                             # xG贡献5分
        ).clip(0, 100)
        df['composite_score'] = score
        return df[['match_id', 'team', 'outcome', 'composite_score']]

高级效率模型（包含多维度）

class AdvancedEfficiencyModel:
    def __init__(self, df):
        self.df = df.copy()
    def calculate_efficiency_matrix(self):
        """计算效率矩阵"""
        df = self.df.copy()
        # 1. 时间效率
        df['time_efficiency'] = self.calculate_time_efficiency(df)
        # 2. 空间效率
        df['space_efficiency'] = self.calculate_space_efficiency(df)
        # 3. 技术效率
        df['tech_efficiency'] = self.calculate_tech_efficiency(df)
        # 4. 决策效率
        df['decision_efficiency'] = self.calculate_decision_efficiency(df)
        # 5. 综合效率指数
        df['overall_efficiency'] = self.composite_index(df)
        return df
    def calculate_time_efficiency(self, df):
        """时间效率：每秒钟创造xG"""
        return df['xG'] / (df['duration'] / 60)  # 每分钟xG产出
    def calculate_space_efficiency(self, df):
        """空间效率：每米推进创造的xG"""
        space_gain = df['end_position'] - df['start_position']
        return df['xG'] / (space_gain / 50)  # 每次进攻推进50米创造的xG
    def calculate_tech_efficiency(self, df):
        """技术效率：传球/控球的精准度"""
        # 简化计算：传球创造xG的效率
        return df['xG'] / (df['passes'] + 1)  # 避免除零
    def calculate_decision_efficiency(self, df):
        """决策效率：是否选择了最好的方案"""
        # 对方半场的进攻成功率更高
        position_factor = np.where(df['start_position'] > 50, 1.2, 0.8)
        return df['xG'] * position_factor
    def composite_index(self, df):
        """综合指数（0-100）"""
        # 归一化处理
        indices = [
            df['time_efficiency'] / df['time_efficiency'].max(),
            df['space_efficiency'] / df['space_efficiency'].max(),
            df['tech_efficiency'] / df['tech_efficiency'].max(),
            df['decision_efficiency'] / df['decision_efficiency'].max()
        ]
        weights = [0.3, 0.3, 0.2, 0.2]  # 权重分布
        composite = sum(w * idx for w, idx in zip(weights, indices))
        return (composite * 100).round(2)

团队/比赛维度分析

class TeamEfficiencyAnalyzer:
    def __init__(self, df):
        self.df = df.copy()
        self.analyzer = AdvancedEfficiencyModel(df)
    def team_summary(self):
        """团队整体效率分析"""
        df_efficiency = self.analyzer.calculate_efficiency_matrix()
        # 按队伍聚合
        team_stats = df_efficiency.groupby('team').agg({
            'xG': ['sum', 'mean'],
            'overall_efficiency': ['mean', 'max'],
            'space_efficiency': 'mean',
            'time_efficiency': 'mean'
        }).round(2)
        # 扁平化列名
        team_stats.columns = ['_'.join(col).strip() for col in team_stats.columns]
        return team_stats
    def match_comparison(self):
        """比赛对比分析"""
        df_efficiency = self.analyzer.calculate_efficiency_matrix()
        # 按比赛和队伍分组
        match_comp = df_efficiency.groupby(['match_id', 'team']).agg({
            'xG': 'sum',
            'overall_efficiency': 'mean',
            'outcome': lambda x: (x == 'goal').sum()  # 反击进球数
        }).rename(columns={'outcome': 'counter_goals'})
        # 评估每场比赛的整体反击效率
        match_comp['match_rating'] = (
            match_comp['xG'] * 3 + 
            match_comp['overall_efficiency'] * 0.5 +
            match_comp['counter_goals'] * 5
        ).round(2)
        return match_comp
    def efficiency_ranking(self):
        """效率排名"""
        df_efficiency = self.analyzer.calculate_efficiency_matrix()
        # 综合排名指标
        ranking_criteria = {
            '进球权重': 0.25,
            '射正权重': 0.20, 
            'xG效率权重': 0.25,
            '推进效率权重': 0.15,
            '比赛胜率权重': 0.15
        }
        # 计算排名分数
        df_efficiency['rank_score'] = (
            df_efficiency['xG'] * 3 * ranking_criteria['xG效率权重'] +
            df_efficiency['overall_efficiency'] * 0.5 * ranking_criteria['推进效率权重'] +
            (df_efficiency['outcome'] == 'goal').astype(int) * ranking_criteria['进球权重'] * 10
        )
        # 按队伍排名
        team_ranking = df_efficiency.groupby('team')['rank_score'].mean().sort_values(ascending=False)
        return team_ranking

可视化分析

import matplotlib.pyplot as plt
import seaborn as sns
class EFFVisualizer:
    def __init__(self, analyzer):
        self.analyzer = analyzer
        self.basic_model = CounterAttackAnalyzer(analyzer.df)
    def plot_efficiency_dashboard(self):
        """效率仪表盘"""
        df_basic = self.basic_model.basic_efficiency()
        df_advanced = self.analyzer.calculate_efficiency_matrix()
        fig, axes = plt.subplots(2, 2, figsize=(14, 10))
        # 1. 基础效率分布
        ax1 = axes[0, 0]
        sns.barplot(data=df_basic, x='team', y='basic_score', ax=ax1, 
                   hue='team', palette='viridis', legend=False)
        ax1.set_title('基础反击效率得分')
        ax1.set_ylabel('效率得分')
        # 2. 综合效率指数对比
        ax2 = axes[0, 1]
        df_team = df_advanced.groupby('team')['overall_efficiency'].mean().reset_index()
        sns.boxplot(data=df_advanced, x='team', y='xG', ax=ax2, 
                   hue='team', palette='muted', legend=False)
        ax2.set_title('反击预期进球分布')
        ax2.set_ylabel('xG')
        # 3. 效率矩阵热力图
        ax3 = axes[1, 0]
        efficiency_matrix = df_advanced.groupby('team')[['time_efficiency', 'space_efficiency', 
                                                          'tech_efficiency', 'decision_efficiency']].mean()
        sns.heatmap(efficiency_matrix, annot=True, cmap='YlOrRd', 
                   fmt='.2f', ax=ax3, linewidths=0.5)
        ax3.set_title('效率维度热力图')
        # 4. 效率随时间变化
        ax4 = axes[1, 1]
        df_advanced['minute_range'] = pd.cut(df_advanced['timestamp'], 
                                           bins=[0, 15, 30, 45, 60, 75, 90], 
                                           labels=['0-15', '15-30', '30-45', '45-60', '60-75', '75-90'])
        time_efficiency = df_advanced.groupby('minute_range')['overall_efficiency'].mean()
        time_efficiency.plot(kind='line', marker='o', ax=ax4)
        ax4.set_title('比赛时间段的效率变化')
        ax4.set_xlabel('时间段（分钟）')
        ax4.set_ylabel('平均效率')
        plt.tight_layout()
        plt.show()

实时分析类（可用于直播分析）

class RealTimeEfficiencyMonitor:
    """实时防守反击效率监控"""
    def __init__(self):
        self.history = []
        self.weights = {
            'speed': 0.3,
            'accuracy': 0.3, 
            'threat': 0.4
        }
    def update_event(self, counter_attack_event):
        """更新反击事件"""
        efficiency = self.calculate_event_efficiency(counter_attack_event)
        event_data = {
            'timestamp': datetime.now(),
            **counter_attack_event,
            'efficiency_score': efficiency
        }
        self.history.append(event_data)
        return event_data
    def calculate_event_efficiency(self, event):
        """计算单次反击事件效率"""
        # 速度指标
        speed_score = min(event['speed'] / 6.0, 1)
        # 精度指标（基于传球成功率）
        accuracy_score = event.get('pass_accuracy', 0.8)  # 默认0.8
        # 威胁指标
        threat_mapping = {
            'goal': 1.0,
            'shot_on_target': 0.7,
            'shot_off_target': 0.4,
            'outside_box': 0.2,
            'failed': 0.1
        }
        threat_score = threat_mapping.get(event['outcome'], 0.1)
        # 加权计算
        total_score = (
            self.weights['speed'] * speed_score +
            self.weights['accuracy'] * accuracy_score +
            self.weights['threat'] * threat_score
        )
        return round(total_score * 100, 2)
    def get_live_summary(self, minutes=5):
        """获取最近X分钟的实时效率"""
        if not self.history:
            return {'message': '暂无数据'}
        recent = [h for h in self.history if 
                 (datetime.now() - h['timestamp']).total_seconds()/60 <= minutes]
        if recent:
            avg_efficiency = np.mean([r['efficiency_score'] for r in recent])
            return {
                'counter_attacks': len(recent),
                'average_efficiency': round(avg_efficiency, 2),
                'top_efficiency': max(recent, key=lambda x: x['efficiency_score'])['efficiency_score']
            }
        else:
            return {'message': '规定时间内无反击事件'}

使用示例

def main():
    # 1. 加载数据
    data = {
        'match_id': ['M1', 'M1', 'M1', 'M2', 'M2', 'M2'],
        'team': ['皇马', '皇马', '皇马', '利物浦', '利物浦', '利物浦'],
        'timestamp': [10, 23, 45, 15, 38, 55],
        'start_position': [30, 25, 40, 35, 28, 45],
        'duration': [8, 6, 10, 7, 5, 9],
        'passes': [3, 2, 4, 3, 2, 5],
        'speed': [5.2, 6.1, 4.8, 5.8, 6.5, 4.2],
        'end_position': [80, 75, 90, 85, 70, 95],
        'outcome': ['goal', 'shot_on_target', 'shot_off_target', 'shot_on_target', 
                   'tackle', 'goal'],
        'xG': [0.35, 0.15, 0.08, 0.12, 0.02, 0.42]
    }
    df = pd.DataFrame(data)
    # 2. 运行分析
    basic_analyzer = CounterAttackAnalyzer(df)
    basic_results = basic_analyzer.basic_efficiency()
    print("基础效率分析:")
    print(basic_results)
    # 3. 高级分析
    advanced = AdvancedEfficiencyModel(df)
    eff_matrix = advanced.calculate_efficiency_matrix()
    print("\n高级效率矩阵:")
    print(eff_matrix[['team', 'time_efficiency', 'space_efficiency', 
                     'tech_efficiency', 'decision_efficiency', 'overall_efficiency']])
    # 4. 团队分析
    team_analyzer = TeamEfficiencyAnalyzer(df)
    team_results = team_analyzer.team_summary()
    print("\n团队效率汇总:")
    print(team_results)
    # 5. 可视化
    visualizer = EFFVisualizer(team_analyzer)
    visualizer.plot_efficiency_dashboard()
    # 6. 实时监控示例
    monitor = RealTimeEfficiencyMonitor()
    event = {
        'speed': 5.5,
        'pass_accuracy': 0.85,
        'outcome': 'shot_on_target'
    }
    result = monitor.update_event(event)
    print(f"\n实时反击效率: {result['efficiency_score']}")
    print(monitor.get_live_summary(minutes=5))
if __name__ == "__main__":
    main()

效率量化公式总结

计算公式: 效率值 = ∑(权重 × 对应指标)

主要指标权重分配：

这个量化框架可以帮助教练和数据分析师：

1. 客观评估每次防守反击的质量
2. 比较不同队伍的反击效率
3. 实时监控比赛的攻防转换效率
4. 识别高效率反击的关键因素