# 双 11 前必须打通的三条数据链路，晚一周就来不及（附 python 源码）

距离大促只剩几天，这个时间点最不该做的事是加功能。

但有三条链路如果现在还没打通，这几天必须打通。不是为了让系统更先进，是为了**不在这三天里出事故**。

大促出事故，从来不是因为系统不够快。是三种最朴素的情况：

- **卖错了价**——活动没生效、券没叠加、改价延迟
- **卖超了货**——多平台同时卖同一批货
- **客户付了钱没发货**——订单漏拉，等你发现时已经被投诉

这三件事在平时都是小概率。在大促它们是必然，因为触发条件全被放大了：

| 放大项    | 平时       | 大促               | 后果              |
| ------ | -------- | ---------------- | --------------- |
| 请求量    | 100 单/小时 | 5,000 单/小时       | 限流从"偶尔撞"变成"必然撞" |
| 变更频率   | 一天改 3 次价 | 一小时改 3 次         | 人工操作的错误率线性上升    |
| 单次错误代价 | 超卖赔 5 块  | 赔 50 + 平台处罚 + 口碑 | 容错空间被压到零        |

所以这篇不讲架构，只讲三件事：**每条链路的红线是什么、崩了会怎样、怎么用一百多行代码守住**。

## 一、先讲一个认知：大促期间你不是在"同步"，是在"核对"

这是全文最重要的一句话，也是大多数人搞错的地方。

日常开发的心智模型是"同步"：我拉取平台数据，写进我的库，保持一致。

大促期间这个模型会崩，因为**写者有两个人**：

```
运营 → 在平台后台改价、改库存、设活动
你   → 在系统里同步价格、同步库存
```

两个写者，必然打架。你 10:00 同步下来的价格，运营 10:01 在后台改了，你的库就是错的；反过来，你写回去的价格可能覆盖掉运营刚设的活动价。

**正确的心智模型是：平台是唯一真相，你的系统只做三件事——读、核对、告警。**

- 你**不写**价格（除了你自己有定价权的那部分）
- 你**不猜**库存（平台的数就是数，你只维护自己占用的部分）
- 你**不假设**订单拉全了（必须用单数对账验证）

这个转变很难，因为它意味着放弃"我控制一切"的安全感。但大促期间，控制欲就是事故源。

下面三条链路，全都建立在这个前提上：**每个 guard 都是"预期 vs 实际"的比对器，不是同步器。**

## 二、先把地基打了：一张告警表 + 一个落库函数

三条链路都要写告警，所以先把这部分抽出来。它只有几十行，但有两个设计决定值得说：

```python
# guard/base.py
"""三条链路共用的地基：表结构 + 告警落库。

为什么告警落库而不是只发到群里：
  大促期间消息会被刷屏,没人看得过来。
  落库后能按 level 过滤、按 code 聚合,
  事后还能复盘"到底哪类问题先出现"。
"""

import sqlite3
import time

SCHEMA = """
create table if not exists alert(
  id integer primary key autoincrement,
  level text not null,          -- P0 立刻处理 / P1 当天处理
  code text not null,           -- 机器可读,用于聚合和静默
  detail text,
  created_at integer,
  acked integer default 0
);
create index if not exists ix_alert_time on alert(created_at);

create table if not exists price_check(
  id integer primary key autoincrement,
  item_id text, sku_id text,
  expected text, actual text, diff text, layers text,
  checked_at integer
);
create index if not exists ix_price_item on price_check(item_id, checked_at);

create table if not exists stock_level(
  item_id text primary key,
  available integer, reserved integer, real integer,
  floor integer, updated_at integer
);

create table if not exists stock_back(
  id integer primary key autoincrement,
  item_id text, qty integer, source text, day text, created_at integer
);

create table if not exists order_gap(
  platform text, hour_start integer,
  local_count integer, remote_count integer, checked_at integer,
  primary key (platform, hour_start)
);

-- 这张是你们业务库里的表，这里给个最小结构好让 order_lag 能跑
create table if not exists orders(
  order_id text primary key,
  platform text,
  paid_at integer,        -- 买家支付时间,平台给
  created_at integer      -- 进入我方系统的时间,我方记
);
create index if not exists ix_orders_created on orders(created_at);
"""


def open_db(path="promo.db"):
    db = sqlite3.connect(path)
    db.executescript(SCHEMA)
    return db


def alert(db, level, code, detail):
    """落库 + 立刻 commit。

    不用 ORM、不进队列,是刻意的:告警是最后一道防线,
    它自身依赖的组件越少越好。大促期间任何中间件抖动
    都不该让告警丢失。
    """
    db.execute(
        "insert into alert(level, code, detail, created_at) values(?,?,?,?)",
        (level, code, detail, int(time.time())),
    )
    db.commit()
    print(f"[{level}] {code} {detail}")
```

两个决定：

1. **`code` 字段是机器可读的**（`PRICE_DRIFT`、`ORDER_GAP`），不是给人看的中文。这样能做聚合——一晚上 300 条 `STOCK_LOW` 应该合成一条"37 个 SKU 触底"，而不是 300 条消息。
2. **告警必须落库，不能只发消息群。** 群消息在大促会被刷屏，落库的能查、能统计、能复盘。

## 三、链路一：价格——红线是偏差不超过 1 分钱

### 大促期间它为什么不一样

平时改价是低频事件，错一次改回来就行。大促不一样：

- 活动价有**生效时间点**（0 点、10 点、20 点），差一分钟就是一批订单按错价成交
- 优惠叠加有三层（标价 → 单品券 → 平台满减），任何一层没生效，到手价就是错的
- 改价窗口集中在开抢前 10 分钟，运营手忙脚乱，改错的概率最高

### 最容易崩的那个点：你核对的是标价，不是到手价

标价对得上，不代表到手价对得上。

```
标价      99.00
单品券    -10.00    ← 券没生效
平台满减  -15.00
到手价    74.00     ← 客户实际付的
```

如果只看标价，99 是对的，一切正常。但客户实际付了 84 而不是 74，客诉从这里开始；反过来如果券提前生效，客户付了 64，每单亏 10 块。

**所以价格核对必须核到"到手价"这一层**，而且要能拆开看是哪一层没生效。

### 代码

