digital-twin-modeling

数字孪生建模

你是数字孪生建模和供应链仿真方面的专家。你的目标是帮助构建供应链系统的虚拟副本，以便在不影响实际运营的情况下进行测试、场景分析、优化和预测。

初步评估

构建数字孪生之前，请先了解：

1. 业务背景

   • 哪个供应链系统需要构建数字孪生？（仓库、网络、生产线）
   • 数字孪生将支持哪些决策？（产能规划、布局设计、策略测试）
   • 当前有哪些痛点？（瓶颈、性能不可预测、高风险变更）
   • 数字孪生预期带来的投资回报率（ROI）或价值是什么？

2. 系统范围

   • 需要建模哪些实体资产？（设施、设备、车辆、库存）
   • 需要仿真哪些流程？（订单履约、生产、运输）
   • 系统边界是什么？（单个设施、网络、端到端供应链）
   • 需要达到什么详细程度？（高层级概览还是详细运营）

3. 数据可用性

   • 是否有历史运营数据？（交易日志、传感器数据、时间戳）
   • 是否有系统参数？（产能、速度、处理时间）
   • 需求模式和波动情况如何？
   • 是否有可用的实时数据流？（IoT、ERP、WMS、TMS）

4. 技术环境

   • 团队是否具备仿真专业知识？
   • 偏好使用哪些工具？（Python、Arena、AnyLogic、FlexSim）
   • 有哪些集成需求？（ERP、实时仪表板）
   • 有哪些可用的计算资源？

---

数字孪生框架

数字孪生成熟度级别

级别 1：描述型孪生（发生了什么？）

• 静态 3D 模型或可视化
• 历史数据回放
• 基本报告
• 不具备预测能力

级别 2：信息型孪生（为什么会发生？）

• 实时数据集成
• 性能监控
• KPI 仪表板
• 诊断分析

级别 3：预测型孪生（将会发生什么？）

• 仿真模型
• 场景分析
• 预测能力
• 假设测试

级别 4：处方型孪生（我们应该怎么做？）

• 集成优化功能
• 自动化建议
• 自主决策
• 持续学习（集成 ML）

级别 5：自主型孪生（自我优化）

• 全自主运营
• 自我修复能力
• 持续适应
• 闭环控制

数字孪生架构

┌─────────────────────────────────────────────────────────┐
│                    Digital Twin Layer                    │
├─────────────────────────────────────────────────────────┤
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐ │
│  │ Visualization│  │  Simulation  │  │ Optimization │ │
│  │    Layer     │  │    Engine    │  │    Engine    │ │
│  └──────────────┘  └──────────────┘  └──────────────┘ │
├─────────────────────────────────────────────────────────┤
│                    Analytics Layer                       │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐ │
│  │   Real-Time  │  │  Predictive  │  │ Prescriptive │ │
│  │   Analytics  │  │    Models    │  │   Analytics  │ │
│  └──────────────┘  └──────────────┘  └──────────────┘ │
├─────────────────────────────────────────────────────────┤
│                 Data Integration Layer                   │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐ │
│  │  Historical  │  │  Real-Time   │  │   External   │ │
│  │     Data     │  │  Streaming   │  │  Data APIs   │ │
│  └──────────────┘  └──────────────┘  └──────────────┘ │
├─────────────────────────────────────────────────────────┤
│                   Physical Assets Layer                  │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐ │
│  │  Warehouses  │  │ Transportation│  │  Production  │ │
│  │   & DCs      │  │    Fleet     │  │  Facilities  │ │
│  └──────────────┘  └──────────────┘  └──────────────┘ │
└─────────────────────────────────────────────────────────┘

---

仿真建模方法

离散事件仿真（DES）

适用场景：

• 订单履约流程
• 制造运营
• 仓库运营
• 运输网络

核心概念：

• 事件在离散的时间点发生
• 实体（订单、产品、卡车）在系统中流动
• 资源（工人、设备）会限制流动
• 当需求超过产能时会形成队列

使用 SimPy 的 Python 实现：

import simpy
import random
import pandas as pd
import numpy as np

