数据可视化模拟题2-生成代码

# -*- coding: utf-8 -*-
"""
电商运营数据分析 - 模拟数据生成脚本
生成电商销售数据和用户数据，并注入脏数据用于数据分析练习
"""

import numpy as np
import pandas as pd
from datetime import datetime, timedelta
import os

# 设置随机种子确保可复现
np.random.seed(42)

# 输出目录
OUTPUT_DIR = "/workspace/mock_exam/第二套_电商运营数据分析"

# ============================================================
# 第一部分：生成电商销售数据 (ecommerce_sales.csv)
# ============================================================

# 商品类别与对应商品名称、价格范围
category_config = {
    "电子产品": {
        "products": [
            "智能手机", "笔记本电脑", "平板电脑", "蓝牙耳机", "智能手表",
            "移动电源", "无线充电器", "机械键盘", "显示器", "摄像头",
            "路由器", "音响", "鼠标", "U盘", "数据线"
        ],
        "price_range": (99, 8999),
    },
    "服装鞋帽": {
        "products": [
            "男士T恤", "女士连衣裙", "运动鞋", "牛仔裤", "羽绒服",
            "衬衫", "卫衣", "短裤", "帽子", "围巾",
            "皮带", "袜子套装", "西装外套", "运动裤", "帆布鞋"
        ],
        "price_range": (39, 1999),
    },
    "食品饮料": {
        "products": [
            "坚果礼盒", "进口牛奶", "有机茶叶", "咖啡豆", "巧克力",
            "薯片", "方便面", "蜂蜜", "果汁", "饼干",
            "牛肉干", "水果罐头", "矿泉水", "气泡水", "燕麦片"
        ],
        "price_range": (9, 399),
    },
    "家居用品": {
        "products": [
            "四件套", "枕头", "毛巾套装", "收纳箱", "台灯",
            "垃圾桶", "拖把", "洗衣液", "纸巾", "保温杯",
            "拖鞋", "雨伞", "挂钩套装", "香薰蜡烛", "洗手液"
        ],
        "price_range": (15, 899),
    },
    "图书文具": {
        "products": [
            "畅销小说", "编程教材", "儿童绘本", "英语词典", "历史读物",
            "钢笔套装", "笔记本", "彩色铅笔", "书架", "文具盒",
            "计算器", "文件夹", "便签纸", "橡皮擦套装", "书包"
        ],
        "price_range": (8, 199),
    },
}

# 中国省份列表
provinces = [
    "北京", "上海", "广东", "浙江", "江苏", "四川", "湖北", "湖南",
    "山东", "河南", "福建", "安徽", "河北", "辽宁", "陕西", "重庆",
    "天津", "云南", "广西", "黑龙江", "吉林", "内蒙古", "山西", "贵州",
    "甘肃", "海南", "宁夏", "青海", "西藏", "新疆", "江西"
]

