📌Emotion Index - (индекс эмоций рынка в реальном времени) 🧑‍💻

🔍 Идея: 🧨

Рынок движется эмоциями: страхом и жадностью. У нас есть «индекс страха и жадности» в интернете, но он статичный и обновляется раз в сутки.
👉 Этот инструмент строит свой локальный индекс эмоций из данных:

волатильность (чем выше → тем больше паники/ажиотажа),

соотношение красных и зелёных свечей,

объёмные всплески,

перекупленность/перепроданность (RSI).

📌 Польза: 💸

Можно быстро видеть, в каком «психологическом режиме» сейчас рынок.

Использовать как фильтр:

если паника → лучше искать лонг (контртренд),

если эйфория → лучше искать шорт.

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

SYMBOL = "BTC/USDT"
TF = "1h"
DAYS = 7

def fetch_data():
    ex = ccxt.binance({"enableRateLimit": True})
    since = int((datetime.now(timezone.utc) - timedelta(days=DAYS)).timestamp() * 1000)
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, since)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    return df

def rsi(series, period=14):
    delta = series.diff()
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
    avg_gain = gain.rolling(period).mean()
    avg_loss = loss.rolling(period).mean()
    rs = avg_gain / avg_loss
    return 100 - (100 / (1 + rs))

def emotion_index(df):
    # компоненты
    vol = df["close"].pct_change().rolling(24).std() * 100
    trend = (df["close"] > df["open"]).rolling(24).mean() * 100
    rsi_val = rsi(df["close"])
    
    # нормализация
    vol_norm = (vol - vol.min()) / (vol.max() - vol.min())
    trend_norm = trend / 100
    rsi_norm = (rsi_val - 30) / 40  # 0 при RSI=30, 1 при RSI=70
    
    # интегральный индекс
    emotion = (vol_norm + trend_norm + rsi_norm) / 3
    df["EmotionIndex"] = emotion
    return df

if __name__ == "__main__":
    df = fetch_data()
    df = emotion_index(df)
    print(df[["dt", "close", "EmotionIndex"]].tail(10))

#инструмент

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

📌Liquidity Vacuum - (играем на “пылесосе ликвидности” после ложного пробоя)🧑‍💻

🔍 Идея: 🧨

Большие игроки часто двигают цену так:

1. Вытягивают ликвидность из рынка ложным пробоем ключевого уровня (high/low дня или диапазона).

2. Забирают стопы толпы.

3. Возвращают цену обратно → начинается импульс в противоположную сторону.

👉 Стратегия ловит именно этот момент — когда «всосали ликвидность» и цена разворачивается.

⚙ Логика: 🛠️

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

2. Если цена пробивает high/low и возвращается внутрь диапазона → сигнал в противоположную сторону.

3. Стоп: за экстремум пробоя.

4. Тейк: 2–3× стоп или середина диапазона.

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

SYMBOL = "BTC/USDT"
TF = "15m"
DAYS = 5

def fetch_data():
    ex = ccxt.binance({"enableRateLimit": True})
    since = int((datetime.now(timezone.utc) - timedelta(days=DAYS)).timestamp() * 1000)
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, since)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    return df

def liquidity_vacuum(df, window=24):
    signals = []
    for i in range(window, len(df)):
        rng_high = df["high"].iloc[i-window:i].max()
        rng_low = df["low"].iloc[i-window:i].min()
        price = df["close"].iloc[i]

        # ложный пробой high
        if df["high"].iloc[i] > rng_high and price < rng_high:
            signals.append((df["dt"].iloc[i], "SHORT", rng_high, price))

        # ложный пробой low
        if df["low"].iloc[i] < rng_low and price > rng_low:
            signals.append((df["dt"].iloc[i], "LONG", rng_low, price))

    return signals

if __name__ == "__main__":
    df = fetch_data()
    sigs = liquidity_vacuum(df)
    for s in sigs[-5:]:
        print(s)

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

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

📌Stop-Loss Heatmap - (тепловая карта вероятных стопов толпы) 🧑‍💻

🔍 Идея: 🧨

Большинство трейдеров ставят стопы «по учебнику»:

