📌"Order Book Velocity" - (скорость изменения стакана) 🧑‍💻

📌 Суть: 🧨

Большинство смотрит на сам стакан (где стоят лимитки).
Но гораздо важнее — как быстро меняется глубина и структура заявок.

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

1. Подключается к Binance через WebSocket и берёт стакан в реальном времени.

2. Каждую секунду считает:

насколько вырос/уменьшился суммарный объём на стороне BID (покупатели),

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

3. Строит метрику

Если отрицательная → продавцы давят сильнее.

import asyncio
import json
import websockets
from collections import defaultdict

# Пара для анализа
SYMBOL = "btcusdt"
DEPTH_LEVEL = 10  # глубина стакана для анализа
INTERVAL = 1      # интервал в секундах для подсчёта velocity

# Храним прошлые значения объёмов
previous = {"bids": 0, "asks": 0}

async def orderbook_velocity():
    url = f"wss://stream.binance.com:9443/ws/{SYMBOL}@depth@100ms"
    async with websockets.connect(url) as ws:
        print(f"✅ Connected to Binance Order Book for {SYMBOL.upper()}")
        buffer_bids = defaultdict(float)
        buffer_asks = defaultdict(float)

        while True:
            msg = await ws.recv()
            data = json.loads(msg)

            # Берем только DEPTH_LEVEL уровней
            bids = data.get("b", [])[:DEPTH_LEVEL]
            asks = data.get("a", [])[:DEPTH_LEVEL]

            # Сумма объёмов
            total_bids = sum(float(b[1]) for b in bids)
            total_asks = sum(float(a[1]) for a in asks)

            # Разница с предыдущим состоянием
            delta_bids = total_bids - previous["bids"]
            delta_asks = total_asks - previous["asks"]

            velocity = delta_bids - delta_asks

            print(f"📊 Velocity: {velocity:.4f} | ΔBids: {delta_bids:.2f} | ΔAsks: {delta_asks:.2f}")

            # Обновляем
            previous["bids"] = total_bids
            previous["asks"] = total_asks

            await asyncio.sleep(INTERVAL)

if __name__ == "__main__":
    try:
        asyncio.run(orderbook_velocity())
    except KeyboardInterrupt:
        print("⛔ Stopped by user")

#инструмент

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

📌"Liquidity Magnet"- (магнит ликвидности — определяет, где “” цена) 🧑‍💻

💡 Идея: 🧨

Цена всегда тянется туда, где сконцентрированы незакрытые ордера и уровни ликвидности.
Инструмент в реальном времени отслеживает ценовые кластеры, где:

происходят частые касания свечей,

но цена не может уйти далеко.

Такие зоны — это “магниты” — туда рынок тянет ликвидность и часто пробивает их импульсом.

🧠 Принцип: 🛠️

1. Для каждой пары получаем последние N свечей (например, 200 × 1m).

2. Разбиваем диапазон цен на сетку (например, шаг 0.1%).

3. Подсчитываем, сколько свечей “касались” каждого уровня.

4. Если уровень набирает аномально много касаний → отмечаем как магнит.

import ccxt
import numpy as np

exchange = ccxt.binance()
symbol = 'BTC/USDT'
candles = exchange.fetch_ohlcv(symbol, timeframe='1m', limit=200)

prices = [c[4] for c in candles]
highs = [c[2] for c in candles]
lows = [c[3] for c in candles]

min_price, max_price = min(lows), max(highs)
bins = np.linspace(min_price, max_price, 100)

touches = np.zeros(len(bins))
for h, l in zip(highs, lows):
    for i, level in enumerate(bins):
        if l <= level <= h:
            touches[i] += 1

threshold = np.percentile(touches, 95)
magnets = [(bins[i], int(t)) for i, t in enumerate(touches) if t >= threshold]

print("🧲 Liquidity Magnets Detected:")
for level, count in magnets:
    print(f"→ Level: {level:.2f} | touches: {count}")

#инструмент

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

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

📌“Market Memory Map” - (анализатор рыночной памяти — где цена «помнит» уровни ликвидности) 🧑‍💻

💡 Идея: 🧨

Рынок часто возвращается к ценам, где происходили сильные обороты — это и есть “память рынка”.
Инструмент создаёт тепловую карту “памяти” — показывает, на каких ценовых уровнях цена останавливалась, ускорялась или меняла волатильность.

Это не индикатор, не сигнал — это инструмент для осознанной визуализации структуры рынка.

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

1. Загружает исторические свечи (klines) за заданный период.

2. Для каждого уровня цены (например, шаг 0.2%) считает:

сколько раз цена в него “заходила”,

сколько раз от него резко отскакивала.

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

4. Показывает ключевые “зоны памяти” — где рынок чаще всего реагировал.


🔍 Интерпретация: 💸

🔥 Высокие пики → цена часто меняла направление → “зоны памяти” рынка.

⚙️ Плоские зоны → нейтральные участки без реакции.

import ccxt
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

symbol = "BTC/USDT"
tf = "1h"
limit = 1000
exchange = ccxt.binance()

