📌Sweep Reaction 🧑‍💻

💡 Идея: 🧨

Когда цена делает резкий прокол (sweep) локального минимума или максимума и мгновенно возвращается обратно в диапазон, это признак того, что:

ликвидность была собрана (сняли стопы),

крупный игрок взял объём против толпы,

импульс в противоположную сторону только начинается.

Мы ловим реакцию на этот сбор ликвидности, а не сам момент прокола.

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

1. Определяем последние N свечей (например, 50) и находим локальный high/low.

2. Если текущая цена:

делает прокол локального экстремума (на X% или пунктов),

и закрывается обратно внутрь диапазона,
то это sweep.

3. После sweep:

Если был прокол низа → открываем лонг при возврате цены внутрь диапазона.

Если прокол верха → открываем шорт при возврате вниз.

4. Стоп — за хвостом свечи, тейк — середина диапазона или 2× стоп.

import ccxt
import pandas as pd
import time

exchange = ccxt.binance({
    'options': {'defaultType': 'future'}
})

symbol = "BTC/USDT"
tf = "5m"
sweep_threshold = 0.001  # 0.1%

def detect_sweep(df):
    high_range = df['high'].iloc[:-1].max()
    low_range = df['low'].iloc[:-1].min()

    last = df.iloc[-1]
    prev_close = df['close'].iloc[-2]

    # Прокол низа
    if last['low'] < low_range * (1 - sweep_threshold) and last['close'] > low_range:
        return 'long', low_range
    
    # Прокол верха
    if last['high'] > high_range * (1 + sweep_threshold) and last['close'] < high_range:
        return 'short', high_range
    
    return None, None

while True:
    ohlcv = exchange.fetch_ohlcv(symbol, tf, limit=100)
    df = pd.DataFrame(ohlcv, columns=['ts','open','high','low','close','vol'])

    signal, level = detect_sweep(df)
    if signal:
        print(f"[⚡ SWEEP DETECTED] {symbol} → {signal.upper()} | уровень: {round(level,2)}")

    time.sleep(60)

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

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

📌“Liquidity Map Generator” — Карта ликвидности в реальном времени 🧑‍💻

💡 Суть: 🧨

Этот инструмент анализирует стакан Binance Futures (USDT-M) и визуализирует уровни, где сосредоточена наибольшая ликвидность (крупные лимитные ордера).

Он помогает видеть, куда “притягивается” цена — к скоплению ликвидности (магниту).

⚙️ Что делает функция: 🛠️

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

2. Группирует ордера по диапазонам цены (например, каждые 0.1% от текущей).

3. Находит топ-зоны концентрации объёма.

4. Возвращает данные и рисует "тепловую карту" ликвидности прямо в консоли.

import ccxt
import numpy as np
import pandas as pd
import time
from rich import print
from rich.table import Table

def liquidity_map(symbol="BTC/USDT", bins=20, limit=200):
    """
    Генератор карты ликвидности Binance Futures (USDT-M).
    
    :param symbol: Торговая пара (пример: 'BTC/USDT')
    :param bins: Кол-во диапазонов вокруг текущей цены
    :param limit: Глубина стакана
    """
    exchange = ccxt.binance({'options': {'defaultType': 'future'}})
    orderbook = exchange.fetch_order_book(symbol, limit=limit)
    last_price = exchange.fetch_ticker(symbol)['last']

    asks = np.array(orderbook['asks'])
    bids = np.array(orderbook['bids'])

    df_asks = pd.DataFrame(asks, columns=['price', 'volume'])
    df_bids = pd.DataFrame(bids, columns=['price', 'volume'])

    df_asks['type'] = 'ask'
    df_bids['type'] = 'bid'
    df = pd.concat([df_asks, df_bids])

    df['dist_%'] = 100 * (df['price'] - last_price) / last_price
    df['dist_bin'] = pd.cut(df['dist_%'], bins=bins)

    grouped = df.groupby(['dist_bin', 'type'])['volume'].sum().reset_index()

    # Визуализация
    table = Table(title=f"💧 Карта ликвидности {symbol} (цена: {round(last_price,2)})")
    table.add_column("Диапазон %", justify="right")
    table.add_column("Bid объём", justify="right")
    table.add_column("Ask объём", justify="right")

    for dist_bin in grouped['dist_bin'].unique():
        bid_vol = grouped[(grouped['dist_bin'] == dist_bin) & (grouped['type'] == 'bid')]['volume'].sum()
        ask_vol = grouped[(grouped['dist_bin'] == dist_bin) & (grouped['type'] == 'ask')]['volume'].sum()
        if bid_vol + ask_vol > 0:
            table.add_row(str(dist_bin), f"{bid_vol:.2f}", f"{ask_vol:.2f}")

    print(table)

