📌"Echo Bounce" (Эхо-отскок)🧑‍💻

💡 Идея: 🧨

Когда на рынке появляется крупный объём покупок или продаж, но цена не продолжает движение, а отскакивает обратно — это часто сигнал ложного пробоя и «эхо»-эффекта ликвидности.

Мы ищем локальные импульсы, где:

- объём резко растёт,

- цена почти не двигается,

- затем появляется быстрый откат.

Это отражает борьбу крупных ордеров и ликвидности — отличная точка для контртрендовой сделки с короткой целью.

📈 Логика: 🛠️

1. Сканируем сделки (trades) через Binance WebSocket.

2. За последние N секунд считаем:

средний объём сделки,

изменение цены.

3. Если средний объём вырос более чем в 5 раз,
но цена изменилась менее чем на 0.1% → ловим “эхо”.

4. Входим в обратную сторону от всплеска.

import json
import websocket
import threading
from datetime import datetime
from collections import deque

SYMBOL = "btcusdt"
WINDOW = 5  # секунд
VOLUME_MULTIPLIER = 5
PRICE_CHANGE_LIMIT = 0.001  # 0.1%

trades = deque(maxlen=500)

def on_message(ws, message):
    global trades
    data = json.loads(message)
    t = data['T'] / 1000
    price = float(data['p'])
    qty = float(data['q'])
    side = 'buy' if data['m'] == False else 'sell'
    trades.append((t, price, qty, side))
    
def detect_echo():
    while True:
        if len(trades) < 10:
            continue
        recent = [t for t in trades if t[0] > trades[-1][0] - WINDOW]
        if len(recent) < 3:
            continue

        prices = [x[1] for x in recent]
        vols = [x[2] for x in recent]
        avg_vol = sum(vols) / len(vols)
        avg_recent_vol = sum([x[2] for x in list(trades)[-100:]]) / 100
        price_change = abs(prices[-1] - prices[0]) / prices[0]

        if avg_vol > avg_recent_vol * VOLUME_MULTIPLIER and price_change < PRICE_CHANGE_LIMIT:
            dominant_side = max(set([x[3] for x in recent]), key=[x[3] for x in recent].count)
            signal = "SELL" if dominant_side == "buy" else "BUY"
            print(f"[{datetime.now().strftime('%H:%M:%S')}] ECHO DETECTED → {signal} | Δprice={price_change:.4f} | vol×{avg_vol/avg_recent_vol:.1f}")

def run_ws():
    ws = websocket.WebSocketApp(f"wss://stream.binance.com:9443/ws/{SYMBOL}@trade", on_message=on_message)
    ws.run_forever()

threading.Thread(target=run_ws).start()
detect_echo()

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

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

📌“Liquidity Loop Scalper” - (циклический скальпер по ликвидности без индикаторов) 🧑‍💻

💡 Идея: 🧨

Большинство пар Binance (особенно спотовые альткоины) имеют повторяющиеся микродвижения внутри лимитной книги: когда ликвидность на одной стороне (bid/ask) резко исчезает, цена почти всегда делает короткий импульс в эту сторону — затем возвращается.

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

📈 Логика: 🛠️

1. Каждую секунду получаем книгу ордеров (fetch_order_book(symbol, 50)).

2. Считаем суммарный объём первых N уровней (например 10).

bid_sum = сумма объёмов на первых 10 уровнях покупки.

ask_sum = сумма объёмов на первых 10 уровнях продажи.

3. Сравниваем текущее состояние с предыдущим:

Если bid_sum ↓ более чем на 40% за <2 секунды, но цена не упала — значит “ликвидность сняли, но не продали”, то есть ложный пробой вниз.
→ Входим в лонг.

Если ask_sum ↓ более чем на 40%, но цена не выросла — ложный пробой вверх.
→ Входим в шорт (или продаём, если спот).

4. Выход — при возвращении ликвидности (bid_sum или ask_sum восстанавливается до 80% прежнего уровня) или по прибыли >0.3%.

import ccxt
import time

exchange = ccxt.binance({'enableRateLimit': True})
symbol = 'DOGE/USDT'

N = 10
DROP_THRESHOLD = 0.4
PROFIT_TARGET = 0.003

last_bid_sum = None
last_ask_sum = None
entry_price = None
position = None

print("Liquidity Loop Scalper running...")

