📌"Liquidity Imbalance Pulse"🧑‍💻

🧠 Идея:🧨

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

- Использует данные ордербука (bid/ask depth)

- Сопоставляет дельту объема (агрессивные покупки/продажи)

- Отслеживает поведение цены в микроструктуре (1s–5s свечи)

- Алгоритм определяет: это ложное движение или начало реального импульса

📊 Сигналы:💸

Лонг:

- Обнаружен сильный дисбаланс: bid size резко увеличился, ask съедается маркет-ордерами

- Spread сузился

- Цена закрепляется выше локального уровня ликвидности

Шорт:

Обратная ситуация: ask усилился, маркет-продажи интенсивны, цена пробивает ликвидность вниз

import ccxt
import time
import datetime

# Настройки
symbol = 'BTC/USDT'
depth_limit = 20  # Глубина анализа стакана
imbalance_threshold = 0.5  # Чувствительность к дисбалансу
poll_interval = 3  # Интервал между проверками (сек)

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

def get_orderbook_metrics():
    try:
        orderbook = exchange.fetch_order_book(symbol, limit=depth_limit)

        bids = orderbook['bids']
        asks = orderbook['asks']

        bid_volume = sum([price * amount for price, amount in bids])
        ask_volume = sum([price * amount for price, amount in asks])

        best_bid = bids[0][0] if bids else 0
        best_ask = asks[0][0] if asks else 0
        spread = best_ask - best_bid
        mid_price = (best_ask + best_bid) / 2 if best_bid and best_ask else 0

        imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume + 1e-9)

        return {
            'bid_volume': bid_volume,
            'ask_volume': ask_volume,
            'imbalance': imbalance,
            'spread': spread,
            'mid_price': mid_price,
            'timestamp': datetime.datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')
        }

    except Exception as e:
        print("Ошибка при получении стакана:", str(e))
        return None

def print_signal(metrics):
    imbalance = metrics['imbalance']
    signal = "⚪ Нейтрально"

    if imbalance >= imbalance_threshold:
        signal = "🟢 ЛОНГ сигнал"
    elif imbalance <= -imbalance_threshold:
        signal = "🔴 ШОРТ сигнал"

    print(f"[{metrics['timestamp']}] {signal} | Imbalance: {imbalance:.2f} | "
          f"BidVol: {metrics['bid_volume']:.0f} | AskVol: {metrics['ask_volume']:.0f} | "
          f"Mid: {metrics['mid_price']:.2f} | Spread: {metrics['spread']:.2f}")

def run():
    print("🚀 Старт стратегии Liquidity Imbalance Pulse\n")
    while True:
        metrics = get_orderbook_metrics()
        if metrics:
            print_signal(metrics)
        time.sleep(poll_interval)

if __name__ == '__main__':
    run()

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

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

📌Стратегия возврата к среднему с подтверждением через ATR и Price Action🧑‍💻

Стратегия возврата к среднему в трейдинге предполагает, что цены финансовых активов имеют тенденцию возвращаться к своим историческим средним значениям после отклонений.

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

Цена временно отклоняется от скользящей средней (SMA) — если отклонение достаточно велико (например, 2×ATR), и свеча подтверждает разворот, то стратегия выдаёт сигнал.

🔧 Настройки:

- SMA: 20-периодная

- ATR: 14-периодная

- Подтверждение: свеча закрылась в сторону возврата

- Сигналы: только когда цена находится вне 2×ATR от SMA

import ccxt
import pandas as pd
import time
import datetime

# Настройки
symbol = 'BTC/USDT'
timeframe = '1h'
sma_period = 20
atr_period = 14
atr_multiplier = 2

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

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

def compute_indicators(df):
    df['sma'] = df['close'].rolling(sma_period).mean()
    tr = pd.concat([
        df['high'] - df['low'],
        (df['high'] - df['close'].shift()).abs(),
        (df['low'] - df['close'].shift()).abs()
    ], axis=1).max(axis=1)
    df['atr'] = tr.rolling(atr_period).mean()
    return df

def check_signal(df):
    latest = df.iloc[-1]
    prev = df.iloc[-2]
    upper_band = latest['sma'] + atr_multiplier * latest['atr']
    lower_band = latest['sma'] - atr_multiplier * latest['atr']

    # Условия для сигнала на шорт
    if latest['close'] > upper_band and latest['close'] < prev['close']:
        print(f"🔻 SHORT сигнал [{symbol}] @ {latest['close']} — возврат от верхней границы")

    # Условия для сигнала на лонг
    elif latest['close'] < lower_band and latest['close'] > prev['close']:
        print(f"🔼 LONG сигнал [{symbol}] @ {latest['close']} — возврат от нижней границы")

