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

💡 Идея:🧨

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

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

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

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

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

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

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

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

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

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

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

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

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

#инструмент

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

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

📌 Идея:🛠️

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

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

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

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

import ccxt
import time

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

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

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

#инструмент

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

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

💡 Идея:🛠️

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

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

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

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

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

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

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

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

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

def detect_volume_void(df):
    global last_signal_time
    df['change'] = df['close'].pct_change()
    df['avg_vol'] = df['volume'].rolling(window=20).mean()
    df['low_volume'] = df['volume'] < df['avg_vol'] * volume_threshold

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

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

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

if __name__ == '__main__':
    main_loop()

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

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

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

import ccxt

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

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

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

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

#инструмент

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

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

🧠 Идея:💸

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

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

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

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

import ccxt
import time

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

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

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

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

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

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

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

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

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

Описание:

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


---

💡 Принцип:

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

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

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

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

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

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

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

import ccxt
import time

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

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

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

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

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

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

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

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

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

    grid = []

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

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

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

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

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

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

📌Position Size Calculator CLI 🧑‍💻 (Калькулятор позиции в консоли)

Описание:🚀

Инструмент, который помогает рассчитывать оптимальный размер позиции на основе баланса, допустимого риска на сделку и стоп-лосса.

✅ Возможности:🛠️

- Поддержка расчёта для спота и фьючерсов.

- Ввод параметров с клавиатуры (CLI).

Рассчитывает:

- Размер позиции (в единицах актива)

- Объём сделки в долларах

- Кол-во допустимых лотов

def position_size_calculator():
    print("\n📊 Position Size Calculator 📊")

    balance = float(input("Введите баланс аккаунта (в USDT): "))
    risk_percent = float(input("Риск на сделку (%) : "))
    entry_price = float(input("Цена входа: "))
    stop_loss = float(input("Цена стоп-лосса: "))

    risk_amount = balance * (risk_percent / 100)
    stop_size = abs(entry_price - stop_loss)
    if stop_size == 0:
        print("❌ Стоп не может быть 0")
        return

    position_size = risk_amount / stop_size
    position_value = position_size * entry_price

    print(f"\n✅ Размер позиции: {round(position_size, 4)} единиц")
    print(f"💰 Объём сделки: {round(position_value, 2)} USDT")
    print(f"🔒 Риск: {round(risk_amount, 2)} USDT")

if __name__ == "__main__":
    position_size_calculator()

#инструмент

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

📌Wick Trap Strategy — ловушка на длинные хвосты свечей🧑‍💻

📌 Описание:🚀

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

🔍 Условия входа (лонг):

- У свечи нижняя тень > 2× тела свечи.

- Закрытие свечи выше открытия (бычья).

- Предыдущая свеча была медвежьей.

- Объём выше среднего (опционально).

Условия для шорта — зеркальные.

🎯 Выход:💸

Take Profit: +1.5× риска.

Stop Loss: под (или над) экстремум тени.

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

# Настройка ccxt
exchange = ccxt.binance({
    'enableRateLimit': True,
    'options': {'defaultType': 'future'}  # для фьючерсов
})

symbol = 'BTC/USDT'
timeframe = '15m'
limit = 100  # кол-во свечей

def fetch_ohlcv():
    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

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

    body = abs(last['close'] - last['open'])
    lower_wick = min(last['open'], last['close']) - last['low']
    upper_wick = last['high'] - max(last['open'], last['close'])

    ts = last['timestamp'].strftime('%Y-%m-%d %H:%M')

    # LONG trap
    if (lower_wick > 2 * body and
        last['close'] > last['open'] and
        prev['close'] < prev['open']):
        print(f"[LONG SIGNAL] 🟢 at {ts} | Price: {last['close']:.2f}")

    # SHORT trap
    if (upper_wick > 2 * body and
        last['close'] < last['open'] and
        prev['close'] > prev['open']):
        print(f"[SHORT SIGNAL] 🔴 at {ts} | Price: {last['close']:.2f}")

def run():
    print(f"📡 Monitoring Wick Trap on {symbol} ({timeframe})")
    while True:
        try:
            df = fetch_ohlcv()
            check_wick_trap(df)
            time.sleep(60)
        except Exception as e:
            print(f"Error: {e}")
            time.sleep(10)

if __name__ == "__main__":
    run()

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

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

📌Volatility Range Estimator (оценка дневного диапазона по волатильности)🧑‍💻

