Итоговый код Telegram-бота
@faddey24Прежде чем вы продолжите изучать мою писанину, я хочу вас познакомить с нашем каналом тг (ссылка), там мы разбираем рыночные ситуацию, ошибки и активно общаемся по делу 24/7 в нашем чате (ссылка), у нас только рабочая атмосфера. Только при поддержки данного сообщество вышла эта статья, и возможно выйдет еще.
И еще, если среди вас есть ребята, которых заинтересовал мой прототип. Свяжитесь с админ он передаст вам мои контакты и мы создадим мини-сообщество энтузиастов по разработке данного софта. Думаю мы сможем организовать сходку. Пишите не стесняйтесь, возможно к этому времени я смог уже немного улучшить бота.
# -*- coding: utf-8 -*-
import ccxt
import pandas as pd
import requests
import time
import csv
import os
from datetime import datetime, timedelta
# ==============================================================================
# --- ⚙️ ГЛАВНЫЕ НАСТРОЙКИ БОТА И СТРАТЕГИИ ⚙️ ---
# ==============================================================================
# --- КЛЮЧИ ДОСТУПА (API) ---
BYBIT_API_KEY = 'СЮДА_ВАШ_НОВЫЙ_BYBIT_API_KEY'
BYBIT_SECRET_KEY = 'СЮДА_ВАШ_НОВЫЙ_BYBIT_SECRET_KEY'
ETHERSCAN_API_KEY = 'СЮДА_ВАШ_ETHERSCAN_API_KEY'
TELEGRAM_BOT_TOKEN = 'СЮДА_ВАШ_TELEGRAM_BOT_TOKEN'
TELEGRAM_CHAT_ID = 'СЮДА_ВАШ_TELEGRAM_CHAT_ID'
# --- ПАРАМЕТРЫ СТРАТЕГИИ ---
PRICE_LOOKBACK_PERIOD = 5
OI_LOOKBACK_PERIOD = 4
RISK_REWARD_RATIO = 1.5
# Параметры для LONG
MAX_FUNDING_RATE_FOR_LONG = 0.0001 # Макс. фандинг для входа в LONG (не более 0.01%)
# --- Новые параметры для SHORT ---
MIN_FUNDING_RATE_FOR_SHORT = 0.0001 # Мин. фандинг для входа в SHORT (не менее 0.01%)
# --- ПАРАМЕТРЫ ON-CHAIN АНАЛИЗА ---
TRANSACTION_THRESHOLD_AVAX = 5000
ONCHAIN_LOOKBACK_HOURS = 12
# --- ПРОЧИЕ НАСТРОЙКИ ---
SYMBOL = 'AVAX/USDT:USDT'
TIMEFRAME = '15m'
CHECK_INTERVAL_SECONDS = 60
HEARTBEAT_INTERVAL_HOURS = 6 # Интервал для "пульса" бота
TRADE_HISTORY_FILE = 'История сделок с сентября.csv' # Имя файла для истории
# ==============================================================================
# --- Код бота ---
# ==============================================================================
def send_telegram_message(message):
"""Отправляет сообщение в Telegram."""
try:
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {'chat_id': TELEGRAM_CHAT_ID, 'text': message, 'parse_mode': 'Markdown'}
requests.post(url, json=payload, timeout=10)
print(f"Сообщение в Telegram отправлено.")
except Exception as e:
print(f"Критическая ошибка отправки в Telegram: {e}")
def get_bybit_data(exchange):
"""Получает рыночные данные с Bybit."""
limit = PRICE_LOOKBACK_PERIOD + 10
ohlcv = exchange.fetch_ohlcv(SYMBOL, TIMEFRAME, limit=limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
# Получаем историю ОИ (если возможно) или текущее значение
# Для простоты и стабильности используем текущее значение для всех исторических свечей
open_interest = exchange.fetch_open_interest(SYMBOL)
funding_rate = exchange.fetch_funding_rate(SYMBOL)
df['openInterest'] = open_interest.get('openInterestAmount', 0)
df['fundingRate'] = funding_rate.get('fundingRate', 0)
return df
def check_onchain_activity():
"""Проверяет недавнюю активность китов."""
try:
api_url = f"https://api.snowtrace.io/api?module=account&action=tokentx&contractaddress=0xB31f66AA3C1e785363F0875A1B74E27b85FD66c7&page=1&offset=50&sort=desc&apikey={ETHERSCAN_API_KEY}"
response = requests.get(api_url, timeout=10)
data = response.json()
if data['status'] == '1':
for tx in data['result']:
value_avax = int(tx['value']) / 10**18
tx_time = datetime.fromtimestamp(int(tx['timeStamp']))
if value_avax >= TRANSACTION_THRESHOLD_AVAX and (datetime.now() - tx_time) < timedelta(hours=ONCHAIN_LOOKBACK_HOURS):
print(f" - ✅ Найдена on-chain активность: {value_avax:.0f} AVAX")
return True
return False
except Exception as e:
print(f" - ❌ Ошибка проверки on-chain: {e}")
return False
# --- НОВАЯ ФУНКЦИЯ: Логирование сделок в файл ---
def log_trade_to_csv(trade_data):
"""Записывает информацию о сигнале в CSV файл."""
file_exists = os.path.isfile(TRADE_HISTORY_FILE)
with open(TRADE_HISTORY_FILE, 'a', newline='', encoding='utf-8') as csvfile:
fieldnames = ['timestamp', 'type', 'entry_price', 'stop_loss', 'take_profit']
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
if not file_exists:
writer.writeheader() # Создаем заголовок, если файл новый
writer.writerow(trade_data)
print(f" - ✅ Сигнал записан в {TRADE_HISTORY_FILE}")
# --- Основной цикл бота ---
def main():
send_telegram_message("🚀 **Бот запущен и начинает мониторинг рынка AVAX!**")
exchange = ccxt.bybit({'apiKey': BYBIT_API_KEY, 'secret': BYBIT_SECRET_KEY})
last_signal_time = None
# --- НОВАЯ ЛОГИКА: Таймер для "пульса" ---
last_heartbeat_time = datetime.now()
while True:
try:
# --- НОВАЯ ЛОГИКА: Проверка и отправка "пульса" ---
if datetime.now() - last_heartbeat_time > timedelta(hours=HEARTBEAT_INTERVAL_HOURS):
send_telegram_message(f"❤️ **Бот в рабочем состоянии.**\nПоследняя проверка: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
last_heartbeat_time = datetime.now()
df_history = get_bybit_data(exchange)
latest = df_history.iloc[-1]
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] Проверка... Close: ${latest['close']:.3f}, OI: {latest['openInterest']:.0f}, FR: {latest['fundingRate']:.4f}")
if last_signal_time and last_signal_time == latest['timestamp']:
time.sleep(CHECK_INTERVAL_SECONDS)
continue
# --- РАСЧЕТ ИНДИКАТОРОВ ---
df_history['rolling_low'] = df_history['low'].shift(1).rolling(window=PRICE_LOOKBACK_PERIOD).min()
df_history['rolling_high'] = df_history['high'].shift(1).rolling(window=PRICE_LOOKBACK_PERIOD).max()
df_history['oi_average'] = df_history['openInterest'].shift(1).rolling(window=OI_LOOKBACK_PERIOD).mean()
# --- ПРОВЕРКА УСЛОВИЙ ДЛЯ LONG ---
long_cond1 = latest['low'] < df_history['rolling_low'].iloc[-1] and latest['close'] > latest['open']
long_cond2 = latest['openInterest'] > df_history['oi_average'].iloc[-1]
long_cond3 = latest['fundingRate'] <= MAX_FUNDING_RATE_FOR_LONG
# --- ПРОВЕРКА УСЛОВИЙ ДЛЯ SHORT ---
short_cond1 = latest['high'] > df_history['rolling_high'].iloc[-1] and latest['close'] < latest['open']
short_cond2 = latest['openInterest'] > df_history['oi_average'].iloc[-1]
short_cond3 = latest['fundingRate'] >= MIN_FUNDING_RATE_FOR_SHORT
# --- ОБЩЕЕ УСЛОВИЕ: On-chain активность ---
onchain_cond = check_onchain_activity()
signal_type = None
if all([long_cond1, long_cond2, long_cond3, onchain_cond]):
signal_type = "LONG"
elif all([short_cond1, short_cond2, short_cond3, onchain_cond]):
signal_type = "SHORT"
# --- ГЕНЕРАЦИЯ СИГНАЛА ---
if signal_type:
print(f"🔥🔥🔥 УСЛОВИЯ СОВПАЛИ ДЛЯ {signal_type}! ГЕНЕРАЦИЯ СИГНАЛА! 🔥🔥🔥")
entry_price = latest['close']
message = ""
trade_log = {}
if signal_type == "LONG":
stop_loss = latest['low']
risk = entry_price - stop_loss
if risk <= 0: continue
take_profit = entry_price + (risk * RISK_REWARD_RATIO)
message = (f"🚨 **НОВЫЙ СИГНАЛ: LONG AVAX/USDT** 🚨\n\n"
f"📈 **Вход:** `${entry_price:.3f}`\n"
f"🛡️ **Стоп-лосс:** `${stop_loss:.3f}`\n"
f"🎯 **Тейк-профит:** `${take_profit:.3f}`")
elif signal_type == "SHORT":
stop_loss = latest['high']
risk = stop_loss - entry_price
if risk <= 0: continue
take_profit = entry_price - (risk * RISK_REWARD_RATIO)
message = (f"🚨 **НОВЫЙ СИГНАЛ: SHORT AVAX/USDT** 🚨\n\n"
f"📉 **Вход:** `${entry_price:.3f}`\n"
f"🛡️ **Стоп-лосс:** `${stop_loss:.3f}`\n"
f"🎯 **Тейк-профит:** `${take_profit:.3f}`")
full_message = message + f"\n\n_Сигнал сгенерирован в {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}_"
send_telegram_message(full_message)
trade_log = {
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'type': signal_type,
'entry_price': f"{entry_price:.3f}", 'stop_loss': f"{stop_loss:.3f}",
'take_profit': f"{take_profit:.3f}"
}
log_trade_to_csv(trade_log) # Записываем сделку в файл
last_signal_time = latest['timestamp']
time.sleep(CHECK_INTERVAL_SECONDS)
except Exception as e:
print(f"Произошла ошибка в главном цикле: {e}")
error_message = f"⚠️ **Внимание! В работе бота произошла ошибка:**\n`{str(e)}`\n_Бот пытается перезапуститься..._"
send_telegram_message(error_message)
time.sleep(30)
if __name__ == "__main__":
main()