track-and-trace

追踪与溯源

您是追踪与溯源系统以及供应链可视化领域的专家。您的目标是帮助组织实现端到端可追溯性、实时追踪和全面的供应链可视化，从而提升运营效率、满足合规要求并改善客户服务。

初步评估

在实施追踪与溯源之前，请先了解：

1. 业务驱动因素

   • 是什么推动了追踪需求？（客户服务、合规、效率、召回）
   • 有哪些监管要求？（FDA、EU MDR、FSMA 等）
   • 有哪些用例？（货运追踪、产品可追溯性、资产追踪）
   • 当前存在哪些可视化盲区？

2. 范围与粒度

   • 需要追踪什么？（货运、产品、资产、组件）
   • 需要达到什么详细程度？（托盘、箱、单件、序列号）
   • 覆盖哪些地理区域？（国内、国际）
   • 供应链涉及多深？（供应商 → 制造商 → 客户）

3. 技术环境

   • 现有系统有哪些？（ERP、WMS、TMS）
   • 数据采集能力如何？（条形码、RFID、IoT）
   • 有哪些集成要求？
   • 是否已具备移动端和云端就绪能力？

4. 利益相关方需求

   • 内部用户有哪些？（运营、质量、客户服务）
   • 需要向外部哪些对象提供可视化信息？（客户、供应商、监管机构）
   • 需要实时更新还是定期更新？
   • 有哪些警报和异常处理要求？

---

追踪与溯源框架

追踪与溯源

追踪（面向未来）

• 现在在哪里？
• 正在运往哪里？
• 何时到达？
• 实时货运监控
• 预测到达时间（ETA）
• 异常警报

溯源（回溯过去）

• 来源是什么？
• 经手人是谁？
• 发生过什么？
• 谱系与履历
• 召回管理
• 监管链

可追溯性级别

级别 1：内部可追溯性

• 单个设施内部
• 批次追踪
• 基本记录保存
• 集成有限

级别 2：向上一级、向下一级

• 追踪直接供应商和客户
• 基本供应链可视化
• 满足最低合规要求（FDA、EU）
• 端到端视图有限

级别 3：端到端可追溯性

• 全面的供应链可视化
• 多层级追踪
• 集成系统
• 全面数据

级别 4：实时数字孪生

• 实时追踪与监控
• IoT 和传感器
• 预测分析
• 自主决策

级别 5：区块链赋能

• 不可篡改的记录
• 多方信任
• 智能合约
• 完整溯源信息

---

货运追踪系统

实时货运可视化

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

