📌"Volatility Snapshot Tool" — монитор волатильности по активам🧑‍💻

📌 Суть инструмента:🛠️

Инструмент показывает текущую реальную (реализованную) волатильность по выбранным активам — позволяет:

- Определить, какие пары сейчас самые активные

- Избегать торговли в слишком "глухих" зонах

- Найти возможности для волатильных стратегий (скальпинг, breakout)

🔧 Что делает:🧨

- Скачивает свечи за последние X минут или часов

- Вычисляет реализованную волатильность (на основе std отклонения log-доходностей или ATR)

- Сортирует пары по волатильности

- Показывает топ-движущихся активов прямо в консоли

import ccxt
import pandas as pd
import numpy as np

symbols = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'LTC/USDT']
exchange = ccxt.binance()
volatility_data = {}

for symbol in symbols:
    ohlcv = exchange.fetch_ohlcv(symbol, '5m', limit=50)
    df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
    df['log_return'] = np.log(df['close'] / df['close'].shift(1))
    realized_vol = df['log_return'].std() * np.sqrt(12)  # за 1 час
    volatility_data[symbol] = realized_vol

sorted_vol = dict(sorted(volatility_data.items(), key=lambda x: x[1], reverse=True))

print("📊 ТОП активов по реализованной волатильности (час):")
for symbol, vol in sorted_vol.items():
    print(f"{symbol}: {vol:.4f}")

#инструмент

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

📌"Anchored VWAP Shift Strategy" стратегия на смещение точки равновесия🧑‍💻

📌 Суть стратегии:🛠️

Вместо классического VWAP (который начинается с открытия дня), используется Anchored VWAP — привязка средней взвешенной цены к определённому событию:

- локальному минимуму

- сильному импульсу

- началу недели и т.п.

> Когда цена уходит выше или ниже Anchored VWAP и удерживается, это сигнал о смене баланса между покупателями и продавцами.

✅ Условия стратегии:🚀

1. Зафиксировать ключевую точку привязки (например, лоу недели)

2. Рассчитать Anchored VWAP с этого момента

3. Когда цена пробивает VWAP и держится выше N свечей → сигнал на вход

4. Тейк по ATR / R:R, стоп — за точку пробоя

import ccxt
import pandas as pd

symbol = "BTC/USDT"
exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv(symbol, '15m', limit=100)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
df["typical_price"] = (df["high"] + df["low"] + df["close"]) / 3

# Привязываем к лоу определённой свечи
anchor_index = df["low"].idxmin()
anchored = df.iloc[anchor_index:]

anchored["cum_vol"] = anchored["volume"].cumsum()
anchored["cum_vwap"] = (anchored["typical_price"] * anchored["volume"]).cumsum() / anchored["cum_vol"]

vwap = anchored["cum_vwap"].iloc[-1]
current_price = df["close"].iloc[-1]

if current_price > vwap:
    print(f"🟢 Цена выше Anchored VWAP ({vwap:.2f}) — сигнал на ЛОНГ")
else:
    print(f"🔴 Цена ниже Anchored VWAP ({vwap:.2f}) — сигнал на ШОРТ")

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

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

“Position Size Calculator CLI”
Консольный калькулятор размера позиции на Python🧑‍💻

📌 Описание🧨

Инструмент помогает рассчитать размер позиции на основе:

- величины капитала

- процента риска

- уровня входа и стоп-лосса

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

def calculate_position_size(account_balance, risk_percent, entry_price, stop_price):
    risk_amount = account_balance * (risk_percent / 100)
    stop_loss_per_unit = abs(entry_price - stop_price)
    if stop_loss_per_unit == 0:
        raise ValueError("Stop loss слишком близко к входу")
    position_size = risk_amount / stop_loss_per_unit
    return position_size

def main():
    print("=== Position Size Calculator ===")
    try:
        balance = float(input("Введите капитал ($): "))
        risk_pct = float(input("Риск на сделку (%): "))
        entry = float(input("Цена входа: "))
        stop = float(input("Цена стоп-лосса: "))

        size = calculate_position_size(balance, risk_pct, entry, stop)
        print(f"\n🔢 Рекомендуемый размер позиции: {size:.4f} единиц актива")
    except Exception as e:
        print(f"[Ошибка] {e}")

if __name__ == "__main__":
    main()

#инструмент

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

📌"Timebox Scalping Strategy" – стратегия торговли по временным блокам🧑‍💻

🧠 Описание:🧨

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

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

