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

📌 VWAP Reversion Strategy – Возврат к VWAP после отклонения🧑‍💻

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

VWAP (Volume Weighted Average Price) отражает среднюю цену сделки с учётом объёма.

Если цена сильно отклоняется от VWAP, велика вероятность, что она вернётся обратно — это и есть стратегия возврата к VWAP🚀

Принцип работы:

Цена выше VWAP + X% → сигнал на шорт

Цена ниже VWAP − X% → сигнал на лонг

Цель — возврат к VWAP

import ccxt
import pandas as pd
import time

# Настройки
EXCHANGE = ccxt.binance()
PAIR = "BTC/USDT"
TIMEFRAME = "5m"
LOOKBACK = 100
DEVIATION = 0.01  # 1% отклонение
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 calculate_vwap(df):
    typical_price = (df["high"] + df["low"] + df["close"]) / 3
    vwap = (typical_price * df["volume"]).cumsum() / df["volume"].cumsum()
    return vwap

def check_vwap_reversion(df):
    vwap_series = calculate_vwap(df)
    vwap = vwap_series.iloc[-1]
    price = df["close"].iloc[-1]
    deviation = (price - vwap) / vwap

    print(f"\nЦена: {price:.2f} | VWAP: {vwap:.2f} | Отклонение: {deviation*100:.2f}%")

    if deviation >= DEVIATION:
        print("📉 Цена перекуплена — возможный откат к VWAP (шорт-сигнал)")
    elif deviation <= -DEVIATION:
        print("📈 Цена перепродана — возможный рост к VWAP (лонг-сигнал)")
    else:
        print("⏳ Цена в пределах нормы")

if __name__ == "__main__":
    while True:
        df = fetch_data()
        check_vwap_reversion(df)
        time.sleep(INTERVAL)

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

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

📌 Multi-Timeframe Arbitrage – Расхождения между таймфреймами одного актива🧑‍💻

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

Анализирует поведение одного актива на разных таймфреймах и ищет несогласованность сигналов, например:

На 1h — восходящий тренд (бычий сигнал)

На 5m — перепроданность и разворот вниз

→ Возможность взять контртрендовую позицию или перезайти по выгодной цене💸

Применение:🧨

Вход в лонг по старшему ТФ (например, 1h),

Ожидание отката на младшем ТФ (например, 5m)

Вход по наилучшей точке против краткосрочного импульса

import ccxt
import pandas as pd
import talib

# Настройки
EXCHANGE = ccxt.binance()
PAIR = "BTC/USDT"
TF_HIGH = "1h"
TF_LOW = "5m"
RSI_PERIOD = 14
RSI_OVERBOUGHT = 70
RSI_OVERSOLD = 30

def fetch_rsi(symbol, timeframe):
    ohlcv = EXCHANGE.fetch_ohlcv(symbol, timeframe=timeframe, limit=RSI_PERIOD + 10)
    df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
    rsi = talib.RSI(df["close"], timeperiod=RSI_PERIOD)
    return rsi.iloc[-1]

def detect_divergence():
    rsi_high = fetch_rsi(PAIR, TF_HIGH)
    rsi_low = fetch_rsi(PAIR, TF_LOW)

    print(f"\n[RSI] {TF_HIGH}: {rsi_high:.2f} | {TF_LOW}: {rsi_low:.2f}")

    if rsi_high > 60 and rsi_low < RSI_OVERSOLD:
        print("📥 Вход в лонг: старший ТФ бычий, младший — перепродан")
    elif rsi_high < 40 and rsi_low > RSI_OVERBOUGHT:
        print("📤 Вход в шорт: старший ТФ медвежий, младший — перекуплен")
    else:
        print("⏳ Сигналов нет, таймфреймы согласованы")

if __name__ == "__main__":
    detect_divergence()

#арбитраж

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

📌 Auto-Trailing Profit – Перемещение тейк-профита при росте цены🧑‍💻

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

Auto-Trailing Profit — это автоматическое перемещение тейк-профита вверх, когда цена растёт, чтобы:

Зафиксировать прибыль, если тренд развернётся,

