<jupyter_start><jupyter_text>Bitcoin Price Prediction - Global Mode * **Description**: COMP4103(Big Data)--Group Project* **Author**: Aaron* **Version**: 1.2 (The whole workflow with results) **Updates:**1. Add the final results **Issues:**N/A **To be done**:1. Statistical Analysis of Residuals - plot histogram of residuals and a QQplot to check for normality(Linear Regression) **Questions:**N/A **Future Work:**1. The **autoTuning()** and **tsCrossValidation()** cost much time because the procedure is not distributed. The possible way is to overwrite the following code.    * CrossValidator: https://spark.apache.org/docs/3.1.1/api/python/_modules/pyspark/ml/tuning.htmlCrossValidator    * TrainValidationSplit: https://spark.apache.org/docs/3.1.1/api/python/_modules/pyspark/ml/tuning.htmlTrainValidationSplit2. Could add more algorithms of sklearn by Spark UDFs.    * Apache Arrow: https://spark.apache.org/docs/latest/api/python/user_guide/arrow_pandas.html3. Could consider Linear model Trees: https://medium.com/convoy-tech/the-best-of-both-worlds-linear-model-trees-7c9ce139767d ****Unified DataSet Format****    **TimeZone**: `UTC`1. `id`: Represent the order of this rows2. `Timestamp` : Represent the time of this record3. `Close`: Represent the original close price without shifting4. `NEXT_BTC_CLOSE`: Represent the close price with shifting. It's our label.  1. Data loading and Preprocessing  1.1. Load related packages<jupyter_code># Apache Spark
from pyspark.sql import SparkSession
import pyspark.sql.functions as F

from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler,StandardScaler
from pyspark.ml.regression import LinearRegression, GeneralizedLinearRegression, DecisionTreeRegressor, RandomForestRegressor, GBTRegressor
from pyspark.ml.evaluation import RegressionEvaluator

# Python
import numpy as np
import pandas as pd
from itertools import product
import time

# Graph packages
# https://plotly.com/python/getting-started/#jupyterlab-support
# https://plotly.com/python/time-series/
import plotly.express as px

# Scikit-learn
from sklearn.metrics import mean_absolute_percentage_error<jupyter_output><empty_output><jupyter_text>1.2. Create a SparkSession<jupyter_code># Start a SparkSession
spark = SparkSession \
    .builder \
    .master("local[*]") \
    .appName("Bitcoin Prediction") \
    .getOrCreate()

sc = spark.sparkContext
# Show the URL of SparkSession
spark<jupyter_output><empty_output><jupyter_text>1.3. DataSet Checking<jupyter_code># Read csv file
filename = "bitcoin_10y_1min_interpolate.csv"

dataset = spark.read.format("csv") \
          .option("inferSchema",'True') \
          .option("header",True) \
          .load(filename)
# Check the schema of loaded dataSet
dataset.printSchema()
# Check the num of partitions
dataset.rdd.getNumPartitions()
# Have a look on the original and shifted label column
dataset.select("id","Timestamp","Close","NEXT_BTC_CLOSE").tail(5)
# Check if there are any "nan", "Null" or empty in numerical columns
all_col = dataset.columns
#dataset.select([F.count(F.when(F.isnan(cols) | dataset[cols].isNull() | (dataset[cols] == ""), cols)).alias(cols) for cols in all_col]).show()
# Total rows
dataset.count()<jupyter_output><empty_output><jupyter_text>2. Feature Engineering  2.1. Transform to MLlib required format<jupyter_code># labels and features
feature_cols = dataset.columns
# Gain the column list of features
non_feature_cols  = ['id',"NEXT_BTC_CLOSE",'Timestamp']
[feature_cols.remove(non_feature) for non_feature in non_feature_cols]
feature_cols
# Form a column contains all the features - VectorAssembler
vector_assembler = VectorAssembler(inputCols=feature_cols, outputCol="features")<jupyter_output><empty_output><jupyter_text>2.2. Feature Standardization<jupyter_code># Standardize the "features" column
# Use this column "scaledFeatures" if the algorithm didn't have the build-in param: Standardization
standard_scaler = StandardScaler(inputCol="features", outputCol="scaledFeatures", withStd=True, withMean=False)<jupyter_output><empty_output><jupyter_text>2.3. Train-test Split(Time series)<jupyter_code>'''
Description: Split and keep the original time-series order
Args:
    dataSet: The dataSet which needs to be splited
    proportion: A number represents the split proportion

Return: 
    train_data: The train dataSet
    test_data: The test dataSet
'''
def trainSplit(dataSet, proportion):
    records_num = dataset.count()
    split_point = round(records_num * proportion)
    
    train_data = dataset.filter(F.col("id") < split_point)
    test_data = dataset.filter(F.col("id") >= split_point)
    
    return (train_data,test_data)
