📌 Wick Ratio Scanner – сканер “фальшивых” свечей (ловушек)🧑‍💻

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

Этот инструмент сканирует свечи и оценивает долю тени (вика) к телу.

Большой фитиль при маленьком теле часто указывает на:

- Ложный пробой уровня

- Сильное отторжение цены

- Потенциальную ловушку для “толпы”

import requests

def get_klines(symbol="BTCUSDT", interval="15m", limit=50):
    url = f"https://api.binance.com/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}"
    return requests.get(url).json()

def wick_ratio_scanner(symbol="BTCUSDT"):
    klines = get_klines(symbol)
    print(f"🕵️‍♂️ Wick Ratio Scanner для {symbol}:")

    for k in klines[-10:]:
        open_, high, low, close = map(float, [k[1], k[2], k[3], k[4]])
        body = abs(close - open_)
        upper_wick = high - max(open_, close)
        lower_wick = min(open_, close) - low
        max_wick = max(upper_wick, lower_wick)

        if body == 0:
            continue  # избежание деления на 0

        ratio = max_wick / body

        if ratio > 2.5:
            direction = "↑ LONG" if lower_wick > upper_wick else "↓ SHORT"
            print(f"⚠️ Обнаружена свеча с wick ratio {ratio:.2f} → {direction}")

wick_ratio_scanner()

#инструмент

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

📌“Contrarian Volume Echo” — стратегия против толпы с задержкой по объёму 🧑‍💻

📌 Идея:🚀

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

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

1. Обнаружить всплеск объёма: объём на текущей свече в N раз выше среднего.

2. Сравнить направление свечи: если свеча "бычья", предполагаем, что большинство покупают.

3. Подождать 1–2 свечи (эхо): если движение не продолжается, открываем контртрендовую позицию.

4. Фиксация по TP или откату: фиксируем на возврате к среднему или при достижении ATR.

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

- Толпа заходит поздно — маркетмейкер уже “вышел”.

- Объём даёт сигнал, но эффект запаздывает.

- Ложные пробои часто сопровождаются всплесками объёма.

import requests
import numpy as np

def get_klines(symbol="BTCUSDT", interval="5m", limit=100):
    url = f"https://api.binance.com/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}"
    data = requests.get(url).json()
    return data

def contrarian_volume_echo(symbol="BTCUSDT"):
    data = get_klines(symbol)
    closes = [float(k[4]) for k in data]
    volumes = [float(k[5]) for k in data]

    avg_vol = np.mean(volumes[:-3])
    current_vol = volumes[-3]
    prev_vol = volumes[-4]

    price_now = closes[-3]
    price_next = closes[-2]
    body = float(data[-3][4]) - float(data[-3][1])

    if current_vol > avg_vol * 1.8:
        if body > 0 and price_next < price_now:
            print("🔻 Вход в SHORT по Contrarian Volume Echo")
        elif body < 0 and price_next > price_now:
            print("🔺 Вход в LONG по Contrarian Volume Echo")

contrarian_volume_echo()

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

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

📌 "Position Sizer Pro" – инструмент расчёта размера позиции с учётом риска🧑‍💻

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

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

🧠 Что учитывает:💻

- Размер депозита

- Допустимый риск на сделку (в %)

- Точка входа и стоп-лосс

- Валюту депозита и пары

- Плечо (опционально)

def position_size(entry_price, stop_loss_price, capital, risk_percent):
    risk_amount = capital * (risk_percent / 100)
    stop_loss_amount = abs(entry_price - stop_loss_price)
    if stop_loss_amount == 0:
        raise ValueError("Stop-loss cannot be zero")
    size = risk_amount / stop_loss_amount
    print(f"💰 Capital: {capital} USDT")
    print(f"📉 Risk per trade: {risk_percent}% = {risk_amount:.2f} USDT")
    print(f"📈 Entry: {entry_price}, 🛑 Stop: {stop_loss_price}")
    print(f"📦 Position size: {size:.4f} units")

