📌"Противоимпульсный выброс (Antipump Trap)"🧑‍💻

📌 Идея:🚀

Когда происходит резкий ценовой импульс вверх на повышенном объёме, многие новички входят в лонг на хаях. Но если цена не закрепляется выше импульса — вероятен разворот и падение. Мы используем это поведение толпы как ловушку.

📊 Условия входа в шорт:

1. Свеча (1m или 5m) с телом > 2× среднего тела за последние 20 свечей.

2. Объём на свече > 2× среднего объёма.

3. Следующая свеча закрывается ниже 50% тела предыдущей (не подтверждён импульс).

4. Открытие шорт-позиции со стопом чуть выше хая импульсной свечи.

🎯 TP/SL:

TP: 1.5× риск

SL: выше хая импульсной свечи

Дополнительный выход при росте объёма на падении (фиксация прибыли)

import ccxt
import pandas as pd
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
tf = '5m'

def fetch_ohlcv():
    data = exchange.fetch_ohlcv(symbol, timeframe=tf, limit=100)
    df = pd.DataFrame(data, columns=['time', 'open', 'high', 'low', 'close', 'volume'])
    df['body'] = abs(df['close'] - df['open'])
    return df

def detect_antipump(df):
    avg_body = df['body'][:-2].tail(20).mean()
    avg_vol = df['volume'][:-2].tail(20).mean()
    
    last = df.iloc[-2]
    confirm = df.iloc[-1]
    
    if last['body'] > 2 * avg_body and last['volume'] > 2 * avg_vol:
        body_mid = (last['open'] + last['close']) / 2
        if confirm['close'] < body_mid:
            print(f"🔻 Antipump сигнал на шорт ({symbol}) @ {confirm['close']}")
            print(f"   SL: {last['high']}, TP: ~{confirm['close'] - (last['high'] - confirm['close'])*1.5}")

while True:
    df = fetch_ohlcv()
    detect_antipump(df)
    time.sleep(15)

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

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

📌Автоматический Дневной Отчёт по Портфелю и Рынку🧑‍💻

Назначение:🧨

Каждое утро (или в выбранное время) скрипт автоматически собирает данные по рынку, активам, новостям, ключевым уровням и составляет краткий отчёт в консоль или в Telegram. Это личный дайджест трейдера — как утренний брифинг на десктопе.

✅ Функции:🛠️

- Портфель и изменение за сутки (% и $)

- Текущие уровни поддержки/сопротивления (по ATR/каналу/SR)

- Волатильность по ключевым монетам

- Горячие монеты дня (по объёму и импульсу)

- Важные экономические события (например, CPI, FOMC — можно через календарь событий)

- Сводка по рынку: BTC, ETH, фондовый индекс, золото

import ccxt
import datetime

exchange = ccxt.binance()
symbol_list = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT']

def fetch_summary(symbol):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe='1d', limit=2)
    today = ohlcv[-1]
    yesterday = ohlcv[-2]

    change = (today[4] - yesterday[4]) / yesterday[4] * 100
    return {
        'pair': symbol,
        'close': today[4],
        'change': round(change, 2)
    }

def daily_report():
    print(f"📅 Daily Market Summary: {datetime.date.today()}")
    for sym in symbol_list:
        summary = fetch_summary(sym)
        print(f"🔹 {summary['pair']}: ${summary['close']} ({summary['change']}%)")

if __name__ == "__main__":
    daily_report()

💡 Расширения:

- Добавить новости через API (например, CryptoPanic)

- Интеграция с Telegram Bot API

- Хранение истории в SQLite/CSV

- Графическое отображение динамики портфеля (через matplotlib или plotly)

- Оповещение, если дневное изменение > 5%

#инструмент

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

📌“Обратная Точка Баланса” (Reverse Balance Point)🧑‍💻

💡 Идея:🚀

Внутридневные и краткосрочные трейдеры часто используют уровни VWAP или среднюю цену как "баланс". Эта стратегия делает противоположное — ищет моменты, когда цена слишком долго избегает баланса, и ловит возможность возврата к средней цене.

⚙️ Условия входа (LONG):

1. Цена открылась и удерживается выше VWAP как минимум 2 часа.

2. После сильного роста (>1.5% от VWAP) цена резко уходит ниже VWAP на повышенном объёме.

3. Вход в LONG на первом отскоке от нижнего стандартного отклонения VWAP (или уровней -1/-2 SD).

4. Цель — возврат к VWAP, Stop — под предыдущий минимум.

📉 SHORT:

То же самое, но в зеркальном отражении (сначала удержание под VWAP, затем пробой вверх и возврат).

import ccxt
import pandas as pd
import numpy as np

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '15m'
limit = 120  # 30 часов

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

# VWAP
df['typical_price'] = (df['high'] + df['low'] + df['close']) / 3
df['cum_tp_vol'] = (df['typical_price'] * df['volume']).cumsum()
df['cum_vol'] = df['volume'].cumsum()
df['vwap'] = df['cum_tp_vol'] / df['cum_vol']

# Отношение к VWAP
df['distance'] = (df['close'] - df['vwap']) / df['vwap'] * 100

# Сигнал на возврат к балансу
last = df.iloc[-1]
prev = df.iloc[-2]

if prev['distance'] > 1.5 and last['distance'] < 0:
    print("🔁 Возможен возврат к VWAP: Рассмотреть LONG")

elif prev['distance'] < -1.5 and last['distance'] > 0:
    print("🔁 Возможен возврат к VWAP: Рассмотреть SHORT")
else:
    print("⏳ Нет сигнала. Ждём аномалии")

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

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