И максимизировать выход, если тренд продолжается.💸

Это похоже на тейк-профит, "привязанный" к цене, как трейлинг-стоп, но с фиксацией прибыли по заданным условиям.

import ccxt
import time

# Настройки
EXCHANGE = ccxt.binance()
PAIR = "BTC/USDT"
POSITION_SIZE = 0.01  # BTC
TRAIL_START = 100.0  # начальная прибыль (в $)
TRAIL_STEP = 50.0    # шаг подтягивания (в $)

def get_price():
    return EXCHANGE.fetch_ticker(PAIR)['last']

def trailing_tp_manager(entry_price):
    max_profit = 0
    tp_price = entry_price + TRAIL_START

    while True:
        price = get_price()
        profit = (price - entry_price) * POSITION_SIZE

        print(f"\nТекущая цена: {price:.2f} | Профит: ${profit:.2f} | Текущий TP: {tp_price:.2f}")

        if profit > max_profit:
            max_profit = profit

        # Перемещаем тейк-профит вверх при достижении следующего шага
        new_tp_price = entry_price + ((int((max_profit - TRAIL_START) / TRAIL_STEP) + 1) * TRAIL_STEP) / POSITION_SIZE
        if new_tp_price > tp_price:
            tp_price = new_tp_price
            print(f"🔄 TP обновлён: {tp_price:.2f}")

        # Эмуляция: цена опустилась ниже TP → фиксация
        if price <= tp_price:
            print(f"✅ Цена достигла TP: {tp_price:.2f} → фиксация прибыли")
            break

        time.sleep(10)

# Пример входа
if __name__ == "__main__":
    entry = get_price()
    print(f"📈 Вход по {entry:.2f}, старт трейлинг TP...")
    trailing_tp_manager(entry)

#ордер

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

📌 PCA Strategy Reducer – Сокращение количества сигналов через анализ главных компонент (PCA)🧑‍💻

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

Когда стратегия использует много технических индикаторов, сигналы могут быть:🧨

избыточными (повторяющимися),

шумными (противоречивыми),

слабо значимыми.

PCA (Principal Component Analysis) позволяет:
→ сжать индикаторы до нескольких основных факторов,
→ уменьшить количество сигналов,
→ улучшить точность входа и устойчивость модели.

Применение:

Преобразовать 5–10 индикаторов (RSI, MACD, ATR, Bollinger Width, Momentum...)

Сократить до 2–3 главных компонент

Строить сигналы на основе этих компонент (например, PCA_1 > порога)

import ccxt
import pandas as pd
import numpy as np
import talib
from sklearn.decomposition import PCA

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

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

def calculate_indicators(df):
    df["rsi"] = talib.RSI(df["close"], timeperiod=14)
    df["macd"], _, _ = talib.MACD(df["close"])
    df["atr"] = talib.ATR(df["high"], df["low"], df["close"], timeperiod=14)
    df["momentum"] = talib.MOM(df["close"], timeperiod=10)
    df["boll_width"] = talib.BBANDS(df["close"])[1] - talib.BBANDS(df["close"])[2]
    df.dropna(inplace=True)
    return df

def apply_pca(df):
    features = df[["rsi", "macd", "atr", "momentum", "boll_width"]]
    pca = PCA(n_components=N_COMPONENTS)
    components = pca.fit_transform(features)
    df["pca_1"] = components[:, 0]
    df["pca_2"] = components[:, 1]
    return df, pca.explained_variance_ratio_

def signal_from_pca(df):
    last_pca = df.iloc[-1][["pca_1", "pca_2"]]
    print(f"\nPCA-1: {last_pca['pca_1']:.2f} | PCA-2: {last_pca['pca_2']:.2f}")

    if last_pca["pca_1"] > 1.0:
        print("📈 Сигнал на покупку (основной компонент указывает на силу)")
    elif last_pca["pca_1"] < -1.0:
        print("📉 Сигнал на продажу")
    else:
        print("⏳ Нет сигнала — нейтральное состояние")

