4. 模拟数据生成：使用 Python 生成供应链订单、库存、物流等模拟时序数据

在供应链控制塔的异常检测链路中，真实业务数据往往难以获取、标注成本高且涉及隐私。因此，使用 Python 生成高质量的模拟时序数据是测试验证的基础。本节将介绍如何构建一个模拟数据生成器，覆盖订单、库存、物流三大核心域，并植入可控的异常模式，为后续的检测算法提供标准化的测试输入。

4.1 模拟数据架构与生成策略

模拟数据生成器采用分层设计，确保数据在时间维度上具有连续性和业务逻辑一致性。生成策略遵循以下原则：

• 时序连续性：以分钟/小时为粒度，生成连续时间戳序列，模拟真实业务流。
• 业务关联性：订单生成后触发库存扣减与物流发货，三个域的数据通过订单 ID 关联。
• 异常注入：在正常数据基础上，按预设比例和时间窗口注入异常模式，用于验证检测算法的召回率。

下表列出了三个数据域的核心字段及其模拟策略：

4.2 订单数据生成实现

订单数据是供应链的起点。以下代码片段展示了如何使用 numpy 和 pandas 生成带有时间戳的订单序列，并植入“订单量突增”异常。

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

def generate_order_data(start_date, end_date, base_rate=10, anomaly_rate=0.05):
    """
    生成订单时序数据
    :param start_date: 起始日期
    :param end_date: 结束日期
    :param base_rate: 基础订单到达率（每小时）
    :param anomaly_rate: 异常注入比例
    """
    date_range = pd.date_range(start=start_date, end=end_date, freq='H')
    records = []
    
    for ts in date_range:
        # 正常订单量：泊松分布
        order_count = np.random.poisson(base_rate)
        
        # 异常注入：订单量突增（3-5倍）
        if np.random.random() < anomaly_rate:
            order_count = int(order_count * np.random.uniform(3, 5))
        
        for _ in range(order_count):
            order_id = f"ORD_{ts.strftime('%Y%m%d%H%M%S')}_{np.random.randint(1000,9999)}"
            sku_id = f"SKU_{np.random.choice(['A','B','C','D'])}{np.random.randint(100,999)}"
            quantity = max(1, int(np.random.normal(10, 3)))
            amount = round(quantity * np.random.uniform(20, 100), 2)
            
            records.append({
                'order_id': order_id,
                'timestamp': ts + timedelta(minutes=np.random.randint(0, 59)),
                'sku_id': sku_id,
                'quantity': quantity,
                'amount': amount
            })
    
    return pd.DataFrame(records)

# 生成7天订单数据
order_df = generate_order_data('2025-01-01', '2025-01-07')
print(f"生成订单记录数: {len(order_df)}")
print(order_df.head())

4.3 库存数据生成与动态更新

库存数据需要与订单消耗联动。我们采用“期初库存 + 补货 - 订单消耗”的模型，并加入库存异常（如库存骤降为负值）。

def generate_inventory_data(order_df, sku_list, initial_stock=500):
    """
    基于订单数据生成库存时序
    :param order_df: 订单DataFrame
    :param sku_list: SKU列表
    :param initial_stock: 每个SKU的初始库存
    """
    inventory_records = []
    # 按SKU和时间排序
    order_df_sorted = order_df.sort_values(['sku_id', 'timestamp'])
    
    current_stock = {sku: initial_stock for sku in sku_list}
    
    for sku in sku_list:
        sku_orders = order_df_sorted[order_df_sorted['sku_id'] == sku]
        stock = initial_stock
        
        for _, row in sku_orders.iterrows():
            # 正常消耗
            stock -= row['quantity']
            
            # 异常注入：库存骤降（额外扣减）
            if np.random.random() < 0.02:
                stock -= int(np.random.uniform(100, 200))
            
            # 补货逻辑：低于安全库存时补货
            if stock < 100:
                stock += 300
            
            inventory_records.append({
                'sku_id': sku,
                'warehouse_id': 'WH_MAIN',
                'timestamp': row['timestamp'],
                'on_hand_qty': max(stock, -50),  # 允许负库存异常
                'in_transit_qty': np.random.randint(0, 50)
            })
    
    return pd.DataFrame(inventory_records)