```python
# guard/price.py
"""价格核对器。注意：它不是同步器，不写平台价格。

定位：运营是唯一写者，我们是观测者。
     发现偏差 → 告警 → 人工确认 → 人工修。
     绝不允许程序自动"修正"平台价格——大促期间自动写价是灾难。
"""

import time
from decimal import Decimal, InvalidOperation

from base import alert

TOL = Decimal("0.01")        # 1 分钱。低于这个数就是数据精度问题，不是业务问题

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def to_dec(v):
    try:
        return Decimal(str(v))
    except (InvalidOperation, TypeError, ValueError):
        return None


def check_price(db, item_id, sku_id, expected, actual, layers=None):
    """核对到手价。

    expected: 我方算出的到手价（标价 - 券 - 满减）
    actual:   平台返回的实际到手价
    layers:   {"list": 99, "coupon": -10, "promo": -15}，用于定位是哪一层错

    返回 (是否通过, 偏差)
    """
    e, a = to_dec(expected), to_dec(actual)
    if e is None or a is None:
        # 拿不到价本身就是故障：大促期间价格字段为空，通常是接口降级了
        alert(db, "P0", "PRICE_MISSING",
              f"{item_id}/{sku_id} 价格不可用 expected={expected} actual={actual}")
        return False, None

    diff = a - e
    db.execute(
        "insert into price_check(item_id, sku_id, expected, actual, diff, layers, checked_at) "
        "values(?,?,?,?,?,?,?)",
        (item_id, sku_id, str(e), str(a), str(diff),
         str(layers or {}), int(time.time())),
    )

    if abs(diff) <= TOL:
        return True, diff

    # 偏差超过 1% 就是资损级，1 分钱到 1% 之间是配置级
    level = "P0" if abs(diff) > e * Decimal("0.01") else "P1"
    where = _locate(layers, e, a) if layers else "未知层"
    alert(db, level, "PRICE_DRIFT",
          f"{item_id}/{sku_id} 预期 {e} 实际 {a} 偏差 {diff} 疑似 {where}")
    return False, diff


def _locate(layers, expected, actual):
    """把偏差归到具体某一层，方便运营直接去改那一项。

    注意：券和满减都存负数，所以这里是加，不是减。
    写成减法会把符号翻两次，明明是券没生效，却报"我方算价错"。
    """
    lst = to_dec(layers.get("list"))
    coupon = to_dec(layers.get("coupon")) or Decimal("0")
    promo = to_dec(layers.get("promo")) or Decimal("0")
    if lst is None:
        return "标价缺失"

    # 先确认我方算得对
    if lst + coupon + promo != expected:
        return "我方算价逻辑"

    # 我方算对了，那就是平台少减了某一层——逐层试，命中哪个就是哪个
    if actual == lst + promo:
        return "券未生效"
    if actual == lst + coupon:
        return "满减未生效"
    if actual == lst:
        return "券和满减都未生效"
    return "平台价格与任一层组合都不符"
```

### 大促期间的核对频率

平时 30 分钟一次就够。大促要按活动时间点加密：

| 时间窗         | 频率   | 说明              |
| ----------- | ---- | --------------- |
| 活动生效前 10 分钟 | 10 秒 | 盯住生效，这是最容易出错的窗口 |
| 活动生效后 10 分钟 | 10 秒 | 确认生效且没有错层       |
| 活动进行中       | 1 分钟 | 运营仍在改价、改券       |
| 凌晨低峰        | 5 分钟 | 省配额             |

**不要全天 10 秒一次**。大促期间配额是最贵的资源，价格核对一秒钟扫一万个 SKU，会把库存和订单链路的额度吃光——这是第 10 篇那个"全局桶"的另一种表现。

## 四、链路二：库存——红线是可售不低于安全水位

### 大促期间它为什么不一样

平时库存是"还剩多少"。大促库存是"我承诺了多少"。

一个 SKU 在大促期间同时被四种东西占用：

```
物理库存 120 件
├── 日常可售      40    ← 期就在卖的
├── 秒杀预留      50    ← 20:00 开抢，此刻不能卖
├── 预售定金      20    ← 已付定金，待付尾款
└── 安全水位      10    ← 预留的缓冲，防超卖
```

如果你只维护一个数字 `120`，那么秒杀开始的瞬间，日常订单会把这 50 件吃掉，等到 20:00 开抢，库存是 0。

**所以库存不是数量，是承诺。**&#x8981;按用途分池，而不是一个总数。

### 最容易崩的那个点：多平台同时卖同一批货

这是超卖的头号原因，也是最多人踩的：

```
实际仓库：120 件
淘宝设置：100 件
京东设置：100 件
→ 两边各卖 100，一共 200，超卖 80 件
```

平台之间不通气，谁都不知道对方卖了多少。你必须在**自己这边**做池化：

```
物理池 120
  ├── 淘宝额度 60
  ├── 京东额度 50
  └── 缓冲 10        ← 谁先卖完可以借，但缓冲不能被借光
```

**缓冲池是关键**。它让你在某个平台爆单时还有得卖，同时给你时间调整分配。

### 代码

