📌Autoencoder Anomaly Detection – Выявление рыночных аномалий через автоэнкодер🧑‍💻

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

Автоэнкодер — это тип нейросети, обучающийся восстанавливать входные данные.

Он отлично подходит для выявления аномалий: если сеть не может точно восстановить поведение рынка, значит текущее состояние аномально.

Применение в трейдинге:💸

- Выявление всплесков/скачков, нехарактерных движений, фейков, манипуляций

- Используется для фильтрации фейковых сигналов или входа в момент выхода за “норму”

import pandas as pd
import numpy as np
from keras.models import Model
from keras.layers import Input, Dense

# Данные
df = pd.read_csv("market_data.csv")  # Предположим, у вас есть OHLCV
features = ["close", "volume", "rsi", "ema", "atr"]
X = df[features].dropna()
X_scaled = (X - X.mean()) / X.std()

# Архитектура автоэнкодера
input_dim = X_scaled.shape[1]
input_layer = Input(shape=(input_dim,))
encoded = Dense(8, activation="relu")(input_layer)
decoded = Dense(input_dim, activation="linear")(encoded)

autoencoder = Model(input_layer, decoded)
autoencoder.compile(optimizer="adam", loss="mse")

# Обучение
autoencoder.fit(X_scaled, X_scaled, epochs=50, batch_size=32, shuffle=True, verbose=0)

# Вычисляем ошибку восстановления
recon = autoencoder.predict(X_scaled)
mse = np.mean((X_scaled - recon) ** 2, axis=1)
df["recon_error"] = mse

# Порог (например, 99-й перцентиль)
threshold = np.percentile(mse, 99)
df["anomaly"] = df["recon_error"] > threshold

print(df[["close", "recon_error", "anomaly"]].tail(10))

#инструмент

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

📌Cross-Exchange Funding Rate Arbitrage – Арбитраж на разнице ставок финансирования между биржами🧑‍💻

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

Funding Rate Arbitrage использует то, что разные биржи назначают разные ставки финансирования (funding rate) на бессрочные фьючерсы (perpetual contracts).

Цель стратегии — открыть противоположные позиции на двух биржах и зарабатывать на разнице funding rate, при этом:

- Сохраняется рыночный нейтралитет (цена не важна),

- Основная прибыль идёт от разницы в выплатах между биржами.💸

import ccxt

binance = ccxt.binance()
bybit = ccxt.bybit()

symbol = "BTC/USDT:USDT"

binance_funding = binance.fetch_funding_rate(symbol)["fundingRate"]
bybit_funding = bybit.fetch_funding_rate(symbol)["fundingRate"]

spread = binance_funding - bybit_funding

print(f"Binance funding: {binance_funding:.5f}")
print(f"Bybit funding:   {bybit_funding:.5f}")
print(f"Разница (спред): {spread:.5f}")

if abs(spread) > 0.0003:  # 0.03%
    if spread > 0:
        print("✅ Лонг на Binance, шорт на Bybit — положительное арбитражное плечо")
    else:
        print("✅ Лонг на Bybit, шорт на Binance — положительное арбитражное плечо")
else:
    print("Разница мала — арбитраж невыгоден")

#арбитраж

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

📌Day of Week Bias Strategy – Торговля по дневной сезонности🧑‍💻

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

Day of Week Bias (или сезонность по дням недели) — это стратегия, основанная на том, что в исторических данных может прослеживаться предсказуемое поведение рынка в определённые дни недели.

Например:💸

- BTC часто падал по понедельникам,

- Альткоины показывали рост по пятницам,

- Минимумы чаще формировались по вторникам и четвергам.

import ccxt
import pandas as pd

exchange = ccxt.binance()
symbol = "BTC/USDT"
ohlcv = exchange.fetch_ohlcv(symbol, timeframe="1d", limit=365)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")
df["day"] = df["ts"].dt.day_name()
df["return"] = df["close"].pct_change()

# Группируем по дням недели
stats = df.groupby("day")["return"].agg(["mean", "std", "count"])
stats["sharpe"] = stats["mean"] / stats["std"]

# Сортируем по средней доходности
stats = stats.sort_values(by="mean", ascending=False)
print(stats)

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

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

