Function get_best_levels_prices_and_labels

utils.py:281–340

Get the best levels (bid and ask) prices and the corresponding discretized labels. Args: dataset (str): Name of the dataset to be used (e.g. nasdaq, lse, ...). history_length (int): Length of the history (each model's sample is a 2D array of shape ( , <feature

(
    dataset: str,
    target_stocks: str,
    history_length: int,
    all_horizons: list[int],
    prediction_horizon: int,
    threshold: float,
)

def get_best_levels_prices_and_labels(
    dataset: str,
    target_stocks: str,
    history_length: int,
    all_horizons: list[int],
    prediction_horizon: int,
    threshold: float,
) -> tuple[Any, ...]:
    """
    Get the best levels (bid and ask) prices and the corresponding discretized labels.
    Args:
        dataset (str): Name of the dataset to be used (e.g. nasdaq, lse, ...).
        history_length (int): Length of the history (each model&#x27;s sample is a 2D array of shape (<history_length>, <features>)).
        all_horizons (list): List all horizons computed in the preprocessing stage.
        prediction_horizon (int): Horizon to be considered.
        threshold (float): Threshold to be used to discretize the labels.

    Returns:
        A tuple containing the best levels (bid and ask) prices and the corresponding discretized labels.
    """

    # List the test files.
    test_files = sorted(glob.glob(f"./data/{dataset}/unscaled_data/test/*{target_stocks[0]}*.csv"))

    best_levels_prices = pd.DataFrame()

    # Get the position of the prediction horizon in the list of all horizons.
    position = next(
        (
            index
            for index, value in enumerate(all_horizons)
            if value == prediction_horizon
        ),
        None,
    )
    all_labels_temp = []

    for file in test_files:
        # Load the file.
        df = pd.read_csv(file).iloc[history_length:, :]
        # Reset the index.
        df.reset_index(drop=True, inplace=True)
        # Get all the labels.
        label_df = df.iloc[:, 41:]
        # Get the label corresponding to the prediction horizon.
        label = label_df.iloc[:, position]
        # Get the best levels (ask and bid) prices and the datetime corresponding to each tick.
        best_levels_prices = pd.concat(
            [best_levels_prices, df[["seconds", "ASKp1", "BIDp1"]]]
        )
        # Append the label to the list of labels.
        all_labels_temp = all_labels_temp + label.tolist()

    # Discretize the labels (0: downtrend, 1: no trend, 2: uptrend).
    all_labels = [
        2 if label >= threshold else 0 if label <= -threshold else 1
        for label in all_labels_temp
    ]