📌Execution Footprint Reclaim (EFR) - (восстановление после «съеденного» стакана — вход на откате после sweep'а) 🧑‍💻

Идея и почему это работает: 🧨

Киты и маркетмейкеры часто снимают лимитную ликвидность (микростенки) — market orders «проходят» через уровни, где стояли лимитки. Это называется sweep / liquidity takeout. Часто после такого снятия ликвидности цена коротко возвращается к зоне, где осталась или воссоздалась глубина (reclaim), потому что:

- прошёл «вынос» стопов/лимиток, сработал наклон рынка, и затем маркет-мейкеры/крупные лимитчики вставляют обратную ликвидность;

- межвременные алгоритмы заполняют книгу обратно, и рынок «отскакивает».

Мы детектируем фактическое исполнение (реально прошедший объём по ценам) и сравниваем с тем, сколько «стояло» в стакане (resting depth).

Когда исполненный объём на уровне (или пройдя уровень) существенно превышает ранее видимую глубину → это sweep.

После sweep ждём восстановления глубины с противоположной стороны и входа обратно в сторону реклайма (reclaim).

Позиции короткие (интрадей), стопы близкие — хорошее RR.

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

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

📌Execution Cost Surface (ECS) - (оценщик стоимости исполнения — моделирует проскальзывание / VWAP / market-impact для разных объёмов и бирж) 🧑‍💻

Что делает: 🛠

Для заданного списка бирж и инструмента (например, BTC/USDT) загружает стаканы (order books).

Для каждого объёма заказа (серии размеров) симулирует исполнение рыночного ордера — «съедает» стакан и считает VWAP, абсолютный и относительный слиппаж.

Опционально строит простую оценку «square-root impact» на основе объёма/волатильности (если доступны OHLCV/volume).

Выводит таблицу/CSV с оценками — удобно выбирать: где и какой объём безопасно исполнять.

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

- Помогает решить: дробить ли ордер, переключиться на другую биржу, или дождаться лучшего времени.

- Работает прямо с реальными стаканами — не устаревшие индикаторы.

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

#инструмент

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

📌Microstructure Transition Predictor (MTP) - (предсказатель микроструктурных переходов — краткосрочная вероятность движения на базе переходов состояния стакана+ленты) 🧑‍💻

Что это и зачем: 🚀

MTP — компактный Bayesian/Markov-подход, который моделирует вероятности краткосрочных переходов цены (↑/↓/flat) исходя из состояний книги ордеров и параметров потока сделок.
Идея: рынок делает не случайные шага — есть повторяющиеся «микросценарии» (слабая книга справа + тиковые покупки → большой шанс подъёма в ближайшие N тиков). Мы строим маленькую модель переходов состояний (transition matrix) и выдаём вероятность движения за следующий тик/бар + рекомендованный «скоростной» размер позиции.

Полезно как:💸

- фильтр для скальп-исполнения (входить только при P(up)>X),

- ординатор для алгоритмов агрессивного входа (когда краткосрочная вероятность > порога),

- компонент ensemble — признак в ML.

Как пользоваться: 🧨

1. Запустите прототип, дайте ему построить transition-матрицу на истории (~несколько минут).

2. В реальном времени инструмент будет печатать P_up/P_down и давать простые сигналы (LONG/SHORT) при высокой краткосрочной вероятности.

3. Комбинируйте с управлением риском: размер позиции ∝ (P - 0.5) / volatility, стоп = ATR или ближайший уровень книги.

#инструмент

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

📌"Liquidity Gradient Ride" - (езда по «градиенту ликвидности») 🧑‍💻

🔍 Идея: 🧨

Цена стремится туда, где меньше ликвидности в стакане, потому что туда проще двигать рынок.
👉 Если на стороне ask (продаж) ликвидности меньше, чем на стороне bid (покупок) → проще толкнуть цену вверх, и наоборот.

Мы анализируем градиент ликвидности: сравниваем плотности объёмов на разных уровнях стакана и входим в сторону, где «дыра».

⚙ Логика: 🛠️

1. Подгружаем стакан (order book).

2. Считаем кумулятивный объём на N уровнях по bid и ask.

3. Если ask_объём ≪ bid_объём → сигнал в лонг.
Если bid_объём ≪ ask_объём → сигнал в шорт.

4. Стоп за ближайший противоположный крупный уровень ликвидности.

5. Тейк по 1–2× величине текущего спреда/волатильности.

import asyncio
import json
import websockets
import numpy as np

SYMBOL = "btcusdt"
DEPTH = 20  # уровни стакана для анализа
THRESHOLD = 0.6  # если одна сторона <60% другой, даём сигнал

async def liquidity_gradient():
    url = f"wss://stream.binance.com:9443/ws/{SYMBOL}@depth@100ms"
    async with websockets.connect(url) as ws:
        print(f"📡 Liquidity Gradient Ride для {SYMBOL.upper()} запущен")

        async for msg in ws:
            data = json.loads(msg)
            bids = [(float(p), float(q)) for p, q in data["bids"][:DEPTH]]
            asks = [(float(p), float(q)) for p, q in data["asks"][:DEPTH]]

            bid_vol = sum(q for _, q in bids)
            ask_vol = sum(q for _, q in asks)

            if ask_vol < bid_vol * THRESHOLD:
                print(f"🟢 Лонг-сигнал: bid={bid_vol:.2f}, ask={ask_vol:.2f}")
            elif bid_vol < ask_vol * THRESHOLD:
                print(f"🔴 Шорт-сигнал: bid={bid_vol:.2f}, ask={ask_vol:.2f}")

if __name__ == "__main__":
    asyncio.run(liquidity_gradient())

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

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

📌Adaptive Slicer & Impact Estimator (ASIE) - (адаптивный планировщик исполнения + оценщик стоимости — умный TWAP/TWAP-слайсер с учётом глубины и текущей ликвидности) 🧑‍💻

Что делает (кратко):🛠

- Берёт целевой объём (в BTC/ETH и т.д.) и тайм-окно исполнения (например, 60 минут).

- Регулярно снимает snapshot стакана (order book) и поток сделок (опционально).

- Для каждого среза времени рассчитывает рекомендуемый объём так, чтобы:

- не «съедать» слишком глубоко книгу (лимит по допустимому slippage на срез),

- при слабой ликвидности дробить сильнее, при высокой — собирать быстрее.

- Симулирует исполнение каждого среза по текущему стакану и считает ожидаемый VWAP / %slippage / потребность в дроблении.

- Печатает план с расписанием срезов и оценкой стоимости; можно экспортировать CSV.

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

- Выполнять крупные ордера с минимальным ущербом капиталу.

- Автоматически решать: дробить дальше или ускорить выполнение, ориентируясь на живую ликвидность.

- Может служить модулем перед real-order execution (решает: market vs limit, размер лимитов и частоту).

#инструмент

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

📌"Exhaustion Burst" - (взрыв после истощения) 🧑‍💻

🔍 Идея: 🧨

Цена двигается рывками, потому что толпа агрессивно жмёт в одну сторону, пока не кончается «топливо».

👉 После серии подряд однонаправленных свечей часто происходит резкий обратный импульс — маркетмейкеры и крупные игроки пользуются истощением толпы.

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

1. Считаем количество подряд идущих свечей одного цвета (например, бычьи).

2. Если таких свечей ≥ N и при этом средний размер тела свечей уменьшается → признак истощения.

3. Входим в противоположную сторону на закрытии последней свечи.

4. Стоп за экстремум серии, тейк = 1.5–2× стопа.

import ccxt
import pandas as pd

SYMBOL = "BTC/USDT"
TF = "5m"
N = 5  # количество свечей в серии

ex = ccxt.binance()

def fetch_ohlcv(symbol, tf="5m", limit=200):
    df = ex.fetch_ohlcv(symbol, tf, limit=limit)
    return pd.DataFrame(df, columns=["ts","open","high","low","close","vol"])

def check_exhaustion(df):
    closes = df["close"].values
    opens = df["open"].values

    streak = 1
    direction = None
    bodies = []

    for i in range(1, len(closes)):
        curr_dir = "up" if closes[i] > opens[i] else "down"
        body = abs(closes[i] - opens[i])

        if direction is None:
            direction = curr_dir

        if curr_dir == direction:
            streak += 1
            bodies.append(body)
        else:
            if streak >= N and len(bodies) >= 2:
                if bodies[-1] < bodies[0] * 0.5:  # тела уменьшаются
                    signal = "SELL" if direction == "up" else "BUY"
                    print(f"🚨 Exhaustion Burst: {signal} на свече {df.iloc[i].ts}")
            direction = curr_dir
            streak = 1
            bodies = [body]

if __name__ == "__main__":
    data = fetch_ohlcv(SYMBOL, TF, limit=200)
    check_exhaustion(data)

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

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

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

🔍 Идея: 🚀

Крупные игроки часто оставляют «следы» — большие лимитные заявки в стакане.
Но они могут:

стоять и держать уровень (поддержка/сопротивление),

сниматься перед пробоем,

или двигаться вместе с ценой, как «магнит».

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

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

Находит самые крупные заявки в стакане (топ-N по объёму).

Смотрит, «живут» ли они долго или исчезают перед движением.

Помечает их как:

🟢 Поддержка (сильный bid-кластер)

🔴 Сопротивление (сильный ask-кластер)

⚡ Ловушка (кластер исчез перед импульсом)

import asyncio
import json
import websockets

SYMBOL = "btcusdt"
DEPTH_LEVELS = 50
BIG_ORDER_THRESHOLD = 50  # условный порог крупной заявки (в BTC)

async def track_liquidity():
    url = f"wss://stream.binance.com:9443/ws/{SYMBOL}@depth{DEPTH_LEVELS}@100ms"
    print(f"📡 Liquidity Footprint Tracker для {SYMBOL.upper()} запущен")

    async with websockets.connect(url) as ws:
        async for msg in ws:
            data = json.loads(msg)
            bids = [(float(p), float(q)) for p, q in data["b"]]
            asks = [(float(p), float(q)) for p, q in data["a"]]

            big_bids = [(p, q) for p, q in bids if q > BIG_ORDER_THRESHOLD]
            big_asks = [(p, q) for p, q in asks if q > BIG_ORDER_THRESHOLD]

            if big_bids:
                print(f"🟢 Поддержка: {big_bids}")
            if big_asks:
                print(f"🔴 Сопротивление: {big_asks}")

if __name__ == "__main__":
    asyncio.run(track_liquidity())

#инструмент

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

📌"Импульс + затухание ликвидности" 🧑‍💻

Логика: 🛠

1. В нормальном рынке при росте объёма сделок ликвидность в стакане должна увеличиваться (новые лимитные ордера).

2. Но бывают моменты, когда идёт импульс сделок в одну сторону, а в стакане ликвидность начинает исчезать:

например, агрессивно покупают, а лимитные продавцы уходят, и книга сверху пустеет.

или наоборот, агрессивно продают, а лимитные покупатели снимают заявки.

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

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

Лонг:

поток сделок показывает >70% агрессивных покупок за последние N секунд,

при этом в верхних 1–2% стакана ликвидность снижается ≥30% по сравнению со средним.


Шорт:

поток сделок показывает >70% агрессивных продаж,

в нижних 1–2% стакана ликвидность снижается ≥30%.

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

1. Частичная фиксация прибыли, когда ликвидность возвращается к норме.

2. Полный выход, если:

объём агрессивных сделок падает ниже 50%,

или появляется крупный лимитный игрок, перекрывающий движение.

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

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

📌Microstate Mirror - (зеркало микросостояний стакана — детектор повторяющихся «микростанций» книги, которые исторически ведут к предсказуемому движению)🧑‍💻

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

Мы превращаем каждый снимок стакана (Top-N уровней) в компактный вектор — микросостояние. Храним недавнюю историю таких микросостояний вместе с тем, что случилось с ценой через фиксированный горизонт (например, +5s или +1 бар).
Когда текущий микроснимок сильно похож (по косинусной/евклидовой близости) на набор исторически «предиктивных» микростанций, инструмент агрегирует исходы и выдаёт вероятностный прогноз — высока ли вероятность движения вверх/вниз и с каким ожиданием.
Это — гибрид orderbook-анализ + case-based reasoning: не индикаторы, а повторение реального поведения рынка.

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

Детектит редко используемый признак: повторимость структур стакана, а не только поток сделок или свечи.

Работает в реальном времени: полезен как фильтр входа/разделения объёма.

Универсален: для скальпинга (секунды), интрадей (минуты) и для нескольких инструментов.

#инструмент

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

📌"Liquidity Fade" - (игра против ложных уровней ликвидности) 🧑‍💻

🔍 Идея: 🚀

Крупные заявки в стакане часто используются для обмана толпы:

- ставят огромный ордер (например, на покупку), трейдеры думают «поддержка», заходят в лонг,

- ордер снимают — и рынок летит вниз.

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

📌 Логика стратегии: 🧨

1. Мониторим стакан и ищем крупные заявки (по объёму).

2. Если такой кластер исчезает перед движением цены:

если это была заявка на покупку → открываем шорт;

если это была заявка на продажу → открываем лонг.

3. Стоп ставим за «фейковый уровень».

4. Тейк фиксируем на ближайшем реальном кластере ликвидности.

import asyncio
import json
import websockets
import time

SYMBOL = "btcusdt"
DEPTH_LEVELS = 50
BIG_ORDER_THRESHOLD = 100  # порог крупного ордера
ORDER_MEMORY = {}

async def liquidity_fade():
    url = f"wss://stream.binance.com:9443/ws/{SYMBOL}@depth{DEPTH_LEVELS}@100ms"
    print(f"🚀 Liquidity Fade strategy запущена для {SYMBOL.upper()}")

    async with websockets.connect(url) as ws:
        async for msg in ws:
            data = json.loads(msg)
            bids = [(float(p), float(q)) for p, q in data["b"]]
            asks = [(float(p), float(q)) for p, q in data["a"]]

            # проверяем крупные заявки
            for p, q in bids + asks:
                if q > BIG_ORDER_THRESHOLD:
                    ORDER_MEMORY[p] = time.time()

            # проверяем исчезновение
            current_prices = {p for p, q in bids + asks}
            vanished = [p for p in list(ORDER_MEMORY.keys()) if p not in current_prices]

            for p in vanished:
                side = "buy" if p < bids[0][0] else "sell"
                age = time.time() - ORDER_MEMORY[p]
                if age < 2:  # если исчез недавно
                    if side == "buy":
                        print(f"⚠️ Фейковая поддержка исчезла @ {p} → сигнал на ШОРТ")
                    else:
                        print(f"⚠️ Фейковое сопротивление исчезло @ {p} → сигнал на ЛОНГ")
                del ORDER_MEMORY[p]

if __name__ == "__main__":
    asyncio.run(liquidity_fade())

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

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

📌"Volatility Cone Tracker" - (реальный конус волатильности в онлайне) 🧑‍💻

🔍 Идея: 🧨

Трейдеру важно понимать: текущая волатильность «нормальная» или рынок ведёт себя экстремально.

👉 Инструмент сравнивает реализованную волатильность на разных таймфреймах с её историческими средними и строит «конус»: диапазон ожидаемых значений.

Если текущая вола выходит за границы конуса → сигнал о необычном режиме рынка (повышенный риск или шанс на сильный тренд).

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

Подгружает свечи (например, 1m, 5m, 15m).

Считает реал. волатильность (стд. отклонение лог-доходностей).

Сравнивает с историческим распределением.

В консоли выводит:

🟢 Нормально

🟡 У верхней границы

🔴 Аномальная волатильность

import ccxt
import numpy as np
import time

SYMBOL = "BTC/USDT"
EXCHANGE = ccxt.binance()
LOOKBACK = 500  # кол-во свечей
TF = "1m"

def realized_volatility(returns):
    return np.sqrt(np.mean(returns**2))

def volatility_cone():
    ohlcv = EXCHANGE.fetch_ohlcv(SYMBOL, timeframe=TF, limit=LOOKBACK)
    closes = np.array([c[4] for c in ohlcv])
    log_ret = np.diff(np.log(closes))

    window = 50
    vols = [realized_volatility(log_ret[i:i+window]) for i in range(len(log_ret)-window)]
    current_vol = realized_volatility(log_ret[-window:])

    mean_vol = np.mean(vols)
    std_vol = np.std(vols)

    low, high = mean_vol - 2*std_vol, mean_vol + 2*std_vol

    print(f"Текущая волатильность: {current_vol:.5f}")
    if current_vol < low:
        print("🟢 Рынок спокойный (ниже конуса)")
    elif current_vol > high:
        print("🔴 Аномально высокая волатильность!")
    else:
        print("🟡 В пределах нормы")

if __name__ == "__main__":
    while True:
        volatility_cone()
        time.sleep(60)

#инструмент

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

📌"Auction Imbalance Play" - (игра на дисбалансе аукциона в момент формирования цены) 🧑‍💻

🔍 Идея: 🧨

Рынок постоянно находится в состоянии «аукциона»: одни хотят купить, другие продать.
Когда дисбаланс между агрессивными покупателями (market buy) и продавцами (market sell) становится слишком большим, цена должна «сдвинуться», чтобы уравновесить спрос и предложение.

👉 Мы отслеживаем поток сделок (tape) и сравниваем:

сколько объёма прошло в сторону покупки,

сколько в сторону продажи.

Если разница > порога → ожидаем движение в сторону победивших.

📌 Логика: 🛠️

1. Подключаемся к потоку сделок (Binance trades stream).

2. В скользящем окне (например, 10 секунд) считаем суммарный buy volume и sell volume.

3. Если buy ≫ sell → открываем лонг.
Если sell ≫ buy → открываем шорт.

4. Закрываем сделку при возврате баланса или через фиксированный тайм-аут.

import asyncio
import websockets
import json
from collections import deque
import time

SYMBOL = "btcusdt"
WINDOW_SEC = 10
IMBALANCE_THRESHOLD = 2.0  # во сколько раз buy > sell или наоборот

trades_window = deque()

async def auction_imbalance():
    url = f"wss://stream.binance.com:9443/ws/{SYMBOL}@trade"
    print(f"📡 Auction Imbalance strategy запущена для {SYMBOL.upper()}")

    async with websockets.connect(url) as ws:
        async for msg in ws:
            data = json.loads(msg)
            price = float(data["p"])
            qty = float(data["q"])
            is_buyer_maker = data["m"]  # True = sell, False = buy
            ts = time.time()

            trades_window.append((ts, qty, is_buyer_maker))

            # чистим старые сделки
            while trades_window and ts - trades_window[0][0] > WINDOW_SEC:
                trades_window.popleft()

            # считаем баланс
            buy_vol = sum(q for t, q, m in trades_window if not m)
            sell_vol = sum(q for t, q, m in trades_window if m)

            if sell_vol > 0 and buy_vol / sell_vol > IMBALANCE_THRESHOLD:
                print(f"🚀 ЛОНГ сигнал @ {price} | Buy/Sell = {buy_vol:.2f}/{sell_vol:.2f}")
            elif buy_vol > 0 and sell_vol / buy_vol > IMBALANCE_THRESHOLD:
                print(f"🔻 ШОРТ сигнал @ {price} | Buy/Sell = {buy_vol:.2f}/{sell_vol:.2f}")

if __name__ == "__main__":
    asyncio.run(auction_imbalance())

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

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

📌"Time Stretch Reversal" - (разворот по растяжению времени) 🧑‍💻

🔍 Идея: 🧨

Рынок движется не только по цене, но и во времени.

Когда импульс длится дольше обычного, но не даёт сильного результата по цене → это признак выдыхания движения.

Когда цена быстро достигает экстремума за необычно короткое время → это сигнал о возможном продолжении (инерция толпы).

👉 Мы анализируем время между локальными максимумами и минимумами.

📌 Правила: 🛠️

1. Выделяем swing-high и swing-low (например, через экстремум за N свечей).

2. Считаем:

ΔPrice = изменение цены,

ΔTime = кол-во баров между экстремумами.

3. Если ΔTime > среднее, но ΔPrice маленькое → рынок слабый, входим в контртренд.

4. Если ΔTime очень короткое, но ΔPrice большое → вероятно, движение продолжится → вход по тренду.

import ccxt
import pandas as pd
import numpy as np

SYMBOL = "BTC/USDT"
TF = "5m"
LOOKBACK = 500

ex = ccxt.binance()

def fetch_data():
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","vol"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    return df

def find_swings(df, n=3):
    swings = []
    for i in range(n, len(df)-n):
        high = df["high"].iloc[i]
        low = df["low"].iloc[i]
        if high == max(df["high"].iloc[i-n:i+n+1]):
            swings.append((df["dt"].iloc[i], df["close"].iloc[i], "H"))
        if low == min(df["low"].iloc[i-n:i+n+1]):
            swings.append((df["dt"].iloc[i], df["close"].iloc[i], "L"))
    return swings

def analyze(swings):
    signals = []
    for i in range(2, len(swings)):
        t1, p1, _ = swings[i-2]
        t2, p2, _ = swings[i-1]
        t3, p3, _ = swings[i]

        dprice = abs(p3 - p2) / p2
        dtime = (t3 - t2).total_seconds() / 60

        avg_time = np.mean([(swings[j][0]-swings[j-1][0]).total_seconds()/60 for j in range(1,i)])

        if dtime > 1.5*avg_time and dprice < 0.003:
            signals.append((t3, p3, "REVERSAL"))
        elif dtime < 0.5*avg_time and dprice > 0.01:
            signals.append((t3, p3, "MOMENTUM"))
    return signals

if __name__ == "__main__":
    df = fetch_data()
    swings = find_swings(df)
    sigs = analyze(swings)
    for s in sigs[-10:]:
        print(s)

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

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

📌"Market Memory Index" - (индекс рыночной памяти) 🧑‍💻

🔍 Идея: 🧨

Цена не всегда «помнит» прошлое одинаково.

В трендовых режимах каждый следующий бар коррелирует с предыдущими (есть инерция).

Во флэте рынок «забывает» направление (корреляция быстро уходит в ноль).

👉 Инструмент измеряет автокорреляцию доходностей на разных лагах и строит индекс «рыночной памяти».

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

Берёт последние N баров.

Считает лаговую корреляцию (например, 1–10 шагов).

Выводит индекс 0…1:

Близко к 1 → рынок «помнит» (инерционный режим → тренд).

Близко к 0 → рынок забывчивый (хаос → флэт).

import ccxt
import pandas as pd
import numpy as np

SYMBOL = "BTC/USDT"
TF = "5m"
LOOKBACK = 300

ex = ccxt.binance()

def fetch_data():
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=["ts","o","h","l","c","v"])
    df["ret"] = np.log(df["c"]).diff()
    return df.dropna()