📌DeFi Gas Spike Strategy – Стратегия на основе всплеска комиссии в сети (Gas) как сигнал к панике или ажиотажу🧑‍💻

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

Комиссия (Gas price) в DeFi-сетях, таких как Ethereum, Arbitrum, BSC, Polygon, — это высокочувствительный индикатор поведения толпы.
Резкое увеличение цены газа (Gas Spike) указывает на:

- ажиотаж — массовое FOMO, запуск новых токенов, NFT, memecoins,

- панику — массовый выход из протоколов, скам, взломы, ликвидации.

Идея стратегии:💸

> Реагировать на всплески газа как на опережающий сигнал рыночной турбулентности:
или входить (при FOMO), или шортить (при панике).

import requests

def get_gas_price():
    url = "https://api.etherscan.io/api"
    params = {
        "module": "gastracker",
        "action": "gasoracle",
        "apikey": "YOUR_ETHERSCAN_API_KEY"
    }
    r = requests.get(url, params=params)
    data = r.json()
    return int(data["result"]["FastGasPrice"])

# Пример сигнала
gas = get_gas_price()
if gas > 100:
    print(f"⚠️ Высокий GAS: {gas} Gwei — возможна паника или ажиотаж")
elif gas < 40:
    print(f"GAS нормальный: {gas} Gwei")

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

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

📌Reinforcement Learning with Risk Constraints – Обучение с подкреплением с ограничением по риску🧑‍💻

Суть подхода:🛠️

Это стратегия, в которой агент обучается на основе Reinforcement Learning (RL) принимать торговые решения, учитывая не только прибыль, но и ограничения по риску:

- Максимальная просадка (Max Drawdown)

- Волатильность позиции (Stdev)

- Value at Risk (VaR)

- Ограничение на потери за эпизод/день

Цель💸 — найти оптимальную торговую политику, не превышая заданный риск-профиль.

Где это применимо:🚀

- Алготрейдинг с лимитами по просадке

- Криптофонды с ограничением на daily VaR

- Автоуправление позицией с учётом волатильности

- Агент для стратегии с мягким контролем риска

Пример среды(псевдо-код)💻

class TradingEnvWithRisk(gym.Env):
    def __init__(self):
        ...
        self.max_drawdown = 0.2  # ограничение
        self.volatility_limit = 0.05
        ...

    def step(self, action):
        ...
        pnl = self.calc_profit()
        risk = self.calc_volatility()  # или drawdown, VaR, CVaR
        penalty = self.lambda_ * max(0, risk - self.volatility_limit)
        reward = pnl - penalty
        ...
        return obs, reward, done, info

#инструмент

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

📌Bayesian Optimization for Hyperparameters – Автонастройка стратегий и ML-моделей🧑‍💻

Суть метода:🛠️

Байесовская оптимизация (Bayesian Optimization) — это метод поиска наилучших гиперпараметров для стратегии или модели,
при котором:

- Строится вероятностная модель зависимости гиперпараметры → результат (обычно Gaussian Process),

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

Метод быстрее и умнее, чем перебор (Grid Search) или случайный поиск (Random Search).

Где применяется в трейдинге:💸

- Оптимизация параметров стратегии (например: RSI период, TP/SL, ATR множитель)

- Настройка моделей ML (например: XGBoost: learning_rate, max_depth, n_estimators)

- Комбинирование индикаторов и фильтров

- Поиск оптимальных весов в портфеле

import optuna

def objective(trial):
    rsi_period = trial.suggest_int("rsi_period", 5, 30)
    tp = trial.suggest_float("take_profit", 0.01, 0.1)
    sl = trial.suggest_float("stop_loss", 0.01, 0.1)

    score = backtest_strategy(rsi_period, tp, sl)  # вернуть, например, Sharpe Ratio
    return score

study = optuna.create_study(direction="maximize")
study.optimize(objective, n_trials=50)

print("Лучшие параметры:", study.best_params)

#инструмент

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

📌Deep Q-Networks for Execution – Оптимизация исполнения сделок с помощью DQN🧑‍💻

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

Deep Q-Network (DQN) — это метод обучения с подкреплением, где агент обучается максимизировать ожидаемую награду, выбирая действия (дискретные) на основе текущего состояния.

