📌Volatility Coil 🧑‍💻 - (катушка волатильности — играем на резком выбросе после «сжатия»)

🔍 Идея: 🚀

Рынок не бывает долго одинаково активным:

после периода низкой волатильности всегда следует выброс волатильности,

направление выбираем с помощью фильтров тренда (EMA или объёмы).

👉 Мы ищем зоны, где рынок «сжимается в катушку» → вход на пробой.

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

1. Считаем ATR (волатильность).

2. Берём скользящее среднее ATR → находим моменты, когда текущая вола сильно ниже среднего (например, < 0.7 * SMA(ATR)).

3. Это сигнал «сжатия».

4. Входим в сделку на пробой High/Low диапазона последних N свечей.

Если цена выше EMA → лонг на пробой.

Если ниже EMA → шорт.

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

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

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

def atr(df, period=14):
    hl = df["high"] - df["low"]
    hc = abs(df["high"] - df["close"].shift())
    lc = abs(df["low"] - df["close"].shift())
    tr = pd.concat([hl,hc,lc], axis=1).max(axis=1)
    return tr.rolling(period).mean()

def generate_signals(df):
    df["ATR"] = atr(df)
    df["ATR_SMA"] = df["ATR"].rolling(50).mean()
    df["EMA"] = df["close"].ewm(span=50).mean()

    signals = []
    for i in range(50, len(df)):
        if df["ATR"].iloc[i] < 0.7 * df["ATR_SMA"].iloc[i]:  # сжатие
            rng_high = df["high"].iloc[i-10:i].max()
            rng_low = df["low"].iloc[i-10:i].min()
            price = df["close"].iloc[i]
            ema = df["EMA"].iloc[i]

            if price > ema:  # тренд вверх
                signals.append((df["dt"].iloc[i], "LONG", rng_high))
            elif price < ema:
                signals.append((df["dt"].iloc[i], "SHORT", rng_low))
    return signals

if __name__ == "__main__":
    df = fetch_ohlcv()
    signals = generate_signals(df)
    for s in signals[-10:]:
        print(s)

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

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

📌Market Stress Monitor 🧑‍💻 - (монитор стресса рынка — индикатор перегрева и паники)

🔍 Идея: 🧨

Рынок двигается не только из-за цены, но и из-за стресса участников.
Когда трейдеры перегружены риском или паникой → начинаются всплески ликвидаций, резкие движения, «свечи-шипы».

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

📈 Funding rate (перегретый лонг/шорт).

🔥 Открытый интерес (OI) — резкие скачки → перегруз позиций.

💧 Ликвидации (массовые стоп-ауты).

📊 Скорость изменения цены (аномально быстрое движение).

На основе этих сигналов строится индекс стресса рынка (0–100).

⚙️ Польза: 💸

- Если стресс высокий (например, >70) → рынок опасен, лучше не открывать новые сделки.

- Если стресс низкий (<30) → рынок «спокоен», хорошее время для аккуратного входа.

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

import ccxt
import pandas as pd
import numpy as np
import time

exchange = ccxt.binance({
    "enableRateLimit": True,
    "options": {"defaultType": "future"}
})

SYMBOL = "BTC/USDT"

def get_funding_rate():
    data = exchange.fapiPublic_get_fundingrate({"symbol": "BTCUSDT", "limit": 1})
    return float(data[0]["fundingRate"])

def get_open_interest():
    data = exchange.fapiPublic_get_openinterest({"symbol": "BTCUSDT"})
    return float(data["openInterest"])

def get_ticker():
    t = exchange.fetch_ticker(SYMBOL)
    return t["last"], t["quoteVolume"]

def stress_index():
    funding = abs(get_funding_rate()) * 10000   # в bps
    oi = get_open_interest()
    price, volume = get_ticker()

    # нормализация (примитивный пример)
    funding_score = min(funding, 50)
    oi_score = np.log(oi) % 100
    vol_score = np.log(volume) % 100

    score = (funding_score*0.3 + oi_score*0.3 + vol_score*0.4)
    return round(score, 2)

while True:
    print("Market Stress:", stress_index())
    time.sleep(60)

#инструмент

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

📌Position Sizer Pro - (умный калькулятор объёма позиции с адаптивным риском) 🧑‍💻

🔍 Идея: 🧨

Обычный калькулятор позиции скучный: трейдер сам вводит риск и плечо.
👉 Здесь мы делаем умный авто-калькулятор, который:

сам подбирает размер позиции на основе волатильности инструмента (ATR),

учитывает баланс трейдера,

автоматически корректирует риск, если рынок в «стрессе».

📌 Польза: 💸

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

Работает как универсальный помощник, который адаптирует риск под рынок.

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

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

SYMBOL = "BTC/USDT"
TF = "1h"
BALANCE = 1000  # USDT
MAX_RISK = 0.02  # 2% от депо

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

def atr(df, period=14):
    hl = df["high"] - df["low"]
    hc = abs(df["high"] - df["close"].shift())
    lc = abs(df["low"] - df["close"].shift())
    tr = pd.concat([hl,hc,lc], axis=1).max(axis=1)
    return tr.rolling(period).mean()

def position_size(df, balance, max_risk):
    df["ATR"] = atr(df)
    atr_val = df["ATR"].iloc[-1]
    price = df["close"].iloc[-1]

    # риск на сделку
    risk_amount = balance * max_risk

    # стоп = ATR
    stop_dist = atr_val

    # размер позиции
    qty = risk_amount / stop_dist
    contracts = qty / price * balance
    return round(contracts, 4), price, atr_val

if __name__ == "__main__":
    df = fetch_data()
    size, price, atr_val = position_size(df, BALANCE, MAX_RISK)
    print(f"Цена: {price:.2f} | ATR: {atr_val:.2f}")
    print(f"Рекомендуемый размер позиции: {size} контрактов")

