双 11 前必须打通的三条数据链路，晚一周就来不及（附 python 源码）

距离大促只剩几天，这个时间点最不该做的事是加功能。

但有三条链路如果现在还没打通，这几天必须打通。不是为了让系统更先进，是为了不在这三天里出事故。

大促出事故，从来不是因为系统不够快。是三种最朴素的情况：

• 卖错了价——活动没生效、券没叠加、改价延迟
• 卖超了货——多平台同时卖同一批货
• 客户付了钱没发货——订单漏拉，等你发现时已经被投诉

这三件事在平时都是小概率。在大促它们是必然，因为触发条件全被放大了：

所以这篇不讲架构，只讲三件事：每条链路的红线是什么、崩了会怎样、怎么用一百多行代码守住。

一、先讲一个认知：大促期间你不是在"同步"，是在"核对"

这是全文最重要的一句话，也是大多数人搞错的地方。

日常开发的心智模型是"同步"：我拉取平台数据，写进我的库，保持一致。

大促期间这个模型会崩，因为写者有两个人：

运营 → 在平台后台改价、改库存、设活动
你   → 在系统里同步价格、同步库存

两个写者，必然打架。你 10:00 同步下来的价格，运营 10:01 在后台改了，你的库就是错的；反过来，你写回去的价格可能覆盖掉运营刚设的活动价。

正确的心智模型是：平台是唯一真相，你的系统只做三件事——读、核对、告警。

• 你不写价格（除了你自己有定价权的那部分）
• 你不猜库存（平台的数就是数，你只维护自己占用的部分）
• 你不假设订单拉全了（必须用单数对账验证）

这个转变很难，因为它意味着放弃"我控制一切"的安全感。但大促期间，控制欲就是事故源。

下面三条链路，全都建立在这个前提上：每个 guard 都是"预期 vs 实际"的比对器，不是同步器。

二、先把地基打了：一张告警表 + 一个落库函数

三条链路都要写告警，所以先把这部分抽出来。它只有几十行，但有两个设计决定值得说：

# guard/base.py
"""三条链路共用的地基：表结构 + 告警落库。

为什么告警落库而不是只发到群里：
  大促期间消息会被刷屏,没人看得过来。
  落库后能按 level 过滤、按 code 聚合,
  事后还能复盘"到底哪类问题先出现"。
"""

import sqlite3
import time

SCHEMA = """
create table if not exists alert(
  id integer primary key autoincrement,
  level text not null,          -- P0 立刻处理 / P1 当天处理
  code text not null,           -- 机器可读,用于聚合和静默
  detail text,
  created_at integer,
  acked integer default 0
);
create index if not exists ix_alert_time on alert(created_at);

create table if not exists price_check(
  id integer primary key autoincrement,
  item_id text, sku_id text,
  expected text, actual text, diff text, layers text,
  checked_at integer
);
create index if not exists ix_price_item on price_check(item_id, checked_at);

create table if not exists stock_level(
  item_id text primary key,
  available integer, reserved integer, real integer,
  floor integer, updated_at integer
);

create table if not exists stock_back(
  id integer primary key autoincrement,
  item_id text, qty integer, source text, day text, created_at integer
);

create table if not exists order_gap(
  platform text, hour_start integer,
  local_count integer, remote_count integer, checked_at integer,
  primary key (platform, hour_start)
);

-- 这张是你们业务库里的表，这里给个最小结构好让 order_lag 能跑
create table if not exists orders(
  order_id text primary key,
  platform text,
  paid_at integer,        -- 买家支付时间,平台给
  created_at integer      -- 进入我方系统的时间,我方记
);
create index if not exists ix_orders_created on orders(created_at);
"""


def open_db(path="promo.db"):
    db = sqlite3.connect(path)
    db.executescript(SCHEMA)
    return db


