Let PD be a Plane/Date couple. For each PD, I would like to predict the forecasted booking curve, i.e. the cumuled amount of bookings registered each day between X days before departure and departure. Let us name JX the day before departure (actually, it has a negative sign for other reasons).
I also need to recompute this curve each day, to account for new informations. Here is an example of what I would obtain for a PD that currently has 100 bookings at JX = -15 (10 days remaining before departure), and will get 10 reservations per day during the 5 remaining days. Let us suppose, for this example, that my model predicts 11 bookings per day. In my following dataframe, TRAF designate the current booking and CURVE_FINAL_TRAF the forecasting curve
PD | JX | TRAF | CURVE_FINAL_TRAF
--------------------------------------------------------------------------------------
1 | -5 | 100 | {-5 : 100, -4 : 111, -3 : 122, -2 : 133, -1: 144, 0: 155}
1 | -4 | 110 | {-4 : 110, -3 : 121, -2 : 132, -1: 143, 0: 154}
1 | -3 | 120 | {-3 : 120, -2 : 131, -1: 142, 0: 153}
1 | -2 | 130 | {-2 : 130, -1 : 141, 0: 152}
1 | -1 | 140 | {-1 : 140, 0 : 151}
1 | 0 | 150 | {0 : 150}
This is contained in a pandas.DataFrame instance. I have considered a "long" storage with the following columns.
PD | JX | JX_CURVE | FORECASTED_VALUE
but the computation time was actually longer, probably due to the amount of other columns in my DataFrame, that needs to be duplicated. But I may not have tried hard enough. So, my first question is is there a standard way to store this kind of data while keeping the wide format while minimizing the amount of storage space/code/etc. ?
My second question has more to do with the way I implement the model. I use a standard API model (In my specific case, this is a LGBMRegressor, but it could be anything with a fit/predict) and wrap it in a new class.
Let us call STATIC_FEATURES the features relative to a given PD, but that will not move before departure and DYNAMIC_FEATURES the features relative to a given PD that will move before departure.
My program does this :
- create a copy of my prediction dataframe, that contains both
FEATURESandDYNAMIC_FEATURES. - while some lines have a JX < 0 (actually 1 in my case, because there are result consolidation after departure, but let us say 0 for the example)
- Compute the "DIFF_TRAF" forecast using the copy of the DataFrame, the amount of booking that occurs at a given JX for a given PD.
- Store this "DIFF_TRAF" in a dictionnary placed inside a column of my DataFrame.
- Given this "DIFF" forecast, update the
DYNAMIC_FEATURESof my DataFrame copy to simulate tomorrow situation.
- At the end of the "while", it computes the curve indicated.
In practice, I have two models, one that compute the TBS_TRAF ("To-Be-Sold") remaining forecast total, and one that compute the "DIFF_TRAF" model explained above. I have a bunch of other options to make my code a bit more flexible, but that are not relevant for the problem.
My second question is : Is there a way to speed up my code ? My unit test for the whole predict function (attached below) takes ~3sec to process 200 lines.
import numpy as np
class Label:
def __init__(self, current, final="", diff="", tbs="", diff_tbs=""):
self.current = current
self.diff = diff if diff else f"DIFF_{current}"
self.final = final if final else f"FINAL_{current}"
self.tbs = tbs if tbs else f"TBS_{current}"
self.diff_tbs = diff_tbs if diff_tbs else f"DIFF_TBS_{current}"
def default_update_rolling_variable_func(temp_prediction_df):
"""should update the """
raise NotImplementedError
class ReservationCurveModel:
def __init__(self, model_class, update_rolling_variable_func=default_update_rolling_variable_func):
self.model_class = model_class
self.update_dynamic_variables = update_rolling_variable_func
self.models = {}
def fit(self, dataframe, features, label: Label):
assert (label.diff in dataframe.columns.values) and (label.tbs in dataframe.columns.values)
self.models[label.diff] = self.model_class().fit(
dataframe[features].values, dataframe[label.diff].values.ravel()
)
self.models[label.tbs] = self.model_class().fit(
dataframe[features].values, dataframe[label.tbs].values.ravel()
)
def predict(self, prediction_df, features, label: Label):
prediction_df[f"PREV_{label.tbs}"] = (
self.models[f"{label.tbs}"]
.predict(prediction_df[features].values)
.round(3)
)
temp_prediction_df = prediction_df.copy()
while temp_prediction_df.JX.min() <= 1:
temp_prediction_df = self.do_the_predict_step(temp_prediction_df, features, label)
prediction_df = self.save_new_prev_in_dict(prediction_df, temp_prediction_df, label)
temp_prediction_df = self.update_dynamic_variables(temp_prediction_df)
prediction_df = self.refine_dicts(prediction_df, label)
return prediction_df
def do_the_predict_step(self, temp_prediction_df, features, label: Label):
temp_prediction_df[f"PREV_{label.diff}"] = (
self.models[label.diff]
.predict(temp_prediction_df[features].values)
.round(3)
)
temp_prediction_df[f"PREV_{label.tbs}"] = (
self.models[label.tbs]
.predict(temp_prediction_df[features].values)
.round(3)
)
return temp_prediction_df
@staticmethod
def update_dict(row, kind, label: Label, temp_prediction_df):
assert kind in ["diff", "tbs"]
dict_to_return = dict(
row[f"CURVE_{getattr(label, kind)}"],
**(
{
str(temp_prediction_df.JX.iloc[row.name]): temp_prediction_df[
f"PREV_{getattr(label, kind)}"
].iloc[row.name]
}
if (temp_prediction_df.JX.iloc[row.name] <= 1)
else {}
),
)
return dict_to_return
def save_new_prev_in_dict(self, prediction_df, temp_prediction_df, label: Label):
for kind in ["diff", "tbs"]:
if not (f"CURVE_{getattr(label, kind)}" in prediction_df.columns.values.tolist()):
prediction_df[f"CURVE_{getattr(label, kind)}"] = prediction_df.apply(
lambda row: {}, axis=1
)
prediction_df[f"CURVE_{getattr(label, kind)}"] = prediction_df.apply(
lambda row: self.update_dict(row, kind, label, temp_prediction_df),
axis=1,
)
return prediction_df
@staticmethod
def compute_diff_tbs_dict(row, label):
return dict(
zip(
list(row[f"CURVE_{label.tbs}"].keys()),
np.abs(
np.clip(
np.round(
np.diff(
np.array(list(row[f"CURVE_{label.tbs}"].values())),
append=0,
),
3,
),
None,
0,
)
),
)
)
@staticmethod
def compute_load_factor(row, label: Label):
return dict(
zip(
list(row[f"CURVE_{label.diff_tbs}"].keys()),
np.round(
np.cumsum(np.array(list(row[f"CURVE_{label.diff_tbs}"].values()))), 3
)
+ round(row[label.current], 3),
)
)
def refine_dicts(self, prediction_df, label):
prediction_df[f"CURVE_{label.diff_tbs}"] = prediction_df.apply(
lambda row: self.compute_diff_tbs_dict(row, label), axis=1
)
prediction_df[f"CURVE_{label.final}"] = prediction_df.apply(
lambda row: self.compute_load_factor(row, label), axis=1
)
return prediction_df
and some unit testing :
import numpy as np
import pandas as pd
from cassioutils import Label, ReservationCurveModel
from pandas.testing import assert_frame_equal
from lightgbm import LGBMRegressor
import os
def generate_dataset():
df = pd.DataFrame(index=list(range(200)))
df["FEATURE_1"] = [1] * 100 + [2] * 100
df["JX"] = [- i for i in range(100)] * 2
df["TRAF"] = [(1000 - 10*i) for i in range(100)] + [(2000 - 20*i) for i in range(100)]
df["DIFF_TRAF"] = [10 for i in range(100)] + [20 for i in range(100)]
df["TBS_TRAF"] = [10*i for i in range(100)] + [20*i for i in range(100)]
return df
class DummyModel:
def __new__(self):
return LGBMRegressor(random_state=0)
def test_label():
label = Label("TRAF", final="TRAF_J_1")
assert (
(label.current == "TRAF")
and (label.tbs == "TBS_TRAF")
and (label.final == "TRAF_J_1")
and (label.diff_tbs == "DIFF_TBS_TRAF")
), "Boom"
def test_reservation_curve_model_fitting():
df = generate_dataset()
model = ReservationCurveModel(DummyModel, lambda x: x)
model.fit(
df,
["FEATURE_1", "JX"],
label=Label("TRAF"),
)
df["PREV_DIFF_TRAF"] = model.models["DIFF_TRAF"].predict(
df[["FEATURE_1", "JX"]].values,
)
assert (df.PREV_DIFF_TRAF.round() == df.DIFF_TRAF.round()).all()
def test_compute_diff_tbs_dict():
row = {"CURVE_TBS_TRAF": {-2: 100, -1: 100, 0: 50, 1: 0}}
model = ReservationCurveModel(DummyModel)
assert model.compute_diff_tbs_dict(row, Label("TRAF")) == {-2: 0, -1: 50, 0: 50, 1: 0}
def test_compute_load_factor():
row = {"CURVE_DIFF_TBS_TRAF": {-2: 0, -1: 50, 0: 50, 1: 0}, "TRAF": 0}
model = ReservationCurveModel(DummyModel)
assert model.compute_load_factor(row, Label("TRAF")) == {-2: 0, -1: 50, 0: 100, 1: 100}
def test_refine_dict():
prediction_df = pd.DataFrame(
[[0, {-2: 100, -1: 100, 0: 50, 1: 0}]],
columns=["TRAF", "CURVE_TBS_TRAF"]
)
model = ReservationCurveModel(DummyModel)
prediction_df = model.refine_dicts(prediction_df, Label("TRAF"))
assert prediction_df.loc[0, "CURVE_DIFF_TBS_TRAF"] == {-2: 0, -1: 50, 0: 50, 1: 0}
assert prediction_df.loc[0, "CURVE_FINAL_TRAF"] == {-2: 0, -1: 50, 0: 100, 1: 100}
def test_save_new_prev_in_dict():
temp_prediction_df = pd.DataFrame(
data=[
[100, 10, -10],
[90, 10, -5],
],
columns=[
"PREV_TBS_TRAF",
"PREV_DIFF_TRAF",
"JX",
],
)
prediction_df = pd.DataFrame(
data=[
[-10],
[-5],
],
columns=[
"JX",
],
)
expected_df = pd.DataFrame(
data=[
[-10, {'-10': 10}, {'-10': 100}],
[-5, {'-5': 10}, {'-5': 90}],
],
columns=[
"JX",
"CURVE_DIFF_TRAF",
"CURVE_TBS_TRAF",
],
)
model = ReservationCurveModel(DummyModel)
prediction_df = model.save_new_prev_in_dict(prediction_df, temp_prediction_df, Label("TRAF"))
assert_frame_equal(expected_df, prediction_df)
def test_do_the_predict_step():
train_df = generate_dataset()
temp_prediction_df = pd.DataFrame(
data=[
[1, -10],
[1, -5],
],
columns=[
"FEATURE_1",
"JX",
],
)
expected_temp_prediction_df = pd.DataFrame(
data=[
[1, -10, 10.0, 881.952],
[1, -5, 10.0, 993.1],
],
columns=[
"FEATURE_1",
"JX",
"PREV_DIFF_TRAF",
"PREV_TBS_TRAF",
],
)
model = ReservationCurveModel(DummyModel)
model.fit(
train_df,
["FEATURE_1", "JX"],
label=Label("TRAF"),
)
temp_prediction_df = model.do_the_predict_step(temp_prediction_df, ["FEATURE_1", "JX"], Label("TRAF"))
assert_frame_equal(expected_temp_prediction_df, temp_prediction_df)
def test_reservation_curve_model_predicting():
df = generate_dataset()
current_dir = os.path.dirname(os.path.abspath(__file__))
expected_df = pd.read_csv(os.path.join(current_dir, "csv/curve_model/expected_predicted_df.csv")).drop(
"Unnamed: 0",
axis=1
)
for col in ["CURVE_DIFF_TRAF", "CURVE_FINAL_TRAF", "CURVE_TBS_TRAF", "CURVE_DIFF_TBS_TRAF"]:
expected_df.loc[:, col] = expected_df[col].apply(
lambda x: ast.literal_eval(x)
)
def update_dynamic_variables(df):
df["JX"] += 1
df["TRAF"] += df["PREV_DIFF_TRAF"]
return df
model = ReservationCurveModel(DummyModel, update_dynamic_variables)
model.fit(
df,
["FEATURE_1", "JX"],
label=Label("TRAF"),
)
predicted_df = model.predict(
df,
["FEATURE_1", "JX"],
label=Label("TRAF"),
)
assert_frame_equal(expected_df, predicted_df)
Please let me know if anything relevant is missing, I will try to provide it ASAP. In my case, expected_df == predicted_df, so you can quite easily reproduce the missing .csv
