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

📌“Market Microstructure Pulse” (MMPulse)🧑‍💻

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

📌 Описание:🛠️

Market Microstructure Pulse (MMPulse) — это Python-инструмент, который анализирует:

1. Скорость изменения Bid/Ask-объёмов (Order Book Imbalance)

2. Спайки агрессивных маркет-ордеров

3. Изменение спреда

4. Аномалии в глубине стакана (на уровне 5+)

5. Скорость отмены лимитных ордеров (Order Book Flip)

И на основе этого выдаёт сигналы в консоль.

📈 Цель инструмента:💸

- Выявить точки временного дисбаланса ликвидности, где цена может резко сдвинуться.

- Используется как индикатор для высокочастотных входов (scalping) или алгосистем со сверхточкой входа.

import ccxt
import time

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

def fetch_orderbook(symbol, depth=5):
    ob = exchange.fetch_order_book(symbol, limit=depth)
    bid_vol = sum([level[1] for level in ob['bids']])
    ask_vol = sum([level[1] for level in ob['asks']])
    imbalance = (bid_vol - ask_vol) / (bid_vol + ask_vol + 1e-9)
    spread = ob['asks'][0][0] - ob['bids'][0][0]
    return {
        'bid_vol': bid_vol,
        'ask_vol': ask_vol,
        'imbalance': imbalance,
        'spread': spread
    }

def monitor_market():
    print("🔍 MMPulse запущен...")
    while True:
        try:
            data = fetch_orderbook(symbol)
            imbalance = data['imbalance']
            spread = data['spread']
            if imbalance > 0.6:
                print(f"📈 Лонг дисбаланс: {imbalance:.2f}, Spread: {spread:.2f}")
            elif imbalance < -0.6:
                print(f"📉 Шорт дисбаланс: {imbalance:.2f}, Spread: {spread:.2f}")
            time.sleep(2)
        except Exception as e:
            print(f"⚠️ Ошибка: {e}")
            time.sleep(3)

monitor_market()

#инструмент

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

📌"Impulse Fade Reversal"🧑‍💻

🧠 Идея:🛠️

Использовать резкий импульс (вверх или вниз) как сигнал на потенциальное истощение движения — и зайти в противоположную сторону при подтверждении “затухания” (fade).

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

1. Импульс:

Цена за последние N свечей выросла/упала более чем на X%.

Например: рост/падение > 2% за 3 свечи (5m таймфрейм).

2. Фэйд (затухание):

Последняя свеча закрывается ниже (при лонге) или выше (при шорте) тела предыдущей свечи.

Объём падает (подтверждение ослабления импульса).

3. Сигнал: 🚀

Вход в противоположную сторону.

Стоп: за экстремум импульса.

Тейк: 1.5–2× риск или возврат к средней цене.

import ccxt
import pandas as pd

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

def fetch_data():
    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 detect_impulse_fade(df, percent=2.0, candles=3):
    df['returns'] = df['close'].pct_change(periods=candles) * 100
    last_ret = df['returns'].iloc[-1]
    fade = df['close'].iloc[-1] < df['close'].iloc[-2]  # для лонга
    signal = None

    if last_ret > percent and fade:
        signal = '📉 SHORT сигнал (импульс вверх, фэйд вниз)'
    elif last_ret < -percent and df['close'].iloc[-1] > df['close'].iloc[-2]:
        signal = '📈 LONG сигнал (импульс вниз, фэйд вверх)'

    return signal

df = fetch_data()
signal = detect_impulse_fade(df)

print(f"🕒 Последняя свеча: {df['timestamp'].iloc[-1]}")
if signal:
    print(signal)
else:
    print("❌ Сигналов нет")

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

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

📌"Alpha Tracker" — Реальное сравнение вашей стратегии с рынком🧑‍💻

📌 Идея:🧨

Инструмент позволяет в реальном времени сравнивать доходность вашей стратегии с базовым активом (например, BTC/USDT). Это помогает:

- Понять, обгоняет ли стратегия рынок (генерирует alpha).

- Измерить эффективность любой стратегии: алгоритмической, ручной, сигнальной и т.п.

- Визуализировать накопленную доходность стратегии и сравнить её с buy & hold.

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

1. Вы подаёте ваши сигналы (например, список входов/выходов).

2. Скрипт рассчитывает доходность вашей стратегии.

3. Скрипт параллельно считает доходность пассивного владения активом (HODL).

4. В консоли (или графике) сравниваются кумулятивные результаты.

import ccxt
import time
import pandas as pd

# Пример входов и выходов (timestamp в секундах, тип сделки, цена будет подставляться)
# Пример: вход и выход по BTC
strategy_signals = [
    {"timestamp": 1720000000, "action": "buy"},
    {"timestamp": 1720005000, "action": "sell"},
    {"timestamp": 1720010000, "action": "buy"},
    {"timestamp": 1720020000, "action": "sell"}
]

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