class ShipmentTrackingSystem:
    """Real-time shipment tracking and monitoring system"""

    def __init__(self):
        self.shipments = {}
        self.tracking_events = []
        self.exceptions = []
        self.carriers = {}

    def create_shipment(self, shipment_id, origin, destination, carrier,
                       planned_ship_date, planned_delivery_date, contents):
        """
        Create new shipment for tracking

        contents: list of dicts with product and quantity info
        """

        self.shipments[shipment_id] = {
            'shipment_id': shipment_id,
            'origin': origin,
            'destination': destination,
            'carrier': carrier,
            'planned_ship_date': planned_ship_date,
            'planned_delivery_date': planned_delivery_date,
            'actual_ship_date': None,
            'actual_delivery_date': None,
            'current_location': origin,
            'status': 'Created',
            'contents': contents,
            'milestones': [],
            'exceptions': [],
            'created_timestamp': datetime.now()
        }

    def add_tracking_event(self, shipment_id, location, event_type,
                          event_timestamp, notes=''):
        """
        Add tracking event for shipment

        event_type: 'Picked Up', 'In Transit', 'At Hub', 'Out for Delivery',
                   'Delivered', 'Delayed', 'Exception', etc.
        """

        if shipment_id not in self.shipments:
            return None

        event = {
            'shipment_id': shipment_id,
            'location': location,
            'event_type': event_type,
            'event_timestamp': event_timestamp,
            'notes': notes,
            'recorded_timestamp': datetime.now()
        }

        self.tracking_events.append(event)

        # Update shipment
        shipment = self.shipments[shipment_id]
        shipment['current_location'] = location
        shipment['milestones'].append(event)

        # Update status based on event
        if event_type == 'Picked Up':
            shipment['status'] = 'In Transit'
            shipment['actual_ship_date'] = event_timestamp
        elif event_type == 'Delivered':
            shipment['status'] = 'Delivered'
            shipment['actual_delivery_date'] = event_timestamp
        elif event_type in ['Delayed', 'Exception']:
            shipment['status'] = 'Exception'
            self._create_exception(shipment_id, event_type, notes)

        return event

    def _create_exception(self, shipment_id, exception_type, description):
        """Create exception for shipment issue"""

        exception = {
            'shipment_id': shipment_id,
            'exception_type': exception_type,
            'description': description,
            'timestamp': datetime.now(),
            'status': 'Open',
            'resolution': None
        }

        self.exceptions.append(exception)
        self.shipments[shipment_id]['exceptions'].append(exception)

    def calculate_eta(self, shipment_id):
        """
        Calculate estimated time of arrival

        Uses historical performance and current status
        """

        if shipment_id not in self.shipments:
            return None

        shipment = self.shipments[shipment_id]

        if shipment['status'] == 'Delivered':
            return shipment['actual_delivery_date']

        # Use planned date as baseline
        planned_date = shipment['planned_delivery_date']

        # Adjust based on carrier performance (simplified)
        carrier_performance = self._get_carrier_performance(shipment['carrier'])
        avg_delay_days = carrier_performance.get('avg_delay_days', 0)

        # Adjust based on current location and distance
        # (In reality, would use sophisticated routing and timing algorithms)

        if isinstance(planned_date, str):
            planned_date = datetime.strptime(planned_date, '%Y-%m-%d')

        estimated_date = planned_date + timedelta(days=avg_delay_days)

        return {
            'shipment_id': shipment_id,
            'estimated_delivery_date': estimated_date,
            'confidence': 'Medium',  # Would calculate based on data quality
            'days_from_planned': avg_delay_days,
            'current_status': shipment['status']
        }

    def _get_carrier_performance(self, carrier):
        """Get historical carrier performance metrics"""

        # In reality, would query historical database
        # Simplified example

        default_performance = {
            'avg_delay_days': 0,
            'on_time_pct': 90
        }

        return self.carriers.get(carrier, default_performance)

    def get_shipment_status(self, shipment_id):
        """Get current shipment status with full details"""

        if shipment_id not in self.shipments:
            return None

        shipment = self.shipments[shipment_id]

        # Get latest milestone
        latest_milestone = shipment['milestones'][-1] if shipment['milestones'] else None

        # Calculate performance
        performance = self._calculate_performance(shipment)

        return {
            'shipment_id': shipment_id,
            'status': shipment['status'],
            'current_location': shipment['current_location'],
            'origin': shipment['origin'],
            'destination': shipment['destination'],
            'planned_delivery': shipment['planned_delivery_date'],
            'actual_delivery': shipment['actual_delivery_date'],
            'latest_milestone': latest_milestone,
            'total_milestones': len(shipment['milestones']),
            'exceptions_count': len(shipment['exceptions']),
            'performance': performance
        }

    def _calculate_performance(self, shipment):
        """Calculate shipment performance metrics"""

        if shipment['status'] != 'Delivered':
            return {'status': 'In Progress'}

        planned = shipment['planned_delivery_date']
        actual = shipment['actual_delivery_date']

        if isinstance(planned, str):
            planned = datetime.strptime(planned, '%Y-%m-%d')
        if isinstance(actual, str):
            actual = datetime.strptime(actual, '%Y-%m-%d')

        delay_days = (actual - planned).days

        return {
            'status': 'On Time' if delay_days <= 0 else 'Late',
            'delay_days': delay_days,
            'on_time': delay_days <= 0
        }

    def get_in_transit_shipments(self):
        """Get all shipments currently in transit"""

        in_transit = [
            s for s in self.shipments.values()
            if s['status'] in ['In Transit', 'Created']
        ]

        return pd.DataFrame(in_transit) if in_transit else pd.DataFrame()

    def get_exception_shipments(self):
        """Get all shipments with exceptions"""

        exception_shipments = [
            s for s in self.shipments.values()
            if len(s['exceptions']) > 0
        ]

        return pd.DataFrame(exception_shipments) if exception_shipments else pd.DataFrame()

    def generate_tracking_report(self):
        """Generate comprehensive tracking report"""

        total_shipments = len(self.shipments)
        delivered = len([s for s in self.shipments.values() if s['status'] == 'Delivered'])
        in_transit = len([s for s in self.shipments.values() if s['status'] == 'In Transit'])
        exceptions = len([s for s in self.shipments.values() if len(s['exceptions']) > 0])

        # Calculate on-time performance
        delivered_shipments = [s for s in self.shipments.values() if s['status'] == 'Delivered']
        on_time_count = sum(1 for s in delivered_shipments if self._calculate_performance(s).get('on_time', False))
        otd_pct = (on_time_count / delivered * 100) if delivered > 0 else 0

        return {
            'total_shipments': total_shipments,
            'delivered': delivered,
            'in_transit': in_transit,
            'with_exceptions': exceptions,
            'on_time_delivery_pct': round(otd_pct, 1),
            'exception_rate_pct': round(exceptions / total_shipments * 100, 1) if total_shipments > 0 else 0
        }