📌“Маркет-сканер волатильности с кластеризацией аномалий”🧑‍💻

🔍 Что делает:🛠️

Этот инструмент в реальном времени анализирует десятки монет (например, с Binance через ccxt) и ищет аномальные ценовые или объёмные всплески, применяя кластеризацию (например, DBSCAN или KMeans).

Он не просто говорит: "Цена выросла на 4%", а выявляет группы активов, которые ведут себя одинаково, но резко отличаются от других.

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

- Находить монеты в начале пампа.

- Выявлять нелогичные движения (арбитражные окна).

- Искать сигналы толпы / инсайдерского спроса.

import ccxt
import pandas as pd
from sklearn.cluster import DBSCAN
from sklearn.preprocessing import StandardScaler
import numpy as np

exchange = ccxt.binance()
symbols = [s['symbol'] for s in exchange.load_markets().values() if '/USDT' in s['symbol'] and s['active']]

data = []

for symbol in symbols[:50]:
    try:
        ohlcv = exchange.fetch_ohlcv(symbol, '5m')[-3:]
        df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        pct_change = (df['close'].iloc[-1] - df['close'].iloc[0]) / df['close'].iloc[0]
        vol_change = df['volume'].iloc[-1] - df['volume'].iloc[0]
        data.append([symbol, pct_change, vol_change])
    except:
        continue

df = pd.DataFrame(data, columns=['symbol', 'pct_change', 'vol_change'])

# Кластеризация
X = StandardScaler().fit_transform(df[['pct_change', 'vol_change']])
clustering = DBSCAN(eps=0.8, min_samples=3).fit(X)
df['cluster'] = clustering.labels_

# Выводим аномалии
anomalies = df[df['cluster'] == -1]
print("📊 Обнаружены аномалии:\n", anomalies)

#инструмент

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

📌“Алгокарта кросс-корреляций в реальном времени”🧑‍💻

📌 Что это?🧨

Скрипт, строящий динамическую тепловую карту кросс-корреляций между несколькими криптоактивами (например, BTC, ETH, SOL, BNB и т.д.) в реальном времени — на малых таймфреймах (1m/5m), и обновляющий её в терминале или графически.

🧠 Как помогает трейдеру/алготрейдеру?💸

Находит активы, которые ведут себя синхронно или наоборот расходятся, что важно:

- Для парного трейдинга

- Для арбитражных стратегий

- Для фильтрации сигналов

- Для контроля риска по скоррелированным активам

import ccxt
import pandas as pd
import time
from colorama import Fore, Style, init

init(autoreset=True)

symbols = ['BTC/USDT', 'ETH/USDT', 'BNB/USDT', 'SOL/USDT', 'XRP/USDT']
exchange = ccxt.binance()
timeframe = '1m'
limit = 100

def fetch_ohlcv(symbol):
    try:
        ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
        df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        return df['close']
    except Exception as e:
        print(f"[{symbol}] Ошибка: {e}")
        return pd.Series(dtype=float)

def print_correlation_matrix(data):
    df = pd.DataFrame(data)
    corr_matrix = df.corr()

    print("\nКросс-корреляция:")
    headers = corr_matrix.columns
    print("     " + "  ".join([f"{h[:4]}" for h in headers]))

    for i, row in corr_matrix.iterrows():
        row_str = f"{i[:4]}  "
        for val in row:
            if val > 0.75:
                row_str += Fore.GREEN + f"{val:.2f}" + Style.RESET_ALL + "  "
            elif val < -0.75:
                row_str += Fore.RED + f"{val:.2f}" + Style.RESET_ALL + "  "
            else:
                row_str += f"{val:.2f}  "
        print(row_str)

if __name__ == "__main__":
    data = {}
    for symbol in symbols:
        closes = fetch_ohlcv(symbol)
        if not closes.empty:
            data[symbol] = closes
        time.sleep(0.5)

    if len(data) >= 2:
        print_correlation_matrix(data)
    else:
        print("Недостаточно данных.")

#инструмент

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

📌“Обратный эхолот ордербука”🧑‍💻

🧠 Идея:🚀

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

⚙️ Как работает:🛠️

1. Сканируем стакан (order book) по выбранной паре с определённой частотой (например, каждые 5 секунд).

2. Ищем крупные заявки (например, 5x выше среднего).

3. Проверяем, исчезают ли они внезапно — это может быть признаком попытки манипулировать.

4. При исчезновении крупной заявки — открываем позицию против этого "давления":

Если исчез крупный buy — открываем short.

Если исчез крупный sell — открываем long.

📈 Почему это работает:

Алгоритмические трейдеры и крупные игроки используют "ложные" заявки для создания видимого давления. Когда заявка исчезает — это сигнал, что цена может пойти в обратную сторону.

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
threshold_multiplier = 5

def fetch_order_book(symbol):
    ob = exchange.fetch_order_book(symbol, limit=100)
    return ob['bids'], ob['asks']

def detect_spoof_disappearance(prev, curr):
    def large_orders_diff(old, new):
        large = []
        avg_size = sum([x[1] for x in old]) / len(old)
        for price, size in old:
            if size > avg_size * threshold_multiplier and not any(abs(price - p[0]) < 1e-4 for p in new):
                large.append((price, size))
        return large
    disappeared_bids = large_orders_diff(prev['bids'], curr['bids'])
    disappeared_asks = large_orders_diff(prev['asks'], curr['asks'])
    return disappeared_bids, disappeared_asks

last_ob = {'bids': [], 'asks': []}
print("⏳ Мониторинг ордербука...")