def fetch_price_at(timestamp):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe='1m', since=timestamp * 1000, limit=1)
    return ohlcv[0][4] if ohlcv else None

def evaluate_strategy(signals):
    returns = []
    for i in range(0, len(signals)-1, 2):
        buy_price = fetch_price_at(signals[i]['timestamp'])
        sell_price = fetch_price_at(signals[i+1]['timestamp'])
        if buy_price and sell_price:
            pnl = (sell_price - buy_price) / buy_price
            returns.append(pnl)
    return returns

def evaluate_hodl(start_time, end_time):
    buy = fetch_price_at(start_time)
    sell = fetch_price_at(end_time)
    return (sell - buy) / buy if buy and sell else None

# Запуск
strategy_returns = evaluate_strategy(strategy_signals)
hodl_return = evaluate_hodl(strategy_signals[0]['timestamp'], strategy_signals[-1]['timestamp'])

print("\n📊 Доходность стратегии:")
for i, r in enumerate(strategy_returns):
    print(f"Сделка {i+1}: {r:.2%}")

print(f"\n📈 Доходность HODL за тот же период: {hodl_return:.2%}")
print(f"📐 Итоговое alpha: {sum(strategy_returns) - hodl_return:.2%}")

#инструмент

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

📌“Panic Bid Reversal” (PBR Strategy)🧑‍💻

📈 Стратегия ловли капитуляций с автоматическим определением "панической свечи" и входом в разворот на повышенном объёме.

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

Когда происходит резкий обвал с повышенным объёмом (часто вызванный ликвидациями), рынок временно "перепродаётся".
PBR — стратегия, отслеживающая такие движения и входящая в лонг на развороте.

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

1. Свеча на 1м/5м таймфрейме:

Упала на >3% от предыдущей закрытия.

Имеет объём X выше среднего (Volume Spike).

Закрылась с фитилём снизу (низкий Close относительно Low).

2. Следующая свеча начинает рост — сигнал на вход.

📤 Выход по:💸

Take Profit = +1.5–2.5%

Stop Loss = ниже фитиля панической свечи.

import ccxt
import time
from datetime import datetime

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '1m'
limit = 50

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

def detect_panic_reversal(data):
    avg_volume = sum([c['volume'] for c in data[-10:-1]]) / 9
    last = data[-2]
    current = data[-1]

    drop = (last['close'] - last['low']) / last['close']
    body = abs(last['close'] - last['open'])
    wick = (last['high'] - last['low']) - body

    if (last['close'] < last['open'] * 0.97 and
        last['volume'] > avg_volume * 2 and
        wick > body and
        current['close'] > last['close']):
        return True
    return False

print("📡 Panic Bid Reversal Strategy запущена...")

while True:
    try:
        candles = fetch_ohlcv()
        if detect_panic_reversal(candles):
            ts = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')
            print(f"✅ [{ts}] Сигнал на лонг: Обнаружена паническая свеча с разворотом!")
        time.sleep(60)
    except Exception as e:
        print(f"⚠️ Ошибка: {e}")
        time.sleep(10)

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

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

📌“Liquidity Sniper” — Торговля на пробой зон ликвидности по паттернам всплесков объёма🧑‍💻

📌 Идея:🧨

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

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

- Выявить свечи с аномально высоким объёмом.

- Определить “зону ликвидности” — ценовой диапазон тела такой свечи.

- Ожидать возврат в зону и входить по направлению предыдущего объёма.

- Стоп за противоположный край зоны, тейк — по ближайшему импульсу.

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

1. Находится свеча с объёмом выше среднего на N свечей.

2. Цена возвращается в тело этой свечи спустя X баров.

3. Объём остаётся высоким — подтверждение интереса.

4. Открывается сделка по направлению исходного движения.

import ccxt
import pandas as pd
import pandas_ta as ta
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '15m'
bars = 150
volume_multiplier = 2

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

def detect_liquidity_zones(df):
    df['avg_vol'] = df['volume'].rolling(20).mean()
    df['is_liquidity_candle'] = df['volume'] > df['avg_vol'] * volume_multiplier

    liquidity_zones = []
    for i in range(1, len(df)):
        if df['is_liquidity_candle'].iloc[i]:
            zone = {
                "index": i,
                "price_low": df['open'].iloc[i],
                "price_high": df['close'].iloc[i],
                "direction": "buy" if df['close'].iloc[i] > df['open'].iloc[i] else "sell"
            }
            liquidity_zones.append(zone)
    return liquidity_zones