# Split the dataSet: Train(70%), test(30%)
proportion = 0.7
train_data,test_data = trainSplit(dataset, proportion)

# Cache it
train_data.cache()
test_data.cache()

# Number of train and test dataSets
print(f"Training data: {train_data.count()}\nTest data: {test_data.count()}")<jupyter_output>Training data: 3399486
Test data: 1456922<jupyter_text>3. Model Building  3.1. LinearRegression(Simple workflow demo)<jupyter_code># Train a LinearRegression model
lr = LinearRegression(featuresCol="features",labelCol="NEXT_BTC_CLOSE",maxIter=5, regParam=0.0, elasticNetParam=0.8)
# Chain assembler and forest in a Pipeline
pipeline = Pipeline(stages=[vector_assembler, lr])
pipeline_model = pipeline.fit(train_data)
# Make predictions
predictions = pipeline_model.transform(test_data)
predictions.select("Timestamp","NEXT_BTC_CLOSE","prediction").show(5)
# Compute test error
rmse_evaluator = RegressionEvaluator(labelCol="NEXT_BTC_CLOSE", predictionCol="prediction", metricName='rmse')
mae_evaluator = RegressionEvaluator(labelCol="NEXT_BTC_CLOSE", predictionCol="prediction", metricName='mae')
r2_evaluator = RegressionEvaluator(labelCol="NEXT_BTC_CLOSE", predictionCol="prediction", metricName='r2')
var_evaluator = RegressionEvaluator(labelCol="NEXT_BTC_CLOSE", predictionCol="prediction", metricName='var')

rmse = rmse_evaluator.evaluate(predictions)
mae = mae_evaluator.evaluate(predictions)
var = var_evaluator.evaluate(predictions)
r2 = r2_evaluator.evaluate(predictions)

predictions_pd = predictions.select("NEXT_BTC_CLOSE","prediction").toPandas()
mape = mean_absolute_percentage_error(predictions_pd["NEXT_BTC_CLOSE"], predictions_pd["prediction"])

# Adjusted R-squared
n = predictions.count()
p = len(predictions.columns)
adj_r2 = 1-(1-r2)*(n-1)/(n-p-1)

# Use dict to store each result
results = {
    "Model": "Linear Regression",
    "Proportion": proportion,
    "RMSE": rmse,
    "MAPE": mape,
    "MAE": mae,
    "Variance": var,
    "R2": r2,
    "Adjusted_R2": adj_r2,
}