📌 Описание:

Полезный инструмент для интрадей трейдеров — оценивает вероятный диапазон движения цены в течение текущего дня на основе исторической истинной волатильности (True Range). Это помогает:

- Планировать уровни входа/выхода

- Оценить риски для стопов и тейков

- Избегать входов в "выжатом" рынке

📊 Логика:🛠️

1. Рассчитывает средний True Range (ATR) за последние N дней.

2. На основе текущей цены строит вероятный дневной диапазон:
диапазон = цена ± ATR * коэффициент

3. Выводит зоны: "нижняя граница", "средняя", "верхняя граница"

import ccxt
import pandas as pd
import time

def fetch_ohlcv(symbol="BTC/USDT", timeframe="1d", limit=30):
    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")
    return df

def calculate_atr(df, period=14):
    df['H-L'] = df['high'] - df['low']
    df['H-PC'] = abs(df['high'] - df['close'].shift(1))
    df['L-PC'] = abs(df['low'] - df['close'].shift(1))
    df['TR'] = df[['H-L', 'H-PC', 'L-PC']].max(axis=1)
    df['ATR'] = df['TR'].rolling(window=period).mean()
    return df

def estimate_range(symbol="BTC/USDT", atr_period=14, atr_mult=1.5):
    df = fetch_ohlcv(symbol)
    df = calculate_atr(df, period=atr_period)
    
    latest_close = df['close'].iloc[-1]
    latest_atr = df['ATR'].iloc[-1]

    upper = latest_close + latest_atr * atr_mult
    lower = latest_close - latest_atr * atr_mult

    print(f"\n🎯 {symbol} — Прогноз дневного диапазона (Binance)")
    print(f"Последняя цена     : {latest_close:.2f}")
    print(f"ATR ({atr_period})           : {latest_atr:.2f}")
    print(f"📈 Верхняя граница : {upper:.2f}")
    print(f"📉 Нижняя граница  : {lower:.2f}")

if __name__ == "__main__":
    estimate_range("BTC/USDT")

#инструмент

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

📌"Divergent Latency Fade" — Торговля на запаздывании реакции между синхронизированными активами🧑‍💻

💡 Идея:🛠️

Эта стратегия основывается на концепции "асинхронной реакции" двух тесно связанных активов (например, BTC/USDT на разных биржах или пары ETH и LDO, или SPOT и PERP одного актива). Цель — ловить расхождения в реакции на одно и то же внешнее событие или движение и открывать позиции на возврат к синхронизации.

📌 Пример:💸

- BTC начинает резко расти на Binance.

- ETH на той же бирже с задержкой начинает движение.

- Стратегия фиксирует расхождение, открывает позицию на ETH в сторону BTC с малым SL и фиксированным TP на возврате спрэда.

import ccxt
import time
import numpy as np

binance = ccxt.binance()
symbol_1 = "BTC/USDT"
symbol_2 = "ETH/USDT"
timeframe = "1m"
lookback = 20

def fetch_close_prices(symbol):
    ohlcv = binance.fetch_ohlcv(symbol, timeframe=timeframe, limit=lookback)
    return [x[4] for x in ohlcv]

def calc_slope(prices):
    x = np.arange(len(prices))
    y = np.array(prices)
    m, _ = np.polyfit(x, y, 1)
    return m

def strategy():
    prices_1 = fetch_close_prices(symbol_1)
    prices_2 = fetch_close_prices(symbol_2)
    
    slope_1 = calc_slope(prices_1)
    slope_2 = calc_slope(prices_2)

    divergence = slope_1 - slope_2
    print(f"📉 BTC Slope: {slope_1:.4f}, ETH Slope: {slope_2:.4f}, Δ: {divergence:.4f}")

    if abs(divergence) > 0.5:
        if divergence > 0:
            print("🔔 Signal: ETH BUY — отстаёт от BTC")
        else:
            print("🔔 Signal: ETH SELL — опережает BTC")

if __name__ == "__main__":
    while True:
        strategy()
        time.sleep(60)

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

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

📌"Sentiment Radar" — локальный Python-индикатор настроения рынка по новостям🧑‍💻

📌 Описание:🚀

Sentiment Radar — это лёгкий скрипт, который периодически анализирует упоминания выбранных криптовалют (например, BTC, ETH) в популярных крипто-новостях, оценивая настроение (позитив/негатив) на основе NLP.

