Preț optim pentru profit maxim folosind Amazon SageMaker

Aceasta este o postare invitată de Viktor Enrico Jeney, inginer senior de învățare automată la Adspert.

Adspert este un ISV cu sediul în Berlin care a dezvoltat un instrument de gestionare a sumelor licitate conceput pentru a optimiza automat campaniile de marketing și publicitate de performanță. Principiul de bază al companiei este de a automatiza maximizarea profitului publicității de comerț electronic cu ajutorul inteligenței artificiale. Dezvoltarea continuă a platformelor de publicitate deschide calea pentru noi oportunități, pe care Adspert le utilizează cu experiență pentru succesul clienților săi.

Scopul principal al Adspert este de a simplifica procesul pentru utilizatori în timp ce optimizează campaniile publicitare pe diferite platforme. Aceasta include utilizarea informațiilor adunate pe diferite platforme, în raport cu bugetul optim stabilit la un nivel deasupra fiecărei platforme. Accentul Adspert este de a optimiza atingerea obiectivului unui client, indiferent de platforma utilizată. Adspert continuă să adauge platforme după cum este necesar pentru a oferi clienților noștri avantaje semnificative.

În această postare, împărtășim modul în care Adspert a creat instrumentul de stabilire a prețurilor de la zero folosind diferite servicii AWS, cum ar fi Amazon SageMaker și modul în care Adspert a colaborat cu Laboratorul de date AWS pentru a accelera acest proiect de la proiectare la construcție în timp record.

Instrumentul de stabilire a prețurilor reprețează un produs selectat de vânzător pe o piață de comerț electronic pe baza vizibilității și a marjei de profit pentru a maximiza profiturile la nivel de produs.

În calitate de vânzător, este esențial ca produsele dvs. să fie întotdeauna vizibile, deoarece acest lucru va crește vânzările. Cel mai important factor în vânzările de comerț electronic este pur și simplu dacă oferta dvs. este vizibilă pentru clienți în loc de oferta unui concurent.

Deși depinde cu siguranță de platforma specifică de comerț electronic, am constatat că prețul produsului este unul dintre cele mai importante cifre cheie care pot afecta vizibilitatea. Cu toate acestea, prețurile se schimbă des și rapid; din acest motiv, instrumentul de stabilire a prețurilor trebuie să acționeze în timp aproape real pentru a crește vizibilitatea.

Prezentare generală a soluției

Următoarea diagramă ilustrează arhitectura soluției.

Soluția conține următoarele componente:

1. Amazon Relational Database Service (Amazon RDS) pentru PostgreSQL este principala sursă de date, care conține informații despre produse care sunt stocate într-o bază de date RDS pentru Postgres.
2. Informațiile privind modificările din lista de produse ajung în timp real într-un Serviciul de coadă simplă Amazon (Amazon SQS) coadă.
3. Informațiile despre produse stocate în Amazon RDS sunt ingerate în timp aproape real în stratul brut folosind modelul de captare a datelor de modificare (CDC) disponibil în Serviciul de migrare a bazelor de date AWS (AWS DMS).
4. Notificările privind lista de produse care provin de la Amazon SQS sunt ingerate în timp aproape real în stratul brut folosind un AWS Lambdas Funcția.
5. Datele sursă originale sunt stocate în Serviciul Amazon de stocare simplă (Amazon S3) găleată cu strat brut folosind formatul de date Parquet. Acest strat este singura sursă de adevăr pentru lacul de date. Partiționarea utilizată pe această stocare acceptă procesarea incrementală a datelor.
6. AWS Adeziv Lucrările de extragere, transformare și încărcare (ETL) curăță datele despre produse, eliminând duplicatele și aplicând consolidări de date și transformări generice care nu sunt legate de un anumit caz de afaceri.
7. Stratul de etapă Amazon S3 primește date pregătite care sunt stocate în format Apache Parquet pentru procesare ulterioară. Partiționarea utilizată în depozitul de scenă sprijină procesarea incrementală a datelor.
8. Joburile AWS Glue create în acest strat folosesc datele disponibile în stratul de etapă Amazon S3. Aceasta include aplicarea regulilor de afaceri specifice cazului de utilizare și calculele necesare. Datele rezultate din aceste joburi sunt stocate în stratul de analiză Amazon S3.
9. Stratul de analiză Amazon S3 este utilizat pentru a stoca datele care sunt utilizate de modelele ML în scopuri de instruire. Partiționarea utilizată în magazinul curatat se bazează pe utilizarea datelor așteptată. Acest lucru poate fi diferit de partiționarea utilizată pe stratul de scenă.
10. Modelul ML de reevaluare este o implementare Scikit-Learn Random Forest în SageMaker Script Mode, care este antrenat folosind datele disponibile în compartimentul S3 (stratul de analiză).
11. O lucrare de procesare a datelor AWS Glue pregătește datele pentru inferența în timp real. Lucrarea procesează datele ingerate în compartimentul S3 (stratul de etapă) și invocă punctul final de inferență SageMaker. Datele sunt pregătite pentru a fi utilizate de modelul de repreciere SageMaker. AWS Glue a fost preferat în locul Lambda, deoarece inferența necesită diferite operațiuni complexe de procesare a datelor, cum ar fi funcții de alăturare și ferestre pentru un volum mare de date (miliarde de tranzacții zilnice). Rezultatele invocărilor modelului de repricing sunt stocate în compartimentul S3 (stratul de inferență).
12. Jobul de instruire SageMaker este implementat folosind un punct final SageMaker. Acest punct final este invocat de procesorul de inferență AWS Glue, generând recomandări de preț aproape în timp real pentru a crește vizibilitatea produsului.
13. Predicțiile generate de punctul final de inferență SageMaker sunt stocate în stratul de inferență Amazon S3.
14. Funcția de optimizare a predicțiilor Lambda procesează recomandările generate de punctul final de inferență SageMaker și generează o nouă recomandare de preț care se concentrează pe maximizarea profitului vânzătorului, aplicând un compromis între volumul vânzărilor și marja vânzărilor.
15. Recomandările de preț generate de optimizatorul de predicții Lambda sunt trimise la API-ul de repricing, care actualizează prețul produsului pe piață.
16. Recomandările de preț actualizate generate de optimizatorul de predicții Lambda sunt stocate în stratul de optimizare Amazon S3.
17. Lucrarea de încărcare a predicțiilor AWS Glue reîncarcă în baza de date RDS sursă pentru Postgres SQL predicțiile generate de modelul ML în scopuri de auditare și raportare. AWS Glue Studio a fost folosit pentru a implementa această componentă; este o interfață grafică care facilitează crearea, rularea și monitorizarea joburilor ETL în AWS Glue.

Pregătirea datelor

Setul de date pentru modelul de vizibilitate Adspert este creat dintr-o coadă SQS și ingerat în stratul brut al lacului nostru de date în timp real cu Lambda. Ulterior, datele brute sunt igienizate prin efectuarea de transformări simple, cum ar fi eliminarea duplicatelor. Acest proces este implementat în AWS Glue. Rezultatul este stocat în stratul de staging al lacului nostru de date. Notificările oferă concurenților pentru un anumit produs, prețurile, canalele de realizare, timpii de livrare și multe alte variabile. Ele oferă, de asemenea, o măsură de vizibilitate dependentă de platformă, care poate fi exprimată ca o variabilă booleană (vizibilă sau invizibilă). Primim o notificare ori de câte ori are loc o modificare a ofertei, ceea ce însumează câteva milioane de evenimente pe lună pentru toate produsele clienților noștri.

Din acest set de date, extragem datele de antrenament după cum urmează: pentru fiecare notificare, asociam ofertele vizibile cu fiecare ofertă invizibilă și invers. Fiecare punct de date reprezintă o competiție între doi vânzători, în care există un câștigător și un învins clar. Această sarcină de procesare este implementată într-o lucrare AWS Glue cu Spark. Setul de date de antrenament pregătit este trimis în compartimentul de analiză S3 pentru a fi utilizat de SageMaker.

Antrenează modelul

Modelul nostru clasifică pentru fiecare pereche de oferte, dacă o anumită ofertă va fi vizibilă. Acest model ne permite să calculăm cel mai bun preț pentru clienții noștri, să creștem vizibilitatea pe baza concurenței și să le maximizăm profitul. În plus, acest model de clasificare ne poate oferi o perspectivă mai profundă asupra motivelor pentru care înregistrările noastre sunt vizibile sau nu. Folosim următoarele caracteristici:

• Raportul dintre prețul nostru și prețurile concurenților
• Diferența în canalele de realizare
• Cantitatea de feedback pentru fiecare vânzător
• Evaluarea feedback-ului fiecărui vânzător
• Diferența în timpii minimi de livrare
• Diferența de timp maxim de livrare
• Disponibilitatea produsului fiecărui vânzător

Adspert folosește SageMaker pentru a antrena și a găzdui modelul. Folosim implementarea Scikit-Learn Random Forest în Modul Script SageMaker. De asemenea, includem unele funcții de preprocesare direct în conducta Scikit-Learn în scriptul de instruire. Vezi următorul cod:

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)