while True:
    try:
        bids, asks = fetch_order_book(symbol)
        if last_ob['bids'] and last_ob['asks']:
            lost_bids, lost_asks = detect_spoof_disappearance(last_ob, {'bids': bids, 'asks': asks})
            if lost_bids:
                print(f"⚠️ Исчезли крупные заявки BUY: {lost_bids} → сигнал на SHORT")
            if lost_asks:
                print(f"⚠️ Исчезли крупные заявки SELL: {lost_asks} → сигнал на LONG")
        last_ob = {'bids': bids, 'asks': asks}
    except Exception as e:
        print("Ошибка:", e)
    time.sleep(5)

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

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

📌"Гиперволатильный ретест"

(авторская идея, направлена на поиск резких выбросов цены и обратного возврата к уровню — со встроенным фильтром на тренд)🧑‍💻

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

1. Вход:

Когда происходит резкий импульс вверх или вниз (например, > 1.5% за 1 минуту), а через 1-3 свечи происходит откат обратно к предыдущему уровню (ретест) — открывается сделка в направлении возврата (на откат).

2. Фильтр тренда:

Стратегия работает только при наличии сильного направленного движения за последние 30 минут (например, изменение > 2.5%). То есть, торгует только в условиях тренда.

3. Индикаторы (опционально):

- EMA(20) и EMA(50) для фильтра тренда.

- ATR для оценки силы импульса.

- Отслеживание свечных паттернов "Pin Bar" или "Engulfing" на ретесте — как дополнительное подтверждение.

import ccxt
import pandas as pd
import time

symbol = 'BTC/USDT'
timeframe = '1m'
limit = 100

exchange = ccxt.binance()

def fetch_ohlcv():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['returns'] = df['close'].pct_change()
    df['ema20'] = df['close'].ewm(span=20).mean()
    df['ema50'] = df['close'].ewm(span=50).mean()
    df['atr'] = df['high'] - df['low']
    return df

def detect_signal(df):
    last = df.iloc[-1]
    prev = df.iloc[-2]

    # Тренд фильтр: EMA20 > EMA50
    in_trend = last['ema20'] > last['ema50']
    recent_movement = (df['close'].iloc[-1] - df['close'].iloc[-30]) / df['close'].iloc[-30] * 100

    if in_trend and recent_movement > 2.5:
        # Импульс был 2 свечи назад
        impulse = df['returns'].iloc[-3]
        retest = df['returns'].iloc[-1]

        if impulse > 0.015 and retest < -0.01:
            print("🔽 Short signal detected (Impulse up → Retest down)")

        elif impulse < -0.015 and retest > 0.01:
            print("🔼 Long signal detected (Impulse down → Retest up)")

if __name__ == "__main__":
    while True:
        try:
            df = fetch_ohlcv()
            detect_signal(df)
            time.sleep(60)
        except Exception as e:
            print("Ошибка:", e)
            time.sleep(10)

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

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

📌"AlgoMarketMap" — Карта активности и плотности алгоритмической торговли в реальном времени🧑‍💻

🔍 Суть инструмента:🚀

Визуализатор и анализатор мест концентрации алготорговой активности (ботов) по:

- частоте обновлений ордербука;

- всплескам объёма на уровнях;

- микродвижениям цены без объёма;

- аномально точным лимиткам (в миллисекунды);

- частым сменам позиции в стакане.

💡 Зачем это нужно:💸

- Помогает понять, где «сидят» алгоритмы, в какие моменты рынок управляется не людьми.

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

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

import asyncio
import websockets
import json
import time
from collections import deque

symbol = "btcusdt"
depth_url = f"wss://stream.binance.com:9443/ws/{symbol}@depth"

event_times = deque(maxlen=100)

async def monitor_depth():
    async with websockets.connect(depth_url) as websocket:
        while True:
            try:
                msg = await websocket.recv()
                data = json.loads(msg)

                now = time.time()
                event_times.append(now)

                # Оцениваем активность обновлений
                if len(event_times) > 10:
                    freq = len(event_times) / (event_times[-1] - event_times[0])
                    if freq > 30:
                        print(f"⚠️ Алготрейдерская активность обнаружена! Частота: {freq:.2f} обновлений/сек")

            except Exception as e:
                print("Ошибка:", e)
                await asyncio.sleep(5)

asyncio.run(monitor_depth())

#инструмент

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

📌"Delta Drift Reversal" — Интегральный разворот по дельте объёма и микроимпульсам🧑‍💻

💡 Идея стратегии:🛠️

Стратегия строится на анализе дельты объёма (разница между агрессивными покупками и продажами) и выявлении затухающих микроимпульсов, после которых вероятен разворот.

📌 Используется только рыночная информация: стакан, лента, объём и цена.

🔬 Условия входа (LONG):💸

1. На кластерной ленте видно:

Сильное агрессивное давление продавцов (дельта < -X);

Но цена не пробивает минимум, остаётся в диапазоне → затухание импульса.

2. В течение последних N минут:

Был хотя бы 1 мощный толчок вниз, но объём при повторном тесте ниже среднего.

3. Формируется "антиимпульс" — небольшое поглощение (delta становится положительной, цена идёт вверх).

4. Цена выше скользящего VWAP или среднего баланса за сессию.

🟢 → Вход в LONG

🔴 → Аналогично работает и в SHORT при обратной конфигурации.

📊 Фильтры:💻

- Не торгуется в низковолатильные сессии (определяется по ATR).

- Обязательно проверяется объём ленты и скорость сделок (tick rate).

- Выход по Take Profit = x2 от ближайшего скопления ликвидности (по дельте)
или после формирования противоположного "антиимпульса".

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
window = 5  # в минутах