def main():
    while True:
        try:
            df = fetch_data()
            df = compute_indicators(df)
            check_signal(df)
            time.sleep(60 * 60)  # Ждём 1 час до следующего бара
        except Exception as e:
            print("Ошибка:", e)
            time.sleep(60)

if __name__ == '__main__':
    main()

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

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

📌“Order Book Pressure Index” (OBPI)🧑‍💻

📌 Идея:🚀

Анализ микроструктуры рынка на основе дисбаланса заявок в стакане. Инструмент вычисляет «давление» покупателей и продавцов по данным ордербука, чтобы выдать сигнал: скрытая аккумуляция / дистрибуция.

🧠 Логика:🧨

1. Подгружаются bid/ask уровни (обычно топ 20).

2. Рассчитываются:

Bid Pressure = сумма объёмов всех заявок на покупку.

Ask Pressure = сумма объёмов всех заявок на продажу.

3. Order Book Pressure Index (OBPI) = (Bid Pressure - Ask Pressure) / (Bid Pressure + Ask Pressure)

Значения ближе к +1 → спрос > предложение → возможный рост.

Значения ближе к -1 → предложение > спрос → возможное падение.

4. Если OBPI стабильно высокий/низкий + цена стоит → сигнал накопления/распродажи.

import ccxt
import time

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

def get_order_book(symbol, depth=20):
    ob = exchange.fetch_order_book(symbol, limit=depth)
    bids = ob['bids']
    asks = ob['asks']
    bid_vol = sum([b[1] for b in bids])
    ask_vol = sum([a[1] for a in asks])
    return bid_vol, ask_vol

def compute_obpi(bid_vol, ask_vol):
    if bid_vol + ask_vol == 0:
        return 0
    return (bid_vol - ask_vol) / (bid_vol + ask_vol)

while True:
    bid_vol, ask_vol = get_order_book(symbol)
    obpi = compute_obpi(bid_vol, ask_vol)
    print(f"[{symbol}] OBPI: {obpi:.4f}")

    if obpi > 0.6:
        print("🔼 Сильное давление покупателей — возможный рост")
    elif obpi < -0.6:
        print("🔽 Сильное давление продавцов — возможное падение")
    else:
        print("⏸️ Баланс, рынок в равновесии")

    time.sleep(10)

#инструмент

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

📌"Live Liquidity Sweep Tracker" (Трекер снятия ликвидности в реальном времени)🧑‍💻

🔍 Что делает:🧨

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

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

Часто перед разворотом маркетмейкеры "снимают стопы", чтобы собрать ликвидность.

Этот инструмент помогает отследить такие моменты — идеально для контртрендовых стратегий или входа на откате.

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

1. Загружает свечи.

2. Находит локальные экстремумы (например, последние 3 high/low).

3. Следит за пробоем этих уровней на аномально высоком объёме.

4. Если пробой сопровождается скачком объёма — сигнал в консоль.

import ccxt
import pandas as pd
import time

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

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

def detect_liquidity_sweep(df):
    highs = df['high'][-4:-1]
    lows = df['low'][-4:-1]
    volume_mean = df['volume'][-20:-1].mean()
    
    last = df.iloc[-1]
    high_break = last['high'] > highs.max() and last['volume'] > volume_mean * 1.5
    low_break = last['low'] < lows.min() and last['volume'] > volume_mean * 1.5

    if high_break:
        return f"🔴 Liquidity Sweep UP detected on {symbol} — possible short setup"
    elif low_break:
        return f"🟢 Liquidity Sweep DOWN detected on {symbol} — possible long setup"
    return None

while True:
    df = fetch_candles()
    signal = detect_liquidity_sweep(df)
    if signal:
        print(signal)
    else:
        print("⚪ No liquidity sweep detected.")
    time.sleep(10)

#инструмент

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

📌"Imbalance Breakout" (Пробой Дисбаланса)🧑‍💻

📌 Идея:🚀

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

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

Лонг:

- Ищется участок из ≥3 бычьих свечей без теней снизу (или почти без).

- Цена консолидируется у верхней границы этой зоны (боковик).

- При пробое верхней границы на увеличенном объёме — вход в лонг.


Шорт:

- Аналогично: участок из ≥3 медвежьих свечей без откатов.

- Консолидация.

- Пробой вниз → вход в шорт.

✅ Подтверждение:

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

- Волатильность расширяется.

- Желательно наличие дивергенции по RSI для раннего сигнала.

import ccxt
import time
import pandas as pd

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

def fetch_candles(symbol, tf='5m', lim=100):
    data = exchange.fetch_ohlcv(symbol, tf, limit=lim)
    df = pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    return df

