# 1688 选品 + 淘宝比价 + 京东竞品监控，一条数据流跑起来（附 python 源码）

> 前面写了很多篇怎么把数据拿下来：签名、限流、稳定性、字段对齐。但拿下来之后呢？
>
> 数据本身不产生价值。**用三平台的数据回答三个问题**才产生价值：这个货最低多少钱能拿到（1688）、同款现在卖多少钱（淘宝）、对手在做什么动作（京东）。
>
> 这篇把三个问题串成一条流，从采集到输出一张「能不能做」的清单。中间最难的环节不是采集——是**同款识别**。三个平台对同一个商品的描述方式完全不同，认错了，后面全错。

## 一、先说清这条流要输出什么

技术方案要服务于一个具体产物。我们做这条流，最后只输出一张表，每天早上推到企业微信：

| 商品           | 1688 拿货价 | 起订量    | 淘宝价格带（P25/中位/P75） | 京东在售竞品 | 预估净利率 | 退货率红线 | 建议 |
| ------------ | -------- | ------ | ----------------- | ------ | ----- | ----- | -- |
| 不锈钢保温杯 500ml | 28.0     | 2 件代发  | 49 / 65 / 89      | 7 家    | 28.8% | 125%  | 可做 |
| 硅胶折叠水壶       | 41.5     | 50 件起批 | 39 / 45 / 52      | 12 家   | -2.1% | —     | 放弃 |
| 玻璃冷水壶 1.2L   | 19.8     | 5 件代发  | 45 / 58 / 79      | 3 家    | 31.5% | 88%   | 重点 |

这张表的四列数据来源不同、更新频率不同，但**必须在同一行上对齐**——这是同款识别要解决的问题。

三段的职责：

- **1688（货源）**：成本价、起订量、是否支持一件代发。日级更新就够，价格不常动。
- **淘宝（比价）**：不是找最低价，是找**价格带**。最低价往往是不能做的货（亏本引流、规格不同、二手翻新）。
- **京东（竞品）**：不是看有多少竞品，是看竞品**做了什么动作**——降价、上新、断货。

## 二、最难的环节：同款识别

同一个保温杯，三个平台可能这样写：

```
淘宝： 【爆款】316不锈钢保温杯男女士大容量水杯便携 500ml 送礼定制
京东： 富光保温杯 316不锈钢 500ml 白色 FGL-500
1688： 厂家直销 316不锈钢真空保温杯 500ml 定制LOGO 2件起批
```

纯标题匹配会把这当成三个完全不同的商品——关键词几乎没有一个完全重合。而实际上它们就是同一个货。

### 三级匹配：类目 → 标题 → 主图

我们的做法是三道筛子，逐级加严：

| 级别 | 手段                   | 作用           | 成本     |
| -- | -------------------- | ------------ | ------ |
| 一级 | 叶子类目 + 价格量级          | 粗筛，砍掉 90% 噪音 | 极低     |
| 二级 | 标题归一 + 字符 2-gram 相似度 | 找候选          | 低      |
| 三级 | 主图 dHash 汉明距离        | 定案           | 需要下载图片 |

实测数据：**纯标题匹配的误判率约 40%，加上主图后降到 8%**。所以主图那一步不能省——图片是唯一跨平台稳定的特征，同一张图在三个平台都会出现（哪怕是供应商给的白底图）。

### 标题归一：去营销词，但必须保留规格词

这一步的规则是**反直觉的**：营销词要全部删掉（它们纯属噪音），但规格词必须留下并归一化——`500ml`、`0.5L`、`500毫升` 是同一个东西，而 `500ml` 和 `750ml` 是两个不同的货，价格能差 30%。

```python
# matching.py —— 标题归一：去噪音，保留并统一规格
import re

NOISE = ["包邮", "爆款", "新款", "正品", "厂家直销", "一件代发", "买一送一",
         "限时", "特价", "热销", "网红", "同款", "批发", "定制", "送礼", "便携"]

UNIT = {"ml": ("ml", 1), "毫升": ("ml", 1), "l": ("ml", 1000), "升": ("ml", 1000),
        "g": ("g", 1), "克": ("g", 1), "kg": ("g", 1000), "千克": ("g", 1000)}

SPEC_RE = re.compile(r"(\d+(?:\.\d+)?)\s*(ml|毫升|l|升|g|克|kg|千克)")


def normalize_title(title: str) -> str:
    t = title.lower()
    for w in NOISE:
        t = t.replace(w, "")
    t = re.sub(r"[【】\[\]（）()《》\s\-_/|,，、!！?？+*~]+", " ", t)

    def repl(m):
        num = float(m.group(1))
        unit, factor = UNIT[m.group(2)]
        return f" {int(num * factor)}{unit} "     # 0.5l / 500毫升 -> 500ml

    return " ".join(SPEC_RE.sub(repl, t).split())
```

