📌"Volatility Cluster Compression & Release"
Анализ сжатий волатильности в виде кластеров и их последующих импульсных разгрузок. Основана на механике накопления энергии рынком и пробоя.🧑‍💻

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

Рынок имеет свойства чередовать фазы спокойствия и всплесков активности. При длительных сжатиях (низкой волатильности в узком диапазоне) происходит накопление ордеров. Выход из такого диапазона часто сопровождается сильным движением.

📌 Ключевые принципы:🚂

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

- Меряем волатильность через ATR или стандартное отклонение.

- Пробой уровня кластера в любую сторону с увеличением объема — сигнал на вход.

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

1. Последние N свечей имеют низкую волатильность (ниже медианы).

2. Цена движется в узком диапазоне (канал не шире X% от цены).

3. Объём начинает увеличиваться.

4. Свеча закрывается за пределами диапазона.

💡 Фильтры:💻

- Учитывать часы сессии (например, пробои чаще на пересечении Лондон/Нью-Йорк).

- Исключать ложные пробои через подтверждение следующей свечи.

🔁 Выход из позиции:🧨

Take Profit — фиксированный % или на первом уровне Fibo.

Stop Loss — внутри зоны кластера или за середину канала

import ccxt
import pandas as pd

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

exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)

df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['range'] = df['high'] - df['low']
df['atr'] = df['range'].rolling(window=14).mean()

# Выделим последние 10 свечей как кластер
cluster = df[-10:]
max_range = cluster['range'].max()
avg_range = df['atr'].mean()

if max_range < avg_range * 0.7:
    high = cluster['high'].max()
    low = cluster['low'].min()
    last_close = df['close'].iloc[-1]
    
    if last_close > high:
        print("🔔 BUY breakout signal")
    elif last_close < low:
        print("🔔 SELL breakout signal")
    else:
        print("⌛ Внутри кластера, ждем пробоя.")
else:
    print("❌ Нет сжатия, нет сигнала.")

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

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

📌"Asymmetrical Cluster Trap"
Использование несимметричных кластеров объёма для выявления ловушек маркет-мейкеров.🧑‍💻

🧠 Идея:🧨

Определять объёмные всплески с несимметричным распределением покупок и продаж в ограниченном ценовом диапазоне. Это может указывать на:

Ловушку для покупателей (резкий рост объёма с преобладанием покупок вверху диапазона перед разворотом вниз).

Ловушку для продавцов (аналогично, но внизу диапазона).

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

⚙️ Принцип работы:🛠️

1. Получаем данные с кластерной платформы (или проксируем с торговых данных через расчёт дельты по стакану).

2. Обнаруживаем участки с:

- узким диапазоном цен;

- резко возрастающим объёмом;

- дельтой сильно смещённой в одну сторону (более 80% покупок или продаж).

3. Сигнал на разворот против толпы.

import ccxt
import time

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

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

def detect_cluster_trap(candles):
    volumes = [c[5] for c in candles]
    highs = [c[2] for c in candles]
    lows = [c[3] for c in candles]

    range_avg = sum(h - l for h, l in zip(highs, lows)) / len(candles)
    vol_avg = sum(volumes) / len(candles)

    last = candles[-1]
    last_range = last[2] - last[3]
    last_volume = last[5]

    if last_range < 0.6 * range_avg and last_volume > 2 * vol_avg:
        print(f"[!] Потенциальная ловушка маркет-мейкера. Объёмный кластер на {symbol}")
        print(f"Цена: {last[4]} | Объём: {last_volume:.2f} | Дата: {time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(last[0]/1000))}")

while True:
    try:
        candles = fetch_candles()
        detect_cluster_trap(candles)
        time.sleep(60)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(10)

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

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

📌"Whale Shadow Tracker"
Отслеживание "теневых" действий крупных кошельков по задержке поступлений на биржи — до появления в стакане🧑‍💻

🧠 Идея:

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

Whale Shadow Tracker — инструмент, который:

- Отслеживает крупные входящие транзакции на биржи (например, > 100 BTC).

- Сравнивает время поступления на биржу с изменениями ликвидности в стакане.

- Сигнализирует, если активы уже пришли, но ещё не выставлены ордерами — возможная подготовка к резкому движению.

⚙️ Что делает инструмент:💸

1. Подключается к публичным блокчейн-источникам.

2. Находит крупные транзакции на биржевые адреса (выборка whitelist биржевых адресов).

3. Анализирует, когда активы пришли, и отслеживает ликвидность этих пар через ccxt.

4. Если ликвидность не изменилась через X минут → сигнал: крупный капитал в режиме ожидания → возможен импульс.

import ccxt
import time
import requests
from datetime import datetime

# Настройки Binance
symbol = 'BTC/USDT'
binance = ccxt.binance()
depth_limit = 10
interval_sec = 300
threshold_usd = 1_000_000

# Настройки Arkham
ARKHAM_API_KEY = 'YOUR_ARKHAM_API_KEY'  # 🔐 Вставь свой API ключ
ARKHAM_API_URL = 'https://api.arkhamintelligence.com/v1/labels/activity'

def get_orderbook_snapshot():
    orderbook = binance.fetch_order_book(symbol, limit=depth_limit)
    bids = orderbook['bids']
    asks = orderbook['asks']
    bid_liquidity = sum([price * amount for price, amount in bids])
    ask_liquidity = sum([price * amount for price, amount in asks])
    return {
        'timestamp': datetime.utcnow().isoformat(),
        'bid_liquidity': bid_liquidity,
        'ask_liquidity': ask_liquidity
    }