def check_retest(df, zones):
    current_price = df['close'].iloc[-1]
    for zone in zones[-5:]:  # последние 5 зон
        if zone["price_low"] <= current_price <= zone["price_high"]:
            print(f"\n📡 Сигнал: Retest зоны ликвидности!")
            print(f"👉 Направление: {zone['direction'].upper()}")
            print(f"📍 Диапазон зоны: {zone['price_low']} - {zone['price_high']}")
            break

# Основной цикл
while True:
    df = fetch_ohlcv()
    zones = detect_liquidity_zones(df)
    check_retest(df, zones)
    time.sleep(60)

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

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

📌"Order Book Dynamics Tracker" — Инструмент отслеживания изменений глубины стакана в реальном времени🧑‍💻

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

Позволяет отслеживать изменения лимитных ордеров (bid/ask) по уровням и строить метрику, показывающую поведение маркетмейкеров и крупных игроков — до появления свечи и сигнала от индикаторов.

🧪 Идея:🧨

Изменения в order book (например, внезапное исчезновение большого лимитника или появление «стены») часто предшествуют импульсным движениям. Это можно использовать для:

- определения фейковых движений,

- входа до свечного сигнала,

- выявления «стоп охоты» и проскальзываний.

import asyncio
import json
import websockets
from collections import deque

symbol = 'btcusdt'
depth_levels = 10
window = deque(maxlen=5)

async def track_orderbook():
    uri = f"wss://fstream.binance.com/ws/{symbol}@depth@100ms"
    
    async with websockets.connect(uri) as websocket:
        print("📡 Подключение к Binance WebSocket...")
        
        while True:
            msg = await websocket.recv()
            data = json.loads(msg)

            bids = data['bids'][:depth_levels]
            asks = data['asks'][:depth_levels]

            bid_vol = sum([float(b[1]) for b in bids])
            ask_vol = sum([float(a[1]) for a in asks])

            imbalance = (bid_vol - ask_vol) / (bid_vol + ask_vol + 1e-6)
            window.append(imbalance)

            avg_imbalance = sum(window) / len(window)

            signal = "🔼 BUY PRESSURE" if avg_imbalance > 0.25 else "🔽 SELL PRESSURE" if avg_imbalance < -0.25 else "⚖️ Neutral"

            print(f"Imbalance: {avg_imbalance:.2f} → {signal}")

asyncio.run(track_orderbook())

#инструмент

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

📌"Объем + Импульсная фильтрация по кластерам"🧑‍💻

📈 Тип стратегии: краткосрочная (intra-day или swing)

📉 Подходит для: спот и фьючерсы

⚙️ Идея: открывать сделки при резком всплеске объема в пределах узкого ценового диапазона, что указывает на скрытую аккумуляцию/распродажу — с подтверждением импульса и выхода из диапазона.

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

1. Фильтр объема (Volume Spike)

Считается средний объем за последние N свечей.

Если текущий объем > x * среднего (например, 3x), переходим к следующему шагу.

2. Фильтр диапазона

Проверяем, что последние N свечей были в узком ценовом диапазоне (например, max/min < 1.5%).

3. Фильтр импульса (Breakout)

Проверяется, что текущая свеча пробивает максимум или минимум диапазона.

4. Сигнал на вход в сторону пробоя

Take-Profit можно поставить на ширину диапазона * 2,

Stop-Loss — внутри диапазона.

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

- Объем указывает на институциональные действия.

- Узкий диапазон + резкий выход — признак движения, а не шума.

- Подходит для алготрейдинга, особенно в консолидациях.

import ccxt
import numpy as np
import time

symbol = 'BTC/USDT'
timeframe = '5m'
volume_multiplier = 3
range_percent_threshold = 1.5  # %

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

def fetch_data():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=30)
    return np.array(ohlcv)

def analyze(data):
    highs = data[:, 2]
    lows = data[:, 3]
    closes = data[:, 4]
    volumes = data[:, 5]

    last_candle = data[-1]
    recent_volumes = volumes[-21:-1]
    avg_volume = np.mean(recent_volumes)
    
    if last_candle[5] < avg_volume * volume_multiplier:
        return None  # Нет всплеска объема

    recent_high = np.max(highs[-21:-1])
    recent_low = np.min(lows[-21:-1])
    price_range_pct = (recent_high - recent_low) / recent_low * 100

    if price_range_pct > range_percent_threshold:
        return None  # Диапазон был слишком широкий

    close = last_candle[4]
    if close > recent_high:
        return 'BUY'
    elif close < recent_low:
        return 'SELL'
    return None

while True:
    try:
        candles = fetch_data()
        signal = analyze(candles)
        if signal:
            print(f"[{time.strftime('%H:%M:%S')}] 📢 Сигнал: {signal} на {symbol}")
        else:
            print(f"[{time.strftime('%H:%M:%S')}] ⏳ Нет сигнала")
    except Exception as e:
        print("Ошибка:", e)
    
    time.sleep(60)

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

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