intelligent-trading-bot/scripts/predict_rolling.py

from pathlib import Path
from datetime import datetime, timezone, timedelta
import time
import click
from concurrent.futures import ProcessPoolExecutor
from joblib import Parallel, delayed
import multiprocessing as mp

import numpy as np
import pandas as pd
import pandas.api.types as ptypes

from service.App import *
from common.model_store import *
from common.gen_features import *
from common.utils import compute_scores_regression, compute_scores
from common.generators import train_feature_set, predict_feature_set

"""
Generate label predictions for the whole input feature matrix by iteratively training models using historic data and predicting labels for some future horizon.
The main parameter is the step of iteration, that is, the future horizon for prediction.
As usual, we can specify past history length used to train a model.
The output file will store predicted labels in addition to all input columns (generated features and true labels).
This file is intended for training signal models (by simulating trade process and computing overall performance for some long period).
The output predicted labels will cover shorter period of time because we need some relatively long history to train the very first model.
"""

#
# Main
#

@click.command()
@click.option('--config_file', '-c', type=click.Path(), default='', help='Configuration file name')
def main(config_file):
    load_config(config_file)
    config = App.config

    App.model_store = ModelStore(config)
    App.model_store.load_models()

    time_column = config["time_column"]

    now = datetime.now()

    symbol = config["symbol"]
    data_path = Path(config["data_folder"]) / symbol

    #
    # Load matrix data with regular time series
    #
    file_path = data_path / config.get("matrix_file_name")
    if not file_path.is_file():
        print(f"ERROR: Input file does not exist: {file_path}")
        return

    print(f"Loading data from source data file {file_path}...")
    if file_path.suffix == ".parquet":
        df = pd.read_parquet(file_path)
    elif file_path.suffix == ".csv":
        df = pd.read_csv(file_path, parse_dates=[time_column], date_format="ISO8601")
    else:
        print(f"ERROR: Unknown extension of the input file '{file_path.suffix}'. Only 'csv' and 'parquet' are supported")
        return

    print(f"Finished loading {len(df)} records with {len(df.columns)} columns.")

    #
    # Limit the source data
    #
    rp_config = config["rolling_predict"]

    data_start = rp_config.get("data_start", None)
    data_end = rp_config.get("data_end", None)

    if data_start:
        if isinstance(data_start, str):
            df = df[ df[time_column] >= data_start ]
        elif isinstance(data_start, int):
            df = df.iloc[data_start:]

    if data_end:
        if isinstance(data_end, str):
            df = df[ df[time_column] < data_end ]
        elif isinstance(data_end, int):
            df = df.iloc[:-data_end]

    df = df.reset_index(drop=True)

    print(f"Input data size {len(df)} records. Range: [{df.iloc[0][time_column]}, {df.iloc[-1][time_column]}]")

    #
    # Determine parameters of the rolling prediction loop
    #

    prediction_start = rp_config.get("prediction_start", None)
    if isinstance(prediction_start, str):
        prediction_start = find_index(df, prediction_start)
    prediction_size = rp_config.get("prediction_size")
    prediction_steps = rp_config.get("prediction_steps")

    # Compute a missing parameter if any
    if not prediction_start:
        if not prediction_size or not prediction_steps:
            raise ValueError(f"Only one of the three rolling prediction loop parameters can be empty.")
        # Where we have to start in order to perform the specified number of steps each having the specified length
        prediction_start = len(df) - prediction_size*prediction_steps
    elif not prediction_size:
        if not prediction_start or not prediction_steps:
            raise ValueError(f"Only one of the three rolling prediction loop parameters can be empty.")
        # Size of one prediction in order to get the specified number of steps with the specified length
        prediction_size = (len(df) - prediction_start) // prediction_steps
    elif not prediction_steps:
        if not prediction_start or not prediction_size:
            raise ValueError(f"Only one of the three rolling prediction loop parameters can be empty.")
        # Number of steps with the specified length with the specified start
        prediction_steps = (len(df) - prediction_start) // prediction_size

    # Check consistency of the loop parameters
    if len(df) - prediction_start < prediction_steps * prediction_size:
        raise ValueError(f"Not enough data for {prediction_steps} steps each of size {prediction_size} starting from {prediction_start}. Available data for prediction: {len(df) - prediction_start}")

    #
    # Prepare data by selecting columns and rows
    #
    train_features_all = config.get("train_features")
    labels_all = config.get("labels")

    # Select necessary features and label
    out_columns = [time_column, 'open', 'high', 'low', 'close', 'volume', 'close_time']
    out_columns = [x for x in out_columns if x in df.columns]
    labels_present = set(labels_all).issubset(df.columns)
    if labels_present:
        all_features = train_features_all + labels_all
    else:
        all_features = train_features_all
    df = df[out_columns + [x for x in all_features if x not in out_columns]]

    for label in labels_all:
        if np.issubdtype(df[label].dtype, bool):
            df[label] = df[label].astype(int)  # For classification tasks we want to use integers

    df.replace([np.inf, -np.inf], np.nan, inplace=True)
    #in_df = in_df.dropna(subset=labels)
    df = df.reset_index(drop=True)  # We must reset index after removing rows to remove gaps

    print(f"Start index: {prediction_start}. Number of steps: {prediction_steps}. Step size: {prediction_size}")

    #
    # Rolling/moving train-predict sequence
    #

    train_feature_sets = config.get("train_feature_sets", [])
    if not train_feature_sets:
        print(f"ERROR: no train feature sets defined. Nothing to process.")
        return

    # TODO: It will be removed because we do not use directly a list of algorithms (it can be empty)
    label_horizon = config["label_horizon"]  # Labels are generated from future data and hence we might want to explicitly remove some tail rows
    train_length = config.get("train_length")

    labels_hat_df = pd.DataFrame()  # Result rows. Here store only rows for which we make predictions

    print(f"Starting rolling predict loop...")

    use_multiprocessing = rp_config.get("use_multiprocessing", False)
    max_workers = rp_config.get("max_workers", None)
    if use_multiprocessing:
        parallel = Parallel(n_jobs=max_workers, backend="loky", verbose=13)  # ['loky', 'multiprocessing', 'sequential', 'threading']
        #parallel = mp.Pool(processes=max_workers)
        #parallel = ProcessPoolExecutor(max_workers=max_workers)
    else:
        parallel = None

    for step in range(prediction_steps):

        # Predict data

        predict_start = prediction_start + (step * prediction_size)
        predict_end = predict_start + prediction_size

        predict_df = df.iloc[predict_start:predict_end]  # Assume iloc equal to index
        df_X_test = predict_df[train_features_all]

        # Train data

        # We exclude recent objects from training, because they do not have labels yet - the labels are in future
        # In real (stream) data, we will have null labels for recent objects. During simulation, labels are available and hence we need to ignore/exclude them manually
        train_end = predict_start - label_horizon - 1
        if train_length:
            train_start = max(0, train_end - train_length)
        else:
            train_start = 0

        train_df = df.iloc[train_start:train_end]  # We assume that iloc is equal to index
        train_df = train_df.dropna(subset=train_features_all)

        print(f"\n===>>> Start step {step}/{prediction_steps}. Train range: [{train_start}, {train_end}]={train_end-train_start}. Prediction range: [{predict_start}, {predict_end}]={predict_end-predict_start}")

        step_start_time = datetime.now()

        #
        # Real execution of one step
        #
        predict_labels_df = execute_train_predict_step(config, train_df, predict_df, parallel)

        # Append predicted rows to the end of previous predicted rows
        labels_hat_df = pd.concat([labels_hat_df, predict_labels_df])

        elapsed = datetime.now() - step_start_time
        print(f"End step {step}/{prediction_steps}. Scores predicted: {len(predict_labels_df.columns)}. Time elapsed: {str(elapsed).split('.')[0]}")

    # End of loop over prediction steps
    print("")
    print(f"Finished all {prediction_steps} prediction steps each with {prediction_size} predicted rows (stride). ")
    print(f"Size of predicted dataframe {len(labels_hat_df)}. Number of rows in all steps {prediction_steps*prediction_size} (steps * stride). ")
    print(f"Number of predicted columns {len(labels_hat_df.columns)}")

    #
    # Store data
    #
    # We do not store features. Only selected original data, labels, and their predictions
    out_df = labels_hat_df.join(df[out_columns + labels_all])

    out_path = data_path / config.get("predict_file_name")

    print(f"Storing predictions with {len(out_df)} records and {len(out_df.columns)} columns in output file {out_path}...")
    if out_path.suffix == ".parquet":
        out_df.to_parquet(out_path, index=False)
    elif out_path.suffix == ".csv":
        out_df.to_csv(out_path, index=False, float_format='%.6f')
    else:
        print(f"ERROR: Unknown extension of the output file '{out_path.suffix}'. Only 'csv' and 'parquet' are supported")
        return

    print(f"Predictions stored in file: {out_path}. Length: {len(out_df)}. Columns: {len(out_df.columns)}")

    #
    # Compute and store scores
    #
    score_lines = []
    # For each predicted column, find the corresponding true label column and then compare them
    for score_column_name in labels_hat_df.columns:
        label_column, _ = score_to_label_algo_pair(score_column_name)

        # Drop nans from scores
        df_scores = pd.DataFrame({"y_true": out_df[label_column], "y_predicted": out_df[score_column_name]})
        df_scores = df_scores.dropna()

        y_true = df_scores["y_true"]
        y_predicted = df_scores["y_predicted"]
        y_predicted_class = np.where(y_predicted.values > 0.5, 1, 0)

        if ptypes.is_float_dtype(y_true) and ptypes.is_float_dtype(y_predicted):
            score = compute_scores_regression(y_true, y_predicted)  # Regression stores
        else:
            score = compute_scores(y_true.astype(int), y_predicted)  # Classification stores

        score_lines.append(f"{score_column_name}: {score}")

    #
    # Store scores
    #
    score_path = out_path.with_suffix('.txt')
    with open(score_path, "a+") as f:
        f.write("\n".join([str(x) for x in score_lines]) + "\n\n")

    print(f"Prediction scores stored in path: {score_path.absolute()}")

    #
    # End
    #
    elapsed = datetime.now() - now
    print(f"Finished rolling prediction in {str(elapsed).split('.')[0]}")


