refactor the structure of processing for online/stream part (batch scripts to be updated)

Author: asavinov

common/feature_generation.py

@@ -106,7 +106,7 @@ def generate_features_talib(df, config: dict, last_rows: int = 0):
     Apply TA functions from talib according to the specified configuration parameters.
 
     config = {
-        "parameters": {"relative": True, "realtive_to_last": True, "percentage": True},
+        "parameters": {"relative": True, "relative_to_last": True, "percentage": True},
         "columns": ["close"],
         "functions": ["SMA"],
         "windows": [2, 3],  # If numbers, then to argument timeperiod. If dict, then

common/model_store.py

@@ -70,11 +70,11 @@ def load_model_pair(model_path, score_column_name: str):
     return (model, scaler)
 
 
-def load_models(model_path, labels: list, train_features: list, algorithms: list):
-    """Load all model pairs for all combinations of the model parameters and return as a dict."""
+def load_models(model_path, labels: list, algorithms: list):
+    """Load all model pairs for all combinations of labels and algorithms and return as a dict."""
     models = {}
-    for predicted_label in itertools.product(labels, train_features, algorithms):
-        score_column_name = predicted_label[0] + label_algo_separator + predicted_label[1][0] + label_algo_separator + predicted_label[2]
+    for label_algorithm in itertools.product(labels, algorithms):
+        score_column_name = label_algorithm[0] + label_algo_separator + label_algorithm[1]["name"]
         model_pair = load_model_pair(model_path, score_column_name)
         models[score_column_name] = model_pair
     return models

common/signal_generation.py

@@ -1,6 +1,6 @@
 import os
 from datetime import datetime, timezone, timedelta
-from typing import Union, List
+from typing import Union, List, Tuple
 import json
 
 import numpy as np
@@ -17,6 +17,54 @@
 """
 
 
+def generate_smoothen_scores(df, config: dict):
+    """
+    Smoothen several columns and rows. Used for smoothing scores.
+
+    The following operations are applied:
+        - find average of the specified input columns (row-wise)
+        - find moving average with the specified window
+        - apply threshold to source buy/sell column(s) according to threshold parameter(s) by producing a boolean column
+
+    Notes:
+        - Input point-wise scores in buy and sell columns are always positive
+    """
+
+    columns = config.get('columns')
+    if not columns:
+        raise ValueError(f"The 'columns' parameter must be a non-empty string. {type(columns)}")
+    elif isinstance(columns, str):
+        columns = [columns]
+
+    # TODO: check that all columns exist
+    #if columns not in df.columns:
+    #    raise ValueError(f"{columns} do not exist  in the input data. Existing columns: {df.columns.to_list()}")
+
+    # Average all buy and sell columns
+    out_column = df[columns].mean(skipna=True, axis=1)
+
+    # Apply thresholds (if specified) and binarize the score
+    point_threshold = config.get("point_threshold")
+    if point_threshold:
+        out_column = out_column >= point_threshold
+
+    # Moving average
+    window = config.get("window")
+    if isinstance(window, int):
+        out_column = out_column.rolling(window, min_periods=window // 2).mean()
+    elif isinstance(window, float):
+        out_column = out_column.ewm(span=window, min_periods=window // 2, adjust=False).mean()
+
+    names = config.get('names')
+    if not isinstance(names, str):
+        raise ValueError(f"'names' parameter must be a non-empty string. {type(names)}")
+
+    df[names] = out_column
+
+    return df, [names]
+
+
+# TODO: DEPRECATED, REMOVE, REPLACE BY generator_smoothen_signals
 def aggregate_scores(df, model, score_column_out: str, score_columns: Union[List[str], str]):
     """
     Add two signal numeric (buy and sell) columns by processing a list of buy and sell point-wise predictions.
@@ -62,6 +110,43 @@ def aggregate_scores(df, model, score_column_out: str, score_columns: Union[List
     return score_column
 
 
+def generate_combine_scores(df, config: dict):
+    """
+    ML algorithms predict score which is always positive and typically within [0,1].
+    One score for price growth and one score for price fall. This function combines pairs
+    of such scores and produce one score within [-1,+1]. Positive values mean growth
+    and negative values mean fall of price.
+    """
+    columns = config.get('columns')
+    if not columns:
+        raise ValueError(f"The 'columns' parameter must be a non-empty string. {type(columns)}")
+    elif not isinstance(columns, list) or len(columns) != 2:
+        raise ValueError(f"'columns' parameter must be a list with buy column name and sell column name. {type(columns)}")
+
+    up_column = columns[0]
+    down_column = columns[1]
+
+    out_column = config.get('names')
+
+    if config.get("combine") == "relative":
+        combine_scores_relative(df, up_column, down_column, out_column)
+    elif config.get("combine") == "difference":
+        combine_scores_difference(df, up_column, down_column, out_column)
+    else:
+        # If buy score is greater than sell score then positive buy, otherwise negative sell
+        df[out_column] = df[[up_column, down_column]].apply(lambda x: x[0] if x[0] >= x[1] else -x[1], raw=True, axis=1)
+
+    # Scale the score distribution to make it symmetric or normalize
+    # Always apply the transformation to buy score. It might be in [0,1] or [-1,+1] depending on combine parameter
+    if config.get("coefficient"):
+        df[out_column] = df[out_column] * config.get("coefficient")
+    if config.get("constant"):
+        df[out_column] = df[out_column] + config.get("constant")
+
+    return df, [out_column]
+
+
+# TODO: DEPRECATED, REMOVE, REPLACE BY generator_smoothen_signals
 def combine_scores(df, model, buy_score_column, sell_score_column, trade_score_column):
     """
     Mutually adjust two independent scores with opposite semantics.
@@ -159,6 +244,32 @@ def linear_regr_fn(X):
 # Signal rules
 #
 
+def generate_threshold_rule(df, config):
+    """
+    Apply rules based on thresholds and generate trade signal buy, sell or do nothing.
+
+    Returns signals in two pre-defined columns: 'buy_signal_column' and 'sell_signal_column'
+    """
+    parameters = config.get("parameters", {})
+
+    columns = config.get("columns")
+    if not columns:
+        raise ValueError(f"The 'columns' parameter must be a non-empty string. {type(columns)}")
+    elif isinstance(columns, list):
+        columns = [columns]
+
+    buy_signal_column = config.get("names")[0]
+    sell_signal_column = config.get("names")[1]
+
+    df[buy_signal_column] = \
+        (df[columns] >= parameters.get("buy_signal_threshold"))
+    df[sell_signal_column] = \
+        (df[columns] <= parameters.get("sell_signal_threshold"))
+
+    return df, [buy_signal_column, sell_signal_column]
+
+
+# TODO: DEPRECATED TO BE REMOVED
 def apply_rule_with_score_thresholds(df, score_column_names, model):
     """
     Apply rules based on thresholds and generate trade signal buy, sell or do nothing.
@@ -178,6 +289,38 @@ def apply_rule_with_score_thresholds(df, score_column_names, model):
         (df[score_column] <= parameters.get("sell_signal_threshold"))
 
 
+def generate_threshold_rule2(df, config):
+    """
+    Assume using difference combination with negative sell scores
+    """
+    parameters = config.get("parameters", {})
+
+    columns = config.get("columns")
+    if not columns:
+        raise ValueError(f"The 'columns' parameter must be a non-empty string. {type(columns)}")
+    elif not isinstance(columns, list) or len(columns) != 2:
+        raise ValueError(f"'columns' parameter must be a list with two column names. {type(columns)}")
+
+    score_column = columns[0]
+    score_column_2 = columns[1]
+
+    buy_signal_column = config.get("names")[0]
+    sell_signal_column = config.get("names")[1]
+
+    # Both buy scores are greater than the corresponding thresholds
+    df[buy_signal_column] = \
+        (df[score_column] >= parameters.get("buy_signal_threshold")) & \
+        (df[score_column_2] >= parameters.get("buy_signal_threshold_2"))
+
+    # Both sell scores are smaller than the corresponding thresholds
+    df[sell_signal_column] = \
+        (df[score_column] <= parameters.get("sell_signal_threshold")) & \
+        (df[score_column_2] <= parameters.get("sell_signal_threshold_2"))
+
+    return df, [buy_signal_column, sell_signal_column]
+
+
+# TODO: DEPRECATED TO BE REMOVED
 def apply_rule_with_score_thresholds_2(df, score_column_names, model):
     """
     Assume using difference combination with negative sell scores

configs/config-sample-v0.7.dev.jsonc

@@ -56,9 +56,33 @@
         }
     ],
 
-    "features_horizon": 100, // Minimum data length for computing features (in online mode). Take it from feature generator parameters
+    "features_horizon": 100, // Online/stream: Minimum data length for computing features. Take it from feature generator parameters
+    "features_last_rows": 5, // Online/stream: Last values which are really needed and have to be computed. All older values are not needed
 
-    // === AGGREGATION AND POST-PROCESSING ===
+    // === SIGNALS ===
+    "signal_sets": [
+        {
+            // Combine two unsigned scores into one signed score
+            "generator": "combine", "config": {
+            "columns": ["high_20_lc", "low_20_lc"],  // 2 columns: with grow score and fall score
+            "names": "trade_score",  // Output column name: positive values - buy, negative values - sell
+            "parameters": {
+                "point_threshold": null, // Produce boolean column (optional)
+                "window": 3, // Aggregate in time
+                "combine": "difference", // "no_combine" (or empty), "relative", "difference"
+                "coefficient": 1.0, "constant": 0.0  // Normalize
+            }
+        }},
+        {
+            // Decide whether to buy or sell depending on the score value
+            "generator": "threshold_rule", "config": {
+            "columns": "trade_score",
+            "names": ["buy_signal_column", "sell_signal_column"],  // Output boolean columns
+            "parameters": {"buy_signal_threshold": 0.05, "sell_signal_threshold": -0.05}
+        }}
+    ],
+
+    // === (DEPRECATED) AGGREGATION AND POST-PROCESSING ===
 
     "score_aggregation_sets": [
         {
@@ -77,26 +101,13 @@
         }
     ],
 
-    // === TRADE SIGNAL MODEL ===
-
-    "trade_model": {
-        "rule_name": "",
-
-        "signal_columns": ["buy_signal_column", "sell_signal_column"],
-
-        "parameters": {
-            "buy_signal_threshold": 0.05,
-            "sell_signal_threshold": -0.05
-        },
-
-        "simulate_trade": true // Trade simulation and notifications
-    },
-
     // === NOTIFICATIONS ===
 
     "score_notification_model": { // When and what score notifications to send
         "score_notification": true,
 
+        "score_column_names": ["trade_score"],
+
         "notify_band_up": true,
         "notify_band_dn": true,
         "positive_bands": [
@@ -110,9 +121,25 @@
             {"edge": -0.04, "frequency": 3, "sign": "〈", "text": "weak"},
             {"edge": -0.05, "frequency": 2, "sign": "〈〈", "bold": false, "text": "strong"},
             {"edge": -1.0, "frequency": 1, "sign": "〈〈〈📉", "bold": true, "text": "SELL ZONE"}
-        ],
+        ]
+    },
+
+    "diagram_notification_model": {
+        // Regularly sending historic data with prices, scores and buy-sell trade decisions
+        "diagram_notification": true,
+
+        // # 1 previous week with hourly aggregation
+        "freq": "H",
+        "nrows": 168
+    },
+
+    // === TRADE MODEL ===
+
+    "trade_model": {
+        "simulate_trade": true, // Trade simulation with notifications
 
-        "notify_diagram": false
+        "buy_signal_column":"buy_signal_column",
+        "sell_signal_column": "sell_signal_column"
     },
 
     // === FINDING BEST TRADE PARAMETERS ===

scripts/features.py

@@ -7,10 +7,12 @@
 
 from service.App import *
 from common.feature_generation import *
-from common.label_generation_highlow import generate_labels_highlow
-from common.label_generation_highlow import generate_labels_highlow2
-from common.label_generation_topbot import generate_labels_topbot
-from common.label_generation_topbot import generate_labels_topbot2
+from common.label_generation_highlow import generate_labels_highlow, generate_labels_highlow2
+from common.label_generation_topbot import generate_labels_topbot, generate_labels_topbot2
+from common.signal_generation import (
+    generate_smoothen_scores, generate_combine_scores,
+    generate_threshold_rule, generate_threshold_rule2
+)
 
 #
 # Parameters
@@ -157,6 +159,17 @@ def generate_feature_set(df: pd.DataFrame, fs: dict, last_rows: int) -> Tuple[pd
         f_df, features = generate_labels_topbot(f_df, column_name, top_level_fracs, bot_level_fracs)
     elif generator == "topbot2":
         f_df, features = generate_labels_topbot2(f_df, gen_config)
+
+    # Signals
+    elif generator == "smoothen":
+        f_df, features = generate_smoothen_scores(f_df, gen_config)
+    elif generator == "combine":
+        f_df, features = generate_combine_scores(f_df, gen_config)
+    elif generator == "threshold_rule":
+        f_df, features = generate_threshold_rule(f_df, gen_config)
+    elif generator == "threshold_rule2":
+        f_df, features = generate_threshold_rule2(f_df, gen_config)
+
     else:
         print(f"Unknown feature generator {generator}")
         return

service/App.py

@@ -39,7 +39,7 @@ class App:
     trade_state_status = 0  # Something wrong with our trading logic (wrong use, inconsistent state etc. what we cannot recover)
 
     signal = None  # Latest signal "BUY", "SELL"
-    feature_df = None  # Data from the latest analysis
+    df = None  # Data from the latest analysis
 
     # Trade status
     transaction = None
@@ -140,10 +140,16 @@ class App:
         # Minimum history length required to compute derived features
         "features_horizon": 10,
 
-        # =======================================
-        # === AGGREGATION AND POST-PROCESSING ===
+        # ===============
+        # === SIGNALS ===
 
-        "score_aggregation_sets": [],
+        "signal_sets": [],
+
+        # =====================
+        # === NOTIFICATIONS ===
+
+        "score_notification_model": {},
+        "diagram_notification_model": {},
 
         # ================================
         # === SIGNAL RULES FOR TRADING ===