📌 "Market Snapshot CLI"🧑‍💻

Описание:🛠️

Консольный Python-скрипт, который делает моментальный снимок рынка: показывает цену, спред, объём, тренд и индикаторы (RSI, EMA) по заданным парам — идеально подходит для быстрого анализа перед входом в сделку.

🔧 Возможности:🚀

- Цена и спред по выбранным парам (например, BTC/USDT, ETH/USDT).

- 24h объём.

- RSI и EMA (TA-анализ).

- Поддержка нескольких бирж (через ccxt).

- Удобный CLI-вывод.

import ccxt
import argparse
from ta.momentum import RSIIndicator
from ta.trend import EMAIndicator
import pandas as pd

def fetch_ohlcv(symbol, exchange_id):
    exchange = getattr(ccxt, exchange_id)()
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe='1h', limit=50)
    df = pd.DataFrame(ohlcv, columns=['time', 'open', 'high', 'low', 'close', 'volume'])
    return df

def analyze(symbol, exchange_id):
    df = fetch_ohlcv(symbol, exchange_id)
    close = df['close']

    rsi = RSIIndicator(close).rsi().iloc[-1]
    ema = EMAIndicator(close).ema_indicator().iloc[-1]
    last_price = close.iloc[-1]
    spread = df['high'].iloc[-1] - df['low'].iloc[-1]

    print(f"\n📌 {symbol} on {exchange_id.upper()}")
    print(f"Last Price: {last_price:.2f}")
    print(f"Spread: {spread:.2f}")
    print(f"EMA: {ema:.2f}")
    print(f"RSI: {rsi:.2f}")
    print("Trend:", "↑ Bullish" if last_price > ema else "↓ Bearish")

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('symbols', nargs='+', help="Trading pairs like BTC/USDT")
    parser.add_argument('--exchange', default='binance', help="Exchange name supported by ccxt")
    args = parser.parse_args()

    for sym in args.symbols:
        try:
            analyze(sym, args.exchange)
        except Exception as e:
            print(f"Ошибка с {sym}: {e}")

#инструмент

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

📌"Shadow Order Echo" - торговля на основе поведения цен после появления крупных лимитных заявок 🧑‍💻

💡 Идея:🛠️

Крупные лимитные ордера, даже если их не исполняют, могут повлиять на поведение других участников. Иногда они исчезают сразу после реакции толпы. Эта стратегия отслеживает резкие реакции цены на появление больших лимитных заявок и торгует «эхо» этого события — движение в обратную сторону, если заявка была фейковой.

📈 Логика:💻

-Отслеживать крупные заявки в стакане (например, > 10 BTC).

- Если цена быстро смещается в сторону крупной заявки, а затем заявка исчезает:

- Ждём отката и входим в позицию в обратную сторону.

- Ставим короткий стоп за край импульса и таргет — до ближайшей ликвидности.

import ccxt
import time

# Настройки
symbol = 'BTC/USDT'
threshold_volume = 10  # Объем в BTC, выше которого считаем заявку крупной
check_interval = 5     # Интервал проверки стакана в секундах
depth_limit = 20       # Глубина стакана (top N цен)

exchange = ccxt.binance()

def get_large_orders(orderbook_side, threshold):
    return [order for order in orderbook_side if order[1] >= threshold]

def fetch_order_book():
    order_book = exchange.fetch_order_book(symbol, limit=depth_limit)
    large_bids = get_large_orders(order_book['bids'], threshold_volume)
    large_asks = get_large_orders(order_book['asks'], threshold_volume)
    return {'bids': large_bids, 'asks': large_asks}

print(f"[START] Monitoring fake walls on {symbol}...")

previous_snapshot = fetch_order_book()

while True:
    try:
        time.sleep(check_interval)
        current_snapshot = fetch_order_book()

        # Проверка: была крупная заявка, но исчезла
        for side in ['bids', 'asks']:
            prev_orders = set((round(price, 2), round(amount, 2)) for price, amount in previous_snapshot[side])
            curr_orders = set((round(price, 2), round(amount, 2)) for price, amount in current_snapshot[side])

            disappeared = prev_orders - curr_orders
            for price, amount in disappeared:
                print(f"[ALERT] {side.upper()} wall disappeared! Price: {price}, Size: {amount} {symbol.split('/')[0]}")

        previous_snapshot = current_snapshot

    except KeyboardInterrupt:
        print("Stopped by user.")
        break
    except Exception as e:
        print(f"[ERROR] {e}")
        time.sleep(10)

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

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

