SKU主数据黄金记录治理 — MDM体系、主键统一与多源冲突消解

Skill-SKU-Master-Data-Golden-Record · 24-标签工程

causalexperimentknowledge_graphmulti_agentpricingvisual_generation广告与投放供应链与补货知识图谱与RAGMAS与智能体工程定价与利润视觉内容生成WF-A 智能补货WF-B 广告优化WF-F 动态定价WF-I 智能体工程WF-J DTC 独立站增长WF-L 内容营销增长

收录于标签工程与本体驱动手册

实现难度⭐⭐⭐⭐☆

业务视角

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

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

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

业务痛点

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

1. 解决的问题

运营面临"Amazon/Shopify/TikTok/ERP四套SKU编码跨渠道数据孤岛"——MDM黄金记录将跨渠道库存准确率从74%提升至99%，新品上市从3天降至4小时

2. 核心算法逻辑

主数据黄金记录（MDM Golden Record）是整个供应链数据体系的"单一真相来源"。没有它，每个系统都用自己的SKU定义，跨系统分析永远是噪声。

3. 业务应用场景

场景A：500+ SKU全量MDM治理 - 现状：Amazon/Shopify/TikTok三套SKU编码体系，ERP另一套，数据孤岛导致库存视图不准 - MDM建立后：500个SKU全部有黄金记录，跨渠道库存合并视图准确率从74%→99% - 年化价值：消除跨渠道超卖风险（每次超卖约损失$500），年均防止12次 = $6,000

**场景B：新品上市MDM快速注册** - 新品从ERP创建到Amazon/TikTok上架通常需要手工录入3次（重复劳动） - MDM引擎：ERP创建 → 自动生成黄金记录 → 自动推送各平台模板 - 新品上市准备时间从3天→4小时

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

ROI预估：500个SKU建立GR后，跨渠道库存合并准确率从74%→99%，防止超卖损失年化$6,000+；新品上市从3天→4小时，每月2个新品节省约2个工作日；消除3个ERP数据重复录入工作，年节省约3万元人力
实施难度：⭐⭐⭐⭐☆（初期建立映射关系工作量大，持续维护成本低）
优先级评分：⭐⭐⭐⭐⭐（没有GR，所有跨系统分析都在沙地建楼）
评估依据：Gartner：MDM项目ROI平均3-5倍，数据质量提升25%可降低运营成本约10%

7. 代码模板

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

"""
SKU主数据黄金记录治理引擎
功能：多源SKU数据摄入 / 匹配融合 / 冲突消解 / 黄金记录生成 / 质量监控
"""
import hashlib
import re
from dataclasses import dataclass, field
from typing import Any, Optional
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')


# 字段优先级配置（数字越大优先级越高）
FIELD_PRIORITY = {
    "unit_cost":    {"erp": 100, "amazon": 0, "shopify": 30, "tiktok": 0, "supplier": 80},
    "title":        {"erp": 40, "amazon": 100, "shopify": 80, "tiktok": 70, "supplier": 20},
    "category":     {"erp": 90, "amazon": 100, "shopify": 60, "tiktok": 40, "supplier": 50},
    "weight_kg":    {"erp": 80, "amazon": 70, "shopify": 60, "tiktok": 0, "supplier": 100},
    "lead_time_days": {"erp": 60, "amazon": 0, "shopify": 0, "tiktok": 0, "supplier": 100},
    "default":      {"erp": 70, "amazon": 80, "shopify": 60, "tiktok": 40, "supplier": 50},
}

REQUIRED_FIELDS = ["internal_id", "canonical_name", "category", "unit_cost", "weight_kg"]


@dataclass
class SourceRecord:
    source: str          # erp / amazon / shopify / tiktok / supplier
    external_id: str
    attributes: dict
    updated_at: datetime = field(default_factory=datetime.now)
    confidence: float = 1.0


@dataclass
class GoldenRecord:
    internal_id: str
    canonical_name: str
    source_records: list = field(default_factory=list)
    master_attributes: dict = field(default_factory=dict)
    platform_ids: dict = field(default_factory=dict)
    conflicts: list = field(default_factory=list)
    version: int = 1
    confidence_score: float = 0.0
    created_at: str = field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d"))
    last_merged_at: str = field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d %H:%M"))

    def completeness_score(self) -> float:
        filled = sum(1 for f in REQUIRED_FIELDS if self.master_attributes.get(f) or f in ("internal_id", "canonical_name"))
        return filled / len(REQUIRED_FIELDS)


class MDMGoldenRecordEngine:

    def __init__(self):
        self.golden_records: dict = {}    # internal_id → GoldenRecord
        self.id_index: dict = {}          # "source:ext_id" → internal_id
        self.merge_log: list = []

8. 论文来源

2310.14823
2402.09234