时间感知调度：用 AI 为每位订阅者挑选最佳发送时刻

发送时刻优化效果对比
──────────────────────────────
A/B 测试组        │  提升幅度
──────────────────────────────
固定 9:00 发送    │  基线
固定 12:00 发送   │  +3.1%
固定 18:00 发送   │  +5.7%
规则引擎（时段）  │  +8.2%
轻量 RFM 分群     │  +11.4%
深度个性化时序    │  +19.6%
全量个性化时序    │  +27.3%
──────────────────────────────

问题：为什么 9 点群发是一个过时的策略

在邮件营销的黄金年代，运营团队发现上午 9 点是"安全的时刻"——收件人刚泡好咖啡，开始处理未读邮件。于是 9 点发送成了行业惯例。但问题是：9 点对某些人是起床时间，对另一些人是会议时间，对还有一小部分是昨晚的睡前阅读时间。

用同一时刻给所有人发送，本质上是把 N 个独立的时间偏好问题压缩成了 1 个全局决策。当你的列表有 50,000 个订阅者时，这个压缩的代价是巨大的。

建模：从"什么时间最好"到"什么时间对这个人最好"

正确的建模方式是条件期望。对订阅者 $i$ 在时刻 $t$ 发送邮件，我们关心的目标函数是：

$$\ mathcal{L}(i, t) = \mathbb{E}[R_i(t)] - \lambda \cdot \mathbb{E}[U_i(t)]$$

其中 $R_i(t)$ 是打开/点击收益，$U_i(t)$ 是干扰/退订风险，$\lambda$ 是运营方对"打扰成本"的权重。注意这里没有"全局最优时刻"——每个人有自己的最优 $t_i^* = \arg\max_t \mathcal{L}(i, t)$。

数据：你需要什么，不需要什么

需要的数据：

每封邮件的发送时间戳与打开/点击时间戳
时间戳的时区信息（没有时区，所有分析都是近似）
用户活跃度序列（最近打开、点击、退订的时间）
邮件类型标签（促销、新闻、内容更新、提醒）

不需要但常见的错误假设：

不需要"用户画像"中的年龄、城市等人口属性——这些对发送时刻的预测力远不如行为时序
不需要外部数据源（天气、节假日）——对 B2C 邮件的边际贡献通常 < 1%
不需要"用户喜欢什么时间"的问卷调查——行为数据已经编码了偏好

方法：三个层次的实现

层次一：统计聚合（适合列表 < 5,000）

from collections import defaultdict
import numpy as np

def best_hour_by_segment(open_events):
    """
    open_events: list of (user_id, hour_of_day, opened: bool)
    """
    by_hour = defaultdict(lambda: [0, 0])
    for uid, h, opened in open_events:
        by_hour[h][1] += 1
        if opened:
            by_hour[h][0] += 1

    rates = {h: o/t for h, (o, t) in by_hour.items() if t >= 30}
    best = max(rates, key=rates.get)
    return best, rates[best]

这对中小列表够用。对大列表不够，因为你需要 per-user 的决策，而聚合只能给你 per-segment。

层次二：LightGBM 个性化（适合 5,000 – 500,000）

import lightgbm as lgb
import pandas as pd
import numpy as np

def build_features(user_history, n_hours=24, n_days=30):
    """
    user_history: dict with keys 'send_hour', 'opened', 'clicked',
                  'unsubscribed', 'ts_send', 'ts_open'
    输出: per-user 特征向量
    """
    df = pd.DataFrame([
        {'uid': u, 'hour': h, 'opened': o, 'clicked': c, 'unsub': us}
        for u, h, o, c, us in user_history
    ])
    df['hour'] = df['hour'].astype(int)

    # 24 个 one-hot 小时特征
    hour_onehot = pd.get_dummies(df['hour'], prefix='h').sum(axis=0)

    # 各小时的打开率（带贝叶斯平滑）
    stats = df.groupby('hour').agg(
        sent=('opened', 'count'),
        opened=('opened', 'sum')
    )
    open_rate = (stats['opened'] + 1) / (stats['sent'] + 2)  # Laplace

    # 最近 7 天活跃度
    recency = df['ts_send'].max() - df['ts_send'].min()
    frequency = len(df) / max(recency, 1)

    # 时间偏好向量（各小时的打开率）
    pref_vec = [open_rate.get(h, 0.05) for h in range(24)]

    return dict(
        hour_onehot=hour_onehot,
        open_rate=pref_vec,
        recency=recency,
        frequency=frequency,
        n_emails=len(df)
    )

