📌"Volatility Heat Scanner" — сканер всплесков волатильности по множеству пар🧑‍💻

📌 Суть инструмента:🛠️

Сканирует выбранные торговые пары и находит те, где за последние X минут/часов произошёл резкий рост волатильности. Это может сигнализировать:

- о начале движения

- об окончании флэта

- о потенциальной возможности входа

🔧 Что делает:💻

- Загружает свечные данные по множеству пар

- Вычисляет стандартное отклонение или диапазон

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

- Сортирует пары по силе всплеска

import ccxt
import pandas as pd

exchange = ccxt.binance()
symbols = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'BNB/USDT', 'XRP/USDT']
timeframe = '15m'
vol_window = 20
results = []

for symbol in symbols:
    try:
        ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=vol_window + 5)
        df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
        df["range"] = df["high"] - df["low"]
        avg_range = df["range"].rolling(vol_window).mean().iloc[-2]
        current_range = df["range"].iloc[-1]
        ratio = current_range / avg_range if avg_range else 0

        results.append((symbol, ratio))
    except Exception as e:
        print(f"⚠️ Ошибка по {symbol}: {e}")

# Вывод топов по всплеску волатильности
sorted_results = sorted(results, key=lambda x: x[1], reverse=True)
print("\n🔥 Пары с наибольшим всплеском волатильности:")
for symbol, r in sorted_results:
    if r > 1.5:
        print(f"{symbol}: волатильность +{r:.2f}x выше нормы")

#инструмент

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

📌"News Latency Detector" — определение отставания реакции рынка на новости🧑‍💻

📌 Суть инструмента:🛠️

Многие активы запаздывают с реакцией на важные новости (например, альткойны после движения BTC).
Инструмент ищет те пары, которые не отреагировали или движутся с задержкой, что даёт шанс войти до основного импульса.

🔧 Что делает:💻

- Мониторит реакцию ключевого актива (например, BTC)

- Сравнивает процентное изменение других активов за последние X минут

- Ищет отстающих слабо реагирующих токенов

- Выдаёт список потенциальных "догоняющих"

import ccxt
import time

exchange = ccxt.binance()
symbols = ["BTC/USDT", "ETH/USDT", "SOL/USDT"]
lookback_minutes = 15

def get_price_change(symbol):
    candles = exchange.fetch_ohlcv(symbol, '1m', limit=lookback_minutes)
    open_price = candles[0][1]
    close_price = candles[-1][4]
    return (close_price - open_price) / open_price * 100

btc_change = get_price_change("BTC/USDT")
print(f"🚀 BTC изменился на {btc_change:.2f}% за {lookback_minutes} мин")

for symbol in symbols[1:]:
    alt_change = get_price_change(symbol)
    delta = btc_change - alt_change
    print(f"{symbol} реагирует на {delta:.2f}% слабее")

    if delta > 1.0:
        print(f"⚡ {symbol} может быть догоняющим кандидатом!")

#инструмент

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

📌Grid Rebalancer между BTC и USDT🧑‍💻

- Бот хранит активы в BTC и USDT

- Когда цена BTC падает и касается уровня — USDT обменивается на BTC

- Когда цена растёт — часть BTC продаётся за USDT

- Баланс постоянно корректируется, оставаясь в рамках заданной сетки

import ccxt
import pandas as pd
import time

# --- Настройки ---
symbol = "BTC/USDT"
grid_step = 100       # расстояние между уровнями
grid_levels = 5       # уровни вверх/вниз от центра
btc_balance = 0.01
usdt_balance = 300
last_price = None
grid = []
rebalanced = set()

exchange = ccxt.binance()

# --- Построение сетки ---
def create_grid(center_price):
    return [round(center_price + i * grid_step, 2) for i in range(-grid_levels, grid_levels + 1)]

# --- Симуляция перераспределения ---
def rebalance(price):
    global btc_balance, usdt_balance
    for level in grid:
        if price <= level and level not in rebalanced:
            # Покупка BTC (добавляем в позицию)
            buy_amount = usdt_balance * 0.1 / price
            btc_balance += buy_amount
            usdt_balance -= buy_amount * price
            print(f"🟢 BUY @ {level} | BTC: {btc_balance:.5f} | USDT: {usdt_balance:.2f}")
            rebalanced.add(level)
        elif price >= level and level not in rebalanced:
            # Продажа BTC (уменьшаем позицию)
            sell_amount = btc_balance * 0.1
            btc_balance -= sell_amount
            usdt_balance += sell_amount * price
            print(f"🔴 SELL @ {level} | BTC: {btc_balance:.5f} | USDT: {usdt_balance:.2f}")
            rebalanced.add(level)

