📌"Alpha Tracker" — Реальное сравнение вашей стратегии с рынком🧑‍💻

📌 Идея:🧨

Инструмент позволяет в реальном времени сравнивать доходность вашей стратегии с базовым активом (например, BTC/USDT). Это помогает:

- Понять, обгоняет ли стратегия рынок (генерирует alpha).

- Измерить эффективность любой стратегии: алгоритмической, ручной, сигнальной и т.п.

- Визуализировать накопленную доходность стратегии и сравнить её с buy & hold.

💼 Что делает инструмент:🛠️

1. Вы подаёте ваши сигналы (например, список входов/выходов).

2. Скрипт рассчитывает доходность вашей стратегии.

3. Скрипт параллельно считает доходность пассивного владения активом (HODL).

4. В консоли (или графике) сравниваются кумулятивные результаты.

import ccxt
import time
import pandas as pd

# Пример входов и выходов (timestamp в секундах, тип сделки, цена будет подставляться)
# Пример: вход и выход по BTC
strategy_signals = [
    {"timestamp": 1720000000, "action": "buy"},
    {"timestamp": 1720005000, "action": "sell"},
    {"timestamp": 1720010000, "action": "buy"},
    {"timestamp": 1720020000, "action": "sell"}
]

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

def fetch_price_at(timestamp):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe='1m', since=timestamp * 1000, limit=1)
    return ohlcv[0][4] if ohlcv else None

def evaluate_strategy(signals):
    returns = []
    for i in range(0, len(signals)-1, 2):
        buy_price = fetch_price_at(signals[i]['timestamp'])
        sell_price = fetch_price_at(signals[i+1]['timestamp'])
        if buy_price and sell_price:
            pnl = (sell_price - buy_price) / buy_price
            returns.append(pnl)
    return returns

def evaluate_hodl(start_time, end_time):
    buy = fetch_price_at(start_time)
    sell = fetch_price_at(end_time)
    return (sell - buy) / buy if buy and sell else None

# Запуск
strategy_returns = evaluate_strategy(strategy_signals)
hodl_return = evaluate_hodl(strategy_signals[0]['timestamp'], strategy_signals[-1]['timestamp'])

print("\n📊 Доходность стратегии:")
for i, r in enumerate(strategy_returns):
    print(f"Сделка {i+1}: {r:.2%}")

print(f"\n📈 Доходность HODL за тот же период: {hodl_return:.2%}")
print(f"📐 Итоговое alpha: {sum(strategy_returns) - hodl_return:.2%}")

#инструмент

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

📌“Panic Bid Reversal” (PBR Strategy)🧑‍💻

📈 Стратегия ловли капитуляций с автоматическим определением "панической свечи" и входом в разворот на повышенном объёме.

🧠 Идея стратегии:🛠️

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

✅ Условия входа в лонг:🧨

1. Свеча на 1м/5м таймфрейме:

Упала на >3% от предыдущей закрытия.

Имеет объём X выше среднего (Volume Spike).

Закрылась с фитилём снизу (низкий Close относительно Low).

2. Следующая свеча начинает рост — сигнал на вход.

📤 Выход по:💸

Take Profit = +1.5–2.5%

Stop Loss = ниже фитиля панической свечи.

import ccxt
import time
from datetime import datetime

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '1m'
limit = 50

def fetch_ohlcv():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    return [{
        'timestamp': x[0],
        'open': x[1],
        'high': x[2],
        'low': x[3],
        'close': x[4],
        'volume': x[5]
    } for x in ohlcv]

def detect_panic_reversal(data):
    avg_volume = sum([c['volume'] for c in data[-10:-1]]) / 9
    last = data[-2]
    current = data[-1]

    drop = (last['close'] - last['low']) / last['close']
    body = abs(last['close'] - last['open'])
    wick = (last['high'] - last['low']) - body

    if (last['close'] < last['open'] * 0.97 and
        last['volume'] > avg_volume * 2 and
        wick > body and
        current['close'] > last['close']):
        return True
    return False

print("📡 Panic Bid Reversal Strategy запущена...")

while True:
    try:
        candles = fetch_ohlcv()
        if detect_panic_reversal(candles):
            ts = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')
            print(f"✅ [{ts}] Сигнал на лонг: Обнаружена паническая свеча с разворотом!")
        time.sleep(60)
    except Exception as e:
        print(f"⚠️ Ошибка: {e}")
        time.sleep(10)

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

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

📌“Liquidity Sniper” — Торговля на пробой зон ликвидности по паттернам всплесков объёма🧑‍💻

📌 Идея:🧨