ЛОНГ:

- В предыдущем блоке был локальный минимум.

- Текущая цена пробила high прошлого блока.

- Объём выше среднего за 5 блоков.

ШОРТ:

- В предыдущем блоке был локальный максимум.

- Текущая цена пробила low прошлого блока.

- Объём выше среднего за 5 блоков.

import time
import pandas as pd
import requests

# Настройки
symbol = 'BTCUSDT'
interval = '15m'
limit = 50
api_url = f"https://api.binance.com/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}"

def get_klines():
    response = requests.get(api_url)
    data = response.json()
    df = pd.DataFrame(data, columns=[
        'timestamp', 'open', 'high', 'low', 'close', 'volume',
        'close_time', 'quote_asset_volume', 'trades',
        'taker_buy_base', 'taker_buy_quote', 'ignore'
    ])
    df = df.astype(float)
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    return df

def check_signal(df):
    # Берём последние 2 блока
    prev_block = df.iloc[-2]
    curr_block = df.iloc[-1]

    prev_high = prev_block['high']
    prev_low = prev_block['low']
    curr_price = curr_block['close']
    curr_vol = curr_block['volume']

    avg_vol = df['volume'].iloc[-6:-1].mean()

    if curr_price > prev_high and curr_vol > avg_vol:
        print(f"[ALERT] LONG сигнал: цена пробила хай блока | Цена: {curr_price}")
    elif curr_price < prev_low and curr_vol > avg_vol:
        print(f"[ALERT] SHORT сигнал: цена пробила лоу блока | Цена: {curr_price}")
    else:
        print("[INFO] Нет сигнала.")

def run_monitor():
    print("⏳ Запуск Timebox Scalping Monitor...")
    while True:
        try:
            df = get_klines()
            check_signal(df)
            time.sleep(60)  # каждые 60 сек, чтобы не спамить
        except Exception as e:
            print(f"[ERROR] {e}")
            time.sleep(30)

if __name__ == "__main__":
    run_monitor()

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

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

📌Стратегия: Price Ladder Climb🧑‍💻

“Лестница цены” — стратегия по поэтапному закреплению на ключевых уровнях

📌 Идея🛠️

Представь, что цена — это альпинист, поднимающийся по склону. Каждый уровень сопротивления — это новая ступень. Если альпинист успешно закрепляется (удерживается) на ступени в течение N минут — он готов лезть выше.

⚙️ Правила входа💸

1. Обнаружение "ступени": цена пробивает локальное сопротивление (high за последние X минут).

2. Закрепление: если в течение следующих 3 свечей цена не опускается ниже пробитого уровня — это "закрепление".

3. Вход: после закрепления — вход в направлении пробоя.

4. Выход: по цели (например, фиксированное расстояние или следующая ступень).

import requests
import time

symbol = "BTCUSDT"
interval = "1m"
limit = 10

def get_ohlcv():
    url = "https://api.binance.com/api/v3/klines"
    params = {"symbol": symbol, "interval": interval, "limit": limit}
    res = requests.get(url, params=params).json()
    return [[float(x[2]), float(x[3]), float(x[4])] for x in res]  # high, low, close

print("[*] Лестница цены: начало мониторинга")

while True:
    try:
        data = get_ohlcv()
        highs = [x[0] for x in data[:-3]]  # до последних 3 свечей
        recent = data[-3:]
        level = max(highs)

        above_level = all(x[2] > level for x in recent)
        if above_level:
            print(f"[!] Закрепление выше {level:.2f}, возможный ЛОНГ")
        else:
            print(f"[ ] Нет закрепления: последний high={level:.2f}")

        time.sleep(30)
    except Exception as e:
        print("[Ошибка]", e)
        time.sleep(10)

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

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

📌 Инструмент: Multi-Asset Correlation Tracker 🧑‍💻

Онлайн-монитор корреляций между активами с динамикой за последние N часов/дней

📌 Описание 🧨

Этот инструмент отслеживает корреляции между активами (например, BTC/ETH, BTC/SP500, золото/фондовые индексы и т.д.) и показывает, как они меняются во времени. Это помогает:

- Понимать перекосы в поведении активов

- Выявлять дивергенции

- Искать неэффективности и сигналы для парного трейдинга

- Адаптировать стратегию под текущую макро-среду

import requests
import numpy as np
import pandas as pd
from scipy.stats import pearsonr
import time

symbols = ["BTCUSDT", "ETHUSDT"]
interval = "1h"
limit = 100