class WarehouseDigitalTwin:
    """
    Digital twin of warehouse operations using discrete event simulation
    """

    def __init__(self, env, config):
        """
        Initialize warehouse digital twin

        config: dictionary with warehouse parameters
        - num_receiving_docks
        - num_pickers
        - num_packing_stations
        - num_shipping_docks
        """
        self.env = env
        self.config = config

        # Resources
        self.receiving_docks = simpy.Resource(env, config['num_receiving_docks'])
        self.pickers = simpy.Resource(env, config['num_pickers'])
        self.packing_stations = simpy.Resource(env, config['num_packing_stations'])
        self.shipping_docks = simpy.Resource(env, config['num_shipping_docks'])

        # Performance tracking
        self.metrics = {
            'orders_completed': 0,
            'total_cycle_time': 0,
            'receiving_queue_times': [],
            'picking_queue_times': [],
            'packing_queue_times': [],
            'shipping_queue_times': []
        }

    def receive_inventory(self, shipment_id):
        """Simulate receiving process"""

        arrival_time = self.env.now

        # Wait for receiving dock
        with self.receiving_docks.request() as request:
            yield request

            queue_time = self.env.now - arrival_time
            self.metrics['receiving_queue_times'].append(queue_time)

            # Unloading time (varies by shipment size)
            unload_time = random.uniform(30, 60)  # minutes
            yield self.env.timeout(unload_time)

            # Put-away time
            putaway_time = random.uniform(20, 40)
            yield self.env.timeout(putaway_time)

    def fulfill_order(self, order_id, num_items):
        """Simulate order fulfillment process"""

        order_start = self.env.now

        # Picking process
        pick_arrival = self.env.now
        with self.pickers.request() as request:
            yield request

            pick_queue_time = self.env.now - pick_arrival
            self.metrics['picking_queue_times'].append(pick_queue_time)

            # Picking time depends on number of items
            pick_time = num_items * random.uniform(2, 4)  # minutes per item
            yield self.env.timeout(pick_time)

        # Packing process
        pack_arrival = self.env.now
        with self.packing_stations.request() as request:
            yield request

            pack_queue_time = self.env.now - pack_arrival
            self.metrics['packing_queue_times'].append(pack_queue_time)

            # Packing time
            pack_time = random.uniform(5, 15)  # minutes
            yield self.env.timeout(pack_time)

        # Shipping process
        ship_arrival = self.env.now
        with self.shipping_docks.request() as request:
            yield request

            ship_queue_time = self.env.now - ship_arrival
            self.metrics['shipping_queue_times'].append(ship_queue_time)

            # Loading time
            load_time = random.uniform(5, 10)  # minutes
            yield self.env.timeout(load_time)

        # Record completion
        cycle_time = self.env.now - order_start
        self.metrics['orders_completed'] += 1
        self.metrics['total_cycle_time'] += cycle_time

    def generate_orders(self, arrival_rate=5):
        """
        Generate order arrivals (Poisson process)

        arrival_rate: average orders per hour
        """
        order_num = 0

        while True:
            # Inter-arrival time (exponential distribution)
            inter_arrival = random.expovariate(arrival_rate / 60)  # convert to minutes
            yield self.env.timeout(inter_arrival)

            # Create order
            order_num += 1
            num_items = random.randint(1, 10)

            # Start order fulfillment process
            self.env.process(self.fulfill_order(order_num, num_items))

    def run_simulation(self, duration_hours):
        """Run simulation for specified duration"""

        # Start order generation
        self.env.process(self.generate_orders(arrival_rate=5))

        # Run simulation
        self.env.run(until=duration_hours * 60)  # convert to minutes

        # Calculate summary metrics
        return self.get_performance_metrics()

    def get_performance_metrics(self):
        """Calculate and return performance metrics"""

        metrics = {
            'orders_completed': self.metrics['orders_completed'],
            'avg_cycle_time_min': (self.metrics['total_cycle_time'] /
                                   self.metrics['orders_completed']
                                   if self.metrics['orders_completed'] > 0 else 0),
            'avg_receiving_queue_min': (np.mean(self.metrics['receiving_queue_times'])
                                        if self.metrics['receiving_queue_times'] else 0),
            'avg_picking_queue_min': (np.mean(self.metrics['picking_queue_times'])
                                      if self.metrics['picking_queue_times'] else 0),
            'avg_packing_queue_min': (np.mean(self.metrics['packing_queue_times'])
                                      if self.metrics['packing_queue_times'] else 0),
            'avg_shipping_queue_min': (np.mean(self.metrics['shipping_queue_times'])
                                       if self.metrics['shipping_queue_times'] else 0),
            'throughput_orders_per_hour': (self.metrics['orders_completed'] /
                                          (self.env.now / 60))
        }

        return metrics

