# source: https://raw.githubusercontent.com/labanux/freq/9db9cead6613729ce236ad871219534a99d067b4/user_data/strategies/sekka-four.py import logging import numpy as np import pandas as pd import talib.abstract as ta from datetime import datetime from pandas import DataFrame from freqtrade.strategy import IStrategy, stoploss_from_absolute, CategoricalParameter, DecimalParameter from freqtrade.persistence import Trade logger = logging.getLogger(__name__) NY_TZ = 'America/New_York' RANGE_START_HOUR = 8 # NY hour — range window opens RANGE_START_MINUTE = 30 # NY minute — range window opens RANGE_END_HOUR = 9 # NY hour — range window closes, trading begins RANGE_END_MINUTE = 30 # NY minute — range window closes, trading begins RR_RATIO = 1 # Risk:Reward = 1:2 TIME_STOP_MINS = 180 # Exit any position older than this TRAILING_STOP_POSITIVE = 0.005 # 0.5% trail once TP offset is reached RANGE_START_MOD = RANGE_START_HOUR * 60 + RANGE_START_MINUTE RANGE_END_MOD = RANGE_END_HOUR * 60 + RANGE_END_MINUTE class Github_labanux_freq__sekka_four__20260612_100257(IStrategy): """ NY-session Range Breakout / Re-entry Strategy. Range definition ---------------- Build HP (high) and LP (low) from the NY-hour window [RANGE_START_HOUR, RANGE_END_HOUR), using candle wicks. Signal logic ------------ After RANGE_END_HOUR (NY): SHORT — a 15m 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 a meaningful distance inside the range — at least `fakeout_buffer_pct` of the range width past the boundary. – Short: open[N+1] < HP - fakeout_buffer_pct × (HP - LP) – Long : open[N+1] > LP + fakeout_buffer_pct × (HP - LP) `fakeout_buffer_pct` is a hyperopt parameter (0.02–0.15, default 0.05). • 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. SL / TP — handled via custom_stoploss as a two-phase stop: • Phase 1 (pre-TP): SL pinned to the signal-derived absolute price. • Phase 2 (post-TP): once current price reaches the RR-based TP level, switch to a 0.5% trailing stop behind current_rate (acts as trailing_stop_positive=0.005 with trailing_only_offset_is_reached=True, where the offset is the per-trade TP 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 = '5m' 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=1, space='buy', optimize=True ) # Minimum fakeout depth at N+1 open, as a fraction of the range width. # Short fires only if open[N+1] < HP - buffer*(HP-LP) # Long fires only if open[N+1] > LP + buffer*(HP-LP) fakeout_buffer_pct = DecimalParameter( 0.02, 0.35, default=0.05, decimals=2, space='buy', optimize=True ) # RSI is computed for plot visibility only; it is not used in the # entry/exit logic. RSI_PERIOD = 14 plot_config = { "main_plot": { "HP": {"color": "red"}, "LP": {"color": "green"}, "vwap_live": {"color": "orange"}, }, "subplots": { "RSI": { "rsi": {"color": "purple"}, }, }, } # ------------------------------------------------------------------ # # Indicators # # ------------------------------------------------------------------ # def _compute_signals(self, df: DataFrame, retry_enabled: bool, fakeout_buffer: float) -> 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_minute_of_day = ny_dt.dt.hour * 60 + ny_dt.dt.minute # ── Range: [RANGE_START, RANGE_END) NY wicks ───────────────────── in_range = (ny_minute_of_day >= RANGE_START_MOD) & (ny_minute_of_day < RANGE_END_MOD) 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) # ── 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() ny_dates = df['ny_date'].to_numpy() ny_mods = ny_minute_of_day.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 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 hp_break_start = -1 lp_break_start = -1 # Only generate signals after the range window closes if ny_mods[i] < RANGE_END_MOD: 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] 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 the range # by at least `fakeout_buffer` × range_size below HP, so # marginal re-crosses don't count as a confirmed fakeout. short_threshold = hp - fakeout_buffer * range_size open_ok = np.isnan(next_open) or next_open < short_threshold # VWAP filter: HP must be above current candle's VWAP. # Use vwaps[i] (freakout candle close) — same value the # live confirm_trade_entry path sees. No look-ahead. cv = vwaps[i] vwap_ok = np.isnan(cv) or (hp > cv) if open_ok and vwap_ok: 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] 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] # Mirror of the short-side fakeout filter: next candle # must open at least `fakeout_buffer` × range_size above LP. long_threshold = lp + fakeout_buffer * range_size open_ok = np.isnan(next_open) or next_open > long_threshold cv = vwaps[i] vwap_ok = np.isnan(cv) or (lp < cv) if open_ok and vwap_ok: 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: df = self._compute_signals( dataframe, bool(self.entry_retry_enabled.value), float(self.fakeout_buffer_pct.value), ) # Plot-only RSI — not referenced anywhere in entry/exit logic. df['rsi'] = ta.RSI(df, timeperiod=self.RSI_PERIOD) return df 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: """ Two-phase stoploss: 1. Before TP offset is reached → pin SL to the signal-derived absolute price (long: lowest-low / fallback fractal; short mirrored). 2. Once price touches the RR-based TP level → switch to a 0.5% trail behind current_rate. Freqtrade only tightens custom_stoploss returns, so this acts as a one-way ratchet (equivalent to trailing_stop_positive=0.005 with trailing_only_offset_is_reached=True, where the offset is the per-trade TP price). Falls back to the static `self.stoploss` (-0.10) if SL price is unavailable. """ sl_price, tp_price = self._ensure_trade_levels(pair, trade) if sl_price is None or current_rate <= 0: return self.stoploss # hard backstop tp_reached = bool(trade.get_custom_data('tp_reached')) if not tp_reached and tp_price is not None: if trade.is_short and current_rate <= tp_price: trade.set_custom_data('tp_reached', True) tp_reached = True elif not trade.is_short and current_rate >= tp_price: trade.set_custom_data('tp_reached', True) tp_reached = True if tp_reached: return -TRAILING_STOP_POSITIVE 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): """ Time stop only — TP is now realised via the trailing stop activated in `custom_stoploss` once price reaches the RR-based TP level. """ elapsed_minutes = (current_time - trade.open_date_utc).total_seconds() / 60 if elapsed_minutes >= TIME_STOP_MINS: return 'time_stop_180min' 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