① 问题 ② 算法 ③ 场景 ④ 代码 ⑤ 关联 ⑥ 价值

跨系统数据对账 — ERP/WMS/OMS三系统库存一致性自动比对与差异处置

Skill-Cross-System-Data-Reconciliation · 24-标签工程

causalexperimentrecommendationmulti_agent供应链与补货推荐与搜索MAS与智能体工程WF-A 智能补货WF-I 智能体工程

收录于供应链全链路智能化手册

年化 ROI5-10万元

实现难度⭐⭐⭐☆☆

业务视角

适用角色数据架构师 / 供应链数字化负责人 · CTO · 数据工程师 · 供应链团队

什么情况下用多平台数据孤岛导致断货识别延迟8小时；标签覆盖率不足使AI决策触发率<30%；想实现分析→行动自动闭环但不知从何下手

成功是什么样的统一 Tag Schema + 传播引擎将标签覆盖率从 30% 提升至 97%；Palantir 风格 Object-Action-Writeback 将补货响应从 2 天缩短至 4 小时自动触发

业务痛点

多平台 SKU 编码混乱无法统一合规标签手工维护遗漏频繁预测模型有了但结果无法自动触发采购标签打了但没有质量监控

1. 解决的问题

运营面临"ERP/WMS/FBA三套库存数字不一致无从判断"——自动对账从月度盘点→每日自动，防止超卖和缺货年化节省5-10万元

2. 核心算法逻辑

跨系统对账解决：ERP显示库存100件，WMS显示95件，Amazon FBA显示88件——哪个是真的？

3. 业务应用场景

未自动抽取；请查看原始 Skill 卡片。

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

ROI预估：自动对账将库存差异发现从"月度盘点"→"每日自动"，减少因库存数据不准导致的超卖/缺货，年化节省约5-10万元；Amazon FBA差异申诉成功率提升，每年追回约$1,000-3,000赔款
实施难度：⭐⭐⭐☆☆（需要各系统API接入，主要是技术集成工作）
优先级评分：⭐⭐⭐⭐⭐（多平台运营的品牌，库存数据不准是日常噩梦，自动对账是基础必须）

7. 代码模板

代码块数量：2 · 路径：未检测到

"""
跨系统数据对账引擎
功能：三系统对账 / 差异计算 / 根因分类 / 自动处置建议 / 对账Tag生成
"""
from dataclasses import dataclass, field
import warnings
warnings.filterwarnings('ignore')


@dataclass
class SystemInventory:
    sku_id: str
    system_name: str
    qty: int
    reserved_qty: int = 0
    in_transit_qty: int = 0
    timestamp: str = ""


@dataclass
class ReconciliationResult:
    sku_id: str
    erp_qty: int
    wms_qty: int
    oms_available: int
    fba_sellable: int
    max_discrepancy: int
    discrepancy_status: str  # OK / MINOR / MAJOR / CRITICAL
    root_cause: str
    recommended_action: str
    tags: dict = field(default_factory=dict)


def reconcile_inventory(erp: SystemInventory, wms: SystemInventory,
                         oms: SystemInventory, fba: SystemInventory = None) -> ReconciliationResult:
    """执行四系统库存对账"""
    values = [erp.qty, wms.qty, oms.available if hasattr(oms, 'available') else oms.qty]
    if fba:
        values.append(fba.qty)

    max_val = max(values)
    min_val = min(values)
    max_discrepancy = max_val - min_val
    discrepancy_pct = max_discrepancy / max(1, max_val) * 100

    # 严重程度
    if discrepancy_pct < 2:
        status = "OK"
    elif discrepancy_pct < 5:
        status = "MINOR"
    elif discrepancy_pct < 15:
        status = "MAJOR"
    else:
        status = "CRITICAL"

    # 根因推断
    if erp.qty > wms.qty + 5:
        root_cause = "ERP_WMS_SYNC_DELAY"
        action = "检查近期入库单是否已在WMS确认扫描"
    elif wms.qty > (oms.qty if hasattr(oms, 'qty') else oms.qty) + 5:

8. 论文来源

2309.11234
2401.09823