В контексте оптимизации исполнения ордеров, DQN обучается решать задачу:💸

> Когда и как исполнять ордер (market/limit/split), чтобы минимизировать издержки (slippage, impact) и максимизировать итоговый результат.

import torch
import torch.nn as nn

class DQN(nn.Module):
    def __init__(self, input_dim, output_dim):
        super(DQN, self).__init__()
        self.net = nn.Sequential(
            nn.Linear(input_dim, 128), nn.ReLU(),
            nn.Linear(128, 128), nn.ReLU(),
            nn.Linear(128, output_dim)
        )

    def forward(self, x):
        return self.net(x)

#инструмент

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

📌High-Frequency Entry Strategy – Стратегия с большим количеством входов в позиции🧑‍💻

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

Данная стратегия предназначена для активной внутридневной торговли и работает по принципу множества краткосрочных входов с минимальной задержкой.
Подходит для скальпинга, ботов и ситуаций с высокой ликвидностью.

Принцип входа:💸

- Используются частые сигналы от осцилляторов и краткосрочных условий волатильности

- Вход при чётком импульсе на малом ТФ

- Выход через фиксированный TP/SL или по времени

- Можно выполнять 20+ входов в сутки

Используемые индикаторы:🚀

- RSI (период 5): краткосрочная перекупленность/перепроданность

- ATR: фильтрация по волатильности

- EMA 21 / EMA 50: подтверждение направления

- Volume Spike (объём выше среднего) — дополнительный триггер

import ccxt
import pandas as pd
import ta
import numpy as np

exchange = ccxt.binance()
symbol = "BTC/USDT"
timeframe = "5m"

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

# Индикаторы
df["rsi"] = ta.momentum.RSIIndicator(df["close"], window=5).rsi()
df["ema21"] = ta.trend.EMAIndicator(df["close"], window=21).ema_indicator()
df["ema50"] = ta.trend.EMAIndicator(df["close"], window=50).ema_indicator()
df["atr"] = ta.volatility.AverageTrueRange(df["high"], df["low"], df["close"], window=14).average_true_range()
df["volume_mean"] = df["volume"].rolling(20).mean()

# Сигналы
df["signal"] = 0

# Лонг-сигнал
df.loc[
    (df["rsi"] < 30) &
    (df["close"] > df["ema21"]) &
    (df["volume"] > df["volume_mean"]) &
    (df["atr"] > df["atr"].rolling(20).mean()),
    "signal"
] = 1

# Шорт-сигнал
df.loc[
    (df["rsi"] > 70) &
    (df["close"] < df["ema21"]) &
    (df["volume"] > df["volume_mean"]) &
    (df["atr"] > df["atr"].rolling(20).mean()),
    "signal"
] = -1

# Последние сигналы
signals = df[df["signal"] != 0][["ts", "close", "signal"]].tail(10)
print(signals)

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

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

📌Variational Autoencoders for Regime Clustering – Выявление рыночных фаз через вариационные автоэнкодеры🧑‍💻

Суть метода:🛠️

Variational Autoencoder (VAE) — это вероятностная нейросетевая модель, которая обучается сжимать данные в компактное латентное пространство,
а затем восстанавливать их с минимальной потерей информации.

Идея для трейдинга:💸

> Пропустить рыночные данные (например, свечи + индикаторы) через VAE,
Изучить распределение в латентном пространстве,
Затем кластеризовать его (например, через KMeans) и определить рыночные фазы:

- Флэт

- Тренд вверх

- Тренд вниз

- Волатильный переходный режим

Спекулятивный всплеск

import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler
from sklearn.cluster import KMeans
from tensorflow.keras import layers, Model

# Данные: окна по 30 свечей (OHLC)
X = np.load("windows_30.npy")  # форма (N, 30, 4)
X = X.reshape((X.shape[0], -1))  # flatten
X = StandardScaler().fit_transform(X)

# VAE: encoder
input_dim = X.shape[1]
inputs = layers.Input(shape=(input_dim,))
h = layers.Dense(64, activation="relu")(inputs)
z_mean = layers.Dense(8)(h)
z_logvar = layers.Dense(8)(h)

