Strategy 0.4501 - Converted from StrategyQuantX
Timeframe
15m
Direction
Long & Short
Stoploss
-2.0%
Trailing Stop
No
ROI
0m: 10.0%
Interface Version
3
Startup Candles
N/A
Indicators
6
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
# isort: skip_file
# --- Do not remove these imports ---
import numpy as np
import pandas as pd
from pandas import DataFrame
from typing import Optional, Union
from freqtrade.strategy import (
BooleanParameter,
CategoricalParameter,
DecimalParameter,
IStrategy,
IntParameter,
merge_informative_pair,
)
# --------------------------------
from datetime import datetime
from freqtrade.persistence import Trade
from freqtrade.exchange import timeframe_to_minutes
import talib.abstract as ta
import pandas_ta as pta
# Suppress pandas_ta FutureWarnings about ChainedAssignment
import warnings
warnings.filterwarnings("ignore", category=FutureWarning, module="pandas_ta")
class Strategy_0_4501(IStrategy):
"""
Strategy 0.4501 - Converted from StrategyQuantX
This strategy uses fuzzy logic for entry/exit signals combining HeikenAshi,
MTATR, Williams %R, Bollinger Bands, and various session indicators.
"""
# Strategy interface version
INTERFACE_VERSION = 3
# Optimal timeframe for the strategy
timeframe = "15m"
# Can this strategy go short?
can_short = True
# Minimal ROI designed for the strategy
minimal_roi = {"0": 0.10}
# Optimal stoploss
stoploss = -0.02 # 2% from parameters
# Trailing stoploss
trailing_stop = False
# Run "populate_indicators()" only for new candle
process_only_new_candles = True
# These values can be overridden in the config
use_exit_signal = True
exit_profit_only = False
ignore_roi_if_entry_signal = False
# Number of candles the strategy requires before producing valid signals
startup_candle_count: int = 500
# Strategy parameters from pseudocode
# Hyperparameters for optimization
indicator_ma_period_1 = IntParameter(50, 85, default=68, space="entry")
indicator_crs_ma_period = IntParameter(20, 40, default=30, space="entry")
mt_atr_period = IntParameter(15, 25, default=20, space="entry")
bb_bar_opens_period = IntParameter(25, 50, default=37, space="entry")
wpr_cross_period = IntParameter(15, 30, default=20, space="entry")
indicator_ma_period_2 = IntParameter(60, 95, default=79, space="entry")
bb_bar_closes_period = IntParameter(35, 65, default=50, space="entry")
ma_period = IntParameter(35, 65, default=50, space="exit")
bs_power_period = IntParameter(10, 20, default=15, space="exit")
# Entry/Exit parameters
price_entry_mult = DecimalParameter(0.1, 0.5, default=0.3, space="entry")
exit_after_bars = IntParameter(3, 8, default=5, space="exit")
profit_target_coef = DecimalParameter(3.5, 6.0, default=4.7, space="exit")
stop_loss_pct = DecimalParameter(1.0, 3.0, default=2.0, space="exit")
keltner_channel_period = IntParameter(15, 30, default=20, space="entry")
atr_period = IntParameter(20, 40, default=30, space="entry")
def informative_pairs(self):
"""
Define additional, informative pair/interval combinations to be cached from the exchange.
"""
pairs = self.dp.current_whitelist()
informative_pairs = [(pair, "1h") for pair in pairs]
return informative_pairs
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Adds several different TA indicators to the given DataFrame
"""
# Calculate main timeframe indicators
# HeikenAshi
ha = pta.ha(
dataframe["open"],
dataframe["high"],
dataframe["low"],
dataframe["close"],
)
dataframe["ha_close"] = ha["HA_close"]
dataframe["ha_open"] = ha["HA_open"]
# MTATR (approximation using standard ATR)
dataframe["mtatr"] = ta.ATR(dataframe, timeperiod=self.mt_atr_period.value)
# EMA for MTATR crossovers
dataframe["mtatr_ema"] = ta.EMA(
dataframe["mtatr"], timeperiod=self.indicator_crs_ma_period.value
)
# Bollinger Bands
bb_upper, bb_middle, bb_lower = ta.BBANDS(
dataframe["close"],
timeperiod=self.bb_bar_opens_period.value,
nbdevup=2.0,
nbdevdn=2.0,
)
dataframe["bb_upper"] = bb_upper
dataframe["bb_lower"] = bb_lower
# Williams %R
dataframe["wpr"] = ta.WILLR(dataframe, timeperiod=self.wpr_cross_period.value)
# True Range
dataframe["true_range"] = ta.TRANGE(dataframe)
# SMA for True Range
dataframe["true_range_sma"] = ta.SMA(
dataframe["true_range"], timeperiod=self.indicator_ma_period_2.value
)
# Monthly High/Low approximation (30 days * 24 hours * 4 15-min candles)
monthly_period = 30 * 24 * 4
dataframe["high_monthly"] = dataframe["high"].rolling(window=monthly_period).max()
dataframe["low_monthly"] = dataframe["low"].rolling(window=monthly_period).min()
dataframe["close_monthly"] = dataframe["close"].shift(monthly_period)
# Weekly High/Low approximation (7 days * 24 hours * 4 15-min candles)
weekly_period = 7 * 24 * 4
dataframe["high_weekly"] = dataframe["high"].rolling(window=weekly_period).max()
dataframe["low_weekly"] = dataframe["low"].rolling(window=weekly_period).min()
# Session High/Low (4:33-12:11 approximation)
dataframe["hour"] = dataframe["date"].dt.hour
dataframe["minute"] = dataframe["date"].dt.minute
def is_in_session(row):
hour = row["hour"]
minute = row["minute"]
# 4:33 to 12:11
if hour == 4 and minute >= 30:
return True
elif 5 <= hour <= 11:
return True
elif hour == 12 and minute <= 15:
return True
return False
dataframe["in_session"] = dataframe.apply(is_in_session, axis=1)
dataframe["session_high"] = dataframe.apply(
lambda row: row["high"] if row["in_session"] else np.nan, axis=1
)
dataframe["session_low"] = dataframe.apply(
lambda row: row["low"] if row["in_session"] else np.nan, axis=1
)
dataframe["session_high"] = dataframe["session_high"].ffill()
dataframe["session_low"] = dataframe["session_low"].ffill()
# Session Close (7:38)
dataframe["session_close"] = dataframe.apply(
lambda row: row["close"] if (row["hour"] == 7 and 30 <= row["minute"] < 45) else np.nan,
axis=1,
).ffill()
# Keltner Channel
kc = pta.kc(
dataframe["high"],
dataframe["low"],
dataframe["close"],
length=self.keltner_channel_period.value,
scalar=2.0,
)
# Keltner channels returns columns: KCLe_<length>_<scalar>, KCBe_<length>_<scalar>, KCUe_<length>_<scalar>
# Access by column position for reliability: 0=lower, 1=basis, 2=upper
if kc is not None and len(kc.columns) >= 3:
dataframe["kc_lower"] = kc.iloc[:, 0] # Lower band
dataframe["kc_upper"] = kc.iloc[:, 2] # Upper band
else:
# Fallback if kc calculation fails
dataframe["kc_upper"] = dataframe["close"] * 1.02
dataframe["kc_lower"] = dataframe["close"] * 0.98
# ATR for entry calculations
dataframe["atr"] = ta.ATR(dataframe, timeperiod=self.atr_period.value)
dataframe["atr_20"] = ta.ATR(dataframe, timeperiod=20) # Fixed ATR
# Smoothed Moving Average
dataframe["smma"] = ta.EMA(
dataframe["close"], timeperiod=self.ma_period.value * 2
) # SMMA approximation
# Bears Power and Bulls Power
ema_period = self.bs_power_period.value
dataframe["ema_bears"] = ta.EMA(dataframe["close"], timeperiod=ema_period)
dataframe["bears_power"] = dataframe["low"] - dataframe["ema_bears"]
dataframe["bulls_power"] = dataframe["high"] - dataframe["ema_bears"]
# Add 1h informative indicators
informative_1h = self.dp.get_pair_dataframe(pair=metadata["pair"], timeframe="1h")
# Calculate 1h indicators
informative_1h["high_weekly_1h"] = informative_1h["high"].rolling(window=7 * 24).max()
informative_1h["low_weekly_1h"] = informative_1h["low"].rolling(window=7 * 24).min()
# Session High/Low for 1h
def is_in_session_1h(row):
hour = row["hour"]
# 4:33 to 12:11
if 4 <= hour <= 12:
return True
return False
informative_1h["hour"] = informative_1h["date"].dt.hour
informative_1h["in_session_1h"] = informative_1h.apply(is_in_session_1h, axis=1)
informative_1h["session_high_1h"] = informative_1h.apply(
lambda row: row["high"] if row["in_session_1h"] else np.nan, axis=1
)
informative_1h["session_low_1h"] = informative_1h.apply(
lambda row: row["low"] if row["in_session_1h"] else np.nan, axis=1
)
informative_1h["session_high_1h"] = informative_1h["session_high_1h"].ffill()
informative_1h["session_low_1h"] = informative_1h["session_low_1h"].ffill()
# Bollinger Bands for 1h
bb_upper, bb_middle, bb_lower = ta.BBANDS(
informative_1h["close"],
timeperiod=self.bb_bar_closes_period.value,
nbdevup=2.1,
nbdevdn=2.1,
)
informative_1h["bb_upper_1h"] = bb_upper
informative_1h["bb_lower_1h"] = bb_lower
# HeikenAshi for 1h
ha_1h = pta.ha(
informative_1h["open"],
informative_1h["high"],
informative_1h["low"],
informative_1h["close"],
)
informative_1h["ha_close_1h"] = ha_1h["HA_close"]
# Merge informative pair
dataframe = merge_informative_pair(
dataframe, informative_1h, self.timeframe, "1h", ffill=True
)
return dataframe
def fuzzy_logic_long_entry(self, dataframe: DataFrame, index: int) -> bool:
"""
Fuzzy logic for long entry - at least 42% of conditions (3 out of 8) must be true
"""
conditions = []
# Condition 1: HeikenAshiClose[1] < HighMonthly[2]
if index > 2:
conditions.append(
dataframe.iloc[index - 1]["ha_close"] < dataframe.iloc[index - 2]["high_monthly"]
)
else:
conditions.append(False)
# Condition 2: -735601814 is above its EMA (ignoring invalid condition)
conditions.append(False)
# Condition 3: MTATR[3] crosses below its EMA
if index > 3:
conditions.append(
(dataframe.iloc[index - 4]["mtatr"] > dataframe.iloc[index - 4]["mtatr_ema"])
and (dataframe.iloc[index - 3]["mtatr"] < dataframe.iloc[index - 3]["mtatr_ema"])
)
else:
conditions.append(False)
# Condition 4: Open[5] above BB Lower[6]
if index > 6:
conditions.append(
dataframe.iloc[index - 5]["open"] > dataframe.iloc[index - 6]["bb_lower"]
)
else:
conditions.append(False)
# Condition 5: Williams %R[5] crosses -80 upwards
if index > 5:
conditions.append(
(dataframe.iloc[index - 6]["wpr"] < -80)
and (dataframe.iloc[index - 5]["wpr"] > -80)
)
else:
conditions.append(False)
# Condition 6: TrueRange[4] is below its SMA
if index > 4:
conditions.append(
dataframe.iloc[index - 4]["true_range"]
< dataframe.iloc[index - 4]["true_range_sma"]
)
else:
conditions.append(False)
# Condition 7: SessionLow[1] < LowWeekly(1h)[2]
if index > 2:
conditions.append(
dataframe.iloc[index - 1]["session_low_1h_1h"]
< dataframe.iloc[index - 2]["low_weekly_1h_1h"]
)
else:
conditions.append(False)
# Condition 8: Close[1] below BB Lower(1h)[1]
if index > 1:
conditions.append(
dataframe.iloc[index - 1]["close"] < dataframe.iloc[index - 1]["bb_lower_1h_1h"]
)
else:
conditions.append(False)
# Return True if at least 42% of conditions are met (3 out of 8)
return sum(conditions) >= 3
def fuzzy_logic_short_entry(self, dataframe: DataFrame, index: int) -> bool:
"""
Fuzzy logic for short entry - at least 42% of conditions (3 out of 8) must be true
"""
conditions = []
# Condition 1: HeikenAshiClose[1] > LowMonthly[2]
if index > 2:
conditions.append(
dataframe.iloc[index - 1]["ha_close"] > dataframe.iloc[index - 2]["low_monthly"]
)
else:
conditions.append(False)
# Condition 2: -735601814 is below its EMA (ignoring invalid condition)
conditions.append(False)
# Condition 3: MTATR[3] crosses above its EMA
if index > 3:
conditions.append(
(dataframe.iloc[index - 4]["mtatr"] < dataframe.iloc[index - 4]["mtatr_ema"])
and (dataframe.iloc[index - 3]["mtatr"] > dataframe.iloc[index - 3]["mtatr_ema"])
)
else:
conditions.append(False)
# Condition 4: Open[5] below BB Upper[6]
if index > 6:
conditions.append(
dataframe.iloc[index - 5]["open"] < dataframe.iloc[index - 6]["bb_upper"]
)
else:
conditions.append(False)
# Condition 5: Williams %R[5] crosses -20 downwards
if index > 5:
conditions.append(
(dataframe.iloc[index - 6]["wpr"] > -20)
and (dataframe.iloc[index - 5]["wpr"] < -20)
)
else:
conditions.append(False)
# Condition 6: TrueRange[4] is above its SMA
if index > 4:
conditions.append(
dataframe.iloc[index - 4]["true_range"]
> dataframe.iloc[index - 4]["true_range_sma"]
)
else:
conditions.append(False)
# Condition 7: SessionHigh[1] > HighWeekly(1h)[2]
if index > 2:
conditions.append(
dataframe.iloc[index - 1]["session_high_1h_1h"]
> dataframe.iloc[index - 2]["high_weekly_1h_1h"]
)
else:
conditions.append(False)
# Condition 8: Close[1] above BB Upper(1h)[1]
if index > 1:
conditions.append(
dataframe.iloc[index - 1]["close"] > dataframe.iloc[index - 1]["bb_upper_1h_1h"]
)
else:
conditions.append(False)
# Return True if at least 42% of conditions are met (3 out of 8)
return sum(conditions) >= 3
def fuzzy_logic_long_exit(self, dataframe: DataFrame, index: int) -> bool:
"""
Fuzzy logic for long exit - at least 80% of conditions (4 out of 5) must be true
"""
conditions = []
# Condition 1: SessionClose[2] crosses below CloseMonthly[2]
if index > 2:
conditions.append(
(
dataframe.iloc[index - 3]["session_close"]
> dataframe.iloc[index - 3]["close_monthly"]
)
and (
dataframe.iloc[index - 2]["session_close"]
< dataframe.iloc[index - 2]["close_monthly"]
)
)
else:
conditions.append(False)
# Condition 2: SMMA[2] is rising
if index > 2:
conditions.append(dataframe.iloc[index - 2]["smma"] > dataframe.iloc[index - 3]["smma"])
else:
conditions.append(False)
# Condition 3: BearsPower[4] is rising for 25 bars at 5 bar ago
if index > 30:
condition_met = True
for i in range(25):
bar_index = index - 5 - i
if (
dataframe.iloc[bar_index]["bears_power"]
<= dataframe.iloc[bar_index - 1]["bears_power"]
):
condition_met = False
break
conditions.append(condition_met)
else:
conditions.append(False)
# Condition 4: Close[3] crosses above Open[3]
if index > 3:
conditions.append(
dataframe.iloc[index - 3]["close"] > dataframe.iloc[index - 3]["open"]
)
else:
conditions.append(False)
# Condition 5: HeikenAshiClose(1h)[3] is falling for 5 bars at 3 bar ago
if index > 8:
condition_met = True
for i in range(5):
bar_index = index - 3 - i
if (
dataframe.iloc[bar_index]["ha_close_1h_1h"]
>= dataframe.iloc[bar_index - 1]["ha_close_1h_1h"]
):
condition_met = False
break
conditions.append(condition_met)
else:
conditions.append(False)
# Return True if at least 80% of conditions are met (4 out of 5)
return sum(conditions) >= 4
def fuzzy_logic_short_exit(self, dataframe: DataFrame, index: int) -> bool:
"""
Fuzzy logic for short exit - at least 80% of conditions (4 out of 5) must be true
"""
conditions = []
# Condition 1: SessionClose[2] crosses above CloseMonthly[2]
if index > 2:
conditions.append(
(
dataframe.iloc[index - 3]["session_close"]
< dataframe.iloc[index - 3]["close_monthly"]
)
and (
dataframe.iloc[index - 2]["session_close"]
> dataframe.iloc[index - 2]["close_monthly"]
)
)
else:
conditions.append(False)
# Condition 2: SMMA[2] is falling
if index > 2:
conditions.append(dataframe.iloc[index - 2]["smma"] < dataframe.iloc[index - 3]["smma"])
else:
conditions.append(False)
# Condition 3: BullsPower[4] is falling for 25 bars at 5 bar ago
if index > 30:
condition_met = True
for i in range(25):
bar_index = index - 5 - i
if (
dataframe.iloc[bar_index]["bulls_power"]
>= dataframe.iloc[bar_index - 1]["bulls_power"]
):
condition_met = False
break
conditions.append(condition_met)
else:
conditions.append(False)
# Condition 4: Close[3] crosses below Open[3]
if index > 3:
conditions.append(
dataframe.iloc[index - 3]["close"] < dataframe.iloc[index - 3]["open"]
)
else:
conditions.append(False)
# Condition 5: HeikenAshiClose(1h)[3] is rising for 5 bars at 3 bar ago
if index > 8:
condition_met = True
for i in range(5):
bar_index = index - 3 - i
if (
dataframe.iloc[bar_index]["ha_close_1h_1h"]
<= dataframe.iloc[bar_index - 1]["ha_close_1h_1h"]
):
condition_met = False
break
conditions.append(condition_met)
else:
conditions.append(False)
# Return True if at least 80% of conditions are met (4 out of 5)
return sum(conditions) >= 4
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, populates the entry signals
"""
dataframe.loc[:, "enter_long"] = 0
dataframe.loc[:, "enter_short"] = 0
# Apply fuzzy logic for each candle
for i in range(len(dataframe)):
if i < self.startup_candle_count:
continue
# Long entry
if self.fuzzy_logic_long_entry(dataframe, i):
dataframe.loc[i, "enter_long"] = 1
# Calculate limit order price
if i > 5:
limit_price = dataframe.iloc[i - 1]["kc_upper"] - (
self.price_entry_mult.value * dataframe.iloc[i - 5]["atr"]
)
dataframe.loc[i, "enter_long_limit"] = limit_price
# Short entry (only if not long entry)
if self.fuzzy_logic_short_entry(dataframe, i) and dataframe.loc[i, "enter_long"] == 0:
dataframe.loc[i, "enter_short"] = 1
# Calculate limit order price
if i > 5:
limit_price = dataframe.iloc[i - 1]["kc_lower"] + (
self.price_entry_mult.value * dataframe.iloc[i - 5]["atr"]
)
dataframe.loc[i, "enter_short_limit"] = limit_price
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, populates the exit signals
"""
dataframe.loc[:, "exit_long"] = 0
dataframe.loc[:, "exit_short"] = 0
# Apply fuzzy logic for each candle
for i in range(len(dataframe)):
if i < self.startup_candle_count:
continue
# Long exit
if self.fuzzy_logic_long_exit(dataframe, i) and dataframe.loc[i, "enter_long"] == 0:
dataframe.loc[i, "exit_long"] = 1
# Short exit
if self.fuzzy_logic_short_exit(dataframe, i) and dataframe.loc[i, "enter_short"] == 0:
dataframe.loc[i, "exit_short"] = 1
return dataframe
def custom_stoploss(
self,
pair: str,
trade: "Trade",
current_time: datetime,
current_rate: float,
current_profit: float,
**kwargs,
) -> float:
"""
Custom stoploss logic - using the stop_loss_pct parameter
"""
return -self.stop_loss_pct.value / 100
def custom_exit(
self,
pair: str,
trade: "Trade",
current_time: "datetime",
current_rate: float,
current_profit: float,
**kwargs,
):
"""
Custom exit logic for profit target and exit after bars
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
# Exit after N bars
trade_duration = (current_time - trade.open_date_utc).total_seconds() / 60
if trade_duration > (self.exit_after_bars.value * timeframe_to_minutes(self.timeframe)):
return "exit_after_bars"
# Profit target based on ATR
if len(dataframe) > 0:
current_atr = dataframe.iloc[-1]["atr_20"]
if current_atr > 0:
profit_target = (self.profit_target_coef.value * current_atr) / trade.open_rate
if current_profit >= profit_target:
return "profit_target_reached"
return None