def check_arkham_whale_activity():
    headers = {'x-api-key': ARKHAM_API_KEY}
    params = {
        "entityType": "whale",  # можно менять на 'exchange', 'fund' и т.д.
        "chain": "bitcoin",
        "timeframe": "last_hour"
    }
    try:
        response = requests.get(ARKHAM_API_URL, headers=headers, params=params)
        if response.status_code == 200:
            data = response.json()
            if data.get('data'):
                print(f"🐋 Найдена ончейн-активность whale-адресов на BTC-сети ({len(data['data'])} записей)")
                return True
            else:
                print("🤫 Нет активности крупных адресов за последний час.")
        else:
            print(f"Ошибка запроса Arkham: {response.status_code}")
    except Exception as e:
        print(f"Ошибка при проверке Arkham: {e}")
    return False

def main():
    print("🚀 Запуск Whale Shadow Tracker...")
    snapshot_before = get_orderbook_snapshot()
    print("🕒 Снимок #1 получен. Ожидание 5 минут...\n")
    time.sleep(interval_sec)
    snapshot_after = get_orderbook_snapshot()

    delta_bid = abs(snapshot_after['bid_liquidity'] - snapshot_before['bid_liquidity'])
    delta_ask = abs(snapshot_after['ask_liquidity'] - snapshot_before['ask_liquidity'])

    if delta_bid < threshold_usd and delta_ask < threshold_usd:
        print("\n⚠️ Обнаружено стабильное состояние ликвидности. Проверяем ончейн-активность...")
        if check_arkham_whale_activity():
            print(f"🔔 ALERT: Whale Shadow — стакан спокоен, но whale-активность зафиксирована.")
        else:
            print("✅ Никаких признаков скрытой активности не найдено.")
    else:
        print(f"{datetime.utcnow().isoformat()} — Ликвидность изменилась значительно. Всё в порядке.")

if __name__ == "__main__":
    main()

#инструмент

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

📌"Sleeping Cluster Breakout"🧑‍💻

💡 Идея:🚀

На низких таймфреймах часто формируются плотности — “спящие кластеры” — уровни, где долгое время не было пробоя (цена касалась, но отскакивала). Когда крупные лимитные ордера около этих зон внезапно исчезают или съедаются — это часто приводит к резкому пробою.

📈 Логика:🛠️

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

2. Мониторим изменения ликвидности в ордербуке вокруг этого уровня.

3. Если:

- цена подходит к уровню;

- лимитная поддержка или сопротивление резко исчезают;

- объём увеличивается —

→ это может быть триггер на вход в позицию по пробою.

import ccxt
import time
from datetime import datetime

symbol = 'BTC/USDT'
exchange = ccxt.binance()
level_price = 110000  # 👈 Уровень, за которым следим
cluster_zone = 10    # Погрешность в $
liquidity_threshold = 500000  # Ликвидность в стакане, USD
check_interval = 10  # секунд

def get_orderbook():
    orderbook = exchange.fetch_order_book(symbol, limit=20)
    bids = orderbook['bids']
    asks = orderbook['asks']
    return bids, asks

def scan_liquidity_near_level(bids, asks):
    near_bids = [b for b in bids if abs(b[0] - level_price) <= cluster_zone]
    near_asks = [a for a in asks if abs(a[0] - level_price) <= cluster_zone]
    total_bid_liquidity = sum([b[0] * b[1] for b in near_bids])
    total_ask_liquidity = sum([a[0] * a[1] for a in near_asks])
    return total_bid_liquidity, total_ask_liquidity

def main():
    print(f"🛡 Мониторинг уровня {level_price}$ с зоной ±{cluster_zone}$")
    while True:
        bids, asks = get_orderbook()
        bid_liq, ask_liq = scan_liquidity_near_level(bids, asks)
        price = exchange.fetch_ticker(symbol)['last']

        print(f"\n[{datetime.utcnow().strftime('%H:%M:%S')}] Цена: {price:.2f}")
        print(f"   🔹 Ликвидность BUY: ${bid_liq:,.0f}")
        print(f"   🔸 Ликвидность SELL: ${ask_liq:,.0f}")

        # Условие пробоя уровня
        if level_price - cluster_zone <= price <= level_price + cluster_zone:
            if bid_liq < liquidity_threshold and ask_liq < liquidity_threshold:
                print("🚨 ALERT: Спящий кластер исчез. Возможен пробой уровня!")

        time.sleep(check_interval)

if __name__ == "__main__":
    main()

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

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

📌"Volatility-Weighted Signal Combiner"🧑‍💻

💡 Описание:🧨

Это модуль, который комбинирует сигналы из нескольких стратегий (например: импульс, откат, пробой уровня) — но взвешивает их в зависимости от текущей волатильности рынка. То есть, если рынок в фазе высокой волатильности — предпочтение отдается импульсным стратегиям, в фазе низкой — откатным и mean-reversion.

✅ Что делает:💸

- Получает волатильность по ATR или стандартному отклонению.

- Оценивает силу сигналов по каждой стратегии.

- Взвешивает силу с учётом волатильности.

