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

📌Инструмент: "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❗️