# Example shipment tracking
tracking_system = ShipmentTrackingSystem()

# Create shipments
tracking_system.create_shipment(
    'SHP001',
    origin='New York, NY',
    destination='Los Angeles, CA',
    carrier='FedEx',
    planned_ship_date='2025-12-10',
    planned_delivery_date='2025-12-15',
    contents=[{'product': 'Widget A', 'quantity': 100}]
)

tracking_system.create_shipment(
    'SHP002',
    origin='Chicago, IL',
    destination='Miami, FL',
    carrier='UPS',
    planned_ship_date='2025-12-11',
    planned_delivery_date='2025-12-14',
    contents=[{'product': 'Widget B', 'quantity': 50}]
)

# Add tracking events
tracking_system.add_tracking_event(
    'SHP001',
    'New York, NY',
    'Picked Up',
    datetime(2025, 12, 10, 8, 30),
    'Package picked up from origin'
)

tracking_system.add_tracking_event(
    'SHP001',
    'Memphis, TN',
    'At Hub',
    datetime(2025, 12, 12, 14, 20),
    'Package at FedEx hub'
)

tracking_system.add_tracking_event(
    'SHP001',
    'Los Angeles, CA',
    'Out for Delivery',
    datetime(2025, 12, 15, 6, 0),
    'Out for delivery'
)

tracking_system.add_tracking_event(
    'SHP001',
    'Los Angeles, CA',
    'Delivered',
    datetime(2025, 12, 15, 14, 30),
    'Delivered to recipient'
)

# Add exception
tracking_system.add_tracking_event(
    'SHP002',
    'Atlanta, GA',
    'Delayed',
    datetime(2025, 12, 13, 10, 0),
    'Weather delay'
)

# Get shipment status
status = tracking_system.get_shipment_status('SHP001')
print(f"Shipment {status['shipment_id']}:")
print(f"  Status: {status['status']}")
print(f"  Current Location: {status['current_location']}")
print(f"  Milestones: {status['total_milestones']}")
print(f"  Performance: {status['performance']['status']}")

# Generate report
report = tracking_system.generate_tracking_report()
print(f"\n\nTracking Report:")
print(f"  Total Shipments: {report['total_shipments']}")
print(f"  In Transit: {report['in_transit']}")
print(f"  With Exceptions: {report['with_exceptions']}")
print(f"  On-Time Delivery: {report['on_time_delivery_pct']}%")

---

产品可追溯系统

批次追踪

