542 lines
19 KiB
Python
542 lines
19 KiB
Python
# @param strategy_lever_rate float 策略风险杠杆系数
|
|
# @param profit_line float 锁盈触发收益率
|
|
# @param lock_profit_rate float 锁盈回撤保护比例
|
|
# @param open_time_interval float 冷却窗口小时数
|
|
# @param up_line_span int 开仓突破通道周期
|
|
# @param up_line_offset float 突破通道偏移倍数
|
|
# @param buy_stop_profit_span int 动态止盈底线周期
|
|
# @param buy_stop_profit_offset float 动态止盈底线偏移倍数
|
|
# @param ma_span_long int 长均线天数
|
|
# @param regime_slope_lookback int 趋势检测回看K线数
|
|
# @param regime_slope_threshold float 趋势斜率阈值
|
|
# @param regime_displace_threshold float 价格偏离阈值
|
|
# @param regime_vol_ema_span int 波动率EMA周期
|
|
# @param regime_compression_threshold float 压缩状态波动阈值
|
|
# @param regime_expansion_threshold float 扩张状态波动阈值
|
|
# @param regime_hysteresis_bars int 状态切换确认K线数
|
|
# @param entry_up_line_span_short int 压缩突破短通道周期
|
|
# @param entry_ma_span_short int 回调短均线天数
|
|
# @param entry_pullback_bars_min int 回调最低K线数
|
|
# @param exit_max_loss_pct float 保护止损最大亏损比例
|
|
# @param exit_breakeven_buffer float 保本止损触发缓冲
|
|
# @param exit_max_hold_bars int 时间止损最大持仓K线数
|
|
# @strategy tradeDirection long
|
|
|
|
SPREAD_SPAN = 6
|
|
N_VALUE_SPAN = 10
|
|
|
|
|
|
def _ema(values):
|
|
value = None
|
|
span = float(len(values))
|
|
for number in values:
|
|
number = float(number)
|
|
if value is None:
|
|
value = number
|
|
else:
|
|
value = 2 * number / (span + 1) + (span - 1) / (span + 1) * value
|
|
return value
|
|
|
|
|
|
def _history_bars(ctx, length, history=None):
|
|
length = int(length)
|
|
if length <= 0:
|
|
return []
|
|
if history is None:
|
|
bars = ctx.bars(length + 1)
|
|
if len(bars) <= 1:
|
|
return []
|
|
history = bars[:-1]
|
|
if len(history) < length:
|
|
return []
|
|
return history[-length:]
|
|
|
|
|
|
def _n_value(history):
|
|
window = _history_bars(None, SPREAD_SPAN * N_VALUE_SPAN, history=history)
|
|
if len(window) < SPREAD_SPAN * N_VALUE_SPAN:
|
|
return None
|
|
spreads = []
|
|
for i in range(N_VALUE_SPAN):
|
|
start = i * SPREAD_SPAN
|
|
chunk = window[start:start + SPREAD_SPAN]
|
|
high = max(bar.high for bar in chunk)
|
|
low = min(bar.low for bar in chunk)
|
|
spreads.append(high - low)
|
|
return _ema(spreads)
|
|
|
|
|
|
def _open_up_line(history, span):
|
|
window = _history_bars(None, span, history=history)
|
|
if len(window) < int(span):
|
|
return None
|
|
return max(bar.high for bar in window)
|
|
|
|
|
|
def _stop_profit_bottom_line(history, span):
|
|
window = _history_bars(None, span, history=history)
|
|
if len(window) < int(span):
|
|
return None
|
|
return min(bar.low for bar in window)
|
|
|
|
|
|
def _ma_long(history, span_days):
|
|
length = int(span_days) * 24 * 6
|
|
window = _history_bars(None, length, history=history)
|
|
if length <= 0 or len(window) < length:
|
|
return None
|
|
return sum(bar.close for bar in window) / float(length)
|
|
|
|
|
|
def _prepare_history(ctx, up_line_span, buy_stop_profit_span, ma_span_long,
|
|
regime_slope_lookback=0, entry_up_line_span_short=0, entry_ma_span_short=0):
|
|
ma_length = int(ma_span_long) * 24 * 6
|
|
ma_short_length = int(entry_ma_span_short) * 24 * 6 if entry_ma_span_short else 0
|
|
required = max(
|
|
SPREAD_SPAN * N_VALUE_SPAN,
|
|
int(up_line_span),
|
|
int(buy_stop_profit_span),
|
|
ma_length,
|
|
int(regime_slope_lookback) + ma_length,
|
|
int(entry_up_line_span_short),
|
|
ma_short_length,
|
|
)
|
|
if required <= 0:
|
|
return []
|
|
return _history_bars(ctx, required)
|
|
|
|
|
|
def _cached_indicators(ctx, params):
|
|
if hasattr(ctx, 'indicator_value'):
|
|
up_line_short = None
|
|
ma_short = None
|
|
if params.get('entry_up_line_span_short'):
|
|
up_line_short = ctx.indicator_value('up_line_short')
|
|
if params.get('entry_ma_span_short'):
|
|
ma_short = ctx.indicator_value('ma_short')
|
|
# Always compute history for regime signal MA slope calculation
|
|
history = _prepare_history(
|
|
ctx,
|
|
params['up_line_span'],
|
|
params['buy_stop_profit_span'],
|
|
params['ma_span_long'],
|
|
params.get('regime_slope_lookback', 0),
|
|
params.get('entry_up_line_span_short', 0),
|
|
params.get('entry_ma_span_short', 0),
|
|
)
|
|
return {
|
|
'n_value': ctx.indicator_value('n_value'),
|
|
'up_line': ctx.indicator_value('up_line'),
|
|
'stop_profit_bottom': ctx.indicator_value('stop_profit_bottom'),
|
|
'ma_long': ctx.indicator_value('ma_long'),
|
|
'up_line_short': up_line_short,
|
|
'ma_short': ma_short,
|
|
'history': history,
|
|
}
|
|
history = _prepare_history(
|
|
ctx,
|
|
params['up_line_span'],
|
|
params['buy_stop_profit_span'],
|
|
params['ma_span_long'],
|
|
params.get('regime_slope_lookback', 0),
|
|
params.get('entry_up_line_span_short', 0),
|
|
params.get('entry_ma_span_short', 0),
|
|
)
|
|
result = {
|
|
'n_value': _n_value(history),
|
|
'up_line': _open_up_line(history, params['up_line_span']),
|
|
'stop_profit_bottom': _stop_profit_bottom_line(history, params['buy_stop_profit_span']),
|
|
'ma_long': _ma_long(history, params['ma_span_long']),
|
|
'history': history,
|
|
}
|
|
if params.get('entry_up_line_span_short'):
|
|
result['up_line_short'] = _open_up_line(history, params['entry_up_line_span_short'])
|
|
if params.get('entry_ma_span_short'):
|
|
result['ma_short'] = _ma_long(history, params['entry_ma_span_short'])
|
|
return result
|
|
|
|
|
|
def _strategy_params(ctx):
|
|
return {
|
|
'strategy_lever_rate': float(ctx.param('strategy_lever_rate')),
|
|
'profit_line': float(ctx.param('profit_line')),
|
|
'lock_profit_rate': float(ctx.param('lock_profit_rate')),
|
|
'open_time_interval': float(ctx.param('open_time_interval')),
|
|
'up_line_span': int(ctx.param('up_line_span')),
|
|
'up_line_offset': float(ctx.param('up_line_offset')),
|
|
'buy_stop_profit_span': int(ctx.param('buy_stop_profit_span')),
|
|
'buy_stop_profit_offset': float(ctx.param('buy_stop_profit_offset')),
|
|
'ma_span_long': int(ctx.param('ma_span_long')),
|
|
'regime_slope_lookback': int(ctx.param('regime_slope_lookback')),
|
|
'regime_slope_threshold': float(ctx.param('regime_slope_threshold')),
|
|
'regime_displace_threshold': float(ctx.param('regime_displace_threshold')),
|
|
'regime_vol_ema_span': int(ctx.param('regime_vol_ema_span')),
|
|
'regime_compression_threshold': float(ctx.param('regime_compression_threshold')),
|
|
'regime_expansion_threshold': float(ctx.param('regime_expansion_threshold')),
|
|
'regime_hysteresis_bars': int(ctx.param('regime_hysteresis_bars')),
|
|
'entry_up_line_span_short': int(ctx.param('entry_up_line_span_short')),
|
|
'entry_ma_span_short': int(ctx.param('entry_ma_span_short')),
|
|
'entry_pullback_bars_min': int(ctx.param('entry_pullback_bars_min')),
|
|
'exit_max_loss_pct': float(ctx.param('exit_max_loss_pct')),
|
|
'exit_breakeven_buffer': float(ctx.param('exit_breakeven_buffer')),
|
|
'exit_max_hold_bars': int(ctx.param('exit_max_hold_bars')),
|
|
}
|
|
|
|
|
|
def _ensure_indicator_cache(ctx, params):
|
|
if hasattr(ctx, 'set_indicator_cache'):
|
|
ctx.set_indicator_cache(params)
|
|
return True
|
|
return False
|
|
|
|
|
|
def _position_size_pct(ctx, n_value, strategy_lever_rate):
|
|
if n_value is None or n_value <= 0:
|
|
return 0.0
|
|
price = ctx.current_price()
|
|
if price <= 0:
|
|
return 0.0
|
|
stop_loss_pct = n_value / price
|
|
if stop_loss_pct <= 0:
|
|
return 0.0
|
|
pct = 0.01 * float(strategy_lever_rate) / stop_loss_pct
|
|
return min(max(pct, 0.0), 1.0)
|
|
|
|
|
|
def _time_diff_ms(current_time, last_close_time):
|
|
if current_time is None or last_close_time is None:
|
|
return None
|
|
delta = current_time - last_close_time
|
|
if hasattr(delta, 'total_seconds'):
|
|
return delta.total_seconds() * 1000.0
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Market regime detection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _regime_state_init(ctx):
|
|
defaults = {
|
|
'effective_regime': 'range',
|
|
'regime_candidate': 'range',
|
|
'regime_candidate_bars': 0,
|
|
'n_value_ema': None,
|
|
'prev_vol_ratio': None,
|
|
}
|
|
for key, val in defaults.items():
|
|
if ctx.get_state(key, None) is None:
|
|
ctx.set_state(key, val)
|
|
|
|
|
|
def _regime_signals(ctx, bar, params, indicators):
|
|
ma_long = indicators['ma_long']
|
|
n_value = indicators['n_value']
|
|
history = indicators.get('history')
|
|
if ma_long is None or n_value is None:
|
|
return None
|
|
|
|
if bar.close == 0:
|
|
return None
|
|
price_displacement = (bar.close - ma_long) / ma_long
|
|
|
|
lookback = int(params['regime_slope_lookback'])
|
|
ma_slope = None
|
|
if history is not None and lookback > 0:
|
|
ma_length = int(params['ma_span_long']) * 24 * 6
|
|
past_history = history[:max(0, len(history) - lookback)]
|
|
ma_past = _ma_long(past_history, params['ma_span_long'])
|
|
if ma_past is not None and ma_past != 0:
|
|
ma_slope = (ma_long - ma_past) / ma_past
|
|
|
|
n_ema = ctx.get_state('n_value_ema', None)
|
|
ema_span = int(params['regime_vol_ema_span'])
|
|
if n_ema is None:
|
|
n_ema = n_value
|
|
else:
|
|
alpha = 2.0 / (ema_span + 1.0)
|
|
n_ema = alpha * n_value + (1.0 - alpha) * n_ema
|
|
ctx.set_state('n_value_ema', n_ema)
|
|
|
|
vol_ratio = n_value / n_ema if n_ema and n_ema > 0 else 1.0
|
|
|
|
prev_vol_ratio = ctx.get_state('prev_vol_ratio', None)
|
|
ctx.set_state('prev_vol_ratio', vol_ratio)
|
|
|
|
return {
|
|
'ma_slope': ma_slope,
|
|
'price_displacement': price_displacement,
|
|
'vol_ratio': vol_ratio,
|
|
'vol_ratio_rising': prev_vol_ratio is not None and vol_ratio > prev_vol_ratio,
|
|
}
|
|
|
|
|
|
def _classify_regime(signals, params):
|
|
if signals is None:
|
|
return 'range'
|
|
|
|
vol_ratio = signals['vol_ratio']
|
|
ma_slope = signals['ma_slope']
|
|
price_displacement = signals['price_displacement']
|
|
|
|
if vol_ratio < float(params['regime_compression_threshold']):
|
|
return 'compression'
|
|
|
|
if vol_ratio > float(params['regime_expansion_threshold']):
|
|
return 'expansion'
|
|
|
|
slope_threshold = float(params['regime_slope_threshold'])
|
|
displace_threshold = float(params['regime_displace_threshold'])
|
|
if (ma_slope is not None
|
|
and abs(ma_slope) > slope_threshold
|
|
and abs(price_displacement) > displace_threshold):
|
|
return 'trend'
|
|
|
|
return 'range'
|
|
|
|
|
|
def _effective_regime(ctx, new_regime, params):
|
|
prev_candidate = ctx.get_state('regime_candidate', 'range')
|
|
if new_regime == prev_candidate:
|
|
bars = ctx.get_state('regime_candidate_bars', 0) + 1
|
|
ctx.set_state('regime_candidate_bars', bars)
|
|
else:
|
|
ctx.set_state('regime_candidate', new_regime)
|
|
ctx.set_state('regime_candidate_bars', 1)
|
|
return ctx.get_state('effective_regime', 'range')
|
|
|
|
hysteresis = int(params['regime_hysteresis_bars'])
|
|
if ctx.get_state('regime_candidate_bars', 0) >= hysteresis:
|
|
ctx.set_state('effective_regime', new_regime)
|
|
return new_regime
|
|
|
|
return ctx.get_state('effective_regime', 'range')
|
|
|
|
|
|
def _current_regime(ctx):
|
|
return ctx.get_state('effective_regime', 'range')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Entry modes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _entry_cooldown_ok(ctx, params):
|
|
last_close_time = ctx.get_state('last_close_time', None)
|
|
time_diff_ms = _time_diff_ms(ctx.current_time, last_close_time)
|
|
if time_diff_ms is None:
|
|
return True
|
|
time_range_ms = max(float(params['open_time_interval']), 0.0) * 60 * 60 * 1000.0
|
|
return time_range_ms <= 0 or time_diff_ms >= time_range_ms
|
|
|
|
|
|
def _entry_breakout_chase(ctx, bar, params, indicators):
|
|
n_value = indicators['n_value']
|
|
up_line = indicators['up_line']
|
|
ma_long = indicators['ma_long']
|
|
if n_value is None or up_line is None or ma_long is None:
|
|
return False
|
|
|
|
if bar.close <= ma_long:
|
|
return False
|
|
|
|
threshold = up_line + n_value * float(params['up_line_offset'])
|
|
if bar.close > threshold:
|
|
position_pct = _position_size_pct(ctx, n_value, float(params['strategy_lever_rate']))
|
|
if position_pct > 0:
|
|
ctx.buy(amount=position_pct)
|
|
return True
|
|
return False
|
|
|
|
|
|
def _entry_compression_breakout(ctx, bar, params, indicators):
|
|
n_value = indicators['n_value']
|
|
up_line_short = indicators.get('up_line_short')
|
|
if n_value is None or up_line_short is None:
|
|
return False
|
|
|
|
n_ema = ctx.get_state('n_value_ema', None)
|
|
if n_ema is None or n_ema <= 0:
|
|
return False
|
|
current_vol = n_value / n_ema
|
|
|
|
prev_vol = ctx.get_state('_prev_vol_saved', None)
|
|
ctx.set_state('_prev_vol_saved', current_vol)
|
|
if prev_vol is None:
|
|
return False
|
|
if current_vol <= prev_vol:
|
|
return False
|
|
|
|
threshold = up_line_short + n_value * float(params['up_line_offset'])
|
|
if bar.close > threshold:
|
|
position_pct = _position_size_pct(ctx, n_value, float(params['strategy_lever_rate']))
|
|
if position_pct > 0:
|
|
ctx.buy(amount=position_pct)
|
|
return True
|
|
return False
|
|
|
|
|
|
def _entry_pullback_reentry(ctx, bar, params, indicators):
|
|
ma_short = indicators.get('ma_short')
|
|
if ma_short is None:
|
|
return False
|
|
|
|
bars_below = ctx.get_state('pullback_bars_below', 0)
|
|
if bar.close < ma_short:
|
|
ctx.set_state('pullback_bars_below', bars_below + 1)
|
|
return False
|
|
|
|
min_bars = int(params['entry_pullback_bars_min'])
|
|
if bars_below >= min_bars and bar.close > ma_short:
|
|
ctx.set_state('pullback_bars_below', 0)
|
|
n_value = indicators['n_value']
|
|
position_pct = _position_size_pct(ctx, n_value, float(params['strategy_lever_rate']))
|
|
if position_pct > 0:
|
|
ctx.buy(amount=position_pct)
|
|
return True
|
|
|
|
ctx.set_state('pullback_bars_below', 0)
|
|
return False
|
|
|
|
|
|
def _entry_router(ctx, bar, params, indicators, regime):
|
|
if not _entry_cooldown_ok(ctx, params):
|
|
return False
|
|
|
|
if regime == 'trend':
|
|
return _entry_breakout_chase(ctx, bar, params, indicators)
|
|
elif regime in ('compression', 'expansion'):
|
|
return _entry_compression_breakout(ctx, bar, params, indicators)
|
|
elif regime == 'range':
|
|
return _entry_pullback_reentry(ctx, bar, params, indicators)
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Exit modules
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _exit_protective_stop(ctx, bar, params):
|
|
loss_limit = (float(params['exit_max_loss_pct'])
|
|
* max(ctx.balance, 0.0)
|
|
* float(params['strategy_lever_rate']))
|
|
unrealized_loss = max(ctx.entry_balance() - ctx.equity, 0.0)
|
|
if unrealized_loss > loss_limit:
|
|
ctx.close_position()
|
|
ctx.set_state('exit_reason', 'protective_stop')
|
|
return True
|
|
return False
|
|
|
|
|
|
def _exit_breakeven_stop(ctx, bar, params, indicators):
|
|
profit = ctx.unrealized_profit_pct(bar.close)
|
|
buffer_val = float(params['exit_breakeven_buffer'])
|
|
|
|
if profit > buffer_val:
|
|
ctx.set_state('breakeven_armed', True)
|
|
|
|
if ctx.get_state('breakeven_armed', False):
|
|
n_value = indicators.get('n_value')
|
|
entry = ctx.entry_price()
|
|
if n_value is not None and entry > 0:
|
|
# Give 0.3 N-value breathing room below entry so noise doesn't trigger exit
|
|
breakeven_level = entry - n_value * 0.3
|
|
else:
|
|
breakeven_level = entry * 0.999
|
|
if bar.close < breakeven_level:
|
|
ctx.close_position()
|
|
ctx.set_state('exit_reason', 'breakeven_stop')
|
|
return True
|
|
return False
|
|
|
|
|
|
def _exit_trailing_stop(ctx, bar, params, indicators):
|
|
stop_profit_bottom = indicators['stop_profit_bottom']
|
|
n_value = indicators['n_value']
|
|
if stop_profit_bottom is None or n_value is None:
|
|
return False
|
|
|
|
buy_stop_profit = stop_profit_bottom + n_value * float(params['buy_stop_profit_offset'])
|
|
if bar.close < buy_stop_profit:
|
|
ctx.close_position()
|
|
ctx.set_state('exit_reason', 'trailing_stop')
|
|
return True
|
|
return False
|
|
|
|
|
|
def _exit_time_stop(ctx, bar, params):
|
|
bars_in_pos = ctx.get_state('bars_in_position', 0) + 1
|
|
ctx.set_state('bars_in_position', bars_in_pos)
|
|
if bars_in_pos >= int(params['exit_max_hold_bars']):
|
|
ctx.close_position()
|
|
ctx.set_state('exit_reason', 'time_stop')
|
|
return True
|
|
return False
|
|
|
|
|
|
def _exit_dispatcher(ctx, bar, params, indicators):
|
|
# Priority: protective → trailing → breakeven → time
|
|
# Trailing before breakeven: when trailing stop rises above entry, it captures
|
|
# trend profits; breakeven only acts as safety net when trailing hasn't activated.
|
|
if _exit_protective_stop(ctx, bar, params):
|
|
return True
|
|
if _exit_trailing_stop(ctx, bar, params, indicators):
|
|
return True
|
|
if _exit_breakeven_stop(ctx, bar, params, indicators):
|
|
return True
|
|
if _exit_time_stop(ctx, bar, params):
|
|
return True
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lifecycle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def on_init(ctx):
|
|
ctx.signal_timing = 'next_bar_open'
|
|
ctx.max_profit = 0.0
|
|
ctx.last_close_time = None
|
|
ctx.last_close_index = None
|
|
params = _strategy_params(ctx)
|
|
_ensure_indicator_cache(ctx, params)
|
|
ctx.set_state('strategy_params_cache', params)
|
|
_regime_state_init(ctx)
|
|
ctx.set_state('breakeven_armed', False)
|
|
ctx.set_state('bars_in_position', 0)
|
|
ctx.set_state('exit_reason', None)
|
|
ctx.set_state('pullback_bars_below', 0)
|
|
ctx.set_state('_prev_vol_saved', None)
|
|
|
|
|
|
def on_bar(ctx, bar):
|
|
params = ctx.get_state('strategy_params_cache') or _strategy_params(ctx)
|
|
indicators = _cached_indicators(ctx, params)
|
|
|
|
n_value = indicators['n_value']
|
|
up_line = indicators['up_line']
|
|
stop_profit_bottom = indicators['stop_profit_bottom']
|
|
ma_long = indicators['ma_long']
|
|
if n_value is None or up_line is None or stop_profit_bottom is None or ma_long is None:
|
|
return
|
|
|
|
# 1. Classify regime
|
|
signals = _regime_signals(ctx, bar, params, indicators)
|
|
new_regime = _classify_regime(signals, params)
|
|
regime = _effective_regime(ctx, new_regime, params)
|
|
|
|
# 2. If in position, check exits
|
|
if ctx.has_position() and ctx.is_long():
|
|
# Track max profit for observability
|
|
profit = ctx.unrealized_profit_pct(bar.close)
|
|
if profit > ctx.max_profit:
|
|
ctx.max_profit = profit
|
|
|
|
_exit_dispatcher(ctx, bar, params, indicators)
|
|
return
|
|
|
|
# 3. If flat, reset exit state then check entry
|
|
ctx.set_state('breakeven_armed', False)
|
|
ctx.set_state('bars_in_position', 0)
|
|
_entry_router(ctx, bar, params, indicators, regime)
|