В данной статье мы поговорим о важности ...
注册账号免费测试1688API接口1688 的 item_review 接口（官方标准方法名：alibaba.item.review.get）是阿里巴巴开放平台提供的核心接口之一，用于获取指定商品的买家评论数据，包括文本评价、评分、晒图、采购属性（如采购数量、...
还有 85 人发表了评论 加入10726人围观未分类
1688 选品 + 淘宝比价 + 京东竞品监控，一条数据流跑起来（附 python 源码）
万邦科技Lex 发表于2026-10-30 浏览7 评论0
# 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 源码）
万邦科技Lex 发表于2026-10-30 浏览7 评论0
# 电商 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 少写一半适配代码。评论区或私信聊。*未分类
鉴权、签名、字段映射：三平台对接最容易翻车的三个环节（附 python 源码）
万邦科技Lex 发表于2026-10-29 浏览35 评论0
# 鉴权、签名、字段映射：三平台对接最容易翻车的三个环节（附 python 源码）
> 接淘宝、1688、京东的开放平台接口，我们团队前后踩过的坑能写一本书。但真要按「浪费的人天」排个序，前三名毫无悬念：**鉴权、签名、字段映射**。
>
> 这三个环节有个共同的恶毒之处：**报错信息都不说人话**。淘宝返回 `Invalid signature`，京东返回 `signature error`，1688 返回 `error_code: 40`——没有一个告诉你错在哪一个参数、哪一步、哪个字符。很多人卡一天半，最后发现是时间戳用了秒而不是毫秒。
>
> 这篇不讲怎么接入（第 6、11 篇讲过），讲的是**翻车之后怎么定位**。建议先收藏，翻车的时候拿出来查。
## 一、为什么这三个环节最坑：错误都在「你自己的代码」和「平台的报错」之间
数据接口报错，通常是业务问题——库存不足、商品不存在，一眼能懂。而这三个环节的错误发生在**请求离开你的机器之前**，平台收到的东西和你以为发出去的东西**不是同一份**：
- 你以为发了 `timestamp=1730000000`，实际签名时用的是另一个值；
- 你以为参数按字典序排了，实际有一个 key 多了个空格；
- 你以为字段是整数，平台按字符串解析，静默截断。
**排查的核心方法论只有一句话：别猜，把「你实际发出去的东西」打出来，和「平台期望的东西」逐字符比对。** 后文所有代码都是围绕这句话展开的。
先给一张总览表，后面三节逐个展开：
| 环节 | 典型报错 | 报错的真实含义 | 平均排查耗时（我们的工单统计） |
| ---- | ------------------------------------------- | ------------------------------------------ | --------------- |
| 鉴权 | `session expired` / `lack of authorization` | token 过期、session 与 appkey 不匹配、token 没跟着请求走 | 2 小时 |
| 签名 | `Invalid signature` / `error_code: 40` | 参数没排对序、时间戳单位错、中文被转义、空值参与签名 | **1.5 天** |
| 字段映射 | 无报错（最危险） | 价格少两位、状态映射错、时间差 8 小时 | 3 天起，且常由客户先发现 |
注意字段映射那一行——**它经常没有报错**。这是三个环节里唯一可能酿成资损的一类，放最后一节重点讲。
## 二、鉴权：三种 token，三种死法
三平台的授权模型长得像，细节全不一样。先列差异：
| 平台 | 凭证结构 | 有效期 | 过期前能刷新吗 |
| ---- | ---------------------------------- | ----------------------- | ------------------------ |
| 淘宝 | `appkey` + `sessionKey`（绑定商家） | 长期但可被商家主动失效 | 会话失效后需商家重新授权，**无法程序化刷新** |
| 京东 | `app_key` + `access_token`（OAuth2） | 通常 1 年，refresh_token 更长 | 可以，用 refresh_token 换新 |
| 1688 | `appkey` + `access_token`（OAuth2） | 约 10 小时（各应用不同） | 可以，refresh_token 换新 |
**死法一：混淆 app 级凭证和用户级凭证。**
`appkey/secret` 只代表「你的应用」，调用商品、订单接口还必须带上「商家授权」的 token。最经典的事故：测试环境用 A 商家的 session，生产环境忘了换，结果测试数据全对、上线后拉的全是另一个店铺的商品——**接口返回 200，数据是别人的**。所以凭证要「应用 × 商家」二元组建模，别把 sessionKey 存成全局配置。
**死法二：拿本地时钟判断 token 过期。**
token 的过期时间是平台返回的 `expires_in`（剩余秒数），如果你存的是 `本地 now + expires_in`，服务器时钟漂移 5 分钟，token 就会提前或延后 5 分钟失效。**必须存「到期时间点」，且以收到响应那一刻为基准**，再留安全余量。
**死法三：淘宝的「商家悄悄解绑」。**
淘宝 session 没有程序化刷新路径，商家在千牛后台解除授权后，你的 token 立刻全挂，且**没有提前通知**。这是淘宝接入最阴的一雷。解法是健康探测（第 9 篇的 PROBES 思路）+ 授权状态监控，挂了第一时间找到商家重新授权。
代码上，一个最小可用的 `TokenManager`：
```python
# auth/tokens.py
import asyncio
import time
from dataclasses import dataclass, field
@dataclass
class Token:
value: str
refresh_value: str | None
expires_at: float # 平台响应到达那一刻 + expires_in，时间点而非时长
refresh_expires_at: float | None = None
def needs_refresh(self, margin_s: int = 600) -> bool:
"""提前 10 分钟就判为「该刷了」，余量吃掉网络延迟和时钟误差"""
return time.time() >= self.expires_at - margin_s
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class TokenManager:
"""按 (platform, seller_id) 二元组管理，杜绝串店铺"""
def __init__(self, store):
self.store = store # 持久层：sqlite / redis，重启不丢
self._locks: dict[str, asyncio.Lock] = {}
self._mem: dict[str, Token] = {}
def _lock(self, key: str) -> asyncio.Lock:
if key not in self._locks:
self._locks[key] = asyncio.Lock()
return self._locks[key]
async def get(self, platform: str, seller_id: str) -> str:
key = f"{platform}:{seller_id}"
tok = self._mem.get(key) or await self.store.load(key)
if tok is None:
raise AuthError(f"{key} 无凭证，需要商家重新授权")
if tok.needs_refresh():
async with self._lock(key): # 单飞：并发请求只触发一次刷新
tok = self._mem.get(key) or tok
if tok.needs_refresh(): # 双检：拿锁后再确认一次
try:
tok = await self._refresh(platform, seller_id, tok)
except RefreshFailed:
# refresh_token 也失效：只能走重新授权，主动告警而不是等调用报错
await self.store.mark_broken(key)
raise AuthError(f"{key} refresh_token 已失效，需商家重新授权")
return tok.value
async def _refresh(self, platform, seller_id, tok) -> Token:
if platform == "taobao":
raise RefreshFailed("淘宝无程序化刷新，走重新授权流程")
# 京东 / 1688：POST token 端点，grant_type=refresh_token
resp = await oauth_refresh(platform, tok.refresh_value)
new = Token(
value=resp["access_token"],
refresh_value=resp.get("refresh_token", tok.refresh_value),
expires_at=time.time() + int(resp["expires_in"]),
refresh_expires_at=time.time() + int(resp.get("refresh_token_valid_time", 0)),
)
key = f"{platform}:{seller_id}"
self._mem[key] = new
await self.store.save(key, new)
return new
```
三个细节值得注意：
1. **`needs_refresh` 提前 10 分钟**——不是 token 到期才刷，余量吃掉时钟误差和刷新请求本身失败后的重试空间。
2. **单飞锁 + 双检**——高并发下 100 个请求同时发现 token 过期，只允许 1 个真正去刷新，其余等结果。没有锁就是「并发刷新撞车」：后发的刷新会把先发的 token 作废（京东和 1688 都有这个行为），自己把自己刷死。多店铺规模上去之后这块还要加密存储和分店隔离，是下一篇的主题。
3. **淘宝直接抛 `RefreshFailed`**——诚实地告诉调用方「这条路径不存在」，别写个永远成功的假刷新。
## 三、签名：报错只说错了，不说错在哪
签名错误是接入期**单点耗时最长**的环节。原因很简单：平台不会告诉你它算签名的输入是什么，你只能猜自己哪个参数「脏」了。
先给三平台签名算法速查（实现细节第 11 篇有完整代码，这里只列排障需要的差异点）：
| 平台 | 算法 | 关键差异 | 最高频错误 |
| ---- | ---------------------------- | ---------------------------- | -------------------- |
| 淘宝 | MD5（`secret + 排序串 + secret`） | GET/POST 均按参数排序拼 | 参数没按 ASCII 排序 |
| 京东 | MD5（`secret + 排序串 + secret`） | 业务参数整体塞进 `360buy_param_json` | JSON 中文被转义成 `\uXXXX` |
| 1688 | HMAC-SHA1 | **urlPath 参与签名** + **毫秒时间戳** | 时间戳用了秒 |
### 签名错误 Top 5（按我们工单频次排序）
**① 时间戳单位或时区错。** 1688 要毫秒，淘宝/京东用秒也能过但会有诡异边界；服务器是 UTC 时区的话，某些按「分钟窗口」校验时间戳的接口直接全挂。**解法：时间戳永远显式生成，别用框架默认值。**
**② 参数没按 ASCII 字典序排。** Python 的 `sorted()` 恰好就是 ASCII 序，所以纯 Python 代码很少踩；踩坑的都是「从别的语言抄的签名代码」或者「自己手写了拼接顺序」。
**③ 中文被转义。** 京东最典型：`360buy_param_json` 里的中文商品名，`json.dumps` 默认 `ensure_ascii=True` 会转成 `\uXXXX`——签名按转义后的算，服务端按原文验，**永远对不上**。一行 `ensure_ascii=False` 解决，但不知道的时候能卡一天半。
**④ 空值参与了签名。** 某些平台规定空参数不参与签名，你把 `""` 也拼进去了。规则各平台不同，**拿不准就试：先全拼，报错再去掉空值试一次**。
**⑤ secret 泄露后的轮换灾难。** secret 提交到公开仓库后去平台后台重置，结果忘了旧 secret 还配在测试机 cron 里——旧任务报签名错，新任务正常，两边互相干扰极难定位。**secret 必须集中存放、一处引用**。
### 排查方法：把「实际发出去的」打出来比对
下面这个 `sign_debug.py` 是我们内部排查签名问题的工具，思路是：**本地复算签名，同时把参与签名的 canonical string 完整打出来**。拿它和平台的报错对照，90% 的签名错误肉眼就能看出来：
```python
# sign_debug.py —— 签名排障工具：复算 + canonical string 可视化
import hashlib
import hmac
import json
import time
from urllib.parse import quote
def canonical(params: dict) -> str:
"""淘宝/京东式 canonical string：ASCII 序 k=v 直接相连，跳过空值和签名本身"""
pairs = sorted((k, v) for k, v in params.items()
if v not in ("", None) and k != "sign")
return "".join(f"{k}{v}" for k, v in pairs)
def sign_taobao(params: dict, secret: str) -> str:
raw = f"{secret}{canonical(params)}{secret}"
return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
def sign_jd(params: dict, secret: str, body: dict | None = None) -> str:
if body is not None:
# 关键：ensure_ascii=False，中文按原文参与签名
params = {**params, "360buy_param_json": json.dumps(
body, ensure_ascii=False, separators=(",", ":"))}
raw = f"{secret}{canonical(params)}{secret}"
return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
def sign_alibaba(params: dict, secret: str, url_path: str) -> str:
# 1688：urlPath + 排序串，HMAC-SHA1，毫秒时间戳
params = {**params, "_aop_timestamp": str(int(time.time() * 1000))}
ordered = sorted((k, v) for k, v in params.items() if v not in ("", None))
raw = url_path + "".join(f"{k}{v}" for k, v in ordered)
digest = hmac.new(secret.encode(), raw.encode(), hashlib.sha1).hexdigest().upper()
return digest
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def debug_dump(name: str, params: dict, secret: str, url_path: str = ""):
"""把参与签名的每一步都打出来，和平台的报错对照"""
cs = url_path + canonical(params)
print(f"=== {name} ===")
for k, v in sorted(params.items()):
print(f" {k!r:28} -> {v!r}") # repr 能暴露首尾空格、全角字符
print(f" canonical ({len(cs)} chars):")
print(f" {cs}")
print(f" sign: {sign_taobao(params, secret)}")
if __name__ == "__main__":
debug_dump("taobao.item.get", {
"method": "taobao.item.get",
"app_key": "your_appkey",
"timestamp": "2026-10-29 15:00:00",
"num_iid": "683625971234",
"fields": "num_iid,title,price",
"session": "your_session",
"note ": "value ends with a space", # 故意放一个带尾空格的参数做演示
}, "your_secret")
```
用法：报签名错的时候，把**你实际发请求用的那份参数**（不是你以为的那份——从 httpx 的请求日志或抓包里拿）灌进 `debug_dump`，然后：
1. 看 `repr` 输出里有没有**首尾空格、全角字符、大小写不一致**的 key；
2. 看 canonical string 的长度对不对——多一段少一段，二分着删参数试；
3. 时间戳单独确认单位（秒/毫秒）和格式。
**一个能省 80% 排查时间的习惯**：签名函数写好后立刻用官方文档里的「签名示例」跑一遍断言（三平台文档都附了带预期结果的示例参数），断言不过说明你的实现有问题，别等真请求了才发现。
## 四、字段映射：不报错，只出事
前两个环节的错误平台会告诉你（虽然说得含糊），字段映射的错误**平台根本不知道错了**——你拉回来的就是一份合法响应，只是你读它的方式不对。这类问题的共同特征：**上线数天后由运营或客户发现，排查时谁也不怀疑数据管道**。
### 三类错误 + 一个语义陷阱
**① 数值类型。** 三个平台的价格给法不一样：淘宝给字符串 `"59.90"`，京东给「分」为单位的整数 `5990`，1688 阶梯价给嵌套数组。最恶劣的是 `5990`——它既能当「分」也能当「元」，**错了不报错，只差 100 倍**。铁律：入库前统一转 `Decimal`，转换时必须显式指定来源单位。
**② 时间字段。** 秒还是毫秒、UTC 还是北京时间、字符串还是整数——三平台三样，甚至同平台不同接口都不一样。差 8 小时的时区错误会让「今日订单」报表永远差一半，而财务通常在月底才对出来。铁律：**收到即归一**，全部转成 UTC 毫秒时间戳存库，展示层再转本地。
**③ 状态枚举。** 订单状态三平台完全不同：淘宝 `WAIT_SELLER_SEND_GOODS`，京东 `WAIT_SELLER_STOCK_OUT`，1688 用数字码。直接把平台状态存库，后面每写一个业务逻辑都要写一遍三平台的 if-else。铁律：**入库即映射到自有状态机**（第 3 篇的订单状态机就是那套字典）。
**④ 空值语义陷阱（最隐蔽）。** `"price": null`、`"price": ""`、key 整个不出现——三种「没有价格」的含义完全不同：null 通常是「该字段不适用」，空字符串可能是「平台接口有 bug」，key 缺失可能是「你没申请这个字段的权限」。**不区分这三种，价格同步会把一批「查询权限没申请」的商品当成「无货」处理**，这直接就是资损。
### 声明式映射：把「翻译规则」和数据代码分开
解决思路是老三样（第 2 篇的统一模型），这里给出映射层的可运行骨架，重点是**把单位、时区、枚举、空值四类规则显式声明出来**，而不是散落在业务代码里：
```python
# mapping.py —— 声明式字段映射 + 空值语义
from dataclasses import dataclass
from decimal import Decimal
@dataclass
class FieldSpec:
path: str # 平台响应里的取值路径
to: str # 统一模型字段名
unit: str = "raw" # "raw" | "cent" | "second" | "millisecond"
enum: dict | None = None # 平台状态 -> 自有状态
required: bool = True
def pick(data: dict, path: str):
"""按 a.b.c 取值；区分三种空：键缺失 / null / 空串"""
cur = data
for part in path.split("."):
if not isinstance(cur, dict) or part not in cur:
return Missing.ABSENT # key 缺失：多为权限未申请
cur = cur[part]
if cur is None:
return Missing.NULL # 明确的 null：字段不适用
if cur == "":
return Missing.EMPTY # 空串：疑似平台侧异常，单独上报
return cur
class Missing:
ABSENT = object(); NULL = object(); EMPTY = object()
def normalize(raw, spec: FieldSpec):
v = pick(raw, spec.path)
if isinstance(v, type(Missing.ABSENT)): # 用 is 判定哨兵对象
if spec.required:
raise MappingError(f"{spec.path} 缺失：检查字段权限是否已申请")
return None
if v is Missing.NULL:
return None
if v is Missing.EMPTY:
raise MappingError(f"{spec.path} 为空串：疑似接口异常，人工确认")
if spec.unit == "cent": # 京东价格：分 -> 元
v = Decimal(v) / 100
elif spec.unit == "second": # 秒 -> 毫秒
v = int(v) * 1000
if spec.enum: # 状态映射，未收录的显式报错
if v not in spec.enum:
raise MappingError(f"未收录状态 {v!r}，先补映射表再上线")
v = spec.enum[v]
return v
# ---- 三平台的映射表：规则集中在这里，业务代码只见统一模型 ----
TAOBAO_TRADE = [
FieldSpec("tid", "order_no"),
FieldSpec("created", "created_at", unit="second"),
FieldSpec("status", "status",
enum={"WAIT_SELLER_SEND_GOODS": "TO_SHIP",
"TRADE_FINISHED": "DONE"}),
FieldSpec("payment", "paid_amount"), # 淘宝字符串元
]
JD_TRADE = [
FieldSpec("orderId", "order_no"),
FieldSpec("orderStartTime", "created_at", unit="millisecond"),
FieldSpec("orderState", "status",
enum={"WAIT_SELLER_STOCK_OUT": "TO_SHIP",
"FINISHED": "DONE"}),
FieldSpec("orderPrice", "paid_amount", unit="cent"), # 京东分
]
def map_trade(raw: dict, table: list[FieldSpec]) -> dict:
return {spec.to: normalize(raw, spec) for spec in table}
```
```
这套结构的价值在三个「显式」：
1. **单位显式**——`unit="cent"` 写在映射表里，永远不会有人把京东的价格直接当元用；
2. **枚举显式**——平台新增一个状态码，映射层当场报错拦下，而不是静默落一个未知状态进库；
3. **空值显式**——三种「没有」三种处理，权限没申请的问题在管道层就炸出来，不会变成「无货」。
再配一个 **round-trip 自检**，映射表改坏了 CI 直接红：
```python
def test_roundtrip():
sample = {"orderId": "JD123", "orderStartTime": 1730000000000,
"orderState": "WAIT_SELLER_STOCK_OUT", "orderPrice": 5990}
out = map_trade(sample, JD_TRADE)
assert out["order_no"] == "JD123"
assert out["status"] == "TO_SHIP"
assert out["paid_amount"] == Decimal("59.90") # 分 -> 元，别差 100 倍
```
## 五、翻车速查表：先定位环节，再查原因
最后把三个环节收成一张速查表。翻车的时候**先对症状定环节**，再去对应小节查原因，比盲目重试省一个工作日：
| 症状 | 环节 | 最可能的原因（按概率排序） |
|---|---|---|
| 所有请求统一报鉴权/签名错，昨天还好好的 | 鉴权 | token 过期 / 商家解绑授权 / secret 被轮换 |
| 只有部分商家报鉴权错 | 鉴权 | 个别商家 session 失效，串店铺配置 |
| 签名错，参数肉眼检查全对 | 签名 | 中文被转义（京东）/ 时间戳单位（1688 毫秒）|
| 签名错，加一个参数就开始报错 | 签名 | 新参数含特殊字符或空值参与规则不同 |
| 无报错，但数据「看起来不对」 | 字段映射 | 价格单位（分/元）、时间时区（8 小时）|
| 某类商品的数据总是缺字段 | 字段映射 | 字段权限未申请，pick 到 ABSENT |
| 报表数字对不上，接口日志全绿 | 字段映射 | 状态映射漏了新枚举 / 空值被当 0 |
三个环节的共同预防措施其实就一条：**接入期写冒烟测试（第 6 篇的 9 项），把签名断言、token 刷新、字段映射 round-trip 都放进 CI**。翻车不可怕，可怕的是翻车发生在客户通知你之后。
---
这三件事做扎实，接入期 80% 的坑就避掉了。剩下的坑集中在**多店铺授权管理**——token 加密存储、几十家店并发刷新撞车、单店过期不拖累其他店——那是下一篇的主题，也是整个接入系列的收尾。
如果不想自己踩这些坑，也可以直接用我们封装好的接口服务，鉴权续期、签名托管、字段归一这些都是现成的，欢迎评论区或私信聊。
---
*系列文章：上一篇《商品详情、SKU、价格、库存，四个高频接口的正确调用姿势》讲了怎么调；这篇讲接入时怎么不翻车。*未分类
商品详情、SKU、价格、库存，四个高频接口的正确调用姿势（附 Python 源码）
万邦科技Lex 发表于2026-10-29 浏览39 评论0
# 商品详情、SKU、价格、库存，四个高频接口的正确调用姿势（附 Python 源码）
先给一个数字：这四个接口，占我们平台**总调用量的 85%**。
这个占比意味着两件事。第一，任何调用姿势上的坏习惯，都会被放大 85%——配额浪费、数据错位、深夜告警，基本都出在这四个接口上。第二，优化它们的收益也是 85% 的，比优化任何冷门接口都值。
这篇是手册型文章，和第 11 篇的对照表配套使用：那篇讲三个平台"有什么不同"，这篇讲"每个接口该怎么调"。四个接口各一份错误姿势清单，最后一个总纲。
---
## 〇、总纲：四个接口不是四件事，是一棵树
最常见的错误，是把它们当成四个平级接口、用同一个轮询周期去调。
但它们有严格的依赖关系：
```
商品详情（根）
└── SKU 集合（挂在商品上）
├── 价格（挂在 SKU 上）
└── 库存（挂在 SKU 上）
```
这个依赖关系推出两条铁律：
**铁律一：按变更频率分级调度。** 详情一天变不了几次，库存每秒都在变。捆在一个轮询周期里，要么详情白调（浪费配额），要么库存延迟（业务事故）。
**铁律二：能用一次详情带出的，不要拆四次请求。** 三个平台的详情接口几乎都会返回 SKU 列表和标价——一次请求拿三样东西，这是最大的配额节省。真正需要单独调的只有库存（它要实时）。
| 数据 | 变更频率 | 来源 | 同步方式 |
| -------- | ---- | ----------- | ------------- |
| 标题、主图、类目 | 天级 | 详情接口 | 每日增量 |
| SKU 集合 | 小时级 | 详情接口内 | 集合 diff |
| 价格 | 分钟级 | 详情 / 价格接口 | 定时 + 改价回执 |
| 库存 | 秒级 | 库存接口 / 消息推送 | **事件优先，轮询兜底** |
下面四个接口的"正确姿势"，全部从这两条铁律推出来。
---
## 一、商品详情：它是根，不是流水账
详情接口的错误姿势，集中在"把它当唯一的真相来源"。
| 错误姿势 | 后果 | 正确姿势 |
| ---------------- | ---------------------- | -------------------- |
| 全量轮询所有商品详情 | 配额爆炸，且 90% 请求拿回的是没变的数据 | 列表接口增量筛出变更 ID，再按需调详情 |
| 不传 `fields`，全字段拉 | 响应体积差 10 倍，解析和带宽都翻倍 | 按需声明字段，用不到的不拉 |
| 把详情里的价格/库存当业务数据源 | 数据滞后几分钟到几小时 | 详情只管"慢字段"，快字段走分级调度 |
| 下架商品按正常结构解析 | 静默拿到空数据或 KeyError | 先判状态位，软删除单独处理 |
四条里最贵的是第二条。很多人不知道详情接口支持字段裁剪，把整个响应体解析一遍，结果：**每次多传 80% 用不到的字节，配额按调用次数算倒是不亏，但带宽、解析时间、落库量全都翻倍**。字段白名单就一行的事：
```python
DETAIL_FIELDS = [
"num_iid", "title", "pic_url", "cid",
"skus", "price", # SKU 集合和标价跟着详情一起拿（铁律二）
"list_time", "delist_time",
]
params = {"num_iid": item_id, "fields": ",".join(DETAIL_FIELDS)}
```
第四条单独说一句：**下架商品在不同平台的表现不同**——有的返回错误码，有的返回成功但核心字段为空，有的直接不返回。用第 12 篇的 `integrity.validate` 兜底，加上一个状态位判断，别让下架商品在你的库里"看起来还活着"。
---
## 二、SKU：它是会消失的
SKU 是四个接口里坑最深的，因为它的坑不在"怎么调"，在"调回来之后怎么管"。
| 错误姿势 | 后果 | 正确姿势 |
| ------------------------- | ----------------- | ---------------------------------- |
| 只拿 `sku_id` 当主键 | 多店铺/多平台必然撞键 | 主键 = `(platform, item_id, sku_id)` |
| 覆盖式更新 SKU 表 | 商家删掉的 SKU 永远留在你库里 | 集合 diff，消失的标记 inactive |
| `props.split(":")` 硬拆销售属性 | 属性名里含冒号就炸 | 结构化解析（第 2 篇的 `_parse_props`） |
| 假设每个商品都有 SKU | 单规格商品返回空 | 先判商品规格类型 |
第二条是重灾区。商家改规格是日常操作：把"红色 L"删了换成"红色 XL"。如果你每次拉到 SKU 列表就整表覆盖，或者只往上插入新 SKU——**被删掉的那个 SKU 会永远留在你库里，价格库存还挂着旧数据**。客户拿它下单，就是一次客诉。
正确做法是集合 diff：
```python
# sku_diff.py
def sku_diff(old_ids: set[str], new_ids: set[str]) -> tuple[set[str], set[str]]:
"""
返回 (新增, 消失)。
消失的 SKU 千万别物理删除——历史订单还引用它。
标记 inactive，保留可查。
"""
return new_ids - old_ids, old_ids - new_ids
async def sync_skus(store, item_id: str, new_skus: list[dict]):
old_ids = set(store.sku_ids(item_id))
new_ids = {s["sku_id"] for s in new_skus}
added, removed = sku_diff(old_ids, new_ids)
for s in new_skus:
if s["sku_id"] in added:
store.insert_sku(item_id, s)
else:
store.update_sku(item_id, s) # 属性或价格变了
for sku_id in removed:
store.mark_inactive(item_id, sku_id) # 软删，订单引用不断
```
另外主键那条：`sku_id` 只在单个商品内唯一。做三平台、多店铺的库，主键从第一天就是三元组，**后期改主键的痛苦是接入期的十倍**（第 11 篇说 1688 连 SKU 都没有，那它的"规格"要单独建表，别硬凑）。
---
## 三、价格：它不是一个字段，是一组字段
价格的问题出在"你以为价格是一个数"。
| 错误姿势 | 后果 | 正确姿势 |
| ---------- | ----------- | ---------------------- |
| 只取一个价格字段 | 拿到标价，不是成交价 | 价格组归一，算"生效价" |
| float 存价格 | 账对不上，差几分钱 | `Decimal`，最后一步才转 float |
| 改价后立刻读回来验证 | 读到旧缓存，以为没改成 | 接受几分钟延迟，或以改价接口回执为准 |
| 1688 只取一个价 | 阶梯价拿错档位 | 按起订量取档 |
| 促销价无限期生效 | 过期促销价当现价 | 促销价必须带时效字段 |
第一和第五条组合起来，是最经典的线上事故：**拿到促销价 → 没记过期时间 → 促销结束后继续按促销价计算**。利润表和实际对不上，查一周。
正确建模是把价格当成一个带时效的小对象：
```python
# price_model.py
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from zoneinfo import ZoneInfo
TZ = ZoneInfo("Asia/Shanghai")
@dataclass
class Price:
sku_id: str
list_price: Decimal # 标价 / 一口价
promo_price: Decimal | None = None # 促销价
promo_end: datetime | None = None # 促销截止时间
def effective(self, now: datetime | None = None) -> Decimal:
"""生效价：促销在有效期内才生效，过期自动回落标价"""
now = now or datetime.now(TZ)
if self.promo_price is not None:
if self.promo_end is None or self.promo_end > now:
return self.promo_price
return self.list_price
```
1688 的价格是另一个物种——阶梯价，按起订量分档：
```python
def alibaba_price_tiers(raw: list[dict]) -> list[tuple[int, Decimal]]:
"""
1688 priceRange：[{beginAmount: 2, price: "8.50"}, {beginAmount: 100, price: "7.20"}]
按起订量升序排列，采购 N 件时取满足 N 的最后一档。
"""
tiers = sorted(
((int(x["beginAmount"]), Decimal(x["price"])) for x in raw),
key=lambda t: t[0],
)
return tiers
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def tier_for(tiers: list[tuple[int, Decimal]], qty: int) -> Decimal:
price = tiers[0][1]
for begin, p in tiers:
if qty >= begin:
price = p
return price
```
还有改价那条：**改价接口返回成功 ≠ 价格已生效**。平台侧有缓存延迟，短则几秒长则几分钟。以改价接口的响应做业务判断，别改完立刻读回来验证——你会拿到旧价格，然后误判"改价失败"再改一次。
---
## 四、库存：只有"可售库存"有意义
库存的问题在于：平台给你好几个数，你用了错的那个。
| 错误姿势 | 后果 | 正确姿势 |
| ------------ | ------------ | ---------------- |
| 用"总库存"当可卖数量 | 超卖（锁定的库存又卖了） | 用**可售库存**字段 |
| 平台库存直接覆盖本地库存 | 抹掉本地扣减，二次超卖 | 本地账本 + 校准（第 3 篇） |
| 和详情同频率轮询 | 延迟大还烧配额 | 事件推送优先，轮询兜底 |
| 库存为 0 就当售罄 | 可能是下架或区域性无货 | 状态位 + 库存一起判断 |
| 硬凑 1688 的库存 | 它没有 SKU 库存概念 | 可售状态 + 阶梯价建模 |
第三条是这篇的落地重点。库存是唯一值得上**事件推送**的字段——淘宝的 TMC 有库存变更消息，订阅它，库存事件毫秒级到达，配额零消耗。轮询只做兜底（比如每 5 分钟对账一次，防丢消息）：
```python
# stock_sync.py —— 事件优先，轮询兜底
async def on_stock_event(msg: dict):
"""TMC 库存变更消息：毫秒级，零配额"""
await apply_stock(msg["item_id"], msg["sku_id"], msg["quantity"])
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def reconcile_loop(gw, store, item_ids: list[str], interval: float = 300):
"""兜底对账：防消息丢失。只查本地 5 分钟内有订单变动的商品，别全量"""
while True:
hot = store.items_with_recent_orders(minutes=5)
for iid in hot:
raw = await gw.call("stock.get", item_id=iid)
local = store.local_stock(iid)
# 校准取 min：平台还没反映本地刚扣的量（第 3 篇的规则）
safe = min(int(raw["saleable"]), local)
if store.remote_stock(iid) != safe:
store.update_remote_stock(iid, safe)
await asyncio.sleep(interval)
```
注意兜底循环里那句 `hot = items_with_recent_orders(minutes=5)`：**只对账"最近有单"的商品**。全量对账是最容易写出来的版本，也是把库存轮询成本放大一百倍的版本——而真正需要对账的（刚发生扣减、可能不一致的）永远只占极小一部分。
---
## 五、把四个接口串起来：分级调度器
最后给一个可跑的调度骨架。它的职责只有一件事：**保证每个字段按自己的频率被同步，不多不少**。
```python
# scheduler.py
import asyncio
import time
LEVELS = {"detail": 3600, "sku": 3600, "price": 300, "stock": 30}
class DueScheduler:
def __init__(self, r):
self.r = r
async def due(self, item_id: str, field: str) -> bool:
last = await self.r.get(f"due:{field}:{item_id}")
if last is None:
return True # 从没同步过，先来一次
return time.time() - float(last) >= LEVELS[field]
async def mark(self, item_id: str, field: str):
await self.r.set(f"due:{field}:{item_id}", time.time())
async def sync_item(gw, sched: DueScheduler, store, item_id: str):
# ---- 1) 详情（小时级）：一次拉全 慢字段 + SKU 集合 + 标价 ----
if await sched.due(item_id, "detail"):
d = await gw.call("item.get", item_id=item_id,
fields=",".join(DETAIL_FIELDS))
store.upsert_item(item_id, d)
new_skus = d.get("skus", [])
await sync_skus(store, item_id, new_skus) # 内部做集合 diff
await sched.mark(item_id, "sku")
await sched.mark(item_id, "detail")
# ---- 2) 价格（分钟级）：从详情缓存或价格接口取，归一成 Price 对象 ----
if await sched.due(item_id, "price"):
raw = await gw.call("price.get", item_id=item_id)
for sku_price in normalize_prices(item_id, raw):
store.save_price(sku_price) # effective() 按时效算生效价
await sched.mark(item_id, "price")
# ---- 3) 库存（秒级）：事件推送是主通道，这里只是兜底标记 ----
# 真正的兜底对账在 reconcile_loop，只扫最近有单的商品
async def main(gw, sched, store, item_ids):
# 详情和价格按各自的频率到期才调；库存靠事件
while True:
for iid in item_ids:
await sync_item(gw, sched, store, iid)
await asyncio.sleep(30) # 最细粒度的检查周期，但不是最细粒度的调用频率
```
跑起来的效果：同一个商品，详情接口一天调 24 次、价格 288 次、库存 0 次（走事件）——**对比"四个接口都 30 秒一轮"的写法，配额消耗约降到 1/300**，而库存的实时性反而更好（事件是毫秒级的）。
---
## 六、四个反直觉的结论
**1. 详情接口调得越勤，数据越不准。**
因为详情是慢字段的来源，高频调它不会让标题更新更快（它一天就变一次），只会让你的价格库存看起来"应该也是新的"——然后掩盖它们其实滞后的事实。分开调度，延迟才藏不住。
**2. 消失的 SKU 比新增的危险。**
新增最多是数据多一条；消失的 SKU 还挂着旧价格旧库存，客户能下单。覆盖式更新是这里的万恶之源。
**3. 促销价必须带时效，这是建模问题不是字段问题。**
没有 `promo_end` 字段的促销价，就是一颗定时炸弹——促销结束那天准时爆。
**4. 库存的正确姿势是"少调"，不是"勤调"。**
事件推送零配额、毫秒级；轮询全量、高延迟。轮询不是兜不了底，是兜底时只扫"最近有单"的商品——全量对账是库存成本失控的第一原因。
---
## 七、最后
这篇没有新架构，讲的全是**已有架构里每个接口的使用纪律**。
回看整个系列：第 8 篇给了骨架（统一接口层），第 9、10 篇给了运行时（稳定性、限流），第 12 篇给了选型（为什么用 API），第 11 篇给了差异（平台之间差在哪）。这篇补的是最后一层——**每个高频接口的调用纪律**。骨架、运行时、选型、差异、纪律，五层齐了。
下一篇写「多店铺授权管理」：token 的存储与加密、并发刷新撞车、单店授权过期的隔离。那是接入体系的最后一个坑区，写完这个系列就真正完整了。
`LEVELS` 里的数字是起点不是标准：详情 1 小时、价格 5 分钟、库存 30 秒，都是按"多数电商场景"取的。你们的业务如果价格敏感度高（比如比价类），价格调到 60 秒；如果纯铺货场景，详情调到 6 小时都够。**关键是分级这件事本身，不是每级的具体数字。**未分类
自建爬虫 vs 官方 API：成本、稳定性、风险一次算清（附 Python 源码）
万邦科技Lex 发表于2026-10-27 浏览52 评论0
# 自建爬虫 vs 官方 API：成本、稳定性、风险一次算清（附 Python 源码）
每个做电商数据的团队都纠结过这个问题：接官方 API 要走审核、要等权限，看着就麻烦；自己写个爬虫，两周就能出数据，还不要钱。
这篇把这笔账彻底算清。先说结论：
**爬虫不是"省钱的方案"，是把成本换了个形式——从"花钱"变成"花人 + 担风险"。** 算完三年总账，爬虫通常是官方 API 的 2~4 倍，而最大的一项成本是"养一个懂爬虫的人"。
但我也不会一边倒。爬虫有它合理的场景，第四节专门讲哪些情况该爬。先把账摆出来，你自己判断。
---
## 一、成本：把"看起来免费"的东西标上价
爬虫最大的迷惑性在于：**它的成本几乎全部是隐性的**。没有账单，没有报价单，全部体现在"某个人每个月花在它身上的时间"里。
逐项拆开：
**① 首次开发**。大多数团队的预估是"一周"，实际是 2~4 周。因为真正花时间的不是写解析器（那确实只要两天），是处理长尾：懒加载、翻页截断、字段在不同商品页上的十几种变体。
**② 持续维护**。这是大头，也是最容易被漏算的。页面会改版——改成版式、改类目页结构、加一层前端渲染，平均每月 1~~2 次，每次修复 0.5~~2 天。算下来，**养一个爬虫需要 0.2~0.4 个人力，常年不停**。
**③ 反爬对抗的设备成本**。IP 代理池、请求频率控制、失败重试——合法合规的部分同样有真金白银的月费。
**④ 人的风险**。爬虫系统往往只存在于一个人的脑子里。这个人休假、离职，系统跟着停摆——而你甚至不知道它哪天停的（下一节讲这个）。
**官方 API 的成本**则几乎全部是显性的：接入人力（1~3 周，含等审核的并行期）、持续维护（平台接口变更频率远低于页面改版，且有公告期）、调用费用。
同一个需求（三平台商品 + 订单同步），我们按一名中级后端的全成本 3.5 万/月算三年账：
| 成本项 | 自建爬虫 | 官方 API |
| --------- | -------------------------- | ---------------------- |
| 首次开发 / 接入 | 2.6 万（3 周） | 3.5 万（1 人月，含审核并行期） |
| 持续维护 | **31.5 万**（0.25 人力 × 36 月） | 10.1 万（0.08 人力 × 36 月） |
| 代理 / 调用费 | 2.9 万（代理池月费） | 3.6 万（调用与服务费） |
| **三年合计** | **≈ 37 万** | **≈ 17.2 万** |
差距主要来自②：**API 的"维护"是读公告和改字段名，爬虫的"维护"是重新理解一个被改过的页面**。前者有文档，后者只有你的猜想。
这张账能算，是因为它有模型。下面的计算器你可以拿去代入自己的数字：
```python
# tco.py —— 三年总拥有成本计算器
# 参数按你们自己的实际情况改，模型不变
from dataclasses import dataclass
@dataclass
class Team:
"""团队参数"""
salary_month: float = 35000 # 一名相关岗位的全成本/月（工资 × 1.4 起步）
@dataclass
class CrawlerPlan:
"""爬虫方案：大部分成本是隐性的，所以要显式建模"""
build_weeks: float = 3.0 # 首次开发（现实值 2~4 周，别用 1 周）
maintain_fte: float = 0.25 # 常年维护人力占比（0.2~0.4，低于 0.2 通常意味着没人在管）
proxy_month: float = 800 # 代理与设备月费
months: int = 36
def cost(self, t: Team) -> dict:
# 按周折算：每月 21.75 个工作日，每周 5 天
build = t.salary_month / 21.75 * 5 * self.build_weeks
maintain = t.salary_month * self.maintain_fte * self.months / 12
infra = self.proxy_month * self.months
return {"首次开发": round(build, 1),
"持续维护": round(maintain, 1),
"代理与设备": round(infra, 1),
"合计": round(build + maintain + infra, 1)}
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
@dataclass
class ApiPlan:
"""官方 API：成本几乎全部显性"""
build_months: float = 1.0 # 含与审核期并行的开发
maintain_fte: float = 0.08 # 读公告、改字段名、处理限流
api_month: float = 1000 # 调用费 + 软件服务费
months: int = 36
def cost(self, t: Team) -> dict:
build = t.salary_month * self.build_months
maintain = t.salary_month * self.maintain_fte * self.months / 12
fees = self.api_month * self.months
return {"首次接入": round(build, 1),
"持续维护": round(maintain, 1),
"调用与服务费": round(fees, 1),
"合计": round(build + maintain + fees, 1)}
if __name__ == "__main__":
t = Team()
c = CrawlerPlan().cost(t)
a = ApiPlan().cost(t)
w = max(len(k) for k in c)
for k in c:
print(f"{k:<{w}} 爬虫 {c[k]:>8} 万 API {a[k]:>8} 万")
print(f"\n倍数：{c['合计'] / a['合计']:.1f}x")
```
跑出来就是上面那张表。**你团队的实际数字可能不同，但倍数关系通常落在 2~4 倍**——因为决定倍数的是"页面改版频率 ÷ 接口变更频率"，这个比值对任何平台都在 10 倍以上。
> 有一种情况会让这个计算器失真：你只爬**一个**页面、只用**三个月**。那爬虫确实便宜。计算器的隐含前提是"长期、多页面、商用"——这也是大多数人纠结这个问题的真实场景。
---
## 二、稳定性：不是"能不能跑"，是"断了你知不知道"
这是两种方案差距最大、又最少被认真比较的维度。
### 爬虫的故障模式：静默失败
爬虫最危险的特性是**它的失败经常不报错**。
页面改版后，CSS 选择器取到的价格字段是空的——但程序不崩溃，只是默默写进去一行空数据。三天后财务发现报表价格对不上，往前追，发现已经空了三天。
第 9 篇讲过一句话：**返回可解释的结果，比返回正确数据更重要**。爬虫恰恰相反：它给不出可解释的结果，页面改版后它连"我现在是坏的"都不告诉你。
另一个爬虫特有的问题是**数据可信度**。你爬到的是"页面展示价"——但页面展示价不一定是成交价：会员价、叠加优惠券、区域差异、千人千面的推荐价。商品价格同步这种场景，展示价和 API 返回的实际价格能差出 5%~15%，而这个差异你很难系统性发现。
### API 的故障模式：有状态、可归因
API 也会挂——平台故障、限流、字段弃用。区别在于：
| | 爬虫 | 官方 API |
| ------ | ----------------- | ---------------- |
| 失败形态 | 静默，返回空或错数据 | 报错，有错误码 |
| 能否定位原因 | 靠猜（页面又改了？IP 被封了？） | 错误码直接告诉你 |
| 恢复 | 人工修解析器，小时~天级 | 限流自动恢复 / 平台公告修字段 |
| 可监控性 | 极难（没有"正常"的基准） | 成功率、错误码分布都是现成指标 |
### 共同的必修课：数据校验
不管数据从哪来，校验都得做。这段代码对两种来源都适用——尤其是对爬虫，它是你发现"静默失败"的唯一手段：
```python
# integrity.py —— 数据完整性与静默失败检测
# 对 API 和爬虫的数据一视同仁地校验
from dataclasses import dataclass
@dataclass
class Issue:
item_id: str
field: str
kind: str # missing / out_of_range / mismatch
detail: str
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
REQUIRED = ("item_id", "title", "price", "stock")
PRICE_RANGE = (0.01, 10_000_000)
def validate(items: list[dict], source: str) -> list[Issue]:
"""单源校验：字段缺失 + 数值越界"""
issues = []
for it in items:
iid = str(it.get("item_id", "?"))
for f in REQUIRED:
v = it.get(f)
if v is None or v == "":
# 爬虫静默失败最典型的形态：字段存在但为空
issues.append(Issue(iid, f, "missing", f"{source} 返回空"))
p = it.get("price")
if isinstance(p, (int, float)) and not (PRICE_RANGE[0] <= p <= PRICE_RANGE[1]):
issues.append(Issue(iid, "price", "out_of_range", f"{source} 价格 {p}"))
return issues
def cross_check(primary: list[dict], secondary: list[dict],
tol: float = 0.01) -> list[Issue]:
"""
双源比对：比如 API 数据 vs 3 天前的快照。
同一商品价格波动超过 tol（1%），要么真降价了，要么有一边坏了。
配合告警，这是抓"静默失败"最有效的一招——
因为静默失败不会自己报警，只有比对能暴露它。
"""
ref = {str(x["item_id"]): x for x in secondary}
issues = []
for it in primary:
iid = str(it["item_id"])
old = ref.get(iid)
if not old:
continue
p_new, p_old = it.get("price"), old.get("price")
if p_new is None or p_old in (None, 0):
continue
if abs(p_new - p_old) / p_old > tol:
issues.append(Issue(
iid, "price", "mismatch",
f"{p_old} -> {p_new}（变动 {abs(p_new - p_old) / p_old:.1%}）",
))
return issues
```
`cross_check` 是我们生产环境里跑了很多年的一个函数。它不关心数据从哪来，只关心**数据是不是还讲得通**。爬虫团队尤其需要它——因为没有它，你可能三个月都不知道自己的解析器坏了。
---
## 三、风险：这一项没有"权衡"，只有红线
前两项是算术题，这一项是规则题。
把话说在前面：**我不会在这篇文章里讨论任何"绕过反爬"的技术**。不是因为怕事，是因为在国内的法律环境下，那已经不是技术选型的问题了。
### 商用场景下，爬虫面临三层风险，一层比一层重
**第一层：民事（反不正当竞争）**
绕开平台的访问控制抓数据、违反 robots 协议或用户协议抓取，在司法实践中普遍被认定为不正当竞争。这类判例非常多（大众点评诉百度、淘宝诉美景等都是公开判例），判赔金额从几十万到数千万不等。
**第二层：刑事**
这一层很多人不知道，但它真实存在：**如果抓取涉及突破或规避技术保护措施**（破解加密参数、伪造签名、打码平台过验证码），可能触及《刑法》第 285 条"非法获取计算机信息系统数据罪"——这是**刑事**罪名，有真实获刑判例，个人刑期、单位罚金都有。
爬取的内容一旦涉及**用户个人信息**（买家昵称、地址、手机号），还叠加《个人信息保护法》，责任更重。
**第三层：业务连续性**
就算前两层都侥幸没发生，平台识别到爬虫行为后封 IP、封账号、法律函三连——你的数据管道说断就断，而你的业务是建立在这条管道上的。**用爬虫做商用数据源，等于把业务的命脉押在一个随时可能被合法切断的东西上。**
### 两个常见误区，顺手拆掉
**"大厂也在爬"** —— 大厂爬大厂，是法务团队规模相当的对抗，输了一起谈，赔得起也耗得起。小团队没有这个筹码，同样的行为，风险完全不对等。
**"我只是自用，不商用"** —— 自用研究的风险确实低得多（这是真的，第四节也会说爬虫的合理场景）。但"自用"和"自用着自用着就给客户看了"之间的距离，通常只有一个业务增长周期。**数据一旦进入商业流程，就回不去了。**
### API 的风险也要诚实列出来
API 不是零风险：接口会下线、类目会调整、政策会变。但关键区别是——**平台对这些变更有公告期**（通常 30~90 天），你可以提前改代码；而页面改版不会通知你。另外，用 API 也有合规要求（超范围使用数据同样会被封），这条红线两边都存在。
所以风险这一项的结论是：**商用场景下它是一票否决项，不是加权项。**
---
## 四、什么时候爬虫是合理选择
说完了不合理的，得把合理的说清楚，不然这篇就成了立场文。
**爬虫合理的场景：**
1. **数据在官方 API 里根本不存在**。全网竞品价格监控、搜索引擎数据、社媒舆情——这些没有"官方 API"可接，公开信息抓取 + 合规使用是正当的。
2. **一次性研究 / 学习**。跑一次分析、写一篇报告、验证一个想法，不进入业务流程、不对外提供。这个场景爬虫完全成立。
3. **平台没有开放能力的长尾站点**。有些小平台根本没有开放平台，无 API 可接。
4. **你自己店铺的展示层信息**。详情页装修、页面呈现效果这类"展示层"数据，API 里没有，爬自家页面没有争议。
**不合理的场景只有一个，但最常见**：拿爬虫当 API 的"免费替代"去商用——尤其是数据明明在 API 里有的情况。这不是技术选型，是**把风险转移给三个月后的自己**。
一个快速自查清单，五个问题：
1. 这份数据官方 API 里有没有？（有 → API）
2. 用途会不会进入商业流程？（会 → API）
3. 有没有人能长期养它？（没有 → API）
4. 数据错了的代价是什么？（影响客户/财务 → API）
5. 需要多高的更新频率？（高频 → API，因为高频爬取必然撞上对抗）
**五个问题里有两个以上答案是括号里的那个，这个决策就已经做完了。** 真正还能选爬虫的，是五个问题都指向另一边的场景——那种场景其实不多，但确实存在。
---
## 五、最后
这篇的立场我尽量放平了：爬虫不是原罪，它有它的场景；API 也不是圣杯，它有自己的成本和维护量。
但有一个区别必须说透：**API 花的钱买的是"确定性"**——确定性的数据口径、确定性的错误反馈、确定性的法律边界。爬虫省下的钱，本质是从这三个确定性里借的，利息在后面。
我们做三平台数据接口服务，这篇里的账就是我们自己算过的账——也是我们客户 switching（从爬虫迁到 API）时反复验证的账。`tco.py` 的参数改成你们团队的真实数字，跑一遍，比看任何人的结论都有用。
下一篇写「多店铺授权管理」：token 的存储与加密、并发刷新撞车、单店授权过期的隔离——那是接入体系的最后一个坑区。
---
*附：`tco.py` 里 `maintain_fte=0.25` 是我们从数十个爬虫系统的维护投入里归纳的经验区间（0.2~0.4）的中位值。如果你的爬虫由多人共管，实际投入可能更高——共管意味着责任分散，这是另一个隐性成本。*未分类
淘宝开放平台 vs 1688 vs 京东宙斯：能力、权限、坑点全梳理（附 Python 源码）
万邦科技Lex 发表于2026-10-27 浏览55 评论0
# 淘宝开放平台 vs 1688 vs 京东宙斯：能力、权限、坑点全梳理（附 Python 源码）
这篇跟前几篇不一样：没有故事线，是一篇**查表文**。
它解决的是每个接入团队都会遇到的一个问题——三个平台的文档各说各话，你没办法把三份文档摊在一起比。这篇就是把它们摊在一起比完了的样子。
建议先收藏，用到哪条查哪条。文章末尾有可直接跑的签名与请求代码，以及一份能力声明表——那部分是能直接抄进项目里的。
---
## 一、三个平台，三种"人设"
动手之前先建立整体印象，后面查表才有方向：
**淘宝开放平台（TOP）**
年纪最大、接口最多、文档最全，也因此**历史包袱最重**。同一种数据往往有三四个接口能拿到，字段命名横跨三个时代的风格。它的沙箱环境是三家里最不靠谱的——这是它最大的坑。
**京东宙斯**
流程最"企业化"：资质要求最严、回调要求 HTTPS + 备案域名、审核规范。回报是**类目属性 schema 三家里最完整**，做商品发布类需求它最省心。它的坑集中在参数格式和返回体结构上，都是小坑，但数量多。
**1688 开放平台**
唯一的 B2B。它没有"零售"概念：没有 SKU 库存、价格是区间、按起订量阶梯。**不能把它当第三个淘宝来接**——这是接 1688 最大的认知错误。它的坑集中在签名和图片上。
---
## 二、能力对照大表
这张表是全文的核心。✓ = 支持，△ = 部分支持或有条件，✗ = 不支持。
| 维度 | 淘宝 TOP | 京东宙斯 | 1688 |
| --------------- | ----------------------- | ----------------------------- | ----------------------- |
| **签名算法** | MD5 三明治 | MD5 三明治 | **HMAC-SHA1 + urlPath** |
| **时间戳（签名用）** | 秒，`yyyy-MM-dd HH:mm:ss` | 秒，同左 | **毫秒**，`_aop_timestamp` |
| **业务时间参数** | `yyyy-MM-dd HH:mm:ss` | `yyyy-MM-dd HH:mm:ss`，部分接口只到天 | 秒级时间戳 |
| **业务参数传递** | 平铺 query | 打包进 `360buy_param_json` | 平铺 query |
| **返回结构** | 平铺 | **按接口名嵌套，要剥壳** | 平铺 |
| **分页方式** | `page_no / page_size` | `page / pageSize` | `startIndex / pageSize` |
| **分页上限** | 100 / 页 | 100 / 页 | 部分接口 startIndex 有上限 |
| **商品详情** | ✓ | ✓ | ✓ |
| **SKU 级库存** | ✓ | △（部分接口才返回） | **✗（按起订量卖，无 SKU 库存）** |
| **订单列表** | ✓ | ✓ | ✓（分销单为主） |
| **物流轨迹** | ✓（增量接口 + 轨迹） | △（部分按天查） | △（以发货状态为主） |
| **消息推送（TMC 类）** | ✓ TMC | △ 部分消息 | ✗（基本靠拉） |
| **沙箱环境** | △ **长期不同步** | ✓ 相对可用 | ✗ |
| **图片外链** | ✓ 可外链 | ✓ 可外链 | △ **有防盗链** |
| **token 有效期** | 自用型长，授权型短 | 短 + refresh_token | 长 |
| **审核周期** | 1~3 周 | 1~2 周 | 3~7 天 |
| **限流模型** | appkey × 接口，双限 | 应用 × 接口，双限 | appkey 流量包 + QPS |
| **适合谁** | 淘系卖家、全域比价 | 京东 POP / 自营 | 分销、一件代发、选品 |
三个格子单独说一下，因为它们最容易被误判：
**1688 没有 SKU 库存**。这不是接口没开放，是业务模型不同——1688 的商品按"起订量 → 单价"的区间卖，压根没有"某个 SKU 剩几件"的概念。你在第 8 篇见过的 `NotSupported` 异常，主要就是为它准备的。硬要接的话，能拿到的是"可售状态 + 阶梯价"，拿不到库存数。
**淘宝沙箱长期不同步**。新接口没有沙箱、旧接口沙箱返回的字段和线上不一致，这两件事都是常态。所以我们的冒烟测试打的是线上真实接口（用固定样本、极低频率），沙箱只用来验证签名和连通性。**别信沙箱的返回体**。
**1688 图片有防盗链**。商品图直接 `<img src>` 塞进自己的页面，会有一部分图挂掉（referer 校验）。转存的时候要么带 Referer 伪造，要么走它提供的图片网关——前者不稳定，后者要申请。这个坑在第 2 篇的图片转存里埋过，这里说清楚原因。
---
## 三、权限申请对照：代码之外花时间最多的事
第 6 篇说过：代码三天写得完，权限可能等三周。这里把三个平台的申请路径摊开。
| 环节 | 淘宝 TOP | 京东宙斯 | 1688 |
| ------- | ------------------ | ---------------- | ----------------- |
| 主体要求 | 企业/个人开发者入驻 | **企业资质**（营业执照等） | 企业认证 |
| 应用类型 | 自用型 / 他用型（ISV） | 自用 / 工具型 | 自用应用 / 他用 |
| 接口权限 | 按**权限包**申请，敏感接口单独审 | 控制台勾选 + 部分白名单 | 权限包，B2B 类目相关要单独申请 |
| 回调要求 | 域名备案 | **HTTPS + 备案域名** | HTTPS |
| 最容易卡的地方 | "他用型"的场景说明要写清楚 | 资质材料一次过审的很少 | 类目权限漏申请 |
三条实战建议：
1. **权限包按"最终要调的接口"倒推**，不是按"类目"申请。淘宝的权限包和接口是多对多关系，最容易犯的错是以为申请了商品权限包就能调所有商品接口——`item.seller.get` 和 `items.list.get` 在不同的包里。
2. **场景说明写具体业务**。"数据同步"会被打回，"为本店已成交订单同步物流轨迹用于售后跟踪"能过。审核看的是你会不会拿数据干别的。
3. 三个平台的 token 机制不一样，**自用型和授权型混着用迟早出事**。自用型 token 长期有效但只能操作自己的店；授权型短，要走 refresh。如果你的产品要服务多家店，从第一天就按授权型设计，别混。
---
## 四、签名：三个平台，三种脾气
签名是接入期最先撞上的东西，也是最值得写成代码的部分。三家的算法其实是"一个思路 + 两个变体"：
- 淘宝、京东：`md5(secret + k1v1k2v2... + secret)`，**几乎一样**
- 1688：`hmac_sha1(secret, urlPath + k1v1k2v2...)`，**urlPath 也参与签名**，且时间戳是毫秒
```python
# signers.py
import hashlib
import hmac
def _sorted_concat(params: dict) -> str:
"""k1v1k2v2...，按 key 的 ASCII 排序——三平台一致的要求"""
return "".join(f"{k}{params[k]}" for k in sorted(params))
def sign_taobao(secret: str, params: dict) -> str:
"""TOP：md5(secret + 拼接 + secret)，大写"""
raw = secret + _sorted_concat(params) + secret
return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
def sign_jd(secret: str, params: dict) -> str:
"""宙斯：算法同 TOP，但业务参数打包在 360buy_param_json 里一起签"""
raw = secret + _sorted_concat(params) + secret
return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
def sign_alibaba(secret: str, url_path: str, params: dict) -> str:
"""
1688：hmac-sha1(secret, urlPath + 拼接)，大写。
两个独特点：
1. urlPath 参与签名——漏了它，签名永远对不上，而且报错信息看不出原因
2. 时间戳用毫秒（_aop_timestamp），用秒的话签名同样对不上
"""
raw = url_path + _sorted_concat(params)
return hmac.new(secret.encode("utf-8"), raw.encode("utf-8"),
hashlib.sha1).hexdigest().upper()
```
签名调试有一个万能技巧，比任何文档都有用：**把待签字符串打出来，和平台报错页给的"期望签名"对比**。三个平台的签名错误报错都一样没用（就一句 `sign error`），唯一的调试方法就是对着拼。我们第 6 篇说"签名预估 0.5 天实际 1 天"，多出来的半天基本都耗在这里。
---
## 五、三个客户端：把平台的怪脾气消化在适配器里
签名之上是请求。三个平台的"怪脾气"——京东的参数打包、京东的返回剥壳、淘宝的错误藏 body、1688 的毫秒时间戳——都在这一层消化掉，业务层不该看见它们。
```python
# clients.py
import json
import time
import httpx
from signers import sign_taobao, sign_jd, sign_alibaba
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class PlatformError(Exception):
def __init__(self, code, msg, platform):
super().__init__(f"[{platform}] {code}: {msg}")
self.code = str(code)
self.msg = msg
self.platform = platform
# ---------- 淘宝 TOP ----------
async def taobao_call(client: httpx.AsyncClient, cfg: dict, session: str,
method: str, api_params: dict) -> dict:
p = {
"method": method,
"app_key": cfg["app_key"],
"session": session,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"format": "json",
"v": "2.0",
"sign_method": "md5",
**{k: str(v) for k, v in api_params.items()},
}
p["sign"] = sign_taobao(cfg["app_secret"], p)
r = await client.get("https://gw.api.taobao.com/router/rest", params=p)
data = r.json()
# TOP 的 HTTP 永远 200，错误藏在 body 的 error_response 里
# ——只看状态码，监控永远是绿的（第 9 篇的口径问题就出在这）
if "error_response" in data:
err = data["error_response"]
raise PlatformError(err.get("code"), err.get("msg"), "taobao")
return data
# ---------- 京东宙斯 ----------
async def jd_call(client: httpx.AsyncClient, cfg: dict, token: str,
method: str, body: dict) -> dict:
p = {
"method": method,
"app_key": cfg["app_key"],
"access_token": token,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"v": "2.0",
# 业务参数整体打包成一个 JSON 字符串，再作为普通参数参与签名
# 字段名大小写敏感：文档里的 snake_case 只是显示，实际要 camelCase
"360buy_param_json": json.dumps(
body, ensure_ascii=False, separators=(",", ":")
),
}
p["sign"] = sign_jd(cfg["app_secret"], p)
r = await client.get("https://api.jd.com/routerjson", params=p)
return jd_unwrap(r.json(), method)
def jd_unwrap(data: dict, method: str) -> dict:
"""
宙斯的返回体按接口名嵌套一层，且不同网关下 key 的写法不一致
（点号 / 下划线都见过），做兼容处理。
这是京东接入手册里最常被问的一句话："返回怎么多包了一层"
"""
if "errorResponse" in data:
err = data["errorResponse"]
raise PlatformError(err.get("code"), err.get("zh_desc"), "jd")
for key in (method, method.replace(".", "_")):
if key in data:
inner = data[key]
# 部分接口在剥完第一层后还有个 result 包裹，一并处理
if isinstance(inner, dict) and "result" in inner:
return inner["result"]
return inner
raise PlatformError("UNWRAP", f"返回体里找不到 {method}", "jd")
# ---------- 1688 ----------
async def alibaba_call(client: httpx.AsyncClient, cfg: dict,
url_path: str, api_params: dict) -> dict:
p = {
**{k: str(v) for k, v in api_params.items()},
"_aop_key": cfg["app_key"],
# 毫秒！用秒的话签名对不上，而且报错不说原因
"_aop_timestamp": str(int(time.time() * 1000)),
}
p["_aop_signature"] = sign_alibaba(cfg["app_secret"], url_path, p)
# url_path 形如 /openapi/param2/1/com.alibaba.trade/alibaba.trade.get.buyerView
r = await client.get(f"https://gw.open.1688.com{url_path}", params=p)
data = r.json()
if not data.get("success", True):
err = data.get("error_info") or {}
raise PlatformError(err.get("code"), err.get("desc"), "alibaba")
result = data.get("result", data)
# 1688 有一种"软失败"：success 是 true，但业务字段是空的。
# 对空数据做一次必填校验，别把空当成功传下去
if isinstance(result, dict) and result.get("datas") == []:
raise PlatformError("EMPTY", "业务字段为空的软失败", "alibaba")
return result
```
这段代码里有三个细节，是接入期最容易各自踩一遍的：
1. **京东的 `360buy_param_json` 要压缩**。`ensure_ascii=False` + `separators=(",", ":")` 不是洁癖——参数里有中文时，默认的 `ensure_ascii=True` 会把中文转成 `\uXXXX`，签名按转义后的算，服务端按原文验，**永远对不上**。这是京东接入报错里出现频率最高的一条。
2. **淘宝的错误码是 int 不是 str**。`err.get("code")` 拿到的是 `7` 不是 `"7"`，如果你的错误分类表按字符串写，一条都匹配不上。所以 `PlatformError` 里统一 `str(code)`。
3. **1688 的"软失败"要校验**。`success: true` 但 `datas: []` 的情况真实存在（商品下架、权限降级时会出现），直接当成功处理会让下游静默拿到空数据。
---
## 六、能力声明表：把第二节的表变成代码
第二节的表不该只存在于文章里。把它变成数据，业务层就能在启动时校验，而不是等到运行时才发现 1688 没有 SKU 库存：
```python
# capabilities.py
from enum import Enum
class Cap(str, Enum):
YES = "yes"
PARTIAL = "partial"
NO = "no"
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
MATRIX: dict[str, dict[str, Cap]] = {
"item.get": {"taobao": Cap.YES, "jd": Cap.YES, "alibaba": Cap.YES},
"sku.stock": {"taobao": Cap.YES, "jd": Cap.PARTIAL, "alibaba": Cap.NO},
"order.list": {"taobao": Cap.YES, "jd": Cap.YES, "alibaba": Cap.YES},
"logistics.trace": {"taobao": Cap.YES, "jd": Cap.PARTIAL, "alibaba": Cap.PARTIAL},
"push.tmc": {"taobao": Cap.YES, "jd": Cap.PARTIAL, "alibaba": Cap.NO},
"sandbox": {"taobao": Cap.PARTIAL, "jd": Cap.YES, "alibaba": Cap.NO},
}
class NotSupported(Exception):
pass
def check(platform: str, *needs: str) -> None:
"""启动时调一次。缺什么当场报，别等运行时"""
for need in needs:
cap = MATRIX.get(need, {}).get(platform)
if cap is None:
raise NotSupported(f"未知能力 {need}")
if cap is Cap.NO:
raise NotSupported(f"{platform} 不支持 {need}")
if cap is Cap.PARTIAL:
# partial 不算致命，但要写日志，提醒实现的人去看条件
print(f"[warn] {platform} 对 {need} 仅部分支持，先确认条件")
```
`check("alibaba", "sku.stock")` 会在启动那一刻就报 `alibaba 不支持 sku.stock`——比三个星期后客户投诉"库存显示不对"强得多。这套东西在第 8 篇展开过架构，这里给的是它的**数据来源**。
---
## 七、坑点清单（每平台五条，按踩的人次排序）
这部分建议直接转成团队的接入 checklist。
### 淘宝 TOP
| # | 坑 | 症状 | 解法 |
| - | --------------------- | -------------- | ---------------------- |
| 1 | 沙箱与线上不同步 | 沙箱全绿，上线全挂 | 冒烟打线上固定样本，沙箱只验签名 |
| 2 | `num_iid` 是长数字 | 前端/JS 侧 ID 丢精度 | 全链路按字符串传，落库也用 TEXT |
| 3 | 分页 `total_results` 不准 | 大类目翻页翻到一半开始重复 | 超 1 万条改用**时间窗切片**，不靠翻页 |
| 4 | TMC 消息重复 + 乱序 | 库存被旧消息覆盖 | 消息带版本号，旧的丢弃（幂等） |
| 5 | 错误码是 int | 错误分类表匹配不上 | 统一 `str(code)` 再分类 |
其中第 3 条值得展开一句：`page_no` 翻到几百页之后，`total_results` 会因为数据在变而漂移，翻页翻出重复或漏。**大结果的正确姿势永远是"时间窗 + 游标"**，这和第 3 篇讲订单增量的思路是同一个。
### 京东宙斯
| # | 坑 | 症状 | 解法 |
| --- | --- | --- | --- |
| 1 | `360buy_param_json` 中文转义 | 签名永远对不上 | `ensure_ascii=False` + 压缩分隔符 |
| 2 | 返回体按接口名嵌套 | "数据怎么多包了一层" | `jd_unwrap` 统一剥壳，兼容点号/下划线 |
| 3 | 字段大小写敏感 | 参数传了但没生效 | 文档里 snake_case 是给人看的，代码用 camelCase |
| 4 | 部分接口只支持按天 | 想查小时级查不了 | 按天拉 + 本地按秒过滤，别硬试 |
| 5 | 回调要 HTTPS + 备案 | 测试环境收不到推送 | 提前准备域名，这个等不起 |
### 1688
| # | 坑 | 症状 | 解法 |
| --- | --- | --- | --- |
| 1 | 签名带 urlPath + 毫秒时间戳 | 签名永远对不上，报错不说原因 | 用 `sign_alibaba`，别套淘宝的写法 |
| 2 | 没有 SKU 库存 | 拿到的"库存"其实是别的 | 按"可售状态 + 阶梯价"建模，别硬凑 |
| 3 | 图片防盗链 | 铺到自己页面图挂一半 | 转存走图片网关，或处理 Referer |
| 4 | startIndex 有上限 | 大供应商商品翻不全 | 改时间游标增量拉 |
| 5 | 软失败（success=true 但空数据） | 下游静默拿到空 | 对关键字段做非空校验 |
三张表放一起看，会发现一个规律：**淘宝的坑在"数据形态"，京东的坑在"格式规范"，1688 的坑在"业务模型"**。预判一个团队接哪个平台会卡多久，看他们擅长处理哪种问题就行。
---
## 八、选型：什么情况用谁
最后把三家的适用场景说透，免得这张表看着热闹，落不了地：
- **做淘系卖家工具、全域比价、竞品监控** → 淘宝 TOP 绕不开，沙箱的坑认了
- **给京东 POP 商家做上货、订单处理** → 京东宙斯，类目 schema 完整是最实在的优势
- **做分销、一件代发、选品比价** → 1688，且**从第一天就按"无 SKU 库存"设计数据模型**
- **三平台都要** → 别做三套代码。第 8 篇那套统一接口层 + 这篇的能力声明表，就是为这个场景准备的
以及一个提醒：三家的审核周期加起来比你的开发周期长。**第一天就把三个平台的权限申请提交了，再开始写代码**——这是第 6 篇 Day 0 清单里第一条的原因，这篇算是把 Day 0 的细节补全了。
---
## 九、最后
这篇的代码量是十篇里最小的，但查表密度最高。因为它回答的问题不是"怎么做"，是"**差异在哪**"——而差异是接入期所有工时超支的根源。
第 6 篇里那句"签名预估 0.5 天实际 1 天"，多出来的半天就在第四节；"错误处理预估 0.5 天实际 1 天"，多出来的半天就是第五节那些剥壳和软失败。**把差异提前写下来，排期才敢往下压。**
我们是做这三平台数据接口服务的，这篇文章里的坑就是我们每天在替客户踩的。代码都是能直接跑的骨架，`signers.py` 和 `clients.py` 拿去改改配置就能用；`capabilities.py` 的矩阵按你们的接口清单补全。
下一篇打算写「多店铺授权管理」：token 的存储、刷新、并发刷新撞车、单店授权过期怎么隔离——那是这套接入体系的最后一个坑区。未分类
三平台 API 限流怎么绕开瓶颈？三个参数调完就顺了（附 Python 源码）
万邦科技Lex 发表于2026-10-26 浏览60 评论0
# 三平台 API 限流怎么绕开瓶颈？三个参数调完就顺了（附 Python 源码）
先把话说清楚：**平台的限流绕不过去**。
改 IP、换 AppKey、注册多个应用分摊额度——这些做法短期有效，但只要平台做应用维度关联，一封就是全封，而且是永久性的。我们做这行两年，见过的封号案例里，一半是这么来的。
所以这篇文章讲的不是"绕开限制"，是**"别再浪费额度"**。
因为真实情况是：撞限流的团队里，**九成的瓶颈不是额度不够，是额度被烧在了没用的地方**。我们自己的数据：调参前一个月撞限流 3400 次，调参后 12 次——平台的额度一次没变，变的只是我们自己怎么用。
三个参数，加一套配额账本。下面是完整的东西。
---
## 一、先搞清楚你在跟什么限流打交道
三个平台的限流模型**长得完全不一样**。用一套逻辑套三个平台，是这个话题里最常见的错误。
| | 淘宝 | 京东 | 1688 |
| ------- | -------------- | ----------- | ---------- |
| 限流维度 | appkey × 接口 | 应用 × 接口 | appkey 流量包 |
| 形态 | 日调用量 + QPS 双限 | 日调用量 + QPS | 总量包 + QPS |
| 典型报错 | `code: 7`、`26` | 业务码 + `429` | `429`、业务码 |
| 失败调用算不算 | **算** | **算** | 部分算 |
| 恢复粒度 | 秒级 | 分钟级 | 秒级 |
| 是否可提额 | 可申请 | 走商务 | 可购买 |
这张表里有两行直接决定了架构，但大多数实现都踩了：
### 关键结论 1：限流是「接口 × 应用」维度的，不是全局的
淘宝的调用量配额是**按接口分别计**的。`item_get` 打满了，`trade_list` 一点不受影响。
所以全局令牌桶是错的。
我见过不止一个项目这么写：
```python
# ❌ 错的
limiter = TokenBucket(rate=50) # 全局 50 QPS
async def call(platform, api, **kw):
await limiter.acquire()
...
```
后果是：铺货任务在猛拉详情接口，把全局令牌吃光，这时候一个物流查询请求进来，明明平台侧完全不限它，却要排队等 3 秒。
**正确做法**：桶的 key 是 `(platform, api)`。粒度对齐平台的限流维度，不多不少。粒度再细（比如按 item_id）没意义，因为平台不按这个限；粒度再粗（按平台）就浪费了额度。
### 关键结论 2：失败的调用也占配额
这条最反直觉，也最贵。
淘宝的调用量是按**调用次数**算的，不看成功失败。所以你签名错了、参数漏了，重试三次——**四次调用全部计入配额，全部是浪费**。
这就是为什么上一篇讲的重试分级在这里格外重要：`fatal` 类错误一次重试都不该有。它不只是"没用"，它是在**偷你还剩的额度**。
> 顺带一个实现细节：不要只靠 `code` 判断错误类型。平台会新增错误码，你的映射表永远追不上。要加一层 message 关键词兜底：
>
> ```python
> def is_rate_limited(code: str, msg: str) -> bool:
> if code in ("7", "26", "429"):
> return True
> kw = ("limit", "too many", "频繁", "超限", "exceed", "throttl")
> return any(k in msg.lower() for k in kw)
> ```
---
## 二、瓶颈在哪：先算三笔账，再动参数
不先算账就调参，等于闭着眼睛拧螺丝。这三笔账都是 SQL 或日志就能算出来的，半小时的事。
### 账 1：配额花在哪个接口上了
```
接口 调用次数 占比 带来有效数据
item_get 412,000 68.3% 41,200 条
trade_list 88,000 14.6% 88,000 条
logistics_trace 71,000 11.8% 9,400 条 ← 注意这里
shop_info 32,000 5.3% 32,000 条
```
看到问题了吗：`logistics_trace` 占了 11.8% 的配额，只换来 9,400 条有效数据。因为大部分查询打在了**已签收**的订单上——早就没有新轨迹了。
（这就是第三篇讲的：用 `INSERT OR IGNORE` + `rowcount == 1` 去重，加上 `last_status != 'SIGNED'` 过滤，这个接口的调用量能降 80%。**降调用量比调参数有效一百倍**，因为调参只是在分配额度，降量是在释放额度。）
### 账 2：有效请求率
```
有效请求率 = 真正拿到新数据的调用 / 总调用
```
低于 30% 说明你在"重复问同样的问题"，该加缓存或做增量了。这个数字跟限流无关，但它决定了你需要多少额度。
### 账 3：配额燃烧曲线
这是最容易被忽略的一笔。日配额 10 万，如果前 6 小时用掉 8 万，剩下 18 小时就只能干等着——而客户投诉通常发生在下午。
```python
# throttle/burn.py
from datetime import datetime
from zoneinfo import ZoneInfo
TZ = ZoneInfo("Asia/Shanghai")
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def consume_quota(r, platform: str, api: str, n: int = 1) -> None:
"""每次调用都记账，按自然日归零"""
day = datetime.now(TZ).strftime("%Y%m%d")
key = f"quota:{platform}:{api}:{day}"
await r.incrby(key, n)
await r.expire(key, 90000) # 25 小时，跨天自动过期
async def burn_ratio(r, platform: str, api: str, daily_quota: int) -> float:
"""
燃烧比 = 已用比例 / 时间进度比例
> 1.3 报警：按这个速度，今天不够用
< 0.5 提示：额度有富余，可以提高并发
"""
day = datetime.now(TZ).strftime("%Y%m%d")
used = int(await r.get(f"quota:{platform}:{api}:{day}") or 0)
now = datetime.now(TZ)
elapsed = (now.hour * 3600 + now.minute * 60 + now.second) / 86400
progress = max(elapsed, 1 / 1440) # 防止零点除零
return (used / daily_quota) / progress
```
这个 `burn_ratio` 是我们监控里最有用的一个指标。**它能提前 4 小时预测"今天配额会不够"**，而不是等到撞墙了才知道。
而且它是双向的：比值 0.5 说明额度有富余，那段时间可以把并发调高一点，把活干完。
---
## 三、参数一：`concurrency`——它不是一个可以随便拍的数
标题说三个参数，第一个就打破你的直觉：**并发不是自变量，它是因变量。**
恒等式：
```
在途请求数 = QPS × 平均耗时
```
也就是说：
```python
concurrency ≈ target_qps × p95_latency_seconds
```
举个具体的：目标 100 QPS，p95 延迟 300ms → `100 × 0.3 = 30`。
**为什么用 p95 不用平均值**：因为撞墙往往发生在长尾。平均 200ms、p95 800ms 的接口，你按平均值算出来 20，实际会遇到"20 个位置里 5 个卡了 8 秒"，然后队列瞬间堆起来，后面全在等。按 p95 算，留的余量正好覆盖长尾。
两个常见的错法：
| 错法 | 现象 | 后果 |
| ---------- | ---------- | ---------------------- |
| 并发给太大（200） | 90% 的请求在排队 | 内存和 fd 被占满，P99 恶化，还撞限流 |
| 并发给太小（10） | 打不满目标 QPS | 吞吐上不去，任务跑不完 |
第二种更容易被忽略。`concurrency=10`、延迟 300ms → 实际吞吐上限 33 QPS。你想跑 100 QPS，永远跑不到，而且看日志全是"正常返回"，没有报错。
```python
# throttle/gate.py
import asyncio
class GateBusy(Exception):
pass
class Gate:
"""
每个 (platform, api) 一个 Gate。
并发 + 令牌桶 + AIMD，三件东西绑在一起，因为它们必须同步调整。
"""
def __init__(self, name: str, bucket, adaptive, concurrency: int):
self.name = name
self.bucket = bucket
self.adaptive = adaptive
self.concurrency = concurrency
self._sem = asyncio.Semaphore(concurrency)
async def __aenter__(self):
# 先占并发槽，再等令牌。
# 反过来的话，等令牌的请求会把槽全占住，
# 明明桶里还有令牌也没人能拿到（真实的坑，别问怎么知道的）
try:
await asyncio.wait_for(self._sem.acquire(), timeout=1.0)
except asyncio.TimeoutError:
raise GateBusy(f"{self.name} concurrency={self.concurrency} 已满")
try:
await self.bucket.wait(1)
except Exception:
self._sem.release()
raise
return self
async def __aexit__(self, exc_type, exc, tb):
self._sem.release()
if exc is None:
await self.adaptive.on_success()
elif getattr(exc, "rate_limited", False):
await self.adaptive.on_limited()
return False
```
注意 `__aenter__` 里那个顺序，以及那个 1 秒超时。并发满了就**立刻告诉调用方"忙"**，别让它排队——排队会掩盖真实压力，等服务崩了你才发现队列已经三千长了。
---
## 四、参数二：令牌桶——`rate` 管长期，`burst` 管瞬时
这两个是文章标题里的第二个参数，但其实是两个数：**速率**决定长期，**桶深**决定瞬时。
### 为什么必须有 burst
电商接口调用是**脉冲式**的，不是均匀分布的。一次铺货 500 个商品，就是突然一批并发请求涌进来。你的令牌桶如果 `burst = rate`（每秒回一个就只能存一个），那这 500 个请求会被均匀摊到 500/rate 秒里——本来 10 秒能干完的活，变成 50 秒。
### burst 该多大
两个约束同时要满足：
```
① burst ≥ 平台单秒允许的突发量 × 0.5 （别浪费突发能力）
② burst + rate × 1s ≤ 平台单秒硬上限 （别打穿）
```
经验值 `burst = rate × 2 ~ 5`。太小失去突发能力，太大直接撞墙。
举个反例：我们最早给淘宝 `item_get` 设 `rate=80, burst=500`。因为那个接口日配额够大，想着"一次多放点没关系"。结果每 6 秒就有一次 500 连发的脉冲，平台侧按秒窗口算，直接 429。**rate 明明只有 80，但瞬时到了 500。**
### 分布式令牌桶（Lua + Redis）
单机令牌桶在多实例部署下是灾难：4 个 worker 各算各的，实际 QPS 是配置值的 4 倍。所以必须集中式。
```lua
-- throttle/bucket.lua
-- KEYS[1] 桶 key
-- ARGV[1] rate（每秒回多少）ARGV[2] burst（最大存量）
-- ARGV[3] now_ms ARGV[4] want（这次要几个）
local rate = tonumber(ARGV[1])
local burst = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local want = tonumber(ARGV[4])
local d = redis.call('HMGET', KEYS[1], 'tokens', 'ts')
local tokens = tonumber(d[1])
local ts = tonumber(d[2])
if tokens == nil then
tokens = burst
ts = now
end
local delta = math.max(0, now - ts) / 1000
tokens = math.min(burst, tokens + delta * rate)
if tokens < want then
local need_ms = math.ceil((want - tokens) / rate * 1000)
return {0, need_ms}
end
tokens = tokens - want
redis.call('HMSET', KEYS[1], 'tokens', tokens, 'ts', now)
redis.call('PEXPIRE', KEYS[1], math.ceil(burst / rate * 1000) + 5000)
return {1, 0}
```
```python
# throttle/bucket.py
import asyncio
import time
from pathlib import Path
class RateLimitTimeout(Exception):
def __init__(self, key: str, wait: float):
super().__init__(f"{key} 排队超时，还需 {wait:.2f}s")
self.rate_limited = True
class TokenBucket:
def __init__(self, redis, key: str, rate: float, burst: int):
self.r = redis
self.key = f"rl:{key}"
self.rate = rate
self.burst = burst
self._lua = Path("throttle/bucket.lua").read_text(encoding="utf-8")
self._sha_id: str | None = None
async def _script_sha(self) -> str:
if self._sha_id is None:
self._sha_id = await self.r.script_load(self._lua)
return self._sha_id
async def acquire(self, want: int = 1) -> float:
"""返回 0 表示拿到；> 0 表示还差多少秒"""
now_ms = int(time.time() * 1000)
ok, wait_ms = await self.r.evalsha(
await self._script_sha(), 1, self.key,
self.rate, self.burst, now_ms, want,
)
return 0.0 if ok else wait_ms / 1000
async def wait(self, want: int = 1, timeout: float = 60.0) -> None:
deadline = time.monotonic() + timeout
while True:
delay = await self.acquire(want)
if delay == 0:
return
if time.monotonic() + delay > deadline:
raise RateLimitTimeout(self.key, delay)
await asyncio.sleep(delay)
```
一个容易忽略的点：**lua 脚本用 `evalsha` 而不是 `eval`**。`eval` 每次都传脚本全文，一个大脚本几百字节，高频调用时这部分网络开销不小；`evalsha` 只传 40 字节的 sha1。用 `script_load` 缓存，节点重启后 `evalsha` 会报 `NOSCRIPT`，需要捕获后重新 load（上面为了简洁省略了，生产环境要补）。
---
## 五、参数三：退避曲线——退多快，恢复多慢
前两个参数决定"会不会撞"，这一个决定"撞了之后多久恢复"。它是一对系数（降幅 + 恢复步长），合成一条曲线。
三种错法，按危害排序：
| 错法 | 代码 | 为什么错 |
| -------- | ------------------------ | ----------------------------------- |
| 固定 sleep | `await asyncio.sleep(1)` | 撞墙期间每秒都撞一次，1 秒变 1 小时；而且**每次失败都烧配额** |
| 直接放弃 | `raise` | 大批任务失败，客户重跑，额度烧得更多 |
| 无脑重试 | `for _ in range(5)` | 雪崩，还可能触发平台的"恶意调用"风控 |
正确做法是 AIMD（加法增、乘法减），但标准 AIMD 有个缺陷：**它不记得上次撞在哪**。
所以每次恢复都要重新"试探"过去，第二次撞在同一个位置。加一个临界点记忆就能解决：
```python
# throttle/adaptive.py
import time
class AdaptiveRate:
"""
AIMD + 临界点记忆
撞限流 → rate *= 0.5 （乘法降，退得快）
连续成功 N → rate += step （加法升，进得慢）
记忆 → 记下撞墙时的 rate，恢复不超过它的 90%
最后这条是关键：没有它，每次恢复都会重新撞在同一个地方。
"""
def __init__(self, rate: float, min_rate: float = 1.0,
max_rate: float = 500.0, recover_after: int = 50,
step: float = 1.0, memory_ratio: float = 0.9):
self.rate = rate
self.min_rate = min_rate
self.max_rate = max_rate
self.recover_after = recover_after
self.step = step
self.memory_ratio = memory_ratio
self.ceiling: float | None = None # 上次撞墙时的速率
self._ok_streak = 0
self._last_hit = 0.0
async def on_success(self):
self._ok_streak += 1
if self._ok_streak < self.recover_after:
return
self._ok_streak = 0
new_rate = self.rate + self.step
if self.ceiling is not None:
new_rate = min(new_rate, self.ceiling * self.memory_ratio)
self.rate = min(self.max_rate, new_rate)
async def on_limited(self):
now = time.monotonic()
# 5 秒内的第二次撞击才更新 ceiling，
# 避免一次偶发毛刺把 ceiling 永久压低
if now - self._last_hit > 5.0 or self.ceiling is None:
self.ceiling = max(self.ceiling or 0, self.rate)
self._last_hit = now
self._ok_streak = 0
self.rate = max(self.min_rate, self.rate * 0.5)
```
`memory_ratio=0.9` 这个值有讲究：太低（0.5）恢复太保守，明明还有余量却上不去；太高（0.99）会频繁在边界试探。0.9 是实测下来最平衡的——撞墙一次后，稳定在临界值的 90%，之后很少再撞。
还有一个**必须做但很少有文章提**的点：`on_limited` 之后，那个请求**不能再立刻重试**。它要退回去排队，等新速率生效。否则你降了速率，但手上这批已经在途的请求还在猛冲，等于没降。
---
## 六、把三个参数装到一起
```python
# throttle/registry.py
import asyncio
class Throttle:
"""
按 (platform, api) 生成 Gate。
参数不是写死的常量，而是从配置里读——因为三个平台的三个参数值完全不同。
"""
def __init__(self, redis, config: dict):
self.r = redis
self.cfg = config
self._gates: dict[tuple[str, str], Gate] = {}
self._lock = asyncio.Lock()
async def get(self, platform: str, api: str) -> Gate:
key = (platform, api)
gate = self._gates.get(key)
if gate is not None:
return gate
async with self._lock: # 双检锁，避免并发建重复桶
gate = self._gates.get(key)
if gate is not None:
return gate
c = self.cfg.get(platform, {}).get(api, self.cfg["default"])
bucket = TokenBucket(self.r, f"{platform}:{api}",
rate=c["rate"], burst=c["burst"])
gate = Gate(
name=f"{platform}/{api}",
bucket=bucket,
adaptive=AdaptiveRate(rate=c["rate"]),
concurrency=c["concurrency"],
)
self._gates[key] = gate
return gate
async def call(self, platform: str, api: str, fn, **kwargs):
gate = await self.get(platform, api)
# 速率被 AIMD 调过之后，令牌桶要跟上
gate.bucket.rate = gate.adaptive.rate
async with gate:
try:
return await fn(**kwargs)
except Exception as e:
if getattr(e, "rate_limited", False):
# 这里退避一下，让降速生效再回来
await asyncio.sleep(1.0)
raise
```
配置长这样（三个平台的值完全不一样，这就是为什么不能写死）：
```yaml
# throttle.yaml
default:
rate: 20
burst: 60
concurrency: 8
taobao:
item_get: { rate: 80, burst: 300, concurrency: 24 } # 日配额大，可冲
trade_list: { rate: 30, burst: 90, concurrency: 9 } # 敏感接口，保守
logistics_trace: { rate: 10, burst: 20, concurrency: 3 } # 限流最严
jd:
item_get: { rate: 40, burst: 120, concurrency: 12 }
order_list: { rate: 15, burst: 45, concurrency: 5 }
alibaba:
item_get: { rate: 25, burst: 75, concurrency: 8 }
```
---
## 七、调完之后的数字
同一批任务（10 万商品铺货 + 3 万订单同步），平台的额度一次没改：
| 指标 | 调参前 | 调参后 | 靠什么 |
| --------- | ------------- | --------------- | ------------------ |
| 撞限流次数 / 天 | 3,400 | **12** | burst 调小 + AIMD 记忆 |
| 平均 QPS | 38（峰值乱冲到 200） | **稳定 92** | 桶速率 + 并发按 p95 算 |
| 任务耗时 | 5h20m（一半在退避） | **1h05m** | 三件叠加 |
| 配额利用率 | 61%（失败也烧） | **97%** | 重试分级 + 分接口桶 |
| 配额燃烧比 | 下午 3 点就 > 1.5 | 全天稳定在 0.95~1.05 | burn_ratio 监控 |
最值得看的是第三行。**耗时从 5 小时 20 分降到 1 小时 05 分，但 QPS 只从 38 涨到 92**——按比例算应该只快 2.4 倍，实际快了 4.9 倍。
多出来的那部分是"不再退避"。调参前那 5 小时里，有一半时间是在 sleep 等重试。**限流最大的成本不是吞吐低，是时间被 sleep 吃掉了。**
---
## 八、四个反直觉的结论
**1. 并发调大不会更快，只会让长尾更慢。**
并发超过 `QPS × p95延迟` 之后，多出来的请求只是在队列里等着，同时占用内存和文件句柄。真正的吞吐由令牌桶的 rate 决定，不由并发决定。
**2. burst 是三个参数里最危险的一个。**
rate 决定长期，出不了大事；burst 决定瞬时，设大了直接打穿平台窗口。而且它出问题的表现很迷惑——平均 QPS 明明没超，但就是天天撞 429。
**3. 撞限流后立刻降速，恢复反而更快。**
因为撞墙期间每次失败都在烧配额，而且失败请求也要等超时。降速看着"退了一步"，实际上是止损。
**4. 减少请求数比调任何参数都有效。**
调参最多把效率从 61% 提到 97%，这是 1.6 倍。而把物流查询从 71,000 次降到 12,000 次，是 6 倍的额度释放。**调参是分配问题，降量是供给问题。** 先做降量。
---
## 九、最后
三个参数，说到底是在回答三个问题：
- `concurrency`：**同时在途多少个**（由 QPS 和延迟推出来，不是拍的）
- `rate / burst`：**每秒放多少个，能突然放多少个**（长期与瞬时的分离）
- `backoff`：**撞了之后退多快、恢复多慢**（决定故障时长，不是决定吞吐）
再加一个不在标题里但最重要的：**先算账**。三笔账算完，你大概会发现真正的瓶颈不在限流上。
最后再说一次开头那句话：这套东西的目标是**把额度用满**，不是**突破额度**。前者是工程，后者是违规。我们做接口服务这些年，见过太多为了省点调用量去搞多 AppKey 分摊的，最后账号全封、客户数据全断——那个损失比省下来的钱大得多。
代码里 `bucket.lua` 和 `adaptive.py` 是不依赖业务的，可以单独抽出来用；`registry.py` 的配置结构按你们自己的接口清单改。所有阈值都建议先跑一周观察日志再定，别照抄我这几个数——平台额度、你的客户结构、调用模式都不一样。
---
*附：文中淘宝 `item_get` 的 `rate=80` 是我们账号的实际额度推导出来的，不同账号、不同应用等级差别很大。接入前先用第七篇那套 `StageTimer` 跑一天，看真实 QPS 曲线和平台报错分布，再定这三个数。*未分类
接口稳定率 99.9% 背后，我们在链路层做了这 5 件事（附 Python 源码）
万邦科技Lex 发表于2026-10-26 浏览69 评论0
# 接口稳定率 99.9% 背后，我们在链路层做了这 5 件事（附 Python 源码）
先说个数字：99.9% 的可用性，摊到 30 天里是 **43 分钟**。
43 分钟，一个月。听起来还有余量对吧？但这是**全部**——三个平台加在一起，所有接口、所有客户、所有时段，一个月累计只能有 43 分钟不达标。
我们做电商数据接口（淘宝 / 1688 / 京东），上游是别人家的系统。上游什么时候抖动、什么时候限流、什么时候半夜改字段，你控制不了。所以这个数字不可能靠"不出事"达成，只能靠**出事的时候影响面被限制住**。
这篇文章讲的就是"限制影响面"这件事：在业务代码和平台 API 之间加的那一层，我们叫**链路层**，里面做了五件事。
代码是 Python + asyncio + httpx，可以直接拿去用。
---
## 一、先把 99.9% 的口径定下来，否则没法谈
这是最多人跳过的一步，也是后面所有优化没意义的原因。
同一个系统，三种算法能算出三个数：
| 口径 | 怎么算 | 数字好看程度 | 有没有意义 |
| --------- | ----------------- | ------ | ------------------ |
| 进程存活率 | 服务没挂就算成功 | 99.99% | 基本没意义，挂了才叫故障 |
| HTTP 成功率 | 返回 200 就算成功 | 99.5% | 有点意义，但 200 里可能有错误体 |
| **业务成功率** | 拿到**正确的业务数据**才算成功 | 99.9% | 有意义，但最难做 |
我们说的是第三种。定义写清楚：
```
可用性 = 成功请求数 / 应当成功的请求数
```
两个关键词要解释：
**"成功"** —— 不是 HTTP 200，是拿到结构完整、字段非空、语义正确的业务数据。淘宝的接口 HTTP 永远 200，错误信息藏在 body 的 `error_response` 里。只统计状态码，你的监控永远是绿的。
**"应当成功"** —— 这个限定词救了我们。客户传错参数、AppKey 拼错、没授权就调接口，这些请求怎么算都是失败，而且**再怎么优化链路也不会成功**。
真实情况是：这类"调用方错误"占我们总请求量的 **6% 左右**。如果把它们算进分母，可用性的理论上限就是 94%，永远上不了 99.9%。
所以口径是：调用方错误**排除出分母**，但**单独监控**。一旦这个比例突然从 6% 涨到 15%，说明文档有问题，或者某个客户在乱调，得去查（通常是我们自己的锅：接口文档写得不清楚）。
> 这一步的价值不在于算出什么数字，而在于**逼你想清楚"失败"到底指什么**。想清楚了，后面五件事才知道该往哪使劲。
---
## 二、链路层在哪，它管什么
```
业务代码（下单、铺货、同步）
│ 它只知道：调 get_item(id)，要么拿到数据，要么拿到异常
▼
┌──────────────────────────────────────────┐
│ 链路层（本文主角） │
│ ① 舱壁隔离 ② 超时预算 ③ 重试分级 │
│ ④ 熔断降级 ⑤ 探测与归一 │
└──────────────────────────────────────────┘
│ 三平台各自 client，各自限流，各自熔断
▼
淘宝开放平台 / 京东宙斯 / 1688 网关
```
三条设计原则，后面每个模块都在守：
1. **业务代码看不见重试、限流、超时**——它只描述"我要什么"，不描述"失败了怎么办"。
2. **故障隔离在平台粒度**——淘宝挂了不能拖累京东，京东限流不能让淘宝的请求排队。
3. **每一层都必须能"快速失败"**——宁可 100ms 返回一个明确的"现在不行"，也不要 30s 后返回一个超时。两者对业务的意义完全不同。
---
## 三、第 1 件事：舱壁隔离，让一个平台的病不传染
最经典的翻车场景：三个平台共用一个 `httpx.AsyncClient`，淘宝那侧开始超时，连接全部卡在淘宝上，连接池占满，京东的请求连连接都拿不到——**京东明明好好的，也跟着挂**。
解法是造船上的"舱壁"：每个平台一套独立的资源，进水只淹一个舱。
```python
# link/bulkhead.py
import asyncio
import httpx
class BulkheadFull(Exception):
def __init__(self, name: str, concurrency: int):
super().__init__(f"bulkhead {name} full (concurrency={concurrency})")
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class Bulkhead:
"""舱壁：每个平台一套并发额度，互不挤占"""
def __init__(self, name: str, concurrency: int = 16):
self.name = name
self.concurrency = concurrency
self._sem = asyncio.Semaphore(concurrency)
async def __aenter__(self):
try:
# 注意这里是 0.5s，不是无限等
# 排队本质上也是一种失败，只是它看起来像"在正常工作"
await asyncio.wait_for(self._sem.acquire(), timeout=0.5)
except asyncio.TimeoutError:
raise BulkheadFull(self.name, self.concurrency)
return self
async def __aexit__(self, *exc):
self._sem.release()
return False
def make_client(platform: str) -> httpx.AsyncClient:
"""每个平台一个 client，参数不是随便填的"""
return httpx.AsyncClient(
http2=True,
limits=httpx.Limits(
max_connections=32,
max_keepalive_connections=16,
# ↓ 这一行是本篇最值钱的一行，下面有血案
keepalive_expiry=25.0,
),
timeout=httpx.Timeout(connect=2.0, read=8.0, write=2.0, pool=1.0),
follow_redirects=False,
)
```
三个参数值得单独说：
**`concurrency` 怎么定**：`QPS × 平均耗时`。假设某平台限流 100 QPS，平均响应 300ms，那理论并发是 30。取 16 是留了余量——并发给多了不会更快，因为**限流才是物理下限**，多出来的只是在网关排队，还白占内存。
**`pool` 超时 1 秒**：这是"拿不到连接"的等待上限。默认不设或者设很长，意味着连接池被占满时新请求会一直等，等到天荒地老，然后一起超时。1 秒拿不到就直接认输，快得多。
**`keepalive_expiry=25`**：这一行有故事，我单独讲。
### 血案复盘：一次因为 35 秒导致的成功率暴跌
某个凌晨，京东侧的成功率从 99.95% 掉到 92%，持续了两个小时。监控上没有任何"错误"，延迟 P99 从 400ms 涨到 8s。
查出来的原因很蠢：
- httpx 默认 `keepalive_expiry=5.0`，我们当时手写成了 `60.0`，觉得"连接复用久一点更好"
- 京东网关侧的空闲连接回收是 **30 秒**
- 于是 30~60 秒这个区间里的连接，是**对端已经关掉、我们以为还活着**的死连接
- 请求发出去，一直等 `read` 超时（8 秒），然后重试，重试又可能拿到另一个死连接
一个死连接 = 8 秒浪费。那晚 60% 的请求撞上了。
修法就是 `keepalive_expiry` 必须**小于**对端的空闲回收时间。拿不到对端确切值时，取 20~30 秒是最安全的区间。另外加一条：`RemoteProtocolError` 必须**不消耗重试次数**——那是连接级别的偶发问题，换条连接就好，不该算成"这次调用失败了一次"。
> 这个坑的可怕之处：**它不报错，只是慢**。你的错误率监控完全正常，只有延迟曲线在涨。如果你只看错误率不看延迟分布，这类问题能藏几个月。
---
## 四、第 2 件事：超时三级拆分 + 预算传递
"超时设 30 秒"是另一个常见错误。它有个隐藏假设：**一次业务调用只有一个请求**。
实际上一次铺货可能串 20 个请求。每个都 30 秒，最坏情况 10 分钟——用户早就关了页面，你的线程还在跑。
两个动作：
**（1）超时按接口类型拆开，不是一刀切**
| 接口类型 | connect | read | 理由 |
| --------- | ------- | ---- | -------------- |
| 列表 / 增量查询 | 2s | 3s | 数据量小，慢了就是有问题 |
| 商品详情 | 2s | 6s | 详情字段多，正常要 1~2s |
| 批量提交 / 导出 | 2s | 15s | 平台侧真的在跑任务 |
| 物流轨迹 | 2s | 4s | 单条查询，快 |
**（2）总预算往下传，剩余不够就立刻放弃**
```python
# link/budget.py
import time
class Budget:
"""
一次业务调用分给链路的剩余时间。
核心思想：超时不是「每个请求各自计时」，是「一共就这么久，你自己分」。
"""
__slots__ = ("_deadline", "_total")
def __init__(self, total_ms: int):
self._total = total_ms / 1000
self._deadline = time.monotonic() + self._total
def left(self) -> float:
return self._deadline - time.monotonic()
def slice(self, want: float) -> float:
"""子请求可用超时：不超过剩余预算，也不低于 200ms"""
return max(0.2, min(want, self.left()))
def enough(self, need: float) -> bool:
"""还剩的时间够不够再试一次？不够就别试了"""
return self.left() >= need
```
用起来是这样：
```python
budget = Budget(total_ms=8000) # 这个业务动作总共 8 秒
for item_id in item_ids:
if not budget.enough(1.0): # 连 1 秒都不剩了
raise BudgetExhausted(f"剩 {budget.left():.2f}s，还有 {len(rest)} 个没处理")
await gw.call("item_get", timeout=budget.slice(6.0), ...)
```
**为什么要这么麻烦**：因为"超时"和"重试"必须联动。预算只剩 1 秒，你还按 6 秒超时去发请求、失败了再重试三次——这 1 秒变成了 20 秒，而且必然失败。有了预算，链路层就能回答一个业务层答不出的问题：**现在还有没有必要试**。
> 一个实现细节：`asyncio.wait_for` 超时之后，底层请求**不一定**立刻取消干净（尤其是 HTTP/2 多路复用下）。所以 httpx 自己的 `timeout` 必须比 `wait_for` 小一点，让 httpx 先超时、自己清理连接；`wait_for` 只做最后兜底。反过来用，会漏连接。
---
## 五、第 3 件事：重试分级——别把所有错误都当"再试一次"
先说最反直觉的一点：**大部分重试是无效的，而且有害**。
签名错了，重试一百次还是签名错，只是把配额烧光。参数少了，重试一百次还是少。权限没开，重试一百次还是没开。这些错误重试的唯一效果是：**把一次明确的失败，变成一次很慢的失败**。
所以第一件事是把三平台几百个错误码压成几类：
```python
# link/errors.py
from enum import Enum
import httpx
class Kind(str, Enum):
OK = "ok"
RETRY = "retry" # 瞬时故障：网络抖动、5xx、网关超时 → 可立刻重试
SLOW = "slow" # 限流、配额耗尽 → 退避久一点再试
FATAL = "fatal" # 参数 / 签名 / 权限 → 重试无意义，直接抛
UNKNOWN = "unknown" # 没见过 → 不重试，立即告警，人工看
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ---- 平台业务码 → 归一 ----
_TAOBAO = {
"7": Kind.SLOW, # app call limited（限流）
"15": Kind.RETRY, # remote service error
"21": Kind.FATAL, # missing params
"22": Kind.FATAL, # invalid params
"26": Kind.SLOW, # 调用量超限
"27": Kind.FATAL, # 权限不足
"28": Kind.FATAL, # 应用被禁用
"33": Kind.FATAL, # 签名错误
"40": Kind.FATAL, # 会话过期
}
_JD = {
"1001": Kind.FATAL,
"1002": Kind.FATAL,
"2001": Kind.RETRY,
"3001": Kind.SLOW,
}
_ALIBABA = {
"400": Kind.FATAL,
"401": Kind.FATAL,
"429": Kind.SLOW,
"500": Kind.RETRY,
}
_BY_PLATFORM = {"taobao": _TAOBAO, "jd": _JD, "alibaba": _ALIBABA}
def classify(platform: str, exc: Exception | None = None,
code: str | None = None) -> Kind:
if exc is not None:
if isinstance(exc, (httpx.ConnectTimeout, httpx.ReadTimeout,
httpx.WriteTimeout, httpx.PoolTimeout,
httpx.ConnectError)):
return Kind.RETRY
# 连接被对端关掉：换条连接就好，不是业务失败
if isinstance(exc, httpx.RemoteProtocolError):
return Kind.RETRY
if isinstance(exc, httpx.HTTPStatusError):
s = exc.response.status_code
if s == 429:
return Kind.SLOW
if 500 <= s < 600:
return Kind.RETRY
return Kind.FATAL
return Kind.UNKNOWN
return _BY_PLATFORM.get(platform, {}).get(str(code), Kind.UNKNOWN)
```
第二件事是给重试加**全局预算**。这一条很多人没做，但它是防止"重试风暴"的唯一手段。
原理解释一下：上游开始抖动 → 10% 请求失败 → 重试 → 变成 30% 请求量 → 上游更扛不住 → 失败率涨到 30% → 重试 → 请求量翻倍……**重试本身成了压垮上游的那根稻草**。这就是雪崩。
```python
# link/retry.py
import random
import time
class RetryBudget:
"""
全局重试预算：按速率发令牌，用完就不许重试。
正常时几乎无感；上游抖动时它会自动把重试量摁住。
"""
def __init__(self, capacity: int = 20, window: float = 10.0):
self.capacity = capacity # 窗口内最多允许多少次重试
self.window = window
self._tokens = float(capacity)
self._last = time.monotonic()
def _refill(self):
now = time.monotonic()
rate = self.capacity / self.window
self._tokens = min(self.capacity, self._tokens + (now - self._last) * rate)
self._last = now
def allow(self) -> bool:
self._refill()
if self._tokens < 1:
return False
self._tokens -= 1
return True
def backoff(attempt: int, base: float = 0.2, cap: float = 4.0) -> float:
"""
指数退避 + full jitter
full jitter 很重要：不加随机的话，一批同时失败的请求会同时重试，
形成周期性的尖峰，把上游二次打挂
"""
return random.uniform(0, min(cap, base * 2 ** attempt))
```
第三件事，也是最容易被忽略的：**写接口不能盲目重试**。
`item_update` 这种接口，超时了不代表没执行成功。可能平台已经改完价了，只是响应没回来。你重试一次，就改了两次——如果平台侧按"每次调用扣一次额度"计费，或者更新操作会触发审核流程，这就是实打实的损失。
规则：
- **读接口**：可以重试，最多 3 次
- **写接口**：必须带业务幂等键（`out_request_id`，用你自己的订单号/任务号），并且只在**确认没送达**的情况下重试（`ConnectError` 可以，`ReadTimeout` 不行）
- **没有幂等键的写接口**：一次都不重试，失败直接抛给业务层，让业务层用业务状态去判断
> 还有一个结构性问题：**重试必须收口在一层**。如果适配器里重试 3 次、网关层再重试 3 次、业务层还重试 3 次，一次调用放大成 27 次。这个问题在代码 review 时看不出来，只有在大促当天能看出来——那时候你已经没时间改了。
---
## 六、第 4 件事：熔断，加上"带标记的降级"
重试解决"偶发失败"，熔断解决"持续失败"。
区别在于：上游只是抖一下，重试有用；上游已经挂了，重试只是在浪费自己的时间。熔断器就是那个替你判断"它是不是已经挂了"的东西。
```python
# link/breaker.py
import time
from collections import deque
class Breaker:
"""
closed → 正常放行，滑窗内失败率超阈值 → open
open → 直接快速失败（不请求上游），冷却 30s → half_open
half_open → 只放 3 个探测，2 个成功 → closed，任一失败 → open
注意 half_open 放 3 个而不是 1 个：
只放 1 个的话，那个请求恰好赶上抖动，就要再等一个完整冷却期，
恢复会变得非常慢。这是实测出来的参数，不是拍脑袋。
"""
def __init__(self, window=10.0, min_samples=20, fail_ratio=0.5,
cooldown=30.0, probe_quota=3, probe_need=2):
self.window = window
self.min_samples = min_samples # 样本太少不算，避免启动就被熔断
self.fail_ratio = fail_ratio
self.cooldown = cooldown
self.probe_quota = probe_quota
self.probe_need = probe_need
self.state = "closed"
self._events: deque[tuple[float, bool]] = deque()
self._opened_at = 0.0
self._probes_left = 0
self._probe_ok = 0
def allow(self) -> bool:
if self.state == "closed":
return True
if self.state == "open":
if time.monotonic() - self._opened_at >= self.cooldown:
self.state = "half_open"
self._probes_left = self.probe_quota
self._probe_ok = 0
return True
return False
if self._probes_left > 0:
self._probes_left -= 1
return True
return False
def record(self, ok: bool):
now = time.monotonic()
if self.state == "half_open":
if ok:
self._probe_ok += 1
if self._probe_ok >= self.probe_need:
self.state = "closed"
self._events.clear()
else:
self._trip(now)
return
self._events.append((now, ok))
cutoff = now - self.window
while self._events and self._events[0][0] < cutoff:
self._events.popleft()
if len(self._events) >= self.min_samples:
fails = sum(1 for _, ok_ in self._events if not ok_)
if fails / len(self._events) > self.fail_ratio:
self._trip(now)
def _trip(self, now: float):
self.state = "open"
self._opened_at = now
self._events.clear()
```
然后是全篇我最想讲的部分：**降级不是返回错误，是返回"能用的旧数据 + 一个标记"**。
```python
# link/degrade.py
import json
import time
from dataclasses import dataclass
@dataclass
class Stale:
value: dict
age: float # 数据多少秒前抓的
reason: str
def as_payload(self) -> dict:
return {
"data": self.value,
"stale": True,
"age": round(self.age, 1),
"reason": self.reason,
}
class StaleCache:
"""
「最后一份好数据」，和业务缓存分开存。
区别：业务缓存可以被主动失效（因为有人改了东西）；
这份不会——它就是降级时唯一的兜底。
"""
def __init__(self, redis):
self.r = redis
async def put(self, key: str, value: dict):
await self.r.set(f"stale:{key}",
json.dumps({"v": value, "ts": time.time()}),
ex=7 * 86400) # 留 7 天，够久了
async def get(self, key: str, max_age: float = 3600) -> Stale | None:
raw = await self.r.get(f"stale:{key}")
if not raw:
return None
d = json.loads(raw)
age = time.time() - d["ts"]
if age > max_age:
return None # 太旧了，宁可不给
return Stale(value=d["v"], age=age, reason="upstream_unavailable")
```
**为什么这个设计比"返回 503"好**，用铺货场景举例：
- 客户要铺 500 个商品，要读每个商品的价格
- 上游挂了，但 3 分钟前的价格明明还有
- 返回 503 → 客户整个任务失败，重跑一遍
- 返回 `{stale: true, age: 180}` → 客户照常铺货，价格是 3 分钟前的，几乎没影响
而且 `stale: true` 和 `age` 这两个字段是**诚实的**。客户自己决定要不要接受——如果他做的是财务对账，他会拒绝；如果是铺货，他会接受。这个决定权应该在他手里，不该由你替他做。
> 一句话总结这个模块：**99.9% 的含义不是"永远返回正确数据"，是"永远返回可解释的结果"**。抛异常是"我不知道发生了什么"，返回 stale 数据是"我知道发生了什么，这是我能给的最好的"。
---
## 七、第 5 件事：主动探测 + 错误码归一
前四件事都在"出事之后"起作用。这一件是**在客户发现之前发现**。
### （1）探测：不要把客户的第一次请求当成探测
```python
# link/probe.py
import asyncio
import time
# 三平台 × 三类资源，每类一个固定样本
# 这三个样本必须提前准备好，且长期存在（找一个自己店铺的老商品）
PROBES = [
("taobao", "item_get", {"num_iid": "654320987654"}),
("taobao", "trade_list", {"start": "2026-10-01 00:00:00"}),
("jd", "item_get", {"skuId": "100012043978"}),
("jd", "order_list", {"startTime": "2026-10-01 00:00:00"}),
("alibaba", "item_get", {"num_iid": "610947572360"}),
("alibaba", "order_list", {"startTime": "2026-10-01 00:00:00"}),
]
async def probe_once(gateways, metrics):
for platform, api, params in PROBES:
t0 = time.monotonic()
ok, detail = True, "ok"
try:
await gateways[platform].call(
api, params, budget_ms=8000, timeout=5.0,
)
except Exception as e:
ok, detail = False, f"{type(e).__name__}: {e}"
cost = (time.monotonic() - t0) * 1000
metrics.observe(platform=platform, api=f"probe.{api}",
ok=ok, cost_ms=cost)
if not ok:
metrics.alert(f"[探测失败] {platform}/{api} | {detail} | {cost:.0f}ms")
async def probe_loop(gateways, metrics, interval: float = 30.0):
while True:
try:
await probe_once(gateways, metrics)
except Exception as e: # 探测器自己不能挂
metrics.alert(f"[探测器异常] {e}")
await asyncio.sleep(interval)
```
30 秒一次，三平台三类资源，一天 8640 次请求——很小，但它能提前发现三件**不主动查就永远发现不了**的事：
1. **授权过期**。token 到期的那一刻不会有人告诉你，客户第二天来投诉你才知道。
2. **平台改字段**。京东某次把 `wareQd` 改名了，字段还是返回、只是空了。如果你的解析不校验必填，数据就静默变空。探测里做一次字段非空校验就能抓到。
3. **限流额度变了**。平台调整配额从 100 QPS 降到 50 QPS，你不探测就只能等客户撞上去。
### （2）错误码归一：一行结构化日志
```python
# link/metrics.py
import json
import time
class Metrics:
def __init__(self, sink):
self.sink = sink
def observe(self, *, platform: str, api: str, ok: bool,
kind: str = "ok", cost_ms: float = 0.0,
trace_id: str = "", retries: int = 0, stale: bool = False):
# 一行 JSON，不要多行日志
# 多行日志在并发下会串行错乱，grep 的时候想死
self.sink.write(json.dumps({
"ts": round(time.time(), 3),
"trace": trace_id,
"platform": platform,
"api": api,
"ok": ok,
"kind": kind, # ok / retry / slow / fatal / unknown
"cost_ms": round(cost_ms, 1),
"retries": retries,
"stale": stale,
}, ensure_ascii=False) + "\n")
def alert(self, msg: str):
self.sink.write(json.dumps(
{"ts": round(time.time(), 3), "level": "alert", "msg": msg},
ensure_ascii=False) + "\n")
```
`kind` 这个字段是全部监控的基石。有了它，你能直接算出：
- `fatal` 占比 → 文档和参数校验有没有问题
- `slow` 占比 → 配额够不够，是不是该提额了
- `retry` 占比 → **上游健康度**（这个指标比错误率更早预警）
- `stale` 占比 → 降级在多大范围内发生
> 一个经验值：**`retry` 占比超过 5% 就该警惕，超过 10% 说明上游已经不正常了**，即便此时业务成功率还是 100%。这个指标比可用性提前 10~30 分钟反应。
---
## 八、串起来：一个请求要过五道关
```python
# link/gateway.py
import asyncio
import time
from .budget import Budget
from .errors import Kind, classify
from .retry import backoff
class Gateway:
"""链路层门面。业务层只调这一个方法。"""
def __init__(self, platform, client, bulkhead, breaker,
retry_budget, stale_cache, metrics):
self.platform = platform
self.client = client
self.bulkhead = bulkhead
self.breaker = breaker
self.retry_budget = retry_budget
self.stale = stale_cache
self.metrics = metrics
async def call(self, api: str, fn, *, budget_ms: int = 8000,
timeout: float = 6.0, cache_key: str | None = None,
idempotent: bool = True, trace_id: str = "", **kwargs):
"""
fn: 适配器函数，签名 fn(timeout=..., **kwargs) -> dict
"""
budget = Budget(budget_ms)
t0 = time.monotonic()
# 第 4 关：熔断（最先，最便宜）
if not self.breaker.allow():
self.metrics.observe(platform=self.platform, api=api, ok=False,
kind="breaker_open", trace_id=trace_id)
return await self._degrade(cache_key, "breaker_open")
last_exc: Exception | None = None
retries = 0
for attempt in range(3):
# 第 2 关：预算不够就别试了
if not budget.enough(0.3):
last_exc = last_exc or TimeoutError("budget exhausted")
break
# 第 3 关：全局重试预算
if attempt > 0:
if not self.retry_budget.allow():
break
await asyncio.sleep(backoff(attempt - 1))
retries += 1
try:
# 第 1 关：舱壁
async with self.bulkhead:
result = await asyncio.wait_for(
fn(timeout=budget.slice(timeout), **kwargs),
timeout=budget.left(),
)
self.breaker.record(True)
self.metrics.observe(
platform=self.platform, api=api, ok=True,
cost_ms=(time.monotonic() - t0) * 1000,
trace_id=trace_id, retries=retries,
)
if cache_key:
await self.stale.put(cache_key, result)
return result
except Exception as e:
kind = classify(self.platform, exc=e)
self.breaker.record(False)
last_exc = e
self.metrics.observe(
platform=self.platform, api=api, ok=False,
kind=kind.value,
cost_ms=(time.monotonic() - t0) * 1000,
trace_id=trace_id, retries=retries,
)
# 参数错 / 签名错 / 没见过：重试是烧配额
if kind in (Kind.FATAL, Kind.UNKNOWN):
break
# 限流：多等一会儿，别急着再来一次
if kind is Kind.SLOW:
await asyncio.sleep(1.0 + backoff(attempt))
# 写接口：除非在适配器里带了幂等键，否则一次都不重试
if not idempotent:
break
# 第 5 关：降级
stale = await self._degrade(cache_key, str(last_exc))
if stale is not None:
return stale
raise last_exc
async def _degrade(self, cache_key, reason):
if not cache_key:
return None
s = await self.stale.get(cache_key)
if s is None:
return None
self.metrics.observe(platform=self.platform, api="degrade",
ok=True, kind="stale", stale=True)
return s.as_payload()
```
注意最后的 `_degrade`：**降级放在抛出异常之前**。这个顺序是有意的——只有"真的没招了"才降级，而不是一失败就降级（那样会把偶发抖动放大成大面积 stale 数据）。
---
## 九、五件事做完，数字变成了什么样
| 指标 | 做之前 | 做之后 | 主要来自哪件事 |
| ------------ | ---------- | -------------- | ----------- |
| 业务成功率 | 99.21% | **99.94%** | 五件叠加 |
| 单平台故障影响面 | 全站（三平台一起挂） | **该平台降级，其余正常** | ① 舱壁 |
| 故障时 P99 延迟 | 31s（重试到超时） | **1.2s**（快速失败） | ② 预算 + ④ 熔断 |
| 重试放大倍数 | 3.7× | 1.12× | ③ 重试预算 |
| 上游异常发现时间 | 客户投诉后（小时级） | **30 秒内** | ⑤ 探测 |
| 深夜 P0 次数（半年） | 7 | 1 | ④ 熔断 + ⑤ 探测 |
最值得看的是第二行：**故障影响面从"全站"变成"一个平台"**。这件事比把 99.21 提到 99.94 更有价值——因为前者意味着"客户换个平台还能干活"，后者只是"数字好看一点"。
---
## 十、五个反直觉的结论
做完这一套，有几个发现跟直觉不太一样，可能比代码更有用：
**1. 加超时不等于更快，超时必须和重试联动。**
单独设超时，只是把"永久卡住"变成"卡一会儿再卡一会儿"。真正起作用的是总预算——它管的不是单个请求，是整个业务动作的时间上限。
**2. 重试率超过 5%，重试就在让系统变慢。**
因为重试是有成本的：占连接、占并发、占配额，还挤压正常请求。要监控**重试率**而不是重试次数——次数看着不多，比率可能已经失控了。
**3. 熔断打开容易，恢复难。**
`half_open` 只放 1 个探测请求的话，那个请求只要恰好失败，就得再等一个完整冷却期。实测放 3 个、成功 2 个才恢复，恢复速度差好几倍。
**4. 连接池不是越大越好，超过限流额度就是纯浪费。**
`max_connections` 给到 200，但平台限流 50 QPS，多出来的 150 个连接只是在网关排队，同时占着你的内存和文件句柄。连接数应该 ≈ `QPS × 平均耗时`，再留 20%。
**5. 最难的不是技术，是口径。**
把"调用方参数错误"算进分母，可用性上限就是 94%，你把代码优化到极致也上不了 99.9%。先想清楚分母，再谈数字。
---
## 十一、最后
这五件事，本质上是同一件事的五个层面：**承认上游一定会出问题，然后把"出问题"的代价控制住**。
- 舱壁控制**范围**——只影响一个平台
- 预算控制**时间**——1 秒认输，不拖 30 秒
- 重试分级控制**放大**——不让一次失败变成十次请求
- 熔断降级控制**形态**——从"报错"变成"给可解释的旧数据"
- 探测归一控制**发现速度**——客户投诉之前就知道
单看每一件事都不复杂，加起来大概一千多行代码。但它们决定了你半夜能不能睡着。
我们把这套东西在三个平台上跑了两年，中间踩过的坑基本都在上面了（`keepalive_expiry` 那条尤其贵）。如果你正在自己接这三个平台，这套代码可以直接拿去改；如果你觉得这套活不值得自己养，那也正常——我们做的就是这件事。
代码里所有类都是独立可测的，`Breaker`、`RetryBudget`、`Budget` 这三个不依赖网络，建议先写单元测试再接入，别拿线上流量试。
---
*附：文中 `keepalive_expiry` 的推荐值取决于各平台网关的实际配置，25 秒是我们在淘宝/京东/1688 上的经验值，接入前建议先用 `curl -v` 观察 `Connection: close` 的时机再定。*未分类
一套接口打三平台，省下 80% 的对接工时（附 Python 源码）
万邦科技Lex 发表于2026-10-23 浏览79 评论0
# 一套接口打三平台，省下 80% 的对接工时（附 Python 源码）
> 80% 不是省在第一次对接，是省在之后每一次改动。这笔账要算清楚，否则你会以为统一接口层是过度设计。
## 一、三套代码的日常
先看一个很常见的项目结构：
```
project/
├── taobao/
│ ├── client.py # 签名 + 请求
│ ├── product.py # 商品：字段映射、SKU 处理
│ ├── order.py # 订单：分页、状态转换
│ └── logistics.py
├── alibaba/
│ ├── client.py # 又是一份签名
│ ├── product.py # 又是一份字段映射
│ ├── order.py
│ └── logistics.py
├── jd/
│ ├── client.py
│ ├── product.py
│ ├── order.py
│ └── logistics.py
└── service/
├── price_sync.py # ← 业务逻辑，里面全是 if platform == "taobao"
├── stock_sync.py # ← 又一次 if platform == ...
└── order_sync.py # ← 第三次
```
问题不在文件多，在于**每个业务需求都要改三处**。
产品说了句「库存要精确到 SKU」，你得改三个 `stock` 模块。发现淘宝有个坑要修，另外两个平台得确认一遍是不是也有同样的问题。加个新字段，三个平台各写一遍映射。
代码量翻三倍，bug 面积也翻三倍——因为三份实现里总有一份写得跟另外两份不一样。
## 二、先算一笔诚实的账
「省 80% 工时」这种话，如果只算首次对接，是站不住的。算细一点：
### 首次对接
| 项目 | 三套独立实现 | 统一接口层 |
| --------- | ------------- | ----------------- |
| 签名/请求基础设施 | 3 × 1 天 = 3 天 | 1 天（抽象基类） |
| 商品资源适配 | 3 × 2 天 = 6 天 | 3 × 1.5 天 = 4.5 天 |
| 订单资源适配 | 3 × 2 天 = 6 天 | 3 × 1.5 天 = 4.5 天 |
| 物流资源适配 | 3 × 1 天 = 3 天 | 3 × 0.5 天 = 1.5 天 |
| 统一层设计 | 0 | 1.5 天 |
| **合计** | **18 天** | **13 天** |
首次对接只省 **28%**。因为适配器还是得一个一个写，平台差异不会因为抽象就消失。
### 后续每一次改动
真正拉开差距的是这里。假设一个持续一年的项目：
| 变更类型 | 频率 | 三套实现 | 统一层 |
| --------- | ------- | ------------------------ | ----------- |
| 加一个业务字段 | 每月 1 次 | 3 处 × 0.5 天 = 1.5 天 | 1 处 × 0.5 天 |
| 修一个平台 bug | 每月 2 次 | 2 × 0.5 天（有时要排查是否影响其他两份） | 0.5 天 |
| 调整业务规则 | 每季度 1 次 | 3 处 × 1 天 | 1 处 × 1 天 |
| 接第四个平台 | 每年 1 次 | 5 天 | 1.5 天 |
| 新人接手理解成本 | 每年 1 次 | 3 天 | 1 天 |
一年下来：三套实现约 **43 天**，统一层约 **16 天**。
**维护期省 63%。** 加上首次的 28%，以及「bug 只修一处就不会漏」带来的隐性收益，综合到 80% 是合理的。
关键结论：**统一接口层的价值随时间增长，项目越久省得越多。** 如果只做一个短期的数据导出脚本，确实没必要。但如果是长期跑的业务系统，这个投入一年内就回本。
## 三、架构：门面 + 适配器 + 能力声明
```
业务代码（只写一次）
↓
CommerceAPI（门面）
├─ 统一重试 / 缓存 / 限流
├─ 能力检查（不支持就明确报错）
└─ 结果归一化
↓
┌─────────┼─────────┐
TaobaoAdapter AlibabaAdapter JdAdapter
↓ ↓ ↓
各自签名、字段、分页、错误码
```
三个角色的职责必须分清：
- **门面**：业务逻辑唯一入口，负责跨平台共性（重试、缓存、能力检查）
- **适配器**：只做「平台方言 → 统一语言」的翻译，不含业务判断
- **能力声明**：显式声明每个平台支持什么，不支持的操作直接抛错
第三条是最容易被忽略、也最容易埋雷的。下面单独讲。
## 四、代码
### 4.1 统一模型和能力声明
```python
# core/models.py
from dataclasses import dataclass, field
from decimal import Decimal
@dataclass
class Product:
platform: str
item_id: str
title: str
price: Decimal
stock: int | None = None # None 表示该平台不提供
images: list[str] = field(default_factory=list)
skus: list["Sku"] = field(default_factory=list)
@dataclass
class Sku:
sku_id: str
props: dict[str, str]
price: Decimal
stock: int | None = None
@dataclass
class Order:
platform: str
order_id: str
status: str # 统一状态，不是平台原始值
pay_amount: Decimal
created_at: str
modified_at: str
buyer: str = ""
@dataclass
class Capability:
"""
每个平台支持什么，显式声明。
这是统一接口里最关键的一层——三平台能力根本不对等，
不声明清楚，业务代码就会在某个平台上拿到空数据还以为正常。
"""
sku_level_stock: bool = False # 库存是否精确到 SKU
tiered_price: bool = False # 是否有阶梯价（1688 特有）
order_increment: bool = False # 订单是否支持按修改时间增量拉
logistics_subscribe: bool = False # 物流是否支持订阅推送
batch_product_query: bool = False # 商品是否支持批量查询
```
为什么不用 `hasattr` 或者 try/except 判断？因为**能力应该是显式契约，不是运行时探针**。业务代码在启动时就能知道「京东这个接口不支持 SKU 级库存」，从而决定降级策略，而不是等跑到一半发现返回全是 `None`。
### 4.2 适配器基类
```python
# core/adapter.py
from abc import ABC, abstractmethod
from decimal import Decimal
from .models import Product, Order, Capability
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class NotSupported(Exception):
"""平台明确不支持该操作。不要用假数据糊弄调用方。"""
def __init__(self, platform: str, operation: str):
super().__init__(f"[{platform}] 不支持操作：{operation}")
self.platform = platform
self.operation = operation
class BaseAdapter(ABC):
platform: str = ""
capability: Capability = Capability()
# --- 子类必须实现 ---
@abstractmethod
async def get_product(self, item_id: str) -> Product:
...
@abstractmethod
async def list_orders(self, start: str, end: str, page: int = 1) -> list[Order]:
...
# --- 子类按能力选择性实现 ---
async def get_stock(self, item_id: str, sku_id: str | None = None) -> int:
if not self.capability.sku_level_stock and sku_id:
raise NotSupported(self.platform, "按 SKU 查询库存")
return await self._get_stock_impl(item_id, sku_id)
async def _get_stock_impl(self, item_id: str, sku_id: str | None) -> int:
raise NotSupported(self.platform, "库存查询")
# --- 状态归一：每个平台状态码不同，统一成五个 ---
STATUS_MAP: dict[str, str] = {}
def normalize_status(self, raw: str) -> str:
return self.STATUS_MAP.get(str(raw), "UNKNOWN")
```
统一状态只定五个：`PENDING`（待付款）、`PAID`（待发货）、`SHIPPED`（已发货）、`DONE`（已完成）、`CLOSED`（已关闭）。多了没用，少了不够用。
### 4.3 三个适配器（只展示关键差异）
```python
# adapters/taobao.py
from decimal import Decimal
from core.adapter import BaseAdapter
from core.models import Product, Order, Capability, Sku
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class TaobaoAdapter(BaseAdapter):
platform = "taobao"
capability = Capability(
sku_level_stock=True,
order_increment=True,
logistics_subscribe=True,
batch_product_query=False,
)
STATUS_MAP = {
"WAIT_BUYER_PAY": "PENDING",
"WAIT_SELLER_SEND_GOODS": "PAID",
"WAIT_BUYER_CONFIRM_GOODS": "SHIPPED",
"TRADE_FINISHED": "DONE",
"TRADE_CLOSED": "CLOSED",
}
def __init__(self, client):
self.c = client
async def get_product(self, item_id: str) -> Product:
d = await self.c.call("taobao.item.seller.get", {
"fields": "num_iid,title,price,num,quantity,pic_url,sku",
"num_iid": item_id,
})
it = d["item_get_response"]["item"]
skus = []
for s in (it.get("skus") or {}).get("sku") or []:
skus.append(Sku(
sku_id=str(s["sku_id"]),
props=_parse_props(s.get("properties_name", "")),
price=Decimal(str(s.get("price", "0"))),
stock=int(s.get("quantity", 0)),
))
return Product(
platform=self.platform,
item_id=str(it["num_iid"]),
title=it.get("title", ""),
price=Decimal(str(it.get("price", "0"))),
stock=int(it.get("quantity", 0)),
images=[it["pic_url"]] if it.get("pic_url") else [],
skus=skus,
)
async def list_orders(self, start, end, page=1) -> list[Order]:
d = await self.c.call("taobao.trades.sold.get", {
"fields": "tid,status,payment,created,modified,buyer_nick",
"start_modified": start, "end_modified": end,
"page_no": page, "page_size": 100,
})
rows = ((d.get("trades_sold_get_response") or {})
.get("trades") or {}).get("trade") or []
return [Order(
platform=self.platform,
order_id=str(t["tid"]),
status=self.normalize_status(t.get("status", "")),
pay_amount=Decimal(str(t.get("payment", "0"))),
created_at=t.get("created", ""),
modified_at=t.get("modified", ""),
buyer=t.get("buyer_nick", ""),
) for t in rows]
def _parse_props(s: str) -> dict[str, str]:
out = {}
for seg in filter(None, s.split(";")):
p = seg.split(":")
if len(p) >= 4:
out[p[2]] = p[3]
return out
```
```python
# adapters/alibaba.py
from decimal import Decimal
from core.adapter import BaseAdapter
from core.models import Product, Order, Capability
from core.adapter import NotSupported
class AlibabaAdapter(BaseAdapter):
platform = "alibaba"
capability = Capability(
sku_level_stock=False, # 1688 按起订量阶梯，没有 SKU 概念
tiered_price=True,
order_increment=True,
logistics_subscribe=False, # 需要单独申请
)
STATUS_MAP = {
"waitbuyerpay": "PENDING",
"waitsellersend": "PAID",
"waitbuyerreceive": "SHIPPED",
"success": "DONE",
"cancel": "CLOSED",
}
def __init__(self, client):
self.c = client
async def get_product(self, item_id: str) -> Product:
d = await self.c.call("com.alibaba.product.get", {
"productID": item_id, "webSite": "1688",
})
r = d.get("result") or {}
ranges = (r.get("saleInfo") or {}).get("priceRanges") or []
price = Decimal(str(ranges[0]["price"])) if ranges else Decimal("0")
return Product(
platform=self.platform,
item_id=str(r.get("productID", item_id)),
title=r.get("subject", ""),
price=price,
stock=int((r.get("saleInfo") or {}).get("amountOnSale", 0) or 0),
images=[i["fullPathImageURI"]
for i in (r.get("image") or {}).get("images", [])],
)
async def get_stock(self, item_id: str, sku_id: str | None = None) -> int:
# 显式拒绝，而不是忽略 sku_id 返回总量
if sku_id:
raise NotSupported(self.platform, "按 SKU 查询库存（1688 无此概念）")
return await self._get_stock_impl(item_id, None)
async def list_orders(self, start, end, page=1) -> list[Order]:
d = await self.c.call("com.alibaba.trade/get.soldList", {
"modifyStartTime": start, "modifyEndTime": end,
"pageNo": page, "pageSize": 50,
})
rows = (d.get("result") or {}).get("orderList") or []
return [Order(
platform=self.platform,
order_id=str(o.get("orderId")),
status=self.normalize_status(o.get("orderStatus", "")),
pay_amount=Decimal(str(o.get("sumPayment") or "0")),
created_at=o.get("createTime", ""),
modified_at=o.get("modifyTime", ""),
buyer=o.get("toLoginId", ""),
) for o in rows]
```
```python
# adapters/jd.py
from decimal import Decimal
from core.adapter import BaseAdapter
from core.models import Product, Order, Capability
class JdAdapter(BaseAdapter):
platform = "jd"
capability = Capability(
sku_level_stock=True,
order_increment=True,
logistics_subscribe=True,
)
STATUS_MAP = {
"1": "PENDING", # 等待付款
"2": "PAID", # 等待发货
"3": "SHIPPED", # 已发货
"4": "DONE", # 已完成
"6": "CLOSED", # 已取消
}
def __init__(self, client):
self.c = client
async def get_product(self, item_id: str) -> Product:
d = await self.c.call("jd.union.open.goods.query", {
"360buy_param_json": '{"skuIds":[%s]}' % item_id,
})
rows = ((d.get("jd_union_open_goods_query_responce") or {})
.get("result") or {}).get("data") or []
if not rows:
return Product(platform=self.platform, item_id=item_id,
title="", price=Decimal("0"))
g = rows[0]
return Product(
platform=self.platform,
item_id=str(g.get("skuId", item_id)),
title=g.get("skuName", ""),
price=Decimal(str(g.get("price", "0"))),
stock=None, # 该接口不返回库存
images=[g["imageUrl"]] if g.get("imageUrl") else [],
)
async def list_orders(self, start, end, page=1) -> list[Order]:
d = await self.c.call("jingdong.pop.order.search", {
"360buy_param_json": (
'{"page":"%d","pageSize":"100",'
'"startDate":"%s","endDate":"%s"}' % (page, start, end)),
})
rows = ((d.get("jingdong_pop_order_search_responce") or {})
.get("searchorderinfo_result") or {}).get("orderInfoList") or []
return [Order(
platform=self.platform,
order_id=str(o.get("orderId")),
status=self.normalize_status(o.get("orderState", "")),
pay_amount=Decimal(str(o.get("orderPayment") or "0")),
created_at=o.get("orderStartTime", ""),
modified_at=o.get("orderModified", ""),
buyer=o.get("pin", ""),
) for o in rows]
```
三个适配器放在一起看，共性一目了然：**都要签名、都要分页、都要状态映射、都要字段归一**。这些「一样的东西」全部沉到基类，适配器只保留「不一样的东西」。
这就是统一接口层真正省事的地方——不是省打字，是**让「不一样」变得可枚举**。
### 4.4 门面
```python
# core/api.py
import asyncio
from .adapter import BaseAdapter, NotSupported
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class CommerceAPI:
"""
业务代码唯一入口。三平台一个门面，业务层不再出现 platform 判断。
"""
def __init__(self, adapters: dict[str, BaseAdapter]):
self.adapters = adapters
def _get(self, platform: str) -> BaseAdapter:
if platform not in self.adapters:
raise KeyError(f"未注册平台：{platform}")
return self.adapters[platform]
# --- 单平台操作 ---
async def product(self, platform: str, item_id: str):
return await self._get(platform).get_product(item_id)
async def orders(self, platform: str, start: str, end: str,
max_pages: int = 50) -> list:
ad = self._get(platform)
if not ad.capability.order_increment:
raise NotSupported(platform, "订单增量拉取")
out, page = [], 1
while page <= max_pages:
batch = await ad.list_orders(start, end, page)
out.extend(batch)
if len(batch) < 100:
break
page += 1
return out
# --- 跨平台操作：这才是统一接口真正的价值 ---
async def products_all(self, item_ids_map: dict[str, list[str]],
concurrency: int = 10) -> dict[str, list]:
"""
一次拿三个平台的商品。业务层不用写循环、不用管并发。
item_ids_map = {"taobao": [...], "alibaba": [...], "jd": [...]}
"""
sem = asyncio.Semaphore(concurrency)
result: dict[str, list] = {}
async def one(platform: str, iid: str):
async with sem:
try:
return platform, await self.product(platform, iid)
except Exception as e:
print(f"[{platform}] {iid} 失败: {e}")
return platform, None
tasks = [one(p, i) for p, ids in item_ids_map.items() for i in ids]
for platform, prod in await asyncio.gather(*tasks):
if prod is not None:
result.setdefault(platform, []).append(prod)
return result
async def order_summary(self, start: str, end: str,
platforms: list[str] | None = None) -> dict:
"""
跨平台订单汇总。业务层写一行，三平台自动并行。
"""
targets = platforms or list(self.adapters.keys())
async def one(p):
try:
orders = await self.orders(p, start, end)
return p, {
"count": len(orders),
"amount": sum(o.pay_amount for o in orders),
"by_status": _group_status(orders),
}
except Exception as e:
return p, {"error": str(e)}
pairs = await asyncio.gather(*[one(p) for p in targets])
return dict(pairs)
def _group_status(orders) -> dict[str, int]:
out: dict[str, int] = {}
for o in orders:
out[o.status] = out.get(o.status, 0) + 1
return out
```
`order_summary` 这个方法值得多看一眼。在没统一接口的项目里，这段逻辑要写三遍，每遍处理不同的分页、不同的状态码、不同的金额字段。统一之后业务层只写一行。
## 五、同一个需求，两种写法
需求：「拉三个平台昨天的订单，算出各平台待发货数量和金额。」
**没有统一接口：**
```python
# 30 多行，而且每加一个平台就再加 10 行
result = {}
# 淘宝
t_orders, page = [], 1
while True:
d = taobao.call("taobao.trades.sold.get", {...})
rows = d["trades_sold_get_response"]["trades"]["trade"]
t_orders += rows
if len(rows) < 100:
break
page += 1
result["taobao"] = {
"pending": sum(1 for o in t_orders if o["status"] == "WAIT_SELLER_SEND_GOODS"),
"amount": sum(float(o["payment"]) for o in t_orders
if o["status"] == "WAIT_SELLER_SEND_GOODS"),
}
# 1688（字段名完全不同）
a_orders, page = [], 1
while True:
d = alibaba.call("com.alibaba.trade/get.soldList", {...})
rows = d["result"]["orderList"]
a_orders += rows
if len(rows) < 50:
break
page += 1
result["alibaba"] = {
"pending": sum(1 for o in a_orders if o["orderStatus"] == "waitsellersend"),
"amount": sum(float(o["sumPayment"] or 0) for o in a_orders
if o["orderStatus"] == "waitsellersend"),
}
# 京东（又是另一套）
...
```
**有统一接口：**
```python
summary = await api.order_summary("2026-10-22 00:00:00", "2026-10-23 00:00:00")
pending = {p: v["by_status"].get("PAID", 0) for p, v in summary.items()}
```
从 30 行到一个方法调用。而且**加第四个平台，业务代码一行都不用改。**
## 六、能力不对等：统一接口最硬的部分
这是统一接口真正难的地方，也是最容易埋雷的地方。
三个平台的能力根本不对等：
| 能力 | 淘宝 | 1688 | 京东 |
| --------- | --- | ----------- | --------- |
| SKU 级库存 | 支持 | 不支持（按起订量阶梯） | 部分接口支持 |
| 阶梯价 | 不支持 | 支持 | 不支持 |
| 订单按修改时间增量 | 支持 | 支持 | 部分接口只支持按天 |
| 物流订阅推送 | 支持 | 需单独申请 | 支持 |
| 商品批量查询 | 不支持 | 支持 | 支持 |
**最坑的做法是「假装都支持」**：调用方传 `sku_id`，1688 适配器忽略这个参数返回商品总量。业务代码以为拿到了 SKU 库存，实际上是错的，而且不报错。
所以 `get_stock` 里那句显式 `raise NotSupported` 才是正确做法：
```python
if sku_id:
raise NotSupported(self.platform, "按 SKU 查询库存（1688 无此概念）")
```
配套的用法是**启动时检查能力**：
```python
def check_capabilities(api: CommerceAPI, required: dict[str, bool]):
"""
启动时校验：业务需要的能力，各平台是否都支持。
不支持就提前暴露，而不是跑到线上才发现。
"""
problems = []
for platform, ad in api.adapters.items():
for cap_name, needed in required.items():
if needed and not getattr(ad.capability, cap_name, False):
problems.append(f"{platform} 不支持 {cap_name}")
if problems:
raise RuntimeError("能力校验失败：\n" + "\n".join(problems))
# 启动时调用
check_capabilities(api, {
"sku_level_stock": True, # 业务强依赖 SKU 库存
"order_increment": True,
})
```
这段会在启动时告诉你：「1688 不支持 sku_level_stock」。你可以立刻决定是降级、还是排除 1688，而不是等三天后客户投诉库存不准。
## 七、一套测试跑三个平台
统一接口的「统一」不能靠嘴说，得靠测试证明。用 pytest 参数化，同一套契约测试跑三个适配器：
```python
# tests/test_contract.py
import pytest
from decimal import Decimal
from core.adapter import NotSupported
@pytest.fixture
def adapters():
"""三个适配器实例，用测试账号"""
return {"taobao": ..., "alibaba": ..., "jd": ...}
@pytest.mark.parametrize("platform", ["taobao", "alibaba", "jd"])
@pytest.mark.asyncio
async def test_product_shape(adapters, platform, sample_ids):
"""契约一：商品必须返回统一结构"""
p = await adapters[platform].get_product(sample_ids[platform])
assert p.platform == platform
assert isinstance(p.title, str)
assert isinstance(p.price, Decimal)
assert p.price >= 0
assert isinstance(p.images, list)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
@pytest.mark.parametrize("platform", ["taobao", "alibaba", "jd"])
@pytest.mark.asyncio
async def test_order_status_normalized(adapters, platform):
"""契约二：状态必须是五个统一值之一"""
orders = await adapters[platform].list_orders(
"2026-10-15 00:00:00", "2026-10-23 00:00:00")
for o in orders:
assert o.status in {"PENDING", "PAID", "SHIPPED", "DONE", "CLOSED", "UNKNOWN"}
assert isinstance(o.pay_amount, Decimal)
@pytest.mark.asyncio
async def test_unsupported_explicit(adapters, sample_ids):
"""契约三：不支持的能力必须明确报错，不能静默降级"""
with pytest.raises(NotSupported):
await adapters["alibaba"].get_stock(
sample_ids["alibaba"], sku_id="any_sku")
```
第三条测试是整套设计里最重要的一条。它把「不能静默降级」这个约定固化成了测试，防止后来的人图省事改成返回 `0`。
## 八、收益复盘
同一套业务代码，服务三个平台：
| 指标 | 三套实现 | 统一接口层 |
| ------ | ----------- | -------------- |
| 首次对接 | 18 天 | 13 天（省 28%） |
| 一年维护 | 约 43 天 | 约 16 天（省 63%） |
| 业务层代码量 | 30 行 / 每个需求 | 3 行 / 每个需求 |
| 加第四个平台 | 约 5 天 | 约 1.5 天（省 70%） |
| 新人上手 | 读三套代码 | 读一个门面 + 三个薄适配器 |
综合下来，**一年周期省 80% 左右是站得住的**。但前提是：真的长期跑，且平台数量 ≥ 2。
只接一个平台，或者只做一次性的数据导出，这个抽象就是过度设计。工程决策永远看场景。
## 九、四个坑
1. **别在适配器里写业务判断。** 适配器只做翻译。一旦出现 `if 订单金额 > 100` 这种代码，说明业务逻辑漏进了适配器，将来三个平台会各写一份不一样的规则。
2. **能力声明要跟着平台文档更新。** 京东的接口能力经常变（同一个能力，不同接口支持度不同）。建议每个季度对照文档核一次 `Capability`，把它当配置维护，不是当常量。
3. **统一状态不要贪多。** 五个状态够用。给三个平台各定十个细分状态，最后业务代码里全是 `if status in [...]`，反而更乱。
4. **门面不要变成上帝类。** `CommerceAPI` 只放跨平台共性（重试、缓存、并发、能力检查）。具体业务（比如「降价 5% 就告警」）放在 service 层，别塞进门面。
## 十、最后
「一套接口打三平台」的价值，不在少写几行代码，在**让「平台差异」变成一个有边界、可枚举、可测试的东西**。
差异集中在适配器里，业务代码里就再也不该出现 `if platform == "taobao"`。一旦出现了，说明抽象漏了。
代码在上面，`BaseAdapter` + `Capability` + 契约测试这三件套可以直接搬到你的项目里。
如果你连适配器都不想写——签名、字段映射、状态归一、能力差异这些事，现成的三平台 API 接口服务已经做完了，输出就是统一格式，接进来只需要写业务层。省下的正是前面算的那 13 天加 16 天。
有在做的可以聊聊，你们的统一层是怎么处理能力不对等的，是抛错还是降级？
---
**标签**：`Python` `架构设计` `电商` `适配器模式` `API`未分类
从 3 小时到 3 分钟：三平台商品同步是怎么提速的（附 Python 源码）
万邦科技Lex 发表于2026-10-23 浏览80 评论0
# 从 3 小时到 3 分钟：三平台商品同步是怎么提速的（附 Python 源码）
> 60 倍提速不是靠某一个技巧，是四个优化叠加。每个贡献多少，本文算给你看。
## 一、一个真实的慢
某个多平台商家，3000 个商品，每天要同步一次三平台的商品数据（标题、价格、库存、状态）。
原来的脚本：串行逐个拉，每个商品调一次详情接口，拉完逐条写库。跑一次 3 小时。
运营的抱怨很直接：「我早上想看昨天的库存，脚本要到中午才跑完，看的是上午的数据。」
更糟的是，**没人说得清 3 小时都花在哪**。有人说接口慢，有人说服务器不行，有人说数据库写不动。都在猜。
## 二、第一步：先测量，别猜
性能优化的第一原则：**先测量，再动手。** 我在这上面吃过亏——曾经花一天优化数据库索引，后来 profile 一跑发现数据库只占 2% 的时间。
给脚本加了个简单的计时器，跑一次全量，结果如下：
| 环节 | 耗时 | 占比 |
| ---------------- | ------ | ---- |
| 调接口等响应（3000 次串行） | 172 分钟 | 96% |
| JSON 解析 + 字段映射 | 4 分钟 | 2% |
| 逐条写数据库 | 3 分钟 | 2% |
| 其他（重试、日志） | 1 分钟 | 0.5% |
结论非常清晰：**96% 的时间在等网络**。
这不是「接口慢」，是**串行等 3000 次**。每个请求平均 3.4 秒，串行乘以 3000，就是 172 分钟。
这个测量直接决定了优化顺序：先解决等待（并发），再减少请求数（增量），写库那 3 分钟排最后。
测量代码很简单，就几十行，但它决定了后面所有力气往哪使：
```python
# timer.py
import time
from collections import defaultdict
class StageTimer:
"""分阶段计时，跑完输出每阶段耗时和占比"""
def __init__(self):
self.spent = defaultdict(float)
self.counts = defaultdict(int)
self._t0 = None
self._current = None
def start(self, stage: str):
self._current = stage
self._t0 = time.perf_counter()
def stop(self):
if self._current and self._t0 is not None:
self.spent[self._current] += time.perf_counter() - self._t0
self.counts[self._current] += 1
self._current = None
def report(self) -> list[tuple[str, float, float]]:
total = sum(self.spent.values()) or 1.0
rows = sorted(self.spent.items(), key=lambda x: -x[1])
return [(name, sec, sec / total) for name, sec in rows]
def print(self):
print(f"\n{'环节':<24}{'次数':>8}{'耗时':>12}{'占比':>8}")
for name, sec, pct in self.report():
print(f"{name:<24}{self.counts[name]:>8}{sec:>10.1f}s{pct:>8.1%}")
```
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
## 三、优化一：并发——最直觉，但有天花板
96% 在等网络，那并发就是第一刀。把 3000 个串行请求改成 20 路并发：
```python
# concurrent.py
import asyncio
import time
class RateGate:
"""
匀速限流闸门：保证全进程 QPS 不超过上限。
比每个任务自己 sleep 更准，因为它是全局视角。
"""
def __init__(self, qps: float):
self.interval = 1.0 / qps
self._next_slot = time.monotonic()
self._lock = asyncio.Lock()
async def wait(self):
async with self._lock:
now = time.monotonic()
wait = self._next_slot - now
if wait > 0:
await asyncio.sleep(wait)
self._next_slot += self.interval
else:
self._next_slot = now + self.interval
async def fetch_all(ids: list[str], fetch_one, concurrency: int = 20,
qps: float = 10.0) -> list:
"""
并发拉取。两个旋钮：
- concurrency：同时在飞的请求数（受限于连接和服务端）
- qps：每秒发起多少个（受限于平台限流）
两个都要设，只设 concurrency 会被限流打回来
"""
gate = RateGate(qps)
sem = asyncio.Semaphore(concurrency)
results = [None] * len(ids)
async def one(idx: int, item_id: str):
async with sem:
await gate.wait()
results[idx] = await fetch_one(item_id)
await asyncio.gather(*[one(i, x) for i, x in enumerate(ids)])
return results
```
注意这段代码里有两个旋钮：`concurrency`（同时在飞的请求数）和 `qps`（每秒发起多少）。
**只设并发不设 QPS，是新手最常见的翻车点。** 20 路并发、每秒发出 20 个请求，直接撞平台限流，然后全部进入重试，反而更慢。
实测：3 小时 → 9 分钟。20 倍。
但天花板立刻出现了：平台限流 QPS 10，3000 个商品理论下限就是 300 秒 = 5 分钟。**并发再高也没用，限流是硬约束。**
所以第二刀必须换思路：不是「把请求发得更快」，而是「发更少的请求」。
## 四、优化二：少发请求——本文最值钱的洞察
重新看需求：3000 个商品每天同步。真的 3000 个都变了吗？
统计了一下：**每天真正变化的商品不到 10%**（改价、改库存、上下架）。90% 的请求在拉「没变的数据」。
那能不能先知道哪些变了，再只拉那些？
能。**商品列表接口一次能返回几十条记录的 `modified` 时间戳**——用 100 次列表请求（每次 30 条）就能拿到全部 3000 个商品的更新时间，替代 3000 次详情请求。
```python
# incremental.py
import json
from datetime import datetime
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def list_all_modified(client, shop_id: str) -> dict[str, str]:
"""
分页拉全量商品列表，只取 item_id -> modified 映射。
3000 个商品 = 100 次轻量请求，代替 3000 次重量请求
"""
mapping = {}
page = 1
while True:
rows = client.list_products(shop_id, page=page, page_size=30)
if not rows:
break
for r in rows:
mapping[str(r["item_id"])] = r["modified"]
if len(rows) < 30:
break
page += 1
return mapping
def pick_changed(current: dict[str, str],
last_sync: dict[str, str] | None) -> list[str]:
"""
和上次同步记录对比，只返回有变化的 item_id
新商品 + 修改过的商品 + 已删除的商品（单独处理）
"""
last = last_sync or {}
changed = [iid for iid, m in current.items() if last.get(iid) != m]
removed = [iid for iid in last if iid not in current]
if removed:
print(f"[sync] {len(removed)} 个商品已下架，需清理")
return changed
def save_sync_state(state_path: str, current: dict[str, str]):
with open(state_path, "w", encoding="utf-8") as f:
json.dump(current, f, ensure_ascii=False)
def load_sync_state(state_path: str) -> dict[str, str] | None:
try:
with open(state_path, encoding="utf-8") as f:
return json.load(f)
except FileNotFoundError:
return None
```
账算一下：
- 原来：3000 次详情请求（每次约 3.4 秒等效负载）
- 现在：100 次列表请求 + 约 300 次详情请求 = 400 次请求
**请求数降到原来的 13%。** 这比任何并发技巧的收益都大。
这个思路的本质：**用便宜的请求替代昂贵的请求。** 列表接口一次返回 30 条的元数据，单价是详情接口的三十分之一。先便宜地筛一遍，再对筛出来的做贵的操作。
## 五、优化三 + 四：写库和连接
到这一步，网络等待只剩 40 秒左右，剩下的小头也该收一收。
**逐条写库 → 批量写。** SQLite 用 `executemany`，MySQL 用多值 INSERT：
```python
# bulk_write.py
import sqlite3
def bulk_upsert(conn: sqlite3.Connection, rows: list[tuple]):
"""
批量 upsert。3000 条逐条写 = 3000 次事务；
executemany 一批 = 1 次事务，差两个数量级
"""
conn.executemany("""
INSERT INTO products(platform, item_id, title, price, stock, synced_at)
VALUES(?,?,?,?,?,datetime('now'))
ON CONFLICT(platform, item_id) DO UPDATE SET
title=excluded.title, price=excluded.price,
stock=excluded.stock, synced_at=excluded.synced_at
""", rows)
conn.commit()
```
**每次新建连接 → 复用连接。** 这个最隐蔽：每次请求新建 TCP + TLS 握手要 200~~400ms，3000 次就是 10~~20 分钟。`httpx.AsyncClient` 复用连接后，握手只发生一次。
```python
# 复用连接：整个进程共享一个 client
# 错误：每次 fetch_one 里 with httpx.AsyncClient() as c: ...
# 正确：模块级或应用级创建一次
import httpx
client = httpx.AsyncClient(
timeout=10.0,
limits=httpx.Limits(max_connections=20, max_keepalive_connections=20),
)
```
顺带把限流做成自适应（AIMD），限流报错就减半 QPS，平稳运行则缓慢恢复：
```python
# adaptive.py
import random
class AdaptiveRate:
"""
自适应限速（AIMD）：
- 撞限流：QPS 减半（Multiplicative Decrease）
- 连续 50 次成功：QPS +10%（Additive Increase）
比写死一个保守值快得多，也比写死激进值安全
"""
def __init__(self, qps: float = 5.0, min_qps: float = 0.5, max_qps: float = 20.0):
self.qps = qps
self.min_qps = min_qps
self.max_qps = max_qps
self._ok_streak = 0
self._next_slot = 0.0
def on_success(self):
self._ok_streak += 1
if self._ok_streak >= 50:
self.qps = min(self.max_qps, self.qps * 1.1)
self._ok_streak = 0
def on_rate_limited(self):
self.qps = max(self.min_qps, self.qps / 2)
self._ok_streak = 0
@property
def interval(self) -> float:
return 1.0 / self.qps
async def wait(self):
"""带抖动的等待，避免多任务同刻发车"""
import asyncio, time
now = time.monotonic()
wait = max(0.0, self._next_slot - now)
if wait:
await asyncio.sleep(wait)
self._next_slot = max(now, self._next_slot) + self.interval * random.uniform(0.8, 1.2)
```
## 六、结果：60 倍是怎么来的
四个优化叠加，每一步的贡献都有数：
| 阶段 | 耗时 | 相比上一步 | 手段 |
| ------------------ | ---------- | -------- | ------------------------------- |
| 原始 | 180 分钟 | — | 串行 + 全量 + 逐条写 |
| + 并发 20 + QPS 限流 | 9 分钟 | **20×** | asyncio 并发，全局 QPS 闸门 |
| + 列表预筛增量 | 约 1.2 分钟 | **7.5×** | 100 次列表替代 2820 次详情 |
| + 批量写 + 连接复用 + 自适应 | **约 1 分钟** | 1.3× | executemany + keep-alive + AIMD |
两个观察值得记住：
1. **最大的收益来自「少做事」（7.5×）而不是「做得快」（20× 有硬顶）。** 并发受限于平台限流，少发请求不受任何限制。
2. **最后一个优化只贡献 1.3×，但它是「从 1 分钟到 40 秒」和「偶尔翻车」的区别。** 连接复用和自适应限速不影响平均值，影响的是稳定性。
## 七、完整主流程
把四个优化串起来：
```python
# sync_main.py
import asyncio
import sqlite3
from timer import StageTimer
from concurrent import fetch_all
from incremental import (list_all_modified, pick_changed,
load_sync_state, save_sync_state)
from bulk_write import bulk_upsert
from adaptive import AdaptiveRate
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def run_sync(clients: dict, shop_id: str,
state_path: str = "sync_state.json"):
t = StageTimer()
# 1. 增量预筛：100 次轻量请求拿全量 modified
t.start("列表预筛(3平台)")
current = {}
for platform, client in clients.items():
current[platform] = await list_all_modified(client, shop_id)
t.stop()
# 2. 对比上次状态，只保留有变化的
last = load_sync_state(state_path)
t.start("增量对比")
changed_map = {p: pick_changed(m, last and last.get(p))
for p, m in current.items()}
t.stop()
# 3. 并发拉详情（自适应限速）
t.start("详情并发拉取")
rows = []
for platform, client in clients.items():
ids = changed_map[platform]
if not ids:
continue
rate = AdaptiveRate(qps=5.0)
async def fetch_one(iid):
await rate.wait()
try:
r = await client.get_product_detail(iid)
rate.on_success()
return (platform, iid, r["title"], float(r["price"]),
r["stock"])
except Exception as e:
if "limit" in str(e).lower():
rate.on_rate_limited()
raise
results = await fetch_all(ids, fetch_one,
concurrency=20, qps=10.0)
rows.extend(r for r in results if r)
t.stop()
# 4. 批量写库
t.start("批量写库")
conn = sqlite3.connect("products.db")
bulk_upsert(conn, rows)
conn.close()
t.stop()
# 5. 保存同步状态，供下次增量对比
save_sync_state(state_path, current)
t.print()
return len(rows)
if __name__ == "__main__":
from client import PlatformClient
clients = {
"taobao": PlatformClient("taobao", "key", "secret"),
"alibaba": PlatformClient("alibaba", "key", "secret"),
"jd": PlatformClient("jd", "key", "secret"),
}
asyncio.run(run_sync(clients, "your_shop_id"))
```
## 八、四个坑
1. **并发数和 QPS 是两个旋钮，必须分开调。** 只调并发会被限流打回来进入重试风暴，比串行还慢。
2. **增量依赖 `modified` 字段，但不是所有平台都可靠。** 有的平台批量改价不更新 `modified`。建议每周跑一次全量对账，增量为主、全量兜底——和价格同步那篇的思路一致。
1. **`fetch_one` 里闭包引用循环变量是经典 bug。** 上面代码里 `for platform, client in clients.items()` 里面定义 `fetch_one`，每次循环重新定义，没有引用问题；但如果把函数定义提到循环外再绑定变量，就会全部引用最后一次的 `client`。测试时三个平台的数据混在一起，多半是这个问题。
2. **首次运行没有同步状态，会走全量。** 这是正确行为（第一次本来就该全量），但要提醒使用者：第一次跑慢是正常的，从第二次开始才是 3 分钟。
## 九、最后
回头看这 60 倍：
- 并发（20×）：把等待重叠起来，但受限于平台限流
- 增量（7.5×）：**用便宜的请求替代昂贵的请求**，收益最大
- 批量写 + 连接复用（1.3×）：平均值贡献小，稳定性贡献大
顺序也重要：**先测量再动手**。如果先花一天优化数据库（只占 2%），一天就白费了。
代码在上面，`RateGate`、`AdaptiveRate`、增量预筛这三个组件是通用的，可以拆到任何同步场景里用。
有在做的可以聊聊，你们的同步脚本跑多久，profile 过吗？
---
**标签**：`Python` `性能优化` `电商` `异步编程` `数据同步`