跑一下上面三个标题，归一后是这样：

```
不锈钢保温杯男女士大容量水杯 500ml
富光保温杯不锈钢 500ml白色 fgl-500
不锈钢真空保温杯 500ml logo
```

现在「保温杯」和「500ml」都浮出来了，2-gram 相似度就能算得比较准。

### 相似度：中文用字符 2-gram，不要用空格分词

中文商品标题没有空格分词，用 `split()` 分出来的往往是一整个长串。字符 2-gram 简单粗暴但效果好，而且零依赖：

```python
def bigrams(s: str) -> set[str]:
    s = re.sub(r"\s+", "", s)
    return {s[i:i + 2] for i in range(len(s) - 1)} if len(s) > 1 else {s}

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def jaccard(a: str, b: str) -> float:
    A, B = bigrams(a), bigrams(b)
    return len(A & B) / len(A | B) if A | B else 0.0
```

### 主图 dHash：64 位指纹，汉明距离 ≤10 判相似

dHash 的思路是把图缩到 9×8 灰度，逐像素和右边邻居比较，得到一个 64 位整数。它只关心**明暗结构的梯度**，所以对缩放、轻微裁剪、压缩都不敏感——正好适合跨平台的主图。

```python
# image_hash.py
from PIL import Image


def dhash(path: str, size: int = 8) -> int:
    img = Image.open(path).convert("L").resize((size + 1, size), Image.LANCZOS)
    px = list(img.getdata())
    bits = 0
    for row in range(size):
        for col in range(size):
            i = row * (size + 1) + col
            bits = (bits << 1) | (1 if px[i] > px[i + 1] else 0)
    return bits


def hamming(a: int, b: int) -> int:
    return bin(a ^ b).count("1")
```

`hamming ≤ 10` 判为相似（64 位里最多差 10 位）。这个阈值我们是拿 500 对人工标注样本调出来的，你可以按自己的品类重调——**服装类要放宽到 14**，因为模特姿势和背景差异更大。

### 组合打分：权重和阈值都要保守

```python
from decimal import Decimal

W = {"title": 0.55, "image": 0.35, "price": 0.10}


def price_plausible(cost: Decimal, retail: Decimal) -> float:
    """成本价与售价的倍数关系：正常 1.5~5 倍；<1 或 >8 基本是不同货或不同规格"""
    if cost <= 0 or retail <= 0:
        return 0.0
    ratio = float(retail / cost)
    if ratio < 1.0 or ratio > 8.0:
        return 0.0
    return 1.0


def match_score(title_sim: float, image_sim: float, price_score: float) -> float:
    """image_sim: 1 - hamming/64；无图时传 None 走保守降权"""
    if image_sim is None:
        # 没图不能直接当 0 分，按标题相似度打折，且总分封顶 0.7（进人工复核）
        return min(title_sim * 0.9, 0.7)
    return (W["title"] * title_sim + W["image"] * image_sim
            + W["price"] * price_score)


def is_same(score: float, threshold: float = 0.72) -> bool:
    return score >= threshold
```

**阈值偏保守是有意为之。** 漏匹配的代价是少一个候选，人工还能补；误匹配的代价是按错的价格算利润，最后备货备错——后者贵得多。所以我们把 0.6~0.72 这段叫「灰区」，全部丢进人工复核队列，一天几十条，运营顺手就过了。

## 三、利润模型：把隐性成本算进去，否则算出来的都是幻觉

选品最容易犯的错是「售价减拿货价，毛利 40%，能做」。真实情况是：**毛利 36% 的大件，退货率 25% 的时候净利率只剩 10.7%**，再扣人工仓储就是负的。

四个必须进模型的隐性成本：

1. **平台佣金**（约 5%，各平台各品类不同）
2. **推广费**（新店起量期 10~20%，成熟期 5%）
3. **退货成本**——不只是退钱，是「发货运费 + 回程运费 + 包装折损 + 商品可能无法二次销售」
4. **资金占用**——批量拿货便宜 20%，但货压在仓里，两个月卖不掉就是负现金流