- Выводит финальный алерт на вход в позицию, если итоговый вес выше порога.

import ccxt
import time
import statistics

# Настройки
symbol = 'BTC/USDT'
timeframe = '1m'
limit = 30  # Количество свечей для анализа

exchange = ccxt.binance()

def fetch_prices():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
    return [candle[4] for candle in ohlcv]  # Закрытия

def momentum_signal(prices):
    return 1 if prices[-1] > prices[-2] * 1.002 else 0

def mean_reversion_signal(prices):
    return 1 if prices[-1] < min(prices[-5:]) else 0

def calc_volatility(prices):
    return statistics.stdev(prices)

def weighted_combiner(momentum, reversion, volatility):
    if volatility > 30:
        return momentum * 0.7 + reversion * 0.3
    else:
        return momentum * 0.3 + reversion * 0.7

while True:
    try:
        prices = fetch_prices()
        momentum = momentum_signal(prices)
        reversion = mean_reversion_signal(prices)
        vol = calc_volatility(prices)
        signal_strength = weighted_combiner(momentum, reversion, vol)

        print(f"🟡 Цена: {round(prices[-1], 2)} | Волатильность: {round(vol, 2)} | Сигнал: {round(signal_strength, 2)}")

        if signal_strength > 0.6:
            print("✅ Сигнал на вход в ЛОНГ!\n")

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

#инструмент

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

📌"Behavioral Clustering" — кластеризация поведения толпы🧑‍💻

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

Эта стратегия опирается на кластеризацию аномального поведения розничных трейдеров — например, массовый FOMO (покупка на росте), panic selling (распродажа на падении), нерешительность (flat). Основная идея: когда большинство трейдеров делает одно и то же — делать наоборот.

Кластеры поведения определяются через:

📊 соотношение лонгов и шортов,

🔁 скорость смены открытых позиций (open interest delta),

🔥 рост объёма и волатильности,

🐳 активность мелких vs крупных адресов (например, через Arkham или Glassnode).

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

1. FOMO-кластер: Резкий рост объёма, цены и открытого интереса со стороны розницы → вход в шорт.

2. Паника-кластер: Массовая распродажа с низкими ценами и всплеском активности мелких адресов → вход в лонг.

3. Нерешительность: Плоский рынок, низкий объём, но рост открытого интереса → возможен прорыв → торгуется пробой.

import ccxt
import time

exchange = ccxt.binance()

symbol = 'BTC/USDT'
timeframe = '1m'

def get_data():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=20)
    prices = [c[4] for c in ohlcv]
    volumes = [c[5] for c in ohlcv]
    return prices, volumes

def detect_fomo(prices, volumes):
    price_jump = prices[-1] > max(prices[:-1]) * 1.01
    volume_spike = volumes[-1] > sum(volumes[-5:-1]) / 2
    return price_jump and volume_spike

def detect_panic(prices, volumes):
    price_drop = prices[-1] < min(prices[:-1]) * 0.99
    volume_spike = volumes[-1] > sum(volumes[-5:-1]) / 2
    return price_drop and volume_spike

while True:
    try:
        prices, volumes = get_data()

        if detect_fomo(prices, volumes):
            print("🔻 Обнаружен FOMO кластер → Сигнал на ШОРТ")

        elif detect_panic(prices, volumes):
            print("🔺 Обнаружена ПАНИКА → Сигнал на ЛОНГ")

        else:
            print("⚪ Рынок без явных кластеров. Ожидание...")

        time.sleep(60)

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

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

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

📌"Wallet Flow Radar" – Радар движения средств между адресами и биржами🧑‍💻

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

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

Особенности:🚀

- Анализ потоков на основе публичных данных (Arkham, Blockchair, Whale Alert, либо через ноды).

- Фильтрация по типу адреса: биржа → личный кошелёк, наоборот, OTC, холодные и горячие кошельки.

- Автоматическое определение направлений и масштабов движения.

- Консольные алерты и логгирование.

- Выдача сигнала: если, например, 1000 BTC отправлены с кошелька → биржа, это может быть сигнал к продаже.

import time
import requests

ARKHAM_API_KEY = 'your_api_key_here'
THRESHOLD_USD = 1_000_000  # отслеживать транзакции свыше $1M

def check_flows():
    url = 'https://api.arkhamintelligence.com/transactions'
    headers = {'Authorization': f'Bearer {ARKHAM_API_KEY}'}

    response = requests.get(url, headers=headers)
    data = response.json()

    for tx in data.get('transactions', []):
        amount_usd = tx['valueUSD']
        direction = tx['direction']  # 'in' or 'out'
        exchange_related = 'exchange' in tx['fromLabel'].lower() or 'exchange' in tx['toLabel'].lower()

        if amount_usd > THRESHOLD_USD and exchange_related:
            if direction == 'in':
                print(f"⚠️ Крупный ввод на биржу: {amount_usd:.0f} USD → возможный дамп")
            elif direction == 'out':
                print(f"📦 Крупный вывод с биржи: {amount_usd:.0f} USD → возможно накопление")

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

#инструмент

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

📌"Volatility Compression Scanner" — Сканер Сжатия Волатильности🧑‍💻

📌 Цель:🚀

Автоматически находить активы, находящиеся в фазе низкой волатильности (консолидации), что часто предшествует сильному движению — вверх или вниз.

💡 Почему это полезно?🧨

