From 29af7509cfbc2277200e94b69e215ddd5bf9e333 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Sun, 10 Aug 2025 17:00:25 +0200 Subject: [PATCH] chore: improve typing MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- ReforceXY/user_data/freqaimodels/ReforceXY.py | 9 ++- .../freqaimodels/QuickAdapterRegressorV3.py | 20 ++++--- quickadapter/user_data/strategies/Utils.py | 60 +++++++++++-------- 3 files changed, 53 insertions(+), 36 deletions(-) diff --git a/ReforceXY/user_data/freqaimodels/ReforceXY.py b/ReforceXY/user_data/freqaimodels/ReforceXY.py index f024094..5ca5093 100644 --- a/ReforceXY/user_data/freqaimodels/ReforceXY.py +++ b/ReforceXY/user_data/freqaimodels/ReforceXY.py @@ -24,6 +24,7 @@ from freqtrade.freqai.tensorboard.TensorboardCallback import TensorboardCallback from freqtrade.strategy import timeframe_to_minutes from gymnasium import Env from gymnasium.spaces import Box +from numpy.typing import NDArray from optuna import Trial, TrialPruned, create_study from optuna.exceptions import ExperimentalWarning from optuna.pruners import HyperbandPruner @@ -788,7 +789,9 @@ class ReforceXY(BaseReinforcementLearningModel): low=-np.inf, high=np.inf, shape=self.shape, dtype=np.float32 ) - def reset(self, seed=None, **kwargs) -> Tuple[np.ndarray, Dict[str, Any]]: + def reset( + self, seed=None, **kwargs + ) -> Tuple[NDArray[np.float32], Dict[str, Any]]: """ Reset is called at the beginning of every episode """ @@ -941,7 +944,7 @@ class ReforceXY(BaseReinforcementLearningModel): return 0.0 - def _get_observation(self) -> np.ndarray: + def _get_observation(self) -> NDArray[np.float32]: """ This may or may not be independent of action types, user can inherit this in their custom "MyRLEnv" @@ -1032,7 +1035,7 @@ class ReforceXY(BaseReinforcementLearningModel): def step( self, action: int - ) -> Tuple[np.ndarray, float, bool, bool, Dict[str, Any]]: + ) -> Tuple[NDArray[np.float32], float, bool, bool, Dict[str, Any]]: """ Take a step in the environment based on the provided action """ diff --git a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py index cf90a7f..6924951 100644 --- a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py +++ b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py @@ -16,6 +16,7 @@ import skimage import sklearn from freqtrade.freqai.base_models.BaseRegressionModel import BaseRegressionModel from freqtrade.freqai.data_kitchen import FreqaiDataKitchen +from numpy.typing import NDArray from Utils import ( calculate_min_extrema, @@ -517,10 +518,11 @@ class QuickAdapterRegressorV3(BaseRegressionModel): def eval_set_and_weights( X_test: pd.DataFrame, y_test: pd.DataFrame, - test_weights: np.ndarray, + test_weights: NDArray[np.float64], test_size: float, ) -> tuple[ - Optional[list[tuple[pd.DataFrame, pd.DataFrame]]], Optional[list[np.ndarray]] + Optional[list[tuple[pd.DataFrame, pd.DataFrame]]], + Optional[list[NDArray[np.float64]]], ]: if test_size == 0: eval_set = None @@ -633,7 +635,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): @staticmethod def apply_skimage_threshold( - series: pd.Series, threshold_func: Callable[[np.ndarray], float] + series: pd.Series, threshold_func: Callable[[NDArray[np.float64]], float] ) -> float: values = series.to_numpy() @@ -721,8 +723,8 @@ class QuickAdapterRegressorV3(BaseRegressionModel): return None def calculate_distances( - normalized_matrix: np.ndarray, metric: str - ) -> np.ndarray: + normalized_matrix: NDArray[np.float64], metric: str + ) -> NDArray[np.float64]: n_objectives = normalized_matrix.shape[1] n_samples = normalized_matrix.shape[0] label_p_order = float(self.ft_params.get("label_p_order", 2.0)) @@ -1193,10 +1195,10 @@ def train_objective( regressor: str, X: pd.DataFrame, y: pd.DataFrame, - train_weights: np.ndarray, + train_weights: NDArray[np.float64], X_test: pd.DataFrame, y_test: pd.DataFrame, - test_weights: np.ndarray, + test_weights: NDArray[np.float64], test_size: float, fit_live_predictions_candles: int, candles_step: int, @@ -1303,10 +1305,10 @@ def hp_objective( regressor: str, X: pd.DataFrame, y: pd.DataFrame, - train_weights: np.ndarray, + train_weights: NDArray[np.float64], X_test: pd.DataFrame, y_test: pd.DataFrame, - test_weights: np.ndarray, + test_weights: NDArray[np.float64], model_training_best_parameters: dict[str, Any], model_training_parameters: dict[str, Any], expansion_ratio: float, diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py index e34f396..40bbc85 100644 --- a/quickadapter/user_data/strategies/Utils.py +++ b/quickadapter/user_data/strategies/Utils.py @@ -10,6 +10,7 @@ import optuna import pandas as pd import scipy as sp import talib.abstract as ta +from numpy.typing import NDArray from technical import qtpylib T = TypeVar("T", pd.Series, float) @@ -44,7 +45,7 @@ def _calculate_coeffs( win_type: Literal["gaussian", "kaiser", "triang"], std: float, beta: float, -) -> np.ndarray: +) -> NDArray[np.float64]: if win_type == "gaussian": coeffs = sp.signal.windows.gaussian(M=window, std=std, sym=True) elif win_type == "kaiser": @@ -235,8 +236,15 @@ def calculate_zero_lag(series: pd.Series, period: int) -> pd.Series: @lru_cache(maxsize=8) -def get_ma_fn(mamode: str) -> Callable[[pd.Series, int], np.ndarray]: - mamodes: dict[str, Callable[[pd.Series, int], np.ndarray]] = { +def get_ma_fn( + mamode: str, +) -> Callable[[pd.Series | NDArray[np.float64], int], pd.Series | NDArray[np.float64]]: + mamodes: dict[ + str, + Callable[ + [pd.Series | NDArray[np.float64], int], pd.Series | NDArray[np.float64] + ], + ] = { "sma": ta.SMA, "ema": ta.EMA, "wma": ta.WMA, @@ -250,7 +258,9 @@ def get_ma_fn(mamode: str) -> Callable[[pd.Series, int], np.ndarray]: @lru_cache(maxsize=8) -def get_zl_ma_fn(mamode: str) -> Callable[[pd.Series, int], np.ndarray]: +def get_zl_ma_fn( + mamode: str, +) -> Callable[[pd.Series | NDArray[np.float64], int], pd.Series | NDArray[np.float64]]: ma_fn = get_ma_fn(mamode) return lambda series, timeperiod: ma_fn( calculate_zero_lag(series, timeperiod), timeperiod=timeperiod @@ -265,7 +275,9 @@ def zlema(series: pd.Series, period: int) -> pd.Series: return zl_series.ewm(alpha=alpha, adjust=False).mean() -def _fractal_dimension(highs: np.ndarray, lows: np.ndarray, period: int) -> float: +def _fractal_dimension( + highs: NDArray[np.float64], lows: NDArray[np.float64], period: int +) -> float: """Original fractal dimension computation implementation per Ehlers' paper.""" if period % 2 != 0: raise ValueError("period must be even") @@ -292,7 +304,7 @@ def _fractal_dimension(highs: np.ndarray, lows: np.ndarray, period: int) -> floa return np.clip(D, 1.0, 2.0) -def frama(df: pd.DataFrame, period: int = 16, zero_lag=False) -> pd.Series: +def frama(df: pd.DataFrame, period: int = 16, zero_lag: bool = False) -> pd.Series: """ Original FRAMA implementation per Ehlers' paper with optional zero lag. """ @@ -372,12 +384,12 @@ def get_price_fn(pricemode: str) -> Callable[[pd.DataFrame], pd.Series]: def ewo( dataframe: pd.DataFrame, - ma1_length=5, - ma2_length=34, - pricemode="close", - mamode="sma", - zero_lag=False, - normalize=False, + ma1_length: int = 5, + ma2_length: int = 34, + pricemode: str = "close", + mamode: str = "sma", + zero_lag: bool = False, + normalize: bool = False, ) -> pd.Series: """ Calculate the Elliott Wave Oscillator (EWO) using two moving averages. @@ -402,14 +414,14 @@ def ewo( def alligator( df: pd.DataFrame, - jaw_period=13, - teeth_period=8, - lips_period=5, - jaw_shift=8, - teeth_shift=5, - lips_shift=3, - pricemode="median", - zero_lag=False, + jaw_period: int = 13, + teeth_period: int = 8, + lips_period: int = 5, + jaw_shift: int = 8, + teeth_shift: int = 5, + lips_shift: int = 3, + pricemode: str = "median", + zero_lag: bool = False, ) -> tuple[pd.Series, pd.Series, pd.Series]: """ Calculate Bill Williams' Alligator indicator lines. @@ -454,7 +466,7 @@ def find_fractals(df: pd.DataFrame, period: int = 2) -> tuple[list[int], list[in return fractal_highs, fractal_lows -def calculate_quantile(values: np.ndarray, value: float) -> float: +def calculate_quantile(values: NDArray[np.float64], value: float) -> float: if values.size == 0: return np.nan @@ -487,7 +499,7 @@ def zigzag( natr_values = (ta.NATR(df, timeperiod=natr_period).bfill() / 100.0).to_numpy() indices: list[int] = df.index.tolist() - thresholds: np.ndarray = natr_values * natr_ratio + thresholds: NDArray[np.float64] = natr_values * natr_ratio closes = df.get("close").to_numpy() highs = df.get("high").to_numpy() lows = df.get("low").to_numpy() @@ -719,9 +731,9 @@ def fit_regressor( regressor: str, X: pd.DataFrame, y: pd.DataFrame, - train_weights: np.ndarray, + train_weights: NDArray[np.float64], eval_set: Optional[list[tuple[pd.DataFrame, pd.DataFrame]]], - eval_weights: Optional[list[np.ndarray]], + eval_weights: Optional[list[NDArray[np.float64]]], model_training_parameters: dict[str, Any], init_model: Any = None, callbacks: Optional[list[Callable]] = None, -- 2.43.0