# --- Основной цикл ---
while True:
    try:
        ticker = exchange.fetch_ticker(symbol)
        current_price = ticker['last']
        if last_price is None:
            grid = create_grid(current_price)
            print("📊 Сетка создана:", grid)

        rebalance(current_price)
        last_price = current_price
        time.sleep(10)

    except Exception as e:
        print(f"Ошибка: {e}")
        time.sleep(5)

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

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

📌"Correlation Shock Strategy" — вход при внезапном разрушении корреляции🧑‍💻

📌 Суть стратегии:🛠️

Многие активы (например, ETH и BTC, или LDO и LSD-токены) часто движутся в сильной корреляции.
В моменты, когда один актив резко уходит в сторону, а другой — остаётся на месте, возникает временной дисбаланс.
Стратегия использует это как возможность на вход: покупать "отстающий" актив до восстановления синхронности.

✅ Условия стратегии:🚀

1. Выбрать две коррелированные пары (например, ETH/USDT и LDO/USDT)

2. Рассчитать изменение цены за последние N минут

3. Если расхождение между изменениями превышает X%, генерируется сигнал

4. Вход в позицию по "отстающему" активу, выход — при восстановлении корреляции

import ccxt
import pandas as pd

exchange = ccxt.binance()
pairs = {"ETH/USDT": None, "LDO/USDT": None}

lookback = 15  # минут
threshold = 2.0  # % расхождение для сигнала

for symbol in pairs.keys():
    candles = exchange.fetch_ohlcv(symbol, '1m', limit=lookback)
    df = pd.DataFrame(candles, columns=["ts", "open", "high", "low", "close", "volume"])
    pairs[symbol] = df["close"].iloc[-1] / df["close"].iloc[0] - 1

eth_move = pairs["ETH/USDT"] * 100
ldo_move = pairs["LDO/USDT"] * 100
delta = eth_move - ldo_move

print(f"📊 ETH: {eth_move:.2f}%, LDO: {ldo_move:.2f}% | Δ: {delta:.2f}%")

if abs(delta) >= threshold:
    lagging = "LDO" if delta > 0 else "ETH"
    print(f"📡 Сигнал: {lagging} отстаёт — потенциал входа на возврат")
else:
    print("✅ Корреляция стабильна — сигнала нет")

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

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

📌"Rolling Correlation Matrix Viewer" — инструмент анализа взаимосвязей активов во времени🧑‍💻

📌 Суть инструмента:🛠️

Позволяет отслеживать, как меняется корреляция между активами во времени. Это важно для:

- Определения диверсификации

- Поиска новых пар для парного трейдинга

- Обнаружения "разрывов" в синхронности

🔧 Что делает:💻

- Загружает исторические цены по выбранным активам

- Считает скользящую корреляцию (например, 30 свечей)

- Выводит матрицу корреляций, которую можно анализировать или визуализировать

import ccxt
import pandas as pd

exchange = ccxt.binance()
symbols = ['BTC/USDT', 'ETH/USDT', 'BNB/USDT', 'SOL/USDT']
timeframe = '1h'
window = 30

price_data = {}

# Получаем данные
for symbol in symbols:
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=window + 10)
    df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
    price_data[symbol] = df["close"].pct_change().dropna().reset_index(drop=True)

# Собираем DataFrame с доходностями
returns_df = pd.DataFrame(price_data)

# Считаем корреляционную матрицу
corr_matrix = returns_df.corr()

print("🔍 Текущая корреляция между активами:")
print(corr_matrix.round(2))

#инструмент

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

📌"Bid-Ask Flip Detector" — стратегия на смену агрессора в стакане🧑‍💻

📌 Суть стратегии:🛠️

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

✅ Условия стратегии:🚀

1. В течение X секунд/циклов наблюдается рост объёма на стороне bid

2. Затем происходит резкий сдвиг: крупный ask появляется, а bid исчезает

3. Это "flip" — сигнал на вход в сторону нового агрессора

import ccxt
import time

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

# Порог объёма для фиксации "flip"
threshold = 1.5  # x раз превышение объёма

