📌 VWAP Trend Analyzer – Анализ трендов с учетом VWAP

Смысл стратегии:

VWAP (Volume Weighted Average Price) – это индикатор, который показывает среднюю цену сделки с учетом объема.

Используется для оценки тренда:

Цена выше VWAP → бычий тренд (поиск покупок).

Цена ниже VWAP → медвежий тренд (поиск продаж).

VWAP в боковом движении → рынок во флэте.

🔹 Применение:

- Подтверждение тренда → если цена выше VWAP, лучше искать покупки.

- Оценка откатов → если цена падает к VWAP в бычьем тренде, можно искать вход.

- Анализ краткосрочного тренда для скальпинга.

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "5m"  # Анализ краткосрочного тренда
VWAP_PERIOD = 20  # Количество свечей для расчета VWAP
CHECK_INTERVAL = 60  # Проверка раз в минуту

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение данных
def fetch_ohlcv(symbol):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=TIMEFRAME, limit=VWAP_PERIOD + 5)
    df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
    return df

# Рассчет VWAP
def calculate_vwap(df):
    typical_price = (df["high"] + df["low"] + df["close"]) / 3
    vwap = (typical_price * df["volume"]).cumsum() / df["volume"].cumsum()
    return vwap.iloc[-1]

# Анализ тренда с VWAP
def vwap_trend_analyzer():
    df = fetch_ohlcv(PAIR)
    vwap = calculate_vwap(df)
    close_price = df["close"].iloc[-1]

    print(f"Цена: {close_price:.2f}, VWAP: {vwap:.2f}")

    if close_price > vwap:
        print("📈 Цена выше VWAP → бычий тренд (лонг при откатах).")
    elif close_price < vwap:
        print("📉 Цена ниже VWAP → медвежий тренд (шорт при откатах).")
    else:
        print("⚡ Цена у VWAP, возможный флэт.")

# Запуск
if __name__ == "__main__":
    while True:
        vwap_trend_analyzer()
        time.sleep(CHECK_INTERVAL)

#индикатор

📌 Подпишись  Crypto Python❗️

📌 Smart Order Execution – Умное исполнение ордеров (разбитие на части)🧑‍💻

Смысл стратегии:

Этот скрипт помогает избежать рыночного проскальзывания, когда объем ордера слишком велик и может повлиять на цену.

Разбивает крупный ордер на несколько мелких.

Использует случайные интервалы между исполнением, чтобы не привлечь внимание алгоритмов HFT.

Оптимизирует исполнение за счет анализа ликвидности.

import ccxt
import time
import random

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TOTAL_AMOUNT = 1  # Общий объем покупки (BTC)
SPLIT_ORDERS = 10  # Количество частей
MIN_WAIT = 5  # Минимальная задержка между ордерами (сек)
MAX_WAIT = 15  # Максимальная задержка между ордерами (сек)

# API-ключи (замените своими)
API_KEY = "your_api_key"
API_SECRET = "your_api_secret"

# Инициализация биржи
exchange = ccxt.binance({
    "apiKey": API_KEY,
    "secret": API_SECRET,
    "enableRateLimit": True,
})

# Получение текущей цены
def get_current_price(symbol):
    ticker = exchange.fetch_ticker(symbol)
    return ticker["last"]

# Умное исполнение ордера
def smart_order_execution():
    part_size = TOTAL_AMOUNT / SPLIT_ORDERS  # Размер одной сделки

    for i in range(SPLIT_ORDERS):
        price = get_current_price(PAIR)
        print(f"📥 Размещение ордера {i+1}/{SPLIT_ORDERS}: Покупка {part_size:.4f} BTC по {price:.2f} USDT")

        try:
            order = exchange.create_market_buy_order(PAIR, part_size)
            print(f"✅ Ордер исполнен: {order}")
        except Exception as e:
            print(f"❌ Ошибка при исполнении ордера: {e}")

        wait_time = random.randint(MIN_WAIT, MAX_WAIT)
        print(f"⏳ Ждем {wait_time} секунд перед следующим ордером...")
        time.sleep(wait_time)

# Запуск
if __name__ == "__main__":
    smart_order_execution()

#ордера

📌 Подпишись  Crypto Python❗️

📌 Liquidity Pool Scanner – Поиск ликвидных активов🧑‍💻

Смысл стратегии:

Ликвидность – это ключевой фактор успешной торговли. Скрипт анализирует объем торгов и глубину стакана, чтобы найти самые ликвидные активы.

Используется для:

📍Поиска монет с высоким объемом и узким спредом.

📍Выбора лучших активов для скальпинга или арбитража.

📍Избегания неликвидных активов с высоким спредом.

import ccxt
import pandas as pd

# Параметры
EXCHANGE_NAME = "binance"
TOP_N = 10  # Количество лучших активов по ликвидности

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()
exchange.load_markets()

# Получение ликвидных активов
def scan_liquidity():
    tickers = exchange.fetch_tickers()
    liquidity_data = []

    for symbol, data in tickers.items():
        if "/USDT" in symbol and "quoteVolume" in data:
            bid = data.get("bid", None)
            ask = data.get("ask", None)
            spread = (ask - bid) / ask * 100 if bid and ask else None

            liquidity_data.append({
                "pair": symbol,
                "volume_24h": data["quoteVolume"],
                "spread_%": spread
            })

    df = pd.DataFrame(liquidity_data)
    df = df.dropna().sort_values(by=["volume_24h"], ascending=False).head(TOP_N)

    print("📊 Топ ликвидных активов:")
    print(df.to_string(index=False))

# Запуск
if __name__ == "__main__":
    scan_liquidity()

#индикатор

📌 Подпишись  Crypto Python❗️

