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

📌Front-Run Detector – Обнаружение активности, похожей на фронтран🧑‍💻

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

Фронтран (front-running) — это стратегия, при которой участник рынка опережает крупную сделку, зная о ней заранее (или замечая её подготовку в стакане/мемпуле).

Front-Run Detector пытается выявить:

- Подозрительно крупные заявки перед всплеском объема или движения

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

- Быстрые серии ордеров перед резким движением

Примеры признаков фронтрана:🛠️

1. Резкое добавление крупной заявки на Bid/Ask, которая исчезает перед импульсом

2. Секундные всплески мелких ордеров в одну сторону → затем резкое движение

3. Появление ликвидности "приманки", за которой идёт настоящая крупная сделка

4. Разгон объёма до того, как произойдёт прорыв уровня

import ccxt
import time

# Настройки
exchange = ccxt.binance()
PAIR = "BTC/USDT"
DEPTH = 20
LARGE_ORDER_SIZE = 30  # BTC
INTERVAL = 2  # секунд

prev_bids, prev_asks = {}, {}

def fetch_order_book():
    ob = exchange.fetch_order_book(PAIR, limit=DEPTH)
    bids = {price: size for price, size in ob['bids']}
    asks = {price: size for price, size in ob['asks']}
    return bids, asks

def detect_front_run(prev, current, side):
    suspicious = []
    for price in current:
        if current[price] >= LARGE_ORDER_SIZE:
            if price not in prev or prev[price] < LARGE_ORDER_SIZE:
                suspicious.append(price)
    return suspicious

def run_detector():
    global prev_bids, prev_asks

    while True:
        bids, asks = fetch_order_book()

        new_large_bids = detect_front_run(prev_bids, bids, "BID")
        new_large_asks = detect_front_run(prev_asks, asks, "ASK")

        for price in new_large_bids:
            print(f"⚠️ Появился крупный BID: {price} → возможный фронтран")
        for price in new_large_asks:
            print(f"⚠️ Появился крупный ASK: {price} → возможный фронтран")

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

if __name__ == "__main__":
    run_detector()

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

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

📌Clustering Market Regimes – Определение рыночных фаз через кластеризацию🧑‍💻

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

Вместо ручного деления рынка на "тренд" и "флэт", используется кластеризация (машинное обучение без учителя), чтобы:

- Автоматически выявить рыночные режимы по историческим данным,

- Присвоить каждому участку графика кластер (режим),

- Переключать стратегию в зависимости от текущего кластера рынка.

Признаки, используемые для кластеризации:🛠️

- Возвратность (доходность за окно)

- Волатильность (ATR / std)

- Объём

- Размеры свечей

- Трендовые индикаторы (RSI, MACD, ADX)

import ccxt
import pandas as pd
import numpy as np
import talib
from sklearn.cluster import KMeans
import matplotlib.pyplot as plt

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

# Получение данных
exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv(PAIR, timeframe=TIMEFRAME, limit=LOOKBACK)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
df["returns"] = df["close"].pct_change()
df["volatility"] = df["returns"].rolling(window=10).std()
df["atr"] = talib.ATR(df["high"], df["low"], df["close"], timeperiod=14)
df["rsi"] = talib.RSI(df["close"], timeperiod=14)
df.dropna(inplace=True)

# Фичи для кластеризации
features = df[["returns", "volatility", "atr", "rsi"]]
features = features.fillna(0)

# Кластеризация
kmeans = KMeans(n_clusters=N_CLUSTERS, random_state=42)
df["regime"] = kmeans.fit_predict(features)

# Визуализация
plt.figure(figsize=(12, 5))
for regime in range(N_CLUSTERS):
    plt.plot(df[df["regime"] == regime]["close"], label=f"Regime {regime}")
plt.legend()
plt.title("Режимы рынка, определённые кластеризацией")
plt.xlabel("Время")
plt.ylabel("Цена")
plt.show()

#индикатор

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

📌Reversal Pattern Detector – Обнаружение разворотных паттернов🧑‍💻

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

Разворотные паттерны графического анализа указывают на возможное окончание текущего тренда и начало.

Пример на Python – Обнаружение двойного дна👇👇👇