def alert(db, level, code, detail):
    """落库 + 立刻 commit。

    不用 ORM、不进队列,是刻意的:告警是最后一道防线,
    它自身依赖的组件越少越好。大促期间任何中间件抖动
    都不该让告警丢失。
    """
    db.execute(
        "insert into alert(level, code, detail, created_at) values(?,?,?,?)",
        (level, code, detail, int(time.time())),
    )
    db.commit()
    print(f"[{level}] {code} {detail}")

两个决定：

1. code 字段是机器可读的（PRICE_DRIFT、ORDER_GAP），不是给人看的中文。这样能做聚合——一晚上 300 条 STOCK_LOW 应该合成一条"37 个 SKU 触底"，而不是 300 条消息。
2. 告警必须落库，不能只发消息群。 群消息在大促会被刷屏，落库的能查、能统计、能复盘。

三、链路一：价格——红线是偏差不超过 1 分钱

大促期间它为什么不一样

平时改价是低频事件，错一次改回来就行。大促不一样：

• 活动价有生效时间点（0 点、10 点、20 点），差一分钟就是一批订单按错价成交
• 优惠叠加有三层（标价 → 单品券 → 平台满减），任何一层没生效，到手价就是错的
• 改价窗口集中在开抢前 10 分钟，运营手忙脚乱，改错的概率最高

最容易崩的那个点：你核对的是标价，不是到手价

标价对得上，不代表到手价对得上。

标价      99.00
单品券    -10.00    ← 券没生效
平台满减  -15.00
到手价    74.00     ← 客户实际付的

如果只看标价，99 是对的，一切正常。但客户实际付了 84 而不是 74，客诉从这里开始；反过来如果券提前生效，客户付了 64，每单亏 10 块。

所以价格核对必须核到"到手价"这一层，而且要能拆开看是哪一层没生效。

代码

# guard/price.py
"""价格核对器。注意：它不是同步器，不写平台价格。

定位：运营是唯一写者，我们是观测者。
     发现偏差 → 告警 → 人工确认 → 人工修。
     绝不允许程序自动"修正"平台价格——大促期间自动写价是灾难。
"""

import time
from decimal import Decimal, InvalidOperation

from base import alert

TOL = Decimal("0.01")        # 1 分钱。低于这个数就是数据精度问题，不是业务问题


def to_dec(v):
    try:
        return Decimal(str(v))
    except (InvalidOperation, TypeError, ValueError):
        return None


def check_price(db, item_id, sku_id, expected, actual, layers=None):
    """核对到手价。

    expected: 我方算出的到手价（标价 - 券 - 满减）
    actual:   平台返回的实际到手价
    layers:   {"list": 99, "coupon": -10, "promo": -15}，用于定位是哪一层错

    返回 (是否通过, 偏差)
    """
    e, a = to_dec(expected), to_dec(actual)
    if e is None or a is None:
        # 拿不到价本身就是故障：大促期间价格字段为空，通常是接口降级了
        alert(db, "P0", "PRICE_MISSING",
              f"{item_id}/{sku_id} 价格不可用 expected={expected} actual={actual}")
        return False, None

    diff = a - e
    db.execute(
        "insert into price_check(item_id, sku_id, expected, actual, diff, layers, checked_at) "
        "values(?,?,?,?,?,?,?)",
        (item_id, sku_id, str(e), str(a), str(diff),
         str(layers or {
   }), int(time.time())),
    )

    if abs(diff) <= TOL:
        return True, diff

    # 偏差超过 1% 就是资损级，1 分钱到 1% 之间是配置级
    level = "P0" if abs(diff) > e * Decimal("0.01") else "P1"
    where = _locate(layers, e, a) if layers else "未知层"
    alert(db, level, "PRICE_DRIFT",
          f"{item_id}/{sku_id} 预期 {e} 实际 {a} 偏差 {diff} 疑似 {where}")
    return False, diff