📌 Market Impact Calculator – Оценка влияния крупных сделок на цену 🧑‍💻

Смысл стратегии:

При размещении крупных ордеров на покупку/продажу цена может измениться, особенно на рынках с низкой ликвидностью.
Этот скрипт анализирует глубину стакана и оценивает, насколько цена изменится при исполнении ордера.

🔹 Метрики влияния:

Объем стакана – сколько ликвидности доступно.

Проскальзывание (Slippage) – насколько изменится цена после сделки.

Рассчет на основе Order Book Depth.

import ccxt

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
ORDER_SIZE = 1  # Объем сделки (в BTC)

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение стакана ордеров
def get_order_book(symbol):
    order_book = exchange.fetch_order_book(symbol, limit=50)
    return order_book

# Оценка влияния сделки
def calculate_market_impact():
    order_book = get_order_book(PAIR)
    bids = order_book["bids"]  # Покупки
    asks = order_book["asks"]  # Продажи

    # Расчет проскальзывания на покупку
    buy_impact = 0
    sell_impact = 0
    buy_volume = ORDER_SIZE
    sell_volume = ORDER_SIZE

    for price, volume in asks:
        if buy_volume > volume:
            buy_impact += volume * price
            buy_volume -= volume
        else:
            buy_impact += buy_volume * price
            break

    for price, volume in bids:
        if sell_volume > volume:
            sell_impact += volume * price
            sell_volume -= volume
        else:
            sell_impact += sell_volume * price
            break

    avg_buy_price = buy_impact / ORDER_SIZE
    avg_sell_price = sell_impact / ORDER_SIZE
    mid_price = (asks[0][0] + bids[0][0]) / 2

    buy_slippage = ((avg_buy_price - mid_price) / mid_price) * 100
    sell_slippage = ((mid_price - avg_sell_price) / mid_price) * 100

    print(f"📊 Средняя цена покупки: {avg_buy_price:.2f}, проскальзывание: {buy_slippage:.4f}%")
    print(f"📉 Средняя цена продажи: {avg_sell_price:.2f}, проскальзывание: {sell_slippage:.4f}%")

# Запуск
if __name__ == "__main__":
    calculate_market_impact()

#ордера

📌 Подпишись  Crypto Python❗️

📌 Delta Volume Analyzer – Анализ дельты объемов для фьючерсов🧑‍💻

Смысл стратегии:

Дельта объема – это разница между объемами покупок по рынку (market buy) и продаж по рынку (market sell).

Положительная дельта (ΔV > 0) → больше рыночных покупок → сильный спрос.

Отрицательная дельта (ΔV < 0) → больше рыночных продаж → сильное предложение.

Используется для оценки силы тренда и поиска ложных пробоев.


🔹 Применение:

Если цена растет, но дельта объема падает → вероятен ложный пробой.

Если цена падает, но дельта объема растет → возможный разворот.

Используется в скальпинге и фьючерсной торговле.

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "1m"  # Используем низкие таймфреймы для анализа потоков объема
CHECK_INTERVAL = 10  # Обновление каждые 10 секунд

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение данных по ленте сделок
def fetch_trades(symbol):
    trades = exchange.fetch_trades(symbol, limit=100)
    df = pd.DataFrame(trades)
    return df

# Рассчет дельты объемов
def calculate_delta_volume():
    df = fetch_trades(PAIR)

    # Фильтруем рыночные покупки и продажи
    buy_volume = df[df["side"] == "buy"]["amount"].sum()
    sell_volume = df[df["side"] == "sell"]["amount"].sum()

    delta_volume = buy_volume - sell_volume  # Разница между рыночными покупками и продажами

    print(f"📊 Дельта объема за последние 100 сделок: {delta_volume:.4f} BTC")
    print(f"📈 Покупки: {buy_volume:.4f} BTC, 📉 Продажи: {sell_volume:.4f} BTC")

    if delta_volume > 0:
        print("🔼 Преобладают рыночные покупки → возможный бычий импульс.")
    elif delta_volume < 0:
        print("🔽 Преобладают рыночные продажи → возможное снижение.")
    else:
        print("⚡ Баланс спроса и предложения.")

# Запуск
if __name__ == "__main__":
    while True:
        calculate_delta_volume()
        time.sleep(CHECK_INTERVAL)

#индикатор

📌 Подпишись  Crypto Python❗️

📌 Dynamic Position Sizing – Адаптивный размер позиций на основе ATR 🧑‍💻

Смысл стратегии:

Adaptive Position Sizing помогает избежать излишней волатильности, регулируя размер позиции в зависимости от ATR (Average True Range).

Высокий ATR → рынок волатильный → уменьшаем размер позиции.

Низкий ATR → рынок спокойный → увеличиваем размер позиции.

Используется в фьючерсной и риск-менеджмент стратегии.

import ccxt
import pandas as pd
import talib
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
ATR_PERIOD = 14  # Период ATR
RISK_PER_TRADE = 100  # Максимальный риск на сделку (в $)
RISK_MULTIPLIER = 1.5  # Коэффициент риска (K)

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение данных
def fetch_ohlcv(symbol):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=TIMEFRAME, limit=ATR_PERIOD + 5)
    df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
    return df

# Рассчет ATR и размера позиции
def calculate_position_size():
    df = fetch_ohlcv(PAIR)
    atr = talib.ATR(df["high"], df["low"], df["close"], timeperiod=ATR_PERIOD).iloc[-1]
    current_price = df["close"].iloc[-1]

    # Рассчитываем размер позиции в BTC
    position_size = RISK_PER_TRADE / (atr * RISK_MULTIPLIER)

    print(f"📊 Цена: {current_price:.2f} USDT, ATR: {atr:.2f}")
    print(f"🔹 Оптимальный размер позиции: {position_size:.4f} BTC")