# 省份与主要城市映射
province_city_map = {
    "北京": ["北京"],
    "上海": ["上海"],
    "广东": ["广州", "深圳", "东莞", "佛山", "珠海"],
    "浙江": ["杭州", "宁波", "温州", "嘉兴", "绍兴"],
    "江苏": ["南京", "苏州", "无锡", "常州", "南通"],
    "四川": ["成都", "绵阳", "德阳", "宜宾", "泸州"],
    "湖北": ["武汉", "宜昌", "襄阳", "荆州", "黄石"],
    "湖南": ["长沙", "株洲", "湘潭", "衡阳", "岳阳"],
    "山东": ["济南", "青岛", "烟台", "潍坊", "临沂"],
    "河南": ["郑州", "洛阳", "开封", "南阳", "新乡"],
    "福建": ["福州", "厦门", "泉州", "漳州", "莆田"],
    "安徽": ["合肥", "芜湖", "蚌埠", "马鞍山", "安庆"],
    "河北": ["石家庄", "唐山", "保定", "邯郸", "秦皇岛"],
    "辽宁": ["沈阳", "大连", "鞍山", "抚顺", "锦州"],
    "陕西": ["西安", "咸阳", "宝鸡", "渭南", "汉中"],
    "重庆": ["重庆"],
    "天津": ["天津"],
    "云南": ["昆明", "大理", "丽江", "曲靖", "玉溪"],
    "广西": ["南宁", "桂林", "柳州", "北海", "梧州"],
    "黑龙江": ["哈尔滨", "齐齐哈尔", "大庆", "牡丹江", "佳木斯"],
    "吉林": ["长春", "吉林", "四平", "延边", "通化"],
    "内蒙古": ["呼和浩特", "包头", "鄂尔多斯", "赤峰", "通辽"],
    "山西": ["太原", "大同", "运城", "临汾", "长治"],
    "贵州": ["贵阳", "遵义", "六盘水", "安顺", "毕节"],
    "甘肃": ["兰州", "天水", "白银", "庆阳", "酒泉"],
    "海南": ["海口", "三亚", "儋州", "琼海", "文昌"],
    "宁夏": ["银川", "吴忠", "固原", "石嘴山", "中卫"],
    "青海": ["西宁", "海东", "海西", "海南州", "海北州"],
    "西藏": ["拉萨", "日喀则", "林芝", "山南", "昌都"],
    "新疆": ["乌鲁木齐", "伊犁", "喀什", "阿克苏", "哈密"],
    "江西": ["南昌", "九江", "赣州", "景德镇", "上饶"],
}

# 支付方式
payment_methods = ["支付宝", "微信", "银行卡"]

# 生成日期范围：2022-01-01 至 2024-12-31
start_date = datetime(2022, 1, 1)
end_date = datetime(2024, 12, 31)
total_days = (end_date - start_date).days + 1

# 生成日期序列（带季节性权重）
# 基础日期
all_dates = [start_date + timedelta(days=d) for d in range(total_days)]

# 季节性权重：双十一(11月)、618(6月)、春节前后(1-2月)、年底(12月)销量增加
date_weights = []
for d in all_dates:
    month = d.month
    day = d.day
    weight = 1.0
    # 双十一期间（11月1日-11月15日）权重增加
    if month == 11 and 1 <= day <= 15:
        weight = 4.0 if day <= 11 else 2.5
    # 618期间（6月1日-6月20日）权重增加
    elif month == 6 and 1 <= day <= 20:
        weight = 3.0 if day <= 18 else 2.0
    # 春节前后（1月15日-2月15日）权重增加
    elif (month == 1 and day >= 15) or (month == 2 and day <= 15):
        weight = 2.5
    # 年底促销（12月10日-12月31日）
    elif month == 12 and day >= 10:
        weight = 2.0
    # 周末权重略高
    elif d.weekday() >= 5:
        weight = 1.3
    date_weights.append(weight)

date_weights = np.array(date_weights)
date_probs = date_weights / date_weights.sum()

# 生成约18000条销售数据
n_sales = 18000

# 按权重随机采样日期
sampled_dates_idx = np.random.choice(len(all_dates), size=n_sales, p=date_probs)
sampled_dates = [all_dates[i] for i in sampled_dates_idx]

# 随机选择商品类别
categories = list(category_config.keys())
category_choices = np.random.choice(categories, size=n_sales)

# 生成订单数据
order_ids = []
order_dates = []
cat_list = []
product_names = []
prices = []
quantities = []
total_amounts = []
province_list = []
city_list = []
payment_list = []