def _locate(layers, expected, actual):
    """把偏差归到具体某一层，方便运营直接去改那一项。

    注意：券和满减都存负数，所以这里是加，不是减。
    写成减法会把符号翻两次，明明是券没生效，却报"我方算价错"。
    """
    lst = to_dec(layers.get("list"))
    coupon = to_dec(layers.get("coupon")) or Decimal("0")
    promo = to_dec(layers.get("promo")) or Decimal("0")
    if lst is None:
        return "标价缺失"

    # 先确认我方算得对
    if lst + coupon + promo != expected:
        return "我方算价逻辑"

    # 我方算对了，那就是平台少减了某一层——逐层试，命中哪个就是哪个
    if actual == lst + promo:
        return "券未生效"
    if actual == lst + coupon:
        return "满减未生效"
    if actual == lst:
        return "券和满减都未生效"
    return "平台价格与任一层组合都不符"

大促期间的核对频率

平时 30 分钟一次就够。大促要按活动时间点加密：

不要全天 10 秒一次。大促期间配额是最贵的资源，价格核对一秒钟扫一万个 SKU，会把库存和订单链路的额度吃光——这是第 10 篇那个"全局桶"的另一种表现。

四、链路二：库存——红线是可售不低于安全水位

大促期间它为什么不一样

平时库存是"还剩多少"。大促库存是"我承诺了多少"。

一个 SKU 在大促期间同时被四种东西占用：

物理库存 120 件
├── 日常可售      40    ← 期就在卖的
├── 秒杀预留      50    ← 20:00 开抢，此刻不能卖
├── 预售定金      20    ← 已付定金，待付尾款
└── 安全水位      10    ← 预留的缓冲，防超卖

如果你只维护一个数字 120，那么秒杀开始的瞬间，日常订单会把这 50 件吃掉，等到 20:00 开抢，库存是 0。

所以库存不是数量，是承诺。要按用途分池，而不是一个总数。

最容易崩的那个点：多平台同时卖同一批货

这是超卖的头号原因，也是最多人踩的：

实际仓库：120 件
淘宝设置：100 件
京东设置：100 件
→ 两边各卖 100，一共 200，超卖 80 件

平台之间不通气，谁都不知道对方卖了多少。你必须在自己这边做池化：

物理池 120
  ├── 淘宝额度 60
  ├── 京东额度 50
  └── 缓冲 10        ← 谁先卖完可以借，但缓冲不能被借光

缓冲池是关键。它让你在某个平台爆单时还有得卖，同时给你时间调整分配。

代码

# guard/stock.py
"""库存守护。两个职责：
   1. 水位熔断——可售低于安全线就自动下架或限购
   2. 池化校验——各平台额度之和不能超过物理池

注意：这里的 available 是平台可售数，reserved 是本系统已占用但
     还没回写到平台的量。真实可售 = available - reserved。
     这两个数分开存，是因为回写有延迟，混在一起会重复扣。
"""

import time

from base import alert


def real_available(db, item_id, platform_available, reserved):
    """扣掉本地占用后的真实可售"""
    return max(0, int(platform_available) - int(reserved))