if __name__ == "__main__":
    df = fetch_data()
    df = calculate_indicators(df)
    df, variance = apply_pca(df)
    print(f"Объяснённая дисперсия компонент: {variance}")
    signal_from_pca(df)

#индикатор

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

📌 Volatility Regime Classifier – Переключение стратегий по типу рынка🧑‍💻

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

Рынок постоянно меняет состояние:

Трендовый (высокая волатильность, направленное движение)

Флэт (низкая волатильность, боковое движение)

Volatility Regime Classifier — это инструмент, который:🚀

1. Определяет текущее состояние рынка (режим),

2. Автоматически включает подходящую стратегию (например, трендовую или флетовую).

import ccxt
import pandas as pd
import talib
import time

# Настройки
EXCHANGE = ccxt.binance()
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
LOOKBACK = 100
ATR_PERIOD = 14
TREND_THRESHOLD = 200  # условный порог волатильности (в USD)

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

def classify_volatility_regime(df):
    atr = talib.ATR(df["high"], df["low"], df["close"], timeperiod=ATR_PERIOD)
    last_atr = atr.iloc[-1]

    print(f"\nТекущий ATR: {last_atr:.2f} USD")

    if last_atr > TREND_THRESHOLD:
        print("📈 Режим: ТРЕНД → активируем трендовую стратегию (например, MA crossover)")
    else:
        print("🔄 Режим: ФЛЭТ → активируем стратегию на возврат (например, Bollinger Reversion)")

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

#индикатор

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

📌 Dynamic Entry Splitter – Адаптивное разбиение позиции по условиям рынка🧑‍💻

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

Вместо того чтобы входить в рынок одним большим ордером, позиция разбивается на несколько частей, которые исполняются в зависимости от:

- текущей ликвидности,
- глубины стакана,
- волатильности,
- объёма,
- или других рыночных условий.

Это позволяет:🚀

→ уменьшить проскальзывание,
→ войти более точно,
→ скрыть крупный вход от других участников (анти-HFT подход).

import ccxt
import time
import random

# Настройки
EXCHANGE = ccxt.binance({
    # "apiKey": "YOUR_API_KEY",
    # "secret": "YOUR_SECRET",
    "enableRateLimit": True
})

PAIR = "BTC/USDT"
TOTAL_AMOUNT = 1.0  # BTC
MIN_PART = 0.05     # минимальная доля
MAX_PART = 0.2      # максимальная доля
SPREAD_THRESHOLD = 20  # в USD
DEPTH = 10
CHECK_INTERVAL = 10  # сек

def get_order_book():
    book = EXCHANGE.fetch_order_book(PAIR, limit=DEPTH)
    best_bid = book['bids'][0][0]
    best_ask = book['asks'][0][0]
    spread = best_ask - best_bid
    return best_bid, best_ask, spread

def adaptive_entry():
    remaining = TOTAL_AMOUNT
    print(f"Начинаем адаптивный вход на сумму {TOTAL_AMOUNT} BTC...\n")
    while remaining > 0:
        bid, ask, spread = get_order_book()
        mid = (bid + ask) / 2

        print(f"Цена BID: {bid:.2f}, ASK: {ask:.2f}, Спред: {spread:.2f}")

        if spread <= SPREAD_THRESHOLD:
            size = min(random.uniform(MIN_PART, MAX_PART), remaining)
            price = bid  # вход лимитом по BID
            print(f"📥 Размещаем ордер на {size:.4f} BTC по {price:.2f}")
            # order = EXCHANGE.create_limit_buy_order(PAIR, size, price)  # для реальной торговли
            remaining -= size
        else:
            print("⏳ Спред широкий, ждём улучшения...")

        print(f"Осталось: {remaining:.4f} BTC\n")
        time.sleep(CHECK_INTERVAL)

if __name__ == "__main__":
    adaptive_entry()

#ордер

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

📌 Price Delay Arbitrage – Арбитраж при задержке обновления цен между биржами🧑‍💻

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

На разных биржах котировки одного и того же актива могут обновляться с задержкой.
Если:

Биржа A уже отреагировала на движение,

А Биржа B ещё нет,
→ Можно войти на медленной бирже, пока цена не догнала.💸

