📌 Выявление скрытых крупных заявок (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❗️

📌 Fibonacci Retracement Trader – Автоматическая торговля по уровням Фибоначчи🧑‍💻

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

Уровни Фибоначчи используются для:

- поиска точек входа на откате,

- установки тейк-профитов,

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

Классические уровни: 0.236, 0.382, 0.5, 0.618, 0.786

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

1. Найти сильный импульс (максимум–минимум на отрезке свечей).

2. Построить сетку уровней Фибоначчи.

3. При возврате цены к ключевому уровню (например, 0.618) — вход в позицию.

4. Выход по 0.382 или 1.0, стоп за 0.786.

import ccxt
import pandas as pd
import time

# Настройки
EXCHANGE = ccxt.binance()
PAIR = 'BTC/USDT'
TIMEFRAME = '1h'
LOOKBACK = 100
FIB_LEVEL = 0.618
CHECK_INTERVAL = 60  # секунд

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 get_fib_levels(df):
    swing_high = df["high"].max()
    swing_low = df["low"].min()
    levels = {
        "0.618": swing_high - (swing_high - swing_low) * 0.618,
        "0.5": swing_high - (swing_high - swing_low) * 0.5,
        "0.382": swing_high - (swing_high - swing_low) * 0.382,
    }
    return levels

def check_entry(df):
    close_price = df["close"].iloc[-1]
    levels = get_fib_levels(df)

    print(f"\nЦена: {close_price:.2f}")
    for level, price in levels.items():
        print(f"Уровень {level}: {price:.2f}")

    if close_price <= levels["0.618"] and close_price > levels["0.5"]:
        print("📥 Сигнал на вход в лонг от уровня 0.618")
    elif close_price >= levels["0.382"] and close_price < levels["0.5"]:
        print("📤 Возможность фиксации прибыли у 0.382")
    else:
        print("⏳ Нет сигнала, цена вне ключевых зон.")

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

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

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

📌 High Volume Doji Strategy – Поиск дожи на высоком объёме🧑‍💻

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

Дожи (Doji) — свеча с очень маленьким телом и длинными тенями, указывающая на нерешённость рынка.
Если дожи формируется на аномально высоком объёме, это может сигнализировать о потенциальном развороте тренда или консолидации перед импульсом.🚀

Условия сигнала:

1. Свеча — дожи: |close − open| очень мал

2. Объём выше среднего (например, в 2 раза)

3. Желательно — на уровне поддержки/сопротивления или после сильного движения

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE = ccxt.binance()
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
LOOKBACK = 100
VOLUME_MULTIPLIER = 2.0
DOJI_BODY_THRESHOLD = 0.001  # 0.1% от цены

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_high_volume_doji(df):
    avg_volume = df["volume"][:-1].mean()
    last = df.iloc[-1]
    body = abs(last["close"] - last["open"])
    body_pct = body / last["open"]
    
    is_doji = body_pct <= DOJI_BODY_THRESHOLD
    is_high_volume = last["volume"] > avg_volume * VOLUME_MULTIPLIER

    print(f"\nЦена: {last['close']:.2f}, Объём: {last['volume']:.2f}, Тело: {body_pct:.4f}")

    if is_doji and is_high_volume:
        print("⚠️ Обнаружена ДОЖИ на высоком объёме! Возможный разворот.")
    else:
        print("✅ Нет сигнала.")

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

#патерн

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

📌 Order Book Spread Analyzer – Торговля на асимметрии в стакане заявок🧑‍💻

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

Анализ асимметрии объёма между лимитными ордерами на покупку (bid) и на продажу (ask).

Если одна сторона стакана значительно доминирует:🚀

Покупатели (bid) сильнее — возможный рост

Продавцы (ask) сильнее — возможное снижение

import ccxt
import time

# Настройки
EXCHANGE = ccxt.binance()
PAIR = "BTC/USDT"
DEPTH = 20
THRESHOLD = 0.2  # 20% дисбаланс
INTERVAL = 5  # секунд

def fetch_order_book():
    order_book = EXCHANGE.fetch_order_book(PAIR, limit=DEPTH)
    bids = order_book['bids']
    asks = order_book['asks']

    bid_volume = sum([v for _, v in bids])
    ask_volume = sum([v for _, v in asks])
    imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume)

    return bid_volume, ask_volume, imbalance

def analyze():
    bid, ask, imbalance = fetch_order_book()

    print(f"\nBID: {bid:.2f} BTC | ASK: {ask:.2f} BTC | Imbalance: {imbalance:.2%}")

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

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

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

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

📌 Stablecoin Arbitrage Monitor – Мониторинг расхождений курсов стейблов на разных платформах 🧑‍💻

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

Стейблкоины (например, USDT, USDC, BUSD, DAI) должны торговаться около $1, но из-за спроса/предложения и разной ликвидности цены иногда отклоняются🫰

→ Это создаёт возможность для арбитража, особенно между:

Разными стейблами (USDC/USDT, DAI/USDT),

Разными биржами (Binance, Kraken, KuCoin и др.).

import ccxt
import time

# Параметры
STABLE_PAIRS = ["USDC/USDT", "DAI/USDT", "TUSD/USDT", "BUSD/USDT"]
THRESHOLD = 0.3  # % отклонения
EXCHANGES = {
    "binance": ccxt.binance(),
    "kucoin": ccxt.kucoin(),
    "kraken": ccxt.kraken()
}
INTERVAL = 30  # сек

def get_price(exchange, symbol):
    try:
        ticker = exchange.fetch_ticker(symbol)
        return ticker['last']
    except:
        return None

def monitor_stable_arbitrage():
    while True:
        print("\n📊 Stablecoin Arbitrage Monitor")
        for pair in STABLE_PAIRS:
            prices = {}
            for name, ex in EXCHANGES.items():
                price = get_price(ex, pair)
                if price:
                    prices[name] = price

            if len(prices) < 2:
                continue

            exchanges = list(prices.keys())
            for i in range(len(exchanges)):
                for j in range(i + 1, len(exchanges)):
                    e1, e2 = exchanges[i], exchanges[j]
                    p1, p2 = prices[e1], prices[e2]
                    spread = abs(p1 - p2) / ((p1 + p2) / 2) * 100

                    if spread >= THRESHOLD:
                        print(f"⚠️ {pair} | {e1}: {p1:.4f} vs {e2}: {p2:.4f} | Spread: {spread:.2f}%")

        time.sleep(INTERVAL)

if __name__ == "__main__":
    monitor_stable_arbitrage()

#арбитраж

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