Una dintre cele mai importante funcții de preprocesare este transform_price, care împarte prețul la minimul dintre prețul concurentului și o coloană de preț extern. Am descoperit că această caracteristică are un impact relevant asupra preciziei modelului. De asemenea, aplicăm logaritmul pentru a lăsa modelul să decidă pe baza diferențelor de preț relative, nu diferențelor de preț absolute.

În training_script.py script, definim mai întâi cum să construim Scikit-Learn ColumnTransformer pentru a aplica transformatoarele specificate pe coloanele unui cadru de date:

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')

În scriptul de antrenament, încărcăm datele din Parquet într-un cadru de date Pandas, definim pipeline-ul ColumnTranformer si RandomForestClassifier, și antrenați modelul. Ulterior, modelul este serializat folosind 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'))

În scriptul de antrenament, trebuie să implementăm și funcții pentru inferență:

• input_fn – Este responsabil pentru analizarea datelor din corpul solicitării încărcăturii utile
• model_fn – Încarcă și returnează modelul care a fost descărcat în secțiunea de instruire a scriptului
• prezice_fn – Conține implementarea noastră pentru a solicita o predicție de la model folosind datele din sarcina utilă
• predict_proba – Pentru a desena curbele de vizibilitate prezise, ​​returnăm probabilitatea clasei folosind predict_proba funcția, în locul predicției binare a clasificatorului

Consultați următorul cod:

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

Următoarea figură arată importanțele caracteristicilor bazate pe impurități returnate de Clasificator de pădure aleatoriu.

Cu SageMaker, am reușit să antrenăm modelul pe o cantitate mare de date (până la 14 miliarde de tranzacții zilnice) fără a încărca instanțele noastre existente sau a trebui să configuram o mașină separată cu suficiente resurse. Mai mult, deoarece instanțele sunt imediat închise după jobul de instruire, instruirea cu SageMaker a fost extrem de eficientă din punct de vedere al costurilor. Implementarea modelului cu SageMaker a funcționat fără încărcătură de lucru suplimentară. Un singur apel de funcție în Python SDK este suficient pentru a găzdui modelul nostru ca punct final de inferență și poate fi solicitat cu ușurință de la alte servicii folosind, de asemenea, SageMaker Python SDK. Vezi următorul cod:

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})

Artefactul model este stocat în Amazon S3 prin funcția de potrivire. După cum se vede în codul următor, modelul poate fi încărcat ca a SKLearnModel obiect folosind artefactul modelului, calea scriptului și alți parametri. Ulterior, poate fi implementat la tipul și numărul de instanțe dorite.

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
)

Evaluați modelul în timp real

Ori de câte ori este trimisă o nouă notificare pentru unul dintre produsele noastre, dorim să calculăm și să trimitem prețul optim. Pentru a calcula prețurile optime, creăm un set de date de predicție în care comparăm propria noastră ofertă cu oferta fiecărui concurent pentru o gamă de prețuri posibile. Aceste puncte de date sunt transmise la punctul final SageMaker, care returnează probabilitatea estimată de a fi vizibile față de fiecare concurent pentru fiecare preț dat. Numim probabilitatea de a fi vizibil vizibilitatea prevăzută. Rezultatul poate fi vizualizat ca o curbă pentru fiecare concurent, prezentând relația dintre prețul nostru și vizibilitate, așa cum se arată în figura următoare.

În acest exemplu, vizibilitatea față de concurentul 1 este aproape o funcție constantă pe bucăți, ceea ce sugerează că în principal trebuie să scădem prețul sub un anumit prag, aproximativ prețul concurentului, pentru a deveni vizibili. Cu toate acestea, vizibilitatea față de Competitor 2 nu scade la fel de abrupt. În plus, avem în continuare șanse de 50% să fim vizibili chiar și cu un preț foarte mare. Analiza datelor de intrare a relevat faptul că concurentul are un număr scăzut de evaluări, care se întâmplă să fie foarte slabe. Modelul nostru a învățat că această platformă de comerț electronic specifică oferă un dezavantaj vânzătorilor cu evaluări slabe de feedback. Am descoperit efecte similare pentru celelalte funcții, cum ar fi canalul de realizare și timpii de livrare.

Transformările de date și inferențe necesare față de punctul final SageMaker sunt implementate în AWS Glue. Lucrarea AWS Glue funcționează în micro-loturi pe datele în timp real ingerate de la Lambda.