def check_stock(db, item_id, platform_available, reserved, floor=10, action="off"):
    """水位检查。返回建议动作：normal / limit / off

    floor: 安全水位。大促建议按日均销量的 0.5 倍设，不要凭感觉。
    action: 触底后的动作，off=下架，limit=限购 1 件
    """
    raw = int(platform_available) - int(reserved)
    avail = max(0, raw)

    db.execute(
        "insert or replace into stock_level"
        "(item_id, available, reserved, real, floor, updated_at) values(?,?,?,?,?,?)",
        (item_id, platform_available, reserved, avail, floor, int(time.time())),
    )
    db.commit()

    # 超卖和售罄必须分开报：售罄就是没货可卖，超卖是要走赔付预案的事故
    if raw < 0:
        alert(db, "P0", "STOCK_NEGATIVE",
              f"{item_id} 超卖 {-raw} 件（平台 {platform_available} 本地占用 {reserved}），走赔付预案")
        return "off"

    if avail <= 0:
        alert(db, "P0", "STOCK_OUT",
              f"{item_id} 可售为 0（平台 {platform_available} 本地占用 {reserved}）")
        return "off"

    if avail <= floor:
        # 大促期间这里必须是 P0：P1 会淹没在其他告警里没人看
        alert(db, "P0", "STOCK_LOW",
              f"{item_id} 可售 {avail} 已到水位 {floor}，建议{action}")
        return action

    return "normal"

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def check_pool(db, item_id, physical, quotas):
    """池化校验：各平台额度之和不能超过物理库存

    quotas: {"taobao": 60, "jd": 50, "pdd": 0}
    超配是超卖的前置条件，必须在开卖前拦住，不是卖的时候才管。
    """
    total = sum(quotas.values())
    if total > physical:
        alert(db, "P0", "POOL_OVERCOMMIT",
              f"{item_id} 物理 {physical} 分配 {total}（{quotas}）超配 {total - physical}")
        return False
    return True


def replenish(db, item_id, qty, source, daily_sales=None):
    """库存回补：取消单、退款、超时未付款都要回补

    但必须设上限：刷单党会反复下单占库存再取消，
    没有上限的回补等于把库存交给对手。

    daily_sales 传当日销量，上限按 20% 算；
    拿不到销量就退回固定上限。**不能用 qty 自己算上限**——
    那等于"回补多少就允许多少"，一点约束都没有。
    """
    used = db.execute(
        "select coalesce(sum(qty),0) from stock_back where item_id=? and day=?",
        (item_id, time.strftime("%Y%m%d")),
    ).fetchone()[0]

    cap = max(20, int(daily_sales * 0.2)) if daily_sales else 50

    if used + qty > cap:
        alert(db, "P1", "REPLENISH_ABNORMAL",
              f"{item_id} 今日已回补 {used}，本次 {qty} 超上限 {cap}，来源 {source}")
        return False

    db.execute(
        "insert into stock_back(item_id, qty, source, day, created_at) values(?,?,?,?,?)",
        (item_id, qty, source, time.strftime("%Y%m%d"), int(time.time())),
    )
    db.commit()
    return True

那个 replenish 的限流容易被忽略，但它是真事：大促期间同行恶意下单占库存再取消，能把你的秒杀库存锁死。回补不设上限，等于把库存交给对手。

五、链路三：订单——红线是漏单为零

大促期间它为什么不一样

订单链路的特殊性在于：漏单不可恢复，重复可以吸收。

• 重复拉一单 → ON CONFLICT 幂等掉，无损失
• 漏拉一单 → 客户付了钱，你没发货，等超时罚款才知道

所以整条链路的设计原则是：宁可重复十次，不可遗漏一次（第 3 篇那条铁律，在大促是生死线）。

大促会同时放大三个坑：

坑一：分页漂移。 你在翻第 3 页的时候，第 1 页又插入了新订单，整个列表往后挤。第 3 页的部分单会跑到第 4 页，被你跳过。

坑二：时间窗边界。 同一秒多单，用 modified 时间戳做游标，等于把同一秒的单全丢了。

坑三：接口超时。 大促期间超时率从 0.1% 涨到 5%，一次超时就是整页丢。

解法：双游标 + 单数对账

双游标（第 3 篇讲过）解决坑一和坑二：游标存 (时间戳, 订单号) 二元组，重叠加窗 60 秒。

真正解决"我到底漏没漏"的，是单数对账：

# guard/order.py
"""订单对账。核心不是"拉",是"验"。

拉订单的代码到处都有，能回答"我拉全了吗"的代码很少。
方法：用平台的订单计数接口，和自己库里的数比。
     不等就说明漏了,重拉整个时间窗。
"""

import time

from base import alert