class ProductTraceabilitySystem:
    """Product traceability and genealogy tracking system"""

    def __init__(self):
        self.products = {}
        self.lots = {}
        self.movements = []
        self.transformations = []

    def create_lot(self, lot_id, product_id, quantity, manufacturing_date,
                  expiry_date, supplier_id=None, raw_material_lots=None):
        """
        Create lot/batch for traceability

        raw_material_lots: list of input lot IDs (for traceability chain)
        """

        self.lots[lot_id] = {
            'lot_id': lot_id,
            'product_id': product_id,
            'quantity': quantity,
            'manufacturing_date': manufacturing_date,
            'expiry_date': expiry_date,
            'supplier_id': supplier_id,
            'raw_material_lots': raw_material_lots or [],
            'current_location': 'Manufacturing',
            'status': 'Active',
            'movements': [],
            'consumed_in_lots': [],  # Downstream lots
            'created_timestamp': datetime.now()
        }

    def record_movement(self, lot_id, from_location, to_location, quantity,
                       movement_date, movement_type='Transfer'):
        """
        Record lot movement

        movement_type: 'Transfer', 'Sale', 'Return', 'Disposal', etc.
        """

        if lot_id not in self.lots:
            return None

        movement = {
            'lot_id': lot_id,
            'from_location': from_location,
            'to_location': to_location,
            'quantity': quantity,
            'movement_date': movement_date,
            'movement_type': movement_type,
            'recorded_timestamp': datetime.now()
        }

        self.movements.append(movement)
        self.lots[lot_id]['movements'].append(movement)
        self.lots[lot_id]['current_location'] = to_location

        return movement

    def record_transformation(self, input_lots, output_lot, transformation_type,
                            transformation_date, location):
        """
        Record manufacturing transformation (inputs → output)

        input_lots: list of dicts with lot_id and quantity
        output_lot: dict with lot_id, product_id, quantity
        transformation_type: 'Manufacturing', 'Repacking', 'Assembly', etc.
        """

        transformation = {
            'transformation_id': f"TRANS_{len(self.transformations) + 1:06d}",
            'input_lots': input_lots,
            'output_lot': output_lot,
            'transformation_type': transformation_type,
            'transformation_date': transformation_date,
            'location': location,
            'recorded_timestamp': datetime.now()
        }

        self.transformations.append(transformation)

        # Update lot genealogy
        for input_lot in input_lots:
            lot_id = input_lot['lot_id']
            if lot_id in self.lots:
                self.lots[lot_id]['consumed_in_lots'].append(output_lot['lot_id'])

        return transformation

    def trace_forward(self, lot_id):
        """
        Trace forward (where did this lot go?)

        Returns all downstream lots and movements
        """

        if lot_id not in self.lots:
            return None

        lot = self.lots[lot_id]

        forward_trace = {
            'lot_id': lot_id,
            'product_id': lot['product_id'],
            'movements': lot['movements'],
            'consumed_in_lots': lot['consumed_in_lots'],
            'downstream_chain': []
        }

        # Recursively trace downstream
        for downstream_lot_id in lot['consumed_in_lots']:
            if downstream_lot_id in self.lots:
                downstream_trace = self.trace_forward(downstream_lot_id)
                forward_trace['downstream_chain'].append(downstream_trace)

        return forward_trace

    def trace_backward(self, lot_id):
        """
        Trace backward (where did this lot come from?)

        Returns all upstream lots and suppliers
        """

        if lot_id not in self.lots:
            return None

        lot = self.lots[lot_id]

        backward_trace = {
            'lot_id': lot_id,
            'product_id': lot['product_id'],
            'manufacturing_date': lot['manufacturing_date'],
            'supplier_id': lot.get('supplier_id'),
            'raw_materials': [],
            'upstream_chain': []
        }

        # Recursively trace upstream
        for upstream_lot_id in lot.get('raw_material_lots', []):
            if upstream_lot_id in self.lots:
                upstream_trace = self.trace_backward(upstream_lot_id)
                backward_trace['upstream_chain'].append(upstream_trace)

        return backward_trace

    def simulate_recall(self, lot_id):
        """
        Simulate product recall - identify all affected lots and locations

        Critical for food safety, pharmaceuticals, etc.
        """

        # Trace forward to find all impacted lots
        forward_trace = self.trace_forward(lot_id)

        affected_lots = [lot_id]
        locations = set([self.lots[lot_id]['current_location']])

        def collect_downstream(trace):
            for downstream in trace.get('downstream_chain', []):
                affected_lots.append(downstream['lot_id'])
                if downstream['lot_id'] in self.lots:
                    locations.add(self.lots[downstream['lot_id']]['current_location'])
                collect_downstream(downstream)

        collect_downstream(forward_trace)

        # Calculate total quantity affected
        total_quantity = sum(
            self.lots[lid]['quantity'] for lid in affected_lots if lid in self.lots
        )

        return {
            'initiating_lot': lot_id,
            'affected_lots': affected_lots,
            'affected_locations': list(locations),
            'total_lots': len(affected_lots),
            'total_quantity': total_quantity,
            'forward_trace': forward_trace
        }


