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

📌"Volatility Cluster Compression & Release"
Анализ сжатий волатильности в виде кластеров и их последующих импульсных разгрузок. Основана на механике накопления энергии рынком и пробоя.🧑‍💻

⚙️ Идея стратегии:🛠️

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

📌 Ключевые принципы:🚂

- Выделяем "волатильные кластеры": зоны, где цена двигалась в узком диапазоне с пониженной волатильностью.

- Меряем волатильность через ATR или стандартное отклонение.

- Пробой уровня кластера в любую сторону с увеличением объема — сигнал на вход.

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

1. Последние N свечей имеют низкую волатильность (ниже медианы).

2. Цена движется в узком диапазоне (канал не шире X% от цены).

3. Объём начинает увеличиваться.

4. Свеча закрывается за пределами диапазона.

💡 Фильтры:💻

- Учитывать часы сессии (например, пробои чаще на пересечении Лондон/Нью-Йорк).

- Исключать ложные пробои через подтверждение следующей свечи.

🔁 Выход из позиции:🧨

Take Profit — фиксированный % или на первом уровне Fibo.

Stop Loss — внутри зоны кластера или за середину канала

import ccxt
import pandas as pd

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

exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)

df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['range'] = df['high'] - df['low']
df['atr'] = df['range'].rolling(window=14).mean()

# Выделим последние 10 свечей как кластер
cluster = df[-10:]
max_range = cluster['range'].max()
avg_range = df['atr'].mean()

if max_range < avg_range * 0.7:
    high = cluster['high'].max()
    low = cluster['low'].min()
    last_close = df['close'].iloc[-1]
    
    if last_close > high:
        print("🔔 BUY breakout signal")
    elif last_close < low:
        print("🔔 SELL breakout signal")
    else:
        print("⌛ Внутри кластера, ждем пробоя.")
else:
    print("❌ Нет сжатия, нет сигнала.")

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

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

📌"Asymmetrical Cluster Trap"
Использование несимметричных кластеров объёма для выявления ловушек маркет-мейкеров.🧑‍💻

🧠 Идея:🧨

Определять объёмные всплески с несимметричным распределением покупок и продаж в ограниченном ценовом диапазоне. Это может указывать на:

Ловушку для покупателей (резкий рост объёма с преобладанием покупок вверху диапазона перед разворотом вниз).

Ловушку для продавцов (аналогично, но внизу диапазона).

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

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

1. Получаем данные с кластерной платформы (или проксируем с торговых данных через расчёт дельты по стакану).

2. Обнаруживаем участки с:

- узким диапазоном цен;

- резко возрастающим объёмом;

- дельтой сильно смещённой в одну сторону (более 80% покупок или продаж).

3. Сигнал на разворот против толпы.

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
interval = '1m'

def fetch_candles():
    candles = exchange.fetch_ohlcv(symbol, timeframe=interval, limit=20)
    return candles

def detect_cluster_trap(candles):
    volumes = [c[5] for c in candles]
    highs = [c[2] for c in candles]
    lows = [c[3] for c in candles]

    range_avg = sum(h - l for h, l in zip(highs, lows)) / len(candles)
    vol_avg = sum(volumes) / len(candles)

    last = candles[-1]
    last_range = last[2] - last[3]
    last_volume = last[5]

    if last_range < 0.6 * range_avg and last_volume > 2 * vol_avg:
        print(f"[!] Потенциальная ловушка маркет-мейкера. Объёмный кластер на {symbol}")
        print(f"Цена: {last[4]} | Объём: {last_volume:.2f} | Дата: {time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(last[0]/1000))}")

while True:
    try:
        candles = fetch_candles()
        detect_cluster_trap(candles)
        time.sleep(60)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(10)

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

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

📌"Whale Shadow Tracker"
Отслеживание "теневых" действий крупных кошельков по задержке поступлений на биржи — до появления в стакане🧑‍💻

🧠 Идея:

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

Whale Shadow Tracker — инструмент, который:

- Отслеживает крупные входящие транзакции на биржи (например, > 100 BTC).

- Сравнивает время поступления на биржу с изменениями ликвидности в стакане.

- Сигнализирует, если активы уже пришли, но ещё не выставлены ордерами — возможная подготовка к резкому движению.

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