def detect_imbalance_breakout(df):
    last = df.iloc[-4:-1]  # Последние 3 свечи до текущей
    current = df.iloc[-1]
    
    bullish_block = all(last['close'] > last['open']) and all((last['low'] - last['open']).abs() < 0.1 * (last['high'] - last['low']))
    bearish_block = all(last['close'] < last['open']) and all((last['high'] - last['open']).abs() < 0.1 * (last['high'] - last['low']))

    if bullish_block and current['close'] > last['high'].max():
        return "📈 Long breakout signal"
    elif bearish_block and current['close'] < last['low'].min():
        return "📉 Short breakout signal"
    else:
        return None

while True:
    candles = fetch_candles(symbol, timeframe, limit)
    signal = detect_imbalance_breakout(candles)
    if signal:
        print(f"[{symbol}] {signal}")
    else:
        print(f"[{symbol}] ⚪ No signal")
    time.sleep(15)

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

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

📌Triple Confirmation Strategy🧑‍💻

Индикаторы:🚀

1. EMA 200 — определение тренда.

2. RSI (14) — фильтрация перекупленности/перепроданности.

3. MACD — подтверждение импульса.

🧠 Логика стратегии:💸

✅ LONG:

- Цена выше EMA 200 (восходящий тренд).

- RSI пересекает снизу вверх уровень 50.

- MACD-гистограмма выше 0 и увеличивается.

❌ SHORT:

- Цена ниже EMA 200 (нисходящий тренд).

- RSI пересекает сверху вниз уровень 50.

- MACD-гистограмма ниже 0 и уменьшается.

import ccxt
import pandas as pd
import time
import ta

symbol = 'BTC/USDT'
timeframe = '5m'
limit = 200

exchange = ccxt.binance()

def fetch_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 apply_indicators(df):
    df['ema200'] = ta.trend.ema_indicator(df['close'], window=200).ema_indicator()
    df['rsi'] = ta.momentum.rsi(df['close'], window=14)
    macd = ta.trend.macd(df['close'])
    df['macd_hist'] = macd.macd_diff()
    return df

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

    # LONG setup
    if (
        last['close'] > last['ema200'] and
        prev['rsi'] < 50 and last['rsi'] > 50 and
        last['macd_hist'] > 0 > prev['macd_hist']
    ):
        return '🟢 LONG signal detected'

    # SHORT setup
    if (
        last['close'] < last['ema200'] and
        prev['rsi'] > 50 and last['rsi'] < 50 and
        last['macd_hist'] < 0 < prev['macd_hist']
    ):
        return '🔴 SHORT signal detected'

    return None

while True:
    df = fetch_ohlcv()
    df = apply_indicators(df)
    signal = check_signals(df)
    if signal:
        print(f"{signal} at {df.iloc[-1]['timestamp']}")
    else:
        print(f"No signal — {df.iloc[-1]['timestamp']}")
    time.sleep(30)

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

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

📌"Market Regime Detector" — Определение рыночного режима в реальном времени🧑‍💻

🧠 Что делает?🧨

Автоматически определяет фазу рынка:

📈 Тренд вверх

📉 Тренд вниз

🔁 Флэт/Боковик


📍 Для чего нужен?

📊 Использовать разные стратегии под разные рыночные режимы.

⛔ Фильтровать сигналы в плохих условиях (например, во флэте).

🔁 Менять параметры индикаторов/рисков при смене режима.

🧠 Логика определения режима:🛠️

Используется ADX (силу тренда) и наклон EMA (направление).

- Флэт: ADX < 20

Тренд вверх: ADX > 20 и EMA наклон вверх

- Тренд вниз: ADX > 20 и EMA наклон вниз

import ccxt
import pandas as pd
import ta
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '1h'
limit = 100

def get_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 detect_market_regime(df):
    df['ema'] = ta.trend.ema_indicator(df['close'], window=50).ema_indicator()
    adx = ta.trend.adx(df['high'], df['low'], df['close'], window=14)
    df['adx'] = adx.adx()

    last = df.iloc[-1]
    prev = df.iloc[-2]

    ema_slope = last['ema'] - prev['ema']

    if last['adx'] < 20:
        return '⚪ Флэт (Low ADX)'
    elif last['adx'] >= 20 and ema_slope > 0:
        return '🟢 Восходящий тренд'
    elif last['adx'] >= 20 and ema_slope < 0:
        return '🔴 Нисходящий тренд'
    else:
        return '❓ Неопределённый режим'

while True:
    df = get_ohlcv()
    regime = detect_market_regime(df)
    print(f"[{df.iloc[-1]['timestamp']}] Рыночный режим: {regime}")
    time.sleep(60)