# Sampling
def sample(args):
    z_mean, z_logvar = args
    eps = np.random.normal(size=z_mean.shape)
    return z_mean + np.exp(0.5 * z_logvar) * eps

z = layers.Lambda(sample)([z_mean, z_logvar])

# Decoder
decoder_h = layers.Dense(64, activation="relu")(z)
outputs = layers.Dense(input_dim)(decoder_h)

vae = Model(inputs, outputs)
vae.compile(optimizer="adam", loss="mse")
vae.fit(X, X, epochs=50, batch_size=64, verbose=0)

# Получение латентных векторов
encoder = Model(inputs, z_mean)
Z = encoder.predict(X)

# Кластеризация в режимы
kmeans = KMeans(n_clusters=4).fit(Z)
regimes = kmeans.labels_

# Теперь можно сопоставить фазу рынку
df = pd.DataFrame({"regime": regimes})
print(df["regime"].value_counts())

#инструмент

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

📌Liquidity Sweep Reversal – стратегия на разворот после сбора ликвидности🧑‍💻

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

Многие внутридневные движения начинаются с "сбора стопов" — это когда цена резко пробивает локальные экстремумы (high/low), активирует стоп-ордера трейдеров,
и затем разворачивается в обратную сторону.

> Стратегия ловит момент, когда рынок сначала выбивает ликвидность, а затем делает разворот.

Условия входа:💸

1. Цена пробивает ближайший локальный high/low (например, за последние 4–8 свечей)

2. Всплеск объёма во время пробоя

3. Следующая свеча закрывается обратно внутрь диапазона (фейк-пробой)

4. Вход в разворот, стоп — за экстремум, тейк — до средней или противоположного края диапазона

import ccxt
import pandas as pd

