Python体育联赛所有可能结果计算的性能优化问题

问题背景 #

我正在开发一款计算体育联赛所有可能结果的应用，联赛包含9支球队，剩余8场五局三胜制比赛，需要计算任意球队跻身特定排名区间（如B队进入前三）的概率。从CSV读取当前积分榜，每场比赛有6种可能结果（3:0、3:1、3:2、0:3、1:3、2:3），需追踪这些结果以处理同分情况下的胜局数排名规则。当前代码可生成所有可能结果，但8场比赛共约170万种场景，计算耗时预估达50分钟，现寻求可行的程序运行时优化方法。

原实现代码 #

import pandas as pd
from itertools import product


# simulates all possible outcomes
def simulate_league(initial_standings, remaining_games):
    possible_outcomes = list()
    for winloss_schedule in product(['www', 'wwwl', 'wwwll', 'lll', 'lllw', 'lllww'], repeat=len(remaining_games)):
        possible_outcomes.append(play_out(initial_standings, remaining_games, winloss_schedule))
    return possible_outcomes


# plays out one possible scenario
def play_out(initial_standings, remaining_games, winloss_schedule):
    standings_frame = initial_standings.copy()

    for (team_a, team_b), outcome in zip(remaining_games, winloss_schedule):
        if outcome == "www":  # 3:0 for team a
            standings_frame.loc[team_a]["points"] += 3
            standings_frame.loc[team_a]["set_win"] += 3
            standings_frame.loc[team_b]["set_loss"] += 3

        elif outcome == "wwwl":  # 3:1 for team a
            standings_frame.loc[team_a]["points"] += 3
            standings_frame.loc[team_a]["set_win"] += 3
            standings_frame.loc[team_a]["set_loss"] += 1
            standings_frame.loc[team_b]["set_win"] += 1
            standings_frame.loc[team_b]["set_loss"] += 3

        elif outcome == "wwwll":  # 3:2 for team a
            standings_frame.loc[team_a]["points"] += 2
            standings_frame.loc[team_a]["set_win"] += 3
            standings_frame.loc[team_a]["set_loss"] += 2
            standings_frame.loc[team_b]["points"] += 1
            standings_frame.loc[team_b]["set_win"] += 2
            standings_frame.loc[team_b]["set_loss"] += 3

        elif outcome == "lll":  # 3:0 for team b
            standings_frame.loc[team_b]["points"] += 3
            standings_frame.loc[team_b]["set_win"] += 3
            standings_frame.loc[team_a]["set_loss"] += 3

        elif outcome == "lllw":  # 3:1 for team b
            standings_frame.loc[team_b]["points"] += 3
            standings_frame.loc[team_b]["set_win"] += 3
            standings_frame.loc[team_b]["set_loss"] += 1
            standings_frame.loc[team_a]["set_win"] += 1
            standings_frame.loc[team_a]["set_loss"] += 3

        elif outcome == "lllww":  # 3:2 for team b
            standings_frame.loc[team_b]["points"] += 2
            standings_frame.loc[team_b]["set_win"] += 3
            standings_frame.loc[team_b]["set_loss"] += 2
            standings_frame.loc[team_a]["points"] += 1
            standings_frame.loc[team_a]["set_win"] += 2
            standings_frame.loc[team_a]["set_loss"] += 3

    return standings_frame


# reads initial standings from csv
def get_initial_standings():
    return pd.read_csv("standings.csv", sep=';', index_col="Team")


def get_remaining_games():
    remaining_games = ["A:B", "C:D", "A:D", "D:B", "E:F", "F:E", "G:E", "E:B"]
    remaining_games = [tuple(s.split(":")) for s in remaining_games]
    return remaining_games


simulate_league(get_initial_standings(), get_remaining_games())

可行优化方案 #

1. 替换Pandas为原生字典减少操作开销 #

Pandas的DataFrame在频繁复制和行级修改时存在大量性能损耗，改用嵌套字典存储积分榜数据，直接操作字典键值对，能大幅降低单场景计算时间：

def get_initial_standings():
    df = pd.read_csv("standings.csv", sep=';', index_col="Team")
    # 转换为嵌套字典：{球队名: {'points': 积分, 'set_win': 胜局数, 'set_loss': 负局数}}
    return df.to_dict('index')

2. 预定义比赛结果增量，消除冗余判断 #

把6种比赛结果对应的积分、胜局、负局增量提前封装成字典，避免循环内的大量if-elif分支判断：