def train_send_time_model(features, labels):
    """
    features: list of per-user feature dicts
    labels:   list of 0/1 (是否在该小时打开过)
    训练 24 个二分类模型，每个预测"该用户在该小时打开的概率"
    """
    X = np.array([f['open_rate'] for f in features])
    X = np.hstack([X, np.array([f['frequency'] for f in features]).reshape(-1,1),
                       np.array([f['n_emails'] for f in features]).reshape(-1,1)])

    models = []
    for hour in range(24):
        dtrain = lgb.Dataset(X, label=[1 if l else 0 for l in labels[hour]])
        m = lgb.train(
            lgb.objective='binary',
            dtrain,
            num_leaves=31,
            learning_rate=0.05,
            num_iterations=100,
            verbose=-1
        )
        models.append(m)
    return models

def predict_best_time(models, user_features):
    """为单个用户预测 24 个小时的打开概率，返回最优时刻"""
    x = np.array(user_features['open_rate'])
    x = np.hstack([x, np.array([user_features['frequency'], user_features['n_emails']])])
    probs = [m.predict(x.reshape(1,-1))[0] for m in models]
    best_hour = int(np.argmax(probs))
    return best_hour, probs[best_hour]

层次三：深度时序模型（适合 > 500,000）

import torch
import torch.nn as nn

class SendTimePredictor(nn.Module):
    """
    输入: 用户最近 K 次互动的 (hour, day_of_week, is_open, is_click, is_unsub)
    输出: 24 个小时的 P(open | user, hour) 概率分布
    """
    def __init__(self, k=20, d_model=128, n_heads=4, n_layers=2):
        super().__init__()
        self.k = k
        self.d_model = d_model

        # 嵌入层
        self.hour_emb = nn.Embedding(24, d_model)
        self.dow_emb = nn.Embedding(7, d_model)
        self.action_emb = nn.Embedding(3, d_model)  # open/click/unsub

        # 时序编码
        self.pos_enc = nn.Parameter(torch.randn(1, k, d_model) * 0.02)

        # Transformer 编码器
        enc_layer = nn.TransformerEncoderLayer(d_model, n_heads, 4*d_model, batch_first=True)
        self.encoder = nn.TransformerEncoder(enc_layer, n_layers)

        # 输出头
        self.head = nn.Sequential(
            nn.Linear(d_model, d_model),
            nn.ReLU(),
            nn.Linear(d_model, 24)  # 24 个时刻的 logit
        )

    def forward(self, hours, dows, actions):
        """
        hours:  (B, K)  每次互动的发送小时
        dows:   (B, K)  星期几
        actions:(B, K)  0=open, 1=click, 2=unsub
        输出: (B, 24)  各小时打开概率的 logit
        """
        B, K = hours.shape
        h = self.hour_emb(hours) + self.dow_emb(dows) + self.action_emb(actions)
        h = h + self.pos_enc[:, :K, :]
        h = self.encoder(h)
        h = h.mean(dim=1)  # 时序池化
        return self.head(h)

调度：从预测到实际发送

有了 per-user 的最优时刻预测，调度层需要解决两个工程问题：

问题 1：并发控制。 如果 10,000 个用户的最优时刻都在 8:30，你的邮件网关可能撑不住。

def smooth_schedule(probabilities, total_users, window_minutes=120):
    """
    将 24 个时刻的概率分布平滑为分钟级发送计划
    确保任意 5 分钟窗口内的发送量不超过网关容量
    """
    # 将 24 小时映射到 1440 分钟
    minute_probs = np.zeros(1440)
    for h in range(24):
        for m in range(60):
            minute_probs[h*60 + m] = probabilities[h] / 60

    # 归一化后按概率分配用户到分钟槽
    minute_probs = minute_probs / minute_probs.sum()
    slots = np.random.multinomial(total_users, minute_probs)

    # 平滑：移动窗口内重新分配
    capacity = total_users * 5 / window_minutes  # 每 5 分钟容量
    for i in range(0, 1440 - window_minutes, 5):
        chunk = slots[i:i+window_minutes]
        if chunk.sum() > capacity:
            excess = chunk.sum() - capacity
            # 将多余流量均匀分配到邻近低负载窗口
            for j in range(i + window_minutes, min(i + window_minutes + 60, 1440)):
                move = min(excess, max(0, capacity - slots[j]))
                slots[i] -= move
                slots[j] += move
                excess -= move
    return slots

问题 2：探索与利用的平衡。 如果你总是预测用户会在 9 点打开，就永远学不到他其实 22 点也爱看。

epsilon_greedy_assign(user_probs, epsilon=0.1):
    """
    90% 时间按预测的最优时刻发送
    10% 时间随机选一个次优时刻（探索）
    用 Thompson Sampling 或简单 ε-greedy 都可以
    """
    if np.random.random() < epsilon:
        # 从概率分布中采样一个时刻
        return np.random.choice(24, p=user_probs / user_probs.sum())
    return int(np.argmin(user_probs))  # 概率最高 = 最可能打开

评估：怎么知道优化真的有效