exchange = ccxt.binance()
symbol = "BTC/USDT"
df = pd.DataFrame(exchange.fetch_ohlcv(symbol, "5m", limit=100),
                  columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")

# Диапазон последних 8 свечей (без текущей)
lookback = 8
df["prev_high"] = df["high"].shift(1).rolling(lookback).max()
df["prev_low"] = df["low"].shift(1).rolling(lookback).min()
df["volume_avg"] = df["volume"].rolling(lookback).mean()

# Условия пробоя и возврата
df["fake_breakout_long"] = (
    (df["high"] > df["prev_high"]) &
    (df["close"] < df["prev_high"]) &
    (df["volume"] > df["volume_avg"])
)

df["fake_breakout_short"] = (
    (df["low"] < df["prev_low"]) &
    (df["close"] > df["prev_low"]) &
    (df["volume"] > df["volume_avg"])
)

# Вывод сигналов
signals = df[(df["fake_breakout_long"] | df["fake_breakout_short"])]
signals["direction"] = signals.apply(
    lambda row: "LONG" if row["fake_breakout_long"] else "SHORT", axis=1)

print(signals[["ts", "close", "direction"]].tail())

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

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

📌Real-Time News Sentiment Dashboard – Панель новостного настроения в реальном времени🧑‍💻

Суть проекта:🛠️

Это веб-приложение или скрипт, которое в реальном времени собирает и анализирует новости,
выделяет ключевые события и определяет настроение (sentiment):

Позитив: рост, партнерства, листинг, инвестиции

Негатив: хак, делистинг, FUD, падение

Нейтрал: общий фон, обновления без эффекта

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

import requests
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer

API_KEY = "YOUR_CRYPTOPANIC_API_KEY"
analyzer = SentimentIntensityAnalyzer()

url = f"https://cryptopanic.com/api/v1/posts/?auth_token={API_KEY}&public=true"

r = requests.get(url)
news = r.json()["results"]

for item in news[:10]:
    title = item["title"]
    score = analyzer.polarity_scores(title)["compound"]
    sentiment = "positive" if score > 0.2 else "negative" if score < -0.2 else "neutral"
    print(f"[{sentiment.upper()}] {title}")

#инструмент

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

📌Net Unrealized Profit/Loss (NUPL) Strategy – Торговля по фазам метрики NUPL🧑‍💻

Что такое NUPL:🧨

NUPL (Net Unrealized Profit/Loss) — это ончейн-метрика, показывающая соотношение нереализованной прибыли и убытков всех участников сети.

> NUPL = (Market Cap – Realized Cap) / Market Cap

Если NUPL близок к 1 → большинство в прибыли (рынок перегрет)

Если NUPL близок к 0 или < 0 → большинство в убытке (капитуляция/дно)

Принцип стратегии:🛠️

- Мониторим значение NUPL, обновляемое ежедневно (с источников вроде Glassnode)

- Действуем при смене фазы (например, когда NUPL переходит из страха в надежду)

- Вход/выход сопровождается фильтром по цене/технике (например, пересечение EMA)

import os
import requests
import pandas as pd
from datetime import datetime
from dotenv import load_dotenv

load_dotenv()
API_KEY = os.getenv("GLASSNODE_API_KEY")

def get_nupl(asset="BTC", currency="USD"):
    url = "https://api.glassnode.com/v1/metrics/market/nupl"
    params = {
        "a": asset,
        "api_key": API_KEY,
        "i": "24h"
    }

    r = requests.get(url, params=params)
    if r.status_code != 200:
        raise Exception(f"Ошибка при запросе: {r.status_code}")

    data = pd.DataFrame(r.json())
    data["t"] = pd.to_datetime(data["t"], unit="s")
    data.set_index("t", inplace=True)
    return data.rename(columns={"v": "nupl"})

def classify_phase(nupl_val: float) -> str:
    if nupl_val > 0.75:
        return "🚨 Эйфория — фиксируем прибыль или шортим"
    elif nupl_val > 0.5:
        return "⚠️ Оптимизм — держим, но осторожно"
    elif nupl_val > 0.25:
        return "🟡 Надежда — можно входить"
    elif nupl_val > 0:
        return "🔵 Страх — фаза накопления"
    else:
        return "🟢 Крайний страх — дно, агрессивная покупка"

def main():
    df = get_nupl()
    today = df.iloc[-1]
    nupl_val = today["nupl"]
    date = today.name.strftime("%Y-%m-%d")

    print(f"[{date}] NUPL: {nupl_val:.4f}")
    print(classify_phase(nupl_val))

if __name__ == "__main__":
    main()

#инструмент

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

📌Volatility Arbitrage with Options – Арбитраж на расхождении implied и realized волатильности🧑‍💻

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

Volatility Arbitrage — это рыночно-нейтральная стратегия, основанная на торговле ожиданиями волатильности.

Суть:🧨

> Покупаешь или продаёшь волатильность через опционы, если implied volatility (IV) сильно отличается от realized volatility (RV) базового актива.

Ключевые понятия:💻

Implied Volatility (IV) Ожидаемая рынком волатильность (в цене опциона)

Realized Volatility (RV) Фактическая историческая волатильность (напр. 30д)

Торговая логика:💸

IV ≫ RV (опционы переоценены) Продавать волатильность (short straddle/strangle)

RV ≫ IV (опционы недооценены) Покупать волатильность (long straddle/strangle)

import ccxt
import pandas as pd
import numpy as np

exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv("BTC/USDT", "1d", limit=60)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])

# Реализованная волатильность (30д)
returns = df["close"].pct_change()
rv = returns.rolling(window=30).std() * np.sqrt(365)

print(f"Realized Volatility (30d): {rv.iloc[-1]:.2%}")

#арбитраж

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

📌Impulse Candle Detector – стратегия по импульсным свечам🧑‍💻

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

Импульсная свеча — это бар, который резко выделяется по длине тела и объёму.
Часто такая свеча указывает на:

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

- пробой уровня,

- ликвидации или вход крупного игрока.

Цель стратегии🚀 — обнаружить такие свечи и входить в направлении импульса или ждать отката.

Критерии импульсной свечи:💸

- Длина тела > 2× среднего за N свечей

- Объём выше среднего

- Тело ≥ 60% от всего диапазона (не просто хвост)

import ccxt
import pandas as pd