#инструмент

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

📌"Adaptive Liquidity Trap"🧑‍💻

🧠 Идея:🚀

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

2. Если цена резко уходит за этот диапазон — вероятен ложный пробой.

3. Возврат в диапазон после прокола — сигнал на вход в противоположную сторону.

📊 Условия:🧨

- Вычисляется средний true spread: (high - low) / close

- Вычисляется волатильность и объём на 20 последних свечах

Если:

- Цена проколола границу диапазона ликвидности

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

- И следующая свеча вернулась обратно

→ сигнал на вход внутрь диапазона

🧩 Логика сигналов:💸

LONG, если был прокол вниз и возврат

SHORT, если был прокол вверх и возврат

import ccxt
import pandas as pd
import time

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

exchange = ccxt.binance()

def fetch_data():
    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 generate_signal(df):
    df['spread'] = (df['high'] - df['low']) / df['close']
    df['avg_spread'] = df['spread'].rolling(window=20).mean()
    df['avg_volume'] = df['volume'].rolling(window=20).mean()

    last = df.iloc[-1]
    prev = df.iloc[-2]

    upper_bound = df['high'][-20:].mean() + df['avg_spread'].iloc[-1]
    lower_bound = df['low'][-20:].mean() - df['avg_spread'].iloc[-1]

    signal = None

    if prev['low'] < lower_bound and last['close'] > lower_bound and last['volume'] > df['avg_volume'].iloc[-1]:
        signal = "🟢 LONG сигнал: возврат в зону ликвидности"
    elif prev['high'] > upper_bound and last['close'] < upper_bound and last['volume'] > df['avg_volume'].iloc[-1]:
        signal = "🔴 SHORT сигнал: возврат в зону ликвидности"

    return signal

while True:
    try:
        df = fetch_data()
        signal = generate_signal(df)
        if signal:
            print(f"[{pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}] {signal}")
        else:
            print(f"[{pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}] 🔍 Нет сигнала")
        time.sleep(60)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(60)

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

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

📌"Сентимент Индикатор из Ордербука (Order Book Sentiment Tracker)"🧑‍💻

📌 Идея:🚀

Создаёт оценку настроений (bullish/bearish) на основе глубины стакана (order book). В отличие от классических индикаторов, он использует реальное распределение заявок на покупку и продажу в режиме реального времени.

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

1. С помощью ccxt подключается к Binance.

2. Загружает глубину стакана (order book).

3. Сравнивает суммарный объём заявок выше текущей цены (ask) и ниже текущей цены (bid).

4. Вычисляет имбаланс и определяет:

Bullish, если ниже цены заявок больше.

Bearish, если выше цены заявок больше.

Neutral, если равновесие.

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
depth_limit = 50  # Кол-во уровней стакана

def get_orderbook_sentiment(symbol):
    orderbook = exchange.fetch_order_book(symbol, limit=depth_limit)
    bids = orderbook['bids']
    asks = orderbook['asks']

    bid_volume = sum([price * amount for price, amount in bids])
    ask_volume = sum([price * amount for price, amount in asks])

    imbalance = bid_volume - ask_volume
    sentiment = "Neutral"
    if imbalance > ask_volume * 0.1:
        sentiment = "Bullish"
    elif imbalance < -bid_volume * 0.1:
        sentiment = "Bearish"

    print(f"Sentiment: {sentiment}")
    print(f"Bid Volume: {bid_volume:.2f}, Ask Volume: {ask_volume:.2f}, Imbalance: {imbalance:.2f}")
    print("-" * 50)

while True:
    try:
        get_orderbook_sentiment(symbol)
        time.sleep(5)
    except Exception as e:
        print("Error:", e)
        time.sleep(5)

#инструмент

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

📌"Импульсная асимметрия" (Asymmetric Momentum Breakout)🧑‍💻

💡 Идея🧨

Большинство стратегий импульсов смотрят только на пробой в одном направлении

Здесь же мы ищем асимметричные всплески волатильности, когда:

- В одну сторону движение сильное,

- А в обратную — слабое и быстро затухает.

Это часто говорит о доминации покупателей или продавцов, и сигнал более надёжен, чем простой пробой.

📊 Логика🚀

1. Берём ATR (средний истинный диапазон) для оценки волатильности.

2. Отслеживаем импульсы:

Если свеча пробивает максимум последних N свечей и закрывается в верхней 20% своего диапазона — фиксируем бычий импульс.

Если пробивает минимум и закрывается в нижней 20% диапазона — медвежий импульс.

3. Проверяем асимметрию:

После импульса в обратную сторону не должно быть свечи с амплитудой > 0.5 от импульсной в течение M свечей.

4. Если условия соблюдены — даём сигнал на вход.

import ccxt
import pandas as pd
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '5m'
lookback = 20
cooldown = 5

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

def atr(df, period=14):
    df['h-l'] = df['high'] - df['low']
    df['h-c'] = abs(df['high'] - df['close'].shift())
    df['l-c'] = abs(df['low'] - df['close'].shift())
    tr = df[['h-l', 'h-c', 'l-c']].max(axis=1)
    return tr.rolling(period).mean()

def detect_signals():
    df = fetch_ohlcv()
    df['ATR'] = atr(df)
    last = df.iloc[-1]
    prev_high = df['high'][-lookback-1:-1].max()
    prev_low = df['low'][-lookback-1:-1].min()

    # Проверка бычьего импульса
    if last['high'] > prev_high and (last['close'] - last['low']) / (last['high'] - last['low']) > 0.8:
        # Проверяем асимметрию
        after_impulse = df.iloc[-cooldown:]
        if all((after_impulse['high'] - after_impulse['low']) < 0.5 * (last['high'] - last['low'])):
            print(f"🐂 Bullish breakout signal at {last['time']} price: {last['close']}")

    # Проверка медвежьего импульса
    if last['low'] < prev_low and (last['high'] - last['close']) / (last['high'] - last['low']) > 0.8:
        after_impulse = df.iloc[-cooldown:]
        if all((after_impulse['high'] - after_impulse['low']) < 0.5 * (last['high'] - last['low'])):
            print(f"🐻 Bearish breakout signal at {last['time']} price: {last['close']}")

while True:
    try:
        detect_signals()
        time.sleep(5)
    except Exception as e:
        print("Error:", e)
        time.sleep(5)

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

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

📌"Индикатор ликвидности и скрытых уровней" (Hidden Liquidity Levels Scanner)🧑‍💻

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

Большинство трейдеров смотрят на цену и индикаторы, но алготрейдеру критически важно знать, где в ордербуке стоят крупные лимитные заявки, потому что они часто:

- Разворачивают цену

- Замедляют импульсы

- Выступают как "магниты" для цены

📊 Наш инструмент будет:🛠️

1. Подключаться к Binance

2. Сканировать стакан ордеров (order book).

3. Выделять аномально крупные кластеры заявок.

4. Отображать их в консоли с пометкой:

🟢 крупная поддержка (buy wall)

🔴 крупное сопротивление (sell wall)

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
threshold = 5  # множитель среднего объема в стакане для выделения

def get_order_book():
    order_book = exchange.fetch_order_book(symbol, limit=50)
    bids = order_book['bids']  # [price, amount]
    asks = order_book['asks']
    
    bid_volumes = [b[1] for b in bids]
    ask_volumes = [a[1] for a in asks]
    
    avg_bid = sum(bid_volumes) / len(bid_volumes)
    avg_ask = sum(ask_volumes) / len(ask_volumes)
    
    large_bids = [(p, v) for p, v in bids if v > avg_bid * threshold]
    large_asks = [(p, v) for p, v in asks if v > avg_ask * threshold]
    
    print("\n=== Hidden Liquidity Levels ===")
    if large_bids:
        for price, vol in large_bids:
            print(f"🟢 BUY WALL: {price:.2f} | {vol:.4f} BTC")
    if large_asks:
        for price, vol in large_asks:
            print(f"🔴 SELL WALL: {price:.2f} | {vol:.4f} BTC")
    if not large_bids and not large_asks:
        print("Нет значимых уровней")

while True:
    try:
        get_order_book()
        time.sleep(3)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(5)

#инструмент

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

📌"Импульс на истощении ликвидности" (Liquidity Exhaustion Breakout).🧑‍💻

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

Вместо того чтобы просто ловить пробои или откаты, мы отслеживаем момент, когда:

1. Ликвидность (глубина ордербука) резко уменьшается в одну сторону.

2. Цена при этом начинает двигаться в направлении ослабевшей стороны.

3. Подтверждаем сигнал с помощью ускорения тикового потока сделок.

📊 Логика работы:🛠️

- Если из стакана исчезает крупная поддержка (buy wall) → цена пробьёт вниз.

- Если исчезает крупное сопротивление (sell wall) → цена пробьёт вверх.

- Входим в сторону исчезнувшей ликвидности, пока маркет-мейкеры не успели восстановить стенку.

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
threshold_ratio = 0.4  # падение объема >60% считается исчезновением ликвидности
limit = 20  # глубина стакана

prev_bids = []
prev_asks = []