# Загружаем свечи
data = exchange.fetch_ohlcv(symbol, tf, limit=limit)
df = pd.DataFrame(data, columns=["time", "open", "high", "low", "close", "volume"])
df["time"] = pd.to_datetime(df["time"], unit="ms")

# Готовим сетку уровней
min_price, max_price = df["low"].min(), df["high"].max()
bins = np.linspace(min_price, max_price, 200)
reactions = np.zeros(len(bins))

# Считаем реакции на каждом уровне
for _, row in df.iterrows():
    for i, lvl in enumerate(bins):
        # если цена прошла уровень и изменила направление — "реакция"
        if row["low"] <= lvl <= row["high"]:
            if (row["close"] - row["open"]) * (lvl - row["open"]) < 0:
                reactions[i] += 1

# Нормализация
reactions = reactions / reactions.max()

# Визуализация
plt.figure(figsize=(10, 6))
plt.bar(bins, reactions, width=(bins[1]-bins[0]), color="cyan", edgecolor="black")
plt.title(f"Market Memory Map — {symbol} ({tf})")
plt.xlabel("Price Level")
plt.ylabel("Reaction Intensity (normalized)")
plt.grid(True, alpha=0.3)
plt.show()

#инструмент

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

📌"Liquidity Prober" - (пробник ликвидности — активное картирование реальной глубины и refill-реакции стакана) 🧑‍💻

🔑 Идея (коротко)🧨

Вместо того чтобы полагаться на статический снимок стакана, этот инструмент активно (и осторожно) „прощупывает“ рынок небольшими рыночными/лимитными пробами, фиксирует реальный проскальзывание, скорость восстановления (refill) и поведение цены. По результатам даёт рекомендацию по стратегии исполнения (сколько слайсов, паузы между ними, безопасная доля от баланса).

Полезно для: исполнение больших заявок, алгоритмической дробки, оценки риска слippage перед «прыжком».

📌Что он делает: 🛠

делает N маленьких проб (probe) в одну сторону (buy или sell) — в dry-run можно только эмулировать; при live — реально отправляет маленькие market/limit ордера;

для каждого пробы фиксирует:

среднюю цену исполнения, проскальзывание vs mid, изменение best bid/ask, и время восстановления объёма в локальной зоне;

- строит статистику: mean/std проскальзывания, среднее время refill;

- на основе результатов выдаёт рекомендацию: slice_count, slice_delay_sec, max_notional_fraction_per_slice.


#инструмент

📌 Подпишись  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❗️

📌“Volatility Pulse” — живой пульс волатильности рынка 🧑‍💻

💡 Идея: 🧨

Инструмент отслеживает реальную рыночную активность — темп изменения цены и объёма за секунду.
Он показывает, какие пары “оживают” прямо сейчас, ещё до того, как начнётся движение.
То есть ты видишь импульс в момент его зарождения, а не после свечи.

import ccxt
import time
import numpy as np

exchange = ccxt.binance({'enableRateLimit': True})
symbols = [s['symbol'] for s in exchange.load_markets().values() if '/USDT' in s['symbol']]

price_data = {s: [] for s in symbols}
volume_data = {s: [] for s in symbols}

print("🔥 Volatility Pulse — живой монитор рыночной активности\n")

while True:
    pulse = {}
    for symbol in symbols[:50]:  # ограничим первые 50 пар
        try:
            ticker = exchange.fetch_ticker(symbol)
            price = ticker['last']
            vol = ticker['baseVolume']

            price_data[symbol].append(price)
            volume_data[symbol].append(vol)

            if len(price_data[symbol]) > 3:
                dp = abs(price_data[symbol][-1] - price_data[symbol][-2]) / price_data[symbol][-2]
                dv = abs(volume_data[symbol][-1] - volume_data[symbol][-2]) / (volume_data[symbol][-2] + 1e-9)
                pulse[symbol] = dp * dv

                price_data[symbol] = price_data[symbol][-3:]
                volume_data[symbol] = volume_data[symbol][-3:]
        except Exception:
            continue

    # Топ 5 по пульсу
    if pulse:
        top = sorted(pulse.items(), key=lambda x: x[1], reverse=True)[:5]
        print("\n🔥 Топ активных пар:")
        for sym, score in top:
            print(f"{sym}: VPI={score:.6f}")
    time.sleep(3)

#инструмент

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

📌Order Flow Gradient (OFG)🧑‍💻

💡 Суть:🧨

Order Flow Gradient — это инструмент, измеряющий наклон дисбаланса ордеров во времени, а не просто их объём.
Он показывает, как быстро смещается сила покупателей и продавцов в сторону одной из сторон стакана.

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

1. Каждую секунду запрашиваются данные стакана (orderbook) с Binance.

2. Для каждой стороны (bid/ask) вычисляется:

G = \frac{\Delta(\text{объём})}{\Delta t}

3. Разница между G_bid и G_ask даёт градиент потока ордеров.

Если G_bid > G_ask, значит покупатели агрессивно наращивают объём — растущий импульс.

Если G_ask > G_bid, значит продавцы усиливаются — возможен откат или дамп.