- Большинство импульсов и прорывов начинаются после сжатия диапазона.

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

⚙️ Принцип:🛠️

1. Загружаем данные по свечам (например, 1h) для множества инструментов.

2. Вычисляем диапазон свечей (high - low) за последние N периодов.

3. Ищем инструменты, у которых текущая волатильность — в X раз ниже средней.

4. Выводим список — возможные кандидаты на пробой.

import ccxt
import time
from statistics import mean

exchange = ccxt.binance()
symbols = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'XRP/USDT']
timeframe = '1h'
lookback = 20
compression_threshold = 0.3  # текущая волатильность < 30% от средней

def get_volatility(symbol):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=lookback)
    ranges = [candle[2] - candle[3] for candle in ohlcv[:-1]]  # high - low
    current = ohlcv[-1][2] - ohlcv[-1][3]
    avg_range = mean(ranges)
    if avg_range == 0:
        return None
    compression = current / avg_range
    return compression

while True:
    print(f"\n🔍 Поиск сжатия волатильности ({time.strftime('%X')}):")
    for symbol in symbols:
        try:
            ratio = get_volatility(symbol)
            if ratio is not None and ratio < compression_threshold:
                print(f"🟡 {symbol} — волатильность сжата ({ratio:.2f})")
        except Exception as e:
            print(f"[Ошибка] {symbol}: {e}")
    time.sleep(300)  # 5 минут

#инструмент

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

📌"Time-Weighted Bid Walls" (TWBW) — вход на основе жизни крупных заявок🧑‍💻

🧠 Идея:🚀

Отслеживать bid-стенки (стены заявок на покупку) в order book и определять, как долго они удерживаются на месте.
Если большая лимитная заявка долго не исчезает и не исполняется, это может сигнализировать о поддержке, за которой стоит реальный спрос.

Если стена держится долго — входим в лонг.

⚙️ Принцип работы:🛠️

1. Сканим стакан с периодичностью

2. Фиксируем крупные заявки (на 10–20x больше среднего объема).

3. Засекаем, сколько времени они остаются на месте.

4. Если заявка живёт дольше N секунд и не исполняется — это сигнал на вход в лонг.

📉 Выход:💸

- Если стена исчезает → выход по стопу.

- Если цена уходит вверх на X% → тейк.

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
depth_limit = 50
threshold_volume = 10  # например, 10 BTC
wall_lifetime = 60  # сек

walls = {}

def fetch_order_book():
    order_book = exchange.fetch_order_book(symbol, limit=depth_limit)
    bids = order_book['bids']
    return bids

def track_bid_walls():
    global walls
    bids = fetch_order_book()

    now = time.time()
    for price, volume in bids:
        if volume >= threshold_volume:
            key = str(price)
            if key not in walls:
                walls[key] = now
            elif now - walls[key] >= wall_lifetime:
                print(f"🟢 Сигнал LONG: удерживается стенка {volume:.2f} BTC @ {price:.2f}")
                walls[key] = now + 99999  # не триггерить повторно
        else:
            # удаляем, если объем упал
            walls.pop(str(price), None)

    # чистим старые
    walls = {k: v for k, v in walls.items() if now - v <= 2 * wall_lifetime}

while True:
    try:
        track_bid_walls()
        time.sleep(5)
    except Exception as e:
        print(f"[Ошибка] {e}")
        time.sleep(10)

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

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

📌"Whale Fingerprint Tracker"🧑‍💻

Отслеживание уникального поведения китов по их характерным шаблонам торгов

💡 Суть идеи:🧨

Крупные адреса (киты) зачастую входят и выходят с рынка по характерным шаблонам, например:

- Разделение ордера на множество мелких покупок

- Повторяющиеся последовательности ордеров

- Частота и время активности

- Стратегии ложных пробоев

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

📌 Что делает скрипт:🛠️

- Сканирует адреса с наибольшим объёмом сделок (можно с Arkham API или Glassnode/Whale-alert)

- Сохраняет шаблон поведения (время, объёмы, направления, частота)

- Сравнивает в реальном времени активность с этими шаблонами

- Выдаёт сигнал в консоль, если похожая активность снова появляется

🐍 Пример архитектуры на Python (упрощённый скелет):

# Псевдокод — пример логики анализа активности
class WhaleTracker:
    def __init__(self):
        self.whale_fingerprints = {}

    def add_whale_pattern(self, address, pattern):
        self.whale_fingerprints[address] = pattern

    def fetch_live_activity(self):
        # Тут может быть API Arkham, Whale Alert или собственный mempool scanner
        return [
            {"address": "0xabc...", "volume": 1_200_000, "type": "buy", "timestamp": 1720000000},
            {"address": "0xdef...", "volume": 950_000, "type": "sell", "timestamp": 1720000050}
        ]

    def check_for_matches(self):
        for tx in self.fetch_live_activity():
            address = tx["address"]
            if address in self.whale_fingerprints:
                print(f"🔍 Whale activity match! Address: {address} — Type: {tx['type']} — Volume: {tx['volume']}")

#инструмент

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

📌"Анализ ликвидности по часам (Smart Liquidity Sweep)"🧑‍💻

💡 Идея:🚀

Использовать временные зоны ликвидности и поведение цены вблизи часовых максимумов/минимумов, чтобы ловить ложные пробои и быстрое возвращение в диапазон (mean reversion).