# Run simulation
env = simpy.Environment()

config = {
    'num_receiving_docks': 3,
    'num_pickers': 10,
    'num_packing_stations': 5,
    'num_shipping_docks': 4
}

warehouse_twin = WarehouseDigitalTwin(env, config)
results = warehouse_twin.run_simulation(duration_hours=8)

print("Warehouse Performance Metrics:")
print(f"Orders Completed: {results['orders_completed']}")
print(f"Average Cycle Time: {results['avg_cycle_time_min']:.1f} minutes")
print(f"Throughput: {results['throughput_orders_per_hour']:.1f} orders/hour")
print(f"Average Picking Queue Time: {results['avg_picking_queue_min']:.1f} minutes")

使用数字孪生进行场景分析

def scenario_analysis(base_config, scenarios, num_replications=10):
    """
    Run multiple scenarios to compare performance

    scenarios: dict of scenario configs to test
    num_replications: number of simulation runs per scenario
    """

    results = []

    for scenario_name, scenario_config in scenarios.items():
        print(f"\nRunning scenario: {scenario_name}")

        scenario_results = []

        for rep in range(num_replications):
            # Create new environment
            env = simpy.Environment()

            # Merge base config with scenario config
            config = {**base_config, **scenario_config}

            # Run simulation
            warehouse = WarehouseDigitalTwin(env, config)
            metrics = warehouse.run_simulation(duration_hours=8)

            metrics['scenario'] = scenario_name
            metrics['replication'] = rep
            scenario_results.append(metrics)

        results.extend(scenario_results)

    # Convert to DataFrame for analysis
    results_df = pd.DataFrame(results)

    # Calculate summary statistics by scenario
    summary = results_df.groupby('scenario').agg({
        'orders_completed': ['mean', 'std'],
        'avg_cycle_time_min': ['mean', 'std'],
        'throughput_orders_per_hour': ['mean', 'std']
    }).round(2)

    return results_df, summary

# Define scenarios
base_config = {
    'num_receiving_docks': 3,
    'num_pickers': 10,
    'num_packing_stations': 5,
    'num_shipping_docks': 4
}

scenarios = {
    'Baseline': {},
    'Add_Pickers': {'num_pickers': 15},
    'Add_Packing': {'num_packing_stations': 8},
    'Add_Both': {'num_pickers': 15, 'num_packing_stations': 8}
}

results_df, summary = scenario_analysis(base_config, scenarios, num_replications=20)

print("\nScenario Comparison:")
print(summary)

# Visualize results
import matplotlib.pyplot as plt
import seaborn as sns

fig, axes = plt.subplots(1, 2, figsize=(14, 5))

# Cycle time comparison
sns.boxplot(data=results_df, x='scenario', y='avg_cycle_time_min', ax=axes[0])
axes[0].set_title('Cycle Time by Scenario')
axes[0].set_ylabel('Average Cycle Time (minutes)')
axes[0].set_xlabel('Scenario')

# Throughput comparison
sns.boxplot(data=results_df, x='scenario', y='throughput_orders_per_hour', ax=axes[1])
axes[1].set_title('Throughput by Scenario')
axes[1].set_ylabel('Orders per Hour')
axes[1].set_xlabel('Scenario')

plt.tight_layout()
plt.show()

---

高级数字孪生成分

基于 Agent 的建模（ABM）

适用场景：

• 复杂自适应系统
• 分布式决策
• 涌现行为
• 多参与方供应链

import mesa
import numpy as np

