← Назад"""
WT Bot v3 — Technical Indicators
===================================
WaveTrend, EMA, ATR — чистые функции.
Работают с raw klines от Binance API.
"""
import numpy as np
from src.config import (
WT_CHANNEL_LEN, WT_AVG_LEN, WT_MA_LEN,
WT_OVERSOLD, WT_OVERBOUGHT, EMA_PERIOD,
)
def _klines_to_arrays(klines):
"""Конвертирует Binance klines в numpy arrays."""
highs = np.array([float(k[2]) for k in klines])
lows = np.array([float(k[3]) for k in klines])
closes = np.array([float(k[4]) for k in klines])
volumes = np.array([float(k[5]) for k in klines])
return highs, lows, closes, volumes
def _ema(data, period):
"""Exponential Moving Average."""
alpha = 2 / (period + 1)
ema = np.zeros_like(data)
ema[0] = data[0]
for i in range(1, len(data)):
ema[i] = alpha * data[i] + (1 - alpha) * ema[i - 1]
return ema
def _sma(data, period):
"""Simple Moving Average (last value)."""
if len(data) < period:
return None
return np.mean(data[-period:])
# ============================================================
# WAVETREND
# ============================================================
def calc_wavetrend(klines):
"""
WaveTrend Oscillator (LazyBear).
Returns: (wt1_last, wt2_last) — последние значения.
"""
if len(klines) < WT_AVG_LEN + WT_MA_LEN + 10:
return None, None
highs, lows, closes, _ = _klines_to_arrays(klines)
hlc3 = (highs + lows + closes) / 3
# ESA = EMA(hlc3, channel_len)
esa = _ema(hlc3, WT_CHANNEL_LEN)
# D = EMA(|hlc3 - ESA|, channel_len)
d = _ema(np.abs(hlc3 - esa), WT_CHANNEL_LEN)
# CI = (hlc3 - ESA) / (0.015 * D)
d_safe = np.where(d == 0, 1, d)
ci = (hlc3 - esa) / (0.015 * d_safe)
# WT1 = EMA(CI, avg_len)
wt1 = _ema(ci, WT_AVG_LEN)
# WT2 = SMA(WT1, ma_len) — берём последнее значение
wt2_last = float(np.mean(wt1[-WT_MA_LEN:]))
wt1_last = float(wt1[-1])
return wt1_last, wt2_last
def calc_wavetrend_series(klines):
"""
WaveTrend — полные серии для детекции кросса.
Returns: (wt1_array, wt2_array) — numpy arrays.
"""
if len(klines) < WT_AVG_LEN + WT_MA_LEN + 10:
return None, None
highs, lows, closes, _ = _klines_to_arrays(klines)
hlc3 = (highs + lows + closes) / 3
esa = _ema(hlc3, WT_CHANNEL_LEN)
d = _ema(np.abs(hlc3 - esa), WT_CHANNEL_LEN)
d_safe = np.where(d == 0, 1, d)
ci = (hlc3 - esa) / (0.015 * d_safe)
wt1 = _ema(ci, WT_AVG_LEN)
# WT2 = SMA(WT1, ma_len) — rolling
wt2 = np.zeros_like(wt1)
for i in range(WT_MA_LEN - 1, len(wt1)):
wt2[i] = np.mean(wt1[i - WT_MA_LEN + 1:i + 1])
return wt1, wt2
# ============================================================
# WT ZONE & CROSS DETECTION
# ============================================================
def is_in_wt_zone(wt1, wt2):
"""
Проверка зоны.
Returns: 1 = oversold (потенциальный long), -1 = overbought (short), 0 = neutral.
"""
if wt1 < WT_OVERSOLD or wt2 < WT_OVERSOLD:
return 1
elif wt1 > WT_OVERBOUGHT or wt2 > WT_OVERBOUGHT:
return -1
return 0
def detect_wt_cross(wt1_series, wt2_series):
"""
Детекция WT кросса на ПОСЛЕДНЕЙ закрытой свече.
Используем [-2] и [-1] (предпоследняя и последняя ЗАКРЫТАЯ).
Returns: 1 = bullish cross, -1 = bearish cross, 0 = none.
"""
if len(wt1_series) < 3 or len(wt2_series) < 3:
return 0
# Используем закрытые свечи: [-3] = предпоследняя, [-2] = последняя закрытая
# [-1] = текущая незакрытая — НЕ используем
prev_wt1 = wt1_series[-3]
prev_wt2 = wt2_series[-3]
curr_wt1 = wt1_series[-2]
curr_wt2 = wt2_series[-2]
# Bullish cross: wt1 пересекает wt2 снизу вверх
cross_up = prev_wt1 <= prev_wt2 and curr_wt1 > curr_wt2
# Bearish cross: wt1 пересекает wt2 сверху вниз
cross_down = prev_wt1 >= prev_wt2 and curr_wt1 < curr_wt2
if cross_up:
return 1
elif cross_down:
return -1
return 0
# ============================================================
# EMA
# ============================================================
def calc_ema(klines, period=EMA_PERIOD):
"""
EMA от close. Возвращает последнее значение.
"""
if len(klines) < period:
return None
closes = np.array([float(k[4]) for k in klines])
ema = _ema(closes, period)
return float(ema[-1])
# ============================================================
# ATR
# ============================================================
def calc_atr_pct(klines, period=14):
"""
ATR в процентах от цены (для скринера).
Использует 1h klines.
"""
if len(klines) < period + 1:
return None
highs, lows, closes, _ = _klines_to_arrays(klines)
trs = []
for i in range(1, len(highs)):
tr = max(
highs[i] - lows[i],
abs(highs[i] - closes[i - 1]),
abs(lows[i] - closes[i - 1]),
)
trs.append(tr)
if len(trs) < period:
return None
atr = np.mean(trs[-period:])
last_price = closes[-1]
if last_price == 0:
return None
return (atr / last_price) * 100