📌Fractal Pattern Scanner - (сканер самоподобных паттернов) 🧑‍💻

🔍 Идея: 🚀

Рынок часто повторяет фрактальные структуры: маленькие паттерны свечей копируют крупные (например, мини-треугольник внутри большого).
👉 Инструмент ищет на истории повторяющиеся самоподобные комбинации свечей и отмечает, когда они возникают снова в реальном времени.

📌 Польза : 💸

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

- Дает ранние сигналы о том, что рынок повторяет старый сценарий.

- Универсален: работает и на минутках, и на дневках.

import ccxt
import pandas as pd
import numpy as np
from scipy.spatial.distance import euclidean

ex = ccxt.binance()
symbol = "BTC/USDT"

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

def pattern_vector(df, i, n=5):
    """создаём вектор паттерна из n свечей"""
    window = df.iloc[i:i+n]
    return (window["close"].values - window["open"].values) / window["open"].values

def find_fractals(df, n=5, threshold=0.1):
    patterns = []
    for i in range(len(df)-n):
        v1 = pattern_vector(df, i, n)
        for j in range(i+n, len(df)-n):
            v2 = pattern_vector(df, j, n)
            if euclidean(v1, v2) < threshold:
                patterns.append((i, j))
    return patterns

df = fetch_ohlcv()
fractals = find_fractals(df)

print("🔎 Найдено фрактальных повторов:", len(fractals))

#инструмент

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

📌Order Imbalance Sniper - (снайпер дисбаланса ордербука) 🧑‍💻

🔍 Идея: 🧨

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

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

⚙ Логика: 🛠️

1. Загружаем ордербук (например, 20 уровней).

2. Считаем сумму заявок на buy и sell.

3. Если buy_volume / sell_volume > K → идём в лонг.

4. Если наоборот sell_volume / buy_volume > K → идём в шорт.

5. Стоп и тейк короткие (это стратегия скальпинга).

import ccxt
import time

ex = ccxt.binance({"enableRateLimit": True})
symbol = "BTC/USDT"
threshold = 2.0  # во сколько раз должен быть дисбаланс

def check_imbalance():
    orderbook = ex.fetch_order_book(symbol, limit=20)
    bids = sum([b[1] for b in orderbook["bids"]])
    asks = sum([a[1] for a in orderbook["asks"]])

    if bids / asks > threshold:
        return "LONG", bids/asks
    elif asks / bids > threshold:
        return "SHORT", asks/bids
    else:
        return None, 1.0

while True:
    signal, ratio = check_imbalance()
    if signal:
        print(f"📊 Сигнал: {signal} | Дисбаланс = {ratio:.2f}")
    time.sleep(3)

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

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

📌"Volatility Pocket Breakout" - (прорыв кармана волатильности) 🧑‍💻

🔍 Идея: 🧨

Цена часто застревает в карманах низкой волатильности (узкие консолидации).
После них почти всегда происходит резкий выход — импульс.

👉 Стратегия ищет такие карманы (низкая ATR и сжатие Боллинджера) и заходит в сторону пробоя.

⚙ Логика: 🛠️

1. Считаем ATR (волатильность) и ширину Bollinger Bands.

2. Если и ATR, и BB ниже среднего уровня → рынок в кармане.

3. Ждём пробоя вверх/вниз (закрытие свечи за границей BB).

4. Входим в сторону пробоя, стоп — за противоположной границей, тейк = 2× стоп

import ccxt
import pandas as pd
import numpy as np

ex = ccxt.binance()
symbol = "BTC/USDT"

def fetch_data(limit=200):
    ohlcv = ex.fetch_ohlcv(symbol, "1h", limit=limit)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","vol"])
    df["ret"] = df["close"].pct_change()
    return df

def bollinger_bands(series, window=20, mult=2):
    ma = series.rolling(window).mean()
    std = series.rolling(window).std()
    upper = ma + mult * std
    lower = ma - mult * std
    return upper, lower

def atr(df, window=14):
    high_low = df["high"] - df["low"]
    high_close = np.abs(df["high"] - df["close"].shift())
    low_close = np.abs(df["low"] - df["close"].shift())
    tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
    return tr.rolling(window).mean()

df = fetch_data()
df["ATR"] = atr(df)
df["BB_upper"], df["BB_lower"] = bollinger_bands(df["close"])

