Timeframe
1h
Direction
Long & Short
Stoploss
-99.0%
Trailing Stop
No
ROI
0m: 10000.0%
Interface Version
3
Startup Candles
4
Indicators
0
freqtrade/freqtrade-strategies
freqtrade/freqtrade-strategies
freqtrade/freqtrade-strategies
this is an example class, implementing a PSAR based trailing stop loss you are supposed to take the `custom_stoploss()` and `populate_indicators()` parts and adapt it to your own strategy
freqtrade/freqtrade-strategies
from __future__ import annotations
from datetime import datetime
from typing import Optional
import pandas as pd
from freqtrade.exchange import timeframe_to_prev_date
from freqtrade.persistence import Order, Trade
from freqtrade.strategy import IStrategy, stoploss_from_absolute
from pandas import DataFrame
class FVG:
def __init__(self, top: float, bottom: float, is_bullish: bool, bar_index: int) -> None:
self.top = top
self.bottom = bottom
self.is_bullish = is_bullish
self.bar_index = bar_index
class SMC_FVG_Confirmation_Freqtrade(IStrategy):
INTERFACE_VERSION = 3
can_short = True
timeframe = "1h"
startup_candle_count = 4
process_only_new_candles = True
minimal_roi = {"0": 100.0}
stoploss = -0.99
use_custom_stoploss = True
use_custom_roi = True
use_exit_signal = False
exit_profit_only = False
ignore_roi_if_entry_signal = False
PIN_BAR_BODY_RATIO = 0.33
PIN_BAR_WICK_TO_BODY = 2.5
PIN_BAR_CLOSE_EXTREME_RATIO = 0.30
FVG_RETRACE_RATIO = 0.45
FVG_CONFIRM_RATIO = 0.55
order_types = {
"entry": "market",
"exit": "market",
"stoploss": "market",
"stoploss_on_exchange": False,
}
order_time_in_force = {"entry": "GTC", "exit": "GTC"}
@staticmethod
def _signal_tag(signal_kind: str, stop: float) -> str:
return f"{signal_kind}|{stop:.10f}"
@staticmethod
def _parse_enter_tag(tag: str | None) -> tuple[Optional[str], Optional[float]]:
if not tag:
return None, None
parts = tag.split("|", 1)
if len(parts) != 2:
return tag, None
try:
return parts[0], float(parts[1])
except ValueError:
return parts[0], None
def _is_pin_bar(self, row: pd.Series, is_bullish: bool) -> bool:
open_price = float(row["open"])
close_price = float(row["close"])
high_price = float(row["high"])
low_price = float(row["low"])
body = abs(close_price - open_price)
upper_wick = high_price - max(close_price, open_price)
lower_wick = min(close_price, open_price) - low_price
total_range = high_price - low_price
if total_range == 0:
return False
if is_bullish:
close_near_high = close_price >= high_price - total_range * self.PIN_BAR_CLOSE_EXTREME_RATIO
body_in_upper_range = min(open_price, close_price) >= low_price + total_range * (
1 - self.PIN_BAR_BODY_RATIO
)
return (
close_price >= open_price
and lower_wick >= self.PIN_BAR_WICK_TO_BODY * body
and upper_wick <= body
and body <= total_range * self.PIN_BAR_BODY_RATIO
and close_near_high
and body_in_upper_range
)
close_near_low = close_price <= low_price + total_range * self.PIN_BAR_CLOSE_EXTREME_RATIO
body_in_lower_range = max(open_price, close_price) <= high_price - total_range * (
1 - self.PIN_BAR_BODY_RATIO
)
return (
close_price <= open_price
and upper_wick >= self.PIN_BAR_WICK_TO_BODY * body
and lower_wick <= body
and body <= total_range * self.PIN_BAR_BODY_RATIO
and close_near_low
and body_in_lower_range
)
@staticmethod
def _is_trend_body(row: pd.Series, is_bullish: bool) -> bool:
open_price = float(row["open"])
close_price = float(row["close"])
high_price = float(row["high"])
low_price = float(row["low"])
body = abs(close_price - open_price)
upper_wick = high_price - max(close_price, open_price)
lower_wick = min(close_price, open_price) - low_price
total_range = high_price - low_price
if total_range == 0:
return False
body_ratio = body / total_range
if is_bullish:
return (
close_price > open_price
and body_ratio >= 0.55
and close_price >= high_price - total_range * 0.15
and upper_wick <= total_range * 0.15
and lower_wick <= total_range * 0.2
)
return (
close_price < open_price
and body_ratio >= 0.55
and close_price <= low_price + total_range * 0.15
and lower_wick <= total_range * 0.15
and upper_wick <= total_range * 0.2
)
@staticmethod
def _is_displacement_break(row: pd.Series, prev_row: pd.Series, is_bullish: bool) -> bool:
open_price = float(row["open"])
close_price = float(row["close"])
high_price = float(row["high"])
low_price = float(row["low"])
prev_high = float(prev_row["high"])
prev_low = float(prev_row["low"])
body = abs(close_price - open_price)
total_range = high_price - low_price
if total_range == 0:
return False
body_ratio = body / total_range
if is_bullish:
return (
close_price > open_price
and body_ratio >= 0.55
and close_price > prev_high
and close_price >= high_price - total_range * 0.25
)
return (
close_price < open_price
and body_ratio >= 0.55
and close_price < prev_low
and close_price <= low_price + total_range * 0.25
)
def _entry_signal_kind(self, row: pd.Series, prev_row: pd.Series, is_bullish: bool) -> Optional[str]:
if self._is_displacement_break(row, prev_row, is_bullish):
return "displacement"
if self._is_trend_body(row, is_bullish):
return "trend_body"
if self._is_pin_bar(row, is_bullish):
return "pin_bar"
return None
@staticmethod
def _signal_matches_fvg(row: pd.Series, fvg: FVG, signal_kind: str, is_bullish: bool) -> bool:
high_price = float(row["high"])
low_price = float(row["low"])
close_price = float(row["close"])
if is_bullish != fvg.is_bullish:
return False
overlaps_fvg = high_price >= fvg.bottom and low_price <= fvg.top
if not overlaps_fvg:
return False
if signal_kind not in {"trend_body", "displacement"}:
return True
fvg_height = fvg.top - fvg.bottom
if fvg_height <= 0:
return False
if is_bullish:
return (
low_price <= fvg.bottom + fvg_height * SMC_FVG_Confirmation_Freqtrade.FVG_RETRACE_RATIO
and close_price >= fvg.bottom + fvg_height * SMC_FVG_Confirmation_Freqtrade.FVG_CONFIRM_RATIO
)
return (
high_price >= fvg.top - fvg_height * SMC_FVG_Confirmation_Freqtrade.FVG_RETRACE_RATIO
and close_price <= fvg.bottom + fvg_height * SMC_FVG_Confirmation_Freqtrade.FVG_RETRACE_RATIO
)
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
dataframe = dataframe.copy()
dataframe["ft_long_signal"] = 0
dataframe["ft_short_signal"] = 0
dataframe["ft_long_stop"] = pd.NA
dataframe["ft_short_stop"] = pd.NA
dataframe["ft_long_tag"] = ""
dataframe["ft_short_tag"] = ""
active_bullish: list[FVG] = []
active_bearish: list[FVG] = []
rows = dataframe.reset_index(drop=True)
for i in range(len(rows)):
row = rows.iloc[i]
current_fvg: Optional[FVG] = None
if i >= 3:
candle_3 = rows.iloc[i - 3]
candle_1 = rows.iloc[i - 1]
high_3 = float(candle_3["high"])
low_3 = float(candle_3["low"])
high_1 = float(candle_1["high"])
low_1 = float(candle_1["low"])
if high_3 < low_1:
current_fvg = FVG(top=low_1, bottom=high_3, is_bullish=True, bar_index=i)
elif low_3 > high_1:
current_fvg = FVG(top=low_3, bottom=high_1, is_bullish=False, bar_index=i)
if current_fvg is not None:
if current_fvg.is_bullish:
active_bullish.append(current_fvg)
else:
active_bearish.append(current_fvg)
current_low = float(row["low"])
current_high = float(row["high"])
active_bullish = [fvg for fvg in active_bullish if current_low > fvg.bottom]
active_bearish = [fvg for fvg in active_bearish if current_high < fvg.top]
if i == 0:
continue
prev_row = rows.iloc[i - 1]
long_kind = self._entry_signal_kind(row, prev_row, is_bullish=True)
if long_kind is not None:
for fvg in reversed(active_bullish):
if self._signal_matches_fvg(row, fvg, long_kind, is_bullish=True):
rows.at[i, "ft_long_signal"] = 1
rows.at[i, "ft_long_stop"] = fvg.bottom
rows.at[i, "ft_long_tag"] = self._signal_tag(long_kind, fvg.bottom)
break
short_kind = self._entry_signal_kind(row, prev_row, is_bullish=False)
if short_kind is not None:
for fvg in reversed(active_bearish):
if self._signal_matches_fvg(row, fvg, short_kind, is_bullish=False):
rows.at[i, "ft_short_signal"] = 1
rows.at[i, "ft_short_stop"] = fvg.top
rows.at[i, "ft_short_tag"] = self._signal_tag(short_kind, fvg.top)
break
rows["ft_long_entry_signal"] = rows["ft_long_signal"].astype(int)
rows["ft_short_entry_signal"] = rows["ft_short_signal"].astype(int)
rows["ft_long_entry_stop"] = rows["ft_long_stop"]
rows["ft_short_entry_stop"] = rows["ft_short_stop"]
rows["ft_long_entry_tag"] = rows["ft_long_tag"]
rows["ft_short_entry_tag"] = rows["ft_short_tag"]
return rows
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
dataframe.loc[
(dataframe["ft_long_entry_signal"] == 1) & (dataframe["volume"] > 0),
"enter_long",
] = 1
dataframe.loc[
(dataframe["ft_short_entry_signal"] == 1) & (dataframe["volume"] > 0),
"enter_short",
] = 1
dataframe["enter_tag"] = ""
dataframe.loc[dataframe["ft_long_entry_signal"] == 1, "enter_tag"] = dataframe["ft_long_entry_tag"]
dataframe.loc[dataframe["ft_short_entry_signal"] == 1, "enter_tag"] = dataframe["ft_short_entry_tag"]
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
dataframe["exit_long"] = 0
dataframe["exit_short"] = 0
return dataframe
def leverage(
self,
pair: str,
current_time: datetime,
current_rate: float,
proposed_leverage: float,
max_leverage: float,
entry_tag: str | None,
side: str,
**kwargs,
) -> float:
configured_leverage = self.config.get("smc_leverage")
if configured_leverage == "max":
return max_leverage
if configured_leverage is None:
return min(proposed_leverage, max_leverage)
return min(float(configured_leverage), max_leverage)
def custom_stake_amount(
self,
pair: str,
current_time: datetime,
current_rate: float,
proposed_stake: float,
min_stake: float | None,
max_stake: float,
leverage: float,
entry_tag: str | None,
side: str,
**kwargs,
) -> float:
dataframe, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe)
current_candle = dataframe.iloc[-1].squeeze()
stop_rate = current_candle["ft_long_entry_stop"] if side == "long" else current_candle["ft_short_entry_stop"]
if pd.isna(stop_rate):
return 0
stop_rate = float(stop_rate)
distance_ratio = abs(current_rate - stop_rate) / current_rate
if distance_ratio <= 0:
return 0
risk_per_trade = self.config.get("smc_risk_per_trade")
capital_cap_ratio = self.config.get("smc_capital_cap")
if risk_per_trade is None or capital_cap_ratio is None:
return min(proposed_stake, max_stake)
if leverage <= 0:
return 0
total_stake = self.wallets.get_total_stake_amount()
risk_stake = (total_stake * float(risk_per_trade)) / (distance_ratio * leverage)
capital_cap = total_stake * float(capital_cap_ratio)
return min(risk_stake, capital_cap, max_stake)
def order_filled(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> None:
if order.ft_order_side != trade.entry_side or trade.nr_of_successful_entries != 1:
return
signal_kind, stop_rate = self._parse_enter_tag(trade.enter_tag)
if stop_rate is None:
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
trade_date = timeframe_to_prev_date(self.timeframe, trade.open_date_utc)
trade_candle = dataframe.loc[dataframe["date"] == trade_date]
if trade_candle.empty:
return
candle = trade_candle.squeeze()
stop_rate = candle["ft_long_entry_stop"] if trade.is_short is False else candle["ft_short_entry_stop"]
signal_kind = signal_kind or "unknown"
if stop_rate is None or pd.isna(stop_rate):
return
stop_rate = float(stop_rate)
risk_ratio = abs(trade.open_rate - stop_rate) / trade.open_rate
target_roi = risk_ratio * trade.leverage
trade.set_custom_data("smc_signal_kind", signal_kind or "unknown")
trade.set_custom_data("smc_stop_rate", stop_rate)
trade.set_custom_data("smc_target_roi", target_roi)
def custom_stoploss(
self,
pair: str,
trade: Trade,
current_time: datetime,
current_rate: float,
current_profit: float,
after_fill: bool,
**kwargs,
) -> float | None:
stop_rate = trade.get_custom_data("smc_stop_rate")
if stop_rate is None:
_, stop_rate = self._parse_enter_tag(trade.enter_tag)
if stop_rate is None:
return None
return stoploss_from_absolute(
float(stop_rate),
current_rate=current_rate,
is_short=trade.is_short,
leverage=trade.leverage,
)
def custom_roi(
self,
pair: str,
trade: Trade,
current_time: datetime,
trade_duration: int,
entry_tag: str | None,
side: str,
**kwargs,
) -> float | None:
target_roi = trade.get_custom_data("smc_target_roi")
if target_roi is not None:
return float(target_roi)
_, stop_rate = self._parse_enter_tag(entry_tag)
if stop_rate is None:
return None
return (abs(trade.open_rate - float(stop_rate)) / trade.open_rate) * trade.leverage