def fetch_volume_delta(symbol, minutes=5):
    ohlcv = exchange.fetch_ohlcv(symbol, '1m', limit=minutes)
    delta = 0
    for candle in ohlcv:
        open_, high, low, close, volume = candle[1:6]
        direction = 1 if close > open_ else -1
        delta += volume * direction
    return delta

while True:
    delta = fetch_volume_delta(symbol, window)
    print(f"Δ объёма за {window} минут: {delta:.2f}")

    if delta < -1500:
        print("⚠️ Мощная распродажа! Следим за затуханием импульса...")
    elif delta > 1500:
        print("⚠️ Активация покупателей! Возможен отскок.")
    else:
        print("Нейтральное давление...")

    time.sleep(60)

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

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

📌SmartVol Scanner — это инструмент, который отслеживает кластерные накопления низкой волатильности (волатильные "зажимы") и прогнозирует вероятное направление разрядки на основе🧑‍💻:

- поведения тикового объёма,

- сужения диапазона (волатильность),

- направления агрессии (дельта),

- времени суток (часы активности),

- соотношения лимитных и рыночных ордеров.

⚙️ Что делает инструмент:🛠️

1. Находит “волатильные зажимы” — периоды, когда диапазон сужается (low volatility).

2. Отслеживает накопление объёма в этих узких диапазонах (squeeze zones).

3. Пытается предсказать, в какую сторону будет "выброс" (breakout):

- по агрессии дельты,

- по направлению объёма,

- по плотностям в стакане (опционально).

4. Генерирует сигналы

5. Идеален для алготрейдинговых моделей на пробой/отбой и построения сеток.

import ccxt
import time
import datetime
import statistics

symbol = 'BTC/USDT'
exchange = ccxt.binance({'enableRateLimit': True})
timeframe = '1m'
lookback_candles = 30
squeeze_threshold = 0.3  # < 0.3% средний диапазон — зажим
volume_growth_threshold = 1.5  # объем вырос в 1.5x
delay_sec = 10

def fetch_candles():
    return exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=lookback_candles)

def analyze(ohlcv):
    ranges = [(c[2] - c[3]) / c[1] * 100 for c in ohlcv if c[1] > 0]
    avg_range = statistics.mean(ranges)
    volumes = [c[5] for c in ohlcv]
    avg_volume = statistics.mean(volumes)
    latest_volume = volumes[-1]
    return avg_range, avg_volume, latest_volume

def detect(avg_range, avg_volume, latest_volume):
    if avg_range < squeeze_threshold:
        if latest_volume > avg_volume * volume_growth_threshold:
            return "⚡ Всплеск объёма после зажима: потенциальный импульс!"
        return "🕵️ Зажим волатильности. Объём пока обычный."
    return "✅ Волатильность нормальная. Сигналов нет."

print("📡 SmartVol Scanner запущен...")

while True:
    try:
        candles = fetch_candles()
        avg_range, avg_volume, last_volume = analyze(candles)
        signal = detect(avg_range, avg_volume, last_volume)
        print(f"[{datetime.datetime.now()}] Диапазон: {avg_range:.2f}% | Объём: {last_volume:.2f} | {signal}")
    except Exception as e:
        print(f"Ошибка: {e}")
    time.sleep(delay_sec)

#инструмент

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

📌"Order Book Mismatch Pressure"🧑‍💻

Идея:🛠️

Вместо технических индикаторов используется дисбаланс в лимитных заявках на покупку/продажу. Стратегия выявляет, когда маркет-участники массово "переставляют" стакан в сторону покупки или продажи — но цена ещё не среагировала. Это момент опережающего входа.

🔍 Условия входа в лонг:🚀

1. В стакане:

суммарный объём bid (покупка) > 3× объёма ask (продажа).

2. Последние 3 свечи — flat или слегка падающие (ещё нет роста).

3. Объёмы в стакане стабильны (нет спуфинга).

→ Вход в лонг по рынку или лимитке.

🛑 Выход:

Тейк: +0.5–1% (можно использовать трейлинг).

Стоп: ниже последнего локального минимума или фиксированный -0.3%.

✅ Почему это эффективно:💸

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

- Аналог предсказания импульса по изменению поведения стакана.

import ccxt
import time

# Конфигурация
SYMBOL = 'BTC/USDT'
THRESHOLD_MULTIPLIER = 3  # BID в 3 раза больше ASK = сигнал на лонг
SLEEP_INTERVAL = 5  # интервал между проверками в секундах

exchange = ccxt.binance({
    'enableRateLimit': True
})

def fetch_order_book(symbol):
    try:
        book = exchange.fetch_order_book(symbol)
        bid_volume = sum([bid[1] for bid in book['bids'][:10]])
        ask_volume = sum([ask[1] for ask in book['asks'][:10]])
        return bid_volume, ask_volume
    except Exception as e:
        print(f"[!] Ошибка получения стакана: {e}")
        return None, None

def monitor_order_book(symbol, threshold):
    while True:
        bid_volume, ask_volume = fetch_order_book(symbol)
        if bid_volume is None:
            time.sleep(SLEEP_INTERVAL)
            continue

        if bid_volume > ask_volume * threshold:
            print(f"[+] Сигнал на ЛОНГ: BID {bid_volume:.2f} > {threshold}× ASK {ask_volume:.2f}")
        elif ask_volume > bid_volume * threshold:
            print(f"[-] Сигнал на ШОРТ: ASK {ask_volume:.2f} > {threshold}× BID {bid_volume:.2f}")
        else:
            print(f"[=] Нет сигнала. BID: {bid_volume:.2f}, ASK: {ask_volume:.2f}")
        time.sleep(SLEEP_INTERVAL)