# сигналы
df["in_pocket"] = (df["ATR"] < df["ATR"].rolling(100).mean()*0.7) & \
                  ((df["BB_upper"]-df["BB_lower"]) < (df["BB_upper"]-df["BB_lower"]).rolling(100).mean()*0.7)

df["long_signal"] = (df["in_pocket"]) & (df["close"] > df["BB_upper"])
df["short_signal"] = (df["in_pocket"]) & (df["close"] < df["BB_lower"])

print(df[["ts","close","long_signal","short_signal"]].tail(20))

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

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

📌Market Symmetry Detector - (детектор рыночной симметрии) 🧑‍💻

🔍 Идея: 🚀

Рынок часто строит симметричные движения:

рост = падение по времени и амплитуде,

зеркальные паттерны на графике (левое «плечо» ≈ правому),

циклы повторяются через одинаковые интервалы.

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

📌 Польза: 💸

- Можно заранее ожидать «отзеркаливание» цены.

- Помогает в определении точек разворота.

- Работает как фильтр для стратегий на паттернах (H&S, W, M).

import ccxt
import pandas as pd
import numpy as np

ex = ccxt.binance()
symbol = "BTC/USDT"

def fetch_data(limit=300):
    ohlcv = ex.fetch_ohlcv(symbol, "1h", limit=limit)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","vol"])
    return df

def symmetry_score(segment1, segment2):
    """сравнение двух отрезков графика (по форме движения)"""
    s1 = (segment1 - segment1[0]) / segment1[0]
    s2 = (segment2 - segment2[-1]) / segment2[-1]  # реверс
    return np.corrcoef(s1, np.flip(s2))[0,1]

df = fetch_data()
window = 20
scores = []

for i in range(len(df)-2*window):
    seg1 = df["close"].iloc[i:i+window].values
    seg2 = df["close"].iloc[i+window:i+2*window].values
    score = symmetry_score(seg1, seg2)
    scores.append((i, score))

signals = [(i, s) for i, s in scores if s > 0.8]
print("🔍 Найдены симметричные паттерны:", signals[:5])

#инструмент

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

📌Liquidity Shock Detector (LSD) - (детектор ликвидностных «провалов» и «всплесков» для алготрейдинга) 🧑‍💻

📌 Суть: 🧨

Большинство индикаторов смотрят на цену или объём. Но важнейший фактор в высокочастотном движении — изменение ликвидности в стакане. Когда она резко уходит (bid/ask сужается, глубина падает) — это предвестник сильного движения. Когда ликвидность «вливают» обратно — это сигнал разворота/успокоения.

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

1. Подключаемся к бирже (Binance, Bybit, OKX) через WebSocket и берём order book (L2).

2. В каждый момент времени считаем:

BidDepth = суммарный объём в N% от mid (например, до 0.5%).

AskDepth = то же для продаж.

Liquidity Imbalance = (BidDepth - AskDepth) / (BidDepth + AskDepth).


3. Дополнительно считаем Liquidity Volatility — дисперсия изменения глубины за последние M секунд.

4. Сигналы: 🚀

Если глубина резко «падает» > k×σ — значит рынок готов к скачку.

Если дисбаланс > +X% → рынок склоняется к росту.

Если дисбаланс < -X% → рынок склоняется к падению.

📊 Использование 💸

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

Фильтр для стратегий: можно использовать LSD как фильтр к любой ТА-стратегии — входить только при адекватной ликвидности.

Маркетмейкинг: инструмент помогает избегать ловушек, когда ликвидность «снимают» и твою лимитку исполняют перед движением.

import asyncio
import json
import websockets
import numpy as np
from collections import deque

SYMBOL = "btcusdt"
DEPTH_LIMIT = 100   # глубина стакана
PRICE_RANGE = 0.003 # 0.3% от mid для расчёта глубины
WINDOW = 20         # окно для волатильности ликвидности

# очередь для хранения истории
bid_depths = deque(maxlen=WINDOW)
ask_depths = deque(maxlen=WINDOW)