def get_closes(symbol):
    url = "https://api.binance.com/api/v3/klines"
    params = {"symbol": symbol, "interval": interval, "limit": limit}
    data = requests.get(url, params=params).json()
    return [float(x[4]) for x in data]

print("[*] Запуск мониторинга корреляций...")

while True:
    try:
        series = {s: get_closes(s) for s in symbols}
        df = pd.DataFrame(series)
        corr = df.corr().iloc[0, 1]
        print(f"[📈] Корреляция {symbols[0]} / {symbols[1]} = {corr:.3f}")
        time.sleep(3600)
    except Exception as e:
        print("[Ошибка]", e)
        time.sleep(60)

#инструмент

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

📌Multi-Timeframe Range Confluence - вход внутри локального флета, если он совпадает с глобальной зоной накопления 🧑‍💻

🔍 Описание 🛠️

Стратегия ищет локальный боковик на младшем таймфрейме (например, 5 минут) внутри более крупного боковика (например, 1 час). Предполагается, что если рынок одновременно находится в консолидации на нескольких уровнях, то пробой будет сильным, и можно войти в направлении импульса.

📌 Условия входа 💸

- Цена находится внутри узкого диапазона (5-мин ТФ, например, последние 20 свечей)

- Диапазон на 1H таймфрейме также плоский (узкий High-Low, нет тренда)

- Уровни локального диапазона совпадают с границами глобального

- Пробой локального диапазона вверх → Лонг

- Пробой вниз → Шорт

import requests
import pandas as pd
import time

def get_ohlcv(symbol="BTCUSDT", interval="5m", limit=100):
    url = "https://api.binance.com/api/v3/klines"
    params = {"symbol": symbol, "interval": interval, "limit": limit}
    data = requests.get(url, params=params).json()
    df = pd.DataFrame(data, columns=[
        "timestamp", "open", "high", "low", "close", "volume", 
        "close_time", "quote_asset_volume", "num_trades", 
        "taker_buy_base", "taker_buy_quote", "ignore"
    ])
    df = df.astype(float)
    return df

def detect_range(df, deviation=0.005):
    high = df["high"].max()
    low = df["low"].min()
    spread = (high - low) / low
    return (spread < deviation), low, high

symbol = "BTCUSDT"
while True:
    try:
        df_5m = get_ohlcv(symbol, "5m", 20)
        df_1h = get_ohlcv(symbol, "1h", 24)

        in_local_range, l_low, l_high = detect_range(df_5m)
        in_global_range, g_low, g_high = detect_range(df_1h)

        if in_local_range and in_global_range:
            last_price = df_5m["close"].iloc[-1]
            if last_price > l_high:
                print(f"[🚀 ЛОНГ] Пробой вверх из локального + глобального флета @ {last_price:.2f}")
            elif last_price < l_low:
                print(f"[🔻 ШОРТ] Пробой вниз из локального + глобального флета @ {last_price:.2f}")
            else:
                print("[⏳] Внутри диапазона, ждём пробоя...")
        else:
            print("[…] Нет совпадения диапазонов")

        time.sleep(300)
    except Exception as e:
        print("[Ошибка]", e)
        time.sleep(60)

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

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

📌 Stop-Loss Clustering Radar

Индикатор потенциальных зон скопления стоп-лоссов по ценовому поведению🧑‍💻

📋 Описание 🛠️

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

Это зоны, где можно ожидать:💸

- резких пробоев ("stop hunt"),

- ликвидности для крупных игроков,

- возможно, разворотов после снятия стопов.

Подходит для тех, кто торгует внутри дня или для контртрендовых подходов🚀

import ccxt
import pandas as pd
import time
from ta.volatility import average_true_range

symbol = 'BTC/USDT'
timeframe = '5m'
lookback_bars = 50
impulse_threshold = 1.5  # процент движения для "импульса"
alert_distance = 0.3  # процент отклонения от стоп-уровня

exchange = ccxt.binance()
exchange.load_markets()

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

def detect_stop_clusters(df):
    clusters = []
    for i in range(3, len(df) - 3):
        low = df['low'][i]
        high = df['high'][i]
        
        if low < df['low'][i-1] and low < df['low'][i+1]:
            drop = ((df['close'][i+3] - low) / low) * 100
            if drop > impulse_threshold:
                clusters.append(('support', low))
        
        if high > df['high'][i-1] and high > df['high'][i+1]:
            jump = ((high - df['close'][i+3]) / high) * 100
            if jump > impulse_threshold:
                clusters.append(('resistance', high))
    
    return clusters