def market_memory_index(df, max_lag=10):
    rets = df["ret"].values
    corrs = []
    for lag in range(1, max_lag+1):
        x = rets[:-lag]
        y = rets[lag:]
        corr = np.corrcoef(x, y)[0,1]
        corrs.append(corr)
    return np.mean([abs(c) for c in corrs])

if __name__ == "__main__":
    df = fetch_data()
    mmi = market_memory_index(df)
    if mmi > 0.5:
        print(f"🟢 Рынок трендовый (MMI={mmi:.2f})")
    elif mmi < 0.2:
        print(f"🔴 Рынок хаотичный (MMI={mmi:.2f})")
    else:
        print(f"🟡 Нейтральный рынок (MMI={mmi:.2f})")

#инструмент

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

📌"Импульсные ловушки"🧑‍💻

🔍 Идея: 🧨

Многие трейдеры «прыгают в рынок» после резкого движения (импульса). Но часто такие импульсы быстро выдыхаются и цена возвращается назад.
Мы используем это против толпы:

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

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

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

1. Берём короткий таймфрейм (1–5 мин).

2. Считаем относительное изменение цены за последние k свечей.

3. Если движение > X% (импульс), то ждём подтверждения:

если цена не обновляет хай/лоу импульса в течение N свечей → это «ловушка».