```python
# profit.py
from dataclasses import dataclass
from decimal import Decimal


@dataclass
class CostModel:
    price: Decimal               # 售价
    supply: Decimal              # 1688 拿货价
    ship_out: Decimal            # 发货运费
    ship_back: Decimal           # 退货回程运费
    pack_loss: Decimal = Decimal("3")      # 退货包装/折损
    commission_rate: Decimal = Decimal("0.05")
    ad_rate: Decimal = Decimal("0.10")
    return_rate: Decimal = Decimal("0.15")

    def breakdown(self) -> dict:
        gross = self.price - self.supply - self.ship_out
        commission = self.price * self.commission_rate
        ad = self.price * self.ad_rate
        per_return = self.ship_out + self.ship_back + self.pack_loss
        return_loss = self.return_rate * per_return
        net = gross - commission - ad - return_loss
        return {
            "毛利": gross,
            "佣金": commission,
            "推广": ad,
            "退货损失": return_loss,
            "净利": net,
            "净利率": net / self.price if self.price else Decimal("0"),
        }

    def breakeven_return_rate(self) -> Decimal:
        """
        退货率红线：净利归零时的退货率。
        选品时看这个数比看净利率有用——它告诉你「这个货能承受多差的售后」
        """
        gross = self.price - self.supply - self.ship_out
        after_fixed = gross - self.price * (self.commission_rate + self.ad_rate)
        per_return = self.ship_out + self.ship_back + self.pack_loss
        if per_return <= 0:
            return Decimal("1")
        return max(Decimal("0"), after_fixed / per_return)
```

拿三个真实的品算一遍，都是我们做过或评估过的（费率统一按佣金 5%、推广 10%）：

```
【保温杯】售价 65，拿货 28，发 6，回 8，折损 3，退货率 15%
毛利      31.00   (47.7%)
佣金+推广  9.75
退货损失   2.55   (15% × 17)
净利      18.70   (28.8%)      退货率红线 125% → 健康

【服装】售价 129，拿货 45，发 5，回 9，折损 8，退货率 40%
毛利      79.00   (61.2%)
佣金+推广 19.35
退货损失   8.80   (40% × 22)
净利      50.85   (39.4%)      退货率红线 271% → 扛得住

【大件家居】售价 599，拿货 320，发 60，回 120，折损 80，退货率 25%
毛利     219.00   (36.6%)
佣金+推广 89.85
退货损失  65.00   (25% × 260)
净利      64.15   (10.7%)      退货率红线 49.7% → 危险
```

第三个才是真正的杀手。**它单件退货成本 260 元，比毛利 219 元还高**——退一单等于白卖一单还倒贴。所以退货率红线只有 50%，而大件的行业退货率经常就在 30% 上下，稍有波动就到线。

注意这三个品的**毛利率排序和净利率排序是反的**：服装毛利 61% 但净利 39%，大件毛利 37% 净利只剩 11%。只看「毛利率」选品，会系统性地偏向高毛利品类，而高毛利品类往往正是退货成本最高的。

所以 `breakeven_return_rate()` 的真正价值不是「算红线」，而是**逼你把退货率填对**。我们内部的做法：新品按类目历史退货率的 1.2 倍填，换供应商时重算，大件类目额外单列「退货物流成本」字段——因为它的运费可能比货值还高。

## 四、竞品监控：监控「变化」，不是监控「数据」

这是另一个常见的浪费：写个脚本每天把竞品价格、销量、评价数存一遍，看板上堆满数字，然后没人看。

**人要的不是数据，是「发生了什么」。** 所以我们只输出五类事件：

| 事件     | 触发条件                         | 业务含义             |
| ------ | ---------------------------- | ---------------- |
| 降价     | 幅度 ≥3% **且** 超过该商品 30 天波动 3σ | 对手在抢量，可能要跟       |
| 涨价     | 同上（反向）                       | 对手在提利润，是机会窗口     |
| SKU 减少 | SKU 集合出现消失项                  | 对手断货或砍款，可能是我们的机会 |
| 断货     | `in_stock` 由 true 变 false    | 立刻可抢排名           |
| 评价增速异常 | 日增评价数的 z-score > 3           | 对手在冲量（刷单或大促）     |

关键在**阈值要用该商品自己的历史波动做基线**，不能拍脑袋定 3%：