def alert_on_proximity(current_price, clusters):
    for zone_type, level in clusters:
        distance = abs(current_price - level) / level * 100
        if distance < alert_distance:
            print(f"[ALERT] Price near {zone_type.upper()} STOP CLUSTER at {level:.2f}")

def main():
    print("Starting Stop-Loss Cluster Radar...\n")
    while True:
        try:
            df = fetch_ohlcv()
            clusters = detect_stop_clusters(df)
            current_price = df['close'].iloc[-1]
            alert_on_proximity(current_price, clusters)
            time.sleep(60)
        except Exception as e:
            print(f"Error: {e}")
            time.sleep(30)

if __name__ == "__main__":
    main()

#инструмент

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

📌 Time-of-Day Volatility Radar🧑‍💻

Описание: инструмент показывает, в какие часы суток для выбранного актива исторически происходит наибольшая волатильность. Это помогает находить лучшие окна для входа в сделки внутри дня.

🎯 Зачем это нужно трейдеру:💸

- Избегать торговли в “мёртвые часы”

- Настраивать стратегии по времени (например, активировать бота только в высоко-волатильные окна)

- Подбирать лучшие часы под стратегии breakout/impulse

🧪 Что делает: 💻

1. Скачивает исторические свечи (например, 5m, 15m).

2. Группирует их по времени суток (например, 09:00, 09:05, 09:10…).

3. Считает средний диапазон свечи (high - low) по каждому временному окну.

4. Выводит топ-часов/минут по волатильности.

import ccxt
import pandas as pd
import datetime

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

def fetch_data():
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
    df['tod'] = df['datetime'].dt.strftime('%H:%M')
    df['range'] = df['high'] - df['low']
    return df

def analyze_volatility(df):
    grouped = df.groupby('tod')['range'].mean().sort_values(ascending=False)
    print("⏱️ Top volatile times of day (UTC):\n")
    for time, rng in grouped.head(10).items():
        print(f"{time}: Avg Range = {rng:.2f}")

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

#инструмент

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

📌Market Memory Echo Strategy🧑‍💻

Концепт: рынок имеет «память» на ценовые уровни, где ранее происходили резкие развороты. Если в течение дня цена подходит к этим «эхо-зонам», вероятность реакции (разворота или импульса) повышается. Стратегия ищет такие зоны по локальным экстремумам предыдущих дней и следит за приближением к ним🛠️

🧠 Логика: 🚀

1. Берём исторические 5-15 минутные свечи за 2–3 дня.

2. Ищем экстремумы, сопровождавшиеся большими объёмами или свечами с тенями.

3. Сохраняем уровни с подтверждённой реакцией как echo levels.

4. В реальном времени следим за приближением цены к этим уровням.

5. Если цена подходит близко — алерт, сигнал на потенциальную реакцию рынка.

import ccxt
import pandas as pd
import time

exchange = ccxt.binance()
symbol = 'ETH/USDT'
timeframe = '15m'
echo_threshold_pct = 0.3  # % расстояния до уровня
history_days = 3
bars_per_day = int(24 * 60 / 15)  # для 15m

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

def find_echo_levels(df):
    echo_levels = []
    for i in range(2, len(df)-2):
        if df['low'][i] < df['low'][i-1] and df['low'][i] < df['low'][i+1]:
            if df['close'][i+1] > df['close'][i] * 1.01:
                echo_levels.append(('support', df['low'][i]))
        if df['high'][i] > df['high'][i-1] and df['high'][i] > df['high'][i+1]:
            if df['close'][i+1] < df['close'][i] * 0.99:
                echo_levels.append(('resistance', df['high'][i]))
    return echo_levels

def monitor_live(echo_levels):
    print("Echo Levels Strategy active...\n")
    while True:
        try:
            live = exchange.fetch_ohlcv(symbol, timeframe, limit=5)
            last_close = live[-1][4]
            for level_type, price in echo_levels:
                distance = abs(last_close - price) / price * 100
                if distance < echo_threshold_pct:
                    print(f"[ALERT] Price {last_close:.2f} near {level_type.upper()} ECHO LEVEL at {price:.2f}")
            time.sleep(60)
        except Exception as e:
            print(f"Error: {e}")
            time.sleep(30)

if __name__ == '__main__':
    df_hist = fetch_history()
    echoes = find_echo_levels(df_hist)
    monitor_live(echoes)

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

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

