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

📌“Обратная Точка Баланса” (Reverse Balance Point)🧑‍💻

💡 Идея:🚀

Внутридневные и краткосрочные трейдеры часто используют уровни VWAP или среднюю цену как "баланс". Эта стратегия делает противоположное — ищет моменты, когда цена слишком долго избегает баланса, и ловит возможность возврата к средней цене.

⚙️ Условия входа (LONG):

1. Цена открылась и удерживается выше VWAP как минимум 2 часа.

2. После сильного роста (>1.5% от VWAP) цена резко уходит ниже VWAP на повышенном объёме.

3. Вход в LONG на первом отскоке от нижнего стандартного отклонения VWAP (или уровней -1/-2 SD).

4. Цель — возврат к VWAP, Stop — под предыдущий минимум.

📉 SHORT:

То же самое, но в зеркальном отражении (сначала удержание под VWAP, затем пробой вверх и возврат).

import ccxt
import pandas as pd
import numpy as np

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '15m'
limit = 120  # 30 часов

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

# VWAP
df['typical_price'] = (df['high'] + df['low'] + df['close']) / 3
df['cum_tp_vol'] = (df['typical_price'] * df['volume']).cumsum()
df['cum_vol'] = df['volume'].cumsum()
df['vwap'] = df['cum_tp_vol'] / df['cum_vol']

# Отношение к VWAP
df['distance'] = (df['close'] - df['vwap']) / df['vwap'] * 100

# Сигнал на возврат к балансу
last = df.iloc[-1]
prev = df.iloc[-2]

if prev['distance'] > 1.5 and last['distance'] < 0:
    print("🔁 Возможен возврат к VWAP: Рассмотреть LONG")

elif prev['distance'] < -1.5 and last['distance'] > 0:
    print("🔁 Возможен возврат к VWAP: Рассмотреть SHORT")
else:
    print("⏳ Нет сигнала. Ждём аномалии")

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

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

📌“Маркет-сканер волатильности с кластеризацией аномалий”🧑‍💻

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

Этот инструмент в реальном времени анализирует десятки монет (например, с Binance через ccxt) и ищет аномальные ценовые или объёмные всплески, применяя кластеризацию (например, DBSCAN или KMeans).

Он не просто говорит: "Цена выросла на 4%", а выявляет группы активов, которые ведут себя одинаково, но резко отличаются от других.

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

- Находить монеты в начале пампа.

- Выявлять нелогичные движения (арбитражные окна).

- Искать сигналы толпы / инсайдерского спроса.

import ccxt
import pandas as pd
from sklearn.cluster import DBSCAN
from sklearn.preprocessing import StandardScaler
import numpy as np

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

data = []

for symbol in symbols[:50]:
    try:
        ohlcv = exchange.fetch_ohlcv(symbol, '5m')[-3:]
        df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        pct_change = (df['close'].iloc[-1] - df['close'].iloc[0]) / df['close'].iloc[0]
        vol_change = df['volume'].iloc[-1] - df['volume'].iloc[0]
        data.append([symbol, pct_change, vol_change])
    except:
        continue

df = pd.DataFrame(data, columns=['symbol', 'pct_change', 'vol_change'])

# Кластеризация
X = StandardScaler().fit_transform(df[['pct_change', 'vol_change']])
clustering = DBSCAN(eps=0.8, min_samples=3).fit(X)
df['cluster'] = clustering.labels_

# Выводим аномалии
anomalies = df[df['cluster'] == -1]
print("📊 Обнаружены аномалии:\n", anomalies)

#инструмент

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

📌“Алгокарта кросс-корреляций в реальном времени”🧑‍💻

📌 Что это?🧨

Скрипт, строящий динамическую тепловую карту кросс-корреляций между несколькими криптоактивами (например, BTC, ETH, SOL, BNB и т.д.) в реальном времени — на малых таймфреймах (1m/5m), и обновляющий её в терминале или графически.

🧠 Как помогает трейдеру/алготрейдеру?💸

Находит активы, которые ведут себя синхронно или наоборот расходятся, что важно:

- Для парного трейдинга

- Для арбитражных стратегий

- Для фильтрации сигналов

- Для контроля риска по скоррелированным активам

import ccxt
import pandas as pd
import time
from colorama import Fore, Style, init

init(autoreset=True)