```python
# monitor.py
from decimal import Decimal


def detect_events(prev: dict, cur: dict, sigma: Decimal) -> list[tuple[str, dict]]:
    """prev/cur: {"price", "in_stock", "sku_ids", "review_count"}"""
    events = []

    # 价格：幅度门槛 + 基线门槛，双条件才报，避免 99 -> 98.9 这种噪音
    if prev["price"] and cur["price"]:
        delta = (cur["price"] - prev["price"]) / prev["price"]
        if abs(delta) >= Decimal("0.03") and abs(delta) >= 3 * sigma:
            events.append(("PRICE_DROP" if delta < 0 else "PRICE_UP",
                           {"delta": round(float(delta), 4),
                            "from": prev["price"], "to": cur["price"]}))

    if prev["in_stock"] and not cur["in_stock"]:
        events.append(("OUT_OF_STOCK", {"price": cur["price"]}))

    gone = set(prev["sku_ids"]) - set(cur["sku_ids"])
    if gone:
        events.append(("SKU_REMOVED", {"sku_ids": sorted(gone)}))

    # 评价增速：用「日增评价数」的 z-score，不是评价总数
    prev_v, cur_v = prev.get("review_velocity"), cur.get("review_velocity")
    review_sigma = prev.get("review_sigma")      # 该商品历史日增评价数的标准差
    if prev_v is not None and cur_v is not None and review_sigma:
        z = (cur_v - prev_v) / review_sigma
        if z > 3:
            events.append(("REVIEW_SURGE", {"z": round(float(z), 2),
                                            "velocity": cur_v}))

    return events
```

这个设计救过我们一次：最初阈值定的是「价格变动 >1%」，结果一个竞品每天微调 1 块钱（99→98→97.5），一天推 300 条告警，运营把通知静音了，三周后对手直接降到 79 没人知道。加了「3σ 基线 + 3% 幅度」双条件之后，一天最多 5 条，每条都值得看。

**监控清单是人定的，不是全类目扫。** 一个类目全扫是几万次调用，其中 99% 你根本不关心。我们的做法是运营每周挑 20~60 个 SKU 进监控池，只盯这些。

## 五、把三段串起来：两级池 + 分级频率

调用量是这条流最大的成本，靠「两级池」收敛：

```
全类目候选 20,000  ──粗筛（日级，1 次/天）──▶  候选池 800
                                              │
                                    ┌─────────┴─────────┐
                          淘宝比价（4 次/天）      京东竞品（48 次/天，仅精选 60）
                                    │                   │
                                    └────▶ 决策表 + 事件推送
```

- **候选池 800 个**：1688 日扫一次（800）+ 淘宝 6 小时比一次（3,200）→ 约 4,000 次/天
- **精选池 60 个**：运营确认过的，京东 30 分钟一次 → 2,880 次/天
- 合计约 6,900 次/天；如果全量按 30 分钟扫 2 万个商品，是 96 万次/天

**调用量降到约 1/140，监控频率反而从 12 小时提升到 30 分钟。** 这就是第 13 篇那条铁律的又一次应用：按变更频率分级调度，只对值得高频的对象高频。

调度骨架可以直接复用第 13 篇的 `DueScheduler`，只需换一张频率表：

```python
LEVELS = {
    "supply":  {"interval_s": 86400, "pool": "candidate"},   # 1688 货源
    "retail":  {"interval_s": 21600, "pool": "candidate"},   # 淘宝比价
    "rival":   {"interval_s": 1800,  "pool": "selected"},    # 京东竞品
}
```

## 六、三个结论

1. **同款识别是地基。** 认错了，比价是错的、利润是错的、备货也是错的。宁可用保守阈值 + 人工复核，也不要让误匹配混进决策表。
2. **利润要算到净利，退货率是隐藏杀手。** 填 5% 退货率的模型得出的结论，不如不填；`breakeven_return_rate()` 的用处是逼你把假设填对。
3. **监控要监控事件，不要监控数字。** 每天的 300 条噪音会让运营关掉通知，那比不监控更糟。


最后一句合规提醒：这套流用的都是平台授权接口返回的公开商品数据，**不采集买家信息、不抓取评价内容原文**。竞品监控是看公开的挂牌价格和库存状态，这和抓取用户数据是两件事，别混在一起做。

---

*如果不想自己维护这套同款识别和分级调度，我们的三平台接口服务已经把同款聚合、价格带统计、竞品事件都做成了标准输出，接进来自带上面这张决策表。评论区或私信聊。*

