📌Grid Break Scalper - (Сеточный пробой) 🧑‍💻

📌Идея: 🧨

- Строим сетку лимитных ордеров выше и ниже текущей цены (например через каждые 0.2–0.5%).

- Когда цена «цепляет» ордер снизу (бай) → закрываем его на ближайшем возврате вверх (+0.2–0.3%).

- Когда цена «цепляет» ордер сверху (селл) → закрываем его на возврате вниз.

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

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

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

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

📌"Market Replay Console" - (реплей рынка прямо в терминале) 🧑‍💻

📌 Идея: 🧨

Большинство трейдеров анализируют рынок по истории свечей или графикам, но редко «проживают» движение в реальном времени.

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

Это даёт:💸

- тренировку реакции на движения без риска;

- тестирование стратегий «на ощущениях»;

- поиск паттернов в динамике, а не только на статичном графике.

🔑 Особенности: 🛠️

- Работает в консоли (без графики, всё текстом).

- Подгружает историю цен с Binance через ccxt.

- Симулирует «живой рынок» с задержкой между свечами.

- Можно поставить паузу или ускорить «прокрутку».

import ccxt
import time

# Настройка
exchange = ccxt.binance({'enableRateLimit': True})
symbol = "BTC/USDT"
timeframe = "1m"
limit = 100  # кол-во свечей в реплее
delay = 0.5  # секунда между свечами (можно менять для ускорения)

# Загружаем историю
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)

print(f"▶ Реплей {symbol}, таймфрейм {timeframe}, свечей: {limit}\n")

for candle in ohlcv:
    ts, open_, high, low, close, vol = candle
    print(f"🕒 Цена {close:.2f} | O:{open_:.2f} H:{high:.2f} L:{low:.2f} V:{vol:.2f}")
    time.sleep(delay)

#инструмент

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

📌"Pattern Replay Trainer" - (тренажёр распознавания рыночных паттернов в консоли)🧑‍💻

📌 Идея: 🧨

Вместо статичных графиков трейдер получает рандомные фрагменты истории цены (например, по 30 свечей).
Он не знает, что будет дальше, и должен принять решение: лонг / шорт / пропуск.
После выбора программа «открывает» будущее (следующие свечи) и показывает, был ли трейдер прав.

Это позволяет:

- тренировать интуицию и скорость реакции;

- оттачивать свои правила входа/выхода;

- «проживать рынок» как в реальности, но без риска.

import ccxt
import pandas as pd
import random
import time

exchange = ccxt.binance({'enableRateLimit': True})
symbol = "BTC/USDT"
timeframe = "1m"
limit = 500

# Загружаем историю
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")

def replay_trainer(window=30, future=10):
    start = random.randint(0, len(df) - window - future - 1)
    sample = df.iloc[start:start+window]
    future_data = df.iloc[start+window:start+window+future]

    print("\n🕒 Новый фрагмент истории!")
    for _, row in sample.iterrows():
        print(f"{row['ts']} | O:{row['open']:.2f} C:{row['close']:.2f}")

    choice = input("\nТвой выбор (long/short/skip): ").strip().lower()

    print("\n📊 Будущее:")
    for _, row in future_data.iterrows():
        print(f"{row['ts']} | O:{row['open']:.2f} C:{row['close']:.2f}")
        time.sleep(0.2)

    first = sample["close"].iloc[-1]
    last = future_data["close"].iloc[-1]
    change = (last - first) / first * 100

    if choice == "long":
        print(f"\n✅ Результат LONG: {change:.2f}%")
    elif choice == "short":
        print(f"\n✅ Результат SHORT: {-change:.2f}%")
    else:
        print(f"\n⏸ Пропуск | рынок изменился на {change:.2f}%")

if __name__ == "__main__":
    while True:
        replay_trainer()
        if input("\nПродолжить? (y/n): ").strip().lower() != "y":
            break

#инструмент

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

📌"Flash Drop Catcher" - (ловим микро-обвалы и быстрые 🧑‍💻 возвраты)

📌 Идея: 🧨

На крипторынке часто происходят мгновенные резкие проливы (1–3% за минуту), после которых следует отскок.

Эта стратегия пытается поймать такие движения:

- Если цена падает слишком быстро за короткое время → входим в лонг.

- Если цена быстро отскакивает после падения → фиксируем профит.

- Если отскока нет → фиксируем небольшой стоп.

🔑 Логика входа и выхода:💸

- Берём 1-минутные свечи.

- Если цена закрывается ниже на >1% по сравнению со средней ценой последних 5 свечей → покупаем.

- Если после входа цена вырастает хотя бы на +0.6% → продаём.

- Если цена падает ещё на -0.5% → стоп-лосс.

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

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