def detect_liquidity_exhaustion():
    global prev_bids, prev_asks
    ob = exchange.fetch_order_book(symbol, limit=limit)
    
    bids = ob['bids']
    asks = ob['asks']
    
    top_bid_volume = sum([v for _, v in bids[:3]])
    top_ask_volume = sum([v for _, v in asks[:3]])
    
    signal = None
    
    if prev_bids:
        if top_bid_volume < sum(prev_bids) * threshold_ratio:
            signal = "🔴 Возможен пробой вниз — исчезла поддержка!"
    
    if prev_asks:
        if top_ask_volume < sum(prev_asks) * threshold_ratio:
            signal = "🟢 Возможен пробой вверх — исчезло сопротивление!"
    
    prev_bids = [v for _, v in bids[:3]]
    prev_asks = [v for _, v in asks[:3]]
    
    if signal:
        print(f"[{symbol}] {signal}")
    else:
        print(f"[{symbol}] Ситуация стабильна.")

while True:
    try:
        detect_liquidity_exhaustion()
        time.sleep(2)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(5)

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

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

📌"Market Heatmap Engine" — движок для построения в реальном времени тепловой карты крипторынка с учётом:

- силы тренда,

- притока/оттока ликвидности,

- скорости изменения цены.

💡 Главная идея — видеть рынок как "карту температур": красные сектора — перегретые активы с высокой вероятностью коррекции, зелёные — активы в фазе накопления, готовые к импульсу.

🔍 Логика работы:🛠️

1. Берём список монет с Binance (например, топ-50 по объёму).

2. Для каждой монеты:

- рассчитываем скорость изменения цены (momentum);

- измеряем изменение глубины стакана за последние секунды;

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

3. На основе этих метрик формируем индекс перегрева от -100 до +100.

4. Отдаём данные в консоль или в браузер как тепловую карту.

import ccxt
import time
import numpy as np
from prettytable import PrettyTable

exchange = ccxt.binance()

symbols = [m['symbol'] for m in exchange.fetch_markets() if '/USDT' in m['symbol']][:15]
prev_prices = {}
prev_bids = {}
prev_asks = {}

def get_heatmap():
    table = PrettyTable(["Symbol", "Momentum", "Liquidity Δ", "Heat Index"])
    for symbol in symbols:
        try:
            # Цена
            ticker = exchange.fetch_ticker(symbol)
            price = ticker['last']
            momentum = 0
            if symbol in prev_prices:
                momentum = (price - prev_prices[symbol]) / prev_prices[symbol] * 100
            
            # Ликвидность
            orderbook = exchange.fetch_order_book(symbol, limit=10)
            bids_vol = sum([v for _, v in orderbook['bids']])
            asks_vol = sum([v for _, v in orderbook['asks']])
            
            liquidity_change = 0
            if symbol in prev_bids and symbol in prev_asks:
                prev_liq = prev_bids[symbol] + prev_asks[symbol]
                curr_liq = bids_vol + asks_vol
                liquidity_change = (curr_liq - prev_liq) / prev_liq * 100 if prev_liq != 0 else 0
            
            # Индекс перегрева
            heat_index = momentum * 0.7 + liquidity_change * 0.3
            
            table.add_row([
                symbol,
                f"{momentum:+.2f}%",
                f"{liquidity_change:+.2f}%",
                f"{heat_index:+.2f}"
            ])
            
            # Сохраняем предыдущие значения
            prev_prices[symbol] = price
            prev_bids[symbol] = bids_vol
            prev_asks[symbol] = asks_vol
        
        except Exception as e:
            pass
    
    print("\033c", end="")  # Очистка консоли
    print(table)

while True:
    get_heatmap()
    time.sleep(3)

#инструмент

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

📌"Радар крупных скрытых сделок (Stealth Trade Radar)".🧑‍💻

💡 Суть:🛠️

Большие игроки часто маскируют свои входы и выходы:

- разбивают ордера на мелкие,

- или используют быстрое выставление и снятие заявок.

Этот инструмент будет отслеживать поток сделок (trades) и искать аномальные паттерны, указывающие на скрытое накопление или сброс позиции.

📊 Логика:🧨

1. Получаем поток сделок с биржи через ccxt (fetch_trades).

2. Группируем их по направлению (покупка/продажа).

3. Считаем суммарный объём за последние N секунд.

4. Если за короткое время прошло аномально много мелких сделок в одну сторону — фиксируем сигнал.

import ccxt
import time
from collections import deque

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

WINDOW_SEC = 5       # окно анализа в секундах
VOLUME_THRESHOLD = 50  # суммарный объем для сигнала
MAX_TRADE_SIZE = 0.1  # "мелкая" сделка