# Запуск
if __name__ == "__main__":
    while True:
        calculate_position_size()
        time.sleep(60)  # Проверка раз в минуту

#ордера

📌 Подпишись  Crypto Python❗️

📌 Order Book Imbalance Detector – Выявление дисбаланса заявок 🧑‍💻

Смысл стратегии:

Дисбаланс стакана заявок – это ситуация, когда в стакане больше лимитных ордеров на покупку (бидов) или продажу (асков).

Дисбаланс в сторону бидов → рынок может пойти вверх (сильный спрос).

Дисбаланс в сторону асков → рынок может упасть (сильное предложение).

Используется в скальпинге, арбитраже и HFT-торговле.

import ccxt
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
ORDER_BOOK_DEPTH = 20  # Глубина анализа (количество уровней в стакане)
CHECK_INTERVAL = 5  # Проверка раз в 5 секунд

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение стакана заявок
def get_order_book(symbol):
    order_book = exchange.fetch_order_book(symbol, limit=ORDER_BOOK_DEPTH)
    return order_book

# Анализ дисбаланса
def detect_order_book_imbalance():
    order_book = get_order_book(PAIR)
    
    bids = order_book["bids"]  # Лимитные покупки (цена, объем)
    asks = order_book["asks"]  # Лимитные продажи (цена, объем)

    # Рассчитываем общий объем заявок
    bid_volume = sum([volume for price, volume in bids])
    ask_volume = sum([volume for price, volume in asks])

    # Дисбаланс стакана
    imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume)

    print(f"📊 Дисбаланс заявок: {imbalance:.4f} | Биды: {bid_volume:.2f} | Аски: {ask_volume:.2f}")

    if imbalance > 0.1:
        print("📈 Преобладание бидов → возможный рост.")
    elif imbalance < -0.1:
        print("📉 Преобладание асков → возможное падение.")
    else:
        print("⚡ Баланс спроса и предложения.")

# Запуск
if __name__ == "__main__":
    while True:
        detect_order_book_imbalance()
        time.sleep(CHECK_INTERVAL)

#торговые_стратегии

📌 Подпишись  Crypto Python❗️

📌 Slippage Estimator – Расчет проскальзывания для разных бирж🧑‍💻

Смысл стратегии:

Проскальзывание (Slippage) – это разница между ожидаемой ценой сделки и фактической ценой исполнения. Оно возникает из-за недостаточной ликвидности или высокой волатильности.

Этот скрипт оценивает проскальзывание на разных биржах, сравнивая рыночные цены и глубину стакана заявок.

import ccxt
import time

# Параметры
EXCHANGES = ["binance", "kucoin"]  # Биржи для сравнения
PAIR = "BTC/USDT"
ORDER_SIZE = 1  # Объем сделки (в BTC)
ORDER_BOOK_DEPTH = 20  # Глубина анализа стакана

# Инициализация бирж
exchanges = {name: getattr(ccxt, name)() for name in EXCHANGES}

# Получение стакана заявок
def get_order_book(exchange, symbol):
    try:
        return exchange.fetch_order_book(symbol, limit=ORDER_BOOK_DEPTH)
    except Exception as e:
        print(f"❌ Ошибка получения данных с {exchange.id}: {e}")
        return None

# Расчет проскальзывания
def calculate_slippage():
    results = []
    
    for name, exchange in exchanges.items():
        order_book = get_order_book(exchange, PAIR)
        if not order_book:
            continue
        
        bids = order_book["bids"]  # Лимитные покупки (цена, объем)
        asks = order_book["asks"]  # Лимитные продажи (цена, объем)

        # Средняя цена покупки (Market Buy)
        buy_impact, sell_impact = 0, 0
        buy_volume, sell_volume = ORDER_SIZE, ORDER_SIZE

        for price, volume in asks:
            if buy_volume > volume:
                buy_impact += volume * price
                buy_volume -= volume
            else:
                buy_impact += buy_volume * price
                break

        for price, volume in bids:
            if sell_volume > volume:
                sell_impact += volume * price
                sell_volume -= volume
            else:
                sell_impact += sell_volume * price
                break

        avg_buy_price = buy_impact / ORDER_SIZE
        avg_sell_price = sell_impact / ORDER_SIZE
        mid_price = (asks[0][0] + bids[0][0]) / 2

        buy_slippage = ((avg_buy_price - mid_price) / mid_price) * 100
        sell_slippage = ((mid_price - avg_sell_price) / mid_price) * 100

        results.append({
            "Биржа": name,
            "Средняя цена покупки": round(avg_buy_price, 2),
            "Проскальзывание покупки %": round(buy_slippage, 4),
            "Средняя цена продажи": round(avg_sell_price, 2),
            "Проскальзывание продажи %": round(sell_slippage, 4),
        })

    # Вывод результатов
    print("\n📊 Сравнение проскальзывания на биржах:")
    for res in results:
        print(f"{res['Биржа'].capitalize()}: Покупка {res['Средняя цена покупки']} USDT ({res['Проскальзывание покупки %']}%), "
              f"Продажа {res['Средняя цена продажи']} USDT ({res['Проскальзывание продажи %']}%)")

# Запуск
if __name__ == "__main__":
    while True:
        calculate_slippage()
        time.sleep(10)  # Проверка каждые 10 секунд

#ордера

📌 Подпишись  Crypto Python❗️

📌 Volume Spike Detector – Выявление резких всплесков объемов 🧑‍💻

Смысл стратегии:

Всплеск объема может указывать на начало сильного движения цены или присутствие крупных игроков.

Резкий рост объема → возможное продолжение тренда.

Объем растет, но цена не меняется → возможен разворот.

