Online Retail - Predicting order cancellations¶

%pip install -q "getml==1.5.0" "pyspark==3.5.2" "ipywidgets==8.1.5"

RUN_SPARK = False

fname = "online_retail.csv"

if not os.path.exists(fname):
    fname, res = request.urlretrieve(
        "https://static.getml.com/datasets/online_retail/" + fname, 
        fname
    )
    
full_data_pandas = pd.read_csv(fname, sep="|")

def add_zero(string):
    if len(string) == 1:
        return "0" + string
    return string

def format_date(string):
    datetime = string.split(" ")
    assert len(datetime) == 2, "Expected date and time"
    
    date_components = datetime[0].split("/")
    assert len(date_components) == 3, "Expected three date components"
    
    date_components = [add_zero(x) for x in date_components]
    
    return "-".join(date_components) + " " + datetime[1]

full_data_pandas["InvoiceDate"] = [
    format_date(string) for string in np.asarray(full_data_pandas["InvoiceDate"])
]

def add_target(df):
    df = df.sort_values(by=["Customer ID", "InvoiceDate"])
    
    cancelled = np.zeros(df.shape[0])

    invoice = np.asarray(df["Invoice"])
    stock_code = np.asarray(df["StockCode"])
    customer_id = np.asarray(df["Customer ID"])

    for i in range(len(invoice)):
        if (invoice[i][0] == 'C') or (i == len(invoice) - 1):
            continue

        j = i + 1

        while customer_id[j] == customer_id[i]:
            if (invoice[j][0] == 'C') and (stock_code[i] == stock_code[j]):
                cancelled[i] = 1.0
                break

            if stock_code[i] == stock_code[j]:
                break

            j += 1
    
    df["cancelled"] = cancelled
    
    return df

def remove_cancellations(df):
    invoice = np.asarray(df["Invoice"])

    is_order = [inv[0] != 'C' for inv in invoice]
    
    df = df[is_order]
    
    return df

full_data_pandas = add_target(full_data_pandas)
full_data_pandas = remove_cancellations(full_data_pandas)

full_data_pandas = full_data_pandas[~np.isnan(full_data_pandas["Customer ID"])]

full_data.set_role("InvoiceDate", getml.data.roles.time_stamp, time_formats=['%Y-%m-%d %H:%M'])
full_data.set_role(["Customer ID", "Invoice"], getml.data.roles.join_key)
full_data.set_role(["cancelled"], getml.data.roles.target)
full_data.set_role(["Quantity", "Price"], getml.data.roles.numerical)
full_data.set_role("Country", getml.data.roles.categorical)
full_data.set_role("Description", getml.data.roles.text)

full_data.set_unit("StockCode", "code")

substr1 = getml.preprocessors.Substring(0, 1, "code")
substr2 = getml.preprocessors.Substring(0, 2, "code")
substr3 = getml.preprocessors.Substring(0, 3, "code")

mapping = getml.preprocessors.Mapping()

text_field_splitter = getml.preprocessors.TextFieldSplitter()

fast_prop = getml.feature_learning.FastProp(
    loss_function=getml.feature_learning.loss_functions.CrossEntropyLoss,
    num_threads=1,
    sampling_factor=0.1,
)

feature_selector = getml.predictors.XGBoostClassifier()

predictor = getml.predictors.XGBoostClassifier()

pipe.features.to_sql(dialect=getml.pipeline.dialect.spark_sql).save("online_retail_spark")

if RUN_SPARK:
    spark = SparkSession.builder.appName(
        "online_retail"
    ).config(
        "spark.driver.maxResultSize","5g"
    ).config(
        "spark.driver.memory", "5g"
    ).config(
        "spark.executor.memory", "5g"
    ).config(
        "spark.sql.execution.arrow.pyspark.enabled", "true"
    ).config(
        "spark.sql.session.timeZone", "UTC"
    ).enableHiveSupport().getOrCreate()

    spark.sparkContext.setLogLevel("ERROR")

if RUN_SPARK:
    population_spark = star_schema.train.population.to_pyspark(spark, name="population")
    peripheral_spark = star_schema.full_data.to_pyspark(spark, name="full_data")

if RUN_SPARK:
    getml.spark.execute(spark, "online_retail_spark")

if RUN_SPARK:
    spark.sql("SELECT * FROM `FEATURES` LIMIT 20").toPandas()

getml.engine.shutdown()