exchange = ccxt.binance()
symbol = "BTC/USDT"
df = pd.DataFrame(exchange.fetch_ohlcv(symbol, "5m", limit=100),
                  columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")

# Расчёт метрик
df["body"] = abs(df["close"] - df["open"])
df["range"] = df["high"] - df["low"]
df["body_ratio"] = df["body"] / df["range"]
df["body_avg"] = df["body"].rolling(20).mean()
df["vol_avg"] = df["volume"].rolling(20).mean()

# Импульс = длинное тело + высокая плотность + объём выше среднего
df["impulse"] = (df["body"] > 2 * df["body_avg"]) & \
                (df["body_ratio"] > 0.6) & \
                (df["volume"] > df["vol_avg"])

# Последние сигналы
signals = df[df["impulse"]].tail(5)
print("Импульсные свечи:")
print(signals[["ts", "open", "close", "volume"]])

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

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

📌VWAP Bounce Strategy – Отскок от VWAP в активных сессиях🧑‍💻

Суть:🛠️

VWAP (Volume Weighted Average Price) — это ключевой уровень для алгоритмической и институциональной торговли.

> Цена, приближающаяся к VWAP после импульса, часто отскакивает от него — особенно в EU и US сессии.

Условия входа:💸

- Цена приближается к VWAP менее чем на X% от VWAP
- Цена ранее отклонялась от VWAP > Y%
- Только в активные часы (например, 10:00–19:00 UTC)

import ccxt
import pandas as pd
from datetime import datetime

# Настройки
symbol = "BTC/USDT"
timeframe = "5m"
threshold_pct = 0.2  # % допустимого приближения к VWAP

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

# Расчёт VWAP
df["tp"] = (df["high"] + df["low"] + df["close"]) / 3
df["vwap_num"] = (df["tp"] * df["volume"]).cumsum()
df["vwap_den"] = df["volume"].cumsum()
df["vwap"] = df["vwap_num"] / df["vwap_den"]

# Сигнал: близость к VWAP + импульс ранее
current = df.iloc[-1]
previous = df.iloc[-10]

price = current["close"]
vwap = current["vwap"]
delta = abs(price - vwap) / vwap * 100
impulse = abs(previous["close"] - previous["vwap"]) / previous["vwap"] * 100

# Проверка времени
now_utc = datetime.utcnow().time()
is_session = now_utc >= datetime.strptime("10:00", "%H:%M").time() and \
             now_utc <= datetime.strptime("19:00", "%H:%M").time()

print("=" * 50)
print(f"[{current['ts']}] VWAP Bounce Monitor")
print(f"Цена: {price:.2f} | VWAP: {vwap:.2f} | Δ: {delta:.2f}% | Импульс до: {impulse:.2f}%")

if delta < threshold_pct and impulse > 0.5 and is_session:
    print("✅ Потенциальный отскок от VWAP! Рассмотреть вход.")
else:
    print("Нет условий для сигнала.")

print("=" * 50)

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

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

📌Session Range Breakout – стратегия на пробой диапазона торговой сессии🧑‍💻

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

Рынки часто "просыпаются" на стыке торговых сессий (Европа → США).
Если цена долгое время торговалась в узком диапазоне, пробой этого диапазона на высоком объёме — надёжный импульсный сигнал.

Цель стратегии — найти утренний флэт и входить при его пробое.🧨

Условия входа:💸

Формируется диапазон из N свечей (например, первые 3 свечи сессии)

- Объём растёт

- Цена пробивает high/low диапазона

- Вход в сторону пробоя

- Стоп — за границей диапазона, тейк — фиксированный R:R или трейлинг

import ccxt
import pandas as pd

exchange = ccxt.binance()
symbol = "BTC/USDT"
df = pd.DataFrame(exchange.fetch_ohlcv(symbol, "5m", limit=100),
                  columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")

# Формируем диапазон первых 3 свечей (например, с 07:00 UTC)
session_start = df.between_time("07:00", "07:15")
session_high = session_start["high"].max()
session_low = session_start["low"].min()
volume_avg = df["volume"].rolling(20).mean()

# Последняя свеча
current = df.iloc[-1]
signal = None

if current["volume"] > volume_avg.iloc[-1]:
    if current["close"] > session_high:
        signal = "BUY breakout"
    elif current["close"] < session_low:
        signal = "SELL breakdown"

# Алерт
print(f"Диапазон: {session_low:.2f} — {session_high:.2f}")
print(f"Цена: {current['close']:.2f}, Объём: {current['volume']:.2f}")
print("Сигнал:", signal or "нет")

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

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

📌Session Labels Generator – автоматическая разметка торговых сессий по времени🧑‍💻

Суть инструмента:🛠️

Многие стратегии зависят от времени суток: азиатская консолидация, европейский импульс, американский объем.
Этот инструмент автоматически помечает для каждой свечи, в какой сессии она была сформирована.

Цель — использовать это как фильтр или визуальный маркер в стратегии/графике/анализе.🧨

Как использовать:💻

- Фильтровать сигналы: например, торговать только в US-сессии

- Выделять фазы по сессиям (флэт Азия → пробой Европа)

- Построить тепловую карту: в какие часы актив выше/ниже

- Создать отдельную стратегию "Session Breakout"

import ccxt
import pandas as pd

exchange = ccxt.binance()
symbol = "BTC/USDT"
df = pd.DataFrame(exchange.fetch_ohlcv(symbol, "1h", limit=100),
                  columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")
df["hour"] = df["ts"].dt.hour

# Функция определения сессии
def label_session(hour):
    if 0 <= hour < 8:
        return "Asia"
    elif 7 <= hour < 15:
        return "Europe"
    elif 13 <= hour < 21:
        return "US"
    else:
        return "Overlap/Other"

df["session"] = df["hour"].apply(label_session)

# Пример вывода
print(df[["ts", "close", "session"]].tail(10))

#инструмент

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

📌Local Extrema Scanner – поиск локальных минимумов и максимумов в реальном времени🧑‍💻

Суть инструмента:🛠️

Местные экстремумы — это точки, где цена останавливает тренд и временно разворачивается.

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

- построения уровней поддержки/сопротивления,

- фиксации прибыли,

- подтверждения дивергенций,

- входа в контртренд.

import ccxt
import pandas as pd

exchange = ccxt.binance()
symbol = "BTC/USDT"
df = pd.DataFrame(exchange.fetch_ohlcv(symbol, "15m", limit=100),
                  columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")

# Параметры
window = 3

# Поиск локальных максимумов и минимумов
df["local_max"] = df["high"][(df["high"].shift(window) < df["high"]) &
                             (df["high"].shift(-window) < df["high"])]
df["local_min"] = df["low"][(df["low"].shift(window) > df["low"]) &
                            (df["low"].shift(-window) > df["low"])]

# Последние точки
extrema = df[["ts", "local_max", "local_min"]].dropna()
print("Локальные экстремумы:")
print(extrema.tail(5))

#инструмент

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

📌Candle Pressure Strategy – стратегия на основе "давления тела свечи"🧑‍💻

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

Каждая свеча содержит скрытую информацию о силе и направлении движения — особенно, если рассчитать:

> Насколько тело свечи жёстко прижато к high или low → это и есть давление, сигнализирующее о вероятном продолжении.

Цель стратегии — входить в сторону свечей с сильным давлением и подтверждением объёма.

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

- Длина тела свечи ≥ 2× среднего

- Тело «прижато» к high (лонг) или low (шорт)

- Объём выше среднего

- Свеча закрылась ближе к high/low ≥ 80%

import ccxt
import pandas as pd

exchange = ccxt.binance()
symbol = "BTC/USDT"
df = pd.DataFrame(exchange.fetch_ohlcv(symbol, "5m", limit=100),
                  columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")

df["body"] = abs(df["close"] - df["open"])
df["range"] = df["high"] - df["low"]
df["body_avg"] = df["body"].rolling(20).mean()
df["vol_avg"] = df["volume"].rolling(20).mean()
df["pressure_long"] = (df["close"] - df["low"]) / df["range"]
df["pressure_short"] = (df["high"] - df["close"]) / df["range"]

# Условия сигнала
df["signal_long"] = (
    (df["body"] > 2 * df["body_avg"]) &
    (df["volume"] > df["vol_avg"]) &
    (df["pressure_long"] > 0.8)
)

df["signal_short"] = (
    (df["body"] > 2 * df["body_avg"]) &
    (df["volume"] > df["vol_avg"]) &
    (df["pressure_short"] > 0.8)
)

# Вывод
signals = df[(df["signal_long"] | df["signal_short"])]
signals["direction"] = signals.apply(
    lambda x: "LONG" if x["signal_long"] else "SHORT", axis=1
)
print(signals[["ts", "close", "direction"]].tail(5))

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

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

📌Daily Range Analyzer – анализ суточного диапазона актива🧑‍💻

Суть инструмента:🛠️

Инструмент рассчитывает дневной диапазон движения (high – low) и сравнивает его:

- со средним значением за последние N дней,

- определяет, насколько день "активен" или "затухший",

помогает:🚀

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

- находить дни для пробойных стратегий,

- отслеживать приближение к границам дневного диапазона.

import ccxt
import pandas as pd

exchange = ccxt.binance()
symbol = "BTC/USDT"

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

# Расчёт диапазона
df["range"] = df["high"] - df["low"]
df["avg_range"] = df["range"].rolling(14).mean()

# Текущий день (если запуск в реальном времени — заменить на 1h и группировку)
current_range = df.iloc[-1]["range"]
avg_range = df["avg_range"].iloc[-1]
activity = current_range / avg_range * 100

print(f"Текущий дневной диапазон: {current_range:.2f}")
print(f"Средний за 14 дней:       {avg_range:.2f}")
print(f"Активность дня:           {activity:.1f}%")

if activity > 130:
    print("⚠️ Слишком активный день — возможен откат.")
elif activity < 70:
    print("⏳ Затухший день — вероятен пробой позже.")
else:
    print("✅ День в пределах нормы — можно использовать стандартные сигналы.")

#инструмент

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

📌Spread Trap Strategy – ловушка на резкое сужение спреда🧑‍💻

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

На низколатентных биржах (Binance, Bybit и др.) спред между bid/ask почти всегда минимален.
Но иногда, перед внезапным движением, крупные участники:

> искусственно сужают спред, создавая иллюзию "тишины", после чего мгновенно проламывают рынок в одну сторону.

Цель стратегии — отследить аномальное сужение спреда + рост дельты → и войти на пробой первым.🚀

Признаки ловушки:🧨

- Спред < 0.01% (искусственно зажат)

- Объём в лимитах по обе стороны внезапно увеличивается

- После нескольких секунд → всплеск дельты или объёма → движение

Алгоритм стратегии:💻

1. Постоянно читаем стакан

2. Если спред стал очень мал (например, < 0.05%) и объём по bid/ask резко вырос

3. Ожидаем всплеска объёма или односторонней агрессии (delta > X)

4. Вход в сторону импульса (обычно туда, где слабее плотность)

import json
import websockets
import asyncio

# Настройки
symbol = "btcusdt"
spread_threshold_pct = 0.02  # макс. допустимый спред (%)
volume_threshold = 100       # суммарный объём в 5 уровнях с обеих сторон

# Подключение к Binance WebSocket depth stream
url = f"wss://stream.binance.com:9443/ws/{symbol}@depth5@100ms"

async def monitor_orderbook():
    async with websockets.connect(url) as websocket:
        print(f"Подключено к Binance. Отслеживается пара {symbol.upper()}")
        while True:
            response = await websocket.recv()
            data = json.loads(response)

            bids = [(float(p), float(q)) for p, q in data["bids"]]
            asks = [(float(p), float(q)) for p, q in data["asks"]]

            if not bids or not asks:
                continue

            top_bid = bids[0][0]
            top_ask = asks[0][0]
            spread_pct = (top_ask - top_bid) / ((top_ask + top_bid) / 2) * 100

            total_bid_vol = sum([q for _, q in bids])
            total_ask_vol = sum([q for _, q in asks])

            if spread_pct < spread_threshold_pct and total_bid_vol > volume_threshold and total_ask_vol > volume_threshold:
                print("\n=== Обнаружена ловушка ликвидности ===")
                print(f"Спред: {spread_pct:.4f}%")
                print(f"Объём BID (5 ур.): {total_bid_vol:.2f}")
                print(f"Объём ASK (5 ур.): {total_ask_vol:.2f}")
                print(f"Цена BID: {top_bid:.2f} | ASK: {top_ask:.2f}")
                print("Сигнал: возможный резкий импульс\n")

# Запуск
asyncio.run(monitor_orderbook())

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

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