monitor_order_book(SYMBOL, THRESHOLD_MULTIPLIER)

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

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

📌"Volatility Clustering Analyzer"
(Анализ кластеризации волатильности с сигналами на повышенную вероятность движения цены)🧑‍💻

📌 Описание идеи:💻

Инструмент основан на эффекте кластеризации волатильности — это эмпирически наблюдаемое свойство рынков, когда периоды высокой волатильности сменяются высокой волатильностью, а низкой — низкой.

Смысл:🚀

- Выделяются «кластерные окна» резких движений.

- Когда появляется "затишье", модель отслеживает вероятность наступления нового импульса.

- Это можно использовать для входа в момент будущего движения цены (breakout или mean-reversion).

⚙️ Возможности инструмента:🛠️

- Расчёт текущего уровня кластеризации волатильности на основе ATR/стандартного отклонения.

- Автоматическое выявление фазы "сжатия" и сигнал о возможном выходе из неё.

- Генерация сигнала (в консоль или файл) при выходе за порог волатильности.

import ccxt
import pandas as pd
import numpy as np
import time
from datetime import datetime

# Инициализация биржи
exchange = ccxt.binance({
    'enableRateLimit': True,
})

symbol = 'BTC/USDT'
timeframe = '5m'
lookback_candles = 120
volatility_threshold = 1.5  # Множитель стандартного отклонения

def fetch_ohlcv(symbol, timeframe, limit=120):
    try:
        ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
        df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
        return df
    except Exception as e:
        print(f"Ошибка при получении данных: {e}")
        return None

def calculate_volatility(df):
    df['log_return'] = np.log(df['close'] / df['close'].shift(1))
    df.dropna(inplace=True)
    return df['log_return'].std() * np.sqrt(len(df)) * 100  # годовая волатильность в %

def main_loop():
    print("🔁 Старт анализа волатильности... (Ctrl+C для остановки)\n")
    prev_vol = None

    while True:
        df = fetch_ohlcv(symbol, timeframe, limit=lookback_candles)
        if df is None:
            time.sleep(10)
            continue

        current_vol = calculate_volatility(df)
        ts = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')

        if prev_vol is not None:
            ratio = current_vol / prev_vol
            if ratio > volatility_threshold:
                print(f"[{ts}] 🚨 Повышение волатильности! (x{ratio:.2f}) Текущая волатильность: {current_vol:.2f}%")
            else:
                print(f"[{ts}] 📊 Волатильность стабильна: {current_vol:.2f}%")
        else:
            print(f"[{ts}] 🔹 Начальное измерение волатильности: {current_vol:.2f}%")

        prev_vol = current_vol
        time.sleep(60)

if __name__ == "__main__":
    main_loop()

#инструмент

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

📌“Импульс с фильтром тренда” — твой фильтр шума рынка🧑‍💻

🔍 Мы заходим в рынок только когда есть чёткое направление, подтверждённый объём и реальный импульс.

⚙️ Фильтры стратегии:🛠️

– EMA 50/200 → определяем тренд

– ADX > 25 → только сильное движение

– Пробой high/low на объёме

– RSI не в зоне перекупленности/перепроданности

✅ Вход — строго по направлению тренда

❌ Нет тренда? Нет сделки.

📊 Работает на крипте, фьючах, акциях.

🔁 Бэктест — положительный.

import ccxt
import pandas as pd
import ta
import time

# --- Параметры стратегии ---
symbol = 'BTC/USDT'
timeframe = '1h'
limit = 300

# --- Подключение к бирже ---
exchange = ccxt.binance()
exchange.load_markets()

# --- Получение исторических свечей ---
def get_ohlcv():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    return df

# --- Расчёт индикаторов ---
def add_indicators(df):
    df['ema50'] = ta.trend.ema_indicator(df['close'], window=50).ema_indicator()
    df['ema200'] = ta.trend.ema_indicator(df['close'], window=200).ema_indicator()
    df['adx'] = ta.trend.adx(df['high'], df['low'], df['close'], window=14).adx()
    df['rsi'] = ta.momentum.RSIIndicator(df['close'], window=14).rsi()
    df['atr'] = ta.volatility.AverageTrueRange(df['high'], df['low'], df['close'], window=14).average_true_range()
    df['volume_ma'] = df['volume'].rolling(20).mean()
    return df

# --- Логика сигналов ---
def get_signal(df):
    latest = df.iloc[-1]
    previous = df.iloc[-2]

    if (
        latest['ema50'] > latest['ema200'] and
        latest['adx'] > 25 and
        latest['close'] > previous['high'] and
        latest['volume'] > latest['volume_ma'] and
        latest['rsi'] < 70
    ):
        return "🔼 LONG SIGNAL"

    if (
        latest['ema50'] < latest['ema200'] and
        latest['adx'] > 25 and
        latest['close'] < previous['low'] and
        latest['volume'] > latest['volume_ma'] and
        latest['rsi'] > 30
    ):
        return "🔽 SHORT SIGNAL"

    return None

# --- Основной цикл ---
def run():
    while True:
        try:
            df = get_ohlcv()
            df = add_indicators(df)
            signal = get_signal(df)

            last_time = df['timestamp'].iloc[-1]

            if signal:
                print(f"[{last_time}] {signal}")
            else:
                print(f"[{last_time}] Нет сигнала")

            time.sleep(60 * 60)  # Пауза на 60 минут

        except Exception as e:
            print("Ошибка:", e)
            time.sleep(10)