📌Инструмент: "Correlation Radar" – Локатор корреляций между активами🧑‍💻

Тип: Аналитический инструмент

Назначение: Быстро находить активы с высокой или аномальной корреляцией (или декорреляцией)

Полезен для:💸

- Выбора хеджей

- Спред-трейдинга

- Определения схожих/расходящихся трендов

- Избежания дублирования позиций по коррелирующим активам

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

exchange = ccxt.binance()
symbols = ['BTC/USDT', 'ETH/USDT', 'BNB/USDT', 'SOL/USDT', 'AVAX/USDT', 'XRP/USDT']
timeframe = '1h'
limit = 200

price_data = {}

for symbol in symbols:
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    price_data[symbol] = df['close']

df_prices = pd.DataFrame(price_data)
cor_matrix = df_prices.pct_change().corr()

# Визуализация
sns.heatmap(cor_matrix, annot=True, cmap='coolwarm')
plt.title("📊 Correlation Matrix Between Assets")
plt.show()

# ТОП-5
corr_pairs = cor_matrix.unstack().reset_index()
corr_pairs.columns = ['Asset1', 'Asset2', 'Correlation']
corr_pairs = corr_pairs[corr_pairs['Asset1'] != corr_pairs['Asset2']]
top = corr_pairs.sort_values(by='Correlation', ascending=False).drop_duplicates().head(5)
print("\n🔗 Most Correlated Pairs:\n", top)
low = corr_pairs.sort_values(by='Correlation', ascending=True).drop_duplicates().head(5)
print("\n🔍 Most De-correlated Pairs:\n", low)

#инструмент

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

📌Стратегия: "Volume Spike Reversal"🧑‍💻

Тип: Внутридневная, контртрендовая

Инструменты: Объём, свечной анализ, RSI

Идея:🚀

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

📈 Условия входа в шорт:

- Последняя свеча закрывается с ростом >1.5% от предыдущей.

- Объём выше среднего за последние 20 свечей на 200%+.

- RSI > 75.

- Входим в шорт на следующей свече, если открытие ниже закрытия предыдущей.

📉 Условия входа в лонг:

- Обратное: падение >1.5%, объём выше в 2 раза, RSI < 25.

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

import ccxt
import pandas as pd
from ta.momentum import RSIIndicator

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

ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['time','open','high','low','close','volume'])
df['rsi'] = RSIIndicator(df['close']).rsi()

# Средний объём
avg_vol = df['volume'][:-1].tail(20).mean()
last = df.iloc[-1]
prev = df.iloc[-2]

# Условия шорт
if (
    (prev['close'] - prev['open']) / prev['open'] > 0.015 and
    prev['volume'] > 2 * avg_vol and
    prev['rsi'] > 75 and
    last['open'] < prev['close']
):
    print("⚠️ SHORT Signal: Volume Spike Reversal")

# Условия лонг
if (
    (prev['open'] - prev['close']) / prev['open'] > 0.015 and
    prev['volume'] > 2 * avg_vol and
    prev['rsi'] < 25 and
    last['open'] > prev['close']
):
    print("⚠️ LONG Signal: Volume Spike Reversal")

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

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

📌Strategy: "Candle Cluster Fade" – Контртрендовая торговля от кластеров длинных свечей🧑‍💻

Тип: Внутридневная, mean-reversion

Подходит для: Волатильных рынков (crypto, золото, нефть, NASDAQ)

Идея:🧨

Если в течение короткого времени возникает кластер из нескольких длинных свечей в одном направлении без существенной коррекции — вероятен откат.

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

1. Обнаружение свечного кластера:🚀

3 и более подряд идущих свечей в одном направлении.

Каждая свеча по размеру > X% от средней за N периодов.


2. Проверка контекста:💻

- Индикатор волатильности (например, ATR) не выходит за пределы — это не "новости".

- Нет сильного тренда (по ADX < 25).

3. Вход:💸

В шорт — если кластер на рост.

В лонг — если кластер на падение.

Стоп — за экстремум последней свечи.

Тейк — в зону начала кластера или фиксированный RR.

import ccxt
import pandas as pd

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

def fetch_data():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['time', 'open', 'high', 'low', 'close', 'volume'])
    df['body'] = abs(df['close'] - df['open'])
    df['direction'] = df['close'] > df['open']
    df['avg_body'] = df['body'].rolling(window=20).mean()
    return df

