📌 Fakeout Detector – определение ложных пробоев уровней🧑‍💻

Идея:🚀

Инструмент анализирует поведение цены после пробоя ключевого уровня (high/low за N баров) и проверяет, вернулась ли цена обратно в диапазон. Это может сигнализировать о "ложном пробое" (fakeout) — одном из сильных сигналов в price action.

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

1. Построение high/low за заданный период lookback.

2. Фиксация момента пробоя вверх/вниз.

3. Проверка: вернулась ли цена обратно за фиксированное количество свечей (reentry_window).

4. Если да — сигнал: ложный пробой.

import pandas as pd
import requests
import time

# Настройки
symbol = "BTCUSDT"
interval = "15m"
limit = 200
lookback = 20
reentry_window = 3
api_url = f"https://api.binance.com/api/v3/klines"

def fetch_ohlcv():
    params = {"symbol": symbol, "interval": interval, "limit": limit}
    res = requests.get(api_url, params=params)
    data = res.json()
    df = pd.DataFrame(data, columns=[
        "timestamp", "open", "high", "low", "close", "volume",
        "close_time", "quote_asset_volume", "number_of_trades",
        "taker_buy_base_volume", "taker_buy_quote_volume", "ignore"
    ])
    df["close"] = df["close"].astype(float)
    df["high"] = df["high"].astype(float)
    df["low"] = df["low"].astype(float)
    return df

def detect_fakeout(df):
    recent = df.iloc[-(reentry_window + 1):]
    prior = df.iloc[-(reentry_window + 1 + lookback):- (reentry_window + 1)]

    high_level = prior["high"].max()
    low_level = prior["low"].min()

    first_break = recent.iloc[0]
    reentries = recent.iloc[1:]

    # Ложный пробой вверх
    if first_break["high"] > high_level and all(reentries["close"] < high_level):
        print("🔻 Fakeout вверх обнаружен!")

    # Ложный пробой вниз
    if first_break["low"] < low_level and all(reentries["close"] > low_level):
        print("🔺 Fakeout вниз обнаружен!")

if __name__ == "__main__":
    while True:
        df = fetch_ohlcv()
        detect_fakeout(df)
        time.sleep(60)

#инструмент

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

📌 Flat Range Scanner🧑‍💻

Идея:🛠️

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

import ccxt
import pandas as pd

exchange = ccxt.binance()

symbols = ['BTC/USDT', 'ETH/USDT', 'BNB/USDT', 'SOL/USDT']
timeframe = '15m'
lookback = 30
volatility_threshold = 0.01  # 1%

def is_flat(df):
    high = df['high'].max()
    low = df['low'].min()
    avg_price = df['close'].mean()
    return (high - low) / avg_price < volatility_threshold

def fetch_and_check(symbol):
    try:
        ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=lookback)
        df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        return symbol if is_flat(df) else None
    except Exception as e:
        print(f"{symbol} — ошибка: {e}")
        return None

if __name__ == "__main__":
    print(f"\n🧭 Поиск флэтов ({timeframe}, последние {lookback} свечей)...\n")
    for symbol in symbols:
        result = fetch_and_check(symbol)
        if result:
            print(f"✅ {result} — во флэте!")

#инструмент

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

📌"Multi-Pivot Trap Strategy" ловушка на многоуровневых локальных экстремумах 🧑‍💻

Идея:🛠️

Стратегия отслеживает "двойные/тройные" локальные максимумы или минимумы, где цена дважды (или трижды) подходила к уровню и не пробила его. При резком пробое этих уровней формируется "ловушка" — сильное движение в сторону пробоя или возврат (false breakout).

🧠 Логика:🚀

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

2. Сравниваем: если два экстремума близки по цене (±x%), формируем уровень сопротивления/поддержки.

3. Если текущая цена резко пробивает этот уровень на объёме — вход в сторону пробоя (или возврат при откате).

4. Используется таймфрейм 15m–1h.

📌 Правила:💸

- Уровень сопротивления: два недавних пика находятся в пределах 0.3% друг от друга.

- Уровень пробивается: свеча закрывается выше уровня + объём выше среднего.

- Фильтр: RSI или CCI — подтверждение перекупленности/перепроданности.

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

# Настройка
symbol = 'BTC/USDT'
timeframe = '15m'
limit = 150
exchange = ccxt.binance()

