📌"Market Heatmap Engine" — движок для построения в реальном времени тепловой карты крипторынка с учётом:

- силы тренда,

- притока/оттока ликвидности,

- скорости изменения цены.

💡 Главная идея — видеть рынок как "карту температур": красные сектора — перегретые активы с высокой вероятностью коррекции, зелёные — активы в фазе накопления, готовые к импульсу.

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

1. Берём список монет с Binance (например, топ-50 по объёму).

2. Для каждой монеты:

- рассчитываем скорость изменения цены (momentum);

- измеряем изменение глубины стакана за последние секунды;

- оцениваем соотношение рыночных ордеров к лимитным.

3. На основе этих метрик формируем индекс перегрева от -100 до +100.

4. Отдаём данные в консоль или в браузер как тепловую карту.

import ccxt
import time
import numpy as np
from prettytable import PrettyTable

exchange = ccxt.binance()

symbols = [m['symbol'] for m in exchange.fetch_markets() if '/USDT' in m['symbol']][:15]
prev_prices = {}
prev_bids = {}
prev_asks = {}

def get_heatmap():
    table = PrettyTable(["Symbol", "Momentum", "Liquidity Δ", "Heat Index"])
    for symbol in symbols:
        try:
            # Цена
            ticker = exchange.fetch_ticker(symbol)
            price = ticker['last']
            momentum = 0
            if symbol in prev_prices:
                momentum = (price - prev_prices[symbol]) / prev_prices[symbol] * 100
            
            # Ликвидность
            orderbook = exchange.fetch_order_book(symbol, limit=10)
            bids_vol = sum([v for _, v in orderbook['bids']])
            asks_vol = sum([v for _, v in orderbook['asks']])
            
            liquidity_change = 0
            if symbol in prev_bids and symbol in prev_asks:
                prev_liq = prev_bids[symbol] + prev_asks[symbol]
                curr_liq = bids_vol + asks_vol
                liquidity_change = (curr_liq - prev_liq) / prev_liq * 100 if prev_liq != 0 else 0
            
            # Индекс перегрева
            heat_index = momentum * 0.7 + liquidity_change * 0.3
            
            table.add_row([
                symbol,
                f"{momentum:+.2f}%",
                f"{liquidity_change:+.2f}%",
                f"{heat_index:+.2f}"
            ])
            
            # Сохраняем предыдущие значения
            prev_prices[symbol] = price
            prev_bids[symbol] = bids_vol
            prev_asks[symbol] = asks_vol
        
        except Exception as e:
            pass
    
    print("\033c", end="")  # Очистка консоли
    print(table)

while True:
    get_heatmap()
    time.sleep(3)

#инструмент

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

📌"Радар крупных скрытых сделок (Stealth Trade Radar)".🧑‍💻

💡 Суть:🛠️

Большие игроки часто маскируют свои входы и выходы:

- разбивают ордера на мелкие,

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

Этот инструмент будет отслеживать поток сделок (trades) и искать аномальные паттерны, указывающие на скрытое накопление или сброс позиции.

📊 Логика:🧨

1. Получаем поток сделок с биржи через ccxt (fetch_trades).

2. Группируем их по направлению (покупка/продажа).

3. Считаем суммарный объём за последние N секунд.

4. Если за короткое время прошло аномально много мелких сделок в одну сторону — фиксируем сигнал.

import ccxt
import time
from collections import deque

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

WINDOW_SEC = 5       # окно анализа в секундах
VOLUME_THRESHOLD = 50  # суммарный объем для сигнала
MAX_TRADE_SIZE = 0.1  # "мелкая" сделка

buy_trades = deque()
sell_trades = deque()

def check_stealth_trades():
    global buy_trades, sell_trades

    trades = exchange.fetch_trades(symbol)
    now = time.time()

    for t in trades:
        ts = t['timestamp'] / 1000
        size = t['amount']
        side = t['side']

        # Учитываем только "мелкие" сделки
        if size <= MAX_TRADE_SIZE:
            if side == 'buy':
                buy_trades.append((ts, size))
            else:
                sell_trades.append((ts, size))

    # Чистим старые сделки
    while buy_trades and now - buy_trades[0][0] > WINDOW_SEC:
        buy_trades.popleft()
    while sell_trades and now - sell_trades[0][0] > WINDOW_SEC:
        sell_trades.popleft()

    buy_volume = sum(s for _, s in buy_trades)
    sell_volume = sum(s for _, s in sell_trades)

    if buy_volume > VOLUME_THRESHOLD:
        print(f"🟢 Обнаружено скрытое накопление! Объем {buy_volume:.4f} {symbol}")
    elif sell_volume > VOLUME_THRESHOLD:
        print(f"🔴 Обнаружен скрытый сброс! Объем {sell_volume:.4f} {symbol}")
    else:
        print(f"[{symbol}] Активных скрытых сделок нет.")

while True:
    try:
        check_stealth_trades()
        time.sleep(2)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(5)

#инструмент

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

📌"Сдвиг ликвидности + фейковый пробой" (Liquidity Shift & Trap Breakout).🧑‍💻

