商品价格总是慢半拍？三平台数据实时同步的做法（附 Python 源码）

淘宝 / 1688 / 京东 三平台商品价格同步，从「每天跑一次脚本」到「分钟级感知」，中间差的不只是一个定时任务。

一、先说场景

做电商的都遇到过这种场面：

• 竞品昨天降了 5 块，你今早才发现，昨天的爆款流量全被吃走了；
• 三个平台各维护一份价格表，运营改完淘宝忘了改 1688，客户来问「为什么你这边贵 12 块」；
• 自己写了个爬虫，每天凌晨 3 点跑一次全量，跑一次两小时，跑完数据已经是昨天的了。

价格是电商里唯一「过期即无效」的字段。 慢半天，基本等于没有。

很多人第一反应是「加个定时任务，跑勤一点」。但把 5 分钟改成 1 分钟，问题往往更严重：接口限流、IP 被封、数据库被 diff 写爆，最后还是慢。

所以「实时同步」真正的难点不是拉得快，而是怎么在限额内，用最少的请求，感知到最小的变化，并且把变化推给正确的人。

二、为什么会「慢半拍」

先复盘一下常见的五个坑，基本能覆盖 90% 的情况：

其中第 4 条最坑。淘宝的 price 是原价，真实成交要看 promotion_price 加优惠券；京东的 jdPrice 和 price 也不是一回事。字段用错，同步再快也是错的。

三、整体方案：三层结构

┌─────────────── 采集层 ───────────────┐
│  API 轮询（增量）  +  消息推送（TMC/订阅） │
└───────────────┬──────────────────────┘
                ↓ 统一成 Offer 模型
┌─────────────── 变更层 ───────────────┐
│  指纹 Hash 比对 → 只处理真正变化的数据   │
└───────────────┬──────────────────────┘
                ↓ 变更事件
┌─────────────── 分发层 ───────────────┐
│  写库 / 刷新缓存 / Webhook / 消息队列   │
└──────────────────────────────────────┘

三个关键设计：

1. 增量而非全量。 只轮询「最近有动销 / 有改价操作」的商品，全量兜底放到凌晨低峰。
2. 指纹比对。 对 (价格, 库存, 活动) 做一次 hash，hash 没变就整条丢弃，不写库、不发事件。
3. 双通道。 接口拉取保证「最终一致」，消息推送负责「实时」。只靠轮询，实时性上限就是轮询周期。

下面上代码。

四、Python 源码

依赖只有三个：

pip install httpx aiohttp  # aiohttp 可选

4.1 统一数据模型

不管来自哪个平台，先归一成一个 Offer。这一步做好，后面所有逻辑都不用再关心平台差异。

# models.py
from dataclasses import dataclass
from decimal import Decimal
import hashlib, json, time


@dataclass
class Offer:
    platform: str            # taobao / alibaba / jd
    item_id: str             # 平台商品 ID
    sku_id: str | None       # 规格 ID，无规格为 None
    title: str
    list_price: Decimal      # 原价 / 划线价
    deal_price: Decimal      # 到手价（真实成交价）
    stock: int | None
    ts: int                  # 平台侧数据时间戳（秒）

    def fingerprint(self) -> str:
        """价格/库存/活动任一变化，指纹就会变"""
        payload = {
   
            "lp": str(self.list_price),
            "dp": str(self.deal_price),
            "st": self.stock,
        }
        raw = json.dumps(payload, sort_keys=True, separators=(",", ":"))
        return hashlib.md5(raw.encode()).hexdigest()

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def now_ts() -> int:
    return int(time.time())

注意 Decimal 不要用 float。价格用浮点做 diff，会出现 99.9 != 99.90000000000001 这种假变更，一天能给你刷出上万条无效事件。

4.2 三平台签名

三个平台的签名规则不一样，这是对接时最花时间的部分，直接抄：

# signers.py
import hashlib
import hmac