Это форма low-latency арбитража, похожая на front-running, но без необходимости HFT.

import ccxt
import time

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

# Биржи
fast = ccxt.binance()
slow = ccxt.kucoin()

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

def check_price_delay_arbitrage():
    price_fast, time_fast = get_price(fast)
    price_slow, time_slow = get_price(slow)

    if not price_fast or not price_slow:
        return

    spread = price_fast - price_slow
    spread_pct = spread / price_fast * 100

    print(f"\nFast: {price_fast:.2f} ({time_fast}) | Slow: {price_slow:.2f} ({time_slow}) | Spread: {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_price_delay_arbitrage()
        time.sleep(INTERVAL)

#арбитраж

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

📌 LSTM Trend Predictor – Прогноз движения цены с помощью LSTM🧑‍💻

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

LSTM (Long Short-Term Memory) — тип рекуррентной нейросети, которая хорошо справляется с временными рядами, в том числе с финансовыми данными.
Модель обучается на исторических свечах и пытается предсказать направление (или цену) на следующий шаг.

Пример задачи:💻

Вход: последовательность из 50 последних цен

Цель: предсказать цену/тренд на следующую свечу

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

- определения тренда (вверх/вниз),

- генерации сигналов для стратегии,

- фильтрации других индикаторов.

import ccxt
import pandas as pd
import numpy as np
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from sklearn.preprocessing import MinMaxScaler

# Параметры
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
LOOKBACK = 500
WINDOW_SIZE = 50

# Получение данных
exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv(PAIR, timeframe=TIMEFRAME, limit=LOOKBACK)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])

# Подготовка данных
scaler = MinMaxScaler()
scaled_close = scaler.fit_transform(df[['close']])

X, y = [], []
for i in range(WINDOW_SIZE, len(scaled_close)):
    X.append(scaled_close[i-WINDOW_SIZE:i])
    y.append(scaled_close[i])
X, y = np.array(X), np.array(y)

# Модель LSTM
model = Sequential()
model.add(LSTM(50, return_sequences=True, input_shape=(X.shape[1], 1)))
model.add(Dropout(0.2))
model.add(LSTM(50))
model.add(Dropout(0.2))
model.add(Dense(1))
model.compile(optimizer='adam', loss='mse')

# Обучение
model.fit(X, y, epochs=10, batch_size=32)

# Прогноз следующей цены
last_seq = scaled_close[-WINDOW_SIZE:]
last_seq = last_seq.reshape((1, WINDOW_SIZE, 1))
predicted = model.predict(last_seq)
predicted_price = scaler.inverse_transform(predicted)

print(f"📈 Прогноз следующей цены: {predicted_price[0][0]:.2f} USDT")

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

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

📌 Adaptive Stop Loss Manager – Управление стоп-лоссом в зависимости от волатильности🧑‍💻

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

Вместо фиксированного стоп-лосса, его размер адаптируется под текущую волатильность рынка, чтобы:

- избежать раннего выбивания,

- учитывать динамику рынка,

- уменьшить ложные срабатывания в "шумном" движении.

Принцип работы:💸

1. Расчёт волатильности (например, через ATR, StdDev или процент отклонения).

2. Установка стоп-лосса на X × волатильность ниже цены входа (для лонга) или выше (для шорта).

3. Возможность перемещения стопа при росте позиции (опционально — трейлинг).

import ccxt
import pandas as pd
import talib

# Настройки
EXCHANGE = ccxt.binance()
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
ATR_PERIOD = 14
ATR_MULTIPLIER = 1.5  # множитель на волатильность
LOOKBACK = 100

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

def calculate_stop_loss(entry_price, side="long"):
    df = fetch_data()
    atr = talib.ATR(df["high"], df["low"], df["close"], timeperiod=ATR_PERIOD).iloc[-1]

    if side == "long":
        stop_loss = entry_price - (atr * ATR_MULTIPLIER)
    elif side == "short":
        stop_loss = entry_price + (atr * ATR_MULTIPLIER)
    else:
        raise ValueError("Сторона сделки должна быть 'long' или 'short'")

    print(f"\nATR: {atr:.2f} | Entry: {entry_price:.2f} | Stop-Loss: {stop_loss:.2f}")
    return stop_loss