💡 Суть стратегии: 🚀

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

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

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

2. Проверяем, что после пробоя тик-флоу сделок идёт в противоположную сторону.

3. Входим в позицию против пробоя (ловим возврат цены в диапазон).

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

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
limit = 20

prev_price = None
prev_bids = []
prev_asks = []

def detect_fake_breakout():
    global prev_price, prev_bids, prev_asks
    ob = exchange.fetch_order_book(symbol, limit=limit)
    ticker = exchange.fetch_ticker(symbol)
    price = ticker['last']

    bids = ob['bids']
    asks = ob['asks']

    top_bid_volume = sum([v for _, v in bids[:3]])
    top_ask_volume = sum([v for _, v in asks[:3]])

    signal = None

    if prev_price:
        # Пробой вверх + внезапное появление крупной стены продаж
        if price > prev_price * 1.002 and top_ask_volume > sum(prev_asks) * 1.5:
            signal = "🔴 Фейковый пробой вверх! Возможен возврат вниз."
        # Пробой вниз + внезапная крупная поддержка
        elif price < prev_price * 0.998 and top_bid_volume > sum(prev_bids) * 1.5:
            signal = "🟢 Фейковый пробой вниз! Возможен возврат вверх."

    prev_price = price
    prev_bids = [v for _, v in bids[:3]]
    prev_asks = [v for _, v in asks[:3]]

    if signal:
        print(f"[{symbol}] {signal}")
    else:
        print(f"[{symbol}] Нет сигналов.")

while True:
    try:
        detect_fake_breakout()
        time.sleep(2)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(5)

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

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

📌"Дивергенция импульса против ликвидности" (Impulse vs Liquidity Divergence).🧑‍💻

💡 Суть:🛠️

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

Часто перед разворотом:

- цена продолжает расти,

- но ликвидность на стороне покупателей падает (и наоборот).

📊 Логика:🚀

1. Каждую секунду получаем:

- последние сделки,

- стакан ордеров (order book).

2. Считаем импульс (цена vs предыдущая цена).

3. Считаем изменение ликвидности на стороне покупателей и продавцов.

4. Если цена растёт, а ликвидность покупателей падает → сигнал на шорт.

Если цена падает, а ликвидность продавцов падает → сигнал на лонг.

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
depth_limit = 20

prev_price = None
prev_bids_liq = None
prev_asks_liq = None

def get_liquidity(orderbook_side):
    return sum(amount for price, amount in orderbook_side)

while True:
    try:
        ticker = exchange.fetch_ticker(symbol)
        orderbook = exchange.fetch_order_book(symbol, depth_limit)

        last_price = ticker['last']
        bids_liq = get_liquidity(orderbook['bids'])
        asks_liq = get_liquidity(orderbook['asks'])

        if prev_price is not None:
            price_change = last_price - prev_price
            bids_change = bids_liq - prev_bids_liq if prev_bids_liq is not None else 0
            asks_change = asks_liq - prev_asks_liq if prev_asks_liq is not None else 0

            if price_change > 0 and bids_change < 0:
                print(f"🔴 {symbol}: Цена растёт, но покупатели уходят → возможный разворот вниз.")
            elif price_change < 0 and asks_change < 0:
                print(f"🟢 {symbol}: Цена падает, но продавцы уходят → возможный разворот вверх.")
            else:
                print(f"{symbol}: Нет дивергенций. Цена: {last_price}")

        prev_price = last_price
        prev_bids_liq = bids_liq
        prev_asks_liq = asks_liq

        time.sleep(2)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(5)

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

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

📌"Детектор скрытой агрессии в ордербуке".🧑‍💻

💡 Суть инструмента:🚀

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

📊 Как работает:🧨

1. Получаем стакан и ленточку сделок в реальном времени.

2. Отслеживаем, что по определённой цене:

- объём не уменьшается заметно, хотя идут сделки;

- ордера на этой цене постоянно восстанавливаются.

3. Если это происходит многократно → сигнал о скрытом крупном игроке.

📌 Как использовать:💸

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

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

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

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
depth_limit = 20

iceberg_tracker = {}
iceberg_threshold = 5  # число "восстановлений" ордера для сигнала

while True:
    try:
        orderbook = exchange.fetch_order_book(symbol, depth_limit)
        bids = orderbook['bids'][:5]
        asks = orderbook['asks'][:5]

        # Анализ обеих сторон
        for side_name, side_data in [('BUY', bids), ('SELL', asks)]:
            for price, amount in side_data:
                key = (side_name, price)

                # Если цена уже была в трекере
                if key in iceberg_tracker:
                    prev_amount, restorations = iceberg_tracker[key]
                    if amount >= prev_amount * 0.95:  
                        iceberg_tracker[key] = (amount, restorations + 1)
                    else:
                        iceberg_tracker[key] = (amount, restorations)
                else:
                    iceberg_tracker[key] = (amount, 0)

                # Сигнал, если много "восстановлений"
                if iceberg_tracker[key][1] >= iceberg_threshold:
                    print(f"🧊 Iceberg {side_name} {symbol} @ {price}, объём {amount}")
                    iceberg_tracker[key] = (amount, 0)

        time.sleep(1)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(5)