# 预定义每种结果对应的两队数据增量
OUTCOME_DELTAS = {
    "www": (
        {"points": 3, "set_win": 3, "set_loss": 0},
        {"points": 0, "set_win": 0, "set_loss": 3}
    ),
    "wwwl": (
        {"points": 3, "set_win": 3, "set_loss": 1},
        {"points": 0, "set_win": 1, "set_loss": 3}
    ),
    "wwwll": (
        {"points": 2, "set_win": 3, "set_loss": 2},
        {"points": 1, "set_win": 2, "set_loss": 3}
    ),
    "lll": (
        {"points": 0, "set_win": 0, "set_loss": 3},
        {"points": 3, "set_win": 3, "set_loss": 0}
    ),
    "lllw": (
        {"points": 0, "set_win": 1, "set_loss": 3},
        {"points": 3, "set_win": 3, "set_loss": 1}
    ),
    "lllww": (
        {"points": 1, "set_win": 2, "set_loss": 3},
        {"points": 2, "set_win": 3, "set_loss": 2}
    )
}

修改play_out函数，直接通过字典获取增量更新数据：

import copy

def play_out(initial_standings, remaining_games, winloss_schedule):
    # 深拷贝初始字典，避免修改原数据
    standings = copy.deepcopy(initial_standings)
    for (team_a, team_b), outcome in zip(remaining_games, winloss_schedule):
        delta_a, delta_b = OUTCOME_DELTAS[outcome]
        # 批量更新两队数据
        for key in delta_a:
            standings[team_a][key] += delta_a[key]
            standings[team_b][key] += delta_b[key]
    return standings

3. 实时统计目标概率，避免全场景存储 #

不需要保存所有170万种场景的完整积分榜，而是在每个场景计算完成后直接判断目标条件（如B队是否进入前三），只统计符合条件的场景数量，能大幅节省内存并减少后续遍历开销：

from itertools import product

def calculate_rank_probability(initial_standings, remaining_games, target_team, min_rank=1, max_rank=3):
    total_scenarios = 6 ** len(remaining_games)
    success_count = 0
    initial_dict = initial_standings.to_dict('index') if isinstance(initial_standings, pd.DataFrame) else initial_standings

    for winloss_schedule in product(OUTCOME_DELTAS.keys(), repeat=len(remaining_games)):
        standings = copy.deepcopy(initial_dict)
        for (team_a, team_b), outcome in zip(remaining_games, winloss_schedule):
            delta_a, delta_b = OUTCOME_DELTAS[outcome]
            for key in delta_a:
                standings[team_a][key] += delta_a[key]
                standings[team_b][key] += delta_b[key]
        # 按积分降序、净胜局降序排序
        sorted_teams = sorted(standings.items(), key=lambda x: (-x[1]['points'], -(x[1]['set_win'] - x[1]['set_loss'])))
        # 获取目标球队排名
        target_rank = next(idx + 1 for idx, (team, _) in enumerate(sorted_teams) if team == target_team)
        if min_rank <= target_rank <= max_rank:
            success_count += 1
    return success_count / total_scenarios

4. 并行计算利用多核CPU #

使用multiprocessing模块将场景拆分到多个进程并行处理，充分利用多核CPU资源，能将计算时间按核心数近似缩短：

from multiprocessing import Pool
import itertools

def process_single_scenario(args):
    initial_dict, remaining_games, outcome_schedule, target_team, rank_range = args
    standings = copy.deepcopy(initial_dict)
    for (team_a, team_b), outcome in zip(remaining_games, outcome_schedule):
        delta_a, delta_b = OUTCOME_DELTAS[outcome]
        for key in delta_a:
            standings[team_a][key] += delta_a[key]
            standings[team_b][key] += delta_b[key]
    # 计算排名
    sorted_teams = sorted(standings.items(), key=lambda x: (-x[1]['points'], -(x[1]['set_win'] - x[1]['set_loss'])))
    target_rank = next(idx + 1 for idx, (team, _) in enumerate(sorted_teams) if team == target_team)
    return 1 if rank_range[0] <= target_rank <= rank_range[1] else 0

def calculate_rank_probability_parallel(initial_standings, remaining_games, target_team, min_rank=1, max_rank=3):
    initial_dict = initial_standings.to_dict('index') if isinstance(initial_standings, pd.DataFrame) else initial_standings
    rank_range = (min_rank, max_rank)
    # 生成所有场景的参数元组
    scenarios = [(initial_dict, remaining_games, schedule, target_team, rank_range) 
                 for schedule in itertools.product(OUTCOME_DELTAS.keys(), repeat=len(remaining_games))]
    # 启动进程池并行处理
    with Pool() as pool:
        results = pool.map(process_single_scenario, scenarios)
    success_count = sum(results)
    total_scenarios = len(scenarios)
    return success_count / total_scenarios

5. 场景剪枝提前终止无效计算 #

对于部分场景，在计算过程中如果已经能确定目标球队的排名必然符合或不符合条件，可以提前终止后续比赛的计算，减少不必要的操作。比如：

• 当目标球队的积分已经远超所有可能的竞争对手，剩余比赛无论结果如何都能进入前三，直接判定为成功场景
• 当目标球队的积分已经落后到即使剩余比赛全赢也无法进入前三，直接判定为失败场景

内容的提问来源于stack exchange，提问作者Axodarap