)
ModelType = Literal["PPO", "RecurrentPPO", "MaskablePPO", "DQN", "QRDQN"]
-ScheduleType = Literal["linear", "constant", "unknown"]
ScheduleTypeKnown = Literal["linear", "constant"]
+ScheduleType = Union[ScheduleTypeKnown, Literal["unknown"]]
ExitPotentialMode = Literal[
"canonical",
"non_canonical",
TransformFunction = Literal["tanh", "softsign", "arctan", "sigmoid", "asinh", "clip"]
ExitAttenuationMode = Literal["legacy", "sqrt", "linear", "power", "half_life"]
ActivationFunction = Literal["tanh", "relu", "elu", "leaky_relu"]
-OptimizerClass = Literal["adam", "adamw", "rmsprop"]
+OptimizerClassOptuna = Literal["adamw", "rmsprop"]
+OptimizerClass = Union[OptimizerClassOptuna, Literal["adam"]]
NetArchSize = Literal["small", "medium", "large", "extra_large"]
StorageBackend = Literal["sqlite", "file"]
SamplerType = Literal["tpe", "auto"]
"DQN",
"QRDQN",
)
- _SCHEDULE_TYPES: Final[tuple[ScheduleType, ...]] = ("linear", "constant", "unknown")
_SCHEDULE_TYPES_KNOWN: Final[tuple[ScheduleTypeKnown, ...]] = ("linear", "constant")
+ _SCHEDULE_TYPES: Final[tuple[ScheduleType, ...]] = (
+ *_SCHEDULE_TYPES_KNOWN,
+ "unknown",
+ )
_EXIT_POTENTIAL_MODES: Final[tuple[ExitPotentialMode, ...]] = (
"canonical",
"non_canonical",
"elu",
"leaky_relu",
)
- _OPTIMIZER_CLASSES: Final[tuple[OptimizerClass, ...]] = ("adam", "adamw", "rmsprop")
- _OPTIMIZER_CLASSES_OPTUNA: Final[tuple[OptimizerClass, ...]] = ("adamw", "rmsprop")
+ _OPTIMIZER_CLASSES_OPTUNA: Final[tuple[OptimizerClassOptuna, ...]] = (
+ "adamw",
+ "rmsprop",
+ )
+ _OPTIMIZER_CLASSES: Final[tuple[OptimizerClass, ...]] = (
+ *_OPTIMIZER_CLASSES_OPTUNA,
+ "adam",
+ )
_NET_ARCH_SIZES: Final[tuple[NetArchSize, ...]] = (
"small",
"medium",
)
model_params["policy_kwargs"]["optimizer_class"] = get_optimizer_class(
model_params.get("policy_kwargs", {}).get(
- "optimizer_class", ReforceXY._OPTIMIZER_CLASSES[1]
+ "optimizer_class", ReforceXY._OPTIMIZER_CLASSES[0]
) # "adamw"
)
Get optimizer class
"""
return {
- ReforceXY._OPTIMIZER_CLASSES[0]: th.optim.Adam, # "adam"
- ReforceXY._OPTIMIZER_CLASSES[1]: th.optim.AdamW, # "adamw"
- ReforceXY._OPTIMIZER_CLASSES[2]: th.optim.RMSprop, # "rmsprop"
+ ReforceXY._OPTIMIZER_CLASSES[0]: th.optim.AdamW, # "adamw"
+ ReforceXY._OPTIMIZER_CLASSES[1]: th.optim.RMSprop, # "rmsprop"
+ ReforceXY._OPTIMIZER_CLASSES[2]: th.optim.Adam, # "adam"
}.get(optimizer_class_name, th.optim.Adam)
import warnings
from functools import cached_property
from pathlib import Path
-from typing import Any, Callable, Final, Literal, Optional
+from typing import Any, Callable, Final, Literal, Optional, Union
import numpy as np
import optuna
ExtremaSelectionMethod = Literal["peak_values", "extrema_rank", "partition"]
OptunaNamespace = Literal["hp", "train", "label"]
+CustomThresholdMethod = Literal["median", "soft_extremum"]
+SkimageThresholdMethod = Literal[
+ "isodata", "li", "mean", "minimum", "otsu", "triangle", "yen"
+]
+ThresholdMethod = Union[CustomThresholdMethod, SkimageThresholdMethod]
debug = False
"extrema_rank",
"partition",
)
+ _CUSTOM_THRESHOLD_METHODS: Final[tuple[CustomThresholdMethod, ...]] = (
+ "median",
+ "soft_extremum",
+ )
+ _SKIMAGE_THRESHOLD_METHODS: Final[tuple[SkimageThresholdMethod, ...]] = (
+ "isodata",
+ "li",
+ "mean",
+ "minimum",
+ "otsu",
+ "triangle",
+ "yen",
+ )
+ _THRESHOLD_METHODS: Final[tuple[ThresholdMethod, ...]] = (
+ *_CUSTOM_THRESHOLD_METHODS,
+ *_SKIMAGE_THRESHOLD_METHODS,
+ )
_OPTUNA_STORAGE_BACKENDS: Final[tuple[str, ...]] = ("sqlite", "file")
_OPTUNA_SAMPLERS: Final[tuple[str, ...]] = ("tpe", "auto")
_OPTUNA_NAMESPACES: Final[tuple[OptunaNamespace, ...]] = ("hp", "train", "label")
def _extrema_selection_methods_set() -> set[ExtremaSelectionMethod]:
return set(QuickAdapterRegressorV3._EXTREMA_SELECTION_METHODS)
+ @staticmethod
+ def _custom_threshold_methods_set() -> set[CustomThresholdMethod]:
+ return set(QuickAdapterRegressorV3._CUSTOM_THRESHOLD_METHODS)
+
+ @staticmethod
+ def _skimage_threshold_methods_set() -> set[SkimageThresholdMethod]:
+ return set(QuickAdapterRegressorV3._SKIMAGE_THRESHOLD_METHODS)
+
+ @staticmethod
+ def _threshold_methods_set() -> set[ThresholdMethod]:
+ return set(QuickAdapterRegressorV3._THRESHOLD_METHODS)
+
@staticmethod
def _optuna_namespaces_set() -> set[OptunaNamespace]:
return set(QuickAdapterRegressorV3._OPTUNA_NAMESPACES)
QuickAdapterRegressorV3._EXTREMA_SELECTION_METHODS[1],
)
)
- if extrema_selection not in self._extrema_selection_methods_set():
+ if (
+ extrema_selection
+ not in QuickAdapterRegressorV3._extrema_selection_methods_set()
+ ):
raise ValueError(
f"Unsupported extrema selection method: {extrema_selection}. "
- f"Supported methods are {', '.join(self._EXTREMA_SELECTION_METHODS)}"
+ f"Supported methods are {', '.join(QuickAdapterRegressorV3._EXTREMA_SELECTION_METHODS)}"
)
thresholds_smoothing = str(
- predictions_extrema.get("thresholds_smoothing", "mean")
- )
- skimage_thresholds_smoothing_methods = {
- "isodata",
- "li",
- "mean",
- "minimum",
- "otsu",
- "triangle",
- "yen",
- }
- thresholds_smoothing_methods = skimage_thresholds_smoothing_methods.union(
- {"soft_extremum"}
+ predictions_extrema.get(
+ "thresholds_smoothing",
+ QuickAdapterRegressorV3._SKIMAGE_THRESHOLD_METHODS[2],
+ )
)
- if thresholds_smoothing == "soft_extremum":
+ if thresholds_smoothing not in QuickAdapterRegressorV3._threshold_methods_set():
+ raise ValueError(
+ f"Unsupported thresholds smoothing method: {thresholds_smoothing}. "
+ f"Supported methods are {', '.join(QuickAdapterRegressorV3._THRESHOLD_METHODS)}"
+ )
+ if (
+ thresholds_smoothing == QuickAdapterRegressorV3._CUSTOM_THRESHOLD_METHODS[0]
+ ): # "median"
+ return QuickAdapterRegressorV3.median_min_max(
+ pred_extrema, extrema_selection
+ )
+ elif (
+ thresholds_smoothing == QuickAdapterRegressorV3._CUSTOM_THRESHOLD_METHODS[1]
+ ): # "soft_extremum"
thresholds_alpha = float(predictions_extrema.get("thresholds_alpha", 12.0))
return QuickAdapterRegressorV3.soft_extremum_min_max(
pred_extrema, thresholds_alpha, extrema_selection
)
- elif thresholds_smoothing in skimage_thresholds_smoothing_methods:
+ else:
return QuickAdapterRegressorV3.skimage_min_max(
pred_extrema, thresholds_smoothing, extrema_selection
)
- else:
- raise ValueError(
- f"Unsupported thresholds smoothing method: {thresholds_smoothing}. Supported methods are {', '.join(thresholds_smoothing_methods)}"
- )
@staticmethod
def get_pred_min_max(
return soft_minimum, soft_maximum
@staticmethod
- def skimage_min_max(
+ def median_min_max(
pred_extrema: pd.Series,
- method: str,
extrema_selection: ExtremaSelectionMethod,
) -> tuple[float, float]:
pred_minima, pred_maxima = QuickAdapterRegressorV3.get_pred_min_max(
pred_extrema, extrema_selection
)
- method_functions = {
- "isodata": QuickAdapterRegressorV3.apply_skimage_threshold,
- "li": QuickAdapterRegressorV3.apply_skimage_threshold,
- "mean": QuickAdapterRegressorV3.apply_skimage_threshold,
- "minimum": QuickAdapterRegressorV3.apply_skimage_threshold,
- "otsu": QuickAdapterRegressorV3.apply_skimage_threshold,
- "triangle": QuickAdapterRegressorV3.apply_skimage_threshold,
- "yen": QuickAdapterRegressorV3.apply_skimage_threshold,
- }
+ if pred_minima.empty:
+ min_val = np.nan
+ else:
+ min_val = np.median(pred_minima.to_numpy())
+ if not np.isfinite(min_val):
+ min_val = QuickAdapterRegressorV3.safe_min_pred(pred_extrema)
- if method not in method_functions:
- raise ValueError(f"Unsupported method: {method}")
+ if pred_maxima.empty:
+ max_val = np.nan
+ else:
+ max_val = np.median(pred_maxima.to_numpy())
+ if not np.isfinite(max_val):
+ max_val = QuickAdapterRegressorV3.safe_max_pred(pred_extrema)
- min_func = method_functions[method]
- max_func = method_functions[method]
+ return min_val, max_val
+
+ @staticmethod
+ def skimage_min_max(
+ pred_extrema: pd.Series,
+ method: str,
+ extrema_selection: ExtremaSelectionMethod,
+ ) -> tuple[float, float]:
+ pred_minima, pred_maxima = QuickAdapterRegressorV3.get_pred_min_max(
+ pred_extrema, extrema_selection
+ )
try:
threshold_func = getattr(skimage.filters, f"threshold_{method}")
except AttributeError:
raise ValueError(f"Unknown skimage threshold function: threshold_{method}")
+ min_func = QuickAdapterRegressorV3.apply_skimage_threshold
+ max_func = QuickAdapterRegressorV3.apply_skimage_threshold
+
min_val = min_func(pred_minima, threshold_func)
if not np.isfinite(min_val):
min_val = QuickAdapterRegressorV3.safe_min_pred(pred_extrema)
self, df: DataFrame, trade: Trade, exit_stage: int
) -> Optional[float]:
natr_ratio_percent = (
- self.partial_exit_stages[exit_stage][0]
- if exit_stage in self.partial_exit_stages
+ QuickAdapterV3.partial_exit_stages[exit_stage][0]
+ if exit_stage in QuickAdapterV3.partial_exit_stages
else 1.0
)
take_profit_distance = self.get_take_profit_distance(
return None
trade_exit_stage = QuickAdapterV3.get_trade_exit_stage(trade)
- if trade_exit_stage not in self.partial_exit_stages:
+ if trade_exit_stage not in QuickAdapterV3.partial_exit_stages:
return None
df, _ = self.dp.get_analyzed_dataframe(
min_stake = 0.0
if min_stake > trade.stake_amount:
return None
- trade_stake_percent = self.partial_exit_stages[trade_exit_stage][1]
+ trade_stake_percent = QuickAdapterV3.partial_exit_stages[trade_exit_stage][
+ 1
+ ]
trade_partial_stake_amount = trade_stake_percent * trade.stake_amount
remaining_stake_amount = trade.stake_amount - trade_partial_stake_amount
if remaining_stake_amount < min_stake:
"""
if df.empty:
return False
- if side not in self._trade_directions_set():
+ if side not in QuickAdapterV3._trade_directions_set():
return False
- if order not in self._order_types_set():
+ if order not in QuickAdapterV3._order_types_set():
return False
trade_direction = side
return "maxima_detected_long"
trade_exit_stage = QuickAdapterV3.get_trade_exit_stage(trade)
- if trade_exit_stage in self.partial_exit_stages:
+ if trade_exit_stage in QuickAdapterV3.partial_exit_stages:
return None
trade_take_profit_price = self.get_take_profit_price(
side: str,
**kwargs,
) -> bool:
- if side not in self._trade_directions_set():
+ if side not in QuickAdapterV3._trade_directions_set():
return False
if (
side == QuickAdapterV3._TRADE_DIRECTIONS[1] and not self.can_short
repositories / freqai-strategies.git / commitdiff