#инструмент

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

📌"Импульсная ловушка на ликвидности" (Liquidity Sweep Reversal).🧑‍💻

📌 Идея стратегии:🧨

Большие игроки часто выносят стопы мелких трейдеров, чтобы собрать ликвидность, а потом разворачивают цену.
Мы можем ловить обратное движение после ложного пробоя ключевого уровня.

🔍 Логика:🛠️

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

2. Если цена резко пробивает экстремум и возвращается обратно в диапазон:

Это значит, что был сбор ликвидности (стопы) и вероятен разворот.

3. Подтверждение — свеча закрывается внутри диапазона.

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

import ccxt
import pandas as pd
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '5m'
lookback = 20  # кол-во свечей для поиска хай/лоу

def get_ohlcv():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=lookback + 3)
    df = pd.DataFrame(ohlcv, columns=['time', 'open', 'high', 'low', 'close', 'volume'])
    return df

while True:
    try:
        df = get_ohlcv()

        recent_high = df['high'][:-1].max()
        recent_low = df['low'][:-1].min()
        last_candle = df.iloc[-1]

        # Ложный пробой вверх
        if last_candle['high'] > recent_high and last_candle['close'] < recent_high:
            print(f"🔻 SHORT сигнал на {symbol} @ {last_candle['close']} (ложный пробой вверх)")

        # Ложный пробой вниз
        if last_candle['low'] < recent_low and last_candle['close'] > recent_low:
            print(f"🔺 LONG сигнал на {symbol} @ {last_candle['close']} (ложный пробой вниз)")

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

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

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

📌DECS: Dynamic Execution Cost & Slippage Simulator🧑‍💻

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

Для любой пары и любого объёма оценивает ожидаемые затраты исполнения (спред + проскальзывание) по текущему стакану и имитирует разные варианты разбиения (slicing) ордера: один маркет, несколько частей (VWAP-подход), лимитные отложки. В выводе — рекомендуемый способ разбивки (сколько частей), ожидаемая средняя цена исполнения и прогнозируемое проскальзывание в валюте и процентах. Логирует результаты в CSV.

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

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

- Помогает выбрать оптимальный размер частей и стратегию исполнения (market vs slice vs limit).

- Уменьшает рыночное воздействие — критично для профессионалов и алготрейдов.

- Работает через REST (ccxt) — можно запускать в cron/долгоживущем процессе и интегрировать в бэктест/реальный бот.

#инструмент

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

📌Tradeability Heat Score (THS) — сканер «здоровья» рынка🧑‍💻

Идея: для десятков USDT-пар на Binance рассчитать сводный балл торгуемости прямо сейчас и выдать топ-кандидатов.

Учитываем микроструктуру (спред, глубина стакана рядом с мидом), недавнюю волатильность, 24h-объём, концентрацию ликвидности.

Что даёт:💸

- Быстрый ответ на вопрос: что сейчас лучше всего торгуется (для скальпа/интрадей).

- Фильтр мусорных пар: широкие спреды, тонкий стакан, «рваная» волатильность.

- Алерты в консоль при резких изменениях «здоровья».

Как интерпретировать:❗

THS ближе к 1.0 — пара сейчас «здоровая»: узкий спред, плотный стакан возле mid, высокая концентрация ликвидности и адекватная внутридневная вола.

⚠️ Алерты — резкие изменения «здоровья» (±15% по умолчанию): повод пересмотреть приоритеты для входа/исполнения.

#инструмент

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

📌"Импульсная волна с адаптивным стопом" — она сочетает логику поиска импульсных движений и динамическое сопровождение сделки на основе волатильности, что делает её живой и самоподстраивающейся.🧑‍💻

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

1. Фильтр тренда — используем EMA(50) и EMA(200) для определения направления.

EMA(50) выше EMA(200) → рассматриваем только покупки.

EMA(50) ниже EMA(200) → рассматриваем только продажи.

2. Детектор импульса — рост или падение цены за последние N свечей ≥ K%.

Например, рост на ≥ 1.5% за последние 5 свечей.

3. Вход:

Лонг: тренд вверх + импульс вверх.

Шорт: тренд вниз + импульс вниз.

4. Адаптивный стоп-лосс — стоп на основе ATR (среднего истинного диапазона), например 1.5 × ATR.

Стоп-лосс подтягивается за ценой, если она движется в прибыль.

5. Выход:

- При срабатывании стопа.

- Или при развороте тренда (EMA50 пересекла EMA200 обратно).

import ccxt
import pandas as pd
import time

# Настройки
symbol = "BTC/USDT"
timeframe = "5m"
impulse_threshold = 1.5  # %
ema_fast = 50
ema_slow = 200
atr_period = 14

exchange = ccxt.binance()