⚙️ Принцип работы:🛠️

1. Определяем локальные high/low на каждом часе.

2. Если текущая цена пробивает прошлый максимум/минимум, но объём не растёт — это потенциальный ложный пробой.

3. Вход — в противоположную сторону от пробоя с коротким стопом.

4. TP — возврат в средний диапазон свечи (mean of high and low).

import ccxt
import time
from datetime import datetime
from statistics import mean

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '1h'
check_tf = '1m'
lookback = 3

def get_hourly_levels():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=lookback)
    last = ohlcv[-2]  # предыдущий полный час
    return {
        "high": last[2],
        "low": last[3],
        "mean": (last[2] + last[3]) / 2
    }

def get_latest_price():
    ohlcv = exchange.fetch_ohlcv(symbol, check_tf, limit=2)
    return ohlcv[-1][4]  # close price

levels = get_hourly_levels()

print(f"⏱ Мониторинг {symbol}...")
print(f"📉 Предыдущий час: high={levels['high']:.2f}, low={levels['low']:.2f}")

while True:
    try:
        price = get_latest_price()
        ts = datetime.utcnow().strftime("%H:%M:%S")

        if price > levels['high']:
            print(f"[{ts}] 🔺 Потенциальный ложный пробой вверх: {price:.2f}")
        elif price < levels['low']:
            print(f"[{ts}] 🔻 Потенциальный ложный пробой вниз: {price:.2f}")
        else:
            print(f"[{ts}] 🔄 Внутри диапазона: {price:.2f}")

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

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

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

📌Универсальный “Dashboard” мониторинга позиций и портфеля🧑‍💻

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

- Показывает все открытые позиции и баланс на нескольких биржах (Binance, Bybit, OKX и т.д.)

- Отображает PnL, среднюю цену входа, объём, цены ликвидации

- Подсвечивает рисковые позиции

- Обновляется каждые N секунд

- Можно использовать в консоли или как веб-интерфейс (Flask/FastAPI + Tailwind)

import ccxt
import time
from tabulate import tabulate

api_keys = {
    'binance': {
        'apiKey': 'your_api_key',
        'secret': 'your_secret',
    },
    # Добавь другие биржи при необходимости
}

def fetch_positions(exchange):
    try:
        exchange.load_markets()
        balance = exchange.fetch_balance()
        positions = []

        if hasattr(exchange, 'fetch_positions'):
            for pos in exchange.fetch_positions():
                if float(pos.get('contracts', 0)) != 0:
                    positions.append({
                        'symbol': pos['symbol'],
                        'size': pos['contracts'],
                        'entry': pos['entryPrice'],
                        'mark': pos['markPrice'],
                        'pnl': pos['unrealizedPnl'],
                        'side': pos['side']
                    })
        return balance, positions
    except Exception as e:
        print(f"Error: {e}")
        return {}, []

def monitor():
    while True:
        print(f"\n=== {time.strftime('%Y-%m-%d %H:%M:%S')} ===")
        for name, creds in api_keys.items():
            print(f"\n📊 {name.upper()}:")

            exchange = getattr(ccxt, name)({
                'apiKey': creds['apiKey'],
                'secret': creds['secret'],
                'enableRateLimit': True,
                'options': {'defaultType': 'future'}
            })

            balance, positions = fetch_positions(exchange)
            if positions:
                print(tabulate(positions, headers="keys", tablefmt="fancy_grid"))
            else:
                print("Нет открытых позиций.")

        time.sleep(30)

monitor()

🛠 Возможности для расширения:💻

- Веб-интерфейс с Flask или FastAPI

- Telegram/Discord-уведомления при превышении PnL или падении баланса

- Графики изменения позиций

- Подключение ко всем типам аккаунтов: spot, futures, margin

📌 Почему это must-have:🚀

- Не нужно логиниться на 3+ биржи

- Всё видно в одном месте

- Быстрая реакция на просадку или успех

- Отлично подходит как трейдерам, так и командам с PAMM-счетами

#инструмент

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

📌"Поведенческое зеркало толпы"-
Торговля против эмоциональных паттернов розничных трейдеров🧑‍💻

📌 Идея:🧨

Поведение толпы на рынке часто подвержено эмоциям — паника в просадке и жадность при росте. Эта стратегия использует метрики соцсигналов и объёмов в сочетании с волатильностью и RSI, чтобы идти против эмоционального перегрева.

🔍 Логика:🛠️

Индикатор перегретости толпы = spike в упоминаниях актива + spike в объёмах + RSI > 70 (или RSI < 30 для паники)

При перегреве — открывается контртрендовая позиция:

- если рост на эмоциях — открытие шорта

- если паника на дне — открытие лонга

💡 Используемые данные:

📢 Соцсигналы: число упоминаний в Twitter, Reddit, Telegram (через API, например LunarCrush, Santiment, CrowdTangle)

📊 Объёмы: рост объёма спота/фьючей

📉 RSI: стандартный индикатор

🔄 Волатильность: ATR, либо std dev цены за период

import ccxt
import time
from datetime import datetime
import random  # Имитация соц.сигналов

exchange = ccxt.binance()
symbol = 'BTC/USDT'
rsi_threshold_high = 70
rsi_threshold_low = 30

def get_mock_social_hype_score():
    # Здесь должен быть реальный парсинг с API
    return random.randint(0, 100)

