75 lines
2.2 KiB
Python
75 lines
2.2 KiB
Python
import asyncio
| |
from pathlib import Path
| |
import click
| |
from tqdm import tqdm
| |
| |
import numpy as np
| |
import pandas as pd
| |
| |
from service.App import *
| |
from common.utils import *
| |
from common.model_store import *
| |
from common.generators import output_feature_set
| |
| |
"""
| |
Execute outputs based on signal file.
| |
"""
| |
| |
| |
@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 data with (rolling) label point-wise predictions and signals generated
| |
#
| |
file_path = data_path / config.get("signal_file_name")
| |
if not file_path.exists():
| |
print(f"ERROR: Input file does not exist: {file_path}")
| |
return
| |
| |
print(f"Loading signals from input 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"Signals loaded. Length: {len(df)}. Width: {len(df.columns)}")
| |
| |
#
| |
# Execute all output sets. Note that they can be async
| |
#
| |
| |
# For data processing, a data frame is supposed to have timestamp in index
| |
df = df.set_index(time_column, inplace=False)
| |
| |
output_sets = App.config.get("output_sets", [])
| |
for os in output_sets:
| |
try:
| |
#await output_feature_set(df, os, App.config, App.model_store)
| |
asyncio.run(output_feature_set(df, os, App.config, App.model_store))
| |
except Exception as e:
| |
log.error(f"Error in output function: {e}. Generator: {os.get("generator")}. Output config: {os}")
| |
return
| |
| |
elapsed = datetime.now() - now
| |
print(f"Finished executing outputs in {str(elapsed).split('.')[0]}")
| |
| |
| |
if __name__ == '__main__':
| |
main()
|