def fetch_data():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=ema_slow + 50)
    df = pd.DataFrame(ohlcv, columns=["time", "open", "high", "low", "close", "volume"])
    df["time"] = pd.to_datetime(df["time"], unit="ms")
    return df

def ema(series, period):
    return series.ewm(span=period, adjust=False).mean()

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

def check_signals():
    df = fetch_data()
    df["EMA_F"] = ema(df["close"], ema_fast)
    df["EMA_S"] = ema(df["close"], ema_slow)
    df["ATR"] = atr(df, atr_period)

    # Детектор импульса
    df["price_change"] = df["close"].pct_change(5) * 100
    last = df.iloc[-1]

    if last["EMA_F"] > last["EMA_S"] and last["price_change"] >= impulse_threshold:
        sl = last["close"] - last["ATR"] * 1.5
        print(f"[LONG] {symbol} Цена: {last['close']:.2f} | SL: {sl:.2f}")
    elif last["EMA_F"] < last["EMA_S"] and last["price_change"] <= -impulse_threshold:
        sl = last["close"] + last["ATR"] * 1.5
        print(f"[SHORT] {symbol} Цена: {last['close']:.2f} | SL: {sl:.2f}")
    else:
        print("[NO SIGNAL]")

while True:
    try:
        check_signals()
        time.sleep(10)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(10)

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

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

📌“Кластерный импульс + задержка входа”, которая ориентируется на высокую точность входа в импульсные движения, но с использованием контролируемой задержки для фильтрации ложных пробоев.🧑‍💻

Логика стратегии🚀

1. Поиск кластера узкого диапазона (NRB — Narrow Range Bars)

Находим участок из N свечей, где диапазон (High − Low) каждой свечи меньше среднего за период.

Это сигнал о накоплении.

2. Определение направления пробоя

Если цена пробивает верх кластера на объёме выше среднего → потенциал лонга.

Если пробивает вниз → потенциал шорта.

3. Задержка входа (Delay Entry)

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

Это отсеивает большинство ложных выходов.

4. Фильтрация тренда

Используем EMA(50) как фильтр:

Лонг — только если цена выше EMA(50)

Шорт — только если ниже EMA(50)

5. Выход:💸

Тейк профит: 1.5× риск (размер диапазона кластера)

Стоп: за противоположной границей диапазона кластера

import ccxt
import pandas as pd
import time

symbol = 'BTC/USDT'
timeframe = '15m'
cluster_size = 5
delay_bars = 2

exchange = ccxt.binance()

def fetch_ohlcv():
    data = exchange.fetch_ohlcv(symbol, timeframe, limit=200)
    df = pd.DataFrame(data, columns=['time', 'open', 'high', 'low', 'close', 'volume'])
    df['time'] = pd.to_datetime(df['time'], unit='ms')
    return df

def strategy_signals(df):
    signals = []
    df['ema50'] = df['close'].ewm(span=50).mean()

    for i in range(cluster_size, len(df) - delay_bars):
        cluster = df.iloc[i-cluster_size:i]
        avg_range = (cluster['high'] - cluster['low']).mean()
        if all((cluster['high'] - cluster['low']) < avg_range * 1.05):
            high_cluster = cluster['high'].max()
            low_cluster = cluster['low'].min()

            breakout_up = df.iloc[i]['close'] > high_cluster
            breakout_down = df.iloc[i]['close'] < low_cluster

            if breakout_up and df.iloc[i]['close'] > df.iloc[i]['ema50']:
                # Ждем delay_bars
                if all(df.iloc[i+1:i+delay_bars+1]['close'] > high_cluster):
                    signals.append((df.iloc[i+delay_bars]['time'], 'LONG', high_cluster, low_cluster))

            elif breakout_down and df.iloc[i]['close'] < df.iloc[i]['ema50']:
                if all(df.iloc[i+1:i+delay_bars+1]['close'] < low_cluster):
                    signals.append((df.iloc[i+delay_bars]['time'], 'SHORT', high_cluster, low_cluster))

    return signals

if __name__ == "__main__":
    while True:
        df = fetch_ohlcv()
        sigs = strategy_signals(df)
        for s in sigs:
            print(f"{s[0]} | SIGNAL: {s[1]} | Cluster High: {s[2]:.2f} | Cluster Low: {s[3]:.2f}")
        time.sleep(60)

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

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

📌Pre-Trade Risk & Filters Sizer (PTRFS)🧑‍💻

Что делает: до отправки ордера рассчитывает корректный объём позиции из риска (в USDT или % депозита), учитывая биржевые фильтры Binance (минимальный нотионал, шаг цены/лот, точности), стоп-лосс, тейк-профит и комиссии. Выводит готовые параметры сделки (округлённые под правила рынка) и предупреждения, если что-то не проходит.

Полезно для: ручной торговли, полу-авто ботов, автогенерации параметров ордера (без его размещения).

#инструмент

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

📌"Двойной фильтр импульса", которая сочетает определение момента входа по резкому изменению объема и фильтрацию по силе тренда, чтобы избегать ложных движений.🧑‍💻

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