Вместо классических уровней поддержки/сопротивления, стратегия ищет зоны ликвидности — участки, где ранее был высокий объём + разворот. Считается, что маркетмейкеры возвращаются туда, чтобы “забрать ликвидность” перед импульсом.

⚙️ Принципы стратегии:🛠️

- Выявить свечи с аномально высоким объёмом.

- Определить “зону ликвидности” — ценовой диапазон тела такой свечи.

- Ожидать возврат в зону и входить по направлению предыдущего объёма.

- Стоп за противоположный край зоны, тейк — по ближайшему импульсу.

🔍 Условия входа в сделку:💸

1. Находится свеча с объёмом выше среднего на N свечей.

2. Цена возвращается в тело этой свечи спустя X баров.

3. Объём остаётся высоким — подтверждение интереса.

4. Открывается сделка по направлению исходного движения.

import ccxt
import pandas as pd
import pandas_ta as ta
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '15m'
bars = 150
volume_multiplier = 2

def fetch_ohlcv():
    ohlcv = exchange.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 detect_liquidity_zones(df):
    df['avg_vol'] = df['volume'].rolling(20).mean()
    df['is_liquidity_candle'] = df['volume'] > df['avg_vol'] * volume_multiplier

    liquidity_zones = []
    for i in range(1, len(df)):
        if df['is_liquidity_candle'].iloc[i]:
            zone = {
                "index": i,
                "price_low": df['open'].iloc[i],
                "price_high": df['close'].iloc[i],
                "direction": "buy" if df['close'].iloc[i] > df['open'].iloc[i] else "sell"
            }
            liquidity_zones.append(zone)
    return liquidity_zones

def check_retest(df, zones):
    current_price = df['close'].iloc[-1]
    for zone in zones[-5:]:  # последние 5 зон
        if zone["price_low"] <= current_price <= zone["price_high"]:
            print(f"\n📡 Сигнал: Retest зоны ликвидности!")
            print(f"👉 Направление: {zone['direction'].upper()}")
            print(f"📍 Диапазон зоны: {zone['price_low']} - {zone['price_high']}")
            break

# Основной цикл
while True:
    df = fetch_ohlcv()
    zones = detect_liquidity_zones(df)
    check_retest(df, zones)
    time.sleep(60)

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

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

📌"Order Book Dynamics Tracker" — Инструмент отслеживания изменений глубины стакана в реальном времени🧑‍💻

🧠 Что делает инструмент:🛠️

Позволяет отслеживать изменения лимитных ордеров (bid/ask) по уровням и строить метрику, показывающую поведение маркетмейкеров и крупных игроков — до появления свечи и сигнала от индикаторов.

🧪 Идея:🧨

Изменения в order book (например, внезапное исчезновение большого лимитника или появление «стены») часто предшествуют импульсным движениям. Это можно использовать для:

- определения фейковых движений,

- входа до свечного сигнала,

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

import asyncio
import json
import websockets
from collections import deque

symbol = 'btcusdt'
depth_levels = 10
window = deque(maxlen=5)

async def track_orderbook():
    uri = f"wss://fstream.binance.com/ws/{symbol}@depth@100ms"
    
    async with websockets.connect(uri) as websocket:
        print("📡 Подключение к Binance WebSocket...")
        
        while True:
            msg = await websocket.recv()
            data = json.loads(msg)

            bids = data['bids'][:depth_levels]
            asks = data['asks'][:depth_levels]

            bid_vol = sum([float(b[1]) for b in bids])
            ask_vol = sum([float(a[1]) for a in asks])

            imbalance = (bid_vol - ask_vol) / (bid_vol + ask_vol + 1e-6)
            window.append(imbalance)

            avg_imbalance = sum(window) / len(window)

            signal = "🔼 BUY PRESSURE" if avg_imbalance > 0.25 else "🔽 SELL PRESSURE" if avg_imbalance < -0.25 else "⚖️ Neutral"

            print(f"Imbalance: {avg_imbalance:.2f} → {signal}")

asyncio.run(track_orderbook())

#инструмент

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

📌"Объем + Импульсная фильтрация по кластерам"🧑‍💻

📈 Тип стратегии: краткосрочная (intra-day или swing)

📉 Подходит для: спот и фьючерсы

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

🔍 Логика стратегии:🛠️

1. Фильтр объема (Volume Spike)

Считается средний объем за последние N свечей.

Если текущий объем > x * среднего (например, 3x), переходим к следующему шагу.

2. Фильтр диапазона

Проверяем, что последние N свечей были в узком ценовом диапазоне (например, max/min < 1.5%).

3. Фильтр импульса (Breakout)

Проверяется, что текущая свеча пробивает максимум или минимум диапазона.