results
# Release Cache
train_data.unpersist()
test_data.unpersist()<jupyter_output><empty_output><jupyter_text>4. Parameter Tuning<jupyter_code>'''
Description: Use Grid Search to tune the Model 
Args:
    dataSet: The dataSet which needs to be splited
    proportion_lst: A list represents the split proportion
    feature_col: The column name of features
    label_col: The column name of label
    ml_model: The module to use
    params: Parameters which want to test 
    assembler: An assembler to dataSet
    scaler: A scaler to dataSet
Return: 
    results_df: The best result in a pandas dataframe
'''
def autoTuning(dataSet, proportion_lst, feature_col, label_col, ml_model, params, assembler, scaler):
    
    # Initialize the best result for comparison
    result_best = {"RMSE": float('inf')}
    
    # Try different proportions 
    for proportion in proportion_lst:
        # Split the dataSet
        train_data,test_data = trainSplit(dataSet, proportion)
    
        # Cache it
        train_data.cache()
        test_data.cache()
    
        # ALL combination of params
        param_lst = [dict(zip(params, param)) for param in product(*params.values())]
    
        for param in param_lst:
            # Chosen Model
            if ml_model == "LinearRegression":
                model = LinearRegression(featuresCol=feature_col, \
                                         labelCol=label_col, \
                                         maxIter=param['maxIter'], \
                                         regParam=param['regParam'], \
                                         elasticNetParam=param['elasticNetParam'])
            
            elif ml_model == "GeneralizedLinearRegression":
                model = GeneralizedLinearRegression(featuresCol=feature_col, \
                                                    labelCol=label_col, \
                                                    maxIter=param['maxIter'], \
                                                    regParam=param['regParam'], \
                                                    family=param['family'], \
                                                    link=param['link'])
            
            elif ml_model == "DecisionTree":
                model = DecisionTreeRegressor(featuresCol=feature_col, \
                                              labelCol=label_col, \
                                              maxDepth = param["maxDepth"], \
                                              seed=0)
            
            elif ml_model == "RandomForest":
                model = RandomForestRegressor(featuresCol=feature_col, \
                                              labelCol=label_col, \
                                              numTrees = param["numTrees"], \
                                              maxDepth = param["maxDepth"], \
                                              seed=0)
            
            elif ml_model == "GBTRegression":
                model = GBTRegressor(featuresCol=feature_col, \
                                     labelCol=label_col, \
                                     maxIter = param['maxIter'], \
                                     maxDepth = param['maxDepth'], \
                                     stepSize = param['stepSize'], \
                                     seed=0)
            
            # Chain assembler and model in a Pipeline
            pipeline = Pipeline(stages=[assembler, model])
            # Train a model and calculate running time
            start = time.time()
            pipeline_model = pipeline.fit(train_data)
            end = time.time()

            # Make predictions
            predictions = pipeline_model.transform(test_data)

            # Compute test error by several evaluators
            # https://spark.apache.org/docs/3.1.1/mllib-evaluation-metrics.html#regression-model-evaluation
            # https://spark.apache.org/docs/3.1.1/api/scala/org/apache/spark/ml/evaluation/RegressionEvaluator.html
            rmse_evaluator = RegressionEvaluator(labelCol=label_col, predictionCol="prediction", metricName='rmse')
            mae_evaluator = RegressionEvaluator(labelCol=label_col, predictionCol="prediction", metricName='mae')
            r2_evaluator = RegressionEvaluator(labelCol=label_col, predictionCol="prediction", metricName='r2')
            var_evaluator = RegressionEvaluator(labelCol=label_col, predictionCol="prediction", metricName='var')
            
            predictions_pd = predictions.select("NEXT_BTC_CLOSE","prediction").toPandas()
            mape = mean_absolute_percentage_error(predictions_pd["NEXT_BTC_CLOSE"], predictions_pd["prediction"])
            
            rmse = rmse_evaluator.evaluate(predictions)
            mae = mae_evaluator.evaluate(predictions)
            var = var_evaluator.evaluate(predictions)
            r2 = r2_evaluator.evaluate(predictions)
            # Adjusted R-squared
            n = predictions.count()
            p = len(predictions.columns)
            adj_r2 = 1-(1-r2)*(n-1)/(n-p-1)
        
            # Use dict to store each result
            results = {
                "Model": ml_model,
                "Proportion": proportion,
                "Parameters": [list(param.values())],
                "RMSE": rmse,
                "MAPE":mape,
                "MAE": mae,
                "Variance": var,
                "R2": r2,
                "Adjusted_R2": adj_r2,
                "Time": end - start,
                "Predictions": predictions.select("NEXT_BTC_CLOSE","prediction",'Timestamp')
            }
            
            # Only store the lowest RMSE
            if results['RMSE'] < result_best['RMSE']:
                result_best = results
                
        # Release Cache
        train_data.unpersist()
        test_data.unpersist()
        
    # Transform dict to pandas dataframe
    results_df = pd.DataFrame(result_best)
    return results_df