Используется для скальпинга, арбитража и следования за трендом🚀

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "1m"
LOOKBACK_PERIOD = 20  # Количество свечей для расчета среднего объема
SPIKE_THRESHOLD = 2  # Минимальный коэффициент всплеска
CHECK_INTERVAL = 10  # Проверка раз в 10 секунд

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение данных
def fetch_ohlcv(symbol):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=TIMEFRAME, limit=LOOKBACK_PERIOD + 5)
    df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
    return df

# Анализ всплесков объема
def detect_volume_spike():
    df = fetch_ohlcv(PAIR)
    
    avg_volume = df["volume"].iloc[-LOOKBACK_PERIOD:-1].mean()  # Средний объем за N свечей
    current_volume = df["volume"].iloc[-1]  # Объем текущей свечи
    spike_ratio = current_volume / avg_volume if avg_volume > 0 else 0

    print(f"📊 Текущий объем: {current_volume:.2f}, Средний объем: {avg_volume:.2f}, Коэффициент всплеска: {spike_ratio:.2f}")

    if spike_ratio > SPIKE_THRESHOLD:
        print("🔥 Всплеск объема! Возможное сильное движение.")
    else:
        print("⚡ Объем в норме.")

# Запуск
if __name__ == "__main__":
    while True:
        detect_volume_spike()
        time.sleep(CHECK_INTERVAL)

#торговые_стратегии

📌 Подпишись  Crypto Python❗️

📌 Market Depth Analyzer – Анализ глубины рынка 🧑‍💻

Смысл стратегии:

Глубина рынка показывает, сколько ликвидности доступно на разных уровнях стакана.

Глубокий рынок → меньше проскальзывания, стабильные цены.

Мелкий рынок → высокое проскальзывание, возможны резкие движения.

Анализ глубины стакана помогает выявлять уровни поддержки и сопротивления🚀

import ccxt
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
ORDER_BOOK_DEPTH = 20  # Количество уровней в стакане
CHECK_INTERVAL = 5  # Проверка раз в 5 секунд

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение стакана заявок
def get_order_book(symbol):
    try:
        return exchange.fetch_order_book(symbol, limit=ORDER_BOOK_DEPTH)
    except Exception as e:
        print(f"❌ Ошибка получения стакана с {exchange.id}: {e}")
        return None

# Анализ глубины рынка
def analyze_market_depth():
    order_book = get_order_book(PAIR)
    if not order_book:
        return

    bids = order_book["bids"]  # Лимитные покупки (цена, объем)
    asks = order_book["asks"]  # Лимитные продажи (цена, объем)

    # Рассчитываем общий объем ликвидности
    bid_depth = sum([volume for price, volume in bids])
    ask_depth = sum([volume for price, volume in asks])

    # Дисбаланс ликвидности
    imbalance = (bid_depth - ask_depth) / (bid_depth + ask_depth)

    print(f"📊 Глубина рынка для {PAIR}:")
    print(f"🔹 Общий объем BID (покупки): {bid_depth:.2f} BTC")
    print(f"🔸 Общий объем ASK (продажи): {ask_depth:.2f} BTC")
    print(f"📉 Дисбаланс: {imbalance:.4f}")

    if imbalance > 0.1:
        print("📈 Преобладает спрос → возможный рост цены.")
    elif imbalance < -0.1:
        print("📉 Преобладает предложение → возможное снижение цены.")
    else:
        print("⚡ Баланс спроса и предложения.")

# Запуск
if __name__ == "__main__":
    while True:
        analyze_market_depth()
        time.sleep(CHECK_INTERVAL)

#индикатор

📌 Подпишись  Crypto Python❗️

📌 Fear and Greed Index Analyzer – Торговля по индексу страха и жадности🧑‍💻

Смысл стратегии:

Индекс страха и жадности (Fear & Greed Index) анализирует настроение рынка:

0-25 → Страх (Fear) → возможная перепроданность → поиск покупок.

25-50 → Нейтральный рынок → нет четкого направления.

50-75 → Жадность (Greed) → рынок перегрет, но тренд может продолжаться.

75-100 → Крайняя жадность (Extreme Greed) → возможная перекупленность → поиск продаж.


⚠️ Важно: Fear & Greed Index лучше работает на среднесрочных таймфреймах (4H, 1D) для определения глобального настроения рынка. 🚀

import ccxt
import requests
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
CHECK_INTERVAL = 3600  # Проверка раз в час
FEAR_GREED_API = "https://api.alternative.me/fng/"  # Источник данных

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение текущего индекса страха и жадности
def get_fear_greed_index():
    try:
        response = requests.get(FEAR_GREED_API)
        data = response.json()
        index_value = int(data["data"][0]["value"])
        return index_value
    except Exception as e:
        print(f"❌ Ошибка получения индекса: {e}")
        return None

# Получение текущей цены BTC
def get_current_price():
    ticker = exchange.fetch_ticker(PAIR)
    return ticker["last"]

# Логика торговли по индексу
def fear_greed_trading():
    index_value = get_fear_greed_index()
    if index_value is None:
        return

    price = get_current_price()

    print(f"📊 Индекс страха и жадности: {index_value}")
    print(f"💰 Текущая цена {PAIR}: {price:.2f} USDT")

    if index_value <= 25:
        print("📈 Рынок в страхе → Возможная точка входа (покупка).")
    elif index_value >= 75:
        print("📉 Рынок в жадности → Возможная точка выхода (продажа).")
    else:
        print("⚡ Рынок нейтральный, ждем сигнал.")

# Запуск
if __name__ == "__main__":
    while True:
        fear_greed_trading()
        time.sleep(CHECK_INTERVAL)

#индикатор