for i in range(n_sales):
    # 订单编号：ORD{年份}{序号}
    year = sampled_dates[i].year
    order_id = f"ORD{year}{i + 1:05d}"
    order_ids.append(order_id)

    # 下单日期
    order_dates.append(sampled_dates[i])

    # 商品类别
    cat = category_choices[i]
    cat_list.append(cat)

    # 商品名称（从对应品类中随机选取）
    config = category_config[cat]
    product = np.random.choice(config["products"])
    product_names.append(product)

    # 商品单价（根据品类价格范围生成）
    p_min, p_max = config["price_range"]
    # 使用对数均匀分布使低价商品更常见
    log_price = np.random.uniform(np.log(p_min), np.log(p_max))
    price = round(np.exp(log_price), 2)
    prices.append(price)

    # 购买数量（1-5）
    qty = np.random.randint(1, 6)
    quantities.append(qty)

    # 订单金额（默认正确）
    total = round(price * qty, 2)
    total_amounts.append(total)

    # 收货省份
    province = np.random.choice(provinces)
    province_list.append(province)

    # 收货城市（根据省份选取）
    cities = province_city_map[province]
    city = np.random.choice(cities)
    city_list.append(city)

    # 支付方式
    payment = np.random.choice(payment_methods)
    payment_list.append(payment)

# 构建DataFrame
sales_df = pd.DataFrame({
    "order_id": order_ids,
    "order_date": order_dates,
    "category": cat_list,
    "product_name": product_names,
    "price": prices,
    "quantity": quantities,
    "total_amount": total_amounts,
    "province": province_list,
    "city": city_list,
    "payment_method": payment_list,
})

# ============================================================
# 注入销售数据脏数据
# ============================================================

# 1. order_date: 5%格式异常（用/分隔）
# 先将所有日期转为字符串格式
sales_df["order_date"] = sales_df["order_date"].apply(lambda x: x.strftime("%Y-%m-%d"))
date_error_mask = np.random.random(n_sales) < 0.05
for idx in sales_df.index[date_error_mask]:
    # 将 - 替换为 /
    sales_df.at[idx, "order_date"] = sales_df.at[idx, "order_date"].replace("-", "/")

# 2. category: 2%为空值或别名
cat_error_mask = np.random.random(n_sales) < 0.02
cat_error_indices = sales_df.index[cat_error_mask]
# 别名映射
alias_map = {
    "电子产品": "电子",
    "服装鞋帽": "服装",
    "食品饮料": "食品",
    "家居用品": "家居",
    "图书文具": "图书",
}
for idx in cat_error_indices:
    original_cat = sales_df.at[idx, "category"]
    if np.random.random() < 0.5:
        # 使用别名
        sales_df.at[idx, "category"] = alias_map.get(original_cat, original_cat)
    else:
        # 设为空值
        sales_df.at[idx, "category"] = np.nan

# 3. total_amount: 3%与price*quantity不一致
amount_error_mask = np.random.random(n_sales) < 0.03
for idx in sales_df.index[amount_error_mask]:
    correct_total = sales_df.at[idx, "price"] * sales_df.at[idx, "quantity"]
    # 随机偏移10%-50%
    deviation = np.random.uniform(0.1, 0.5) * correct_total
    if np.random.random() < 0.5:
        sales_df.at[idx, "total_amount"] = round(correct_total + deviation, 2)
    else:
        sales_df.at[idx, "total_amount"] = round(correct_total - deviation, 2)

# 4. province: 2%缺失值
province_error_mask = np.random.random(n_sales) < 0.02
sales_df.loc[sales_df.index[province_error_mask], "province"] = np.nan
# 省份缺失时对应城市也设为空
sales_df.loc[sales_df["province"].isna(), "city"] = np.nan

# 保存销售数据
sales_path = os.path.join(OUTPUT_DIR, "ecommerce_sales.csv")
sales_df.to_csv(sales_path, index=False, encoding="utf-8-sig")
print(f"销售数据已保存: {sales_path}")
print(f"  总行数: {len(sales_df)}")

# ============================================================
# 第二部分：生成用户数据 (ecommerce_users.csv)
# ============================================================

n_users = 4000