async def liquidity_shock_detector():
    url = f"wss://stream.binance.com:9443/ws/{SYMBOL}@depth{DEPTH_LIMIT}@100ms"

    async with websockets.connect(url) as ws:
        print(f"🚀 Liquidity Shock Detector запущен для {SYMBOL.upper()}")

        async for msg in ws:
            data = json.loads(msg)

            bids = np.array(data["bids"], dtype=float)
            asks = np.array(data["asks"], dtype=float)

            best_bid = float(bids[0][0])
            best_ask = float(asks[0][0])
            mid = (best_bid + best_ask) / 2

            # фильтрация по диапазону цен
            bid_depth = bids[bids[:,0] > mid * (1 - PRICE_RANGE), 1].sum()
            ask_depth = asks[asks[:,0] < mid * (1 + PRICE_RANGE), 1].sum()

            bid_depths.append(bid_depth)
            ask_depths.append(ask_depth)

            if len(bid_depths) >= WINDOW:
                # считаем дисбаланс и волатильность
                imbalance = (bid_depth - ask_depth) / (bid_depth + ask_depth + 1e-9)
                liq_vol = np.std(list(bid_depths) + list(ask_depths))

                # сигналы
                if abs(bid_depth - np.mean(bid_depths)) > 2 * liq_vol:
                    print(f"⚡ SHOCK по BID! Новая глубина: {bid_depth:.2f}, дисбаланс: {imbalance:.2f}")

                if abs(ask_depth - np.mean(ask_depths)) > 2 * liq_vol:
                    print(f"⚡ SHOCK по ASK! Новая глубина: {ask_depth:.2f}, дисбаланс: {imbalance:.2f}")

                if imbalance > 0.3:
                    print(f"🟢 Сильное смещение ликвидности в сторону покупателей! (imbalance={imbalance:.2f})")
                elif imbalance < -0.3:
                    print(f"🔴 Сильное смещение ликвидности в сторону продавцов! (imbalance={imbalance:.2f})")

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

#инструмент

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

📌Spectral Momentum - (анализ частот возвращений → адаптивный режим: тренд / контртренд) 🧑‍💻

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

Вместо привычных индикаторов смотрим спектр коротких периодов доходностей (STFT / спектральная плотность).

Если в спектре доминируют низкие частоты → движение «плавное» → лучше работает тренд-фолловер.

Если доминируют высокие частоты → рынок шумный → лучше mean-reversion.


Стратегия вычисляет: 🛠

спектральную метрику (например, спектральный центроид + flatness) на окне последних N баров и динамически выбирает тактику:

Trend Mode: вход по пробою EMA/Donchian, стоп = ATR, тейк = k×ATR.

Mean-Revert Mode: вход при экстремуме RSI или отклонении от VWAP, стоп = ATR, тейк = k×ATR.

Плюсы: ловит смену «характера» рынка, адаптируется к разным инструментам/ТФ, комбинирует частотный анализ и проверенные правила исполнения.

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

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

📌Adaptive Correlation Matrix (ACM) - (адаптивная матрица корреляций между монетами)

🔍 Идея: 🚀

Корреляции на крипторынке сильно меняются:

- в спокойные периоды многие альткоины «ходят за BTC»,

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

👉 Инструмент строит скользящую матрицу корреляций для выбранного пула монет и подсвечивает:

- какие пары монет максимально синхронны (корр > 0.8),

- какие — в антикорреляции (корр < -0.3),

- какие связи резко изменились (разрыв корреляции).

📌 Польза: 💸

- Фильтр для арбитражных/парных стратегий.

- Помогает искать независимые активы для диверсификации.

- Может показывать «сигнал домино»: если BTC ↔ ETH корреляция падает, часто это предвестник сильных движений.

import ccxt
import pandas as pd
import numpy as np
from datetime import datetime, timedelta, timezone

ex = ccxt.binance()

SYMBOLS = ["BTC/USDT", "ETH/USDT", "BNB/USDT", "SOL/USDT", "XRP/USDT"]
TF = "1h"
DAYS = 10

def fetch_data(symbol, days=10, tf="1h"):
    since = int((datetime.now(timezone.utc) - timedelta(days=days)).timestamp() * 1000)
    ohlcv = ex.fetch_ohlcv(symbol, timeframe=tf, since=since)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","vol"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    df.set_index("dt", inplace=True)
    return df["close"]

# собираем цены
data = pd.DataFrame()
for s in SYMBOLS:
    data[s] = fetch_data(s, DAYS, TF)

# считаем доходности
returns = data.pct_change().dropna()

# скользящая матрица корреляций
window = 24
rolling_corrs = returns.rolling(window).corr()

# текущая матрица
corr_matrix = returns.tail(window).corr()

print("📊 Текущая матрица корреляций (за последние 24 часа):")
print(corr_matrix.round(2))

