当前位置：首页 > news >正文

基于分组规则的Excel数据分组优化系统设计与实现

news 2025/7/29 15:27:06

基于分组规则的Excel数据分组优化系统设计与实现

前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住分享一下给大家，觉得好请收藏。点击跳转到网站。

1. 引言

在现代数据分析和业务决策中，经常需要对数据进行分组并计算最优组合。本文介绍一个基于Python开发的Excel数据分组优化系统，该系统能够根据给定的分组规则对数据进行分组，并通过组合计算找出最大总金额的解决方案。系统支持小数据集的穷举法和大数据集的优化算法，使用pandas进行数据处理，pulp进行线性优化，具有高效、灵活的特点。

2. 系统设计与架构

2.1 系统需求分析

系统需要满足以下核心需求：

从Excel文件中读取数据并进行预处理
根据用户定义的分组规则对数据进行分组
对于小数据集，使用穷举法找到最优分组组合
对于大数据集，使用线性规划优化算法高效求解
输出分组结果、摘要信息、最大金额和验证结果
提供友好的用户交互界面

2.2 系统架构设计

系统采用分层架构设计，主要分为以下几层：

数据访问层：负责Excel文件的读取和写入
业务逻辑层：包含数据预处理、分组规则处理、优化算法实现
应用层：提供用户界面和结果展示

2.3 技术选型

Python：作为主要开发语言，具有丰富的数据处理库
pandas：用于高效处理表格数据
pulp：用于线性规划优化问题求解
openpyxl：用于Excel文件读写操作

3. 核心算法实现

3.1 数据预处理

首先实现数据读取和预处理功能：

import pandas as pd
from pulp import *
import itertools
import time
from openpyxl import load_workbook
from openpyxl.styles import Font, Alignment
from openpyxl.utils import get_column_letterclass DataGroupOptimizer:def __init__(self, file_path):"""初始化优化器，加载Excel文件:param file_path: Excel文件路径"""self.file_path = file_pathself.data = Noneself.group_rules = {}self.max_amount = 0self.best_groups = []self.summary = {}self.load_data()def load_data(self):"""从Excel文件加载数据"""try:self.data = pd.read_excel(self.file_path)print("数据加载成功，共{}条记录".format(len(self.data)))# 数据清洗：去除空值self.data.dropna(inplace=True)except Exception as e:print("数据加载失败:", str(e))def set_group_rules(self, rules):"""设置分组规则:param rules: 字典形式的分组规则，如{'类别': ['A', 'B'], '地区': ['东', '西']}"""self.group_rules = rulesprint("分组规则设置成功:", rules)def preprocess_data(self):"""数据预处理，确保所需字段存在且格式正确"""# 检查必须字段required_columns = ['金额'] + list(self.group_rules.keys())for col in required_columns:if col not in self.data.columns:raise ValueError(f"数据中缺少必要列: {col}")# 确保金额是数值类型self.data['金额'] = pd.to_numeric(self.data['金额'], errors='coerce')self.data.dropna(subset=['金额'], inplace=True)# 对分组列进行标准化处理for col in self.group_rules.keys():self.data[col] = self.data[col].astype(str).str.strip()

3.2 穷举法实现（小数据集）

对于小数据集，我们可以使用穷举法找到最优解：

    def exhaustive_search(self):"""穷举法寻找最优分组组合（适用于小数据集）"""start_time = time.time()# 获取所有可能的分组组合group_combinations = self._generate_group_combinations()max_amount = 0best_groups = []# 遍历所有组合for combo in group_combinations:current_amount = 0valid = True# 检查组合是否满足互斥条件for i in range(len(combo)):for j in range(i+1, len(combo)):if not self._check_exclusive(combo[i], combo[j]):valid = Falsebreakif not valid:break# 如果组合有效，计算总金额if valid:current_amount = sum([self._get_group_amount(g) for g in combo])if current_amount > max_amount:max_amount = current_amountbest_groups = comboself.max_amount = max_amountself.best_groups = best_groupsself.summary['method'] = '穷举法'self.summary['time'] = time.time() - start_timeself.summary['combinations'] = len(group_combinations)return best_groups, max_amountdef _generate_group_combinations(self):"""生成所有可能的分组组合"""# 首先生成所有可能的分组all_groups = []for col, values in self.group_rules.items():for val in values:group = {col: val}all_groups.append(group)# 生成所有可能的非空子集combinations = []for r in range(1, len(all_groups)+1):combinations.extend(itertools.combinations(all_groups, r))return combinationsdef _check_exclusive(self, group1, group2):"""检查两个分组是否互斥（共享同一分类的不同值）"""for key in group1:if key in group2:if group1[key] != group2[key]:return Falsereturn Truedef _get_group_amount(self, group):"""计算特定分组的金额总和"""query_parts = []for key, value in group.items():query_parts.append(f"{key} == '{value}'")query = " & ".join(query_parts)return self.data.query(query)['金额'].sum()

3.3 线性规划优化实现（大数据集）