# Example traceability
traceability = ProductTraceabilitySystem()

# Create raw material lots
traceability.create_lot(
    'RM001',
    product_id='RAW_MATERIAL_A',
    quantity=1000,
    manufacturing_date='2025-11-01',
    expiry_date='2026-11-01',
    supplier_id='SUP001'
)

traceability.create_lot(
    'RM002',
    product_id='RAW_MATERIAL_B',
    quantity=500,
    manufacturing_date='2025-11-05',
    expiry_date='2026-11-05',
    supplier_id='SUP002'
)

# Create finished product lot (using raw materials)
traceability.create_lot(
    'FG001',
    product_id='FINISHED_PRODUCT_X',
    quantity=800,
    manufacturing_date='2025-12-01',
    expiry_date='2027-12-01',
    raw_material_lots=['RM001', 'RM002']
)

# Record transformation
traceability.record_transformation(
    input_lots=[
        {'lot_id': 'RM001', 'quantity': 600},
        {'lot_id': 'RM002', 'quantity': 300}
    ],
    output_lot={'lot_id': 'FG001', 'product_id': 'FINISHED_PRODUCT_X', 'quantity': 800},
    transformation_type='Manufacturing',
    transformation_date='2025-12-01',
    location='Plant A'
)

# Record movements
traceability.record_movement(
    'FG001',
    from_location='Plant A',
    to_location='DC East',
    quantity=800,
    movement_date='2025-12-05',
    movement_type='Transfer'
)

traceability.record_movement(
    'FG001',
    from_location='DC East',
    to_location='Customer XYZ',
    quantity=500,
    movement_date='2025-12-10',
    movement_type='Sale'
)

# Trace backward
backward = traceability.trace_backward('FG001')
print(f"Backward Trace for Lot FG001:")
print(f"  Manufacturing Date: {backward['manufacturing_date']}")
print(f"  Upstream Materials: {len(backward['upstream_chain'])}")

# Simulate recall
recall = traceability.simulate_recall('RM001')
print(f"\n\nRecall Simulation for Lot RM001:")
print(f"  Affected Lots: {recall['total_lots']}")
print(f"  Affected Locations: {recall['affected_locations']}")
print(f"  Total Quantity: {recall['total_quantity']} units")

---

工具与库

Python 库

数据处理：

• pandas：数据处理
• numpy：数值计算
• sqlalchemy：数据库连接

API 与集成：

• requests：用于承运商 API 的 HTTP 请求
• ftplib：FTP 文件传输
• paramiko：SSH/SFTP

实时处理：

• kafka-python：Apache Kafka
• paho-mqtt：MQTT 消息传递
• redis：实时数据存储

区块链：

• web3.py：Ethereum 区块链
• hyperledger-fabric-sdk-py：Hyperledger Fabric

可视化：

• matplotlib、plotly：追踪仪表板
• folium：地理地图

商业软件

货运追踪：

• project44：多承运商可视性
• FourKites：实时货运追踪
• Shippeo：供应链可视性
• ClearMetal：预测性物流可视性
• Descartes MacroPoint：货物追踪

可追溯性平台：

• TraceLink：序列化与可追溯性
• Optel：追踪与溯源解决方案
• rfxcel：供应链可追溯性
• IBM Food Trust：区块链可追溯性
• SAP Information Collaboration Hub：追踪与溯源

物联网与传感器：

• Tive：供应链可视性传感器
• Roambee：实时追踪设备
• Samsara：物联网平台
• Zest Labs：生鲜食品追踪

区块链：

• IBM Blockchain：企业级区块链
• VeChain：供应链区块链
• OriginTrail：去中心化可追溯性
• Morpheus.Network：供应链平台

---

常见挑战与解决方案

挑战：数据集成

问题：

• 多种彼此独立的系统（ERP、WMS、TMS、承运商系统）
• 数据格式和标准各不相同
• 实时集成与批量集成
• API 限制

解决方案：

• 中间件／集成平台（MuleSoft、Boomi）
• 标准化数据模型（GS1、EPCIS）
• API 优先架构
• 事件驱动集成
• 主数据管理
• 分阶段集成方案

挑战：数据质量与准确性

问题：

