📌 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❗️

📌 Opening Range Breakout – Стратегия пробоя утреннего диапазона🧑‍💻

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

Opening Range Breakout (ORB) — это стратегия, основанная на диапазоне цен в первые 30-60 минут сессии. Она предполагает, что рынок будет двигаться в том направлении, в котором произойдёт пробой утреннего диапазона.

Логика стратегии:

1. Утренний диапазон (Opening Range):

Рассчитывается как разница между ценой открытия и ценой максимума/минимума в первые 30-60 минут.

2. Пробой вверх (Buy):

Если цена пробивает верхнюю границу диапазона, открываем лонг.

3. Пробой вниз (Sell):

Если цена пробивает нижнюю границу диапазона, открываем шорт.

4. Цели и стопы:

Стоп-лосс часто устанавливается небольшим и пропорциональным диапазону ORB.

Тейк-профит можно установить в зависимости от размера диапазона ORB или использовать Trailing Stop.

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "1m"
LOOKBACK = 60  # 60 минут для расчета диапазона
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'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    return df

# Определение утреннего диапазона и сигналов
def opening_range_breakout(df):
    # Утренний диапазон: первые 60 минут (можно настроить на 30 минут)
    morning_range = df.head(60)

    high_range = morning_range['high'].max()
    low_range = morning_range['low'].min()
    open_price = df.iloc[0]['open']
    current_price = df.iloc[-1]['close']

    print(f"📊 Утренний диапазон: {low_range:.2f} - {high_range:.2f}")
    print(f"Текущая цена: {current_price:.2f}")

    if current_price > high_range:
        print("📈 Пробой вверх: открытие лонга!")
        # Здесь можно добавить код для размещения ордера на покупку
    elif current_price < low_range:
        print("📉 Пробой вниз: открытие шорта!")
        # Здесь можно добавить код для размещения ордера на продажу
    else:
        print("⚡ Нет сигнала для входа, ожидаем пробой.")

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

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

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

📌 Index Basket Tracker – Построение собственного индекса из группы токенов🧑‍💻

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

Индекс корзины токенов — это пользовательский портфель, в котором несколько активов объединяются в единый взвешенный индекс.

Можно отслеживать:

Суммарную доходность группы монет,

Сравнение производительности с BTC/ETH,

Использовать как ориентир для входа в рынок, хеджирования или торговли по тренду всей ниши (например, DeFi, AI, GameFi и т.д.).

import ccxt
import pandas as pd
import time

# Настройки
EXCHANGE = ccxt.binance()
TIMEFRAME = '1h'
TOKENS = {
    'BTC/USDT': 0.4,
    'ETH/USDT': 0.3,
    'SOL/USDT': 0.2,
    'LINK/USDT': 0.1
}  # Токены и веса (сумма = 1.0)

INDEX_NAME = 'Custom Index'

def fetch_prices():
    prices = {}
    for symbol in TOKENS.keys():
        ticker = EXCHANGE.fetch_ticker(symbol)
        prices[symbol] = ticker['last']
    return prices

def calculate_index(prices, base_prices):
    returns = {}
    index_value = 0
    for symbol, weight in TOKENS.items():
        change = prices[symbol] / base_prices[symbol]
        returns[symbol] = round((change - 1) * 100, 2)
        index_value += change * weight
    index_change = round((index_value - 1) * 100, 2)
    return index_value, index_change, returns

# Запуск
if __name__ == '__main__':
    print(f"🔄 Получаем начальные цены для индекса '{INDEX_NAME}'...")
    base_prices = fetch_prices()
    print("Базовые цены зафиксированы. Индекс отслеживается...\n")

    while True:
        current_prices = fetch_prices()
        index_val, index_delta, token_returns = calculate_index(current_prices, base_prices)

        print(f"\n📊 {INDEX_NAME} → Значение: {index_val:.3f} | Изменение: {index_delta}%")
        for sym, ret in token_returns.items():
            print(f"• {sym}: {ret}%")

        time.sleep(60 * 5)  # обновление каждые 5 минут