```python
# guard/stock.py
"""库存守护。两个职责：
   1. 水位熔断——可售低于安全线就自动下架或限购
   2. 池化校验——各平台额度之和不能超过物理池

注意：这里的 available 是平台可售数，reserved 是本系统已占用但
     还没回写到平台的量。真实可售 = available - reserved。
     这两个数分开存，是因为回写有延迟，混在一起会重复扣。
"""

import time

from base import alert


def real_available(db, item_id, platform_available, reserved):
    """扣掉本地占用后的真实可售"""
    return max(0, int(platform_available) - int(reserved))

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def check_stock(db, item_id, platform_available, reserved, floor=10, action="off"):
    """水位检查。返回建议动作：normal / limit / off

    floor: 安全水位。大促建议按日均销量的 0.5 倍设，不要凭感觉。
    action: 触底后的动作，off=下架，limit=限购 1 件
    """
    raw = int(platform_available) - int(reserved)
    avail = max(0, raw)

    db.execute(
        "insert or replace into stock_level"
        "(item_id, available, reserved, real, floor, updated_at) values(?,?,?,?,?,?)",
        (item_id, platform_available, reserved, avail, floor, int(time.time())),
    )
    db.commit()

    # 超卖和售罄必须分开报：售罄就是没货可卖，超卖是要走赔付预案的事故
    if raw < 0:
        alert(db, "P0", "STOCK_NEGATIVE",
              f"{item_id} 超卖 {-raw} 件（平台 {platform_available} 本地占用 {reserved}），走赔付预案")
        return "off"

    if avail <= 0:
        alert(db, "P0", "STOCK_OUT",
              f"{item_id} 可售为 0（平台 {platform_available} 本地占用 {reserved}）")
        return "off"

    if avail <= floor:
        # 大促期间这里必须是 P0：P1 会淹没在其他告警里没人看
        alert(db, "P0", "STOCK_LOW",
              f"{item_id} 可售 {avail} 已到水位 {floor}，建议{action}")
        return action

    return "normal"


def check_pool(db, item_id, physical, quotas):
    """池化校验：各平台额度之和不能超过物理库存

    quotas: {"taobao": 60, "jd": 50, "pdd": 0}
    超配是超卖的前置条件，必须在开卖前拦住，不是卖的时候才管。
    """
    total = sum(quotas.values())
    if total > physical:
        alert(db, "P0", "POOL_OVERCOMMIT",
              f"{item_id} 物理 {physical} 分配 {total}（{quotas}）超配 {total - physical}")
        return False
    return True


def replenish(db, item_id, qty, source, daily_sales=None):
    """库存回补：取消单、退款、超时未付款都要回补

    但必须设上限：刷单党会反复下单占库存再取消，
    没有上限的回补等于把库存交给对手。

    daily_sales 传当日销量，上限按 20% 算；
    拿不到销量就退回固定上限。**不能用 qty 自己算上限**——
    那等于"回补多少就允许多少"，一点约束都没有。
    """
    used = db.execute(
        "select coalesce(sum(qty),0) from stock_back where item_id=? and day=?",
        (item_id, time.strftime("%Y%m%d")),
    ).fetchone()[0]

    cap = max(20, int(daily_sales * 0.2)) if daily_sales else 50

    if used + qty > cap:
        alert(db, "P1", "REPLENISH_ABNORMAL",
              f"{item_id} 今日已回补 {used}，本次 {qty} 超上限 {cap}，来源 {source}")
        return False

    db.execute(
        "insert into stock_back(item_id, qty, source, day, created_at) values(?,?,?,?,?)",
        (item_id, qty, source, time.strftime("%Y%m%d"), int(time.time())),
    )
    db.commit()
    return True
```

那个 `replenish` 的限流容易被忽略，但它是真事：大促期间同行恶意下单占库存再取消，能把你的秒杀库存锁死。回补不设上限，等于把库存交给对手。

## 五、链路三：订单——红线是漏单为零

### 大促期间它为什么不一样

订单链路的特殊性在于：**漏单不可恢复，重复可以吸收**。

- 重复拉一单 → `ON CONFLICT` 幂等掉，无损失
- 漏拉一单 → 客户付了钱，你没发货，等超时罚款才知道

所以整条链路的设计原则是：**宁可重复十次，不可遗漏一次**（第 3 篇那条铁律，在大促是生死线）。

大促会同时放大三个坑：

**坑一：分页漂移。** 你在翻第 3 页的时候，第 1 页又插入了新订单，整个列表往后挤。第 3 页的部分单会跑到第 4 页，被你跳过。

**坑二：时间窗边界。** 同一秒多单，用 `modified` 时间戳做游标，等于把同一秒的单全丢了。

**坑三：接口超时。** 大促期间超时率从 0.1% 涨到 5%，一次超时就是整页丢。

### 解法：双游标 + 单数对账

双游标（第 3 篇讲过）解决坑一和坑二：游标存 `(时间戳, 订单号)` 二元组，重叠加窗 60 秒。

真正解决"我到底漏没漏"的，是**单数对账**：

```python
# guard/order.py
"""订单对账。核心不是"拉",是"验"。

拉订单的代码到处都有，能回答"我拉全了吗"的代码很少。
方法：用平台的订单计数接口，和自己库里的数比。
     不等就说明漏了,重拉整个时间窗。
"""

import time

from base import alert


def reconcile(db, platform, hour_start, local_count, remote_count):
    """按小时对账。

    为什么按小时不按天：
      按天对账，漏单要等到次日凌晨才发现，那时候客户已经投诉了。
      按小时对账，最晚一小时发现，还有救。
    """
    db.execute(
        "insert or replace into order_gap"
        "(platform, hour_start, local_count, remote_count, checked_at) values(?,?,?,?,?)",
        (platform, hour_start, local_count, remote_count, int(time.time())),
    )
    db.commit()

    if local_count == remote_count:
        # 注意：0 比 0 也会走到这里。如果拉单任务本身死了，
        # 两边都是 0，"对账通过"毫无意义。这种情况只能靠拉单任务的
        # 心跳发现，不能靠对账。第一次跑之前务必确认拉单是活的。
        return False

    gap = remote_count - local_count
    if gap > 0:
        # 漏单，P0。重拉这个小时的全部订单，不要试图"补差额"——
        # 你不知道漏的是哪几单，只能整窗重拉。
        alert(db, "P0", "ORDER_GAP",
              f"{platform} {_fmt(hour_start)} 本地 {local_count} 平台 {remote_count} "
              f"漏 {gap} 单，需整窗重拉")
        return True

    # 本地比平台多：重复或脏数据，P1，可以慢慢查
    alert(db, "P1", "ORDER_DUP",
          f"{platform} {_fmt(hour_start)} 本地 {local_count} 平台 {remote_count} 多 {-gap}")
    return False


def _fmt(ts):
    return time.strftime("%m-%d %H:%M", time.localtime(ts))


def order_lag(db, limit=20):
    """订单延迟：从支付到进入本系统的时间差

    这个指标比"成功率"重要。大促发货时限是硬指标，
    延迟 30 分钟意味着你有 30 分钟的订单还不知道要发。
    """
    rows = db.execute(
        "select order_id, paid_at, created_at from orders "
        "order by created_at desc limit ?", (limit,),
    ).fetchall()
    lags = [c - p for _, p, c in rows if p and c > p]
    if not lags:
        return None
    lags.sort()
    p95 = lags[max(0, int(len(lags) * 0.95) - 1)]
    return {"p50": lags[len(lags) // 2], "p95": p95, "max": lags[-1]}


def check_lag(db, warn=300, crit=900):
    """延迟超阈值告警。阈值按平台的发货时限倒推：
      24 小时发货 → 延迟 15 分钟是危急（crit）
      48 小时发货 → 30 分钟告警就够
    """
    lag = order_lag(db)
    if not lag:
        return
    if lag["p95"] > crit:
        alert(db, "P0", "ORDER_LAG", f"订单延迟 P95 {lag['p95']}s 超过 {crit}s")
    elif lag["p95"] > warn:
        alert(db, "P1", "ORDER_LAG", f"订单延迟 P95 {lag['p95']}s 超过 {warn}s")
```


