# 双 11 前必须打通的三条数据链路，晚一周就来不及（附 python 源码）
距离大促只剩几天，这个时间点最不该做的事是加功能。
但有三条链路如果现在还没打通，这几天必须打通。不是为了让系统更先进，是为了**不在这三天里出事故**。
大促出事故，从来不是因为系统不够快。是三种最朴素的情况：
- **卖错了价**——活动没生效、券没叠加、改价延迟
- **卖超了货**——多平台同时卖同一批货
- **客户付了钱没发货**——订单漏拉，等你发现时已经被投诉
这三件事在平时都是小概率。在大促它们是必然，因为触发条件全被放大了：
| 放大项 | 平时 | 大促 | 后果 |
| ------ | -------- | ---------------- | --------------- |
| 请求量 | 100 单/小时 | 5,000 单/小时 | 限流从"偶尔撞"变成"必然撞" |
| 变更频率 | 一天改 3 次价 | 一小时改 3 次 | 人工操作的错误率线性上升 |
| 单次错误代价 | 超卖赔 5 块 | 赔 50 + 平台处罚 + 口碑 | 容错空间被压到零 |
所以这篇不讲架构，只讲三件事：**每条链路的红线是什么、崩了会怎样、怎么用一百多行代码守住**。
## 一、先讲一个认知：大促期间你不是在"同步"，是在"核对"
这是全文最重要的一句话，也是大多数人搞错的地方。
日常开发的心智模型是"同步"：我拉取平台数据，写进我的库，保持一致。
大促期间这个模型会崩，因为**写者有两个人**：
```
运营 → 在平台后台改价、改库存、设活动
你 → 在系统里同步价格、同步库存
```
两个写者，必然打架。你 10:00 同步下来的价格，运营 10:01 在后台改了，你的库就是错的；反过来，你写回去的价格可能覆盖掉运营刚设的活动价。
**正确的心智模型是：平台是唯一真相，你的系统只做三件事——读、核对、告警。**
- 你**不写**价格（除了你自己有定价权的那部分）
- 你**不猜**库存（平台的数就是数，你只维护自己占用的部分）
- 你**不假设**订单拉全了（必须用单数对账验证）
这个转变很难，因为它意味着放弃"我控制一切"的安全感。但大促期间，控制欲就是事故源。
下面三条链路，全都建立在这个前提上：**每个 guard 都是"预期 vs 实际"的比对器，不是同步器。**
## 二、先把地基打了：一张告警表 + 一个落库函数
三条链路都要写告警，所以先把这部分抽出来。它只有几十行，但有两个设计决定值得说：
```python
# 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 块。
**所以价格核对必须核到"到手价"这一层**，而且要能拆开看是哪一层没生效。
### 代码
```python
# guard/price.py
"""价格核对器。注意：它不是同步器，不写平台价格。
定位：运营是唯一写者，我们是观测者。
发现偏差 → 告警 → 人工确认 → 人工修。
绝不允许程序自动"修正"平台价格——大促期间自动写价是灾难。
"""
import time
from decimal import Decimal, InvalidOperation
from base import alert
TOL = Decimal("0.01") # 1 分钱。低于这个数就是数据精度问题，不是业务问题
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
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 分钟 | 10 秒 | 盯住生效，这是最容易出错的窗口 |
| 活动生效后 10 分钟 | 10 秒 | 确认生效且没有错层 |
| 活动进行中 | 1 分钟 | 运营仍在改价、改券 |
| 凌晨低峰 | 5 分钟 | 省配额 |
**不要全天 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 ← 谁先卖完可以借，但缓冲不能被借光
```
**缓冲池是关键**。它让你在某个平台爆单时还有得卖，同时给你时间调整分配。
### 代码
```python
# 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))
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
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"
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 秒。
真正解决"我到底漏没漏"的，是**单数对账**：
```python
# 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
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 也不用上消息队列。如果你现在离大促只剩三天，从订单对账开始——两小时能上线，是这三件事里性价比最高的。
需要完整的表结构和告警落库代码，可以找我们要，也可以直接拿这套思路套你们自己的库。大促加油。