1. Определяем импульс по объему — если текущий объем выше среднего за N свечей на заданный коэффициент (например, 2×).

2. Фильтруем тренд по ADX — торгуем только, если тренд действительно сильный (ADX > 25).

3. Вход в сделку:

Лонг: Цена выше EMA(50), импульс на покупку.

Шорт: Цена ниже EMA(50), импульс на продажу.

4. Выход из сделки — по достижению заданного соотношения риск/прибыль или при обратном сигнале импульса.

import ccxt
import pandas as pd
import time

# Параметры
symbol = "BTC/USDT"
timeframe = "5m"
lookback = 100
volume_factor = 2
adx_period = 14
ema_period = 50

# Binance
exchange = ccxt.binance({'enableRateLimit': True})

def fetch_data():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=lookback)
    df = pd.DataFrame(ohlcv, columns=["time","open","high","low","close","volume"])
    df["time"] = pd.to_datetime(df["time"], unit="ms")
    return df

def calculate_indicators(df):
    df["EMA"] = df["close"].ewm(span=ema_period).mean()
    df["VolumeAvg"] = df["volume"].rolling(window=adx_period).mean()

    # ADX
    high = df["high"]
    low = df["low"]
    close = df["close"]

    plus_dm = high.diff()
    minus_dm = low.diff().abs()

    plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0)
    minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0.0)

    tr1 = high - low
    tr2 = (high - close.shift()).abs()
    tr3 = (low - close.shift()).abs()
    tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)

    atr = tr.rolling(window=adx_period).mean()

    plus_di = 100 * (plus_dm.rolling(window=adx_period).mean() / atr)
    minus_di = 100 * (minus_dm.rolling(window=adx_period).mean() / atr)

    dx = (abs(plus_di - minus_di) / (plus_di + minus_di)) * 100
    df["ADX"] = dx.rolling(window=adx_period).mean()
    return df

def check_signals(df):
    latest = df.iloc[-1]
    prev = df.iloc[-2]

    if latest["volume"] > latest["VolumeAvg"] * volume_factor and latest["ADX"] > 25:
        if latest["close"] > latest["EMA"]:
            print(f"[{latest['time']}] ЛОНГ сигнал — сильный импульс вверх, цена {latest['close']}")
        elif latest["close"] < latest["EMA"]:
            print(f"[{latest['time']}] ШОРТ сигнал — сильный импульс вниз, цена {latest['close']}")

while True:
    try:
        df = fetch_data()
        df = calculate_indicators(df)
        check_signals(df)
        time.sleep(10)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(5)

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

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

📌"Мульти-таймфреймовый тепловой индикатор тренда"🧑‍💻

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

Он в реальном времени показывает в консоли, насколько согласованы тренды на разных таймфреймах (например, 1m, 5m, 15m, 1h) для выбранной пары.
Это полезно, потому что многие стратегии эффективны, когда тренды совпадают на нескольких интервалах — вероятность успеха сделки выше.

import ccxt
import pandas as pd
import time

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

symbol = "BTC/USDT"
timeframes = ["1m", "5m", "15m", "1h"]
ema_period = 50

def get_trend(df):
    df["EMA"] = df["close"].ewm(span=ema_period).mean()
    if df["close"].iloc[-1] > df["EMA"].iloc[-1]:
        return "UP"
    elif df["close"].iloc[-1] < df["EMA"].iloc[-1]:
        return "DOWN"
    return "FLAT"

def fetch_data(symbol, timeframe):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=ema_period + 5)
    df = pd.DataFrame(ohlcv, columns=["time","open","high","low","close","volume"])
    return df

while True:
    try:
        trends = {}
        for tf in timeframes:
            df = fetch_data(symbol, tf)
            trends[tf] = get_trend(df)

        up_count = list(trends.values()).count("UP")
        down_count = list(trends.values()).count("DOWN")

        print(f"\n=== {symbol} ===")
        for tf, tr in trends.items():
            print(f"{tf}: {tr}")

        if up_count == len(timeframes):
            print("🔥 ВСЕ ТАЙМФРЕЙМЫ ВВЕРХ — сильный бычий тренд")
        elif down_count == len(timeframes):
            print("❄ ВСЕ ТАЙМФРЕЙМЫ ВНИЗ — сильный медвежий тренд")
        else:
            print("⚠ Смешанный тренд — возможная неопределенность")

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

#инструмент

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

📌"Стратегия сжимающегося диапазона (Volatility Squeeze Breakout)".🧑‍💻

📌 Идея стратегии:🚀

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

⚙️ Логика:🛠️

1. Считаем Bollinger Bands и Keltner Channels:

Когда полосы Боллинджера сжимаются внутрь каналов Кельтнера → рынок в сжатии.

2. Ждём выхода из диапазона:

Цена закрывается выше верхней границы Боллинджера → сигнал на покупку.

Цена закрывается ниже нижней границы Боллинджера → сигнал на продажу.


3. Дополнительно проверяем объём: пробой должен сопровождаться повышенным объёмом.

4. Сигналы выводим в консоль.

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

