免费试用 7 天：三平台接口文档 + 示例代码全给（附 Python 源码）

先说清楚一件事：7 天试用最容易白费。

我们后台看得到试用账号的调用曲线，一个很规律的现象是——第 1 天调几十次，把「能不能通」验证完了，然后就没有然后了；第 6 天想起来还剩一天，慌忙压一下并发，第 7 天到期，什么结论也没拿到。

所以这篇不讲「我们多好用」，讲这 7 天该怎么用：先验什么、后验什么、每一验用什么代码、什么数字算过、什么数字算不过。文末的四个脚本你可以直接拿去跑，跑完会生成一份决策报告——那份报告才是试用的产出，不是「跑通了」这三个字。

顺便说，这套脚本对任何一家接口服务商都适用。你要是正在对比几家，拿它挨个跑一遍，比听销售讲一小时有用。

一、先划清 7 天能做什么、不能做什么

试用不是「免费版产品」，它的边界必须提前讲明，否则第 3 天你就会撞上：

一条经验：试用的目标是拿到「买不买」的决策依据，不是「先跑起来再说」。这两件事的顺序反了，7 天一定白费。

二、文档：三平台接口文档，先看这四节

接口文档动辄几百页、上千个接口，全看完是浪费时间。真正决定你能不能落地的只有四节，按这个顺序看：

三个平台的文档风格差异很大，顺着看会踩坑：

我们把这些整理成了一份接口对照索引：三个平台的核心接口（商品、SKU、价格、库存、订单、物流，各平台 12 个左右）按「同一业务动作」横向对齐，左边是动作，右边是三平台的接口名和关键差异。你按业务动作查，不用记三套命名。文档和示例代码一起给，不用申请就能看。

三、示例代码：四个脚本，覆盖 7 天全部验证

代码是一套的：client.py 打底，另外四个脚本各管一件事。全部用 httpx + asyncio，pip install httpx 就能跑。

3.1 统一的试用客户端

三个平台签名逻辑不同（淘宝和京东是 MD5 包夹，1688 是 HMAC-SHA1 且 urlPath 参与签名），但试用期你不需要关心这些——网关做了归一，你只面对一套参数和一套返回。

# 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)


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)

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
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 分钟内看到三个平台的数据长什么样、差在哪。这比逐页读文档快得多。

# trial/quickstart.py
import asyncio

from client import TrialClient, TrialConfig

TARGETS = [
    # (平台, 商品/订单标识, 说明)
    ("taobao", "654321098765", "自家店在卖的"),
    ("alibaba", "678901234567", "同款或近似同款"),
    ("jd", "100012043978", "在售竞品"),
]


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")。

# 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 出现的时候，你已经有很多请求在超时了。生产上的表现不是「报限流错误」，是「莫名其妙的超时和变慢」。

正确的做法是阶梯加压，找延迟开始抬升的那个点：

# trial/bench.py
import asyncio
import time
import statistics

from client import TrialClient, TrialConfig

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
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

跑出来的表长这样（我们内部跑的真实形态，数字随账号等级变）：

注意 8 到 16 那一行：错误率只有 0.4%，看着还行，但 P95 已经从 420ms 涨到 1,850ms。如果你只看错误率，会以为 16 并发是安全的，上线之后就是一片超时告警。真正的安全值在 8 那级，甚至更低一点。

这也是我建议你在试用期一定要跑压测的原因：这个拐点每个账号都不一样，问销售也只能给个大概值，只有自己跑才知道。

3.5 verify.py：数据准不准，只能和商家后台对

前三步验的是「通不通、快不快」，这一步验的是「对不对」，也是最容易糊弄过去的一步——因为很多服务商的接口「返回 200、字段齐全、数值是错的」，你不做对账永远发现不了。

最硬的验证方式是：让商家从自己后台导一份 CSV，跟接口返回逐条比。抽样 50 个商品就够，覆盖有活动价、有多个 SKU、刚改过价的。

# 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 天怎么排

四个脚本写完了，剩下的就是排期。这张表可以直接照着做：

额度大概这么分（按 5,000 次/天算）：冒烟 60 次、验准 300 次、压测 2,000 次、真实场景 2,500 次。第一天别把额度烧完，这是最常见的错误。

D6 那天值得单独说：故意把 app_secret 改错一位，看返回是 AUTH 还是 UNKNOWN；故意把商品 ID 改成一个不存在的，看是 PARAM 还是空数据。错误可解释性决定了你以后排障要花多少小时——返回 UNKNOWN 的服务商，出问题只能找人问，一问就是半天。

五、trial_report.py：把结果汇总成一页决策报告

7 天结束，你不该拿着一堆 json 去汇报。这个脚本把前面所有结果拼成一份 markdown，带明确的判定：

# 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」这条建议别改，它是唯一一个接入之后没法补救的指标。