1. Подключается к публичным блокчейн-источникам.

2. Находит крупные транзакции на биржевые адреса (выборка whitelist биржевых адресов).

3. Анализирует, когда активы пришли, и отслеживает ликвидность этих пар через ccxt.

4. Если ликвидность не изменилась через X минут → сигнал: крупный капитал в режиме ожидания → возможен импульс.

import ccxt
import time
import requests
from datetime import datetime

# Настройки Binance
symbol = 'BTC/USDT'
binance = ccxt.binance()
depth_limit = 10
interval_sec = 300
threshold_usd = 1_000_000

# Настройки Arkham
ARKHAM_API_KEY = 'YOUR_ARKHAM_API_KEY'  # 🔐 Вставь свой API ключ
ARKHAM_API_URL = 'https://api.arkhamintelligence.com/v1/labels/activity'

def get_orderbook_snapshot():
    orderbook = binance.fetch_order_book(symbol, limit=depth_limit)
    bids = orderbook['bids']
    asks = orderbook['asks']
    bid_liquidity = sum([price * amount for price, amount in bids])
    ask_liquidity = sum([price * amount for price, amount in asks])
    return {
        'timestamp': datetime.utcnow().isoformat(),
        'bid_liquidity': bid_liquidity,
        'ask_liquidity': ask_liquidity
    }

def check_arkham_whale_activity():
    headers = {'x-api-key': ARKHAM_API_KEY}
    params = {
        "entityType": "whale",  # можно менять на 'exchange', 'fund' и т.д.
        "chain": "bitcoin",
        "timeframe": "last_hour"
    }
    try:
        response = requests.get(ARKHAM_API_URL, headers=headers, params=params)
        if response.status_code == 200:
            data = response.json()
            if data.get('data'):
                print(f"🐋 Найдена ончейн-активность whale-адресов на BTC-сети ({len(data['data'])} записей)")
                return True
            else:
                print("🤫 Нет активности крупных адресов за последний час.")
        else:
            print(f"Ошибка запроса Arkham: {response.status_code}")
    except Exception as e:
        print(f"Ошибка при проверке Arkham: {e}")
    return False

def main():
    print("🚀 Запуск Whale Shadow Tracker...")
    snapshot_before = get_orderbook_snapshot()
    print("🕒 Снимок #1 получен. Ожидание 5 минут...\n")
    time.sleep(interval_sec)
    snapshot_after = get_orderbook_snapshot()

    delta_bid = abs(snapshot_after['bid_liquidity'] - snapshot_before['bid_liquidity'])
    delta_ask = abs(snapshot_after['ask_liquidity'] - snapshot_before['ask_liquidity'])

    if delta_bid < threshold_usd and delta_ask < threshold_usd:
        print("\n⚠️ Обнаружено стабильное состояние ликвидности. Проверяем ончейн-активность...")
        if check_arkham_whale_activity():
            print(f"🔔 ALERT: Whale Shadow — стакан спокоен, но whale-активность зафиксирована.")
        else:
            print("✅ Никаких признаков скрытой активности не найдено.")
    else:
        print(f"{datetime.utcnow().isoformat()} — Ликвидность изменилась значительно. Всё в порядке.")

if __name__ == "__main__":
    main()

#инструмент

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

📌"Sleeping Cluster Breakout"🧑‍💻

💡 Идея:🚀

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

📈 Логика:🛠️

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

2. Мониторим изменения ликвидности в ордербуке вокруг этого уровня.

3. Если:

- цена подходит к уровню;

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

- объём увеличивается —

→ это может быть триггер на вход в позицию по пробою.

import ccxt
import time
from datetime import datetime

symbol = 'BTC/USDT'
exchange = ccxt.binance()
level_price = 110000  # 👈 Уровень, за которым следим
cluster_zone = 10    # Погрешность в $
liquidity_threshold = 500000  # Ликвидность в стакане, USD
check_interval = 10  # секунд

def get_orderbook():
    orderbook = exchange.fetch_order_book(symbol, limit=20)
    bids = orderbook['bids']
    asks = orderbook['asks']
    return bids, asks

