2026-07-26 · 风控实战
时间窗口特征设计风控实战
特征工程：风控领域的时间窗口特征设计
时间窗口特征是信贷风控模型中最重要的特征类型之一。一个借款人的历史行为——近 7 天申请了几次、近 30 天登录了几个设备、近 90 天逾期了几次——这些窗口特征的区分能力，往往超过静态特征（年龄、收入、学历）好几个量级。但窗口特征的设计也是一把双刃剑：窗口选错了，要么信息不足，要么引入噪声；更致命的是，窗口特征的计算如果不做"时间穿越防护"，离线验证的 AUC 0.85 上线后直接腰斩。
这篇文章从四个维度讲时间窗口特征：窗口设计原则、常用窗口模式、防时间穿越的工程实现、以及特征存储与回填。
一、窗口设计的三个核心维度
设计一个窗口特征，本质上是在回答三个问题：
1.1 时间跨度（窗口长度）
| 窗口长度 | 典型用途 | 数据量 | 注意事项 |
|---|---|---|---|
| 1-7 天 | 短期行为密集度（申请频率、登录次数） | 少 | 对近期异常敏感，适合反欺诈 |
| 7-30 天 | 中期行为模式（设备切换、联系人变更） | 中 | 最常用的窗口，兼顾时效性和稳定性 |
| 30-90 天 | 中期信用表现（还款行为、额度使用率） | 中-多 | 信用评估的核心窗口 |
| 90-365 天 | 长期行为趋势（历史逾期、多头借贷） | 多 | 对历史数据完整性要求高 |
| 全量 | 累计统计（总借款次数、历史最大逾期天数） | 全部 | 兜底特征，不受窗口选择影响 |
选窗口长度有一个核心原则：窗口要覆盖"业务上认为有意义"的时间段。比如反欺诈场景，欺诈团伙通常在 1-3 天内集中作案，所以窗口设 7 天足够。信用评估场景，借款人的还款行为以月为周期，30-90 天更合理。
1.2 统计粒度（聚合函数）
同一个窗口，不同的聚合方式产出完全不同的信息：
import pandas as pd
import numpy as np
# 假设 df 是某用户在窗口内的行为明细表
# 列：user_id, event_time, amount, event_type, device_id
def build_window_features(df, window_col='event_time', base_col='user_id'):
"""对行为明细做多粒度聚合"""
agg = df.groupby(base_col).agg(
# 计数类
cnt_total=('amount', 'count'), # 总次数
cnt_days=('event_time', 'nunique'), # 活跃天数
# 数值统计类
amt_sum=('amount', 'sum'), # 总金额
amt_mean=('amount', 'mean'), # 平均金额
amt_std=('amount', 'std'), # 金额波动
amt_max=('amount', 'max'), # 最大单笔
amt_min=('amount', 'min'), # 最小单笔
# 去重计数
device_cnt=('device_id', 'nunique'), # 使用设备数
# 时间间隔
last_gap_days=('event_time', lambda x:
(pd.Timestamp.now() - x.max()).days) # 距上次行为天数
).reset_index()
# 衍生比率特征
agg['amt_per_day'] = agg['amt_sum'] / agg['cnt_days'].replace(0, 1)
agg['avg_gap_hours'] = 24 * agg['cnt_days'] / agg['cnt_total'].replace(0, 1)
return agg
1.3 观察点（Reference Point）
窗口的"截止时间"选在哪里，直接影响特征是否发生时间穿越。常见的观察点选择：
- 以申请时间为观察点：最安全，所有特征只看申请时刻之前的窗口。这是标准做法。
- 以当前时间为观察点：在线推理时使用，等于申请时间。离线训练时必须回退到每条样本对应的申请时间。
- 滑动窗口：每天/每小时产出一版特征快照，推理时取最新快照。适合实时风控。
二、六种高频窗口特征模式
以下六种模式覆盖了风控特征工程中 80% 的窗口特征场景。
模式 1：申请频率特征
"借款人在过去 N 天内申请了多少次？"——这是反欺诈和多头借贷的核心特征。
-- 近 7/30/90 天申请次数
SELECT
a.user_id,
a.apply_date,
COUNT(CASE WHEN b.apply_date BETWEEN a.apply_date - 7 AND a.apply_date - 1
THEN 1 END) AS apply_cnt_7d,
COUNT(CASE WHEN b.apply_date BETWEEN a.apply_date - 30 AND a.apply_date - 1
THEN 1 END) AS apply_cnt_30d,
COUNT(CASE WHEN b.apply_date BETWEEN a.apply_date - 90 AND a.apply_date - 1
THEN 1 END) AS apply_cnt_90d
FROM apply_record a
LEFT JOIN apply_record b
ON a.user_id = b.user_id
AND b.apply_date < a.apply_date
GROUP BY a.user_id, a.apply_date;
注意 SQL 中的 b.apply_date < a.apply_date——窗口必须截止到当前申请的前一天，不能包含当天。包含当天就是时间穿越（当天后续的申请不应该用来预测当天的申请）。
模式 2：逾期滚动统计
"过去 N 天内有几次逾期？最大逾期天数是多少？"
def build_overdue_features(repay_df, windows=[7, 30, 90, 180]):
"""
repay_df: 还款记录表，包含 user_id, due_date, actual_date, apply_date
以每笔借款的申请时间为观察点，统计历史逾期
"""
features = []
for _, row in repay_df.iterrows():
uid = row['user_id']
ref_date = row['apply_date'] # 观察点 = 本次申请时间
history = repay_df[
(repay_df['user_id'] == uid) &
(repay_df['due_date'] < ref_date) # 只看申请前的还款记录
]
feat = {'user_id': uid, 'apply_date': ref_date}
for w in windows:
window_data = history[history['due_date'] >= ref_date - pd.Timedelta(days=w)]
feat[f'overdue_cnt_{w}d'] = (window_data['actual_date'] >
window_data['due_date']).sum()
feat[f'overdue_days_max_{w}d'] = (
window_data['actual_date'] - window_data['due_date']
).dt.days.max()
feat[f'overdue_days_max_{w}d'] = (
feat[f'overdue_days_max_{w}d']
if pd.notna(feat[f'overdue_days_max_{w}d']) else 0
)
features.append(feat)
return pd.DataFrame(features)
模式 3：设备/环境变更特征
反欺诈领域最重要的特征之一：用户是否频繁切换设备、IP、地理位置。
def build_device_features(login_df, windows=[7, 30]):
"""
login_df: 登录日志，含 user_id, login_time, device_id, ip_city
"""
features = login_df.groupby('user_id').apply(
lambda g: pd.Series({
**{f'device_cnt_{w}d': (
g[g['login_time'] >= g['login_time'].max() - pd.Timedelta(days=w)]
['device_id'].nunique()
) for w in windows},
**{f'city_cnt_{w}d': (
g[g['login_time'] >= g['login_time'].max() - pd.Timedelta(days=w)]
['ip_city'].nunique()
) for w in windows},
'device_cnt_total': g['device_id'].nunique(),
'city_cnt_total': g['ip_city'].nunique()
})
).reset_index()
return features
模式 4：行为趋势特征（斜率）
不只是看"过去 N 天怎么样"，还要看"趋势是在变好还是变坏"。
from scipy import stats
def build_trend_features(time_series_df):
"""
time_series_df: 用户某行为的时间序列，含 date, value
用线性回归斜率判断趋势方向
"""
x = np.arange(len(time_series_df))
y = time_series_df['value'].values
if len(x) < 2:
return {'trend_slope': 0, 'trend_r2': 0}
slope, intercept, r_value, _, _ = stats.linregress(x, y)
return {
'trend_slope': slope, # 正=上升趋势, 负=下降趋势
'trend_r2': r_value ** 2, # 拟合优度, 接近 1 说明趋势显著
}
模式 5：首次/最近一次的时间间隔
def build_recency_features(df, user_col, time_col):
"""
距离首次行为和最近一次行为的天数
"""
agg = df.groupby(user_col).agg(
first_date=(time_col, 'min'),
last_date=(time_col, 'max')
)
ref_date = pd.Timestamp.now()
agg['days_since_first'] = (ref_date - agg['first_date']).dt.days
agg['days_since_last'] = (ref_date - agg['last_date']).dt.days
agg['active_span_days'] = (agg['last_date'] - agg['first_date']).dt.days
return agg.reset_index()
模式 6：窗口间对比（环比/同比）
def build_window_compare_features(feature_df):
"""
对比不同窗口的同一指标，检测行为变化幅度
"""
feature_df['apply_cnt_change_7d_vs_30d'] = (
feature_df['apply_cnt_7d'] /
feature_df['apply_cnt_30d'].replace(0, 1)
)
feature_df['device_cnt_change_7d_vs_30d'] = (
feature_df['device_cnt_7d'] /
feature_df['device_cnt_30d'].replace(0, 1)
)
# >1 表示近期更活跃（可能在集中作案），<1 表示近期收敛
return feature_df
三、防时间穿越的工程实现
时间穿越是窗口特征的头号杀手。核心原则就一句话：训练时，每条样本只能使用该样本"观察时间点之前"的数据。
3.1 离线训练的正确姿势
def build_window_features_safe(behavior_df, sample_df, windows=[7, 30, 90]):
"""
防时间穿越的窗口特征构建
behavior_df: 全量行为明细，含 user_id, event_time, ...
sample_df: 训练样本，含 user_id, sample_time（每条的观察时间）
"""
results = []
for _, sample in sample_df.iterrows():
uid = sample['user_id']
ref_time = sample['sample_time'] # 本样本的观察时间点
# 关键：只看 ref_time 之前的行为
history = behavior_df[
(behavior_df['user_id'] == uid) &
(behavior_df['event_time'] < ref_time)
]
feat = {'user_id': uid, 'sample_time': ref_time}
for w in windows:
window_data = history[
history['event_time'] >= ref_time - pd.Timedelta(days=w)
]
feat[f'event_cnt_{w}d'] = len(window_data)
feat[f'event_days_{w}d'] = window_data['event_time'].nunique()
results.append(feat)
return pd.DataFrame(results)
3.2 离线训练和在线推理的一致性
这是另一个容易出问题的点。离线训练用历史数据按样本时间回退计算，在线推理用实时数据按当前时间计算——两者逻辑要完全一致。
Java 侧的在线特征计算骨架：
public class WindowFeatureCalculator {
/**
* 在线计算某用户某窗口内的特征
* @param userId 用户 ID
* @param refTime 观察时间点（申请时间）
* @param windowDays 窗口天数
* @param featureStore 特征存储服务
*/
public Map<String, Double> calculate(
String userId,
LocalDateTime refTime,
int windowDays,
FeatureStore featureStore) {
LocalDateTime windowStart = refTime.minusDays(windowDays);
// 从特征存储中查询窗口内的行为明细
List<BehaviorEvent> events = featureStore.queryEvents(
userId, windowStart, refTime // [windowStart, refTime)
);
Map<String, Double> features = new LinkedHashMap<>();
// 计数类
features.put("event_cnt_" + windowDays + "d",
(double) events.size());
// 去重天数
long distinctDays = events.stream()
.map(e -> e.getEventTime().toLocalDate())
.distinct()
.count();
features.put("active_days_" + windowDays + "d",
(double) distinctDays);
// 金额统计（仅统计有金额的事件）
DoubleSummaryStatistics amtStats = events.stream()
.filter(e -> e.getAmount() != null)
.mapToDouble(BehaviorEvent::getAmount)
.summaryStatistics();
if (amtStats.getCount() > 0) {
features.put("amt_sum_" + windowDays + "d", amtStats.getSum());
features.put("amt_avg_" + windowDays + "d", amtStats.getAverage());
features.put("amt_max_" + windowDays + "d", amtStats.getMax());
} else {
features.put("amt_sum_" + windowDays + "d", 0.0);
features.put("amt_avg_" + windowDays + "d", 0.0);
features.put("amt_max_" + windowDays + "d", 0.0);
}
return features;
}
}
3.3 自检清单
每次上线一批窗口特征之前，跑以下三个验证：
- 同一用户同一申请时间的特征值，训练环境和推理环境计算结果是否一致？——随机抽 100 条对比，偏差 > 0 就是 bug。
- 时间穿越检查：对每个窗口特征，用特征重要性或 IV 排个序，如果某个窗口特征的 IV 异常高（> 1.0），大概率是标签泄漏或时间穿越。
- 窗口边界检查：取
ref_time - 1 秒和ref_time + 1 秒两个时间点的特征值，前者应该稳定，后者应该不等于前者（因为前者多了一个边界事件）。如果两者一致，说明窗口的闭区间/开区间处理有问题。
四、特征存储与回填
窗口特征的计算依赖历史行为明细，这些明细数据通常存储在 OLAP 引擎中（ClickHouse、Doris 等）。离线训练时可以直接从 OLAP 查，但在线推理时不能每次实时去 OLAP 做聚合——延迟太高。
标准做法是"离线预计算 + 在线实时读取"：
┌──────────────┐ ┌──────────────────┐ ┌──────────────┐
│ 行为明细数据 │ ──► │ 离线特征计算任务 │ ──► │ 特征存储 │
│ (OLAP) │ │ (Spark/Flink 批) │ │ (Redis/HBase) │
└──────────────┘ └──────────────────┘ └──────┬───────┘
│
┌──────▼───────┐
│ 在线推理服务 │
│ 直接读取特征 │
└──────────────┘
特征回填（给历史样本补充窗口特征）也是一样的逻辑：用样本的申请时间作为观察点，从 OLAP 按时间条件查询窗口内的明细，做聚合后写入样本表。这个过程可以用 Spark 批量跑，核心就是把上面的 Python 单条循环改成 Spark 的窗口函数或 broadcast join。
窗口特征设计是特征工程里"简单但不容易"的一类——概念简单，但窗口选多大、聚合用什么函数、怎么防时间穿越、怎么保证训练推理一致性，每个环节都有坑。把这几个坑踩明白了，窗口特征的威力才能完全释放。