📌"Trade Journal AutoLogger" - (автоматический торговый дневник для сделок с Binance через ccxt) 🧑‍💻

📌 Идея: 🧨

У большинства трейдеров нет дисциплины вести дневник сделок вручную.

Этот инструмент делает всё сам:

- собирает сделки с Binance (история + новые ордера),

- сохраняет их в CSV,

- считает прибыль/убыток, комиссии, % доходности,

- формирует «дневник» трейдера.

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

# ⚠️ Укажи свои API ключи Binance
api_key = "YOUR_API_KEY"
api_secret = "YOUR_API_SECRET"

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

def fetch_trades(symbol="BTC/USDT", since=None, limit=50):
    try:
        trades = exchange.fetch_my_trades(symbol, since=since, limit=limit)
        return trades
    except Exception as e:
        print("Ошибка:", e)
        return []

def log_trades_to_csv(symbol="BTC/USDT", filename="trade_journal.csv"):
    trades = fetch_trades(symbol)
    if not trades:
        print("Нет новых сделок")
        return

    df = pd.DataFrame(trades)
    df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms")
    df = df[["datetime", "symbol", "side", "price", "amount", "cost", "fee"]]

    try:
        old = pd.read_csv(filename)
        df = pd.concat([old, df]).drop_duplicates()
    except FileNotFoundError:
        pass

    df.to_csv(filename, index=False)
    print(f"💾 Журнал обновлён: {len(df)} сделок сохранено")

if __name__ == "__main__":
    print("▶ Авто-журнал сделок запущен (Ctrl+C для выхода)")
    while True:
        log_trades_to_csv("BTC/USDT")
        time.sleep(60)  # обновляем каждую минуту

#инструмент

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

📌"Impulse Reversal Hunter" - (охота на импульсные всплески и быстрый откат) 🧑‍💻

📌 Идея: 🧨

На крипторынке часто случаются короткие импульсы:

цена за 1–3 минуты выстреливает на +1–2%,

но затем так же быстро откатывается.

Большинство участников опаздывают и входят в конце импульса → мы играем против них.

🔑 Логика: 🛠️

1. Берём 1-минутные свечи.

2. Если текущая свеча закрылась выше предыдущей на ≥1%, считаем это импульсом.

3. Входим в шорт на следующей свече.

4. Закрываем при откате -0.5% от цены входа.

5. Если импульс продолжился и цена выросла ещё на +0.5% от входа → стоп-лосс.

import ccxt
import pandas as pd
import time

exchange = ccxt.binance({'enableRateLimit': True})
symbol = "BTC/USDT"
timeframe = "1m"

print(f"🚀 Старт стратегии 'Impulse Reversal Hunter' для {symbol}")

def check_impulse(symbol):
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=3)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
    prev = df.iloc[-2]["close"]
    last = df.iloc[-1]["close"]

    change = (last - prev) / prev * 100
    if change >= 1.0:
        return ("SHORT", last, change)
    return None

while True:
    signal = check_impulse(symbol)
    if signal:
        side, price, change = signal
        print(f"⚡️ Сигнал: {side} @ {price:.2f} (импульс {change:.2f}%)")
    time.sleep(60)

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

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

📌«Market Rhythm Analyzer» - (анализатор ритма рынка в реальном времени) 🧑‍💻

📌 Идея: 🧨

Рынок двигается импульсами: серия быстрых свечей → пауза → новый рывок.

Этот «ритм» можно измерить и использовать:

- для фильтрации сделок (не входить, если рынок «устал»),

- для поиска фаз накопления/импульса,

- для оценки волатильности без индикаторов.

🔑 Как работает: 🛠️

1. Считывает поток цен с Binance.

2. Измеряет средний интервал между движениями цены (например, каждые 0.1% изменения).

3. Строит в консоли или файле «пульсограмму» рынка.

4. Если ритм ускоряется → значит рынок в фазе импульса.

5. Если ритм замедляется → значит рынок в накоплении/передышке.

import ccxt
import time
from collections import deque

symbol = "BTC/USDT"
price_step = 0.001   # 0.1% шаг для фиксации изменения
window = 10          # усреднение по N шагам
exchange = ccxt.binance()

last_price = None
last_event_time = None
intervals = deque(maxlen=window)

