2022-08-07 14:04:47 +02:00 | | | from pathlib import Path
|
| | | from datetime import datetime, timezone, timedelta
|
| | | import click
|
| | | from tqdm import tqdm
|
| | |
|
| | | import numpy as np
|
| | | import pandas as pd
|
| | |
|
| | | from service.App import *
|
| | | from common.model_store import *
|
2024-03-24 13:22:52 +01:00 | | | from common.generators import predict_feature_set
|
2022-08-07 14:04:47 +02:00 | | |
|
| | | """
|
| | | Apply models to (previously generated) features and compute prediction scores.
|
| | | """
|
| | |
|
| | |
|
| | | #
|
| | | # Parameters
|
| | | #
|
| | | class P:
|
| | | in_nrows = 100_000_000 # For debugging
|
| | | tail_rows = 0 # How many last rows to select (for debugging)
|
| | |
|
| | |
|
| | | @click.command()
|
| | | @click.option('--config_file', '-c', type=click.Path(), default='', help='Configuration file name')
|
| | | def main(config_file):
|
| | | load_config(config_file)
|
| | |
|
| | | time_column = App.config["time_column"]
|
| | |
|
| | | now = datetime.now()
|
| | |
|
| | | #
|
| | | # Load feature matrix
|
| | | #
|
| | | symbol = App.config["symbol"]
|
| | | data_path = Path(App.config["data_folder"]) / symbol
|
| | |
|
2024-03-16 13:08:25 +01:00 | | | file_path = data_path / App.config.get("matrix_file_name")
|
2022-08-07 14:04:47 +02:00 | | | 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}...")
|
2024-03-16 13:08:25 +01:00 | | | 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", nrows=P.in_nrows)
|
| | | else:
|
2025-04-17 17:22:55 +02:00 | | | print(f"ERROR: Unknown extension of the input file '{file_path.suffix}'. Only 'csv' and 'parquet' are supported")
|
2024-03-16 13:08:25 +01:00 | | | return
|
2022-08-07 14:04:47 +02:00 | | | print(f"Finished loading {len(df)} records with {len(df.columns)} columns.")
|
| | |
|
| | | df = df.iloc[-P.tail_rows:]
|
| | | df = df.reset_index(drop=True)
|
| | |
|
2023-09-02 11:42:12 +02:00 | | | print(f"Input data size {len(df)} records. Range: [{df.iloc[0][time_column]}, {df.iloc[-1][time_column]}]")
|
| | |
|
2022-08-07 14:04:47 +02:00 | | | #
|
| | | # Prepare data by selecting columns and rows
|
| | | #
|
| | | train_features = App.config.get("train_features")
|
| | | labels = App.config["labels"]
|
| | | algorithms = App.config.get("algorithms")
|
| | |
|
| | | # Select necessary features and label
|
2025-03-25 20:04:36 +01:00 | | | out_columns = [time_column, 'open', 'high', 'low', 'close', 'volume', 'close_time']
|
2022-08-07 14:04:47 +02:00 | | | out_columns = [x for x in out_columns if x in df.columns]
|
| | | labels_present = set(labels).issubset(df.columns)
|
| | | if labels_present:
|
| | | all_features = train_features + labels
|
| | | else:
|
| | | all_features = train_features
|
2023-03-11 17:31:16 +01:00 | | | df = df[out_columns + [x for x in all_features if x not in out_columns]]
|
2022-08-07 14:04:47 +02:00 | | |
|
2024-03-16 12:01:49 +01:00 | | | df.replace([np.inf, -np.inf], np.nan, inplace=True)
|
2024-03-23 18:13:57 +01:00 | | | if len(df) == 0:
|
| | | print(f"ERROR: Empty data set after removing NULLs in feature columns. Some features might all have NULL values.")
|
| | | # print(train_df.isnull().sum().sort_values(ascending=False))
|
2022-08-07 14:04:47 +02:00 | | |
|
2025-01-11 18:44:28 +01:00 | | | df = df.dropna(subset=train_features)
|
| | | df = df.reset_index(drop=True) # We must reset index after removing rows to remove gaps
|
| | |
|
2022-08-07 14:04:47 +02:00 | | | #
|
| | | # Load models for all score columns
|
| | | #
|
2023-03-11 10:05:43 +01:00 | | | model_path = Path(App.config["model_folder"])
|
2022-08-07 14:04:47 +02:00 | | | if not model_path.is_absolute():
|
2023-03-11 10:05:43 +01:00 | | | model_path = data_path / model_path
|
2022-08-07 14:04:47 +02:00 | | | model_path = model_path.resolve()
|
| | |
|
2025-01-11 18:44:28 +01:00 | | | models = load_models_for_generators(App.config, model_path)
|
2022-08-07 14:04:47 +02:00 | | |
|
| | | #
|
2024-03-23 18:13:57 +01:00 | | | # Generate/predict train features
|
2022-08-07 14:04:47 +02:00 | | | #
|
2024-03-23 18:13:57 +01:00 | | | train_feature_sets = App.config.get("train_feature_sets", [])
|
| | | if not train_feature_sets:
|
| | | print(f"ERROR: no train feature sets defined. Nothing to process.")
|
| | | return
|
| | |
|
| | | print(f"Start generating trained features for {len(df)} input records.")
|
| | |
|
2024-03-24 13:22:52 +01:00 | | | out_df = pd.DataFrame() # Collect predictions
|
2024-03-23 18:13:57 +01:00 | | | features = []
|
2022-08-07 14:04:47 +02:00 | | | scores = dict()
|
2024-03-23 18:13:57 +01:00 | | |
|
| | | for i, fs in enumerate(train_feature_sets):
|
| | | fs_now = datetime.now()
|
| | | print(f"Start train feature set {i}/{len(train_feature_sets)}. Generator {fs.get('generator')}...")
|
| | |
|
2024-03-24 13:22:52 +01:00 | | | fs_out_df, fs_features, fs_scores = predict_feature_set(df, fs, App.config, models)
|
2024-03-23 18:13:57 +01:00 | | |
|
2024-03-24 13:22:52 +01:00 | | | out_df = pd.concat([out_df, fs_out_df], axis=1)
|
2024-03-23 18:13:57 +01:00 | | | features.extend(fs_features)
|
| | | scores.update(fs_scores)
|
| | |
|
| | | fs_elapsed = datetime.now() - fs_now
|
| | | print(f"Finished train feature set {i}/{len(train_feature_sets)}. Generator {fs.get('generator')}. Time: {str(fs_elapsed).split('.')[0]}")
|
| | |
|
| | | print(f"Finished generating trained features.")
|
2022-08-07 14:04:47 +02:00 | | |
|
| | | #
|
| | | # Store scores
|
| | | #
|
| | | if labels_present:
|
| | | lines = list()
|
| | | for score_column_name, score in scores.items():
|
| | | line = score_column_name + ", " + str(score)
|
| | | lines.append(line)
|
| | |
|
| | | metrics_file_name = f"prediction-metrics.txt"
|
| | | metrics_path = (data_path / metrics_file_name).resolve()
|
| | | with open(metrics_path, 'a+') as f:
|
| | | f.write("\n".join(lines) + "\n\n")
|
| | |
|
| | | print(f"Metrics stored in path: {metrics_path.absolute()}")
|
| | |
|
| | | #
|
| | | # Store predictions
|
| | | #
|
| | | # Store only selected original data, labels, and their predictions
|
| | | out_df = out_df.join(df[out_columns + (labels if labels_present else [])])
|
| | |
|
| | | out_path = data_path / App.config.get("predict_file_name")
|
| | |
|
2024-03-16 13:46:08 +01:00 | | | 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:
|
2025-04-17 17:22:55 +02:00 | | | print(f"ERROR: Unknown extension of the output file '{out_path.suffix}'. Only 'csv' and 'parquet' are supported")
|
2024-03-16 13:46:08 +01:00 | | | return
|
| | |
|
2022-08-07 14:04:47 +02:00 | | | print(f"Predictions stored in file: {out_path}. Length: {len(out_df)}. Columns: {len(out_df.columns)}")
|
| | |
|
| | | #
|
| | | # End
|
| | | #
|
| | | elapsed = datetime.now() - now
|
2025-04-26 11:13:57 +02:00 | | | print(f"Finished predicting in {str(elapsed).split('.')[0]}")
|
2022-08-07 14:04:47 +02:00 | | |
|
| | |
|
| | | if __name__ == '__main__':
|
| | | main()
|