📌Multi-Timeframe Range Confluence - вход внутри локального флета, если он совпадает с глобальной зоной накопления 🧑‍💻

🔍 Описание 🛠️

Стратегия ищет локальный боковик на младшем таймфрейме (например, 5 минут) внутри более крупного боковика (например, 1 час). Предполагается, что если рынок одновременно находится в консолидации на нескольких уровнях, то пробой будет сильным, и можно войти в направлении импульса.

📌 Условия входа 💸

- Цена находится внутри узкого диапазона (5-мин ТФ, например, последние 20 свечей)

- Диапазон на 1H таймфрейме также плоский (узкий High-Low, нет тренда)

- Уровни локального диапазона совпадают с границами глобального

- Пробой локального диапазона вверх → Лонг

- Пробой вниз → Шорт

import requests
import pandas as pd
import time

def get_ohlcv(symbol="BTCUSDT", interval="5m", limit=100):
    url = "https://api.binance.com/api/v3/klines"
    params = {"symbol": symbol, "interval": interval, "limit": limit}
    data = requests.get(url, params=params).json()
    df = pd.DataFrame(data, columns=[
        "timestamp", "open", "high", "low", "close", "volume", 
        "close_time", "quote_asset_volume", "num_trades", 
        "taker_buy_base", "taker_buy_quote", "ignore"
    ])
    df = df.astype(float)
    return df

def detect_range(df, deviation=0.005):
    high = df["high"].max()
    low = df["low"].min()
    spread = (high - low) / low
    return (spread < deviation), low, high

symbol = "BTCUSDT"
while True:
    try:
        df_5m = get_ohlcv(symbol, "5m", 20)
        df_1h = get_ohlcv(symbol, "1h", 24)

        in_local_range, l_low, l_high = detect_range(df_5m)
        in_global_range, g_low, g_high = detect_range(df_1h)

        if in_local_range and in_global_range:
            last_price = df_5m["close"].iloc[-1]
            if last_price > l_high:
                print(f"[🚀 ЛОНГ] Пробой вверх из локального + глобального флета @ {last_price:.2f}")
            elif last_price < l_low:
                print(f"[🔻 ШОРТ] Пробой вниз из локального + глобального флета @ {last_price:.2f}")
            else:
                print("[⏳] Внутри диапазона, ждём пробоя...")
        else:
            print("[…] Нет совпадения диапазонов")

        time.sleep(300)
    except Exception as e:
        print("[Ошибка]", e)
        time.sleep(60)

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

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

📌 Stop-Loss Clustering Radar

Индикатор потенциальных зон скопления стоп-лоссов по ценовому поведению🧑‍💻

📋 Описание 🛠️

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

Это зоны, где можно ожидать:💸

- резких пробоев ("stop hunt"),

- ликвидности для крупных игроков,

- возможно, разворотов после снятия стопов.

Подходит для тех, кто торгует внутри дня или для контртрендовых подходов🚀

import ccxt
import pandas as pd
import time
from ta.volatility import average_true_range

symbol = 'BTC/USDT'
timeframe = '5m'
lookback_bars = 50
impulse_threshold = 1.5  # процент движения для "импульса"
alert_distance = 0.3  # процент отклонения от стоп-уровня

exchange = ccxt.binance()
exchange.load_markets()

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

def detect_stop_clusters(df):
    clusters = []
    for i in range(3, len(df) - 3):
        low = df['low'][i]
        high = df['high'][i]
        
        if low < df['low'][i-1] and low < df['low'][i+1]:
            drop = ((df['close'][i+3] - low) / low) * 100
            if drop > impulse_threshold:
                clusters.append(('support', low))
        
        if high > df['high'][i-1] and high > df['high'][i+1]:
            jump = ((high - df['close'][i+3]) / high) * 100
            if jump > impulse_threshold:
                clusters.append(('resistance', high))
    
    return clusters

def alert_on_proximity(current_price, clusters):
    for zone_type, level in clusters:
        distance = abs(current_price - level) / level * 100
        if distance < alert_distance:
            print(f"[ALERT] Price near {zone_type.upper()} STOP CLUSTER at {level:.2f}")

def main():
    print("Starting Stop-Loss Cluster Radar...\n")
    while True:
        try:
            df = fetch_ohlcv()
            clusters = detect_stop_clusters(df)
            current_price = df['close'].iloc[-1]
            alert_on_proximity(current_price, clusters)
            time.sleep(60)
        except Exception as e:
            print(f"Error: {e}")
            time.sleep(30)

if __name__ == "__main__":
    main()

#инструмент

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

📌 Time-of-Day Volatility Radar🧑‍💻

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

🎯 Зачем это нужно трейдеру:💸

- Избегать торговли в “мёртвые часы”

- Настраивать стратегии по времени (например, активировать бота только в высоко-волатильные окна)

- Подбирать лучшие часы под стратегии breakout/impulse

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

1. Скачивает исторические свечи (например, 5m, 15m).

2. Группирует их по времени суток (например, 09:00, 09:05, 09:10…).

3. Считает средний диапазон свечи (high - low) по каждому временному окну.