4. Входим против импульса.

5. Тейк-профит = середина диапазона импульса, стоп = за его экстремумом.

import ccxt
import pandas as pd
import numpy as np

SYMBOL = "BTC/USDT"
TF = "1m"
LOOKBACK = 200

ex = ccxt.binance()

def fetch_data():
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=["ts","o","h","l","c","v"])
    return df

def impulse_trap(df, k=5, threshold=0.004, hold=3):
    signals = []
    for i in range(k, len(df)-hold):
        change = (df["c"].iloc[i] - df["c"].iloc[i-k]) / df["c"].iloc[i-k]
        if abs(change) > threshold:
            if change > 0:  # импульс вверх
                high = df["h"].iloc[i]
                if all(df["h"].iloc[i+1:i+1+hold] < high):
                    signals.append((df["ts"].iloc[i], "SELL", df["c"].iloc[i]))
            else:  # импульс вниз
                low = df["l"].iloc[i]
                if all(df["l"].iloc[i+1:i+1+hold] > low):
                    signals.append((df["ts"].iloc[i], "BUY", df["c"].iloc[i]))
    return signals

if __name__ == "__main__":
    df = fetch_data()
    traps = impulse_trap(df)
    for t, side, price in traps:
        print(f"{pd.to_datetime(t, unit='ms')} → {side} @ {price}")

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

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

