from technical.pivots_points import pivots_points
from Utils import (
DEFAULTS_EXIT_THRESHOLDS_CALIBRATION,
+ _CACHE_MAXSIZE_LARGE,
_OPTUNA_NAMESPACES,
EXTREMA_COLUMN,
EXTREMA_DIRECTION_COLUMN,
@staticmethod
def _df_signature(df: DataFrame) -> DfSignature:
+ """Candle-cache key ``(row_count, last_date)``; assumes existing rows
+ stay immutable (holds under ``process_only_new_candles = True``).
+ """
n = len(df)
if n == 0:
return (0, None)
return {}
@staticmethod
- @lru_cache(maxsize=128)
+ @lru_cache(maxsize=_CACHE_MAXSIZE_LARGE)
def _td_format(
delta: datetime.timedelta, pattern: str = "{sign}{d}:{h:02d}:{m:02d}:{s:02d}"
) -> str:
return trade.open_date_utc - offset_timedelta
@staticmethod
- @lru_cache(maxsize=128)
+ @lru_cache(maxsize=_CACHE_MAXSIZE_LARGE)
def is_trade_duration_valid(trade_duration: Optional[int | float]) -> bool:
return isinstance(trade_duration, (int, float)) and not (
isna(trade_duration) or trade_duration <= 0
)
@staticmethod
- @lru_cache(maxsize=128)
+ @lru_cache(maxsize=_CACHE_MAXSIZE_LARGE)
def get_stoploss_factor(trade_duration_candles: int) -> float:
return 2.75 / (1.2675 + math.atan(0.25 * trade_duration_candles))
)
@staticmethod
- @lru_cache(maxsize=128)
+ @lru_cache(maxsize=_CACHE_MAXSIZE_LARGE)
def get_take_profit_factor(trade_duration_candles: int) -> float:
return math.log10(9.75 + 0.25 * trade_duration_candles)
)
@staticmethod
- @lru_cache(maxsize=128)
def _t_statistic(mean: float, std: float, n: int) -> float:
"""Compute t-statistic for H0: mu = 0 as ``mean * sqrt(n) / std``.
return mean * math.sqrt(n) / std
@staticmethod
- @lru_cache(maxsize=128)
+ @lru_cache(maxsize=_CACHE_MAXSIZE_LARGE)
def is_isoformat(string: str) -> bool:
if not isinstance(string, str):
return False
return True
@staticmethod
- @lru_cache(maxsize=128)
def _effective_df(x: tuple[float, ...]) -> float:
"""Effective degrees of freedom with Bartlett's autocorrelation correction.
return df_eff
@staticmethod
- @lru_cache(maxsize=128)
def _t_critical(q: float, df: float, default_t: float) -> float:
"""Critical t-value from Student's t-distribution at quantile ``q``.
T = TypeVar("T", pd.Series, float)
+# lru_cache sizes: SMALL for bounded key spaces (windows, mode strings),
+# LARGE for open numeric/string keys (formatting, rounding, statistics).
+_CACHE_MAXSIZE_SMALL: Final[int] = 8
+_CACHE_MAXSIZE_LARGE: Final[int] = 128
+
@dataclass(frozen=True, slots=True)
class FiniteSample:
_FREQAI_LABEL_SIGIL_PATTERN: Final[re.Pattern[str]] = re.compile(r"^&-?")
-@lru_cache(maxsize=64)
+@lru_cache(maxsize=_CACHE_MAXSIZE_SMALL)
def _label_aux_column_name(label_col: str, suffix: str) -> str:
"""Derive a freqtrade-safe auxiliary column name from a label column.
return diff.where(diff != 0, np.finfo(float).eps)
-@lru_cache(maxsize=8)
+@lru_cache(maxsize=_CACHE_MAXSIZE_SMALL)
def get_odd_window(window: int) -> int:
if window < 1:
raise ValueError(f"Invalid window value {window!r}: must be > 0")
return window if window % 2 == 1 else window + 1
-@lru_cache(maxsize=8)
+@lru_cache(maxsize=_CACHE_MAXSIZE_SMALL)
def get_even_window(window: int) -> int:
if window < 1:
raise ValueError(f"Invalid window value {window!r}: must be > 0")
return window if window % 2 == 0 else window + 1
-@lru_cache(maxsize=8)
+@lru_cache(maxsize=_CACHE_MAXSIZE_SMALL)
def get_gaussian_std(window: int) -> float:
return (window - 1) / 6.0 if window > 1 else 0.5
-@lru_cache(maxsize=8)
+@lru_cache(maxsize=_CACHE_MAXSIZE_SMALL)
def get_savgol_params(
window: int, polyorder: int, mode: SmoothingMode
) -> tuple[int, int, str]:
return window, polyorder, mode
-@lru_cache(maxsize=8)
+@lru_cache(maxsize=_CACHE_MAXSIZE_SMALL)
def _calculate_coeffs(
window: int,
win_type: SmoothingKernel,
f"Invalid window type value {win_type!r}: "
f"supported values are {', '.join(SMOOTHING_KERNELS)}"
)
- return coeffs / np.sum(coeffs)
+ normalized_coeffs = coeffs / np.sum(coeffs)
+ normalized_coeffs.setflags(write=False)
+ return normalized_coeffs
def zero_phase_filter(
_SCIENTIFIC_THRESHOLD_LOW = 1e-6
-@lru_cache(maxsize=128)
+@lru_cache(maxsize=_CACHE_MAXSIZE_LARGE)
def format_number(value: int | float, significant_digits: int = 5) -> str:
if not isinstance(value, (int, float, np.integer, np.floating)):
return str(value)
return f"{{{joined}}}" if style == "dict" else joined
-@lru_cache(maxsize=128)
+@lru_cache(maxsize=_CACHE_MAXSIZE_LARGE)
def calculate_min_extrema(
length: int, fit_live_predictions_candles: int, min_extrema: int = 2
) -> int:
return 2 * series - series.shift(int(lag))
-@lru_cache(maxsize=8)
+@lru_cache(maxsize=_CACHE_MAXSIZE_SMALL)
def get_ma_fn(
mamode: str,
) -> Callable[
return mamodes.get(mamode, mamodes["sma"])
-@lru_cache(maxsize=8)
+@lru_cache(maxsize=_CACHE_MAXSIZE_SMALL)
def get_zl_ma_fn(
mamode: str,
) -> Callable[
return smma
-@lru_cache(maxsize=8)
+@lru_cache(maxsize=_CACHE_MAXSIZE_SMALL)
def get_price_fn(pricemode: str) -> Callable[[pd.DataFrame], pd.Series]:
pricemodes = {
"average": ta.AVGPRICE,
raise ValueError(enum_error_message("regressor", regressor, REGRESSORS))
-@lru_cache(maxsize=128)
+@lru_cache(maxsize=_CACHE_MAXSIZE_LARGE)
def largest_divisor_to_step(integer: int, step: int) -> int | None:
if not isinstance(integer, int) or integer <= 0:
raise ValueError(
return nan_average(values, weights=shifted_exponentials)
-@lru_cache(maxsize=8)
+@lru_cache(maxsize=_CACHE_MAXSIZE_SMALL)
def get_min_max_label_period_candles(
fit_live_predictions_candles: int,
candles_step: int,
raise ValueError(f"Invalid step value {step!r}: must be a positive integer")
-@lru_cache(maxsize=128)
+@lru_cache(maxsize=_CACHE_MAXSIZE_LARGE)
def round_to_step(value: float | int, step: int) -> int:
"""
Round a value to the nearest multiple of a given step.
return int(round(float(value) / step) * step)
-@lru_cache(maxsize=128)
+@lru_cache(maxsize=_CACHE_MAXSIZE_LARGE)
def ceil_to_step(value: float | int, step: int) -> int:
_validate_step_args(value, step)
if isinstance(value, (int, np.integer)):
return int(math.ceil(float(value) / step) * step)
-@lru_cache(maxsize=128)
+@lru_cache(maxsize=_CACHE_MAXSIZE_LARGE)
def floor_to_step(value: float | int, step: int) -> int:
_validate_step_args(value, step)
if isinstance(value, (int, np.integer)):
repositories / freqai-strategies.git / commitdiff