def get_order_book_snapshot(symbol):
    order_book = exchange.fetch_order_book(symbol, limit=5)
    top_bid = order_book['bids'][0][1] if order_book['bids'] else 0
    top_ask = order_book['asks'][0][1] if order_book['asks'] else 0
    return top_bid, top_ask

last_bid, last_ask = get_order_book_snapshot(symbol)

while True:
    try:
        current_bid, current_ask = get_order_book_snapshot(symbol)

        bid_flip = current_bid < last_bid / threshold and current_ask > last_ask * threshold
        ask_flip = current_ask < last_ask / threshold and current_bid > last_bid * threshold

        if bid_flip:
            print("🔴 Flip detected: BID -> ASK dominance. Возможный СИГНАЛ НА ШОРТ.")
        elif ask_flip:
            print("🟢 Flip detected: ASK -> BID dominance. Возможный СИГНАЛ НА ЛОНГ.")
        else:
            print("⏳ Наблюдение... Нет резкой смены доминирования.")

        last_bid, last_ask = current_bid, current_ask
        time.sleep(5)

    except Exception as e:
        print(f"⚠️ Ошибка: {e}")
        time.sleep(10)

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

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

📌"Candle Range Classifier" — классификатор силы свечей🧑‍💻

📌 Суть инструмента:🛠️

Оценивает каждую свечу не просто как "рост" или "падение", а по её реальной энергии — размеру тела, теней, объёму (если есть), и относительному положению.
Помогает фильтровать фейковые импульсы, находить сильные свечи пробоя или слабые разворотные.

🔧 Что делает:

Сканирует последние свечи по выбранной паре

Вычисляет:🧨

- соотношение тела к общему диапазону

- силу закрытия (ближе к high или low)

- наличие хвостов (поглощения, до́жи)

Классифицирует:💻

- strong bullish, weak bullish,

- strong bearish, weak bearish,

- indecision / doji

import ccxt
import pandas as pd

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

ohlcv = exchange.fetch_ohlcv(symbol, '15m', limit=2)
candle = ohlcv[-2]  # предпоследняя свеча

open_, high, low, close = candle[1:5]
body = abs(close - open_)
range_ = high - low
upper_wick = high - max(open_, close)
lower_wick = min(open_, close) - low

body_ratio = body / range_ if range_ else 0

# Классификация
if body_ratio > 0.6:
    if close > open_:
        label = "strong bullish"
    else:
        label = "strong bearish"
elif body_ratio < 0.2:
    label = "indecision / doji"
else:
    label = "weak bullish" if close > open_ else "weak bearish"

print(f"🕯️ Свеча: {label.upper()} | Body: {body_ratio:.2f}")

#инструмент

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

📌"Volatility Snapshot Tool" — монитор волатильности по активам🧑‍💻

📌 Суть инструмента:🛠️

Инструмент показывает текущую реальную (реализованную) волатильность по выбранным активам — позволяет:

- Определить, какие пары сейчас самые активные

- Избегать торговли в слишком "глухих" зонах

- Найти возможности для волатильных стратегий (скальпинг, breakout)

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

- Скачивает свечи за последние X минут или часов

- Вычисляет реализованную волатильность (на основе std отклонения log-доходностей или ATR)

- Сортирует пары по волатильности

- Показывает топ-движущихся активов прямо в консоли

import ccxt
import pandas as pd
import numpy as np

symbols = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'LTC/USDT']
exchange = ccxt.binance()
volatility_data = {}

for symbol in symbols:
    ohlcv = exchange.fetch_ohlcv(symbol, '5m', limit=50)
    df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
    df['log_return'] = np.log(df['close'] / df['close'].shift(1))
    realized_vol = df['log_return'].std() * np.sqrt(12)  # за 1 час
    volatility_data[symbol] = realized_vol

sorted_vol = dict(sorted(volatility_data.items(), key=lambda x: x[1], reverse=True))

print("📊 ТОП активов по реализованной волатильности (час):")
for symbol, vol in sorted_vol.items():
    print(f"{symbol}: {vol:.4f}")

#инструмент

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

📌"Anchored VWAP Shift Strategy" стратегия на смещение точки равновесия🧑‍💻

📌 Суть стратегии:🛠️

Вместо классического VWAP (который начинается с открытия дня), используется Anchored VWAP — привязка средней взвешенной цены к определённому событию:

- локальному минимуму

- сильному импульсу

- началу недели и т.п.

> Когда цена уходит выше или ниже Anchored VWAP и удерживается, это сигнал о смене баланса между покупателями и продавцами.

✅ Условия стратегии:🚀

1. Зафиксировать ключевую точку привязки (например, лоу недели)

2. Рассчитать Anchored VWAP с этого момента

3. Когда цена пробивает VWAP и держится выше N свечей → сигнал на вход

4. Тейк по ATR / R:R, стоп — за точку пробоя

import ccxt
import pandas as pd

symbol = "BTC/USDT"
exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv(symbol, '15m', limit=100)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
df["typical_price"] = (df["high"] + df["low"] + df["close"]) / 3

# Привязываем к лоу определённой свечи
anchor_index = df["low"].idxmin()
anchored = df.iloc[anchor_index:]

anchored["cum_vol"] = anchored["volume"].cumsum()
anchored["cum_vwap"] = (anchored["typical_price"] * anchored["volume"]).cumsum() / anchored["cum_vol"]

vwap = anchored["cum_vwap"].iloc[-1]
current_price = df["close"].iloc[-1]

if current_price > vwap:
    print(f"🟢 Цена выше Anchored VWAP ({vwap:.2f}) — сигнал на ЛОНГ")
else:
    print(f"🔴 Цена ниже Anchored VWAP ({vwap:.2f}) — сигнал на ШОРТ")

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

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

“Position Size Calculator CLI”
Консольный калькулятор размера позиции на Python🧑‍💻

📌 Описание🧨

Инструмент помогает рассчитать размер позиции на основе:

- величины капитала

- процента риска

- уровня входа и стоп-лосса

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

def calculate_position_size(account_balance, risk_percent, entry_price, stop_price):
    risk_amount = account_balance * (risk_percent / 100)
    stop_loss_per_unit = abs(entry_price - stop_price)
    if stop_loss_per_unit == 0:
        raise ValueError("Stop loss слишком близко к входу")
    position_size = risk_amount / stop_loss_per_unit
    return position_size

def main():
    print("=== Position Size Calculator ===")
    try:
        balance = float(input("Введите капитал ($): "))
        risk_pct = float(input("Риск на сделку (%): "))
        entry = float(input("Цена входа: "))
        stop = float(input("Цена стоп-лосса: "))

        size = calculate_position_size(balance, risk_pct, entry, stop)
        print(f"\n🔢 Рекомендуемый размер позиции: {size:.4f} единиц актива")
    except Exception as e:
        print(f"[Ошибка] {e}")

if __name__ == "__main__":
    main()

#инструмент

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

📌"Timebox Scalping Strategy" – стратегия торговли по временным блокам🧑‍💻

🧠 Описание:🧨

Эта стратегия разбивает день на фиксированные временные блоки (например, по 15 или 30 минут) и ищет локальный экстремум в прошлом блоке, а затем торгует на отскок или пробой в следующем блоке. Идея в том, что определённые паттерны повторяются в пределах фиксированных временных структур (особенно во флэте или на переломах).

⚙️ Условия входа:🚀

ЛОНГ:

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

- Текущая цена пробила high прошлого блока.

- Объём выше среднего за 5 блоков.

ШОРТ:

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

- Текущая цена пробила low прошлого блока.

- Объём выше среднего за 5 блоков.

import time
import pandas as pd
import requests

# Настройки
symbol = 'BTCUSDT'
interval = '15m'
limit = 50
api_url = f"https://api.binance.com/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}"

def get_klines():
    response = requests.get(api_url)
    data = response.json()
    df = pd.DataFrame(data, columns=[
        'timestamp', 'open', 'high', 'low', 'close', 'volume',
        'close_time', 'quote_asset_volume', 'trades',
        'taker_buy_base', 'taker_buy_quote', 'ignore'
    ])
    df = df.astype(float)
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    return df

def check_signal(df):
    # Берём последние 2 блока
    prev_block = df.iloc[-2]
    curr_block = df.iloc[-1]

    prev_high = prev_block['high']
    prev_low = prev_block['low']
    curr_price = curr_block['close']
    curr_vol = curr_block['volume']

    avg_vol = df['volume'].iloc[-6:-1].mean()

    if curr_price > prev_high and curr_vol > avg_vol:
        print(f"[ALERT] LONG сигнал: цена пробила хай блока | Цена: {curr_price}")
    elif curr_price < prev_low and curr_vol > avg_vol:
        print(f"[ALERT] SHORT сигнал: цена пробила лоу блока | Цена: {curr_price}")
    else:
        print("[INFO] Нет сигнала.")