📌"Market Microstructure Lag Detector" - (детектор микроструктурного лага) 🧑‍💻

🔍 Идея: 🚀

На биржах иногда возникает лаг между движением цены и реакцией объёма.

В норме: рост цены сопровождается ростом объёмов.

В аномальных ситуациях: цена движется, но объём запаздывает (или наоборот).

👉 Этот инструмент измеряет сдвиг во времени между изменением цены и объёмом → показывает, есть ли «запаздывание реакции рынка».

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

1. Берёт последние N свечей.

2. Строит кросс-корреляцию между:

изменением цены,

изменением объёма.

3. Находит лаг (в свечах), где корреляция максимальна.

4. Выводит результат:

Лаг = 0 → рынок синхронен.

Лаг > 0 → объём опаздывает за ценой.

Лаг < 0 → объём идёт впереди (предупреждает движение).

Польза:💸

- Можно выявлять режимы, когда объём становится опережающим сигналом.

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

- Отличный фильтр для стратегий на пробои и импульсы.

import ccxt
import pandas as pd
import numpy as np

SYMBOL = "BTC/USDT"
TF = "1m"
LOOKBACK = 500

ex = ccxt.binance()

def fetch_data():
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=["ts","o","h","l","c","v"])
    df["ret"] = df["c"].pct_change().fillna(0)
    df["vol_chg"] = df["v"].pct_change().fillna(0)
    return df