```

### 一个容易被忽略的点：订单链路的速度比完整性更容易崩

完整性靠对账能补。**速度补不回来。**

客户 10:00 付款，你 10:40 才拉到，那 40 分钟里你不知道要发什么货。大促发货时限按支付时间算，不是按你拉到的时间算。

所以订单链路要盯两个指标，不是一个：

| 指标 | 含义 | 大促红线 |
|---|---|---|
| 漏单率 | 本地单数 / 平台单数 | 必须 100%，差一单就是 P0 |
| 延迟 P95 | 支付到入库的时间 | 24h 发货时限 → ≤ 5 分钟 |

**只看漏单率会漏掉最痛的那个问题**：一单没漏，但都晚到 40 分钟，照样发不出去。

## 六、只剩 72 小时，先做哪条

现实一点：如果现在离大促只有三天，三条全做完不现实。按这个顺序砍：

| 优先级 | 做什么 | 工作量 | 为什么先做它 |
|---|---|---|---|
| **1** | 订单单数对账 | 2 小时 | 成本最低、收益最高。漏单是唯一"不可恢复"的事故 |
| **2** | 库存水位熔断 | 3 小时 | 超卖要赔付 + 平台处罚，且影响店铺权重 |
| **3** | 价格到手价核对 | 1 天 | 工作量最大（要拆三层），但错价的损失有上限，且能事后追 |

三条里只有第 1 条是"两小时就能上线"的——一个定时任务，拉平台单数，比本地单数，不等就告警。**如果你今天只能做一件事，做这个。**

价格核对放最后不是因为不重要，是因为它最费工：要把标价、券、满减三层都拆开，还要处理不同类目的规则差异。三天做不完，做一半反而危险（核对逻辑错了会误导运营）。

## 七、T-14 到 T+0 的时间表

这张表是给还没到大促的团队用的，存下来，618 还能用。

| 节点 | 该做什么 | 判断标准 |
|---|---|---|
| **T-14** | 三条链路打通 + 单链路压测 | 每条链路都能在峰值 QPS 下跑通，不含告警 |
| **T-7** | 全链路演练（模拟真实峰值 2 倍） | 演练中出现的每个告警都要有处置动作，不能只"知道了" |
| **T-3** | **代码冻结** | 只修 P0 bug，不加功能。这条最难执行，也最该执行 |
| **T-1** | 值守排班 + 降级预案确认 | 谁在 0 点看屏幕，出事了按哪张表操作，必须写下来 |
| **T+0** | 实时监控 + 手工兜底通道 | 系统挂了要能手导单、手改价，这条通道平时就要演练 |

T-3 的代码冻结值得单独说。大促前三天上线新功能，是事故的经典成因——新代码没有经历过峰值，你不知道它在哪里会崩，而这时候已经没有时间修了。**冻结不是保守，是止损。**

## 八、降级预案：出事时按这张表做，不要临场想

大促值守最怕的不是出事，是出事了大家在群里讨论"要不要关"。

提前把这张表写下来，贴在显示器上：

| 触发条件 | 动作 | 谁来决定 |
|---|---|---|
| 订单接口错误率 > 5%，持续 5 分钟 | 停止自动拉单，切备用数据源；同时开手工导单通道 | 值班开发直接执行，事后报备 |
| 订单延迟 P95 > 15 分钟 | 降低库存/价格核对频率，把配额让给订单 | 值班开发 |
| 库存对账失败（平台值 < 本地值 20% 以上） | 全店限购 1 件，人工核查 | 运营负责人 |
| 价格偏差 > 1% 且影响 SKU > 50 个 | 暂停所有自动调价，转人工 | 运营负责人 |
| 某个平台完全不可用 | 该平台商品全部下架（不是"等它恢复"） | 值班开发 + 运营 |

最后一条最反直觉：**平台挂了，第一反应应该是下架，不是等**。

因为平台挂了但你的商品还在卖，客户下单你收不到单，等平台恢复时你已经积累了一堆超时未发货。下架是止损，等待是赌博。

## 九、三条链路的共同点

写完这三段代码，你会发现它们长得几乎一样：

```python
def guard(expected, actual, threshold):
    if abs(expected - actual) > threshold:
        alert(...)