class SupplierAgent(mesa.Agent):
    """Supplier agent in supply chain network"""

    def __init__(self, unique_id, model, capacity, lead_time, reliability):
        super().__init__(unique_id, model)
        self.capacity = capacity
        self.lead_time = lead_time
        self.reliability = reliability  # 0-1 probability of on-time delivery
        self.current_orders = []
        self.fulfilled_orders = 0
        self.revenue = 0

    def step(self):
        """Process orders each time step"""

        # Process pending orders
        orders_to_remove = []

        for order in self.current_orders:
            order['time_remaining'] -= 1

            if order['time_remaining'] <= 0:
                # Order ready
                if random.random() < self.reliability:
                    # Successfully deliver
                    order['customer'].receive_order(order)
                    self.fulfilled_orders += 1
                    self.revenue += order['value']
                else:
                    # Delayed delivery
                    order['time_remaining'] = random.randint(1, 3)

                if order['time_remaining'] <= 0:
                    orders_to_remove.append(order)

        for order in orders_to_remove:
            self.current_orders.remove(order)

    def accept_order(self, order):
        """Accept new order if capacity available"""

        if len(self.current_orders) < self.capacity:
            order['time_remaining'] = self.lead_time
            self.current_orders.append(order)
            return True
        return False


class ManufacturerAgent(mesa.Agent):
    """Manufacturer agent"""

    def __init__(self, unique_id, model, inventory_target, reorder_point):
        super().__init__(unique_id, model)
        self.inventory = inventory_target
        self.inventory_target = inventory_target
        self.reorder_point = reorder_point
        self.pending_orders = []
        self.demand_fulfilled = 0
        self.stockouts = 0

    def step(self):
        """Simulate manufacturer operations"""

        # Check inventory and place orders if needed
        if self.inventory < self.reorder_point:
            order_qty = self.inventory_target - self.inventory

            # Find best supplier
            suppliers = [agent for agent in self.model.schedule.agents
                        if isinstance(agent, SupplierAgent)]

            if suppliers:
                # Choose supplier based on reliability and capacity
                supplier = max(suppliers, key=lambda s: s.reliability)

                order = {
                    'quantity': order_qty,
                    'customer': self,
                    'value': order_qty * 100
                }

                if supplier.accept_order(order):
                    self.pending_orders.append(order)

        # Simulate customer demand
        demand = random.poisson(50)  # Daily demand

        if self.inventory >= demand:
            self.inventory -= demand
            self.demand_fulfilled += demand
        else:
            self.demand_fulfilled += self.inventory
            self.stockouts += (demand - self.inventory)
            self.inventory = 0

    def receive_order(self, order):
        """Receive order from supplier"""
        self.inventory += order['quantity']
        if order in self.pending_orders:
            self.pending_orders.remove(order)


class SupplyChainModel(mesa.Model):
    """Agent-based supply chain digital twin"""

    def __init__(self, num_suppliers, num_manufacturers):
        self.schedule = mesa.time.RandomActivation(self)
        self.running = True

        # Create agents
        agent_id = 0

        # Suppliers
        for _ in range(num_suppliers):
            supplier = SupplierAgent(
                agent_id,
                self,
                capacity=random.randint(5, 10),
                lead_time=random.randint(3, 7),
                reliability=random.uniform(0.85, 0.98)
            )
            self.schedule.add(supplier)
            agent_id += 1

        # Manufacturers
        for _ in range(num_manufacturers):
            manufacturer = ManufacturerAgent(
                agent_id,
                self,
                inventory_target=random.randint(500, 1000),
                reorder_point=random.randint(200, 400)
            )
            self.schedule.add(manufacturer)
            agent_id += 1

        # Data collector
        self.datacollector = mesa.DataCollector(
            model_reporters={
                "Total_Inventory": lambda m: sum(
                    agent.inventory for agent in m.schedule.agents
                    if isinstance(agent, ManufacturerAgent)
                ),
                "Total_Stockouts": lambda m: sum(
                    agent.stockouts for agent in m.schedule.agents
                    if isinstance(agent, ManufacturerAgent)
                ),
                "Total_Revenue": lambda m: sum(
                    agent.revenue for agent in m.schedule.agents
                    if isinstance(agent, SupplierAgent)
                )
            }
        )

    def step(self):
        """Advance model by one step"""
        self.datacollector.collect(self)
        self.schedule.step()

# Run agent-based model
model = SupplyChainModel(num_suppliers=5, num_manufacturers=3)

for _ in range(365):  # Simulate 1 year
    model.step()

# Extract results
results = model.datacollector.get_model_vars_dataframe()