def scan_liquidity_near_level(bids, asks):
    near_bids = [b for b in bids if abs(b[0] - level_price) <= cluster_zone]
    near_asks = [a for a in asks if abs(a[0] - level_price) <= cluster_zone]
    total_bid_liquidity = sum([b[0] * b[1] for b in near_bids])
    total_ask_liquidity = sum([a[0] * a[1] for a in near_asks])
    return total_bid_liquidity, total_ask_liquidity

def main():
    print(f"🛡 Мониторинг уровня {level_price}$ с зоной ±{cluster_zone}$")
    while True:
        bids, asks = get_orderbook()
        bid_liq, ask_liq = scan_liquidity_near_level(bids, asks)
        price = exchange.fetch_ticker(symbol)['last']

        print(f"\n[{datetime.utcnow().strftime('%H:%M:%S')}] Цена: {price:.2f}")
        print(f"   🔹 Ликвидность BUY: ${bid_liq:,.0f}")
        print(f"   🔸 Ликвидность SELL: ${ask_liq:,.0f}")

        # Условие пробоя уровня
        if level_price - cluster_zone <= price <= level_price + cluster_zone:
            if bid_liq < liquidity_threshold and ask_liq < liquidity_threshold:
                print("🚨 ALERT: Спящий кластер исчез. Возможен пробой уровня!")

        time.sleep(check_interval)

if __name__ == "__main__":
    main()

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

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

📌"Volatility-Weighted Signal Combiner"🧑‍💻

💡 Описание:🧨

Это модуль, который комбинирует сигналы из нескольких стратегий (например: импульс, откат, пробой уровня) — но взвешивает их в зависимости от текущей волатильности рынка. То есть, если рынок в фазе высокой волатильности — предпочтение отдается импульсным стратегиям, в фазе низкой — откатным и mean-reversion.

✅ Что делает:💸

- Получает волатильность по ATR или стандартному отклонению.

- Оценивает силу сигналов по каждой стратегии.

- Взвешивает силу с учётом волатильности.

- Выводит финальный алерт на вход в позицию, если итоговый вес выше порога.

import ccxt
import time
import statistics

# Настройки
symbol = 'BTC/USDT'
timeframe = '1m'
limit = 30  # Количество свечей для анализа

exchange = ccxt.binance()

def fetch_prices():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
    return [candle[4] for candle in ohlcv]  # Закрытия

def momentum_signal(prices):
    return 1 if prices[-1] > prices[-2] * 1.002 else 0

def mean_reversion_signal(prices):
    return 1 if prices[-1] < min(prices[-5:]) else 0

def calc_volatility(prices):
    return statistics.stdev(prices)

def weighted_combiner(momentum, reversion, volatility):
    if volatility > 30:
        return momentum * 0.7 + reversion * 0.3
    else:
        return momentum * 0.3 + reversion * 0.7

while True:
    try:
        prices = fetch_prices()
        momentum = momentum_signal(prices)
        reversion = mean_reversion_signal(prices)
        vol = calc_volatility(prices)
        signal_strength = weighted_combiner(momentum, reversion, vol)

        print(f"🟡 Цена: {round(prices[-1], 2)} | Волатильность: {round(vol, 2)} | Сигнал: {round(signal_strength, 2)}")

        if signal_strength > 0.6:
            print("✅ Сигнал на вход в ЛОНГ!\n")

        time.sleep(60)
    except Exception as e:
        print(f"Ошибка: {e}")
        time.sleep(60)

#инструмент

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

📌"Behavioral Clustering" — кластеризация поведения толпы🧑‍💻

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

Эта стратегия опирается на кластеризацию аномального поведения розничных трейдеров — например, массовый FOMO (покупка на росте), panic selling (распродажа на падении), нерешительность (flat). Основная идея: когда большинство трейдеров делает одно и то же — делать наоборот.

Кластеры поведения определяются через:

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

🔁 скорость смены открытых позиций (open interest delta),

🔥 рост объёма и волатильности,

🐳 активность мелких vs крупных адресов (например, через Arkham или Glassnode).

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

1. FOMO-кластер: Резкий рост объёма, цены и открытого интереса со стороны розницы → вход в шорт.

2. Паника-кластер: Массовая распродажа с низкими ценами и всплеском активности мелких адресов → вход в лонг.