while True:
    try:
        ticker = exchange.fetch_ticker(symbol)
        price = ticker['last']

        if last_price is None:
            last_price = price
            last_event_time = time.time()
            continue

        # проверяем шаг
        if abs(price - last_price) / last_price >= price_step:
            now = time.time()
            interval = now - last_event_time
            intervals.append(interval)
            last_event_time = now
            last_price = price

            if len(intervals) == window:
                avg_interval = sum(intervals) / len(intervals)
                print(f"📊 {symbol} | Средний ритм: {avg_interval:.2f} сек/шаг")

                if avg_interval < 5:
                    print("⚡ Рынок в импульсе!")
                elif avg_interval > 15:
                    print("😴 Рынок в накоплении...")
                else:
                    print("📈 Нормальный ритм.")
        time.sleep(1)

    except Exception as e:
        print("Ошибка:", e)
        time.sleep(5)

#инструмент

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

📌«Фазовый импульс»🧑‍💻

🔑 Идея: 🧨

Цена на рынке не двигается равномерно — она идёт фазами:

1. Периоды тихого накопления (малый диапазон, сжатие).

2. Резкий импульс (вверх или вниз).

3. Затухание и коррекция, после чего цикл повторяется.

Мы торгуем не направление, а сам факт перехода между фазами, что делает стратегию нейтральной к рынку.

📜 Правила стратегии: 🛠️

📌Вход: 💸

1. Следим за ценой на 1m или 5m таймфрейме.

2. Если диапазон за последние N свечей (например 20) меньше X% от средней дневной волатильности → значит, идёт накопление.

3. Ставим два стоп-ордера:

Buy Stop чуть выше накопления,

Sell Stop чуть ниже.

📌Выход: 💸

Если активирован Buy → цель = 1.5–2× ширина накопления.

Если активирован Sell → то же самое.

Второй ордер автоматически отменяется (OCO-логика).

import ccxt
import time
import statistics

# === Настройки ===
symbol = "BTC/USDT"
timeframe = "5m"
lookback = 20         # кол-во свечей для поиска накопления
accum_threshold = 0.004  # порог узости (0.4% от цены)
take_profit_mult = 2     # цель в x ширины диапазона
sleep_time = 30          # задержка между проверками (сек)

exchange = ccxt.binance()

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

def analyze(candles):
    highs = [c[2] for c in candles]
    lows = [c[3] for c in candles]

    max_h = max(highs)
    min_l = min(lows)
    rng = max_h - min_l
    mid = (max_h + min_l) / 2
    last_close = candles[-1][4]

    # проверка на "накопление"
    if rng / last_close < accum_threshold:
        buy_level = max_h * 1.001  # чуть выше
        sell_level = min_l * 0.999 # чуть ниже
        target_up = buy_level + rng * take_profit_mult
        target_down = sell_level - rng * take_profit_mult

        return {
            "accumulation": True,
            "range": rng,
            "buy": buy_level,
            "sell": sell_level,
            "target_up": target_up,
            "target_down": target_down,
            "last_close": last_close
        }
    return {"accumulation": False}

def main():
    while True:
        try:
            candles = fetch_candles()
            signal = analyze(candles)

            if signal["accumulation"]:
                print(f"\n📊 Найдено накопление по {symbol}")
                print(f"   Диапазон: {signal['range']:.2f} USDT")
                print(f"   Цена сейчас: {signal['last_close']:.2f}")
                print(f"   👉 Buy Stop: {signal['buy']:.2f}, цель: {signal['target_up']:.2f}")
                print(f"   👉 Sell Stop: {signal['sell']:.2f}, цель: {signal['target_down']:.2f}")
            else:
                print(f"{symbol}: нет накопления...")

        except Exception as e:
            print("Ошибка:", e)

        time.sleep(sleep_time)

if __name__ == "__main__":
    main()

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

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

📌«Spread Pulse Monitor» - (мониторинг пульса спреда в реальном времени) 🧑‍💻

📌 Идея: 🧨

Спред (разница между bid и ask) — живой индикатор ликвидности и настроения рынка.
Чем спред:

уже → выше ликвидность, часто перед сильным движением,

шире → паника/низкая ликвидность, высокий риск.

Инструмент отслеживает «дыхание» спреда и выводит сигналы.

🔑 Функции: 🛠️

1. Подключается к Binance (через ccxt).

2. Каждую секунду берёт order_book по выбранной паре.

3. Считает:

текущий спред в %

средний спред за последние N минут

«пульс» (расширяется или сужается).

4. Печатает сигналы:

⚡ «Спред резко сжался — возможен импульс!»

🚨 «Спред расширился — осторожно, низкая ликвидность».

import ccxt
import time
from collections import deque

symbol = "BTC/USDT"
exchange = ccxt.binance()
window = 20  # усреднение по N точкам
spread_history = deque(maxlen=window)

def get_spread():
    order_book = exchange.fetch_order_book(symbol, limit=5)
    bid = order_book['bids'][0][0]
    ask = order_book['asks'][0][0]
    spread = (ask - bid) / bid
    return spread, bid, ask

