数据治理与主数据管理

同一个供应商在 ERP 里叫"深圳精密"，在财务系统叫"深圳精密制造有限公司"——这不只是命名问题，它会导致对账失败、报表错误和系统无法打通。主数据管理解决的就是这类"数据真实性"问题。

供应链主数据体系

数据质量检查系统

"""
主数据质量检查器
"""
from dataclasses import dataclass
from typing import Any
@dataclass
class DataRecord:
"""数据记录"""
id: str
fields: dict[str, Any]
source_system: str
class DataQualityChecker:
"""数据质量检查器（5 维度）"""
@staticmethod
def completeness(
records: list[DataRecord],
required_fields: list[str],
) -> dict:
"""完整性：必填字段是否都有值"""
issues = []
for r in records:
missing = [
f for f in required_fields
if not r.fields.get(f)
]
if missing:
issues.append({
"记录": r.id,
"缺失字段": missing,
})
score = (len(records) - len(issues)) / len(records) if records else 0
return {
"维度": "完整性",
"得分": f"{score*100:.1f}%",
"问题数": len(issues),
"问题样本": issues[:3],
}
@staticmethod
def uniqueness(
records: list[DataRecord],
unique_field: str,
) -> dict:
"""唯一性：关键字段是否有重复"""
seen = {}
duplicates = []
for r in records:
val = r.fields.get(unique_field)
if val:
if val in seen:
duplicates.append({
"重复值": val,
"记录": [seen[val], r.id],
"来源": [
seen.get(f"src_{val}", "unknown"),
r.source_system,
],
})
else:
seen[val] = r.id
seen[f"src_{val}"] = r.source_system
score = (len(records) - len(duplicates)) / len(records) if records else 0
return {
"维度": "唯一性",
"字段": unique_field,
"得分": f"{score*100:.1f}%",
"重复数": len(duplicates),
"重复样本": duplicates[:3],
}
@staticmethod
def consistency(
records: list[DataRecord],
consistency_rules: list[dict],
) -> dict:
"""一致性：跨字段逻辑是否一致"""
violations = []
for r in records:
for rule in consistency_rules:
try:
if not rule["check"](r.fields):
violations.append({
"记录": r.id,
"规则": rule["name"],
"说明": rule["description"],
})
except (KeyError, TypeError):
pass
score = (len(records) - len(violations)) / len(records) if records else 0
return {
"维度": "一致性",
"得分": f"{score*100:.1f}%",
"违规数": len(violations),
"样本": violations[:3],
}
@classmethod
def full_report(
cls,
records: list[DataRecord],
config: dict,
) -> dict:
"""完整数据质量报告"""
results = []
if "required_fields" in config:
results.append(
cls.completeness(records, config["required_fields"])
)
if "unique_field" in config:
results.append(
cls.uniqueness(records, config["unique_field"])
)
if "consistency_rules" in config:
results.append(
cls.consistency(records, config["consistency_rules"])
)
scores = [
float(r["得分"][:-1]) for r in results if "得分" in r
]
overall = sum(scores) / len(scores) if scores else 0
return {
"总体质量评分": f"{overall:.1f}%",
"评级": (
"优秀" if overall >= 95
else "良好" if overall >= 85
else "需改善" if overall >= 70
else "需立即整治"
),
"维度报告": results,
}
# 演示 — 供应商主数据质量检查
supplier_records = [
DataRecord("SUP-001", {
"name": "深圳精密制造",
"tax_id": "91440300XXXXXXXX",
"bank_account": "6225XXXXXXXXXXXX",
"contact": "李明",
"email": "li@szmfg.com",
}, "ERP"),
DataRecord("SUP-002", {
"name": "东莞快捷电子",
"tax_id": "",               # 缺失！
"bank_account": "6228XXXXXXXXXXXX",
"contact": "",              # 缺失！
"email": "sales@dgjd.com",
}, "SRM"),
DataRecord("SUP-003", {
"name": "深圳精密制造有限公司",   # 与 SUP-001 重复！
"tax_id": "91440300XXXXXXXX",    # 税号重复！
"bank_account": "6225XXXXXXXXXXXX",
"contact": "王华",
"email": "wang@szmfg.com",
}, "FINANCE"),
]
checker = DataQualityChecker()
config = {
"required_fields": ["name", "tax_id", "bank_account", "contact", "email"],
"unique_field": "tax_id",
"consistency_rules": [
{
"name": "邮箱格式",
"description": "邮箱必须包含@符号",
"check": lambda f: "@" in (f.get("email") or ""),
},
],
}
print("=== 供应商主数据质量报告 ===")
report = checker.full_report(supplier_records, config)
print(f"\n总体评分: {report['总体质量评分']} [{report['评级']}]")
for dim in report["维度报告"]:
print(f"\n{dim['维度']}: {dim['得分']}")
if dim.get("问题数", 0) > 0 or dim.get("重复数", 0) > 0:
issues = dim.get("问题样本") or dim.get("重复样本") or dim.get("样本") or []
for issue in issues:
print(f"  问题: {issue}")

主数据治理框架

主数据 Golden Record 策略

"""
多系统主数据合并 — Golden Record
"""
def merge_to_golden(records: list[dict], field_priority: dict) -> dict:
"""
多系统数据合并为黄金记录
field_priority: {field_name: [system_priority_order]}
"""
golden = {}
for field, priority_systems in field_priority.items():
for system in priority_systems:
val = next(
(r.get(field) for r in records if r.get("source") == system and r.get(field)),
None,
)
if val:
golden[field] = val
golden[f"{field}_source"] = system
break
else:
golden[field] = None
return golden
# 同一供应商的多系统记录
supplier_records_raw = [
{
"source": "ERP",
"name": "深圳精密制造",
"tax_id": "91440300XXXXXXXX",
"bank_account": None,
"rating": "A",
},
{
"source": "FINANCE",
"name": "深圳精密制造有限公司",
"tax_id": "91440300XXXXXXXX",
"bank_account": "6225XXXXXXXXXXXX",
"rating": None,
},
{
"source": "SRM",
"name": "深圳精密",
"tax_id": None,
"bank_account": "6225XXXXXXXXXXXX",
"rating": "A+",
},
]
priority = {
"name": ["FINANCE", "ERP", "SRM"],     # 财务系统名称最规范
"tax_id": ["ERP", "FINANCE", "SRM"],   # ERP 税号最可靠
"bank_account": ["FINANCE", "SRM", "ERP"],
"rating": ["SRM", "ERP", "FINANCE"],   # SRM 评级最新
}
golden = merge_to_golden(supplier_records_raw, priority)
print("=== Golden Record ===")
for k, v in golden.items():
if not k.endswith("_source"):
print(f"  {k}: {v} (来源: {golden.get(k+'_source', '无')})")

行动清单

• [ ] 统计当前供应商主数据中有多少"同一供应商多个编码"的情况，启动去重项目
• [ ] 为物料、供应商、客户分别制定编码规范和必填字段清单
• [ ] 建立主数据变更审批流程：关键字段（税号、银行账户）变更需双人审核
• [ ] 对主数据进行季度质量扫描：完整性/唯一性/一致性三个维度
• [ ] 指定各主数据域的数据 Owner（物料 → 供应链，供应商 → 采购，客户 → 销售）
• [ ] 评估是否需要引入 MDM 系统（SAP MDG、Informatica MDM 等）或先用 ERP 主数据模块整治

下一章：风险管理 → — 为不确定性做好准备。