<jupyter_start><jupyter_text>actual and Predicted for given day<jupyter_code>def get_aggregate(df):
    list = ["date","hour"]
    df = df.groupBy(list).agg(sum("label"),sum("prediction"))    
    return df

def select_predicted_actual(df,date,LCLid=None):
    list = []
    if LCLid != None:
        list = df.where((df["LCLid"] == LCLid) & (df["date"] == date)).select("label","prediction").collect()
    else:
        list = df.where((df["date"] == date)).select("label","prediction").collect()
    actual = [int(row['label']) for row in list]
    predicted = [int(row['prediction']) for row in list]
    return actual,predicted

def mean_absolute_percentage_error(y_true, y_pred): 
    y_true, y_pred = np.array(y_true), np.array(y_pred)
    return np.mean(np.abs((y_true - y_pred) / y_true)) * 100

def root_mean_squared_error(y_true, y_pred): 
    y_true, y_pred = np.array(y_true), np.array(y_pred)
    return np.sqrt(np.mean(np.abs((y_true - y_pred)**2)))<jupyter_output><empty_output><jupyter_text>Train-Test Split<jupyter_code>train_df_cluster = []
val_df_cluster = []
test_df_cluster = []
num_of_cluster = 8
for cluster_id in range(num_of_cluster):
    train_df = df_feature.where((df_feature["cluster_id"] == cluster_id) & (df_feature["date"] <= date(2013,10,15)))
    val_df = df_feature.where((df_feature["cluster_id"] == cluster_id) & (df_feature["date"] <= date(2013,10,31)) & (df_feature["date"] >= date(2013,10,15)))
    test_df = df_feature.where((df_feature["cluster_id"] == cluster_id) & (df_feature["date"] > date(2013,10,31)))# & (df_feature["date"] <= date(2013,1,2)))
    print("Train_point = {}, Test_point = {}, cId = {}".format(train_df.count(),test_df.count(),cluster_id))
    train_df_cluster.append(train_df)
    val_df_cluster.append(val_df)
    test_df_cluster.append(test_df)<jupyter_output>Train_point = 6904, Test_point = 1464, cId = 0
Train_point = 6904, Test_point = 1464, cId = 1
Train_point = 6904, Test_point = 1464, cId = 2
Train_point = 6904, Test_point = 1464, cId = 3
Train_point = 6904, Test_point = 1464, cId = 4
Train_point = 6904, Test_point = 1464, cId = 5
Train_point = 6904, Test_point = 1464, cId = 6
Train_point = 6904, Test_point = 1464, cId = 7<jupyter_text>RF<jupyter_code>pred_val = []
#val_pred_val = []
flag = 0
for cluster_id in range(num_of_cluster):
    rf = RandomForestRegressor(numTrees=20,maxDepth=10,maxBins=128,seed=4)
    model = rf.fit(train_df_cluster[cluster_id])
    rf_pred_val = model.transform(test_df_cluster[cluster_id])
    #rf_val_pred_val = model.transform(val_df_cluster[cluster_id])
    if flag == 0:
        #val_pred_val = sqlcontext.createDataFrame([],rf_val_pred_val.schema)
        pred_val = sqlcontext.createDataFrame([],rf_pred_val.schema)
        flag = 1
    pred_val = pred_val.union(rf_pred_val)
    pred_val = pred_val.cache()
    #val_pred_val = val_pred_val.union(rf_val_pred_val)
    #val_pred_val = val_pred_val.cache()<jupyter_output><empty_output><jupyter_text>GBT<jupyter_code>pred_val = []
val_pred_val = []
flag = 0
for cluster_id in range(num_of_cluster):
    gbt = GBTRegressor(maxBins=128,maxDepth=10)
    gbt_model = gbt.fit(train_df_cluster[cluster_id])
    gbt_pred_val = model.transform(test_df_cluster[cluster_id])
    gbt_val_pred_val = model.transform(val_df_cluster[cluster_id])
    if flag == 0:
        val_pred_val = sqlcontext.createDataFrame([],gbt_val_pred_val.schema)
        pred_val = sqlcontext.createDataFrame([],gbt_pred_val.schema)
        flag = 1
    pred_val = pred_val.union(gbt_pred_val)
    pred_val = pred_val.cache()
    val_pred_val = val_pred_val.union(gbt_val_pred_val)
    val_pred_val = val_pred_val.cache()