def fetch_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 find_multi_pivots(df, threshold=0.003):
    highs = df['high'].rolling(5, center=True).apply(
        lambda x: x[2] if x[2] == max(x) else np.nan
    )
    pivot_highs = df[~highs.isna()]
    
    for i in range(len(pivot_highs) - 1):
        p1 = pivot_highs.iloc[i]
        p2 = pivot_highs.iloc[i + 1]
        if abs(p1['high'] - p2['high']) / p1['high'] < threshold:
            return p1['high']
    return None

def detect_breakout(df):
    level = find_multi_pivots(df)
    if level is None:
        return

    last_close = df['close'].iloc[-1]
    last_vol = df['volume'].iloc[-1]
    avg_vol = df['volume'].rolling(20).mean().iloc[-2]

    if last_close > level and last_vol > avg_vol * 1.2:
        print(f"[{datetime.now()}] 🚀 Breakout вверх уровня {level:.2f} на объёме!")
    elif last_close < level and last_vol > avg_vol * 1.2:
        print(f"[{datetime.now()}] 🔻 Пробой вниз уровня {level:.2f} на объёме!")

if __name__ == "__main__":
    while True:
        try:
            df = fetch_ohlcv()
            detect_breakout(df)
            time.sleep(60)  # Проверка раз в минуту
        except Exception as e:
            print(f"Ошибка: {e}")
            time.sleep(60)

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

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

📌Price Shock Recovery 🧑‍💻

Идея:🛠️

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

🧠 Логика:💻

- Берём свечи (например, 5m).

- Ищем “шок-свечу” — изменение цены больше X% за свечу.

- Если следующая свеча идёт в обратную сторону — входим в контртренд.

- Тейк — возврат на 50–70% шока, стоп — за хай/лоу шок-свечи.

import ccxt
import pandas as pd

exchange = ccxt.binance()

symbol = 'BTC/USDT'
timeframe = '5m'
shock_threshold = 0.02  # 2%

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

def check_shock_reversal(df):
    last = df.iloc[-2]
    current = df.iloc[-1]

    change = (last['close'] - last['open']) / last['open']

    # шок-свеча вверх
    if change >= shock_threshold and current['close'] < last['close']:
        print(f"🔻 Потенциал шорт-реверсала на {symbol} ({timeframe})")
    # шок-свеча вниз
    elif change <= -shock_threshold and current['close'] > last['close']:
        print(f"🔺 Потенциал лонг-реверсала на {symbol} ({timeframe})")

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

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

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

📌 Volume Spike Scanner 🧑‍💻

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

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

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

- Мониторит пары (например, BTC/USDT, ETH/USDT).

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

- Сигнализирует, если текущий объём значительно выше среднего (в 2–3 раза).

- Можно использовать как оповещатель возможного начала движения.

import ccxt
import pandas as pd

exchange = ccxt.binance()
symbols = ['BTC/USDT', 'ETH/USDT']
timeframe = '5m'
lookback = 10
volume_multiplier = 2.5  # текущий объём должен быть выше среднего в X раз

def get_volume_alerts():
    for symbol in symbols:
        ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=lookback + 1)
        df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        
        avg_vol = df['volume'][:-1].mean()
        current_vol = df['volume'].iloc[-1]

        if current_vol > avg_vol * volume_multiplier:
            print(f"⚠️ Объёмный всплеск на {symbol}: {current_vol:.2f} > {avg_vol:.2f}")

if __name__ == "__main__":
    get_volume_alerts()

#инструмент

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

📌 "Market Snapshot CLI"🧑‍💻

Описание:🛠️

Консольный Python-скрипт, который делает моментальный снимок рынка: показывает цену, спред, объём, тренд и индикаторы (RSI, EMA) по заданным парам — идеально подходит для быстрого анализа перед входом в сделку.

🔧 Возможности:🚀

- Цена и спред по выбранным парам (например, BTC/USDT, ETH/USDT).

- 24h объём.

- RSI и EMA (TA-анализ).

- Поддержка нескольких бирж (через ccxt).

- Удобный CLI-вывод.

import ccxt
import argparse
from ta.momentum import RSIIndicator
from ta.trend import EMAIndicator
import pandas as pd

def fetch_ohlcv(symbol, exchange_id):
    exchange = getattr(ccxt, exchange_id)()
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe='1h', limit=50)
    df = pd.DataFrame(ohlcv, columns=['time', 'open', 'high', 'low', 'close', 'volume'])
    return df