exchange = ccxt.binance({'enableRateLimit': True})
symbol = "BTC/USDT"
timeframe = "15m"
limit = 150

def fetch_data():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=["time","open","high","low","close","volume"])
    return df

def bollinger_keltner_strategy(df):
    df["MA20"] = df["close"].rolling(20).mean()
    df["STD20"] = df["close"].rolling(20).std()

    df["UpperBB"] = df["MA20"] + 2 * df["STD20"]
    df["LowerBB"] = df["MA20"] - 2 * df["STD20"]

    df["ATR"] = (df["high"] - df["low"]).rolling(20).mean()
    df["UpperKC"] = df["MA20"] + 1.5 * df["ATR"]
    df["LowerKC"] = df["MA20"] - 1.5 * df["ATR"]

    df["squeeze_on"] = (df["UpperBB"] < df["UpperKC"]) & (df["LowerBB"] > df["LowerKC"])
    df["squeeze_off"] = (df["UpperBB"] > df["UpperKC"]) & (df["LowerBB"] < df["LowerKC"])

    return df

while True:
    try:
        df = fetch_data()
        df = bollinger_keltner_strategy(df)

        last = df.iloc[-1]

        print(f"\n=== {symbol} {timeframe} ===")
        if last["squeeze_on"]:
            print("🔵 Рынок в сжатии (ожидается пробой)")
        elif last["squeeze_off"]:
            if last["close"] > last["UpperBB"] and last["volume"] > df["volume"].rolling(20).mean().iloc[-1]:
                print("🟢 Сигнал BUY — пробой вверх с объёмом")
            elif last["close"] < last["LowerBB"] and last["volume"] > df["volume"].rolling(20).mean().iloc[-1]:
                print("🔴 Сигнал SELL — пробой вниз с объёмом")
            else:
                print("⚪ Нет подтверждения пробоя")
        else:
            print("⏸ Рынок без особых сигналов")

        time.sleep(20)

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

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

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

📌Trade Readiness Score (TRS) Scanner — «сканер готовности к сделке».🧑‍💻

Он до запуска любой стратегии показывает, насколько “торгуем” сейчас конкретный инструмент, складывая в один балл ключевые микро- и макро-метрики рынка.

Что делает:🧨

- Снимает по REST (ccxt/Binance): спред, ликвидность в стакане (глубина в bps), дисбаланс заявок, волатильность (ATR/Close), объём vs средний, “колючесть” свечей.

- Считает сводный балл TRS (0–100) и печатает причины: что хорошо/плохо.

Помогает:💸

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

- выбрать лучший тикер для входа среди списка;

- ставить фильтр «не торговать ниже N баллов».

#инструмент

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

📌"Liquidity Vacuum" (вакуум ликвидности) 🧑‍💻

Идея:🧨

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

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

1. Находим диапазон за последние N свечей.

2. Ищем свечу, которая пробивает границы диапазона с аномально низким объёмом (то есть пробой в "пустоту").

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

4. Выход: на середине диапазона или по фиксированному соотношению риск/прибыль (например 1:2).

📌 Почему работает:💸

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

- Это смесь false breakout и анализа ликвидности.

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

# Конфиг
SYMBOL = "BTC/USDT"
TIMEFRAME = "15m"
BARS = 200
RANGE_LOOKBACK = 48   # размер диапазона (например сутки на 15m)
VOLUME_FACTOR = 0.7   # пробой считается слабым, если объём < 70% среднего
SLEEP = 30            # пауза между обновлениями

ex = ccxt.binance()

def fetch_data():
    ohlcv = ex.fetch_ohlcv(SYMBOL, TIMEFRAME, limit=BARS)
    df = pd.DataFrame(ohlcv, columns=["time","open","high","low","close","volume"])
    df["time"] = pd.to_datetime(df["time"], unit="ms")
    return df

def check_signal(df):
    # диапазон последних N свечей (без текущей)
    recent = df.iloc[-RANGE_LOOKBACK-2:-2]
    rng_high, rng_low = recent["high"].max(), recent["low"].min()
    avg_vol = recent["volume"].mean()

    # пробойная свеча (предыдущая)
    br = df.iloc[-2]
    # подтверждающая свеча (последняя)
    cf = df.iloc[-1]

    # проверка на пробой вверх
    if br["close"] > rng_high and br["volume"] < avg_vol*VOLUME_FACTOR:
        if cf["close"] < rng_high:  # возврат внутрь диапазона
            print(f"[{cf['time']}] SELL signal ⚠️ fake breakout up → {SYMBOL}")
            return

    # пробой вниз
    if br["close"] < rng_low and br["volume"] < avg_vol*VOLUME_FACTOR:
        if cf["close"] > rng_low:  # возврат внутрь
            print(f"[{cf['time']}] BUY signal ⚡ fake breakout down → {SYMBOL}")
            return

if __name__ == "__main__":
    while True:
        try:
            df = fetch_data()
            check_signal(df)
            time.sleep(SLEEP)
        except Exception as e:
            print("Ошибка:", e)
            time.sleep(5)

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

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