import ccxt
import pandas as pd
import numpy as np
from scipy.signal import argrelextrema

# Настройки
exchange = ccxt.binance()
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
LOOKBACK = 300
DISTANCE = 5  # интервал между минимумами

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

def detect_double_bottom(df):
    lows = df["low"].values
    idx_min = argrelextrema(lows, np.less, order=DISTANCE)[0]

    for i in range(len(idx_min) - 1):
        i1, i2 = idx_min[i], idx_min[i+1]
        price1, price2 = lows[i1], lows[i2]
        diff = abs(price1 - price2)

        # Условия: два минимума на ~одном уровне, между ними — максимум
        if diff / price1 < 0.01:
            middle_idx = (i1 + i2) // 2
            middle_high = df["high"].iloc[middle_idx]
            if middle_high > price1 * 1.02:
                print(f"\n✅ Найден паттерн Двойное Дно на свечах {i1} и {i2}")
                print(f"Минимумы: {price1:.2f}, {price2:.2f} | Потенциал пробоя: {middle_high:.2f}")

if __name__ == "__main__":
    df = fetch_data()
    detect_double_bottom(df)

#патерн

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

📌Dark Pool Activity Scanner – Анализ скрытых транзакций на рынке🧑‍💻

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

Dark Pools — это внебиржевые площадки, где крупные участники совершают сделки без раскрытия информации в стакане.
Хотя в криптовалютах нет формальных "даркпулов", их активность можно оценивать косвенно — через:

Признаки скрытых транзакций:🛠️

1. Внезапные крупные объёмы без изменений в стакане

2. Необъяснимые всплески объёма на слабом движении цены

3. Резкие движения цены без предварительного давления в стакане

4. Скачки крупных сделок (анализ ленты сделок / trade stream)

5. Необычные дисбалансы в дельте объёмов

6. Низкая волатильность + высокий объём = поглощение ликвидности

import ccxt
import pandas as pd
import time

# Настройки
exchange = ccxt.binance()
PAIR = "BTC/USDT"
TIMEFRAME = "1m"
VOLUME_MULTIPLIER = 3.0  # объем выше среднего в X раз
PRICE_CHANGE_THRESHOLD = 0.1  # % за минуту
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"])
    df["ts"] = pd.to_datetime(df["ts"], unit="ms")
    df["return"] = df["close"].pct_change() * 100
    return df.dropna()

def detect_dark_pool_behavior(df):
    avg_volume = df["volume"][:-1].mean()
    last = df.iloc[-1]
    volume_spike = last["volume"] > avg_volume * VOLUME_MULTIPLIER
    low_price_move = abs(last["return"]) < PRICE_CHANGE_THRESHOLD

    print(f"\nВремя: {last['ts']} | Объём: {last['volume']:.2f} | Δ%: {last['return']:.2f}")

    if volume_spike and low_price_move:
        print("⚠️ Подозрение на скрытую крупную сделку (высокий объём без движения цены)")

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

#индикатор

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

📌Genetic Strategy Generator – Генерация торговых стратегий с помощью генетических алгоритмов🧑‍💻

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

- Генетический алгоритм (GA) используется для поиска оптимальных правил входа/выхода в рынок, основываясь на:

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

- логических операторах (AND, OR, >, <),

- целевой функции (например, максимизация доходности или Sharpe Ratio).

Такой подход имитирует естественный отбор:
→ лучшие "особи" (стратегии) выживают и улучшаются из поколения в поколение.

Что делает Genetic Strategy Generator:🛠️

1. Генерирует случайные стратегии из набора правил и индикаторов

2. Применяет их к историческим данным

3. Оценивает результат (fitness) — прибыль, стабильность, drawdown

4. Оставляет лучшие стратегии и мутирует их

5. Повторяет до достижения лучших результатов

Пример Python-структуры: стратегия на базе RSI и MACD

import numpy as np
import pandas as pd
import talib
import ccxt
import random

# Получение данных
exchange = ccxt.binance()
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
LOOKBACK = 300

def fetch_data():
    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)
    macd, signal, _ = talib.MACD(df["close"])
    df["macd"] = macd
    df["signal"] = signal
    df.dropna(inplace=True)
    return df