while True:
    ob = exchange.fetch_order_book(symbol, limit=50)
    bid_sum = sum([q for _, q in ob['bids'][:N]])
    ask_sum = sum([q for _, q in ob['asks'][:N]])
    mid = (ob['bids'][0][0] + ob['asks'][0][0]) / 2

    if last_bid_sum and last_ask_sum:
        bid_change = (bid_sum - last_bid_sum) / last_bid_sum
        ask_change = (ask_sum - last_ask_sum) / last_ask_sum

        # вход в лонг при исчезновении бидов без падения цены
        if bid_change < -DROP_THRESHOLD and position is None:
            print(f"[{symbol}] ⚡ Лонг — исчезли биды, но цена держится ({bid_change:.1%})")
            entry_price = mid
            position = "long"

        # вход в шорт при исчезновении асков без роста цены
        elif ask_change < -DROP_THRESHOLD and position is None:
            print(f"[{symbol}] ⚡ Продажа — исчезли аски, но цена не растёт ({ask_change:.1%})")
            entry_price = mid
            position = "short"

        # выход из позиции
        if position == "long" and (mid - entry_price) / entry_price > PROFIT_TARGET:
            print(f"[{symbol}] ✅ Выход из лонга: прибыль {((mid-entry_price)/entry_price):.2%}")
            position = None
        elif position == "short" and (entry_price - mid) / entry_price > PROFIT_TARGET:
            print(f"[{symbol}] ✅ Выход из шорта: прибыль {((entry_price-mid)/entry_price):.2%}")
            position = None

    last_bid_sum, last_ask_sum = bid_sum, ask_sum
    time.sleep(1)

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

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

📌"Liquidity Vacuum"- (Пылесос ликвидности) 🧑‍💻

💡 Идея🧨

Цена не движется сама по себе — она падает туда, где пусто в стакане, и растёт туда, где продавцы исчезли.
Когда крупные лимитки внезапно исчезают (cancel walls) — рынок обычно движется в сторону исчезновения ликвидности, потому что это «дырка», куда притягиваются рыночные ордера.

Мы не угадываем направление — мы следим за исчезновением плотности в стакане и входим туда, куда исчезла поддержка (или сопротивление).

🧩 Логика:🛠️

1. Раз в секунду получаем стакан.

2. Считаем локальную плотность (сумма объёмов) в радиусе ±0.2% от mid-price.

3. Если в верхней половине (asks) ликвидность внезапно упала > X% — ждём всплеск рыночных покупок → входим в лонг.

4. Если в нижней половине (bids) ликвидность внезапно исчезла — ждём падения → входим в шорт.

5. Выход — при обратном восстановлении плотности или при изменении цены на Y%.

import ccxt
import time
import statistics

SYMBOL = 'BTC/USDT'
EXCHANGE = ccxt.binance({'enableRateLimit': True})
INTERVAL = 1.0
THRESHOLD_DROP = 0.35   # ликвидность упала на 35%
PRICE_MOVE_TP = 0.002   # 0.2% take profit
PRICE_MOVE_SL = 0.0015  # 0.15% stop loss

def get_book():
    ob = EXCHANGE.fetch_order_book(SYMBOL, limit=50)
    bids = ob['bids']
    asks = ob['asks']
    mid = (bids[0][0] + asks[0][0]) / 2
    return bids, asks, mid

def local_density(bids, asks, mid, pct=0.002):
    low = mid * (1 - pct)
    high = mid * (1 + pct)
    bid_vol = sum(v for p, v in bids if p >= low)
    ask_vol = sum(v for p, v in asks if p <= high)
    return bid_vol, ask_vol

history = []

print("Liquidity Vacuum strategy started...\n")

pos = None
entry_price = 0

while True:
    bids, asks, mid = get_book()
    bid_vol, ask_vol = local_density(bids, asks, mid)
    history.append((bid_vol, ask_vol, mid))
    if len(history) < 3:
        time.sleep(INTERVAL)
        continue

    prev_bid, prev_ask, _ = history[-2]
    drop_bids = (prev_bid - bid_vol) / prev_bid if prev_bid else 0
    drop_asks = (prev_ask - ask_vol) / prev_ask if prev_ask else 0

    # ---- Вход ----
    if not pos:
        if drop_asks > THRESHOLD_DROP:
            pos = 'LONG'
            entry_price = mid
            print(f"💚 LONG entry {mid:.2f} (asks collapsed {drop_asks*100:.1f}%)")
        elif drop_bids > THRESHOLD_DROP:
            pos = 'SHORT'
            entry_price = mid
            print(f"❤️ SHORT entry {mid:.2f} (bids collapsed {drop_bids*100:.1f}%)")

    # ---- Выход ----
    elif pos == 'LONG':
        if mid >= entry_price * (1 + PRICE_MOVE_TP):
            print(f"✅ LONG TP at {mid:.2f}")
            pos = None
        elif mid <= entry_price * (1 - PRICE_MOVE_SL):
            print(f"❌ LONG SL at {mid:.2f}")
            pos = None
    elif pos == 'SHORT':
        if mid <= entry_price * (1 - PRICE_MOVE_TP):
            print(f"✅ SHORT TP at {mid:.2f}")
            pos = None
        elif mid >= entry_price * (1 + PRICE_MOVE_SL):
            print(f"❌ SHORT SL at {mid:.2f}")
            pos = None

    time.sleep(INTERVAL)

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

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