```

对，核心就是一个比对加一个告警。**大促的稳定性不是靠复杂架构，是靠"每个关键假设都有代码在验证"**。

三条链路对应三个假设：

| 链路 | 被验证的假设 | 假设错了会怎样 |
|---|---|---|
| 价格 | 客户看到的价 = 我算的价 | 资损，且不可追回 |
| 库存 | 平台可售 = 真实可售 | 超卖，赔付 + 处罚 |
| 订单 | 我拉到的单 = 平台所有的单 | 漏发，投诉 + 超时罚款 |

这三个假设，平时错了没人发现（因为量小、能人工兜），大促错了就是事故。

所以大促前该做的不是"优化性能"，是**把这三个假设写进代码，让它每小时替你看一遍**。

上面四段加起来三百多行，纯标准库加 SQLite，不用起 Redis 也不用上消息队列。如果你现在离大促只剩三天，从订单对账开始——两小时能上线，是这三件事里性价比最高的。

需要完整的表结构和告警落库代码，可以找我们要，也可以直接拿这套思路套你们自己的库。大促加油。

# 服务商接单前先看：客户最常要的 8 个数据字段（附 python 源码）

接单最容易亏钱的地方不是报价报低了，是**字段没问清**。

我们踩过一次。客户说"我要商品的销量"，我们说"有"。交付那天他把数据接进选品系统，两周后打电话来：排序是乱的，销量高的排不上去。他要的是**近 30 天销量**，我们给的是**累计销量**——一个卖了三年但最近没人买的老品，累计销量五万，排在新品前面。

这个锅没法甩。合同里写的是"销量"，我们确实提供了"销量"。

后来我们把这件事做成了一张表：**客户说的字段名，和数据库里的字段，是两个东西，中间必须翻译一次**。这篇就把最常见的八个字段的翻译写出来，附一段可以跑的自检代码，接单前先跑一遍。

## 一、八个字段，三类落差

先说结论：这八个字段之所以容易出事，不是因为接口难，是因为它们在三个不同的地方让客户失望。

| 落差类型     | 客户什么时候失望        | 字段             |
| -------- | --------------- | -------------- |
| **口径落差** | 拿到数了，但和以为的不是一回事 | 价格、库存、销量       |
| **形态落差** | 拿到东西了，但塞不进他的系统  | SKU 属性、图片、物流编码 |
| **合规落差** | 拿到东西了，但不能随便用    | 收件信息、商品标题      |

口径落差最危险，因为**数据看起来是对的**。五万销量确实是销量，不会报错，不会异常，只有业务逻辑错了才发现。  
形态落差最费工，报价时按"给个字段"算的，实际要做一次转换层。  
合规落差最容易变成法律问题，也是最容易被服务商忽略的一类。

下面逐个说，每个字段给三样：客户通常怎么说、你该追问什么、最贵的那个坑。

## 二、口径落差：数是对的，但意思不对

### 1. 价格

客户说"价格"的时候，脑子里可能是六种东西之一：

| 客户说的   | 实际对应的字段                      |
| ------ | ---------------------------- |
| 页面价格   | 标价（`price`）                  |
| 现在买多少钱 | 促销价 / 活动价（`promotion_price`） |
| 我实际付多少 | 到手价 = 促销价 − 店铺券 − 平台券 + 运费   |
| 会员多少钱  | 会员价（部分平台单独返回，多数要自己算）         |
| 批发多少钱  | 1688 阶梯价，按起订量取档              |
| 含税多少钱  | 多数平台不含税，需要自己加                |

**该追问的三句**：

1. 含不含券？含不含运费？
2. 会员价要不要单独一列，还是并进到手价？
3. 是给用户看，还是给财务算账？（前者要到手价，后者要标价+优惠明细，能对得上账）

**最贵的坑**：促销价没有时效字段。客户按促销价算了一个月的利润，活动结束了数据还在返回促销价，利润表对不上，回来找你查一周。第 13 篇讲过价格要建模成带 `promo_end` 的对象，这是同一个问题的另一面——**服务商交付时如果只给一个数字，就必须同时给这个数字的有效期**。

### 2. 库存

客户说"库存"，实际可能是：

- **总库存**：仓库里有多少件
- **可售库存**：总库存 − 已占用 − 安全库存，这才是"还能卖几件"
- **锁定库存**：下单未付款占用的
- **在途库存**：采购中，还没入库

这四个数能差出好几倍。做活动备货的客户要总库存，做防超卖的客户要可售库存，两个数给错了都是事故。

**该追问的一句**：你要的是"还剩几件"，还是"还能卖几件"？

**最贵的坑**：1688 没有 SKU 库存这个概念。它的业务模型是按起订量阶梯卖，返回的是商品的**可售状态**，不是一个数字。你用淘宝的思维去接，适配器里会写一句"忽略 sku_id，返回总量"——**调用方以为拿到了 SKU 库存，实际上是错的，而且不报错**。第 8 篇把这叫"假装都支持"，在我看来这是服务商最容易犯的错：不是不会做，是不好意思说不支持。

### 3. 销量

这个字段水最深，因为三个平台给的东西根本不是一回事：

| 平台   | 通常返回                  | 说明                |
| ---- | --------------------- | ----------------- |
| 淘宝   | 累计销量（`sold_quantity`） | 商品生命周期累计，老品数字很大   |
| 京东   | 评价数为主，销量部分类目不返回       | 很多接口根本不给你销量       |
| 1688 | 成交件数 / 成交笔数           | 批发的"笔"和零售的"件"完全不同 |

**该追问的一句**：这个销量你拿去干什么？

答案决定你该给什么：

- **排序/选品** → 用**评价增速**或**近 N 天增量**，比累计销量准得多。累计销量对老品有严重偏袒。
- **备货预测** → 必须近 30 天，而且要和退货率一起给，不然备多了。
- **给客户看**（比如"本店已售 10 万件"）→ 累计销量就够，甚至平台前台展示的那个数就行。

**最贵的坑**：**答应了给"月销"，实际平台只给累计**。这种承诺在售前最容易出口，因为客户问的时候你觉得"销量不就是销量"。交付时你只能自己算：`累计(t) − 累计(t−30d)`。这需要你从接单第一天就开始存历史快照——**如果你第三个月才想起来算，前三个月的数据永远补不回来**。

## 三、形态落差：东西给了，但塞不进去

### 4. SKU 属性

客户说"我要颜色和尺码两列"，听起来像是在表格上加两列。实际拿到的是这个：

```
1627207:28320:颜色分类:黑色;1627208:34117:尺码:M
```

这是淘宝的属性串格式：`属性ID:值ID:属性名:值`，多组用分号隔开。京东是另一套结构（对象数组），1688 又是另一套（带 `skuId` 和自定义属性）。

三个平台三种给法，这就是"必须做归一化"的全部理由——不是洁癖，是客户的 ERP 只认两列。

**该追问的两句**：

1. 要"颜色/尺码"这两列，还是要完整属性集（含材质、产地、执行标准）？
2. 属性值要不要标准化？客户系统里是"黑色"，平台给的是"黑色/深黑/曜石黑"，不统一就合并不了。

**最贵的坑**：**属性值不稳定**。同一个商品，这次返回"黑色"，下次商家改了叫"经典黑"，你的 SKU 匹配逻辑就断了。做组合装、做对账的客户会遇到"同一个 SKU 变成两个"。

### 5. 图片

客户说"给我图片链接"，他以为拿到 URL 就能用。实际有三个坑：

1. **防盗链**：1688 的图片直接外链会 403，必须转存到自己的存储
2. **尺寸**：主图有 800×800、1200×1200 多个规格，详情图是长图（可能 750×8000）
3. **有效期**：部分 CDN 链接带签名参数，过期就失效

**该追问的一句**：你是要**链接**，还是要**文件**？

要链接，你得保证这个链接在他那边能打开、能长期打开；要文件，你得做转存，这是存储成本和带宽成本，报价时必须算进去。

**最贵的坑**：交付了链接，客户上线一个月后图片全挂。原因是 CDN 签名过期。这种事不报错，就是图片一个个变成灰块，客户发现的时候往往是客户投诉商品页打不开。

### 6. 物流编码

客户说"物流状态"，实际拿到的是 `WAIT_SELLER_SEND_GOODS`、`STOP_ONE_STAGE`、`GOT` 这一堆东西，三个平台三套编码，同一个"已揽收"在三家可能叫三个名字。

更要命的是**节点的粒度不一样**：有的平台给你 7 个状态，有的给你 20 个子状态（运输中、到达分拨中心、离开分拨中心……）。

**该追问的一句**：你要几个状态？

做客服系统的客户要细（"包裹现在在哪"），做订单看板的客户要粗（"发出/在途/签收"）。给粗了客服答不了，给细了客户系统没有对应字段，会全部落到"其他"。

**最贵的坑**：**只给原始编码，不给映射后的标准状态**。客户得自己写映射表，写完还得跟着平台改。这个工作量不小，而且一旦映射错了，"已签收"被当成"在途"，自动确认收货的逻辑就永远不触发。

## 四、合规落差：拿到了，但不能随便用

### 7. 收件信息

这个字段是八个里唯一会让你承担法律责任的。

客户说"我要订单的收件人姓名、电话、地址"，多数平台返回的是**脱敏值**：

```
收件人：张*
手机：138****1234
地址：浙江省杭州市西湖区****
```

要拿到明文，得走**解密接口**，而且要单独申请权限、有调用额度限制、有审计日志。这不是技术门槛，是平台在履行个人信息保护义务。

**该追问的两句**：

1. 你要明文干什么？（发货打单可以，做用户画像不行）
2. 你有没有做过去标识化处理？

**最贵的坑**：服务商把明文收件信息落库、落日志、甚至缓存到 Redis 里，客户的数据泄露了，追责链条上是**你**。第 18 篇讲数据服务的风险传导，这一条是最短的那根引线：你手里有十万条真实手机号，出事的时候没有"我只是个技术供应商"这种说法。

建议做法：明文只在内存里过一遍，直接推给客户的打单接口，**不落库、不打日志**。日志里要有 `138****1234` 这种打码，而不是全量。

### 8. 商品标题

客户说"标题照搬过来就行"，你照搬了，然后被平台驳回。

标题里的雷有四个：

1. **极限词**："最"、"第一"、"国家级"——违反广告法，平台会拦
2. **平台痕迹**："淘宝爆款"、"天猫同款"、"京东专供"——跨平台铺货必挂
3. **促销词**："【限时秒杀】"、"满 199 减 50"——价格变了标题没变，属于虚假宣传
4. **他人商标**：品牌词没授权就写上去

**该追问的一句**：这个标题你是要**原样展示**，还是要**重新发布**？

原样展示（做比价、做监控）没问题，但要注意展示合规；重新发布（铺货）必须清洗，第 5 篇那套正则就是干这个的。

**最贵的坑**：清洗时**删词不删句**。"【爆款】不锈钢保温杯 500ml 大容量"，删掉"【爆款】"就好，别把整句重写——重写会丢规格词，500ml 变成"大容量"，客户那边同款匹配全乱。

## 五、代码：字段契约 + 填充率探测

上面八个字段的问题是"说不清"，代码解决的是"说不清就没法验收"。

思路：把每个字段写成一份**契约**（客户可能怎么叫、对应哪几个实际字段、口径是什么、填充率下限多少），然后拿真实数据跑一遍，输出一份可以直接发给客户的报告。

### 5.1 字段契约

```python
# contract.py
"""字段契约：接单前和客户确认的内容，落成代码。

每条契约四件事：
  aliases    客户可能怎么叫（用于把需求清单映射到契约）
  candidates 我方数据里可能对应的字段路径（按优先级）
  semantic   口径说明——这句要写进合同附件
  min_fill   填充率下限，低于此值必须在报价前告知客户
"""