# 电商 ERP 对接三平台，90% 的问题都出在字段和超时上（附 python 源码）

> 我们做过十几个 ERP 对接项目，复盘下来有个很稳定的分布：**字段问题 62%，超时问题 28%，剩下的 10% 才是签名、限流、权限这些**。
>
> 这个分布有点反直觉——工程师通常以为对接难在协议和加密，实际难的地方在**语义**和**时间窗口**。
>
> 原因也简单：普通的数据同步，是把平台的数据搬到自己库里；ERP 对接，是让**平台的交易记录**和**企业的会计账**在同一个时间窗口内闭合。前者错了解析字段重跑就行，后者错了要调账。
>
> 这篇讲两件事：为什么 ERP 的字段问题特别多，以及 ERP 说的「超时」到底指什么。

先说清楚边界：单位和时区那类错误（分/元、秒/毫秒）第 14 篇讲过，这里不再重复。这篇讲的是**更靠上的语义层**——字段含义对不上，以及业务时限兜不住。

## 一、ERP 与普通同步的根本差别：交易视图 vs 凭证视图

平台接口返回的是**交易视图**：这笔交易发生了什么，买家付了多少钱，货发到哪。

ERP 需要的是**凭证视图**：这笔业务记哪个科目、算哪个成本中心、归哪个账期、冲销哪张单。

两种视图之间不是字段改名，是**结构变形**。下面四个变形，是 62% 字段问题的来源（按出现频次排序）：

| # | 变形          | 平台给的             | ERP 要的                   | 错了的后果            |
| - | ----------- | ---------------- | ------------------------ | ---------------- |
| 1 | 金额拆分        | 一个 `payment`（实付） | 商品金额 + 运费 + 优惠 + 应收 + 佣金 | 毛利算不对，优惠成本全压在订单上 |
| 2 | SKU → 物料    | 一个销售 SKU         | 多个物料 + 数量（组合装）           | 库存只扣一部分，静默虚高     |
| 3 | 时间 → 账期     | 创建时间、支付时间        | 业务日期（可配日结切点）             | 跨月订单归错月，GMV 差几个点 |
| 4 | 订单状态 → 单据状态 | 5 个交易状态          | 审单 / 配货 / 出库 / 开票多张单据    | 状态机错乱，已出库单被重复发货  |

### 变形一：金额是一个数，但账需要五个数

平台给 `payment`（买家实付），ERP 要做三件事：拆出商品金额与运费；把订单级优惠**分摊到行**（否则算不出单品行毛利）；算出应收（扣平台佣金后的金额）。

其中最容易出错、也最能体现专业度的是**优惠分摊**：

```python
# erp/money.py
from decimal import Decimal, ROUND_HALF_UP

CENT = Decimal("0.01")


def allocate_discount(total_discount: Decimal, lines: list[dict]) -> list[Decimal]:
    """
    把订单级优惠按行金额比例分摊到行。
    lines: [{"line_id": ..., "amount": Decimal}]  —— amount 是分摊前的行金额
    返回: 与 lines 等长的分摊额列表（正数表示优惠额）

    两个必须处理的边界：
      1. 尾差：整除不了时，差额补在最后一行，否则各行相加 ≠ 订单优惠，对账天天差 0.01
      2. 最后一行被补成负数：极端比例下尾差可能超过该行金额，要回退给金额最大的行
    """
    if total_discount == 0:
        return [Decimal("0")] * len(lines)

    base = sum(l["amount"] for l in lines)
    if base == 0:                       # 全额赠品单：无金额可分摊，全部记到首行
        out = [Decimal("0")] * len(lines)
        out[0] = total_discount
        return out

    shares, allocated = [], Decimal("0")
    for l in lines[:-1]:
        share = (l["amount"] * total_discount / base).quantize(CENT, ROUND_HALF_UP)
        shares.append(share)
        allocated += share
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    tail = total_discount - allocated    # 尾差兜底
    last = lines[-1]["amount"]
    if shares and tail > last:           # 最后一行承受不住，转给金额最大行
        idx = max(range(len(lines)), key=lambda i: lines[i]["amount"])
        shares[idx] += tail
        shares.append(Decimal("0"))
    else:
        shares.append(tail)
    return shares
```

**那个尾差不是理论问题。** 我们遇到过一个客户，对账表每天差 0.01，财务查了两周，最后发现是三个行按比例分摊，四舍五入后多出 1 分钱没人认领。ERP 对账是**分毫必究**的，普通业务系统里可以忽略的四舍五入，在这里会每天生成一条差异记录。