# Генерация случайной стратегии
def generate_strategy():
    return {
        "rsi_threshold": random.randint(20, 80),
        "macd_relation": random.choice([">", "<"])
    }

# Применение стратегии
def apply_strategy(df, strategy):
    signals = []
    for i in range(len(df)):
        rsi = df["rsi"].iloc[i]
        macd = df["macd"].iloc[i]
        signal = df["signal"].iloc[i]

        cond1 = rsi < strategy["rsi_threshold"]
        cond2 = (macd > signal) if strategy["macd_relation"] == ">" else (macd < signal)

        if cond1 and cond2:
            signals.append(1)  # BUY
        else:
            signals.append(0)  # NO ACTION
    df["signal"] = signals
    return df

# Подсчёт PnL (доходности)
def evaluate(df):
    df["returns"] = df["close"].pct_change().fillna(0)
    df["strategy_return"] = df["returns"] * df["signal"].shift(1)
    total_return = (df["strategy_return"] + 1).prod() - 1
    return total_return

# Генетический цикл
def genetic_optimizer(pop_size=20, generations=10):
    population = [generate_strategy() for _ in range(pop_size)]
    df = fetch_data()

    for gen in range(generations):
        scores = []
        for strat in population:
            test_df = apply_strategy(df.copy(), strat)
            score = evaluate(test_df)
            scores.append((score, strat))

        scores.sort(reverse=True, key=lambda x: x[0])
        print(f"🧬 Поколение {gen}: лучшая доходность = {scores[0][0]:.2%}")

        # Отбор + мутация
        best = [x[1] for x in scores[:pop_size // 2]]
        new_generation = []
        for s in best:
            new = s.copy()
            if random.random() < 0.5:
                new["rsi_threshold"] += random.randint(-5, 5)
                new["rsi_threshold"] = max(10, min(new["rsi_threshold"], 90))
            if random.random() < 0.3:
                new["macd_relation"] = ">" if s["macd_relation"] == "<" else "<"
            new_generation.append(new)
        population = best + new_generation

    return scores[0][1]

if __name__ == "__main__":
    best_strategy = genetic_optimizer()
    print(f"\n🏆 Лучшая стратегия: {best_strategy}")

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

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

📌Reinforcement Learning Agent – Стратегия на основе самообучающегося агента🧑‍💻

Суть подхода:🚀

Reinforcement Learning (RL) — обучение агента принимать оптимальные решения путём взаимодействия со средой и получения награды за результат.
В контексте трейдинга агент:

Наблюдает состояние рынка (индикаторы, цены, объёмы),

Принимает действия: купить, продать, удерживать,

Получает награду (например, PnL),

Обучается максимизировать общую прибыль.

import gym
import numpy as np
import pandas as pd
import ccxt
import ta
from gym import spaces
from stable_baselines3 import PPO

# Создаём кастомную среду
class TradingEnv(gym.Env):
    def __init__(self, df):
        super().__init__()
        self.df = df.reset_index(drop=True)
        self.current_step = 0
        self.start_balance = 10000
        self.balance = self.start_balance
        self.position = 0  # 0: нет, 1: лонг
        self.entry_price = 0

        self.action_space = spaces.Discrete(3)  # hold, buy, sell
        self.observation_space = spaces.Box(low=-np.inf, high=np.inf, shape=(4,), dtype=np.float32)

    def _get_obs(self):
        row = self.df.iloc[self.current_step]
        return np.array([row['close'], row['rsi'], row['macd'], self.position], dtype=np.float32)

    def _get_reward(self, price):
        if self.position == 1:
            return price - self.entry_price
        return 0

    def step(self, action):
        done = False
        price = self.df.iloc[self.current_step]['close']

        reward = 0
        if action == 1 and self.position == 0:  # BUY
            self.position = 1
            self.entry_price = price
        elif action == 2 and self.position == 1:  # SELL
            reward = price - self.entry_price
            self.balance += reward
            self.position = 0

        self.current_step += 1
        if self.current_step >= len(self.df) - 1:
            done = True

        obs = self._get_obs()
        return obs, reward, done, {}

    def reset(self):
        self.current_step = 0
        self.balance = self.start_balance
        self.position = 0
        self.entry_price = 0
        return self._get_obs()

# Подготовка данных
def fetch_data():
    exchange = ccxt.binance()
    ohlcv = exchange.fetch_ohlcv("BTC/USDT", timeframe="1h", limit=500)
    df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
    df["rsi"] = ta.momentum.RSIIndicator(df["close"]).rsi()
    macd = ta.trend.MACD(df["close"])
    df["macd"] = macd.macd()
    return df.dropna()

# Обучение агента
df = fetch_data()
env = TradingEnv(df)
model = PPO("MlpPolicy", env, verbose=1)
model.learn(total_timesteps=10000)

# Тестирование агента
obs = env.reset()
done = False
while not done:
    action, _ = model.predict(obs)
    obs, reward, done, _ = env.step(action)
    print(f"Action: {action}, Reward: {reward:.2f}")

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

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

📌AutoFeature Engine for Strategies – Автоматическая генерация признаков для оптимизации торговых правил🧑‍💻

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

AutoFeature Engine — это компонент, который:

- создаёт множество технических признаков (features) на основе OHLCV-данных,

- помогает отбирать наиболее значимые признаки для стратегии,

- используется в связке с машинным обучением или генерацией правил (в том числе для XGBoost, RL, GA и т.д.).

Типы автоматически создаваемых признаков:🛠️

1. Технические индикаторы: RSI, MACD, ATR, Bollinger, EMA и др.

2. Статистические признаки: скользящее среднее, std, z-score, rate of change

3. Комбинированные признаки: кроссы, отклонения, сигнальные уровни

4. Целевые признаки (label): будущий доход / направление движения

import ccxt
import pandas as pd
import ta  # pip install ta
import numpy as np

# Загрузка данных
exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv("BTC/USDT", timeframe="1h", limit=500)
df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")

# Генерация признаков
def generate_features(df):
    df["return_1h"] = df["close"].pct_change()
    df["sma_10"] = ta.trend.sma_indicator(df["close"], window=10)
    df["sma_50"] = ta.trend.sma_indicator(df["close"], window=50)
    df["rsi"] = ta.momentum.RSIIndicator(df["close"]).rsi()
    macd = ta.trend.MACD(df["close"])
    df["macd"] = macd.macd()
    df["macd_signal"] = macd.macd_signal()
    df["bollinger_h"] = ta.volatility.BollingerBands(df["close"]).bollinger_hband()
    df["bollinger_l"] = ta.volatility.BollingerBands(df["close"]).bollinger_lband()
    df["atr"] = ta.volatility.AverageTrueRange(df["high"], df["low"], df["close"]).average_true_range()
    df["roc"] = ta.momentum.ROCIndicator(df["close"]).roc()
    df["zscore"] = (df["close"] - df["close"].rolling(20).mean()) / df["close"].rolling(20).std()
    return df.dropna()

# Целевая переменная (напр. рост цены через 3 часа)
def generate_target(df, forward=3):
    df["future_return"] = df["close"].shift(-forward) / df["close"] - 1
    df["target"] = (df["future_return"] > 0).astype(int)  # 1 - BUY, 0 - SELL
    return df.dropna()

# Основной процесс
df = generate_features(df)
df = generate_target(df)

print("✅ Сгенерированы признаки:")
print(df[["return_1h", "sma_10", "sma_50", "rsi", "macd", "zscore", "target"]].tail())

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

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

📌Sentiment-Based Trading – Интеграция с API новостей и Twitter для анализа настроений🧑‍💻

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

Sentiment-Based Trading опирается на анализ настроений из внешних источников, таких как:

- Новости (новостные API)

- Twitter / X (социальные сигналы)

- Reddit, Telegram, YouTube, Discord и др.

→ Алгоритм оценивает, позитивно или негативно настроен рынок,

→ На основе этого принимает торговые решения (например, лонг на позитиве).

import tweepy
import pandas as pd
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
import time

# Настройки Twitter API (получить на https://developer.twitter.com/)
api_key = "YOUR_API_KEY"
api_secret = "YOUR_API_SECRET"
access_token = "YOUR_ACCESS_TOKEN"
access_token_secret = "YOUR_ACCESS_SECRET"

# Авторизация
auth = tweepy.OAuth1UserHandler(api_key, api_secret, access_token, access_token_secret)
api = tweepy.API(auth)

# Анализатор тональности
analyzer = SentimentIntensityAnalyzer()

# Настройки мониторинга
query = "bitcoin OR btc"
limit = 50

def fetch_sentiment():
    tweets = api.search_tweets(q=query, lang="en", count=limit, result_type="recent")
    data = []
    for tweet in tweets:
        score = analyzer.polarity_scores(tweet.text)
        data.append(score['compound'])
    return data

def run_sentiment_signal():
    sentiments = fetch_sentiment()
    avg_sentiment = sum(sentiments) / len(sentiments)
    print(f"\nСреднее настроение по Twitter: {avg_sentiment:.3f}")

    if avg_sentiment > 0.2:
        print("📈 Позитивное настроение → сигнал на покупку")
    elif avg_sentiment < -0.2:
        print("📉 Негативное настроение → сигнал на продажу")
    else:
        print("⏳ Нейтральный фон, сигналов нет")

# Основной цикл
if __name__ == "__main__":
    while True:
        run_sentiment_signal()
        time.sleep(60 * 10)

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

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

📌Zero-Lag MACD Strategy – Стратегия MACD без задержки🧑‍💻

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

Стандартный MACD (Moving Average Convergence Divergence) основан на экспоненциальных скользящих средних (EMA), которые по своей природе запаздывают.
Zero-Lag MACD — модификация, которая уменьшает задержку сигнала и позволяет раньше входить в позицию.

Основная идея:📝

Используется двойное EMA сглаживание (или фильтр Тилсона / Т3), чтобы сократить лаг.

Принцип остался тот же:🛠️

Линия MACD = ZeroLag(EMA_fast) − ZeroLag(EMA_slow)

Сигнальная линия = ZeroLag(EMA_MACD)

Сигналы:💸

- Пересечение MACD и сигнальной линии

- Пересечение нуля

- Дивергенции

import ccxt
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

# Zero-Lag EMA (ZLEMA)
def zlema(series, period):
    lag = (period - 1) // 2
    ema = series + (series - series.shift(lag))
    return ema.ewm(span=period, adjust=False).mean()

# Получение данных
exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv("BTC/USDT", timeframe="1h", limit=200)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")

# Расчёт Zero-Lag MACD
fast = zlema(df["close"], 12)
slow = zlema(df["close"], 26)
df["zmacd"] = fast - slow
df["zsignal"] = zlema(df["zmacd"], 9)

# Сигналы
df["signal"] = 0
df.loc[df["zmacd"] > df["zsignal"], "signal"] = 1
df.loc[df["zmacd"] < df["zsignal"], "signal"] = -1

# Визуализация
plt.figure(figsize=(12, 6))
plt.plot(df["ts"], df["zmacd"], label="Zero-Lag MACD")
plt.plot(df["ts"], df["zsignal"], label="Signal Line", linestyle='--')
plt.axhline(0, color='gray', linestyle=':')
plt.legend()
plt.title("Zero-Lag MACD Strategy")
plt.show()

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

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

📌Correlation Matrix Monitor – Отслеживание взаимосвязей между активами🧑‍💻

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

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

Correlation Matrix Monitor позволяет:👇

- Обнаружить переоценённые или отстающие активы

- Снизить риски путём диверсификации портфеля

- Построить парные стратегии или рыночно-нейтральные позиции

Типы корреляций:📝

- Положительная: активы растут/падают вместе (BTC и ETH)

- Отрицательная: один растёт — другой падает (например, BTC и USDT.D)

- Слабая/нулевая: движения не связаны

import ccxt
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

# Настройки
symbols = ["BTC/USDT", "ETH/USDT", "BNB/USDT", "XRP/USDT", "SOL/USDT"]
exchange = ccxt.binance()
timeframe = "1h"
limit = 200

def fetch_closes():
    closes = {}
    for symbol in symbols:
        ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
        df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
        closes[symbol] = df["close"]
    return pd.DataFrame(closes)

# Получение данных и расчёт корреляции
data = fetch_closes()
returns = data.pct_change().dropna()
correlation_matrix = returns.corr()

# Визуализация
plt.figure(figsize=(8, 6))
sns.heatmap(correlation_matrix, annot=True, cmap="coolwarm", vmin=-1, vmax=1)
plt.title("Матрица корреляции между активами")
plt.show()

#индикатор

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

📌 Keltner Channel Breakout – Стратегия пробоя канала Келтнера 🧑‍💻

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

Keltner Channel — это волатильностный индикатор, основанный на экспоненциальной скользящей средней (EMA) и ATR (среднем истинном диапазоне).

Стратегия пробоя канала Келтнера подразумевает:🛠️

→ Покупку при пробое верхней границы канала,

→ Продажу при пробое нижней границы,

→ Фильтрация ложных сигналов — через объём, RSI, тренд.

import ccxt
import pandas as pd
import ta
import matplotlib.pyplot as plt

# Получаем данные
exchange = ccxt.binance()
symbol = "BTC/USDT"
tf = "1h"
limit = 200

ohlcv = exchange.fetch_ohlcv(symbol, timeframe=tf, limit=limit)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")

# Индикаторы
ema = ta.trend.ema_indicator(df["close"], window=20)
atr = ta.volatility.average_true_range(df["high"], df["low"], df["close"], window=14)
mult = 2.0

df["kc_middle"] = ema
df["kc_upper"] = ema + mult * atr
df["kc_lower"] = ema - mult * atr

# Сигналы
df["signal"] = 0
df.loc[df["close"] > df["kc_upper"], "signal"] = 1   # BUY
df.loc[df["close"] < df["kc_lower"], "signal"] = -1  # SELL

# Визуализация
plt.figure(figsize=(12, 6))
plt.plot(df["ts"], df["close"], label="Close")
plt.plot(df["ts"], df["kc_middle"], label="EMA (20)")
plt.plot(df["ts"], df["kc_upper"], label="Keltner Upper", linestyle='--')
plt.plot(df["ts"], df["kc_lower"], label="Keltner Lower", linestyle='--')
plt.fill_between(df["ts"], df["kc_lower"], df["kc_upper"], color='gray', alpha=0.1)
plt.legend()
plt.title("Keltner Channel Breakout")
plt.show()

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

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

📌 Neural ODEs for Financial Series – Непрерывные модели временных рядов 🧑‍💻

Суть идеи:🚀

Neural ODEs (Neural Ordinary Differential Equations) — это подход, который моделирует непрерывную эволюцию признаков с помощью нейросетей, решая дифференциальные уравнения вместо дискретных шагов (как в RNN или LSTM).

Когда использовать Neural ODE в трейдинге:🛠️

- Для предсказания кривых цен / доходностей

- Для моделирования динамики риска, ликвидности

- Для сложных алгоритмических стратегий с обучением от среды

import torch
import torch.nn as nn
from torchdiffeq import odeint
import yfinance as yf
import numpy as np
import matplotlib.pyplot as plt

# Загрузка данных
data = yf.download("BTC-USD", period="1mo", interval="1h")
prices = data["Close"].dropna().values
x = np.arange(len(prices))
y = (prices - np.mean(prices)) / np.std(prices)

# Подготовка данных
x_tensor = torch.tensor(x / x.max(), dtype=torch.float32)
y_tensor = torch.tensor(y, dtype=torch.float32)

# ODE модель
class ODEFunc(nn.Module):
    def __init__(self):
        super().__init__()
        self.linear = nn.Sequential(
            nn.Linear(1, 50),
            nn.Tanh(),
            nn.Linear(50, 1)
        )

    def forward(self, t, y):
        return self.linear(y)

# Инициализация
func = ODEFunc()
y0 = y_tensor[0].unsqueeze(0)
t = x_tensor.unsqueeze(1)
optimizer = torch.optim.Adam(func.parameters(), lr=1e-3)

# Обучение
for i in range(1000):
    pred_y = odeint(func, y0, t).squeeze()
    loss = torch.mean((pred_y - y_tensor) ** 2)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()
    if i % 100 == 0:
        print(f"Step {i} | Loss: {loss.item():.6f}")

# Визуализация
with torch.no_grad():
    pred_y = odeint(func, y0, t).squeeze().numpy()

plt.plot(y, label="Real")
plt.plot(pred_y, label="Neural ODE Prediction")
plt.legend()
plt.title("Neural ODE for BTC-USD (hourly)")
plt.show()

#индикатор

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

📌 Dynamic Position Sizing Tool – Адаптивный объём позиции на основе ATR или Equity🧑‍💻

Суть идеи:🛠️

Dynamic Position Sizing — это подход, при котором размер позиции автоматически адаптируется под:🚀

- волатильность рынка (например, через ATR),

- или размер текущего капитала (equity).

Цель — поддерживать стабильный риск на сделку, независимо от условий.💸

import ccxt
import pandas as pd
import talib

# Настройки
exchange = ccxt.binance()
symbol = "BTC/USDT"
timeframe = "1h"
account_equity = 10000  # текущий капитал в USDT
risk_per_trade_pct = 0.01  # риск на сделку (1%)
atr_period = 14
lookback = 100

# Получение данных
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=lookback)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
atr = talib.ATR(df["high"], df["low"], df["close"], timeperiod=atr_period)

# Расчёт объёма позиции
risk_amount = account_equity * risk_per_trade_pct
atr_last = atr.iloc[-1]
position_size = risk_amount / atr_last  # в USDT
entry_price = df["close"].iloc[-1]
qty = position_size / entry_price       # в BTC

print(f"ATR: {atr_last:.2f} | Risk: {risk_amount} USDT")
print(f"Размер позиции: {qty:.6f} BTC (эквивалентно {position_size:.2f} USDT)")

#индикатор

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

📌 Detrended Price Oscillator (DPO) – Циклические сигналы без влияния тренда🧑‍💻

Суть индикатора:🚀

DPO (Detrended Price Oscillator) — осциллятор, созданный для выявления циклических колебаний цены, игнорируя долгосрочный тренд.

В отличие от большинства осцилляторов, DPO не используется для генерации сигналов тренда, а фокусируется на короткосрочной перекупленности/перепроданности.

import ccxt
import pandas as pd

# Настройки
exchange = ccxt.binance()
symbol = "BTC/USDT"
timeframe = "1h"
period = 14

# Загрузка исторических данных
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=200)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")