📌 Подпишись  Crypto Python❗️

📌 Open Interest Analyzer – Анализ открытого интереса на фьючерсах🧑‍💻

Смысл стратегии:

Открытый интерес (Open Interest, OI) – это общее количество открытых контрактов на фьючерсном рынке.

Рост OI → новые деньги входят в рынок → усиление текущего тренда.

Падение OI → закрытие позиций → возможный разворот тренда.

Используется в фьючерсной торговле для анализа силы тренда и настроения рынка🚀

import ccxt
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
CHECK_INTERVAL = 60  # Проверка раз в минуту

# Инициализация биржи
exchange = ccxt.binance({
    "options": {"defaultType": "future"}  # Работаем с фьючерсами
})

# Получение открытого интереса
def get_open_interest(symbol):
    try:
        oi_data = exchange.fapiPublicGetOpenInterest({"symbol": symbol.replace("/", "")})
        return float(oi_data["openInterest"])
    except Exception as e:
        print(f"❌ Ошибка получения OI: {e}")
        return None

# Получение текущей цены
def get_current_price(symbol):
    ticker = exchange.fetch_ticker(symbol)
    return ticker["last"]

# Логика анализа OI
def open_interest_analyzer():
    prev_oi = None

    while True:
        oi = get_open_interest(PAIR)
        price = get_current_price(PAIR)

        if oi is None:
            time.sleep(CHECK_INTERVAL)
            continue

        print(f"📊 Цена: {price:.2f} USDT | Открытый интерес: {oi:.2f}")

        if prev_oi is not None:
            delta_oi = oi - prev_oi
            print(f"🔍 Изменение OI: {delta_oi:.2f}")

            if delta_oi > 0:
                print("📈 OI растет → В рынок заходят новые деньги, тренд усиливается.")
            elif delta_oi < 0:
                print("📉 OI падает → Закрытие позиций, возможный разворот.")

        prev_oi = oi
        time.sleep(CHECK_INTERVAL)

# Запуск
if __name__ == "__main__":
    open_interest_analyzer()

#торговые_стратегии

📌 Подпишись  Crypto Python❗️

📌 Hidden Order Placement – Алгоритм скрытых ордеров для больших объемов🧑‍💻

Смысл стратегии:

Когда крупный ордер размещается в стакане, он может влиять на цену и привлекать внимание других трейдеров.

Этот алгоритм разбивает ордер на мелкие части и размещает их случайно в течение времени, имитируя естественные рыночные покупки/продажи.

import ccxt
import time
import random

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TOTAL_AMOUNT = 5  # Общий объем ордера (BTC)
SPLIT_ORDERS = 20  # Количество частей
MIN_WAIT = 10  # Минимальная задержка между ордерами (сек)
MAX_WAIT = 60  # Максимальная задержка между ордерами (сек)

# API-ключи (замените своими)
API_KEY = "your_api_key"
API_SECRET = "your_api_secret"

# Инициализация биржи
exchange = ccxt.binance({
    "apiKey": API_KEY,
    "secret": API_SECRET,
    "enableRateLimit": True,
})

# Получение текущей цены
def get_current_price(symbol):
    ticker = exchange.fetch_ticker(symbol)
    return ticker["last"]

# Размещение скрытых ордеров
def hidden_order_execution():
    part_size = TOTAL_AMOUNT / SPLIT_ORDERS  # Размер одной сделки

    for i in range(SPLIT_ORDERS):
        price = get_current_price(PAIR)
        print(f"📥 Размещение ордера {i+1}/{SPLIT_ORDERS}: Покупка {part_size:.4f} BTC по {price:.2f} USDT")

        try:
            order = exchange.create_limit_buy_order(PAIR, part_size, price)
            print(f"✅ Ордер размещен: {order}")
        except Exception as e:
            print(f"❌ Ошибка при размещении ордера: {e}")

        wait_time = random.randint(MIN_WAIT, MAX_WAIT)
        print(f"⏳ Ждем {wait_time} секунд перед следующим ордером...")
        time.sleep(wait_time)

# Запуск
if __name__ == "__main__":
    hidden_order_execution()

#ордера

📌 Подпишись  Crypto Python❗️

📌 Cumulative Delta Strategy – Стратегия на основе кумулятивной дельты объема🧑‍💻

Смысл стратегии:

Кумулятивная дельта объема (CVD, Cumulative Volume Delta) — это накопленная разница между объемом рыночных покупок и продаж.

CVD растет → в рынок входят рыночные покупатели (бычий сигнал).

CVD падает → доминируют рыночные продавцы (медвежий сигнал).

Дивергенция между ценой и CVD → возможный разворот.


🔹 Применение стратегии:🚀

CVD + цена растут вместе → подтверждение восходящего тренда (лонг).

CVD падает, а цена растёт → дивергенция, возможный разворот вниз.

CVD растёт, а цена падает → скрытый спрос, возможный разворот вверх.

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "1m"
CVD_LENGTH = 100  # Количество последних трейдов
CHECK_INTERVAL = 10  # Проверка каждые 10 секунд

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение последних сделок
def fetch_trades(symbol):
    trades = exchange.fetch_trades(symbol, limit=CVD_LENGTH)
    df = pd.DataFrame(trades)
    return df

# Расчет кумулятивной дельты
def calculate_cvd(trades_df):
    trades_df['delta'] = trades_df.apply(
        lambda row: row['amount'] if row['side'] == 'buy' else -row['amount'], axis=1
    )
    trades_df['cvd'] = trades_df['delta'].cumsum()
    return trades_df['cvd'].iloc[-1], trades_df['cvd']

# Получение текущей цены
def get_price(symbol):
    return exchange.fetch_ticker(symbol)['last']