def calculate_rsi(prices, period=14):
    deltas = [prices[i+1]-prices[i] for i in range(len(prices)-1)]
    ups = [delta for delta in deltas if delta > 0]
    downs = [-delta for delta in deltas if delta < 0]
    avg_gain = sum(ups[-period:])/period if ups else 0.001
    avg_loss = sum(downs[-period:])/period if downs else 0.001
    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

while True:
    candles = exchange.fetch_ohlcv(symbol, timeframe='5m', limit=100)
    close_prices = [c[4] for c in candles]
    rsi = calculate_rsi(close_prices)
    social_score = get_mock_social_hype_score()

    if rsi > rsi_threshold_high and social_score > 80:
        print(f"[{datetime.now()}] ⚠️ ПЕРЕГРЕВ! RSI={rsi:.2f}, Hype={social_score} — сигнал на ШОРТ")
    elif rsi < rsi_threshold_low and social_score > 80:
        print(f"[{datetime.now()}] 😱 ПАНИКА! RSI={rsi:.2f}, Hype={social_score} — сигнал на ЛОНГ")
    else:
        print(f"[{datetime.now()}] 🕒 RSI={rsi:.2f}, Hype={social_score} — нет сигнала")

    time.sleep(60)

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

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

📌"Fakeout Hunter" (Охотник за ложными пробоями)🧑‍💻

💡 Идея:

Многие трейдеры теряют деньги на пробоях, которые оказываются ложными. Стратегия “Fakeout Hunter” строится на поиске и входе в рынок в момент, когда цена возвращается обратно после ложного пробоя уровня.

🧠 Логика:🛠️

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

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

3. Ждём возврата под уровень (или над уровень, если это ложный пробой вниз).

4. Входим в противоположную сторону (на откате) с коротким стопом.

5. Фильтруем по объёму и ATR — нужна высокая волатильность.

📊 Пример условий входа:💸

- Цена пробила уровень сопротивления, но закрытие свечи ниже уровня → ложный пробой

- Объём выше среднего → подтверждение интереса

- RSI в зоне перекупленности → усиливает сигнал на шорт

import ccxt
import pandas as pd
import time

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

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

def detect_fakeout(df):
    last = df.iloc[-1]
    prev = df.iloc[-2]
    
    resistance = df['high'][-20:-1].max()
    support = df['low'][-20:-1].min()

    fakeout_up = prev['high'] > resistance and last['close'] < resistance
    fakeout_down = prev['low'] < support and last['close'] > support

    if fakeout_up:
        print(f"[{last['timestamp']}] 🚨 Fakeout вверх! Потенциал шорта: {symbol}")
    elif fakeout_down:
        print(f"[{last['timestamp']}] 🚨 Fakeout вниз! Потенциал лонга: {symbol}")

while True:
    df = fetch_ohlcv()
    detect_fakeout(df)
    time.sleep(60)

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

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

📌Risk Manager CLI — Консольный помощник по управлению рисками🧑‍💻

📌 Что это?

Это консольное приложение на Python, которое помогает трейдеру автоматически рассчитывать размер позиции, расставлять цели и показывать риск/профит-профиль до входа в сделку.

Работает с клавиатуры — вводишь цену входа, стоп и риск в %, а инструмент показывает, сколько нужно купить, на какой объём, где тейк и каков RR (risk/reward).

🧠 Для чего нужен:🚀

- Избавляет от ручных расчётов при открытии позиции

- Учит дисциплине и риск-менеджменту

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

def risk_manager():
    balance = float(input("💰 Введите сумму депозита (USDT): "))
    risk_percent = float(input("📉 Риск на сделку (%) например 1: ")) / 100
    entry_price = float(input("📥 Цена входа: "))
    stop_price = float(input("🛑 Стоп-лосс: "))
    take_profit = float(input("🎯 Тейк-профит: "))

    risk_amount = balance * risk_percent
    stop_distance = abs(entry_price - stop_price)
    position_size = risk_amount / stop_distance
    position_value = position_size * entry_price
    potential_profit = abs(take_profit - entry_price) * position_size
    rr_ratio = potential_profit / risk_amount

    print("\n📊 Результаты:")
    print(f"🔹 Риск на сделку: {risk_amount:.2f} USDT")
    print(f"🔹 Размер позиции: {position_size:.4f} BTC")
    print(f"🔹 Объём позиции: {position_value:.2f} USDT")
    print(f"🔹 Потенциальная прибыль: {potential_profit:.2f} USDT")
    print(f"🔹 Соотношение риск/прибыль: 1:{rr_ratio:.2f}")

if __name__ == "__main__":
    risk_manager()

#инструмент

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

📌"Алгоритм охоты за ликвидностью" (Liquidity Sweep Sniper)🧑‍💻

🧠 Идея:🚀

Стратегия ориентирована на поведение крупных участников (смарт-денег), которые охотятся за стопами розницы. Мы ищем зоны ложного пробоя локальных экстремумов (high/low), где рынок "съедает" ликвидность, и заходим против направления прокола после подтверждения.

📊 Сценарий входа (лонг):💸

1. Цена пробивает локальный минимум → забирает стопы → формируется хвост (пинбар, молот).

2. Возврат выше уровня прокола → подтверждение ложности.

3. Вход в лонг по закрытию подтверждающей свечи.