📌“Micro-Impulse Trap” — Ловушка микроимпульсов🧑‍💻

💡 Идея: 🧨

На микросекундных движениях часто наблюдается закономерность:
после короткого сильного импульса, за которым следует резкая пауза без продолжения, рынок возвращается к точке старта импульса — это реакция алгоритмов, которые не нашли ликвидности для продолжения.

Мы отслеживаем:

быстрые скачки цены (например, >0.15% за секунду);

отсутствие обновления high/low в следующие 2 секунды;

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

🧩 Алгоритм: 🛠️

1. Каждую секунду фиксируем текущую цену.

2. Если за 1 секунду цена выросла >0.15% — запоминаем уровень импульса.

3. Через 2 секунды:

если цена не выше импульсного high → вход в шорт (ожидание возврата).

4. Если цена упала >0.15%:

через 2 секунды, если не ниже low → вход в лонг.

5. Цель — возврат на 0.1–0.2% к средней.

import ccxt
import time
from collections import deque

exchange = ccxt.binance({'enableRateLimit': True})
symbol = 'ETH/USDT'
window = deque(maxlen=3)
position = None
entry = 0

THR_IMPULSE = 0.0015  # 0.15%
TP = 0.001            # 0.1%
SL = 0.0015           # 0.15%

print("Micro-Impulse Trap started...\n")

while True:
    price = exchange.fetch_ticker(symbol)['last']
    window.append(price)

    if len(window) == 3:
        p_now = window[-1]
        p_prev = window[-2]
        p_old = window[0]

        impulse = (p_prev - p_old) / p_old

        # восходящий импульс без продолжения
        if impulse > THR_IMPULSE and position is None:
            time.sleep(2)
            new_price = exchange.fetch_ticker(symbol)['last']
            if new_price <= p_prev:  # нет обновления high
                position = 'short'
                entry = new_price
                print(f"📉 SHORT {symbol} at {entry:.3f} — импульс вверх погашен")

        # нисходящий импульс без продолжения
        elif impulse < -THR_IMPULSE and position is None:
            time.sleep(2)
            new_price = exchange.fetch_ticker(symbol)['last']
            if new_price >= p_prev:  # нет обновления low
                position = 'long'
                entry = new_price
                print(f"📈 LONG {symbol} at {entry:.3f} — импульс вниз погашен")

        # выход
        if position == 'long':
            if price >= entry * (1 + TP):
                print(f"✅ TP LONG {symbol} at {price:.3f}")
                position = None
            elif price <= entry * (1 - SL):
                print(f"❌ SL LONG {symbol} at {price:.3f}")
                position = None

        elif position == 'short':
            if price <= entry * (1 - TP):
                print(f"✅ TP SHORT {symbol} at {price:.3f}")
                position = None
            elif price >= entry * (1 + SL):
                print(f"❌ SL SHORT {symbol} at {price:.3f}")
                position = None

    time.sleep(1)

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

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

📌“Echo Rebound” (Эхо-отскок цены)🧑‍💻

Тип: скальпинг/интрадей

Платформа: спот Binance

Идея:🧨

рынок имеет “эхо” — если импульс вверх быстро поглощается встречными ордерами, часто через 5–15 секунд следует возврат точно в противоположную сторону.

💡 Логика:🛠️

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

2. Если за последние 3 секунды цена выросла >0.25%,
но в следующие 2 секунды не обновляет high, значит импульс поглощён — это ловушка покупателей.
→ Вход в шорт.

3. Если за последние 3 секунды цена упала >0.25%,
но в следующие 2 секунды не обновляет low, значит импульс поглощён — это ловушка продавцов.
→ Вход в лонг.

4. Выход:

- при обратном движении ±0.15–0.25%,

- или через 10 секунд, если цена не сдвинулась.

import ccxt
import time
from collections import deque

symbol = 'BTC/USDT'
exchange = ccxt.binance({'enableRateLimit': True})

WINDOW_SEC = 5
price_history = deque(maxlen=WINDOW_SEC)
position = None
entry_price = 0

print("Echo Rebound Strategy running...")