def analyze(symbol, exchange_id):
    df = fetch_ohlcv(symbol, exchange_id)
    close = df['close']

    rsi = RSIIndicator(close).rsi().iloc[-1]
    ema = EMAIndicator(close).ema_indicator().iloc[-1]
    last_price = close.iloc[-1]
    spread = df['high'].iloc[-1] - df['low'].iloc[-1]

    print(f"\n📌 {symbol} on {exchange_id.upper()}")
    print(f"Last Price: {last_price:.2f}")
    print(f"Spread: {spread:.2f}")
    print(f"EMA: {ema:.2f}")
    print(f"RSI: {rsi:.2f}")
    print("Trend:", "↑ Bullish" if last_price > ema else "↓ Bearish")

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('symbols', nargs='+', help="Trading pairs like BTC/USDT")
    parser.add_argument('--exchange', default='binance', help="Exchange name supported by ccxt")
    args = parser.parse_args()

    for sym in args.symbols:
        try:
            analyze(sym, args.exchange)
        except Exception as e:
            print(f"Ошибка с {sym}: {e}")

#инструмент

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

📌"Shadow Order Echo" - торговля на основе поведения цен после появления крупных лимитных заявок 🧑‍💻

💡 Идея:🛠️

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

📈 Логика:💻

-Отслеживать крупные заявки в стакане (например, > 10 BTC).

- Если цена быстро смещается в сторону крупной заявки, а затем заявка исчезает:

- Ждём отката и входим в позицию в обратную сторону.

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

import ccxt
import time

# Настройки
symbol = 'BTC/USDT'
threshold_volume = 10  # Объем в BTC, выше которого считаем заявку крупной
check_interval = 5     # Интервал проверки стакана в секундах
depth_limit = 20       # Глубина стакана (top N цен)

exchange = ccxt.binance()

def get_large_orders(orderbook_side, threshold):
    return [order for order in orderbook_side if order[1] >= threshold]

def fetch_order_book():
    order_book = exchange.fetch_order_book(symbol, limit=depth_limit)
    large_bids = get_large_orders(order_book['bids'], threshold_volume)
    large_asks = get_large_orders(order_book['asks'], threshold_volume)
    return {'bids': large_bids, 'asks': large_asks}

print(f"[START] Monitoring fake walls on {symbol}...")

previous_snapshot = fetch_order_book()

while True:
    try:
        time.sleep(check_interval)
        current_snapshot = fetch_order_book()

        # Проверка: была крупная заявка, но исчезла
        for side in ['bids', 'asks']:
            prev_orders = set((round(price, 2), round(amount, 2)) for price, amount in previous_snapshot[side])
            curr_orders = set((round(price, 2), round(amount, 2)) for price, amount in current_snapshot[side])

            disappeared = prev_orders - curr_orders
            for price, amount in disappeared:
                print(f"[ALERT] {side.upper()} wall disappeared! Price: {price}, Size: {amount} {symbol.split('/')[0]}")

        previous_snapshot = current_snapshot

    except KeyboardInterrupt:
        print("Stopped by user.")
        break
    except Exception as e:
        print(f"[ERROR] {e}")
        time.sleep(10)

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

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

📌Инструмент: "Correlation Radar" – Локатор корреляций между активами🧑‍💻

Тип: Аналитический инструмент

Назначение: Быстро находить активы с высокой или аномальной корреляцией (или декорреляцией)

Полезен для:💸

- Выбора хеджей

- Спред-трейдинга

- Определения схожих/расходящихся трендов

- Избежания дублирования позиций по коррелирующим активам

import ccxt
import pandas as pd
import numpy as np
import seaborn as sns
import matplotlib.pyplot as plt

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

price_data = {}

for symbol in symbols:
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    price_data[symbol] = df['close']

df_prices = pd.DataFrame(price_data)
cor_matrix = df_prices.pct_change().corr()

# Визуализация
sns.heatmap(cor_matrix, annot=True, cmap='coolwarm')
plt.title("📊 Correlation Matrix Between Assets")
plt.show()