val_pred_val = val_pred_val.select("date","hour","cluster_id","prediction","label")
val_pred_val.printSchema()
#pred_val = pred_val.select("date","hour","cluster_id","prediction","label").orderBy("cluster_id")
#pred_val.printSchema()
LR_pivot = val_pred_val.groupBy("date","hour").pivot("cluster_id").agg(first("prediction"))
#LR_pivot = pred_val.groupBy("date","hour").pivot("cluster_id").agg(first("prediction"))
LR_pivot.printSchema()
pred_val.where((pred_val["date"] == date(2013,11,10)) & (pred_val["hour"] == 23)).show()
LR_pivot.where((LR_pivot["date"] == date(2013,11,10)) & (LR_pivot["hour"] == 23)).show(2)

LR_actual = val_pred_val.groupBy("date","hour").sum("label")
#LR_actual = pred_val.groupBy("date","hour").sum("label")
LR_actual.printSchema()
LR_feature = LR_pivot.join(LR_actual,["date","hour"])
LR_feature = LR_feature.withColumnRenamed("sum(label)","label")
LR_feature.printSchema()
feature_col = LR_feature.columns
feature_col = feature_col[2:10]
feature_col
vecAssembler = VectorAssembler(inputCols=feature_col, outputCol="features")
df_feature = vecAssembler.transform(LR_feature)
df_feature.where((df_feature["date"] == date(2013,11,10)) & (df_feature["hour"] == 23)).show()
scaler = StandardScaler(withMean=True,inputCol="features",outputCol="scaled_features")
df_feature = scaler.fit(df_feature).transform(df_feature)
df_feature = df_feature.drop("features")
df_feature = df_feature.withColumnRenamed("scaled_features","features")
# from keras.models import Sequential
# from keras.layers.core import Dense, Dropout
# from elephas.ml_model import ElephasEstimator
# model = Sequential()
# model.add(Dense(128,input_dim=8,activation="relu"))
# model.add(Dense(64,activation="relu"))
# model.add(Dense(1,activation="linear"))
# model.compile(loss="mse",optimizer="adam",metrics=["mae","mse"])
# model.summary()
# from elephas.ml_model import ElephasEstimator
# estimator = ElephasEstimator(model,epochs=50,batch_size=32)
# # lrModel = estimator.fit(df_feature)
# # from elephas.spark_model import SparkModel
# # spark_model = SparkModel(model,frequency="epoch")
# # spark_model.fit(df_feature.select("features","label").rdd,epochs=20,batch_size=32)
lr = LinearRegression(solver="auto",loss = "huber")
lrModel = lr.fit(df_feature)
df_feature.take(1)<jupyter_output><empty_output><jupyter_text>Perform CLuster+RF+LR<jupyter_code>def get_aggregate_LR(pred_val,lrModel):
    pred_val = pred_val.select("date","hour","cluster_id","prediction","label")
    LR_pivot = pred_val.groupBy("date","hour").pivot("cluster_id").agg(first("prediction"))
    LR_actual = pred_val.groupBy("date","hour").sum("label")
    LR_feature = LR_pivot.join(LR_actual,["date","hour"])
    LR_feature = LR_feature.withColumnRenamed("sum(label)","label")
    feature_col = LR_feature.columns
    feature_col = feature_col[2:10]
    vecAssembler = VectorAssembler(inputCols=feature_col, outputCol="features")
    df_feature = vecAssembler.transform(LR_feature)
    scaler = StandardScaler(withMean=True,inputCol="features",outputCol="scaled_features")
    df_feature = scaler.fit(df_feature).transform(df_feature)
    df_feature = df_feature.drop("features")
    df_feature = df_feature.withColumnRenamed("scaled_features","features")
    lr_prediction = lrModel.transform(df_feature)
    return lr_prediction