4. Сигнал на вход в сторону пробоя

Take-Profit можно поставить на ширину диапазона * 2,

Stop-Loss — внутри диапазона.

🧠 Преимущества:🧨

- Объем указывает на институциональные действия.

- Узкий диапазон + резкий выход — признак движения, а не шума.

- Подходит для алготрейдинга, особенно в консолидациях.

import ccxt
import numpy as np
import time

symbol = 'BTC/USDT'
timeframe = '5m'
volume_multiplier = 3
range_percent_threshold = 1.5  # %

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

def fetch_data():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=30)
    return np.array(ohlcv)

def analyze(data):
    highs = data[:, 2]
    lows = data[:, 3]
    closes = data[:, 4]
    volumes = data[:, 5]

    last_candle = data[-1]
    recent_volumes = volumes[-21:-1]
    avg_volume = np.mean(recent_volumes)
    
    if last_candle[5] < avg_volume * volume_multiplier:
        return None  # Нет всплеска объема

    recent_high = np.max(highs[-21:-1])
    recent_low = np.min(lows[-21:-1])
    price_range_pct = (recent_high - recent_low) / recent_low * 100

    if price_range_pct > range_percent_threshold:
        return None  # Диапазон был слишком широкий

    close = last_candle[4]
    if close > recent_high:
        return 'BUY'
    elif close < recent_low:
        return 'SELL'
    return None

while True:
    try:
        candles = fetch_data()
        signal = analyze(candles)
        if signal:
            print(f"[{time.strftime('%H:%M:%S')}] 📢 Сигнал: {signal} на {symbol}")
        else:
            print(f"[{time.strftime('%H:%M:%S')}] ⏳ Нет сигнала")
    except Exception as e:
        print("Ошибка:", e)
    
    time.sleep(60)

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

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

📌"Анализатор Необычной Волатильности и Объема (UVVA)"🧑‍💻

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

- Мониторит в реальном времени сразу несколько криптовалют на предмет необычного поведения:

- резких всплесков объема,

- расширения/сжатия диапазона,

- резких изменений в волатильности,

- появления «гепов» на мелком таймфрейме.

Сигналы можно использовать для:💸

- входа в пробой,

- скальпинга,

- детектирования фейковых движений или манипуляций.

import ccxt
import numpy as np
import time

exchange = ccxt.binance({'enableRateLimit': True})
symbols = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT']
timeframe = '1m'
lookback = 50
volume_threshold = 2.5  # В 2.5 раза выше среднего
range_threshold = 2.0   # В 2 раза выше среднего диапазона

def fetch_ohlcv(symbol):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=lookback)
    return np.array(ohlcv)

def analyze(symbol, data):
    volumes = data[:, 5]
    highs = data[:, 2]
    lows = data[:, 3]
    closes = data[:, 4]

    avg_volume = np.mean(volumes[:-1])
    last_volume = volumes[-1]
    avg_range = np.mean(highs[:-1] - lows[:-1])
    last_range = highs[-1] - lows[-1]

    volume_spike = last_volume > avg_volume * volume_threshold
    range_spike = last_range > avg_range * range_threshold

    if volume_spike and range_spike:
        print(f"[{symbol}] ⚡ Обнаружено необычное поведение!")
        print(f"  Объем: {last_volume:.2f} (> {avg_volume:.2f})")
        print(f"  Диапазон: {last_range:.4f} (> {avg_range:.4f})")
        return True
    return False

while True:
    print(f"\n🔍 Проверка ({time.strftime('%H:%M:%S')})")
    for symbol in symbols:
        try:
            data = fetch_ohlcv(symbol)
            analyze(symbol, data)
        except Exception as e:
            print(f"Ошибка с {symbol}: {e}")
    time.sleep(60)

#инструмент

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

📌"Импульс на паузе (Impulse Pause Entry)"🧑‍💻

💡 Идея:🚀

Вход не во время импульса, а сразу после него, когда цена "замирает" в узком диапазоне. Это часто происходит перед продолжением движения.

Используется для:🧨

- Торговли пробоев с минимальным риском

- Снижения количества ложных входов

✅ Условия входа в сделку (лонг):

1. Импульс: свеча > X% роста (например, 1% на 1м таймфрейме)

2. Пауза: N следующих свечей с малым диапазоном и объемом

3. Выход вверх из диапазона паузы = сигнал на покупку

(для шорта — наоборот)

📉 Пример условий:💸

- Импульс: свеча выросла >1%

- Пауза: 3 свечи подряд, у которых диапазон < 0.3% и объём ниже среднего

- Если цена пробивает high импульсной свечи — входим в сделку

import ccxt
import time
import numpy as np