📌"Latency Arbitrage Simulator"🧑‍💻

Идея:🚀

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

🔍 Принцип:💸

1. Вместо того чтобы сразу совершать арбитражные сделки (что требует скорости), инструмент наблюдает, где запаздывает обновление цены.

2. Если, например, Binance Futures отстаёт от Bybit, это может быть признаком подготовки к движению.

3. Используется как предиктор волатильности или направления, а не как прямой арбитраж.

import time
import requests

symbol = "BTCUSDT"
interval = 5  # Интервал между запросами в секундах
iterations = 20

binance_url = f"https://api.binance.com/api/v3/ticker/price?symbol={symbol}"
bybit_url = "https://api.bybit.com/v2/public/tickers"

def get_binance_price():
    try:
        r = requests.get(binance_url)
        return float(r.json()['price'])
    except Exception as e:
        print("Binance error:", e)
        return None

def get_bybit_price():
    try:
        r = requests.get(bybit_url)
        for ticker in r.json()['result']:
            if ticker['symbol'] == symbol:
                return float(ticker['last_price'])
    except Exception as e:
        print("Bybit error:", e)
        return None

for i in range(iterations):
    print(f"\nИтерация {i+1}")

    binance_price = get_binance_price()
    bybit_price = get_bybit_price()

    if binance_price and bybit_price:
        diff = binance_price - bybit_price
        print(f"Binance: {binance_price}, Bybit: {bybit_price}, Разница: {diff:.2f}")
    else:
        print("Ошибка получения данных")

    time.sleep(interval)

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

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

📌 Market Microstructure Analyzer🧑‍💻

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

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

- Считает, как долго цена "залипает" на каждом уровне (time at price).

- Фиксирует аптики и даунтики — сколько подряд свечей в одном направлении.

- Обнаруживает микроимпульсы — быстрые движения на низких таймфреймах.

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

📄 Возможности применения:💸

Вход на раннем этапе импульса.

- Выход при "залипании" цены.

- Фильтрация флетов перед пробоем.

- Построение микроуровней поддержки/сопротивления.

import requests
import time
from collections import deque

# Пара и настройки
SYMBOL = 'BTCUSDT'
INTERVAL = 3  # сек между запросами
IMPULSE_LENGTH = 3
STUCK_TIME = 60  # сек без движения

price_history = deque(maxlen=IMPULSE_LENGTH + 1)
last_price = None
last_move_time = time.time()

def get_price(symbol):
    url = f'https://api.binance.com/api/v3/ticker/price?symbol={symbol}'
    try:
        res = requests.get(url, timeout=5)
        res.raise_for_status()
        return float(res.json()['price'])
    except Exception as e:
        print(f"⛔ Error fetching price: {e}")
        return None

def detect_impulse(prices):
    if len(prices) < IMPULSE_LENGTH + 1:
        return None
    diffs = [prices[i+1] - prices[i] for i in range(len(prices)-1)]
    if all(d > 0 for d in diffs):
        return "📈 Micro Uptrend"
    elif all(d < 0 for d in diffs):
        return "📉 Micro Downtrend"
    return None

print(f"🚀 Monitoring {SYMBOL} for micro impulses...\n")

while True:
    price = get_price(SYMBOL)
    if price is None:
        time.sleep(INTERVAL)
        continue

    price_history.append(price)
    now = time.time()

    if last_price is not None and price == last_price:
        if now - last_move_time > STUCK_TIME:
            print(f"[{time.strftime('%H:%M:%S')}] ⚠️ Price stuck at {price} for {int(now - last_move_time)}s")
    else:
        last_move_time = now

    impulse = detect_impulse(list(price_history))
    if impulse:
        print(f"[{time.strftime('%H:%M:%S')}] {impulse} → {price_history}")

    last_price = price
    time.sleep(INTERVAL)

#инструмент

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

📌 Trading Strategy Navigator – подсказчик подходящих стратегий под рыночный режим🧑‍💻

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

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

🧠 Как работает:

1. Анализ рыночного режима:

Направление и угол наклона скользящей средней (EMA)

Уровень волатильности (ATR)

Коэффициент автокорреляции

2. Классификация:

Тренд вверх → рекомендованы трендовые стратегии

Флэт + низкая волатильность → стратегии отскока и границ диапазона

Импульс → breakout и объёмные стратегии

3. Вывод рекомендаций: список стратегий под текущую фазу.