# 🔍 Пример вызова
if __name__ == "__main__":
    while True:
        liquidity_map("BTC/USDT")
        time.sleep(15)

#инструмент

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

📌“Liquidity Gravity Map” - (Карта гравитации ликвидности) 🧑‍💻

💡 Идея: 🧨

Цена на рынке тянется туда, где накапливается ликвидность — лимитные ордера, стопы, зоны плотного объёма.

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

🧠 Применение: 💸

Определить, куда с большей вероятностью пойдёт цена (в сторону сильной ликвидности).

Найти “магниты цены” — уровни, где часто происходят отскоки или ложные пробои.

Использовать для входа в направлении притяжения или для контр-трейдов от зон концентрации ордеров.

import ccxt
import numpy as np
import time

def liquidity_gravity(symbol="BTC/USDT", depth_limit=100):
    exchange = ccxt.binance({'options': {'defaultType': 'future'}})
    orderbook = exchange.fetch_order_book(symbol, limit=depth_limit)

    bids = np.array(orderbook['bids'])
    asks = np.array(orderbook['asks'])
    mid_price = (asks[0, 0] + bids[0, 0]) / 2

    def gravity_side(side, direction):
        g_total = 0
        for price, volume in side:
            distance = abs(price - mid_price)
            if distance == 0: 
                continue
            g_total += volume / (distance ** 2)
        if direction == "up":
            return g_total, np.average([p for p, _ in side], weights=[v for _, v in side])
        else:
            return g_total, np.average([p for p, _ in side], weights=[v for _, v in side])

    g_bids, bid_center = gravity_side(bids, "up")
    g_asks, ask_center = gravity_side(asks, "down")

    bias = "UP" if g_bids > g_asks else "DOWN"
    gravity_center = (bid_center * g_bids + ask_center * g_asks) / (g_bids + g_asks)

    print(f"\n📊 {symbol}")
    print(f"Текущая цена: {mid_price:.2f}")
    print(f"Центр гравитации: {gravity_center:.2f}")
    print(f"Сила гравитации BID: {g_bids:.2f}")
    print(f"Сила гравитации ASK: {g_asks:.2f}")
    print(f"👉 Преобладание: {bias}")

if __name__ == "__main__":
    while True:
        try:
            liquidity_gravity("BTC/USDT", 100)
            time.sleep(5)
        except Exception as e:
            print("Ошибка:", e)
            time.sleep(2)

#инструмент

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

📌“Micro-Rebound Hunter” - (Охотник за микровозвратами) 🧑‍💻

💡 Идея: 🧨

Цена на спотовом рынке Binance часто делает быстрые ложные пробои (wicks) при повышенной волатильности, особенно на младших таймфреймах.

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

⚙️ Логика входа: 🛠️

1. Находим свечу, где:

диапазон high - low ≥ X% (например, 0.5%)

цена закрытия находится в нижних 20% диапазона (для лонга)

либо в верхних 20% диапазона (для шорта)

2. Входим в противоположную сторону импульса, если следующая свеча подтверждает слабость (не обновляет экстремум).

3. Тейк фиксированный (например, +0.2%), стоп за экстремумом предыдущей свечи.

import ccxt
import pandas as pd
import time

def micro_rebound_hunter(symbol="BTC/USDT", timeframe="1m", percent_range=0.5, tp=0.25, sl=0.2):
    exchange = ccxt.binance()
    bars = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=200)
    df = pd.DataFrame(bars, columns=['time','open','high','low','close','vol'])
    df['time'] = pd.to_datetime(df['time'], unit='ms')
    
    last = df.iloc[-1]
    prev = df.iloc[-2]
    body_range = (last['high'] - last['low']) / last['low'] * 100

    if body_range >= percent_range:
        # Потенциальный лонг
        if last['close'] <= last['low'] + 0.2 * (last['high'] - last['low']):
            print(f"🟢 [{symbol}] Потенциальный ЛОНГ на {last['close']}")
            print(f"Стоп: {last['low']:.2f}, Тейк: {last['close']*(1+tp/100):.2f}")
        
        # Потенциальный шорт
        elif last['close'] >= last['high'] - 0.2 * (last['high'] - last['low']):
            print(f"🔴 [{symbol}] Потенциальный ШОРТ на {last['close']}")
            print(f"Стоп: {last['high']:.2f}, Тейк: {last['close']*(1-tp/100):.2f}")

# 🔁 Запуск цикла сканера
if __name__ == "__main__":
    while True:
        try:
            micro_rebound_hunter("BTC/USDT")
            time.sleep(30)
        except Exception as e:
            print("Ошибка:", e)
            time.sleep(5)

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

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