Накопления и распределения (A/D Line) для определения тренда

Анализ рынка – ключевой элемент успешной торговли на криптовалютных биржах. Один из эффективных инструментов – линия накопления/распределения (A/D Line), которая помогает определить, в какую сторону движется тренд. В этой статье мы рассмотрим код на Python, который позволяет автоматически получать данные с Bybit и анализировать рынок с использованием A/D Line.

Что делает этот скрипт?

Данный код автоматически:
✅ Получает данные свечей с Bybit через API.
✅ Вычисляет линию накопления/распределения (A/D Line).
✅ Определяет направление тренда (восходящий или нисходящий).
✅ Выводит обновленную информацию каждые 30 секунд.

Теперь разберем его подробнее.

Получение данных с Bybit

Первый шаг – это запрос к API Bybit, чтобы получить 15-минутные свечи по BTCUSDT:

def get_kline_data():
    """Получает исторические свечи (open, high, low, close, volume)"""
    params = {"category": CATEGORY, "symbol": SYMBOL, "interval": INTERVAL, "limit": LIMIT}
    response = requests.get(KLINE_URL, params=params)
    
    if response.status_code == 200:
        result = response.json().get("result", {}).get("list", [])
        
        candles = []
        for c in result:
            try:
                open_price = float(c[1])
                high = float(c[2])
                low = float(c[3])
                close = float(c[4])
                volume = float(c[5])
                candles.append((open_price, high, low, close, volume))
            except (ValueError, IndexError):
                print(f"Ошибка обработки свечи: {c}")
        
        return candles
    else:
        raise Exception(f"Ошибка API Kline: {response.status_code}, {response.text}")

Разбор:

1. Отправляется GET-запрос к API Bybit.
2. Если запрос успешен, обрабатываем данные свечей.
3. Вытягиваем open, high, low, close, volume.
4. Возвращаем список свечей для анализа.

Расчет линии накопления/распределения

Теперь вычисляем A/D Line:

def calculate_ad_line(candles):
    """Вычисляет линию накопления/распределения (A/D)"""
    ad_line = 0
    ad_values = []
    
    for open_price, high, low, close, volume in candles:
        if high != low:
            money_flow_multiplier = ((close - low) - (high - close)) / (high - low)
        else:
            money_flow_multiplier = 0  # Защита от деления на ноль
            
        money_flow_volume = money_flow_multiplier * volume
        ad_line += money_flow_volume
        ad_values.append(ad_line)

    return ad_values

Разбор:

• Money Flow Multiplier – показывает, находится ли цена ближе к минимуму или максимуму.
• Money Flow Volume – рассчитывает объем на основе мультипликатора.
• A/D Line – суммирует значения и формирует линию тренда.

Определение тренда

Последний шаг – определение восходящего или нисходящего тренда:

def detect_trend():
    """Анализ тренда на основе A/D Line"""
    try:
        print(f"Начинаем анализ {SYMBOL}...\n")
        while True:
            candles = get_kline_data()
            if not candles:
                print("Не удалось получить свечи, пробуем снова...")
                time.sleep(30)
                continue

            ad_values = calculate_ad_line(candles)
            current_price = candles[-1][3]  # Цена закрытия последней свечи
            
            ad_trend = "⬆️ Восходящий" if ad_values[-1] > ad_values[-2] else "⬇️ Нисходящий"
            
            print(f"Цена: {current_price:.2f}, A/D Line: {ad_values[-1]:.2f}, Тренд: {ad_trend}")
            print("-" * 40)
            
            time.sleep(30)  # Обновление каждые 30 секунд
    except Exception as e:
        print(f"Ошибка: {e}")

Разбор:

1. Запускаем бесконечный цикл анализа.
2. Получаем свечи, рассчитываем A/D Line.
3. Сравниваем последние два значения A/D:
   • Если текущее значение выше предыдущего – восходящий тренд.
   • Если ниже – нисходящий тренд.
4. Выводим результаты в консоль.

Полный код

Вот весь код, который можно запустить:

import requests
import time

# Константы
KLINE_URL = "https://api.bybit.com/v5/market/kline"
SYMBOL = "BTCUSDT"
CATEGORY = "spot"
INTERVAL = "15"  # 15-минутные свечи
LIMIT = 50  # Количество свечей для анализа

def get_kline_data():
    """Получает исторические свечи (open, high, low, close, volume)"""
    params = {"category": CATEGORY, "symbol": SYMBOL, "interval": INTERVAL, "limit": LIMIT}
    response = requests.get(KLINE_URL, params=params)
    
    if response.status_code == 200:
        result = response.json().get("result", {}).get("list", [])
        
        candles = []
        for c in result:
            try:
                open_price = float(c[1])
                high = float(c[2])
                low = float(c[3])
                close = float(c[4])
                volume = float(c[5])
                candles.append((open_price, high, low, close, volume))
            except (ValueError, IndexError):
                print(f"Ошибка обработки свечи: {c}")
        
        return candles
    else:
        raise Exception(f"Ошибка API Kline: {response.status_code}, {response.text}")

def calculate_ad_line(candles):
    """Вычисляет линию накопления/распределения (A/D)"""
    ad_line = 0
    ad_values = []
    
    for open_price, high, low, close, volume in candles:
        if high != low:
            money_flow_multiplier = ((close - low) - (high - close)) / (high - low)
        else:
            money_flow_multiplier = 0  # Защита от деления на ноль
            
        money_flow_volume = money_flow_multiplier * volume
        ad_line += money_flow_volume
        ad_values.append(ad_line)

    return ad_values

def detect_trend():
    """Анализ тренда на основе A/D Line"""
    try:
        print(f"Начинаем анализ {SYMBOL}...\n")
        while True:
            candles = get_kline_data()
            if not candles:
                print("Не удалось получить свечи, пробуем снова...")
                time.sleep(30)
                continue

            ad_values = calculate_ad_line(candles)
            current_price = candles[-1][3]  # Цена закрытия последней свечи
            
            ad_trend = "⬆️ Восходящий" if ad_values[-1] > ad_values[-2] else "⬇️ Нисходящий"
            
            print(f"Цена: {current_price:.2f}, A/D Line: {ad_values[-1]:.2f}, Тренд: {ad_trend}")
            print("-" * 40)
            
            time.sleep(30)  # Обновление каждые 30 секунд
    except Exception as e:
        print(f"Ошибка: {e}")

if __name__ == "__main__":
    detect_trend()