import requests
import numpy as np

def get_prices(symbol="BTCUSDT", interval="15m", limit=100):
    url = f"https://api.binance.com/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}"
    data = requests.get(url).json()
    return [float(k[4]) for k in data]  # Close prices

def detect_market_regime(prices):
    ema = np.mean(prices[-20:])
    slope = ema - np.mean(prices[-40:-20])
    atr = np.mean([abs(prices[i] - prices[i-1]) for i in range(1, len(prices))])

    if slope > 0.1 and atr > 10:
        return "Восходящий тренд", ["Следование за трендом", "Пирамидинг", "EMA pullback"]
    elif slope < -0.1 and atr > 10:
        return "Нисходящий тренд", ["Шорт по тренду", "Breakdown", "SuperTrend"]
    elif atr < 5:
        return "Флэт", ["Range Reversion", "Bollinger bounce", "Mean reversion"]
    else:
        return "Импульс / Смена режима", ["Breakout", "Volume spike", "Delta surge"]

prices = get_prices()
regime, strategies = detect_market_regime(prices)

print(f"📈 Режим рынка: {regime}")
print("🔍 Рекомендуемые стратегии:")
for strat in strategies:
    print("–", strat)

#инструмент

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

📌Mean Shift Clustering Price Bands🧑‍💻

Идея: вместо фиксированных индикаторов типа Bollinger Bands использовать метод кластеризации Mean Shift для динамического определения зон концентрации цены. Это позволяет находить актуальные уровни скопления/разворота без предположений о форме распределения.

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

- Используем скользящее окно последних N цен (например, закрытие за 1 час с интервалом в 1 минуту).

- Применяем алгоритм MeanShift из sklearn к этим точкам.

- Получаем динамические "моды" (центры плотности цен).

- Если текущая цена резко уходит от ближайшего центра — вероятен возврат.

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

import numpy as np
import pandas as pd
import requests
from sklearn.cluster import MeanShift
import time

def get_binance_klines(symbol="BTCUSDT", interval="1m", limit=60):
    url = f'https://api.binance.com/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}'
    r = requests.get(url)
    data = r.json()
    closes = [float(i[4]) for i in data]
    return closes

def detect_price_shift(closes, bandwidth=50):
    X = np.array(closes).reshape(-1, 1)
    ms = MeanShift(bandwidth=bandwidth)
    ms.fit(X)
    center = ms.cluster_centers_[0][0]
    current_price = closes[-1]
    deviation = current_price - center
    return deviation, center, current_price

while True:
    closes = get_binance_klines()
    deviation, center, price = detect_price_shift(closes)
    
    print(f"[INFO] Price: {price:.2f} | Center: {center:.2f} | Δ: {deviation:.2f}")

    if deviation > 100:
        print("🔻 Возможен шорт – цена выше плотности рынка")
    elif deviation < -100:
        print("🔺 Возможен лонг – цена ниже плотности рынка")

    time.sleep(60)

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

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

📌 Wick Ratio Scanner – сканер “фальшивых” свечей (ловушек)🧑‍💻

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

Этот инструмент сканирует свечи и оценивает долю тени (вика) к телу.

Большой фитиль при маленьком теле часто указывает на:

- Ложный пробой уровня

- Сильное отторжение цены

- Потенциальную ловушку для “толпы”

import requests

def get_klines(symbol="BTCUSDT", interval="15m", limit=50):
    url = f"https://api.binance.com/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}"
    return requests.get(url).json()

def wick_ratio_scanner(symbol="BTCUSDT"):
    klines = get_klines(symbol)
    print(f"🕵️‍♂️ Wick Ratio Scanner для {symbol}:")

    for k in klines[-10:]:
        open_, high, low, close = map(float, [k[1], k[2], k[3], k[4]])
        body = abs(close - open_)
        upper_wick = high - max(open_, close)
        lower_wick = min(open_, close) - low
        max_wick = max(upper_wick, lower_wick)

        if body == 0:
            continue  # избежание деления на 0

        ratio = max_wick / body

        if ratio > 2.5:
            direction = "↑ LONG" if lower_wick > upper_wick else "↓ SHORT"
            print(f"⚠️ Обнаружена свеча с wick ratio {ratio:.2f} → {direction}")

wick_ratio_scanner()

#инструмент

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

📌“Contrarian Volume Echo” — стратегия против толпы с задержкой по объёму 🧑‍💻

📌 Идея:🚀

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

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

