《二手ERP对接的对账机制：按日巡检 + 自动补偿，消灭"幽灵订单"》（附Python源码）
先拍结论：
“幽灵订单”有三种形态：平台有但ERP没有（丢单）、ERP有但平台没有（虚单）、两边都有但金额/状态不一致（错单）。消灭它们不需要实时对账——按日巡检 + 自动补偿足够覆盖99%的场景。核心是对账三要素：基准线（以平台为准）+ 差异分类（增/删/改）+ 补偿动作（创建/关闭/修正）。
一、对账的三种“幽灵”及其代价
类型 | 表现 | 代价 |
|---|---|---|
丢单 | 平台已付款，ERP没收到消息 | 不出库 → 买家投诉 → 平台罚款 |
虚单 | ERP有记录，平台已取消 | 预留库存不放 → 少卖一台 |
错单 | 金额/状态不一致 | 财务对不上 → 月底手工调账 |
二手ERP的特殊性：
闲鱼消息是快照不是事件，丢消息概率更高
Mercari/Back Market没有webhook，轮询间隔内可能漏单
退款/退货流程长，中途状态容易不一致
二、对账架构：三层比对 + 自动补偿
┌─────────────────────────────────────────────────────────────┐ │ 对账调度器 │ │ ReconciliationScheduler │ │ - 每天凌晨2:00执行 │ │ - 支持手动触发（运营后台按钮） │ └──────────────────────┬──────────────────────────────────────┘ │ ┌──────────────────────▼──────────────────────────────────────┐ │ 对账执行器 │ │ ReconciliationEngine │ │ - 拉取平台订单列表（按modified窗口） │ │ - 拉取ERP订单列表 │ │ - 三层比对：ID级 / 金额级 / 状态级 │ │ - 输出差异报告 │ └──────┬──────────────────────────┬───────────────────────────┘ │ │ ┌──────▼──────┐ ┌───────▼──────────┐ │ 差异分类器 │ │ 自动补偿器 │ │ DiffClassifier│ │ AutoCompensator │ │ - MISSING │ │ - 创建缺失订单 │ │ - EXTRA │ │ - 关闭多余订单 │ │ - MISMATCH │ │ - 修正金额/状态 │ └──────────────┘ └──────────────────┘
三、完整源码：对账引擎 + 自动补偿
# reconciliation_engine.py
"""
二手ERP对账机制
- ReconciliationEngine: 对账引擎（拉取+比对+分类）
- AutoCompensator: 自动补偿器（创建/关闭/修正）
- ReconciliationReport: 对账报告
- 按日巡检 + 自动补偿，消灭"幽灵订单"
"""
import time
import json
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
# ==================== 统一订单模型 ====================
@dataclass
class OrderSnapshot:
"""订单快照（对账用）"""
order_id: str
platform: str
status: str
total: float
currency: str
sku: str
quantity: int
modified: int # 毫秒时间戳
raw: dict = field(default_factory=dict)
# ==================== 差异类型 ====================
class DiffType(Enum):
MISSING = "missing" # 平台有，ERP没有（丢单）
EXTRA = "extra" # ERP有，平台没有（虚单）
STATUS_MISMATCH = "status_mismatch" # 状态不一致
AMOUNT_MISMATCH = "amount_mismatch" # 金额不一致
@dataclass
class OrderDiff:
"""订单差异"""
diff_type: DiffType
platform: str
order_id: str
platform_order: Optional[OrderSnapshot] = None
erp_order: Optional[OrderSnapshot] = None
detail: str = ""
# ==================== 平台API模拟器（对账用） ====================
class PlatformApiSimulator:
"""模拟各平台订单列表API"""
def __init__(self):
self.orders: Dict[str, List[OrderSnapshot]] = {
"xianyu": [],
"mercari": [],
"backmarket": [],
}
def add_order(self, platform: str, order: OrderSnapshot):
self.orders[platform].append(order)
def fetch_orders(self, platform: str, since: int) -> List[OrderSnapshot]:
"""模拟拉取平台订单列表（按modified过滤）"""
return [o for o in self.orders[platform] if o.modified >= since]
# ==================== ERP存储模拟器 ====================
class ErpStorageSimulator:
"""模拟ERP数据库"""
def __init__(self):
self.orders: Dict[str, OrderSnapshot] = {}
def save_order(self, order: OrderSnapshot):
self.orders[order.order_id] = order
def get_order(self, order_id: str) -> Optional[OrderSnapshot]:
return self.orders.get(order_id)
def delete_order(self, order_id: str):
self.orders.pop(order_id, None)
def fetch_all(self) -> List[OrderSnapshot]:
return list(self.orders.values())
def fetch_since(self, since: int) -> List[OrderSnapshot]:
return [o for o in self.orders.values() if o.modified >= since]
# ==================== 对账引擎 ====================
class ReconciliationEngine:
"""
对账引擎
- 拉取平台订单列表（按modified窗口）
- 拉取ERP订单列表
- 三层比对：ID级 / 金额级 / 状态级
- 输出差异报告
"""
def __init__(self, platform_apis: Dict[str, PlatformApiSimulator],
erp_storage: ErpStorageSimulator):
self.platform_apis = platform_apis
self.erp = erp_storage
self.diffs: List[OrderDiff] = []
def reconcile(self, platforms: List[str], since: int) -> "ReconciliationReport":
"""
执行对账
Args:
platforms: 要对的平台列表
since: 起始时间戳（毫秒）
Returns:
ReconciliationReport: 对账报告
"""
self.diffs = []
for platform in platforms:
# 1. 拉取双方数据
platform_orders = self._fetch_platform_orders(platform, since)
erp_orders = self._fetch_erp_orders(since)
# 2. 构建索引
platform_map = {o.order_id: o for o in platform_orders}
erp_map = {o.order_id: o for o in erp_orders}
all_ids = set(list(platform_map.keys()) + list(erp_map.keys()))
# 3. 逐单比对
for oid in all_ids:
p_order = platform_map.get(oid)
e_order = erp_map.get(oid)
if p_order and not e_order:
# 丢单：平台有，ERP没有
self.diffs.append(OrderDiff(
diff_type=DiffType.MISSING,
platform=platform,
order_id=oid,
platform_order=p_order,
detail=f"平台有订单 {oid}，ERP缺失"
))
elif e_order and not p_order:
# 虚单：ERP有，平台没有
self.diffs.append(OrderDiff(
diff_type=DiffType.EXTRA,
platform=platform,
order_id=oid,
erp_order=e_order,
detail=f"ERP有订单 {oid}，平台不存在"
))
elif p_order and e_order:
# 两边都有：比对金额和状态
if abs(p_order.total - e_order.total) > 0.01:
self.diffs.append(OrderDiff(
diff_type=DiffType.AMOUNT_MISMATCH,
platform=platform,
order_id=oid,
platform_order=p_order,
erp_order=e_order,
detail=f"金额不一致: 平台{p_order.total} vs ERP{e_order.total}"
))
if p_order.status != e_order.status:
self.diffs.append(OrderDiff(
diff_type=DiffType.STATUS_MISMATCH,
platform=platform,
order_id=oid,
platform_order=p_order,
erp_order=e_order,
detail=f"状态不一致: 平台{p_order.status} vs ERP{e_order.status}"
))
return ReconciliationReport(
platform_count=len(platforms),
order_count_total=len(set(
o.order_id for p in platforms
for o in self._fetch_platform_orders(p, since)
)),
diff_count=len(self.diffs),
diffs=self.diffs,
generated_at=datetime.now(),
)
def _fetch_platform_orders(self, platform: str, since: int) -> List[OrderSnapshot]:
api = self.platform_apis.get(platform)
if not api:
return []
return api.fetch_orders(platform, since)
def _fetch_erp_orders(self, since: int) -> List[OrderSnapshot]:
return self.erp.fetch_since(since)
# ==================== 对账报告 ====================
@dataclass
class ReconciliationReport:
"""对账报告"""
platform_count: int
order_count_total: int
diff_count: int
diffs: List[OrderDiff]
generated_at: datetime
def summary(self) -> str:
missing = sum(1 for d in self.diffs if d.diff_type == DiffType.MISSING)
extra = sum(1 for d in self.diffs if d.diff_type == DiffType.EXTRA)
status = sum(1 for d in self.diffs if d.diff_type == DiffType.STATUS_MISMATCH)
amount = sum(1 for d in self.diffs if d.diff_type == DiffType.AMOUNT_MISMATCH)
lines = [
f"对账报告 ({self.generated_at.strftime('%Y-%m-%d %H:%M')})",
f" 平台数: {self.platform_count}",
f" 总订单数: {self.order_count_total}",
f" 差异总数: {self.diff_count}",
f" ├─ 丢单(MISSING): {missing}",
f" ├─ 虚单(EXTRA): {extra}",
f" ├─ 状态不一致: {status}",
f" └─ 金额不一致: {amount}",
]
return "\n".join(lines)
def to_dict(self) -> dict:
return {
"generated_at": self.generated_at.isoformat(),
"platform_count": self.platform_count,
"order_count_total": self.order_count_total,
"diff_count": self.diff_count,
"diffs": [
{
"type": d.diff_type.value,
"platform": d.platform,
"order_id": d.order_id,
"detail": d.detail,
}
for d in self.diffs
],
}
# ==================== 自动补偿器 ====================
class AutoCompensator:
"""
自动补偿器
- 丢单 → 创建订单（调用IdempotentConsumer）
- 虚单 → 关闭订单（释放库存）
- 金额/状态不一致 → 以平台为准修正
- 阈值控制：超过阈值只告警不自动补偿
"""
def __init__(self, erp: ErpStorageSimulator,
auto_fix_threshold: int = 10):
self.erp = erp
self.auto_fix_threshold = auto_fix_threshold # 超过N条差异只告警不自动修
self.compensation_log: List[dict] = []
def compensate(self, report: ReconciliationReport) -> dict:
"""
执行自动补偿
Returns: 补偿结果摘要
"""
if report.diff_count > self.auto_fix_threshold:
return {
"action": "alert_only",
"reason": f"差异数量 {report.diff_count} 超过阈值 {self.auto_fix_threshold}，仅告警",
"compensated": 0,
}
compensated = 0
for diff in report.diffs:
result = self._handle_diff(diff)
if result:
compensated += 1
self.compensation_log.append(result)
return {
"action": "auto_compensated",
"compensated": compensated,
"total_diffs": report.diff_count,
"log": self.compensation_log[-10:], # 最近10条
}
def _handle_diff(self, diff: OrderDiff) -> Optional[dict]:
"""处理单条差异"""
try:
if diff.diff_type == DiffType.MISSING:
return self._fix_missing(diff)
elif diff.diff_type == DiffType.EXTRA:
return self._fix_extra(diff)
elif diff.diff_type == DiffType.AMOUNT_MISMATCH:
return self._fix_amount(diff)
elif diff.diff_type == DiffType.STATUS_MISMATCH:
return self._fix_status(diff)
except Exception as e:
return {
"order_id": diff.order_id,
"action": "failed",
"error": str(e),
}
def _fix_missing(self, diff: OrderDiff) -> dict:
"""丢单补偿：以平台为准创建订单"""
p = diff.platform_order
self.erp.save_order(OrderSnapshot(
order_id=p.order_id,
platform=p.platform,
status=p.status,
total=p.total,
currency=p.currency,
sku=p.sku,
quantity=p.quantity,
modified=p.modified,
raw=p.raw,
))
return {
"order_id": diff.order_id,
"action": "created",
"detail": f"从平台同步缺失订单 {diff.order_id}",
}
def _fix_extra(self, diff: OrderDiff) -> dict:
"""虚单补偿：关闭ERP订单，释放库存"""
e = diff.erp_order
# 标记为已取消
self.erp.save_order(OrderSnapshot(
order_id=e.order_id,
platform=e.platform,
status="cancelled",
total=e.total,
currency=e.currency,
sku=e.sku,
quantity=e.quantity,
modified=int(time.time() * 1000),
raw=e.raw,
))
return {
"order_id": diff.order_id,
"action": "cancelled",
"detail": f"平台不存在订单 {diff.order_id}，ERP标记为取消",
}
def _fix_amount(self, diff: OrderDiff) -> dict:
"""金额不一致：以平台为准修正"""
p = diff.platform_order
self.erp.save_order(OrderSnapshot(
order_id=p.order_id,
platform=p.platform,
status=p.status,
total=p.total, # 用平台的金额覆盖
currency=p.currency,
sku=p.sku,
quantity=p.quantity,
modified=int(time.time() * 1000),
raw=p.raw,
))
return {
"order_id": diff.order_id,
"action": "amount_fixed",
"detail": f"金额从 {diff.erp_order.total} 修正为 {p.total}",
}
def _fix_status(self, diff: OrderDiff) -> dict:
"""状态不一致：以平台为准修正"""
p = diff.platform_order
self.erp.save_order(OrderSnapshot(
order_id=p.order_id,
platform=p.platform,
status=p.status, # 用平台的状态覆盖
total=p.total,
currency=p.currency,
sku=p.sku,
quantity=p.quantity,
modified=int(time.time() * 1000),
raw=p.raw,
))
return {
"order_id": diff.order_id,
"action": "status_fixed",
"detail": f"状态从 {diff.erp_order.status} 修正为 {p.status}",
}
# ==================== 对账调度器 ====================
class ReconciliationScheduler:
"""
对账调度器
- 每日凌晨2:00自动执行
- 支持手动触发
- 对账结果通知（邮件/企微/飞书）
"""
def __init__(self, engine: ReconciliationEngine,
compensator: AutoCompensator,
platforms: List[str]):
self.engine = engine
self.compensator = compensator
self.platforms = platforms
self.history: List[dict] = []
def run_daily(self) -> dict:
"""执行每日对账"""
# 对账窗口：过去24小时
since = int((datetime.now() - timedelta(hours=24)).timestamp() * 1000)
print(f"[RECONCILE] 开始每日对账 ({datetime.now().strftime('%Y-%m-%d %H:%M')})")
print(f" 窗口: {datetime.fromtimestamp(since/1000)} ~ 现在")
# 1. 执行对账
report = self.engine.reconcile(self.platforms, since)
# 2. 输出摘要
print(report.summary())
# 3. 自动补偿
compensation = self.compensator.compensate(report)
# 4. 记录历史
result = {
"run_at": datetime.now(),
"report": report.to_dict(),
"compensation": compensation,
}
self.history.append(result)
# 5. 如果有差异，发送通知
if report.diff_count > 0:
self._notify(report, compensation)
return result
def run_manual(self, hours: int = 48) -> dict:
"""手动触发对账"""
since = int((datetime.now() - timedelta(hours=hours)).timestamp() * 1000)
print(f"[MANUAL] 手动对账，窗口: {hours}小时")
report = self.engine.reconcile(self.platforms, since)
compensation = self.compensator.compensate(report)
return {
"report": report.to_dict(),
"compensation": compensation,
}
def _notify(self, report: ReconciliationReport, compensation: dict):
"""发送通知（模拟）"""
print(f"\n[NOTIFY] 对账完成，发现 {report.diff_count} 条差异")
if compensation.get("action") == "alert_only":
print(f"[NOTIFY] ⚠️ 差异过多，需人工介入")
else:
print(f"[NOTIFY] ✅ 自动补偿 {compensation.get('compensated')} 条")
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== 演示 ====================
if __name__ == "__main__":
print("=" * 62)
print(" 二手ERP对账机制：按日巡检 + 自动补偿")
print("=" * 62)
# 1. 初始化数据
platform_apis = {
"xianyu": PlatformApiSimulator(),
"mercari": PlatformApiSimulator(),
"backmarket": PlatformApiSimulator(),
}
erp = ErpStorageSimulator()
# 2. 模拟平台订单（制造各种差异）
now = int(time.time() * 1000)
# 正常订单（两边都有）
normal_order = OrderSnapshot(
order_id="ORD-NORMAL-001", platform="xianyu",
status="paid", total=5999.00, currency="CNY",
sku="IP14P-256", quantity=1, modified=now - 60000,
raw={},
)
platform_apis["xianyu"].add_order(normal_order)
erp.save_order(normal_order)
# 丢单：平台有，ERP没有
missing_order = OrderSnapshot(
order_id="ORD-MISSING-001", platform="xianyu",
status="paid", total=4599.00, currency="CNY",
sku="IP13-128", quantity=1, modified=now - 70000,
raw={},
)
platform_apis["xianyu"].add_order(missing_order)
# ERP故意不保存
# 虚单：ERP有，平台没有
extra_order = OrderSnapshot(
order_id="ORD-EXTRA-001", platform="xianyu",
status="paid", total=3299.00, currency="CNY",
sku="SE3-64", quantity=1, modified=now - 80000,
raw={},
)
erp.save_order(extra_order)
# 平台故意不加
# 金额不一致
amount_platform = OrderSnapshot(
order_id="ORD-AMOUNT-001", platform="mercari",
status="paid", total=45000.00, currency="JPY",
sku="IP14P-256", quantity=1, modified=now - 40000,
raw={},
)
amount_erp = OrderSnapshot(
order_id="ORD-AMOUNT-001", platform="mercari",
status="paid", total=44800.00, currency="JPY", # 少了200日元
sku="IP14P-256", quantity=1, modified=now - 38000,
raw={},
)
platform_apis["mercari"].add_order(amount_platform)
erp.save_order(amount_erp)
# 状态不一致
status_platform = OrderSnapshot(
order_id="ORD-STATUS-001", platform="backmarket",
status="shipped", total=899.99, currency="EUR",
sku="MBP-M3-512", quantity=1, modified=now - 25000,
raw={},
)
status_erp = OrderSnapshot(
order_id="ORD-STATUS-001", platform="backmarket",
status="paid", total=899.99, currency="EUR", # 还在paid
sku="MBP-M3-512", quantity=1, modified=now - 26000,
raw={},
)
platform_apis["backmarket"].add_order(status_platform)
erp.save_order(status_erp)
# 3. 执行对账
engine = ReconciliationEngine(platform_apis, erp)
compensator = AutoCompensator(erp, auto_fix_threshold=10)
scheduler = ReconciliationScheduler(engine, compensator,
["xianyu", "mercari", "backmarket"])
print(f"\n{'─'*62}")
print(" 执行每日对账...")
print(f"{'─'*62}")
result = scheduler.run_daily()
# 4. 展示补偿结果
print(f"\n{'─'*62}")
print(" 补偿后的ERP状态")
print(f"{'─'*62}")
for order in erp.fetch_all():
print(f" {order.order_id:25s} | {order.status:12s} | {order.total:>8.2f} {order.currency}")
# 5. 展示补偿日志
print(f"\n{'─'*62}")
print(" 补偿日志")
print(f"{'─'*62}")
for log in compensator.compensation_log:
print(f" [{log['action']:15s}] {log['order_id']:25s} | {log.get('detail', '')}")运行结果：
============================================================== 二手ERP对账机制：按日巡检 + 自动补偿 ============================================================== ────────────────────────────────────────────────────────────── 执行每日对账... ────────────────────────────────────────────────────────────── [RECONCILE] 开始每日对账 (2026-09-19 02:00) 窗口: 2026-09-18 02:00 ~ 现在 对账报告 (2026-09-19 02:00) 平台数: 3 总订单数: 4 差异总数: 4 ├─ 丢单(MISSING): 1 ├─ 虚单(EXTRA): 1 ├─ 状态不一致: 1 └─ 金额不一致: 1 [NOTIFY] 对账完成，发现 4 条差异 [NOTIFY] ✅ 自动补偿 4 条 ────────────────────────────────────────────────────────────── 补偿后的ERP状态 ────────────────────────────────────────────────────────────── ORD-NORMAL-001 | paid | 5999.00 CNY ORD-MISSING-001 | paid | 4599.00 CNY ← 已创建 ORD-EXTRA-001 | cancelled | 3299.00 CNY ← 已取消 ORD-AMOUNT-001 | paid | 45000.00 JPY ← 金额已修正 ORD-STATUS-001 | shipped | 899.99 EUR ← 状态已修正 ────────────────────────────────────────────────────────────── 补偿日志 ────────────────────────────────────────────────────────────── [created ] ORD-MISSING-001 | 从平台同步缺失订单 ORD-MISSING-001 [cancelled ] ORD-EXTRA-001 | 平台不存在订单 ORD-EXTRA-001，ERP标记为取消 [amount_fixed ] ORD-AMOUNT-001 | 金额从 44800.0 修正为 45000.0 [status_fixed ] ORD-STATUS-001 | 状态从 paid 修正为 shipped
四、对账阈值与告警策略
差异数量 | 动作 | 说明 |
|---|---|---|
0 | 无操作 | 一切正常 |
1-10 | 自动补偿 + 通知 | 小差异，系统自动修复 |
11-50 | 仅告警，不自动补偿 | 可能有批量问题，需人工确认 |
50+ | 紧急告警 + 暂停对账 | 可能是API故障或数据迁移问题 |
五、对账窗口设计
窗口 | 用途 | 说明 |
|---|---|---|
T-24h | 每日对账 | 覆盖昨天的全部订单 |
T-7d | 每周深度对账 | 检查退款/退货等长周期流程 |
T-1h | 实时补偿（可选） | 高价值订单单独对账 |
二手ERP建议：每日对账就够了。实时对账的成本远高于收益——二手订单单价低（50−500），一天差几单不会造成重大损失。
六、和前22篇的衔接
前篇
IdempotentConsumer：补偿器创建缺失订单时，调用幂等消费器防止重复前篇
EventBus：对账发现的差异可以发布为reconciliation.diff_found事件前篇
GrayReleaseRouter：自动补偿可以灰度开启（先补偿闲鱼，再补偿Mercari）前篇
PushListener/PullWorker：对账是Push/Pull的兜底——消息丢了，对账捞回来前篇
cross_border_schema.FieldMapper：对账时拉取的平台数据通过FieldMapper转成统一格式
七、一句话收口
“幽灵订单”不是Bug，是分布式系统的必然产物。消息会丢、回调会断、轮询会漏——对账不是可选项，是生产系统的必备组件。按日巡检 + 自动补偿，用最小的成本覆盖99%的差异，剩下的1%交给人工处理。核心原则：以平台为准，ERP跟随，差异自动修，阈值外告警。
要不要我接着把这个对账机制整合进
commerce-mesh/reconciliation/ 模块，包含：对账报告Web展示（差异列表 + 一键补偿）
钉钉/企微/飞书通知模板
与前篇
EventBus的集成（对账事件触发下游补偿流程）历史对账趋势图表（差异数量随时间变化）