def sign_taobao(params: dict, app_secret: str) -> str:
    """淘宝 TOP：MD5(secret + 排序后的 k+v 拼接 + secret)，结果大写"""
    keys = sorted(params.keys())
    raw = app_secret + "".join(f"{k}{params[k]}" for k in keys) + app_secret
    return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()


def sign_alibaba(params: dict, app_secret: str) -> str:
    """1688 开放平台：HMAC-SHA1，key 为 appSecret，结果大写 HEX"""
    keys = sorted(params.keys())
    raw = "".join(f"{k}{params[k]}" for k in keys)
    return hmac.new(
        app_secret.encode("utf-8"), raw.encode("utf-8"), hashlib.sha1
    ).hexdigest().upper()

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def sign_jd(params: dict, app_secret: str) -> str:
    """京东宙斯：MD5(secret + 排序后的 k+v 拼接 + secret)，结果大写"""
    keys = sorted(params.keys())
    raw = app_secret + "".join(f"{k}{params[k]}" for k in keys) + app_secret
    return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

三个平台里淘宝和京东规则几乎一样（都是 MD5 包夹），1688 用 HMAC-SHA1，别搞混。签名错的时候接口一般只回一个 sign error，不告诉你哪错了，建议本地先用官方 SDK 的签名结果对一遍。

4.3 平台适配器

每个平台一个类，对外只暴露 fetch(item_ids) -> list[Offer]。

# adapters.py
import asyncio
import httpx
from decimal import Decimal

from models import Offer, now_ts
from signers import sign_taobao, sign_alibaba, sign_jd


class BaseAdapter:
    name = ""
    sem: asyncio.Semaphore

    def __init__(self, app_key: str, app_secret: str, qps: int = 5):
        self.app_key = app_key
        self.app_secret = app_secret
        self.sem = asyncio.Semaphore(qps)   # 每平台独立限流
        self.client = httpx.AsyncClient(timeout=10.0)

    async def fetch(self, item_ids: list[str]) -> list[Offer]:
        raise NotImplementedError

    async def close(self):
        await self.client.aclose()


class TaobaoAdapter(BaseAdapter):
    name = "taobao"
    URL = "https://eco.taobao.com/router/rest"

    async def _one(self, item_id: str) -> list[Offer]:
        params = {
   
            "method": "taobao.item.seller.get",
            "app_key": self.app_key,
            "timestamp": now_ts(),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            "fields": "num_iid,title,price,sku,quantity",
            "num_iid": item_id,
        }
        params["sign"] = sign_taobao(params, self.app_secret)

        async with self.sem:
            r = await self.client.post(self.URL, data=params)
        data = r.json().get("item_get_response", {
   }).get("item")
        if not data:
            return []

        # 有 SKU 时以 SKU 为准，没有则退化到商品级
        skus = (data.get("skus") or {
   }).get("sku") or []
        if not skus:
            return [Offer(
                platform=self.name,
                item_id=str(data["num_iid"]),
                sku_id=None,
                title=data.get("title", ""),
                list_price=Decimal(str(data.get("price", "0"))),
                deal_price=Decimal(str(data.get("price", "0"))),
                stock=int(data.get("quantity", 0)),
                ts=now_ts(),
            )]

        out = []
        for s in skus:
            out.append(Offer(
                platform=self.name,
                item_id=str(data["num_iid"]),
                sku_id=str(s.get("sku_id")),
                title=f'{data.get("title","")} / {s.get("properties_name","")}',
                list_price=Decimal(str(s.get("price", "0"))),
                deal_price=Decimal(str(s.get("promotion_price") or s.get("price", "0"))),
                stock=int(s.get("quantity", 0)),
                ts=now_ts(),
            ))
        return out

    async def fetch(self, item_ids):
        results = await asyncio.gather(*[self._one(i) for i in item_ids])
        return [o for group in results for o in group]

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class AlibabaAdapter(BaseAdapter):
    name = "alibaba"
    BASE = "https://gw.open.1688.com/openapi/param2/1/com.alibaba.product"

    async def _one(self, item_id: str) -> list[Offer]:
        api = "alibaba.product.get"
        params = {
   
            "access_token": self.app_key,
            "productID": item_id,
            "webSite": "1688",
        }
        params["_aop_signature"] = sign_alibaba(params, self.app_secret)

        async with self.sem:
            r = await self.client.post(f"{self.BASE}/{api}/{self.app_key}", data=params)
        d = r.json().get("result", {
   }) or {
   }
        if not d:
            return []

        ranges = d.get("saleInfo", {
   }).get("priceRanges") or []
        price = Decimal(str(ranges[0]["price"])) if ranges else Decimal("0")
        return [Offer(
            platform=self.name,
            item_id=str(d.get("productID", item_id)),
            sku_id=None,
            title=d.get("subject", ""),
            list_price=price,
            deal_price=price,     # 1688 阶梯价，按起订量取首档，业务侧再折算
            stock=int(d.get("saleInfo", {
   }).get("amountOnSale", 0) or 0),
            ts=now_ts(),
        )]

    async def fetch(self, item_ids):
        results = await asyncio.gather(*[self._one(i) for i in item_ids])
        return [o for group in results for o in group]