while True:
    try:
        spread, bid, ask = get_spread()
        spread_history.append(spread)
        avg_spread = sum(spread_history) / len(spread_history)

        print(f"{symbol} | Bid: {bid:.2f} Ask: {ask:.2f} | Спред: {spread*100:.3f}% (средн: {avg_spread*100:.3f}%)")

        if spread < avg_spread * 0.5:
            print("⚡ Спред резко сжался — возможен импульс!")
        elif spread > avg_spread * 2:
            print("🚨 Спред расширился — осторожно, низкая ликвидность!")
        
        time.sleep(1)
    except Exception as e:
        print("Ошибка:", e)
        time.sleep(5)

#инструмент

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

📌«Перехват агрессора»🧑‍💻

📌 Идея: 🧨

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

Мы ловим момент, когда агрессор «перегнал цену», и зарабатываем на откате.

🔑 Правила: ❗

Условие входа (шорт):

1. Находим серию из 3+ свечей подряд в одну сторону.

2. Суммарное движение ≥ 0.8–1.2% за короткое время (1–5m).

3. После серии ждём маленькую свечу или паузу.

4. Входим в противоположную сторону (шорт).

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

Зеркальные правила (сильное падение, затем откат вверх).

Выход:💸

Take Profit = 0.5–0.7 от движения агрессора.

Stop Loss = за экстремумом серии.

import ccxt
import time

# === Настройки ===
symbol = "BTC/USDT"
timeframe = "1m"
lookback = 20              # сколько свечей загружаем
min_series = 3             # минимальное число свечей подряд
impulse_threshold = 0.01   # суммарное движение (1%)
sleep_time = 30            # задержка проверки (сек)

exchange = ccxt.binance()

def fetch_candles():
    return exchange.fetch_ohlcv(symbol, timeframe, limit=lookback)

def detect_aggressor(candles):
    closes = [c[4] for c in candles]
    opens = [c[1] for c in candles]

    # определяем направление каждой свечи
    directions = [1 if c > o else -1 for o, c in zip(opens, closes)]
    last_dir = directions[-1]

    # считаем подряд идущие свечи
    streak = 1
    for d in directions[-2::-1]:
        if d == last_dir:
            streak += 1
        else:
            break

    if streak >= min_series:
        total_move = (closes[-1] - opens[-streak]) / opens[-streak]
        if abs(total_move) >= impulse_threshold:
            return {
                "direction": "long" if last_dir > 0 else "short",
                "streak": streak,
                "move_pct": total_move * 100,
                "entry": closes[-1],
                "tp": closes[-1] - (total_move / 2) * closes[-1] if last_dir > 0 else closes[-1] + (abs(total_move) / 2) * closes[-1],
                "sl": max(closes[-streak:]) if last_dir > 0 else min(closes[-streak:])
            }
    return None

def main():
    while True:
        try:
            candles = fetch_candles()
            signal = detect_aggressor(candles)
            if signal:
                print("\n🚀 Найден паттерн 'Перехват агрессора'")
                print(f"Направление агрессора: {signal['direction']}")
                print(f"Серия свечей: {signal['streak']} | Движение: {signal['move_pct']:.2f}%")
                print(f"👉 Вход против тренда: {signal['entry']:.2f}")
                print(f"🎯 TP: {signal['tp']:.2f} | 🛑 SL: {signal['sl']:.2f}")
            else:
                print(f"{symbol}: агрессор не найден...")
        except Exception as e:
            print("Ошибка:", e)

        time.sleep(sleep_time)

if __name__ == "__main__":
    main()

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

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

📌"Order Book Velocity" - (скорость изменения стакана) 🧑‍💻

📌 Суть: 🧨

Большинство смотрит на сам стакан (где стоят лимитки).
Но гораздо важнее — как быстро меняется глубина и структура заявок.

🔑 Как работает: 🛠️

1. Подключается к Binance через WebSocket и берёт стакан в реальном времени.

2. Каждую секунду считает:

насколько вырос/уменьшился суммарный объём на стороне BID (покупатели),

насколько изменился объём на стороне ASK (продавцы).

3. Строит метрику

Если отрицательная → продавцы давят сильнее.

import asyncio
import json
import websockets
from collections import defaultdict

# Пара для анализа
SYMBOL = "btcusdt"
DEPTH_LEVEL = 10  # глубина стакана для анализа
INTERVAL = 1      # интервал в секундах для подсчёта velocity

# Храним прошлые значения объёмов
previous = {"bids": 0, "asks": 0}

