使用 Amazon SageMaker 获得最大利润的最优定价

这是 Adspert 的高级机器学习工程师 Viktor Enrico Jeney 的客座文章。

广告专家 是一家位于柏林的 ISV，开发了一种投标管理工具，旨在自动优化绩效营销和广告活动。 该公司的核心原则是借助人工智能实现电子商务广告利润的自动化最大化。 广告平台的不断发展为新机会铺平了道路，Adspert 熟练地利用这些机会帮助客户取得成功。

Adspert 的主要目标是为用户简化流程，同时优化跨平台的广告活动。 这包括使用在各个平台上收集的信息与在每个平台上一级设置的最佳预算相平衡。 Adspert 的重点是优化客户的目标达成，无论使用什么平台。 Adspert 继续根据需要添加平台，为我们的客户提供显着优势。

在这篇文章中，我们分享了 Adspert 如何使用不同的 AWS 服务从头开始创建定价工具，例如 亚马逊SageMaker 以及 Adspert 如何与 AWS数据实验室 加速该项目从设计到建造的创纪录时间。

定价工具根据可见性和利润率对电子商务市场上卖家选择的产品重新定价，以最大限度地提高产品级别的利润。

作为卖家，您的产品必须始终可见，因为这会增加销售额。 电子商务销售中最重要的因素就是您的报价是否对客户可见，而不是竞争对手的报价。

虽然它肯定取决于特定的电子商务平台，但我们发现产品价格是影响知名度的最重要的关键数据之一。 然而，价格变化频繁且迅速； 出于这个原因，定价工具需要近乎实时地采取行动以提高知名度。

解决方案概述

下图说明了解决方案体系结构。

该解决方案包含以下组件：

1. 适用于 PostgreSQL 的 Amazon 关系数据库服务 (Amazon RDS) 是主要数据源，包含存储在 RDS for Postgres 数据库中的产品信息。
2. 产品列表更改信息实时到达 Amazon Simple Queue服务 （Amazon SQS）队列。
3. 存储在 Amazon RDS 中的产品信息使用可在 AWS 数据库迁移服务 (AWS DMS)。
4. 来自 Amazon SQS 的产品列表通知使用 AWS Lambda 功能。
5. 原始源数据存储在 亚马逊简单存储服务 (Amazon S3) 使用 Parquet 数据格式的原始层存储桶。 该层是数据湖的唯一真实来源。 此存储上使用的分区支持数据的增量处理。
6. AWS胶水 提取、转换和加载 (ETL) 作业清理产品数据，删除重复数据，并应用与特定业务案例无关的数据整合和通用转换。
7. Amazon S3 阶段层接收准备好的数据，这些数据以 Apache Parquet 格式存储以供进一步处理。 阶段存储上使用的分区支持数据的增量处理。
8. 在此层中创建的 AWS Glue 作业使用 Amazon S3 阶段层中可用的数据。 这包括应用特定于用例的业务规则和所需的计算。 这些作业的结果数据存储在 Amazon S3 分析层中。
9. Amazon S3 分析层用于存储 ML 模型用于训练目的的数据。 策划存储上使用的分区基于预期的数据使用情况。 这可能与舞台层上使用的分区不同。
10. 重新定价 ML 模型是 SageMaker 脚本模式下的 Scikit-Learn 随机森林实施，它使用 S3 存储桶（分析层）中可用的数据进行训练。
11. AWS Glue 数据处理作业为实时推理准备数据。 该作业处理在 S3 存储桶（阶段层）中摄取的数据并调用 SageMaker 推理端点。 数据已准备好供 SageMaker 重新定价模型使用。 AWS Glue 比 Lambda 更受欢迎，因为推理需要不同的复杂数据处理操作，例如对大量数据（数十亿的日常事务）进行连接和窗口函数。 重新定价模型调用的结果存储在 S3 存储桶（推理层）中。
12. SageMaker 训练作业是使用 SageMaker 端点部署的。 该终端节点由 AWS Glue 推理处理器调用，生成近乎实时的价格建议以提高产品知名度。
13. SageMaker 推理终端节点生成的预测存储在 Amazon S3 推理层中。
14. Lambda 预测优化器函数处理 SageMaker 推理端点生成的建议，并生成一个新的价格建议，该建议侧重于最大化卖家利润，在销量和销售利润率之间进行权衡。
15. Lambda 预测优化器生成的价格建议被提交给重新定价 API，该 API 会更新市场上的产品价格。
16. Lambda 预测优化器生成的更新价格建议存储在 Amazon S3 优化层中。
17. AWS Glue 预测加载器作业将 ML 模型生成的预测重新加载到源 RDS for Postgres SQL 数据库中，以用于审计和报告目的。 AWS Glue Studio 用于实现该组件； 它是一个图形界面，可让您轻松在 AWS Glue 中创建、运行和监控 ETL 作业。