# Define a function to plot line-like graph
# https://plotly.com/python/time-series/#time-series-with-range-selector-buttons
'''
Description: Plot the line graph by plotly(custom design)
Args:
    data: The data(pandas dataframe) which you want to ploy by line
    graph_title: The title of the graph
Return: None
'''
def line_plot(data,graph_title):
    plot = px.line(data,title=graph_title)
    plot.update_xaxes(
        rangeslider_visible=True,
        rangeselector=dict(
            buttons=list([
                dict(count=7, label="1w", step="day", stepmode="backward"),
                dict(count=1, label="1m", step="month", stepmode="backward"),
                dict(count=6, label="6m", step="month", stepmode="backward"),
                dict(count=1, label="1y", step="year", stepmode="backward"),
                dict(step="all")
            ])
        )
    )
    plot.show()
# Draw the Prediction Graph
'''
Description: Plot a line graph of the Prediction
Args:
    result: The result from the "autoTuning" Func
    graph_title: The title of the graph
Return: None
'''
def drawPrediction(result, graph_title):
    result_pd = result['Predictions'][0].withColumn("Time", F.to_timestamp("Timestamp", 'yyyy-MM-dd HH:mm:ss')) \
                                        .drop("Timestamp") \
                                        .toPandas() \
                                        .set_index('Time')
    # Display the info of the best trained Model
    print(result.iloc[0,:-1])
    # Draw by plotly
    line_plot(result_pd, graph_title)
# Parameter choosing - only use the most popular/useful params
# https://spark.apache.org/docs/3.1.1/ml-classification-regression.html#regression
# The explanations of params are form the above document
# Split proportion list
proportion_lst = [0.6, 0.7, 0.8, 0.9]<jupyter_output><empty_output><jupyter_text>4.1. Linear Regression Model - Prediction<jupyter_code># LinearRegression
lr_params = {
    'maxIter' : [5, 10, 50, 80, 100], # max number of iterations (>=0), default:100
    'regParam' : np.arange(0,1,0.2).round(decimals=2),# regularization parameter (>=0), default:0.0
    'elasticNetParam' : np.arange(0,1,0.2).round(decimals=2) # the ElasticNet mixing parameter, [0, 1], default:0.0 
}
result_lr = autoTuning(dataset, proportion_lst, "features", "NEXT_BTC_CLOSE", "LinearRegression", lr_params, vector_assembler ,standard_scaler)
# Virtualization
drawPrediction(result_lr,"Predict by LinearRegression")
# Generalized linear regression
glr_params = {
    'maxIter' : [5, 10, 50, 80], # max number of iterations (>=0), default:25
    'regParam' : [0, 0.1, 0.2],# regularization parameter (>=0), default:0.0
    'family': ['gaussian', 'gamma'], # The name of family which is a description of the error distribution to be used in the model.
    'link': ['identity', 'inverse'] # which provides the relationship between the linear predictor and the mean of the distribution function.
}
result_glr = autoTuning(dataset, proportion_lst, "features", "NEXT_BTC_CLOSE", "GeneralizedLinearRegression", glr_params, vector_assembler ,standard_scaler)
# Virtualization
drawPrediction(result_glr,"Predict by GeneralizedLinearRegression")<jupyter_output><empty_output><jupyter_text>4.2. Tree based Model - Prediction<jupyter_code># RandomForest
rf_params = {
    'numTrees' : [3, 5, 10, 20, 30],# Number of trees to train, >=1, default:20
    'maxDepth' : [3, 5, 10] # Maximum depth of the tree, <=30, default:5
}
result_rf = autoTuning(dataset, proportion_lst, "features", "NEXT_BTC_CLOSE", "RandomForest", rf_params, vector_assembler ,standard_scaler)
# Virtualization
drawPrediction(result_rf,"Predict by RandomForest")
# GBTRegression
gb_params = {
    'maxIter' : [20, 40, 60], # max number of iterations (>=0), default:20
    'maxDepth' : [5, 8, 10], # Maximum depth of the tree (>=0), <=30, default:5
    'stepSize': [0.1, 0.3, 0.5, 0.7] # learning rate, [0,1], default:0.1
}
result_gb = autoTuning(dataset, proportion_lst, "features", "NEXT_BTC_CLOSE", "GBTRegression", gb_params, vector_assembler ,standard_scaler)
# Virtualization
drawPrediction(result_gb,"Predict by GBTRegression")<jupyter_output>Model          GBTRegression
Proportion               0.7
Parameters      [40, 5, 0.3]
RMSE                 8090.42
MAPE                0.128075
MAE                  2977.23
Variance         2.78414e+07
R2                  0.486053
Adjusted_R2         0.486048
Time                 168.555
Name: 0, dtype: object<jupyter_text>5. Time Series Cross Validation > https://machinelearningmastery.com/backtest-machine-learning-models-time-series-forecasting/     > https://hub.packtpub.com/cross-validation-strategies-for-time-series-forecasting-tutorial/        > https://scikit-learn.org/stable/modules/generated/sklearn.model_selection.TimeSeriesSplit.html     5.1. Multiple Splits Time Series Cross Validation<jupyter_code>'''
Description: Multiple Splits Cross Validation on Time Series data
Args:
    num: Number of DataSet
    n_splits: Split times
