动态定价算法在电商平台中的实战：从原理到 200 行可上线代码

引言：为什么“价格”成了电商增长第二曲线

流量红利见顶后，平台从“卖更多”转向“卖得更聪明”。在供给端 SKU 爆炸、需求端实时波动的双重压力下，静态价签的毛利率损失可达 7–15%（麦肯锡 2023 报告）。动态定价（Dynamic Pricing）因此成为电商增长的新杠杆：它能在毫秒级响应库存、竞对、用户意图等多维信号，把 GMV、毛利或清仓率优化到 Pareto 前沿。

本文以某头部生鲜电商“ 30 分钟极速达”频道为蓝本，拆解其 2024 年 Q2 上线的动态定价系统。全文包含：

1. 业务约束与目标函数
2. 算法选型：从 Thompson Sampling 到深度强化学习
3. 200 行 Python 代码（含特征工程、离线训练、在线推理、A/B 测试）
4. 灰度实验结果与踩坑复盘

所有代码已脱敏，可在单机上复现；如需对接真实订单系统，文末给出扩展指南。

业务场景：30 分钟极速达的价格博弈

商品池与生命周期

• 日均在线 SKU ≈ 4,200，其中 60% 为短保商品（保质期 ≤ 3 天）。
• 库存曲线呈“早高晚低”：0 点大仓铺货，上午 9 点起按门店销量实时补货，22 点后进入清仓时段。

目标函数（多目标优化）

管理层要求“同时保住 GMV、毛利、用户体验”，我们将其转化为带约束的标量目标：

maxₜ Σᵢ (pᵢₜ · qᵢₜ · (1 – α · discᵢₜ))
s.t.

1. 清仓率约束：短保商品日清仓率 ≥ 95%
2. 价格一致性：同城市同商品 1 小时内价差 ≤ 5%
3. 用户体验：调价频率 ≤ 15 min/次，单日最大涨跌 ≤ 20%

α 为管理层对折扣敏感度的超参（实验取 0.3）。

算法架构：三级火箭

Level 1 基线规则引擎（上线 3 天）

• 短保商品：每过 6 小时线性降价 5%。
• 竞对监控：若竞对价低于我方 3%，则跟价到竞对 –1%。
  规则引擎作为兜底，保证业务可解释性。

Level 2 Contextual Bandit（上线第 4–30 天）

将“调价”视为动作，reward = 毛利 + λ·GMV – γ·用户投诉。
选用 Thompson Sampling + 特征分桶，解决冷启动与不确定性。

Level 3 深度强化学习（DRL）

当 Level 2 累积 10w+ 样本后，切换到 DQN + 库存状态机。网络结构：

• 状态 s：库存、竞对价、用户实时搜索量、天气、节假日、配送运力饱和度
• 动作 a：{–6%, –3%, –1%, 0, +1%, +3%, +6%} 七档离散调价
• Reward r：与 Level 2 相同，但引入“保质期惩罚”项：exp(–剩余小时/24)

200 行可运行代码：端到端 demo

下面代码模拟“上海某门店 SKU=12345 的西红柿”在 2024-06-01 一天的动态定价。
依赖：pandas、numpy、torch、gymnasium。

环境安装

pip install pandas numpy torch gymnasium scikit-learn tqdm

数据模拟器

import numpy as np
from datetime import datetime, timedelta
import gymnasium as gym
from gymnasium import spaces

class TomatoEnv(gym.Env):
    """
    模拟西红柿一天的价格与库存
    state: [stock, competitor_price, hour_of_day, weekday, weather, search_volume]
    action: 0~6 映射到 {-6%,-3%,-1%,0,+1%,+3%,+6%}
    """
    def __init__(self):
        super().__init__()
        self.action_space = spaces.Discrete(7)
        self.observation_space = spaces.Box(low=0, high=1, shape=(6,), dtype=np.float32)
        self.reset()

    def _price2action(self, p):
        return int(np.round((p - 0.94) * 100 / 2))

    def reset(self, seed=None, options=None):
        super().reset(seed=seed)
        self.stock = np.random.randint(50, 200)
        self.base_price = 5.8  # 元/500g
        self.hour = 6
        self.weekday = datetime(2024, 6, 1).weekday() / 6  # 0~1
        self.weather = np.random.choice([0.2, 0.5, 0.8])  # 雨 0.2，多云 0.5，晴 0.8
        self.search_volume = np.random.beta(2, 5)  # 0~1
        self.competitor_price = np.random.normal(self.base_price * 0.98, 0.1)
        return self._get_obs(), {}

    def _get_obs(self):
        return np.array([
            self.stock / 200,
            self.competitor_price / 10,
            self.hour / 24,
            self.weekday,
            self.weather,
            self.search_volume
        ], dtype=np.float32)

    def step(self, action):
        assert self.action_space.contains(action)
        delta = [-0.06, -0.03, -0.01, 0, 0.01, 0.03, 0.06][action]
        my_price = self.base_price * (1 + delta)
        self.hour += 1
        # 需求模型：价格弹性 + 天气/搜索量影响
        demand = max(0, (1.5 - 2 * my_price / self.competitor_price) *
                     (self.weather + 0.5) * (self.search_volume + 0.5) * 20)
        sold = min(self.stock, int(demand))
        self.stock -= sold
        revenue = sold * my_price
        cost = sold * 3.5  # 假设进货成本 3.5
        profit = revenue - cost
        # 保质期惩罚
        spoil_penalty = max(0, self.stock * 3.5 * np.exp(-(24 - self.hour) / 24))
        reward = profit - spoil_penalty
        done = self.hour >= 24 or self.stock == 0
        return self._get_obs(), reward, done, False, {}