# Пример вызова
if __name__ == "__main__":
    entry = 30000.0  # примерная цена входа
    calculate_stop_loss(entry_price=entry, side="long")

#ордер

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

📌Market Microstructure Scanner – Анализ микродвижений в стакане (order book)🧑‍💻

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

Market Microstructure — это анализ поведения участников в стакане и потоке ордеров для выявления:

- скрытых покупок/продаж,

- спуфинга (spoofing),

- перестроения ликвидности,

- импульсных дисбалансов.

import ccxt
import time

# Настройки
EXCHANGE = ccxt.binance()
PAIR = "BTC/USDT"
DEPTH = 20
THRESHOLD = 30  # BTC – крупная заявка
INTERVAL = 2  # сек

def scan_microstructure():
    prev_bids, prev_asks = None, None

    while True:
        book = EXCHANGE.fetch_order_book(PAIR, limit=DEPTH)
        bids = {price: volume for price, volume in book['bids']}
        asks = {price: volume for price, volume in book['asks']}

        print("\n--- Скан стакана ---")
        # Отслеживаем крупные заявки
        for side, data in [("BID", bids), ("ASK", asks)]:
            for price, volume in data.items():
                if volume >= THRESHOLD:
                    print(f"🧱 {side}: {volume:.2f} BTC на {price:.2f}")

        # Отслеживаем исчезновение крупных заявок (спуфинг)
        if prev_bids:
            disappeared = [
                price for price in prev_bids
                if prev_bids[price] >= THRESHOLD and price not in bids
            ]
            for price in disappeared:
                print(f"❗ Крупный BID исчез: {price}")

        if prev_asks:
            disappeared = [
                price for price in prev_asks
                if prev_asks[price] >= THRESHOLD and price not in asks
            ]
            for price in disappeared:
                print(f"❗ Крупный ASK исчез: {price}")

        prev_bids, prev_asks = bids, asks
        time.sleep(INTERVAL)

if __name__ == "__main__":
    scan_microstructure()

#ордер

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

📌Position Reversal Trigger – Переворот позиции при пробое уровня🧑‍💻

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

Если цена пробивает ключевой уровень (поддержки/сопротивления), текущая позиция:

Закрывается,

И тут же открывается в противоположную сторону.

Это позволяет моментально реагировать на изменение рыночной структуры, особенно при:

- ложных пробоях,

- переломах тренда,

- волатильных импульсах.

import ccxt
import time

# Настройки
EXCHANGE = ccxt.binance({
    # 'apiKey': 'your_api_key',
    # 'secret': 'your_api_secret',
    'enableRateLimit': True
})

PAIR = "BTC/USDT"
LEVEL = 30000.0  # ключевой уровень
POSITION = None  # текущая позиция: None, 'long', 'short'
AMOUNT = 0.01
INTERVAL = 10  # секунд

def get_price():
    return EXCHANGE.fetch_ticker(PAIR)['last']

def open_position(direction, amount, price):
    print(f"🚀 Открываем {direction.upper()} по {price}")
    # Пример: отправка рыночного ордера
    # if direction == "long":
    #     EXCHANGE.create_market_buy_order(PAIR, amount)
    # else:
    #     EXCHANGE.create_market_sell_order(PAIR, amount)

def close_position(current_price):
    print(f"❌ Закрытие позиции по {current_price}")
    # Здесь должна быть логика закрытия ордера (обратная операция)

def manage_position(price):
    global POSITION

    if POSITION == "long" and price < LEVEL:
        close_position(price)
        open_position("short", AMOUNT, price)
        POSITION = "short"

    elif POSITION == "short" and price > LEVEL:
        close_position(price)
        open_position("long", AMOUNT, price)
        POSITION = "long"

    elif POSITION is None:
        # Начальная позиция по текущему тренду
        if price > LEVEL:
            open_position("long", AMOUNT, price)
            POSITION = "long"
        elif price < LEVEL:
            open_position("short", AMOUNT, price)
            POSITION = "short"