Результат отображается в консоли и может быть использован в стратегиях как фильтр:
– "Не открывать позицию, если новостной фон негативный"
– "Открыть при всплеске позитивного настроения"

import requests
from textblob import TextBlob
import time

KEYWORDS = ["Bitcoin", "BTC", "Crypto"]
NEWS_SOURCES = [
    "https://cryptopanic.com/api/posts/?auth_token=demo&public=true",
]

def fetch_headlines():
    headlines = []
    for url in NEWS_SOURCES:
        try:
            response = requests.get(url)
            articles = response.json().get("results", [])
            for a in articles:
                headlines.append(a['title'])
        except Exception as e:
            print("Ошибка загрузки новостей:", e)
    return headlines

def analyze_sentiment(texts):
    polarity = [TextBlob(text).sentiment.polarity for text in texts]
    avg = sum(polarity) / len(polarity) if polarity else 0
    return avg

def sentiment_radar():
    headlines = fetch_headlines()
    relevant = [h for h in headlines if any(k.lower() in h.lower() for k in KEYWORDS)]

    score = analyze_sentiment(relevant)
    print(f"\n🛰 Sentiment Score: {score:.2f} | 📰 {len(relevant)} релевантных новостей")

    if score > 0.2:
        print("✅ Настроение: позитивное")
    elif score < -0.2:
        print("⚠️ Настроение: негативное")
    else:
        print("🔄 Настроение: нейтральное")

if __name__ == "__main__":
    while True:
        sentiment_radar()
        time.sleep(600)  # каждые 10 минут

#инструмент

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

📌"VWAP Pressure Breakout"🧑‍💻

Идея:🛠️

Цена, долгое время торгующаяся ниже VWAP, но с нарастающим объёмом и быстрой попыткой пробоя VWAP вверх — потенциальный сигнал к лонгу. Аналогично для шорта. Это стратегия на прорыв после давления возле важного уровня (VWAP).

📌 Правила входа:💸

📥 Лонг:

- Цена минимум 20 свечей ниже VWAP.

- Объём за последние 3 свечи выше среднего.

- Цена пересекает VWAP вверх — сигнал к покупке.

📤 Шорт:

- Цена минимум 20 свечей выше VWAP.

- Объём за последние 3 свечи выше среднего.

- Цена пересекает VWAP вниз — сигнал к продаже.

import ccxt
import pandas as pd
import numpy as np

symbol = "BTC/USDT"
timeframe = "5m"
limit = 100

exchange = ccxt.binance()

def fetch_data():
    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')
    return df

def calculate_vwap(df):
    df['tp'] = (df['high'] + df['low'] + df['close']) / 3
    df['vwap'] = (df['tp'] * df['volume']).cumsum() / df['volume'].cumsum()
    return df

def detect_signal(df):
    last = df.iloc[-1]
    prev = df.iloc[-2]
    
    below_vwap = (df['close'][-20:] < df['vwap'][-20:]).all()
    above_vwap = (df['close'][-20:] > df['vwap'][-20:]).all()
    
    avg_vol = df['volume'][-20:].mean()
    recent_vol = df['volume'][-3:].mean()
    
    if below_vwap and last['close'] > last['vwap'] and recent_vol > avg_vol:
        print("📈 Лонг сигнал (пробой VWAP вверх после давления)")
    elif above_vwap and last['close'] < last['vwap'] and recent_vol > avg_vol:
        print("📉 Шорт сигнал (пробой VWAP вниз после давления)")
    else:
        print("— Сигналов нет")

if __name__ == "__main__":
    df = fetch_data()
    df = calculate_vwap(df)
    detect_signal(df)

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

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

📌Multi-Timeframe Signal Scanner🧑‍💻

Описание:🛠️

Инструмент сканирует заданный торговый инструмент (или список) на нескольких таймфреймах и отображает сигналы по ключевым индикаторам — например, RSI, MACD, EMA crossover — одновременно. Это позволяет видеть совпадения или противоречия на разных ТФ.

import ccxt
import pandas as pd
import talib
import time

symbol = "BTC/USDT"
timeframes = ["15m", "1h", "4h"]
limit = 100

exchange = ccxt.binance()

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