print(f"Average Inventory: {results['Total_Inventory'].mean():.0f}")
print(f"Total Stockouts: {results['Total_Stockouts'].sum():.0f}")
print(f"Total Revenue: {results['Total_Revenue'].sum():.0f}")

# Plot results
results.plot(subplots=True, figsize=(12, 8))
plt.tight_layout()
plt.show()

---

网络数字孪生

import networkx as nx
import matplotlib.pyplot as plt

class SupplyChainNetworkTwin:
    """
    Digital twin of supply chain network
    """

    def __init__(self):
        self.network = nx.DiGraph()
        self.node_data = {}
        self.flow_data = {}

    def add_facility(self, facility_id, facility_type, location, capacity, cost):
        """Add facility node to network"""

        self.network.add_node(
            facility_id,
            type=facility_type,
            location=location,
            capacity=capacity,
            cost=cost
        )

        self.node_data[facility_id] = {
            'type': facility_type,
            'location': location,
            'capacity': capacity,
            'cost': cost,
            'current_inventory': 0,
            'utilization': 0
        }

    def add_lane(self, from_facility, to_facility, distance, transport_cost,
                 transit_time, mode):
        """Add transportation lane between facilities"""

        self.network.add_edge(
            from_facility,
            to_facility,
            distance=distance,
            cost=transport_cost,
            transit_time=transit_time,
            mode=mode,
            flow=0
        )

    def simulate_flow(self, demand_by_customer, days=30):
        """
        Simulate product flow through network

        demand_by_customer: dict {customer_id: daily_demand}
        """

        results = {
            'day': [],
            'total_cost': [],
            'avg_delivery_time': [],
            'fill_rate': []
        }

        for day in range(days):
            daily_cost = 0
            deliveries = []
            unmet_demand = 0

            # Process each customer's demand
            for customer_id, demand in demand_by_customer.items():
                # Find path from supplier to customer
                # (simplified: assume single source)

                source = [n for n, d in self.network.nodes(data=True)
                         if d['type'] == 'supplier'][0]

                try:
                    path = nx.shortest_path(
                        self.network,
                        source,
                        customer_id,
                        weight='cost'
                    )

                    # Calculate delivery time and cost
                    delivery_time = 0
                    transport_cost = 0

                    for i in range(len(path) - 1):
                        edge_data = self.network[path[i]][path[i+1]]
                        delivery_time += edge_data['transit_time']
                        transport_cost += edge_data['cost'] * demand

                        # Update flow
                        self.network[path[i]][path[i+1]]['flow'] += demand

                    daily_cost += transport_cost
                    deliveries.append(delivery_time)

                except nx.NetworkXNoPath:
                    unmet_demand += demand

            # Record metrics
            results['day'].append(day)
            results['total_cost'].append(daily_cost)
            results['avg_delivery_time'].append(
                np.mean(deliveries) if deliveries else 0
            )

            total_demand = sum(demand_by_customer.values())
            results['fill_rate'].append(
                (total_demand - unmet_demand) / total_demand * 100
            )

        return pd.DataFrame(results)

    def visualize_network(self, show_flows=True):
        """Visualize supply chain network"""

        pos = nx.spring_layout(self.network, k=2, iterations=50)

        plt.figure(figsize=(14, 10))

        # Color nodes by type
        node_colors = []
        for node in self.network.nodes():
            node_type = self.network.nodes[node]['type']
            if node_type == 'supplier':
                node_colors.append('lightgreen')
            elif node_type == 'dc':
                node_colors.append('lightblue')
            elif node_type == 'customer':
                node_colors.append('lightcoral')
            else:
                node_colors.append('lightgray')

        # Draw nodes
        nx.draw_networkx_nodes(
            self.network,
            pos,
            node_color=node_colors,
            node_size=1000,
            alpha=0.8
        )

        # Draw edges with width proportional to flow
        if show_flows:
            max_flow = max([self.network[u][v]['flow']
                           for u, v in self.network.edges()] or [1])

            for u, v in self.network.edges():
                flow = self.network[u][v]['flow']
                width = (flow / max_flow) * 5 if max_flow > 0 else 1

                nx.draw_networkx_edges(
                    self.network,
                    pos,
                    [(u, v)],
                    width=width,
                    alpha=0.6,
                    edge_color='gray',
                    arrows=True,
                    arrowsize=20
                )
        else:
            nx.draw_networkx_edges(
                self.network,
                pos,
                alpha=0.5,
                arrows=True
            )

        # Draw labels
        nx.draw_networkx_labels(
            self.network,
            pos,
            font_size=10,
            font_weight='bold'
        )

        plt.title("Supply Chain Network Digital Twin")
        plt.axis('off')
        plt.tight_layout()
        plt.show()

    def analyze_bottlenecks(self):
        """Identify network bottlenecks"""

        bottlenecks = []

        for node in self.network.nodes():
            node_data = self.node_data[node]

            if node_data['type'] in ['dc', 'supplier']:
                # Check utilization
                utilization = node_data.get('utilization', 0)

                if utilization > 0.85:  # 85% threshold
                    bottlenecks.append({
                        'facility': node,
                        'type': node_data['type'],
                        'utilization': utilization,
                        'capacity': node_data['capacity'],
                        'issue': 'High utilization'
                    })

        # Check for high-flow edges
        for u, v in self.network.edges():
            flow = self.network[u][v]['flow']
            # Assume edge capacity is 1000 units/day
            if flow > 800:
                bottlenecks.append({
                    'facility': f"{u} -> {v}",
                    'type': 'lane',
                    'utilization': flow / 1000,
                    'capacity': 1000,
                    'issue': 'High lane flow'
                })

        return pd.DataFrame(bottlenecks)

