📌Стратегия: "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❗️

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