#индикатор

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

📌 Correlation Divergence Finder – Поиск расхождений у коррелированных пар 🧑‍💻

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

Некоторые активы исторически движутся синхронно (например, ETH/BTC, SOL/AVAX, MATIC/ARB).
Дивергенция коррелированных пар возникает, когда:

Один из активов отклоняется от общего направления,

Возникает возможность для парной торговли (long/short):
→ шорт перекупленного, лонг отстающего🚀

import ccxt
import pandas as pd
import time

# Настройки
EXCHANGE = ccxt.binance()
PAIR_1 = 'ETH/USDT'
PAIR_2 = 'BNB/USDT'
TIMEFRAME = '1h'
LOOKBACK = 100
DIVERGENCE_THRESHOLD = 0.03  # 3% расхождение

def fetch_data(pair):
    ohlcv = EXCHANGE.fetch_ohlcv(pair, timeframe=TIMEFRAME, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['return'] = df['close'].pct_change()
    df.dropna(inplace=True)
    return df

def check_divergence():
    df1 = fetch_data(PAIR_1)
    df2 = fetch_data(PAIR_2)

    # Краткосрочная доходность (последние 5 свечей)
    ret1 = df1['return'].tail(5).sum()
    ret2 = df2['return'].tail(5).sum()
    diff = ret1 - ret2

    print(f"\n📊 Доходность за 5ч:")
    print(f"{PAIR_1}: {ret1*100:.2f}%, {PAIR_2}: {ret2*100:.2f}% | Расхождение: {diff*100:.2f}%")

    if abs(diff) >= DIVERGENCE_THRESHOLD:
        lead = PAIR_1 if diff > 0 else PAIR_2
        lag = PAIR_2 if diff > 0 else PAIR_1
        print(f"⚠️ Обнаружена дивергенция! {lead} опережает {lag}")
        print("💡 Возможность парной торговли: Long отстающий, Short лидирующий")
    else:
        print("✅ Нет значимого расхождения")

if __name__ == '__main__':
    while True:
        check_divergence()
        time.sleep(60 * 5)  # обновление каждые 5 минут

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

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

📌 Trendline Breakout Detector – Определение пробоев вручную построенных трендлайнов

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

Инструмент отслеживает пробои пользовательских (вручную заданных) трендлайнов:

Восходящая линия тренда (поддержка)

Нисходящая линия тренда (сопротивление)

→ При пробое таких линий можно ловить развороты или импульсы по тренду🚀

import ccxt
import pandas as pd
import time
from datetime import datetime

# Настройки
EXCHANGE = ccxt.binance()
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
LOOKBACK = 100

# Трендлайн: 2 точки (индекс свечи от конца, цена)
# Пример: точка A — 20 свечей назад по цене 30000, точка B — 5 свечей назад по цене 35000
x1, y1 = -20, 30000
x2, y2 = -5, 35000

# Уравнение тренда: y = kx + b
def compute_trendline(x1, y1, x2, y2):
    k = (y2 - y1) / (x2 - x1)
    b = y1 - k * x1
    return k, b

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 check_breakout(df, k, b):
    current_idx = len(df) - 1
    x = 0  # Текущая свеча (x=0 соответствует последнему бару)
    trend_price = k * x + b
    current_price = df["close"].iloc[-1]

    print(f"\nТекущая цена: {current_price:.2f} | Линия тренда: {trend_price:.2f}")

    if current_price > trend_price:
        print("📈 Пробой ВЫШЕ линии тренда!")
    elif current_price < trend_price:
        print("📉 Пробой НИЖЕ линии тренда!")
    else:
        print("⚖️ Цена около линии тренда (нет сигнала)")

if __name__ == "__main__":
    print(f"Запущен трекер пробоя трендлайна на {PAIR} [{TIMEFRAME}]")
    k, b = compute_trendline(x1, y1, x2, y2)

    while True:
        df = fetch_data()
        check_breakout(df, k, b)
        time.sleep(60)

#индикатор

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

📌 Premium Index Arbitrage – Спотово-фьючерсная стратегия🧑‍💻

Суть стратегии:

На фьючерсных рынках цена может отличаться от спотовой (особенно на perpetual futures).
Эта разница называется премией (premium).

Стратегия использует эти расхождения:

Фьючерс дороже спота → продать фьючерс, купить спот.💸

Фьючерс дешевле спота → купить фьючерс, продать спот.💸

import ccxt
import time

# Настройки
PAIR = "BTC/USDT"
THRESHOLD = 0.5  # В % — минимальная премия для сигнала

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

def get_prices():
    spot_price = spot.fetch_ticker(PAIR)['last']
    fut_price = futures.fetch_ticker(PAIR)['last']
    premium = (fut_price - spot_price) / spot_price * 100
    return spot_price, fut_price, premium

while True:
    spot_price, fut_price, premium = get_prices()

    print(f"\n📊 Spot: {spot_price:.2f} | Futures: {fut_price:.2f} | Premium: {premium:.2f}%")

    if premium >= THRESHOLD:
        print("📉 Фьючерс перекуплен → ПРОДАЖА фьючерса / ПОКУПКА спота")
    elif premium <= -THRESHOLD:
        print("📈 Фьючерс перепродан → ПОКУПКА фьючерса / ПРОДАЖА спота")
    else:
        print("✅ Премия в норме, без сигнала")

    time.sleep(60)

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

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

📌 Volume-Based Reversal – Разворотная стратегия на основе резкого объёма🧑‍💻

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

Эта стратегия ищет точки разворота тренда, основываясь на всплесках объёма, когда:

Цена достигает локального экстремума (максимума или минимума),

И одновременно происходит аномальный объём,
→ Это может свидетельствовать о развороте или истощении тренда🚀

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE = ccxt.binance()
PAIR = "BTC/USDT"
TIMEFRAME = "15m"
LOOKBACK = 100
VOLUME_MULTIPLIER = 2.0

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 detect_volume_reversal(df):
    avg_volume = df['volume'][:-1].mean()
    last = df.iloc[-1]
    prev = df.iloc[-2]

    # Условие: свеча с большим объемом и возможный разворот
    if last['volume'] > avg_volume * VOLUME_MULTIPLIER:
        # Локальный минимум
        if last['low'] < prev['low'] and last['close'] > last['open']:
            print(f"📈 Объёмный разворот вверх! Цена: {last['close']:.2f}, Объём: {last['volume']:.2f}")
        # Локальный максимум
        elif last['high'] > prev['high'] and last['close'] < last['open']:
            print(f"📉 Объёмный разворот вниз! Цена: {last['close']:.2f}, Объём: {last['volume']:.2f}")
        else:
            print("⚠️ Всплеск объёма, но нет явного сигнала.")
    else:
        print("✅ Объём в норме, сигналов нет.")

if __name__ == "__main__":
    while True:
        df = fetch_data()
        detect_volume_reversal(df)
        time.sleep(60 * 5)

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

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

📌 Price Action Pattern Bot – поиск свечных паттернов 🧑‍💻

Суть стратегии:

Бот ищет паттерны price action на основе анализа свечей — без индикаторов.

Используется для определения точек входа/разворота на чистом графике.

1. Пин-бар (Pin Bar) — свеча с длинным хвостом и маленьким телом (разворот).

2. Поглощение (Engulfing) — одна свеча полностью перекрывает предыдущую.

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE = ccxt.binance()
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
LOOKBACK = 100

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 is_pin_bar(candle, threshold=0.6):
    body = abs(candle['close'] - candle['open'])
    tail = candle['high'] - candle['low']
    upper_wick = candle['high'] - max(candle['close'], candle['open'])
    lower_wick = min(candle['close'], candle['open']) - candle['low']

    # длинный хвост минимум в X раз больше тела
    return (
        tail > 0 and body / tail < 0.3 and
        (upper_wick / tail > threshold or lower_wick / tail > threshold)
    )

# Поглощение: тело полностью покрывает предыдущее
def is_engulfing(curr, prev):
    return (
        (curr['open'] < curr['close'] and prev['open'] > prev['close'] and
         curr['open'] < prev['close'] and curr['close'] > prev['open']) or
        (curr['open'] > curr['close'] and prev['open'] < prev['close'] and
         curr['open'] > prev['close'] and curr['close'] < prev['open'])
    )

def detect_patterns(df):
    last = df.iloc[-1]
    prev = df.iloc[-2]

    if is_pin_bar(last):
        print(f"📍 Найден пин-бар на {PAIR}")
    if is_engulfing(last, prev):
        print(f"🔥 Найден паттерн поглощения на {PAIR}")

# Запуск
if __name__ == "__main__":
    while True:
        df = fetch_data()
        detect_patterns(df)
        time.sleep(60 * 5)

#патерн

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

📌 Cross Exchange Spread Monitor – Мониторинг спредов между биржами🧑‍💻

Суть стратегии:

Мониторинг разницы цен (спреда) одного и того же актива на двух и более биржах.

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

- Арбитража (если спред покрывает комиссии),

- Поиска расхождений в ликвидности,

- Реалтайм-анализа рыночной неэффективности.

import ccxt
import time

# Настройки
SYMBOL = "BTC/USDT"
THRESHOLD = 0.5  # в % — минимальный спред для сигнала
INTERVAL = 10  # сек

# Инициализация бирж
binance = ccxt.binance()
kucoin = ccxt.kucoin()

def get_price(exchange, symbol):
    try:
        return exchange.fetch_ticker(symbol)["last"]
    except Exception as e:
        print(f"{exchange.id} error:", e)
        return None

def check_spread():
    price_binance = get_price(binance, SYMBOL)
    price_kucoin = get_price(kucoin, SYMBOL)

    if price_binance and price_kucoin:
        spread = abs(price_binance - price_kucoin)
        spread_pct = spread / ((price_binance + price_kucoin) / 2) * 100

        print(f"\nBinance: {price_binance:.2f} | KuCoin: {price_kucoin:.2f} | Spread: {spread_pct:.2f}%")

        if spread_pct >= THRESHOLD:
            more = "Binance" if price_binance > price_kucoin else "KuCoin"
            print(f"⚡ Возможность арбитража: {more} дороже на {spread_pct:.2f}%")

if __name__ == "__main__":
    while True:
        check_spread()
        time.sleep(INTERVAL)

#арбитраж

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

📌 Latency Arbitrage – Торговля с преимуществом по задержке (задержка котировок)🧑‍💻

Суть стратегии:🚀

Latency Arbitrage использует разницу во времени обновления котировок между двумя биржами (или спот/фьючерс), чтобы:

Быстро купить там, где цена ещё не обновилась,

Продать там, где цена уже ушла вверх/вниз.

Часто применяется в:

- HFT (high-frequency trading),

- На рынках с низкой ликвидностью,

- При разной скорости обновления API/WebSocket.

import ccxt
import time

# Настройки
SYMBOL = "BTC/USDT"
THRESHOLD = 0.3  # % расхождения
INTERVAL = 1  # сек

# Биржи (одна медленнее, например)
fast_exchange = ccxt.binance()
slow_exchange = ccxt.kucoin()

def get_price(exchange):
    try:
        return exchange.fetch_ticker(SYMBOL)["last"]
    except:
        return None

def check_latency_arbitrage():
    fast_price = get_price(fast_exchange)
    slow_price = get_price(slow_exchange)

    if fast_price and slow_price:
        spread = fast_price - slow_price
        spread_pct = spread / slow_price * 100

        print(f"\nFast: {fast_price:.2f} | Slow: {slow_price:.2f} | Δ: {spread_pct:.2f}%")

        if abs(spread_pct) >= THRESHOLD:
            direction = "BUY slow / SELL fast" if spread > 0 else "BUY fast / SELL slow"
            print(f"⚡ Возможность арбитража: {direction}")

if __name__ == "__main__":
    while True:
        check_latency_arbitrage()
        time.sleep(INTERVAL)

#арбитраж

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