# Основной цикл стратегии
def cumulative_delta_strategy():
    prev_cvd = None
    while True:
        trades_df = fetch_trades(PAIR)
        current_price = get_price(PAIR)
        cvd_value, cvd_series = calculate_cvd(trades_df)

        print(f"Цена: {current_price:.2f} USDT | CVD: {cvd_value:.4f}")

        # Примитивная логика входа на основе CVD
        if prev_cvd is not None:
            delta_cvd = cvd_value - prev_cvd

            if delta_cvd > 0:
                print("📈 Кумулятивная дельта растёт → доминируют покупатели (лонг сигнал).")
            elif delta_cvd < 0:
                print("📉 Кумулятивная дельта падает → доминируют продавцы (шорт сигнал).")
            else:
                print("⚖️ Баланс между покупателями и продавцами.")

        prev_cvd = cvd_value
        time.sleep(CHECK_INTERVAL)

# Запуск
if __name__ == "__main__":
    cumulative_delta_strategy()

#индикатор

📌 Подпишись  Crypto Python❗️

📌 Выявление скрытых крупных заявок (Iceberg Orders)🧑‍💻

Что это и зачем?

Iceberg-заявки — это ордера, которые выглядят как маленькие, но при исполнении обновляются, показывая новую "скрытую" часть.

🔥 Как это использовать?

📌 Если Iceberg в BID → крупный игрок накапливает позицию, рост вероятен

📌 Если Iceberg в ASK → крупный игрок разгружает позицию, может быть падение

import requests
import time

# API Bybit (глубина стакана)
ORDER_BOOK_URL = "https://api.bybit.com/v5/market/orderbook"
SYMBOL = "BTCUSDT"
DEPTH = 50  # Берём топ-50 заявок

def get_order_book():
    """Получает стакан заявок"""
    params = {"category": "spot", "symbol": SYMBOL, "limit": DEPTH}
    response = requests.get(ORDER_BOOK_URL, params=params)
    if response.status_code == 200:
        data = response.json().get("result", {})
        bids = [(float(price), float(size)) for price, size in data.get("b", [])]
        asks = [(float(price), float(size)) for price, size in data.get("a", [])]
        return bids, asks
    else:
        print(f"Ошибка API: {response.status_code}")
        return None, None

def detect_iceberg_orders():
    """Выявляет скрытые Iceberg-заявки"""
    prev_bids, prev_asks = get_order_book()
    if not prev_bids or not prev_asks:
        print("❌ Ошибка получения стакана.")
        return

    print("📡 Начинаем мониторинг Iceberg-заявок...")
    while True:
        time.sleep(2)  # Ждём обновления стакана
        bids, asks = get_order_book()
        if not bids or not asks:
            continue

        # Анализ заявок (BID - покупатели)
        for i in range(min(len(bids), len(prev_bids))):
            if bids[i][0] == prev_bids[i][0] and bids[i][1] > prev_bids[i][1] * 1.5:
                print(f"❄️ Iceberg-заявка в BID: {bids[i][0]} - новый объём {bids[i][1]} BTC (было {prev_bids[i][1]} BTC)")

        # Анализ заявок (ASK - продавцы)
        for i in range(min(len(asks), len(prev_asks))):
            if asks[i][0] == prev_asks[i][0] and asks[i][1] > prev_asks[i][1] * 1.5:
                print(f"❄️ Iceberg-заявка в ASK: {asks[i][0]} - новый объём {asks[i][1]} BTC (было {prev_asks[i][1]} BTC)")

        prev_bids, prev_asks = bids, asks  # Обновляем стакан

if __name__ == "__main__":
    detect_iceberg_orders()

#индикатор

📌 Подпишись  Crypto Python❗️

📌 Whale Tracker – Отслеживание крупных сделок и ордеров в стакане🧑‍💻

Смысл стратегии:

Whale Tracker анализирует поток сделок и стакан заявок, чтобы обнаружить действия крупных игроков («китов»), которые могут вызвать движение цены.

Крупные рыночные сделки → сильное движение, импульс.

Крупные лимитные ордера → вероятные уровни поддержки/сопротивления.

Используется для скальпинга, интрадей и подтверждения трендов🚀

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
MIN_TRADE_SIZE = 10  # Минимальный объем сделки (в BTC), чтобы считать её «китовой»
MIN_ORDER_VOLUME = 50  # Объем заявки в стакане для отслеживания
CHECK_INTERVAL = 5  # Проверка раз в 5 секунд

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение последних сделок
def fetch_large_trades():
    trades = exchange.fetch_trades(PAIR, limit=100)
    df = pd.DataFrame(trades)
    large_trades = df[df['amount'] >= MIN_TRADE_SIZE]

    for _, row in large_trades.iterrows():
        side = row['side'].upper()
        print(f"🐋 Крупная {side} сделка: {row['amount']} BTC по {row['price']} USDT")

# Анализ стакана ордеров
def check_order_book():
    order_book = exchange.fetch_order_book(PAIR, limit=50)
    bids = order_book["bids"]
    asks = order_book["asks"]

    print("🔍 Поиск крупных заявок в стакане...")
    for price, volume in bids[:10]:
        if volume >= MIN_ORDER_VOLUME:
            print(f"⬇️ BID: {volume:.2f} BTC на {price:.2f} USDT")

    for price, volume in asks[:10]:
        if volume >= MIN_ORDER_VOLUME:
            print(f"⬆️ ASK: {volume:.2f} BTC на {price:.2f} USDT")

# Основной цикл
def whale_tracker():
    while True:
        print("\n--- Обновление ---")
        fetch_large_trades()
        check_order_book()
        time.sleep(CHECK_INTERVAL)

# Запуск
if __name__ == "__main__":
    whale_tracker()