📌Regime Shift Early Warning (EWS)🧑‍💻

Сенсор раннего предупреждения о смене рыночного режима: считает набор устойчивых метрик (волатильность, автокорреляция AR(1), херст H, “тяжёлые хвосты”, всплеск кросс-корреляций) и даёт сводный балл EWS. При превышении порога печатает алерт по тикеру.

Что даёт:💸

- Ранние сигналы «критического замедления» → возможная смена тренда/режима.

- Фильтр для включения/выключения стратегий (не торговать при высоком EWS).

- Кросс-секционный радар: синхронные сдвиги по рынку.

#инструмент

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

📌Market Microstructure Entropy (MME)🧑‍💻

📌 Суть:🚀

Это алгоритм, который в реальном времени вычисляет энтропию рыночных тиков — насколько рынок хаотичен или упорядочен.

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

1. Через ccxt или прямой WebSocket получаем поток сделок (trades).

2. Для каждого окна (например, 1000 последних сделок) строим распределение направлений и размеров сделок.

- 70% покупок и все одинакового размера → низкая энтропия (рынок «упорядочен», скорее идёт тренд).

- хаотичное распределение, покупки/продажи перемешаны, размеры разные → высокая энтропия (рынок «шумный», возможен разворот).

3. Вычисляем Shannon entropy для распределения.

4. На основе этого подаём сигнал:

🔴 Низкая энтропия → лучше торговать по тренду.

🟢 Высокая энтропия → лучше использовать mean reversion (возврат к среднему).

5. Можно строить индикатор "энтропия → стратегия".

import time
import math
from collections import Counter, deque
import numpy as np
import ccxt

class MarketMicrostructureEntropy:
    def __init__(self, window_size=1000, size_bins=None, low_thresh=0.4, high_thresh=0.75, ema_alpha=0.1):
        self.window_size = window_size
        self.trades = deque(maxlen=window_size)
        self.size_bins = size_bins or [0.5, 1.0, 2.0, 5.0]
        self.low_thresh = low_thresh
        self.high_thresh = high_thresh
        self.ema_alpha = ema_alpha
        self.ema_entropy = None

    def add_trade(self, size, side):
        self.trades.append({"size": float(size), "side": side})

    def _size_ratio(self):
        sizes = np.array([t["size"] for t in self.trades], dtype=float)
        if len(sizes) == 0:
            return np.array([])
        median = np.median(sizes) or 1e-9
        return sizes / median

    def _size_category(self, ratio):
        for i, b in enumerate(self.size_bins):
            if ratio <= b:
                return i
        return len(self.size_bins)

    def compute_entropy(self, base=2.0):
        n = len(self.trades)
        if n == 0:
            return 0.0, 0.0

        ratios = self._size_ratio()
        categories = []
        for i, t in enumerate(self.trades):
            cat_size = self._size_category(ratios[i])
            categories.append((t["side"], cat_size))

        cnt = Counter(categories)
        probs = np.array(list(cnt.values()), dtype=float) / n
        with np.errstate(divide="ignore", invalid="ignore"):
            logs = np.log(probs) / np.log(base)
        H = -np.sum(probs * logs)

        M = 2 * (len(self.size_bins) + 1)
        maxH = math.log(M) / math.log(base)
        norm = H / maxH if maxH > 0 else 0.0

        if self.ema_entropy is None:
            self.ema_entropy = norm
        else:
            self.ema_entropy = (1 - self.ema_alpha) * self.ema_entropy + self.ema_alpha * norm

        return H, norm

    def get_signal(self):
        if self.ema_entropy is None:
            return "neutral"
        if self.ema_entropy < self.low_thresh:
            return "trend"
        elif self.ema_entropy > self.high_thresh:
            return "mean_reversion"
        return "neutral"

def run_polling(exchange_id="binance", symbol="BTC/USDT", poll_interval=2.0, window_size=1000):
    ex = getattr(ccxt, exchange_id)()
    mme = MarketMicrostructureEntropy(window_size=window_size)

    print(f"Polling {exchange_id} {symbol} every {poll_interval}s...")

    while True:
        try:
            trades = ex.fetch_trades(symbol, limit=200)
            for t in trades:
                side = t.get("side", "buy")
                size = t.get("amount", 0.0)
                mme.add_trade(size=size, side=side)

            H, norm = mme.compute_entropy()
            sig = mme.get_signal()
            print(f"[{time.strftime('%H:%M:%S')}] trades={len(mme.trades)} H={H:.4f} norm={norm:.4f} ema={mme.ema_entropy:.4f} signal={sig}")

        except Exception as e:
            print("Error:", e)

        time.sleep(poll_interval)


if __name__ == "__main__":
    run_polling(exchange_id="binance", symbol="BTC/USDT", poll_interval=2.0, window_size=1000)

#инструмент

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

📌"Асимметричный риск-реверс" (Asymmetric Risk-Reversal)🧑‍💻

🔑 Суть💸

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

- Падение обычно резкое и короткое.

- Рост более плавный и долгий.

