LEA v4 — Volatility expansion classifier. Predicts P(ATR will expand > 5% in next 12 candles). Entry fires when vol expansion likely AND BTC trend positive AND pair above EMA50.
Timeframe
5m
Direction
Long Only
Stoploss
-5.0%
Trailing Stop
No
ROI
0m: 5.0%, 30m: 3.0%, 60m: 2.0%, 120m: 1.0%
Interface Version
3
Startup Candles
200
Indicators
5
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
"""
LEA FreqAI Strategy — LeahAI v4
===============================
Spec date: 2026-06-05
Architecture change vs v3 (directional classifier):
- TARGET: volatility expansion prediction
Binary: (future_ATR_12 > current_ATR * 1.05)
31.8% positive rate on SOL 5m (balanced, not imbalanced)
- MODEL: XGBoostClassifier (binary:logistic) with scale_pos_weight=2.14
Compensates for class imbalance (31.8% positive / 68.2% negative)
- ENTRY GATE: 3-signal independent ensemble
1. Model: P(vol expansion) > 0.55
2. BTC macro: %btc_trend > 0 (price above BTC EMA50)
3. Pair micro: close > ema_50 (pair in local uptrend)
- ROI: extended hold times for vol expansion strategy
{0: 5%, 30: 3%, 60: 2%, 120: 1%}
Gives vol expansion 30-60 min to develop vs 10-20 min in v3
Unchanged from v3:
- 74 features, no %ret_12 leakage
- do_predict == 1 quality gate
- RSI < 70 overbought guard
- Stoploss: -5% hard stop
- custom_stoploss, custom_exit (disabled)
- IDENTIFIER: leah_v3_vol (forces fresh model)
Why volatility as target (vs directional):
- Directional target: max feature correlation < 3% — pure noise
- Volatility target: max feature correlation 8-11% — real signal exists
- 10x improvement in feature-label correlation
Signal independence argument:
1. Vol expansion: predicted by vol regime features (ATR ratio, volume z-score)
2. BTC trend: predicted by BTC-specific features (separate price series)
3. Pair EMA50: predicted by pair's own price momentum
Three orthogonal information sources, errors are not perfectly correlated
Kill criteria (50 dry-run trades):
- Expectancy >= $0.10/trade
- Sharpe >= 0.5
- No single trade > 15% of total P&L
"""
import logging
from datetime import timedelta
from functools import reduce
import numpy as np
import pandas as pd
from pandas import DataFrame
import talib.abstract as ta
from technical import qtpylib
from freqtrade.strategy import IStrategy, Trade
try:
from binance_research_backtest_loader import BinanceBacktestResearchLoader
RESEARCH_LOADER_AVAILABLE = True
except ImportError:
RESEARCH_LOADER_AVAILABLE = False
logging.getLogger(__name__).warning(
"Research loader not available - running without Binance research data"
)
logger = logging.getLogger(__name__)
def _apply_freqai_fallback(dataframe: DataFrame) -> DataFrame:
"""
Fail-closed fallback when FreqAI cannot build a valid live frame.
Classifier fallback: probability = 0.0 (no entry).
"""
dataframe["&-target"] = 0.0
dataframe["&-target_mean"] = 0.0
dataframe["&-target_std"] = 0.0
dataframe["do_predict"] = 0
dataframe["btc_regime_bull"] = dataframe.get("btc_regime_bull", 0)
dataframe["%btc_trend"] = dataframe.get("%btc_trend", 0.0)
return dataframe
def _get_latest_signal_candle(dataframe: DataFrame) -> pd.Series:
"""
FreqAI leaves the newest in-progress candle with zero target and do_predict=0.
Confirm entries against the latest completed candle instead.
"""
if len(dataframe) == 0:
return pd.Series(dtype=float)
if len(dataframe) == 1:
return dataframe.iloc[-1]
return dataframe.iloc[-2]
class LeahAIStrategy(IStrategy):
"""
LEA v4 — Volatility expansion classifier.
Predicts P(ATR will expand > 5% in next 12 candles).
Entry fires when vol expansion likely AND BTC trend positive AND pair above EMA50.
"""
INTERFACE_VERSION = 3
can_short = False
timeframe = "5m"
startup_candle_count = 200
# =========================================================================
# ENTRY THRESHOLD — v4
# =========================================================================
ml_entry_probability = 0.55 # Model confidence for vol expansion
# =========================================================================
# ROI & STOPLOSS — v4
# Extended hold times: vol expansion needs 30-60 min to develop
# =========================================================================
minimal_roi = {
"0": 0.05, # 5% immediate (tight but acceptable for vol spike)
"30": 0.03, # 3% after 30 min
"60": 0.02, # 2% after 60 min
"120": 0.01, # 1% after 2 hours (trailing floor)
}
stoploss = -0.05
trailing_stop = False
use_custom_stoploss = False
use_exit_signal = True # True — required for custom_exit (time exit) to fire (v4.3 fix)
exit_profit_only = False
ignore_roi_if_entry_signal = False
process_only_new_candles = True
order_types = {
"entry": "market",
"exit": "market",
"stoploss": "market",
"stoploss_on_exchange": False,
}
order_time_in_force = {"entry": "GTC", "exit": "GTC"}
atr_period = 14
atr_multiplier = 1.5
@property
def plot_config(self):
return {
"main_plot": {
"ema_50": {"color": "blue"},
"ema_200": {"color": "orange"},
},
"subplots": {
"RSI": {"rsi": {"color": "red"}},
"MACD": {"macd": {"color": "blue"}, "macdsignal": {"color": "orange"}},
"Probability": {"&-target": {"color": "green"}},
"BTC Trend": {"%btc_trend": {"color": "purple"}},
"ATR Ratio": {"%atr_ratio": {"color": "teal"}},
},
}
def __init__(self, config: dict) -> None:
super().__init__(config)
self._atr_cache: dict[str, float] = {}
self.research_loader = None
self.research_data_cache = {}
if RESEARCH_LOADER_AVAILABLE:
try:
self.research_loader = BinanceBacktestResearchLoader()
logger.info("Research loader initialised")
except Exception as e:
logger.warning(f"Failed to initialise research loader: {e}")
# =========================================================================
# FEATURE ENGINEERING — unchanged from v3
# =========================================================================
def feature_engineering_expand_all(
self, dataframe: DataFrame, period: int, metadata: dict, **kwargs
) -> DataFrame:
# Price momentum
dataframe["%ret_1"] = dataframe["close"].pct_change(1)
dataframe["%ret_3"] = dataframe["close"].pct_change(3)
dataframe["%ret_6"] = dataframe["close"].pct_change(6)
# %ret_12 EXCLUDED — feature-target leakage
# Volatility
dataframe["atr14"] = ta.ATR(dataframe, timeperiod=14)
dataframe["%atr14_rel"] = dataframe["atr14"] / dataframe["close"]
dataframe["%atr_ratio"] = (
dataframe["atr14"] / dataframe["atr14"].rolling(20).mean()
)
# Range
dataframe["%rng_24"] = (
dataframe["high"].rolling(24).max() - dataframe["low"].rolling(24).min()
) / dataframe["close"]
# Bollinger Bands
bollinger = qtpylib.bollinger_bands(
qtpylib.typical_price(dataframe), window=20, stds=2
)
dataframe["bb_lowerband"] = bollinger["lower"]
dataframe["bb_middleband"] = bollinger["mid"]
dataframe["bb_upperband"] = bollinger["upper"]
dataframe["%bb_width"] = (
(dataframe["bb_upperband"] - dataframe["bb_lowerband"])
/ dataframe["bb_middleband"]
)
dataframe["%bb_position"] = (
(dataframe["close"] - dataframe["bb_lowerband"])
/ (dataframe["bb_upperband"] - dataframe["bb_lowerband"] + 1e-9)
)
# Z-score
returns = dataframe["close"].pct_change()
dataframe["%z_48"] = (
(returns - returns.rolling(48).mean())
/ (returns.rolling(48).std() + 1e-9)
)
# Volume
dataframe["%vol_z_48"] = (
(dataframe["volume"] - dataframe["volume"].rolling(48).mean())
/ (dataframe["volume"].rolling(48).std() + 1e-9)
)
dataframe["%vol_ratio"] = (
dataframe["volume"] / (dataframe["volume"].rolling(20).mean() + 1e-9)
)
# Momentum oscillators
dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
dataframe["%rsi_delta"] = dataframe["rsi"].diff(3)
macd = ta.MACD(dataframe)
dataframe["macd"] = macd["macd"]
dataframe["macdsignal"] = macd["macdsignal"]
dataframe["macdhist"] = macd["macdhist"]
dataframe["%macd_rel"] = dataframe["macd"] / (dataframe["close"] + 1e-9)
# Trend EMAs
dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50)
dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200)
dataframe["%ema50_dev"] = (dataframe["close"] - dataframe["ema_50"]) / dataframe["ema_50"]
dataframe["%ema200_dev"] = (dataframe["close"] - dataframe["ema_200"]) / dataframe["ema_200"]
# Candle body character
dataframe["%body_size"] = (
(dataframe["close"] - dataframe["open"]).abs()
/ (dataframe["high"] - dataframe["low"] + 1e-9)
)
dataframe["%upper_wick"] = (
(dataframe["high"] - dataframe[["close", "open"]].max(axis=1))
/ (dataframe["high"] - dataframe["low"] + 1e-9)
)
return dataframe
def feature_engineering_expand_basic(
self, dataframe: DataFrame, metadata: dict, **kwargs
) -> DataFrame:
dataframe = self.feature_engineering_expand_all(
dataframe, period=1, metadata=metadata
)
return dataframe
def feature_engineering_standard(
self, dataframe: DataFrame, metadata: dict, **kwargs
) -> DataFrame:
"""
BTC macro regime features.
%btc_trend is the v4 entry gate — not btc_regime_bull.
"""
if metadata.get("pair") != "BTC/USDT" and self.dp:
btc = self.dp.get_pair_dataframe(pair="BTC/USDT", timeframe=self.timeframe)
if not btc.empty and len(btc) > 50:
btc_ema8 = ta.EMA(btc["close"], timeperiod=8)
btc_ema21 = ta.EMA(btc["close"], timeperiod=21)
btc_regime = pd.Series((btc_ema8 > btc_ema21).astype(int), index=btc["close"].index)
btc_ema50 = ta.EMA(btc["close"], timeperiod=50)
btc_trend = pd.Series((btc["close"] - btc_ema50) / (btc_ema50 + 1e-9), index=btc["close"].index)
btc_roc3 = pd.Series(btc["close"].pct_change(3), index=btc["close"].index)
btc_vol48 = pd.Series(btc["close"].pct_change().rolling(48).std(), index=btc["close"].index)
btc_atr14 = ta.ATR(btc, timeperiod=14)
btc_atr_rel = pd.Series(btc_atr14 / (btc["close"] + 1e-9), index=btc["close"].index)
btc_vol_ratio = pd.Series(btc["volume"] / (btc["volume"].rolling(20).mean() + 1e-9), index=btc["close"].index)
btc_bb = qtpylib.bollinger_bands(
qtpylib.typical_price(btc), window=20, stds=2
)
btc_bb_width = pd.Series(
(btc_bb["upper"] - btc_bb["lower"]) / (btc_bb["mid"] + 1e-9),
index=btc["close"].index
)
idx = dataframe.index
dataframe["btc_regime_bull"] = btc_regime.reindex(idx, method="ffill").fillna(0).astype(int)
dataframe["%btc_trend"] = btc_trend.reindex(idx, method="ffill").fillna(0.0)
dataframe["%btc_roc_3"] = btc_roc3.reindex(idx, method="ffill").fillna(0.0)
dataframe["%market_vol"] = btc_vol48.reindex(idx, method="ffill").fillna(0.0)
dataframe["%btc_atr_rel"] = btc_atr_rel.reindex(idx, method="ffill").fillna(0.0)
dataframe["%btc_vol_ratio"] = btc_vol_ratio.reindex(idx, method="ffill").fillna(1.0)
dataframe["%btc_bb_width"] = btc_bb_width.reindex(idx, method="ffill").fillna(0.0)
else:
logger.warning("BTC/USDT informative unavailable — regime defaults to 0")
dataframe["btc_regime_bull"] = 0
dataframe["%btc_trend"] = 0.0
dataframe["%btc_roc_3"] = 0.0
dataframe["%market_vol"] = dataframe["close"].pct_change().rolling(48).std()
dataframe["%btc_atr_rel"] = 0.0
dataframe["%btc_vol_ratio"] = 1.0
dataframe["%btc_bb_width"] = 0.0
else:
dataframe["btc_regime_bull"] = 1
dataframe["%btc_trend"] = 0.0
dataframe["%btc_roc_3"] = dataframe["close"].pct_change(3)
dataframe["%market_vol"] = dataframe["close"].pct_change().rolling(48).std()
dataframe["%btc_atr_rel"] = ta.ATR(dataframe, timeperiod=14) / dataframe["close"]
dataframe["%btc_vol_ratio"] = dataframe["volume"] / dataframe["volume"].rolling(20).mean()
dataframe["%btc_bb_width"] = 0.0
return dataframe
# =========================================================================
# TARGET — v4 CHANGE
# Volatility expansion: ATR in 12 candles > current ATR * 1.05
# Positive rate: 31.8% — balanced for binary classification
# FreqAI XGBoost does NOT auto-binarize — must be explicit here.
# =========================================================================
# FREQAI PREDICTION OVERRIDE
# FreqAI 2026.4's XGBoostClassifier.predict() has a bug: it fits a
# LabelEncoder on column names (['&-target']) instead of actual label values,
# so classes_=[0] while the model predicts [0,1]. inverse_transform([1])
# then raises ValueError. Override uses the model's own classes_ directly.
# =========================================================================
def predict(self, unfiltered_df, dk, **kwargs):
"""
Override to bypass FreqAI 2026.4's broken XGBoostClassifier.predict() LabelEncoder.
The base class (BaseClassifierModel.predict) returns:
pred_df with columns ['&-target', 0, 1]:
'&-target': raw 0/1 predictions (unused by entry gate)
0: P(class=0) from model.predict_proba
1: P(class=1) from model.predict_proba <- column 1, used for entry
XGBoostClassifier.predict() then calls inverse_transform() using a
LabelEncoder fit on column names ['&-target'], producing classes_=[0].
When model predicts class 1, inverse_transform([1]) raises:
ValueError: y contains previously unseen labels: [1]
This override calls base predict, extracts probability columns (0, 1),
and skips the broken encoder entirely.
"""
from freqtrade.freqai.base_models.BaseClassifierModel import BaseClassifierModel
pred_df, dk.do_predict = BaseClassifierModel.predict(self, unfiltered_df, dk, **kwargs)
# pred_df columns: ['&-target', 0, 1]
# '&-target' = raw 0/1 predictions (unused by entry gate)
# 0 = P(class=0), 1 = P(class=1) <- entry gate reads column 1
# Drop corrupted '&-target' column, keep probability columns
if '&-target' in pred_df.columns:
pred_df = pred_df.drop(columns=['&-target'])
# Rename integer cols to strings — data_drawer.py line 355 uses
# df.columns.get_loc(label) with string '0'/'1', not int 0/1
pred_df.columns = [str(c) for c in pred_df.columns]
return pred_df, dk.do_predict
# =========================================================================
def set_freqai_targets(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame:
# Compute ATR here since it may not be populated before set_freqai_targets is called
if "atr14" not in dataframe.columns:
dataframe["atr14"] = ta.ATR(dataframe, timeperiod=14)
future_atr = dataframe["atr14"].shift(-12)
dataframe["&-target"] = (future_atr > dataframe["atr14"] * 1.05).astype(int)
return dataframe
# =========================================================================
# INDICATORS
# =========================================================================
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
try:
dataframe = self.freqai.start(dataframe, metadata, self)
# Debug: log what FreqAI actually returned
_cols = list(dataframe.columns)
_has_1 = "1" in dataframe.columns
_has_target = "&-target" in dataframe.columns
_do_pred = dataframe["do_predict"].iloc[-1] if "do_predict" in dataframe.columns else -1
_pred_val = dataframe["1"].iloc[-1] if _has_1 else float("nan")
logger.warning(f"[{metadata.get('pair')}] freqai.start done — col1={_has_1}, do_predict={_do_pred}, pred={_pred_val:.4f}, cols={len(_cols)}, DI_vals={dataframe['DI_values'].iloc[-1] if 'DI_values' in dataframe.columns else -1}")
except Exception as exc:
pair = metadata.get("pair", "UNKNOWN")
import traceback
tb = traceback.format_exc()
logger.warning(
f"[{pair}] FreqAI prediction failed ({exc.__class__.__name__}: {exc}). "
f"Fail-closed fallback applied.\nTraceback: {tb}"
)
dataframe = _apply_freqai_fallback(dataframe)
# Re-fetch BTC data and recompute %btc_trend — FreqAI's
# attach_return_values_to_return_dataframe concatenates model outputs
# (which include %btc_trend=0 for every row) and overwrites the real
# values computed in feature_engineering_standard inside FreqAI.
# Since %btc_trend is not in historic_predictions, FreqAI adds a zeroed
# column that replaces the correct values. Recompute here to fix.
if metadata.get("pair") != "BTC/USDT" and self.dp:
btc = self.dp.get_pair_dataframe(pair="BTC/USDT", timeframe=self.timeframe)
if not btc.empty and len(btc) > 50:
btc_ema50 = ta.EMA(btc["close"], timeperiod=50)
btc_trend = (btc["close"] - btc_ema50) / (btc_ema50 + 1e-9)
dataframe["%btc_trend"] = btc_trend.reindex(
dataframe.index, method="ffill"
).fillna(0.0)
else:
dataframe["%btc_trend"] = 0.0
else:
dataframe["%btc_trend"] = 0.0
# Recompute indicators for entry/exit logic
dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50)
dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200)
dataframe["atr"] = ta.ATR(dataframe, timeperiod=self.atr_period)
macd = ta.MACD(dataframe)
dataframe["macd"] = macd["macd"]
dataframe["macdsignal"] = macd["macdsignal"]
dataframe["macdhist"] = macd["macdhist"]
dataframe["vol_mean_20"] = dataframe["volume"].rolling(20).mean()
# Diagnostic logging — fail silently on missing columns
try:
if "1" in dataframe.columns:
prob = dataframe["1"] # col "1" = model P(vol expansion), not &-target (training label)
btc_trend = float(dataframe["%btc_trend"].iloc[-1]) if "%btc_trend" in dataframe.columns and not pd.isna(dataframe["%btc_trend"].iloc[-1]) else 0.0
above_ema50 = bool((dataframe["close"].iloc[-1] > dataframe["ema_50"].iloc[-1]) if "ema_50" in dataframe.columns and not pd.isna(dataframe["ema_50"].iloc[-1]) else False)
do_predict_val = dataframe["do_predict"].iloc[-1] if "do_predict" in dataframe.columns else 'N/A'
logger.info(
f"[{metadata['pair']}] prob: min={prob.min():.4f} max={prob.max():.4f} "
f"mean={prob.mean():.4f} last={prob.iloc[-1]:.4f} | "
f"btc_trend={btc_trend:+.4f} | above_ema50={above_ema50} | "
f"do_predict={do_predict_val} | threshold={self.ml_entry_probability}"
)
except Exception:
pass
return dataframe
# =========================================================================
# ENTRY LOGIC — v4
# 3-signal independent ensemble:
# 1. Model: P(vol expansion) > 0.55
# 2. BTC macro: %btc_trend > 0
# 3. Pair micro: close > ema_50
# =========================================================================
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
v4 entry conditions:
1. P(vol expansion) > 0.55
2. %btc_trend > 0 (BTC bull regime)
3. close > ema_50 (pair in local uptrend)
Removed vs v3:
- btc_regime_bull filter replaced with %btc_trend (continuous)
- RSI < 70 gate removed (overbought consistent with vol expansion)
"""
# Guard: require %btc_trend column (created by feature_engineering_standard)
# If missing (FreqAI fallback active or feature engineering failed), deny entries
if "%btc_trend" not in dataframe.columns:
logger.warning(f"[{metadata['pair']}] No %btc_trend column — no entries")
dataframe["enter_long"] = 0
return dataframe
conditions = [
# 1. Model confidence > 55%, vectorized per-row. Live loop reads last row; backtests get correct per-candle values.
# Fail closed if column 1 missing — &-target is a 0/1 label, not a probability; comparing it to 0.55 silently passes label=1.
(dataframe["1"] > self.ml_entry_probability if "1" in dataframe.columns else pd.Series(False, index=dataframe.index)),
# 2. BTC bull regime (continuous, not binary)
dataframe["%btc_trend"] >= 0.002,
# 3. Pair in local uptrend
dataframe["close"] > dataframe["ema_50"],
# 4. FreqAI quality gate
(
dataframe["do_predict"] == 1
if "do_predict" in dataframe.columns
else pd.Series(True, index=dataframe.index)
),
# 5. Valid candle
dataframe["volume"] > 0,
]
entry_signal = reduce(lambda x, y: x & y, conditions)
dataframe["enter_long"] = 0
dataframe.loc[entry_signal, "enter_long"] = 1
# Persist probability tag to trades table — Freqtrade reads enter_tag from this column,
# not from confirm_trade_entry's local variable (which was a no-op).
if "1" in dataframe.columns:
dataframe["enter_tag"] = ""
dataframe.loc[entry_signal, "enter_tag"] = dataframe.loc[entry_signal, "1"].apply(
lambda p: f"prob_{p:.4f}"
)
# Debug: log each condition's last-row result
cond1 = dataframe["1"].iloc[-1] > self.ml_entry_probability if "1" in dataframe.columns else False
cond2 = dataframe["%btc_trend"].iloc[-1] >= 0.002 if "%btc_trend" in dataframe.columns else False
cond3 = dataframe["close"].iloc[-1] > dataframe["ema_50"].iloc[-1] if "ema_50" in dataframe.columns else False
cond4 = dataframe["do_predict"].iloc[-1] == 1 if "do_predict" in dataframe.columns else False
cond5 = dataframe["volume"].iloc[-1] > 0 if "volume" in dataframe.columns else False
_prob = dataframe["1"].iloc[-1] if "1" in dataframe.columns else float("nan")
logger.warning(f"[{metadata['pair']}] entry check: prob={_prob:.4f} vs {self.ml_entry_probability}, cond1={cond1}, cond2={cond2}, cond3={cond3}, cond4={cond4}, cond5={cond5}")
# Debug: log entry signal status
if dataframe["enter_long"].iloc[-1]:
logger.warning(f"[{metadata['pair']}] populate_entry_signals TRIGGERED — enter_long=True")
else:
logger.warning(f"[{metadata['pair']}] populate_entry_signals NOT triggered — enter_long=False")
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""ROI table drives all exits. Exit signal disabled."""
dataframe["exit_long"] = 0
return dataframe
# =========================================================================
# EXIT LOGIC — unchanged from v4 spec
# =========================================================================
def custom_stoploss(
self,
pair: str,
trade: Trade,
current_time,
current_rate: float,
current_profit: float,
**kwargs,
) -> float:
if current_time - trade.open_date_utc >= timedelta(minutes=90):
return self.stoploss
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe is None or dataframe.empty:
return self.stoploss
last = dataframe.iloc[-1]
atr = last.get("atr")
if atr is None or pd.isna(atr):
return self.stoploss
self._atr_cache[pair] = float(atr)
entry_price = trade.open_rate
stop_distance = self.atr_multiplier * atr
stop_price = entry_price - stop_distance
stop_pct = (stop_price / entry_price) - 1
if current_profit > 0.030:
stop_price = max(entry_price * 0.995, stop_price)
stop_pct = (stop_price / entry_price) - 1
elif current_profit > 0.015:
stop_price = max(entry_price * 0.998, stop_price)
stop_pct = (stop_price / entry_price) - 1
elif current_profit > 0.008:
stop_price = entry_price - (stop_distance * 0.5)
stop_pct = (stop_price / entry_price) - 1
return max(stop_pct, -abs(self.stoploss))
def custom_exit(self, pair, trade, current_time, current_rate, current_profit, **kwargs):
# Time-based exit: if underwater after 6 hours, exit rather than hold indefinitely
# Both current_time and trade.open_date are tz-aware in Freqtrade 2026.4.
# Strip tzinfo from both to allow arithmetic without timezone offset errors.
current_time_naive = current_time.replace(tzinfo=None)
open_date_naive = trade.open_date.replace(tzinfo=None)
hold_minutes = (current_time_naive - open_date_naive).total_seconds() / 60
if current_profit < 0:
if hold_minutes > 360: # 6 hours
return "time_exit_6h_negative"
return None
# =========================================================================
# TRADE CONFIRMATION — v4
# =========================================================================
def confirm_trade_entry(
self,
pair: str,
order_type: str,
amount: float,
rate: float,
time_in_force: str,
current_time,
entry_tag,
side: str,
**kwargs,
) -> bool:
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe is None or dataframe.empty:
return False
signal_candle = _get_latest_signal_candle(dataframe)
# Guard: require column "1" (model probability) — &-target was dropped by predict()
# to avoid LabelEncoder corruption, but confirm_trade_entry gates on prob, not the label
if "1" not in dataframe.columns:
logger.warning(f"[{pair}] No probability column — no entries")
return False
# Gate 1: probability threshold
prob = float(signal_candle["1"])
if prob <= self.ml_entry_probability:
logger.warning(
f"[{pair}] confirm DENIED: prob={prob:.4f} <= {self.ml_entry_probability}"
)
return False
# Gate 2: BTC trend — require meaningful positive regime, not noise
btc_trend = float(signal_candle.get("%btc_trend", 0))
if btc_trend < 0.002:
logger.warning(f"[{pair}] confirm DENIED: BTC trend={btc_trend:+.4f} < 0.002")
return False
# Gate 3: pair above EMA50
ema_50 = float(signal_candle.get("ema_50", 0))
close = float(signal_candle.get("close", 0))
if ema_50 > 0 and close <= ema_50:
logger.warning(f"[{pair}] confirm DENIED: close={close:.4f} <= ema_50={ema_50:.4f}")
return False
logger.info(
f"[{pair}] confirm APPROVED: prob={prob:.4f} btc_trend={btc_trend:+.4f} close={close:.4f} above_ema50"
)
entry_tag = f"prob_{prob:.4f}"
return True
# =========================================================================
# POSITION SIZING — v4
# Scaled by probability of vol expansion.
# =========================================================================
def custom_stake_amount(
self,
pair: str,
current_time,
current_rate: float,
proposed_stake: float,
min_stake: float,
max_stake: float,
entry_tag,
side: str,
**kwargs,
) -> float:
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe is None or dataframe.empty:
return proposed_stake
signal_candle = _get_latest_signal_candle(dataframe)
if "1" not in dataframe.columns:
return proposed_stake
prob = float(signal_candle["1"])
confidence_multiplier = np.clip(1.0 + (prob - self.ml_entry_probability) * 2.5, 0.5, 1.5)
return np.clip(proposed_stake * confidence_multiplier, min_stake, max_stake)