# ТОП-5
corr_pairs = cor_matrix.unstack().reset_index()
corr_pairs.columns = ['Asset1', 'Asset2', 'Correlation']
corr_pairs = corr_pairs[corr_pairs['Asset1'] != corr_pairs['Asset2']]
top = corr_pairs.sort_values(by='Correlation', ascending=False).drop_duplicates().head(5)
print("\n🔗 Most Correlated Pairs:\n", top)
low = corr_pairs.sort_values(by='Correlation', ascending=True).drop_duplicates().head(5)
print("\n🔍 Most De-correlated Pairs:\n", low)

#инструмент

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

📌Стратегия: "Volume Spike Reversal"🧑‍💻

Тип: Внутридневная, контртрендовая

Инструменты: Объём, свечной анализ, RSI

Идея:🚀

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

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

- Последняя свеча закрывается с ростом >1.5% от предыдущей.

- Объём выше среднего за последние 20 свечей на 200%+.

- RSI > 75.

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

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

- Обратное: падение >1.5%, объём выше в 2 раза, RSI < 25.

- Вход в лонг при признаках разворота вверх.

import ccxt
import pandas as pd
from ta.momentum import RSIIndicator

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

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

# Средний объём
avg_vol = df['volume'][:-1].tail(20).mean()
last = df.iloc[-1]
prev = df.iloc[-2]

# Условия шорт
if (
    (prev['close'] - prev['open']) / prev['open'] > 0.015 and
    prev['volume'] > 2 * avg_vol and
    prev['rsi'] > 75 and
    last['open'] < prev['close']
):
    print("⚠️ SHORT Signal: Volume Spike Reversal")

# Условия лонг
if (
    (prev['open'] - prev['close']) / prev['open'] > 0.015 and
    prev['volume'] > 2 * avg_vol and
    prev['rsi'] < 25 and
    last['open'] > prev['close']
):
    print("⚠️ LONG Signal: Volume Spike Reversal")

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

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

📌Strategy: "Candle Cluster Fade" – Контртрендовая торговля от кластеров длинных свечей🧑‍💻

Тип: Внутридневная, mean-reversion

Подходит для: Волатильных рынков (crypto, золото, нефть, NASDAQ)

Идея:🧨

Если в течение короткого времени возникает кластер из нескольких длинных свечей в одном направлении без существенной коррекции — вероятен откат.

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

1. Обнаружение свечного кластера:🚀

3 и более подряд идущих свечей в одном направлении.

Каждая свеча по размеру > X% от средней за N периодов.


2. Проверка контекста:💻

- Индикатор волатильности (например, ATR) не выходит за пределы — это не "новости".

- Нет сильного тренда (по ADX < 25).

3. Вход:💸

В шорт — если кластер на рост.

В лонг — если кластер на падение.

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

Тейк — в зону начала кластера или фиксированный RR.

import ccxt
import pandas as pd

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

def fetch_data():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['time', 'open', 'high', 'low', 'close', 'volume'])
    df['body'] = abs(df['close'] - df['open'])
    df['direction'] = df['close'] > df['open']
    df['avg_body'] = df['body'].rolling(window=20).mean()
    return df

def detect_cluster(df, min_size=3, body_thresh=1.2):
    cluster = []
    for i in range(len(df) - min_size):
        candles = df.iloc[i:i+min_size]
        if all(candles['direction']) or all(~candles['direction']):
            if all(candles['body'] > body_thresh * candles['avg_body']):
                cluster.append((i, candles))
    return cluster

df = fetch_data()
clusters = detect_cluster(df)

if clusters:
    for i, cl in clusters:
        direction = "UP" if cl['direction'].iloc[0] else "DOWN"
        print(f"[ALERT] Candle Cluster detected ({direction}) at index {i}, time: {pd.to_datetime(df['time'].iloc[i], unit='ms')}")
else:
    print("No clusters detected.")

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

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

📌Real-Time RSI Divergence Detector🧑‍💻

Описание:🚀

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

🔍 Что делает скрипт:

- Загружает свечи с Binance через ccxt.

- Рассчитывает RSI.

Ищет:

- Бычью дивергенцию: цена обновляет минимум, RSI — нет.

- Медвежью дивергенцию: цена обновляет максимум, RSI — нет.

import ccxt
import pandas as pd
import time
import datetime

# === Настройки ===
symbol = 'BTC/USDT'
timeframe = '15m'
lookback = 20          # окно анализа дивергенции
sleep_interval = 900   # 15 минут (900 сек)
log_file = 'rsi_divergence_log.csv'

