# 刚起步的电商小团队，第一批该接哪几个接口（附 Python 源码）
> 起步团队最容易犯的错，是把「接口清单」当成任务清单，一条条往下接。
>
> 三平台文档里有上千个接口，你真正需要的第一批只有 **4 个**。剩下的不是不重要，是**现在接它们会拖慢你赚到第一笔钱的时间**。
>
> 这篇给一个明确的顺序：第一批 4 个、第二批 3 个、第三批 4 个，以及三个「现在千万别接」的。判断标准只有一条——**这个接口能不能让货卖出去、或者让钱收回来**。
>
> 代码是一个 150 行的单文件，跑完就是一条完整闭环：商品上架 → 客户下单 → 订单进库 → 扣库存 → 日终对账。
## 一、先定一个判断标准
接口的价值不是「技术上重要」，是**它对收入的贡献**。按这个标准，起步期能碰的接口只分三类：
| 类型 | 作用 | 起步期优先级 |
| ------- | --------- | ---------- |
| **换收入** | 商品能上架、能卖 | 最高，第一批 |
| **保现金** | 订单不漏、账对得上 | 最高，第一批 |
| **补体验** | 物流可查、售后顺畅 | 第二批，等有单了再说 |
| **纯优化** | 评价、竞品、素材 | 第三批，或者永远不接 |
一个反例：很多团队第一批就去接「物流轨迹查询」，理由是「客户会问快递到哪了」。但起步期你一天可能就 5 单，客户问了你手动查一下，30 秒的事。**为一天 5 次的查询，写一套订阅 + 去重 + 状态机，是典型的提前优化**（第 3 篇那套方案是对的，但不是现在该做的）。
## 二、第一批：4 个接口，按这个顺序接
| 顺序 | 接口 | 为什么在这个位置 |
| -- | ----------------------- | --------------------------------------- |
| ① | **商品详情**（含 SKU、标价、可售库存） | 没商品就没得卖。三平台的详情接口都一次返回 SKU 列表和标价，别拆成四次请求 |
| ② | **订单列表**（按更新时间增量） | 收入的证据。也是「有没有单」的唯一可靠来源 |
| ③ | **订单详情**（含收货地址、支付时间） | 发货必需。**列表接口通常不含完整地址**，这个坑第一天就会撞上 |
| ④ | **库存**（可售库存） | 防超卖。但起步期用「简单账本」撑住就行，别做实时同步 |
### 为什么「先订单、后库存」
这是这篇最反直觉的一条，值得单独说。
按技术难度排，库存比订单简单（一个数字 vs 一堆状态）。但按**收益**排，订单在前：
- **订单**接进来，你立刻能发货、能算账、能知道今天赚了多少。这是正反馈。
- **库存**接进来，你防的是「超卖」。而起步期 SKU 少、单量小，超卖的实际损失远小于「商品还没上架」的损失。
所以正确的做法是：**库存先用静态值撑住，下单即扣本地账本，宁可少卖**。具体就是三行逻辑：
```
可卖数 = 平台可售 − 本地已占 （查询时算，不写库）
下单 → 够就 reserved += n，不够就拦住这笔单
日终 → 平台库存重新拉过之后，reserved 归零重算
```
这么做会偶尔少卖（账面比实际紧），但不会超卖，而且**没有并发问题**——因为你从来不做「实时同步」这件难事。
等单量上来了（比如日单过百，或者开始做多仓），再把第 3 篇那套 `total - reserved - safety` 加上事件校准换上。那时候你已经有收入了，花两周做对它是合理的。
反过来做的团队我见过不少：花两周把库存并发扣减做对，商品还没上架，一分钱没赚。**技术正确，商业上错了。**
## 三、第二批：有第一笔订单之后
大概在第一笔订单之后的 2~4 周，接这三个：
| 接口 | 触发条件 | 不接的代价 |
| ----------- | --------------------- | ---------------------- |
| **物流轨迹** | 客户开始问「到哪了」，且每天超过 10 次 | 客服时间被吃掉，但还不致命 |
| **售后 / 退款** | 出现第一笔退款 | **账永远对不上**，你会以为是接口有问题 |
| **发货回写** | 你做代发，或对接了 ERP | 只能在平台后台手工点发货，一天几十单就撑不住 |
退款这个特别值得说：起步团队常常只拉「已付款」的订单，退款不管。结果月底一算，GMV 和实际到账差一截，财务来问，你查三天。
**退款不是异常，是正常业务**。它该在第二批，不该在「以后再说」。
## 四、第三批：月订单过千再看
| 接口 | 什么时候值得 |
| --------- | --------------------------------- |
| 活动价 / 优惠券 | 你开始做促销，且发现「到手价」算不对（第 13 篇的价格对象建模） |
| 类目属性 | 你开始批量铺货、被审核驳回（第 5 篇） |
| 评价 | 你要做选品或口碑分析（第 16 篇） |
| 素材 / 图片 | 你要做自动生成主图、多平台分发 |
这四个都是「业务推着你接」，不是「文档里有所以接」。
## 五、三个现在千万别接的
**① TMC 消息推送。**
听起来很美：订单来了实时推给你。实际要维护长连接、断线重连、消息确认、重复消费——四件事都能坏，坏了都要你查。起步期用「5 分钟轮询 + 时间窗重叠」完全够，而且失败就是这一轮没跑，下一轮补上，心智负担接近于零。
等你日单过千、或者对时效真的有要求了再换。切换成本是一次可控的重构，不是重写。
**② 批量 / 异步接口。**
批量接口的特点是：提交任务 → 拿 task_id → 轮询结果 → 处理部分失败。这套东西要状态机、要重试、要存储中间态。起步期你的商品可能就 200 个，循环调 200 次详情，10 分钟跑完，够用。
**③ 全站 / 竞品数据。**
先把自己店的货卖好。比价、竞品监控是第二阶段的事（第 16 篇），而且它需要的数据源和自营接口是两套逻辑（联盟接口 / 聚合服务），混在一起做会让你的第一版架构变形。
## 六、代码：150 行，一条闭环
起步版的原则：**小到一屏能看懂**。不要配置层、不要抽象层、不要依赖注入——这些东西等你有了第二个平台再谈。
```python
# mvp.py —— 起步版：4 个接口，1 条闭环
# 依赖：pip install httpx
import asyncio
import sqlite3
import time
import httpx
BASE = "https://api.example.com" # 换成你的网关
KEY = "your_app_key"
SECRET = "your_app_secret"
SHOP = "shop_001"
SCHEMA = """
pragma journal_mode = wal;
create table if not exists items(
item_id text primary key, title text, price real,
stock int, skus text, updated_at int);
create table if not exists orders(
tid text primary key, status text, pay_time int,
amount real, receiver text, updated_at int);
create table if not exists ledger(
item_id text primary key, reserved int default 0, last_sync int);
"""
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def call(client, path, **params):
"""一个函数搞定所有接口：拼参数、判错、返回 data"""
r = await client.get(path, params={"shop_id": SHOP, **params})
r.raise_for_status()
body = r.json()
if body.get("code") not in (0, 200, "0", None):
raise RuntimeError(f"{body.get('code')} {body.get('msg')}")
return body.get("data", body)
async def pull_items(client, db):
"""① 商品详情：一次带回 SKU、标价、可售库存。
三平台详情接口都支持，别拆成 SKU 接口 + 价格接口 + 库存接口"""
ids = [r[0] for r in db.execute("select item_id from items")]
if not ids: # 第一次：从列表接口拿 ID
lst = await call(client, "/v1/item/list", page_size=100)
ids = [it["item_id"] for it in lst.get("items", [])]
for item_id in ids:
d = await call(client, "/v1/item", item_id=item_id)
skus = d.get("skus") or []
stock = sum(int(s.get("stock") or 0) for s in skus) if skus \
else int(d.get("stock") or 0)
db.execute(
"insert into items(item_id,title,price,stock,skus,updated_at) "
"values(?,?,?,?,?,?) on conflict(item_id) do update set "
"title=excluded.title, price=excluded.price, stock=excluded.stock, "
"skus=excluded.skus, updated_at=excluded.updated_at",
(item_id, d.get("title", ""), float(d.get("price") or 0),
stock, str(len(skus)), int(time.time())))
db.commit()
return len(ids)
async def pull_orders(client, db, since: int):
"""② 订单列表：按更新时间增量，重叠加窗（第 3 篇的边界坑）"""
start = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(since - 600))
page, got = 1, 0
while True:
lst = await call(client, "/v1/orders",
updated_start=start, page=page, page_size=100)
rows = lst.get("orders", [])
if not rows:
break
for o in rows:
db.execute(
"insert into orders(tid,status,pay_time,amount,updated_at) "
"values(?,?,?,?,?) on conflict(tid) do update set "
"status=excluded.status, updated_at=excluded.updated_at",
(o["tid"], o["status"], int(o.get("pay_time") or 0),
float(o.get("payment") or 0), int(time.time())))
got += 1
page += 1
db.commit()
return got
async def fill_receiver(client, db):
"""③ 订单详情：列表接口没有地址，发货前补。
只补「待发货」的，别全量拉——这一步最容易把额度烧光"""
todo = db.execute(
"select tid from orders where receiver is null and status='WAIT_SELLER_SEND'"
).fetchall()
for (tid,) in todo:
d = await call(client, "/v1/order", tid=tid)
addr = d.get("receiver") or {}
db.execute("update orders set receiver=? where tid=?",
(f"{addr.get('name','')}|{addr.get('mobile','')}|"
f"{addr.get('address','')}", tid))
db.commit()
return len(todo)
def available(db, item_id: str) -> int:
"""④ 可卖数 = 平台可售 − 本地已占。
查询时算，不写库：`items.stock` 是平台镜像，每次拉取覆盖；
本地占用量单独记在 ledger，两者相减才是能卖的数。
不写回的好处是不存在「谁覆盖谁」的问题——第 3 篇那个
「校准取 min」的原则，是给「要把库存写回平台」的场景用的，
起步期我们不写回，所以用不上。"""
item = db.execute("select stock from items where item_id=?",
(item_id,)).fetchone()
if item is None:
return 0
row = db.execute("select reserved from ledger where item_id=?",
(item_id,)).fetchone()
return max(0, item[0] - (row[0] if row else 0))
def reserve_stock(db, item_id: str, num: int = 1) -> bool:
"""下单占库存：够就扣，不够就拦住这笔单（宁可少卖，不超卖）。
注意先判后扣——先扣再判，被拦下的单也会留下占用，账面越推越紧。"""
if available(db, item_id) < num:
return False
row = db.execute("select reserved from ledger where item_id=?",
(item_id,)).fetchone()
if row is None:
db.execute("insert into ledger(item_id,reserved,last_sync) "
"values(?,?,?)", (item_id, num, int(time.time())))
else:
db.execute("update ledger set reserved=reserved+?, last_sync=? "
"where item_id=?", (num, int(time.time()), item_id))
db.commit()
return True
def reset_reserved(db):
"""日终：平台库存已经在 pull_items 里重新拉过，本地占用量归零重算。
必须在拉取成功之后调，否则会把当天的占用抹掉——那就是超卖。"""
n = db.execute("update ledger set reserved=0 where reserved>0").rowcount
db.commit()
return n
async def main():
db = sqlite3.connect("mvp.db")
db.executescript(SCHEMA)
async with httpx.AsyncClient(base_url=BASE, timeout=10,
headers={"X-Secret": SECRET, "X-Key": KEY}) as c:
n_items = await pull_items(c, db) # 先拉最新库存
n_orders = await pull_orders(c, db, int(time.time()) - 3600)
n_addr = await fill_receiver(c, db)
n_reset = reset_reserved(db) # 拉成功之后才清零
print(f"商品 {n_items} · 订单 {n_orders} · 补地址 {n_addr} · 归零 {n_reset}")
if __name__ == "__main__":
asyncio.run(main())
```
这 150 行就是起步版的全部。它故意没有的东西，是按重要性排的：
| 没有 | 为什么现在不加 | 什么时候加 |
| --------- | ------------------- | ---------------- |
| 配置层（YAML） | 5 个常量改起来比读配置快 | 有第二个部署环境 |
| 多平台适配器 | 先接一个平台，等第二个出现重复代码 | 第二个平台接的时候（第 8 篇） |
| 并发 / 限流 | 200 个商品串行跑 4 分钟，能接受 | 商品上千，或单量上来 |
| 重试 / 熔断 | 失败就下一轮再跑，起步期能忍 | 开始有客户依赖你的数据 |
| 消息推送 | 轮询 5 分钟够用 | 日单过千 |
## 七、验收标准：一条能跑完的闭环
「接口接完了」不等于「能用了」。第一批的正确验收标准是一个动作：**把你自己的一个商品，从头到尾跑一遍**。
1. 商品信息从接口拉下来，落到你的库，页面能展示
2. 用另一个账号在你店里下一单（1 元商品，自己买）
3. 轮询把订单拉进来，地址补全，状态是「待发货」
4. 本地库存被扣掉 1，页面显示的可售数减 1
5. 你在平台后台点发货，回写状态（第二批才有，现在手工）
这五步跑通，第一批就算完成。**注意第 2 步必须真下单**——只用接口返回的模拟数据，你永远不知道字段口径对不对。1 元的成本，换来的是「整条链路真的通」，这是起步期最划算的一笔投入。
跑完你会发现至少两个问题：某个字段为空（通常是权限没开）、某个时间格式不对（时区）。这两个坑每个团队都会踩，早踩比上线后踩好。
## 八、起步期三个最常见的错误
**① 三个平台一起接。**
看起来平行推进效率高，实际是把「三个平台各自的怪癖」同时砸在你头上。第 2 篇那个坑表里的东西——1688 没有 SKU 库存、京东返回体多套一层、淘宝错误藏在 body 里——同时遇到三个，你会怀疑人生。
正确顺序：**一个平台跑通闭环 → 再复制到第二个**。第二个平台的成本大约是第一个的 1/3，因为业务逻辑已经验证过了。
**② 一开始就做增量和高并发。**
起步期商品 200 个、订单一天 20 个，全量拉一遍 4 分钟。这个阶段做增量调度、做并发优化，是把第 7 篇那套 60 倍提速用在 4 分钟上——**优化一个你根本感知不到的问题**。
**③ 一开始就自建。**
第 12 篇算过账：自建爬虫三年 37 万，官方 API 17 万。起步团队还有一笔更贵的隐藏成本：**你在写同步代码的那两周，没有在卖货**。
起步期最该花钱买的就是数据接口，因为它把你从「修管道」里解放出来，去做只有你能做的事——选品、内容、投放。这笔账在收入为零的时候最容易被算错。
## 九、时间预算
按这篇的顺序做，第一周（约 20 小时）应该能到：
| 天 | 内容 | 小时 |
| ----- | ---------------- | -- |
| D1 | 申请、拿 key、跑通商品详情 | 4 |
| D2 | 商品落库 + 页面展示 | 4 |
| D3-D4 | 订单增量 + 地址补全 + 幂等 | 8 |
| D5 | 库存账本 + 日终校准 | 4 |
对比「全都接」的版本：三平台 + 物流 + 售后 + 推送，两个月起步，而且大概率在第二个月放弃。
**起步团队真正的资源不是技术能力，是「还没被复杂度拖住」的那段时间。**
---
这一篇放在系列里的位置比较特殊：前面二十一篇讲的是「怎么把系统做对」，这篇讲「什么时候该做多少」。如果你现在只有一个店、一天十几单，上面 150 行代码就够了——第 3 篇的库存账本、第 7 篇的并发、第 9 篇的五道关，都等你需要的时候再看，那时候你会看得更懂。
