gen_sample_data.py 2.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263
  1. # -*- coding: utf-8 -*-
  2. """生成示例销售数据集 sample_sales.csv(100 条记录,<1MB)"""
  3. import csv
  4. import random
  5. from datetime import date, timedelta
  6. random.seed(42)
  7. REGIONS = ["华东", "华北", "华南", "西南", "东北"]
  8. PRODUCTS = {
  9. "智能手机": ("数码电子", 3999),
  10. "笔记本电脑": ("数码电子", 5999),
  11. "无线耳机": ("数码电子", 499),
  12. "电饭煲": ("家用电器", 399),
  13. "空气净化器": ("家用电器", 1899),
  14. "电动牙刷": ("个护健康", 299),
  15. "按摩仪": ("个护健康", 899),
  16. }
  17. CHANNELS = ["线上商城", "线下门店", "直播带货"]
  18. rows = []
  19. start = date(2025, 1, 1)
  20. for i in range(97):
  21. product, (category, price) = random.choice(list(PRODUCTS.items()))
  22. quantity = random.randint(1, 20)
  23. # 区域系数:华东/华南销售额偏高
  24. region = random.choices(REGIONS, weights=[30, 22, 25, 13, 10])[0]
  25. region_factor = {"华东": 1.2, "华北": 1.0, "华南": 1.15, "西南": 0.8, "东北": 0.7}[region]
  26. sales = round(price * quantity * region_factor * random.uniform(0.85, 1.15), 2)
  27. d = start + timedelta(days=random.randint(0, 180))
  28. rows.append({
  29. "order_id": f"ORD2025{i:04d}",
  30. "order_date": d.isoformat(),
  31. "region": region,
  32. "channel": random.choice(CHANNELS),
  33. "product": product,
  34. "category": category,
  35. "unit_price": float(price),
  36. "quantity": quantity,
  37. "sales_amount": sales,
  38. "customer_satisfaction": round(random.uniform(3.0, 5.0), 1) if random.random() > 0.08 else "",
  39. })
  40. # 人为制造数据质量问题:缺失值 + 重复行
  41. rows[5]["sales_amount"] = "" # 缺失销售额
  42. rows[20]["region"] = "" # 缺失区域
  43. rows[40]["customer_satisfaction"] = ""
  44. rows[66]["sales_amount"] = ""
  45. rows.append(dict(rows[10])) # 3 条重复行
  46. rows.append(dict(rows[33]))
  47. rows.append(dict(rows[71]))
  48. random.shuffle(rows)
  49. fieldnames = ["order_id", "order_date", "region", "channel", "product", "category",
  50. "unit_price", "quantity", "sales_amount", "customer_satisfaction"]
  51. out = "data/sample_sales.csv"
  52. with open(out, "w", newline="", encoding="utf-8-sig") as f:
  53. writer = csv.DictWriter(f, fieldnames=fieldnames)
  54. writer.writeheader()
  55. writer.writerows(rows)
  56. print(f"已生成 {out},共 {len(rows)} 条记录")