În cele din urmă, dorim să calculăm curba de vizibilitate agregată, care este vizibilitatea prezisă pentru fiecare preț posibil. Oferta noastră este vizibilă dacă este mai bună decât ofertele tuturor celorlalți vânzători. Presupunând independența dintre probabilitățile de a fi vizibil față de fiecare vânzător dat fiind prețul nostru, probabilitatea de a fi vizibil față de toți vânzătorii este produsul probabilităților respective. Aceasta înseamnă că curba de vizibilitate agregată poate fi calculată prin înmulțirea tuturor curbelor.

Următoarele figuri arată vizibilitățile estimate returnate de la punctul final SageMaker.

Următoarea figură prezintă curba de vizibilitate agregată.

Pentru a calcula prețul optim, curba de vizibilitate este mai întâi netezită și apoi înmulțită cu marjă. Pentru a calcula marja, folosim costurile mărfurilor și taxele. Costul bunurilor vândute și taxele sunt informațiile statice despre produs sincronizate prin AWS DMS. Pe baza funcției de profit, Adspert calculează prețul optim și îl trimite pe platforma de comerț electronic prin API-ul platformei.

Acest lucru este implementat în optimizatorul de predicții AWS Lambda.

Următoarea figură arată relația dintre vizibilitatea estimată și preț.

Figura următoare arată relația dintre preț și profit.

Concluzie

Abordarea existentă a Adspert pentru maximizarea profitului se concentrează pe gestionarea sumelor licitate pentru a crește profiturile din publicitate. Cu toate acestea, pentru a obține performanțe superioare pe piețele de comerț electronic, vânzătorii trebuie să ia în considerare atât publicitatea, cât și prețurile competitive ale produselor lor. Cu acest nou model ML pentru a prezice vizibilitatea, ne putem extinde funcționalitatea pentru a ajusta și prețurile clienților.

Noul instrument de stabilire a prețurilor trebuie să fie capabil de instruire automată a modelului ML pe o cantitate mare de date, precum și transformări de date în timp real, predicții și optimizări de preț. În această postare, am parcurs pașii principali ai motorului nostru de optimizare a prețurilor și arhitectura AWS pe care am implementat-o ​​în colaborare cu AWS Data Lab pentru a atinge aceste obiective.

Preluarea modelelor ML de la concept la producție este de obicei complexă și necesită timp. Trebuie să gestionați cantități mari de date pentru a antrena modelul, să alegeți cel mai bun algoritm pentru antrenamentul acestuia, să gestionați capacitatea de calcul în timp ce îl antrenați și apoi să implementați modelul într-un mediu de producție. SageMaker a redus această complexitate făcând mult mai simplă construirea și implementarea modelului ML. După ce am ales algoritmii și cadrele potrivite din gama largă de opțiuni disponibile, SageMaker a gestionat toată infrastructura de bază pentru a ne instrui modelul și a-l implementa în producție.

Dacă doriți să începeți să vă familiarizați cu SageMaker, Atelier de zi de imersiune vă poate ajuta să obțineți o înțelegere de la capăt la capăt a modului de a construi cazuri de utilizare ML din ingineria caracteristicilor, diferiții algoritmi încorporați și cum să antrenați, să reglați și să implementați modelul ML într-un scenariu asemănător producției. Vă îndrumă să vă aduceți propriul model și să efectuați o creștere și schimbare a sarcinii de lucru ML la nivel local către platforma SageMaker. Demonstrează în continuare concepte avansate precum depanarea modelului, monitorizarea modelului și AutoML și vă ajută să vă evaluați volumul de lucru ML prin prisma AWS ML Well-Architected.

Dacă doriți ajutor pentru accelerarea implementării cazurilor de utilizare care implică date, analiză, AI și ML, fără server și modernizarea containerelor, vă rugăm să contactați Laboratorul de date AWS.

Despre autori

Viktor Enrico Jeney este inginer senior de învățare automată la Adspert, cu sediul în Berlin, Germania. El creează soluții pentru probleme de predicție și optimizare pentru a crește profiturile clienților. Viktor are experiență în matematică aplicată și îi place să lucreze cu date. În timpul liber, îi place să învețe limba maghiară, să practice artele marțiale și să cânte la chitară.

Ennio Pastore este un arhitect de date în echipa AWS Data Lab. Este un pasionat de tot ceea ce are legătură cu noile tehnologii care au un impact pozitiv asupra afacerilor și a mijloacelor de trai în general. Ennio are peste 9 ani de experiență în analiza datelor. El ajută companiile să definească și să implementeze platforme de date în diverse industrii, cum ar fi telecomunicații, bancar, jocuri, retail și asigurări.