# === RSI расчёт ===
def rsi(series, period=14):
    delta = series.diff()
    gain = delta.where(delta > 0, 0).rolling(period).mean()
    loss = -delta.where(delta < 0, 0).rolling(period).mean()
    rs = gain / loss
    return 100 - (100 / (1 + rs))

# === Поиск дивергенции ===
def detect_divergence(df):
    recent = df.iloc[-lookback:]
    price_min_idx = recent['close'].idxmin()
    rsi_min_idx = recent['rsi'].idxmin()
    price_max_idx = recent['close'].idxmax()
    rsi_max_idx = recent['rsi'].idxmax()

    signal = None
    if price_min_idx > rsi_min_idx and df['close'][price_min_idx] < df['close'][rsi_min_idx]:
        signal = 'Bullish Divergence'
    elif price_max_idx > rsi_max_idx and df['close'][price_max_idx] > df['close'][rsi_max_idx]:
        signal = 'Bearish Divergence'

    return signal

# === Логирование сигнала ===
def log_signal(signal, df):
    timestamp = df['time'].iloc[-1]
    price = df['close'].iloc[-1]
    row = pd.DataFrame([[timestamp, signal, price]], columns=['time', 'signal', 'price'])
    row.to_csv(log_file, mode='a', header=not pd.io.common.file_exists(log_file), index=False)
    print(f"[🔔] {signal} | {timestamp} | Цена: {price:.2f}")

# === Основной цикл ===
def run_monitor():
    exchange = ccxt.binance()

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

            signal = detect_divergence(df)
            if signal:
                log_signal(signal, df)
            else:
                print(f"[{datetime.datetime.now().strftime('%H:%M:%S')}] Нет сигнала.")
        except Exception as e:
            print(f"[Ошибка] {e}")
        time.sleep(sleep_interval)

# === Запуск ===
if __name__ == "__main__":
    run_monitor()

#инструмент

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

📌Strategy: "Dark Pool Footprint"

Опережающий вход на основе крупной скрытой активности 🧑‍💻

📌 Идея:🚀

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

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

1. Сканируем поток сделок на наличие повторяющихся объёмов (например, 100+ ордеров по 500 ETH подряд).

2. Сравниваем частоту и распределение цен — если эти блоки проходят по одной цене — признак dark pool-аккумуляции.

3. Если в течение X минут появляется кластер крупных сделок — заходим в ту же сторону.

4. Фильтр на отсутствие видимого волатильного движения (чтобы это не были маркет-движения).

✅ Вход:💸

- Найден блок из ≥5 сделок с объёмом выше среднего (по инструменту).

- Все сделки одной стороной (buy/sell).

- Цена не двигается резко → вероятная "тихая" позиция.

- Вход в ту же сторону с коротким стопом.

import ccxt
import time
import pandas as pd

# Настройка
symbol = 'BTC/USDT'
min_trade_volume = 5  # минимальный объём в BTC для крупной сделки
block_window = 10     # сколько последних сделок анализировать
same_price_threshold = 5  # сколько сделок должно быть по одной цене
poll_delay = 3        # пауза между опросами

exchange = ccxt.binance()

def fetch_trades():
    trades = exchange.fetch_trades(symbol)
    df = pd.DataFrame(trades)
    df['price'] = df['price'].astype(float)
    df['amount'] = df['amount'].astype(float)
    df['side'] = df['side'].astype(str)
    df['volume'] = df['price'] * df['amount']
    return df

def detect_block_trades(trades_df):
    recent = trades_df.tail(block_window)
    big_trades = recent[recent['amount'] > min_trade_volume]

    if big_trades.empty:
        return None

    # Группируем по цене
    grouped = big_trades.groupby('price').size().reset_index(name='count')
    clusters = grouped[grouped['count'] >= same_price_threshold]

    if not clusters.empty:
        price = clusters.iloc[0]['price']
        print(f"[ALERT] Возможен скрытый интерес на {price} USDT — кластер из крупных сделок!")
        return price

    return None

if __name__ == "__main__":
    print(f"🟡 Запуск мониторинга {symbol} на крупные скрытые сделки (через REST)...")
    while True:
        try:
            trades_df = fetch_trades()
            detect_block_trades(trades_df)
        except Exception as e:
            print(f"Ошибка: {e}")
        time.sleep(poll_delay)

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

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

📌Volatility Regime Classifier🧑‍💻