symbols = ['BTC/USDT', 'ETH/USDT', 'BNB/USDT', 'SOL/USDT', 'XRP/USDT']
exchange = ccxt.binance()
timeframe = '1m'
limit = 100

def fetch_ohlcv(symbol):
    try:
        ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
        df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        return df['close']
    except Exception as e:
        print(f"[{symbol}] Ошибка: {e}")
        return pd.Series(dtype=float)

def print_correlation_matrix(data):
    df = pd.DataFrame(data)
    corr_matrix = df.corr()

    print("\nКросс-корреляция:")
    headers = corr_matrix.columns
    print("     " + "  ".join([f"{h[:4]}" for h in headers]))

    for i, row in corr_matrix.iterrows():
        row_str = f"{i[:4]}  "
        for val in row:
            if val > 0.75:
                row_str += Fore.GREEN + f"{val:.2f}" + Style.RESET_ALL + "  "
            elif val < -0.75:
                row_str += Fore.RED + f"{val:.2f}" + Style.RESET_ALL + "  "
            else:
                row_str += f"{val:.2f}  "
        print(row_str)

if __name__ == "__main__":
    data = {}
    for symbol in symbols:
        closes = fetch_ohlcv(symbol)
        if not closes.empty:
            data[symbol] = closes
        time.sleep(0.5)

    if len(data) >= 2:
        print_correlation_matrix(data)
    else:
        print("Недостаточно данных.")

#инструмент

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

📌“Обратный эхолот ордербука”🧑‍💻

🧠 Идея:🚀

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

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

1. Сканируем стакан (order book) по выбранной паре с определённой частотой (например, каждые 5 секунд).

2. Ищем крупные заявки (например, 5x выше среднего).

3. Проверяем, исчезают ли они внезапно — это может быть признаком попытки манипулировать.

4. При исчезновении крупной заявки — открываем позицию против этого "давления":

Если исчез крупный buy — открываем short.

Если исчез крупный sell — открываем long.

📈 Почему это работает:

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

import ccxt
import time

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

def fetch_order_book(symbol):
    ob = exchange.fetch_order_book(symbol, limit=100)
    return ob['bids'], ob['asks']

def detect_spoof_disappearance(prev, curr):
    def large_orders_diff(old, new):
        large = []
        avg_size = sum([x[1] for x in old]) / len(old)
        for price, size in old:
            if size > avg_size * threshold_multiplier and not any(abs(price - p[0]) < 1e-4 for p in new):
                large.append((price, size))
        return large
    disappeared_bids = large_orders_diff(prev['bids'], curr['bids'])
    disappeared_asks = large_orders_diff(prev['asks'], curr['asks'])
    return disappeared_bids, disappeared_asks

last_ob = {'bids': [], 'asks': []}
print("⏳ Мониторинг ордербука...")

while True:
    try:
        bids, asks = fetch_order_book(symbol)
        if last_ob['bids'] and last_ob['asks']:
            lost_bids, lost_asks = detect_spoof_disappearance(last_ob, {'bids': bids, 'asks': asks})
            if lost_bids:
                print(f"⚠️ Исчезли крупные заявки BUY: {lost_bids} → сигнал на SHORT")
            if lost_asks:
                print(f"⚠️ Исчезли крупные заявки SELL: {lost_asks} → сигнал на LONG")
        last_ob = {'bids': bids, 'asks': asks}
    except Exception as e:
        print("Ошибка:", e)
    time.sleep(5)

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

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

📌"Гиперволатильный ретест"

(авторская идея, направлена на поиск резких выбросов цены и обратного возврата к уровню — со встроенным фильтром на тренд)🧑‍💻

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

1. Вход:

Когда происходит резкий импульс вверх или вниз (например, > 1.5% за 1 минуту), а через 1-3 свечи происходит откат обратно к предыдущему уровню (ретест) — открывается сделка в направлении возврата (на откат).

2. Фильтр тренда:

Стратегия работает только при наличии сильного направленного движения за последние 30 минут (например, изменение > 2.5%). То есть, торгует только в условиях тренда.

3. Индикаторы (опционально):

- EMA(20) и EMA(50) для фильтра тренда.

- ATR для оценки силы импульса.

- Отслеживание свечных паттернов "Pin Bar" или "Engulfing" на ретесте — как дополнительное подтверждение.