# Пример вызова
position_size(entry_price=25000, stop_loss_price=24800, capital=1000, risk_percent=2)

#инструмент

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

📌"Range Fade Exhaustion" – контртрендовая стратегия на истощение в боковике🧑‍💻

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

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

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

1. Найти локальный диапазон за N свечей (например, 20).

2. Определить верхнюю и нижнюю границы.

3. При пробое границы вверх/вниз:

Если свеча с большим фитилём и объёмом, и цена быстро возвращается внутрь диапазона → сигнал в обратную сторону.

4. Цель — середина диапазона (mean reversion), стоп — за экстремум свечи.

import requests

def get_klines(symbol="BTCUSDT", interval="15m", limit=50):
    url = f"https://api.binance.com/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}"
    return requests.get(url).json()

def range_fade_exhaustion(symbol="BTCUSDT"):
    klines = get_klines(symbol)
    highs = [float(k[2]) for k in klines[:-1]]
    lows = [float(k[3]) for k in klines[:-1]]
    recent = klines[-1]

    range_high = max(highs[-20:])
    range_low = min(lows[-20:])
    mid = (range_high + range_low) / 2

    open_, high, low, close = map(float, [recent[1], recent[2], recent[3], recent[4]])

    wick_up = high - max(open_, close)
    wick_down = min(open_, close) - low
    wick_ratio = max(wick_up, wick_down) / max(abs(close - open_), 1e-5)

    if close < range_low and wick_up > wick_down and wick_ratio > 2.5:
        print("🔁 LONG Signal: ложный пробой вниз")
    elif close > range_high and wick_down > wick_up and wick_ratio > 2.5:
        print("🔁 SHORT Signal: ложный пробой вверх")
    else:
        print("🔍 Нет сигнала (внутри диапазона)")

range_fade_exhaustion()

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

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

📌"Sudden Spread Compression" – стратегия на резком сжатии спреда🧑‍💻

💡 Суть:🛠️

Мониторим спред между бидом и аском. При внезапном резком сжатии (например, с 5–7 тиков до 1–2), это может сигнализировать о готовности к резкому движению – маркет-мейкеры уходят, или наоборот, заходят на проталкивание цены.

📈 Логика:💻

1. Измеряется средний спред за N секунд (например, 20 секунд).

2. Если текущий спред резко сужается на X% ниже среднего, срабатывает сигнал.

3. Фильтрация по направлению последней сделки (buy/sell) или импульсу объёма.

4. Ставим стоп за локальную консолидацию.

import requests
import time
import numpy as np

# Настройки стратегии
SYMBOL = "BTCUSDT"
INTERVAL = 1  # задержка между запросами в секундах
WINDOW = 20  # скользящее окно в отслеживании спреда
THRESHOLD = 0.4  # 40% от среднего спреда

spread_window = []

def get_orderbook(symbol):
    url = f"https://api.binance.com/api/v3/depth?symbol={symbol}&limit=5"
    response = requests.get(url)
    data = response.json()
    bid = float(data['bids'][0][0])
    ask = float(data['asks'][0][0])
    return bid, ask

def monitor_spread():
    while True:
        try:
            bid, ask = get_orderbook(SYMBOL)
            spread = ask - bid
            spread_window.append(spread)
            if len(spread_window) > WINDOW:
                spread_window.pop(0)

            avg_spread = np.mean(spread_window)

            if avg_spread > 0 and spread < avg_spread * THRESHOLD:
                print(f"\n🚨 Spread compression detected!")
                print(f"🔻 Spread: {spread:.2f} < Avg: {avg_spread:.2f}")
                print(f"📌 Bid: {bid}, Ask: {ask}, Time: {time.strftime('%X')}")
        except Exception as e:
            print(f"⚠️ Error: {e}")
        
        time.sleep(INTERVAL)

if __name__ == "__main__":
    print(f"🔍 Monitoring spread compression for {SYMBOL}...\n")
    monitor_spread()

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

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

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