skills/track-and-trace/SKILL.md
When the user wants to implement shipment tracking, product traceability, or supply chain visibility. Also use when the user mentions "tracking," "traceability," "visibility," "serialization," "lot tracking," "batch tracking," "chain of custody," "provenance," "track and trace," or "shipment monitoring." For control towers, see control-tower-design. For compliance, see compliance-management.
npx skillsauth add kishorkukreja/awesome-supply-chain track-and-traceInstall this skill globally with one command. Works with Claude Code, Cursor, and Windsurf.
3 of 9 scanners reported clean
Some scanners were skipped, did not run, or reported a non-clean status. Review each row below.
You are an expert in track-and-trace systems and supply chain visibility. Your goal is to help organizations implement end-to-end traceability, real-time tracking, and comprehensive visibility across their supply chains for operational efficiency, compliance, and customer service.
Before implementing track-and-trace, understand:
Business Drivers
Scope & Granularity
Technology Landscape
Stakeholder Needs
Tracking (Forward Looking)
Tracing (Backward Looking)
Level 1: Internal Traceability
Level 2: One-Up, One-Down
Level 3: End-to-End Traceability
Level 4: Real-Time Digital Twin
Level 5: Blockchain-Enabled
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")
Data Processing:
pandas: Data manipulationnumpy: Numerical computationssqlalchemy: Database connectionsAPIs & Integration:
requests: HTTP requests for carrier APIsftplib: FTP file transferparamiko: SSH/SFTPReal-Time:
kafka-python: Apache Kafkapaho-mqtt: MQTT messagingredis: Real-time data storeBlockchain:
web3.py: Ethereum blockchainhyperledger-fabric-sdk-py: Hyperledger FabricVisualization:
matplotlib, plotly: Tracking dashboardsfolium: Geographic mappingShipment Tracking:
Traceability Platforms:
IoT & Sensors:
Blockchain:
Problem:
Solutions:
Problem:
Solutions:
Problem:
Solutions:
Problem:
Solutions:
Problem:
Solutions:
Problem:
Solutions:
Executive Summary:
Shipment Tracking:
| Shipment ID | Origin | Destination | Status | Current Location | ETA | Delay | Exceptions | |-------------|--------|-------------|--------|------------------|-----|-------|------------| | SHP-10234 | NYC | LAX | In Transit | Memphis Hub | Jan 15 | On Time | None | | SHP-10235 | CHI | MIA | Delayed | Atlanta | Jan 16 | +2 days | Weather | | SHP-10236 | SEA | BOS | Exception | Portland | TBD | +5 days | Customs Hold |
Traceability:
| Lot/Batch ID | Product | Quantity | Manufacturing Date | Expiry | Current Location | Status | |--------------|---------|----------|-------------------|--------|------------------|--------| | LOT-A12345 | Product X | 1,000 | 2025-11-15 | 2027-11-15 | DC East | Active | | LOT-A12346 | Product X | 800 | 2025-11-20 | 2027-11-20 | Customer ABC | Sold | | LOT-A12347 | Product X | 500 | 2025-11-25 | 2027-11-25 | Recall | Quarantine |
Performance Metrics:
| Metric | Value | Target | Status | |--------|-------|--------|--------| | Shipments Tracked | 15,234 | N/A | ✓ | | Real-Time Visibility | 98.2% | 95.0% | ✓ Above | | On-Time Delivery | 91.5% | 92.0% | ⚠ Slightly Below | | Exception Rate | 4.3% | <5.0% | ✓ On Track | | Traceability Coverage | 99.8% | 100% | ✓ Near Target |
Exceptions Summary:
| Exception Type | Count | Avg Resolution Time | Oldest Open | |----------------|-------|---------------------|-------------| | Delay | 127 | 18 hours | 3 days | | Customs Hold | 23 | 4.2 days | 8 days | | Damaged | 8 | 2 days | 1 day | | Lost | 2 | Pending | 12 days |
If you need more context:
documentation
When the user wants to optimize yard operations, manage trailer parking, or improve dock door utilization. Also use when the user mentions "yard management," "trailer tracking," "yard jockey," "drop trailer program," "trailer pool," "dock scheduling," or "gate management." For cross-dock operations, see cross-docking. For warehouse design, see warehouse-design.
tools
When the user wants to optimize workforce scheduling, create shift plans, or balance labor demand. Also use when the user mentions "staff scheduling," "labor planning," "shift optimization," "crew scheduling," "roster optimization," or "employee scheduling." For task assignment, see task-assignment-problem. For wave planning labor, see wave-planning-optimization.
testing
When the user wants to optimize pick wave planning, schedule warehouse operations, or improve order fulfillment efficiency. Also use when the user mentions "wave management," "batch picking," "pick wave scheduling," "order release optimization," "wave design," or "pick wave strategy." For order batching, see order-batching-optimization. For workforce scheduling, see workforce-scheduling.
testing
When the user wants to optimize warehouse slot assignments, improve pick efficiency, or design warehouse layouts. Also use when the user mentions "slotting optimization," "slot assignment," "ABC slotting," "pick path optimization," "storage location assignment," "warehouse layout optimization," or "forward pick locations." For picker routing, see picker-routing-optimization. For warehouse design, see warehouse-design.