def run_monitor():
    print("⏳ Запуск Timebox Scalping Monitor...")
    while True:
        try:
            df = get_klines()
            check_signal(df)
            time.sleep(60)  # каждые 60 сек, чтобы не спамить
        except Exception as e:
            print(f"[ERROR] {e}")
            time.sleep(30)

if __name__ == "__main__":
    run_monitor()

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

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

📌Стратегия: Price Ladder Climb🧑‍💻

“Лестница цены” — стратегия по поэтапному закреплению на ключевых уровнях

📌 Идея🛠️

Представь, что цена — это альпинист, поднимающийся по склону. Каждый уровень сопротивления — это новая ступень. Если альпинист успешно закрепляется (удерживается) на ступени в течение N минут — он готов лезть выше.

⚙️ Правила входа💸

1. Обнаружение "ступени": цена пробивает локальное сопротивление (high за последние X минут).

2. Закрепление: если в течение следующих 3 свечей цена не опускается ниже пробитого уровня — это "закрепление".

3. Вход: после закрепления — вход в направлении пробоя.

4. Выход: по цели (например, фиксированное расстояние или следующая ступень).

import requests
import time

symbol = "BTCUSDT"
interval = "1m"
limit = 10

def get_ohlcv():
    url = "https://api.binance.com/api/v3/klines"
    params = {"symbol": symbol, "interval": interval, "limit": limit}
    res = requests.get(url, params=params).json()
    return [[float(x[2]), float(x[3]), float(x[4])] for x in res]  # high, low, close

print("[*] Лестница цены: начало мониторинга")

while True:
    try:
        data = get_ohlcv()
        highs = [x[0] for x in data[:-3]]  # до последних 3 свечей
        recent = data[-3:]
        level = max(highs)

        above_level = all(x[2] > level for x in recent)
        if above_level:
            print(f"[!] Закрепление выше {level:.2f}, возможный ЛОНГ")
        else:
            print(f"[ ] Нет закрепления: последний high={level:.2f}")

        time.sleep(30)
    except Exception as e:
        print("[Ошибка]", e)
        time.sleep(10)

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

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

📌 Инструмент: Multi-Asset Correlation Tracker 🧑‍💻

Онлайн-монитор корреляций между активами с динамикой за последние N часов/дней

📌 Описание 🧨

Этот инструмент отслеживает корреляции между активами (например, BTC/ETH, BTC/SP500, золото/фондовые индексы и т.д.) и показывает, как они меняются во времени. Это помогает:

- Понимать перекосы в поведении активов

- Выявлять дивергенции

- Искать неэффективности и сигналы для парного трейдинга

- Адаптировать стратегию под текущую макро-среду

import requests
import numpy as np
import pandas as pd
from scipy.stats import pearsonr
import time

symbols = ["BTCUSDT", "ETHUSDT"]
interval = "1h"
limit = 100

def get_closes(symbol):
    url = "https://api.binance.com/api/v3/klines"
    params = {"symbol": symbol, "interval": interval, "limit": limit}
    data = requests.get(url, params=params).json()
    return [float(x[4]) for x in data]

print("[*] Запуск мониторинга корреляций...")

while True:
    try:
        series = {s: get_closes(s) for s in symbols}
        df = pd.DataFrame(series)
        corr = df.corr().iloc[0, 1]
        print(f"[📈] Корреляция {symbols[0]} / {symbols[1]} = {corr:.3f}")
        time.sleep(3600)
    except Exception as e:
        print("[Ошибка]", e)
        time.sleep(60)

#инструмент

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

📌Multi-Timeframe Range Confluence - вход внутри локального флета, если он совпадает с глобальной зоной накопления 🧑‍💻

🔍 Описание 🛠️

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

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

- Цена находится внутри узкого диапазона (5-мин ТФ, например, последние 20 свечей)

- Диапазон на 1H таймфрейме также плоский (узкий High-Low, нет тренда)

- Уровни локального диапазона совпадают с границами глобального

- Пробой локального диапазона вверх → Лонг

- Пробой вниз → Шорт

