Timeframe
15m
Direction
Long & Short
Stoploss
-15.0%
Trailing Stop
No
ROI
0m: 1000.0%
Interface Version
N/A
Startup Candles
100
Indicators
3
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
# ================================================================
# SekkaElastic 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 SekkaElastic(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"[SekkaElastic] 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