diff --git a/models/marts/warehouse/dim__dates.py b/models/marts/warehouse/dim__dates.py new file mode 100644 index 0000000..4e17fdc --- /dev/null +++ b/models/marts/warehouse/dim__dates.py @@ -0,0 +1,80 @@ +# Python script to compute advance working dates + +# Weekend Example: using Saturday or Sunday and add 2 working days: +# Monday = 1 working day, Tuesday = 1 working day, Wednesday = Delivery Date +# Assumption for weekends, base date is NOT considered as one working day + +# Weekday Example: using Monday and adding 2 working days: +# Monday = 1 working day, Tuesday = 1 working day, Wednesday = Delivery Date +# Assumption for weekdays, base date is considered as one working day + +# Forecasting on future dates without information on holidays will not be accurate. + + +import pandas as pd +from datetime import datetime as dt, timedelta + + +# Function to check if a given date is a weekend [Saturday(5) or Sunday(6)] +def is_weekend(date): + return (date.weekday() == 5) | (date.weekday() == 6) + + +# Function to compute the in advance working days +def compute_working_days(start_date, num_working_days, holiday_list, after_cut_off = 0): + # after_cut_off = 0 for orders before 4pm + # after_cut_off = 1 for orders after 4pm + + counter = 0 + + # For dates on weekend or holiday or after cut off time, additional 1 working day to the loop + if (after_cut_off == 1) | is_weekend(start_date) | (start_date in holiday_list): + counter -= 1 + + # Loop to increase n number of working days, if weekend/holiday, skip counter + while counter < num_working_days: + start_date = start_date + timedelta(days=1) + + if is_weekend(start_date) | (start_date in holiday_list): + continue + + counter += 1 + + return start_date + + +# Main function +def model(dbt, session): + + # Setting configuration + dbt.config(materialized="table", + packages = ["pandas"]) + + # Import data from upstream dbt model + df_date = dbt.ref("int__dates") + + # Extract holiday dates into a list + df_work = df_date.to_pandas() + df_new = df_work[df_work["IS_COMPANY_HOLIDAY"] == 1] + df_holiday_list = df_new["DATE_DAY"].tolist() + + # Apply function to df + df_work["ADD_1_WORKING_DAY_INCLUDED_BASE_DATE"] = df_work.apply(lambda row: compute_working_days(row["DATE_DAY"], 1, df_holiday_list, 0), axis=1) + df_work["ADD_2_WORKING_DAY_INCLUDED_BASE_DATE"] = df_work.apply(lambda row: compute_working_days(row["DATE_DAY"], 2, df_holiday_list, 0), axis=1) + df_work["ADD_3_WORKING_DAY_INCLUDED_BASE_DATE"] = df_work.apply(lambda row: compute_working_days(row["DATE_DAY"], 3, df_holiday_list, 0), axis=1) + df_work["ADD_5_WORKING_DAY_INCLUDED_BASE_DATE"] = df_work.apply(lambda row: compute_working_days(row["DATE_DAY"], 5, df_holiday_list, 0), axis=1) + df_work["ADD_30_WORKING_DAY_INCLUDED_BASE_DATE"] = df_work.apply(lambda row: compute_working_days(row["DATE_DAY"], 30, df_holiday_list, 0), axis=1) + df_work["ADD_90_WORKING_DAY_INCLUDED_BASE_DATE"] = df_work.apply(lambda row: compute_working_days(row["DATE_DAY"], 90, df_holiday_list, 0), axis=1) + df_work["ADD_365_WORKING_DAY_INCLUDED_BASE_DATE"] = df_work.apply(lambda row: compute_working_days(row["DATE_DAY"], 365, df_holiday_list, 0), axis=1) + + # For orders after cut off time 4pm + # Affects only the weekdays + df_work["ADD_1_WORKING_DAY_EXCLUDED_BASE_DATE"] = df_work.apply(lambda row: compute_working_days(row["DATE_DAY"], 1, df_holiday_list, 1), axis=1) + df_work["ADD_2_WORKING_DAY_EXCLUDED_BASE_DATE"] = df_work.apply(lambda row: compute_working_days(row["DATE_DAY"], 2, df_holiday_list, 1), axis=1) + df_work["ADD_3_WORKING_DAY_EXCLUDED_BASE_DATE"] = df_work.apply(lambda row: compute_working_days(row["DATE_DAY"], 3, df_holiday_list, 1), axis=1) + df_work["ADD_5_WORKING_DAY_EXCLUDED_BASE_DATE"] = df_work.apply(lambda row: compute_working_days(row["DATE_DAY"], 5, df_holiday_list, 1), axis=1) + df_work["ADD_30_WORKING_DAY_EXCLUDED_BASE_DATE"] = df_work.apply(lambda row: compute_working_days(row["DATE_DAY"], 30, df_holiday_list, 1), axis=1) + df_work["ADD_90_WORKING_DAY_EXCLUDED_BASE_DATE"] = df_work.apply(lambda row: compute_working_days(row["DATE_DAY"], 90, df_holiday_list, 1), axis=1) + df_work["ADD_365_WORKING_DAY_EXCLUDED_BASE_DATE"] = df_work.apply(lambda row: compute_working_days(row["DATE_DAY"], 365, df_holiday_list, 1), axis=1) + + return df_work \ No newline at end of file