Когда мы работаем с криптовалютами, важнейшая задача — распознать точки разворота тренда, особенно после резких скачков. Это может быть шок-свеча вверх, за которой идёт коррекция вниз, или резкое падение, после которого цена начинает восстанавливаться.
В этой статье я покажу, как написать детектор шок-свечей на Python, используя библиотеку ccxt для подключения к Binance и pandas для анализа данных. Скрипт автоматически определяет потенциальные разворотные сигналы и записывает их в файл.
🔧 Что такое «шок-свеча» и как её отследить?
Шок-свеча — это свеча, где изменение цены превышает определённый порог, например 2% за 5 минут. Такие свечи могут быть триггером для крупных движений, особенно если после них цена идет в обратную сторону. Пример:
-
Цена резко выросла на 2.5%, а следующая свеча показывает откат — возможный шорт-сигнал.
-
Цена резко упала на 3%, а следующая свеча закрывается выше — потенциальный лонг-сигнал.
🧠 Как работает скрипт?
Вот логика:
-
Загружаются последние 10 свечей для выбранных криптовалют и таймфреймов.
-
Анализируется предпоследняя свеча:
-
если она выросла или упала более чем на 2%,
-
и следующая свеча пошла в обратную сторону,
-
и при этом тело свечи большое (сравнивается с ATR),
-
и объём выше среднего,
...тогда выводится сигнал на возможный разворот.
-
-
Все сигналы записываются в
signals.csv, чтобы можно было их потом анализировать или подключить Telegram-бота для уведомлений.
🛠 Используемые технологии
-
Python 3.10+
-
ccxt — библиотека для подключения к криптобиржам.
-
pandas — работа с временными рядами и таблицами.
-
CSV логирование — простая запись сигналов в файл.
📈 Результат запуска
[BTC/USDT 5m] Проверка свечи от 2025-06-24 14:35:00: Δ=2.17%, Объём x1.54, Тело/ATR=242.33/215.11
🔻 Потенциал *шорт-реверсала* на BTC/USDT (5m)⚙️ Как адаптировать под себя?
-
Измени список
symbolsиtimeframes, если хочешь следить за другими активами. -
Измени
shock_threshold, чтобы ловить более или менее агрессивные сигналы. -
Подключи Telegram-бота или Discord Webhook для алертов в реальном времени.
🏁 Вывод
Этот инструмент может стать отличной основой для более сложной стратегии или просто как фильтр для ручной торговли. Даже такой простой алгоритм помогает вовремя замечать аномальные движения на рынке и реагировать быстрее.
import ccxt
import pandas as pd
import time
from datetime import datetime
# Настройки
symbols = ['BTC/USDT', 'ETH/USDT']
timeframes = ['5m', '15m']
shock_threshold = 0.02 # 2%
exchange = ccxt.binance()
def fetch_data(symbol, timeframe, limit=10):
try:
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
return df
except Exception as e:
print(f"[{symbol} {timeframe}] Ошибка получения данных: {e}")
return pd.DataFrame()
def check_shock_reversal(df, symbol, timeframe):
if len(df) < 2:
print(f"[{symbol} {timeframe}] Недостаточно данных для анализа.")
return
last = df.iloc[-2]
current = df.iloc[-1]
change = (last['close'] - last['open']) / last['open']
body_size = abs(last['close'] - last['open'])
atr = (df['high'] - df['low']).mean()
volume_avg = df['volume'].mean()
volume_ratio = last['volume'] / volume_avg
print(f"[{symbol} {timeframe}] Проверка свечи от {last['timestamp']}: "
f"Δ={change:.2%}, Объём x{volume_ratio:.2f}, Тело/ATR={body_size:.2f}/{atr:.2f}")
if body_size < atr or volume_ratio < 1:
print(f"⚠️ [{symbol}] Сигнал ослаблен (низкий объём или короткое тело)")
return
if change >= shock_threshold and current['close'] < last['close']:
print(f"🔻 Потенциал *шорт-реверсала* на {symbol} ({timeframe})")
log_signal("short", symbol, timeframe, last['timestamp'])
elif change <= -shock_threshold and current['close'] > last['close']:
print(f"🔺 Потенциал *лонг-реверсала* на {symbol} ({timeframe})")
log_signal("long", symbol, timeframe, last['timestamp'])
def log_signal(direction, symbol, timeframe, timestamp):
with open('signals.csv', 'a') as f:
f.write(f"{datetime.now()},{direction},{symbol},{timeframe},{timestamp}\n")
if __name__ == "__main__":
for symbol in symbols:
for timeframe in timeframes:
df = fetch_data(symbol, timeframe)
check_shock_reversal(df, symbol, timeframe)
time.sleep(1) # пауза между запросами, чтобы избежать rate limit
Пока нет комментариев. Будьте первым!