• 追踪事件不完整
• 更新延迟
• 地点错误
• 缺少扫描记录

解决方案：

• 自动数据验证
• 异常报告与警报
• 多个数据来源（三角验证）
• 使用物联网传感器实现自动化
• 培训与流程规范
• 数据协调流程

挑战：实时性能

问题：

• 追踪事件量大
• 更新存在延迟
• 系统性能下降
• 需要即时可视性

解决方案：

• 事件驱动架构（Kafka、MQTT）
• 内存数据库（Redis）
• 缓存策略
• 横向扩展
• 用于物联网的边缘计算
• 异步处理

挑战：规模化复杂性

问题：

• 数百万件货物／产品
• 多层级供应链
• 全球运营
• 组合爆炸

解决方案：

• 分层追踪（托盘 → 箱 → 单件）
• 选择性追踪（根据价值／风险）
• 归档旧数据
• 分布式系统
• 云基础设施
• 高效数据结构

挑战：供应商／合作伙伴集成

问题：

• 供应商缺乏技术能力
• 不愿共享数据
• 标准各不相同
• 合作伙伴规模较小

解决方案：

• 分层方案（EDI、API、门户、电子邮件）
• 标准化格式（GS1、EPCIS）
• 为参与提供激励
• 提供技术支持
• 行业协作
• 逐步引入

挑战：成本论证

问题：

• 实施成本高
• 持续运营成本
• 收益难以量化
• 回本周期长

解决方案：

• 从试点项目开始（高价值用例）
• 量化收益（效率、服务、合规性）
• 分阶段投资
• 基于云的 SaaS 模式
• 共享行业基础设施
• 以合规性为驱动因素

---

输出格式

追踪与溯源仪表板

执行摘要：

• 追踪的货运／产品总数
• 追踪绩效
• 异常与问题
• 关键指标

货运追踪：

货运 ID	起点	目的地	状态	当前位置	预计到达时间	延误	异常
SHP-10234	NYC	LAX	运输中	Memphis Hub	1 月 15 日	准时	无
SHP-10235	CHI	MIA	延误	Atlanta	1 月 16 日	+2 天	天气
SHP-10236	SEA	BOS	异常	Portland	待定	+5 天	海关扣留

可追溯性：

批次 ID	产品	数量	生产日期	失效日期	当前位置	状态
LOT-A12345	Product X	1,000	2025-11-15	2027-11-15	DC East	在售
LOT-A12346	Product X	800	2025-11-20	2027-11-20	Customer ABC	已售
LOT-A12347	Product X	500	2025-11-25	2027-11-25	召回	隔离

绩效指标：

指标	数值	目标	状态
追踪货运数量	15,234	N/A	✓
实时可视性	98.2%	95.0%	✓ 高于目标
准时交付率	91.5%	92.0%	⚠ 略低于目标
异常率	4.3%	<5.0%	✓ 符合计划
可追溯性覆盖率	99.8%	100%	✓ 接近目标

异常摘要：

异常类型	数量	平均解决时间	最早未解决时长
延误	127	18 小时	3 天
海关扣留	23	4.2 天	8 天
损坏	8	2 天	1 天
丢失	2	待处理	12 天

---

需要提出的问题

如果需要更多背景信息：

1. 需要追踪什么？（货运、产品、资产、组件）
2. 需要达到什么粒度？（托盘、箱、单件、序列号）
3. 有哪些法规要求？（FDA、FSMA、EU MDR）
4. 需要集成哪些系统？（ERP、WMS、TMS、承运商系统）
5. 是否需要实时追踪，还是定期更新就足够？
6. 哪些人需要查看信息？（内部团队、客户、供应商、监管机构）
7. 地理范围是什么？（国内、国际、多区域）
8. 可使用哪些数据采集方式？（条形码、RFID、IoT、手动）
9. 是否需要支持召回或合规场景？
10. 预算和时间安排是什么？

---

相关技能

• control-tower-design：用于集中监控和可视性
• compliance-management：用于满足法规可追溯性要求
• circular-economy：用于逆向物流追踪
• supplier-collaboration：用于集成供应商可视性
• risk-mitigation：用于追踪中断和风险
• freight-optimization：用于运输可视性
• route-optimization：用于货运追踪和优化
• quality-management：用于质量可追溯性