还有个容易漏的：**赠品行的 `amount` 是 0，但必须保留**。很多人清洗数据时把 0 元行删了，看起来数据更干净，实际是把这个赠品的**成本丢出了订单**——月末算单均成本时偏低，越卖越亏还找不到原因。

### 变形二：一个销售 SKU 不等于一个物料

组合装是最典型的坑。平台上一个「三件套礼盒」是一个 SKU，ERP 里是三个物料各若干件：

```python
# erp/bom.py
# 平台 SKU = 销售单元；ERP 物料 = 库存单元。两者不是一一对应
BOM: dict[str, list[tuple[str, int]]] = {
    "SKU-GIFT-3IN1": [("MUG-001", 2), ("COASTER-002", 4), ("GIFTBOX-003", 1)],
    "SKU-CUP-PAIR":  [("MUG-001", 2)],
}


def expand(sku_id: str, qty: int) -> list[tuple[str, int]]:
    """把销售数量展开成物料需求；非组合装原样返回"""
    bom = BOM.get(sku_id)
    if not bom:
        return [(sku_id, qty)]
    return [(mat, n * qty) for mat, n in bom]
```

不展开的后果是**静默的**：卖了 100 套，只扣了 100 个杯子的库存（如果 SKU 编码恰好和主物料同名），包装盒和杯垫根本没扣。系统不报错，账面看着正常，直到盘点发现包装盒少了 100 个——**而这时已经严重缺货，还在接单**。

比这更隐蔽的是反方向：**平台侧修改了组合装成分**（运营把三件套改成了两件套），你本地的 BOM 没更新。新订单按旧 BOM 扣料，多扣一个杯子，日积月累库存对不上，看谁都像偷了货。所以 BOM 要有版本和生效时间，且**跟着商品详情一起增量同步**（第 13 篇的详情同步应该带出组合装信息）。

### 变形三：ERP 要的不是「时间」，是「账期」

平台给你 `created`（下单）和 `pay_time`（支付），ERP 要的是**业务日期**——这笔钱算哪个月的。三条规矩：

1. **用支付时间，不用创建时间。** 上月 30 号下单、本月 1 号付款的订单，按创建时间归上月，上月 GMV 虚高、本月偏低，两个月的报表都要解释。
2. **日结切点可配。** 很多 ERP 的日结在凌晨 1~2 点（财务第二天早上结前一天的账），0 点到 2 点之间的订单应该算前一个业务日。
3. **时区必须显式。** 这条第 14 篇讲过，但 ERP 里更要命：财务的月报是以北京时间的自然月为准的。

```python
# erp/calendar.py
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

CN = ZoneInfo("Asia/Shanghai")


def business_date(paid_at: datetime, cutoff_hour: int = 0) -> str:
    """业务日期：按支付时间归属账期，支持自定义日结切点"""
    t = paid_at.astimezone(CN) if paid_at.tzinfo else paid_at.replace(tzinfo=CN)
    if cutoff_hour and t.hour < cutoff_hour:
        t -= timedelta(days=1)          # 凌晨单归前一业务日
    return t.strftime("%Y-%m-%d")
```

### 变形四：订单状态机不能直接搬

第 3 篇讲过订单状态机怎么设计。ERP 里还要多一层：一个平台订单在 ERP 里会**生成多张单据**（销售订单 → 出库单 → 应收单 → 发票），每张单据有自己的状态。把平台状态直接映射成 ERP 主单状态，会出现「已出库但未开票」这类中间态无处安放的问题。

**规矩是：平台状态映射到「事件」，事件驱动 ERP 单据流转。** 不要映射状态，映射事件。`WAIT_SELLER_SEND_GOODS` 不是一个状态，是「可以配货了」这个事件。

## 二、超时：ERP 说的超时，八成不是接口超时

这是最容易被技术人员误解的部分。工程师听到「超时」，第一反应是调 `timeout=` 参数；而在 ERP 场景里，真正的超时发生在三层，而且**接口层是最好解决的那层**：

| 层级      | 时限量级        | 超时的后果        | 谁来兜     |
| ------- | ----------- | ------------ | ------- |
| 接口超时    | 秒           | 页面转圈、用户重试    | 工程师     |
| 审单/发货时限 | 小时~天        | 平台扣分、罚款、流量降权 | 运营 + 系统 |
| 对账窗口    | 天（T+1 / 月结） | 账不平，财务停工     | 系统 + 财务 |