def detect_cluster(df, min_size=3, body_thresh=1.2):
    cluster = []
    for i in range(len(df) - min_size):
        candles = df.iloc[i:i+min_size]
        if all(candles['direction']) or all(~candles['direction']):
            if all(candles['body'] > body_thresh * candles['avg_body']):
                cluster.append((i, candles))
    return cluster

df = fetch_data()
clusters = detect_cluster(df)

if clusters:
    for i, cl in clusters:
        direction = "UP" if cl['direction'].iloc[0] else "DOWN"
        print(f"[ALERT] Candle Cluster detected ({direction}) at index {i}, time: {pd.to_datetime(df['time'].iloc[i], unit='ms')}")
else:
    print("No clusters detected.")

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

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

📌Real-Time RSI Divergence Detector🧑‍💻

Описание:🚀

Скрипт отслеживает дивергенции между RSI и ценой в реальном времени. Это один из классических, но всё ещё сильных сигналов для разворота тренда, особенно на высоких таймфреймах. Уведомления приходят в консоль.

🔍 Что делает скрипт:

- Загружает свечи с Binance через ccxt.

- Рассчитывает RSI.

Ищет:

- Бычью дивергенцию: цена обновляет минимум, RSI — нет.

- Медвежью дивергенцию: цена обновляет максимум, RSI — нет.

import ccxt
import pandas as pd
import time
import datetime

# === Настройки ===
symbol = 'BTC/USDT'
timeframe = '15m'
lookback = 20          # окно анализа дивергенции
sleep_interval = 900   # 15 минут (900 сек)
log_file = 'rsi_divergence_log.csv'

# === RSI расчёт ===
def rsi(series, period=14):
    delta = series.diff()
    gain = delta.where(delta > 0, 0).rolling(period).mean()
    loss = -delta.where(delta < 0, 0).rolling(period).mean()
    rs = gain / loss
    return 100 - (100 / (1 + rs))

# === Поиск дивергенции ===
def detect_divergence(df):
    recent = df.iloc[-lookback:]
    price_min_idx = recent['close'].idxmin()
    rsi_min_idx = recent['rsi'].idxmin()
    price_max_idx = recent['close'].idxmax()
    rsi_max_idx = recent['rsi'].idxmax()

    signal = None
    if price_min_idx > rsi_min_idx and df['close'][price_min_idx] < df['close'][rsi_min_idx]:
        signal = 'Bullish Divergence'
    elif price_max_idx > rsi_max_idx and df['close'][price_max_idx] > df['close'][rsi_max_idx]:
        signal = 'Bearish Divergence'

    return signal

# === Логирование сигнала ===
def log_signal(signal, df):
    timestamp = df['time'].iloc[-1]
    price = df['close'].iloc[-1]
    row = pd.DataFrame([[timestamp, signal, price]], columns=['time', 'signal', 'price'])
    row.to_csv(log_file, mode='a', header=not pd.io.common.file_exists(log_file), index=False)
    print(f"[🔔] {signal} | {timestamp} | Цена: {price:.2f}")

# === Основной цикл ===
def run_monitor():
    exchange = ccxt.binance()

    while True:
        try:
            ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=100)
            df = pd.DataFrame(ohlcv, columns=['time','open','high','low','close','volume'])
            df['time'] = pd.to_datetime(df['time'], unit='ms')
            df['rsi'] = rsi(df['close'])

            signal = detect_divergence(df)
            if signal:
                log_signal(signal, df)
            else:
                print(f"[{datetime.datetime.now().strftime('%H:%M:%S')}] Нет сигнала.")
        except Exception as e:
            print(f"[Ошибка] {e}")
        time.sleep(sleep_interval)

# === Запуск ===
if __name__ == "__main__":
    run_monitor()

#инструмент

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

📌Strategy: "Dark Pool Footprint"

Опережающий вход на основе крупной скрытой активности 🧑‍💻

📌 Идея:🚀

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

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

1. Сканируем поток сделок на наличие повторяющихся объёмов (например, 100+ ордеров по 500 ETH подряд).

2. Сравниваем частоту и распределение цен — если эти блоки проходят по одной цене — признак dark pool-аккумуляции.

3. Если в течение X минут появляется кластер крупных сделок — заходим в ту же сторону.

4. Фильтр на отсутствие видимого волатильного движения (чтобы это не были маркет-движения).

✅ Вход:💸

- Найден блок из ≥5 сделок с объёмом выше среднего (по инструменту).

- Все сделки одной стороной (buy/sell).

- Цена не двигается резко → вероятная "тихая" позиция.

- Вход в ту же сторону с коротким стопом.

import ccxt
import time
import pandas as pd