class JdAdapter(BaseAdapter):
    name = "jd"
    URL = "https://api.jd.com/routerjson"

    async def _one(self, sku: str) -> list[Offer]:
        params = {
   
            "method": "jd.price.getSellPrice",
            "app_key": self.app_key,
            "access_token": self.app_key,
            "timestamp": str(now_ts()),
            "format": "json",
            "v": "1.0",
            "360buy_param_json": '{"skuIds":"%s"}' % sku,
        }
        params["sign"] = sign_jd(params, self.app_secret)

        async with self.sem:
            r = await self.client.post(self.URL, data=params)
        resp = r.json().get("jingdong_price_getSellPrice_responce", {
   })
        rows = (resp.get("result") or {
   }).get("data") or []
        if not rows:
            return []

        row = rows[0]
        return [Offer(
            platform=self.name,
            item_id=str(sku),
            sku_id=str(sku),
            title="",
            list_price=Decimal(str(row.get("price", "0"))),
            deal_price=Decimal(str(row.get("jdPrice", row.get("price", "0")))),
            stock=None,
            ts=now_ts(),
        )]

    async def fetch(self, item_ids):
        results = await asyncio.gather(*[self._one(i) for i in item_ids])
        return [o for group in results for o in group]

上面代码里的接口名和字段是按各平台开放平台文档写的，真正跑之前把 app_key / app_secret / access_token 换成自己的，字段名以你申请到的那版文档为准（淘宝有 item.seller.get 和 item.get 两套，权限不同）。

4.4 同步引擎：diff + 分发

核心就是「比指纹，只处理变化」。

# sync.py
import asyncio
import sqlite3
from datetime import datetime

from adapters import TaobaoAdapter, AlibabaAdapter, JdAdapter