### 2.1 接口层：ERP 前台是同步阻塞的，所以不能长重试

普通后台同步可以慢慢重试（第 9 篇的预算式重试），但 ERP 不一样：**用户在页面上点「同步订单」，鼠标在那儿等着**。一次重试 30 秒，用户就以为卡死了，会连点五次——于是同一批订单被并发拉五遍。

解法不是把超时调长，而是**把长事务拆成「提交 + 查结果」**：

```python
# erp/jobs.py —— 提交-轮询模式：前台 3 秒拿到 task_id，重活交后台
import asyncio
import uuid


class JobService:
    def __init__(self, store, worker):
        self.store = store          # 持久化任务表（对应第 2 篇的 SQLite 任务表思路）
        self.worker = worker

    async def submit(self, kind: str, idem_key: str, payload: dict) -> str:
        """幂等键：同一批订单连点五次，只产生一个任务"""
        exist = await self.store.find_by_idem(idem_key)
        if exist:
            return exist.task_id
        task_id = uuid.uuid4().hex
        await self.store.create(task_id, kind, idem_key, payload, status="pending")
        self.worker.wake()           # 不 await 执行，立即返回
        return task_id

    async def status(self, task_id: str) -> dict:
        task = await self.store.get(task_id)
        return {"task_id": task_id, "status": task.status,
                "done": task.done, "failed": task.failed,
                "errors": task.errors[:10]}     # 只回传前 10 条，前端可展开

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# 后台 worker：每项任务有独立的业务截止时间（DDL），到点没做完就切下一批
async def run_batch(items, handler, ddl_seconds: int = 300):
    loop = asyncio.get_running_loop()
    deadline = loop.time() + ddl_seconds
    skipped = []
    for it in items:
        if loop.time() >= deadline:
            skipped.append(it)          # 超预算的放回队列，下一轮继续，不算失败
            continue
        await handler(it)
    return skipped
```

三个设计点：

1. **幂等键 = 业务主键**（比如 `sync_order:2026-10-29:shop_123`）。用户连点、两个运营同时点、定时任务和手动任务撞在一起，都只产生一个任务。
2. **接口 3 秒内一定返回**。返回 `task_id` 而不是结果，前端轮询 `status`。这比把 `timeout` 调到 60 秒体验好得多，也避免了长连接占资源。
3. **后台任务有 DDL，超了就把剩下的放回队列**，不要硬跑完——因为下一次调度很快就到，硬跑完只会让两轮任务叠在一起（这就是第 10 篇说的「任务堆积」）。

### 2.2 业务层：库存回写要两段式，不能「等接口成功」

超卖的根因在 ERP 里通常不是接口慢，而是**扣库存的时机错了**。

想一想 ERP 的出库流程：审单 → 生成出库单 → 拣货 → 出库 → 回写平台库存。如果只在「出库完成」时才扣库存，那么从审单到出库这段时间，库存是**虚高**的，平台还在卖，卖出去的单进来又排队等出库——超卖就是这么堆起来的。

正确做法是**预占 + 确认**（第 3 篇讲过 `total - reserved - safety` 的账本，这里给 ERP 的落法）：

```python
# erp/stock.py —— 预占 + 确认，两段式
async def reserve(order_id: str, sku_id: str, qty: int, ttl_s: int = 3600) -> bool:
    """审单通过就预占，不等到出库。返回 False 表示可用库存不足"""
    ok = await db.execute(
        "UPDATE stock SET reserved = reserved + :q "
        "WHERE sku_id = :s AND total - reserved - safety >= :q",
        {"q": qty, "s": sku_id},
    )
    if ok:
        await db.execute(
            "INSERT INTO reservation(order_id, sku_id, qty, expires_at) "
            "VALUES (:o, :s, :q, now() + :ttl)",
            {"o": order_id, "s": sku_id, "q": qty, "ttl": ttl_s},
        )
    return bool(ok)


async def confirm(order_id: str):
    """出库成功：预占转实扣，同步回写平台"""
    await db.execute(
        "UPDATE stock SET total = total - r.qty, reserved = reserved - r.qty "
        "FROM reservation r WHERE r.order_id = :o AND r.sku_id = stock.sku_id",
        {"o": order_id},
    )
    await db.execute("DELETE FROM reservation WHERE order_id = :o", {"o": order_id})
    await push_stock_to_platform(order_id)


async def release_expired():
    """TTL 到期的预占自动释放——取消单、审核不通过、接口超时都靠这个兜底"""
    await db.execute("DELETE FROM reservation WHERE expires_at < now() RETURNING sku_id, qty")
```