if __name__ == "__main__":
    print(f"🎯 Запущен переворот по уровню {LEVEL}")
    while True:
        price = get_price()
        print(f"Цена: {price:.2f} | Позиция: {POSITION}")
        manage_position(price)
        time.sleep(INTERVAL)

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

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

📌Anomaly Detection Alert Bot – Обнаружение аномального поведения цены или объёма 🧑‍💻

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

Бот отслеживает внезапные и нетипичные изменения на рынке:

- всплески объёма,

- резкие ценовые движения,

-высокую волатильность,

→ и выдает сигналы/уведомления при обнаружении аномалий 🧨

Типы аномалий:🌐

1. Ценовые импульсы: резкое движение за 1 свечу

2. Всплеск объёма: объём > X × среднего

3. Отклонение от скользящего среднего (цены или объёма)

4. Волатильность выше нормы (по ATR/StdDev)

import ccxt
import pandas as pd
import time

# Настройки
EXCHANGE = ccxt.binance()
PAIR = "BTC/USDT"
TIMEFRAME = "5m"
LOOKBACK = 100
VOLUME_THRESHOLD = 2.5  # раз выше среднего
PRICE_JUMP_PCT = 1.0    # в % за свечу
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'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    return df

def detect_anomalies(df):
    last = df.iloc[-1]
    prev = df.iloc[-2]
    
    # Аномалия по объёму
    avg_volume = df['volume'][:-1].mean()
    volume_spike = last['volume'] > avg_volume * VOLUME_THRESHOLD
    
    # Аномалия по цене
    price_change_pct = abs(last['close'] - prev['close']) / prev['close'] * 100
    price_spike = price_change_pct >= PRICE_JUMP_PCT

    print(f"\nЦена: {last['close']:.2f} | Объём: {last['volume']:.2f} | Δ%: {price_change_pct:.2f}")

    if volume_spike:
        print(f"⚠️ Всплеск объёма: {last['volume']:.2f} > {avg_volume:.2f}")
    if price_spike:
        direction = "вверх" if last['close'] > prev['close'] else "вниз"
        print(f"⚡ Резкое движение {direction}: {price_change_pct:.2f}% за свечу")
    if not volume_spike and not price_spike:
        print("✅ Аномалий не обнаружено.")

# Основной цикл
if __name__ == "__main__":
    while True:
        df = fetch_data()
        detect_anomalies(df)
        time.sleep(CHECK_INTERVAL)

#индикатор

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

📌Kill Switch Bot – Автоматическое закрытие всех позиций при достижении лимита риска🧑‍💻

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

Kill Switch Bot — это защитный механизм, который:

Постоянно отслеживает PnL (прибыль/убыток) по всем позициям.

И при достижении заданного максимального убытка (например, −5%) или долларового лимита (например, −$500):

→ автоматически закрывает все открытые позиции

→ Можно использовать как антислив защиту или "паник-кнопку" в алгоритмах.

import ccxt
import time

# Настройки
MAX_DRAWDOWN_USD = -300  # максимальный убыток в долларах
CHECK_INTERVAL = 30  # сек

# Инициализация Binance Futures
exchange = ccxt.binance({
    'apiKey': 'YOUR_API_KEY',
    'secret': 'YOUR_SECRET_KEY',
    'enableRateLimit': True,
    'options': {'defaultType': 'future'}
})

def fetch_total_unrealized_pnl():
    balance = exchange.fapiPrivate_get_account()
    total_pnl = 0
    for pos in balance['positions']:
        upnl = float(pos['unrealizedProfit'])
        total_pnl += upnl
    return total_pnl

def close_all_positions():
    positions = exchange.fapiPrivate_get_positionrisk()
    for pos in positions:
        symbol = pos['symbol']
        amt = float(pos['positionAmt'])
        if amt == 0:
            continue
        side = 'SELL' if amt > 0 else 'BUY'
        print(f"⚠️ Закрытие позиции {symbol}: {side} {abs(amt)}")
        try:
            exchange.create_order(
                symbol=symbol,
                type='MARKET',
                side=side,
                amount=abs(amt),
                params={'reduceOnly': True}
            )
        except Exception as e:
            print(f"Ошибка закрытия {symbol}: {e}")

