- ``timeseries_split``: chronological final-fold split.
Both paths compose per-row weights via ``_compose_per_row_weights``
before splitting and feed them to ``model.fit(sample_weight=...)``
- through ``_train_common``. Train and test weights are renormalized
- to mean=1 after ``feature_pipeline.fit_transform`` to preserve the
- invariant despite pipeline-level row drops.
+ through ``_train_common``.
"""
method = self.data_split_parameters.get(
"method", QuickAdapterRegressorV3.DATA_SPLIT_METHOD_DEFAULT
dk: FreqaiDataKitchen,
pair: str,
) -> dict:
- """
- Apply feature and label pipelines to train/test data.
-
- This helper reduces code duplication between train() methods that need
- custom data splitting but share the same pipeline application logic.
-
- :param dd: data_dictionary with train/test features/labels/weights
- :param dk: FreqaiDataKitchen instance
- :param pair: Trading pair (for error messages)
- :return: data_dictionary with transformed features/labels
- """
+ """Apply feature and label pipelines; renormalize weights post-transform."""
dk.feature_pipeline = self.define_data_pipeline(threads=dk.thread_count)
dk.label_pipeline = self.define_label_pipeline(threads=dk.thread_count)
weights: NDArray[np.floating],
dk: FreqaiDataKitchen,
) -> dict:
- """
- Chronological train/test split using the final fold from sklearn's TimeSeriesSplit.
-
- n_splits controls train/test proportions (higher = larger train set).
- gap excludes samples between train/test; when 0, auto-calculated from
- label_period_candles. max_train_size enables sliding window mode.
-
- :param filtered_dataframe: Feature data to split
- :param labels: Label data to split
- :param weights: Pre-computed per-row sample weights aligned to
- filtered_dataframe rows by position; sliced via
- ``weights[train_idx]`` / ``weights[test_idx]``.
- :param dk: FreqaiDataKitchen instance for data building
- :return: data_dictionary with train/test features/labels/weights
+ """Chronological train/test split using sklearn's TimeSeriesSplit final fold.
+
+ ``n_splits`` controls train/test proportions (higher = larger train).
+ ``gap`` excludes samples between train and test; when 0, auto-derived
+ from ``label_period_candles``. ``max_train_size`` enables sliding
+ window mode. ``weights`` is sliced positionally via ``train_idx`` /
+ ``test_idx``.
"""
feat_dict = self.ft_params
if feat_dict.get("shuffle_after_split", False):
def fit(
self, data_dictionary: dict[str, Any], dk: FreqaiDataKitchen, **kwargs
) -> Any:
- """
- User sets up the training and test data to fit their desired model here
- :param data_dictionary: the dictionary constructed by DataHandler to hold
- all the training and test data/labels.
- :param dk: the FreqaiDataKitchen object
- """
-
X = data_dictionary.get("train_features")
y = data_dictionary.get("train_labels")
train_weights = data_dictionary.get("train_weights")
return timeframe_to_prev_date(self.config.get("timeframe"), trade.open_date_utc)
def get_trade_duration_candles(self, df: DataFrame, trade: Trade) -> Optional[int]:
- """
- Get the number of candles since the trade entry.
- :param df: DataFrame with the current data
- :param trade: Trade object
- :return: Number of candles since the trade entry
- """
entry_date = self.get_trade_entry_date(trade)
dates = df.get("date")
if dates is None or dates.empty:
min_natr_multiplier_fraction: float,
max_natr_multiplier_fraction: float,
) -> bool:
- """Confirm a directional reversal using a volatility-adaptive current-candle
- threshold and optionally a backward confirmation chain with geometric decay.
-
- Overview
- --------
- 1. Compute a deviation-based threshold on the latest candle (-1). The current
- rate must strictly break it (long: rate > threshold; short: rate < threshold).
- 2. If lookback_period_candles > 0, for each k = 1..lookback_period_candles:
- - Decay (min_natr_multiplier_fraction, max_natr_multiplier_fraction) by
- (decay_fraction ** k), clamped to [0, 1].
- - Recompute the threshold on candle index -(k+1).
- - Require close[-k] to have strictly broken that historical threshold.
- 3. If an intermediate close or threshold is non-finite, chain evaluation aborts
- and the function falls back to step 1 result only (permissive fallback).
-
- Parameters
- ----------
- df : DataFrame
- Must contain 'open', 'close' and the NATR label series used indirectly.
- pair : str
- Trading pair identifier.
- side : {'long','short'}
- Direction to confirm.
- order : {'entry','exit'}
- Context (affects log wording only).
- rate : float
- Candidate execution price; must break the current threshold.
- lookback_period_candles : int
- Number of historical confirmation steps requested; truncated to history.
- decay_fraction : float
- Geometric decay factor per step (0 < decay_fraction <= 1); 1.0 disables decay.
- min_natr_multiplier_fraction : float
- Lower-bound fraction (e.g. 0.009 = 0.9%).
- max_natr_multiplier_fraction : float
- Upper-bound fraction (>= lower bound).
-
- Returns
- -------
- bool
- True iff the current threshold is broken AND (lookback chain succeeded OR
- a permissive fallback occurred). False otherwise.
-
- Fallback Semantics
- ------------------
- Missing / non-finite intermediate data -> stop chain; return current candle result.
- This may yield True on partial history, weakening strict multi-candle guarantees.
-
- Rejection Conditions
- --------------------
- Empty dataframe, invalid side/order, non-finite rate, negative lookback,
- decay_fraction outside (0,1], invalid min/max ordering, failure to break current
- threshold, or failed historical step comparison.
-
- Complexity
- ----------
- O(lookback_period_candles) threshold computations.
-
- Logging
- -------
- Logs rejection reasons (invalid decay_fraction, threshold not broken, failed step).
- Fallback aborts are silent.
-
- Limitations
- -----------
- No strict mode; partial data may still confirm.
+ """Confirm a directional reversal using a volatility-adaptive threshold.
+
+ Computes a deviation-based threshold on the latest candle (-1); ``rate``
+ must strictly break it (long: ``rate > threshold``; short: ``rate <
+ threshold``). When ``lookback_period_candles > 0``, requires that for
+ each ``k = 1..lookback_period_candles`` the close at ``-k`` strictly
+ broke the threshold recomputed at ``-(k+1)`` with the natr-multiplier
+ bounds geometrically decayed by ``decay_fraction ** k`` clamped to
+ ``[0, 1]``. Non-finite intermediate close or threshold aborts the chain
+ and falls back permissively to the current-candle result, which may
+ weaken strict multi-candle guarantees. Returns False on empty
+ dataframe, invalid side/order, non-finite rate, negative lookback,
+ ``decay_fraction`` outside ``(0, 1]``, or invalid min/max ordering.
"""
if df.empty:
return False
float,
float,
]:
- """Compute velocity and acceleration from PnL history.
-
- Velocity is the first derivative of PnL, acceleration is the second.
-
- Args:
- unrealized_pnl_history: PnL values sequence.
- window_size: Recent window size (0 = no windowing).
+ """Compute velocity (first derivative) and acceleration (second) from PnL history.
- Returns:
- (velocity_values, velocity_mean, velocity_std,
- acceleration_values, acceleration_mean, acceleration_std)
+ ``window_size > 0`` truncates to the most recent window before
+ differencing. Returns
+ ``(velocity_values, velocity_mean, velocity_std, acceleration_values,
+ acceleration_mean, acceleration_std)``.
"""
unrealized_pnl_history_array = np.asarray(unrealized_pnl_history, dtype=float)
@staticmethod
@lru_cache(maxsize=128)
def _t_statistic(mean: float, std: float, n: int) -> float:
- """Compute t-statistic for H₀: μ = 0.
+ """Compute t-statistic for H0: mu = 0 as ``mean * sqrt(n) / std``.
- Formula: t = mean * √n / std
-
- Args:
- mean: Sample mean.
- std: Sample standard deviation (ddof=1).
- n: Sample size.
-
- Returns:
- t-statistic, or NaN if n < 2 or std ≈ 0.
+ Returns NaN when ``n < 2``, ``std`` is approximately zero, or any
+ input is non-finite.
"""
if n < 2:
return np.nan
@staticmethod
@lru_cache(maxsize=128)
def _effective_df(x: tuple[float, ...]) -> float:
- """Compute effective degrees of freedom with Bartlett's autocorrelation correction.
-
- Formula: df_eff = (n - 1) * (1 - ρ₁) / (1 + ρ₁), where ρ₁ is lag-1 autocorrelation.
+ """Effective degrees of freedom with Bartlett's autocorrelation correction.
- Args:
- x: Observations tuple.
-
- Returns:
- Effective df (≥ 1). Falls back to n - 1 if n < 4 or on error.
+ Computes ``df_eff = (n - 1) * (1 - rho1) / (1 + rho1)`` where ``rho1``
+ is the lag-1 autocorrelation clamped to ``[-0.99, 0.99]``. Falls back
+ to ``n - 1`` when ``n < 4`` or pearsonr fails. Result is bounded
+ below by 1.
"""
n = len(x)
if n < 4:
@staticmethod
@lru_cache(maxsize=128)
def _t_critical(q: float, df: float, default_t: float) -> float:
- """Compute critical t-value from Student's t-distribution.
-
- Args:
- q: Quantile in (0, 1), e.g. 0.75.
- df: Degrees of freedom.
- default_t: Fallback value on error.
+ """Critical t-value from Student's t-distribution at quantile ``q``.
- Returns:
- t.ppf(q, df), or default_t if invalid inputs.
+ Returns ``default_t`` on invalid inputs or scipy failure.
"""
if not (0.0 < q < 1.0):
return default_t
side: str,
**kwargs: Any,
) -> float:
- """
- Customize leverage for each new trade. This method is only called in trading modes
- which allow leverage (margin / futures). The strategy is expected to return a
- leverage value between 1.0 and max_leverage.
-
- :param pair: Pair that's currently analyzed
- :param current_time: datetime object, containing the current datetime
- :param current_rate: Rate, calculated based on pricing settings in exit_pricing.
- :param proposed_leverage: A leverage proposed by the bot.
- :param max_leverage: Max leverage allowed on this pair
- :param entry_tag: Optional entry_tag (buy_tag) if provided with the buy signal.
- :param side: 'long' or 'short' - indicating the direction of the proposed trade
- :return: A leverage amount, which will be between 1.0 and max_leverage.
- """
return min(self.config.get("leverage", proposed_leverage), max_leverage)
def plot_annotations(
dataframe: DataFrame,
**kwargs: Any,
) -> list[AnnotationType]:
- """
- Plot annotations.
-
- :param pair: Pair that's currently being plotted
- :param start_date: Start date of the chart range
- :param end_date: End date of the chart range
- :param dataframe: DataFrame with analyzed data for this pair
- :param **kwargs: Additional arguments
- :return: List of annotations to display on the chart
- """
annotations: list[AnnotationType] = []
open_trades = Trade.get_trades_proxy(pair=pair, is_open=True)
_set_path(config, new_path, old_value)
_delete_path(config, old_path)
if old_section == new_section:
- logger.warning(f"{old_path} is deprecated, use {new_key} instead")
+ logger.warning(f"{old_path!r} is deprecated, use {new_key!r} instead")
else:
- logger.warning(f"{old_path} has moved to {new_path}")
+ logger.warning(f"{old_path!r} has moved to {new_path!r}")
else:
_delete_path(config, old_path)
if old_section == new_section:
logger.warning(
- f"{new_section} has both {new_key} and deprecated {old_path.rsplit('.', 1)[-1]}, using {new_key}"
+ f"{new_section!r} has both {new_key!r} and deprecated {old_path.rsplit('.', 1)[-1]!r}, using {new_key!r}"
)
else:
logger.warning(
- f"{new_section} has {new_key} and deprecated {old_path}, using {new_path}"
+ f"{new_section!r} has {new_key!r} and deprecated {old_path!r}, using {new_path!r}"
)
def top_log_return(dataframe: pd.DataFrame, period: int) -> pd.Series:
- """
- Logarithmic return from rolling maximum: log(close / rolling_max).
-
- Measures distance below the highest close in previous `period` bars.
- Returns ≤ 0 (e.g., -0.10 ≈ -9.5% below peak). Zero when at peak.
+ """Logarithmic return from rolling maximum: ``log(close / rolling_max)``.
- :param dataframe: OHLCV DataFrame with 'close' column
- :param period: Lookback window (>=1)
- :return: Log return series (≤ 0)
+ Measures distance below the highest close in previous ``period`` bars.
+ Returns <= 0 (e.g. -0.10 ~ -9.5% below peak), zero when at peak.
"""
if period < 1:
raise ValueError(f"Invalid period value {period!r}: must be >= 1")
def bottom_log_return(dataframe: pd.DataFrame, period: int) -> pd.Series:
- """
- Logarithmic return from rolling minimum: log(close / rolling_min).
+ """Logarithmic return from rolling minimum: ``log(close / rolling_min)``.
- Measures distance above the lowest close in previous `period` bars.
- Returns ≥ 0 (e.g., +0.10 ≈ +10.5% above bottom). Zero when at bottom.
-
- :param dataframe: OHLCV DataFrame with 'close' column
- :param period: Lookback window (>=1)
- :return: Log return series (≥ 0)
+ Measures distance above the lowest close in previous ``period`` bars.
+ Returns >= 0 (e.g. +0.10 ~ +10.5% above bottom), zero when at bottom.
"""
if period < 1:
raise ValueError(f"Invalid period value {period!r}: must be >= 1")
def price_retracement_percent(dataframe: pd.DataFrame, period: int) -> pd.Series:
- """
- Normalized position (0-1) of close within rolling high/low range, using log scale.
+ """Normalized log-scale position of close within rolling high/low range.
- Formula: log(close / low) / log(high / low)
-
- Returns 0 at bottom, 1 at top, 0.5 at geometric midpoint (not arithmetic).
- Example: range [100, 200] → midpoint at ~141, not 150.
-
- :param dataframe: OHLCV DataFrame with 'close' column
- :param period: Lookback window (>=1)
- :return: Normalized position (0 to 1)
+ Formula: ``log(close / low) / log(high / low)``. Returns 0 at bottom, 1
+ at top, 0.5 at geometric (not arithmetic) midpoint; e.g. range [100,
+ 200] has midpoint at ~141.
"""
if period < 1:
raise ValueError(f"Invalid period value {period!r}: must be >= 1")
model_path: Optional[Path] = None,
trial: Optional[optuna.trial.Trial] = None,
) -> Any:
- """Fit a regressor model."""
fit_callbacks = list(callbacks) if callbacks else []
has_eval_set = (
Optional[list[tuple[pd.DataFrame, pd.DataFrame]]],
Optional[list[NDArray[np.floating]]],
]:
+ """Wrap test data for ``model.fit`` ``eval_set`` when ``test_size > 0``.
+
+ Returns ``(None, None)`` when ``test_size <= 0`` to suppress evaluation.
+ """
if test_size <= 0:
return None, None
repositories / freqai-strategies.git / commitdiff