def weighted_avg(predval):
    predval = predval.withColumn("prediction",8*(predval["0"]*(801/3930)+predval["1"]*(940/3930)+predval["2"]*(738/3930)+predval["3"]*(652/3930)+predval["4"]*(501/3930)+predval["5"]*(209/3930)+predval["6"]*(75/3930)+predval["7"]*(14/3930)))
    return predval
aggregate_df = get_aggregate_LR(pred_val,lrModel)
#aggregate_df = weighted_avg(df_feature)
#print(lrModel.coefficients)
aggregate_df.printSchema()
# aggregate_df = aggregate_df.withColumnRenamed("sum(label)","label")
# aggregate_df = aggregate_df.withColumnRenamed("sum(prediction)","prediction")
aggregate_df.where((aggregate_df["date"] == date(2013,11,10)) & (aggregate_df["hour"]==23)).show(5)
train_start_date = date(2013,11,1)
train_end_date = date(2013,12,31)
y_date = []
Mape_date = []
rmse_date = []
while train_start_date <= train_end_date:
    print(train_start_date)
    y_actual,y_pred = select_predicted_actual(aggregate_df,train_start_date)
    if len(y_actual) == 0:
        train_start_date = train_start_date + timedelta(1)
        continue
    Mape_date.append(mean_absolute_percentage_error(y_actual,y_pred))
    rmse_date.append(root_mean_squared_error(y_actual,y_pred))
    y_date.append(train_start_date)
    train_start_date = train_start_date + timedelta(1)
#del y_date[2]
# fig, (ax1,ax2) = plt.subplots(1,2, figsize =(8,6))
# ax1.xaxis.set_major_formatter(mdates.DateFormatter('%m/%d/%Y'))
# ax1.xaxis.set_major_locator(mdates.DayLocator())
# ax2.xaxis.set_major_formatter(mdates.DateFormatter('%m/%d/%Y'))
# ax2.xaxis.set_major_locator(mdates.DayLocator())
# ax1.plot(y_date,Mape_date)
# ax2.plot(y_date,rmse_date)
# fig.autofmt_xdate()
# ax1.set_xlabel('k')
# ax1.set_ylabel('cost')
date_time = pd.to_datetime(y_date)
DF = pd.DataFrame()
DF['Mape_date'] = Mape_date
DF = DF.set_index(date_time)

DF1 = pd.DataFrame()
DF1['rmse_date'] = rmse_date
DF1 = DF1.set_index(date_time)
fig, (ax1,ax2) = plt.subplots(1,2, figsize =(8,6))
fig.subplots_adjust(bottom=0.3,wspace = 0.3)
error = ["MAPE","RMSE"]
cnt = 0
for ax in fig.axes:
    plt.sca(ax)
    plt.xticks(rotation=90)
    plt.xlabel("Day")
    plt.ylabel(error[cnt])
    cnt += 1
ax1.plot(DF)
ax2.plot(DF1)
from statistics import mean
print("Mean RMSE = {}, Mean Mape = {}".format(mean(rmse_date),mean(Mape_date)))
actual_predicted = aggregate_df.where((aggregate_df.date==date(2013,11,29))).select("label","prediction").toPandas()
actual_predicted.head()
ax = actual_predicted.plot()
ax.set_xlabel("Hour of Day")
ax.set_ylabel("Load (Kw/h)")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.16303317535545023, 'fraction_numerical': 0.03401790416008425, 'mean_word_length': 6.809210526315789, 'pattern_counts': {'":': 0, '<': 24, '<?xml version=': 0, '>': 22, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '156087', 'license': 'Apache-2.0', 'n_tokens_mistral': 3853, 'n_tokens_neox': 3583, 'n_words': 452, 'path': 'Novel_cluster_forecasting.ipynb', 'repo_name': 'Ayush-IITP/London_Smart_Meter_Load_Forecasting', 'size': '9495'}