3. Нерешительность: Плоский рынок, низкий объём, но рост открытого интереса → возможен прорыв → торгуется пробой.

import ccxt
import time

exchange = ccxt.binance()

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

def get_data():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=20)
    prices = [c[4] for c in ohlcv]
    volumes = [c[5] for c in ohlcv]
    return prices, volumes

def detect_fomo(prices, volumes):
    price_jump = prices[-1] > max(prices[:-1]) * 1.01
    volume_spike = volumes[-1] > sum(volumes[-5:-1]) / 2
    return price_jump and volume_spike

def detect_panic(prices, volumes):
    price_drop = prices[-1] < min(prices[:-1]) * 0.99
    volume_spike = volumes[-1] > sum(volumes[-5:-1]) / 2
    return price_drop and volume_spike

while True:
    try:
        prices, volumes = get_data()

        if detect_fomo(prices, volumes):
            print("🔻 Обнаружен FOMO кластер → Сигнал на ШОРТ")

        elif detect_panic(prices, volumes):
            print("🔺 Обнаружена ПАНИКА → Сигнал на ЛОНГ")

        else:
            print("⚪ Рынок без явных кластеров. Ожидание...")

        time.sleep(60)

    except Exception as e:
        print(f"Ошибка: {e}")
        time.sleep(60)

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

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

📌"Wallet Flow Radar" – Радар движения средств между адресами и биржами🧑‍💻

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

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

Особенности:🚀

- Анализ потоков на основе публичных данных (Arkham, Blockchair, Whale Alert, либо через ноды).

- Фильтрация по типу адреса: биржа → личный кошелёк, наоборот, OTC, холодные и горячие кошельки.

- Автоматическое определение направлений и масштабов движения.

- Консольные алерты и логгирование.

- Выдача сигнала: если, например, 1000 BTC отправлены с кошелька → биржа, это может быть сигнал к продаже.

import time
import requests

ARKHAM_API_KEY = 'your_api_key_here'
THRESHOLD_USD = 1_000_000  # отслеживать транзакции свыше $1M

def check_flows():
    url = 'https://api.arkhamintelligence.com/transactions'
    headers = {'Authorization': f'Bearer {ARKHAM_API_KEY}'}

    response = requests.get(url, headers=headers)
    data = response.json()

    for tx in data.get('transactions', []):
        amount_usd = tx['valueUSD']
        direction = tx['direction']  # 'in' or 'out'
        exchange_related = 'exchange' in tx['fromLabel'].lower() or 'exchange' in tx['toLabel'].lower()

        if amount_usd > THRESHOLD_USD and exchange_related:
            if direction == 'in':
                print(f"⚠️ Крупный ввод на биржу: {amount_usd:.0f} USD → возможный дамп")
            elif direction == 'out':
                print(f"📦 Крупный вывод с биржи: {amount_usd:.0f} USD → возможно накопление")

while True:
    try:
        check_flows()
        time.sleep(60)
    except Exception as e:
        print(f"Ошибка: {e}")
        time.sleep(60)

#инструмент

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

📌"Volatility Compression Scanner" — Сканер Сжатия Волатильности🧑‍💻

📌 Цель:🚀

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

💡 Почему это полезно?🧨

- Большинство импульсов и прорывов начинаются после сжатия диапазона.

- Можно подготовиться заранее: выставить лимитные ордера, настроить алерты или построить торговую идею.

⚙️ Принцип:🛠️

1. Загружаем данные по свечам (например, 1h) для множества инструментов.

2. Вычисляем диапазон свечей (high - low) за последние N периодов.

3. Ищем инструменты, у которых текущая волатильность — в X раз ниже средней.

4. Выводим список — возможные кандидаты на пробой.

import ccxt
import time
from statistics import mean

exchange = ccxt.binance()
symbols = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'XRP/USDT']
timeframe = '1h'
lookback = 20
compression_threshold = 0.3  # текущая волатильность < 30% от средней

def get_volatility(symbol):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=lookback)
    ranges = [candle[2] - candle[3] for candle in ohlcv[:-1]]  # high - low
    current = ohlcv[-1][2] - ohlcv[-1][3]
    avg_range = mean(ranges)
    if avg_range == 0:
        return None
    compression = current / avg_range
    return compression