# выделим сильные связи
for a in SYMBOLS:
    for b in SYMBOLS:
        if a != b:
            c = corr_matrix.loc[a, b]
            if c > 0.8:
                print(f"🟢 {a} и {b} сильно коррелируют ({c:.2f})")
            elif c < -0.3:
                print(f"🔴 {a} и {b} антикоррелируют ({c:.2f})")

#инструмент

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

📌Order Flow Inertia - (инерция потока заявок) 🧑‍💻

🔍 Идея: 🧨

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

👉 Мы отслеживаем кластеры агрессивных покупок/продаж и заходим в сторону потока.

⚙ Логика: 🛠️

1. Подключаемся к ленте сделок (trades) с биржи.

2. В скользящем окне (например, 30 секунд):

Суммируем объём агрессивных покупок (taker buy).

Суммируем объём агрессивных продаж (taker sell).

3. Если доля покупок >70% и объём > среднего ×2 → LONG.

4. Если доля продаж >70% и объём > среднего ×2 → SHORT.

5. Выход:

стоп = за ближайший swing low/high или ATR,

тейк = 2× стоп либо выход по развороту потока.

import asyncio
import json
import websockets
import numpy as np
from collections import deque
import time

SYMBOL = "btcusdt"
WINDOW_SEC = 30
MIN_VOLUME = 5    # минимальный объём для фильтра сделок

trades_buffer = deque()

async def order_flow_inertia():
    url = f"wss://stream.binance.com:9443/ws/{SYMBOL}@trade"

    async with websockets.connect(url) as ws:
        print(f"🚀 Order Flow Inertia запущена для {SYMBOL.upper()}")

        async for msg in ws:
            data = json.loads(msg)
            ts = data["T"] / 1000
            price = float(data["p"])
            qty = float(data["q"])
            is_buyer_maker = data["m"]  # True = sell (taker sell), False = buy (taker buy)

            if qty < MIN_VOLUME:
                continue

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

            # чистим буфер по окну времени
            while trades_buffer and ts - trades_buffer[0][0] > WINDOW_SEC:
                trades_buffer.popleft()

            if len(trades_buffer) > 10:
                buys = sum(q for t, q, m in trades_buffer if not m)
                sells = sum(q for t, q, m in trades_buffer if m)
                total = buys + sells

                if total == 0:
                    continue

                buy_share = buys / total
                sell_share = sells / total

                avg_vol = np.mean([q for _, q, _ in trades_buffer])

                if buy_share > 0.7 and buys > 2 * avg_vol:
                    print(f"🟢 LONG сигнал | Цена={price} | Покупки={buys:.2f}, Доля={buy_share:.2f}")
                elif sell_share > 0.7 and sells > 2 * avg_vol:
                    print(f"🔴 SHORT сигнал | Цена={price} | Продажи={sells:.2f}, Доля={sell_share:.2f}")

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

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

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

📌"Fractal Dimension Index (FDI) Monitor" - (монитор фрактальной размерности рынка) 🧑‍💻

🔍 Идея: 🧨

Рынок можно описывать через фрактальную размерность:

Когда FDI ≈ 1.2–1.4 → рынок трендовый, движения «линейные».

Когда FDI ≈ 1.5–1.7 → рынок хаотичный, ближе к флэту.

👉 Этот инструмент в реальном времени считает фрактальную размерность цены и даёт сигнал:

«Идёт тренд» → использовать тренд-следящие стратегии.

«Флэт» → применять mean-reversion, сетку или скальпинг.

📌 Польза: 💸

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

- Работает на любых таймфреймах.

- Можно совмещать с другими стратегиями как «режимный фильтр».

import ccxt
import pandas as pd
import numpy as np

ex = ccxt.binance()
symbol = "BTC/USDT"

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

def fractal_dimension(series, k_max=10):
    N = len(series)
    L = []
    k_values = range(2, k_max)
    for k in k_values:
        Lk = sum(abs(series[i] - series[i-k]) for i in range(k, N)) * (N-1)/(k*( (N-k+1)) )
        L.append(np.log(Lk))
    coeffs = np.polyfit(np.log(list(k_values)), L, 1)
    return coeffs[0]  # фрактальная размерность

df = fetch_data(symbol)
df["fdi"] = df["close"].rolling(200).apply(fractal_dimension, raw=False)

for i, row in df.tail(20).iterrows():
    if row["fdi"] < 1.4:
        print(f"📈 Трендовый режим | Цена={row['close']:.2f} | FDI={row['fdi']:.2f}")
    elif row["fdi"] > 1.5:
        print(f"🔄 Флэтовый режим | Цена={row['close']:.2f} | FDI={row['fdi']:.2f}")