Return: 
    split_position_df: All set of splits position in a Pandas dataframe
'''
def mulTsCrossValidation(num, n_splits):
    split_position_lst = []
    # Calculate the split position for each time 
    for i in range(1, n_splits+1):
        # Calculate train size and test size
        train_size = i * num // (n_splits + 1) + num % (n_splits + 1)
        test_size = num //(n_splits + 1)

        # Calculate the start/split/end point for each fold
        start = 0
        split = train_size
        end = train_size + test_size
        
        # Avoid to beyond the whole number of dataSet
        if end > num:
            end = num
        split_position_lst.append((start,split,end))
        
    # Transform the split position list to a Pandas Dataframe
    split_position_df = pd.DataFrame(split_position_lst,columns=['start','split','end'])
    return split_position_df<jupyter_output><empty_output><jupyter_text>5.2. Blocked Time Series Cross Validation<jupyter_code>'''
Description: Blocked Time Series Cross Validation
Args:
    num: Number of DataSet
    n_splits: Split times
Return: 
    split_position_df: All set of splits position in a Pandas dataframe
'''
def blockedTsCrossValidation(num, n_splits):
    kfold_size = num // n_splits

    split_position_lst = []
    # Calculate the split position for each time 
    for i in range(n_splits):
        # Calculate the start/split/end point for each fold
        start = i * kfold_size
        end = start + kfold_size
        # Manually set train-test split proportion in each fold
        split = int(0.8 * (end - start)) + start
        split_position_lst.append((start,split,end))
        
    # Transform the split position list to a Pandas Dataframe
    split_position_df = pd.DataFrame(split_position_lst,columns=['start','split','end'])
    return split_position_df<jupyter_output><empty_output><jupyter_text>5.3. Walk Forward Validation<jupyter_code>'''
Description: Walk Forward Validation on Time Series data
Args:
    num: Number of DataSet
    min_obser: Minimum Number of Observations
    expand_window: Sliding or Expanding Window
Return: 
    split_position_df: All set of splits position in a Pandas dataframe
'''
def wfTsCrossValidation(num, min_obser, expand_window):
    split_position_lst = []
    # Calculate the split position for each time 
    for i in range(min_obser,num,expand_window):
        # Calculate the start/split/end point for each fold
        start = 0
        split = i
        end = split + expand_window
        
        # Avoid to beyond the whole number of dataSet
        if end > num:
            end = num
        split_position_lst.append((start,split,end))
        
    # Transform the split position list to a Pandas Dataframe
    split_position_df = pd.DataFrame(split_position_lst,columns=['start','split','end'])
    return split_position_df<jupyter_output><empty_output><jupyter_text>5.4. Time Series Cross Validation<jupyter_code>'''