if __name__ == "__main__":
    run()

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

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

📌Smart Position Recovery Simulator (SPRS)🧑‍💻
Симулятор восстановления убыточных позиций по стратегиям усреднения и реверса с расчётом оптимальных шагов и оценки риска.

💡 Идея🚀

Оригинальный инструмент, необходимый для алготрейдеров, торгующих с частыми усреднениями или вынужденными реверсами. Вместо тупого «доливания» — он помогает протестировать и подобрать оптимальные уровни входа и объёмы, чтобы минимизировать убыток или даже выйти в прибыль.

📈 Что делает инструмент:🛠️

1. Загружает исторические данные по цене (через ccxt или локально).

2. Позволяет задать начальный убыточный вход.

3. Строит сценарии восстановления позиции:

- шаговое усреднение,

- агрессивное удвоение,

- реверс с последующим возвратом.

import ccxt
import time
from datetime import datetime, timedelta

# === Настройки ===
api_key = 'YOUR_API_KEY'
api_secret = 'YOUR_SECRET_KEY'
symbol = 'BTC/USDT'
amount = 0.001
price_drop_threshold = 1.5  # %
price_rise_target = 2.0     # %
check_interval = 60         # сек

# === Инициализация биржи ===
exchange = ccxt.binance({
    'apiKey': api_key,
    'secret': api_secret,
    'enableRateLimit': True,
})

# === История входов ===
entry_price = None

def get_recent_prices(symbol, minutes=5):
    since = exchange.milliseconds() - minutes * 60 * 1000
    candles = exchange.fetch_ohlcv(symbol, timeframe='1m', since=since)
    return [c[4] for c in candles]  # Закрытия

while True:
    try:
        print(f'\n[{datetime.now().strftime("%H:%M:%S")}] Проверка рынка...')
        prices = get_recent_prices(symbol)
        if len(prices) < 2:
            print("Недостаточно данных.")
            time.sleep(check_interval)
            continue

        current_price = prices[-1]
        max_recent = max(prices)
        price_drop = (max_recent - current_price) / max_recent * 100

        if entry_price:
            profit = (current_price - entry_price) / entry_price * 100
            print(f"Актуальная цена: {current_price:.2f} | Прибыль: {profit:.2f}%")

            if profit >= price_rise_target:
                print("🎯 Цель достигнута. Продаем...")
                exchange.create_market_sell_order(symbol, amount)
                print("✅ Продажа выполнена.")
                entry_price = None
        else:
            print(f"Актуальная цена: {current_price:.2f} | Падение за 5м: {price_drop:.2f}%")
            if price_drop >= price_drop_threshold:
                print("📉 Условие входа выполнено. Покупаем...")
                exchange.create_market_buy_order(symbol, amount)
                entry_price = current_price
                print(f"✅ Покупка по цене {entry_price:.2f}")

    except Exception as e:
        print(f"[Ошибка] {e}")

    time.sleep(check_interval)

#инструмент

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

📌Delta Pressure Breakout🧑‍💻

💡 Суть🧨:

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

🔍 Основные условия:🚀

Используется кластерный объем (bid vs ask).

Вычисляется дельта: delta = buy_volume - sell_volume.

Вход в сторону преобладающего объема при превышении порога.

Тейк и стоп — фиксированные или по ближайшим уровням ликвидности.

✅ Условия входа в сделку:💸

- Объём дельты за 1 минуту превышает среднюю дельту за последние 20 свечей × коэффициент (например, 2.5).

- Цена пробивает локальный уровень (high/low консолидации).

- Подтверждение: увеличение общего объема (vol) и удержание цены выше/ниже пробитого уровня.

import ccxt
import time
from datetime import datetime, timedelta

# Конфигурация
symbol = 'BTC/USDT'
timeframe = '1m'
lookback = 20
volume_multiplier = 2.5

exchange = ccxt.binance()

# Функция получения OHLCV

def get_ohlcv(symbol, timeframe, limit):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    return [
        {
            'timestamp': o[0],
            'open': o[1],
            'high': o[2],
            'low': o[3],
            'close': o[4],
            'volume': o[5]
        }
        for o in ohlcv
    ]

# Функция расчета дельты (proxy через close-open * volume)

def calculate_deltas(candles):
    deltas = []
    for c in candles:
        price_delta = c['close'] - c['open']
        delta = price_delta * c['volume']
        deltas.append(delta)
    return deltas

# Основная логика

def strategy_loop():
    candles = get_ohlcv(symbol, timeframe, lookback + 1)
    deltas = calculate_deltas(candles[:-1])  # исключить последнюю свечу

    avg_delta = sum(map(abs, deltas)) / len(deltas)

    last_candle = candles[-1]
    last_delta = (last_candle['close'] - last_candle['open']) * last_candle['volume']

    if abs(last_delta) > avg_delta * volume_multiplier:
        direction = 'LONG' if last_delta > 0 else 'SHORT'
        print(f"[{datetime.utcnow()}] Сигнал {direction}! Дельта: {last_delta:.2f}, Средняя: {avg_delta:.2f}")
    else:
        print(f"[{datetime.utcnow()}] Нет сигнала. Дельта: {last_delta:.2f}, Средняя: {avg_delta:.2f}")

# Цикл стратегии
if __name__ == '__main__':
    while True:
        try:
            strategy_loop()
            time.sleep(60)
        except Exception as e:
            print(f"[Ошибка] {e}")
            time.sleep(10)

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

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

📌AI Market Regime Detector -
идентификатор рыночного режима с автоадаптацией стратегий 🧑‍💻