чуть ниже локального минимума,

чуть выше локального максимума,

за круглым уровнем (например, 30 000, 25 000).

👉 Этот инструмент строит тепловую карту зон, где скапливаются вероятные стопы.
Когда цена подходит к такой зоне — велика вероятность:

или резкого пробоя (снос стопов),

или ложного прокола и возврата (идеальная точка входа).

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

1. Берём историю свечей (например, 14 дней).

2. Ищем локальные минимумы/максимумы + круглые уровни.

3. Считаем «плотность стопов» вокруг них.

4. Рисуем тепловую карту уровней.

import ccxt
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta, timezone

SYMBOL = "BTC/USDT"
TF = "1h"
DAYS = 10

def fetch_data():
    ex = ccxt.binance({"enableRateLimit": True})
    since = int((datetime.now(timezone.utc) - timedelta(days=DAYS)).timestamp() * 1000)
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, since)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    return df

def stop_loss_heatmap(df, round_step=500):
    levels = []

    # локальные экстремумы
    for i in range(2, len(df)-2):
        if df["low"].iloc[i] < df["low"].iloc[i-1] and df["low"].iloc[i] < df["low"].iloc[i+1]:
            levels.append(df["low"].iloc[i])
        if df["high"].iloc[i] > df["high"].iloc[i-1] and df["high"].iloc[i] > df["high"].iloc[i+1]:
            levels.append(df["high"].iloc[i])

    # круглые уровни
    price_min, price_max = df["low"].min(), df["high"].max()
    round_levels = np.arange(int(price_min), int(price_max), round_step)
    levels.extend(round_levels)

    # тепловая карта (гистограмма)
    hist, bins = np.histogram(levels, bins=50)

    plt.figure(figsize=(10,6))
    plt.hist(levels, bins=50, color="red", alpha=0.6)
    plt.title("Stop-Loss Heatmap")
    plt.xlabel("Price Level")
    plt.ylabel("Stop Density")
    plt.show()

if __name__ == "__main__":
    df = fetch_data()
    stop_loss_heatmap(df)

#инструмент

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

📌Time-Shift Arbitrage - (арбитраж времени между таймфреймами) 🧑‍💻

🔍 Идея: 🚀

Большинство стратегий смотрят только один таймфрейм. Но рынок часто даёт асинхронные сигналы:

На младшем ТФ (5m–15m) формируется локальный импульс.

На старшем ТФ (4h–1d) тренд ещё не развернулся.

👉 Если входить в сделку тогда, когда младший ТФ уже даёт сигнал, а старший ещё “не в курсе”, можно поймать движение на ранней стадии. Это и есть «арбитраж времени».

⚙ Логика: 🛠️

1. Считаем RSI или MACD на двух разных ТФ: например, 15m и 4h.

2. Если на младшем ТФ появляется сигнал (например, RSI < 30 = перепроданность), а на старшем ещё нет — входим заранее.

3. Выход → когда старший ТФ подтверждает тот же сигнал.

4. Это даёт преимущество «форвард-сигнала» в несколько часов.

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

SYMBOL = "BTC/USDT"

def fetch_data(tf="15m", days=7):
    ex = ccxt.binance({"enableRateLimit": True})
    since = int((datetime.now(timezone.utc) - timedelta(days=days)).timestamp() * 1000)
    ohlcv = ex.fetch_ohlcv(SYMBOL, tf, since)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    return df

def rsi(series, period=14):
    delta = series.diff()
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
    avg_gain = gain.rolling(period).mean()
    avg_loss = loss.rolling(period).mean()
    rs = avg_gain / avg_loss
    return 100 - (100 / (1 + rs))