buy_trades = deque()
sell_trades = deque()

def check_stealth_trades():
    global buy_trades, sell_trades

    trades = exchange.fetch_trades(symbol)
    now = time.time()

    for t in trades:
        ts = t['timestamp'] / 1000
        size = t['amount']
        side = t['side']

        # Учитываем только "мелкие" сделки
        if size <= MAX_TRADE_SIZE:
            if side == 'buy':
                buy_trades.append((ts, size))
            else:
                sell_trades.append((ts, size))

    # Чистим старые сделки
    while buy_trades and now - buy_trades[0][0] > WINDOW_SEC:
        buy_trades.popleft()
    while sell_trades and now - sell_trades[0][0] > WINDOW_SEC:
        sell_trades.popleft()

    buy_volume = sum(s for _, s in buy_trades)
    sell_volume = sum(s for _, s in sell_trades)

    if buy_volume > VOLUME_THRESHOLD:
        print(f"🟢 Обнаружено скрытое накопление! Объем {buy_volume:.4f} {symbol}")
    elif sell_volume > VOLUME_THRESHOLD:
        print(f"🔴 Обнаружен скрытый сброс! Объем {sell_volume:.4f} {symbol}")
    else:
        print(f"[{symbol}] Активных скрытых сделок нет.")

while True:
    try:
        check_stealth_trades()
        time.sleep(2)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(5)

#инструмент

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

📌"Сдвиг ликвидности + фейковый пробой" (Liquidity Shift & Trap Breakout).🧑‍💻

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

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

📊 Логика работы:🛠️

1. Отслеживаем момент, когда цена пробивает уровень и в стакане резко появляется противоположная стена ликвидности (новые крупные ордера).

2. Проверяем, что после пробоя тик-флоу сделок идёт в противоположную сторону.

3. Входим в позицию против пробоя (ловим возврат цены в диапазон).

4. Выходим, когда цена возвращается в зону до пробоя.

import ccxt
import time

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

prev_price = None
prev_bids = []
prev_asks = []

def detect_fake_breakout():
    global prev_price, prev_bids, prev_asks
    ob = exchange.fetch_order_book(symbol, limit=limit)
    ticker = exchange.fetch_ticker(symbol)
    price = ticker['last']

    bids = ob['bids']
    asks = ob['asks']

    top_bid_volume = sum([v for _, v in bids[:3]])
    top_ask_volume = sum([v for _, v in asks[:3]])

    signal = None

    if prev_price:
        # Пробой вверх + внезапное появление крупной стены продаж
        if price > prev_price * 1.002 and top_ask_volume > sum(prev_asks) * 1.5:
            signal = "🔴 Фейковый пробой вверх! Возможен возврат вниз."
        # Пробой вниз + внезапная крупная поддержка
        elif price < prev_price * 0.998 and top_bid_volume > sum(prev_bids) * 1.5:
            signal = "🟢 Фейковый пробой вниз! Возможен возврат вверх."

    prev_price = price
    prev_bids = [v for _, v in bids[:3]]
    prev_asks = [v for _, v in asks[:3]]

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

while True:
    try:
        detect_fake_breakout()
        time.sleep(2)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(5)

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

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

📌"Дивергенция импульса против ликвидности" (Impulse vs Liquidity Divergence).🧑‍💻

💡 Суть:🛠️

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

Часто перед разворотом:

- цена продолжает расти,

- но ликвидность на стороне покупателей падает (и наоборот).

📊 Логика:🚀

1. Каждую секунду получаем:

- последние сделки,

- стакан ордеров (order book).

2. Считаем импульс (цена vs предыдущая цена).

3. Считаем изменение ликвидности на стороне покупателей и продавцов.

4. Если цена растёт, а ликвидность покупателей падает → сигнал на шорт.

Если цена падает, а ликвидность продавцов падает → сигнал на лонг.

import ccxt
import time

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

prev_price = None
prev_bids_liq = None
prev_asks_liq = None

def get_liquidity(orderbook_side):
    return sum(amount for price, amount in orderbook_side)

while True:
    try:
        ticker = exchange.fetch_ticker(symbol)
        orderbook = exchange.fetch_order_book(symbol, depth_limit)

        last_price = ticker['last']
        bids_liq = get_liquidity(orderbook['bids'])
        asks_liq = get_liquidity(orderbook['asks'])

        if prev_price is not None:
            price_change = last_price - prev_price
            bids_change = bids_liq - prev_bids_liq if prev_bids_liq is not None else 0
            asks_change = asks_liq - prev_asks_liq if prev_asks_liq is not None else 0

            if price_change > 0 and bids_change < 0:
                print(f"🔴 {symbol}: Цена растёт, но покупатели уходят → возможный разворот вниз.")
            elif price_change < 0 and asks_change < 0:
                print(f"🟢 {symbol}: Цена падает, но продавцы уходят → возможный разворот вверх.")
            else:
                print(f"{symbol}: Нет дивергенций. Цена: {last_price}")

        prev_price = last_price
        prev_bids_liq = bids_liq
        prev_asks_liq = asks_liq

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

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

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