💡 Суть:🚀

Инструмент автоматически определяет текущий рыночный режим — тренд, флет, высокая волатильность, паника и пр. — и переключает используемую стратегию под этот режим в реальном времени. Использует технические индикаторы, волатильность, корреляции, объёмы и ML-модель (например, RandomForestClassifier или LSTM), чтобы определить:

🟢 Восходящий тренд → трендовая стратегия (breakout)

🔴 Нисходящий тренд → short-приоритет

🟡 Боковик → стратегия на возврат к среднему

🔵 Высокая волатильность → ожидание и широкий стоп

⚫ Паника (чёрный лебедь) → стоп всех торгов

⚙️ Что делает скрипт:🛠️

- Скачивает данные с Binance через ccxt

- Обрабатывает индикаторы (ATR, Bollinger Bands, EMA)

- Распознаёт режим рынка по ML-классификатору

- В консоли пишет: режим = "восходящий тренд" и предлагает подходящие действия

- Можно подключить к основному боту через API/файл

import ccxt
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from datetime import datetime
import time

symbol = 'BTC/USDT'
timeframe = '1h'
limit = 500

exchange = ccxt.binance()

def fetch_ohlcv(symbol, timeframe='1h', limit=500):
    data = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    return df

def add_features(df):
    df['returns'] = df['close'].pct_change()
    df['volatility'] = df['returns'].rolling(window=10).std()
    df['ema'] = df['close'].ewm(span=20).mean()
    df.dropna(inplace=True)
    return df

def label_market(df):
    labels = []
    for i in range(len(df)):
        r = df['returns'].iloc[i]
        v = df['volatility'].iloc[i]
        if v > 0.01:
            labels.append("volatile")
        elif r > 0.002:
            labels.append("bullish")
        elif r < -0.002:
            labels.append("bearish")
        else:
            labels.append("sideways")
    df['label'] = labels
    return df

def train_model(df):
    features = df[['returns', 'volatility', 'ema']].copy()
    features = (features - features.mean()) / features.std()
    model = RandomForestClassifier(n_estimators=100)
    model.fit(features, df['label'])
    return model

def predict_current_state(model, df):
    last_row = df.iloc[-1:]
    features = last_row[['returns', 'volatility', 'ema']]
    features = (features - df[['returns', 'volatility', 'ema']].mean()) / df[['returns', 'volatility', 'ema']].std()
    prediction = model.predict(features)[0]
    return prediction

def main():
    df = fetch_ohlcv(symbol, timeframe, limit)
    df = add_features(df)
    df = label_market(df)
    model = train_model(df)

    while True:
        df = fetch_ohlcv(symbol, timeframe, limit)
        df = add_features(df)
        current_signal = predict_current_state(model, df)
        timestamp = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')
        print(f"[{timestamp}] Текущий рыночный режим: {current_signal}")
        time.sleep(60 * 60)  # обновление каждый час

if __name__ == '__main__':
    main()

#инструмент

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

📌Liquidity Pulse Divergence (LPD)🧑‍💻

💡 Идея:🚀

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

📈 Основные сигналы:💸

Buy signal: объём маркет-покупок значительно превышает продажи, но лимитные ордера на продажу продолжают расти (у кого-то крупная пассивная подача ликвидности → рынок может развернуться).

Sell signal: аналогично, но наоборот.

import ccxt
import time
from datetime import datetime

symbol = 'BTC/USDT'
exchange = ccxt.binance()
depth_limit = 5
interval_sec = 10

def get_order_book():
    ob = exchange.fetch_order_book(symbol)
    top_bids = sum([order[1] for order in ob['bids'][:depth_limit]])
    top_asks = sum([order[1] for order in ob['asks'][:depth_limit]])
    return top_bids, top_asks

def get_recent_trades():
    since = exchange.milliseconds() - 60 * 1000  # последняя минута
    trades = exchange.fetch_trades(symbol, since=since)
    buy_volume = sum(t['amount'] for t in trades if t['side'] == 'buy')
    sell_volume = sum(t['amount'] for t in trades if t['side'] == 'sell')
    return buy_volume, sell_volume

def analyze_and_signal():
    buy_vol, sell_vol = get_recent_trades()
    bid_liq, ask_liq = get_order_book()

    signal = "NEUTRAL"
    if buy_vol > sell_vol * 2 and ask_liq > bid_liq * 1.5:
        signal = "SHORT"
    elif sell_vol > buy_vol * 2 and bid_liq > ask_liq * 1.5:
        signal = "LONG"

    print(f"[{datetime.now().strftime('%H:%M:%S')}] "
          f"BuyVol: {buy_vol:.2f}, SellVol: {sell_vol:.2f}, "
          f"BidLiq: {bid_liq:.2f}, AskLiq: {ask_liq:.2f} → Signal: {signal}")

try:
    while True:
        analyze_and_signal()
        time.sleep(interval_sec)
except KeyboardInterrupt:
    print("Остановка...")

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

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

📌Volatility-Zone Auto-Mapper🧑‍💻

🔍 Назначение: Автоматически определяет и помечает зоны волатильности (импульсные и затухающие участки) на основе ATR и плотности свечей.

⚙️ Тип: Аналитический помощник для визуального и алгоритмического определения зон интереса (breakout, консолидация, зоны входа).

📈 Цель: Предиктивная оценка вероятности движения цены после сжатия или расширения волатильности.

🔧 Как это работает:🛠️

1. Получает исторические свечи (через ccxt).

2. Считает ATR, и отмечает участки с:

Резким ростом ATR → зона импульса

