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

📌 Volatility Regime Classifier – Классификация рынка на флет и тренд🧑‍💻

Суть идеи:🚀

Рынок постоянно переходит между двумя основными режимами:

Тренд — направленное движение (вверх или вниз),

Флет — боковик, консолидация, низкая волатильность.

Цель стратегии:🛠️
автоматически определить текущий режим рынка, чтобы:

- применять подходящую стратегию (например, трендовая/флетовая),

- фильтровать ложные сигналы,

- управлять рисками.

import ccxt
import pandas as pd
import talib

# Параметры
exchange = ccxt.binance()
symbol = "BTC/USDT"
timeframe = "1h"
atr_period = 14
lookback = 150
volatility_threshold = 100  # например, 100 USDT по BTC

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

# Расчёт ATR (волатильности)
df["atr"] = talib.ATR(df["high"], df["low"], df["close"], timeperiod=atr_period)

# Классификация режима
df["regime"] = "flat"
df.loc[df["atr"] > volatility_threshold, "regime"] = "trend"

# Последнее состояние
last = df.iloc[-1]
print(f"Последний режим рынка: {last['regime'].upper()} (ATR = {last['atr']:.2f})")

#индикатор

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

📌 Trade Execution Optimizer – Оптимизация входа с учётом комиссий и проскальзывания🧑‍💻

Суть идеи:🚀

Цель Trade Execution Optimizer — минимизировать издержки при входе/выходе в позицию за счёт:

- правильного выбора типа ордера (market / limit / TWAP / VWAP),

- учёта комиссий и проскальзывания,

- адаптации к ликвидности и объёму актива,

- разбиения крупных ордеров на части (execution slicing).

Факторы, которые учитываются:🛠️

1. Тип ордера:

Market – быстро, но с проскальзыванием

Limit – без проскальзывания, но не факт что исполнится

TWAP/VWAP – равномерное исполнение во времени/объёму

2. Комиссия биржи (taker/maker fees)

3. Глубина стакана (Order Book)

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

4. Объём сделки относительно ликвидности

- если объём слишком большой → разбивать на части

import ccxt

# Параметры
exchange = ccxt.binance()
symbol = "BTC/USDT"
trade_amount_usdt = 10000
maker_fee = 0.0001
taker_fee = 0.0004

# Получение стакана
order_book = exchange.fetch_order_book(symbol, limit=20)

def estimate_market_execution_cost(order_book, amount_usdt):
    total_cost = 0
    remaining = amount_usdt
    for price, volume in order_book['asks']:  # покупка по market
        vol_usdt = price * volume
        if vol_usdt >= remaining:
            total_cost += price * (remaining / price)
            break
        else:
            total_cost += vol_usdt
            remaining -= vol_usdt
    avg_price = total_cost / amount_usdt
    return avg_price

def estimate_limit_execution_cost(price, amount_usdt):
    return price  # если исполнится – цена фиксированная

# Расчёты
market_price = estimate_market_execution_cost(order_book, trade_amount_usdt)
limit_price = order_book['asks'][0][0]  # best ask

cost_market = market_price * (1 + taker_fee)
cost_limit = limit_price * (1 + maker_fee)

print(f"Market исполнение: {market_price:.2f} USDT (с учётом комиссии: {cost_market:.2f})")
print(f"Limit исполнение:  {limit_price:.2f} USDT (с учётом комиссии: {cost_limit:.2f})")

if cost_market < cost_limit:
    print("✅ Выгоднее использовать MARKET ордер")
else:
    print("✅ Выгоднее использовать LIMIT ордер")

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

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

📌 Donchian Channel Reversal – Стратегия разворота от границ канала Дончиана🧑‍💻

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

Канал Дончиана (Donchian Channel) строится по максимуму и минимуму за определённое количество баров (обычно 20).
Вместо пробойной логики, в этой стратегии используется отскок от границ канала, что подходит для флетового рынка или контртрендовой торговли.

Правила стратегии:💸

- Покупка, когда цена коснулась нижней границы и начинает отскакивать

- Продажа, когда цена достигла верхней границы и начинает разворачиваться вниз

import ccxt
import pandas as pd

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

# Получение данных
ohlcv = exchange.fetch_ohlcv(symbol, 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["donchian_high"] = df["high"].rolling(window=channel_period).max()
df["donchian_low"] = df["low"].rolling(window=channel_period).min()

# Сигналы разворота
df["signal"] = 0
df.loc[
    (df["low"] <= df["donchian_low"]) & (df["close"] > df["open"]),
    "signal"
] = 1  # Покупка после отскока от нижней границы

df.loc[
    (df["high"] >= df["donchian_high"]) & (df["close"] < df["open"]),
    "signal"
] = -1  # Продажа после отскока от верхней границы

# Последний сигнал
last = df.iloc[-1]
print(f"Последняя свеча: {last['ts']}")
print(f"Donchian High: {last['donchian_high']:.2f} | Low: {last['donchian_low']:.2f}")
print(f"Цена закрытия: {last['close']:.2f}")
print(f"Сигнал: {'BUY' if last['signal'] == 1 else 'SELL' if last['signal'] == -1 else 'NONE'}")

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

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