async def orderbook_velocity():
    url = f"wss://stream.binance.com:9443/ws/{SYMBOL}@depth@100ms"
    async with websockets.connect(url) as ws:
        print(f"✅ Connected to Binance Order Book for {SYMBOL.upper()}")
        buffer_bids = defaultdict(float)
        buffer_asks = defaultdict(float)

        while True:
            msg = await ws.recv()
            data = json.loads(msg)

            # Берем только DEPTH_LEVEL уровней
            bids = data.get("b", [])[:DEPTH_LEVEL]
            asks = data.get("a", [])[:DEPTH_LEVEL]

            # Сумма объёмов
            total_bids = sum(float(b[1]) for b in bids)
            total_asks = sum(float(a[1]) for a in asks)

            # Разница с предыдущим состоянием
            delta_bids = total_bids - previous["bids"]
            delta_asks = total_asks - previous["asks"]

            velocity = delta_bids - delta_asks

            print(f"📊 Velocity: {velocity:.4f} | ΔBids: {delta_bids:.2f} | ΔAsks: {delta_asks:.2f}")

            # Обновляем
            previous["bids"] = total_bids
            previous["asks"] = total_asks

            await asyncio.sleep(INTERVAL)

if __name__ == "__main__":
    try:
        asyncio.run(orderbook_velocity())
    except KeyboardInterrupt:
        print("⛔ Stopped by user")

#инструмент

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

📌"Liquidity Magnet"- (магнит ликвидности — определяет, где “” цена) 🧑‍💻

💡 Идея: 🧨

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

происходят частые касания свечей,

но цена не может уйти далеко.

Такие зоны — это “магниты” — туда рынок тянет ликвидность и часто пробивает их импульсом.

🧠 Принцип: 🛠️

1. Для каждой пары получаем последние N свечей (например, 200 × 1m).

2. Разбиваем диапазон цен на сетку (например, шаг 0.1%).

3. Подсчитываем, сколько свечей “касались” каждого уровня.

4. Если уровень набирает аномально много касаний → отмечаем как магнит.

import ccxt
import numpy as np

exchange = ccxt.binance()
symbol = 'BTC/USDT'
candles = exchange.fetch_ohlcv(symbol, timeframe='1m', limit=200)

prices = [c[4] for c in candles]
highs = [c[2] for c in candles]
lows = [c[3] for c in candles]

min_price, max_price = min(lows), max(highs)
bins = np.linspace(min_price, max_price, 100)

touches = np.zeros(len(bins))
for h, l in zip(highs, lows):
    for i, level in enumerate(bins):
        if l <= level <= h:
            touches[i] += 1

threshold = np.percentile(touches, 95)
magnets = [(bins[i], int(t)) for i, t in enumerate(touches) if t >= threshold]

print("🧲 Liquidity Magnets Detected:")
for level, count in magnets:
    print(f"→ Level: {level:.2f} | touches: {count}")

#инструмент

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

📌"Echo Bounce" (Эхо-отскок)🧑‍💻

💡 Идея: 🧨

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

Мы ищем локальные импульсы, где:

- объём резко растёт,

- цена почти не двигается,

- затем появляется быстрый откат.

Это отражает борьбу крупных ордеров и ликвидности — отличная точка для контртрендовой сделки с короткой целью.

📈 Логика: 🛠️

1. Сканируем сделки (trades) через Binance WebSocket.

2. За последние N секунд считаем:

средний объём сделки,

изменение цены.

3. Если средний объём вырос более чем в 5 раз,
но цена изменилась менее чем на 0.1% → ловим “эхо”.

4. Входим в обратную сторону от всплеска.

import json
import websocket
import threading
from datetime import datetime
from collections import deque

SYMBOL = "btcusdt"
WINDOW = 5  # секунд
VOLUME_MULTIPLIER = 5
PRICE_CHANGE_LIMIT = 0.001  # 0.1%

trades = deque(maxlen=500)

def on_message(ws, message):
    global trades
    data = json.loads(message)
    t = data['T'] / 1000
    price = float(data['p'])
    qty = float(data['q'])
    side = 'buy' if data['m'] == False else 'sell'
    trades.append((t, price, qty, side))
    
def detect_echo():
    while True:
        if len(trades) < 10:
            continue
        recent = [t for t in trades if t[0] > trades[-1][0] - WINDOW]
        if len(recent) < 3:
            continue

        prices = [x[1] for x in recent]
        vols = [x[2] for x in recent]
        avg_vol = sum(vols) / len(vols)
        avg_recent_vol = sum([x[2] for x in list(trades)[-100:]]) / 100
        price_change = abs(prices[-1] - prices[0]) / prices[0]

        if avg_vol > avg_recent_vol * VOLUME_MULTIPLIER and price_change < PRICE_CHANGE_LIMIT:
            dominant_side = max(set([x[3] for x in recent]), key=[x[3] for x in recent].count)
            signal = "SELL" if dominant_side == "buy" else "BUY"
            print(f"[{datetime.now().strftime('%H:%M:%S')}] ECHO DETECTED → {signal} | Δprice={price_change:.4f} | vol×{avg_vol/avg_recent_vol:.1f}")