#инструмент

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

📌"Entropy Breakout" - (пробой энтропии) 🧑‍💻

🔍 Идея: 🧨

Обычно анализируют цену, объём, стакан. Но редко кто смотрит на энтропию распределения движений цены.

👉 Смысл: 🚀

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

Когда энтропия высокая (движения хаотичны, свечи разнонаправленные) → рынок готов к сжатию/консолидации.

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

1. Берём скользящее окно из N свечей.

2. Строим распределение направлений/амплитуд свечей.

3. Считаем Shannon Entropy:

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

import ccxt
import pandas as pd
import numpy as np
from math import log2

ex = ccxt.binance()
symbol = "BTC/USDT"

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

def shannon_entropy(series, bins=10):
    hist, _ = np.histogram(series, bins=bins, density=True)
    hist = hist[hist > 0]
    return -np.sum(hist * np.log2(hist))

df = fetch_data(symbol)

window = 30
df["entropy"] = df["ret"].rolling(window).apply(shannon_entropy, raw=False)

# сигналы
low_thr = df["entropy"].quantile(0.2)
high_thr = df["entropy"].quantile(0.8)

for i, row in df.iterrows():
    if row["entropy"] < low_thr:
        print(f"⚡ Возможен пробой! {row['close']}")
    elif row["entropy"] > high_thr:
        print(f"🔄 Возможна консолидация! {row['close']}")

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

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

📌"Stop Hunt Reversal"- (разворот на выбивании стопов) 🧑‍💻

🔍 Идея: 🧨

Рынок часто двигается так:

1. Сначала выбивает стопы за локальными экстремумами (false breakout).

2. Потом идёт в противоположную сторону — туда, где реальная ликвидность.

👉 Мы ловим именно этот «шип ликвидности» и входим против ложного пробоя.

⚙ Логика: 🛠️

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

2. Если цена пробила хай/лоу и сразу вернулась обратно в диапазон → это «охота за стопами».

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

ложный пробой вверх → SHORT,

ложный пробой вниз → LONG.

4. Стоп — чуть выше/ниже ложного пробоя.

5. Тейк — середина диапазона или 2× стоп.

import ccxt
import pandas as pd

ex = ccxt.binance()
symbol = "BTC/USDT"

def fetch_data(symbol, tf="15m", limit=300):
    ohlcv = ex.fetch_ohlcv(symbol, tf, limit=limit)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","vol"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    return df

df = fetch_data(symbol)

LOOKBACK = 20
signals = []

for i in range(LOOKBACK, len(df)):
    window = df.iloc[i-LOOKBACK:i]
    high_level = window["high"].max()
    low_level = window["low"].min()
    price = df.iloc[i]["close"]
    high_break = df.iloc[i]["high"] > high_level
    low_break = df.iloc[i]["low"] < low_level

    # Ложный пробой вверх → шорт
    if high_break and price < high_level:
        signals.append((df.iloc[i]["dt"], price, "SHORT"))
    # Ложный пробой вниз → лонг
    elif low_break and price > low_level:
        signals.append((df.iloc[i]["dt"], price, "LONG"))

print("🚨 Сигналы ложных пробоев:")
for s in signals[-10:]:
    print(s)

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

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

📌"Order Flow Replay" - (реплей потока ордеров) 🧑‍💻

🔍 Идея: 🚀

Большинство трейдеров видят только свечи, но свечи — это уже «сжатая» история.

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

📌 Польза: 💸

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

- Можно «перематывать назад» и тестировать реакции на поток ордеров.

- Отлично подходит для изучения HFT-паттернов (резкие вливания ликвидности, айсберг-заявки, стоп-ханты).

import asyncio
import json
import websockets
import pandas as pd
from datetime import datetime

SYMBOL = "btcusdt"
trades = []

async def record_trades():
    url = f"wss://stream.binance.com:9443/ws/{SYMBOL}@trade"
    async with websockets.connect(url) as ws:
        print(f"🎥 Запись сделок {SYMBOL.upper()}...")
        async for msg in ws:
            data = json.loads(msg)
            trades.append({
                "time": datetime.fromtimestamp(data["T"]/1000),
                "price": float(data["p"]),
                "qty": float(data["q"]),
                "side": "BUY" if not data["m"] else "SELL"
            })
            if len(trades) % 100 == 0:
                pd.DataFrame(trades).to_csv("trades_record.csv", index=False)

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

