Estrategia long-only con confirmacion ML mediante FreqAI.
Timeframe
1h
Direction
Long Only
Stoploss
-7.0%
Trailing Stop
Yes
ROI
0m: 10.0%, 120m: 5.0%, 240m: 2.0%
Interface Version
3
Startup Candles
N/A
Indicators
2
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
"""
Estrategia FreqAI con LightGBMClassifier (Fase 3 - Con ML).
Objetivo del archivo
-------------------
Integrar FreqAI siguiendo el patron oficial para clasificadores:
- Features empiezan con "%".
- Target/clase empieza con "&s-".
- `self.freqai.class_names` define las clases discretas del clasificador.
- La prediccion vuelve en la misma columna target (`&s-up_or_down`).
- `do_predict == 1` indica que FreqAI considera confiable la prediccion.
Principio de riesgo
-------------------
ML no decide solo. Primero debe existir un setup tecnico razonable
(EMA/RSI/regimen); luego FreqAI confirma si la prediccion es "up".
"""
import logging
import numpy as np
from pandas import DataFrame
import talib.abstract as ta
from freqtrade.strategy import IStrategy
logger = logging.getLogger(__name__)
class FreqaiLightgbm(IStrategy):
"""
Estrategia long-only con confirmacion ML mediante FreqAI.
Flujo operativo:
1) Crear features tecnicas para FreqAI.
2) Crear target clasificador `&s-up_or_down` con clases `down` y `up`.
3) Ejecutar `self.freqai.start(...)` dentro de `populate_indicators`.
4) Entrar solo si setup tecnico + `do_predict == 1` + clase predicha `up`.
5) Salir por senal tecnica o clase predicha `down`.
"""
INTERFACE_VERSION = 3
timeframe = "1h"
stoploss = -0.07
trailing_stop = True
trailing_stop_positive = 0.01
trailing_stop_positive_offset = 0.03
trailing_only_offset_is_reached = True
minimal_roi = {
"0": 0.10,
"120": 0.05,
"240": 0.02,
}
process_only_new_candles = True
use_exit_signal = True
startup_candle_count: int = 220
@property
def protections(self):
"""
Protecciones de riesgo usadas en backtest, dry-run y live.
Mantener las mismas protecciones que la baseline permite comparar
FreqAI contra EmaCrossRsi bajo un regimen de riesgo equivalente.
"""
return [
{
"method": "CooldownPeriod",
"stop_duration_candles": 2,
},
{
"method": "StoplossGuard",
"trade_limit": 3,
"lookback_period_candles": 24,
"stop_duration_candles": 12,
"required_profit": 0.0,
"only_per_pair": False,
},
{
"method": "MaxDrawdown",
"lookback_period_candles": 72,
"trade_limit": 20,
"stop_duration_candles": 48,
"max_allowed_drawdown": 0.12,
},
{
"method": "LowProfitPairs",
"lookback_period_candles": 96,
"trade_limit": 4,
"stop_duration_candles": 48,
"required_profit": 0.0,
},
]
def feature_engineering_expand_all(
self, dataframe: DataFrame, period: int, metadata: dict, **kwargs
) -> DataFrame:
"""
Genera features que FreqAI expandira por periodo/timeframe/par/shift.
Args:
dataframe: Velas OHLCV del par actual.
period: Periodo inyectado por FreqAI desde `indicator_periods_candles`.
metadata: Metadata del par.
Returns:
DataFrame con features `%` reconocidas por FreqAI.
Nota:
Todo lo definido aqui puede multiplicarse mucho por la configuracion
FreqAI. Mantener este set pequeno reduce ruido y overfitting.
"""
dataframe["%-rsi-period"] = ta.RSI(dataframe, timeperiod=period)
dataframe["%-pct-change-period"] = dataframe["close"].pct_change(periods=period)
dataframe["%-close-ema-ratio-period"] = dataframe["close"] / ta.EMA(dataframe, timeperiod=period)
dataframe["%-volatility-period"] = dataframe["close"].pct_change().rolling(period).std()
dataframe["%-volume-ratio-period"] = dataframe["volume"] / dataframe["volume"].rolling(period).mean()
dataframe["%-candle-range-period"] = (
(dataframe["high"] - dataframe["low"]) / dataframe["close"]
).rolling(period).mean()
return dataframe
def feature_engineering_expand_basic(
self, dataframe: DataFrame, metadata: dict, **kwargs
) -> DataFrame:
"""
Genera features que FreqAI expande por timeframe/par/shift, pero no por periodo.
Args:
dataframe: Velas OHLCV del par actual.
metadata: Metadata del par.
Returns:
DataFrame con features `%` basicas.
"""
dataframe["%-raw-return"] = dataframe["close"].pct_change()
dataframe["%-is-green"] = (dataframe["close"] > dataframe["open"]).astype(int)
dataframe["%-volume-change"] = dataframe["volume"].pct_change()
return dataframe
def feature_engineering_standard(
self, dataframe: DataFrame, metadata: dict, **kwargs
) -> DataFrame:
"""
Genera features de base timeframe que NO deben autoexpandirse.
Args:
dataframe: Velas OHLCV del timeframe base.
metadata: Metadata del par.
Returns:
DataFrame con features temporales normalizadas.
Uso:
Hora y dia de semana se ubican aqui porque no tiene sentido
duplicarlas por cada periodo de indicador.
"""
dataframe["%-day_of_week"] = (dataframe["date"].dt.dayofweek + 1) / 7
dataframe["%-hour_of_day"] = (dataframe["date"].dt.hour + 1) / 25
return dataframe
def set_freqai_targets(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame:
"""
Define target clasificador oficial para FreqAI.
Target:
`&s-up_or_down` = "up" si el retorno futuro supera 1%.
`&s-up_or_down` = "down" en caso contrario.
Args:
dataframe: Velas OHLCV del par actual.
metadata: Metadata del par.
Returns:
DataFrame con columna target `&s-up_or_down`.
Nota:
`self.freqai.class_names` es requerido por clasificadores FreqAI
para mapear labels string a clases del modelo.
"""
self.freqai.class_names = ["down", "up"]
label_period = self.freqai_info["feature_parameters"]["label_period_candles"]
future_return = dataframe["close"].shift(-label_period) / dataframe["close"] - 1
dataframe["&s-up_or_down"] = np.where(future_return > 0.01, "up", "down")
return dataframe
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Calcula indicadores tecnicos y ejecuta FreqAI.
Args:
dataframe: Velas OHLCV del par actual.
metadata: Metadata del par.
Returns:
DataFrame con indicadores tecnicos y prediccion FreqAI.
Importante:
`self.freqai.start(...)` debe llamarse aqui, no en entry/exit.
"""
dataframe["ema_fast"] = ta.EMA(dataframe, timeperiod=20)
dataframe["ema_slow"] = ta.EMA(dataframe, timeperiod=50)
dataframe["ema_regime"] = ta.EMA(dataframe, timeperiod=200)
dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
dataframe = self.freqai.start(dataframe, metadata, self)
if "&s-up_or_down" in dataframe.columns and metadata.get("pair") == "BTC/USDT":
distribution = dataframe["&s-up_or_down"].value_counts(normalize=True).to_dict()
logger.info("FreqAI class distribution/prediction sample %s: %s", metadata.get("pair"), distribution)
return dataframe
@staticmethod
def _do_predict_ok(dataframe: DataFrame):
"""
Retorna mascara booleana para aceptar predicciones FreqAI.
Args:
dataframe: DataFrame retornado por `self.freqai.start(...)`.
Returns:
Serie booleana cuando existe `do_predict`; True si no existe.
Regla oficial:
`do_predict == 1` significa que la prediccion es confiable.
"""
if "do_predict" in dataframe.columns:
return dataframe["do_predict"] == 1
return True
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Define entradas long con setup tecnico + confirmacion FreqAI.
Args:
dataframe: DataFrame con indicadores y prediccion FreqAI.
metadata: Metadata del par.
Returns:
DataFrame con `enter_long` y `enter_tag` cuando aplica.
"""
if "&s-up_or_down" not in dataframe.columns:
return dataframe
technical_setup = (
(dataframe["ema_fast"] > dataframe["ema_slow"])
& (dataframe["close"] > dataframe["ema_regime"])
& (dataframe["rsi"] < 70)
)
ml_setup = self._do_predict_ok(dataframe) & (dataframe["&s-up_or_down"] == "up")
dataframe.loc[
technical_setup & ml_setup & (dataframe["volume"] > 0),
["enter_long", "enter_tag"],
] = (1, "tech_plus_freqai_up")
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Define salidas long por criterio tecnico o prediccion FreqAI bajista.
Args:
dataframe: DataFrame con indicadores y prediccion FreqAI.
metadata: Metadata del par.
Returns:
DataFrame con `exit_long` y `exit_tag` cuando aplica.
"""
technical_exit = (dataframe["ema_fast"] < dataframe["ema_slow"]) | (dataframe["rsi"] > 80)
ml_exit = False
if "&s-up_or_down" in dataframe.columns:
ml_exit = self._do_predict_ok(dataframe) & (dataframe["&s-up_or_down"] == "down")
dataframe.loc[
technical_exit | ml_exit,
["exit_long", "exit_tag"],
] = (1, "tech_or_freqai_down")
return dataframe