# source: https://raw.githubusercontent.com/labanux/freq/b573987c312315003c6ed9cd1bcf8cd009ec500b/user_data/strategies/sekka-elastic.py # ================================================================ # Github_labanux_freq__sekka_elastic__20260505_044916 v0.2 — Daily BB Mean-Reversion + M15 MSS # ---------------------------------------------------------------- # Strategy Plan: user_data/strategies/plan/elasticity_engine_freqtrade_plan.md # # Entry (long): Daily BB squeeze → wick below lower BB → # M15 MSS above SLH → FVG retest (fallback: MSS candle) # Entry (short): Daily BB squeeze → wick above upper BB → # M15 MSS below SHL → FVG retest (fallback: MSS candle) # # Exit: T1 (60%) at Daily SMA → trail M15 swing lows (runner) # T2 (40%) at opposite Daily BB or 48h timeout # Invalidation: BB expansion + touch, 2 consecutive closes outside BB # # CVD: loads trades feather → 15m delta/cvd; graceful fallback to True # if trades file missing (plan §11 failure-mode mitigation) # ================================================================ import logging import numpy as np import pandas as pd import talib.abstract as ta from functools import reduce from datetime import datetime from pathlib import Path from typing import Optional from freqtrade.strategy import (IStrategy, IntParameter, DecimalParameter, merge_informative_pair, stoploss_from_absolute, stoploss_from_open) from freqtrade.persistence import Trade from pandas import DataFrame logger = logging.getLogger(__name__) class Github_labanux_freq__sekka_elastic__20260505_044916(IStrategy): timeframe = '15m' can_short = True process_only_new_candles = True use_exit_signal = True exit_profit_only = False ignore_buying_expiry = True position_adjustment_enable = True # Required for T1 partial close # BB(20) on 1d needs 20 daily candles = 1920 M15 candles; add buffer startup_candle_count = 100 # Hard fallback only — real exits managed via custom_stoploss / custom_exit stoploss = -0.15 trailing_stop = False minimal_roi = {"0": 10.0} # Effectively disabled # ---------------------------------------------------------------- # Hyperopt Parameters # ---------------------------------------------------------------- # Regime filters adx_threshold = IntParameter(15, 45, default=25, space='buy', optimize=True) bbw_delta_threshold = DecimalParameter(0.10, 0.35, decimals=2, default=0.20, space='buy', optimize=True) bbw_min_threshold = DecimalParameter(0.02, 0.12, decimals=2, default=0.03, space='buy', optimize=True) bb_slope_threshold = DecimalParameter(0.001, 0.050, decimals=3, default=0.003, space='buy', optimize=True) # Entry quality rr_guardrail = DecimalParameter(0.50, 0.60, decimals=2, default=0.50, space='buy', optimize=True) # MSS timing τ in M15 candles (32 = 8h, 64 = 16h, 96 = 24h) mss_candle_window = IntParameter(16, 128, default=64, space='buy', optimize=True) # Delta intensity — SMA₂₀(|Δ|) multiplier (plan §7.1) delta_intensity_multiplier = DecimalParameter(0.5, 3.0, decimals=1, default=1.5, space='buy', optimize=True) # T1 partial close fraction t1_partial_close_pct = DecimalParameter(0.50, 0.75, decimals=2, default=0.60, space='sell', optimize=True) # ---------------------------------------------------------------- # Internal trade state — keyed by trade.id (int) # Lazily initialised on first call to custom_stoploss # ---------------------------------------------------------------- _trade_info: dict = {} def bot_start(self, **kwargs): self._trade_info = {} self.cvd_cache = {} # ---------------------------------------------------------------- # CVD loader — resample raw trades to 15m delta/cvd columns # Returns empty DataFrame if trades file missing (graceful fallback) # ---------------------------------------------------------------- def compute_true_cvd(self, pair: str) -> pd.DataFrame: pair_filename = pair.replace('/', '_').replace(':', '_') data_dir = Path(self.config['datadir']) if self.config.get('trading_mode', 'spot') == 'futures' and data_dir.name != 'futures': data_dir = data_dir / 'futures' trades_path = data_dir / f"{pair_filename}-trades.feather" if not trades_path.exists(): logger.warning( f"[Github_labanux_freq__sekka_elastic__20260505_044916] Trades file not found: {trades_path} " f"— CVD stubs active (all True)" ) return pd.DataFrame() trades_df = pd.read_feather(trades_path) if not pd.api.types.is_datetime64_any_dtype(trades_df['timestamp']): trades_df['timestamp'] = pd.to_datetime(trades_df['timestamp'], unit='ms', utc=True) trades_df.set_index('timestamp', inplace=True) trades_df['taker_buy'] = trades_df['side'] == 'buy' trades_df['buy_vol'] = trades_df['amount'].where(trades_df['taker_buy'], 0) trades_df['sell_vol'] = trades_df['amount'].where(~trades_df['taker_buy'], 0) resampled = trades_df.resample('15min').agg( buy_vol=('buy_vol', 'sum'), sell_vol=('sell_vol', 'sum'), ).fillna(0) resampled['delta'] = resampled['buy_vol'] - resampled['sell_vol'] resampled['total_vol'] = resampled['buy_vol'] + resampled['sell_vol'] resampled['cvd'] = resampled['delta'].cumsum() return resampled[['delta', 'cvd', 'total_vol']] # ---------------------------------------------------------------- # Informative pairs # ---------------------------------------------------------------- def informative_pairs(self): return [(pair, '1d') for pair in self.dp.current_whitelist()] # ---------------------------------------------------------------- # Indicators # ---------------------------------------------------------------- def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pair = metadata['pair'] tau = self.mss_candle_window.value # ============================================================ # 1. DAILY (1d) REGIME INDICATORS # ============================================================ inf = self.dp.get_pair_dataframe(pair=pair, timeframe='1d') # Bollinger Bands (20, 2σ) bb = ta.BBANDS(inf, timeperiod=20, nbdevup=2.0, nbdevdn=2.0, matype=0) inf['bb_upper'] = bb['upperband'] inf['bb_lower'] = bb['lowerband'] inf['bb_mid'] = bb['middleband'] # std = (upper - mid) / 2 (exact for 2σ bands) bb_std = (bb['upperband'] - bb['middleband']) / 2.0 inf['bb_3sigma_lower'] = inf['bb_mid'] - 3 * bb_std inf['bb_3sigma_upper'] = inf['bb_mid'] + 3 * bb_std # BB Width and 3-candle delta (squeeze filter) inf['bbw'] = (bb['upperband'] - bb['lowerband']) / bb['middleband'] inf['delta_bbw'] = (inf['bbw'] - inf['bbw'].shift(3)) / inf['bbw'].shift(3) # ADX (14) inf['adx'] = ta.ADX(inf, timeperiod=14) # BB slope — normalised by SMA to be price-invariant across assets inf['bb_slope'] = inf['bb_mid'] - inf['bb_mid'].shift(1) inf['bb_slope_pct'] = inf['bb_slope'].abs() / inf['bb_mid'] # Merge into 15m frame — all columns get '_1d' suffix dataframe = merge_informative_pair(dataframe, inf, self.timeframe, '1d', ffill=True) # ============================================================ # 2. M15 SWING STRUCTURE # ============================================================ lb = 3 # candles each side to confirm a swing roll = 2 * lb + 1 dataframe['swing_high'] = ( dataframe['high'] == dataframe['high'].rolling(roll, center=True).max() ) dataframe['swing_low'] = ( dataframe['low'] == dataframe['low'].rolling(roll, center=True).min() ) # Running last-confirmed swing level (forward-filled) dataframe['last_swing_high'] = dataframe['high'].where(dataframe['swing_high']).ffill() dataframe['last_swing_low'] = dataframe['low'].where(dataframe['swing_low']).ffill() # ============================================================ # 3. LONG SETUP — wick below lower BB # ============================================================ # Outside-in trigger: M15 wicks below daily lower BB, closes back inside dataframe['wick_event_long'] = ( (dataframe['low'] < dataframe['bb_lower_1d']) & (dataframe['close'] > dataframe['bb_lower_1d']) ) dataframe['wick_low'] = dataframe['low'].where(dataframe['wick_event_long']).ffill() # SLH: last swing high level at the moment of the wick event dataframe['slh_level'] = ( dataframe['last_swing_high'].where(dataframe['wick_event_long']).ffill() ) # Recent wick: trigger fired within last τ candles dataframe['recent_wick_long'] = ( dataframe['wick_event_long'].rolling(tau).max().fillna(0).astype(bool) ) # MSS long: close crosses above SLH within τ candles of wick event dataframe['mss_candle_long'] = ( dataframe['recent_wick_long'] & (dataframe['close'] > dataframe['slh_level']) & (dataframe['close'].shift(1) <= dataframe['slh_level'].shift(1)) ) dataframe['mss_valid_long'] = ( dataframe['mss_candle_long'].rolling(tau).max().fillna(0).astype(bool) ) # Bullish FVG: gap between candle[i-1].high (bottom) and candle[i+1].low (top) # Confirmed on candle[i+1] — no lookahead bias. # Geometry: high[T-1] < low[T+1] means price gapped up through the MSS candle. dataframe['is_fvg_confirmed_long'] = ( (dataframe['high'].shift(2) < dataframe['low']) & dataframe['mss_candle_long'].shift(1, fill_value=False) ) # fvg_upper = top of zone = low[T+1] (higher price) # fvg_lower = bottom of zone = high[T-1] (lower price) dataframe['fvg_upper_long'] = ( dataframe['low'].where(dataframe['is_fvg_confirmed_long']).ffill() ) dataframe['fvg_lower_long'] = ( dataframe['high'].shift(2).where(dataframe['is_fvg_confirmed_long']).ffill() ) dataframe['fvg_valid_long'] = dataframe['fvg_upper_long'] > dataframe['fvg_lower_long'] dataframe['fvg_retest_long'] = ( dataframe['fvg_valid_long'] & (dataframe['low'] <= dataframe['fvg_upper_long']) & (dataframe['close'] >= dataframe['fvg_lower_long']) ) # Candles elapsed since last MSS long (resets on each new MSS) _mss_groups_long = dataframe['mss_candle_long'].cumsum() dataframe['candles_since_mss_long'] = ( _mss_groups_long.groupby(_mss_groups_long).cumcount() ) # FVG retest already happened within the current MSS validity window? fvg_retested_long = ( dataframe['fvg_retest_long'].rolling(tau).max().fillna(0).astype(bool) ) # Entry trigger: # [preferred] FVG retest within τ candles of MSS # [fallback] 2h (8 candles) passed with no retest → enter on any valid candle until τ dataframe['entry_trigger_long'] = ( (dataframe['fvg_retest_long'] & dataframe['mss_valid_long']) | (dataframe['mss_valid_long'] & (dataframe['candles_since_mss_long'] >= 8) & ~fvg_retested_long) ) # R:R guardrail: reject if price already too far toward SMA move_in_long = (dataframe['close'] - dataframe['wick_low']).clip(lower=0) total_range_long = (dataframe['bb_mid_1d'] - dataframe['wick_low']).replace(0, np.nan) dataframe['rr_ratio_long'] = move_in_long / total_range_long # ============================================================ # 4. SHORT SETUP — wick above upper BB # ============================================================ # Outside-in trigger: M15 wicks above daily upper BB, closes back inside dataframe['wick_event_short'] = ( (dataframe['high'] > dataframe['bb_upper_1d']) & (dataframe['close'] < dataframe['bb_upper_1d']) ) dataframe['wick_high'] = dataframe['high'].where(dataframe['wick_event_short']).ffill() # SHL (Significant Higher Low): last swing low level at the wick event dataframe['shl_level'] = ( dataframe['last_swing_low'].where(dataframe['wick_event_short']).ffill() ) dataframe['recent_wick_short'] = ( dataframe['wick_event_short'].rolling(tau).max().fillna(0).astype(bool) ) # MSS short: close crosses below SHL within τ candles of wick event dataframe['mss_candle_short'] = ( dataframe['recent_wick_short'] & (dataframe['close'] < dataframe['shl_level']) & (dataframe['close'].shift(1) >= dataframe['shl_level'].shift(1)) ) dataframe['mss_valid_short'] = ( dataframe['mss_candle_short'].rolling(tau).max().fillna(0).astype(bool) ) # Bearish FVG: gap between candle[i-1].low (top) and candle[i+1].high (bottom) # Confirmed on candle[i+1] — no lookahead bias. # Geometry: low[T-1] > high[T+1] means price gapped down through the MSS candle. dataframe['is_fvg_confirmed_short'] = ( (dataframe['low'].shift(2) > dataframe['high']) & dataframe['mss_candle_short'].shift(1, fill_value=False) ) # fvg_upper = top of zone = low[T-1] (higher price) # fvg_lower = bottom of zone = high[T+1] (lower price) dataframe['fvg_upper_short'] = ( dataframe['low'].shift(2).where(dataframe['is_fvg_confirmed_short']).ffill() ) dataframe['fvg_lower_short'] = ( dataframe['high'].where(dataframe['is_fvg_confirmed_short']).ffill() ) dataframe['fvg_valid_short'] = dataframe['fvg_lower_short'] < dataframe['fvg_upper_short'] dataframe['fvg_retest_short'] = ( dataframe['fvg_valid_short'] & (dataframe['high'] >= dataframe['fvg_lower_short']) & (dataframe['close'] <= dataframe['fvg_upper_short']) ) # Candles elapsed since last MSS short (resets on each new MSS) _mss_groups_short = dataframe['mss_candle_short'].cumsum() dataframe['candles_since_mss_short'] = ( _mss_groups_short.groupby(_mss_groups_short).cumcount() ) # FVG retest already happened within the current MSS validity window? fvg_retested_short = ( dataframe['fvg_retest_short'].rolling(tau).max().fillna(0).astype(bool) ) # Entry trigger: # [preferred] FVG retest within τ candles of MSS # [fallback] 2h (8 candles) passed with no retest → enter on any valid candle until τ dataframe['entry_trigger_short'] = ( (dataframe['fvg_retest_short'] & dataframe['mss_valid_short']) | (dataframe['mss_valid_short'] & (dataframe['candles_since_mss_short'] >= 8) & ~fvg_retested_short) ) move_in_short = (dataframe['wick_high'] - dataframe['close']).clip(lower=0) total_range_short = (dataframe['wick_high'] - dataframe['bb_mid_1d']).replace(0, np.nan) dataframe['rr_ratio_short'] = move_in_short / total_range_short # ============================================================ # 5. EXIT INVALIDATION COLUMNS # ============================================================ # Two consecutive M15 closes outside Daily BB close_below_lb = dataframe['close'] < dataframe['bb_lower_1d'] close_above_ub = dataframe['close'] > dataframe['bb_upper_1d'] dataframe['two_close_below_lb'] = close_below_lb & close_below_lb.shift(1, fill_value=False) dataframe['two_close_above_ub'] = close_above_ub & close_above_ub.shift(1, fill_value=False) # BB expanding AND price touching the band bb_expanding = dataframe['delta_bbw_1d'] > 0 dataframe['bb_expanding_touch_lower'] = ( bb_expanding & (dataframe['low'] <= dataframe['bb_lower_1d']) ) dataframe['bb_expanding_touch_upper'] = ( bb_expanding & (dataframe['high'] >= dataframe['bb_upper_1d']) ) # ============================================================ # 6. CVD DIVERGENCE & DELTA INTENSITY (plan §4.3 + appendix) # ============================================================ # Rolling window for divergence: 96 × 15m = 24h CVD_DIV_WINDOW = 96 # Delta intensity baseline: SMA₂₀(|Δ|) per plan appendix formula DELTA_SMA_WINDOW = 20 if pair not in self.cvd_cache: self.cvd_cache[pair] = self.compute_true_cvd(pair) cvd_df = self.cvd_cache[pair] if cvd_df.empty: # Trades file unavailable — graceful fallback (plan §11 failure modes) dataframe['cvd_divergence_long'] = True dataframe['cvd_divergence_short'] = True dataframe['delta_intensity_pass_long'] = True dataframe['delta_intensity_pass_short'] = True else: # --- Merge delta/cvd onto 15m frame --- df_idx = dataframe.set_index(pd.to_datetime(dataframe['date'], utc=True)) df_idx = df_idx.join(cvd_df[['delta', 'cvd', 'total_vol']], how='left') df_idx['cvd'] = df_idx['cvd'].ffill() df_idx['delta'] = df_idx['delta'].fillna(0) df_idx['total_vol'] = df_idx['total_vol'].fillna(0) dataframe['cvd'] = df_idx['cvd'].values dataframe['delta'] = df_idx['delta'].values dataframe['total_vol'] = df_idx['total_vol'].values # --- Rolling price & CVD extremes for divergence --- price_rolling_low = dataframe['low'].rolling(CVD_DIV_WINDOW).min() price_rolling_high = dataframe['high'].rolling(CVD_DIV_WINDOW).max() cvd_rolling_low = dataframe['cvd'].rolling(CVD_DIV_WINDOW).min() cvd_rolling_high = dataframe['cvd'].rolling(CVD_DIV_WINDOW).max() # Bullish CVD divergence: price at N-candle low (±0.1% buffer), CVD above its low # Vectorized approximation of plan's detect_cvd_divergence(low_idx_1, low_idx_2) cvd_bull_div = ( (dataframe['low'] <= price_rolling_low * 1.001) & (dataframe['cvd'] > cvd_rolling_low) ) # Active when a recent wick-below-BB occurred within τ candles dataframe['cvd_divergence_long'] = ( dataframe['recent_wick_long'] & cvd_bull_div ) # Bearish CVD divergence: price at N-candle high, CVD below its high cvd_bear_div = ( (dataframe['high'] >= price_rolling_high * 0.999) & (dataframe['cvd'] < cvd_rolling_high) ) dataframe['cvd_divergence_short'] = ( dataframe['recent_wick_short'] & cvd_bear_div ) # --- Delta intensity: SMA₂₀(|Δ|), shift(1) excludes current candle # per plan note: "rolling window must exclude the wick candle itself" delta_sma20 = ( dataframe['delta'].abs() .rolling(DELTA_SMA_WINDOW, min_periods=1) .mean() .shift(1) .clip(lower=1e-8) ) # Long: MSS candle must have aggressive buy delta ≥ multiplier × SMA₂₀(|Δ|) mss_intense_long = ( dataframe['mss_candle_long'] & (dataframe['delta'] > 0) & (dataframe['delta'] >= self.delta_intensity_multiplier.value * delta_sma20) ) dataframe['delta_intensity_pass_long'] = ( mss_intense_long.rolling(tau).max().fillna(0).astype(bool) ) # Short: MSS candle must have aggressive sell delta ≥ multiplier × SMA₂₀(|Δ|) mss_intense_short = ( dataframe['mss_candle_short'] & (dataframe['delta'] < 0) & (dataframe['delta'].abs() >= self.delta_intensity_multiplier.value * delta_sma20) ) dataframe['delta_intensity_pass_short'] = ( mss_intense_short.rolling(tau).max().fillna(0).astype(bool) ) return dataframe # ---------------------------------------------------------------- # Entry # ---------------------------------------------------------------- def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # --- Long --- long_conditions = [ #dataframe['adx_1d'] < self.adx_threshold.value, dataframe['delta_bbw_1d'] < self.bbw_delta_threshold.value, dataframe['bbw_1d'] > self.bbw_min_threshold.value, dataframe['bb_slope_pct_1d'] < self.bb_slope_threshold.value, dataframe['recent_wick_long'], dataframe['rr_ratio_long'] < self.rr_guardrail.value, dataframe['entry_trigger_long'], dataframe['cvd_divergence_long'], dataframe['delta_intensity_pass_long'], dataframe['volume'] > 0, ] dataframe.loc[ reduce(lambda x, y: x & y, long_conditions), ['enter_long', 'enter_tag'] ] = (1, 'elastic_long') # --- Short --- short_conditions = [ dataframe['adx_1d'] < self.adx_threshold.value, dataframe['delta_bbw_1d'] < self.bbw_delta_threshold.value, dataframe['bbw_1d'] > self.bbw_min_threshold.value, dataframe['bb_slope_pct_1d'] < self.bb_slope_threshold.value, dataframe['recent_wick_short'], dataframe['rr_ratio_short'] < self.rr_guardrail.value, dataframe['entry_trigger_short'], dataframe['cvd_divergence_short'], dataframe['delta_intensity_pass_short'], dataframe['volume'] > 0, ] dataframe.loc[ reduce(lambda x, y: x & y, short_conditions), ['enter_short', 'enter_tag'] ] = (1, 'elastic_short') return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # All exits managed via custom_stoploss and custom_exit return dataframe # ---------------------------------------------------------------- # Trade info helper — lazy init on first call, keyed by trade.id # ---------------------------------------------------------------- def _get_trade_info(self, trade: Trade, df: DataFrame) -> dict: """Return state dict for this trade, initialising from df on first call.""" tid = trade.id if tid not in self._trade_info: last = df.iloc[-1] if trade.is_short: wick_ref = float(last.get('wick_high', trade.open_rate * 1.10)) else: wick_ref = float(last.get('wick_low', trade.open_rate * 0.90)) self._trade_info[tid] = { 'wick_low': float(last.get('wick_low', trade.open_rate * 0.90)), 'wick_high': float(last.get('wick_high', trade.open_rate * 1.10)), 'wick_ref': wick_ref, 'daily_sma': float(last.get('bb_mid_1d', trade.open_rate)), 'daily_upper': float(last.get('bb_upper_1d', trade.open_rate * 1.10)), 'daily_lower': float(last.get('bb_lower_1d', trade.open_rate * 0.90)), 't1_hit': False, 't1_closed': False, 't1_time': None, } return self._trade_info[tid] # ---------------------------------------------------------------- # Custom stoploss — 3-phase stop management # ---------------------------------------------------------------- def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: try: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) info = self._get_trade_info(trade, df) last = df.iloc[-1] except Exception: return self.stoploss t1_hit = info['t1_hit'] if trade.is_short: wick_ref = info['wick_high'] bb_3sigma_stop = float(last.get('bb_3sigma_upper_1d', current_rate * 1.15)) last_swing_ref = float(last.get('last_swing_high', wick_ref)) if not t1_hit: stop = min(wick_ref * 1.001, bb_3sigma_stop) return stoploss_from_absolute(stop, current_rate, is_short=True) # After T1: trail M15 swing highs for short runner if last_swing_ref and last_swing_ref > current_rate: return stoploss_from_absolute(last_swing_ref, current_rate, is_short=True) else: wick_ref = info['wick_low'] bb_3sigma_stop = float(last.get('bb_3sigma_lower_1d', current_rate * 0.85)) last_swing_ref = float(last.get('last_swing_low', wick_ref)) if not t1_hit: stop = max(wick_ref * 0.999, bb_3sigma_stop) return stoploss_from_absolute(stop, current_rate, is_short=False) # After T1: trail M15 swing lows for long runner if last_swing_ref and last_swing_ref < current_rate: return stoploss_from_absolute(last_swing_ref, current_rate, is_short=False) # Fallback after T1: breakeven + 0.2% return stoploss_from_open(0.002, current_profit) # ---------------------------------------------------------------- # Custom exit — T1 / T2 / time / structure / invalidation # ---------------------------------------------------------------- def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> Optional[str]: try: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) info = self._get_trade_info(trade, df) last = df.iloc[-1] except Exception: return None t1_hit = info['t1_hit'] daily_sma = float(last.get('bb_mid_1d', 0)) daily_upper = float(last.get('bb_upper_1d', 0)) daily_lower = float(last.get('bb_lower_1d', 0)) if trade.is_short: # T1 short: price drops to Daily SMA if not t1_hit and daily_sma and current_rate <= daily_sma: info['t1_hit'] = True info['t1_time'] = current_time return None # partial close fires in adjust_trade_position if t1_hit: # T2 short: price drops to opposite (lower) Daily BB if daily_lower and current_rate <= daily_lower: return 'target_2_full' # Time exit: 48h after T1 t1_time = info.get('t1_time') if t1_time and (current_time - t1_time).total_seconds() > 48 * 3600: return 'time_invalidation_runner' # M15 structure invalidation: swing high forms above entry if last.get('swing_high') and last['high'] > last.get('last_swing_high', 0): return 'structure_invalidation' # Invalidation: BB expanding AND touching upper band if last.get('bb_expanding_touch_upper'): return 'volatility_expansion_exit' # Invalidation: 2 consecutive closes above upper BB if last.get('two_close_above_ub'): return 'two_candle_invalidation' else: # T1 long: price rises to Daily SMA if not t1_hit and daily_sma and current_rate >= daily_sma: info['t1_hit'] = True info['t1_time'] = current_time return None # partial close fires in adjust_trade_position if t1_hit: # T2 long: price rises to opposite (upper) Daily BB if daily_upper and current_rate >= daily_upper: return 'target_2_full' # Time exit: 48h after T1 t1_time = info.get('t1_time') if t1_time and (current_time - t1_time).total_seconds() > 48 * 3600: return 'time_invalidation_runner' # M15 structure invalidation: swing low forms below entry if last.get('swing_low') and last['low'] < last.get('last_swing_low', float('inf')): return 'structure_invalidation' # Invalidation: BB expanding AND touching lower band if last.get('bb_expanding_touch_lower'): return 'volatility_expansion_exit' # Invalidation: 2 consecutive closes below lower BB if last.get('two_close_below_lb'): return 'two_candle_invalidation' return None # ---------------------------------------------------------------- # T1 partial close (default 60% of position) # ---------------------------------------------------------------- def adjust_trade_position(self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, min_stake: Optional[float], max_stake: float, **kwargs) -> Optional[float]: try: df, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) info = self._get_trade_info(trade, df) except Exception: return None if not info.get('t1_hit') or info.get('t1_closed'): return None # Negative stake = reduce position (sell for long, buy-back for short) sell_stake = -(trade.stake_amount * self.t1_partial_close_pct.value) info['t1_closed'] = True return sell_stake # ---------------------------------------------------------------- # Leverage — 1x until strategy is validated in backtesting # ---------------------------------------------------------------- def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: return 1.0