4. Стоп — под ложный прокол, Тейк — в область предыдущей консолидации или по R:R 1:2+.

🤖 Особенности:💻

- Работает на любом таймфрейме от 5m до 4H, особенно хорошо в азиатскую сессию.

- Не требует индикаторов, только цена и объёмы (если есть).

- Подходит как для мануального, так и алготрейдинга.

- Можно расширить до автоопределения "стопохантов".

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

- Заходит в момент смены дисбаланса (где много ликвидности)

- Позволяет точно выставлять стоп

- Даёт высокое соотношение риск/прибыль (1:2 — 1:5)

import ccxt
import time

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

def get_candles(symbol, tf='5m', limit=50):
    ohlcv = exchange.fetch_ohlcv(symbol, tf, limit=limit)
    return [{'time': c[0], 'open': c[1], 'high': c[2], 'low': c[3], 'close': c[4]} for c in ohlcv]

def detect_liquidity_sweep(candles):
    last = candles[-1]
    prev = candles[-2]

    # Условие: был ложный прокол минимума
    if last['low'] < prev['low'] and last['close'] > prev['low']:
        print("🟢 ВОЗМОЖЕН ЛОНГ (поглощение ликвидности вниз)")
        print(f"Цена: {last['close']}")
    elif last['high'] > prev['high'] and last['close'] < prev['high']:
        print("🔴 ВОЗМОЖЕН ШОРТ (поглощение ликвидности вверх)")
        print(f"Цена: {last['close']}")

def run_monitor():
    print("🔍 Мониторинг ликвидности...")
    while True:
        try:
            candles = get_candles(symbol, timeframe)
            detect_liquidity_sweep(candles)
            time.sleep(60)  # каждую минуту
        except Exception as e:
            print(f"Ошибка: {e}")
            time.sleep(60)

if __name__ == "__main__":
    run_monitor()

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

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

📌“Универсальный калькулятор риска и позиции с журналом сделок”🧑‍💻

📌 Назначение:🛠️

Позволяет в режиме консоли:

- рассчитывать объём позиции и стоп-лосс в $ и %

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

- логировать каждую сделку в CSV

- собирать статистику: winrate, avg R, expectancy и др.

- быстро переключаться между биржами/активами

import csv
import datetime

def calculate_position(balance, risk_pct, entry_price, stop_price, leverage=1):
    risk_usd = balance * (risk_pct / 100)
    stop_loss_per_unit = abs(entry_price - stop_price)
    position_size = (risk_usd / stop_loss_per_unit) * leverage
    return round(position_size, 4), round(risk_usd, 2)

def log_trade(symbol, direction, entry, stop, tp, size, risk_usd, result=None):
    with open('trade_journal.csv', 'a', newline='') as f:
        writer = csv.writer(f)
        writer.writerow([
            datetime.datetime.now().strftime('%Y-%m-%d %H:%M'),
            symbol, direction, entry, stop, tp, size, risk_usd, result
        ])

def main():
    print("📊 Расчёт позиции и логгер сделок")
    symbol = input("Тикер (например, BTC/USDT): ")
    direction = input("Направление (long/short): ")
    balance = float(input("Баланс аккаунта ($): "))
    risk_pct = float(input("Риск на сделку (%): "))
    entry = float(input("Цена входа: "))
    stop = float(input("Цена стоп-лосса: "))
    tp = float(input("Тейк-профит (по желанию, можно 0): "))
    leverage = float(input("Плечо (по умолчанию 1): ") or 1)

    size, risk_usd = calculate_position(balance, risk_pct, entry, stop, leverage)
    print(f"👉 Объём позиции: {size} единиц актива\n📉 Риск в долларах: ${risk_usd}")
    
    log_trade(symbol, direction, entry, stop, tp, size, risk_usd)
    print("✅ Сделка записана в журнал (trade_journal.csv)")

if __name__ == "__main__":
    main()

#инструмент

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

📌“Индикаторная флуктуация + импульсный фильтр на объёме”🧑‍💻

🎯 Идея:🧨

Комбинируется флуктуация RSI в нейтральной зоне (45–55) с импульсным скачком объёма, что говорит о потенциальной скорой смене направления тренда. Это контр-интуитивная стратегия, основанная не на перекупленности/перепроданности, а на “вялой зоне”, из которой рынок выходит резко.

📌 Условия входа (Long):💸

1. RSI(14) пересекает снизу вверх уровень 50

2. RSI находится в диапазоне 45–55 не менее 20 свечей (накапливается энергия)

3. Происходит скачок объёма (Volume > SMA(20) * 1.8)

4. Подтверждение: цена пробивает локальный high последних 10 свечей

💥 Вход на пробой high
📍 SL — под low накопления
🎯 TP1 — R=1.5, TP2 — трейлинг или exit по RSI>70

import ccxt
import pandas as pd
import ta

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

ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['rsi'] = ta.momentum.RSIIndicator(close=df['close']).rsi()
df['vol_sma'] = df['volume'].rolling(20).mean()

# Условия
rsi_range = (df['rsi'] > 45) & (df['rsi'] < 55)
df['in_range'] = rsi_range.rolling(20).apply(lambda x: x.all(), raw=True)
df['volume_spike'] = df['volume'] > df['vol_sma'] * 1.8
df['signal'] = (df['rsi'].shift(1) < 50) & (df['rsi'] > 50) & df['in_range'] & df['volume_spike']