def run_kill_switch():
    while True:
        pnl = fetch_total_unrealized_pnl()
        print(f"\n🔍 Текущий общий PnL: {pnl:.2f} USDT")
        if pnl <= MAX_DRAWDOWN_USD:
            print(f"🛑 Убыток превышен ({pnl:.2f} USDT) — запускаем Kill Switch!")
            close_all_positions()
            break
        time.sleep(CHECK_INTERVAL)

if __name__ == "__main__":
    run_kill_switch()

#ордер

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

📌XGBoost Classifier Strategy – Обучение модели на индикаторах (сигналы Buy/Sell)🧑‍💻

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

Используем технические индикаторы как признаки (features) и обучаем XGBoost-классификатор, который:

- анализирует исторические данные,

- учится предсказывать сигналы: покупка / продажа / держать,

- может применяться в алготрейдинге и сигнальных ботах.

import ccxt
import pandas as pd
import numpy as np
import talib
import xgboost as xgb
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report

# Параметры
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
LOOKBACK = 500

# Получение данных
exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv(PAIR, timeframe=TIMEFRAME, limit=LOOKBACK)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])

# Технические индикаторы
df["rsi"] = talib.RSI(df["close"], timeperiod=14)
df["macd"], _, _ = talib.MACD(df["close"])
df["atr"] = talib.ATR(df["high"], df["low"], df["close"], timeperiod=14)
df["mom"] = talib.MOM(df["close"], timeperiod=10)
df.dropna(inplace=True)

# Целевая переменная: рост цены через N свечей
N = 3
df["future_return"] = df["close"].shift(-N) - df["close"]
df["target"] = (df["future_return"] > 0).astype(int)
df.dropna(inplace=True)

# Обучающая выборка
features = ["rsi", "macd", "atr", "mom"]
X = df[features]
y = df["target"]

X_train, X_test, y_train, y_test = train_test_split(X, y, shuffle=False, test_size=0.2)

# Модель
model = xgb.XGBClassifier(n_estimators=100, max_depth=3, use_label_encoder=False, eval_metric="logloss")
model.fit(X_train, y_train)

# Прогноз
preds = model.predict(X_test)
print(classification_report(y_test, preds))

# Последний сигнал
last_features = X.iloc[[-1]]
signal = model.predict(last_features)[0]
print(f"\n📈 Текущий сигнал: {'BUY' if signal == 1 else 'SELL'}")

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

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

📌Order Layering Strategy – Построение лестницы лимитных ордеров🧑‍💻

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

Order Layering — это тактика размещения нескольких лимитных ордеров на разных ценовых уровнях, чтобы:

- войти в позицию частями по более выгодной цене,

- разместить ликвидность в стакане,

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

- или обмануть других участников рынка

import ccxt
import time

# Настройки
exchange = ccxt.binance({
    # 'apiKey': 'YOUR_API_KEY',
    # 'secret': 'YOUR_SECRET',
    'enableRateLimit': True
})

PAIR = "BTC/USDT"
TOTAL_AMOUNT = 0.1  # BTC
LAYERS = 5
START_PRICE = 79500
PRICE_STEP = 100
ORDER_SIDE = 'buy'  # или 'sell'

def build_order_ladder():
    amount_per_order = TOTAL_AMOUNT / LAYERS

    for i in range(LAYERS):
        price = START_PRICE - i * PRICE_STEP if ORDER_SIDE == 'buy' else START_PRICE + i * PRICE_STEP
        print(f"📥 Размещение лимитного ордера: {ORDER_SIDE.upper()} {amount_per_order:.4f} BTC по {price:.2f}")
        
        # exchange.create_limit_buy_order(PAIR, amount_per_order, price)  # активировать при торговле

        time.sleep(1)  # небольшая пауза между ордерами

if __name__ == "__main__":
    build_order_ladder()

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

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