# 免费试用 7 天：三平台接口文档 + 示例代码全给（附 Python 源码）
> 先说清楚一件事：**7 天试用最容易白费**。
>
> 我们后台看得到试用账号的调用曲线，一个很规律的现象是——第 1 天调几十次，把「能不能通」验证完了，然后就没有然后了；第 6 天想起来还剩一天，慌忙压一下并发，第 7 天到期，什么结论也没拿到。
>
> 所以这篇不讲「我们多好用」，讲**这 7 天该怎么用**：先验什么、后验什么、每一验用什么代码、什么数字算过、什么数字算不过。文末的四个脚本你可以直接拿去跑，跑完会生成一份决策报告——**那份报告才是试用的产出，不是「跑通了」这三个字**。
>
> 顺便说，这套脚本对任何一家接口服务商都适用。你要是正在对比几家，拿它挨个跑一遍，比听销售讲一小时有用。
## 一、先划清 7 天能做什么、不能做什么
试用不是「免费版产品」，它的边界必须提前讲明，否则第 3 天你就会撞上：
| 能做 | 做不到 |
| ----------------- | ------------------------- |
| 调通全部核心接口，验证字段是否够用 | 全量数据迁移（额度不够，也不该在试用期做） |
| 抽样验证数据准确性（对商家后台） | 长期稳定性验证（7 天看不出季度级故障） |
| 测出限流拐点和延迟分布 | 拿到和正式开通完全一致的权限（试用权限通常是子集） |
| 走通一个真实业务场景的端到端流程 | 把试用期的数据当生产依赖（到期即停，别埋雷） |
一条经验：**试用的目标是拿到「买不买」的决策依据，不是「先跑起来再说」**。这两件事的顺序反了，7 天一定白费。
## 二、文档：三平台接口文档，先看这四节
接口文档动辄几百页、上千个接口，全看完是浪费时间。真正决定你能不能落地的只有四节，按这个顺序看：
| 顺序 | 看什么 | 为什么先看这个 |
| -- | --------------- | ---------------------------------------------- |
| ① | **接口列表 + 权限归属** | 先确认「你要的接口存不存在、在哪个权限包里」。这一步能筛掉一半方案 |
| ② | **请求参数与返回字段** | 重点看返回字段有没有 `null` 之外的「缺失」——缺 key 通常是权限问题，不是没数据 |
| ③ | **错误码表** | 只看前 15 个高频码。剩下的你一辈子不会遇到 |
| ④ | **限流与配额说明** | 很多文档只写「有频率限制」，具体数值要问。这一条决定你的调度策略 |
三个平台的文档风格差异很大，顺着看会踩坑：
| | 淘宝开放平台 | 1688 开放平台 | 京东宙斯 |
| ---- | ------------------ | ----------------- | ---------------------------- |
| 文档组织 | 按**权限包**分组，接口散在各包下 | 按**业务域**分组，中文描述更细 | 按**接口名**平铺，中文文档偏薄 |
| 参数命名 | 下划线（`num_iid`） | 下划线，字段偏业务语义 | 驼峰 + `360buy_param_json` 包一层 |
| 返回结构 | 顶层直接给业务字段 | 与淘宝接近 | **按接口名再套一层**，key 写法还不统一 |
| 最该注意 | 沙箱常与线上不同步 | 部分接口只有中文文档，字段含义要试 | 中文序列化转义问题（见第 11 篇） |
我们把这些整理成了一份**接口对照索引**：三个平台的核心接口（商品、SKU、价格、库存、订单、物流，各平台 12 个左右）按「同一业务动作」横向对齐，左边是动作，右边是三平台的接口名和关键差异。你按业务动作查，不用记三套命名。文档和示例代码一起给，不用申请就能看。
## 三、示例代码：四个脚本，覆盖 7 天全部验证
代码是一套的：`client.py` 打底，另外四个脚本各管一件事。全部用 `httpx` + `asyncio`，`pip install httpx` 就能跑。
### 3.1 统一的试用客户端
三个平台签名逻辑不同（淘宝和京东是 MD5 包夹，1688 是 HMAC-SHA1 且 urlPath 参与签名），但试用期你不需要关心这些——网关做了归一，你只面对一套参数和一套返回。
```python
# trial/client.py
from __future__ import annotations
import asyncio
import time
from dataclasses import dataclass, field
from typing import Any
import httpx
@dataclass(frozen=True)
class TrialConfig:
base_url: str # 试用网关地址
app_key: str
app_secret: str
daily_quota: int = 5000 # 试用额度，每天重置
timeout: float = 8.0
@dataclass
class Result:
ok: bool
data: Any = None
code: str = "" # 归一后的错误码，见下表
message: str = ""
cost_ms: int = 0
request_id: str = "" # 出问题时报给服务商用
raw: dict = field(default_factory=dict)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class TrialClient:
"""试用期最小客户端：只管调用、计时、错误归一，不含业务逻辑。"""
def __init__(self, cfg: TrialConfig):
self.cfg = cfg
self._client = httpx.AsyncClient(
base_url=cfg.base_url,
timeout=httpx.Timeout(cfg.timeout),
limits=httpx.Limits(max_connections=32, max_keepalive_connections=16),
)
self.calls = 0
async def __aenter__(self):
return self
async def __aexit__(self, *exc):
await self._client.aclose()
async def get(self, path: str, **params) -> Result:
self.calls += 1
t0 = time.monotonic()
try:
r = await self._client.get(
path,
params={"app_key": self.cfg.app_key, **params},
headers={"X-Trial-Secret": self.cfg.app_secret},
)
except httpx.TimeoutException:
return Result(False, code="TIMEOUT",
message=f">{self.cfg.timeout}s",
cost_ms=int((time.monotonic() - t0) * 1000))
except httpx.HTTPError as e:
return Result(False, code="NETWORK", message=str(e)[:120],
cost_ms=int((time.monotonic() - t0) * 1000))
cost = int((time.monotonic() - t0) * 1000)
body = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
# 关键：HTTP 200 不等于成功。三个平台都有「200 + body 里报错」的返回
code = str(body.get("error_code") or body.get("code") or "")
if r.status_code >= 400 or (code and code not in ("0", "200", "success")):
return Result(False, code=normalize(code, r.status_code),
message=str(body.get("error_msg") or body.get("message") or "")[:200],
cost_ms=cost, request_id=r.headers.get("x-request-id", ""),
raw=body)
return Result(True, data=body.get("data", body), cost_ms=cost,
request_id=r.headers.get("x-request-id", ""), raw=body)
ERROR_KIND = {
"AUTH": {"401", "403", "TOKEN_EXPIRED", "SIGN_ERROR", "INVALID_SIGN"},
"PARAM": {"400", "PARAM_MISSING", "PARAM_INVALID"},
"LIMIT": {"429", "THROTTLED", "QUOTA_EXCEEDED"},
"PLATFORM":{"500", "502", "503", "504", "PLATFORM_ERROR"},
"NOT_SUPPORT": {"404", "API_NOT_FOUND", "CAPABILITY_MISSING"},
}
def normalize(code: str, status: int) -> str:
raw = code or str(status)
for kind, codes in ERROR_KIND.items():
if raw in codes:
return kind
return "UNKNOWN"
```
`normalize` 这一层看着简单，但它是整个验证的前提——**如果错误码不归一，你没法统计「这 7 天里失败都是什么原因」**，最后只能说「偶尔报错」，等于没验。
### 3.2 `quickstart.py`：5 分钟跑通，输出一张对齐表
第一个脚本的目的不是「能调通」，而是**让你在 5 分钟内看到三个平台的数据长什么样、差在哪**。这比逐页读文档快得多。
```python
# trial/quickstart.py
import asyncio
from client import TrialClient, TrialConfig
TARGETS = [
# (平台, 商品/订单标识, 说明)
("taobao", "654321098765", "自家店在卖的"),
("alibaba", "678901234567", "同款或近似同款"),
("jd", "100012043978", "在售竞品"),
]
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def main(cfg: TrialConfig):
async with TrialClient(cfg) as c:
rows = []
for platform, item_id, note in TARGETS:
item = await c.get(f"/v1/{platform}/item", item_id=item_id)
if not item.ok:
rows.append([platform, item_id, f"失败:{item.code}", "", "", ""])
continue
d = item.data
skus = d.get("skus") or []
prices = [s.get("price") for s in skus if s.get("price") is not None]
stocks = [s.get("stock") for s in skus if s.get("stock") is not None]
rows.append([
platform, item_id, str(d.get("title", ""))[:24],
f"{min(prices) if prices else d.get('price', '-')}",
f"{sum(stocks) if stocks else '无SKU库存'}",
f"{len(skus)}个SKU / {item.cost_ms}ms",
])
print(f"{'平台':<9}{'商品ID':<15}{'标题':<26}{'最低价':<10}{'总库存':<12}{'结构'}")
print("-" * 88)
for r in rows:
print(f"{r[0]:<9}{r[1]:<15}{r[2]:<26}{r[3]:<10}{r[4]:<12}{r[5]}")
print(f"\n本次消耗 {c.calls} 次额度")
if __name__ == "__main__":
asyncio.run(main(TrialConfig(
base_url="https://api.example.com",
app_key="trial_xxxxxxxx",
app_secret="trial_secret_xxxxxxxx",
)))
```
跑完你会立刻发现三件事：1688 那个商品「总库存」是空的、京东的 SKU 结构和淘宝不一样、同一个商品三个平台的标题几乎无法直接比对。这三件事正是后面几天要处理的。
### 3.3 `smoke.py`：9 项冒烟，出一份可粘贴的报告
冒烟的意义不在于「今天通了」，在于**以后每天跑一次，能提前发现三类变化**：授权过期、平台改字段、额度规则调整。所以它必须输出结构化结果，不能只 `print("ok")`。
```python
# trial/smoke.py
import asyncio
import json
from dataclasses import dataclass, asdict
from client import TrialClient, TrialConfig
PROBES = [
("taobao", "item", {"item_id": "654321098765"}, ["title", "skus"]),
("taobao", "orders", {"start": "2026-10-01", "end": "2026-10-07"}, ["orders"]),
("taobao", "logistics", {"tid": "1234567890123"}, ["status"]),
("alibaba", "item", {"item_id": "678901234567"}, ["title", "price_ranges"]),
("alibaba", "orders", {"start": "2026-10-01", "end": "2026-10-07"}, ["orders"]),
("alibaba", "logistics", {"tid": "1234567890"}, ["status"]),
("jd", "item", {"sku_id": "100012043978"}, ["name", "price"]),
("jd", "orders", {"start": "2026-10-01", "end": "2026-10-07"}, ["orders"]),
("jd", "logistics", {"order_id": "23456789012"}, ["status"]),
]
@dataclass
class Probe:
platform: str
resource: str
status: str # PASS / FAIL / EMPTY / SKIP
detail: str
cost_ms: int
async def run(cfg: TrialConfig) -> list[Probe]:
out = []
async with TrialClient(cfg) as c:
for platform, resource, params, required in PROBES:
res = await c.get(f"/v1/{platform}/{resource}", **params)
if not res.ok:
out.append(Probe(platform, resource, "FAIL", res.code, res.cost_ms))
continue
# 空数组不等于成功：无数据、无权限、参数错都长这样
payload = res.data if isinstance(res.data, dict) else {}
missing = [k for k in required if k not in payload]
if missing:
out.append(Probe(platform, resource, "EMPTY",
f"缺字段 {','.join(missing)}", res.cost_ms))
elif not any(payload.get(k) for k in required):
out.append(Probe(platform, resource, "EMPTY", "返回空", res.cost_ms))
else:
out.append(Probe(platform, resource, "PASS", "", res.cost_ms))
return out
def report(probes: list[Probe]) -> str:
lines = ["| 平台 | 资源 | 结果 | 说明 | 耗时 |", "|---|---|---|---|---|"]
for p in probes:
icon = {"PASS": "✅", "FAIL": "❌", "EMPTY": "⚠️", "SKIP": "—"}[p.status]
lines.append(f"| {p.platform} | {p.resource} | {icon} {p.status} | "
f"{p.detail or '—'} | {p.cost_ms}ms |")
return "\n".join(lines)
if __name__ == "__main__":
cfg = TrialConfig(base_url="https://api.example.com",
app_key="trial_xxxxxxxx", app_secret="trial_secret_xxxxxxxx")
probes = asyncio.run(run(cfg))
print(report(probes))
json.dump([asdict(p) for p in probes], open("smoke_result.json", "w"), ensure_ascii=False, indent=2)
```
`EMPTY` 这个状态是刻意加的：**「调用成功但没数据」比「调用失败」危险得多**。失败你会去看，空数据你会以为这个商品就是没库存。淘宝的 `trade_list` 在时间参数超出保留期时就是这么返回的。
### 3.4 `bench.py`：找限流拐点，不是测「最高 QPS」
这是四个脚本里最有价值的一个，也是 90% 的试用会漏掉的一个。
多数人压测是「一路加压，直到报 429，那个数值就是上限」。这错了——**429 出现的时候，你已经有很多请求在超时了**。生产上的表现不是「报限流错误」，是「莫名其妙的超时和变慢」。
正确的做法是**阶梯加压，找延迟开始抬升的那个点**：
```python
# trial/bench.py
import asyncio
import time
import statistics
from client import TrialClient, TrialConfig
async def one_stage(cfg, concurrency: int, seconds: float, path: str) -> dict:
"""固定并发压一段时间，收集延迟样本"""
latencies, errors, wall_end = [], 0, time.monotonic() + seconds
async with TrialClient(cfg) as c:
sem = asyncio.Semaphore(concurrency)
async def worker():
nonlocal errors
while time.monotonic() < wall_end:
async with sem:
r = await c.get(path, item_id="654321098765")
latencies.append(r.cost_ms)
if not r.ok:
errors += 1
await asyncio.gather(*[worker() for _ in range(concurrency)])
total = c.calls
lat = sorted(latencies)
return {
"concurrency": concurrency,
"qps": round(total / seconds, 1),
"p50": lat[int(len(lat) * 0.50)] if lat else 0,
"p95": lat[int(len(lat) * 0.95)] if lat else 0,
"p99": lat[int(len(lat) * 0.99)] if lat else 0,
"err_rate": round(errors / total, 3) if total else 0,
"samples": total,
}
async def find_cliff(cfg: TrialConfig, path="/v1/taobao/item"):
base = None
rows = []
for c in (1, 2, 4, 8, 16, 32):
st = await one_stage(cfg, c, seconds=20, path=path)
rows.append(st)
if base is None:
base = st["p95"]
# 拐点判定：P95 涨到基线的 2 倍，或错误率超 1%
if st["p95"] > base * 2 or st["err_rate"] > 0.01:
break
print(f"{'并发':<6}{'QPS':<8}{'P50':<8}{'P95':<8}{'P99':<8}{'错误率':<8}")
for r in rows:
print(f"{r['concurrency']:<6}{r['qps']:<8}{r['p50']:<8}{r['p95']:<8}"
f"{r['p99']:<8}{r['err_rate']:<8}")
safe = rows[-2]["qps"] if len(rows) > 1 else rows[0]["qps"]
print(f"\n安全 QPS 建议：{safe}（拐点前一级，约占平台上限的 70%）")
return rows, safe
```
跑出来的表长这样（我们内部跑的真实形态，数字随账号等级变）：
| 并发 | QPS | P50 | P95 | P99 | 错误率 |
|---|---|---|---|---|---|
| 1 | 3.4 | 280 | 310 | 420 | 0% |
| 2 | 6.7 | 290 | 330 | 460 | 0% |
| 4 | 13.1 | 300 | 350 | 520 | 0% |
| 8 | 25.6 | 320 | **420** | 980 | 0% |
| 16 | 38.2 | 410 | **1,850** | 4,200 | 0.4% |
注意 8 到 16 那一行：**错误率只有 0.4%，看着还行，但 P95 已经从 420ms 涨到 1,850ms**。如果你只看错误率，会以为 16 并发是安全的，上线之后就是一片超时告警。真正的安全值在 8 那级，甚至更低一点。
这也是我建议你**在试用期一定要跑压测**的原因：这个拐点每个账号都不一样，问销售也只能给个大概值，只有自己跑才知道。
### 3.5 `verify.py`：数据准不准，只能和商家后台对
前三步验的是「通不通、快不快」，这一步验的是**「对不对」**，也是最容易糊弄过去的一步——因为很多服务商的接口「返回 200、字段齐全、数值是错的」，你不做对账永远发现不了。
最硬的验证方式是：**让商家从自己后台导一份 CSV，跟接口返回逐条比**。抽样 50 个商品就够，覆盖有活动价、有多个 SKU、刚改过价的。
```python
# trial/verify.py
import csv
from decimal import Decimal
from client import TrialClient, TrialConfig
def load_truth(path: str) -> dict[str, dict]:
"""商家后台导出的 CSV：item_id,title,price,stock"""
truth = {}
with open(path, encoding="utf-8-sig") as f:
for row in csv.DictReader(f):
truth[row["item_id"].strip()] = {
"title": row["title"].strip(),
"price": Decimal(row["price"]),
"stock": int(row["stock"]),
}
return truth
async def verify(cfg: TrialConfig, truth: dict[str, dict], platform="taobao"):
ok, diffs = 0, []
async with TrialClient(cfg) as c:
for item_id, expect in truth.items():
res = await c.get(f"/v1/{platform}/item", item_id=item_id)
if not res.ok:
diffs.append((item_id, "调用失败", res.code, ""))
continue
got_price = Decimal(str(res.data.get("price", "0")))
got_stock = int(res.data.get("stock", 0) or 0)
if got_price != expect["price"]:
kind = "价格口径" if abs(got_price - expect["price"]) < expect["price"] * Decimal("0.2") else "价格错误"
diffs.append((item_id, kind, str(expect["price"]), str(got_price)))
elif got_stock != expect["stock"]:
# 库存差 5% 以内通常是时点差，不算错
if abs(got_stock - expect["stock"]) > max(3, expect["stock"] * 0.05):
diffs.append((item_id, "库存时点", str(expect["stock"]), str(got_stock)))
else:
ok += 1
else:
ok += 1
total = len(truth)
print(f"准确率 {(ok / total):.2%} ({ok}/{total})")
for d in diffs[:15]:
print(f" {d[0]:<15}{d[1]:<8}预期{d[2]:<10}实际{d[3]}")
return ok / total, diffs
```
差异分类比准确率本身更有用：
- **价格口径**（差 20% 以内）——通常是券后价 / 活动价 / 含运费的差异。这不算错，是口径没对齐，谈的时候问清楚就行。
- **价格错误**（差异很大）——真问题，直接排除。
- **库存时点**——你导 CSV 和接口取值差了几分钟，正常。差 5% 以上才需要查。
我们内部的验收线是 **准确率 ≥ 99.5%，且「价格错误」为 0**。低于这个数，接入之后每天都会有人找你。
## 四、7 天怎么排
四个脚本写完了，剩下的就是排期。这张表可以直接照着做：
| 天 | 做什么 | 用哪个脚本 | 产出 |
|---|---|---|---|
| D1 | 跑通 + 冒烟 | `quickstart.py` / `smoke.py` | 9 项探针结果，知道哪些接口能用 |
| D2 | 文档对表 | — | 你要的 12 个接口，哪几个缺、缺在哪个权限包 |
| D3 | 数据准确性 | `verify.py` | 准确率 + 差异清单（抽样 50 个） |
| D4 | 限流与延迟 | `bench.py` | 安全 QPS、P95 基线 |
| D5 | 真实场景端到端 | 自己写 | 一次全量铺货，或一份日报，跑完 |
| D6 | 异常与降级 | 手动 | 改错 key、断网、超配额，看返回能不能解释 |
| D7 | 出报告、做决策 | `trial_report.py` | 一页纸：过 / 不过 / 有条件过 |
额度大概这么分（按 5,000 次/天算）：冒烟 60 次、验准 300 次、压测 2,000 次、真实场景 2,500 次。**第一天别把额度烧完**，这是最常见的错误。
D6 那天值得单独说：故意把 `app_secret` 改错一位，看返回是 `AUTH` 还是 `UNKNOWN`；故意把商品 ID 改成一个不存在的，看是 `PARAM` 还是空数据。**错误可解释性决定了你以后排障要花多少小时**——返回 `UNKNOWN` 的服务商，出问题只能找人问，一问就是半天。
## 五、`trial_report.py`：把结果汇总成一页决策报告
7 天结束，你不该拿着一堆 json 去汇报。这个脚本把前面所有结果拼成一份 markdown，带明确的判定：
```python
# trial/trial_report.py
import json
from datetime import date
# 验收阈值（可按业务调，但建议不要放宽「价格错误为 0」）
THRESHOLDS = {
"accuracy": 0.995, # 数据准确率
"p95_ms": 800, # 常规接口 P95
"safe_qps_ratio": 1.5, # 安全 QPS / 业务峰值
"smoke_pass": 9, # 冒烟通过项
}
def build(smoke: list[dict], verify_acc: float, price_errors: int,
bench: list[dict], safe_qps: float, peak_qps: float) -> str:
passed = sum(1 for p in smoke if p["status"] == "PASS")
p95 = bench[-2]["p95"] if len(bench) > 1 else bench[-1]["p95"]
checks = [
("冒烟通过", f"{passed}/9", passed >= THRESHOLDS["smoke_pass"]),
("数据准确率", f"{verify_acc:.2%}", verify_acc >= THRESHOLDS["accuracy"]),
("价格错误", str(price_errors), price_errors == 0),
("P95 延迟", f"{p95}ms", p95 <= THRESHOLDS["p95_ms"]),
("QPS 余量", f"{safe_qps / peak_qps:.1f}x",
safe_qps >= peak_qps * THRESHOLDS["safe_qps_ratio"]),
]
verdict = "通过" if all(c[2] for c in checks) else (
"有条件通过" if sum(c[2] for c in checks) >= 4 else "不通过")
lines = [f"# 接口试用评估报告（{date.today()}）", "",
f"**结论：{verdict}**", "", "| 检查项 | 实测 | 判定 |", "|---|---|---|"]
for name, val, ok in checks:
lines.append(f"| {name} | {val} | {'✅' if ok else '❌'} |")
lines += ["", "## 未通过项的处理建议", ""]
for name, val, ok in checks:
if not ok:
lines.append(f"- **{name}**：实测 {val}。"
"若是口径差异请与服务商确认定义；若是真差异，换方案比接入后打补丁便宜。")
return "\n".join(lines)
```
这份报告有个副作用：**它同时是给老板看的材料**。技术人做选型最难的不是判断，是说服——「我觉得这家还行」没有说服力，「准确率 99.8%、P95 420ms、QPS 余量 3.2 倍、五个检查项全过」有。
## 六、试用结束后最容易被忽略的两个坑
**① 试用期的数据不能当生产依赖。** 到期停服，你的任务表里那些外键、缓存里那些商品、报表里那些历史，全部失去来源。正确的做法是试用期只做「读 + 验证」，不做数据落地。要落，就落在你自己库里并标清来源和有效期（见第 18 篇那个 `provenance` 字段）。
**② 试用权限通常是正式权限的子集。** 你试用期能调的接口，正式开通不一定一次全给（要看资质和权限包）；反过来，试用期调不通的，也可能是权限没开而不是接口没有。**这一点必须在 D2 那天和商务确认清楚**，别等到接入时才发现「哦这个要另外申请」。
## 七、文中用到的东西
- 六个文件：`client.py`（打底客户端）、`quickstart.py`、`smoke.py`、`bench.py`、`verify.py`、`trial_report.py`（可直接跑，改配置即可）
- 三平台接口对照索引：核心业务动作 → 三平台接口名 + 关键差异，一页表
- 三平台错误码对照：把三套错误码归一成 5 类，判断「该重试还是该改代码」
- 试用账号：7 天，5,000 次/天，核心接口全开，不需要企业资质
我们把这些打包成了一份试用包，文档和代码不申请也能先看。**建议的顺序是：先看文档对表，确认接口够用，再领试用账号跑脚本**——反过来做，容易在第 2 天发现最关键的接口没有，白折腾。
---
前面十八篇讲的是「怎么把这件事做成」，这篇讲「怎么判断手上的工具行不行」。判断标准我写在代码里了，阈值你可以改成自己的——**但「价格错误为 0」这条建议别改**，它是唯一一个接入之后没法补救的指标。
