《1688采购单API接口边界：purchase.order.* 与 alibaba.trade.* 的差异与选择》（附Python源码）
先拍结论：
1688 开放平台里并没有一套官方主推的purchase.order.*交易族；真正跑采购/代发/分销单的是alibaba.trade.*（买家视角交易域）。你在某些聚合网关/ERP 里看到的purchase.order.get / purchase.order.create，通常是“采购业务语义层”的封装名，不是 1688 原生 method。选型原则：原生用alibaba.trade.*，内部领域模型叫purchase_order，别把“业务名”和“平台 method”混成一个东西。
一、概念先拆：平台 method ≠ 业务对象
名字 | 是什么 | 谁用 |
|---|---|---|
alibaba.trade.* | 1688 官方交易 API 族（下单/查单/支付/退款） | 买家/分销商/ISV 用买家 token 调 |
alibaba.trade.fastCreateOrder | 快速建单：批发 / 分销 / 精选货源 | 代发 ERP 首选 |
alibaba.trade.fenxiaoOrder.create | 分销/代发回流单（带下游渠道+下游单号） | 抖音/快手/闲鱼/Mercari 代发 |
alibaba.trade.order.get / get.buyerView | 买家视角订单真相源 | 查状态/子单/物流/退款 |
purchase.order.* | 业务层/中间件命名，不是 1688 官方主族 | 你自己的 ERP / 聚合 API 网关 |
⚠️ 如果你接的“1688 供应商”其实是聚合平台（万邦/快递鸟/自建网关），它们会暴露purchase.order.list / purchase.order.detail / purchase.order.create。底层还是转成alibaba.trade.*调 1688。聚合层方便，但限流/字段裁剪/退款语义会和原生 API 漂。
二、能力边界对照
维度 | alibaba.trade.*（原生） | purchase.order.*（封装/聚合） |
|---|---|---|
下单 | fastCreateOrder / order.create / fenxiaoOrder.create | 通常只暴露“创建采购单” |
查单 | order.get / get.buyerView / buyer.list | purchase.order.get / list |
价格预览 | alibaba.createOrder.preview（强推荐） | 聚合层可能跳过 → 超卖/错价 |
分销 flow | general/fenxiao/saleproxy/boutiquefenxiao/boutiquepifa | 封装层常写死，严选价会丢 |
密文地址 | encryptOutOrderInfo（白名单） | 很多聚合层不支持 |
支付 | payWay.query / protocolPay.preparePay / crossBorderPay.url.get | 聚合层很少给免密/跨境宝 |
退款 | trade.refund.get / 售后回传 | 封装层常只返状态，不返退款流水 |
消息推送 | 官方订单/退款 topic | 聚合层自己发 webhook，语义不全 |
限流可控性 | 自己管 AppKey/QPS | 受聚合商总配额拖累 |
三、什么时候用哪个
✅ 用 alibaba.trade.*
做生产级代发/分销
要严选价（
boutiquefenxiao）要密文面单
要免密支付 / 跨境宝
要对账到子单/退款流水
要自己管 token / 限流 / 幂等
⚅ 可以用 purchase.order.*（聚合封装）
跑 MVP / 内部工具
只查“我买了啥、发没发”
不碰密文/跨境/免密
供应商数量少、单量小
不想申 1688 交易类权限（交易接口要人工审/按月订购）
❌ 别这么做
用
purchase.order.create当“1688 官方下单”写进架构文档以为
purchase.order.get比trade.order.get更权威（反了）聚合层返回
status=shipped就信，不回trade.order.get拉真相用聚合层价格直接上架下游（没过
createOrder.preview）
四、统一适配层：让业务代码不关心底层是原生还是聚合
# procurement/transport.py
from enum import Enum
from abc import ABC, abstractmethod
from dataclasses import dataclass
class PurchaseChannelKind(str, Enum):
ALI1688_NATIVE = "ali1688_native" # alibaba.trade.*
AGGREGATION = "aggregation" # purchase.order.* 网关
@dataclass
class PurchaseOrderDTO:
erp_order_id: str
platform_order_id: str # 1688 purchase_order_id
flow: str # general / fenxiao / boutiquefenxiao
status: str # wait_pay/paid/shipped/signed/closed
pay_amount: float
freight: float
supplier_id: str
outer_order_no: str # 下游平台单号
raw: dict = None
class PurchaseTransport(ABC):
kind: PurchaseChannelKind
@abstractmethod
def create_order(self, req: dict) -> PurchaseOrderDTO: ...
@abstractmethod
def get_order(self, platform_order_id: str) -> PurchaseOrderDTO: ...
@abstractmethod
def list_orders(self, modified_start: int, page: int = 1) -> list[PurchaseOrderDTO]: ...
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ---------------- 原生 1688 ----------------
class Ali1688NativeTransport(PurchaseTransport):
kind = PurchaseChannelKind.ALI1688_NATIVE
def __init__(self, top_client):
self.top = top_client
def create_order(self, req: dict) -> PurchaseOrderDTO:
# 1) 预览（原生优势）
pre = self.top.execute("alibaba.createOrder.preview", req)
if not pre.get("success"):
raise RuntimeError(pre.get("errorMsg"))
# 2) 快速建单 / 分销单
method = "alibaba.trade.fenxiaoOrder.create" if req.get("is_fenxiao") \
else "alibaba.trade.fastCreateOrder"
resp = self.top.execute(method, {
"flow": req["flow"], # boutiquefenxiao / fenxiao / general
"outerOrderNo": req["erp_order_id"],
"cargoParamList": req["cargo_list"],
"addressParam": req["address"],
"encryptOutOrderInfo": req.get("encrypt_out_order_info"),
})
return PurchaseOrderDTO(
erp_order_id=req["erp_order_id"],
platform_order_id=resp["orderId"],
flow=req["flow"],
status="wait_pay",
pay_amount=resp["totalAmount"],
freight=resp.get("freight", 0.0),
supplier_id=resp["supplier_login_id"],
outer_order_no=req["erp_order_id"],
raw=resp,
)
def get_order(self, platform_order_id: str) -> PurchaseOrderDTO:
resp = self.top.execute("alibaba.trade.get.buyerView",
{"orderId": platform_order_id})
return PurchaseOrderDTO(
erp_order_id=resp.get("outerOrderNo", ""),
platform_order_id=platform_order_id,
flow=resp.get("flow", ""),
status=resp["statusInfo"]["orderStatus"],
pay_amount=float(resp["payAmount"]),
freight=float(resp.get("freight", 0)),
supplier_id=resp.get("supplier_login_id", ""),
outer_order_no=resp.get("outerOrderNo", ""),
raw=resp,
)
def list_orders(self, modified_start: int, page: int = 1):
resp = self.top.execute("alibaba.trade.buyer.list",
{"modifyStartDate": modified_start, "page": page})
return [self._norm(o) for o in resp.get("orderList", [])]
# ---------------- 聚合网关 ----------------
class AggregationPurchaseTransport(PurchaseTransport):
kind = PurchaseChannelKind.AGGREGATION
def __init__(self, http_client, base_url: str, app_key: str, app_secret: str):
self.http = http_client
self.base = base_url
self.app_key = app_key
self.app_secret = app_secret
def create_order(self, req: dict) -> PurchaseOrderDTO:
# 聚合层通常没有 preview，业务层要自己先算价
resp = self.http.post(f"{self.base}/purchase.order.create", json=req)
if not resp.json().get("success"):
raise RuntimeError(resp.json().get("msg"))
d = resp.json()["data"]
return PurchaseOrderDTO(
erp_order_id=req["erp_order_id"],
platform_order_id=d["purchase_order_id"],
flow=d.get("flow", "general"),
status=d.get("status", "wait_pay"),
pay_amount=float(d["amount"]),
freight=float(d.get("freight", 0)),
supplier_id=d.get("supplier_id", ""),
outer_order_no=req["erp_order_id"],
raw=d,
)
def get_order(self, platform_order_id: str) -> PurchaseOrderDTO:
d = self.http.get(f"{self.base}/purchase.order.get",
params={"purchase_order_id": platform_order_id}).json()["data"]
return PurchaseOrderDTO(
erp_order_id=d.get("erp_order_id", ""),
platform_order_id=platform_order_id,
flow=d.get("flow", "general"),
status=d["status"],
pay_amount=float(d["amount"]),
freight=float(d.get("freight", 0)),
supplier_id=d.get("supplier_id", ""),
outer_order_no=d.get("erp_order_id", ""),
raw=d,
)
def list_orders(self, modified_start: int, page: int = 1):
d = self.http.get(f"{self.base}/purchase.order.list",
params={"modified_start": modified_start, "page": page}).json()
return [self._norm(x) for x in d.get("list", [])]五、编排层：永远以原生为真相源
class PurchaseOrderService: def __init__(self, transport: PurchaseTransport): self.t = transport def sync_one(self, platform_order_id: str, registry): dto = self.t.get_order(platform_order_id) # 聚合层也别全信：原生可再核一次 if self.t.kind == PurchaseChannelKind.AGGREGATION: native = native_transport.get_order(platform_order_id) # 可选 dto = native or dto registry.upsert_purchase_order(dto) return dto
规则：
写操作：原生优先；聚合只用于“能省事但不关键”的场景
读操作：聚合可以看，但支付/退款/发货前必须回
alibaba.trade.*拉一次状态机：只用
trade.order.get / get.buyerView推状态，purchase.order.*只做缓存视图
六、flow 选择（这是 alibaba.trade.* 的隐藏主线）
general 普通批发 fenxiao 老代销 saleproxy 分销一件代发（校验分销关系） boutiquefenxiao 新严选 1 件包邮价 boutiquepifa 新严选 多件批发价 paired 天天特卖 repurchase 复购合约
用错 flow：
严选 SKU 走
general→ 价格高、不包邮代发单不走
fenxiao/saleproxy→ 下游单号/密文面单不回流新严选不走
boutiquefenxiao→ “无 retailPrice / 下单失败”
七、和前几篇收口
《1688 订单 API》：
trade.order.get是真相源；本文说“别把聚合层当真相源”《1688 跨境分销》：
boutiquefenxiao/isv_fxgl/fenxiaoMedia都在alibaba.trade.*里《统一采购适配层》：
offerId+skuId → erp_order_id最终落成的就是PurchaseOrderDTO《两套接口边界》：1688 原生 = 供货侧真理；
purchase.order.*= 中间商抽象
八、一句话收口
alibaba.trade.*是 1688 的“采购真相层”；purchase.order.*只是你或聚合商给老板看的“采购业务层”。生产系统里：下单/支付/退款/密文/严选价 → 必须走alibaba.trade.*；内部看板/低代码/MVP → 可以叫purchase.order.*，但落库前一定要回原生核验。