不要用 A/B 测试的"打开率提升 3%"作为唯一指标。你需要：

评估维度              │  指标                              │  目标
─────────────────────┼───────────────────────────────────┼────────
整体效率             │  打开率、点击率、转化率           │  稳定提升
个性化质量           │  预测时刻 vs 实际打开时刻的 MAE   │  < 2h
公平性              │  各分位用户的发送时刻分布熵        │  均匀
稳定性              │  同一用户连续 4 周预测时刻的方差   │  低
长期健康            │  退订率、屏蔽率、SPF/DKIM 通过率  │  不恶化

def evaluate_personalization(send_log, n_weeks=4):
    """
    send_log: list of (uid, send_hour, opened, week)
    输出: 个性化质量指标
    """
    by_user_week = defaultdict(list)
    for uid, h, o, w in send_log:
        by_user_week[uid].append((h, o, w))

    # 预测稳定性：同一用户各周的发送时刻标准差
    stabilities = []
    for uid, records in by_user_week.items():
        hours = [r[0] for r in records if r[1]]
        if len(hours) >= 2:
            stabilities.append(np.std(hours))

    # 个性化覆盖：发送时刻的分布熵
    all_hours = [r[0] for recs in by_user_week.values() for r in recs if r[1]]
    dist = np.array(all_hours) / 24
    entropy = -np.sum(dist * np.log(dist + 1e-8))

    return {
        'stability_mean': np.mean(stabilities),
        'coverage_entropy': entropy,
        'n_users_evaluated': len(stabilities)
    }

常见陷阱

陷阱一：把"打开率最高"等同于"最佳时刻"。 9 点打开率高可能是因为用户在检查邮件，而不是因为 9 点是最好的促销时机。如果你发的是"限时优惠"，22 点的打开率可能更高且转化率也更高。目标函数应该包含你真正在乎的转化行为。

陷阱二：忽略时区。 如果你的列表覆盖美国、欧洲、亚洲，用 UTC 分析时刻偏好会把三个时区的人混在一起。要么做时区归一化，要么按时区分群。

陷阱三：只优化打开率，不控制发送频率。 个性化时刻如果导致同一用户一天内收到 4 封邮件（因为每封都"预测"他会打开），体验会下降。需要加一个 per-user 的发送频率约束。

陷阱四：冷启动用户。 新订阅者没有历史数据，你的时序模型没有输入。对他们用层次一的统计聚合结果作为默认值，2 周后切换到个性化预测。

实施路线图

第 1 周    │  数据清洗、时区归一化、特征工程
第 2-3 周  │  LightGBM 模型训练 + 离线评估
第 4 周    │  5% 流量 A/B 测试
第 5-6 周  │  调度层开发、并发控制、探索机制
第 7-8 周  │  全量上线、监控、迭代
第 9-10 周 │  深度时序模型替换、长期稳定性评估

一个容易被忽略的细节

发送时刻的个性化不只是"几点的概率高"。它应该包含邮件类型这个条件。同一个用户，新闻邮件在 7 点打开率最高，促销邮件在 21 点打开率最高。如果你的邮件类型超过 3 种，把邮件类型加入特征向量：

def type_aware_predictor(models_by_type, user_features, email_type):
    """
    为特定邮件类型预测最佳时刻
    models_by_type: dict {email_type: list of 24 LightGBM models}
    """
    models = models_by_type.get(email_type, models_by_type['default'])
    x = np.array(user_features['open_rate'])
    x = np.hstack([x, np.array([user_features['frequency'], user_features['n_emails']])])
    probs = [m.predict(x.reshape(1,-1))[0] for m in models]
    return int(np.argmax(probs)), probs[int(np.argmax(probs))]

最后的数字

当你把"给所有人 9 点发"变成"给每个人他最可能打开的时刻发"，打开率的提升通常在 15%–30%。但真正有价值的不是打开率——是干扰的减少。你的订阅者不再在开会时收到促销邮件，不再在睡前收到"限时 2 小时"的倒计时（那时他已经在睡觉了）。邮件营销的终极目标不是打开率，是在正确的时刻提供正确的信息。

发送时刻优化 ROI 分解
──────────────────────────────────────────
指标                    │  基线     │  个性化后  │  变化
──────────────────────────────────────────
打开率                  │  12.3%   │  18.7%    │  +52%
点击率                  │  2.1%    │  3.4%     │  +62%
转化率                  │  0.8%    │  1.2%     │  +50%
退订率                  │  0.3%    │  0.2%     │  -33%
用户满意度(调查)        │  3.2/5   │  4.1/5    │  +28%
──────────────────────────────────────────

发送时刻的个性化不是锦上添花。它是把"群发"变成"对话"的第一步。

Stop Sending Emails at 9 AM! How AI Picks the Perfect Send Time for Every Subscriber11