4. Выводит топ-часов/минут по волатильности.

import ccxt
import pandas as pd
import datetime

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

def fetch_data():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
    df['tod'] = df['datetime'].dt.strftime('%H:%M')
    df['range'] = df['high'] - df['low']
    return df

def analyze_volatility(df):
    grouped = df.groupby('tod')['range'].mean().sort_values(ascending=False)
    print("⏱️ Top volatile times of day (UTC):\n")
    for time, rng in grouped.head(10).items():
        print(f"{time}: Avg Range = {rng:.2f}")

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

#инструмент

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

📌Market Memory Echo Strategy🧑‍💻

Концепт: рынок имеет «память» на ценовые уровни, где ранее происходили резкие развороты. Если в течение дня цена подходит к этим «эхо-зонам», вероятность реакции (разворота или импульса) повышается. Стратегия ищет такие зоны по локальным экстремумам предыдущих дней и следит за приближением к ним🛠️

🧠 Логика: 🚀

1. Берём исторические 5-15 минутные свечи за 2–3 дня.

2. Ищем экстремумы, сопровождавшиеся большими объёмами или свечами с тенями.

3. Сохраняем уровни с подтверждённой реакцией как echo levels.

4. В реальном времени следим за приближением цены к этим уровням.

5. Если цена подходит близко — алерт, сигнал на потенциальную реакцию рынка.

import ccxt
import pandas as pd
import time

exchange = ccxt.binance()
symbol = 'ETH/USDT'
timeframe = '15m'
echo_threshold_pct = 0.3  # % расстояния до уровня
history_days = 3
bars_per_day = int(24 * 60 / 15)  # для 15m

def fetch_history():
    total_bars = bars_per_day * history_days
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=total_bars)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    return df

def find_echo_levels(df):
    echo_levels = []
    for i in range(2, len(df)-2):
        if df['low'][i] < df['low'][i-1] and df['low'][i] < df['low'][i+1]:
            if df['close'][i+1] > df['close'][i] * 1.01:
                echo_levels.append(('support', df['low'][i]))
        if df['high'][i] > df['high'][i-1] and df['high'][i] > df['high'][i+1]:
            if df['close'][i+1] < df['close'][i] * 0.99:
                echo_levels.append(('resistance', df['high'][i]))
    return echo_levels

def monitor_live(echo_levels):
    print("Echo Levels Strategy active...\n")
    while True:
        try:
            live = exchange.fetch_ohlcv(symbol, timeframe, limit=5)
            last_close = live[-1][4]
            for level_type, price in echo_levels:
                distance = abs(last_close - price) / price * 100
                if distance < echo_threshold_pct:
                    print(f"[ALERT] Price {last_close:.2f} near {level_type.upper()} ECHO LEVEL at {price:.2f}")
            time.sleep(60)
        except Exception as e:
            print(f"Error: {e}")
            time.sleep(30)

if __name__ == '__main__':
    df_hist = fetch_history()
    echoes = find_echo_levels(df_hist)
    monitor_live(echoes)

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

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

📌"Latency Arbitrage Simulator"🧑‍💻

Идея:🚀

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

🔍 Принцип:💸

1. Вместо того чтобы сразу совершать арбитражные сделки (что требует скорости), инструмент наблюдает, где запаздывает обновление цены.

2. Если, например, Binance Futures отстаёт от Bybit, это может быть признаком подготовки к движению.

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

import time
import requests

symbol = "BTCUSDT"
interval = 5  # Интервал между запросами в секундах
iterations = 20

binance_url = f"https://api.binance.com/api/v3/ticker/price?symbol={symbol}"
bybit_url = "https://api.bybit.com/v2/public/tickers"

def get_binance_price():
    try:
        r = requests.get(binance_url)
        return float(r.json()['price'])
    except Exception as e:
        print("Binance error:", e)
        return None

def get_bybit_price():
    try:
        r = requests.get(bybit_url)
        for ticker in r.json()['result']:
            if ticker['symbol'] == symbol:
                return float(ticker['last_price'])
    except Exception as e:
        print("Bybit error:", e)
        return None

for i in range(iterations):
    print(f"\nИтерация {i+1}")

    binance_price = get_binance_price()
    bybit_price = get_bybit_price()

    if binance_price and bybit_price:
        diff = binance_price - bybit_price
        print(f"Binance: {binance_price}, Bybit: {bybit_price}, Разница: {diff:.2f}")
    else:
        print("Ошибка получения данных")

    time.sleep(interval)

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

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

📌 Market Microstructure Analyzer🧑‍💻

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

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

- Считает, как долго цена "залипает" на каждом уровне (time at price).

- Фиксирует аптики и даунтики — сколько подряд свечей в одном направлении.

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

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

📄 Возможности применения:💸

Вход на раннем этапе импульса.

- Выход при "залипании" цены.

- Фильтрация флетов перед пробоем.

- Построение микроуровней поддержки/сопротивления.

import requests
import time
from collections import deque

# Пара и настройки
SYMBOL = 'BTCUSDT'
INTERVAL = 3  # сек между запросами
IMPULSE_LENGTH = 3
STUCK_TIME = 60  # сек без движения