Мы строим стратегию, где правила входа/выхода разные для лонга и для шорта.

📐 Логика:🚀

1. Лонг (основная часть стратегии):

Вход: после локальной консолидации (N свечей с низкой волатильностью).

Фильтр: тренд по EMA(100) вверх.

Выход: по тейку в 1.5–2 ATR.

2. Шорт (агрессивный, но короткий):

Вход: при всплеске волатильности (цена падает >2σ за свечу).

Выход: быстрый тейк = 0.5–1 ATR.

Стоп короткий (ниже последнего high).

3. Таким образом:

В лонгах мы "сидим на росте".

В шортах — "снимаем сливки с паники".

🎯 Почему работает:🧨

- Использует структурную асимметрию рынка.

- Даёт меньше убытков на ложных пробоях вниз.

- Хорошо комбинируется с фьючерсами и плечом.

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

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

📌"Анализ уникальности торгового сигнала"🧑‍💻

Большинство стратегий в алготрейдинге страдают от переоптимизации (overfitting). Сигналы часто повторяются в "шуме", и бот гоняет туда-сюда без реального edge.

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

Это позволяет:🚀

- Отсечь повторяющиеся сигналы без смысла.

- Искать реально редкие паттерны, где вероятность отработки выше.

- Добавить "фильтр качества" к любой стратегии.

🔧 Описание логики:🛠️

1. Собираем исторические данные OHLCV.

2. Вычисляем вектор признаков свечи / паттерна (волатильность, позиция цены в диапазоне, ATR, RSI, объём и т.д.).

3. Сравниваем текущий сигнал с историческими через метрику расстояния (cosine similarity / евклидово расстояние).

4. Если сигнал слишком похож на уже бывшие убыточные, игнорируем его.

5. Если сигнал уникален (редкий паттерн) — помечаем как "высокого качества".

import ccxt
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.metrics.pairwise import cosine_similarity

# ========== Параметры ==========
exchange = ccxt.binance()
symbol = "BTC/USDT"
timeframe = "1h"
limit = 500

# === Шаг 1: загрузка данных ===
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=["time", "open", "high", "low", "close", "volume"])
df["time"] = pd.to_datetime(df["time"], unit="ms")

# === Шаг 2: генерация признаков ===
df["return"] = df["close"].pct_change()
df["range"] = (df["high"] - df["low"]) / df["close"]
df["pos_in_range"] = (df["close"] - df["low"]) / (df["high"] - df["low"] + 1e-9)
df["volatility"] = df["return"].rolling(10).std()
df["volume_z"] = (df["volume"] - df["volume"].rolling(20).mean()) / df["volume"].rolling(20).std()

# оставляем чистые признаки
features = df[["return", "range", "pos_in_range", "volatility", "volume_z"]].dropna()

# === Шаг 3: нормализация ===
scaler = StandardScaler()
X = scaler.fit_transform(features)

# === Шаг 4: функция уникальности ===
def signal_uniqueness(current_idx, lookback=100):
    """
    Возвращает "уникальность сигнала" от 0 до 1.
    0 = полностью копия старых, 1 = абсолютно новый паттерн
    """
    if current_idx < lookback:
        return 0.5  # мало истории
    
    current_vec = X[current_idx].reshape(1, -1)
    history_vecs = X[current_idx - lookback: current_idx]
    
    sims = cosine_similarity(current_vec, history_vecs)[0]
    uniqueness = 1 - sims.max()  # чем меньше сходство — тем больше уникальность
    return float(uniqueness)

# === Шаг 5: применяем ===
df = df.iloc[len(df) - len(X):].copy()
df["uniqueness"] = [signal_uniqueness(i) for i in range(len(X))]

# === Пример: фильтр сигналов ===
threshold = 0.3  # ниже этого — сигнал игнорируется
df["signal"] = np.where(df["uniqueness"] > threshold, "TRADE", "IGNORE")

print(df[["time", "close", "uniqueness", "signal"]].tail(20))

#инструмент

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

📌MME VWAP Regime Switch (MRS)🧑‍💻

Ключ: сочетать микроструктурную энтропию (MME) из тиков с поведенческими паттернами дня (Opening Range) и средним притяжения (VWAP).

Режимы по энтропии:🧨

- Низкая энтропия (упорядоченность) → торгуем пробой/продолжение:

- После формирования Opening Range (первые OR_MIN минут) берём пробой ORH/ORL в сторону относительно VWAP.

- Высокая энтропия (хаос) → торгуем mean-reversion:

Когда цена отклоняется от VWAP больше, чем K * ATR(1m), играем возврат к VWAP.

Фильтры & выходы:💸

- В трендовом режиме — вход только если цена по сторону VWAP (для лонга выше, для шорта ниже).

- В MR-режиме — вход против отклонения, выход у VWAP или по стопу.

- Переключение режима (из-за MME) закрывает позицию.

- Стоп-лосс — кратный ATR.

- Торгуем 1 инструмент (по умолчанию BTC/USDT), сигналы только в консоль (без выставления ордеров).

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

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