def reconcile(db, platform, hour_start, local_count, remote_count):
    """按小时对账。

    为什么按小时不按天：
      按天对账，漏单要等到次日凌晨才发现，那时候客户已经投诉了。
      按小时对账，最晚一小时发现，还有救。
    """
    db.execute(
        "insert or replace into order_gap"
        "(platform, hour_start, local_count, remote_count, checked_at) values(?,?,?,?,?)",
        (platform, hour_start, local_count, remote_count, int(time.time())),
    )
    db.commit()

    if local_count == remote_count:
        # 注意：0 比 0 也会走到这里。如果拉单任务本身死了，
        # 两边都是 0，"对账通过"毫无意义。这种情况只能靠拉单任务的
        # 心跳发现，不能靠对账。第一次跑之前务必确认拉单是活的。
        return False

    gap = remote_count - local_count
    if gap > 0:
        # 漏单，P0。重拉这个小时的全部订单，不要试图"补差额"——
        # 你不知道漏的是哪几单，只能整窗重拉。
        alert(db, "P0", "ORDER_GAP",
              f"{platform} {_fmt(hour_start)} 本地 {local_count} 平台 {remote_count} "
              f"漏 {gap} 单，需整窗重拉")
        return True

    # 本地比平台多：重复或脏数据，P1，可以慢慢查
    alert(db, "P1", "ORDER_DUP",
          f"{platform} {_fmt(hour_start)} 本地 {local_count} 平台 {remote_count} 多 {-gap}")
    return False

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def _fmt(ts):
    return time.strftime("%m-%d %H:%M", time.localtime(ts))