🧠 Идея: определить, в каком волатильностном режиме находится рынок:
Low, Medium, High — для настройки агрессии стратегии (например, менять стопы, тейки, частоту сделок).

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

- Загружает исторические данные (например, с Binance через ccxt);

- Расчёт Rolling Volatility по ATR или StdDev;

- Классификация волатильности на три уровня;

- Простая визуализация и экспорт в CSV.

import ccxt
import pandas as pd
import ta
import matplotlib.pyplot as plt

symbol = 'BTC/USDT'
timeframe = '1h'
limit = 300

exchange = ccxt.binance()
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')

# Расчёт ATR
df['atr'] = ta.volatility.AverageTrueRange(df['high'], df['low'], df['close'], window=14).average_true_range()

# Классификация волатильности
q_low = df['atr'].quantile(0.33)
q_high = df['atr'].quantile(0.66)

def classify_volatility(x):
    if x < q_low:
        return 'Low'
    elif x < q_high:
        return 'Medium'
    else:
        return 'High'

df['vol_regime'] = df['atr'].apply(classify_volatility)

# Визуализация
plt.figure(figsize=(12, 6))
plt.plot(df['timestamp'], df['close'], label='Price', color='black')
colors = {'Low': 'green', 'Medium': 'orange', 'High': 'red'}
for regime, group in df.groupby('vol_regime'):
    plt.scatter(group['timestamp'], group['close'], label=regime, color=colors[regime], s=5)
plt.title('Volatility Regime Classification')
plt.legend()
plt.show()

#инструмент

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

📌"Adaptive Momentum Fade" Strategy 🧑‍💻

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

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

- Ищем краткосрочный импульс вверх/вниз, измеряемый через Rate of Change (ROC);

- Если импульс > адаптивного порога, ищем признаки истощения: свечной откат, слабый объём;

- Заходим в противоположную сторону с коротким тейком и tight стопом.

import ccxt
import pandas as pd
import ta

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

exchange = ccxt.binance()
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')

# Индикаторы
df['roc'] = ta.momentum.ROCIndicator(df['close'], window=5).roc()
df['atr'] = ta.volatility.AverageTrueRange(df['high'], df['low'], df['close'], window=14).average_true_range()

# Условия входа в шорт при истощении импульса вверх
df['signal'] = (
    (df['roc'] > df['roc'].rolling(50).mean() + df['roc'].rolling(50).std()) &  # сильный вверх
    (df['close'].shift(1) > df['open'].shift(1)) &  # была зеленая свеча
    (df['close'] < df['open'])                     # текущая свеча красная
).astype(int)

# Вывод сигналов
for i in range(1, len(df)):
    if df['signal'].iloc[i] == 1:
        print(f"🔻 Fade SHORT сигнал {df['timestamp'].iloc[i]} @ {df['close'].iloc[i]:.2f}")

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

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

📌Индикатор Консенсуса Рынка через Корреляции🧑‍💻

💡 Идея:🧨

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

⚙️ Что делает скрипт:🚀

- Загружает исторические данные по списку активов (через ccxt);

- Строит матрицу корреляций (по закрытиям);

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

- Отправляет алерт в консоль, если консенсус выше/ниже порога.

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

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

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

def calculate_market_consensus():
    closes = []
    for symbol in symbols:
        try:
            close_series = fetch_ohlcv(symbol)
            closes.append(close_series)
        except Exception as e:
            print(f"Ошибка загрузки {symbol}: {e}")
            continue

    if len(closes) < 2:
        print("Недостаточно данных")
        return

    df = pd.concat(closes, axis=1)
    df.columns = symbols
    corr_matrix = df.pct_change().corr()
    avg_corr = corr_matrix.where(~np.eye(corr_matrix.shape[0], dtype=bool)).mean().mean()
    
    print(f"[ALERT] Средняя корреляция: {avg_corr:.2f}")
    if avg_corr > 0.75:
        print("⚠️ Высокий рыночный консенсус — возможный тренд.")
    elif avg_corr < 0.3:
        print("🌀 Низкий консенсус — распределение или смена фазы.")

if __name__ == "__main__":
    while True:
        calculate_market_consensus()
        time.sleep(3600)  # Запуск каждый час

#инструмент

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

📌"Пульс Рынка" – Сканер Аномальных Активов🧑‍💻

📌 Идея:🛠️

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