sku_list = [f"SKU_{cat}{num}" for cat in ['A','B'] for num in range(100, 105)]
inv_df = generate_inventory_data(order_df, sku_list)
print(f"生成库存记录数: {len(inv_df)}")

4.4 物流数据生成与状态机

物流数据采用有限状态机模型，每个订单经历“已揽收 → 运输中 → 已签收”三个状态，并加入状态停滞和轨迹偏移异常。

def generate_logistics_data(order_df):
    """
    基于订单生成物流轨迹数据
    """
    logistics_records = []
    status_flow = ['已揽收', '运输中', '已签收']
    
    for _, order in order_df.iterrows():
        shipment_id = f"SHP_{order['order_id']}"
        current_time = order['timestamp']
        
        for i, status in enumerate(status_flow):
            # 正常时间间隔：1-3小时
            time_delta = timedelta(hours=np.random.randint(1, 4))
            
            # 异常注入：状态停滞（时间间隔异常长）
            if status == '运输中' and np.random.random() < 0.03:
                time_delta = timedelta(hours=np.random.randint(24, 48))
            
            current_time += time_delta
            
            # 模拟位置（经纬度）
            lat = 31.2304 + np.random.uniform(-0.1, 0.1)
            lng = 121.4737 + np.random.uniform(-0.1, 0.1)
            
            logistics_records.append({
                'shipment_id': shipment_id,
                'order_id': order['order_id'],
                'status': status,
                'location': f"{lat:.4f},{lng:.4f}",
                'timestamp': current_time
            })
    
    return pd.DataFrame(logistics_records)

log_df = generate_logistics_data(order_df.sample(100))  # 取100个订单生成物流
print(f"生成物流记录数: {len(log_df)}")

4.5 数据质量验证与异常标注

生成模拟数据后，需要对其进行质量验证，并记录异常标注信息，以便后续评估检测算法的性能。

异常标注采用“标签列”方式，在数据集中增加 is_anomaly 和 anomaly_type 字段，记录每条记录是否异常以及异常类型。这样在测试阶段可以直接计算混淆矩阵。

# 异常标注示例
def label_anomalies(df, anomaly_conditions):
    """
    根据条件标注异常
    :param df: 数据DataFrame
    :param anomaly_conditions: 异常条件字典 {列名: (操作符, 阈值)}
    """
    df['is_anomaly'] = 0
    df['anomaly_type'] = ''
    
    for col, (op, val) in anomaly_conditions.items():
        if op == 'gt':
            mask = df[col] > val
        elif op == 'lt':
            mask = df[col] < val
        elif op == 'eq':
            mask = df[col] == val
        
        df.loc[mask, 'is_anomaly'] = 1
        df.loc[mask, 'anomaly_type'] = f'{col}_{op}_{val}'
    
    return df

# 对库存数据标注：库存小于0为异常
inv_df_labeled = label_anomalies(inv_df, {'on_hand_qty': ('lt', 0)})
print(f"异常记录数: {inv_df_labeled['is_anomaly'].sum()}")

4.6 模拟数据生成最佳实践

在实际测试验证项目中，模拟数据生成需要遵循以下最佳实践：

1. 参数化配置：将基础到达率、异常比例、时间范围等参数抽取为配置文件（YAML/JSON），便于批量实验。
2. 随机种子固定：使用 np.random.seed(42) 固定随机种子，确保每次生成的数据可复现。
3. 数据分片存储：按天或按小时分片存储数据，模拟真实数据管道中的批处理场景。
4. 异常模式多样化：至少覆盖 5 种以上异常模式（突增、突降、缺失、重复、延迟），避免单一模式导致过拟合。
5. 与检测算法解耦：生成器只负责输出数据和标签，不依赖任何检测算法实现，保证测试的公平性。