def order_lag(db, limit=20):
    """订单延迟：从支付到进入本系统的时间差

    这个指标比"成功率"重要。大促发货时限是硬指标，
    延迟 30 分钟意味着你有 30 分钟的订单还不知道要发。
    """
    rows = db.execute(
        "select order_id, paid_at, created_at from orders "
        "order by created_at desc limit ?", (limit,),
    ).fetchall()
    lags = [c - p for _, p, c in rows if p and c > p]
    if not lags:
        return None
    lags.sort()
    p95 = lags[max(0, int(len(lags) * 0.95) - 1)]
    return {
   "p50": lags[len(lags) // 2], "p95": p95, "max": lags[-1]}


def check_lag(db, warn=300, crit=900):
    """延迟超阈值告警。阈值按平台的发货时限倒推：
      24 小时发货 → 延迟 15 分钟是危急（crit）
      48 小时发货 → 30 分钟告警就够
    """
    lag = order_lag(db)
    if not lag:
        return
    if lag["p95"] > crit:
        alert(db, "P0", "ORDER_LAG", f"订单延迟 P95 {lag['p95']}s 超过 {crit}s")
    elif lag["p95"] > warn:
        alert(db, "P1", "ORDER_LAG", f"订单延迟 P95 {lag['p95']}s 超过 {warn}s")

### 一个容易被忽略的点：订单链路的速度比完整性更容易崩

完整性靠对账能补。**速度补不回来。**

客户 10:00 付款，你 10:40 才拉到，那 40 分钟里你不知道要发什么货。大促发货时限按支付时间算，不是按你拉到的时间算。

所以订单链路要盯两个指标，不是一个：

| 指标 | 含义 | 大促红线 |
|---|---|---|
| 漏单率 | 本地单数 / 平台单数 | 必须 100%，差一单就是 P0 |
| 延迟 P95 | 支付到入库的时间 | 24h 发货时限 → ≤ 5 分钟 |

**只看漏单率会漏掉最痛的那个问题**：一单没漏，但都晚到 40 分钟，照样发不出去。

## 六、只剩 72 小时，先做哪条

现实一点：如果现在离大促只有三天，三条全做完不现实。按这个顺序砍：

| 优先级 | 做什么 | 工作量 | 为什么先做它 |
|---|---|---|---|
| **1** | 订单单数对账 | 2 小时 | 成本最低、收益最高。漏单是唯一"不可恢复"的事故 |
| **2** | 库存水位熔断 | 3 小时 | 超卖要赔付 + 平台处罚，且影响店铺权重 |
| **3** | 价格到手价核对 | 1 天 | 工作量最大（要拆三层），但错价的损失有上限，且能事后追 |

三条里只有第 1 条是"两小时就能上线"的——一个定时任务，拉平台单数，比本地单数，不等就告警。**如果你今天只能做一件事，做这个。**

价格核对放最后不是因为不重要，是因为它最费工：要把标价、券、满减三层都拆开，还要处理不同类目的规则差异。三天做不完，做一半反而危险（核对逻辑错了会误导运营）。

## 七、T-14 到 T+0 的时间表

这张表是给还没到大促的团队用的，存下来，618 还能用。

| 节点 | 该做什么 | 判断标准 |
|---|---|---|
| **T-14** | 三条链路打通 + 单链路压测 | 每条链路都能在峰值 QPS 下跑通，不含告警 |
| **T-7** | 全链路演练（模拟真实峰值 2 倍） | 演练中出现的每个告警都要有处置动作，不能只"知道了" |
| **T-3** | **代码冻结** | 只修 P0 bug，不加功能。这条最难执行，也最该执行 |
| **T-1** | 值守排班 + 降级预案确认 | 谁在 0 点看屏幕，出事了按哪张表操作，必须写下来 |
| **T+0** | 实时监控 + 手工兜底通道 | 系统挂了要能手导单、手改价，这条通道平时就要演练 |

T-3 的代码冻结值得单独说。大促前三天上线新功能，是事故的经典成因——新代码没有经历过峰值，你不知道它在哪里会崩，而这时候已经没有时间修了。**冻结不是保守，是止损。**

## 八、降级预案：出事时按这张表做，不要临场想

大促值守最怕的不是出事，是出事了大家在群里讨论"要不要关"。

提前把这张表写下来，贴在显示器上：

| 触发条件 | 动作 | 谁来决定 |
|---|---|---|
| 订单接口错误率 > 5%，持续 5 分钟 | 停止自动拉单，切备用数据源；同时开手工导单通道 | 值班开发直接执行，事后报备 |
| 订单延迟 P95 > 15 分钟 | 降低库存/价格核对频率，把配额让给订单 | 值班开发 |
| 库存对账失败（平台值 < 本地值 20% 以上） | 全店限购 1 件，人工核查 | 运营负责人 |
| 价格偏差 > 1% 且影响 SKU > 50 个 | 暂停所有自动调价，转人工 | 运营负责人 |
| 某个平台完全不可用 | 该平台商品全部下架（不是"等它恢复"） | 值班开发 + 运营 |

最后一条最反直觉：**平台挂了，第一反应应该是下架，不是等**。

因为平台挂了但你的商品还在卖，客户下单你收不到单，等平台恢复时你已经积累了一堆超时未发货。下架是止损，等待是赌博。

## 九、三条链路的共同点

写完这三段代码，你会发现它们长得几乎一样：

```python
def guard(expected, actual, threshold):
    if abs(expected - actual) > threshold:
        alert(...)

对，核心就是一个比对加一个告警。大促的稳定性不是靠复杂架构，是靠"每个关键假设都有代码在验证"。

三条链路对应三个假设：

这三个假设，平时错了没人发现（因为量小、能人工兜），大促错了就是事故。

所以大促前该做的不是"优化性能"，是把这三个假设写进代码，让它每小时替你看一遍。

上面四段加起来三百多行，纯标准库加 SQLite，不用起 Redis 也不用上消息队列。如果你现在离大促只剩三天，从订单对账开始——两小时能上线，是这三件事里性价比最高的。

需要完整的表结构和告警落库代码，可以找我们要，也可以直接拿这套思路套你们自己的库。大促加油。