#индикатор

📌 Подпишись  Crypto Python❗️

📌 VWAP Order Execution – Исполнение крупных ордеров по VWAP🧑‍💻

Смысл стратегии:

VWAP (Volume Weighted Average Price) – это средняя цена, по которой актив торговался с учётом объёма.

Цель стратегии:

Исполнить большой ордер максимально близко к VWAP,

Избежать проскальзывания,

Не влиять на рынок, как при рыночной сделке.

🔹 VWAP Execution = Разбиение большого ордера на части, распределённых во времени и ориентированных на текущую VWAP-цену.

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TOTAL_AMOUNT = 1  # Всего купить 1 BTC
SPLIT = 10  # Разбить на 10 частей
TIMEFRAME = "1m"
VWAP_PERIOD = 20
INTERVAL = 60  # Интервал между проверками (сек)

# API-ключи (если нужно реальное исполнение)
API_KEY = "your_api_key"
API_SECRET = "your_api_secret"

# Биржа
exchange = ccxt.binance({
    "apiKey": API_KEY,
    "secret": API_SECRET,
    "enableRateLimit": True,
})

# Получение данных
def fetch_ohlcv(symbol):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=TIMEFRAME, limit=VWAP_PERIOD + 5)
    df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
    return df

# Расчет VWAP
def calculate_vwap(df):
    tp = (df["high"] + df["low"] + df["close"]) / 3
    vwap = (tp * df["volume"]).cumsum() / df["volume"].cumsum()
    return vwap.iloc[-1]

# Получение текущей цены
def get_price(symbol):
    return exchange.fetch_ticker(symbol)["last"]

# Основной цикл
def vwap_order_execution():
    part_size = TOTAL_AMOUNT / SPLIT
    executed = 0

    while executed < SPLIT:
        df = fetch_ohlcv(PAIR)
        vwap = calculate_vwap(df)
        price = get_price(PAIR)

        print(f"Цена: {price:.2f} | VWAP: {vwap:.2f}")

        if price <= vwap * 1.002:  # Допускаем отклонение 0.2% вверх
            print(f"📥 Покупка части {executed + 1}/{SPLIT} по цене {price:.2f}")
            try:
                order = exchange.create_limit_buy_order(PAIR, part_size, price)
                print("✅ Ордер отправлен:", order["id"])
                executed += 1
            except Exception as e:
                print("❌ Ошибка размещения ордера:", e)
        else:
            print("⏳ Цена выше VWAP, ждем...")

        time.sleep(INTERVAL)

# Запуск
if __name__ == "__main__":
    vwap_order_execution()

#ордера

📌 Подпишись  Crypto Python❗️

📌 Support/Resistance Auto Detector – Автоматическое определение уровней поддержки и сопротивления🧑‍💻

Смысл стратегии:

Автоматическое определение ключевых уровней, где:

Поддержка — цена неоднократно отскакивала вверх.

Сопротивление — цена не могла пробить уровень вверх.

Подобные уровни используются для:

- Входа по отскоку или на пробой.

- Установки стоп-лоссов и тейк-профитов.

- Фильтрации сигналов в стратегиях.

import ccxt
import pandas as pd
import numpy as np
import time

# Параметры
EXCHANGE = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
LOOKBACK = 200
TOLERANCE = 0.005  # 0.5% допуск для объединения уровней
TOUCHES_MIN = 2    # Минимум касаний уровня

exchange = getattr(ccxt, EXCHANGE)()

