"retain_previous",
}
+# Supported trading modes
+TRADING_MODES: Tuple[str, ...] = ("spot", "margin", "futures")
+
+# Supported p-value adjustment methods
+ADJUST_METHODS: Tuple[str, ...] = ("none", "benjamini_hochberg")
+# Alias without underscore for convenience
+_ADJUST_METHODS_ALIASES: frozenset[str] = frozenset({"benjaminihochberg"})
+
DEFAULT_MODEL_REWARD_PARAMETERS: RewardParams = {
"invalid_action": -2.0,
if text in {"false", "0", "no", "n", "off"}:
return False
# Unsupported type
- raise ValueError(f"Unrecognized boolean literal: {value!r}")
+ raise ValueError(f"Param: unrecognized boolean literal {value!r}")
def _get_bool_param(params: RewardParams, key: str, default: Optional[bool] = None) -> bool:
if "min" in effective_bounds and adjusted < float(effective_bounds["min"]):
if strict:
raise ValueError(
- f"Parameter '{key}'={adjusted} below min {float(effective_bounds['min'])}"
+ f"Param: '{key}'={adjusted} below min {float(effective_bounds['min'])}"
)
adjusted = float(effective_bounds["min"])
reason_parts.append(f"min={float(effective_bounds['min'])}")
if "max" in effective_bounds and adjusted > float(effective_bounds["max"]):
if strict:
raise ValueError(
- f"Parameter '{key}'={adjusted} above max {float(effective_bounds['max'])}"
+ f"Param: '{key}'={adjusted} above max {float(effective_bounds['max'])}"
)
adjusted = float(effective_bounds["max"])
reason_parts.append(f"max={float(effective_bounds['max'])}")
if not np.isfinite(adjusted):
if strict:
- raise ValueError(f"Parameter '{key}' is non-finite: {adjusted}")
+ raise ValueError(f"Param: '{key}' is non-finite: {adjusted}")
adjusted = float(effective_bounds.get("min", 0.0))
reason_parts.append("non_finite_reset")
def _is_short_allowed(trading_mode: str) -> bool:
mode = trading_mode.lower()
- if mode in {"margin", "futures"}:
+ if mode in TRADING_MODES[1:]: # "margin", "futures"
return True
- if mode == "spot":
+ if mode == TRADING_MODES[0]: # "spot"
return False
- raise ValueError("Unsupported trading mode. Expected one of: spot, margin, futures")
+ raise ValueError(
+ f"Config: unsupported trading mode '{mode}'. Expected one of: {list(TRADING_MODES)}"
+ )
def _fail_safely(reason: str) -> float:
}
continue
if strict:
- raise ValueError(f"Parameter '{key}' is non-numeric or invalid: {original_val!r}")
+ raise ValueError(f"Param: '{key}' is non-numeric or invalid: {original_val!r}")
adjusted = bounds.get("min", 0.0)
sanitized[key] = adjusted
adjustments[key] = {
parsed: RewardParams = {}
for override in overrides:
if "=" not in override:
- raise ValueError(f"Invalid override format: '{override}'")
+ raise ValueError(f"CLI: invalid override format '{override}'. Expected 'key=value'")
key, value = override.split("=", 1)
try:
parsed[key] = float(value)
"""Return count/mean/std/min/max of target grouped by clipped bins of column."""
bins_arr = np.asarray(list(bins), dtype=float)
if bins_arr.ndim != 1 or bins_arr.size < 2:
- raise ValueError("bins must contain at least two edges")
+ raise ValueError("Stats: bins must contain at least two edges")
clipped = df[column].clip(lower=float(bins_arr[0]), upper=float(bins_arr[-1]))
categories = pd.cut(
clipped,
or permutation_importance is None
or r2_score is None
):
- raise ImportError("scikit-learn is not available; skipping feature analysis.")
+ raise ImportError("Feature analysis: scikit-learn is not available")
canonical_features = [
"pnl",
with path.open("rb") as f:
episodes_data = pickle.load(f)
except Exception as e:
- raise ValueError(f"Failed to unpickle '{path}': {e!r}") from e
+ raise ValueError(f"Data: failed to unpickle '{path}': {e!r}") from e
# Top-level dict with 'transitions'
if isinstance(episodes_data, dict) and "transitions" in episodes_data:
df = pd.DataFrame(list(candidate))
except TypeError:
raise ValueError(
- f"Top-level 'transitions' in '{path}' is not iterable (type {type(candidate)!r})."
+ f"Data: 'transitions' in '{path}' is not iterable (type {type(candidate)!r})"
)
except Exception as e:
raise ValueError(
- f"Could not build DataFrame from top-level 'transitions' in '{path}': {e!r}"
+ f"Data: could not build DataFrame from 'transitions' in '{path}': {e!r}"
) from e
# List of episodes where some entries have 'transitions'
elif isinstance(episodes_data, list) and any(
all_transitions.extend(list(trans))
except TypeError:
raise ValueError(
- f"Episode 'transitions' is not iterable in file '{path}'; found type {type(trans)!r}"
+ f"Data: episode 'transitions' is not iterable in '{path}' (type {type(trans)!r})"
)
else:
skipped += 1
df = pd.DataFrame(all_transitions)
except Exception as e:
raise ValueError(
- f"Could not build DataFrame from flattened transitions in '{path}': {e!r}"
+ f"Data: could not build DataFrame from transitions in '{path}': {e!r}"
) from e
else:
try:
df = pd.DataFrame(episodes_data)
except Exception as e:
raise ValueError(
- f"Could not convert pickled object from '{path}' to DataFrame: {e!r}"
+ f"Data: could not convert pickled object from '{path}' to DataFrame: {e!r}"
) from e
# Coerce common numeric fields; warn when values are coerced to NaN
if missing_required:
if enforce_columns:
raise ValueError(
- f"Loaded episodes data is missing required columns: {sorted(missing_required)}. "
- f"Found columns: {sorted(list(df.columns))}."
+ f"Data: missing required columns {sorted(missing_required)}. "
+ f"Found: {sorted(list(df.columns))}"
)
warnings.warn(
f"Missing columns {sorted(missing_required)}; filled with NaN when loading (enforce_columns=False)",
def statistical_hypothesis_tests(
- df: pd.DataFrame, *, adjust_method: str = "none", seed: int = 42
+ df: pd.DataFrame, *, adjust_method: str = ADJUST_METHODS[0], seed: int = 42
) -> Dict[str, Any]:
"""Statistical hypothesis tests (Spearman, Kruskal-Wallis, Mann-Whitney).
}
# Optional multiple testing correction (Benjamini-Hochberg)
- if adjust_method not in {"none", "benjamini_hochberg", "benjaminihochberg"}:
+ _valid_adjust = set(ADJUST_METHODS) | _ADJUST_METHODS_ALIASES
+ if adjust_method not in _valid_adjust:
raise ValueError(
- "Unsupported adjust_method. Use 'none', 'benjamini_hochberg', or 'benjaminihochberg'."
+ f"Stats: unsupported adjust_method '{adjust_method}'. "
+ f"Expected one of: {list(ADJUST_METHODS)}"
)
- if adjust_method in {"benjamini_hochberg", "benjaminihochberg"} and results:
+ if adjust_method in _valid_adjust - {ADJUST_METHODS[0]} and results:
# Collect p-values
items = list(results.items())
pvals = np.array([v[1]["p_value"] for v in items])
parser.add_argument(
"--trading_mode",
type=str.lower,
- choices=["spot", "margin", "futures"],
- default="spot",
+ choices=list(TRADING_MODES),
+ default=TRADING_MODES[0],
help=("Trading mode to simulate (spot disables shorts). Default: spot."),
)
parser.add_argument(
parser.add_argument(
"--pvalue_adjust",
type=str.lower,
- choices=["none", "benjamini_hochberg"],
- default="none",
+ choices=list(ADJUST_METHODS),
+ default=ADJUST_METHODS[0],
help="Multiple testing correction method for hypothesis tests (default: none).",
)
parser.add_argument(
seed: int,
real_df: Optional[pd.DataFrame] = None,
*,
- adjust_method: str = "none",
+ adjust_method: str = ADJUST_METHODS[0],
stats_seed: Optional[int] = None,
strict_diagnostics: bool = False,
bootstrap_resamples: int = 10000,
self.pairs: List[str] = self.config.get("exchange", {}).get("pair_whitelist")
if not self.pairs:
raise ValueError(
- "FreqAI model requires StaticPairList method defined in pairlists configuration and pair_whitelist defined in exchange section configuration"
+ "Config: missing 'pair_whitelist' in exchange section "
+ "or StaticPairList method not defined in pairlists configuration"
)
self.action_masking: bool = (
self.model_type == ReforceXY._MODEL_TYPES[2]
train_df = data_dictionary.get("train_features")
train_timesteps = len(train_df)
if train_timesteps <= 0:
- raise ValueError("train_features dataframe has zero length")
+ raise ValueError("Training: train_features dataframe has zero length")
test_df = data_dictionary.get("test_features")
eval_timesteps = len(test_df)
train_cycles = max(1, int(self.rl_config.get("train_cycles", 25)))
)
else:
raise ValueError(
- f"Unsupported storage backend: {storage_backend}. Supported backends are: {', '.join(ReforceXY._STORAGE_BACKENDS)}"
+ f"Hyperopt: unsupported storage backend '{storage_backend}'. "
+ f"Expected one of: {list(ReforceXY._STORAGE_BACKENDS)}"
)
return storage
)
else:
raise ValueError(
- f"Unsupported sampler: {sampler}. Supported samplers: {', '.join(ReforceXY._SAMPLER_TYPES)}"
+ f"Hyperopt: unsupported sampler '{sampler}'. "
+ f"Expected one of: {list(ReforceXY._SAMPLER_TYPES)}"
)
@staticmethod
elif ReforceXY._MODEL_TYPES[3] in self.model_type:
return sample_params_dqn(trial)
else:
- raise NotImplementedError(f"{self.model_type} not supported for hyperopt")
+ raise NotImplementedError(
+ f"Hyperopt: model type '{self.model_type}' not supported"
+ )
def objective(
self, trial: Trial, dk: FreqaiDataKitchen, total_timesteps: int
lr = optuna_params.get("learning_rate")
if lr is None:
- raise ValueError(f"missing {'learning_rate'} in optuna params for {model_type}")
+ raise ValueError(f"Optuna: missing 'learning_rate' in params for {model_type}")
lr = get_schedule(
optuna_params.get("lr_schedule", ReforceXY._SCHEDULE_TYPES[1]), float(lr)
) # default: "constant"
]
for param in required_ppo_params:
if optuna_params.get(param) is None:
- raise ValueError(f"missing '{param}' in optuna params for {model_type}")
+ raise ValueError(
+ f"Optuna: missing '{param}' in params for {model_type}"
+ )
cr = optuna_params.get("clip_range")
cr = get_schedule(
optuna_params.get("cr_schedule", ReforceXY._SCHEDULE_TYPES[1]),
]
for param in required_dqn_params:
if optuna_params.get(param) is None:
- raise ValueError(f"missing '{param}' in optuna params for {model_type}")
+ raise ValueError(
+ f"Optuna: missing '{param}' in params for {model_type}"
+ )
train_freq = optuna_params.get("train_freq")
subsample_steps = optuna_params.get("subsample_steps")
gradient_steps = compute_gradient_steps(train_freq, subsample_steps)
): # "QRDQN"
policy_kwargs["n_quantiles"] = int(optuna_params["n_quantiles"])
else:
- raise ValueError(f"Model {model_type} not supported")
+ raise ValueError(f"Optuna: model type '{model_type}' not supported")
if optuna_params.get("net_arch"):
net_arch_value = str(optuna_params["net_arch"])
elif trading_mode == RLAgentStrategy._TRADING_MODES[2]:
return False
else:
- raise ValueError(f"Invalid trading_mode: {trading_mode}")
+ raise ValueError(
+ f"Config: invalid trading_mode '{trading_mode}'. "
+ f"Expected one of: {list(RLAgentStrategy._TRADING_MODES)}"
+ )
ExtremaSelectionMethod = Literal["rank_extrema", "rank_peaks", "partition"]
OptunaNamespace = Literal["hp", "train", "label"]
+ClusterSelectionMethod = Literal["medoid", "min"]
CustomThresholdMethod = Literal["median", "soft_extremum"]
SkimageThresholdMethod = Literal[
"mean", "isodata", "li", "minimum", "otsu", "triangle", "yen"
*_CUSTOM_THRESHOLD_METHODS,
)
- _OPTUNA_STORAGE_BACKENDS: Final[tuple[str, ...]] = ("file", "sqlite")
- _OPTUNA_SAMPLERS: Final[tuple[str, ...]] = ("tpe", "auto")
- _OPTUNA_NAMESPACES: Final[tuple[OptunaNamespace, ...]] = ("hp", "train", "label")
+ _CLUSTER_SELECTION_METHODS: Final[tuple[ClusterSelectionMethod, ...]] = (
+ "medoid",
+ "min",
+ )
_OPTUNA_LABEL_N_OBJECTIVES: Final[int] = 7
_OPTUNA_LABEL_DIRECTIONS: Final[tuple[optuna.study.StudyDirection, ...]] = (
optuna.study.StudyDirection.MAXIMIZE,
) * _OPTUNA_LABEL_N_OBJECTIVES
+ _OPTUNA_STORAGE_BACKENDS: Final[tuple[str, ...]] = ("file", "sqlite")
+ _OPTUNA_SAMPLERS: Final[tuple[str, ...]] = ("tpe", "auto")
+ _OPTUNA_NAMESPACES: Final[tuple[OptunaNamespace, ...]] = ("hp", "train", "label")
+
_SCIPY_METRICS: Final[tuple[str, ...]] = (
# "braycurtis",
# "canberra",
*_CUSTOM_METRICS,
)
+ _UNSUPPORTED_CLUSTER_METRICS: Final[tuple[str, ...]] = (
+ "mahalanobis",
+ "seuclidean",
+ "jensenshannon",
+ )
+
PREDICTIONS_EXTREMA_THRESHOLD_OUTLIER_DEFAULT: Final[float] = 0.999
PREDICTIONS_EXTREMA_THRESHOLDS_ALPHA_DEFAULT: Final[float] = 12.0
PREDICTIONS_EXTREMA_EXTREMA_FRACTION_DEFAULT: Final[float] = 1.0
def _metrics_set() -> set[str]:
return set(QuickAdapterRegressorV3._METRICS)
+ @staticmethod
+ def _unsupported_cluster_metrics_set() -> set[str]:
+ return set(QuickAdapterRegressorV3._UNSUPPORTED_CLUSTER_METRICS)
+
+ @staticmethod
+ def _cluster_selection_methods_set() -> set[ClusterSelectionMethod]:
+ return set(QuickAdapterRegressorV3._CLUSTER_SELECTION_METHODS)
+
@staticmethod
def _get_label_p_order_default(metric: str) -> Optional[float]:
if metric == QuickAdapterRegressorV3._SCIPY_METRICS[5]: # "minkowski"
QuickAdapterRegressorV3._CUSTOM_METRICS[10], # "kmeans2"
}:
logger.info(
- f" label_kmeans_selection: min (default for {label_metric})"
+ f" label_kmeans_selection: {QuickAdapterRegressorV3._CLUSTER_SELECTION_METHODS[1]} (default for {label_metric})"
)
label_kmedoids_metric_config = self.ft_params.get("label_kmedoids_metric")
if label_kmedoids_metric_config is not None:
label_metric == QuickAdapterRegressorV3._CUSTOM_METRICS[11]
): # "kmedoids"
logger.info(
- f" label_kmedoids_selection: min (default for {label_metric})"
+ f" label_kmedoids_selection: {QuickAdapterRegressorV3._CLUSTER_SELECTION_METHODS[1]} (default for {label_metric})"
)
label_knn_metric_config = self.ft_params.get("label_knn_metric")
params = self._optuna_label_params.get(pair)
else:
raise ValueError(
- f"Invalid namespace: {namespace}. "
- f"Expected {', '.join(QuickAdapterRegressorV3._OPTUNA_NAMESPACES)}"
+ f"Invalid namespace '{namespace}'. "
+ f"Supported: {', '.join(QuickAdapterRegressorV3._OPTUNA_NAMESPACES)}"
)
return params
self._optuna_label_params[pair] = params
else:
raise ValueError(
- f"Invalid namespace: {namespace}. "
- f"Expected {', '.join(QuickAdapterRegressorV3._OPTUNA_NAMESPACES)}"
+ f"Invalid namespace '{namespace}'. "
+ f"Supported: {', '.join(QuickAdapterRegressorV3._OPTUNA_NAMESPACES)}"
)
def get_optuna_value(self, pair: str, namespace: str) -> float:
value = self._optuna_train_value.get(pair)
else:
raise ValueError(
- f"Invalid namespace: {namespace}. "
- f"Expected {', '.join(QuickAdapterRegressorV3._OPTUNA_NAMESPACES[:2])}" # Only hp and train
+ f"Invalid namespace '{namespace}'. "
+ f"Supported: {', '.join(QuickAdapterRegressorV3._OPTUNA_NAMESPACES[:2])}" # Only hp and train
)
return value
self._optuna_train_value[pair] = value
else:
raise ValueError(
- f"Invalid namespace: {namespace}. "
- f"Expected {', '.join(QuickAdapterRegressorV3._OPTUNA_NAMESPACES[:2])}" # Only hp and train
+ f"Invalid namespace '{namespace}'. "
+ f"Supported: {', '.join(QuickAdapterRegressorV3._OPTUNA_NAMESPACES[:2])}" # Only hp and train
)
def get_optuna_values(self, pair: str, namespace: str) -> list[float | int]:
values = self._optuna_label_values.get(pair)
else:
raise ValueError(
- f"Invalid namespace: {namespace}. "
- f"Expected {QuickAdapterRegressorV3._OPTUNA_NAMESPACES[2]}" # Only label
+ f"Invalid namespace '{namespace}'. "
+ f"Supported: {QuickAdapterRegressorV3._OPTUNA_NAMESPACES[2]}" # Only label
)
return values
self._optuna_label_values[pair] = values
else:
raise ValueError(
- f"Invalid namespace: {namespace}. "
- f"Expected {QuickAdapterRegressorV3._OPTUNA_NAMESPACES[2]}" # Only label
+ f"Invalid namespace '{namespace}'. "
+ f"Supported: {QuickAdapterRegressorV3._OPTUNA_NAMESPACES[2]}" # Only label
)
def init_optuna_label_candle_pool(self) -> None:
QuickAdapterRegressorV3._OPTUNA_NAMESPACES[2]
}: # Only "label"
raise ValueError(
- f"Invalid namespace: {namespace}. "
- f"Expected {QuickAdapterRegressorV3._OPTUNA_NAMESPACES[2]}" # Only label
+ f"Invalid namespace '{namespace}'. "
+ f"Supported: {QuickAdapterRegressorV3._OPTUNA_NAMESPACES[2]}" # Only label
)
if not callable(callback):
- raise ValueError("callback must be callable")
+ raise ValueError("Invalid callback: must be callable")
self._optuna_label_candles[pair] += 1
if pair not in self._optuna_label_incremented_pairs:
self._optuna_label_incremented_pairs.append(pair)
pred_minima = pred_extrema[pred_extrema < -eps]
else:
raise ValueError(
- f"Unsupported extrema selection method: {extrema_selection}. "
- f"Supported methods are {', '.join(QuickAdapterRegressorV3._EXTREMA_SELECTION_METHODS)}"
+ f"Invalid extrema_selection '{extrema_selection}'. "
+ f"Supported: {', '.join(QuickAdapterRegressorV3._EXTREMA_SELECTION_METHODS)}"
)
return pred_minima, pred_maxima
extrema_fraction: float = 1.0,
) -> tuple[float, float]:
if alpha < 0:
- raise ValueError("alpha must be non-negative")
+ raise ValueError(f"Invalid alpha {alpha}: must be >= 0")
pred_minima, pred_maxima = QuickAdapterRegressorV3.get_pred_min_max(
pred_extrema, extrema_selection, extrema_fraction
)
try:
threshold_func = getattr(skimage.filters, f"threshold_{method}")
except AttributeError:
- raise ValueError(f"Unknown skimage threshold function: threshold_{method}")
+ raise ValueError(
+ f"Invalid skimage threshold method '{method}'. "
+ f"Supported: {', '.join(QuickAdapterRegressorV3._SKIMAGE_THRESHOLD_METHODS)}"
+ )
min_func = QuickAdapterRegressorV3.apply_skimage_threshold
max_func = QuickAdapterRegressorV3.apply_skimage_threshold
Must contain only finite values (no NaN or inf).
- metric: distance metric name accepted by scipy.spatial.distance.pdist.
- weights: optional weight vector per feature (passed as 'w' to pdist).
- Not supported by mahalanobis, seuclidean, jensenshannon.
+ Not supported by metrics in _UNSUPPORTED_CLUSTER_METRICS.
Must have size equal to n_features and contain finite non-negative values.
- p: optional Minkowski order (default 2.0 if metric=='minkowski').
array([2. , 2.41421356, 2.41421356])
"""
if matrix.ndim != 2:
- raise ValueError("matrix must be 2-dimensional")
+ raise ValueError("Invalid matrix: must be 2-dimensional")
if matrix.shape[1] == 0:
- raise ValueError("matrix must have at least one feature")
+ raise ValueError("Invalid matrix: must have at least one feature")
if not np.all(np.isfinite(matrix)):
- raise ValueError("matrix must contain only finite values (no NaN or inf)")
+ raise ValueError(
+ "Invalid matrix: must contain only finite values (no NaN or inf)"
+ )
if weights is not None:
if weights.size != matrix.shape[1]:
raise ValueError(
- f"weights size {weights.size} must match number of features {matrix.shape[1]}"
+ f"Invalid weights: size {weights.size} must match number of features {matrix.shape[1]}"
)
if not np.all(np.isfinite(weights)) or np.any(weights < 0):
- raise ValueError("weights must be finite and non-negative")
- if metric in {
- QuickAdapterRegressorV3._SCIPY_METRICS[4], # "mahalanobis"
- QuickAdapterRegressorV3._SCIPY_METRICS[6], # "seuclidean"
- QuickAdapterRegressorV3._SCIPY_METRICS[3], # "jensenshannon"
- }:
- raise ValueError(f"weights not supported for metric '{metric}'")
+ raise ValueError("Invalid weights: must be finite and non-negative")
+ if metric in QuickAdapterRegressorV3._unsupported_cluster_metrics_set():
+ raise ValueError(
+ f"Invalid weights: not supported for metric '{metric}'"
+ )
matrix = np.asarray(matrix, dtype=np.float64)
if weights is not None:
directions: list[optuna.study.StudyDirection],
) -> NDArray[np.floating]:
if objective_values_matrix.ndim != 2:
- raise ValueError("objective_values_matrix must be 2-dimensional")
+ raise ValueError("Invalid objective_values_matrix: must be 2-dimensional")
n_samples, n_objectives = objective_values_matrix.shape
if n_samples == 0 or n_objectives == 0:
raise ValueError(
- "objective_values_matrix must have at least one sample and one objective"
+ "Invalid objective_values_matrix: must have at least one sample and one objective"
)
if len(directions) != n_objectives:
raise ValueError(
- f"Number of directions ({len(directions)}) must match number of objectives ({n_objectives})"
+ f"Invalid directions: length ({len(directions)}) must match number of objectives ({n_objectives})"
)
normalized_matrix = np.zeros_like(objective_values_matrix, dtype=float)
metrics: set[str],
) -> NDArray[np.floating]:
if normalized_matrix.ndim != 2:
- raise ValueError("normalized_matrix must be 2-dimensional")
+ raise ValueError("Invalid normalized_matrix: must be 2-dimensional")
n_objectives = normalized_matrix.shape[1]
n_samples = normalized_matrix.shape[0]
if n_samples == 0 or n_objectives == 0:
raise ValueError(
- "normalized_matrix must have at least one sample and one objective"
+ "Invalid normalized_matrix: must have at least one sample and one objective"
)
if not np.all(np.isfinite(normalized_matrix)):
raise ValueError(
- "normalized_matrix must contain only finite values (no NaN or inf)"
+ "Invalid normalized_matrix: must contain only finite values (no NaN or inf)"
)
label_p_order = self.ft_params.get("label_p_order")
label_weights = self.ft_params.get("label_weights")
np_weights = np.array(label_weights, dtype=float)
else:
raise ValueError(
- f"label_weights must be a list, tuple, or array, got {type(label_weights).__name__}"
+ f"Invalid label_weights: must be a list, tuple, or array, got {type(label_weights).__name__}"
)
if np_weights.size != n_objectives:
- raise ValueError("label_weights length must match number of objectives")
+ raise ValueError(
+ "Invalid label_weights: length must match number of objectives"
+ )
if not np.all(np.isfinite(np_weights)):
- raise ValueError("label_weights must contain only finite values")
+ raise ValueError("Invalid label_weights: must contain only finite values")
if np.any(np_weights < 0):
- raise ValueError("label_weights values must be non-negative")
+ raise ValueError("Invalid label_weights: values must be non-negative")
label_weights_sum = np.nansum(np.abs(np_weights))
if np.isclose(label_weights_sum, 0.0):
- raise ValueError("label_weights sum cannot be zero")
+ raise ValueError("Invalid label_weights: sum cannot be zero")
np_weights = np_weights / label_weights_sum
ideal_point = np.ones(n_objectives)
if metric in QuickAdapterRegressorV3._scipy_metrics_set():
cdist_kwargs: dict[str, Any] = {}
- if metric not in {
- QuickAdapterRegressorV3._SCIPY_METRICS[4], # "mahalanobis"
- QuickAdapterRegressorV3._SCIPY_METRICS[6], # "seuclidean"
- QuickAdapterRegressorV3._SCIPY_METRICS[3], # "jensenshannon"
- }:
+ if metric not in QuickAdapterRegressorV3._unsupported_cluster_metrics_set():
cdist_kwargs["w"] = np_weights
if metric == QuickAdapterRegressorV3._SCIPY_METRICS[5]: # "minkowski"
cdist_kwargs["p"] = (
variances = np.nanvar(np_sqrt_normalized_matrix, axis=0, ddof=1)
if np.any(variances <= 0):
raise ValueError(
- "shellinger metric requires non-zero variance for all objectives"
+ "Invalid data for shellinger metric: requires non-zero variance for all objectives"
)
np_weights = 1 / variances
return (
"label_medoid_metric",
QuickAdapterRegressorV3._SCIPY_METRICS[2], # "euclidean"
)
- if label_medoid_metric in {
- QuickAdapterRegressorV3._SCIPY_METRICS[4], # "mahalanobis"
- QuickAdapterRegressorV3._SCIPY_METRICS[6], # "seuclidean"
- QuickAdapterRegressorV3._SCIPY_METRICS[3], # "jensenshannon"
- }:
+ if (
+ label_medoid_metric
+ in QuickAdapterRegressorV3._unsupported_cluster_metrics_set()
+ ):
raise ValueError(
- f"Unsupported label_medoid_metric: {label_medoid_metric}. Supported are euclidean/minkowski/cityblock/chebyshev/..."
+ f"Invalid label_medoid_metric '{label_medoid_metric}'. "
+ f"Unsupported: {', '.join(QuickAdapterRegressorV3._UNSUPPORTED_CLUSTER_METRICS)}"
)
p = None
if (
"label_kmeans_metric",
QuickAdapterRegressorV3._SCIPY_METRICS[2], # "euclidean"
)
- if label_kmeans_metric in {
- QuickAdapterRegressorV3._SCIPY_METRICS[4], # "mahalanobis"
- QuickAdapterRegressorV3._SCIPY_METRICS[6], # "seuclidean"
- QuickAdapterRegressorV3._SCIPY_METRICS[3], # "jensenshannon"
- }:
+ if (
+ label_kmeans_metric
+ in QuickAdapterRegressorV3._unsupported_cluster_metrics_set()
+ ):
raise ValueError(
- f"Unsupported label_kmeans_metric: {label_kmeans_metric}. Supported are euclidean/minkowski/cityblock/chebyshev/..."
+ f"Invalid label_kmeans_metric '{label_kmeans_metric}'. "
+ f"Unsupported: {', '.join(QuickAdapterRegressorV3._UNSUPPORTED_CLUSTER_METRICS)}"
)
cdist_kwargs: dict[str, Any] = {}
if (
metric=label_kmeans_metric,
**cdist_kwargs,
).flatten()
- label_kmeans_selection = self.ft_params.get("label_kmeans_selection", "min")
+ label_kmeans_selection = self.ft_params.get(
+ "label_kmeans_selection",
+ QuickAdapterRegressorV3._CLUSTER_SELECTION_METHODS[1], # "min"
+ )
ordered_cluster_indices = np.argsort(cluster_center_distances_to_ideal)
best_cluster_indices = None
for cluster_index in ordered_cluster_indices:
if best_cluster_indices is not None and best_cluster_indices.size > 0:
if (
label_kmeans_selection
- == QuickAdapterRegressorV3._CUSTOM_METRICS[16] # "medoid"
+ == QuickAdapterRegressorV3._CLUSTER_SELECTION_METHODS[0] # "medoid"
):
p = None
if (
**cdist_kwargs,
).item()
trial_distances[best_trial_index] = best_trial_distance
- elif label_kmeans_selection == "min":
+ elif (
+ label_kmeans_selection
+ == QuickAdapterRegressorV3._CLUSTER_SELECTION_METHODS[1] # "min"
+ ):
best_cluster_distances = sp.spatial.distance.cdist(
normalized_matrix[best_cluster_indices],
ideal_point_2d,
]
else:
raise ValueError(
- f"Unsupported label_kmeans_selection: {label_kmeans_selection}. Supported are medoid/min"
+ f"Invalid label_kmeans_selection '{label_kmeans_selection}'. "
+ f"Supported: {', '.join(QuickAdapterRegressorV3._CLUSTER_SELECTION_METHODS)}"
)
return trial_distances
elif metric == QuickAdapterRegressorV3._CUSTOM_METRICS[11]: # "kmedoids"
"label_kmedoids_metric",
QuickAdapterRegressorV3._SCIPY_METRICS[2], # "euclidean"
)
- if label_kmedoids_metric in {
- QuickAdapterRegressorV3._SCIPY_METRICS[4], # "mahalanobis"
- QuickAdapterRegressorV3._SCIPY_METRICS[6], # "seuclidean"
- QuickAdapterRegressorV3._SCIPY_METRICS[3], # "jensenshannon"
- }:
+ if (
+ label_kmedoids_metric
+ in QuickAdapterRegressorV3._unsupported_cluster_metrics_set()
+ ):
raise ValueError(
- f"Unsupported label_kmedoids_metric: {label_kmedoids_metric}. Supported are euclidean/minkowski/cityblock/chebyshev/..."
+ f"Invalid label_kmedoids_metric '{label_kmedoids_metric}'. "
+ f"Unsupported: {', '.join(QuickAdapterRegressorV3._UNSUPPORTED_CLUSTER_METRICS)}"
)
kmedoids_kwargs: dict[str, Any] = {
"metric": label_kmedoids_metric,
**cdist_kwargs,
).flatten()
label_kmedoids_selection = self.ft_params.get(
- "label_kmedoids_selection", "min"
+ "label_kmedoids_selection",
+ QuickAdapterRegressorV3._CLUSTER_SELECTION_METHODS[1], # "min"
)
best_medoid_distance_position = np.nanargmin(medoid_distances_to_ideal)
best_medoid_index = medoid_indices[best_medoid_distance_position]
if best_cluster_indices.size > 0:
if (
label_kmedoids_selection
- == QuickAdapterRegressorV3._CUSTOM_METRICS[16] # "medoid"
+ == QuickAdapterRegressorV3._CLUSTER_SELECTION_METHODS[0] # "medoid"
):
trial_distances[best_medoid_index] = medoid_distances_to_ideal[
best_medoid_distance_position
]
- elif label_kmedoids_selection == "min":
+ elif (
+ label_kmedoids_selection
+ == QuickAdapterRegressorV3._CLUSTER_SELECTION_METHODS[1] # "min"
+ ):
if best_cluster_indices.size == 1:
best_trial_index = best_cluster_indices[0]
trial_distances[best_trial_index] = medoid_distances_to_ideal[
]
else:
raise ValueError(
- f"Unsupported label_kmedoids_selection: {label_kmedoids_selection}. Supported are medoid/min"
+ f"Invalid label_kmedoids_selection '{label_kmedoids_selection}'. "
+ f"Supported: {', '.join(QuickAdapterRegressorV3._CLUSTER_SELECTION_METHODS)}"
)
return trial_distances
elif metric in {
return np.nanmax(neighbor_distances, axis=1)
else:
raise ValueError(
- f"Unsupported label metric: {metric}. Supported metrics are {', '.join(metrics)}"
+ f"Invalid label metric '{metric}'. Supported: {', '.join(metrics)}"
)
def _get_multi_objective_study_best_trial(
QuickAdapterRegressorV3._OPTUNA_NAMESPACES[2]
}: # Only "label"
raise ValueError(
- f"Invalid namespace: {namespace}. "
- f"Expected {QuickAdapterRegressorV3._OPTUNA_NAMESPACES[2]}" # Only label
+ f"Invalid namespace '{namespace}'. "
+ f"Supported: {QuickAdapterRegressorV3._OPTUNA_NAMESPACES[2]}" # Only label
)
n_objectives = len(study.directions)
if n_objectives < 2:
) # "euclidean"
if label_metric not in metrics:
raise ValueError(
- f"Unsupported label metric: {label_metric}. Supported metrics are {', '.join(metrics)}"
+ f"Invalid label_metric '{label_metric}'. Supported: {', '.join(metrics)}"
)
best_trials = [
)
else:
raise ValueError(
- f"Unsupported optuna storage backend: {storage_backend}. "
- f"Supported backends are {', '.join(QuickAdapterRegressorV3._OPTUNA_STORAGE_BACKENDS)}"
+ f"Invalid optuna storage_backend '{storage_backend}'. "
+ f"Supported: {', '.join(QuickAdapterRegressorV3._OPTUNA_STORAGE_BACKENDS)}"
)
return storage
)
else:
raise ValueError(
- f"Unsupported sampler: {sampler}. "
- f"Supported samplers are {', '.join(QuickAdapterRegressorV3._OPTUNA_SAMPLERS)}"
+ f"Invalid optuna sampler '{sampler}'. "
+ f"Supported: {', '.join(QuickAdapterRegressorV3._OPTUNA_SAMPLERS)}"
)
def optuna_create_study(
def get_label_natr_ratio_percent(self, pair: str, percent: float) -> float:
if not isinstance(percent, float) or not (0.0 <= percent <= 1.0):
raise ValueError(
- f"Invalid percent value: {percent}. It should be a float between 0 and 1"
+ f"Invalid percent {percent}: must be a float between 0 and 1"
)
return self.get_label_natr_ratio(pair) * percent
trade_price_target_fn = trade_price_target_methods.get(trade_price_target)
if trade_price_target_fn is None:
raise ValueError(
- f"Invalid trade_price_target: {trade_price_target}. Available: {', '.join(sorted(TRADE_PRICE_TARGETS))}"
+ f"Invalid trade_price_target '{trade_price_target}'. "
+ f"Supported: {', '.join(sorted(TRADE_PRICE_TARGETS))}"
)
return trade_price_target_fn()
) -> Optional[float]:
if not (0.0 <= natr_ratio_percent <= 1.0):
raise ValueError(
- f"natr_ratio_percent must be in [0, 1], got {natr_ratio_percent}"
+ f"Invalid natr_ratio_percent {natr_ratio_percent}: must be in [0, 1]"
)
trade_duration_candles = self.get_trade_duration_candles(df, trade)
if not QuickAdapterV3.is_trade_duration_valid(trade_duration_candles):
) -> Optional[float]:
if not (0.0 <= natr_ratio_percent <= 1.0):
raise ValueError(
- f"natr_ratio_percent must be in [0, 1], got {natr_ratio_percent}"
+ f"Invalid natr_ratio_percent {natr_ratio_percent}: must be in [0, 1]"
)
trade_duration_candles = self.get_trade_duration_candles(df, trade)
if not QuickAdapterV3.is_trade_duration_valid(trade_duration_candles):
callback: Callable[[], None],
) -> None:
if not callable(callback):
- raise ValueError("callback must be callable")
+ raise ValueError("Invalid callback: must be callable")
timestamp = int(current_time.timestamp())
candle_duration_secs = max(1, int(self._candle_duration_secs))
candle_start_secs = (timestamp // candle_duration_secs) * candle_duration_secs
)
else:
raise ValueError(
- f"Invalid interpolation_direction: {interpolation_direction}. "
- f"Expected {', '.join(QuickAdapterV3._INTERPOLATION_DIRECTIONS)}"
+ f"Invalid interpolation_direction '{interpolation_direction}'. "
+ f"Supported: {', '.join(QuickAdapterV3._INTERPOLATION_DIRECTIONS)}"
)
candle_deviation = (
candle_label_natr_value / 100.0
candle_threshold = base_price * (1 - current_deviation)
else:
raise ValueError(
- f"Invalid side: {side}. Expected {', '.join(QuickAdapterV3._TRADE_DIRECTIONS)}"
+ f"Invalid side '{side}'. Supported: {', '.join(QuickAdapterV3._TRADE_DIRECTIONS)}"
)
self._candle_threshold_cache[cache_key] = candle_threshold
return self._candle_threshold_cache[cache_key]
return False
else:
raise ValueError(
- f"Invalid trading_mode: {trading_mode}. "
- f"Expected {', '.join(QuickAdapterV3._TRADING_MODES)}"
+ f"Invalid trading_mode '{trading_mode}'. "
+ f"Supported: {', '.join(QuickAdapterV3._TRADING_MODES)}"
)
def leverage(
@lru_cache(maxsize=8)
def get_odd_window(window: int) -> int:
if window < 1:
- raise ValueError("Window size must be greater than 0")
+ raise ValueError(f"Invalid window {window}: must be > 0")
return window if window % 2 == 1 else window + 1
elif win_type == SMOOTHING_METHODS[2]: # "triang"
coeffs = sp.signal.windows.triang(M=window, sym=True)
else:
- raise ValueError(f"Unknown window type: {win_type}")
+ raise ValueError(
+ f"Invalid window type '{win_type}'. "
+ f"Supported: {', '.join(SMOOTHING_METHODS[:3])}"
+ )
return coeffs / np.sum(coeffs)
return _standardize_mmad(weights, scaling_factor=mmad_scaling_factor)
else:
- raise ValueError(f"Unknown standardization method: {method}")
+ raise ValueError(
+ f"Invalid standardization method '{method}'. "
+ f"Supported: {', '.join(STANDARDIZATION_TYPES)}"
+ )
def _normalize_sigmoid(
elif normalization == NORMALIZATION_TYPES[5]: # "rank"
normalized_weights = _normalize_rank(standardized_weights, method=rank_method)
else:
- raise ValueError(f"Unknown normalization method: {normalization}")
+ raise ValueError(
+ f"Invalid normalization method '{normalization}'. "
+ f"Supported: {', '.join(NORMALIZATION_TYPES)}"
+ )
# Phase 3: Post-processing
if not np.isclose(gamma, 1.0) and np.isfinite(gamma) and gamma > 0:
weights=source_weights_array[:, np.newaxis],
)
else:
- raise ValueError(f"Unknown hybrid aggregation method: {aggregation}")
+ raise ValueError(
+ f"Invalid hybrid aggregation method '{aggregation}'. "
+ f"Supported: {', '.join(HYBRID_AGGREGATIONS)}"
+ )
if aggregation_normalization != NORMALIZATION_TYPES[6]: # "none"
combined_source_weights_array = normalize_weights(
default_weight=np.nanmedian(normalized_weights),
)
- raise ValueError(f"Unknown extrema weighting strategy: {strategy}")
+ raise ValueError(
+ f"Invalid extrema weighting strategy '{strategy}'. "
+ f"Supported: {', '.join(WEIGHT_STRATEGIES)}"
+ )
def _apply_weights(
def get_callable_sha256(fn: Callable[..., Any]) -> str:
if not callable(fn):
- raise ValueError("fn must be callable")
+ raise ValueError("Invalid fn: must be callable")
code = getattr(fn, "__code__", None)
if code is None and isinstance(fn, functools.partial):
fn = fn.func
if code is None and hasattr(fn, "__call__"):
code = getattr(fn.__call__, "__code__", None)
if code is None:
- raise ValueError("Unable to retrieve code object from fn")
+ raise ValueError("Invalid fn: unable to retrieve code object")
return hashlib.sha256(code.co_code).hexdigest()
:return: The top change percentage series
"""
if period < 1:
- raise ValueError("period must be greater than or equal to 1")
+ raise ValueError(f"Invalid period {period}: must be >= 1")
previous_close_top = (
dataframe.get("close").rolling(period, min_periods=period).max().shift(1)
:return: The bottom change percentage series
"""
if period < 1:
- raise ValueError("period must be greater than or equal to 1")
+ raise ValueError(f"Invalid period {period}: must be >= 1")
previous_close_bottom = (
dataframe.get("close").rolling(period, min_periods=period).min().shift(1)
:return: Retracement percentage series
"""
if period < 1:
- raise ValueError("period must be greater than or equal to 1")
+ raise ValueError(f"Invalid period {period}: must be >= 1")
previous_close_low = (
dataframe.get("close").rolling(period, min_periods=period).min().shift(1)
) -> float:
"""Original fractal dimension computation implementation per Ehlers' paper."""
if period % 2 != 0:
- raise ValueError("period must be even")
+ raise ValueError(f"Invalid period {period}: must be even")
half_period = period // 2
Original FRAMA implementation per Ehlers' paper with optional zero lag.
"""
if period % 2 != 0:
- raise ValueError("period must be even")
+ raise ValueError(f"Invalid period {period}: must be even")
n = len(df)
https://www.sierrachart.com/index.php?page=doc/StudiesReference.php&ID=173&Name=Moving_Average_-_Smoothed
"""
if period <= 0:
- raise ValueError("period must be greater than 0")
+ raise ValueError(f"Invalid period {period}: must be > 0")
n = len(series)
if n < period:
return pd.Series(index=series.index, dtype=float)
]
else:
raise ValueError(
- f"Unsupported regressor model: {regressor} (supported: {', '.join(REGRESSORS)})"
+ f"Invalid regressor '{regressor}'. Supported: {', '.join(REGRESSORS)}"
)
return callbacks
)
else:
raise ValueError(
- f"Unsupported regressor model: {regressor} (supported: {', '.join(REGRESSORS)})"
+ f"Invalid regressor '{regressor}'. Supported: {', '.join(REGRESSORS)}"
)
return model
) -> dict[str, Any]:
if regressor not in set(REGRESSORS):
raise ValueError(
- f"Unsupported regressor model: {regressor} (supported: {', '.join(REGRESSORS)})"
+ f"Invalid regressor '{regressor}'. Supported: {', '.join(REGRESSORS)}"
)
if not isinstance(expansion_ratio, (int, float)) or not (
0.0 <= expansion_ratio <= 1.0
):
raise ValueError(
- f"expansion_ratio must be a float between 0 and 1, got {expansion_ratio}"
+ f"Invalid expansion_ratio {expansion_ratio}: must be a float between 0 and 1"
)
default_ranges: dict[str, tuple[float, float]] = {
"n_estimators": (100, 2000),
@lru_cache(maxsize=128)
def largest_divisor_to_step(integer: int, step: int) -> Optional[int]:
if not isinstance(integer, int) or integer <= 0:
- raise ValueError("integer must be a positive integer")
+ raise ValueError(f"Invalid integer {integer!r}: must be a positive integer")
if not isinstance(step, int) or step <= 0:
- raise ValueError("step must be a positive integer")
+ raise ValueError(f"Invalid step {step!r}: must be a positive integer")
if step == 1 or integer % step == 0:
return integer
) -> tuple[int, int, int]:
if min_label_period_candles > max_label_period_candles:
raise ValueError(
- "min_label_period_candles must be less than or equal to max_label_period_candles"
+ f"Invalid label_period_candles range: min ({min_label_period_candles}) "
+ f"must be <= max ({max_label_period_candles})"
)
capped_period_candles = max(1, floor_to_step(max_period_candles, candles_step))
:raises ValueError: If step is not a positive integer or value is not finite.
"""
if not isinstance(value, (int, float)):
- raise ValueError("value must be an integer or float")
+ raise ValueError(f"Invalid value {value!r}: must be an integer or float")
if not isinstance(step, int) or step <= 0:
- raise ValueError("step must be a positive integer")
+ raise ValueError(f"Invalid step {step!r}: must be a positive integer")
if isinstance(value, (int, np.integer)):
q, r = divmod(value, step)
twice_r = r * 2
return (q + 1) * step
return int(round(value / step) * step)
if not np.isfinite(value):
- raise ValueError("value must be finite")
+ raise ValueError(f"Invalid value {value!r}: must be finite")
return int(round(float(value) / step) * step)
@lru_cache(maxsize=128)
def ceil_to_step(value: float | int, step: int) -> int:
if not isinstance(value, (int, float)):
- raise ValueError("value must be an integer or float")
+ raise ValueError(f"Invalid value {value!r}: must be an integer or float")
if not isinstance(step, int) or step <= 0:
- raise ValueError("step must be a positive integer")
+ raise ValueError(f"Invalid step {step!r}: must be a positive integer")
if isinstance(value, (int, np.integer)):
return int(-(-int(value) // step) * step)
if not np.isfinite(value):
- raise ValueError("value must be finite")
+ raise ValueError(f"Invalid value {value!r}: must be finite")
return int(math.ceil(float(value) / step) * step)
@lru_cache(maxsize=128)
def floor_to_step(value: float | int, step: int) -> int:
if not isinstance(value, (int, float)):
- raise ValueError("value must be an integer or float")
+ raise ValueError(f"Invalid value {value!r}: must be an integer or float")
if not isinstance(step, int) or step <= 0:
- raise ValueError("step must be a positive integer")
+ raise ValueError(f"Invalid step {step!r}: must be a positive integer")
if isinstance(value, (int, np.integer)):
return int((int(value) // step) * step)
if not np.isfinite(value):
- raise ValueError("value must be finite")
+ raise ValueError(f"Invalid value {value!r}: must be finite")
return int(math.floor(float(value) / step) * step)
repositories / freqai-strategies.git / commitdiff