#инструмент

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

📌"Market Friction Meter" - (измеритель «трения» рынка) 🧑‍💻

🔍 Идея: 🚀

Обычный трейдер видит только цену и объём. Но реальный рынок состоит из неэффективностей — задержек в движении цены из-за скрытых лимитных ордеров и «вязкости» ликвидности.

👉 Этот инструмент измеряет, насколько легко или трудно рынку двигаться в данный момент:

Если много агрессивных сделок в одну сторону → а цена почти не двигается → высокая «фрикция» (рынок вязнет).

Если цена летит даже на малом объёме → низкая «фрикция» (рынок «скользкий»).

📌 Польза: 💸

- Помогает отличить настоящий импульс от «ложного шума».

- Можно использовать как фильтр:

Высокая фрикция → не входим (рынок «вязкий»).

Низкая фрикция → входы эффективнее.

- Отлично подходит для скальпинга и поиска «тонкого льда» на рынке.

import asyncio
import json
import websockets
from collections import deque
import numpy as np
from datetime import datetime

SYMBOL = "btcusdt"
WINDOW = 100  # сколько сделок анализируем

trades = deque(maxlen=WINDOW)

async def friction_meter():
    url = f"wss://stream.binance.com:9443/ws/{SYMBOL}@trade"
    async with websockets.connect(url) as ws:
        print(f"⚙️ Market Friction Meter для {SYMBOL.upper()} запущен")
        async for msg in ws:
            data = json.loads(msg)
            price = float(data["p"])
            qty = float(data["q"])
            side = "BUY" if not data["m"] else "SELL"
            ts = datetime.fromtimestamp(data["T"]/1000)

            trades.append((price, qty, side))

            if len(trades) == WINDOW:
                prices = [t[0] for t in trades]
                qtys = [t[1] for t in trades]

                price_move = abs(prices[-1] - prices[0])
                volume = sum(qtys)

                # коэффициент трения = объем / движение цены
                friction = volume / (price_move+1e-9)

                if friction > 5000:
                    status = "🟥 Высокая фрикция (рынок вязкий)"
                elif friction < 500:
                    status = "🟩 Низкая фрикция (рынок скользкий)"
                else:
                    status = "🟨 Нейтрально"

                print(f"[{ts}] ΔP={price_move:.2f}, V={volume:.2f}, F={friction:.1f} → {status}")

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

#инструмент

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

📌"Adaptive Range Compression" - (адаптивная торговля на сжатии диапазона) 🧑‍💻

🔍 Идея: 🧨

Перед крупным движением рынок часто «сжимается» — диапазон свечей становится всё меньше, волатильность падает. Это сигнал накопления позиций.

👉 Мы ждём момента, когда диапазон становится аномально узким, и торгуем на выход из этого «сжатия».

⚙ Логика: 🛠️

1. Берём последние N свечей и считаем их средний диапазон (High–Low).

2. Если текущий диапазон меньше, чем, скажем, 40% среднего → фиксируем «сжатие».

3. Ставим условные ордера:

Buy stop выше high текущей свечи.

Sell stop ниже low текущей свечи.

4. Вход в сторону пробоя.

5. Стоп-лосс — за противоположной границей свечи сжатия.

6. Тейк — 2–3× размер диапазона сжатия.

import ccxt
import pandas as pd

ex = ccxt.binance()
symbol = "BTC/USDT"

def fetch_data(symbol, tf="15m", limit=200):
    ohlcv = ex.fetch_ohlcv(symbol, tf, limit=limit)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","vol"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    df["range"] = df["high"] - df["low"]
    return df

df = fetch_data(symbol)

N = 20
signals = []

for i in range(N, len(df)):
    avg_range = df.iloc[i-N:i]["range"].mean()
    curr_range = df.iloc[i]["range"]
    
    if curr_range < 0.4 * avg_range:
        compression = df.iloc[i]
        high, low = compression["high"], compression["low"]
        signals.append({
            "time": compression["dt"],
            "long_entry": high,
            "short_entry": low,
            "stop_long": low,
            "stop_short": high,
            "take_profit_long": high + 2 * curr_range,
            "take_profit_short": low - 2 * curr_range
        })

print("🚀 Сигналы по сжатию диапазона:")
for s in signals[-5:]:
    print(s)

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

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

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