Build an explainable next-best-product recommendation system for banking on AWS

import gc
from concurrent.futures import ProcessPoolExecutor

chunksize = 5_000_000
n_workers = 4

for batch_start in range(0, total_chunks, n_workers):
    with ProcessPoolExecutor(max_workers=n_workers) as executor:
        futures = [
            executor.submit(process_chunk_range, input_path, output_path, i, start_row, end_row)
            for i in range(batch_start, min(batch_start + n_workers, total_chunks))
        ]
        for future in futures:
            future.result()
    gc.collect()  # 批次间强制垃圾回收

class SequenceTower(nn.Module):
    def __init__(self, num_products, embedding_dim=32, hidden_dim=64, dropout=0.2):
        super().__init__()
        self.embedding = nn.Embedding(num_products + 1, embedding_dim, padding_idx=0)
        self.gru = nn.GRU(
            input_size=embedding_dim, hidden_size=hidden_dim,
            num_layers=2, batch_first=True, dropout=dropout
        )
        self.active_count_layer = nn.Sequential(
            nn.Linear(1, hidden_dim // 2), nn.ReLU(), nn.Dropout(dropout)
        )
        self.fusion = nn.Sequential(
            nn.Linear(hidden_dim + hidden_dim // 2, hidden_dim),
            nn.ReLU(), nn.Dropout(dropout)
        )

    def forward(self, sequence, seq_length, active_count):
        embedded = self.embedding(sequence)
        packed = nn.utils.rnn.pack_padded_sequence(
            embedded, seq_length.cpu().clamp(min=1),
            batch_first=True, enforce_sorted=False
        )
        _, hidden = self.gru(packed)
        seq_features = hidden[-1]
        active_features = self.active_count_layer(active_count)
        return self.fusion(torch.cat([seq_features, active_features], dim=1))

class TowerAttentionMechanism(nn.Module):
    def __init__(self, hidden_dim=64, num_heads=4, dropout=0.1):
        super().__init__()
        self.tower_attention = nn.MultiheadAttention(
            embed_dim=hidden_dim, num_heads=num_heads,
            dropout=dropout, batch_first=True
        )
        self.context_weighting = nn.Sequential(
            nn.Linear(hidden_dim * 4, 4), nn.Softmax(dim=1)
        )

    def forward(self, tower_outputs):
        stacked = torch.stack(tower_outputs, dim=1)  # [batch, 4, 64]
        attended, _ = self.tower_attention(stacked, stacked, stacked)
        stacked = stacked + attended  # 残差连接
        concat = torch.cat(tower_outputs, dim=1)  # [batch, 256]
        tower_weights = self.context_weighting(concat)  # [batch, 4]
        weighted_outputs = [
            tower_outputs[i] * tower_weights[:, i:i+1]
            for i in range(4)
        ]
        return weighted_outputs, tower_weights

class ContextAwareFusion(nn.Module):
    def __init__(self, hidden_dim=64, dropout=0.2):
        super().__init__()
        self.initial_projection = nn.Linear(hidden_dim * 4, hidden_dim)
        self.fusion1 = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim * 2), nn.LayerNorm(hidden_dim * 2),
            nn.ReLU(), nn.Dropout(dropout), nn.Linear(hidden_dim * 2, hidden_dim)
        )
        self.layer_norm1 = nn.LayerNorm(hidden_dim)
        self.fusion2 = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Dropout(dropout)
        )
        self.layer_norm2 = nn.LayerNorm(hidden_dim)

    def forward(self, weighted_outputs):
        concat = torch.cat(weighted_outputs, dim=1)
        projected = self.initial_projection(concat)
        out1 = self.layer_norm1(projected + self.fusion1(projected))  # Residual
        out2 = self.layer_norm2(out1 + self.fusion2(out1))  # Residual
        return out2

class FeatureImportanceModule(nn.Module):
    def __init__(self, hidden_dim=64):
        super().__init__()
        self.feature_contribution = nn.Sequential(
            nn.Linear(hidden_dim, 4), nn.Softmax(dim=1)
        )

    def forward(self, fused_features, tower_weights):
        feature_importance = self.feature_contribution(fused_features)
        return feature_importance * tower_weights

from sagemaker.pytorch import PyTorch

## 评估指标

模型使用与业务价值直接对应的指标进行评估：

| 指标         | 测量内容             | 业务相关性                     |
|--------------|----------------------|------------------------------|
| Top-1准确率  | 精确预测准确率       | “模型是否预测了完全正确的商品？” |
| Top-3准确率  | 正确商品在前3名中    | “正确的商品是否在客户经理的短名单中？” |
| Top-5准确率  | 正确商品在前5名中    | “它是否出现在推荐轮播中？”       |
| MRR（平均倒数排名） | 正确商品的平均倒数排名 | “正确商品平均出现在多高位置？”   |
| 加权F1       | 每类精确率/召回率平衡 | “模型是否能良好预测所有商品类型？” |

生产模型在所有指标上均表现出色，正确商品始终出现在前3推荐中。特征重要性模块确认序列塔（商品采用历史）贡献最大信号，其次是交易模式、客户人口统计和行为细分。

## 推理与部署

推理流程支持使用SageMaker AI进行批量评分和实时预测。

对于批量评分，SageMaker AI批量转换功能每晚处理整个客户群，为每位客户生成带可解释性评分的Top-k推荐。结果以JSON格式存储在Amazon S3中，供CRM系统和客户经理仪表板使用。

对于实时预测，SageMaker AI实时端点在客户登录移动银行应用或客户经理打开客户档案时提供按需推荐。