price_history = deque(maxlen=IMPULSE_LENGTH + 1)
last_price = None
last_move_time = time.time()

def get_price(symbol):
    url = f'https://api.binance.com/api/v3/ticker/price?symbol={symbol}'
    try:
        res = requests.get(url, timeout=5)
        res.raise_for_status()
        return float(res.json()['price'])
    except Exception as e:
        print(f"⛔ Error fetching price: {e}")
        return None

def detect_impulse(prices):
    if len(prices) < IMPULSE_LENGTH + 1:
        return None
    diffs = [prices[i+1] - prices[i] for i in range(len(prices)-1)]
    if all(d > 0 for d in diffs):
        return "📈 Micro Uptrend"
    elif all(d < 0 for d in diffs):
        return "📉 Micro Downtrend"
    return None

print(f"🚀 Monitoring {SYMBOL} for micro impulses...\n")

while True:
    price = get_price(SYMBOL)
    if price is None:
        time.sleep(INTERVAL)
        continue

    price_history.append(price)
    now = time.time()

    if last_price is not None and price == last_price:
        if now - last_move_time > STUCK_TIME:
            print(f"[{time.strftime('%H:%M:%S')}] ⚠️ Price stuck at {price} for {int(now - last_move_time)}s")
    else:
        last_move_time = now

    impulse = detect_impulse(list(price_history))
    if impulse:
        print(f"[{time.strftime('%H:%M:%S')}] {impulse} → {price_history}")

    last_price = price
    time.sleep(INTERVAL)

#инструмент

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

📌 Trading Strategy Navigator – подсказчик подходящих стратегий под рыночный режим🧑‍💻

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

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

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

1. Анализ рыночного режима:

Направление и угол наклона скользящей средней (EMA)

Уровень волатильности (ATR)

Коэффициент автокорреляции

2. Классификация:

Тренд вверх → рекомендованы трендовые стратегии

Флэт + низкая волатильность → стратегии отскока и границ диапазона

Импульс → breakout и объёмные стратегии

3. Вывод рекомендаций: список стратегий под текущую фазу.

import requests
import numpy as np

def get_prices(symbol="BTCUSDT", interval="15m", limit=100):
    url = f"https://api.binance.com/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}"
    data = requests.get(url).json()
    return [float(k[4]) for k in data]  # Close prices

def detect_market_regime(prices):
    ema = np.mean(prices[-20:])
    slope = ema - np.mean(prices[-40:-20])
    atr = np.mean([abs(prices[i] - prices[i-1]) for i in range(1, len(prices))])

    if slope > 0.1 and atr > 10:
        return "Восходящий тренд", ["Следование за трендом", "Пирамидинг", "EMA pullback"]
    elif slope < -0.1 and atr > 10:
        return "Нисходящий тренд", ["Шорт по тренду", "Breakdown", "SuperTrend"]
    elif atr < 5:
        return "Флэт", ["Range Reversion", "Bollinger bounce", "Mean reversion"]
    else:
        return "Импульс / Смена режима", ["Breakout", "Volume spike", "Delta surge"]

prices = get_prices()
regime, strategies = detect_market_regime(prices)

print(f"📈 Режим рынка: {regime}")
print("🔍 Рекомендуемые стратегии:")
for strat in strategies:
    print("–", strat)

#инструмент

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

📌Mean Shift Clustering Price Bands🧑‍💻

Идея: вместо фиксированных индикаторов типа Bollinger Bands использовать метод кластеризации Mean Shift для динамического определения зон концентрации цены. Это позволяет находить актуальные уровни скопления/разворота без предположений о форме распределения.

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

- Используем скользящее окно последних N цен (например, закрытие за 1 час с интервалом в 1 минуту).

- Применяем алгоритм MeanShift из sklearn к этим точкам.

- Получаем динамические "моды" (центры плотности цен).

- Если текущая цена резко уходит от ближайшего центра — вероятен возврат.

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

import numpy as np
import pandas as pd
import requests
from sklearn.cluster import MeanShift
import time

def get_binance_klines(symbol="BTCUSDT", interval="1m", limit=60):
    url = f'https://api.binance.com/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}'
    r = requests.get(url)
    data = r.json()
    closes = [float(i[4]) for i in data]
    return closes

def detect_price_shift(closes, bandwidth=50):
    X = np.array(closes).reshape(-1, 1)
    ms = MeanShift(bandwidth=bandwidth)
    ms.fit(X)
    center = ms.cluster_centers_[0][0]
    current_price = closes[-1]
    deviation = current_price - center
    return deviation, center, current_price

while True:
    closes = get_binance_klines()
    deviation, center, price = detect_price_shift(closes)
    
    print(f"[INFO] Price: {price:.2f} | Center: {center:.2f} | Δ: {deviation:.2f}")

    if deviation > 100:
        print("🔻 Возможен шорт – цена выше плотности рынка")
    elif deviation < -100:
        print("🔺 Возможен лонг – цена ниже плотности рынка")

    time.sleep(60)

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

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

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