def analyze(df):
    close = df['close'].values
    rsi = talib.RSI(close, timeperiod=14)
    macd, macdsignal, _ = talib.MACD(close)
    ema_fast = talib.EMA(close, timeperiod=12)
    ema_slow = talib.EMA(close, timeperiod=26)

    signal = []
    if rsi[-1] < 30:
        signal.append("🔻 RSI перепроданность")
    elif rsi[-1] > 70:
        signal.append("🔺 RSI перекупленность")

    if macd[-1] > macdsignal[-1] and macd[-2] < macdsignal[-2]:
        signal.append("📈 MACD бычий кросс")
    elif macd[-1] < macdsignal[-1] and macd[-2] > macdsignal[-2]:
        signal.append("📉 MACD медвежий кросс")

    if ema_fast[-1] > ema_slow[-1] and ema_fast[-2] < ema_slow[-2]:
        signal.append("⚡ EMA бычий кросс")
    elif ema_fast[-1] < ema_slow[-1] and ema_fast[-2] > ema_slow[-2]:
        signal.append("💥 EMA медвежий кросс")

    return signal

def scan():
    for tf in timeframes:
        print(f"\n⏱ Таймфрейм: {tf}")
        df = fetch_ohlcv(symbol, tf)
        signals = analyze(df)
        if signals:
            for s in signals:
                print(s)
        else:
            print("— Нет сигналов")

if __name__ == "__main__":
    scan()

#инструмент

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

📌"Shadow Price Reversion" 🧑‍💻

Идея:🛠️

Стратегия отслеживает разницу между текущей ценой актива и средней ценой закрытия за короткий период. При сильном отклонении происходит вход в противоположную сторону (mean reversion). Это "тень" цены — временный выброс, который, как предполагается, откатится назад.

Подходит для: внутридневной торговли на высоколиквидных парах.

⚙️ Условия входа:💸

- Если цена ниже SMA(n) более чем на X%, — покупка.

- Если цена выше SMA(n) более чем на X%, — продажа.

import ccxt
import pandas as pd
import time

symbol = "BTC/USDT"
timeframe = "5m"
sma_period = 20
threshold = 0.005  # 0.5%

def fetch_ohlcv():
    exchange = ccxt.binance()
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=sma_period + 1)
    df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
    df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
    return df

def check_signal():
    df = fetch_ohlcv()
    df["SMA"] = df["close"].rolling(window=sma_period).mean()
    current_price = df["close"].iloc[-1]
    sma = df["SMA"].iloc[-1]

    deviation = (current_price - sma) / sma

    print(f"\n🔍 {symbol} | Цена: {current_price:.2f} | SMA({sma_period}): {sma:.2f} | Отклонение: {deviation*100:.2f}%")

    if deviation <= -threshold:
        print("📈 Сигнал: ПОКУПКА (Цена ниже SMA)")
    elif deviation >= threshold:
        print("📉 Сигнал: ПРОДАЖА (Цена выше SMA)")
    else:
        print("⏳ Нет сигнала")

if __name__ == "__main__":
    while True:
        try:
            check_signal()
            time.sleep(60)  # 1 минута
        except KeyboardInterrupt:
            break
        except Exception as e:
            print(f"Ошибка: {e}")
            time.sleep(60)

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

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

📌"Cluster Anomaly Radar" – Радар кластерных аномалий объёма и дельты🧑‍💻

💡 Описание:🚀

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

📍Цель — обнаружение скрытых действий маркет-мейкеров или крупных игроков, таких как:

- агрессивное накопление/разгрузка без движения цены,

- резкие вбросы дельты на уровнях,

- "ложные" всплески объёма перед разворотом.

import ccxt
import pandas as pd
import numpy as np

exchange = ccxt.binance()
symbol = "BTC/USDT"
timeframe = "1m"
limit = 500

def fetch_data():
    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")
    return df

def detect_anomalies(df):
    df["delta"] = df["close"] - df["open"]
    df["abs_delta"] = abs(df["delta"])
    vol_mean = df["volume"].rolling(50).mean()
    vol_std = df["volume"].rolling(50).std()

    df["vol_zscore"] = (df["volume"] - vol_mean) / vol_std
    df["delta_ratio"] = df["abs_delta"] / df["volume"]

    for i in range(-30, 0):
        if df["vol_zscore"].iloc[i] > 2 and df["delta_ratio"].iloc[i] < 0.001:
            print(f"⚠️ Аномалия: {df['timestamp'].iloc[i]} — высокий объём без движения (возможное накопление/разгрузка)")

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