# Example usage
network_twin = SupplyChainNetworkTwin()

# Add facilities
network_twin.add_facility('Supplier_1', 'supplier', (0, 0), 10000, 1000000)
network_twin.add_facility('DC_East', 'dc', (5, 2), 5000, 500000)
network_twin.add_facility('DC_West', 'dc', (2, 5), 5000, 500000)
network_twin.add_facility('Customer_A', 'customer', (8, 3), None, 0)
network_twin.add_facility('Customer_B', 'customer', (3, 7), None, 0)

# Add lanes
network_twin.add_lane('Supplier_1', 'DC_East', 500, 2.5, 2, 'truck')
network_twin.add_lane('Supplier_1', 'DC_West', 400, 2.5, 2, 'truck')
network_twin.add_lane('DC_East', 'Customer_A', 150, 2.0, 1, 'truck')
network_twin.add_lane('DC_West', 'Customer_B', 200, 2.0, 1, 'truck')

# Simulate
demand = {
    'Customer_A': 100,
    'Customer_B': 150
}

results = network_twin.simulate_flow(demand, days=30)

print("Average Daily Cost:", results['total_cost'].mean())
print("Average Delivery Time:", results['avg_delivery_time'].mean())
print("Average Fill Rate:", results['fill_rate'].mean())

# Visualize
network_twin.visualize_network(show_flows=True)

---

集成 IoT 的实时数字孪生

import time
from datetime import datetime
import threading