def micro_lag(df, max_lag=10):
    x = df["ret"].values
    y = df["vol_chg"].values
    lags = range(-max_lag, max_lag+1)
    corrs = []
    for lag in lags:
        if lag < 0:
            corr = np.corrcoef(x[:lag], y[-lag:])[0,1]
        elif lag > 0:
            corr = np.corrcoef(x[lag:], y[:-lag])[0,1]
        else:
            corr = np.corrcoef(x, y)[0,1]
        corrs.append(corr)
    best_idx = int(np.argmax(np.abs(corrs)))
    return lags, corrs, lags[best_idx], corrs[best_idx]

if __name__ == "__main__":
    df = fetch_data()
    lags, corrs, best_lag, best_corr = micro_lag(df)
    print(f"📊 Лучший лаг: {best_lag} свечей | Корреляция={best_corr:.2f}")
    if best_lag > 0:
        print("⚡ Объём запаздывает за ценой")
    elif best_lag < 0:
        print("🔮 Объём предсказывает цену")
    else:
        print("⚖ Цена и объём синхронны")

#инструмент

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

📌"Noise-to-Signal Ratio" - (коэффициент шума к сигналу) 🧑‍💻

🔍 Идея: 🚀

Не всегда рынок двигается «осмысленно». Иногда цена просто «шумит» в узком диапазоне, и любые сигналы стратегий становятся ложными.