import requests
import pandas as pd
import time

def get_ohlcv(symbol="BTCUSDT", interval="5m", limit=100):
    url = "https://api.binance.com/api/v3/klines"
    params = {"symbol": symbol, "interval": interval, "limit": limit}
    data = requests.get(url, params=params).json()
    df = pd.DataFrame(data, columns=[
        "timestamp", "open", "high", "low", "close", "volume", 
        "close_time", "quote_asset_volume", "num_trades", 
        "taker_buy_base", "taker_buy_quote", "ignore"
    ])
    df = df.astype(float)
    return df

def detect_range(df, deviation=0.005):
    high = df["high"].max()
    low = df["low"].min()
    spread = (high - low) / low
    return (spread < deviation), low, high

symbol = "BTCUSDT"
while True:
    try:
        df_5m = get_ohlcv(symbol, "5m", 20)
        df_1h = get_ohlcv(symbol, "1h", 24)

        in_local_range, l_low, l_high = detect_range(df_5m)
        in_global_range, g_low, g_high = detect_range(df_1h)

        if in_local_range and in_global_range:
            last_price = df_5m["close"].iloc[-1]
            if last_price > l_high:
                print(f"[🚀 ЛОНГ] Пробой вверх из локального + глобального флета @ {last_price:.2f}")
            elif last_price < l_low:
                print(f"[🔻 ШОРТ] Пробой вниз из локального + глобального флета @ {last_price:.2f}")
            else:
                print("[⏳] Внутри диапазона, ждём пробоя...")
        else:
            print("[…] Нет совпадения диапазонов")

        time.sleep(300)
    except Exception as e:
        print("[Ошибка]", e)
        time.sleep(60)

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

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

📌 Stop-Loss Clustering Radar

Индикатор потенциальных зон скопления стоп-лоссов по ценовому поведению🧑‍💻

📋 Описание 🛠️

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

Это зоны, где можно ожидать:💸

- резких пробоев ("stop hunt"),

- ликвидности для крупных игроков,

- возможно, разворотов после снятия стопов.

Подходит для тех, кто торгует внутри дня или для контртрендовых подходов🚀

import ccxt
import pandas as pd
import time
from ta.volatility import average_true_range

symbol = 'BTC/USDT'
timeframe = '5m'
lookback_bars = 50
impulse_threshold = 1.5  # процент движения для "импульса"
alert_distance = 0.3  # процент отклонения от стоп-уровня

exchange = ccxt.binance()
exchange.load_markets()

def fetch_ohlcv():
    data = exchange.fetch_ohlcv(symbol, timeframe, limit=lookback_bars)
    df = pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    return df

def detect_stop_clusters(df):
    clusters = []
    for i in range(3, len(df) - 3):
        low = df['low'][i]
        high = df['high'][i]
        
        if low < df['low'][i-1] and low < df['low'][i+1]:
            drop = ((df['close'][i+3] - low) / low) * 100
            if drop > impulse_threshold:
                clusters.append(('support', low))
        
        if high > df['high'][i-1] and high > df['high'][i+1]:
            jump = ((high - df['close'][i+3]) / high) * 100
            if jump > impulse_threshold:
                clusters.append(('resistance', high))
    
    return clusters

def alert_on_proximity(current_price, clusters):
    for zone_type, level in clusters:
        distance = abs(current_price - level) / level * 100
        if distance < alert_distance:
            print(f"[ALERT] Price near {zone_type.upper()} STOP CLUSTER at {level:.2f}")

def main():
    print("Starting Stop-Loss Cluster Radar...\n")
    while True:
        try:
            df = fetch_ohlcv()
            clusters = detect_stop_clusters(df)
            current_price = df['close'].iloc[-1]
            alert_on_proximity(current_price, clusters)
            time.sleep(60)
        except Exception as e:
            print(f"Error: {e}")
            time.sleep(30)

if __name__ == "__main__":
    main()

#инструмент

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

📌 Time-of-Day Volatility Radar🧑‍💻

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

🎯 Зачем это нужно трейдеру:💸

- Избегать торговли в “мёртвые часы”

- Настраивать стратегии по времени (например, активировать бота только в высоко-волатильные окна)

- Подбирать лучшие часы под стратегии breakout/impulse

🧪 Что делает: 💻

1. Скачивает исторические свечи (например, 5m, 15m).

