OSIRIS DAY TRADE — 10 operações/dia em BTC/USDT, LONG + SHORT. Regime detection + structural entry patterns + structural stops.
Timeframe
5m
Direction
Long & Short
Stoploss
-4.0%
Trailing Stop
No
ROI
0m: 20.0%, 60m: 5.0%, 120m: 2.0%, 240m: 0.5%
Interface Version
3
Startup Candles
200
Indicators
8
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
"""
OSIRIS DAY TRADE STRATEGY v1.0
================================================================
DAY TRADE puro em BTC/USDT — LONG e SHORT.
MANIFESTO INVIOLÁVEL:
- Ativo: BTC/USDT APENAS
- Tipo: DAY TRADE (max 4h por operação)
- Meta: 10 operações/dia
- Direção: LONG e SHORT
- WR alvo: 80%+
- R:R: 2:1 mínimo
- Timeframe: 5m (execução) + 15m/1h (contexto)
ALGORITMO:
O mercado alterna entre RUÍDO e FLUXO. Este algoritmo detecta
a TRANSIÇÃO de um para o outro e entra na primeira confirmação
estrutural após a virada de momentum.
REGIME (filtro diário):
- SHORT: MACD cruzou pra baixo (última hora) + RSI>55 + BB>40%
- LONG: MACD cruzou pra cima (última hora) + RSI<45 + BB<60%
- NEUTRO: Nenhuma condição → NÃO OPERA
ENTRADA (padrão estrutural no 5m):
- Engulfing (bearish/bullish)
- Pin Bar / Shooting Star / Hammer
- Liquidity Grab (sweep de swing + reversão)
- Exhaustion Candle (range > 1.5×ATR + wick dominante)
STOPS:
- SL: Acima/abaixo do swing high/low estrutural + buffer
- TP: 2× risco (mínimo), ajustado por swing oposto
- Trailing: break-even em 1R, lock 0.5R em 1.5R, TP em 2R
COOLDOWN:
- Mínimo 3 candles (15 min) entre entradas
- Máximo 10 entradas/dia (hard cap)
- Se 3 losses consecutivos → pausa de 1h
100% proprietário. OSIRIS Day Trade — Disciplina É Lucro.
"""
import logging
import numpy as np
import pandas as pd
from pandas import DataFrame
from typing import Optional
from datetime import datetime, timedelta
from freqtrade.strategy import IStrategy, merge_informative_pair
from freqtrade.strategy import DecimalParameter, IntParameter
from freqtrade.persistence import Trade
import talib.abstract as ta
try:
from freqtrade.strategy import stoploss_from_open
except ImportError:
def stoploss_from_open(open_relative_stop, current_profit, is_short=False):
if current_profit == 0:
return 1
if is_short:
return -1 + ((1 - open_relative_stop) / (1 - current_profit))
return 1 - ((1 + open_relative_stop) / (1 + current_profit))
logger = logging.getLogger(__name__)
class OsirisDayTradeStrategy(IStrategy):
"""
OSIRIS DAY TRADE — 10 operações/dia em BTC/USDT, LONG + SHORT.
Regime detection + structural entry patterns + structural stops.
"""
INTERFACE_VERSION = 3
can_short = True
timeframe = "5m"
# ROI: wide — custom_exit handles R:R exits
minimal_roi = {
"0": 0.20,
"60": 0.05,
"120": 0.02,
"240": 0.005,
}
# Hard stoploss safety net
stoploss = -0.04
# Trailing — custom_stoploss handles R-based trailing
trailing_stop = False
use_custom_stoploss = True
startup_candle_count = 200
process_only_new_candles = True
# Trade counter per day (runtime state)
_daily_trades = {}
_consecutive_losses = 0
_last_loss_time = None
# ===================================================================
# HYPEROPT PARAMETERS
# ===================================================================
# Regime sensitivity
buy_rsi_short_min = IntParameter(52, 68, default=55, space="buy", optimize=True)
buy_rsi_long_max = IntParameter(32, 48, default=45, space="buy", optimize=True)
buy_macd_lookback = IntParameter(6, 36, default=12, space="buy", optimize=True)
# BB position filter
buy_bb_short_min = DecimalParameter(0.3, 0.7, default=0.40, decimals=2, space="buy", optimize=True)
buy_bb_long_max = DecimalParameter(0.3, 0.7, default=0.60, decimals=2, space="buy", optimize=True)
# Stop/TP
buy_sl_buffer = DecimalParameter(0.05, 0.30, default=0.15, decimals=2, space="buy", optimize=True)
buy_rr_target = DecimalParameter(1.5, 3.5, default=2.0, decimals=1, space="buy", optimize=True)
# Cooldown between entries (candles)
buy_cooldown = IntParameter(2, 10, default=3, space="buy", optimize=True)
# Max trades per day
buy_max_daily = IntParameter(5, 15, default=10, space="buy", optimize=True)
# Max hold time (candles of 5m)
buy_max_hold = IntParameter(24, 96, default=48, space="buy", optimize=True)
# ADX filter: don't counter-trend when ADX very strong
buy_adx_max_counter = IntParameter(25, 50, default=40, space="buy", optimize=True)
# Exit
sell_rsi_exit_long = IntParameter(68, 85, default=75, space="sell", optimize=True)
sell_rsi_exit_short = IntParameter(15, 32, default=25, space="sell", optimize=True)
# ===================================================================
# INFORMATIVE PAIRS — Multi-TF
# ===================================================================
def informative_pairs(self):
return [
("BTC/USDT:USDT", "15m"),
("BTC/USDT:USDT", "1h"),
]
# ===================================================================
# STANDARD INDICATORS
# ===================================================================
def _calc_standard(self, df: DataFrame) -> DataFrame:
df["rsi"] = ta.RSI(df, timeperiod=14)
df["adx"] = ta.ADX(df, timeperiod=14)
df["plus_di"] = ta.PLUS_DI(df, timeperiod=14)
df["minus_di"] = ta.MINUS_DI(df, timeperiod=14)
macd = ta.MACD(df, fastperiod=12, slowperiod=26, signalperiod=9)
df["macd"] = macd["macd"]
df["macd_signal"] = macd["macdsignal"]
df["macd_hist"] = macd["macdhist"]
df["ema_9"] = ta.EMA(df, timeperiod=9)
df["ema_21"] = ta.EMA(df, timeperiod=21)
df["ema_50"] = ta.EMA(df, timeperiod=50)
df["atr"] = ta.ATR(df, timeperiod=14)
df["atr_50"] = ta.ATR(df, timeperiod=50)
bb = ta.BBANDS(df, timeperiod=20, nbdevup=2.0, nbdevdn=2.0)
df["bb_upper"] = bb["upperband"]
df["bb_mid"] = bb["middleband"]
df["bb_lower"] = bb["lowerband"]
df["bb_width"] = (df["bb_upper"] - df["bb_lower"]) / df["bb_mid"]
df["bb_pos"] = (df["close"] - df["bb_lower"]) / (
(df["bb_upper"] - df["bb_lower"]).replace(0, 0.0001)
)
# Volume
df["vol_sma"] = ta.SMA(df["volume"], timeperiod=20)
df["vol_ratio"] = df["volume"] / df["vol_sma"].replace(0, 1)
# VWAP proxy
tp = (df["high"] + df["low"] + df["close"]) / 3
df["vwap"] = (tp * df["volume"]).rolling(50).sum() / df["volume"].rolling(50).sum().replace(0, 1)
return df
# ===================================================================
# SWING DETECTION
# ===================================================================
def _calc_swings(self, df: DataFrame) -> DataFrame:
h = df["high"].values
l = df["low"].values
n = len(df)
# 3-bar swing highs/lows
sh = np.zeros(n, dtype=bool)
sl = np.zeros(n, dtype=bool)
for i in range(3, n - 3):
if h[i] >= max(h[max(0, i-3):i]) and h[i] >= max(h[i+1:min(n, i+4)]):
sh[i] = True
if l[i] <= min(l[max(0, i-3):i]) and l[i] <= min(l[i+1:min(n, i+4)]):
sl[i] = True
# Track last confirmed swing high/low (confirmed = 3 bars ago)
last_sh = np.full(n, np.nan)
last_sl = np.full(n, np.nan)
last_sh_val = np.nan
last_sl_val = np.nan
for i in range(6, n):
if sh[i - 3]:
last_sh_val = h[i - 3]
if sl[i - 3]:
last_sl_val = l[i - 3]
last_sh[i] = last_sh_val
last_sl[i] = last_sl_val
# Second-to-last swing for wider context
prev_sh = np.full(n, np.nan)
prev_sl = np.full(n, np.nan)
sh_history = []
sl_history = []
for i in range(6, n):
if sh[i - 3]:
sh_history.append(h[i - 3])
if len(sh_history) > 5:
sh_history = sh_history[-5:]
if sl[i - 3]:
sl_history.append(l[i - 3])
if len(sl_history) > 5:
sl_history = sl_history[-5:]
if len(sh_history) >= 2:
prev_sh[i] = sh_history[-2]
if len(sl_history) >= 2:
prev_sl[i] = sl_history[-2]
df["swing_high"] = last_sh
df["swing_low"] = last_sl
df["prev_swing_high"] = prev_sh
df["prev_swing_low"] = prev_sl
# Recent highest high / lowest low (10 bars)
df["recent_hh"] = df["high"].rolling(10).max()
df["recent_ll"] = df["low"].rolling(10).min()
return df
# ===================================================================
# REGIME DETECTION
# ===================================================================
def _calc_regime(self, df: DataFrame) -> DataFrame:
macd_h = df["macd_hist"].values
rsi = df["rsi"].values
bb_pos = df["bb_pos"].values
n = len(df)
lookback = self.buy_macd_lookback.value
regime = np.zeros(n, dtype=int) # 0=neutral, 1=long, -1=short
for i in range(lookback + 1, n):
if np.isnan(macd_h[i]) or np.isnan(rsi[i]) or np.isnan(bb_pos[i]):
continue
# === SHORT REGIME ===
# MACD hist negativo agora E foi positivo recentemente
if macd_h[i] < 0:
had_pos = False
for k in range(1, lookback + 1):
if i - k >= 0 and not np.isnan(macd_h[i-k]) and macd_h[i-k] > 0:
had_pos = True
break
if had_pos and rsi[i] > self.buy_rsi_short_min.value and bb_pos[i] > self.buy_bb_short_min.value:
regime[i] = -1
# === LONG REGIME ===
# MACD hist positivo agora E foi negativo recentemente
if macd_h[i] > 0:
had_neg = False
for k in range(1, lookback + 1):
if i - k >= 0 and not np.isnan(macd_h[i-k]) and macd_h[i-k] < 0:
had_neg = True
break
if had_neg and rsi[i] < self.buy_rsi_long_max.value and bb_pos[i] < self.buy_bb_long_max.value:
regime[i] = 1
df["regime"] = regime
return df
# ===================================================================
# PATTERN DETECTION — Structural Entries
# ===================================================================
def _detect_patterns(self, df: DataFrame) -> DataFrame:
c = df["close"].values
o = df["open"].values
h = df["high"].values
l = df["low"].values
atr = df["atr"].values
n = len(df)
# Pattern arrays: 1=bullish, -1=bearish, 0=none
engulf = np.zeros(n, dtype=int)
pin = np.zeros(n, dtype=int)
liq_grab = np.zeros(n, dtype=int)
exhaust = np.zeros(n, dtype=int)
sh = df["swing_high"].values
sl_arr = df["swing_low"].values
for i in range(1, n):
if np.isnan(atr[i]) or atr[i] < 0.5:
continue
body = abs(c[i] - o[i])
upper_w = h[i] - max(c[i], o[i])
lower_w = min(c[i], o[i]) - l[i]
candle_range = h[i] - l[i]
is_bearish = c[i] < o[i]
is_bullish = c[i] > o[i]
prev_body = abs(c[i-1] - o[i-1])
prev_bullish = c[i-1] > o[i-1]
prev_bearish = c[i-1] < o[i-1]
# === BEARISH PATTERNS ===
if is_bearish:
# Bearish Engulfing
if prev_bullish and o[i] >= c[i-1] and c[i] <= o[i-1]:
engulf[i] = -1
# Bearish Liquidity Grab (sweep swing high + close below)
if not np.isnan(sh[i]) and h[i] > sh[i] and c[i] < sh[i]:
liq_grab[i] = -1
# Bearish Pin Bar (long upper wick)
if upper_w > 2 * max(body, 0.01) and upper_w > 2 * lower_w and candle_range > 0.3 * atr[i]:
pin[i] = -1
# Bearish Exhaustion
if candle_range > 1.5 * atr[i] and upper_w > 0.4 * candle_range:
exhaust[i] = -1
# === BULLISH PATTERNS ===
if is_bullish:
# Bullish Engulfing
if prev_bearish and o[i] <= c[i-1] and c[i] >= o[i-1]:
engulf[i] = 1
# Bullish Liquidity Grab (sweep swing low + close above)
if not np.isnan(sl_arr[i]) and l[i] < sl_arr[i] and c[i] > sl_arr[i]:
liq_grab[i] = 1
# Bullish Pin Bar / Hammer (long lower wick)
if lower_w > 2 * max(body, 0.01) and lower_w > 2 * upper_w and candle_range > 0.3 * atr[i]:
pin[i] = 1
# Bullish Exhaustion
if candle_range > 1.5 * atr[i] and lower_w > 0.4 * candle_range:
exhaust[i] = 1
# Shooting star (can close slightly green — bearish reversal)
if not is_bearish and upper_w > 2.5 * max(body, 0.01) and upper_w > 3 * lower_w:
pin[i] = -1
# Inverted hammer (can close slightly red — bullish reversal)
if not is_bullish and lower_w > 2.5 * max(body, 0.01) and lower_w > 3 * upper_w:
pin[i] = 1
df["pat_engulf"] = engulf
df["pat_pin"] = pin
df["pat_liq_grab"] = liq_grab
df["pat_exhaust"] = exhaust
# Any pattern (direction-specific)
df["pat_bearish"] = ((engulf == -1) | (pin == -1) | (liq_grab == -1) | (exhaust == -1)).astype(int)
df["pat_bullish"] = ((engulf == 1) | (pin == 1) | (liq_grab == 1) | (exhaust == 1)).astype(int)
return df
# ===================================================================
# STRUCTURAL STOP LEVELS
# ===================================================================
def _calc_structural_stops(self, df: DataFrame) -> DataFrame:
"""Calculate stop-loss levels based on market structure."""
c = df["close"].values
h = df["high"].values
l = df["low"].values
atr = df["atr"].values
sh = df["recent_hh"].values
sl_arr = df["recent_ll"].values
buf = self.buy_sl_buffer.value
n = len(df)
# Short SL = above recent swing high + buffer
sl_short = np.full(n, np.nan)
# Long SL = below recent swing low - buffer
sl_long = np.full(n, np.nan)
for i in range(10, n):
if np.isnan(atr[i]) or atr[i] < 0.5:
continue
# Short: stop above recent highs
ref_high = sh[i] if not np.isnan(sh[i]) else h[i]
sl_short[i] = ref_high + buf * atr[i]
# Long: stop below recent lows
ref_low = sl_arr[i] if not np.isnan(sl_arr[i]) else l[i]
sl_long[i] = ref_low - buf * atr[i]
df["sl_short"] = sl_short
df["sl_long"] = sl_long
# Risk in percentage terms
df["risk_short_pct"] = (df["sl_short"] - df["close"]) / df["close"]
df["risk_long_pct"] = (df["close"] - df["sl_long"]) / df["close"]
return df
# ===================================================================
# WHALE ABSORPTION FINGERPRINT (from Genesis)
# ===================================================================
def _calc_whale(self, df: DataFrame) -> DataFrame:
vol_z = (
(df["volume"] - df["volume"].rolling(50).mean())
/ df["volume"].rolling(50).std().replace(0, 1)
)
abs_ret = df["close"].pct_change().abs()
price_z = (
(abs_ret - abs_ret.rolling(50).mean())
/ abs_ret.rolling(50).std().replace(0, 0.0001)
)
df["waf"] = np.maximum(0, vol_z - price_z)
close_pos = (df["close"] - df["low"]) / (
(df["high"] - df["low"]).replace(0, 0.0001)
)
df["waf_dir"] = np.where(df["waf"] > 1, 2 * close_pos - 1, 0)
return df
# ===================================================================
# ORDER FLOW ORACLE (from Genesis)
# ===================================================================
def _calc_flow(self, df: DataFrame) -> DataFrame:
hl_range = (df["high"] - df["low"]).replace(0, 0.0001)
close_pos = (df["close"] - df["low"]) / hl_range
imbalance = 2 * close_pos - 1
acl1 = imbalance.rolling(30).corr(imbalance.shift(1)).fillna(0)
acl2 = imbalance.rolling(30).corr(imbalance.shift(2)).fillna(0)
acl3 = imbalance.rolling(30).corr(imbalance.shift(3)).fillna(0)
df["flow_regularity"] = (acl1.abs() + acl2.abs() + acl3.abs()) / 3
df["flow_dir"] = imbalance.rolling(20).mean()
return df
# ===================================================================
# POPULATE INDICATORS
# ===================================================================
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# Standard TA
dataframe = self._calc_standard(dataframe)
# Market structure (swings)
dataframe = self._calc_swings(dataframe)
# Regime detection (MACD crossover + RSI + BB position)
dataframe = self._calc_regime(dataframe)
# Structural entry patterns
dataframe = self._detect_patterns(dataframe)
# Structural stop levels
dataframe = self._calc_structural_stops(dataframe)
# Whale absorption + Order flow (from Genesis)
dataframe = self._calc_whale(dataframe)
dataframe = self._calc_flow(dataframe)
# Multi-timeframe
if self.dp:
pair = metadata["pair"]
inf_15m = self.dp.get_pair_dataframe(pair=pair, timeframe="15m")
if not inf_15m.empty:
inf_15m["rsi"] = ta.RSI(inf_15m, timeperiod=14)
inf_15m["ema_21"] = ta.EMA(inf_15m, timeperiod=21)
inf_15m["ema_50"] = ta.EMA(inf_15m, timeperiod=50)
macd15 = ta.MACD(inf_15m, fastperiod=12, slowperiod=26, signalperiod=9)
inf_15m["macd_hist"] = macd15["macdhist"]
dataframe = merge_informative_pair(
dataframe, inf_15m, self.timeframe, "15m", ffill=True
)
inf_1h = self.dp.get_pair_dataframe(pair=pair, timeframe="1h")
if not inf_1h.empty:
inf_1h["rsi"] = ta.RSI(inf_1h, timeperiod=14)
inf_1h["ema_21"] = ta.EMA(inf_1h, timeperiod=21)
inf_1h["ema_50"] = ta.EMA(inf_1h, timeperiod=50)
inf_1h["adx"] = ta.ADX(inf_1h, timeperiod=14)
dataframe = merge_informative_pair(
dataframe, inf_1h, self.timeframe, "1h", ffill=True
)
return dataframe
# ===================================================================
# ENTRY LOGIC — LONG
# ===================================================================
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# === LONG CONDITIONS ===
long_regime = dataframe["regime"] == 1
# Pattern confirmation
long_pattern = dataframe["pat_bullish"] == 1
# Volume confirmation (above average)
vol_ok = dataframe["vol_ratio"] > 0.8
# Risk/Reward check: risk must be reasonable
risk_ok_long = (dataframe["risk_long_pct"] > 0.001) & (dataframe["risk_long_pct"] < 0.03)
# HTF alignment (15m)
htf_long = pd.Series(True, index=dataframe.index)
if "rsi_15m" in dataframe.columns:
htf_long = dataframe["rsi_15m"] < 60 # Not overbought on 15m
if "macd_hist_15m" in dataframe.columns:
htf_long = htf_long & (dataframe["macd_hist_15m"] > -50) # 15m not deeply bearish
# Don't go long against strong downtrend
adx_filter_long = pd.Series(True, index=dataframe.index)
if "adx_1h" in dataframe.columns:
adx_filter_long = ~(
(dataframe.get("adx_1h", pd.Series(0, index=dataframe.index)) > self.buy_adx_max_counter.value)
& (dataframe.get("ema_21_1h", dataframe["close"]) < dataframe.get("ema_50_1h", dataframe["close"]))
)
# Whale/flow bonus (relaxes conditions)
whale_long = (dataframe["waf"] > 1) & (dataframe["waf_dir"] > 0.2)
flow_long = (dataframe["flow_regularity"] > 0.3) & (dataframe["flow_dir"] > 0.1)
smart_money_long = whale_long | flow_long
# === LONG ENTRY ===
long_strict = long_regime & long_pattern & vol_ok & risk_ok_long & htf_long & adx_filter_long
long_bonus = long_regime & long_pattern & smart_money_long & risk_ok_long
dataframe.loc[
(long_strict | long_bonus) & (dataframe["volume"] > 0),
"enter_long",
] = 1
# === SHORT CONDITIONS ===
short_regime = dataframe["regime"] == -1
# Pattern confirmation
short_pattern = dataframe["pat_bearish"] == 1
# Risk/Reward check
risk_ok_short = (dataframe["risk_short_pct"] > 0.001) & (dataframe["risk_short_pct"] < 0.03)
# HTF alignment (15m)
htf_short = pd.Series(True, index=dataframe.index)
if "rsi_15m" in dataframe.columns:
htf_short = dataframe["rsi_15m"] > 40 # Not oversold on 15m
if "macd_hist_15m" in dataframe.columns:
htf_short = htf_short & (dataframe["macd_hist_15m"] < 50) # 15m not deeply bullish
# Don't go short against strong uptrend
adx_filter_short = pd.Series(True, index=dataframe.index)
if "adx_1h" in dataframe.columns:
adx_filter_short = ~(
(dataframe.get("adx_1h", pd.Series(0, index=dataframe.index)) > self.buy_adx_max_counter.value)
& (dataframe.get("ema_21_1h", dataframe["close"]) > dataframe.get("ema_50_1h", dataframe["close"]))
)
# Whale/flow bonus
whale_short = (dataframe["waf"] > 1) & (dataframe["waf_dir"] < -0.2)
flow_short = (dataframe["flow_regularity"] > 0.3) & (dataframe["flow_dir"] < -0.1)
smart_money_short = whale_short | flow_short
# === SHORT ENTRY ===
short_strict = short_regime & short_pattern & vol_ok & risk_ok_short & htf_short & adx_filter_short
short_bonus = short_regime & short_pattern & smart_money_short & risk_ok_short
dataframe.loc[
(short_strict | short_bonus) & (dataframe["volume"] > 0),
"enter_short",
] = 1
return dataframe
# ===================================================================
# EXIT SIGNALS
# ===================================================================
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# Long exit: RSI overbought OR regime flipped to short
dataframe.loc[
(dataframe["rsi"] > self.sell_rsi_exit_long.value)
| (dataframe["regime"] == -1),
"exit_long",
] = 1
# Short exit: RSI oversold OR regime flipped to long
dataframe.loc[
(dataframe["rsi"] < self.sell_rsi_exit_short.value)
| (dataframe["regime"] == 1),
"exit_short",
] = 1
return dataframe
# ===================================================================
# CUSTOM STOPLOSS — Structural + R-Multiple Trailing
# ===================================================================
def custom_stoploss(
self,
pair: str,
trade: Trade,
current_time,
current_rate: float,
current_profit: float,
**kwargs,
) -> float:
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) == 0:
return -0.03
last = dataframe.iloc[-1]
atr = last.get("atr", 0)
if atr == 0 or trade.open_rate == 0:
return -0.03
is_short = trade.is_short if hasattr(trade, 'is_short') else False
if is_short:
# Short: SL above entry
sl = last.get("sl_short", 0)
if sl > 0:
stop_dist_pct = (sl - trade.open_rate) / trade.open_rate
else:
stop_dist_pct = 1.5 * atr / trade.open_rate
else:
# Long: SL below entry
sl = last.get("sl_long", 0)
if sl > 0:
stop_dist_pct = (trade.open_rate - sl) / trade.open_rate
else:
stop_dist_pct = 1.5 * atr / trade.open_rate
# Progressive R-multiple trailing
if stop_dist_pct > 0 and current_profit > 0:
r_mult = current_profit / stop_dist_pct
if r_mult >= 2.0:
# Lock 1R profit — approaching TP
return stoploss_from_open(
1.0 * stop_dist_pct, current_profit, is_short=is_short
)
elif r_mult >= 1.5:
# Lock 0.5R profit
return stoploss_from_open(
0.5 * stop_dist_pct, current_profit, is_short=is_short
)
elif r_mult >= 1.0:
# Break-even
return stoploss_from_open(
0.001, current_profit, is_short=is_short
)
return max(-stop_dist_pct, -0.04)
# ===================================================================
# CUSTOM EXIT — R:R Target + Time Management + Daily Limits
# ===================================================================
def custom_exit(
self,
pair: str,
trade: Trade,
current_time,
current_rate: float,
current_profit: float,
**kwargs,
) -> Optional[str]:
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) == 0:
return None
last = dataframe.iloc[-1]
atr = last.get("atr", 0)
if atr == 0 or trade.open_rate == 0:
return None
is_short = trade.is_short if hasattr(trade, 'is_short') else False
# Calculate risk distance
if is_short:
sl = last.get("sl_short", 0)
stop_dist = (sl - trade.open_rate) if sl > 0 else 1.5 * atr
else:
sl = last.get("sl_long", 0)
stop_dist = (trade.open_rate - sl) if sl > 0 else 1.5 * atr
stop_dist_pct = stop_dist / trade.open_rate
target_pct = stop_dist_pct * self.buy_rr_target.value
# R:R TARGET HIT — TAKE PROFIT
if current_profit >= target_pct:
return "dt_tp_target"
# Partial at 60% of target if momentum fading
if current_profit >= target_pct * 0.6:
rsi = last.get("rsi", 50)
macd_hist = last.get("macd_hist", 0)
if is_short:
if rsi < 30 or macd_hist > 0:
return "dt_tp_momentum_fade"
else:
if rsi > 70 or macd_hist < 0:
return "dt_tp_momentum_fade"
# Time management — Day Trade: max 4 hours
hours = (current_time - trade.open_date_utc).total_seconds() / 3600
max_candles_h = self.buy_max_hold.value * 5 / 60 # convert candles to hours
if hours > max_candles_h and current_profit > 0.001:
return "dt_time_profit"
# Hard time limit
if hours > max_candles_h * 1.5:
return "dt_time_force"
return None
# ===================================================================
# CONFIRM ENTRY — Daily trade limit + cooldown + loss streak
# ===================================================================
def confirm_trade_entry(
self,
pair: str,
order_type: str,
amount: float,
rate: float,
time_in_force: str,
current_time: datetime,
entry_tag: Optional[str],
side: str,
**kwargs,
) -> bool:
# Daily trade limit
today = current_time.strftime("%Y-%m-%d")
if today not in self._daily_trades:
self._daily_trades = {today: 0} # Reset + clear old days
if self._daily_trades[today] >= self.buy_max_daily.value:
logger.info(f"DAY TRADE LIMIT: {self._daily_trades[today]} trades today, max={self.buy_max_daily.value}")
return False
# Loss streak pause: 3 consecutive losses → wait 1 hour
if self._consecutive_losses >= 3 and self._last_loss_time:
pause_until = self._last_loss_time + timedelta(hours=1)
if current_time < pause_until:
logger.info(f"LOSS STREAK PAUSE: {self._consecutive_losses} losses, paused until {pause_until}")
return False
else:
self._consecutive_losses = 0
self._daily_trades[today] = self._daily_trades.get(today, 0) + 1
return True
# ===================================================================
# CONFIRM EXIT — Track consecutive losses
# ===================================================================
def confirm_trade_exit(
self,
pair: str,
trade: Trade,
order_type: str,
amount: float,
rate: float,
time_in_force: str,
exit_reason: str,
current_time: datetime,
**kwargs,
) -> bool:
# Track win/loss for streak management
if trade.calc_profit_ratio(rate) < 0:
self._consecutive_losses += 1
self._last_loss_time = current_time
else:
self._consecutive_losses = 0
return True