FIELDS = {
    "price": {
        "label": "价格",
        "aliases": ["价格", "售价", "单价", "页面价", "标价"],
        "candidates": ["price", "sku.price", "sku_list[0].price"],
        "semantic": "商品标价，不含优惠券、不含运费；活动价见 promotion_price",
        "must_ask": ["含不含券", "含不含运费", "会员价是否单列"],
        "min_fill": 0.99,
    },
    "promo_price": {
        "label": "促销价",
        "aliases": ["活动价", "到手价", "现价", "最低价"],
        "candidates": ["promotion_price", "sku.promotion_price"],
        "semantic": "活动期内的成交价，必须与 promo_end 一起交付",
        "must_ask": ["活动结束时间要不要一起给"],
        "min_fill": 0.60,      # 没活动的商品本来就没有，60% 是正常的
    },
    "stock": {
        "label": "库存",
        "aliases": ["库存", "剩余", "可售", "数量"],
        "candidates": ["stock", "quantity", "sku.quantity"],
        "semantic": "可售库存（已扣占用与安全库存），非仓库总库存",
        "must_ask": ["要总库存还是可售", "1688 无 SKU 库存，是否接受商品级"],
        "min_fill": 0.95,
    },
    "sales": {
        "label": "销量",
        "aliases": ["销量", "月销", "已售", "成交"],
        "candidates": ["sold_quantity", "sales", "volume"],
        "semantic": "累计销量，非近 30 天；近 30 天需历史快照计算",
        "must_ask": ["用来排序还是备货", "能否接受累计口径"],
        "min_fill": 0.80,
    },
    "sku_props": {
        "label": "SKU 属性",
        "aliases": ["规格", "颜色尺码", "SKU", "属性"],
        "candidates": ["skus", "sku_list", "props"],
        "semantic": "已归一为 {spec_name: spec_value} 结构；原始属性串另存 raw_props",
        "must_ask": ["要两列还是要完整属性集", "属性值是否需标准化"],
        "min_fill": 0.90,
    },
    "images": {
        "label": "图片",
        "aliases": ["主图", "详情图", "图片", "素材"],
        "candidates": ["images", "pic_url", "item_imgs"],
        "semantic": "已转存至自有存储的直链，非平台原始 CDN 地址",
        "must_ask": ["要链接还是要文件", "主图还是详情长图"],
        "min_fill": 0.98,
    },
    "logistics_status": {
        "label": "物流状态",
        "aliases": ["物流", "轨迹", "快递状态"],
        "candidates": ["status", "trace_list[-1].status"],
        "semantic": "已映射为 5 个标准状态；原始编码见 raw_status",
        "must_ask": ["要几个状态粒度", "要不要节点时间"],
        "min_fill": 0.85,
    },
    "receiver": {
        "label": "收件信息",
        "aliases": ["地址", "收件人", "电话", "收货信息"],
        "candidates": ["receiver_name", "receiver_mobile", "receiver_address"],
        "semantic": "默认脱敏值；明文需单独授权，仅透传不落库",
        "must_ask": ["要明文做什么", "是否接受脱敏"],
        "min_fill": 1.00,
    },
    "title": {
        "label": "商品标题",
        "aliases": ["标题", "名称", "商品名"],
        "candidates": ["title", "name"],
        "semantic": "原始标题，含营销词与平台痕迹；用于重新发布需清洗",
        "must_ask": ["原样展示还是重新发布"],
        "min_fill": 0.99,
    },
}
```

（八个字段，契约里为什么有九条：`price` 和 `promo_price` 是同一个字段的两种口径，拆开写是因为**客户说"价格"的时候，你必须先问清是哪一个**。）

### 5.2 填充率探测

这段是整篇最实用的部分。多数人接单只验证"接口通不通、字段在不在"，但真正让客户失望的是**字段在，值是空的**。

```python
# probe.py
"""拿真实样本跑一遍，输出填充率、口径分布、字段可用性。

用法：
    python probe.py --file sample.json        # 本地样本回放
    python probe.py --file sample.json --md   # 输出 markdown 报告

样本格式：一个 JSON 数组，每个元素是一个商品/订单的原始返回。
"""