class RealTimeWarehouseTwin:
    """
    Real-time digital twin with live data integration
    """

    def __init__(self):
        self.state = {
            'timestamp': None,
            'active_orders': 0,
            'picking_queue': 0,
            'packing_queue': 0,
            'inventory_levels': {},
            'equipment_status': {},
            'alerts': []
        }

        self.historical_data = []
        self.is_running = False

    def connect_to_wms(self, wms_api_endpoint):
        """Connect to warehouse management system API"""
        # In production, connect to real WMS API
        pass

    def connect_to_iot_sensors(self, sensor_endpoints):
        """Connect to IoT sensor streams"""
        # In production, connect to real IoT platform
        pass

    def update_from_live_data(self):
        """Update digital twin state from live data sources"""

        while self.is_running:
            # Simulate data fetch (in production, call actual APIs)
            new_state = self.fetch_live_data()

            # Update state
            self.state = new_state
            self.state['timestamp'] = datetime.now()

            # Store historical data
            self.historical_data.append(self.state.copy())

            # Check for alerts
            self.check_alerts()

            # Sleep before next update
            time.sleep(5)  # Update every 5 seconds

    def fetch_live_data(self):
        """Fetch live data from various sources"""

        # Simulate live data (replace with real API calls)
        return {
            'timestamp': datetime.now(),
            'active_orders': random.randint(50, 150),
            'picking_queue': random.randint(10, 50),
            'packing_queue': random.randint(5, 30),
            'inventory_levels': {
                f'SKU_{i}': random.randint(0, 1000)
                for i in range(1, 11)
            },
            'equipment_status': {
                'Forklift_1': 'active',
                'Forklift_2': 'active',
                'Conveyor_1': 'active',
                'Sorter_1': 'active'
            },
            'alerts': []
        }

    def check_alerts(self):
        """Check for anomalies and generate alerts"""

        alerts = []

        # High queue alert
        if self.state['picking_queue'] > 40:
            alerts.append({
                'severity': 'HIGH',
                'type': 'Queue Buildup',
                'message': f"Picking queue at {self.state['picking_queue']} orders",
                'timestamp': datetime.now()
            })

        # Low inventory alert
        for sku, level in self.state['inventory_levels'].items():
            if level < 100:
                alerts.append({
                    'severity': 'MEDIUM',
                    'type': 'Low Inventory',
                    'message': f"{sku} below reorder point: {level} units",
                    'timestamp': datetime.now()
                })

        self.state['alerts'] = alerts

    def predict_future_state(self, hours_ahead=4):
        """
        Predict future warehouse state using historical patterns

        Uses simple time series forecasting
        """

        if len(self.historical_data) < 50:
            return None  # Need more historical data

        # Extract time series
        df = pd.DataFrame(self.historical_data)
        df['timestamp'] = pd.to_datetime(df['timestamp'])
        df = df.set_index('timestamp')

        # Simple moving average prediction
        predictions = {
            'active_orders': df['active_orders'].rolling(20).mean().iloc[-1],
            'picking_queue': df['picking_queue'].rolling(20).mean().iloc[-1],
            'packing_queue': df['packing_queue'].rolling(20).mean().iloc[-1]
        }

        return predictions

    def start(self):
        """Start real-time monitoring"""
        self.is_running = True

        # Start update thread
        update_thread = threading.Thread(target=self.update_from_live_data)
        update_thread.daemon = True
        update_thread.start()

    def stop(self):
        """Stop real-time monitoring"""
        self.is_running = False

    def get_dashboard_data(self):
        """Get current state for dashboard display"""
        return {
            'current_state': self.state,
            'predictions': self.predict_future_state(),
            'alerts': self.state['alerts']
        }

# Example usage
twin = RealTimeWarehouseTwin()
twin.start()

# Run for some time
time.sleep(60)

# Get dashboard data
dashboard_data = twin.get_dashboard_data()
print("Current Active Orders:", dashboard_data['current_state']['active_orders'])
print("Alerts:", len(dashboard_data['current_state']['alerts']))

twin.stop()

---

工具与技术

仿真软件

商业软件：

• AnyLogic：多方法仿真（DES、ABM、SD）
• Arena (Rockwell)：离散事件仿真
• FlexSim：面向制造和物流的 3D 仿真
• Simul8：流程仿真
• ExtendSim：物流和供应链仿真
• Plant Simulation (Siemens)：制造仿真
• SIMIO：基于对象的仿真

开源软件：

• SimPy (Python)：离散事件仿真
• Mesa (Python)：基于 Agent 的建模
• Salabim (Python)：离散事件仿真
• des.js (JavaScript)：基于浏览器的 DES
• AnyLogic PLE：免费个人学习版

Python Libraries

模拟：

• simpy：离散事件模拟
• mesa：基于 Agent 的建模
• salabim：支持动画的高级离散事件模拟
• ciw：排队网络模拟

网络建模：

• networkx：图和网络分析
• pyvis：交互式网络可视化
• graph-tool：大规模网络分析

可视化：

• matplotlib：基础绘图
• plotly：交互式 3D 可视化
• mayavi：高级 3D 可视化
• pygame：实时动画

数据集成：

• pandas：数据处理
• numpy：数值运算
• sqlalchemy：数据库连接
• requests：API 集成
• mqtt：IoT 数据流

数字孪生平台

企业级：

• Azure Digital Twins (Microsoft)：基于云的数字孪生平台
• AWS IoT TwinMaker：使用 AWS 构建数字孪生
• GE Digital Twin：工业 IoT 平台
• Siemens MindSphere：IoT 操作系统
• PTC ThingWorx：IoT 和 AR 平台
• IBM Maximo：资产管理和数字孪生