👉 Этот инструмент измеряет отношение шумовых колебаний к направленному движению.

Если шум >> тренд → лучше не торговать (рынок «грязный»).

Если тренд >> шум → высокая вероятность эффективных сделок.

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

1. Берёт окно последних N свечей.

2. Считает:

Общее движение;

Сумму всех колебаний;

3. Рассчитывает коэффициент NSR

NSR ≈ 1 → движение чистое (трендовое).

NSR >> 1 → рынок шумный, нет смысла торговать пробои.

import ccxt
import pandas as pd
import numpy as np

SYMBOL = "BTC/USDT"
TF = "5m"
LOOKBACK = 200

ex = ccxt.binance()

def fetch_data():
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=["ts","o","h","l","c","v"])
    return df

def noise_signal_ratio(df, window=50):
    closes = df["c"].iloc[-window:].values
    total_move = abs(closes[-1] - closes[0])
    swings = np.sum(abs(np.diff(closes)))
    if total_move == 0:
        return float("inf")
    return swings / total_move

if __name__ == "__main__":
    df = fetch_data()
    nsr = noise_signal_ratio(df)
    print(f"📊 Noise-to-Signal Ratio: {nsr:.2f}")
    if nsr < 2:
        print("🔥 Чистый трендовый рынок — пробойные стратегии в приоритете")
    elif nsr < 5:
        print("⚖ Смешанный режим — подойдёт скальпинг")
    else:
        print("🌪 Шумный рынок — лучше не торговать пробои")