🧠 Польза для трейдера:💸

🔍 Быстрая фильтрация “горячих” активов

📊 Идеально для дейтрейдеров, ловящих движения

import ccxt
import time

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

def fetch_anomalies():
    results = []
    for symbol in symbols:
        try:
            ohlcv = exchange.fetch_ohlcv(symbol, timeframe='1h', limit=24)
            closes = [c[4] for c in ohlcv]
            volumes = [c[5] for c in ohlcv]
            change = (closes[-1] - closes[0]) / closes[0] * 100
            avg_volume = sum(volumes) / len(volumes)
            if abs(change) > 5 or avg_volume > 1_000_000:
                results.append((symbol, round(change, 2), round(avg_volume, 2)))
        except Exception:
            continue
    return sorted(results, key=lambda x: abs(x[1]), reverse=True)

while True:
    print("🔍 Аномальные активы:")
    anomalies = fetch_anomalies()
    for sym, chg, vol in anomalies[:10]:
        print(f"{sym}: {chg}% | Объём: {vol}")
    print('-' * 40)
    time.sleep(300)  # обновление каждые 5 минут

#инструмент

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

📌Volume Void Strategy – торговля на зонах "пустого" объёма🧑‍💻

💡 Идея:🛠️

На фьючерсных и спотовых рынках бывают участки, где цена резко пролетает зону без значимых сделок (low volume node). Эти зоны часто притягивают цену обратно – как магниты. Стратегия ищет резкие движения с объёмным «провалом» и ждет ретеста этой зоны для входа.

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

1. Цена резко уходит вверх или вниз с низким объёмом на участке;

2. На ретесте этой зоны – вход в направлении импульса;

3. TP: верх/низ импульса; SL: за границей зоны void.

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

# Настройки
symbol = 'BTC/USDT'
timeframe = '1m'
limit = 200
volume_threshold = 0.7  # Уровень объема по отношению к скользящему среднему
check_interval = 60  # секунд

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

def fetch_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_volume_void(df):
    global last_signal_time
    df['change'] = df['close'].pct_change()
    df['avg_vol'] = df['volume'].rolling(window=20).mean()
    df['low_volume'] = df['volume'] < df['avg_vol'] * volume_threshold

    latest_idx = len(df) - 3  # На всякий случай оставляем буфер
    row = df.iloc[latest_idx]

    # Условие: сильное движение и после него - низкий объем
    if abs(df.loc[latest_idx, 'change']) > 0.01 and df.loc[latest_idx + 1, 'low_volume']:
        signal_time = df.loc[latest_idx, 'timestamp']
        if last_signal_time != signal_time:
            last_signal_time = signal_time
            print(f"📢 VOID Volume Зона! [{signal_time}] Движение от {df.loc[latest_idx, 'close']:.2f} с объёмом {df.loc[latest_idx+1, 'volume']:.2f}")

def main_loop():
    print("📡 Мониторинг рынка (Volume Void Zones)...")
    while True:
        try:
            df = fetch_ohlcv()
            detect_volume_void(df)
        except Exception as e:
            print(f"❌ Ошибка: {e}")
        time.sleep(check_interval)

if __name__ == '__main__':
    main_loop()

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

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

📌Простой пример, как открыть лимитный ордер на покупку или продажу на спотовом рынке Binance с помощью библиотеки ccxt🧑‍💻

import ccxt

# ✅ API-ключи
api_key = 'YOUR_API_KEY'
secret = 'YOUR_SECRET_KEY'

# ✅ Инициализация биржи
binance = ccxt.binance({
    'apiKey': api_key,
    'secret': secret,
    'enableRateLimit': True,
    'options': {'defaultType': 'spot'}  # важно: spot, не futures
})

# ✅ Параметры ордера
symbol = 'BTC/USDT'        # торговая пара
side = 'buy'               # или 'sell'
price = 100000              # лимитная цена
amount = 0.001             # количество BTC

# ✅ Отправка лимитного ордера
try:
    order = binance.create_limit_order(symbol, side, amount, price)
    print(f"Ордер создан: {order['id']} | {side.upper()} {amount} {symbol} @ {price}")
except Exception as e:
    print("Ошибка при создании ордера:", e)

#инструмент

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

📌Relative Volume Climax (Относительный Объёмный Климакс)🧑‍💻

🧠 Идея:💸