exchange = ccxt.binance({'enableRateLimit': True})
symbol = 'BTC/USDT'
timeframe = '1m'

impulse_threshold = 0.01      # 1% рост
pause_range_threshold = 0.003  # 0.3%
pause_candles = 3

def fetch_ohlcv(symbol):
    return exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=20)

def check_impulse_pause(data):
    closes = np.array([c[4] for c in data])
    highs = np.array([c[2] for c in data])
    lows = np.array([c[3] for c in data])
    volumes = np.array([c[5] for c in data])

    returns = (closes[1:] - closes[:-1]) / closes[:-1]
    last_impulse_index = np.where(returns > impulse_threshold)[0]
    
    if len(last_impulse_index) == 0:
        return False

    idx = last_impulse_index[-1] + 1
    if idx + pause_candles >= len(closes):
        return False

    pause_ranges = highs[idx:idx+pause_candles] - lows[idx:idx+pause_candles]
    avg_range = np.mean(closes[:-pause_candles] * pause_range_threshold)
    if np.all(pause_ranges < avg_range):
        breakout_price = highs[idx-1]  # high импульсной свечи
        current_price = closes[-1]
        if current_price > breakout_price:
            print(f"🚀 [{symbol}] Сигнал ЛОНГ: пробой после паузы на импульсе!")
            return True
    return False

while True:
    try:
        candles = fetch_ohlcv(symbol)
        check_impulse_pause(candles)
    except Exception as e:
        print(f"Ошибка: {e}")
    time.sleep(60)

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

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

📌"Smart Level Radar" — Умный сканер горизонтальных уровней поддержки/сопротивления с визуальной тепловой картой"🧑‍💻

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

Инструмент автоматически сканирует исторические данные (OHLCV) и находит ключевые уровни, где:

- Цена останавливалась, отскакивала или пробивала

- Объем был на пиках

- Уровень тестировался многократно

- Затем он визуализирует уровни как тепловую карту (heatmap), показывая, где на графике сконцентрировано наибольшее "внимание" рынка.

🧠 Зачем трейдеру:💸

- Позволяет быстро оценить важные уровни поддержки/сопротивления

- Упрощает принятие решений по входу/выходу

- Работает как дневной "сканер рынка" перед сессией

- Может использоваться для любых активов: BTC, альты, акции

import ccxt
import numpy as np
import matplotlib.pyplot as plt

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '1h'
limit = 1000

def fetch_ohlcv(symbol, timeframe, limit):
    data = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    return np.array(data)

def detect_levels(candles, threshold=5):
    highs = candles[:, 2]
    lows = candles[:, 3]
    levels = []

    for i in range(2, len(candles) - 2):
        high = highs[i]
        low = lows[i]
        count_high = np.sum(np.abs(highs[i-2:i+3] - high) < 0.001 * high)
        count_low = np.sum(np.abs(lows[i-2:i+3] - low) < 0.001 * low)

        if count_high >= threshold:
            levels.append(high)
        if count_low >= threshold:
            levels.append(low)

    return sorted(set([round(lvl, 2) for lvl in levels]))

def plot_levels(candles, levels):
    timestamps = [c[0] for c in candles]
    closes = [c[4] for c in candles]

    plt.figure(figsize=(15, 7))
    plt.plot(timestamps, closes, label='Price', color='black')
    for lvl in levels:
        plt.axhline(lvl, color='red', alpha=0.3)
        plt.text(timestamps[0], lvl, str(lvl), color='red', fontsize=8)
    plt.title('Smart Level Radar')
    plt.xlabel('Time')
    plt.ylabel('Price')
    plt.grid(True)
    plt.legend()
    plt.show()

candles = fetch_ohlcv(symbol, timeframe, limit)
levels = detect_levels(candles)
plot_levels(candles, levels)

#инструмент

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

📌Bid-Ask Spread Heatmap (Монитор ликвидности по рынку)🧑‍💻

Описание:🚀

Этот скрипт показывает спред между bid и ask ценами на топ-ликвидных парах Binance, сортируя их по уровню "узости" спреда.

Это полезно для:💸

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

- оценки "фрикционных" затрат при входе/выходе;

- настройки стратегий арбитража и маркет-мейкинга.

import ccxt
import time

exchange = ccxt.binance()
markets = exchange.load_markets()
symbols = [s for s in markets if s.endswith('/USDT') and markets[s]['active']]

def get_spread(symbol):
    try:
        orderbook = exchange.fetch_order_book(symbol, limit=5)
        bid = orderbook['bids'][0][0] if orderbook['bids'] else None
        ask = orderbook['asks'][0][0] if orderbook['asks'] else None

        if bid is None or ask is None:
            return None

        spread_pct = (ask - bid) / ((ask + bid) / 2) * 100
        return round(spread_pct, 4)
    except Exception:
        return None