while True:
    print(f"\n🔍 Поиск сжатия волатильности ({time.strftime('%X')}):")
    for symbol in symbols:
        try:
            ratio = get_volatility(symbol)
            if ratio is not None and ratio < compression_threshold:
                print(f"🟡 {symbol} — волатильность сжата ({ratio:.2f})")
        except Exception as e:
            print(f"[Ошибка] {symbol}: {e}")
    time.sleep(300)  # 5 минут

#инструмент

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

📌"Time-Weighted Bid Walls" (TWBW) — вход на основе жизни крупных заявок🧑‍💻

🧠 Идея:🚀

Отслеживать bid-стенки (стены заявок на покупку) в order book и определять, как долго они удерживаются на месте.
Если большая лимитная заявка долго не исчезает и не исполняется, это может сигнализировать о поддержке, за которой стоит реальный спрос.

Если стена держится долго — входим в лонг.

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

1. Сканим стакан с периодичностью

2. Фиксируем крупные заявки (на 10–20x больше среднего объема).

3. Засекаем, сколько времени они остаются на месте.

4. Если заявка живёт дольше N секунд и не исполняется — это сигнал на вход в лонг.

📉 Выход:💸

- Если стена исчезает → выход по стопу.

- Если цена уходит вверх на X% → тейк.

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
depth_limit = 50
threshold_volume = 10  # например, 10 BTC
wall_lifetime = 60  # сек

walls = {}

def fetch_order_book():
    order_book = exchange.fetch_order_book(symbol, limit=depth_limit)
    bids = order_book['bids']
    return bids

def track_bid_walls():
    global walls
    bids = fetch_order_book()

    now = time.time()
    for price, volume in bids:
        if volume >= threshold_volume:
            key = str(price)
            if key not in walls:
                walls[key] = now
            elif now - walls[key] >= wall_lifetime:
                print(f"🟢 Сигнал LONG: удерживается стенка {volume:.2f} BTC @ {price:.2f}")
                walls[key] = now + 99999  # не триггерить повторно
        else:
            # удаляем, если объем упал
            walls.pop(str(price), None)

    # чистим старые
    walls = {k: v for k, v in walls.items() if now - v <= 2 * wall_lifetime}

while True:
    try:
        track_bid_walls()
        time.sleep(5)
    except Exception as e:
        print(f"[Ошибка] {e}")
        time.sleep(10)

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

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

📌"Whale Fingerprint Tracker"🧑‍💻

Отслеживание уникального поведения китов по их характерным шаблонам торгов

💡 Суть идеи:🧨

Крупные адреса (киты) зачастую входят и выходят с рынка по характерным шаблонам, например:

- Разделение ордера на множество мелких покупок

- Повторяющиеся последовательности ордеров

- Частота и время активности

- Стратегии ложных пробоев

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

📌 Что делает скрипт:🛠️

- Сканирует адреса с наибольшим объёмом сделок (можно с Arkham API или Glassnode/Whale-alert)

- Сохраняет шаблон поведения (время, объёмы, направления, частота)

- Сравнивает в реальном времени активность с этими шаблонами

- Выдаёт сигнал в консоль, если похожая активность снова появляется

🐍 Пример архитектуры на Python (упрощённый скелет):

# Псевдокод — пример логики анализа активности
class WhaleTracker:
    def __init__(self):
        self.whale_fingerprints = {}

    def add_whale_pattern(self, address, pattern):
        self.whale_fingerprints[address] = pattern

    def fetch_live_activity(self):
        # Тут может быть API Arkham, Whale Alert или собственный mempool scanner
        return [
            {"address": "0xabc...", "volume": 1_200_000, "type": "buy", "timestamp": 1720000000},
            {"address": "0xdef...", "volume": 950_000, "type": "sell", "timestamp": 1720000050}
        ]

    def check_for_matches(self):
        for tx in self.fetch_live_activity():
            address = tx["address"]
            if address in self.whale_fingerprints:
                print(f"🔍 Whale activity match! Address: {address} — Type: {tx['type']} — Volume: {tx['volume']}")

#инструмент

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