# 注册日期（2023年内）
reg_start = datetime(2023, 1, 1)
reg_end = datetime(2023, 12, 31)
reg_days = (reg_end - reg_start).days + 1
reg_dates = [reg_start + timedelta(days=d) for d in range(reg_days)]

# 城市等级
city_levels = ["一线", "二线", "三线", "四线及以下"]

# 会员等级
member_levels = ["普通", "银卡", "金卡", "钻石"]

# 偏好品类
preference_categories = categories + ["无偏好"]

# 生成用户数据
user_ids = []
register_dates = []
ages = []
genders = []
city_level_list = []
member_level_list = []
total_orders_list = []
total_spending_list = []
avg_monthly_spending_list = []
category_pref_list = []
last_purchase_dates = []

for i in range(n_users):
    # 用户ID：U{年份}{序号}
    user_id = f"U2023{i + 1:05d}"
    user_ids.append(user_id)

    # 注册日期
    reg_date = reg_start + timedelta(days=np.random.randint(0, reg_days))
    register_dates.append(reg_date)

    # 年龄（18-70）
    age = np.random.randint(18, 71)
    ages.append(age)

    # 性别
    gender = np.random.choice(["男", "女"])
    genders.append(gender)

    # 城市等级
    city_level = np.random.choice(city_levels, p=[0.15, 0.30, 0.30, 0.25])
    city_level_list.append(city_level)

    # 会员等级（先随机，后面根据消费金额调整）
    member_level = np.random.choice(member_levels, p=[0.40, 0.30, 0.20, 0.10])
    member_level_list.append(member_level)

    # 年度订单总数
    total_orders = np.random.randint(1, 101)
    total_orders_list.append(total_orders)

    # 年度消费总额（与会员等级关联）
    if member_level == "普通":
        total_spending = round(np.random.uniform(100, 5000), 2)
    elif member_level == "银卡":
        total_spending = round(np.random.uniform(5000, 20000), 2)
    elif member_level == "金卡":
        total_spending = round(np.random.uniform(20000, 50000), 2)
    else:  # 钻石
        total_spending = round(np.random.uniform(50000, 200000), 2)
    total_spending_list.append(total_spending)

    # 月均消费额（基于年度消费总额计算，先正确计算）
    avg_monthly = round(total_spending / 12, 2)
    avg_monthly_spending_list.append(avg_monthly)

    # 偏好品类
    pref = np.random.choice(preference_categories)
    category_pref_list.append(pref)

    # 最近购买日期（在注册日期之后，2024年内）
    last_purchase = datetime(2024, 1, 1) + timedelta(days=np.random.randint(0, 365))
    last_purchase_dates.append(last_purchase)

# 构建用户DataFrame
users_df = pd.DataFrame({
    "user_id": user_ids,
    "register_date": register_dates,
    "age": ages,
    "gender": genders,
    "city_level": city_level_list,
    "member_level": member_level_list,
    "total_orders": total_orders_list,
    "total_spending": total_spending_list,
    "avg_monthly_spending": avg_monthly_spending_list,
    "category_preference": category_pref_list,
    "last_purchase_date": last_purchase_dates,
})

# ============================================================
# 注入用户数据脏数据
# ============================================================

# 1. age: 3%异常值（负数或>120）
age_error_mask = np.random.random(n_users) < 0.03
for idx in users_df.index[age_error_mask]:
    if np.random.random() < 0.5:
        users_df.at[idx, "age"] = np.random.randint(-20, 0)  # 负数
    else:
        users_df.at[idx, "age"] = np.random.randint(121, 200)  # 超大值

# 2. member_level: 3%空值
member_error_mask = np.random.random(n_users) < 0.03
users_df.loc[users_df.index[member_error_mask], "member_level"] = np.nan

# 3. total_spending: 4%缺失值
spending_error_mask = np.random.random(n_users) < 0.04
users_df.loc[users_df.index[spending_error_mask], "total_spending"] = np.nan