Description: Cross Validation on Time Series data
Args:
    dataSet: The dataSet which needs to be splited
    feature_col: The column name of features
    label_col: The column name of label
    ml_model: The module to use
    params: Parameters which want to test 
    assembler: An assembler to dataSet
    scaler: A scaler to dataSet
    cv_info: The type of Cross Validation
Return: 
    tsCv_df: All the splits performance of each model in a pandas dataframe
'''
def tsCrossValidation(dataSet, feature_col, label_col, ml_model, params, assembler, scaler, cv_info):
    
    # Get the number of samples
    num = dataSet.count()
    
    # Save results in a list
    result_lst = []
    
    # ALL combination of params
    param_lst = [dict(zip(params, param)) for param in product(*params.values())]

    for param in param_lst:
        # Chosen Model
        if ml_model == "LinearRegression":
            model = LinearRegression(featuresCol=feature_col, \
                                     labelCol=label_col, \
                                     maxIter=param['maxIter'], \
                                     regParam=param['regParam'], \
                                     elasticNetParam=param['elasticNetParam'])

        elif ml_model == "GeneralizedLinearRegression":
            model = GeneralizedLinearRegression(featuresCol=feature_col, \
                                                labelCol=label_col, \
                                                maxIter=param['maxIter'], \
                                                regParam=param['regParam'], \
                                                family=param['family'], \
                                                link=param['link'])

        elif ml_model == "DecisionTree":
            model = DecisionTreeRegressor(featuresCol=feature_col, \
                                          labelCol=label_col, \
                                          maxDepth = param["maxDepth"], \
                                          seed=0)

        elif ml_model == "RandomForest":
            model = RandomForestRegressor(featuresCol=feature_col, \
                                          labelCol=label_col, \
                                          numTrees = param["numTrees"], \
                                          maxDepth = param["maxDepth"], \
                                          seed=0)

        elif ml_model == "GBTRegression":
            model = GBTRegressor(featuresCol=feature_col, \
                                 labelCol=label_col, \
                                 maxIter = param['maxIter'], \
                                 maxDepth = param['maxDepth'], \
                                 stepSize = param['stepSize'], \
                                 seed=0)
            
        
        
        # Identify the type of Cross Validation 
        if cv_info['cv_type'] == 'mulTs':
            split_position_df = mulTsCrossValidation(num, cv_info['kSplits'])
        elif cv_info['cv_type'] == 'blkTs':
            split_position_df = blockedTsCrossValidation(num, cv_info['kSplits'])
        elif cv_info['cv_type'] == 'wfTs':
            split_position_df = wfTsCrossValidation(num, cv_info['min_obser'], cv_info['expand_window'])
            

        for position in split_position_df.itertuples():
            # Get the start/split/end position from a kind of Time Series Cross Validation
            start = getattr(position, 'start')
            splits = getattr(position, 'split')
            end = getattr(position, 'end')
            idx  = getattr(position, 'Index')
            
            # Train/Test size
            train_size = splits - start
            test_size = end - splits

            # Get training data and test data
            train_data = dataSet.filter(F.col("id").between(start, splits-1))
            test_data = dataSet.filter(F.col("id").between(splits, end-1))

            # Cache it
            train_data.cache()
            test_data.cache()

            # Chain assembler and model in a Pipeline
            pipeline = Pipeline(stages=[assembler, model])
            # Train a model and calculate running time
            start = time.time()
            pipeline_model = pipeline.fit(train_data)
            end = time.time()

            # Make predictions
            predictions = pipeline_model.transform(test_data)

            # Compute test error by several evaluator
            rmse_evaluator = RegressionEvaluator(labelCol=label_col, predictionCol="prediction", metricName='rmse')
            mae_evaluator = RegressionEvaluator(labelCol=label_col, predictionCol="prediction", metricName='mae')
            r2_evaluator = RegressionEvaluator(labelCol=label_col, predictionCol="prediction", metricName='r2')
            var_evaluator = RegressionEvaluator(labelCol=label_col, predictionCol="prediction", metricName='var')
            
            predictions_pd = predictions.select("NEXT_BTC_CLOSE","prediction").toPandas()
            mape = mean_absolute_percentage_error(predictions_pd["NEXT_BTC_CLOSE"], predictions_pd["prediction"])

            rmse = rmse_evaluator.evaluate(predictions)
            mae = mae_evaluator.evaluate(predictions)
            var = var_evaluator.evaluate(predictions)
            r2 = r2_evaluator.evaluate(predictions)
            # Adjusted R-squared
            n = predictions.count()
            p = len(predictions.columns)
            adj_r2 = 1-(1-r2)*(n-1)/(n-p-1)

            # Use dict to store each result
            results = {
                "Model": ml_model,
                'CV_type': cv_info['cv_type'],
                "Splits": idx + 1,
                "Train&Test": (train_size,test_size),
                "Parameters": list(param.values()),
                "RMSE": rmse,
                "MAPE": mape,
                "MAE": mae,
                "Variance": var,
                "R2": r2,
                "Adjusted_R2": adj_r2,
                "Time": end - start
            }
            
            # Store each splits result
            result_lst.append(results)
            
            # Release Cache
            train_data.unpersist()
            test_data.unpersist()

    # Transform dict to pandas dataframe
    tsCv_df = pd.DataFrame(result_lst)
    return tsCv_df
## Cross Validation Parameter
# Multiple Splits Time Series Cross Validation
mul_cv = {'cv_type':'mulTs',
          'kSplits': 5}

# Blocked Time Series Cross Validation
blk_cv = {'cv_type':'blkTs',
          'kSplits': 10}

# Walk Forward Validation, Last 50 steps
wf_cv = {'cv_type':'wfTs',
         'min_obser': 4856359,
         'expand_window': 1}<jupyter_output><empty_output><jupyter_text>5.4.1. LinearRegression Validation<jupyter_code># LinearRegression
lr_params = {
    'maxIter' : [5], # max number of iterations (>=0), default:100
    'regParam' : [0.0],# regularization parameter (>=0), default:0.0
    'elasticNetParam' : [0.8] # the ElasticNet mixing parameter, [0, 1], default:0.0 
}
lr_mul_cv = tsCrossValidation(dataset, "features", "NEXT_BTC_CLOSE", "LinearRegression", lr_params, vector_assembler ,standard_scaler, mul_cv)
lr_mul_cv
lr_blk_cv = tsCrossValidation(dataset, "features", "NEXT_BTC_CLOSE", "LinearRegression", lr_params, vector_assembler ,standard_scaler, blk_cv)
lr_blk_cv
lr_wf_cv = tsCrossValidation(dataset, "features", "NEXT_BTC_CLOSE", "LinearRegression", lr_params, vector_assembler ,standard_scaler, wf_cv)
lr_wf_cv<jupyter_output><empty_output><jupyter_text>5.4.2. Generalized linear regression Validation<jupyter_code># Generalized linear regression
glr_params = {
    'maxIter' : [10], # max number of iterations (>=0), default:25
    'regParam' : [0],# regularization parameter (>=0), default:0.0
    'family': ['gaussian'], # The name of family which is a description of the error distribution to be used in the model.
    'link': ['identity'] # which provides the relationship between the linear predictor and the mean of the distribution function.
}
glr_mul_cv = tsCrossValidation(dataset, "features", "NEXT_BTC_CLOSE", "GeneralizedLinearRegression", glr_params, vector_assembler ,standard_scaler, mul_cv)
glr_mul_cv
glr_blk_cv = tsCrossValidation(dataset, "features", "NEXT_BTC_CLOSE", "GeneralizedLinearRegression", glr_params, vector_assembler ,standard_scaler, blk_cv)
glr_blk_cv
glr_wf_cv = tsCrossValidation(dataset, "features", "NEXT_BTC_CLOSE", "GeneralizedLinearRegression", glr_params, vector_assembler ,standard_scaler, wf_cv)
glr_wf_cv<jupyter_output><empty_output><jupyter_text>5.4.3. RandomForest Validation<jupyter_code>rf_params = {
    'numTrees' : [5],# Number of trees to train, >=1, default:20
    'maxDepth' : [10] # Maximum depth of the tree, <=30, default:5
}
rf_mul_cv = tsCrossValidation(dataset, "features", "NEXT_BTC_CLOSE", "RandomForest", rf_params, vector_assembler ,standard_scaler, mul_cv)
rf_mul_cv
rf_blk_cv = tsCrossValidation(dataset, "features", "NEXT_BTC_CLOSE", "RandomForest", rf_params, vector_assembler ,standard_scaler, blk_cv)
rf_blk_cv
rf_wf_cv = tsCrossValidation(dataset, "features", "NEXT_BTC_CLOSE", "RandomForest", rf_params, vector_assembler ,standard_scaler, wf_cv)
rf_wf_cv<jupyter_output><empty_output><jupyter_text>5.4.4. GBTRegression Validation<jupyter_code>gb_params = {
    'maxIter' : [40], # max number of iterations (>=0), default:20
    'maxDepth' : [5], # Maximum depth of the tree (>=0), <=30, default:5
    'stepSize': [0.3] # learning rate, [0,1], default:0.1
}
gb_mul_cv = tsCrossValidation(dataset, "features", "NEXT_BTC_CLOSE", "GBTRegression", gb_params, vector_assembler ,standard_scaler, mul_cv)
gb_mul_cv
gb_blk_cv = tsCrossValidation(dataset, "features", "NEXT_BTC_CLOSE", "GBTRegression", gb_params, vector_assembler ,standard_scaler, blk_cv)
gb_blk_cv
gb_wf_cv = tsCrossValidation(dataset, "features", "NEXT_BTC_CLOSE", "GBTRegression", gb_params, vector_assembler ,standard_scaler, wf_cv)
gb_wf_cv<jupyter_output><empty_output><jupyter_text>6.Summary    6.1. Model Comparison Table<jupyter_code>'''
Description: Apply calculations on Time Series Cross Validation results to form the final Model Comparison Table
Args:
    cv_result: The results from tsCrossValidation()
    model_info: The model information which you would like to show
    evaluator_lst: The evaluator metrics which you would like to show