def main():
    spreads = []
    for symbol in symbols:
        spread = get_spread(symbol)
        if spread is not None:
            spreads.append((symbol, spread))

    spreads.sort(key=lambda x: x[1])  # от самого узкого спреда
    print(f"\nTop 10 пар с самым узким спредом:\n{'-'*40}")
    for symbol, spread in spreads[:10]:
        print(f"{symbol:15} → Спред: {spread:.4f}%")

if __name__ == "__main__":
    while True:
        main()
        time.sleep(60)  # обновление раз в минуту

#инструмент

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

📌"Liquidity Imbalance Pulse"🧑‍💻

🧠 Идея:🧨

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

- Использует данные ордербука (bid/ask depth)

- Сопоставляет дельту объема (агрессивные покупки/продажи)

- Отслеживает поведение цены в микроструктуре (1s–5s свечи)

- Алгоритм определяет: это ложное движение или начало реального импульса

📊 Сигналы:💸

Лонг:

- Обнаружен сильный дисбаланс: bid size резко увеличился, ask съедается маркет-ордерами

- Spread сузился

- Цена закрепляется выше локального уровня ликвидности

Шорт:

Обратная ситуация: ask усилился, маркет-продажи интенсивны, цена пробивает ликвидность вниз

import ccxt
import time
import datetime

# Настройки
symbol = 'BTC/USDT'
depth_limit = 20  # Глубина анализа стакана
imbalance_threshold = 0.5  # Чувствительность к дисбалансу
poll_interval = 3  # Интервал между проверками (сек)

# Инициализация биржи
exchange = ccxt.binance({
    'enableRateLimit': True,
})

def get_orderbook_metrics():
    try:
        orderbook = exchange.fetch_order_book(symbol, limit=depth_limit)

        bids = orderbook['bids']
        asks = orderbook['asks']

        bid_volume = sum([price * amount for price, amount in bids])
        ask_volume = sum([price * amount for price, amount in asks])

        best_bid = bids[0][0] if bids else 0
        best_ask = asks[0][0] if asks else 0
        spread = best_ask - best_bid
        mid_price = (best_ask + best_bid) / 2 if best_bid and best_ask else 0

        imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume + 1e-9)

        return {
            'bid_volume': bid_volume,
            'ask_volume': ask_volume,
            'imbalance': imbalance,
            'spread': spread,
            'mid_price': mid_price,
            'timestamp': datetime.datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')
        }

    except Exception as e:
        print("Ошибка при получении стакана:", str(e))
        return None

def print_signal(metrics):
    imbalance = metrics['imbalance']
    signal = "⚪ Нейтрально"

    if imbalance >= imbalance_threshold:
        signal = "🟢 ЛОНГ сигнал"
    elif imbalance <= -imbalance_threshold:
        signal = "🔴 ШОРТ сигнал"

    print(f"[{metrics['timestamp']}] {signal} | Imbalance: {imbalance:.2f} | "
          f"BidVol: {metrics['bid_volume']:.0f} | AskVol: {metrics['ask_volume']:.0f} | "
          f"Mid: {metrics['mid_price']:.2f} | Spread: {metrics['spread']:.2f}")

def run():
    print("🚀 Старт стратегии Liquidity Imbalance Pulse\n")
    while True:
        metrics = get_orderbook_metrics()
        if metrics:
            print_signal(metrics)
        time.sleep(poll_interval)

if __name__ == '__main__':
    run()

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

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

📌Стратегия возврата к среднему с подтверждением через ATR и Price Action🧑‍💻

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

Основная идея:🚀

Цена временно отклоняется от скользящей средней (SMA) — если отклонение достаточно велико (например, 2×ATR), и свеча подтверждает разворот, то стратегия выдаёт сигнал.

🔧 Настройки:

- SMA: 20-периодная

- ATR: 14-периодная

- Подтверждение: свеча закрылась в сторону возврата

- Сигналы: только когда цена находится вне 2×ATR от SMA

import ccxt
import pandas as pd
import time
import datetime

# Настройки
symbol = 'BTC/USDT'
timeframe = '1h'
sma_period = 20
atr_period = 14
atr_multiplier = 2

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

def fetch_data():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=max(sma_period, atr_period) + 50)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    return df

def compute_indicators(df):
    df['sma'] = df['close'].rolling(sma_period).mean()
    tr = pd.concat([
        df['high'] - df['low'],
        (df['high'] - df['close'].shift()).abs(),
        (df['low'] - df['close'].shift()).abs()
    ], axis=1).max(axis=1)
    df['atr'] = tr.rolling(atr_period).mean()
    return df