def time_shift_arbitrage(df_fast, df_slow):
    df_fast["RSI"] = rsi(df_fast["close"])
    df_slow["RSI"] = rsi(df_slow["close"])
    signals = []

    for i in range(len(df_fast)):
        if df_fast["RSI"].iloc[i] < 30:  # перепроданность на младшем
            # проверим старший
            last_slow = df_slow[df_slow["dt"] <= df_fast["dt"].iloc[i]].iloc[-1]
            if last_slow["RSI"] > 40:  # старший ещё не перепродан
                signals.append((df_fast["dt"].iloc[i], "LONG", df_fast["close"].iloc[i]))

        if df_fast["RSI"].iloc[i] > 70:  # перекупленность на младшем
            last_slow = df_slow[df_slow["dt"] <= df_fast["dt"].iloc[i]].iloc[-1]
            if last_slow["RSI"] < 60:  # старший ещё не перекуплен
                signals.append((df_fast["dt"].iloc[i], "SHORT", df_fast["close"].iloc[i]))

    return signals

if __name__ == "__main__":
    fast = fetch_data("15m", days=5)
    slow = fetch_data("4h", days=30)
    sigs = time_shift_arbitrage(fast, slow)
    for s in sigs[-5:]:
        print(s)

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

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

📌Position Stress Tester - (стресс-тестер позиции в реальном времени) 🧑‍💻

🔍 Идея : 🚀

Большинство трейдеров видят только текущую цену и PnL. Но гораздо полезнее знать:

что будет с позицией, если цена упадёт на −5% или вырастет на +5%;

при каком сценарии тебя ликвиднёт;

сколько реально потеряешь/заработаешь при изменении волатильности.

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

📌 Польза: 💸

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

- Помогает решать: усилить позицию, захеджировать или выйти.

- Отлично подходит для риск-менеджмента.

import numpy as np
import matplotlib.pyplot as plt

# параметры позиции
entry_price = 27000
position_size = 0.5  # BTC
leverage = 5
margin = (entry_price * position_size) / leverage

# диапазон сценариев: -15% ... +15%
scenarios = np.linspace(-0.15, 0.15, 50)
prices = entry_price * (1 + scenarios)

# расчёт PnL
pnl = (prices - entry_price) * position_size * leverage

# точка ликвидации (приблизительно)
liq_price = entry_price - margin / (position_size / entry_price)

plt.figure(figsize=(10,5))
plt.plot(prices, pnl, label="PnL")
plt.axvline(entry_price, color="blue", linestyle="--", label="Entry")
plt.axvline(liq_price, color="red", linestyle="--", label="Liq Price")
plt.axhline(0, color="black", linewidth=0.7)

plt.title("Stress Test позиции")
plt.xlabel("Цена BTC")
plt.ylabel("PnL (USDT)")
plt.legend()
plt.grid(True)
plt.show()

#инструмент

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

📌Liquidity Sweep Reversal - (разворот после сноса ликвидности) 🧑‍💻

🔍 Идея: 🧨

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

цена выбивает стопы за локальными минимумами/максимумами,

после чего идёт быстрый возврат в диапазон.

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

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

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

2. Если цена пробивает уровень и через 1–2 свечи возвращается обратно — это сигнал.

3. Входим в лонг, если снесли low и вернулись выше уровня.
Входим в шорт, если снесли high и вернулись ниже уровня.

4. Стоп — чуть дальше экстремума.

5. Тейк — 2–3х от риска.

import ccxt
import pandas as pd

SYMBOL = "BTC/USDT"
TF = "15m"
LOOKBACK = 20

ex = ccxt.binance({"enableRateLimit": True})

def fetch_data():
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, limit=500)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
    return df

def liquidity_sweep_strategy(df):
    signals = []
    for i in range(LOOKBACK, len(df)):
        high_level = df["high"].iloc[i-LOOKBACK:i].max()
        low_level  = df["low"].iloc[i-LOOKBACK:i].min()
        c, o = df["close"].iloc[i], df["open"].iloc[i]

        # Лонг после сноса минимума
        if df["low"].iloc[i] < low_level and c > low_level:
            signals.append((df["ts"].iloc[i], "LONG", c, low_level))

        # Шорт после сноса максимума
        if df["high"].iloc[i] > high_level and c < high_level:
            signals.append((df["ts"].iloc[i], "SHORT", c, high_level))

    return signals

if __name__ == "__main__":
    df = fetch_data()
    sigs = liquidity_sweep_strategy(df)
    for s in sigs[-5:]:
        print(s)

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

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

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