如何将时序得分数据集划分为每日均值相近的三个子集？

针对将时间序列得分数据集划分为3组、使每组各日期均值尽可能接近的需求，以下是几种可行的Python实现思路，涵盖暴力解法（小数据集）和启发式近似解法（大数据集）：

一、暴力解法（仅适用于小数据集） #

如果你的数据集样本量很小（比如n≤10），可以直接枚举所有可能的分组组合，筛选出损失最小的方案。核心思路是遍历每个样本分到3组的所有可能，计算每组各日期的均值差异，选择最优解。

实现代码 #

import pandas as pd
import itertools
import numpy as np

# 示例数据集
data = pd.DataFrame({
    'id': [0,1,2,3],
    'day1': [0,0,0,0],
    'day2': [0.5,0.25,0,0.5],
    'day3': [1,0.75,0.5,1],
    'day4': [3,2.25,2.75,2]
})
scores = data.drop('id', axis=1).values
n_samples = len(scores)
n_groups = 3

# 定义损失函数：所有日期的组间均值方差之和（值越小，均值越接近）
def calculate_loss(group_assignments):
    group_sums = np.zeros((n_groups, scores.shape[1]))
    group_counts = np.zeros(n_groups)
    for idx, g in enumerate(group_assignments):
        group_sums[g] += scores[idx]
        group_counts[g] += 1
    # 跳过有空组的情况
    if 0 in group_counts:
        return float('inf')
    group_means = group_sums / group_counts[:, np.newaxis]
    # 计算每个日期的组间方差，求和得到总损失
    date_variances = np.var(group_means, axis=0)
    return date_variances.sum()

best_loss = float('inf')
best_assignment = None

# 枚举所有分组可能（仅适合小样本）
for assignment in itertools.product(range(n_groups), repeat=n_samples):
    current_loss = calculate_loss(assignment)
    if current_loss < best_loss:
        best_loss = current_loss
        best_assignment = assignment

# 输出结果
data['group'] = best_assignment
print("最优分组结果：")
print(data)
print(f"总损失值：{best_loss:.4f}")

局限性 #

当样本量超过15时，3^15=1400万+的组合数会导致计算时间急剧增加，仅适合极小数据集验证结果。

二、启发式近似解法（适合大数据集） #

对于中等或大规模数据集，以下几种启发式方法可以在可接受的时间内得到优质近似解：

1. 贪心迭代优化（交换样本） #

基于随机初始分组，通过迭代移动样本到其他组来降低损失，直到无法优化为止，是对随机划分的直接改进。

实现代码

import pandas as pd
import numpy as np

data = pd.DataFrame({
    'id': [0,1,2,3],
    'day1': [0,0,0,0],
    'day2': [0.5,0.25,0,0.5],
    'day3': [1,0.75,0.5,1],
    'day4': [3,2.25,2.75,2]
})
scores = data.drop('id', axis=1).values
n_samples = len(scores)
n_groups = 3

# 初始化分组：随机分配，确保每组非空
group_assignments = np.random.randint(0, n_groups, size=n_samples)
while len(set(group_assignments)) < n_groups:
    group_assignments = np.random.randint(0, n_groups, size=n_samples)

# 预计算初始组总和与样本数
group_sums = np.zeros((n_groups, scores.shape[1]))
group_counts = np.zeros(n_groups, dtype=int)
for idx, g in enumerate(group_assignments):
    group_sums[g] += scores[idx]
    group_counts[g] += 1

def get_current_loss():
    group_means = group_sums / group_counts[:, np.newaxis]
    return np.var(group_means, axis=0).sum()

best_loss = get_current_loss()
improved = True
iterations = 0

# 迭代优化
while improved and iterations < 1000:
    improved = False
    iterations += 1
    for idx in range(n_samples):
        current_group = group_assignments[idx]
        # 尝试将当前样本移到其他组
        for target_group in range(n_groups):
            if target_group == current_group:
                continue
            # 临时更新组统计数据
            temp_sums = group_sums.copy()
            temp_counts = group_counts.copy()
            temp_sums[current_group] -= scores[idx]
            temp_counts[current_group] -= 1
            temp_sums[target_group] += scores[idx]
            temp_counts[target_group] += 1
            # 计算临时损失
            temp_means = temp_sums / temp_counts[:, np.newaxis]
            temp_loss = np.var(temp_means, axis=0).sum()
            # 如果损失降低则保留移动
            if temp_loss < best_loss:
                best_loss = temp_loss
                group_sums = temp_sums
                group_counts = temp_counts
                group_assignments[idx] = target_group
                improved = True
                break
        if improved:
            continue

# 输出结果
data['group'] = group_assignments
print("迭代优化后的分组结果：")
print(data)
print(f"总损失值：{best_loss:.4f}")

2. 遗传算法（适合复杂场景） #

通过模拟自然选择过程，迭代优化分组方案，适合对近似解质量要求较高的场景。需要安装deap库：pip install deap

实现代码

import pandas as pd
import numpy as np
from deap import base, creator, tools, algorithms

data = pd.DataFrame({
    'id': [0,1,2,3],
    'day1': [0,0,0,0],
    'day2': [0.5,0.25,0,0.5],
    'day3': [1,0.75,0.5,1],
    'day4': [3,2.25,2.75,2]
})
scores = data.drop('id', axis=1).values
n_samples = len(scores)
n_groups = 3

# 创建遗传算法核心类：最小化损失
creator.create("FitnessMin", base.Fitness, weights=(-1.0,))
creator.create("Individual", list, fitness=creator.FitnessMin)

toolbox = base.Toolbox()
toolbox.register("attr_group", np.random.randint, 0, n_groups)
toolbox.register("individual", tools.initRepeat, creator.Individual, toolbox.attr_group, n=n_samples)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)

# 定义适应度计算（即损失函数）
def evaluate(individual):
    group_sums = np.zeros((n_groups, scores.shape[1]))
    group_counts = np.zeros(n_groups, dtype=int)
    for idx, g in enumerate(individual):
        group_sums[g] += scores[idx]
        group_counts[g] += 1
    # 空组惩罚
    if len(set(individual)) < n_groups:
        return (float('inf'),)
    group_means = group_sums / group_counts[:, np.newaxis]
    loss = np.var(group_means, axis=0).sum()
    return (loss,)

toolbox.register("mate", tools.cxTwoPoint)
toolbox.register("mutate", tools.mutUniformInt, low=0, up=n_groups-1, indpb=0.1)
toolbox.register("select", tools.selTournament, tournsize=3)
toolbox.register("evaluate", evaluate)

def main():
    pop = toolbox.population(n=50)  # 种群大小
    hof = tools.HallOfFame(1)  # 保存最优个体
    stats = tools.Statistics(lambda ind: ind.fitness.values)
    stats.register("avg", np.mean)
    stats.register("min", np.min)

    # 运行遗传算法
    pop, log = algorithms.eaSimple(pop, toolbox, cxpb=0.5, mutpb=0.2, ngen=100, stats=stats, halloffame=hof, verbose=True)

    # 输出结果
    best_assignment = hof[0]
    data['group'] = best_assignment
    print("\n遗传算法最优分组结果：")
    print(data)
    print(f"总损失值：{evaluate(best_assignment)[0]:.4f}")

if __name__ == "__main__":
    main()

三、损失函数的可选调整 #

可以根据实际需求替换损失函数：

• 组间方差之和：默认方案，衡量均值的离散程度
• 组均值与全局均值的平方差之和：让每组均值尽可能贴近整体平均水平
• 最大组间差异：最小化所有日期中"最大均值-最小均值"的最大值，适合要求均值绝对接近的场景

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