对于大数据集，我们使用线性规划方法：

    def linear_programming_optimization(self):"""使用线性规划寻找最优分组组合（适用于大数据集）"""start_time = time.time()# 生成所有可能的分组groups = []group_amounts = {}group_indices = {}# 为每个分组规则创建分组index = 0for col, values in self.group_rules.items():for val in values:group = {col: val}groups.append(group)amount = self._get_group_amount(group)group_amounts[index] = amountgroup_indices[index] = groupindex += 1# 创建问题实例prob = LpProblem("GroupOptimization", LpMaximize)# 创建决策变量x = LpVariable.dicts('x', group_indices.keys(), cat='Binary')# 目标函数：最大化总金额prob += lpSum([group_amounts[i] * x[i] for i in group_indices.keys()])# 约束条件：互斥分组不能同时选择# 首先找出所有互斥的分组对exclusive_pairs = []for i in group_indices.keys():for j in group_indices.keys():if i < j and not self._check_exclusive(group_indices[i], group_indices[j]):exclusive_pairs.append((i, j))# 添加互斥约束for i, j in exclusive_pairs:prob += x[i] + x[j] <= 1, f"Exclusive_{i}_{j}"# 求解问题prob.solve(PULP_CBC_CMD(msg=False))# 解析结果selected_groups = []total_amount = 0for i in group_indices.keys():if x[i].value() == 1:selected_groups.append(group_indices[i])total_amount += group_amounts[i]self.max_amount = total_amountself.best_groups = selected_groupsself.summary['method'] = '线性规划'self.summary['time'] = time.time() - start_timeself.summary['variables'] = len(group_indices)self.summary['constraints'] = len(exclusive_pairs)return selected_groups, total_amount

3.4 自动选择算法

系统根据数据规模自动选择合适算法：

    def optimize(self, threshold=1000):"""自动选择优化方法并执行:param threshold: 使用穷举法的最大分组组合数阈值"""# 计算可能的分组组合数total_groups = sum([len(v) for v in self.group_rules.values()])total_combinations = 2**total_groups - 1if total_combinations <= threshold:print(f"分组组合数{total_combinations}小于阈值{threshold}，使用穷举法")return self.exhaustive_search()else:print(f"分组组合数{total_combinations}大于阈值{threshold}，使用线性规划")return self.linear_programming_optimization()

4. 结果输出与验证

4.1 结果输出到Excel

    def export_results(self, output_path):"""将优化结果导出到Excel文件"""try:# 创建新的工作簿wb = load_workbook(self.file_path)if '优化结果' in wb.sheetnames:del wb['优化结果']if '摘要' in wb.sheetnames:del wb['摘要']# 添加优化结果工作表ws_result = wb.create_sheet('优化结果')# 写入标题行headers = list(self.data.columns) + ['是否选中']for col_num, header in enumerate(headers, 1):ws_result.cell(row=1, column=col_num, value=header).font = Font(bold=True)# 标记被选中的记录selected_indices = []for group in self.best_groups:query_parts = []for key, value in group.items():query_parts.append(f"{key} == '{value}'")query = " & ".join(query_parts)selected = self.data.query(query)selected_indices.extend(selected.index.tolist())# 写入数据for row_num, (_, row) in enumerate(self.data.iterrows(), 2):for col_num, value in enumerate(row, 1):ws_result.cell(row=row_num, column=col_num, value=value)# 标记是否选中ws_result.cell(row=row_num, column=len(headers), value='是' if row.name in selected_indices else '否')# 添加摘要工作表ws_summary = wb.create_sheet('摘要')# 写入摘要信息ws_summary.append(['优化方法', self.summary.get('method', '')])ws_summary.append(['最大金额', self.max_amount])ws_summary.append(['计算时间(秒)', self.summary.get('time', '')])ws_summary.append(['分组组合数', self.summary.get('combinations', self.summary.get('variables', ''))])ws_summary.append(['约束条件数', self.summary.get('constraints', '')])ws_summary.append(['最优分组数', len(self.best_groups)])# 写入最优分组详情ws_summary.append([])ws_summary.append(['最优分组详情:'])for i, group in enumerate(self.best_groups, 1):group_desc = ", ".join([f"{k}:{v}" for k, v in group.items()])amount = self._get_group_amount(group)ws_summary.append([f"分组{i}", group_desc, f"金额: {amount}"])# 设置样式for row in ws_summary.iter_rows():for cell in row:cell.font = Font(bold=True) if cell.row <= 6 else None# 保存文件wb.save(output_path)print(f"结果已成功导出到 {output_path}")return Trueexcept Exception as e:print("导出结果时出错:", str(e))return False

4.2 结果验证

    def validate_results(self):"""验证优化结果的正确性"""# 检查是否有重叠的分组for i in range(len(self.best_groups)):for j in range(i+1, len(self.best_groups)):if not self._check_exclusive(self.best_groups[i], self.best_groups[j]):print(f"验证失败: 分组{self.best_groups[i]}和{self.best_groups[j]}存在冲突")return False# 检查总金额计算是否正确calculated_amount = sum([self._get_group_amount(g) for g in self.best_groups])if abs(calculated_amount - self.max_amount) > 0.01:print(f"验证失败: 计算金额{calculated_amount}与报告金额{self.max_amount}不符")return Falseprint("验证通过: 所有分组互不冲突，总金额计算正确")return True

5. 使用示例与性能测试

5.1 基本使用示例

# 示例使用代码
if __name__ == "__main__":# 创建优化器实例optimizer = DataGroupOptimizer('sales_data.xlsx')# 设置分组规则group_rules = {'产品类别': ['电子产品', '家居用品', '服装'],'地区': ['东部', '西部', '北部', '南部'],'季度': ['Q1', 'Q2', 'Q3', 'Q4']}optimizer.set_group_rules(group_rules)# 数据预处理optimizer.preprocess_data()# 执行优化best_groups, max_amount = optimizer.optimize(threshold=10000)# 输出结果print(f"\n最优分组方案(共{len(best_groups)}个分组):")for group in best_groups:print(group)print(f"\n最大总金额: {max_amount}")# 验证结果optimizer.validate_results()# 导出结果optimizer.export_results('sales_data_optimized.xlsx')