""" 模拟数据生成脚本 为Python程序开发竞赛数据清洗模块生成4套模拟数据 """ import pandas as pd import numpy as np import random import os import glob OUTPUT_DIR = "/workspace/模拟数据" os.makedirs(OUTPUT_DIR, exist_ok=True) random.seed(42) np.random.seed(42) # ============================================================ # 模拟题一：电商订单数据 # ============================================================ def generate_order_data(): """生成电商订单模拟数据（含脏数据）""" categories = { "手机数码": ["iPhone 14 Pro", "华为Mate50", "小米13", "OPPO Find X6", "vivo X90"], "电脑办公": ["联想小新Pro16", "MacBook Air M2", "戴尔灵越15", "华为MateBook D16"], "家用电器": ["美的空调", "格力冰箱", "海尔洗衣机", "苏泊尔电饭煲", "九阳豆浆机"], "服装鞋帽": ["Nike Air Max", "Adidas Ultraboost", "优衣库羽绒服", "李宁运动鞋"], "食品饮料": ["三只松鼠坚果", "蒙牛纯牛奶", "农夫山泉", "百事可乐", "良品铺子零食"], "美妆个护": ["兰蔻小黑瓶", "雅诗兰黛眼霜", "欧莱雅面膜", "资生堂洗面奶"], "图书文具": ["Python编程从入门到精通", "数据结构与算法", "晨光中性笔套装"], "家居家装": ["宜家书架", "全友家居沙发", "罗莱家纺四件套"], } pay_methods = ["微信支付", "支付宝", "银行卡支付", "京东白条", "花呗"] statuses = ["已完成", "已完成", "已完成", "待发货", "已取消", "已退款", "待收货"] cities = ["湖南省长沙市岳麓区", "湖南省长沙市天心区", "湖南省株洲市天元区", "广东省深圳市南山区", "北京市朝阳区", "上海市浦东新区", "浙江省杭州市西湖区", "湖北省武汉市洪山区", "四川省成都市武侯区"] weekdays = ["星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日"] records = [] order_id = 1 for month in range(1, 13): for day in range(1, 29): n_orders = random.randint(3, 8) for _ in range(n_orders): cat = random.choice(list(categories.keys())) product = random.choice(categories[cat]) qty = random.randint(1, 10) price = round(random.uniform(5, 15000), 2) total = round(qty * price, 2) date_str = f"2023年{month:02d}月{day:02d}日" weekday = weekdays[random.randint(0, 6)] order_time = f"{date_str} {weekday}" user_id = f"U{random.randint(10000, 99999)}" order_no = f"DD2023{month:02d}{day:02d}-{order_id:04d}" pay = random.choice(pay_methods) status = random.choice(statuses) address = f"{random.choice(cities)}XX路{random.randint(1,200)}号" records.append([order_no, order_time, user_id, product, cat, qty, price, total, pay, status, address]) order_id += 1 df = pd.DataFrame(records, columns=[ "订单编号", "下单时间", "用户ID", "商品名称", "商品类别", "购买数量", "单价(元)", "订单金额(元)", "支付方式", "订单状态", "收货地址" ]) # ===== 注入脏数据 ===== # 1. 注入缺失值（约5%的行） n_rows = len(df) for _ in range(int(n_rows * 0.05)): idx = random.randint(0, n_rows - 1) col = random.choice(["用户ID", "商品名称", "支付方式", "收货地址"]) df.at[idx, col] = np.nan # 2. 注入日期格式错误 for _ in range(15): idx = random.randint(0, n_rows - 1) df.at[idx, "下单时间"] = random.choice([ "2023年02月30日 星期五", # 不存在的日期 "2023年十三月01日 星期一", # 不存在的月份 "日期格式错误", "2023/01/15", "N/A", ]) # 3. 注入异常值（负数、超大值） for _ in range(10): idx = random.randint(0, n_rows - 1) field = random.choice(["购买数量", "单价(元)", "订单金额(元)"]) if field == "购买数量": df.at[idx, field] = random.choice([-1, -5, -10, 150, 200, 500]) else: df.at[idx, field] = random.choice([-99.99, -1000, -5000]) # 4. 注入重复订单 dup_indices = random.sample(range(n_rows), 20) for idx in dup_indices: dup_row = df.iloc[idx].copy() df = pd.concat([df, pd.DataFrame([dup_row])], ignore_index=True) # 5. 打乱数据顺序 df = df.sample(frac=1, random_state=42).reset_index(drop=True) filepath = os.path.join(OUTPUT_DIR, "order_data.csv") df.to_csv(filepath, index=False, encoding="utf-8-sig") print(f"[✓] 模拟题一数据已生成: {filepath} ({len(df)} 条记录)") return df # ============================================================ # 模拟题二：学生成绩数据 # ============================================================ def generate_student_score_data(): """生成学生成绩模拟数据（含脏数据）""" colleges = { "信息工程学院": { "软件技术": ["软工2101班", "软工2102班", "大数据2101班"], "计算机网络技术": ["网络2101班", "网络2102班"], }, "机电工程学院": { "机电一体化": ["机电2101班", "机电2102班"], "工业机器人": ["机器人2101班"], }, "经济管理学院": { "电子商务": ["电商2101班", "电商2102班"], "会计": ["会计2101班"], }, "建筑工程学院": { "建筑工程技术": ["建工2101班"], "工程造价": ["造价2101班"], }, } subjects = ["Python程序开发", "数据结构与算法", "数据库原理", "Web前端开发", "计算机网络", "Java程序设计", "高等数学", "大学英语"] weekdays = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"] exam_dates = [ "2023/01/10", "2023/03/15", "2023/06/20", "2023/06/21", "2023/09/12", "2023/12/25", "2024/01/08", "2024/03/20" ] levels = ["优秀", "良好", "良好", "中等", "中等", "及格", "不及格"] surnames = ["张", "李", "王", "刘", "陈", "杨", "黄", "赵", "周", "吴", "徐", "孙", "胡", "朱", "高", "林", "何", "郭", "马", "罗"] given_names = ["伟", "芳", "娜", "秀英", "敏", "静", "丽", "强", "磊", "军", "洋", "勇", "艳", "杰", "娟", "涛", "明", "超", "秀兰", "霞"] records = [] student_id = 20210101001 for college, majors in colleges.items(): for major, classes in majors.items(): for cls in classes: n_students = random.randint(30, 50) for _ in range(n_students): name = random.choice(surnames) + random.choice(given_names) for _ in range(random.randint(2, 5)): subject = random.choice(subjects) date = random.choice(exam_dates) weekday = weekdays[random.randint(0, 6)] exam_date = f"{date} {weekday}" usual = random.randint(40, 100) midterm = random.randint(30, 100) final = random.randint(25, 100) total = round(usual * 0.2 + midterm * 0.3 + final * 0.5, 1) if total >= 90: level = "优秀" elif total >= 80: level = "良好" elif total >= 70: level = "中等" elif total >= 60: level = "及格" else: level = "不及格" records.append([ str(student_id), name, college, major, cls, exam_date, subject, usual, midterm, final, total, level ]) student_id += 1 df = pd.DataFrame(records, columns=[ "学号", "姓名", "学院", "专业", "班级", "考试日期", "科目名称", "平时成绩", "期中成绩", "期末成绩", "总评成绩", "等级" ]) # ===== 注入脏数据 ===== n_rows = len(df) # 1. 缺失值 for _ in range(int(n_rows * 0.04)): idx = random.randint(0, n_rows - 1) col = random.choice(["姓名", "专业", "平时成绩", "期中成绩", "期末成绩"]) df.at[idx, col] = np.nan # 2. 日期格式错误 for _ in range(12): idx = random.randint(0, n_rows - 1) df.at[idx, "考试日期"] = random.choice([ "2023/02/30 周四", # 不存在的日期 "2023/13/01 周一", # 不存在的月份 "日期错误", "2023-06-20", "未知", ]) # 3. 异常值 for _ in range(15): idx = random.randint(0, n_rows - 1) field = random.choice(["平时成绩", "期中成绩", "期末成绩", "总评成绩"]) if field == "总评成绩": df.at[idx, field] = random.choice([-10, -5.5, -20]) else: df.at[idx, field] = random.choice([-5, -10, 105, 120, 150, -20]) # 4. 重复记录 dup_indices = random.sample(range(n_rows), 25) for idx in dup_indices: dup_row = df.iloc[idx].copy() df = pd.concat([df, pd.DataFrame([dup_row])], ignore_index=True) # 5. 打乱顺序 df = df.sample(frac=1, random_state=42).reset_index(drop=True) filepath = os.path.join(OUTPUT_DIR, "student_score.csv") df.to_csv(filepath, index=False, encoding="utf-8-sig") print(f"[✓] 模拟题二数据已生成: {filepath} ({len(df)} 条记录)") return df # ============================================================ # 练习题三：多文件销售数据 # ============================================================ def generate_multi_file_sales_data(): """生成6个月的超市销售数据文件""" months_info = [ ("01", "一月"), ("02", "二月"), ("03", "三月"), ("04", "四月"), ("05", "五月"), ("06", "六月"), ] stores = [ ("S001", "长沙岳麓店"), ("S002", "长沙天心店"), ("S003", "株洲中心店"), ("S004", "湘潭雨湖店"), ("S005", "衡阳蒸湘店"), ] product_categories = { "食品饮料": ["可口可乐330ml", "百事可乐500ml", "农夫山泉550ml", "蒙牛纯牛奶250ml", "康师傅方便面", "三只松鼠坚果", "良品铺子零食", "旺旺雪饼"], "日用百货": ["维达抽纸", "蓝月亮洗衣液", "舒肤佳香皂", "佳洁士牙膏", "清风卷纸", "立白洗洁精"], "生鲜水果": ["红富士苹果", "海南香蕉", "广西砂糖橘", "山东大葱", "本地白菜", "新鲜草莓"], "酒水茶饮": ["青岛啤酒500ml", "长城干红", "西湖龙井", "金典有机奶"], } records_all = {} for month_num, month_name in months_info: records = [] days_in_month = [31, 28, 31, 30, 31, 30][int(month_num) - 1] for day in range(1, days_in_month + 1): if random.random() < 0.1: # 10%的天数没有数据 continue date_str = f"2023-{month_num}-{day:02d}" n_transactions = random.randint(5, 20) for _ in range(n_transactions): store_id, store_name = random.choice(stores) cat = random.choice(list(product_categories.keys())) product = random.choice(product_categories[cat]) qty = random.randint(1, 30) price = round(random.uniform(2, 200), 2) amount = round(qty * price, 2) records.append([date_str, store_id, store_name, cat, product, qty, amount, month_name]) df = pd.DataFrame(records, columns=[ "日期", "门店编号", "门店名称", "商品类别", "商品名称", "销售数量", "销售金额", "月份" ]) # 注入脏数据 n = len(df) if n > 0: # 缺失值 for _ in range(max(1, int(n * 0.03))): idx = random.randint(0, n - 1) col = random.choice(["商品名称", "销售数量", "销售金额"]) df.at[idx, col] = np.nan # 异常值 for _ in range(max(1, int(n * 0.02))): idx = random.randint(0, n - 1) if random.random() < 0.5: df.at[idx, "销售数量"] = random.choice([-5, -10, -1]) else: df.at[idx, "销售金额"] = random.choice([-50.5, -100, -20]) # 重复行 if n > 5: dup_indices = random.sample(range(n), min(5, n)) dup_rows = df.iloc[dup_indices] df = pd.concat([df, dup_rows], ignore_index=True) filepath = os.path.join(OUTPUT_DIR, f"sales_2023_{month_num}.csv") df.to_csv(filepath, index=False, encoding="utf-8-sig") records_all[month_num] = len(df) print(f"[✓] 练习题三数据已生成: {filepath} ({len(df)} 条记录)") return records_all # ============================================================ # 练习题四：中文数字数据 # ============================================================ def chinese_num(n): """将阿拉伯数字转为中文数字（用于生成数据）""" if n == 0: return "零" units = ["", "十", "百", "千", "万", "十", "百", "千", "亿"] digits = ["零", "一", "二", "三", "四", "五", "六", "七", "八", "九"] result = "" s = str(n) length = len(s) for i, ch in enumerate(s): d = int(ch) pos = length - i - 1 if d == 0: if result and result[-1] != "零": result += "零" else: result += digits[d] + units[pos] # 清理多余的零 result = result.replace("零零", "零") if result.endswith("零"): result = result[:-1] # 处理"一十"开头的简写 if result.startswith("一十"): result = result[1:] # "一十二" -> "十二" return result def generate_chinese_number_data(): """生成含中文数字的数据""" projects = [ "城市道路改造工程", "智慧校园建设", "图书馆数字化升级", "校园网络改造", "实训基地建设", "体育馆翻新工程", "学生宿舍楼建设", "教学楼节能改造", "校园安防系统升级", "实验室设备采购", "绿化景观工程", "供水管网改造", ] records = [] for i, project in enumerate(projects): # 生成各种范围的金额（万元） contract = random.choice([50, 80, 100, 120, 150, 200, 250, 300, 350, 500, 800, 1000, 1500, 2000]) actual = random.randint(int(contract * 0.6), int(contract * 1.05)) days = random.choice([30, 45, 60, 90, 120, 150, 180, 200, 240, 365]) people = random.choice([8, 10, 12, 15, 20, 25, 30, 35, 40, 45, 50, 60, 80, 100]) contract_cn = chinese_num(contract) + "万元" actual_cn = chinese_num(actual) + "万元" days_cn = chinese_num(days) + "天" people_cn = chinese_num(people) + "人" records.append([ f"R{i+1:03d}", project, contract_cn, actual_cn, days_cn, people_cn ]) df = pd.DataFrame(records, columns=[ "编号", "项目名称", "合同金额(中文)", "实际支出(中文)", "工期(中文)", "参与人数(中文)" ]) # 注入少量缺失值 n = len(df) for _ in range(2): idx = random.randint(0, n - 1) col = random.choice(["合同金额(中文)", "工期(中文)"]) df.at[idx, col] = np.nan filepath = os.path.join(OUTPUT_DIR, "chinese_number_data.csv") df.to_csv(filepath, index=False, encoding="utf-8-sig") print(f"[✓] 练习题四数据已生成: {filepath} ({len(df)} 条记录)") return df # ============================================================ # 主函数 # ============================================================ if __name__ == "__main__": print("=" * 60) print("Python程序开发竞赛 - 数据清洗模块模拟数据生成") print("=" * 60) print() generate_order_data() print() generate_student_score_data() print() generate_multi_file_sales_data() print() generate_chinese_number_data() print() print("=" * 60) print("所有模拟数据生成完毕！") print(f"数据保存目录: {OUTPUT_DIR}") print("=" * 60)