def execute_train_predict_step(config: dict, train_df: pd.DataFrame, predict_df: pd.DataFrame, parallel):
    """
    This function is supposed to be used in one step of rolling prediction.
    It gets train data (which is supposed to move along larger data set) and data to be used for predictions.
    The predict data set is supposed to be selected from future data relative to the train data.

    :param config: global config
    :param train_df: data to be used for training with all features and all true labels
    :param predict_df: data to be used for prediction with all features and true labels used for computing score
    :return: predictions for all labels for objects in predict_df
    """
    rp_config = config["rolling_predict"]

    train_feature_sets = config.get("train_feature_sets", [])

    #
    # 1 Execute only training (copy from train script)
    #

    fs_now = datetime.now()
    print(f"Start train all models from {len(train_feature_sets)} feature sets {"sequentially" if not parallel else "in parallel"}. Train set size:  {len(train_df)} ")

    models = dict()
    if isinstance(parallel, Parallel):
        results = parallel(delayed(train_feature_set)(train_df, fs, config) for fs in train_feature_sets)
        for fs_models in results:
            models.update(fs_models)
    elif isinstance(parallel, mp.pool.Pool):
        #results = parallel.starmap(train_feature_set, [(train_df, fs, config) for fs in train_feature_sets])
        results = [parallel.apply(train_feature_set, args=(train_df, fs, config)) for fs in train_feature_sets]
    elif isinstance(parallel, ProcessPoolExecutor):
        # Submit all in a loop
        execution_results = dict()  # Futures for each label
        for i, fs in enumerate(train_feature_sets):
            score_column_name = f"label_{i}"
            execution_results[score_column_name] = parallel.submit(train_feature_set, train_df, fs, config)

        results = dict()
        for score_column_name, future in execution_results.items():
            results[score_column_name] = future.result()
            if future.exception():
                print(f"Exception while train-predict {score_column_name}.")
                return

    else:  # No multiprocessing - sequential execution
        for i, fs in enumerate(train_feature_sets):  #  Execute sequentially
            fs_models = train_feature_set(train_df, fs, config)
            models.update(fs_models)

    fs_elapsed = datetime.now() - fs_now
    print(f"Finished train all. Time: {str(fs_elapsed).split('.')[0]}")

    #
    # 2 Store all collected models in files
    #

    # NOTE: train generator does NOT store models in model store but only returns them
    #   yet, predict generator, expects the models to be in model store - it constructs the necessary model name and extracts it from model store
    #   so we need to put the trained models in the model store (they will be stored automatically and overwrite previously trained models which is ok)
    for score_column_name, model_pair in models.items():
        App.model_store.put_model_pair(score_column_name, model_pair)

    print(f"Models stored in path: {App.model_store.model_path.absolute()}")

    # 3 Execute only predict where df for the generator is a different predict df
    # (Copy from predict script)

    fs_now = datetime.now()
    print(f"Start predictions for {len(predict_df)} input records.")

    out_df = pd.DataFrame()  # Collect predictions
    features = []

    for i, fs in enumerate(train_feature_sets):
        fs_out_df, fs_features = predict_feature_set(predict_df, fs, config, App.model_store)
        out_df = pd.concat([out_df, fs_out_df], axis=1)
        features.extend(fs_features)

    fs_elapsed = datetime.now() - fs_now
    print(f"Finished predictions. Time: {str(fs_elapsed).split('.')[0]}")

    return out_df


if __name__ == '__main__':
    main()