离线训练

import torch, random, numpy as np
from collections import deque
from env import TomatoEnv

class DQN(torch.nn.Module):
    def __init__(self, n_state, n_action):
        super().__init__()
        self.net = torch.nn.Sequential(
            torch.nn.Linear(n_state, 64),
            torch.nn.ReLU(),
            torch.nn.Linear(64, 64),
            torch.nn.ReLU(),
            torch.nn.Linear(64, n_action)
        )

    def forward(self, x):
        return self.net(x)

BATCH = 32
GAMMA = 0.99
EPS = 0.1
env = TomatoEnv()
n_state = env.observation_space.shape[0]
n_action = env.action_space.n
q_net, tgt_net = DQN(n_state, n_action), DQN(n_state, n_action)
optimizer = torch.optim.Adam(q_net.parameters(), lr=1e-3)
replay = deque(maxlen=5000)

def act(obs):
    if random.random() < EPS:
        return env.action_space.sample()
    obs = torch.tensor(obs).unsqueeze(0).float()
    with torch.no_grad():
        return q_net(obs).argmax().item()

for episode in range(500):
    obs, _ = env.reset()
    done = False
    while not done:
        action = act(obs)
        next_obs, reward, done, _, _ = env.step(action)
        replay.append((obs, action, reward, next_obs, done))
        obs = next_obs
        if len(replay) >= BATCH:
            batch = random.sample(replay, BATCH)
            s, a, r, s2, d = map(np.array, zip(*batch))
            s, s2 = torch.tensor(s).float(), torch.tensor(s2).float()
            r, d = torch.tensor(r).float(), torch.tensor(d).float()
            q = q_net(s).gather(1, torch.tensor(a).unsqueeze(1)).squeeze()
            with torch.no_grad():
                q_next = tgt_net(s2).max(1)[0]
                target = r + GAMMA * q_next * (1 - d)
            loss = torch.nn.functional.mse_loss(q, target)
            optimizer.zero_grad(); loss.backward(); optimizer.step()
    if episode % 50 == 0:
        tgt_net.load_state_dict(q_net.state_dict())

torch.save(q_net.state_dict(), "tomato_dqn.pt")

在线推理

import torch, json, time
from env import TomatoEnv
from train import DQN

env = TomatoEnv()
model = DQN(env.observation_space.shape[0], env.action_space.n)
model.load_state_dict(torch.load("tomato_dqn.pt"))
model.eval()

def price_serve(stock, competitor_price, hour, weekday, weather, search_volume):
    obs = np.array([stock/200, competitor_price/10, hour/24, weekday/6, weather, search_volume], dtype=np.float32)
    obs = torch.tensor(obs).unsqueeze(0).float()
    with torch.no_grad():
        action = model(obs).argmax().item()
    delta = [-0.06, -0.03, -0.01, 0, 0.01, 0.03, 0.06][action]
    return round(5.8 * (1 + delta), 2)

if __name__ == "__main__":
    # 模拟实时调用
    for h in range(6, 24):
        print("hour", h, "price", price_serve(stock=np.random.randint(20, 100),
                                              competitor_price=5.9,
                                              hour=h,
                                              weekday=5,
                                              weather=0.7,
                                              search_volume=0.6))
        time.sleep(1)

线上 A/B 实验结果

实验设计

• 桶：上海 30 家门店，随机 50% 走 DRL，50% 走 Level 2 Bandit。
• 周期：2024-06-10 至 2024-06-30，共 21 天。
• 指标：GMV、毛利、短保清仓率、用户投诉率。

核心结论

案例分析

• 6 月 15 日暴雨：Bandit 在 13–15 点降价过猛（–15%），DRL 通过天气特征仅降 8%，既保证 GMV 又减少毛利损失。
• 6 月 20 日 22 点：DRL 提前识别库存积压，在 21:30 启动“清仓价”动作，使西红柿在 23:45 前售罄；Bandit 因缺乏库存状态机，22:30 仍有 20% 库存未清。

踩坑与治理

1. 竞对价格爬虫异常

竞对价突然拉高 3 倍，导致 DRL 动作异常。解决：

• 引入“价格漂移检测”——当日均竞对价 > 过去 7 天 1.5σ 时，触发人工审核。
• 训练集加入对抗样本（±20% 噪声）。

2. 用户侧价格一致性感知

部分用户发现 30 分钟内同门店价差 7%，投诉“杀熟”。解决：

• 在 reward 中加入“价差惩罚”：|pₜ – pₜ₋₁| > 5% 时，reward -= 0.02·GMV。
• 前端展示“价格保护倒计时”提升透明度。

3. 门店执行延迟

调价指令通过门店 ERP 下发，平均延迟 3–5 分钟，导致策略漂移。解决：

• 将“延迟”作为状态维度输入模型。
• 关键门店升级实时 API，延迟降至 < 1 分钟。

未来展望：从单商品到全链路

1. 跨 SKU 联合定价：用多智能体 RL 解决“西红柿降价→鸡蛋销量上升”的替代/互补效应。
2. 用户级个性化：引入用户分层（价格敏感、品质敏感），用 Contextual DQN 输出千人千价。
3. 实时竞价：接入广告流量，把“商品坑位”也纳入动作空间，实现“货找人 + 价找人”双引擎。