while True:
    ticker = exchange.fetch_ticker(symbol)
    price = ticker['last']
    price_history.append(price)
    
    if len(price_history) == WINDOW_SEC:
        p_now = price_history[-1]
        p_3s = price_history[-3]
        
        move = (p_now - p_3s) / p_3s

        # вход в шорт
        if move > 0.0025 and position is None:
            # ждём 2 сек — если high не обновился, значит импульс поглощён
            time.sleep(2)
            new_price = exchange.fetch_ticker(symbol)['last']
            if new_price <= price:
                position = 'short'
                entry_price = new_price
                print(f"📉 SHORT {symbol} at {entry_price:.2f} (echo down)")

        # вход в лонг
        elif move < -0.0025 and position is None:
            time.sleep(2)
            new_price = exchange.fetch_ticker(symbol)['last']
            if new_price >= price:
                position = 'long'
                entry_price = new_price
                print(f"📈 LONG {symbol} at {entry_price:.2f} (echo up)")

        # выход
        if position == 'long':
            if price >= entry_price * 1.0025:
                print(f"✅ Take profit LONG at {price:.2f}")
                position = None
            elif price <= entry_price * 0.9985:
                print(f"❌ Stop loss LONG at {price:.2f}")
                position = None

        elif position == 'short':
            if price <= entry_price * 0.9975:
                print(f"✅ Take profit SHORT at {price:.2f}")
                position = None
            elif price >= entry_price * 1.0015:
                print(f"❌ Stop loss SHORT at {price:.2f}")
                position = None

    time.sleep(1)

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

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

📌“Micro Rebound Engine” — реакция на микро-дисбалансы после резких сбросов🧑‍💻

💡 Идея: 🧨

Эта стратегия наблюдает не за свечами и не за индикаторами, а за поведением микродвижений цены — как рынок реагирует на мгновенные сбросы (mini-dump).
Когда происходит серия быстрых падений, но объём продаж начинает иссякать, рынок часто делает технический микровыкуп — короткий отскок.

Ты ловишь именно этот первый микровыкуп, пока толпа ещё в панике.

🧩 Принцип: 🛠️

1. Отслеживаем цену раз в 0.5–1 секунду.

2. Если за последние 10 секунд:

цена упала на ≥ X % (например, 0.4%),

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

3. Закрываем сделку, как только цена отрастает на 0.3% или проходит 20 секунд (всё — микро-реакция).

import ccxt
import time
import numpy as np

exchange = ccxt.binance({'enableRateLimit': True})
symbol = 'BTC/USDT'
prices = []

print(f"🔥 Запуск стратегии Micro Rebound Engine для {symbol}")

while True:
    ticker = exchange.fetch_ticker(symbol)
    price = ticker['last']
    prices.append(price)

    # Храним последние 20 цен
    if len(prices) > 20:
        prices.pop(0)

        recent_drop = (prices[-10] - prices[-1]) / prices[-10] * 100
        last_deltas = np.diff(prices[-5:])
        delta_change = np.mean(last_deltas[:-1]) - np.mean(last_deltas[-2:])

        if recent_drop <= -0.4 and delta_change > 0:
            entry = prices[-1]
            print(f"[LONG] Потенциальный микро-отскок: {entry}")
            t0 = time.time()

            while time.time() - t0 < 20:
                p = exchange.fetch_ticker(symbol)['last']
                if (p - entry) / entry * 100 >= 0.3:
                    print(f"[EXIT] Микро-профит зафиксирован: {p}")
                    break
                time.sleep(1)

    time.sleep(1)

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

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

📌“Echo Reversal” — стратегия отзвуков инерции🧑‍💻

💡 Идея: 🧨

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

Ты ловишь этот момент, когда импульс выдыхается, а рынок ещё не перешёл в боковик.

🧩 Принцип работы: 🛠️

1. Измеряем скорость движения цены (разницу между текущей и предыдущей ценой).

2. Измеряем ускорение — насколько быстро меняется скорость.

3. Если:

скорость положительная (цена растёт),

но ускорение становится отрицательным (рост замедляется),
— значит импульс исчерпывается → вход в шорт.

4. Аналогично, если цена падает, но ускорение становится положительным → лонг.

import ccxt
import time
import numpy as np

exchange = ccxt.binance({'enableRateLimit': True})
symbol = 'BTC/USDT'
prices = []

print(f"🚀 Запуск стратегии Echo Reversal для {symbol}")

while True:
    ticker = exchange.fetch_ticker(symbol)
    price = ticker['last']
    prices.append(price)

    if len(prices) > 6:
        speeds = np.diff(prices[-4:])  # скорость
        accels = np.diff(speeds)       # ускорение

        speed = np.mean(speeds[-2:])
        accel = np.mean(accels)

        if speed > 0 and accel < 0:
            print(f"[SHORT] Импульс затухает — потенциальный разворот вниз при {price}")
        elif speed < 0 and accel > 0:
            print(f"[LONG] Импульс выдыхается — потенциальный разворот вверх при {price}")

        prices = prices[-6:]

    time.sleep(2)

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

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