Затухающим ATR и высокой плотностью свечей → зона консолидации

3. Выводит в консоль участки времени и ценовые зоны.

4. Генерирует сигналы: BREAKOUT INCOMING, LOW VOLATILITY, HIGH RISK ENTRY, и др.

import ccxt
import pandas as pd
import numpy as np

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '5m'
limit = 500

def fetch_ohlcv():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    return df

def calculate_atr(df, period=14):
    df['hl'] = df['high'] - df['low']
    df['hc'] = abs(df['high'] - df['close'].shift())
    df['lc'] = abs(df['low'] - df['close'].shift())
    df['tr'] = df[['hl', 'hc', 'lc']].max(axis=1)
    df['atr'] = df['tr'].rolling(window=period).mean()
    return df

def detect_zones(df):
    zones = []
    for i in range(20, len(df)):
        recent = df.iloc[i-10:i]
        atr_slope = (recent['atr'].iloc[-1] - recent['atr'].iloc[0]) / 10
        if atr_slope > 0.05:
            zones.append((df['timestamp'].iloc[i], 'IMPULSE ZONE'))
        elif recent['atr'].mean() < recent['atr'].max() * 0.5:
            zones.append((df['timestamp'].iloc[i], 'CONSOLIDATION ZONE'))
    return zones

df = fetch_ohlcv()
df = calculate_atr(df)
zones = detect_zones(df)

print(f"🔍 Анализ зон для {symbol} ({timeframe}):\n")
for z in zones[-10:]:
    print(f"{z[0]} — {z[1]}")

#инструмент

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

📌Граф Анализа Корреляций с Аномалиями🧑‍💻

Инструмент строит живой граф взаимосвязей между криптоактивами и ищет аномалии в корреляции.

💡 Суть:🚀

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

🧠 Что делает инструмент:🛠️

- Загружает котировки 10–50 криптоактивов с Binance через ccxt.

- Строит матрицу корреляций на заданном таймфрейме.

- Находит "стабильные пары" (корреляция > 0.85).

- Следит в реальном времени за изменениями этой корреляции.

- Генерирует сигнал, если происходит "декорреляция" (например, корреляция пары BTC/ETH → 0.9 → 0.3).

import ccxt
import pandas as pd
import networkx as nx
import matplotlib.pyplot as plt

# Настройки
symbols = ['BTC/USDT', 'ETH/USDT', 'BNB/USDT', 'XRP/USDT', 'ADA/USDT']
timeframe = '1h'
limit = 100
correlation_threshold = 0.85
decorrelation_threshold = 0.3

exchange = ccxt.binance()

# Получение закрытий
def fetch_closes(symbol):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('timestamp', inplace=True)
    return df['close']

# Основной процесс
def main():
    prices = {}
    for symbol in symbols:
        try:
            prices[symbol] = fetch_closes(symbol)
        except Exception as e:
            print(f"Ошибка при загрузке {symbol}: {e}")

    df = pd.DataFrame(prices)
    corr = df.corr()

    # Граф корреляций
    G = nx.Graph()
    for i in corr.columns:
        for j in corr.columns:
            if i != j:
                c = corr[i][j]
                if abs(c) > correlation_threshold:
                    G.add_edge(i, j, weight=round(c, 2))

    # Визуализация
    nx.draw(G, with_labels=True, node_color='skyblue', edge_color='gray', font_weight='bold')
    plt.title("Корреляции криптовалют")
    plt.show()

    # Поиск декорреляций
    print("\n🚨 Сигналы декорреляции:")
    for i in corr.columns:
        for j in corr.columns:
            if i != j:
                c = corr[i][j]
                if abs(c) < decorrelation_threshold:
                    print(f"{i} и {j} имеют слабую корреляцию: {c:.2f}")

if __name__ == "__main__":
    main()

#инструмент

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

📌“Аномалия ускорения объёма”🧑‍💻

Отслеживает резкое увеличение торговой активности (объёма) за последние N интервалов и сравнивает с текущим значением.
Когда текущий объём в 3+ раза выше среднего — генерируется сигнал.

⚙️ Логика стратегии:🛠️

- Собираем исторические свечи (например, 1m).

- Рассчитываем средний объём за последние N свечей.

- Сравниваем текущий объём со средним.

- Если превышение в 3 раза — сигнал "LONG" (если свеча зелёная) или "SHORT" (если свеча красная).

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

🧠 Преимущества:🧨

- Подходит для интрадей/скальпинга.

- Эффективна при низкой волатильности до начала тренда.

- Простота и скорость анализа.

import ccxt
import time
from datetime import datetime

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '1m'
volume_multiplier = 3
lookback_candles = 20
sleep_interval = 60  # 1 минута

def fetch_candles():
    candles = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=lookback_candles + 1)
    return candles

def analyze_volume_spike():
    candles = fetch_candles()

    volumes = [c[5] for c in candles[:-1]]  # объёмы без последней свечи
    avg_volume = sum(volumes) / len(volumes)
    last = candles[-1]
    last_open, last_close, last_volume = last[1], last[4], last[5]

    signal = None
    if last_volume > avg_volume * volume_multiplier:
        if last_close > last_open:
            signal = 'LONG'
        elif last_close < last_open:
            signal = 'SHORT'

    print(f"[{datetime.now().strftime('%H:%M:%S')}] "
          f"LastVol: {last_volume:.2f}, AvgVol: {avg_volume:.2f} → Signal: {signal or 'NO SIGNAL'}")

while True:
    try:
        analyze_volume_spike()
        time.sleep(sleep_interval)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(10)

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

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