def check_signal(df):
    latest = df.iloc[-1]
    prev = df.iloc[-2]
    upper_band = latest['sma'] + atr_multiplier * latest['atr']
    lower_band = latest['sma'] - atr_multiplier * latest['atr']

    # Условия для сигнала на шорт
    if latest['close'] > upper_band and latest['close'] < prev['close']:
        print(f"🔻 SHORT сигнал [{symbol}] @ {latest['close']} — возврат от верхней границы")

    # Условия для сигнала на лонг
    elif latest['close'] < lower_band and latest['close'] > prev['close']:
        print(f"🔼 LONG сигнал [{symbol}] @ {latest['close']} — возврат от нижней границы")

def main():
    while True:
        try:
            df = fetch_data()
            df = compute_indicators(df)
            check_signal(df)
            time.sleep(60 * 60)  # Ждём 1 час до следующего бара
        except Exception as e:
            print("Ошибка:", e)
            time.sleep(60)

if __name__ == '__main__':
    main()

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

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

📌“Order Book Pressure Index” (OBPI)🧑‍💻

📌 Идея:🚀

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

🧠 Логика:🧨

1. Подгружаются bid/ask уровни (обычно топ 20).

2. Рассчитываются:

Bid Pressure = сумма объёмов всех заявок на покупку.

Ask Pressure = сумма объёмов всех заявок на продажу.

3. Order Book Pressure Index (OBPI) = (Bid Pressure - Ask Pressure) / (Bid Pressure + Ask Pressure)

Значения ближе к +1 → спрос > предложение → возможный рост.

Значения ближе к -1 → предложение > спрос → возможное падение.

4. Если OBPI стабильно высокий/низкий + цена стоит → сигнал накопления/распродажи.

import ccxt
import time

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

def get_order_book(symbol, depth=20):
    ob = exchange.fetch_order_book(symbol, limit=depth)
    bids = ob['bids']
    asks = ob['asks']
    bid_vol = sum([b[1] for b in bids])
    ask_vol = sum([a[1] for a in asks])
    return bid_vol, ask_vol

def compute_obpi(bid_vol, ask_vol):
    if bid_vol + ask_vol == 0:
        return 0
    return (bid_vol - ask_vol) / (bid_vol + ask_vol)

while True:
    bid_vol, ask_vol = get_order_book(symbol)
    obpi = compute_obpi(bid_vol, ask_vol)
    print(f"[{symbol}] OBPI: {obpi:.4f}")

    if obpi > 0.6:
        print("🔼 Сильное давление покупателей — возможный рост")
    elif obpi < -0.6:
        print("🔽 Сильное давление продавцов — возможное падение")
    else:
        print("⏸️ Баланс, рынок в равновесии")

    time.sleep(10)

#инструмент

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

📌"Live Liquidity Sweep Tracker" (Трекер снятия ликвидности в реальном времени)🧑‍💻

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

Этот инструмент мониторит в реальном времени резкие всплески объёмов и цену, которая прорывает важные уровни (локальные хаи/лоу) — это может указывать на то, что была снята ликвидность (стопы) и дальше произойдёт откат или разворот.

📈 Зачем это нужно:💸

Часто перед разворотом маркетмейкеры "снимают стопы", чтобы собрать ликвидность.

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

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

1. Загружает свечи.

2. Находит локальные экстремумы (например, последние 3 high/low).

3. Следит за пробоем этих уровней на аномально высоком объёме.

4. Если пробой сопровождается скачком объёма — сигнал в консоль.

import ccxt
import pandas as pd
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '1m'
limit = 100

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

def detect_liquidity_sweep(df):
    highs = df['high'][-4:-1]
    lows = df['low'][-4:-1]
    volume_mean = df['volume'][-20:-1].mean()
    
    last = df.iloc[-1]
    high_break = last['high'] > highs.max() and last['volume'] > volume_mean * 1.5
    low_break = last['low'] < lows.min() and last['volume'] > volume_mean * 1.5

    if high_break:
        return f"🔴 Liquidity Sweep UP detected on {symbol} — possible short setup"
    elif low_break:
        return f"🟢 Liquidity Sweep DOWN detected on {symbol} — possible long setup"
    return None

while True:
    df = fetch_candles()
    signal = detect_liquidity_sweep(df)
    if signal:
        print(signal)
    else:
        print("⚪ No liquidity sweep detected.")
    time.sleep(10)

#инструмент

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

📌"Imbalance Breakout" (Пробой Дисбаланса)🧑‍💻