关键在 `release_expired`：**所有异常路径最终都收敛成「预占超时自动释放」**。审核卡住、出库单作废、平台接口 502，不需要每种情况都写补偿逻辑，等 TTL 到就行。这是把分布式一致性问题降级成一个定时任务。

### 2.3 对账层：日切和差异表

T+1 对账是 ERP 的硬需求，坑集中在两处：

- **日切边界**：ERP 按业务日切（前面 `business_date`），平台按自然日，两边窗口不一样，边界上一定有差异。解法是**对账窗口取两边的交集并额外多取一天**，差异在窗口内消化。
- **差异要有表，不能只有日志**。差异记录必须能落到 `recon_diff` 表里（订单号、金额差、状态差、处理结果），财务才能逐条勾对。只在日志里喊一句「对账不平」，等于没做对账。

## 三、把两类问题收成一份契约表

字段和超时，看起来是两个问题，本质是一个：**你和 ERP 之间对「这笔业务的含义」和「多久必须到达」没有书面约定**。

所以我们每个 ERP 项目都会产出一份 `contract.yaml`，既是配置，也是交接文档：

```yaml
# contract.yaml —— ERP 对接契约：含义 + 单位 + 缺失处理 + 超时语义
fields:
  paid_amount:
    source: { taobao: payment, jd: orderPrice, alibaba: totalAmount }
    semantics: 买家实付，含运费、已扣优惠
    unit: CNY            # jd 原始单位为分，由适配器归一（见第 14 篇）
    missing: error       # 缺失即报错，不允许落 0
  discount:
    semantics: 订单级优惠总额
    allocate: by_line_amount   # 必须分摊到行，规则写进契约
  sku_id:
    semantics: 销售单元，可能对应多个物料
    expand: bom          # 出库前必须展开，见 bom.py
  pay_time:
    semantics: 业务日期基准（不是创建时间）
    business_day_cutoff: 2      # 凌晨 2 点日结

timeouts:
  sync_order:
    deadline_ms: 3000            # 前台同步接口上限
    on_timeout: enqueue_job      # 超时不重试，转异步任务
    idem_key: "sync_order:{shop_id}:{date}"
  push_stock:
    deadline_ms: 2000
    on_timeout: retry_queue      # 后台任务可重试，退避由链路层负责
    consistency: reserve_confirm # 两段式，不允许直写
  reconcile:
    window: T+1
    diff_table: recon_diff
```

这张表的价值在项目交付后：**换人接手时，不用读代码猜「这个字段到底什么意思」，表里写着。** 半年后平台改字段，也是从这张表开始改，改完跑一遍启动自检确认没有漏配。

```python
# erp/contract_check.py —— 启动自检：契约里声明的，代码里必须都实现
def check(contract: dict, adapters: dict):
    problems = []
    for name, spec in contract["fields"].items():
        for platform in spec["source"]:
            if platform not in adapters:
                problems.append(f"{name}: 缺 {platform} 适配器")
            elif not hasattr(adapters[platform], f"map_{name}"):
                problems.append(f"{name}: {platform} 未实现映射")
        if spec.get("allocate") and not spec.get("unit"):
            problems.append(f"{name}: 声明了分摊但没声明单位")
    if problems:
        raise SystemExit("契约定稿失败:\n  " + "\n  ".join(problems))
```

启动就炸，比上线三天后发现某个平台少了个字段好得多。

## 四、一句话总结

ERP 对接难，不难在协议。**难在平台描述的是「交易」，ERP 记的是「账」；平台保证的是「最终一致」，ERP 需要的是「账期内闭合」。**

字段是语义翻译问题，超时是时间窗口问题。把这两件事都写进契约表，90% 的问题在发生之前就消掉了。

剩下 10% 是签名、限流、授权那些，前面几篇都写过：签名和鉴权看第 14 篇，限流看第 10 篇，链路层稳定性看第 9 篇。

---

*如果不想在这一层反复踩坑，我们提供的三平台数据接口已经把金额拆分、物料展开、业务日期这三件事做成了标准字段，直接对接 ERP 少写一半适配代码。评论区或私信聊。*