#инструмент

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

📌Echo Levels - (эхо-уровни — торговля на повторных отскоках скрытых уровней) 🧑‍💻

🔍 Идея: 🧨

Большинство стратегий ищут пробои или тренды.
Мы используем эффект памяти рынка:

Когда цена резко отскакивает от какого-то уровня (например, на сильном объёме),

Этот уровень остаётся «акустическим эхом» → рынок часто повторно реагирует на него спустя часы/дни.

👉 Мы строим «эхо-уровни» на основе сильных отскоков и ловим повторные касания.

⚙️ Логика: 🛠️

1. Ищем свечи с аномальным объёмом + длинным хвостом (шипом).
→ это значит, что там «защищали уровень».

2. Запоминаем цену отскока (например, минимум хвоста).

3. Если в будущем цена снова касается этой зоны ±0.2% → входим в сторону отскока.

Лонг при повторном касании «эхо-дна».

Шорт при повторном касании «эхо-верха».

4. Стоп → чуть за уровень (например, 0.5% ниже).

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

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

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

def find_echo_levels(df, vol_mult=2.5):
    levels = []
    vol_mean = df["volume"].rolling(20).mean()
    for i in range(20, len(df)):
        if df["volume"].iloc[i] > vol_mult * vol_mean.iloc[i]:
            body = abs(df["close"].iloc[i] - df["open"].iloc[i])
            wick_low = df["low"].iloc[i]
            wick_high = df["high"].iloc[i]
            # если свеча с хвостом вниз
            if wick_low < df["close"].iloc[i] - body:
                levels.append(("LONG", wick_low))
            # если свеча с хвостом вверх
            if wick_high > df["close"].iloc[i] + body:
                levels.append(("SHORT", wick_high))
    return levels

def check_signals(df, levels, tolerance=0.2):
    signals = []
    for _, row in df.iterrows():
        price = row["close"]
        for side, lvl in levels:
            if side == "LONG" and abs(price - lvl) / lvl < tolerance/100:
                signals.append((row["dt"], "LONG", lvl))
            if side == "SHORT" and abs(price - lvl) / lvl < tolerance/100:
                signals.append((row["dt"], "SHORT", lvl))
    return signals

if __name__ == "__main__":
    df = fetch_ohlcv()
    levels = find_echo_levels(df)
    signals = check_signals(df, levels)
    for s in signals[-10:]:
        print(s)

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

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

📌Correlation Radar - (онлайн-радар корреляций между монетами)🧑‍💻

🔍 Идея: 🧨

Трейдеры часто смотрят на один актив (например, BTC или ETH), но забывают, что альткоины и даже разные сектора рынка двигаются синхронно или расходятся.
👉 Инструмент в реальном времени считает корреляции между монетами и показывает:

кто сейчас ходит «в унисон» с BTC,

кто отклеился и может дать уникальный трейд,

кто образует «кластер» синхронных движений.

📌 Польза: 💸

Легко находить альткоины, которые опаздывают за BTC → точка входа.

Видеть «аномально» независимые монеты → диверсификация.

Использовать как фильтр: торговать только то, что слабо связано с BTC (для хеджей).

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

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

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

def fetch_data(symbol):
    since = int((datetime.utcnow() - timedelta(days=DAYS)).timestamp() * 1000)
    ohlcv = exchange.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")
    df.set_index("dt", inplace=True)
    return df["close"]

def correlation_matrix(symbols):
    data = {}
    for s in symbols:
        data[s.split("/")[0]] = fetch_data(s)
    df = pd.DataFrame(data)
    corr = df.pct_change().corr()
    return corr

if __name__ == "__main__":
    corr = correlation_matrix(SYMBOLS)
    print("Корреляции (за последние 7 дней):")
    print(corr)

#инструмент

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

📌Volatility Harvest- (сборщик волатильности — зарабатываем на чередовании «спокойных» и «бурных» фаз рынка) 🧑‍💻

🔍 Идея: 🚀

Рынок двигается циклами:

периоды спокойствия (узкий диапазон, маленькие свечи),

за ними почти всегда идут взрывы волатильности (пробой, импульс).

👉 Задача стратегии — войти в самом конце «тишины» и поймать «бурю».

⚙ Логика: 🛠️

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

берем ATR (средний диапазон свечей),

сравниваем его с историческим средним.

2. Если ATR сильно ниже нормы (рынок «уснул») → ставим ловушки на пробой в обе стороны.

3. Вход:

Buy Stop чуть выше максимума последних N свечей,

Sell Stop чуть ниже минимума.

4. Когда одна из ловушек срабатывает → вторую отменяем.

5. Стоп = противоположный край диапазона, тейк = 2–3× стоп.

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

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

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 atr(df, period=14):
    hl = df["high"] - df["low"]
    hc = abs(df["high"] - df["close"].shift())
    lc = abs(df["low"] - df["close"].shift())
    tr = pd.concat([hl,hc,lc], axis=1).max(axis=1)
    return tr.rolling(period).mean()

def find_signals(df, atr_period=14, window=20, threshold=0.5):
    df["ATR"] = atr(df, atr_period)
    signals = []
    for i in range(window, len(df)):
        atr_now = df["ATR"].iloc[i]
        atr_mean = df["ATR"].iloc[i-window:i].mean()

        # рынок "уснул"
        if atr_now < threshold * atr_mean:
            rng_high = df["high"].iloc[i-window:i].max()
            rng_low = df["low"].iloc[i-window:i].min()
            signals.append((df["dt"].iloc[i], "SET_TRAP", rng_low, rng_high))
    return signals

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

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

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

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