Стратегия отслеживает аномальные всплески объёма относительно среднего за N свечей. Сигнал возникает, когда:

- Объём текущей свечи превышает средний в X раз

- И при этом свеча имеет сильное тело (ценовое движение выше Y%)

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

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '1m'
volume_multiplier = 3       # Объём должен быть выше среднего в X раз
body_threshold_percent = 0.5  # Размер тела свечи в процентах
bars = 20  # Кол-во свечей для среднего

def fetch_candles():
    return exchange.fetch_ohlcv(symbol, timeframe, limit=bars + 1)

def check_climax():
    candles = fetch_candles()
    last_candle = candles[-1]
    previous_candles = candles[:-1]

    avg_volume = sum([c[5] for c in previous_candles]) / len(previous_candles)
    current_volume = last_candle[5]
    open_price = last_candle[1]
    close_price = last_candle[4]
    body_size = abs(close_price - open_price)
    body_percent = (body_size / open_price) * 100

    if current_volume > avg_volume * volume_multiplier and body_percent > body_threshold_percent:
        direction = "🔼 LONG" if close_price > open_price else "🔽 SHORT"
        print(f"[ALERT] {symbol} | {direction} | Vol x{round(current_volume / avg_volume, 2)} | Body: {round(body_percent, 2)}%")

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

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

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

📌"Умная сетка заявок с усреднением и Take Profit"

Описание:

Скрипт автоматически расставляет серию лимитных ордеров на покупку по убывающим ценам (сеткой), чтобы усреднять вход. После исполнения — выставляется общий Take Profit ордер на весь объём.
Подходит для трендовых откатов или накоплений.


---

💡 Принцип:

1. Проверяет текущую цену.

2. Выставляет, например, 5 лимиток ниже рынка (step, например -0.5%).

3. Когда ордера исполняются — рассчитывает среднюю цену и выставляет единый TP-ордер на продажу.

🧠 Чем полезен:💸

- Автоматизирует вход по частям.

- Снижает риск при входе в неудачную точку.

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

import ccxt
import time

# Настройки API
api_key = 'YOUR_API_KEY'
api_secret = 'YOUR_API_SECRET'

exchange = ccxt.binance({
    'apiKey': api_key,
    'secret': api_secret,
    'enableRateLimit': True
})

symbol = 'BTC/USDT'
grid_levels = 5              # Количество уровней сетки
grid_step = 100              # Шаг в USD
order_size = 0.001           # Размер каждой заявки
profit_target = 150          # Профит в USD с одной позиции

def get_price():
    ticker = exchange.fetch_ticker(symbol)
    return ticker['last']

def place_limit_buy(price):
    try:
        order = exchange.create_limit_buy_order(symbol, order_size, price)
        print(f"[BUY] Лимитный ордер размещён по {price}")
        return order
    except Exception as e:
        print(f"[ERROR] BUY: {e}")

def place_limit_sell(price):
    try:
        order = exchange.create_limit_sell_order(symbol, order_size, price)
        print(f"[SELL] Лимитный ордер размещён по {price}")
        return order
    except Exception as e:
        print(f"[ERROR] SELL: {e}")

def get_open_orders():
    return exchange.fetch_open_orders(symbol)

def cancel_all_orders():
    orders = get_open_orders()
    for o in orders:
        exchange.cancel_order(o['id'], symbol)

# Главный цикл
def grid_bot():
    base_price = get_price()
    print(f"[INFO] Текущая цена: {base_price}, начинаем размещение ордеров...")

    grid = []

    for i in range(1, grid_levels + 1):
        buy_price = round(base_price - i * grid_step, 2)
        buy_order = place_limit_buy(buy_price)
        if buy_order:
            grid.append({'buy_price': buy_price, 'status': 'open'})

    print("[INFO] Ожидание исполнения...")

    while True:
        time.sleep(10)
        orders = get_open_orders()

        for level in grid:
            # Если ордер был исполнен
            if level['status'] == 'open':
                matching = [o for o in orders if abs(o['price'] - level['buy_price']) < 1e-2]
                if not matching:
                    # BUY исполнен — размещаем SELL
                    sell_price = round(level['buy_price'] + profit_target, 2)
                    place_limit_sell(sell_price)
                    level['status'] = 'filled'
                    print(f"[FILLED] Куплено по {level['buy_price']} → выставлен SELL по {sell_price}")

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

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