📌 Идея:🚀

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

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

Лонг:

- Ищется участок из ≥3 бычьих свечей без теней снизу (или почти без).

- Цена консолидируется у верхней границы этой зоны (боковик).

- При пробое верхней границы на увеличенном объёме — вход в лонг.


Шорт:

- Аналогично: участок из ≥3 медвежьих свечей без откатов.

- Консолидация.

- Пробой вниз → вход в шорт.

✅ Подтверждение:

- Объём выше среднего.

- Волатильность расширяется.

- Желательно наличие дивергенции по RSI для раннего сигнала.

import ccxt
import time
import pandas as pd

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

def fetch_candles(symbol, tf='5m', lim=100):
    data = exchange.fetch_ohlcv(symbol, tf, limit=lim)
    df = pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    return df

def detect_imbalance_breakout(df):
    last = df.iloc[-4:-1]  # Последние 3 свечи до текущей
    current = df.iloc[-1]
    
    bullish_block = all(last['close'] > last['open']) and all((last['low'] - last['open']).abs() < 0.1 * (last['high'] - last['low']))
    bearish_block = all(last['close'] < last['open']) and all((last['high'] - last['open']).abs() < 0.1 * (last['high'] - last['low']))

    if bullish_block and current['close'] > last['high'].max():
        return "📈 Long breakout signal"
    elif bearish_block and current['close'] < last['low'].min():
        return "📉 Short breakout signal"
    else:
        return None

while True:
    candles = fetch_candles(symbol, timeframe, limit)
    signal = detect_imbalance_breakout(candles)
    if signal:
        print(f"[{symbol}] {signal}")
    else:
        print(f"[{symbol}] ⚪ No signal")
    time.sleep(15)

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

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

📌Triple Confirmation Strategy🧑‍💻

Индикаторы:🚀

1. EMA 200 — определение тренда.

2. RSI (14) — фильтрация перекупленности/перепроданности.

3. MACD — подтверждение импульса.

🧠 Логика стратегии:💸

✅ LONG:

- Цена выше EMA 200 (восходящий тренд).

- RSI пересекает снизу вверх уровень 50.

- MACD-гистограмма выше 0 и увеличивается.

❌ SHORT:

- Цена ниже EMA 200 (нисходящий тренд).

- RSI пересекает сверху вниз уровень 50.

- MACD-гистограмма ниже 0 и уменьшается.

import ccxt
import pandas as pd
import time
import ta

symbol = 'BTC/USDT'
timeframe = '5m'
limit = 200

exchange = ccxt.binance()

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

def apply_indicators(df):
    df['ema200'] = ta.trend.ema_indicator(df['close'], window=200).ema_indicator()
    df['rsi'] = ta.momentum.rsi(df['close'], window=14)
    macd = ta.trend.macd(df['close'])
    df['macd_hist'] = macd.macd_diff()
    return df

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

    # LONG setup
    if (
        last['close'] > last['ema200'] and
        prev['rsi'] < 50 and last['rsi'] > 50 and
        last['macd_hist'] > 0 > prev['macd_hist']
    ):
        return '🟢 LONG signal detected'

    # SHORT setup
    if (
        last['close'] < last['ema200'] and
        prev['rsi'] > 50 and last['rsi'] < 50 and
        last['macd_hist'] < 0 < prev['macd_hist']
    ):
        return '🔴 SHORT signal detected'

    return None

while True:
    df = fetch_ohlcv()
    df = apply_indicators(df)
    signal = check_signals(df)
    if signal:
        print(f"{signal} at {df.iloc[-1]['timestamp']}")
    else:
        print(f"No signal — {df.iloc[-1]['timestamp']}")
    time.sleep(30)

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

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

📌"Market Regime Detector" — Определение рыночного режима в реальном времени🧑‍💻

🧠 Что делает?🧨

Автоматически определяет фазу рынка:

📈 Тренд вверх

📉 Тренд вниз

🔁 Флэт/Боковик


📍 Для чего нужен?

📊 Использовать разные стратегии под разные рыночные режимы.

⛔ Фильтровать сигналы в плохих условиях (например, во флэте).

🔁 Менять параметры индикаторов/рисков при смене режима.

🧠 Логика определения режима:🛠️

Используется ADX (силу тренда) и наклон EMA (направление).

- Флэт: ADX < 20

Тренд вверх: ADX > 20 и EMA наклон вверх

- Тренд вниз: ADX > 20 и EMA наклон вниз

import ccxt
import pandas as pd
import ta
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '1h'
limit = 100

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