#инструмент

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

📌"Импульсные ритмы" - (торговля по ритмике микродвижений) 🧑‍💻

🔍 Идея: 🧨

Рынок двигается «рывками» — серии маленьких импульсов в одном направлении часто означают, что алгоритмы двигают цену.

Если появляется серия импульсов одной стороны (например, несколько свечей подряд с телом > X%), это значит, что активные участники толкают рынок.

Если после серии идёт затухание (свечи становятся меньше) → сигнал к развороту.

Если серия усиливается → продолжаем в сторону движения.

📌 Логика входа: 💸

1. Находим последовательности из k свечей с телами больше threshold.

2. Если каждая следующая свеча меньше предыдущей → ждём разворот.

3. Если каждая следующая больше → входим в сторону импульса.

4. Выход: по противоположному импульсу или при смене «ритма».

import ccxt
import pandas as pd

SYMBOL = "BTC/USDT"
TF = "1m"
LOOKBACK = 200

ex = ccxt.binance()

def fetch_data():
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=["ts","o","h","l","c","v"])
    df["body"] = abs(df["c"] - df["o"])
    df["dir"] = df["c"] > df["o"]
    return df

def impulse_rhythm(df, k=3, threshold=0.001):
    signals = []
    for i in range(k, len(df)):
        seq = df.iloc[i-k+1:i+1]
        if all(seq["body"] > seq["c"].iloc[-1] * threshold):
            if seq["dir"].all():  # все вверх
                if seq["body"].is_monotonic_increasing:
                    signals.append((df["ts"].iloc[i], "BUY", df["c"].iloc[i]))
                elif seq["body"].is_monotonic_decreasing:
                    signals.append((df["ts"].iloc[i], "SELL", df["c"].iloc[i]))
            elif (~seq["dir"]).all():  # все вниз
                if seq["body"].is_monotonic_increasing:
                    signals.append((df["ts"].iloc[i], "SELL", df["c"].iloc[i]))
                elif seq["body"].is_monotonic_decreasing:
                    signals.append((df["ts"].iloc[i], "BUY", df["c"].iloc[i]))
    return signals