#инструмент

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

📌"High Volume Squeeze Entry" — вход после сжатия с объёмным прорывом🧑‍💻

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

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

📊 Условия входа:🚀

Лонг:

1. Цена находится в узком диапазоне последние N свечей (например, 20).

2. Волатильность (например, ATR или разница high-low) падает ниже заданного порога.

3. Происходит пробой вверх диапазона с резким увеличением объёма (более X% от среднего).

4. Объём подтверждает движение (больше среднего за 20 свечей).

Шорт: зеркально, при пробое вниз.

import ccxt
import pandas as pd
import time

symbol = 'BTC/USDT'
timeframe = '5m'
range_window = 20
vol_multiplier = 1.5
volatility_threshold = 0.005

exchange = ccxt.binance()

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

def check_signal(df):
    recent = df.iloc[-range_window:]
    avg_volume = recent['volume'].mean()
    avg_range = (recent['high'] - recent['low']).mean()
    latest = df.iloc[-1]
    
    squeeze = avg_range / latest['close'] < volatility_threshold
    breakout_up = latest['close'] > recent['high'].max()
    breakout_down = latest['close'] < recent['low'].min()
    vol_spike = latest['volume'] > avg_volume * vol_multiplier

    if squeeze and vol_spike:
        if breakout_up:
            print("🔼 LONG Signal! Пробой вверх с объёмом.")
        elif breakout_down:
            print("🔽 SHORT Signal! Пробой вниз с объёмом.")
        else:
            print("💤 Сжатие и объём есть, но пробоя нет.")
    else:
        print("⌛ Нет сигнала.")

while True:
    try:
        df = fetch_data()
        check_signal(df)
        time.sleep(300)  # 5 минут
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(60)

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

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

📌"Fear-Greed Multi-Asset Radar" — мульти-рыночный радар страха и жадности🧑‍💻

🧠 Что делает:🚀

Создаёт реальный индекс страха и жадности в реальном времени по нескольким криптовалютам, а не только по BTC.

Он комбинирует:🛠️

- долю зелёных/красных свечей,

- всплески объёма,

- волатильность,

- соотношение лонгов/шортов,

-активность Twitter/Reddit (опционально)

- визуализирует результат в виде таблицы или консольной панели.

Можно применять как для оценки глобального настроения рынка, так и для поиска перекупленных/перепроданных активов.

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

symbols = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'BNB/USDT']
exchange = ccxt.binance()
timeframe = '1h'
window = 24  # 24 часа

def calculate_sentiment(df):
    green_ratio = (df['close'] > df['open']).sum() / len(df)
    avg_volume = df['volume'].mean()
    volume_spike = df.iloc[-1]['volume'] / avg_volume
    atr = (df['high'] - df['low']).mean()
    atr_pct = atr / df['close'].iloc[-1]

    # Условная формула индекса страха/жадности
    score = green_ratio * 50 + min(volume_spike, 3) * 20 - atr_pct * 100
    return round(score, 2)

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

def print_dashboard():
    print(f"\nFear-Greed Radar — {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}")
    print("-" * 60)
    for sym in symbols:
        try:
            df = fetch_data(sym)
            score = calculate_sentiment(df)
            mood = "😱 FEAR" if score < 40 else "😐 NEUTRAL" if score < 60 else "🚀 GREED"
            print(f"{sym:<10} | Score: {score:<6} | Mood: {mood}")
        except Exception as e:
            print(f"{sym:<10} | Error: {e}")

while True:
    print_dashboard()
    time.sleep(3600)  # обновление каждый час

#инструмент

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

📌“Volume Delta Heatmap” – Тепловая карта дисбаланса объёмов покупки и продажи🧑‍💻

Полезный аналитический инструмент для трейдеров, особенно на фьючерсах и в кластерном анализе.

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

Анализирует дисбаланс между объёмами ask (покупок) и bid (продаж) на каждой свече.

Визуализирует результат в виде цветной тепловой карты, где:

🔴 Красные ячейки — доминируют продажи (продавцы агрессивны).

🟢 Зелёные ячейки — доминируют покупки (покупатели давят вверх).

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

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

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

exchange = ccxt.binance()

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

def plot_heatmap(df):
    data = df['delta'].values.reshape(10, 10)  # 10x10 сетка
    plt.figure(figsize=(8, 6))
    plt.title("Volume Delta Heatmap")
    plt.imshow(data, cmap='RdYlGn', interpolation='nearest')
    plt.colorbar(label='Delta (close - open) * volume')
    plt.show()