每条推荐包含：

- 商品ID和概率评分。
- 特征重要性分解：每个塔的百分比贡献。
- 置信度指标：基于概率分布熵。

关于使用IAM认证、限速和虚拟私有云（VPC）隔离保护端点的指导，请参见安全考虑部分。

注意：本文中的代码片段仅用于说明架构模式，不适用于生产环境。生产环境需添加输入验证（张量形状、NaN检查、序列长度边界等）、错误处理和推理日志记录。配置Amazon SageMaker Model Monitor以监控输入分布漂移。

## 关键设计决策

- 为什么选择多塔结构而不是单一网络？单一网络需要同时学习处理序列、聚合交易、编码人口统计和解释行为细分。独立的塔可以让每个模块专注于其数据类型，然后基于注意力的融合层学习如何按客户最优地组合它们。

- 为什么在生产环境中选择GRU而不是Transformer？Transformer在处理长序列（100+）时表现优异。对于20个条目或更少的序列，GRU已经足够，相比Transformer的注意力图提供了更清晰的可解释性，避免了二次方注意力计算，并生成更小的模型（约5MB，而Transformer约为15MB）。

- 为什么使用学习的塔权重而不是拼接？使用拼接时，模型对所有客户的所有塔都一视同仁。而使用学习的注意力权重时，模型会根据客户进行自适应：交易历史丰富的客户会获得更高的交易塔权重，而新客户则会获得更高的人口统计学塔权重。

- 为什么使用时间窗口交易特征？不同时间窗口能捕捉不同信号：7天窗口捕捉即时意图，30天窗口捕捉月度模式，180天窗口捕捉季节性模式，365天窗口捕捉年度模式。过去7天突然增加交易频率的客户，与过去365天保持稳定活动的客户意图存在显著差异。

## 操作注意事项

PyTorch、NumPy和CUDA的所有随机种子均固定，以实现确定性训练。模型组件、超参数和数据版本通过Amazon SageMaker Experiments进行跟踪。

Amazon SageMaker Model Monitor可检测数据漂移（输入特征分布变化）、模型漂移（预测质量下降）和潜在偏见（过度依赖人口统计学特征）。

Amazon SageMaker Pipelines工作流每月使用最新客户数据重新训练模型。该流程自动化完整工作流：数据处理、训练、评估和条件部署（仅在指标优于当前生产模型时部署）。

## 安全注意事项

- 部署此解决方案时，若使用真实银行数据，请实施最小权限IAM角色，仅限于必需的SageMaker AI、Amazon S3、AWS Glue和CloudWatch操作。

- 使用AWS KMS客户管理密钥对S3存储桶和SageMaker AI训练卷中的静态数据进行加密，并通过S3存储桶策略强制实施传输中数据的TLS加密。

- 在无互联网网关的私有VPC子网中部署训练作业和端点，使用VPC端点（AWS PrivateLink）进行AWS服务通信，并在训练作业上设置enable_network_isolation=True。

- 使用AWS Signature Version 4对推断端点进行签名，考虑使用Amazon Cognito和Amazon API Gateway进行消费者身份验证和速率限制。

- 在数据治理方面，评估监管义务（PCI-DSS、GDPR、CCPA），实施数据最小化，并定义保留策略。

- 启用AWS CloudTrail进行API审计日志记录，使用S3版本控制确保模型组件完整性。

如需完整实现指南，请参阅SageMaker AI安全文档。

## 清理资源

为避免持续产生AWS费用，在完成此解决方案评估后，请删除以下资源：

- SageMaker AI端点和端点配置。

- SageMaker AI模型和Amazon S3中的训练作业输出。

- Amazon SageMaker Processing作业输出在Amazon S3中的内容。

- SageMaker AI批量转换作业和配置。

- Amazon SageMaker Pipelines定义。

- Amazon SageMaker Model Monitor计划。

- Amazon SageMaker Experiments试验和实验数据。

- 包含训练数据、模型组件和批量转换结果的Amazon S3存储桶。

- AWS Glue ETL作业和Data Catalog资源。

- 由SageMaker AI和AWS Glue创建的CloudWatch日志组

您可以通过AWS管理控制台或使用AWS命令行界面（AWS CLI）删除这些资源。

警告：删除这些资源是不可逆的操作。在继续操作前：

- 导出需要保留的模型构件或评估结果
- 确认删除操作符合您的数据保留政策和监管要求
- 如果启用了版本控制，请删除所有S3对象版本
- 移除为该解决方案创建的IAM角色和策略
- 删除S3存储桶将永久移除所有训练数据、模型构件和批量转换结果

## 结论

本文演示了如何使用PyTorch和SageMaker AI为银行业构建Next-Best-Product推荐系统。该系统采用带学习注意力融合的多塔架构，在实现高预测准确率的同时满足银行监管机构对可解释性的要求。

关键要点包括：

- 针对不同数据类型的专用塔架构在处理异构银行数据时优于单一架构
- 基于GRU的序列处理能够捕捉平面特征聚合无法发现的时间序列产品采纳模式
- 学习塔注意力机制在提升准确率的同时实现客户级可解释性，无需依赖事后解释方法
- SageMaker AI提供了从训练、模型管理到规模化推理的端到端基础设施

您可以将这种多塔架构适配到自己的产品目录和客户数据中。要开始使用，请查阅SageMaker AI文档。如需更多AWS上PyTorch的示例，请参阅PyTorch on AWS。如需帮助构建推荐系统，请联系AWS代表。

## 作者介绍

'"
 /