def run_ws():
    ws = websocket.WebSocketApp(f"wss://stream.binance.com:9443/ws/{SYMBOL}@trade", on_message=on_message)
    ws.run_forever()

threading.Thread(target=run_ws).start()
detect_echo()

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

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

📌“Market Memory Map” - (анализатор рыночной памяти — где цена «помнит» уровни ликвидности) 🧑‍💻

💡 Идея: 🧨

Рынок часто возвращается к ценам, где происходили сильные обороты — это и есть “память рынка”.
Инструмент создаёт тепловую карту “памяти” — показывает, на каких ценовых уровнях цена останавливалась, ускорялась или меняла волатильность.

Это не индикатор, не сигнал — это инструмент для осознанной визуализации структуры рынка.

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

1. Загружает исторические свечи (klines) за заданный период.

2. Для каждого уровня цены (например, шаг 0.2%) считает:

сколько раз цена в него “заходила”,

сколько раз от него резко отскакивала.

3. Строит тепловую матрицу уровней по количеству реакций.

4. Показывает ключевые “зоны памяти” — где рынок чаще всего реагировал.


🔍 Интерпретация: 💸

🔥 Высокие пики → цена часто меняла направление → “зоны памяти” рынка.

⚙️ Плоские зоны → нейтральные участки без реакции.

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

symbol = "BTC/USDT"
tf = "1h"
limit = 1000
exchange = ccxt.binance()

# Загружаем свечи
data = exchange.fetch_ohlcv(symbol, tf, limit=limit)
df = pd.DataFrame(data, columns=["time", "open", "high", "low", "close", "volume"])
df["time"] = pd.to_datetime(df["time"], unit="ms")

# Готовим сетку уровней
min_price, max_price = df["low"].min(), df["high"].max()
bins = np.linspace(min_price, max_price, 200)
reactions = np.zeros(len(bins))

# Считаем реакции на каждом уровне
for _, row in df.iterrows():
    for i, lvl in enumerate(bins):
        # если цена прошла уровень и изменила направление — "реакция"
        if row["low"] <= lvl <= row["high"]:
            if (row["close"] - row["open"]) * (lvl - row["open"]) < 0:
                reactions[i] += 1

# Нормализация
reactions = reactions / reactions.max()

# Визуализация
plt.figure(figsize=(10, 6))
plt.bar(bins, reactions, width=(bins[1]-bins[0]), color="cyan", edgecolor="black")
plt.title(f"Market Memory Map — {symbol} ({tf})")
plt.xlabel("Price Level")
plt.ylabel("Reaction Intensity (normalized)")
plt.grid(True, alpha=0.3)
plt.show()

#инструмент

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

📌"Liquidity Prober" - (пробник ликвидности — активное картирование реальной глубины и refill-реакции стакана) 🧑‍💻

🔑 Идея (коротко)🧨

Вместо того чтобы полагаться на статический снимок стакана, этот инструмент активно (и осторожно) „прощупывает“ рынок небольшими рыночными/лимитными пробами, фиксирует реальный проскальзывание, скорость восстановления (refill) и поведение цены. По результатам даёт рекомендацию по стратегии исполнения (сколько слайсов, паузы между ними, безопасная доля от баланса).

Полезно для: исполнение больших заявок, алгоритмической дробки, оценки риска слippage перед «прыжком».

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

делает N маленьких проб (probe) в одну сторону (buy или sell) — в dry-run можно только эмулировать; при live — реально отправляет маленькие market/limit ордера;

для каждого пробы фиксирует:

среднюю цену исполнения, проскальзывание vs mid, изменение best bid/ask, и время восстановления объёма в локальной зоне;

- строит статистику: mean/std проскальзывания, среднее время refill;

- на основе результатов выдаёт рекомендацию: slice_count, slice_delay_sec, max_notional_fraction_per_slice.


#инструмент

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

📌“Liquidity Loop Scalper” - (циклический скальпер по ликвидности без индикаторов) 🧑‍💻

💡 Идея: 🧨

Большинство пар Binance (особенно спотовые альткоины) имеют повторяющиеся микродвижения внутри лимитной книги: когда ликвидность на одной стороне (bid/ask) резко исчезает, цена почти всегда делает короткий импульс в эту сторону — затем возвращается.

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

📈 Логика: 🛠️

1. Каждую секунду получаем книгу ордеров (fetch_order_book(symbol, 50)).

2. Считаем суммарный объём первых N уровней (например 10).