1. Обнаружить всплеск объёма: объём на текущей свече в N раз выше среднего.

2. Сравнить направление свечи: если свеча "бычья", предполагаем, что большинство покупают.

3. Подождать 1–2 свечи (эхо): если движение не продолжается, открываем контртрендовую позицию.

4. Фиксация по TP или откату: фиксируем на возврате к среднему или при достижении ATR.

🧠 Почему работает:💸

- Толпа заходит поздно — маркетмейкер уже “вышел”.

- Объём даёт сигнал, но эффект запаздывает.

- Ложные пробои часто сопровождаются всплесками объёма.

import requests
import numpy as np

def get_klines(symbol="BTCUSDT", interval="5m", limit=100):
    url = f"https://api.binance.com/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}"
    data = requests.get(url).json()
    return data

def contrarian_volume_echo(symbol="BTCUSDT"):
    data = get_klines(symbol)
    closes = [float(k[4]) for k in data]
    volumes = [float(k[5]) for k in data]

    avg_vol = np.mean(volumes[:-3])
    current_vol = volumes[-3]
    prev_vol = volumes[-4]

    price_now = closes[-3]
    price_next = closes[-2]
    body = float(data[-3][4]) - float(data[-3][1])

    if current_vol > avg_vol * 1.8:
        if body > 0 and price_next < price_now:
            print("🔻 Вход в SHORT по Contrarian Volume Echo")
        elif body < 0 and price_next > price_now:
            print("🔺 Вход в LONG по Contrarian Volume Echo")

contrarian_volume_echo()

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

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

📌 "Position Sizer Pro" – инструмент расчёта размера позиции с учётом риска🧑‍💻

📌 Назначение:💸

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

🧠 Что учитывает:💻

- Размер депозита

- Допустимый риск на сделку (в %)

- Точка входа и стоп-лосс

- Валюту депозита и пары

- Плечо (опционально)

def position_size(entry_price, stop_loss_price, capital, risk_percent):
    risk_amount = capital * (risk_percent / 100)
    stop_loss_amount = abs(entry_price - stop_loss_price)
    if stop_loss_amount == 0:
        raise ValueError("Stop-loss cannot be zero")
    size = risk_amount / stop_loss_amount
    print(f"💰 Capital: {capital} USDT")
    print(f"📉 Risk per trade: {risk_percent}% = {risk_amount:.2f} USDT")
    print(f"📈 Entry: {entry_price}, 🛑 Stop: {stop_loss_price}")
    print(f"📦 Position size: {size:.4f} units")

# Пример вызова
position_size(entry_price=25000, stop_loss_price=24800, capital=1000, risk_percent=2)

#инструмент

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

📌"Range Fade Exhaustion" – контртрендовая стратегия на истощение в боковике🧑‍💻

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

Стратегия работает в рамках узкого диапазона (range). Она ищет признаки истощения при попытке пробоя, чтобы входить против движения — на возврат в середину диапазона. Особенно эффективна в периоды низкой волатильности и отсутствия тренда.

🧠 Логика стратегии:🚀

1. Найти локальный диапазон за N свечей (например, 20).

2. Определить верхнюю и нижнюю границы.

3. При пробое границы вверх/вниз:

Если свеча с большим фитилём и объёмом, и цена быстро возвращается внутрь диапазона → сигнал в обратную сторону.

4. Цель — середина диапазона (mean reversion), стоп — за экстремум свечи.

import requests

def get_klines(symbol="BTCUSDT", interval="15m", limit=50):
    url = f"https://api.binance.com/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}"
    return requests.get(url).json()

def range_fade_exhaustion(symbol="BTCUSDT"):
    klines = get_klines(symbol)
    highs = [float(k[2]) for k in klines[:-1]]
    lows = [float(k[3]) for k in klines[:-1]]
    recent = klines[-1]

    range_high = max(highs[-20:])
    range_low = min(lows[-20:])
    mid = (range_high + range_low) / 2

    open_, high, low, close = map(float, [recent[1], recent[2], recent[3], recent[4]])

    wick_up = high - max(open_, close)
    wick_down = min(open_, close) - low
    wick_ratio = max(wick_up, wick_down) / max(abs(close - open_), 1e-5)

    if close < range_low and wick_up > wick_down and wick_ratio > 2.5:
        print("🔁 LONG Signal: ложный пробой вниз")
    elif close > range_high and wick_down > wick_up and wick_ratio > 2.5:
        print("🔁 SHORT Signal: ложный пробой вверх")
    else:
        print("🔍 Нет сигнала (внутри диапазона)")