2. Группирует их по времени суток (например, 09:00, 09:05, 09:10…).

3. Считает средний диапазон свечи (high - low) по каждому временному окну.

4. Выводит топ-часов/минут по волатильности.

import ccxt
import pandas as pd
import datetime

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '5m'
limit = 2000

def fetch_data():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
    df['tod'] = df['datetime'].dt.strftime('%H:%M')
    df['range'] = df['high'] - df['low']
    return df

def analyze_volatility(df):
    grouped = df.groupby('tod')['range'].mean().sort_values(ascending=False)
    print("⏱️ Top volatile times of day (UTC):\n")
    for time, rng in grouped.head(10).items():
        print(f"{time}: Avg Range = {rng:.2f}")

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

#инструмент

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

📌Market Memory Echo Strategy🧑‍💻

Концепт: рынок имеет «память» на ценовые уровни, где ранее происходили резкие развороты. Если в течение дня цена подходит к этим «эхо-зонам», вероятность реакции (разворота или импульса) повышается. Стратегия ищет такие зоны по локальным экстремумам предыдущих дней и следит за приближением к ним🛠️

🧠 Логика: 🚀

1. Берём исторические 5-15 минутные свечи за 2–3 дня.

2. Ищем экстремумы, сопровождавшиеся большими объёмами или свечами с тенями.

3. Сохраняем уровни с подтверждённой реакцией как echo levels.

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

5. Если цена подходит близко — алерт, сигнал на потенциальную реакцию рынка.

import ccxt
import pandas as pd
import time

exchange = ccxt.binance()
symbol = 'ETH/USDT'
timeframe = '15m'
echo_threshold_pct = 0.3  # % расстояния до уровня
history_days = 3
bars_per_day = int(24 * 60 / 15)  # для 15m

def fetch_history():
    total_bars = bars_per_day * history_days
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=total_bars)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    return df

def find_echo_levels(df):
    echo_levels = []
    for i in range(2, len(df)-2):
        if df['low'][i] < df['low'][i-1] and df['low'][i] < df['low'][i+1]:
            if df['close'][i+1] > df['close'][i] * 1.01:
                echo_levels.append(('support', df['low'][i]))
        if df['high'][i] > df['high'][i-1] and df['high'][i] > df['high'][i+1]:
            if df['close'][i+1] < df['close'][i] * 0.99:
                echo_levels.append(('resistance', df['high'][i]))
    return echo_levels

def monitor_live(echo_levels):
    print("Echo Levels Strategy active...\n")
    while True:
        try:
            live = exchange.fetch_ohlcv(symbol, timeframe, limit=5)
            last_close = live[-1][4]
            for level_type, price in echo_levels:
                distance = abs(last_close - price) / price * 100
                if distance < echo_threshold_pct:
                    print(f"[ALERT] Price {last_close:.2f} near {level_type.upper()} ECHO LEVEL at {price:.2f}")
            time.sleep(60)
        except Exception as e:
            print(f"Error: {e}")
            time.sleep(30)

if __name__ == '__main__':
    df_hist = fetch_history()
    echoes = find_echo_levels(df_hist)
    monitor_live(echoes)

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

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

📌"Latency Arbitrage Simulator"🧑‍💻

Идея:🚀

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

🔍 Принцип:💸

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

2. Если, например, Binance Futures отстаёт от Bybit, это может быть признаком подготовки к движению.

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

import time
import requests

symbol = "BTCUSDT"
interval = 5  # Интервал между запросами в секундах
iterations = 20

binance_url = f"https://api.binance.com/api/v3/ticker/price?symbol={symbol}"
bybit_url = "https://api.bybit.com/v2/public/tickers"

def get_binance_price():
    try:
        r = requests.get(binance_url)
        return float(r.json()['price'])
    except Exception as e:
        print("Binance error:", e)
        return None

def get_bybit_price():
    try:
        r = requests.get(bybit_url)
        for ticker in r.json()['result']:
            if ticker['symbol'] == symbol:
                return float(ticker['last_price'])
    except Exception as e:
        print("Bybit error:", e)
        return None

for i in range(iterations):
    print(f"\nИтерация {i+1}")

    binance_price = get_binance_price()
    bybit_price = get_bybit_price()

    if binance_price and bybit_price:
        diff = binance_price - bybit_price
        print(f"Binance: {binance_price}, Bybit: {bybit_price}, Разница: {diff:.2f}")
    else:
        print("Ошибка получения данных")

    time.sleep(interval)

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

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