def fetch_ohlcv():
    ohlcv = exchange.fetch_ohlcv(PAIR, timeframe=TIMEFRAME, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
    return df

def is_level_touched(level, price):
    return abs(level - price) / price < TOLERANCE

def detect_levels(df):
    highs = df["high"]
    lows = df["low"]
    levels = []

    # Проверка локальных максимумов и минимумов
    for i in range(2, len(df) - 2):
        if highs[i] > highs[i - 1] and highs[i] > highs[i + 1]:
            level = highs[i]
        elif lows[i] < lows[i - 1] and lows[i] < lows[i + 1]:
            level = lows[i]
        else:
            continue

        # Проверка, не существует ли уже близкий уровень
        if not any(is_level_touched(existing, level) for existing in levels):
            levels.append(level)

    # Группировка уровней по количеству касаний
    grouped = {}
    for lvl in levels:
        count = sum(
            df["high"].apply(lambda x: is_level_touched(lvl, x)).sum() +
            df["low"].apply(lambda x: is_level_touched(lvl, x)).sum()
        )
        if count >= TOUCHES_MIN:
            grouped[round(lvl, 2)] = count

    return grouped

# Основной запуск
if __name__ == "__main__":
    df = fetch_ohlcv()
    levels = detect_levels(df)

    print(f"\n📊 Автоматически найденные уровни поддержки/сопротивления для {PAIR}:")
    for level, touches in sorted(levels.items()):
        print(f"• Уровень: {level:.2f} USDT | Касаний: {touches}")

#индикатор

📌 Подпишись  Crypto Python❗️

📌 Breakout Retest Strategy – вход после подтверждённого пробоя и возврата к уровню🧑‍💻

Смысл стратегии:

Стратегия основана на классической логике технического анализа:

1. Цена пробивает уровень сопротивления/поддержки.

2. Возвращается к этому уровню (ретест).

3. Уровень удерживается, и начинается новое движение.
→ Вход происходит на ретесте после подтверждённого пробоя.

import ccxt
import pandas as pd
import numpy as np
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
LOOKBACK = 100  # Количество свечей для анализа
LEVEL_TOLERANCE = 0.003  # Допуск к уровню (0.3%)
CHECK_INTERVAL = 60  # сек

exchange = getattr(ccxt, EXCHANGE_NAME)()

def fetch_data():
    ohlcv = exchange.fetch_ohlcv(PAIR, timeframe=TIMEFRAME, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    return df

def find_breakout_retest(df):
    # Ищем локальные максимумы как потенциальные уровни
    highs = df['high'].rolling(window=5, center=True).apply(lambda x: x[2] if x[2] == max(x) else np.nan)
    levels = highs.dropna().round(2).unique()

    last_candle = df.iloc[-1]
    prev_candle = df.iloc[-2]

    for level in levels:
        tolerance = level * LEVEL_TOLERANCE

        # Условие: пробой — свеча закрылась выше уровня
        breakout = prev_candle['close'] > level + tolerance and prev_candle['open'] < level

        # Условие: ретест — текущая свеча коснулась уровня и закрылась выше него
        retest = abs(last_candle['low'] - level) < tolerance and last_candle['close'] > level

        if breakout and retest:
            print(f"📈 Найден сигнал Breakout + Retest у уровня {level} на {PAIR} ({TIMEFRAME})")
            return

    print("⚠️ Сигналов не найдено.")

if __name__ == "__main__":
    while True:
        df = fetch_data()
        find_breakout_retest(df)
        time.sleep(CHECK_INTERVAL)

#торговые_стратегии

📌 Подпишись  Crypto Python❗️

📌 Gap Finder – Поиск ценовых гэпов на рынке🧑‍💻

Смысл стратегии:

Ценовой гэп (gap) — это разрыв между свечами, когда цена открытия новой свечи существенно выше или ниже цены закрытия предыдущей.

Гэпы возникают из-за:

- Резких новостей,

- Низкой ликвидности,

- Задержек в обновлении цен (в крипте — реже, но бывают на младших таймфреймах и низкокапитальных активах).

Типы гэпов:

Gap Up — свеча открылась выше предыдущей.

Gap Down — свеча открылась ниже предыдущей.

Открытый гэп — ещё не перекрыт.

Закрытый гэп — цена вернулась в зону разрыва.

Gap Finder используется для 🚀:

1. Обнаружение рыночных аномалий


2. Поиск точек входа по стратегии закрытия гэпа


3. Подтверждение импульса после гэпа


4. Фильтрация нестабильных условий рынка


5. Определение уровней поддержки/сопротивления

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
LOOKBACK = 200
GAP_THRESHOLD = 0.01  # Гэп ≥ 1%

# Инициализация
exchange = getattr(ccxt, EXCHANGE_NAME)()

def fetch_data():
    ohlcv = exchange.fetch_ohlcv(PAIR, timeframe=TIMEFRAME, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    return df

def find_gaps(df):
    gaps = []
    for i in range(1, len(df)):
        prev_close = df.loc[i-1, 'close']
        curr_open = df.loc[i, 'open']
        gap_percent = abs(curr_open - prev_close) / prev_close

        if gap_percent >= GAP_THRESHOLD:
            gap_type = "Gap Up" if curr_open > prev_close else "Gap Down"
            gaps.append({
                "time": df.loc[i, 'timestamp'],
                "gap_%": round(gap_percent * 100, 2),
                "type": gap_type,
                "from": prev_close,
                "to": curr_open
            })
    return gaps

# Основной запуск
if __name__ == "__main__":
    df = fetch_data()
    gaps = find_gaps(df)

    print(f"\n📉 Найдено {len(gaps)} гэпов на {PAIR} ({TIMEFRAME}):\n")
    for g in gaps:
        print(f"• {g['time']} | {g['type']} | {g['gap_%']}% | {g['from']} → {g['to']}")

#торговые_стратегии

📌 Подпишись  Crypto Python❗️

📌 Funding Rate Arbitrage-использование положительных funding rate на фьючерсах🧑‍💻

Смысл стратегии:

Фьючерсные рынки (особенно perpetual) используют механизм funding rate для баланса между ценой фьючерса и спота.
Трейдеры платят или получают ежечасную/ежедневную комиссию, в зависимости от их позиции:

Funding Rate положительный → лонги платят шортам

Funding Rate отрицательный → шорты платят лонгам

Как зарабатывать💸:

Бессрисковый арбитраж между спотовым и фьючерсным рынком:

1. Покупка актива на спотовом рынке

2. Открытие шорта на фьючерсах тем же объёмом

3. Получение funding'а каждые 8 ч (Binance, Bybit и др.)

import ccxt
import time

# Параметры
EXCHANGE_NAME = "binance"
SYMBOL = "BTC/USDT"
FUNDING_THRESHOLD = 0.01  # 1% за период — условный минимум для интереса
CHECK_INTERVAL = 60 * 60  # Проверка каждый час

# Инициализация
exchange = ccxt.binance({
    'options': {'defaultType': 'future'}
})

def fetch_funding_rate(symbol):
    try:
        market = exchange.market(symbol)
        symbol_id = market['id']
        rates = exchange.fapiPublic_get_fundingrate({'symbol': symbol_id, 'limit': 1})
        funding = float(rates[0]['fundingRate']) * 100  # в процентах
        time_str = rates[0]['fundingTime']
        return funding, time_str
    except Exception as e:
        print("Ошибка:", e)
        return None, None

# Запуск
if __name__ == "__main__":
    while True:
        rate, ftime = fetch_funding_rate(SYMBOL)
        if rate is not None:
            print(f"📊 Funding Rate для {SYMBOL}: {rate:.4f}% | Время: {ftime}")
            if rate >= FUNDING_THRESHOLD:
                print("💡 Возможность арбитража: открыть SHORT + держать спот LONG")
        time.sleep(CHECK_INTERVAL)

#торговые_стратегии

📌 Подпишись  Crypto Python❗️