import ccxt
import time

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

def calc_gradient(prev_bids, prev_asks, bids, asks):
    bid_vol_prev = sum(v for _, v in prev_bids[:10])
    ask_vol_prev = sum(v for _, v in prev_asks[:10])
    bid_vol = sum(v for _, v in bids[:10])
    ask_vol = sum(v for _, v in asks[:10])
    return (bid_vol - bid_vol_prev), (ask_vol - ask_vol_prev)

prev = exchange.fetch_order_book(symbol)
time.sleep(1)

print(f"Order Flow Gradient tracking {symbol}...\n")

while True:
    ob = exchange.fetch_order_book(symbol)
    bid_grad, ask_grad = calc_gradient(prev['bids'], prev['asks'], ob['bids'], ob['asks'])
    diff = bid_grad - ask_grad

    if diff > 0:
        print(f"🟢 Покупатели усиливаются: градиент {diff:.4f}")
    elif diff < 0:
        print(f"🔴 Продавцы усиливаются: градиент {diff:.4f}")
    else:
        print(f"⚪ Баланс сил")

    prev = ob
    time.sleep(1)

#инструмент

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

📌“Latency Pulse Monitor” — монитор сетевого пульса рынка🧑‍💻

💡 Идея: 🧨

Большинство трейдеров смотрят только на цену и объём, но немногие отслеживают задержку между запросом и ответом от биржи.
А ведь скачки latency — это ранний признак:

перегрузки API (рынок в панике),

приближающихся резких движений (повышенная активность),

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

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

⚙️ Принцип: 🛠️

1. Каждые 1–2 секунды отправляется запрос fetch_ticker() к бирже.

2. Измеряется время отклика.

3. Рассчитывается латентность-среднее и скачки отклонений.

4. Если latency растёт >50% от среднего — выводится предупреждение ⚠️, что рынок переходит в “нагруженное состояние”.

import ccxt
import time
import numpy as np

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

latencies = []

print("⚙️ Latency Pulse Monitor запущен...\n")

while True:
    start = time.time()
    try:
        exchange.fetch_ticker(symbol)
    except Exception:
        continue
    latency = time.time() - start
    latencies.append(latency)

    if len(latencies) > 30:
        latencies.pop(0)

        mean_lat = np.mean(latencies)
        std_lat = np.std(latencies)
        last_lat = latencies[-1]

        if last_lat > mean_lat + 2 * std_lat:
            print(f"⚠️ Всплеск сетевой задержки ({last_lat:.3f}s) — возможен всплеск активности на рынке!")
        else:
            print(f"✅ Нормальный пульс: {last_lat:.3f}s (среднее {mean_lat:.3f}s)")
    time.sleep(2)

#инструмент

📌 Подпишись  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❗️

📌“Market Friction Meter” — измеритель трения рынка🧑‍💻

💡 Идея: 🧨

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

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

import ccxt
import time
import numpy as np

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

print(f"🧭 Market Friction Meter активен для {symbol}")

prev_mid = None
prev_bid_vol = None
prev_ask_vol = None

while True:
    ob = exchange.fetch_order_book(symbol)
    bids = np.array(ob['bids'][:20])
    asks = np.array(ob['asks'][:20])

    bid_vol = bids[:,1].sum()
    ask_vol = asks[:,1].sum()
    mid_price = (bids[0,0] + asks[0,0]) / 2

    if prev_mid is not None:
        delta_p = abs(mid_price - prev_mid)
        delta_v = abs((bid_vol - prev_bid_vol) + (ask_vol - prev_ask_vol))
        FI = delta_p / (delta_v + 1e-9)

        if FI < 1e-5:
            print(f"⚠️ Высокое рыночное трение ({FI:.2e}) — готовится резкий пробой")
        elif FI > 1e-3:
            print(f"💨 Низкое трение ({FI:.2e}) — цена движется свободно")
        else:
            print(f"ℹ️ Среднее трение ({FI:.2e}) — рынок стабилен")

    prev_mid = mid_price
    prev_bid_vol = bid_vol
    prev_ask_vol = ask_vol
    time.sleep(2)

#инструмент

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

📌Cross-Pair Basket Mean-Reverter
- (формирует корзину похожих монет и торгует отклонения спреда между базовой парой и корзиной)🧑‍💻

Идея (коротко): 🧨

Берём целевую пару (например, ABC/USDT) и автоматически подбираем небольшую корзину коррелированных инструментов (например, 3–5 пар).
Строим взвешенный средний цены корзины и считаем спред (price_target / basket_price).
Если спред отклоняется далеко от среднего (z-score > порог) — торгуем на возврат:

спред в верхней хвостовой части → шорт target / лонг корзину,

в нижней хвостовой части → лонг target / шорт корзину.

Это позволяет ловить арбитраж/mean-reversion между похожими активами без сложных ML.

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

Работает на реальном поведении рынка (корреляции) — не индикаторы.

Подходит для спота и «парного» размещения лимитных/market ордеров.

Уменьшает риск направления рынка: открываются две противоположные позиции (target vs basket).

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

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