# 4. avg_monthly_spending: 部分计算不准确（约8%）
avg_error_mask = np.random.random(n_users) < 0.08
for idx in users_df.index[avg_error_mask]:
    if pd.notna(users_df.at[idx, "total_spending"]):
        # 故意使用错误的除数（如除以10或13）或加减偏差
        correct_avg = users_df.at[idx, "total_spending"] / 12
        error_type = np.random.choice(["wrong_divisor", "offset"])
        if error_type == "wrong_divisor":
            wrong_divisor = np.random.choice([10, 11, 13, 6])
            users_df.at[idx, "avg_monthly_spending"] = round(
                users_df.at[idx, "total_spending"] / wrong_divisor, 2
            )
        else:
            offset = np.random.uniform(-500, 500)
            users_df.at[idx, "avg_monthly_spending"] = round(correct_avg + offset, 2)

# 保存用户数据
users_path = os.path.join(OUTPUT_DIR, "ecommerce_users.csv")
users_df.to_csv(users_path, index=False, encoding="utf-8-sig")
print(f"用户数据已保存: {users_path}")
print(f"  总行数: {len(users_df)}")

# ============================================================
# 第三部分：验证脏数据比例
# ============================================================
print("\n" + "=" * 60)
print("脏数据验证报告")
print("=" * 60)

# 重新读取CSV验证
sales_check = pd.read_csv(sales_path)
users_check = pd.read_csv(users_path)

print(f"\n【ecommerce_sales.csv】总行数: {len(sales_check)}")

# 验证order_date格式异常
date_format_errors = sales_check["order_date"].astype(str).str.contains("/", na=False)
print(f"  order_date 格式异常(/分隔): {date_format_errors.sum()} 条 ({date_format_errors.sum()/len(sales_check)*100:.1f}%)")

# 验证category异常
cat_errors = sales_check["category"].isna() | ~sales_check["category"].isin(categories)
print(f"  category 空值或别名: {cat_errors.sum()} 条 ({cat_errors.sum()/len(sales_check)*100:.1f}%)")

# 验证total_amount不一致
correct_totals = (sales_check["price"] * sales_check["quantity"]).round(2)
amount_errors = (sales_check["total_amount"] - correct_totals).abs() > 0.01
print(f"  total_amount 与 price*quantity 不一致: {amount_errors.sum()} 条 ({amount_errors.sum()/len(sales_check)*100:.1f}%)")

# 验证province缺失
province_errors = sales_check["province"].isna()
print(f"  province 缺失值: {province_errors.sum()} 条 ({province_errors.sum()/len(sales_check)*100:.1f}%)")

print(f"\n【ecommerce_users.csv】总行数: {len(users_check)}")

# 验证age异常
age_errors = (users_check["age"] < 18) | (users_check["age"] > 70)
print(f"  age 异常值(负数或>120): {age_errors.sum()} 条 ({age_errors.sum()/len(users_check)*100:.1f}%)")

# 验证member_level空值
member_errors = users_check["member_level"].isna()
print(f"  member_level 空值: {member_errors.sum()} 条 ({member_errors.sum()/len(users_check)*100:.1f}%)")

# 验证total_spending缺失
spending_errors = users_check["total_spending"].isna()
print(f"  total_spending 缺失值: {spending_errors.sum()} 条 ({spending_errors.sum()/len(users_check)*100:.1f}%)")

# 验证avg_monthly_spending计算不准确
valid_mask = users_check["total_spending"].notna()
correct_avg = (users_check.loc[valid_mask, "total_spending"] / 12).round(2)
avg_errors = (users_check.loc[valid_mask, "avg_monthly_spending"] - correct_avg).abs() > 0.01
print(f"  avg_monthly_spending 计算不准确: {avg_errors.sum()} 条 ({avg_errors.sum()/valid_mask.sum()*100:.1f}%)")

print("\n数据生成完成！")