供应链控制塔实施

控制塔不是一个软件，而是一种能力——让供应链全链路状态实时可见，异常可以在 5 分钟内被感知并响应。

控制塔架构层次

异常预警引擎

"""
供应链控制塔预警系统
"""
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
class AlertLevel(Enum):
INFO = "信息"
WARNING = "警告"
CRITICAL = "严重"
EMERGENCY = "紧急"
@dataclass
class Alert:
id: str
level: AlertLevel
category: str
message: str
impact: str
recommended_action: str
created_at: str = field(
default_factory=lambda: datetime.now().strftime("%Y-%m-%d %H:%M")
)
acknowledged: bool = False
class AlertEngine:
"""预警引擎"""
def __init__(self):
self.alerts: list[Alert] = []
self.rules: list[dict] = []
self._setup_default_rules()
def _setup_default_rules(self):
"""设置默认预警规则"""
self.rules = [
{
"name": "库存低于安全线",
"condition": lambda data: (
data.get("stock_days", 99) < data.get("safety_stock_days", 7)
),
"level": AlertLevel.WARNING,
"category": "库存",
"message_template": "{sku} 库存 {stock_days} 天，低于安全线 {safety_stock_days} 天",
"action": "立即触发补货，检查供应商交期",
},
{
"name": "发货严重延迟",
"condition": lambda data: data.get("delay_days", 0) > 3,
"level": AlertLevel.CRITICAL,
"category": "物流",
"message_template": "订单 {order_id} 延迟 {delay_days} 天，影响客户 {customer}",
"action": "联系物流商，通知客户，评估空运替代",
},
{
"name": "供应商交货率下降",
"condition": lambda data: data.get("otd_rate", 1.0) < 0.85,
"level": AlertLevel.WARNING,
"category": "供应商",
"message_template": "{supplier} 本月准时交货率 {otd_rate:.0%}，低于 85% 目标",
"action": "约谈供应商，要求整改计划，评估备选供应商",
},
{
"name": "需求异常突增",
"condition": lambda data: (
data.get("actual_demand", 0)
> data.get("forecast_demand", 0) * 1.30
),
"level": AlertLevel.WARNING,
"category": "需求",
"message_template": "SKU {sku} 实际需求超预测 {excess_pct:.0%}",
"action": "紧急评估库存缺口，加快补货，与销售确认是否持续",
},
]
def evaluate(self, event_data: dict) -> list[Alert]:
"""评估事件，生成预警"""
new_alerts = []
for rule in self.rules:
try:
if rule["condition"](event_data):
message = rule["message_template"].format(**event_data)
alert = Alert(
id=f"ALT-{len(self.alerts)+1:04d}",
level=rule["level"],
category=rule["category"],
message=message,
impact=f"业务影响：{rule['category']} 类风险",
recommended_action=rule["action"],
)
self.alerts.append(alert)
new_alerts.append(alert)
except (KeyError, TypeError):
pass
return new_alerts
def dashboard(self) -> dict:
"""控制塔仪表板摘要"""
by_level = {}
by_category = {}
for alert in self.alerts:
level_name = alert.level.value
by_level[level_name] = by_level.get(level_name, 0) + 1
by_category[alert.category] = (
by_category.get(alert.category, 0) + 1
)
active = [a for a in self.alerts if not a.acknowledged]
return {
"总预警数": len(self.alerts),
"未处理": len(active),
"按级别": by_level,
"按类别": by_category,
"最新预警": [
f"[{a.level.value}] {a.message}"
for a in sorted(
active,
key=lambda x: list(AlertLevel).index(x.level),
reverse=True,
)[:3]
],
}
# 演示
engine = AlertEngine()
events = [
{"sku": "CPU-X100", "stock_days": 4, "safety_stock_days": 7},
{"order_id": "ORD-20240315", "delay_days": 5, "customer": "华为终端"},
{"supplier": "东莞精密", "otd_rate": 0.78},
{"sku": "B002", "actual_demand": 1600, "forecast_demand": 1200, "excess_pct": 0.33},
]
for event in events:
new_alerts = engine.evaluate(event)
for alert in new_alerts:
print(f"[{alert.level.value}] {alert.message}")
print(f"  建议: {alert.recommended_action}\n")
print("=== 控制塔仪表板 ===")
for k, v in engine.dashboard().items():
print(f"  {k}: {v}")

控制塔实施路线图

成功的控制塔需要什么

行动清单

• [ ] 梳理现有系统（ERP/WMS/TMS）能提供哪些实时数据，绘制数据流图
• [ ] 定义优先级最高的 5 类预警场景（库存低于安全线、发货延迟等）
• [ ] 建立预警处理 SOP：预警 → 责任人 → 响应时限 → 关闭标准
• [ ] 在 MVP 阶段先用 Excel 或 BI 工具搭建基础控制塔，验证业务价值
• [ ] 设置控制塔 KPI：MTTR（平均响应时间）、预警准确率、误报率
• [ ] 评估商业控制塔软件（Kinaxis、Blue Yonder、o9 等）与自建的成本对比

下一节：03-数据治理与主数据管理 — 数据质量是控制塔的地基。