df = fetch_ohlcv()
plot_heatmap(df)

#инструмент

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

📌“Адаптивный RSI-Баланс на импульсах ликвидности”
Оригинальная внутридневная стратегия, сочетающая адаптивные зоны RSI с ликвидными всплесками объёма🧑‍💻

🧩 Идея🛠️

Большинство стратегий на RSI используют фиксированные уровни 30/70. Эта стратегия делает уровни RSI динамическими, в зависимости от волатильности, и ищет точки входа на разворотах после всплесков объёма, когда цена отскакивает от локального дисбаланса спроса/предложения.

⚙️ Принцип работы🚀

1. Адаптивный RSI:

Уровни перекупленности/перепроданности вычисляются как:

overbought = 70 + std(RSI, n)

oversold = 30 - std(RSI, n) где n — окно (например, 14 свечей).

2. Объёмный фильтр:
Только когда объём последней свечи > средний * 1.5.

3. Точка входа:

Покупка: RSI пробивает адаптивную нижнюю границу снизу вверх.

Продажа: RSI пробивает верхнюю границу сверху вниз.

- Только при всплеске объёма.

4. Выход:

TP: фиксированная цель в процентах (например, 1%)

SL: по обратному сигналу RSI или волатильному стопу (ATR)

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

symbol = 'BTC/USDT'
timeframe = '5m'
limit = 100
rsi_period = 14
vol_multiplier = 1.5

exchange = ccxt.binance()

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

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 check_signal(df):
    df['RSI'] = rsi(df['close'], rsi_period)
    std_rsi = df['RSI'].rolling(rsi_period).std()
    df['overbought'] = 70 + std_rsi
    df['oversold'] = 30 - std_rsi

    avg_volume = df['volume'].rolling(rsi_period).mean()
    last = df.iloc[-1]
    prev = df.iloc[-2]

    signal = None

    if last['volume'] > avg_volume.iloc[-1] * vol_multiplier:
        if prev['RSI'] < prev['oversold'] and last['RSI'] > last['oversold']:
            signal = f"🔼 LONG | Цена: {last['close']:.2f} | RSI: {last['RSI']:.2f}"
        elif prev['RSI'] > prev['overbought'] and last['RSI'] < last['overbought']:
            signal = f"🔽 SHORT | Цена: {last['close']:.2f} | RSI: {last['RSI']:.2f}"

    return signal

while True:
    try:
        df = fetch_data()
        signal = check_signal(df)
        if signal:
            print(f"[{time.strftime('%H:%M:%S')}] {signal}")
        else:
            print(f"[{time.strftime('%H:%M:%S')}] — сигналов нет")
        time.sleep(60)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(60)

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

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

📌"Market Memory Heatmap"🧑‍💻

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

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

Анализирует историю свечей и объёмов.🚀

Находит ценовые зоны, где🧨:

- происходили резкие развороты,

- был резкий рост объёмов,

- происходило зависание цены (высокое время пребывания).

- Выводит тепловую карту уровней, которые рынок часто «уважает».

🎯 Для чего это полезно трейдеру:

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

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

- Отличная основа для отложенных ордеров, тейк-профитов и стоп-лоссов.

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

# --- Настройки ---
symbol = 'BTC/USDT'
timeframe = '15m'
limit = 500
bins = 50  # Кол-во уровней на ценовой оси

# --- Загрузка данных ---
exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])

# --- Считаем плотность нахождения цены ---
all_prices = []
for _, row in df.iterrows():
    levels = np.linspace(row['low'], row['high'], 10)  # разбиваем свечу по уровням
    all_prices.extend(levels)

# --- Гистограмма плотности ---
hist, edges = np.histogram(all_prices, bins=bins)

# --- Отображаем как тепловую карту ---
plt.figure(figsize=(8, 6))
plt.barh(range(len(hist)), hist, height=1, color=plt.cm.inferno(hist / max(hist)))
plt.yticks(range(len(hist)), [f"{e:.2f}" for e in edges[:-1]])
plt.title(f"🧠 Market Memory Heatmap — {symbol}")
plt.xlabel("Memory Strength (Hits)")
plt.tight_layout()
plt.gca().invert_yaxis()
plt.show()

#инструмент

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