# Настройка
symbol = 'BTC/USDT'
min_trade_volume = 5  # минимальный объём в BTC для крупной сделки
block_window = 10     # сколько последних сделок анализировать
same_price_threshold = 5  # сколько сделок должно быть по одной цене
poll_delay = 3        # пауза между опросами

exchange = ccxt.binance()

def fetch_trades():
    trades = exchange.fetch_trades(symbol)
    df = pd.DataFrame(trades)
    df['price'] = df['price'].astype(float)
    df['amount'] = df['amount'].astype(float)
    df['side'] = df['side'].astype(str)
    df['volume'] = df['price'] * df['amount']
    return df

def detect_block_trades(trades_df):
    recent = trades_df.tail(block_window)
    big_trades = recent[recent['amount'] > min_trade_volume]

    if big_trades.empty:
        return None

    # Группируем по цене
    grouped = big_trades.groupby('price').size().reset_index(name='count')
    clusters = grouped[grouped['count'] >= same_price_threshold]

    if not clusters.empty:
        price = clusters.iloc[0]['price']
        print(f"[ALERT] Возможен скрытый интерес на {price} USDT — кластер из крупных сделок!")
        return price

    return None

if __name__ == "__main__":
    print(f"🟡 Запуск мониторинга {symbol} на крупные скрытые сделки (через REST)...")
    while True:
        try:
            trades_df = fetch_trades()
            detect_block_trades(trades_df)
        except Exception as e:
            print(f"Ошибка: {e}")
        time.sleep(poll_delay)

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

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

📌Volatility Regime Classifier🧑‍💻

🧠 Идея: определить, в каком волатильностном режиме находится рынок:
Low, Medium, High — для настройки агрессии стратегии (например, менять стопы, тейки, частоту сделок).

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

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

- Расчёт Rolling Volatility по ATR или StdDev;

- Классификация волатильности на три уровня;

- Простая визуализация и экспорт в CSV.

import ccxt
import pandas as pd
import ta
import matplotlib.pyplot as plt

symbol = 'BTC/USDT'
timeframe = '1h'
limit = 300

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

# Расчёт ATR
df['atr'] = ta.volatility.AverageTrueRange(df['high'], df['low'], df['close'], window=14).average_true_range()

# Классификация волатильности
q_low = df['atr'].quantile(0.33)
q_high = df['atr'].quantile(0.66)

def classify_volatility(x):
    if x < q_low:
        return 'Low'
    elif x < q_high:
        return 'Medium'
    else:
        return 'High'

df['vol_regime'] = df['atr'].apply(classify_volatility)

# Визуализация
plt.figure(figsize=(12, 6))
plt.plot(df['timestamp'], df['close'], label='Price', color='black')
colors = {'Low': 'green', 'Medium': 'orange', 'High': 'red'}
for regime, group in df.groupby('vol_regime'):
    plt.scatter(group['timestamp'], group['close'], label=regime, color=colors[regime], s=5)
plt.title('Volatility Regime Classification')
plt.legend()
plt.show()

#инструмент

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

📌"Adaptive Momentum Fade" Strategy 🧑‍💻

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

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

- Ищем краткосрочный импульс вверх/вниз, измеряемый через Rate of Change (ROC);

- Если импульс > адаптивного порога, ищем признаки истощения: свечной откат, слабый объём;

- Заходим в противоположную сторону с коротким тейком и tight стопом.

import ccxt
import pandas as pd
import ta

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

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

df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')

# Индикаторы
df['roc'] = ta.momentum.ROCIndicator(df['close'], window=5).roc()
df['atr'] = ta.volatility.AverageTrueRange(df['high'], df['low'], df['close'], window=14).average_true_range()

# Условия входа в шорт при истощении импульса вверх
df['signal'] = (
    (df['roc'] > df['roc'].rolling(50).mean() + df['roc'].rolling(50).std()) &  # сильный вверх
    (df['close'].shift(1) > df['open'].shift(1)) &  # была зеленая свеча
    (df['close'] < df['open'])                     # текущая свеча красная
).astype(int)

# Вывод сигналов
for i in range(1, len(df)):
    if df['signal'].iloc[i] == 1:
        print(f"🔻 Fade SHORT сигнал {df['timestamp'].iloc[i]} @ {df['close'].iloc[i]:.2f}")

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

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

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

💡 Идея:🧨

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

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

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

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

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

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

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

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

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

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

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

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

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

#инструмент

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

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

📌 Идея:🛠️

Отслеживает список активов (например, топ-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❗️