class PriceSync:
    def __init__(self, db_path="price.db"):
        self.db = sqlite3.connect(db_path, check_same_thread=False)
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS offer_state (
                platform TEXT, item_id TEXT, sku_id TEXT,
                fp TEXT, deal_price TEXT, stock INTEGER,
                updated_at TEXT,
                PRIMARY KEY (platform, item_id, sku_id)
            )
        """)
        self.db.commit()

    def _load_fp(self, o) -> str | None:
        cur = self.db.execute(
            "SELECT fp FROM offer_state WHERE platform=? AND item_id=? AND sku_id IS ?",
            (o.platform, o.item_id, o.sku_id),
        )
        row = cur.fetchone()
        return row[0] if row else None

    def _save(self, o, fp, changed):
        self.db.execute(
            """INSERT INTO offer_state(platform,item_id,sku_id,fp,deal_price,stock,updated_at)
               VALUES(?,?,?,?,?,?,?)
               ON CONFLICT(platform,item_id,sku_id) DO UPDATE SET
                 fp=excluded.fp, deal_price=excluded.deal_price,
                 stock=excluded.stock, updated_at=excluded.updated_at""",
            (o.platform, o.item_id, o.sku_id, fp, str(o.deal_price),
             o.stock, datetime.now().isoformat(timespec="seconds")),
        )
        self.db.commit()

    async def run_once(self, adapters, targets: dict[str, list[str]]) -> list[dict]:
        offers = []
        for ad in adapters:
            ids = targets.get(ad.name, [])
            if not ids:
                continue
            got = await ad.fetch(ids)
            offers.extend(got)
            print(f"[{ad.name}] fetched {len(got)} offers")

        events = []
        for o in offers:
            fp = o.fingerprint()
            old_fp = self._load_fp(o)
            if old_fp == fp:
                continue                      # 没变，直接丢
            self._save(o, fp, changed=True)
            events.append({
   
                "platform": o.platform,
                "item_id": o.item_id,
                "sku_id": o.sku_id,
                "title": o.title,
                "old_price": None,            # 需要旧价就再查一次
                "new_price": str(o.deal_price),
                "stock": o.stock,
                "is_new": old_fp is None,
                "ts": o.ts,
            })
        return events

跑起来：

```python
# main.py
import asyncio, json
from adapters import TaobaoAdapter, AlibabaAdapter, JdAdapter
from sync import PriceSync

TAOBAO_ITEMS = ["654321098765"]
ALIBABA_ITEMS = ["612345678901"]
JD_ITEMS = ["100012043978"]

async def main():
    adapters = [
        TaobaoAdapter("your_appkey", "your_secret", qps=5),
        AlibabaAdapter("your_appkey", "your_secret", qps=3),
        JdAdapter("your_appkey", "your_secret", qps=5),
    ]
    targets = {
        "taobao": TAOBAO_ITEMS,
        "alibaba": ALIBABA_ITEMS,
        "jd": JD_ITEMS,
    }
    sync = PriceSync()

    while True:
        events = await sync.run_once(adapters, targets)
        for e in events:
            print("[CHANGED]", json.dumps(e, ensure_ascii=False))
        await asyncio.sleep(60)      # 轮询周期，按平台限额调

if __name__ == "__main__":
    asyncio.run(main())

跑通之后，把 print 换成发 Webhook 或写消息队列，链路就完整了：

async def push(events, url):
    if not events:
        return
    async with httpx.AsyncClient(timeout=5) as c:
        await c.post(url, json={
   "count": len(events), "events": events})

五、实测效果

一个 3000 个 SKU 的店，改造前后对比：

请求数降了 85%，延迟反而降了两个数量级——关键就在指纹比对把无效写入全砍掉了。

六、四个必须注意的坑

1. 限流要按平台分开算。 淘宝、京东、1688 的 QPS 配额完全是三套体系，共用一个信号量会导致快的被慢的拖死。上面代码里每个 Adapter 自带 Semaphore，就是为这个。
2. 时间戳对齐。 三个平台返回的字段里，有的是服务器时间，有的干脆没有。建议统一用本地拉取时间打 ts，再额外存平台活动时间，别混用。
3. 活动价的判定要单独做。 大促期间 deal_price 会低于 list_price，如果直接拿 deal_price 当日常价，活动结束你的数据会集体偏高。
4. 一定要留全量兜底。 增量靠「动销列表」驱动，一旦动销接口漏数据，就会长期不同步。每天凌晨跑一次全量对账，把指纹不一致的挑出来修正。

七、最后

上面这套代码是能直接跑的最小实现，核心逻辑都在：统一模型 → 指纹比对 → 增量拉取 → 变化推送。真实业务里再往上加三样东西，基本就够用了：变更历史表（用于价格趋势）、告警规则（降价超过 X% 才推）、以及一个简单的看板。

如果你不想自己维护签名、限流、字段映射这一堆琐事，也可以用现成的三平台 API 接口服务，淘宝 / 1688 / 京东的商品、价格、SKU、库存一次拿全，格式统一，省下的时间放在业务逻辑上更划算。

有在做的可以评论区聊聊，尤其是 1688 阶梯价那块的折算，各家做法差别挺大。

标签：Python 电商 API 数据同步 爬虫