📌 Market Microstructure Analyzer🧑‍💻

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

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

- Считает, как долго цена "залипает" на каждом уровне (time at price).

- Фиксирует аптики и даунтики — сколько подряд свечей в одном направлении.

- Обнаруживает микроимпульсы — быстрые движения на низких таймфреймах.

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

📄 Возможности применения:💸

Вход на раннем этапе импульса.

- Выход при "залипании" цены.

- Фильтрация флетов перед пробоем.

- Построение микроуровней поддержки/сопротивления.

import requests
import time
from collections import deque

# Пара и настройки
SYMBOL = 'BTCUSDT'
INTERVAL = 3  # сек между запросами
IMPULSE_LENGTH = 3
STUCK_TIME = 60  # сек без движения

price_history = deque(maxlen=IMPULSE_LENGTH + 1)
last_price = None
last_move_time = time.time()

def get_price(symbol):
    url = f'https://api.binance.com/api/v3/ticker/price?symbol={symbol}'
    try:
        res = requests.get(url, timeout=5)
        res.raise_for_status()
        return float(res.json()['price'])
    except Exception as e:
        print(f"⛔ Error fetching price: {e}")
        return None

def detect_impulse(prices):
    if len(prices) < IMPULSE_LENGTH + 1:
        return None
    diffs = [prices[i+1] - prices[i] for i in range(len(prices)-1)]
    if all(d > 0 for d in diffs):
        return "📈 Micro Uptrend"
    elif all(d < 0 for d in diffs):
        return "📉 Micro Downtrend"
    return None

print(f"🚀 Monitoring {SYMBOL} for micro impulses...\n")

while True:
    price = get_price(SYMBOL)
    if price is None:
        time.sleep(INTERVAL)
        continue

    price_history.append(price)
    now = time.time()

    if last_price is not None and price == last_price:
        if now - last_move_time > STUCK_TIME:
            print(f"[{time.strftime('%H:%M:%S')}] ⚠️ Price stuck at {price} for {int(now - last_move_time)}s")
    else:
        last_move_time = now

    impulse = detect_impulse(list(price_history))
    if impulse:
        print(f"[{time.strftime('%H:%M:%S')}] {impulse} → {price_history}")

    last_price = price
    time.sleep(INTERVAL)

#инструмент

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

📌 Trading Strategy Navigator – подсказчик подходящих стратегий под рыночный режим🧑‍💻

📌 Описание:🛠️

Инструмент определяет текущий рыночный режим (тренд, флэт, высокая/низкая волатильность и пр.) и автоматически подбирает подходящие типы стратегий для трейдера.

🧠 Как работает:

1. Анализ рыночного режима:

Направление и угол наклона скользящей средней (EMA)

Уровень волатильности (ATR)

Коэффициент автокорреляции

2. Классификация:

Тренд вверх → рекомендованы трендовые стратегии

Флэт + низкая волатильность → стратегии отскока и границ диапазона

Импульс → breakout и объёмные стратегии

3. Вывод рекомендаций: список стратегий под текущую фазу.

import requests
import numpy as np

def get_prices(symbol="BTCUSDT", interval="15m", limit=100):
    url = f"https://api.binance.com/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}"
    data = requests.get(url).json()
    return [float(k[4]) for k in data]  # Close prices

def detect_market_regime(prices):
    ema = np.mean(prices[-20:])
    slope = ema - np.mean(prices[-40:-20])
    atr = np.mean([abs(prices[i] - prices[i-1]) for i in range(1, len(prices))])

    if slope > 0.1 and atr > 10:
        return "Восходящий тренд", ["Следование за трендом", "Пирамидинг", "EMA pullback"]
    elif slope < -0.1 and atr > 10:
        return "Нисходящий тренд", ["Шорт по тренду", "Breakdown", "SuperTrend"]
    elif atr < 5:
        return "Флэт", ["Range Reversion", "Bollinger bounce", "Mean reversion"]
    else:
        return "Импульс / Смена режима", ["Breakout", "Volume spike", "Delta surge"]

prices = get_prices()
regime, strategies = detect_market_regime(prices)

print(f"📈 Режим рынка: {regime}")
print("🔍 Рекомендуемые стратегии:")
for strat in strategies:
    print("–", strat)

#инструмент

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