bid_sum = сумма объёмов на первых 10 уровнях покупки.

ask_sum = сумма объёмов на первых 10 уровнях продажи.

3. Сравниваем текущее состояние с предыдущим:

Если bid_sum ↓ более чем на 40% за <2 секунды, но цена не упала — значит “ликвидность сняли, но не продали”, то есть ложный пробой вниз.
→ Входим в лонг.

Если ask_sum ↓ более чем на 40%, но цена не выросла — ложный пробой вверх.
→ Входим в шорт (или продаём, если спот).

4. Выход — при возвращении ликвидности (bid_sum или ask_sum восстанавливается до 80% прежнего уровня) или по прибыли >0.3%.

import ccxt
import time

exchange = ccxt.binance({'enableRateLimit': True})
symbol = 'DOGE/USDT'

N = 10
DROP_THRESHOLD = 0.4
PROFIT_TARGET = 0.003

last_bid_sum = None
last_ask_sum = None
entry_price = None
position = None

print("Liquidity Loop Scalper running...")

while True:
    ob = exchange.fetch_order_book(symbol, limit=50)
    bid_sum = sum([q for _, q in ob['bids'][:N]])
    ask_sum = sum([q for _, q in ob['asks'][:N]])
    mid = (ob['bids'][0][0] + ob['asks'][0][0]) / 2

    if last_bid_sum and last_ask_sum:
        bid_change = (bid_sum - last_bid_sum) / last_bid_sum
        ask_change = (ask_sum - last_ask_sum) / last_ask_sum

        # вход в лонг при исчезновении бидов без падения цены
        if bid_change < -DROP_THRESHOLD and position is None:
            print(f"[{symbol}] ⚡ Лонг — исчезли биды, но цена держится ({bid_change:.1%})")
            entry_price = mid
            position = "long"

        # вход в шорт при исчезновении асков без роста цены
        elif ask_change < -DROP_THRESHOLD and position is None:
            print(f"[{symbol}] ⚡ Продажа — исчезли аски, но цена не растёт ({ask_change:.1%})")
            entry_price = mid
            position = "short"

        # выход из позиции
        if position == "long" and (mid - entry_price) / entry_price > PROFIT_TARGET:
            print(f"[{symbol}] ✅ Выход из лонга: прибыль {((mid-entry_price)/entry_price):.2%}")
            position = None
        elif position == "short" and (entry_price - mid) / entry_price > PROFIT_TARGET:
            print(f"[{symbol}] ✅ Выход из шорта: прибыль {((entry_price-mid)/entry_price):.2%}")
            position = None

    last_bid_sum, last_ask_sum = bid_sum, ask_sum
    time.sleep(1)

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

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

📌"Liquidity Vacuum"- (Пылесос ликвидности) 🧑‍💻

💡 Идея🧨

Цена не движется сама по себе — она падает туда, где пусто в стакане, и растёт туда, где продавцы исчезли.
Когда крупные лимитки внезапно исчезают (cancel walls) — рынок обычно движется в сторону исчезновения ликвидности, потому что это «дырка», куда притягиваются рыночные ордера.

Мы не угадываем направление — мы следим за исчезновением плотности в стакане и входим туда, куда исчезла поддержка (или сопротивление).

🧩 Логика:🛠️

1. Раз в секунду получаем стакан.

2. Считаем локальную плотность (сумма объёмов) в радиусе ±0.2% от mid-price.

3. Если в верхней половине (asks) ликвидность внезапно упала > X% — ждём всплеск рыночных покупок → входим в лонг.

4. Если в нижней половине (bids) ликвидность внезапно исчезла — ждём падения → входим в шорт.

5. Выход — при обратном восстановлении плотности или при изменении цены на Y%.

import ccxt
import time
import statistics

SYMBOL = 'BTC/USDT'
EXCHANGE = ccxt.binance({'enableRateLimit': True})
INTERVAL = 1.0
THRESHOLD_DROP = 0.35   # ликвидность упала на 35%
PRICE_MOVE_TP = 0.002   # 0.2% take profit
PRICE_MOVE_SL = 0.0015  # 0.15% stop loss

def get_book():
    ob = EXCHANGE.fetch_order_book(SYMBOL, limit=50)
    bids = ob['bids']
    asks = ob['asks']
    mid = (bids[0][0] + asks[0][0]) / 2
    return bids, asks, mid

def local_density(bids, asks, mid, pct=0.002):
    low = mid * (1 - pct)
    high = mid * (1 + pct)
    bid_vol = sum(v for p, v in bids if p >= low)
    ask_vol = sum(v for p, v in asks if p <= high)
    return bid_vol, ask_vol

history = []

print("Liquidity Vacuum strategy started...\n")