import ccxt
import pandas as pd
import time

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

exchange = ccxt.binance()

def fetch_ohlcv():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['returns'] = df['close'].pct_change()
    df['ema20'] = df['close'].ewm(span=20).mean()
    df['ema50'] = df['close'].ewm(span=50).mean()
    df['atr'] = df['high'] - df['low']
    return df

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

    # Тренд фильтр: EMA20 > EMA50
    in_trend = last['ema20'] > last['ema50']
    recent_movement = (df['close'].iloc[-1] - df['close'].iloc[-30]) / df['close'].iloc[-30] * 100

    if in_trend and recent_movement > 2.5:
        # Импульс был 2 свечи назад
        impulse = df['returns'].iloc[-3]
        retest = df['returns'].iloc[-1]

        if impulse > 0.015 and retest < -0.01:
            print("🔽 Short signal detected (Impulse up → Retest down)")

        elif impulse < -0.015 and retest > 0.01:
            print("🔼 Long signal detected (Impulse down → Retest up)")

if __name__ == "__main__":
    while True:
        try:
            df = fetch_ohlcv()
            detect_signal(df)
            time.sleep(60)
        except Exception as e:
            print("Ошибка:", e)
            time.sleep(10)

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

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

📌"AlgoMarketMap" — Карта активности и плотности алгоритмической торговли в реальном времени🧑‍💻

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

Визуализатор и анализатор мест концентрации алготорговой активности (ботов) по:

- частоте обновлений ордербука;

- всплескам объёма на уровнях;

- микродвижениям цены без объёма;

- аномально точным лимиткам (в миллисекунды);

- частым сменам позиции в стакане.

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

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

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

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

import asyncio
import websockets
import json
import time
from collections import deque

symbol = "btcusdt"
depth_url = f"wss://stream.binance.com:9443/ws/{symbol}@depth"

event_times = deque(maxlen=100)

async def monitor_depth():
    async with websockets.connect(depth_url) as websocket:
        while True:
            try:
                msg = await websocket.recv()
                data = json.loads(msg)

                now = time.time()
                event_times.append(now)

                # Оцениваем активность обновлений
                if len(event_times) > 10:
                    freq = len(event_times) / (event_times[-1] - event_times[0])
                    if freq > 30:
                        print(f"⚠️ Алготрейдерская активность обнаружена! Частота: {freq:.2f} обновлений/сек")

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

asyncio.run(monitor_depth())

#инструмент

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

📌"Delta Drift Reversal" — Интегральный разворот по дельте объёма и микроимпульсам🧑‍💻

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

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

📌 Используется только рыночная информация: стакан, лента, объём и цена.

🔬 Условия входа (LONG):💸

1. На кластерной ленте видно:

Сильное агрессивное давление продавцов (дельта < -X);

Но цена не пробивает минимум, остаётся в диапазоне → затухание импульса.

2. В течение последних N минут:

Был хотя бы 1 мощный толчок вниз, но объём при повторном тесте ниже среднего.

3. Формируется "антиимпульс" — небольшое поглощение (delta становится положительной, цена идёт вверх).

4. Цена выше скользящего VWAP или среднего баланса за сессию.

🟢 → Вход в LONG

🔴 → Аналогично работает и в SHORT при обратной конфигурации.

📊 Фильтры:💻

- Не торгуется в низковолатильные сессии (определяется по ATR).

- Обязательно проверяется объём ленты и скорость сделок (tick rate).

- Выход по Take Profit = x2 от ближайшего скопления ликвидности (по дельте)
или после формирования противоположного "антиимпульса".

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
window = 5  # в минутах

def fetch_volume_delta(symbol, minutes=5):
    ohlcv = exchange.fetch_ohlcv(symbol, '1m', limit=minutes)
    delta = 0
    for candle in ohlcv:
        open_, high, low, close, volume = candle[1:6]
        direction = 1 if close > open_ else -1
        delta += volume * direction
    return delta

while True:
    delta = fetch_volume_delta(symbol, window)
    print(f"Δ объёма за {window} минут: {delta:.2f}")

    if delta < -1500:
        print("⚠️ Мощная распродажа! Следим за затуханием импульса...")
    elif delta > 1500:
        print("⚠️ Активация покупателей! Возможен отскок.")
    else:
        print("Нейтральное давление...")

    time.sleep(60)

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

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