NY-session Range Breakout / Re-entry Strategy.
Timeframe
15m
Direction
Long & Short
Stoploss
-10.0%
Trailing Stop
No
ROI
N/A
Interface Version
N/A
Startup Candles
300
Indicators
3
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
import logging
import numpy as np
import pandas as pd
from datetime import datetime
from pandas import DataFrame
from freqtrade.strategy import IStrategy, stoploss_from_absolute, CategoricalParameter
from freqtrade.persistence import Trade
logger = logging.getLogger(__name__)
NY_TZ = 'America/New_York'
RANGE_START_HOUR = 0 # 00:00 NY — anchor for VWAP and range
RANGE_END_HOUR = 4 # 04:00 NY — range window closes, trading begins
RR_RATIO = 2.0 # Risk:Reward = 1:2
TIME_STOP_MINS = 180 # Exit any position older than this
class SekkaFourExp(IStrategy):
"""
NY-session Range Breakout / Re-entry Strategy.
Range definition
----------------
Build HP (high) and LP (low) from the first 4 hours of the NY day
(00:00–04:00 NY), using candle wicks.
Signal logic
------------
After 04:00 NY:
SHORT — a 5m candle closes above HP → arm short state, track the
highest HIGH seen during the breakout. When a later candle
closes back below HP (freakout), a short signal fires.
Entry is a limit order placed at exactly HP.
SL = highest HIGH during the breakout (or fractal fallback
when breakout exceeds range — see below).
TP = entry − RR_RATIO × (SL − entry).
LONG — mirror of the above using LP.
Filters applied at signal time
------------------------------
• Large-breakout fractal fallback: if the breakout magnitude exceeds
the range, the SL is swapped from running_max_high/running_min_low
to the nearest confirmed 5-candle fractal high/low close inside the
breakout; signal is skipped if no qualifying fractal exists.
• VWAP filter: the limit entry price (HP for shorts, LP for longs) must
be on the correct side of the live day-anchored VWAP at signal time.
Uses the freakout candle's VWAP — same value seen in live trading.
– Short: HP > VWAP (price elevated vs volume-weighted fair value)
– Long : LP < VWAP (price depressed vs volume-weighted fair value)
• Fakeout filter: the candle immediately after the freakout must open
inside the range (open < HP for shorts, open > LP for longs).
• Entry retry (hyperopt): when `entry_retry_enabled = 1`, a freakout
that gets rejected by the fakeout/VWAP filters preserves breakout
state so a later candle can retry against the same level. When `0`,
state is reset on every freakout.
Exit logic
----------
1. Time stop — if the trade is older than TIME_STOP_MINS, exit at market.
2. TP — exit when price reaches the pre-computed TP level.
3. SL — handled via custom_stoploss (pinned to an absolute price).
Multiple trades per day are allowed once a prior position is closed.
Setups are discarded at NY midnight; open positions may run past midnight.
"""
timeframe = '15m'
can_short = True
minimal_roi = {} # TP handled entirely in custom_exit
stoploss = -0.10 # Hard backstop if custom_stoploss fails
trailing_stop = False
use_custom_stoploss = True
use_exit_signal = True
exit_profit_only = False
process_only_new_candles = True
startup_candle_count = 300
# Limit entry at the range boundary (HP for shorts, LP for longs).
# The signal fires on the freakout candle; the limit sits at HP/LP so
# it fills on the very next candle when price re-enters the range.
order_types = {
'entry': 'limit',
'exit': 'market',
'stoploss': 'market',
'stoploss_on_exchange': False,
}
# Entry limits expire quickly; check_entry_timeout enforces NY-midnight
# cancellation as a hard upper bound.
unfilledtimeout = {
'entry': 15, # minutes — one 15m candle to fill on N+1
'exit': 30,
}
# ---- Hyperopt parameters ----
# 1 = preserve breakout state when fakeout/VWAP filters reject the
# signal, allowing a later candle to retry against the same level.
# 0 = always reset state after any freakout.
entry_retry_enabled = CategoricalParameter(
[0, 1], default=0, space='buy', optimize=True
)
# NEW FILTERS FOR TESTING
filter_min_sl = CategoricalParameter([0, 1], default=1, space='buy', optimize=True)
filter_htf_trend = CategoricalParameter([0, 1], default=1, space='buy', optimize=True)
filter_vol_exhaust = CategoricalParameter([0, 1], default=0, space='buy', optimize=True)
filter_strong_fakeout = CategoricalParameter([0, 1], default=0, space='buy', optimize=True)
# ------------------------------------------------------------------ #
# Indicators #
# ------------------------------------------------------------------ #
def _compute_signals(self, df: DataFrame, retry_enabled: bool) -> DataFrame:
"""
Adds columns:
ny_date, HP, LP, vwap_live,
enter_long, enter_short, enter_tag,
long_sl_price, short_sl_price
"""
df = df.copy()
ny_dt = df['date'].dt.tz_convert(NY_TZ)
df['ny_date'] = ny_dt.dt.date
ny_hour = ny_dt.dt.hour
# ── Range: 00:00–04:00 NY wicks ──────────────────────────────────
in_range = (ny_hour >= RANGE_START_HOUR) & (ny_hour < RANGE_END_HOUR)
range_df = df[in_range]
day_hp = range_df.groupby('ny_date')['high'].max()
day_lp = range_df.groupby('ny_date')['low'].min()
df['HP'] = df['ny_date'].map(day_hp)
df['LP'] = df['ny_date'].map(day_lp)
# ── Live VWAP anchored at 00:00 NY each day ───────────────────────
# typical_price × volume, cumulated from midnight NY, divided by
# cumulative volume. Resets every calendar day in NY time.
df['_tp'] = (df['high'] + df['low'] + df['close']) / 3
df['_tp_vol'] = df['_tp'] * df['volume']
df['_cum_vol'] = df.groupby('ny_date')['volume'].cumsum()
df['_cum_tpvol'] = df.groupby('ny_date')['_tp_vol'].cumsum()
df['vwap_live'] = df['_cum_tpvol'] / df['_cum_vol']
df.drop(columns=['_tp', '_tp_vol', '_cum_vol', '_cum_tpvol'], inplace=True)
# ── 5m Volume SMA ────────────────────────────────────────────────
df['vol_sma'] = df['volume'].rolling(window=20).mean()
# ── HTF Trend Filter (1h EMA 50) ─────────────────────────────────
# Use simple pandas mapping if possible, or just calculate on 5m
# 1 hour = 12 * 5m candles. 50 EMA on 1h is approx 600 EMA on 5m.
df['ema_1h'] = df['close'].ewm(span=600, adjust=False).mean()
# ── Numpy arrays for the signal loop ─────────────────────────────
n = len(df)
closes = df['close'].to_numpy()
highs = df['high'].to_numpy()
lows = df['low'].to_numpy()
opens = df['open'].to_numpy()
vwaps = df['vwap_live'].to_numpy()
vol_smas = df['vol_sma'].to_numpy()
ema_1hs = df['ema_1h'].to_numpy()
volumes = df['volume'].to_numpy()
ny_dates = df['ny_date'].to_numpy()
ny_hours = ny_hour.to_numpy()
hps = df['HP'].to_numpy()
lps = df['LP'].to_numpy()
# next_opens[i] = open of candle i+1 (fakeout filter).
# NaN at the last row — live edge defers to confirm_trade_entry.
next_opens = np.full(n, np.nan)
if n > 1:
next_opens[:-1] = opens[1:]
# ── 5-candle fractal swing flags on closes ────────────────────────
swing_high = np.zeros(n, dtype=bool)
swing_low = np.zeros(n, dtype=bool)
for k in range(2, n - 2):
ck = closes[k]
if (ck > closes[k-1] and ck > closes[k-2]
and ck > closes[k+1] and ck > closes[k+2]):
swing_high[k] = True
if (ck < closes[k-1] and ck < closes[k-2]
and ck < closes[k+1] and ck < closes[k+2]):
swing_low[k] = True
# ── Output arrays ─────────────────────────────────────────────────
enter_long = np.zeros(n, dtype=np.int8)
enter_short = np.zeros(n, dtype=np.int8)
long_sl = np.full(n, np.nan)
short_sl = np.full(n, np.nan)
tags = np.full(n, None, dtype=object)
# ── Signal state machine ──────────────────────────────────────────
current_date = None
hp_broken = False
lp_broken = False
running_max_close = -np.inf
running_min_close = np.inf
running_max_high = -np.inf
running_min_low = np.inf
running_max_vol = 0
running_max_vol_sma = 0
hp_break_start = -1
lp_break_start = -1
for i in range(n):
d = ny_dates[i]
if d != current_date:
current_date = d
hp_broken = False
lp_broken = False
running_max_close = -np.inf
running_min_close = np.inf
running_max_high = -np.inf
running_min_low = np.inf
running_max_vol = 0
running_max_vol_sma = 0
hp_break_start = -1
lp_break_start = -1
# Only generate signals after the range window closes
if ny_hours[i] < RANGE_END_HOUR:
continue
hp = hps[i]
lp = lps[i]
if np.isnan(hp) or np.isnan(lp):
continue
c = closes[i]
range_size = hp - lp
# ── SHORT side ────────────────────────────────────────────────
if c > hp:
if not hp_broken:
hp_break_start = i
hp_broken = True
if c > running_max_close:
running_max_close = c
if highs[i] > running_max_high:
running_max_high = highs[i]
if volumes[i] > running_max_vol:
running_max_vol = volumes[i]
running_max_vol_sma = vol_smas[i]
elif hp_broken and c < hp:
# Freakout detected — determine SL
breakout_mag = running_max_close - hp
sl_value = running_max_high # default: highest wick
no_fractal = False
if range_size > 0 and breakout_mag > range_size:
# Breakout too large — fall back to nearest fractal swing
# high close between HP and running_max_close
fallback = np.nan
lowest = np.inf
for k in range(hp_break_start, i - 1):
if swing_high[k]:
ck = closes[k]
if hp < ck < running_max_close and ck < lowest:
lowest = ck
fallback = ck
sl_value = fallback
no_fractal = np.isnan(fallback)
# Decide whether to emit and whether to reset state
emitted = False
if not no_fractal:
next_open = next_opens[i]
# Fakeout filter: next candle must open inside range
open_ok = np.isnan(next_open) or next_open < hp
# VWAP filter: HP must be above current candle's VWAP.
cv = vwaps[i]
vwap_ok = np.isnan(cv) or (hp > cv)
# New filters
filter_pass = True
if getattr(self, 'filter_min_sl', CategoricalParameter([0, 1], default=0)).value == 1:
if ((sl_value - hp) / hp) < 0.005: filter_pass = False
if getattr(self, 'filter_htf_trend', CategoricalParameter([0, 1], default=0)).value == 1:
if c > ema_1hs[i]: filter_pass = False # Only short if below HTF EMA
if getattr(self, 'filter_vol_exhaust', CategoricalParameter([0, 1], default=0)).value == 1:
if running_max_vol > (2.0 * running_max_vol_sma): filter_pass = False
if getattr(self, 'filter_strong_fakeout', CategoricalParameter([0, 1], default=0)).value == 1:
if (hp - c) / hp < 0.001: filter_pass = False # Must close at least 0.1% into range
if open_ok and vwap_ok and filter_pass:
enter_short[i] = 1
short_sl[i] = sl_value
tags[i] = 'four_short_hp'
emitted = True
# Reset rules:
# • emitted → reset (signal consumed)
# • no fractal → reset (retry can't help)
# • filters rejected → retry_enabled controls
if emitted or no_fractal or not retry_enabled:
hp_broken = False
running_max_close = -np.inf
running_max_high = -np.inf
hp_break_start = -1
# ── LONG side ─────────────────────────────────────────────────
if c < lp:
if not lp_broken:
lp_break_start = i
lp_broken = True
if c < running_min_close:
running_min_close = c
if lows[i] < running_min_low:
running_min_low = lows[i]
if volumes[i] > running_max_vol:
running_max_vol = volumes[i]
running_max_vol_sma = vol_smas[i]
elif lp_broken and c > lp:
breakout_mag = lp - running_min_close
sl_value = running_min_low
no_fractal = False
if range_size > 0 and breakout_mag > range_size:
fallback = np.nan
highest = -np.inf
for k in range(lp_break_start, i - 1):
if swing_low[k]:
ck = closes[k]
if running_min_close < ck < lp and ck > highest:
highest = ck
fallback = ck
sl_value = fallback
no_fractal = np.isnan(fallback)
emitted = False
if not no_fractal:
next_open = next_opens[i]
open_ok = np.isnan(next_open) or next_open > lp
cv = vwaps[i]
vwap_ok = np.isnan(cv) or (lp < cv)
filter_pass = True
if getattr(self, 'filter_min_sl', CategoricalParameter([0, 1], default=0)).value == 1:
if ((lp - sl_value) / lp) < 0.005: filter_pass = False
if getattr(self, 'filter_htf_trend', CategoricalParameter([0, 1], default=0)).value == 1:
if c < ema_1hs[i]: filter_pass = False # Only long if above HTF EMA
if getattr(self, 'filter_vol_exhaust', CategoricalParameter([0, 1], default=0)).value == 1:
if running_max_vol > (2.0 * running_max_vol_sma): filter_pass = False
if getattr(self, 'filter_strong_fakeout', CategoricalParameter([0, 1], default=0)).value == 1:
if (c - lp) / lp < 0.001: filter_pass = False # Must close at least 0.1% into range
if open_ok and vwap_ok and filter_pass:
enter_long[i] = 1
long_sl[i] = sl_value
tags[i] = 'four_long_lp'
emitted = True
if emitted or no_fractal or not retry_enabled:
lp_broken = False
running_min_close = np.inf
running_min_low = np.inf
lp_break_start = -1
df['enter_long'] = enter_long
df['enter_short'] = enter_short
df['long_sl_price'] = long_sl
df['short_sl_price'] = short_sl
df['enter_tag'] = tags
return df
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
return self._compute_signals(dataframe, bool(self.entry_retry_enabled.value))
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# Signals are fully populated in populate_indicators.
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
return dataframe
# ------------------------------------------------------------------ #
# Entry price — limit at the range boundary #
# ------------------------------------------------------------------ #
def custom_entry_price(self, pair: str, trade, current_time: datetime,
proposed_rate: float, entry_tag, side: str, **kwargs) -> float:
"""
Return the range boundary as the limit price:
Short → HP (sell limit at the top of the range)
Long → LP (buy limit at the bottom of the range)
Freshness: pulls HP/LP from the most recent signal row and only
accepts it if that row's NY date matches the current NY date.
Falls back to proposed_rate when no current-day signal is found.
"""
df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if df is None or df.empty:
return proposed_rate
signal_col = 'enter_short' if side == 'short' else 'enter_long'
price_col = 'HP' if side == 'short' else 'LP'
signals = df[df[signal_col] == 1]
if signals.empty:
return proposed_rate
signal_row = signals.iloc[-1]
cur_ts = pd.Timestamp(current_time)
if cur_ts.tzinfo is None:
cur_ts = cur_ts.tz_localize('UTC')
cur_ny_date = cur_ts.tz_convert(NY_TZ).date()
if signal_row.get('ny_date') != cur_ny_date:
return proposed_rate
price = signal_row.get(price_col, np.nan)
if pd.isna(price) or price <= 0:
return proposed_rate
return float(price)
# ------------------------------------------------------------------ #
# Entry timeout — cancel pending limits at NY midnight #
# ------------------------------------------------------------------ #
def check_entry_timeout(self, pair: str, trade: Trade, order,
current_time: datetime, **kwargs) -> bool:
"""
Cancel a pending entry order once the NY day rolls over.
Setups are tied to the NY-day's range; an unfilled limit that
crosses into the next NY day no longer represents a valid setup.
"""
if trade.open_date_utc is None:
return False
open_ts = pd.Timestamp(trade.open_date_utc)
if open_ts.tzinfo is None:
open_ts = open_ts.tz_localize('UTC')
cur_ts = pd.Timestamp(current_time)
if cur_ts.tzinfo is None:
cur_ts = cur_ts.tz_localize('UTC')
return open_ts.tz_convert(NY_TZ).date() != cur_ts.tz_convert(NY_TZ).date()
# ------------------------------------------------------------------ #
# Entry confirmation — live fakeout + VWAP re-check #
# ------------------------------------------------------------------ #
def confirm_trade_entry(self, pair, order_type, amount, rate,
time_in_force, current_time, entry_tag, side, **kwargs):
df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if df is None or df.empty:
return False
last = df.iloc[-1]
hp = last.get('HP', np.nan)
lp = last.get('LP', np.nan)
vwap = last.get('vwap_live', np.nan)
if pd.isna(hp) or pd.isna(lp):
return False
if side == 'short':
# rate = HP (our limit price). Confirm it's still inside the range
# and still above VWAP (both are already guaranteed by the signal,
# but VWAP can drift in live trading before the order fires).
return rate <= hp and (pd.isna(vwap) or rate > vwap)
else:
return rate >= lp and (pd.isna(vwap) or rate < vwap)
# ------------------------------------------------------------------ #
# SL / TP levels — resolved once and cached in custom_data #
# ------------------------------------------------------------------ #
def _ensure_trade_levels(self, pair: str, trade: Trade) -> tuple:
"""
Return (sl_price, tp_price). Values are computed once from the
signal row and stored in trade.custom_data for all subsequent calls.
"""
sl_price = trade.get_custom_data('sl_price')
tp_price = trade.get_custom_data('tp_price')
if sl_price is not None and tp_price is not None:
return float(sl_price), float(tp_price)
df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if df is None or df.empty:
return None, None
sl_col = 'short_sl_price' if trade.is_short else 'long_sl_price'
signal_col = 'enter_short' if trade.is_short else 'enter_long'
prior = df[df['date'] < trade.open_date_utc]
if prior.empty:
return None, None
signals = prior[prior[signal_col] == 1]
if signals.empty:
return None, None
signal_row = signals.iloc[-1]
sl_raw = signal_row[sl_col]
if pd.isna(sl_raw):
return None, None
sl_price = float(sl_raw)
# Entry is the limit fill price = HP (short) or LP (long)
entry = float(trade.open_rate)
if trade.is_short:
tp_price = entry - RR_RATIO * (sl_price - entry)
else:
tp_price = entry + RR_RATIO * (entry - sl_price)
trade.set_custom_data('sl_price', sl_price)
trade.set_custom_data('tp_price', float(tp_price))
return sl_price, float(tp_price)
# ------------------------------------------------------------------ #
# Custom stoploss — absolute price pin for both sides #
# ------------------------------------------------------------------ #
def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
current_rate: float, current_profit: float, **kwargs) -> float:
"""
Pins the stoploss to the signal-derived absolute SL price for both
longs and shorts. Falls back to the static `self.stoploss` (-0.10)
if the SL price cannot be retrieved.
"""
sl_price, _ = self._ensure_trade_levels(pair, trade)
if sl_price is None or current_rate <= 0:
return self.stoploss # hard backstop
return stoploss_from_absolute(
sl_price,
current_rate,
is_short=trade.is_short,
leverage=trade.leverage,
)
# ------------------------------------------------------------------ #
# Custom exit — time stop first, then TP #
# ------------------------------------------------------------------ #
def custom_exit(self, pair: str, trade: Trade, current_time: datetime,
current_rate: float, current_profit: float, **kwargs):
"""
Exit priority:
1. Time stop — close after TIME_STOP_MINS regardless of P&L.
2. Take profit — close when price reaches the pre-computed TP level.
SL is handled by custom_stoploss / the exchange stoploss order.
"""
# 1. Time stop — use total_seconds() so the comparison is correct
# for trades older than 24h and immune to negative timedelta quirks.
elapsed_minutes = (current_time - trade.open_date_utc).total_seconds() / 60
if elapsed_minutes >= TIME_STOP_MINS:
return 'time_stop_180min'
# 2. Take profit
_, tp_price = self._ensure_trade_levels(pair, trade)
if tp_price is None:
return None
if trade.is_short and current_rate <= tp_price:
return 'tp_1to2'
if not trade.is_short and current_rate >= tp_price:
return 'tp_1to2'
return None
# ------------------------------------------------------------------ #
# Leverage #
# ------------------------------------------------------------------ #
def leverage(self, pair: str, current_time: datetime, current_rate: float,
proposed_leverage: float, max_leverage: float,
side: str, **kwargs) -> float:
return 1.0