资料准备

Adspert 可见性模型的数据集是从 SQS 队列创建的，并使用 Lambda 实时摄取到我们数据湖的原始层中。 之后，通过执行简单的转换（例如删除重复项）对原始数据进行清理。 此过程在 AWS Glue 中实施。 结果存储在我们数据湖的暂存层中。 通知为给定产品的竞争对手提供价格、履行渠道、运输时间和更多变量。 它们还提供依赖于平台的可见性度量，可以表示为布尔变量（可见或不可见）。 每当优惠发生变化时，我们都会收到通知，这会在我们所有客户的产品中每月增加数百万个事件。

从这个数据集中，我们提取训练数据如下：对于每个通知，我们将可见的报价与每个不可见的报价配对，反之亦然。 每个数据点都代表两个卖家之间的竞争，其中有明显的赢家和输家。 此处理作业是在使用 Spark 的 AWS Glue 作业中实现的。 准备好的训练数据集被推送到分析 S3 存储桶以供 SageMaker 使用。

训练模型

如果给定的报价将是可见的，我们的模型会为每对报价进行分类。 该模型使我们能够为客户计算最优惠的价格，根据竞争提高知名度，并最大限度地提高他们的利润。 最重要的是，这种分类模型可以让我们更深入地了解我们的列表可见或不可见的原因。 我们使用以下功能：

• 我们的价格与竞争对手价格的比率
• 履行渠道的差异
• 每个卖家的反馈量
• 每个卖家的反馈评分
• 最短运输时间的差异
• 最长运输时间的差异
• 每个卖家产品的可用性

Adspert 使用 SageMaker 训练和托管模型。 我们使用 Scikit-Learn 随机森林实现 SageMaker 脚本模式. 我们还在训练脚本的 Scikit-Learn 管道中直接包含一些特征预处理。 请参阅以下代码：

import numpy as np

def transform_price(X):
    X = X.to_numpy()
    return np.log(
        X[:, 0] / np.nanmin([X[:, 1], X[:, 2]], axis=0),
    ).reshape(-1, 1)

def difference(X):
    X = X.to_numpy()
    return (X[:, 0] - X[:, 1]).reshape(-1, 1)

def fulfillment_difference(X):
    X = X.astype(int)
    return difference(X)

最重要的预处理功能之一是 transform_price，它将价格除以竞争对手价格和外部价格列的最小值。 我们发现这个特征对模型的准确性有相关的影响。 我们还应用对数让模型根据相对价格差异而不是绝对价格差异做出决定。

在 training_script.py 脚本，我们首先定义如何构建 Scikit-Learn ColumnTransformer 将指定的转换器应用于数据框的列：

import argparse
import os
from io import StringIO

import joblib
import numpy as np
import pandas as pd
from custom_transformers import difference
from custom_transformers import fulfillment_difference
from custom_transformers import transform_price
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import RandomForestClassifier
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import FunctionTransformer
from sklearn.preprocessing import OneHotEncoder