import argparse
import json
from collections import Counter
from decimal import Decimal, InvalidOperation
from pathlib import Path

from contract import FIELDS


MISSING = object()

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def pick(obj, path):
    """按 "sku_list[0].price" 取值，取不到返回 MISSING（区别于 None）

    为什么区分 MISSING 和 None：
      key 不存在 = 我方没给这个字段（或没权限）
      key 存在但值是 null = 平台没给（该商品不适用）
    这两种在验收时是两回事，混在一起客服会答错。
    """
    cur = obj
    for seg in path.split("."):
        if "[" in seg:
            name, idx = seg.rstrip("]").split("[")
            cur = cur.get(name, MISSING) if isinstance(cur, dict) else MISSING
            if cur is MISSING:
                return MISSING
            try:
                cur = cur[int(idx)]
            except (IndexError, TypeError):
                return MISSING
        else:
            if not isinstance(cur, dict) or seg not in cur:
                return MISSING
            cur = cur[seg]
    return cur


def filled(v):
    """空值的判定比想象中麻烦：0 是有效值，空串不是"""
    if v is MISSING or v is None:
        return False
    if isinstance(v, str):
        return v.strip() != ""
    if isinstance(v, (list, dict)):
        return len(v) > 0
    return True


def normalize_price(v):
    """价格比较必须用 Decimal，float 会把 19.9 和 19.90 判成不同"""
    try:
        return Decimal(str(v)).quantize(Decimal("0.01"))
    except (InvalidOperation, TypeError, ValueError):
        return None


def probe(samples, field_key):
    spec = FIELDS[field_key]
    stats = {p: 0 for p in spec["candidates"]}
    samples_with_any = 0
    values = Counter()

    for row in samples:
        hit = False
        for path in spec["candidates"]:
            v = pick(row, path)
            if filled(v):
                stats[path] += 1
                hit = True
                if field_key in ("price", "promo_price"):
                    d = normalize_price(v)
                    values[str(d) if d else str(v)] += 1
                elif field_key == "sku_props" and isinstance(v, list):
                    values[f"{len(v)}个SKU"] += 1
        if hit:
            samples_with_any += 1

    total = len(samples) or 1
    fill = samples_with_any / total

    # 只有命中率接近才叫"多口径并存"，否则就是主备关系
    live = {p: c for p, c in stats.items() if c > 0}

    return {
        "label": spec["label"],
        "fill": fill,
        "ok": fill >= spec["min_fill"],
        "min_fill": spec["min_fill"],
        "hit": live,
        "distinct": len(values),
        "top_values": values.most_common(3),
        "semantic": spec["semantic"],
        "must_ask": spec["must_ask"],
    }


def stability(a, b, field_key):
    """同一批样本拉两次，看字段值变了多少。

    变了不一定是坏事（价格本来就该变），但要知道是哪个字段在变：
      title / sku_props 频繁变 = 商家在改，你的匹配逻辑会断
      price 变 = 正常，但交付时必须带更新时间
    """
    spec = FIELDS[field_key]
    path = spec["candidates"][0]
    changed = 0
    compared = 0
    for x, y in zip(a, b):
        vx, vy = pick(x, path), pick(y, path)
        if not filled(vx) or not filled(vy):
            continue
        compared += 1
        if vx != vy:
            changed += 1
    return changed / compared if compared else 0.0