def detect_market_regime(df):
    df['ema'] = ta.trend.ema_indicator(df['close'], window=50).ema_indicator()
    adx = ta.trend.adx(df['high'], df['low'], df['close'], window=14)
    df['adx'] = adx.adx()

    last = df.iloc[-1]
    prev = df.iloc[-2]

    ema_slope = last['ema'] - prev['ema']

    if last['adx'] < 20:
        return '⚪ Флэт (Low ADX)'
    elif last['adx'] >= 20 and ema_slope > 0:
        return '🟢 Восходящий тренд'
    elif last['adx'] >= 20 and ema_slope < 0:
        return '🔴 Нисходящий тренд'
    else:
        return '❓ Неопределённый режим'

while True:
    df = get_ohlcv()
    regime = detect_market_regime(df)
    print(f"[{df.iloc[-1]['timestamp']}] Рыночный режим: {regime}")
    time.sleep(60)

#инструмент

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

📌"Adaptive Liquidity Trap"🧑‍💻

🧠 Идея:🚀

1. Вычисляется динамический диапазон ликвидности на основе объёма и спреда.

2. Если цена резко уходит за этот диапазон — вероятен ложный пробой.

3. Возврат в диапазон после прокола — сигнал на вход в противоположную сторону.

📊 Условия:🧨

- Вычисляется средний true spread: (high - low) / close

- Вычисляется волатильность и объём на 20 последних свечах

Если:

- Цена проколола границу диапазона ликвидности

- Объём выше среднего

- И следующая свеча вернулась обратно

→ сигнал на вход внутрь диапазона

🧩 Логика сигналов:💸

LONG, если был прокол вниз и возврат

SHORT, если был прокол вверх и возврат

import ccxt
import pandas as pd
import time

symbol = 'BTC/USDT'
timeframe = '15m'
limit = 100

exchange = ccxt.binance()

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

def generate_signal(df):
    df['spread'] = (df['high'] - df['low']) / df['close']
    df['avg_spread'] = df['spread'].rolling(window=20).mean()
    df['avg_volume'] = df['volume'].rolling(window=20).mean()

    last = df.iloc[-1]
    prev = df.iloc[-2]

    upper_bound = df['high'][-20:].mean() + df['avg_spread'].iloc[-1]
    lower_bound = df['low'][-20:].mean() - df['avg_spread'].iloc[-1]

    signal = None

    if prev['low'] < lower_bound and last['close'] > lower_bound and last['volume'] > df['avg_volume'].iloc[-1]:
        signal = "🟢 LONG сигнал: возврат в зону ликвидности"
    elif prev['high'] > upper_bound and last['close'] < upper_bound and last['volume'] > df['avg_volume'].iloc[-1]:
        signal = "🔴 SHORT сигнал: возврат в зону ликвидности"

    return signal

while True:
    try:
        df = fetch_data()
        signal = generate_signal(df)
        if signal:
            print(f"[{pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}] {signal}")
        else:
            print(f"[{pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}] 🔍 Нет сигнала")
        time.sleep(60)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(60)

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

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

📌"Сентимент Индикатор из Ордербука (Order Book Sentiment Tracker)"🧑‍💻

📌 Идея:🚀

Создаёт оценку настроений (bullish/bearish) на основе глубины стакана (order book). В отличие от классических индикаторов, он использует реальное распределение заявок на покупку и продажу в режиме реального времени.

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

1. С помощью ccxt подключается к Binance.

2. Загружает глубину стакана (order book).

3. Сравнивает суммарный объём заявок выше текущей цены (ask) и ниже текущей цены (bid).

4. Вычисляет имбаланс и определяет:

Bullish, если ниже цены заявок больше.

Bearish, если выше цены заявок больше.

Neutral, если равновесие.

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
depth_limit = 50  # Кол-во уровней стакана

def get_orderbook_sentiment(symbol):
    orderbook = exchange.fetch_order_book(symbol, limit=depth_limit)
    bids = orderbook['bids']
    asks = orderbook['asks']

    bid_volume = sum([price * amount for price, amount in bids])
    ask_volume = sum([price * amount for price, amount in asks])

    imbalance = bid_volume - ask_volume
    sentiment = "Neutral"
    if imbalance > ask_volume * 0.1:
        sentiment = "Bullish"
    elif imbalance < -bid_volume * 0.1:
        sentiment = "Bearish"

    print(f"Sentiment: {sentiment}")
    print(f"Bid Volume: {bid_volume:.2f}, Ask Volume: {ask_volume:.2f}, Imbalance: {imbalance:.2f}")
    print("-" * 50)

while True:
    try:
        get_orderbook_sentiment(symbol)
        time.sleep(5)
    except Exception as e:
        print("Error:", e)
        time.sleep(5)

#инструмент

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