range_fade_exhaustion()

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

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

📌"Sudden Spread Compression" – стратегия на резком сжатии спреда🧑‍💻

💡 Суть:🛠️

Мониторим спред между бидом и аском. При внезапном резком сжатии (например, с 5–7 тиков до 1–2), это может сигнализировать о готовности к резкому движению – маркет-мейкеры уходят, или наоборот, заходят на проталкивание цены.

📈 Логика:💻

1. Измеряется средний спред за N секунд (например, 20 секунд).

2. Если текущий спред резко сужается на X% ниже среднего, срабатывает сигнал.

3. Фильтрация по направлению последней сделки (buy/sell) или импульсу объёма.

4. Ставим стоп за локальную консолидацию.

import requests
import time
import numpy as np

# Настройки стратегии
SYMBOL = "BTCUSDT"
INTERVAL = 1  # задержка между запросами в секундах
WINDOW = 20  # скользящее окно в отслеживании спреда
THRESHOLD = 0.4  # 40% от среднего спреда

spread_window = []

def get_orderbook(symbol):
    url = f"https://api.binance.com/api/v3/depth?symbol={symbol}&limit=5"
    response = requests.get(url)
    data = response.json()
    bid = float(data['bids'][0][0])
    ask = float(data['asks'][0][0])
    return bid, ask

def monitor_spread():
    while True:
        try:
            bid, ask = get_orderbook(SYMBOL)
            spread = ask - bid
            spread_window.append(spread)
            if len(spread_window) > WINDOW:
                spread_window.pop(0)

            avg_spread = np.mean(spread_window)

            if avg_spread > 0 and spread < avg_spread * THRESHOLD:
                print(f"\n🚨 Spread compression detected!")
                print(f"🔻 Spread: {spread:.2f} < Avg: {avg_spread:.2f}")
                print(f"📌 Bid: {bid}, Ask: {ask}, Time: {time.strftime('%X')}")
        except Exception as e:
            print(f"⚠️ Error: {e}")
        
        time.sleep(INTERVAL)

if __name__ == "__main__":
    print(f"🔍 Monitoring spread compression for {SYMBOL}...\n")
    monitor_spread()

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

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

📌 Fakeout Detector – определение ложных пробоев уровней🧑‍💻

Идея:🚀

Инструмент анализирует поведение цены после пробоя ключевого уровня (high/low за N баров) и проверяет, вернулась ли цена обратно в диапазон. Это может сигнализировать о "ложном пробое" (fakeout) — одном из сильных сигналов в price action.

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

1. Построение high/low за заданный период lookback.

2. Фиксация момента пробоя вверх/вниз.

3. Проверка: вернулась ли цена обратно за фиксированное количество свечей (reentry_window).

4. Если да — сигнал: ложный пробой.

import pandas as pd
import requests
import time

# Настройки
symbol = "BTCUSDT"
interval = "15m"
limit = 200
lookback = 20
reentry_window = 3
api_url = f"https://api.binance.com/api/v3/klines"

def fetch_ohlcv():
    params = {"symbol": symbol, "interval": interval, "limit": limit}
    res = requests.get(api_url, params=params)
    data = res.json()
    df = pd.DataFrame(data, columns=[
        "timestamp", "open", "high", "low", "close", "volume",
        "close_time", "quote_asset_volume", "number_of_trades",
        "taker_buy_base_volume", "taker_buy_quote_volume", "ignore"
    ])
    df["close"] = df["close"].astype(float)
    df["high"] = df["high"].astype(float)
    df["low"] = df["low"].astype(float)
    return df

def detect_fakeout(df):
    recent = df.iloc[-(reentry_window + 1):]
    prior = df.iloc[-(reentry_window + 1 + lookback):- (reentry_window + 1)]

    high_level = prior["high"].max()
    low_level = prior["low"].min()

    first_break = recent.iloc[0]
    reentries = recent.iloc[1:]

    # Ложный пробой вверх
    if first_break["high"] > high_level and all(reentries["close"] < high_level):
        print("🔻 Fakeout вверх обнаружен!")

    # Ложный пробой вниз
    if first_break["low"] < low_level and all(reentries["close"] > low_level):
        print("🔺 Fakeout вниз обнаружен!")

if __name__ == "__main__":
    while True:
        df = fetch_ohlcv()
        detect_fakeout(df)
        time.sleep(60)

#инструмент

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