def make_preprocessor():
    return ColumnTransformer([
        ('price_by_smallest_cp', FunctionTransformer(transform_price),
         ['price', 'competitor_price', 'external_price']),
        (fulfillment_difference, FunctionTransformer(fulfillment_difference),
         ['fulfillment', 'competitor_'fulfillment']),
        ('feedback_count', 'passthrough',
         ['feedback_count', 'competitor_feedback_count']),
        ('feedback_rating', 'passthrough',
         ['feedback_rating', 'competitor_feedback_rating']),
        (
            'availability_type',
            OneHotEncoder(categories=[['NOW'], ['NOW']],
                          handle_unknown='ignore'),
            ['availability_type', 'competitor_availability_type'],
        ),
        ('min_shipping', FunctionTransformer(difference),
         ['minimum_shipping_hours', 'competitor_min_shipping_hours']),
        ('max_shipping', FunctionTransformer(difference),
         ['maximum_shipping_hours', 'competitor_max_shipping_hours']),
    ], remainder='drop')

在训练脚本中，我们将 Parquet 中的数据加载到 Pandas 数据框中，定义 ColumnTranformer 和 RandomForestClassifier，并训练模型。 之后，模型使用序列化 joblib:

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--output-data-dir', type=str,
                        default=os.environ['SM_OUTPUT_DATA_DIR'])
    parser.add_argument('--model-dir', type=str,
                        default=os.environ['SM_MODEL_DIR'])
    parser.add_argument('--train', type=str,
                        default=os.environ['SM_CHANNEL_TRAIN'])

    args = parser.parse_args()

    # load training data
    input_files = [os.path.join(args.train, file)
                   for file in os.listdir(args.train)]
    if len(input_files) == 0:
        raise ValueError
    raw_data = [pd.read_parquet(file) for file in input_files]
    train_data = pd.concat(raw_data)

    # split data set into x and y values
    train_y = train_data.loc[:, 'is_visible']

    if train_y.dtype != 'bool':
        raise ValueError(f'Label 'is_visible' has to be dtype bool but is'
                         f' {train_y.dtype}')

    train_X = train_data.drop('is_visible', axis=1)

    # fit the classifier pipeline and store the fitted model
    clf = Pipeline([
        ('preprocessor', make_preprocessor()),
        ('classifier', RandomForestClassifier(random_state=1)),
    ])
    clf.fit(train_X, train_y)
    joblib.dump(clf, os.path.join(args.model_dir, 'model.joblib'))

在训练脚本中，我们还必须实现推理功能：

• 输入_fn – 负责从payload的请求体中解析数据
• 模型_fn – 加载并返回在脚本的训练部分中转储的模型
• 预测函数 – 包含我们的实现，以使用来自有效负载的数据从模型请求预测
• 预测概率 – 为了绘制预测的可见性曲线，我们使用 predict_proba 函数，而不是分类器的二元预测

请参见以下代码：

def input_fn(request_body, request_content_type):
    """Parse input data payload"""
    if request_content_type == 'text/csv':
        df = pd.read_csv(StringIO(request_body))
        return df
    else:
        raise ValueError(f'{request_content_type} not supported by script!')


def predict_fn(input_data, model):
    """Predict the visibilities"""
    classes = model.classes_

    if len(classes) != 2:
        raise ValueError('Model has more than 2 classes!')

    # get the index of the winning class
    class_index = np.where(model.classes_ == 1)[0][0]

    output = model.predict_proba(input_data)
    return output[:, class_index]


def model_fn(model_dir):
    """Deserialized and return fitted model

    Note that this should have the same name as the serialized model in the
    main method
    """
    clf = joblib.load(os.path.join(model_dir, 'model.joblib'))
    return clf

下图显示了基于杂质的特征重要性 随机森林分类器.

借助 SageMaker，我们能够在大量数据（每天多达 14 亿次交易）上训练模型，而无需在现有实例上增加负载，也无需设置具有足够资源的单独机器。 此外，由于实例在训练作业后立即关闭，因此使用 SageMaker 进行训练极具成本效益。 使用 SageMaker 进行模型部署无需任何额外工作负载。 Python SDK 中的单个函数调用足以将我们的模型作为推理端点托管，并且也可以使用 SageMaker Python SDK 从其他服务轻松请求。 请参阅以下代码：

from sagemaker.sklearn.estimator import SKLearn

FRAMEWORK_VERSION = "0.23-1"
script_path = 'training_script.py'
output_location = f's3://{bucket}/{folder}/output'
source_dir = 'source_dir'

sklearn = SKLearn(
    entry_point=script_path,
    source_dir=source_dir,
    framework_version=FRAMEWORK_VERSION,
    instance_type='ml.m5.large',
    role=role,
    sagemaker_session=sagemaker_session,
    output_path=output_location)

sklearn.fit({'train': training_path})

模型人工制品通过 fit 函数存储在 Amazon S3 中。 如以下代码所示，模型可以加载为 SKLearnModel 使用模型工件、脚本路径和一些其他参数的对象。 之后，可以将其部署到所需的实例类型和实例数量。

model = sagemaker.sklearn.model.SKLearnModel(
    model_data=f'{output_location}/sagemaker-scikit-learn-2021-02-23-11-13-30-036/output/model.tar.gz',
    source_dir=source_dir,
    entry_point=script_path,
    framework_version=FRAMEWORK_VERSION,
    sagemaker_session=sagemaker_session,
    role=role
)
ENDPOINT_NAME = 'visibility-model-v1'
model.deploy(
    initial_instance_count=1,
    instance_type='ml.m5.large',
    endpoint_name=ENDPOINT_NAME
)

实时评估模型

每当为我们的一种产品发送新通知时，我们都希望计算并提交最优价格。 为了计算最优价格，我们创建了一个预测数据集，在其中我们将我们自己的报价与每个竞争对手的报价进行比较，以获得一系列可能的价格。 这些数据点被传递到 SageMaker 端点，该端点返回每个给定价格对每个竞争对手可见的预测概率。 我们称可见的概率为 预测能见度. 结果可以可视化为每个竞争对手的曲线，描绘了我们的价格和知名度之间的关系，如下图所示。

在这个例子中，针对竞争对手 1 的可见性几乎是一个分段常数函数，这表明我们主要必须将价格降低到某个阈值以下，大致是竞争对手的价格，才能变得可见。 但是，对竞争对手 2 的能见度并没有急剧下降。 最重要的是，即使价格非常高，我们仍有 50% 的机会被看到。 分析输入数据显示，竞争对手的收视率很低，恰好很差。 我们的模型了解到，这个特定的电子商务平台对反馈评分较差的卖家不利。 我们发现其他功能也有类似的影响，例如履行渠道和运输时间。

针对 SageMaker 终端节点的必要数据转换和推断在 AWS Glue 中实施。 AWS Glue 作业对从 Lambda 摄取的实时数据进行微批量处理。

最后，我们要计算聚合可见度曲线，即每个可能价格的预测可见度。 如果它比所有其他卖家的报价更好，我们的报价是可见的。 假设给定我们的价格，对每个卖家可见的概率之间是独立的，那么对所有卖家可见的概率是各自概率的乘积。 这意味着可以通过将所有曲线相乘来计算聚合的可见性曲线。

下图显示了从 SageMaker 端点返回的预测可见性。

下图显示了聚合的可见性曲线。

为了计算最优价格，可见性曲线首先被平滑，然后乘以保证金。 为了计算保证金，我们使用商品成本和费用。 销售成本和费用是通过 AWS DMS 同步的静态产品信息。 Adspert 根据利润函数计算出最优价格，并通过平台的 API 提交给电商平台。

这是在 AWS Lambda 预测优化器中实现的。

下图显示了预测可见性和价格之间的关系。

下图显示了价格和利润之间的关系。

结论

Adspert 现有的利润最大化方法侧重于投标管理，以增加广告回报。 然而，为了在电子商务市场上取得卓越的业绩，卖家必须同时考虑广告和产品的竞争定价。 使用这个新的机器学习模型来预测可见性，我们可以扩展我们的功能来调整客户的价格。

新的定价工具必须能够在大量数据上自动训练 ML 模型，以及实时数据转换、预测和价格优化。 在这篇文章中，我们介绍了价格优化引擎的主要步骤，以及我们与 AWS 数据实验室合作实施的 AWS 架构以实现这些目标。

将 ML 模型从概念转变为生产通常是复杂且耗时的。 您必须管理大量数据来训练模型，选择最佳算法进行训练，在训练时管理计算能力，然后将模型部署到生产环境中。 SageMaker 通过更直接地构建和部署 ML 模型来降低这种复杂性。 在我们从众多可用选项中选择了正确的算法和框架后，SageMaker 管理所有底层基础设施来训练我们的模型并将其部署到生产中。

如果您想开始熟悉 SageMaker， 沉浸日工作坊 可以帮助您端到端地了解如何从特征工程构建 ML 用例、各种内置算法，以及如何在类似生产的场景中训练、调整和部署 ML 模型。 它指导您引入自己的模型，并将本地 ML 工作负载迁移到 SageMaker 平台。 它进一步展示了模型调试、模型监控和 AutoML 等高级概念，并帮助您通过 AWS ML Well-Architected 镜头评估您的 ML 工作负载。

如果您希望帮助加快实施涉及数据、分析、人工智能和机器学习、无服务器和容器现代化的用例，请联系 AWS数据实验室.

关于作者

维克多·恩里科·杰尼 是位于德国柏林的 Adspert 的高级机器学习工程师。 他为预测和优化问题创建解决方案，以增加客户的利润。 Viktor 拥有应用数学背景，喜欢处理数据。 在空闲时间，他喜欢学习匈牙利语、练习武术和弹吉他。

恩尼奥·帕斯托雷 是 AWS 数据实验室团队的数据架构师。 他热衷于与对企业和一般生活产生积极影响的新技术相关的一切。 Ennio 在数据分析方面拥有超过 9 年的经验。 他帮助公司定义和实施跨行业的数据平台，例如电信、银行、游戏、零售和保险。