if __name__ == "__main__":
    df = fetch_data()
    sigs = impulse_rhythm(df)
    for t, side, price in sigs:
        print(f"{pd.to_datetime(t, unit='ms')} → {side} @ {price}")

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

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

📌"Ценовой маятник" 🧑‍💻

🔍 Идея: 🚀

Цена на коротких интервалах ведёт себя как маятник:

резкий рывок → накопление «кинетической энергии»,

затем возврат к равновесию.

Мы ловим эти отклонения от равновесной точки и зарабатываем на возврате.

📌 Логика входа: 💸

1. Считаем «центр равновесия» цены = скользящее среднее по N свечам.

2. Измеряем отклонение от центра в процентах.

3. Если отклонение > X%, открываем позицию в обратную сторону (ждём возврата маятника).

4. Выход — возврат цены к центру (тейк), стоп — за пределами расширенного диапазона.

import ccxt
import pandas as pd
import numpy as np

SYMBOL = "BTC/USDT"
TF = "5m"
LOOKBACK = 300

ex = ccxt.binance()

def fetch_data():
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=["ts","o","h","l","c","v"])
    return df

def pendulum_strategy(df, n=50, threshold=0.02):
    df["center"] = df["c"].rolling(n).mean()
    df = df.dropna()
    signals = []
    for i in range(len(df)):
        price = df["c"].iloc[i]
        center = df["center"].iloc[i]
        deviation = (price - center) / center
        if deviation > threshold:
            signals.append((df["ts"].iloc[i], "SELL", price))
        elif deviation < -threshold:
            signals.append((df["ts"].iloc[i], "BUY", price))
    return signals

if __name__ == "__main__":
    df = fetch_data()
    sigs = pendulum_strategy(df)
    for t, side, price in sigs[-5:]:
        print(f"{pd.to_datetime(t, unit='ms')} → {side} @ {price}")

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

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

📌"Радар крупных ордеров" 🧑‍💻

📌 Идея: 🧨

На биржах часто выставляют айсберг-ордера или «стены» в стакане, которые могут:

- удерживать цену,

- создавать иллюзию спроса/предложения,

- сигнализировать о возможном развороте.


👉 Этот инструмент в реальном времени сканирует ордербук и подсвечивает аномально крупные заявки.

import ccxt
import time

SYMBOL = "BTC/USDT"
DEPTH = 20
THRESHOLD = 50  # минимальный размер ордера в BTC для "аномалии"

ex = ccxt.binance()

def radar():
    orderbook = ex.fetch_order_book(SYMBOL, limit=DEPTH)
    bids = orderbook["bids"]
    asks = orderbook["asks"]

    big_bids = [(p, v) for p, v in bids if v > THRESHOLD]
    big_asks = [(p, v) for p, v in asks if v > THRESHOLD]

    if big_bids:
        print("🟢 Крупные покупки:")
        for p, v in big_bids:
            print(f"   {v:.2f} BTC @ {p}")

    if big_asks:
        print("🔴 Крупные продажи:")
        for p, v in big_asks:
            print(f"   {v:.2f} BTC @ {p}")

if __name__ == "__main__":
    while True:
        radar()
        time.sleep(3)

#инструмент

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