如何用Pandas实现满足三类约束的任务公平再分配算法？

实现思路 #

• 任务已经按days_in_prog、days_after_upd降序排序，直接按现有顺序逐个处理，满足优先分配进展时间更长任务的要求
• 维护动态更新的处理人持有任务数字典，每次分配时先过滤掉当前任务的原处理人，再从剩余处理人中选择持有任务数最少的对象，同时更新其持有任务数，同时满足新处理人不可与原处理人相同和优先分配给任务数更少处理人的要求

完整实现代码 #

import pandas as pd
import numpy as np
np.random.seed(0)

# 构造任务表
tasks = pd.DataFrame()
tasks['task_id'] = np.arange(1, 101).astype('object')
tasks['assignee_id'] = np.random.randint(low=1, high=11, size=100).astype('object')
tasks['days_in_prog'] = np.random.randint(low=11, high=16, size=100)
tasks['days_after_upd'] = np.random.randint(low=6, high=11, size=100)
tasks = tasks.sort_values(['days_in_prog', 'days_after_upd'], ascending=False).reset_index(drop=True)

# 构造处理人表
assignees = pd.DataFrame()
assignees['assignee_id'] = np.arange(1, 11).astype('object')
assignees['tasks'] = np.random.randint(low=5, high=16, size=10)
assignees = assignees.sort_values('tasks').reset_index(drop=True)

# 初始化处理人任务数字典
assignee_task_cnt = assignees.set_index('assignee_id')['tasks'].to_dict()
new_assignees = []

# 按优先级逐个分配任务
for _, row in tasks.iterrows():
    original_assignee = row['assignee_id']
    # 过滤原处理人，得到可选候选人
    candidates = [a for a in assignee_task_cnt if a != original_assignee]
    # 选择任务数最少的候选人
    selected = min(candidates, key=lambda x: assignee_task_cnt[x])
    new_assignees.append(selected)
    # 更新处理人任务数
    assignee_task_cnt[selected] += 1

# 写入新处理人字段
tasks['new_assignee'] = new_assignees

# 查看前10条分配结果
print(tasks.head(10))
# 验证新老处理人无重复
print("新老处理人重复数：", sum(tasks['assignee_id'] == tasks['new_assignee']))

结果验证 #

运行代码后可确认：

1. 输出新老处理人重复数：0，符合新处理人不能与原处理人相同的要求
2. 最终所有处理人持有任务数差值最大为1，分配公平
3. 优先处理了进展时间更长、距上次更新更久的任务，符合优先级规则

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