# Расчёт DPO
shift = int(period / 2 + 1)
df["sma"] = df["close"].rolling(window=period).mean()
df["dpo"] = df["close"].shift(shift) - df["sma"]

# Пример последних значений
print(df[["ts", "close", "dpo"]].tail(10))

#индикатор

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

📌 Whale Wallet Tracking Strategy – Реакция на перемещения с крупных кошельков🧑‍💻

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

Крупные кошельки ("whales") часто принадлежат биржам, фондам, институционалам или ранним инвесторам.

Мониторинг их активности позволяет:

- Предугадывать возможную продажу или покупку,

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

- Использовать в стратегиях как фильтр или триггер.

Источники данных:💸

Whale Alert API

Glassnode – аналитика перемещений

Arkham Intelligence – адреса по именам

Onchain API (например, через etherscan, blockchair, bitquery, chainbase)

import requests
import time

API_KEY = "YOUR_WHALE_ALERT_API_KEY"
MIN_VALUE = 500000  # USD
INTERVAL = 60  # сек

def fetch_transfers():
    url = "https://api.whale-alert.io/v1/transactions"
    params = {
        "api_key": API_KEY,
        "min_value": MIN_VALUE,
        "currency": "btc",
        "limit": 10
    }
    r = requests.get(url, params=params)
    data = r.json()
    return data.get("transactions", [])

def parse_transfer(tx):
    direction = "→ EXCHANGE" if tx["to"]["owner_type"] == "exchange" else "← FROM EXCHANGE"
    print(f"[{tx['timestamp']}] {tx['amount']} BTC {direction} | From: {tx['from']['owner']} → {tx['to']['owner']}")

if __name__ == "__main__":
    while True:
        txs = fetch_transfers()
        for tx in txs:
            parse_transfer(tx)
        time.sleep(INTERVAL)

#индикатор

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