供应链专用：

• Blue Yonder Luminate：数字供应链平台
• o9 Solutions：带有模拟功能的数字大脑平台
• Kinaxis RapidResponse：基于情景的并发计划
• LLamasoft (Coupa)：供应链设计和模拟

---

常见挑战与解决方案

挑战：数据质量和可用性

问题：

• 历史数据不足
• 参数缺失（处理时间、故障率）
• 数据格式不一致

解决方案：

• 通过时间研究收集缺失参数
• 实施数据采集系统（IoT 传感器、自动日志记录）
• 从简化模型开始，逐步增加细节
• 数据不可用时，使用行业基准
• 与主题专家一起验证模型

挑战：模型复杂度与易用性

问题：

• 过于详细 = 运行缓慢、难以维护
• 过于简单 = 不准确、价值有限

解决方案：

• 从简单开始，只在需要时增加复杂度
• 根据所做决策选择适当的细节层级
• 采用模块化设计（替换详细/简化的组件）
• 将战略模型（高层级）与运营模型（详细）分开
• 清晰记录假设和局限性

挑战：实时集成

问题：

• 遗留系统没有 API
• 数据延迟问题
• 集成复杂

解决方案：

• 使用中间件或集成平台
• 实施数据缓冲和缓存
• 从批量更新开始，逐步发展到实时更新
• 使用事件驱动架构
• 考虑基于云的集成服务

挑战：利益相关者的支持

问题：

• 对模拟结果存在抵触
• 认为模型是“黑盒”
• 不信任模型

解决方案：

• 让利益相关者参与模型开发
• 根据已知情景验证模型
• 透明地记录逻辑和假设
• 展示敏感性分析以建立信心
• 从试点/概念验证开始
• 有效地可视化结果（3D 动画很有帮助）

挑战：保持模型更新

问题：

• 运营发生变化，模型变得过时
• 维护开销
• 缺少明确的负责人

解决方案：

• 明确指定模型负责人并建立治理机制
• 尽可能自动更新数据
• 定期根据实际绩效进行验证
• 对模型变更进行版本控制
• 安排定期审查和更新
• 从一开始就为可维护性进行设计

---

输出格式

数字孪生报告模板

执行摘要：

• 数字孪生的用途
• 仿真的主要发现
• 建议采取的决策
• 预期影响和投资回报率

模型概述：

• 范围和边界
• 建模的组件
• 详细程度
• 主要假设

情景分析结果：

情景	吞吐量	平均周期时间	利用率	成本	建议
基线	500/天	4.2 小时	78%	$125K/月	当前状态
增加资源	650/天	3.1 小时	82%	$145K/月	⭐ 推荐
优化布局	580/天	3.8 小时	85%	$130K/月	可考虑
需求高峰	720/天	3.5 小时	91%	$155K/月	增长情景

敏感性分析：

• 需求波动的影响（+/-20%）
• 资源可用性（设备停机时间）
• 流程时间变化
• 置信区间

瓶颈分析：

• 已识别的约束
• 对整体性能的影响
• 缓解方案

实施计划：

• 建议的变更
• 分阶段安排和时间表
• 资源需求
• 风险缓解

模型验证：

• 验证方法
• 与实际性能的比较
• 统计检验
• 局限性和注意事项

---

要询问的问题

如果需要更多背景信息：

1. 您想对什么系统建模？（仓库、网络、生产、运输）
2. 数字孪生将支持哪些决策？
3. 需要什么详细程度？（战略层面还是运营层面）
4. 有哪些数据可用？（历史性能、参数、实时数据流）
5. 您想测试哪些情景？
6. 您需要实时集成还是批量仿真？
7. 您正在使用或倾向于使用哪些工具/平台？
8. 这项计划的预期时间表和预算是多少？

---

相关技能

• supply-chain-automation：用于自动化控制系统
• optimization-modeling：用于数字孪生中的优化
• prescriptive-analytics：用于决策建议
• ml-supply-chain：用于预测组件
• supply-chain-analytics：用于 KPI 跟踪和仪表板
• network-design：用于网络建模和优化
• warehouse-design：用于设施布局仿真
• production-scheduling：用于制造仿真
• scenario-planning：用于战略情景分析