| freqai.label_smoothing.window_candles | 5 | int >= 3 | Smoothing window length (candles). |
| freqai.label_smoothing.beta | 8.0 | float > 0 | Shape parameter for `kaiser` and `kaiser_bessel_derived` kernels. |
| freqai.label_smoothing.polyorder | 3 | int >= 0 | Polynomial order for `savgol` smoothing. |
-| freqai.label_smoothing.mode | `mirror` | enum {`mirror`,`constant`,`nearest`,`wrap`,`interp`} | Boundary mode for `savgol` and `gaussian_filter1d`. |
+| freqai.label_smoothing.mode | `mirror` | `savgol`: enum {`mirror`,`constant`,`nearest`,`wrap`,`interp`}; `gaussian_filter1d`: enum {`mirror`,`constant`,`nearest`,`wrap`}; ignored otherwise | Boundary mode for `savgol` and `gaussian_filter1d`. |
| freqai.label_smoothing.sigma | 1.0 | float > 0 | Gaussian `sigma` for `gaussian_filter1d` smoothing. |
| _Label weighting_ | | | |
| freqai.label_weighting.strategy | `none` | enum {`none`,`uniform`,`amplitude`,`amplitude_threshold_ratio`,`volume_rate`,`speed`,`efficiency_ratio`,`volume_weighted_efficiency_ratio`,`combined`} | Label weighting metric: none (`none`), uniform unit weight on every detected pivot (`uniform`), swing amplitude (`amplitude`), swing amplitude / median volatility-threshold ratio (`amplitude_threshold_ratio`), swing volume per candle (`volume_rate`), swing speed (`speed`), swing efficiency ratio (`efficiency_ratio`), swing volume-weighted efficiency ratio (`volume_weighted_efficiency_ratio`), or combined metrics aggregation (`combined`). Switching between `none` and any other strategy requires deleting trained models to realign training emphasis. |
)
from Utils import (
- as_dict,
enum_error_message,
- DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES,
DEFAULT_MAX_LABEL_NATR_MULTIPLIER,
DEFAULT_MAX_LABEL_PERIOD_CANDLES,
DEFAULT_MIN_LABEL_NATR_MULTIPLIER,
format_dict,
format_number,
get_causal_mode,
+ get_fit_live_predictions_candles,
get_label_defaults,
get_label_horizon_candles,
get_label_pipeline_config,
}
_POWER_MEAN_METRICS_SET: Final[frozenset[str]] = frozenset(_POWER_MEAN_MAP)
- FIT_LIVE_PREDICTIONS_CANDLES_DEFAULT: Final[int] = (
- DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES
- )
MIN_LABEL_PERIOD_CANDLES_DEFAULT: Final[int] = DEFAULT_MIN_LABEL_PERIOD_CANDLES
MAX_LABEL_PERIOD_CANDLES_DEFAULT: Final[int] = DEFAULT_MAX_LABEL_PERIOD_CANDLES
MIN_LABEL_NATR_MULTIPLIER_DEFAULT: Final[float] = DEFAULT_MIN_LABEL_NATR_MULTIPLIER
@cached_property
def label_weighting(self) -> dict[str, Any]:
return get_label_weighting_config(
- as_dict(self.freqai_info.get("label_weighting")), logger
+ self.freqai_info.get("label_weighting"), logger
)
@cached_property
def label_pipeline(self) -> dict[str, Any]:
- return get_label_pipeline_config(
- as_dict(self.freqai_info.get("label_pipeline")), logger
- )
+ return get_label_pipeline_config(self.freqai_info.get("label_pipeline"), logger)
@cached_property
def label_prediction(self) -> dict[str, Any]:
return get_label_prediction_config(
- as_dict(self.freqai_info.get("label_prediction")), logger
+ self.freqai_info.get("label_prediction"), logger
)
@cached_property
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
migrate_config(self.config, logger)
+ self._fit_live_predictions_candles: int = get_fit_live_predictions_candles(
+ self.freqai_info, logger
+ )
self.pairs: list[str] = self.config.get("exchange", {}).get("pair_whitelist")
if not self.pairs:
raise ValueError(
)
logger.info("Label Hyperparameters:")
logger.info(
- f" fit_live_predictions_candles: {self.freqai_info.get('fit_live_predictions_candles', QuickAdapterRegressorV3.FIT_LIVE_PREDICTIONS_CANDLES_DEFAULT)}"
+ f" fit_live_predictions_candles: {self._fit_live_predictions_candles}"
)
if self._optuna_hyperopt:
logger.info(
def fit_live_predictions(self, dk: FreqaiDataKitchen, pair: str) -> None:
warmed_up = True
- fit_live_predictions_candles = self.freqai_info.get(
- "fit_live_predictions_candles",
- QuickAdapterRegressorV3.FIT_LIVE_PREDICTIONS_CANDLES_DEFAULT,
- )
+ fit_live_predictions_candles = self._fit_live_predictions_candles
if self._optuna_hyperopt:
self.optuna_throttle_callback(
"wrap",
"interp",
)
+SMOOTHING_METHOD_MODES: Final[dict[SmoothingMethod, tuple[SmoothingMode, ...]]] = {
+ SMOOTHING_METHODS[7]: SMOOTHING_MODES, # "savgol"
+ SMOOTHING_METHODS[8]: SMOOTHING_MODES[:-1], # "gaussian_filter1d"
+}
DEFAULTS_LABEL_SMOOTHING: Final[dict[str, Any]] = {
"method": SMOOTHING_METHODS[1], # "gaussian"
COMBINED_AGGREGATIONS,
FILL_METHODS,
SMOOTHING_METHODS,
+ SMOOTHING_METHOD_MODES,
SMOOTHING_MODES,
WEIGHT_STRATEGIES,
get_label_column_config,
from scipy.stats import pearsonr, t
from technical.pivots_points import pivots_points
from Utils import (
- as_dict,
+ DEFAULTS_EXIT_THRESHOLDS_CALIBRATION,
_OPTUNA_NAMESPACES,
- DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES,
EXTREMA_COLUMN,
EXTREMA_DIRECTION_COLUMN,
EXTREMA_DIRECTION_SMOOTHED_COLUMN,
get_callable_sha256,
get_causal_mode,
get_distance,
+ get_custom_protections_config,
get_exit_pricing_config,
+ get_exit_thresholds_calibration_config,
+ get_fit_live_predictions_candles,
get_label_defaults,
get_label_horizon_candles,
get_label_smoothing_config,
_TRADING_MODE_FUTURES: Final[str] = _TRADING_MODES[2]
_SMOOTHING_SMM: Final[str] = SMOOTHING_METHODS[5]
_SMOOTHING_SAVGOL: Final[str] = SMOOTHING_METHODS[7]
- _SMOOTHING_GAUSSIAN_FILTER1D: Final[str] = SMOOTHING_METHODS[8]
_FILL_EPSILON: Final[str] = FILL_METHODS[1]
_FILL_GAUSSIAN: Final[str] = FILL_METHODS[2]
_FILL_EPSILON_GAUSSIAN: Final[str] = FILL_METHODS[3]
"t_decl_a": 0.675,
}
- default_exit_thresholds_calibration: ClassVar[dict[str, float]] = {
- "decline_quantile": 0.5,
- }
+ default_exit_thresholds_calibration: ClassVar[dict[str, float]] = (
+ DEFAULTS_EXIT_THRESHOLDS_CALIBRATION
+ )
position_adjustment_enable = True
# strict `remaining < min_exit_stake` guard.
_PARTIAL_EXIT_MIN_STAKE_MARGIN: Final[float] = 1e-3
- minimal_roi = {str(timeframe_minutes * 864): -1}
-
# FreqAI is crashing if minimal_roi is a property
- # @property
- # def minimal_roi(self) -> dict[str, Any]:
- # timeframe_minutes = self.timeframe_minutes
- # fit_live_predictions_candles = int(
- # self.config.get("freqai", {}).get(
- # "fit_live_predictions_candles", DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES
- # )
- # )
- # return {str(timeframe_minutes * fit_live_predictions_candles): -1}
-
- # @minimal_roi.setter
- # def minimal_roi(self, value: dict[str, Any]) -> None:
- # pass
+ minimal_roi = {str(timeframe_minutes * 864): -1}
process_only_new_candles = True
},
}
- @property
+ @cached_property
+ def _fit_live_predictions_candles(self) -> int:
+ return get_fit_live_predictions_candles(self.config.get("freqai"), logger)
+
+ @cached_property
def protections(self) -> list[dict[str, Any]]:
- fit_live_predictions_candles = int(
- self.config.get("freqai", {}).get(
- "fit_live_predictions_candles", DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES
- )
- )
- protections = self.config.get("custom_protections", {})
- trade_duration_candles = int(protections.get("trade_duration_candles", 72))
- lookback_period_fraction = float(
- protections.get("lookback_period_fraction", 0.5)
+ fit_live_predictions_candles = self._fit_live_predictions_candles
+ protections = get_custom_protections_config(
+ self.config.get("custom_protections"), logger
)
+ trade_duration_candles = protections["trade_duration_candles"]
+ lookback_period_fraction = protections["lookback_period_fraction"]
lookback_period_candles = max(
1, int(round(fit_live_predictions_candles * lookback_period_fraction))
)
- cooldown = protections.get("cooldown", {})
- cooldown_stop_duration_candles = int(cooldown.get("stop_duration_candles", 4))
+ cooldown = protections["cooldown"]
+ cooldown_stop_duration_candles = cooldown["stop_duration_candles"]
stoploss_stop_duration_candles = max(
cooldown_stop_duration_candles, trade_duration_candles
)
protections_list = []
- if cooldown.get("enabled", True):
+ if cooldown["enabled"]:
protections_list.append(
{
"method": "CooldownPeriod",
}
)
- drawdown = protections.get("drawdown", {})
- if drawdown.get("enabled", True):
+ drawdown = protections["drawdown"]
+ if drawdown["enabled"]:
protections_list.append(
{
"method": "MaxDrawdown",
"lookback_period_candles": lookback_period_candles,
"trade_limit": 2 * max_open_trades,
"stop_duration_candles": drawdown_stop_duration_candles,
- "max_allowed_drawdown": float(
- drawdown.get("max_allowed_drawdown", 0.2)
- ),
+ "max_allowed_drawdown": drawdown["max_allowed_drawdown"],
}
)
- stoploss = protections.get("stoploss", {})
- if stoploss.get("enabled", True):
+ stoploss = protections["stoploss"]
+ if stoploss["enabled"]:
protections_list.append(
{
"method": "StoplossGuard",
@property
def startup_candle_count(self) -> int:
# Match the predictions warmup period
- return self.config.get("freqai", {}).get(
- "fit_live_predictions_candles", DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES
- )
+ return self._fit_live_predictions_candles
@property
def max_open_trades_per_side(self) -> int:
@cached_property
def label_weighting(self) -> dict[str, Any]:
return get_label_weighting_config(
- as_dict(self.freqai_info.get("label_weighting")), logger
+ self.freqai_info.get("label_weighting"), logger
)
@cached_property
def label_smoothing(self) -> dict[str, Any]:
return get_label_smoothing_config(
- as_dict(self.freqai_info.get("label_smoothing")), logger
+ self.freqai_info.get("label_smoothing"), logger
)
@cached_property
def trade_price_target_method(self) -> str:
- return get_exit_pricing_config(
- as_dict(self.config.get("exit_pricing")), logger
- )["trade_price_target_method"]
+ return get_exit_pricing_config(self.config.get("exit_pricing"), logger)[
+ "trade_price_target_method"
+ ]
@cached_property
def reversal_confirmation(self) -> dict[str, int | float]:
return get_reversal_confirmation_config(
- as_dict(self.config.get("reversal_confirmation")), logger
+ self.config.get("reversal_confirmation"), logger
)
@cached_property
label_col, label_smoothing["default"], label_smoothing["columns"]
)
if (
- col_smoothing_config["method"]
- in (
- QuickAdapterV3._SMOOTHING_SAVGOL,
- QuickAdapterV3._SMOOTHING_GAUSSIAN_FILTER1D,
- )
+ col_smoothing_config["method"] in SMOOTHING_METHOD_MODES
and col_smoothing_config["mode"] == SMOOTHING_MODES[3]
): # "wrap"
raise ValueError(
f"{self._pnl_momentum_window_size} candles "
f"(~{velocity_span_minutes} min velocity span)."
)
- self._exit_thresholds_calibration: dict[str, float] = {
- **QuickAdapterV3.default_exit_thresholds_calibration,
- **self.config.get("exit_pricing", {}).get("thresholds_calibration", {}),
- }
+ self._exit_thresholds_calibration: dict[str, float] = (
+ get_exit_thresholds_calibration_config(
+ self.config.get("exit_pricing"),
+ logger,
+ self.default_exit_thresholds_calibration,
+ )
+ )
self._candle_deviation_cache: dict[CandleDeviationCacheKey, float] = {}
self._candle_threshold_cache: dict[CandleThresholdCacheKey, float] = {}
self._cached_df_signature: dict[str, DfSignature] = {}
f"supported values are {', '.join(QuickAdapterV3._TRADING_MODES)}"
)
+ @cached_property
+ def _configured_leverage(self) -> Optional[float]:
+ leverage = self.config.get("leverage")
+ if leverage is None:
+ return None
+ if not is_finite_number(leverage):
+ logger.warning(
+ f"Invalid leverage value {leverage!r}: must be a finite number, "
+ "using proposed_leverage"
+ )
+ return None
+ leverage = float(leverage)
+ if leverage < 1.0:
+ logger.warning(
+ f"Invalid leverage value {leverage}: must be >= 1.0, clamping to 1.0"
+ )
+ return leverage
+
def leverage(
self,
pair: str,
side: str,
**kwargs: Any,
) -> float:
- return min(self.config.get("leverage", proposed_leverage), max_leverage)
+ configured_leverage = self._configured_leverage
+ if configured_leverage is None:
+ configured_leverage = proposed_leverage
+ return float(max(1.0, min(configured_leverage, max_leverage)))
def plot_annotations(
self,
LABEL_WEIGHT_SUPPORT_POLICIES,
NORMALIZATION_TYPES,
PREDICTION_METHODS,
+ SMOOTHING_METHOD_MODES,
SMOOTHING_METHODS,
SMOOTHING_MODES,
STANDARDIZATION_TYPES,
FillEpsilonBaseline,
SmoothingMethod,
SmoothingMode,
+ get_label_column_config,
)
from numpy.typing import NDArray
from scipy.ndimage import gaussian_filter1d
require_int: bool = False
def __call__(self, value: Any) -> bool:
+ if isinstance(value, bool):
+ return False
if self.require_int and not isinstance(value, int):
return False
if not isinstance(value, (int, float)) or not _is_finite_value(value):
return "must be a mapping"
-_Validator = _EnumValidator | _NumericValidator | _RangeValidator | _DictValidator
+@dataclass(frozen=True, slots=True)
+class _BoolValidator:
+ def __call__(self, value: Any) -> bool:
+ return isinstance(value, bool)
+
+ def message(self, param: str) -> str:
+ return "must be a boolean"
+
+
+_Validator = (
+ _EnumValidator
+ | _NumericValidator
+ | _RangeValidator
+ | _DictValidator
+ | _BoolValidator
+)
@dataclass(frozen=True, slots=True)
return value if isinstance(value, dict) else {}
+def as_config_section(value: Any, name: str, logger: Logger) -> dict[str, Any]:
+ if value is not None and not isinstance(value, dict):
+ logger.warning(
+ f"Invalid {name} value {value!r}: must be a mapping, using defaults"
+ )
+ return as_dict(value)
+
+
def enum_error_message(ctx: str, value: Any, options: Sequence[str]) -> str:
return f"Invalid {ctx} value {value!r}: supported values are {', '.join(options)}"
ValidateParamsFn = Callable[[dict[str, Any], Logger, str], dict[str, Any]]
+CrossFieldValidatorFn = Callable[[dict[str, Any], str], None]
_MISSING: Final = object()
return {"default": validated_default, "columns": {}}
-_LABEL_KIND_REGISTRY: Final[dict[str, tuple[dict[str, _ParamSpec], dict[str, Any]]]] = {
- "label_weighting": (_WEIGHTING_SPECS, DEFAULTS_LABEL_WEIGHTING),
- "label_pipeline": (_PIPELINE_SPECS, DEFAULTS_LABEL_PIPELINE),
- "label_smoothing": (_SMOOTHING_SPECS, DEFAULTS_LABEL_SMOOTHING),
- "label_prediction": (_PREDICTION_SPECS, DEFAULTS_LABEL_PREDICTION),
+def _validate_smoothing_method_mode(
+ config: dict[str, Any],
+ config_name: str,
+) -> None:
+ method = config["method"]
+ valid_modes = SMOOTHING_METHOD_MODES.get(method)
+ if valid_modes is not None and config["mode"] not in valid_modes:
+ raise ValueError(
+ f"Invalid {config_name} mode value {config['mode']!r} for "
+ f"method {method!r}: supported values are {', '.join(valid_modes)}"
+ )
+
+
+_LABEL_KIND_REGISTRY: Final[
+ dict[
+ str,
+ tuple[dict[str, _ParamSpec], dict[str, Any], CrossFieldValidatorFn | None],
+ ]
+] = {
+ "label_weighting": (_WEIGHTING_SPECS, DEFAULTS_LABEL_WEIGHTING, None),
+ "label_pipeline": (_PIPELINE_SPECS, DEFAULTS_LABEL_PIPELINE, None),
+ "label_smoothing": (
+ _SMOOTHING_SPECS,
+ DEFAULTS_LABEL_SMOOTHING,
+ _validate_smoothing_method_mode,
+ ),
+ "label_prediction": (_PREDICTION_SPECS, DEFAULTS_LABEL_PREDICTION, None),
}
def _label_kind_validator(kind: str) -> ValidateParamsFn:
- specs, defaults = _LABEL_KIND_REGISTRY[kind]
+ specs, defaults, _ = _LABEL_KIND_REGISTRY[kind]
def validate(
config: dict[str, Any],
def get_label_kind_config(
kind: str,
- config: dict[str, Any],
+ config: Any,
logger: Logger,
) -> dict[str, Any]:
if kind not in _LABEL_KIND_REGISTRY:
f"Unknown label kind {kind!r}: supported values are "
f"{', '.join(_LABEL_KIND_REGISTRY)}"
)
- _, defaults = _LABEL_KIND_REGISTRY[kind]
- return _get_label_config(
+ config = as_config_section(config, kind, logger)
+ _, defaults, cross_field_validator = _LABEL_KIND_REGISTRY[kind]
+ validated = _get_label_config(
config, logger, kind, _label_kind_validator(kind), defaults
)
+ if cross_field_validator is not None:
+ for label_col in LABEL_COLUMNS:
+ cross_field_validator(
+ get_label_column_config(
+ label_col, validated["default"], validated["columns"]
+ ),
+ f"{kind} for label {label_col!r}",
+ )
+ return validated
def get_label_weighting_config(
}
-def get_exit_pricing_config(config: dict[str, Any], logger: Logger) -> dict[str, str]:
+def get_exit_pricing_config(config: Any, logger: Logger) -> dict[str, str]:
return _validate_params(
- config, logger, "exit_pricing", _EXIT_PRICING_SPECS, DEFAULTS_EXIT_PRICING
+ as_config_section(config, "exit_pricing", logger),
+ logger,
+ "exit_pricing",
+ _EXIT_PRICING_SPECS,
+ DEFAULTS_EXIT_PRICING,
)
+DEFAULTS_EXIT_THRESHOLDS_CALIBRATION: Final[dict[str, Any]] = {
+ "decline_quantile": 0.5,
+}
+
+_EXIT_THRESHOLDS_CALIBRATION_SPECS: Final[dict[str, _ParamSpec]] = {
+ "decline_quantile": _ParamSpec(
+ _NumericValidator(
+ min_value=0, max_value=1, min_exclusive=True, max_exclusive=True
+ ),
+ output_type=float,
+ ),
+}
+
+
+def get_exit_thresholds_calibration_config(
+ config: Any,
+ logger: Logger,
+ overrides: dict[str, Any] | None = None,
+) -> dict[str, float]:
+ # exit_pricing mapping warning owned by get_exit_pricing_config (avoid double-warn)
+ config = as_dict(config)
+ # validate the override so an invalid subclass default falls back to the canonical default
+ defaults = _validate_params(
+ overrides or {},
+ logger,
+ "exit_pricing.thresholds_calibration",
+ _EXIT_THRESHOLDS_CALIBRATION_SPECS,
+ DEFAULTS_EXIT_THRESHOLDS_CALIBRATION,
+ )
+ return _validate_params(
+ as_config_section(
+ config.get("thresholds_calibration"),
+ "exit_pricing.thresholds_calibration",
+ logger,
+ ),
+ logger,
+ "exit_pricing.thresholds_calibration",
+ _EXIT_THRESHOLDS_CALIBRATION_SPECS,
+ defaults,
+ )
+
+
+DEFAULTS_CUSTOM_PROTECTIONS: Final[dict[str, Any]] = {
+ "trade_duration_candles": 72,
+ "lookback_period_fraction": 0.5,
+}
+
+DEFAULTS_COOLDOWN_PROTECTION: Final[dict[str, Any]] = {
+ "enabled": True,
+ "stop_duration_candles": 4,
+}
+
+DEFAULTS_DRAWDOWN_PROTECTION: Final[dict[str, Any]] = {
+ "enabled": True,
+ "max_allowed_drawdown": 0.2,
+}
+
+DEFAULTS_STOPLOSS_PROTECTION: Final[dict[str, Any]] = {
+ "enabled": True,
+}
+
+_CUSTOM_PROTECTIONS_SPECS: Final[dict[str, _ParamSpec]] = {
+ "trade_duration_candles": _ParamSpec(
+ _NumericValidator(min_value=1, require_int=True), output_type=int
+ ),
+ "lookback_period_fraction": _ParamSpec(
+ _NumericValidator(min_value=0, max_value=1, min_exclusive=True),
+ output_type=float,
+ ),
+}
+
+_COOLDOWN_PROTECTION_SPECS: Final[dict[str, _ParamSpec]] = {
+ "enabled": _ParamSpec(_BoolValidator()),
+ "stop_duration_candles": _ParamSpec(
+ _NumericValidator(min_value=1, require_int=True), output_type=int
+ ),
+}
+
+_DRAWDOWN_PROTECTION_SPECS: Final[dict[str, _ParamSpec]] = {
+ "enabled": _ParamSpec(_BoolValidator()),
+ "max_allowed_drawdown": _ParamSpec(
+ _NumericValidator(
+ min_value=0, max_value=1, min_exclusive=True, max_exclusive=True
+ ),
+ output_type=float,
+ ),
+}
+
+_STOPLOSS_PROTECTION_SPECS: Final[dict[str, _ParamSpec]] = {
+ "enabled": _ParamSpec(_BoolValidator()),
+}
+
+
+def get_custom_protections_config(config: Any, logger: Logger) -> dict[str, Any]:
+ config = as_config_section(config, "custom_protections", logger)
+ validated = _validate_params(
+ config,
+ logger,
+ "custom_protections",
+ _CUSTOM_PROTECTIONS_SPECS,
+ DEFAULTS_CUSTOM_PROTECTIONS,
+ )
+ validated["cooldown"] = _validate_params(
+ as_config_section(
+ config.get("cooldown"), "custom_protections.cooldown", logger
+ ),
+ logger,
+ "custom_protections.cooldown",
+ _COOLDOWN_PROTECTION_SPECS,
+ DEFAULTS_COOLDOWN_PROTECTION,
+ )
+ validated["drawdown"] = _validate_params(
+ as_config_section(
+ config.get("drawdown"), "custom_protections.drawdown", logger
+ ),
+ logger,
+ "custom_protections.drawdown",
+ _DRAWDOWN_PROTECTION_SPECS,
+ DEFAULTS_DRAWDOWN_PROTECTION,
+ )
+ validated["stoploss"] = _validate_params(
+ as_config_section(
+ config.get("stoploss"), "custom_protections.stoploss", logger
+ ),
+ logger,
+ "custom_protections.stoploss",
+ _STOPLOSS_PROTECTION_SPECS,
+ DEFAULTS_STOPLOSS_PROTECTION,
+ )
+ return validated
+
+
+_FIT_LIVE_PREDICTIONS_SPECS: Final[dict[str, _ParamSpec]] = {
+ "fit_live_predictions_candles": _ParamSpec(
+ _NumericValidator(min_value=1, require_int=True), output_type=int
+ ),
+}
+
+
+def get_fit_live_predictions_candles(config: Any, logger: Logger) -> int:
+ return _validate_params(
+ as_config_section(config, "freqai", logger),
+ logger,
+ "freqai",
+ _FIT_LIVE_PREDICTIONS_SPECS,
+ {"fit_live_predictions_candles": DEFAULT_FIT_LIVE_PREDICTIONS_CANDLES},
+ )["fit_live_predictions_candles"]
+
+
DEFAULTS_REVERSAL_CONFIRMATION: Final[dict[str, Any]] = {
"lookback_period_candles": 0,
"decay_fraction": 0.5,
def get_reversal_confirmation_config(
- config: dict[str, Any], logger: Logger
+ config: Any, logger: Logger
) -> dict[str, int | float]:
+ config = as_config_section(config, "reversal_confirmation", logger)
validated = _validate_params(
config,
logger,
repositories / freqai-strategies.git / commitdiff