signals = df[df['signal'] == True]
print(signals[['timestamp', 'close', 'rsi', 'volume']].tail())

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

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

"Тепловая карта стакана (Order Book Heatmap) в терминале"🧑‍💻

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

Показывает визуализацию лимитных ордеров в стакане — где стоят крупные заявки, на каких ценовых уровнях есть кластеры. Это помогает видеть уровни сопротивления/поддержки, спуфинг, а также реакции на крупные лимитные ордера.

import ccxt
import time
import numpy as np
import os

exchange = ccxt.binance()
symbol = 'BTC/USDT'
depth_limit = 50  # Глубина стакана

def fetch_orderbook():
    order_book = exchange.fetch_order_book(symbol, limit=depth_limit)
    bids = order_book['bids']
    asks = order_book['asks']
    return bids, asks

def render_heatmap(bids, asks):
    os.system('cls' if os.name == 'nt' else 'clear')
    print(f"🔥 Order Book Heatmap: {symbol}\n")

    # Конвертация объема в "яркость" символов
    def format_line(side, color):
        for price, volume in side:
            intensity = int(min(volume, 5) * 10)  # Ограничить яркость
            bar = "#" * intensity
            print(f"{color}{price:>12.2f} {bar:<50} ({volume:.3f})\033[0m")

    print("🟥 Asks:")
    format_line(reversed(asks[:depth_limit]), "\033[91m")  # Красный

    print("\n🟩 Bids:")
    format_line(bids[:depth_limit], "\033[92m")  # Зелёный

while True:
    bids, asks = fetch_orderbook()
    render_heatmap(bids, asks)
    time.sleep(3)

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

#инструмент

📌"Противоимпульсный выброс (Antipump Trap)"🧑‍💻

📌 Идея:🚀

Когда происходит резкий ценовой импульс вверх на повышенном объёме, многие новички входят в лонг на хаях. Но если цена не закрепляется выше импульса — вероятен разворот и падение. Мы используем это поведение толпы как ловушку.

📊 Условия входа в шорт:

1. Свеча (1m или 5m) с телом > 2× среднего тела за последние 20 свечей.

2. Объём на свече > 2× среднего объёма.

3. Следующая свеча закрывается ниже 50% тела предыдущей (не подтверждён импульс).

4. Открытие шорт-позиции со стопом чуть выше хая импульсной свечи.

🎯 TP/SL:

TP: 1.5× риск

SL: выше хая импульсной свечи

Дополнительный выход при росте объёма на падении (фиксация прибыли)

import ccxt
import pandas as pd
import time

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

def fetch_ohlcv():
    data = exchange.fetch_ohlcv(symbol, timeframe=tf, limit=100)
    df = pd.DataFrame(data, columns=['time', 'open', 'high', 'low', 'close', 'volume'])
    df['body'] = abs(df['close'] - df['open'])
    return df

def detect_antipump(df):
    avg_body = df['body'][:-2].tail(20).mean()
    avg_vol = df['volume'][:-2].tail(20).mean()
    
    last = df.iloc[-2]
    confirm = df.iloc[-1]
    
    if last['body'] > 2 * avg_body and last['volume'] > 2 * avg_vol:
        body_mid = (last['open'] + last['close']) / 2
        if confirm['close'] < body_mid:
            print(f"🔻 Antipump сигнал на шорт ({symbol}) @ {confirm['close']}")
            print(f"   SL: {last['high']}, TP: ~{confirm['close'] - (last['high'] - confirm['close'])*1.5}")

while True:
    df = fetch_ohlcv()
    detect_antipump(df)
    time.sleep(15)

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

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

📌Автоматический Дневной Отчёт по Портфелю и Рынку🧑‍💻

Назначение:🧨

Каждое утро (или в выбранное время) скрипт автоматически собирает данные по рынку, активам, новостям, ключевым уровням и составляет краткий отчёт в консоль или в Telegram. Это личный дайджест трейдера — как утренний брифинг на десктопе.

✅ Функции:🛠️

- Портфель и изменение за сутки (% и $)

- Текущие уровни поддержки/сопротивления (по ATR/каналу/SR)

- Волатильность по ключевым монетам

- Горячие монеты дня (по объёму и импульсу)

- Важные экономические события (например, CPI, FOMC — можно через календарь событий)

- Сводка по рынку: BTC, ETH, фондовый индекс, золото

import ccxt
import datetime

exchange = ccxt.binance()
symbol_list = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT']

def fetch_summary(symbol):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe='1d', limit=2)
    today = ohlcv[-1]
    yesterday = ohlcv[-2]

    change = (today[4] - yesterday[4]) / yesterday[4] * 100
    return {
        'pair': symbol,
        'close': today[4],
        'change': round(change, 2)
    }

def daily_report():
    print(f"📅 Daily Market Summary: {datetime.date.today()}")
    for sym in symbol_list:
        summary = fetch_summary(sym)
        print(f"🔹 {summary['pair']}: ${summary['close']} ({summary['change']}%)")

if __name__ == "__main__":
    daily_report()

💡 Расширения:

- Добавить новости через API (например, CryptoPanic)

- Интеграция с Telegram Bot API

- Хранение истории в SQLite/CSV

- Графическое отображение динамики портфеля (через matplotlib или plotly)

- Оповещение, если дневное изменение > 5%

#инструмент

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