def render(results, total_samples, n_days=None):
    lines = [
        f"# 字段可用性报告（样本 {total_samples} 条）",
        "",
        "| 字段 | 填充率 | 门槛 | 判定 | 备注 |",
        "|---|---|---|---|---|",
    ]
    for r in results:
        mark = "通过" if r["ok"] else "**不足**"
        if not r["ok"]:
            note = "需在报价前告知客户"
        elif r["distinct"] > 1:
            note = f"{r['distinct']} 种取值"      # 口径可能不唯一，值得问一句
        else:
            note = ""
        if r.get("drift", 0) > 0.3:
            note += f"；两次抽样差异 {r['drift']:.0%}"
        lines.append(
            f"| {r['label']} | {r['fill']:.1%} | {r['min_fill']:.0%} | {mark} | {note} |"
        )

    bad = [r for r in results if not r["ok"]]
    if bad:
        lines += ["", "## 需要提前说明的字段", ""]
        for r in bad:
            lines.append(f"**{r['label']}** 填充率 {r['fill']:.1%}，低于 {r['min_fill']:.0%}。")
            lines.append(f"- 口径：{r['semantic']}")
            lines.append(f"- 必须确认：{'；'.join(r['must_ask'])}")
            lines.append("")

    lines += ["", "## 各字段口径", ""]
    for r in results:
        lines.append(f"- **{r['label']}**：{r['semantic']}")
    return "\n".join(lines)


def _pad(text, width):
    """中文占两格。str.ljust 按字符数补空格，表格会歪，必须按显示宽度算"""
    w = sum(2 if ord(c) > 127 else 1 for c in text)
    return text + " " * max(0, width - w)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--file", required=True, help="样本 JSON 文件")
    ap.add_argument("--compare", help="第二次抽样，用于比对字段漂移")
    ap.add_argument("--md", action="store_true", help="输出 markdown 报告")
    ap.add_argument("--out", default="field_report.md")
    args = ap.parse_args()

    samples = json.loads(Path(args.file).read_text(encoding="utf-8"))
    if not isinstance(samples, list) or not samples:
        raise SystemExit("样本必须是 JSON 数组，且不能为空")

    results = [probe(samples, k) for k in FIELDS]

    if args.compare:
        later = json.loads(Path(args.compare).read_text(encoding="utf-8"))
        for r, k in zip(results, FIELDS):
            r["drift"] = stability(samples, later, k)

    if args.md:
        Path(args.out).write_text(render(results, len(samples)), encoding="utf-8")
        print(f"已写出 {args.out}（{len(samples)} 条样本）")
        return

    print(_pad("字段", 14) + _pad("填充率", 10) + _pad("门槛", 8) + "判定")
    print("-" * 46)
    for r in results:
        flag = "OK" if r["ok"] else "LOW"
        line = (_pad(r["label"], 14) + _pad(f"{r['fill']:.1%}", 10)
                + _pad(f"{r['min_fill']:.0%}", 8) + flag)
        if "drift" in r:
            line += f"   漂移 {r['drift']:.0%}"
        print(line)
        if not r["ok"]:
            print(f"  └ {r['semantic']}")


if __name__ == "__main__":
    main()
```


```

### 5.3 怎么用

三步，接单前跑完，只要半天：

```bash
# 1. 从客户的目标商品里抽 200 条，存成 sample.json
python dump_samples.py --shop shop_123 --limit 200 > sample.json

# 2. 跑一遍，先看哪些字段填不满
python probe.py --file sample.json

# 3. 隔一天再抽一次，看哪些字段在漂（可选，但强烈建议）
python probe.py --file sample.json --compare sample_day2.json

# 4. 生成报告，直接发给客户确认
python probe.py --file sample.json --md --out 字段确认单.md
```

输出大概是这样：

```
字段              填充率      门槛  判定
----------------------------------------
价格              100.0%      99%  OK
促销价             62.5%      60%  OK
库存               98.5%      95%  OK
销量               41.0%      80%  LOW
  └ 累计销量，非近 30 天；近 30 天需历史快照计算
SKU 属性           93.0%      90%  OK
图片               99.5%      98%  OK
物流状态           86.0%      85%  OK
收件信息          100.0%     100%  OK
商品标题          100.0%      99%  OK
```

那个 41% 就是这篇的价值。**如果你在报价前跑出这个数，就可以跟客户说"销量这个字段有一半商品拿不到，我们建议用评价增速替代，或者加价做历史快照"**。如果等交付了才发现，客户只会记住"你说有销量的"。

顺带一个观察：跑完这份报告，你会发现自己对数据的了解比读三天文档还多。文档告诉你字段存在，报告告诉你字段**有值**。

## 六、接单前该给客户的三张纸

代码解决的是"我知道"，三张纸解决的是"客户知道"。少一张，后面就会有一次扯皮。

**第一张：字段确认单**

就是上面代码生成的那份报告，每个字段要写清四件事：口径（一句话）、示例值（真的从数据里取的）、填充率、更新频率。**示例值必须是真的**，从样本里截一段；自己编的示例，客户拿去验证会发现对不上。

**第二张：不保证清单**

这张最容易被跳过，也最保命。写清楚哪些事不承诺：

- 平台字段变更不另行通知，但我们会在 24 小时内适配
- 部分类目的可选属性，不保证有值
- 历史数据只能从接入日开始（除非另行采购）
- 平台侧的限流、审核、政策变化导致的延迟，不算 SLA 违约

**第三张：验收标准**

不要写"以客户满意为准"，要写成可测的：

- 字段准确率 ≥ 99.5%（抽 50 条人工比对）
- 主键准确率 100%（`item_id` / `order_id` 不能错，错了就是数据串店）
- 填充率达到确认单标注的门槛
- 数据延迟 ≤ N 分钟（P95）

这三张纸加起来不到两页，但它们把"你当时说的是这个意思"这类争执，提前到签约之前解决。

## 七、最后一句

这八个字段里，没有一个技术上难。它们难在**客户用的词和系统里的字段之间，隔着一层业务理解**。

服务商真正卖的不是接口访问权，是这层翻译。翻译得好，客户觉得你懂他的业务；翻译得不好，你会一直在解释"我给你的确实是销量"。

而翻译这件事，是可以提前做、并且用代码验证的。上面那段 `probe.py` 两百行出头，但它比任何一份字段文档都更能避免交付期的争执——**因为它测的不是字段存不存在，是字段有没有值，值是什么口径**。

如果你正在评估要不要自己接这三个平台，或者手上有一批客户需求要报，可以拿这套契约模板去套一遍。跑完你会知道哪些能承诺、哪些要提前说清——这个清单本身就是报价的依据。

有具体的字段拿不准口径，也可以直接问我们，我们把三平台的对照表摊开一起看。