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
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
"""
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"
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
"""
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,
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
@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()
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))
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,
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,
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)
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":
@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,
@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
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")
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.
"""
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.
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.
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
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()
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,
repositories / freqai-strategies.git / commitdiff