📌"Детектор скрытой агрессии в ордербуке".🧑‍💻

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

Большие игроки часто используют iceberg-ордера — они разбивают крупную заявку на маленькие куски, чтобы скрыть реальный объём.
Но по паттернам обновления стакана и частоте сделок можно вычислить, что там стоит крупный скрытый участник.

📊 Как работает:🧨

1. Получаем стакан и ленточку сделок в реальном времени.

2. Отслеживаем, что по определённой цене:

- объём не уменьшается заметно, хотя идут сделки;

- ордера на этой цене постоянно восстанавливаются.

3. Если это происходит многократно → сигнал о скрытом крупном игроке.

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

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

- Можно отслеживать только уровни поддержки/сопротивления с айсбергами.

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

import ccxt
import time

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

iceberg_tracker = {}
iceberg_threshold = 5  # число "восстановлений" ордера для сигнала

while True:
    try:
        orderbook = exchange.fetch_order_book(symbol, depth_limit)
        bids = orderbook['bids'][:5]
        asks = orderbook['asks'][:5]

        # Анализ обеих сторон
        for side_name, side_data in [('BUY', bids), ('SELL', asks)]:
            for price, amount in side_data:
                key = (side_name, price)

                # Если цена уже была в трекере
                if key in iceberg_tracker:
                    prev_amount, restorations = iceberg_tracker[key]
                    if amount >= prev_amount * 0.95:  
                        iceberg_tracker[key] = (amount, restorations + 1)
                    else:
                        iceberg_tracker[key] = (amount, restorations)
                else:
                    iceberg_tracker[key] = (amount, 0)

                # Сигнал, если много "восстановлений"
                if iceberg_tracker[key][1] >= iceberg_threshold:
                    print(f"🧊 Iceberg {side_name} {symbol} @ {price}, объём {amount}")
                    iceberg_tracker[key] = (amount, 0)

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

#инструмент

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

📌"Импульсная ловушка на ликвидности" (Liquidity Sweep Reversal).🧑‍💻

📌 Идея стратегии:🧨

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

🔍 Логика:🛠️

1. Определяем локальные экстремумы (хай/лоу) последних N свечей.

2. Если цена резко пробивает экстремум и возвращается обратно в диапазон:

Это значит, что был сбор ликвидности (стопы) и вероятен разворот.

3. Подтверждение — свеча закрывается внутри диапазона.

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

import ccxt
import pandas as pd
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '5m'
lookback = 20  # кол-во свечей для поиска хай/лоу

def get_ohlcv():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=lookback + 3)
    df = pd.DataFrame(ohlcv, columns=['time', 'open', 'high', 'low', 'close', 'volume'])
    return df

while True:
    try:
        df = get_ohlcv()

        recent_high = df['high'][:-1].max()
        recent_low = df['low'][:-1].min()
        last_candle = df.iloc[-1]

        # Ложный пробой вверх
        if last_candle['high'] > recent_high and last_candle['close'] < recent_high:
            print(f"🔻 SHORT сигнал на {symbol} @ {last_candle['close']} (ложный пробой вверх)")

        # Ложный пробой вниз
        if last_candle['low'] < recent_low and last_candle['close'] > recent_low:
            print(f"🔺 LONG сигнал на {symbol} @ {last_candle['close']} (ложный пробой вниз)")

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

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

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

📌DECS: Dynamic Execution Cost & Slippage Simulator🧑‍💻

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

Для любой пары и любого объёма оценивает ожидаемые затраты исполнения (спред + проскальзывание) по текущему стакану и имитирует разные варианты разбиения (slicing) ордера: один маркет, несколько частей (VWAP-подход), лимитные отложки. В выводе — рекомендуемый способ разбивки (сколько частей), ожидаемая средняя цена исполнения и прогнозируемое проскальзывание в валюте и процентах. Логирует результаты в CSV.

Почему полезен:🚀

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

- Помогает выбрать оптимальный размер частей и стратегию исполнения (market vs slice vs limit).

- Уменьшает рыночное воздействие — критично для профессионалов и алготрейдов.

- Работает через REST (ccxt) — можно запускать в cron/долгоживущем процессе и интегрировать в бэктест/реальный бот.

#инструмент

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