pos = None
entry_price = 0

while True:
    bids, asks, mid = get_book()
    bid_vol, ask_vol = local_density(bids, asks, mid)
    history.append((bid_vol, ask_vol, mid))
    if len(history) < 3:
        time.sleep(INTERVAL)
        continue

    prev_bid, prev_ask, _ = history[-2]
    drop_bids = (prev_bid - bid_vol) / prev_bid if prev_bid else 0
    drop_asks = (prev_ask - ask_vol) / prev_ask if prev_ask else 0

    # ---- Вход ----
    if not pos:
        if drop_asks > THRESHOLD_DROP:
            pos = 'LONG'
            entry_price = mid
            print(f"💚 LONG entry {mid:.2f} (asks collapsed {drop_asks*100:.1f}%)")
        elif drop_bids > THRESHOLD_DROP:
            pos = 'SHORT'
            entry_price = mid
            print(f"❤️ SHORT entry {mid:.2f} (bids collapsed {drop_bids*100:.1f}%)")

    # ---- Выход ----
    elif pos == 'LONG':
        if mid >= entry_price * (1 + PRICE_MOVE_TP):
            print(f"✅ LONG TP at {mid:.2f}")
            pos = None
        elif mid <= entry_price * (1 - PRICE_MOVE_SL):
            print(f"❌ LONG SL at {mid:.2f}")
            pos = None
    elif pos == 'SHORT':
        if mid <= entry_price * (1 - PRICE_MOVE_TP):
            print(f"✅ SHORT TP at {mid:.2f}")
            pos = None
        elif mid >= entry_price * (1 + PRICE_MOVE_SL):
            print(f"❌ SHORT SL at {mid:.2f}")
            pos = None

    time.sleep(INTERVAL)

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

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

📌“Micro-Impulse Trap” — Ловушка микроимпульсов🧑‍💻

💡 Идея: 🧨

На микросекундных движениях часто наблюдается закономерность:
после короткого сильного импульса, за которым следует резкая пауза без продолжения, рынок возвращается к точке старта импульса — это реакция алгоритмов, которые не нашли ликвидности для продолжения.

Мы отслеживаем:

быстрые скачки цены (например, >0.15% за секунду);

отсутствие обновления high/low в следующие 2 секунды;

и заходим в противоположную сторону на возврат.

🧩 Алгоритм: 🛠️

1. Каждую секунду фиксируем текущую цену.

2. Если за 1 секунду цена выросла >0.15% — запоминаем уровень импульса.

3. Через 2 секунды:

если цена не выше импульсного high → вход в шорт (ожидание возврата).

4. Если цена упала >0.15%:

через 2 секунды, если не ниже low → вход в лонг.

5. Цель — возврат на 0.1–0.2% к средней.

import ccxt
import time
from collections import deque

exchange = ccxt.binance({'enableRateLimit': True})
symbol = 'ETH/USDT'
window = deque(maxlen=3)
position = None
entry = 0

THR_IMPULSE = 0.0015  # 0.15%
TP = 0.001            # 0.1%
SL = 0.0015           # 0.15%

print("Micro-Impulse Trap started...\n")

while True:
    price = exchange.fetch_ticker(symbol)['last']
    window.append(price)

    if len(window) == 3:
        p_now = window[-1]
        p_prev = window[-2]
        p_old = window[0]

        impulse = (p_prev - p_old) / p_old

        # восходящий импульс без продолжения
        if impulse > THR_IMPULSE and position is None:
            time.sleep(2)
            new_price = exchange.fetch_ticker(symbol)['last']
            if new_price <= p_prev:  # нет обновления high
                position = 'short'
                entry = new_price
                print(f"📉 SHORT {symbol} at {entry:.3f} — импульс вверх погашен")

        # нисходящий импульс без продолжения
        elif impulse < -THR_IMPULSE and position is None:
            time.sleep(2)
            new_price = exchange.fetch_ticker(symbol)['last']
            if new_price >= p_prev:  # нет обновления low
                position = 'long'
                entry = new_price
                print(f"📈 LONG {symbol} at {entry:.3f} — импульс вниз погашен")

        # выход
        if position == 'long':
            if price >= entry * (1 + TP):
                print(f"✅ TP LONG {symbol} at {price:.3f}")
                position = None
            elif price <= entry * (1 - SL):
                print(f"❌ SL LONG {symbol} at {price:.3f}")
                position = None

        elif position == 'short':
            if price <= entry * (1 - TP):
                print(f"✅ TP SHORT {symbol} at {price:.3f}")
                position = None
            elif price >= entry * (1 + SL):
                print(f"❌ SL SHORT {symbol} at {price:.3f}")
                position = None

    time.sleep(1)

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

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