Return: 
    comparison_df: A pandas dataframe of a model on a type of Time Series Cross Validation
'''
def modelComparison(cv_result, model_info, evaluator_lst):
    # Calculate mean of all splits on chosen evaluator 
    col_mean_df = cv_result[evaluator_lst].mean().to_frame().T
    # Extract model info
    model_info_df = cv_result[model_info][:1]
    # Concatenate by row
    comparison_df = pd.concat([model_info_df,col_mean_df],axis=1)
    return comparison_df
# Define what model_info and evaluators in the Model Comparison Table
model_info = ['Model','CV_type','Parameters']
evaluator_lst = ['RMSE','MAPE','MAE','Variance','R2','Adjusted_R2','Time']

# The the Cross Validation results would like to compare
comparison_lst = [lr_mul_cv,lr_blk_cv,lr_wf_cv,glr_mul_cv,glr_blk_cv,glr_wf_cv,rf_mul_cv,rf_blk_cv,rf_wf_cv,gb_mul_cv,gb_blk_cv,gb_wf_cv]
# Show the Comparison Table
pd.concat([modelComparison(cv_result,model_info,evaluator_lst) for cv_result in comparison_lst])<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10292876566748976, 'fraction_numerical': 0.012909073859906854, 'mean_word_length': 2.85145058384495, 'pattern_counts': {'":': 41, '<': 81, '<?xml version=': 0, '>': 95, 'https://': 13, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 73, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '19', 'dir': 'jupyter-structured-clean-dedup', 'id': '644204', 'license': 'MIT', 'n_tokens_mistral': 9740, 'n_tokens_neox': 9331, 'n_words': 2260, 'path': 'global_mode.ipynb', 'repo_name': 'AaronOS0/Bitcoin-Price-Prediction-PySpark', 'size': '31993'}