📌"Пружина"🧑‍💻

🔍 Идея: 🚀

Цена на рынке часто зажимается в узкий диапазон («сжимается как пружина»), после чего следует резкий выход.

👉 Мы ловим момент сжатия и торгуем пробой в сторону выхода.

📌 Логика: 🧨

1. Считаем диапазон свечей = High – Low.

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

3. Когда цена выходит за границу диапазона этих свечей → открываем позицию в сторону пробоя.

4. Стоп — за противоположной границей диапазона, тейк — в 2–3 раза больше.

import ccxt
import pandas as pd
import numpy as np

SYMBOL = "BTC/USDT"
TF = "5m"
LOOKBACK = 200

ex = ccxt.binance()

def fetch_data():
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=["ts","o","h","l","c","v"])
    return df

def spring_strategy(df, window=20, compress_bars=5, thr=0.5):
    df["range"] = df["h"] - df["l"]
    avg_range = df["range"].rolling(window).mean()
    signals = []

    for i in range(window+compress_bars, len(df)):
        recent = df.iloc[i-compress_bars:i]
        if (recent["range"] < avg_range.iloc[i]*thr).all():
            box_high = recent["h"].max()
            box_low = recent["l"].min()
            price = df["c"].iloc[i]
            ts = df["ts"].iloc[i]

            if price > box_high:
                signals.append((ts, "BUY", price))
            elif price < box_low:
                signals.append((ts, "SELL", price))

    return signals

if __name__ == "__main__":
    df = fetch_data()
    sigs = spring_strategy(df)
    for t, side, price in sigs[-5:]:
        print(f"{pd.to_datetime(t, unit='ms')} → {side} @ {price}")

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

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

📌"Давка ликвидности"🧑‍💻

🔍 Идея: 🧨

Когда свеча делает длинный хвост (тенень), это часто сигнал, что рынок собрал стопы и вернулся обратно.

👉 Мы входим в сторону противоположную хвосту, так как там уже снята ликвидность.

📌 Логика: 💸

1. Определяем свечи с длинным хвостом (тело < 30% от всей свечи, тень > 60%).

2. Если длинная нижняя тень → сигнал на покупку.

3. Если длинная верхняя тень → сигнал на продажу.

4. Стоп за тенью, тейк = 2× стоп.

import ccxt
import pandas as pd

SYMBOL = "BTC/USDT"
TF = "5m"
LOOKBACK = 300

ex = ccxt.binance()

def fetch_data():
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=["ts","o","h","l","c","v"])
    return df

def liquidity_squeeze(df, body_thr=0.3, wick_thr=0.6):
    signals = []
    for i in range(len(df)):
        o, h, l, c = df.iloc[i][["o","h","l","c"]]
        body = abs(c - o)
        rng = h - l
        upper = h - max(o, c)
        lower = min(o, c) - l

        if rng == 0: 
            continue

        body_ratio = body / rng
        upper_ratio = upper / rng
        lower_ratio = lower / rng

        # длинная нижняя тень
        if body_ratio < body_thr and lower_ratio > wick_thr:
            signals.append((df["ts"].iloc[i], "BUY", c))
        # длинная верхняя тень
        elif body_ratio < body_thr and upper_ratio > wick_thr:
            signals.append((df["ts"].iloc[i], "SELL", c))

    return signals

if __name__ == "__main__":
    df = fetch_data()
    sigs = liquidity_squeeze(df)
    for t, side, price in sigs[-5:]:
        print(f"{pd.to_datetime(t, unit='ms')} → {side} @ {price}")

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

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

📌"Асинхронный компас рынков"🧑‍💻

📌 Идея: 🧨

Часто одни монеты начинают движение раньше других (например, ETH идёт за BTC, а альты идут за ETH).

👉 Этот инструмент ищет лидера и ведомых: какая пара двигается первой, а какие повторяют её с лагом.

🎯 Польза: 💸

- Показывает, кто ведёт рынок, а кто только догоняет.

- Можно заранее входить в альты, если «лидер» уже дал сигнал.

- Подходит для построения парных и межрыночных стратегий.

import ccxt
import pandas as pd
import numpy as np

ex = ccxt.binance()

SYMBOLS = ["BTC/USDT", "ETH/USDT", "SOL/USDT", "BNB/USDT"]
TF = "5m"
LOOKBACK = 100

def fetch(symbol):
    ohlcv = ex.fetch_ohlcv(symbol, TF, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=["ts","o","h","l","c","v"])
    df["ret"] = df["c"].pct_change()
    return df[["ts","ret"]]

def market_compass(symbols):
    data = {s: fetch(s)["ret"].dropna().values for s in symbols}
    matrix = pd.DataFrame(index=symbols, columns=symbols)

    for s1 in symbols:
        for s2 in symbols:
            if s1 == s2: 
                matrix.loc[s1,s2] = 1.0
            else:
                corr = np.corrcoef(data[s1], data[s2])[0,1]
                matrix.loc[s1,s2] = round(corr,3)

    return matrix

if __name__ == "__main__":
    compass = market_compass(SYMBOLS)
    print("📊 Корреляции движений (лидеры vs ведомые):")
    print(compass)

#инструмент

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

📌«Охота на толпу»🧑‍💻

🔑 Идея: 💸

Большинство трейдеров заходит слишком поздно — после резкого движения.

👉 Эта стратегия ищет точку, где толпа уже в позиции, и работает против неё, если рынок показывает слабость.

📌 Логика: 🧨

1. Отслеживаем резкий импульс вверх/вниз (например, движение >2% за ≤3 минуты).

2. Смотрим, появилась ли стабильность объёма после импульса (толпа входит).

3. Если цена не продолжает движение (консолидация), а начинает давать обратные свечи,
→ входим против направления импульса.

4. Выход — при возврате хотя бы на 30–50% от движения импульса.

import ccxt
import pandas as pd

ex = ccxt.binance()

PAIR = "BTC/USDT"
TF = "1m"
LOOKBACK = 200

def fetch_data():
    ohlcv = ex.fetch_ohlcv(PAIR, TF, limit=LOOKBACK)
    df = pd.DataFrame(ohlcv, columns=["ts","o","h","l","c","v"])
    df["ret"] = df["c"].pct_change()
    return df

def detect_crowd_trap(df):
    # шаг 1: ищем резкий импульс
    df["impulse"] = abs(df["ret"]) > 0.02  # 2% за свечу
    signals = []
    
    for i in range(2, len(df)):
        if df["impulse"].iloc[i]:
            direction = "long" if df["ret"].iloc[i] > 0 else "short"
            # шаг 2: проверяем слабость
            if (df["c"].iloc[i] < df["c"].iloc[i-1] and direction=="long") or \
               (df["c"].iloc[i] > df["c"].iloc[i-1] and direction=="short"):
                signals.append((df["ts"].iloc[i], "reverse_" + direction))
    
    return signals

if __name__ == "__main__":
    df = fetch_data()
    sigs = detect_crowd_trap(df)
    for s in sigs:
        print("📌 Сигнал:", s)

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

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

📌"Market Balance Radar" - (радар баланса покупателей и продавцов в реальном времени)🧑‍💻

🔑 Идея: 🧨

Большинство трейдеров смотрят только цену или индикаторы.

👉 Но цена — это результат борьбы агрессивных покупателей (market buy) и агрессивных продавцов (market sell).

Инструмент в реальном времени показывает:

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

- насколько «перекошен» рынок,

- сигналит, когда дисбаланс достигает критического уровня (часто → разворот или импульс).

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

1. Берём поток сделок (trades) с биржи.

2. Считаем отдельно объём buy и sell.

3. Строим баланс: imbalance = buy_volume / (buy_volume + sell_volume).

4. Если imbalance > 0.7 → рынок перегрет покупателями (часто шорт-сетап).

5. Если imbalance < 0.3 → рынок перегрет продавцами (часто лонг-сетап).

import websocket, json

PAIR = "btcusdt"
URL = f"wss://stream.binance.com:9443/ws/{PAIR}@trade"

buy_volume = 0
sell_volume = 0

def on_message(ws, msg):
    global buy_volume, sell_volume
    data = json.loads(msg)
    price = float(data["p"])
    qty = float(data["q"])
    is_buyer_maker = data["m"]

    # если maker — продавец, значит сделка buy
    if not is_buyer_maker:
        buy_volume += qty
    else:
        sell_volume += qty

    total = buy_volume + sell_volume
    if total > 0:
        imbalance = buy_volume / total
        print(f"Цена: {price:.2f} | Buy: {buy_volume:.2f} | Sell: {sell_volume:.2f} | Баланс: {imbalance:.2f}")
        
        # сигналы
        if imbalance > 0.7:
            print("⚠️ Перегрев покупателями — возможный шорт!")
        elif imbalance < 0.3:
            print("⚠️ Перегрев продавцами — возможный лонг!")

def on_open(ws):
    print("✅ Market Balance Radar запущен")

ws = websocket.WebSocketApp(URL, on_message=on_message, on_open=on_open)
ws.run_forever()

#инструмент

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

📌"Time Pressure Index" (Индекс временного давления) 🧑‍💻

🔑 Идея: 🧨

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

👉 Индекс показывает давление времени: насколько быстро меняются сделки и свечи.

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

1. Считаем количество сделок за фиксированный промежуток (например, 10 секунд).

2. Нормализуем относительно среднего значения.

3. Если значение резко ↑ (ускорение потока сделок) → вероятен импульс.

4. Если значение резко ↓ (замедление при высоком объёме) → рынок готов к развороту.

import websocket, json, time
from collections import deque

PAIR = "btcusdt"
URL = f"wss://stream.binance.com:9443/ws/{PAIR}@trade"

window = deque(maxlen=10)  # последние 10 секунд
last_ts = int(time.time())

def on_message(ws, msg):
    global last_ts
    data = json.loads(msg)
    ts = int(time.time())

    # записываем количество сделок в секунду
    if ts != last_ts:
        window.append(0)
        last_ts = ts
    if len(window) == 0:
        window.append(0)
    window[-1] += 1

    if len(window) == window.maxlen:
        avg = sum(window) / len(window)
        current = window[-1]
        tpi = current / avg if avg > 0 else 1
        print(f"TPI: {tpi:.2f} (сделок в сек: {current}, среднее: {avg:.1f})")

        if tpi > 2:
            print("⚡ Всплеск активности! Возможный импульс.")
        elif tpi < 0.5:
            print("⏳ Замедление. Возможна разворотная точка.")

def on_open(ws):
    print("✅ Time Pressure Index запущен")

ws = websocket.WebSocketApp(URL, on_message=on_message, on_open=on_open)
ws.run_forever()

#инструмент

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

📌«Импульс–Усталость»🧑‍💻

🔑 Идея: 🧨

Каждый импульс рано или поздно теряет силу.

👉 Мы ловим момент, когда рынок «выдыхается» — объём ещё высокий, а движение цены замедляется. Это часто предвещает разворот или глубокий откат.

📌 Логика входа: 💸

1. Находим импульс: свеча > X% от среднего размера последних N свечей.

2. Проверяем, что за импульсом идут 2–3 свечи:

объём остаётся высоким,

но каждая новая свеча делает всё меньший прогресс (разница high-low и close-close уменьшается).

3. Если цена замедляется при высоком объёме → входим против импульса.

импульс вверх → открываем шорт,

импульс вниз → открываем лонг.

4. Тейк — возврат хотя бы на 30–50% от импульса.

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

import ccxt
import pandas as pd

ex = ccxt.binance()
PAIR = "BTC/USDT"

def fetch_data():
    ohlcv = ex.fetch_ohlcv(PAIR, "5m", limit=200)
    df = pd.DataFrame(ohlcv, columns=["ts","o","h","l","c","v"])
    return df

def detect_fade(df, lookback=20, factor=2):
    df["range"] = df["h"] - df["l"]
    avg_range = df["range"].rolling(lookback).mean()
    signals = []
    
    for i in range(lookback+3, len(df)):
        # импульсная свеча
        if df["range"].iloc[i] > factor * avg_range.iloc[i]:
            direction = "up" if df["c"].iloc[i] > df["o"].iloc[i] else "down"
            # следующие свечи замедляются при высоком объёме
            slowdown = (
                df["range"].iloc[i+1] < df["range"].iloc[i] and
                df["range"].iloc[i+2] < df["range"].iloc[i+1] and
                df["v"].iloc[i+1] > df["v"].rolling(lookback).mean().iloc[i] and
                df["v"].iloc[i+2] > df["v"].rolling(lookback).mean().iloc[i]
            )
            if slowdown:
                sig = ("short" if direction=="up" else "long", df["ts"].iloc[i])
                signals.append(sig)
    return signals

if __name__ == "__main__":
    df = fetch_data()
    signals = detect_fade(df)
    for s in signals:
        print("📌 Сигнал:", s)

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

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

📌"Order Flow Velocity" - (скорость потока ордеров) 🧑‍💻

🔑 Идея: 🧨

Большинство трейдеров смотрят на объём или цену.

👉 Но скорость, с которой льются рыночные ордера, даёт ещё более точную картину:

когда поток резко ускоряется → начинается импульс,

когда поток замедляется при росте объёма → вероятен разворот.

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

1. Подключаемся к WebSocket биржи.

2. Считаем количество сделок в единицу времени (например, за 1 секунду).

3. Вычисляем «ускорение» потока (разница между текущей и средней скоростью).

4. Генерируем сигналы:

⚡ всплеск скорости → начало импульса,

⏳ замедление при больших объёмах → рынок выдыхается.

import websocket, json, time
from collections import deque

PAIR = "btcusdt"
URL = f"wss://stream.binance.com:9443/ws/{PAIR}@trade"

trades_per_sec = deque(maxlen=20)  # последние 20 секунд
last_sec = int(time.time())

def on_message(ws, msg):
    global last_sec
    ts = int(time.time())
    if ts != last_sec:
        trades_per_sec.append(0)
        last_sec = ts
    if len(trades_per_sec) == 0:
        trades_per_sec.append(0)
    trades_per_sec[-1] += 1

    if len(trades_per_sec) == trades_per_sec.maxlen:
        avg = sum(trades_per_sec) / len(trades_per_sec)
        cur = trades_per_sec[-1]
        velocity = cur / avg if avg > 0 else 1
        print(f"⏱️ Скорость сделок: {cur}/с | Среднее: {avg:.1f} | Индекс: {velocity:.2f}")
        
        if velocity > 2:
            print("⚡ Импульсный всплеск — возможный тренд!")
        elif velocity < 0.5 and cur > avg*0.8:
            print("⏳ Замедление при объёме — возможный разворот.")

def on_open(ws):
    print("✅ Order Flow Velocity запущен")

ws = websocket.WebSocketApp(URL, on_message=on_message, on_open=on_open)
ws.run_forever()

#инструмент

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

📌"Трёхходовка"🧑‍💻

🔑 Идея: 🧨

Очень часто рынок перед реальным движением делает ложные шаги:

1. Импульс в одну сторону (всех заманили).

2. Быстрый разворот в обратку (сняли стопы первых).

3. Настоящее движение начинается в том же направлении, что и первый импульс.

👉 Мы входим на третьем ходе, когда толпа уже выбита.

📌 Логика входа: 💸

1. Отслеживаем свечу-импульс (X% от среднего диапазона).

2. Если следующая свеча перекрывает импульс более чем на 50% → это «второй ход» (снятие стопов).

3. Если после этого цена снова начинает двигаться в сторону первого импульса → входим.

4. Стоп — за минимум/максимум второго хода.

5. Тейк — диапазон первого импульса или больше.

import ccxt
import pandas as pd

ex = ccxt.binance()
PAIR = "BTC/USDT"

def fetch_data():
    ohlcv = ex.fetch_ohlcv(PAIR, "5m", limit=200)
    df = pd.DataFrame(ohlcv, columns=["ts","o","h","l","c","v"])
    df["range"] = df["h"] - df["l"]
    return df

def detect_three_moves(df, lookback=20, factor=1.5):
    avg_range = df["range"].rolling(lookback).mean()
    signals = []
    for i in range(lookback+3, len(df)-2):
        # шаг 1: импульс
        if df["range"].iloc[i] > factor * avg_range.iloc[i]:
            direction = "up" if df["c"].iloc[i] > df["o"].iloc[i] else "down"
            
            # шаг 2: обратка
            if direction == "up" and df["c"].iloc[i+1] < (df["o"].iloc[i] + df["c"].iloc[i])/2:
                reversal = True
            elif direction == "down" and df["c"].iloc[i+1] > (df["o"].iloc[i] + df["c"].iloc[i])/2:
                reversal = True
            else:
                reversal = False

            # шаг 3: подтверждение
            if reversal:
                if direction == "up" and df["c"].iloc[i+2] > df["c"].iloc[i]:
                    signals.append(("long", df["ts"].iloc[i+2]))
                elif direction == "down" and df["c"].iloc[i+2] < df["c"].iloc[i]:
                    signals.append(("short", df["ts"].iloc[i+2]))
    return signals

if __name__ == "__main__":
    df = fetch_data()
    sigs = detect_three_moves(df)
    for s in sigs:
        print("📌 Сигнал:", s)

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

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

📌"Перегрев стакана"🧑‍💻

🔑 Идея: 🧨

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

1. Если стена настоящая → цена отскакивает.

2. Если стена фейковая (spoofing) → цену пробивают, и движение ускоряется.

👉 Мы используем это как торговый сигнал.

📌 Логика входа: 🛠️

1. Подключаемся к ордербуку.

2. Считаем суммарный объём в первых X уровнях стакана.

3. Если на одной стороне (buy/sell) объём резко > среднего (например, ×3), фиксируем «перегрев».

4. Дальше смотрим реакцию цены:

- если цена стоит и отскакивает → торгуем в сторону стены,

- если цена пробивает → торгуем в сторону пробоя.

import websocket, json

PAIR = "btcusdt"
DEPTH_URL = f"wss://stream.binance.com:9443/ws/{PAIR}@depth20@100ms"

def on_message(ws, msg):
    data = json.loads(msg)
    bids = [(float(p), float(q)) for p, q in data["bids"]]
    asks = [(float(p), float(q)) for p, q in data["asks"]]

    total_bids = sum(q for _, q in bids[:5])
    total_asks = sum(q for _, q in asks[:5])

    imbalance = total_bids / (total_bids + total_asks)

    print(f"Buy: {total_bids:.1f} | Sell: {total_asks:.1f} | Баланс: {imbalance:.2f}")

    if total_bids > total_asks * 3:
        print("🟢 Стена покупателей — возможный отскок вверх!")
    elif total_asks > total_bids * 3:
        print("🔴 Стена продавцов — возможный слив вниз!")

def on_open(ws):
    print("✅ Стратегия 'Перегрев стакана' запущена")

ws = websocket.WebSocketApp(DEPTH_URL, on_message=on_message, on_open=on_open)
ws.run_forever()

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

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

📌"Order Replay" - (реплей ордеров в реальном времени) 🧑‍💻

🔑 Идея: 🧨

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

👉 Можно исследовать:

- как цена строила импульс,

- где были «резкие вбросы» объёма,

- как формировался диапазон свечи.

📌 Польза: 💸

- Анализ микроструктуры рынка.

- Поиск аномальных сделок и скрытых манипуляций.

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

import time
import random

# имитация тиков (обычно берём с биржи через websocket)
ticks = [100 + random.uniform(-0.5, 0.5) for _ in range(200)]

def replay(ticks, speed=0.05):
    print("▶ Начинаем реплей ордеров...")
    for i, price in enumerate(ticks):
        print(f"Tick {i+1}: {price:.2f}")
        time.sleep(speed)
    print("⏹ Реплей окончен.")

replay(ticks, speed=0.1)

#инструмент

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

📌"Market Stress Meter" - (измеритель рыночного стресса) 🧑‍💻

🔑 Идея: 🧨

Вместо индикаторов, мы оцениваем напряжённость рынка через поведение цены и объёмов.
Рынок в «стрессе», когда:

свечи становятся длинными,

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

спред (High–Low) увеличивается относительно среднего.

👉 Такой инструмент помогает понять, когда рынок готов к импульсу или коррекции.

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

1. Считает «стресс-индекс» как взвешенную комбинацию:

- отклонение текущего диапазона свечи от среднего,

- рост объёма,

- скорость изменения цены.

2. Выводит предупреждение, если индекс выше порога.

3. Можно использовать как фильтр для стратегий: торговать только при «высоком стрессе» или наоборот — в спокойные периоды


Интерпретация: 🚀

StressIndex > 1.5 → рынок в состоянии «стресса» (готов к импульсу).

StressIndex < 1 → рынок спокоен, можно торговать mean reversion.

import pandas as pd
import ccxt

# === параметры ===
symbol = "BTC/USDT"
timeframe = "15m"
limit = 200

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

df = pd.DataFrame(ohlcv, columns=["ts","Open","High","Low","Close","Volume"])
df["Range"] = df["High"] - df["Low"]

# средние значения
avg_range = df["Range"].rolling(20).mean()
avg_vol = df["Volume"].rolling(20).mean()

# стресс-индекс
df["StressIndex"] = (df["Range"]/avg_range + df["Volume"]/avg_vol) / 2

# сигнал
threshold = 1.5
df["StressSignal"] = df["StressIndex"] > threshold

print(df[["ts","Close","StressIndex","StressSignal"]].tail(20))

#инструмент

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

📌"Auction Range Breakout" - (прорыв аукционного диапазона)🧑‍💻

🔑 Идея: 🧨

Рынок часто ведёт себя как аукцион:

внутри дня формируется основной диапазон справедливой цены (где проходят торги),

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

👉 Мы фиксируем «аукционный диапазон» (fair range) и торгуем пробой.

📌 Логика: 🛠️

1. Берём первые X часов (например, 4ч после открытия дня).

2. Определяем диапазон High–Low за этот период → это наша «аукционная зона».

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

4. Стоп внутри диапазона, тейк = ширина диапазона * 1.5–2.

import ccxt
import pandas as pd
from datetime import datetime, timedelta

# === ПАРАМЕТРЫ ===
symbol = "BTC/USDT"
timeframe = "1h"
lookback_days = 5       # сколько дней анализируем
session_hours = 4       # аукционный диапазон (первые 4 часа)

# === API (можно добавить ключи для реальной торговли) ===
binance = ccxt.binance({
    "enableRateLimit": True
})

# === ЗАГРУЗКА СВЕЧЕЙ ===
since = binance.parse8601((datetime.utcnow() - timedelta(days=lookback_days)).isoformat())
ohlcv = binance.fetch_ohlcv(symbol, timeframe=timeframe, since=since)

df = pd.DataFrame(ohlcv, columns=["timestamp", "Open", "High", "Low", "Close", "Volume"])
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
df.set_index("timestamp", inplace=True)
df["hour"] = df.index.hour

# === ЛОГИКА АУКЦИОННОГО ПРОБОЯ ===
signals = []
for day, df_day in df.groupby(df.index.date):
    session = df_day[df_day["hour"] < session_hours]
    if session.empty:
        continue
    low, high = session["Low"].min(), session["High"].max()

    for i in range(session_hours, len(df_day)):
        row = df_day.iloc[i]
        if row["Close"] > high:
            signals.append((row.name, "LONG", high, (high - low) * 2))
            break
        elif row["Close"] < low:
            signals.append((row.name, "SHORT", low, (high - low) * 2))
            break

# === ВЫВОД СИГНАЛОВ ===
print("📊 Сигналы стратегии Auction Breakout:")
for ts, side, stop, target in signals:
    print(f"{ts} | {side} | Стоп: {stop:.2f} | Тейк: {target:.2f}")

# === ДЛЯ РЕАЛЬНОЙ ТОРГОВЛИ ===
# пример ордера:
# order = binance.create_order(symbol, "market", "buy", 0.001)
# print(order)

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

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

📌"Pattern Hunter" - (охотник за уникальными паттернами цены) 🧑‍💻

🔑 Идея: 🧨

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

👉 Этот инструмент автоматически ищет повторяющиеся последовательности свечей (например, 3–5 подряд), чтобы находить статистически значимые закономерности.

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

1. Берёт историю свечей.

2. Кодирует их в паттерны (например, U – рост, D – падение, N – нейтральная).

3. Считает частоту появления паттернов.

4. Строит статистику: как часто после каждого паттерна цена растёт или падает.

import ccxt
import pandas as pd

# === Параметры ===
symbol = "BTC/USDT"
timeframe = "15m"
limit = 500
pattern_len = 3

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

df = pd.DataFrame(ohlcv, columns=["ts","Open","High","Low","Close","Volume"])

# кодируем свечи
df["dir"] = df.apply(lambda r: "U" if r["Close"]>r["Open"] else ("D" if r["Close"]<r["Open"] else "N"), axis=1)

# формируем паттерны
patterns = []
outcomes = []
for i in range(len(df)-pattern_len):
    pat = "".join(df["dir"].iloc[i:i+pattern_len])
    next_move = "UP" if df["Close"].iloc[i+pattern_len] > df["Close"].iloc[i+pattern_len-1] else "DOWN"
    patterns.append(pat)
    outcomes.append(next_move)

stats = pd.DataFrame({"pattern":patterns,"outcome":outcomes})
result = stats.groupby(["pattern","outcome"]).size().unstack(fill_value=0)

print("🔎 Найденные паттерны и их исходы:")
print(result)

#инструмент

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

📌"Exhaustion Wick Reversal" - (разворот по свечам-выбросам)🧑‍💻

🔑 Идея: 🧨

Часто рынок делает резкий «выброс» цены — длинный хвост свечи (wick), за которым следует разворот.
Это показывает исчерпание агрессивных участников:

длинный нижний хвост = продавцы выдохлись → сигнал на лонг,

длинный верхний хвост = покупатели выдохлись → сигнал на шорт.

📌 Логика входа 💸

1. Берём свечи (например, 15m или 1h).

2. Если нижний хвост > 2× тела свечи и свеча закрылась выше середины → LONG.

3. Если верхний хвост > 2× тела и закрытие ниже середины → SHORT.

4. Стоп-лосс за хвостом.

5. Тейк-профит = 1.5–2× риск.

import pandas as pd
import ccxt

# === Параметры ===
symbol = "BTC/USDT"
timeframe = "15m"
limit = 300

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

df = pd.DataFrame(ohlcv, columns=["ts","Open","High","Low","Close","Volume"])

def wick_reversal(row):
    body = abs(row["Close"] - row["Open"])
    upper_wick = row["High"] - max(row["Close"], row["Open"])
    lower_wick = min(row["Close"], row["Open"]) - row["Low"]

    # long-сигнал
    if lower_wick > 2 * body and row["Close"] > (row["Open"] + row["Low"]) / 2:
        return "LONG"
    # short-сигнал
    if upper_wick > 2 * body and row["Close"] < (row["Open"] + row["High"]) / 2:
        return "SHORT"
    return None

df["Signal"] = df.apply(wick_reversal, axis=1)

print(df[["ts","Open","High","Low","Close","Signal"]].dropna())

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

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

📌"Impulse & Absorption" - (импульс и поглощение) 🧑‍💻

🔑 Идея: 🧨

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

👉 Логика:🛠️

- импульс выбил стопы, но ликвидности на продолжение нет,

- рынок «разворачивается на пустоте».

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

1. Импульсная свеча — тело больше 2× среднего за N свечей.

2. Поглощение — следующая свеча закрывается в противоположную сторону и перекрывает минимум 70% предыдущего тела.

3. Вход в направлении свечи-поглощения.

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

5. Тейк-профит = 2× риск или до локального уровня.

import ccxt
import pandas as pd

# === параметры ===
symbol = "BTC/USDT"
timeframe = "15m"
limit = 300

binance = ccxt.binance()
ohlcv = binance.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=["ts","Open","High","Low","Close","Volume"])

# среднее тело свечи
df["body"] = abs(df["Close"] - df["Open"])
avg_body = df["body"].rolling(20).mean()

signals = []
for i in range(21, len(df)):
    body = df["body"].iloc[i-1]
    if body > 2 * avg_body.iloc[i-1]:
        prev_open, prev_close = df["Open"].iloc[i-1], df["Close"].iloc[i-1]
        curr_open, curr_close = df["Open"].iloc[i], df["Close"].iloc[i]

        # бычий импульс + медвежье поглощение
        if prev_close > prev_open and curr_close < curr_open and curr_close < (prev_open + 0.3*(prev_close-prev_open)):
            signals.append((df["ts"].iloc[i], "SHORT"))

        # медвежий импульс + бычье поглощение
        elif prev_close < prev_open and curr_close > curr_open and curr_close > (prev_open - 0.3*(prev_close-prev_open)):
            signals.append((df["ts"].iloc[i], "LONG"))

print("📊 Сигналы стратегии:")
for s in signals:
    print(s)

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

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

📌Order Book Resilience Index (OBRI) - (индекс устойчивости книги — насколько быстро и в каком объёме книга восстанавливается после потребления ликвидности) 🧑‍💻

📌 Идея (коротко): 🧨

Когда маркет-ордеры «съедают» уровень в книге, важно знать: заполнилась ли книга снова (resilient) или осталась пустой (vulnerable).

Быстрая регенерация — маркетмейкеры/лимитники возвращаются → движение менее рискованно.

Долгая/слабая регенерация — малый запас лимитов → пробой/импульс может продолжиться с меньшим сопротивлением.

OBRI измеряет скорость и долю восстановления объёма в локальной зоне (±X% от mid) после искусственного (или естественного) потребления уровня. Выдаёт индекс 0..1: 1 — полностью быстрый откат/восстановление; 0 — книга «пустая».

📌 Чем полезен: 💸

Фильтр для исполнения (если OBRI низкий → дробить ордера, не давать market-impact).

Фильтр для пробойных стратегий (низкий OBRI → пробой вероятнее и продолжительнее).

Риск-метрика: высокая вероятность сломанных уровней → повышать осторожность.

#инструмент

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

📌"Time-of-Absorption" (TOA) - (стратегия на основе времени удержания цены после агрессивного удара по стакану) 🧑‍💻

🔑 Идея: 🧨

На реальном рынке важно не только то, что цену пробили, а как долго она там удержалась.

Если крупный рыночный ордер выбил уровень, но цена сразу вернулась обратно — это ложный пробой.

Если цена удерживается и объём лимитов «поглощает» последующие удары — это сигнал к продолжению движения.

То есть время поглощения (time-of-absorption) выступает главным фильтром.

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

1. Мониторим стакан и поток сделок (trades).

2. Когда цена делает импульс (съедается заметный уровень в стакане, ≥ X% локальной глубины) → фиксируем новый уровень.

3. Засекаем время удержания:

Если в течение T_hold секунд цена не возвращается за пробитый уровень → считаем уровень «поглощённым».

Если цена откатывает назад до старого диапазона быстрее T_fail → ложный пробой.

4. Торговое правило:

Лонг, если пробит уровень сверху вниз (bid side) и цена удержалась выше.

Шорт, если пробит уровень снизу вверх (ask side) и цена удержалась ниже.

5. Stop-loss = за пробитым уровнем.

6. Take-profit = риск:прибыль 1:2 или по достижении нового ближайшего кластера ликвидности.

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

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

📌"Trade Memory Tracker" - (Трекер повторяющихся ошибок и паттернов трейдера) 🧑‍💻

🔑 Идея: 🧨

Многие трейдеры совершают одни и те же ошибки:

ранний выход,

перезаход,

вход без сигнала,

завышенный риск.

👉 Инструмент автоматически ведёт журнал сделок и выявляет повторяющиеся ошибки/паттерны трейдера, которые стоят ему денег.

📌 Логика работы: 🛠️

1. После каждой сделки трейдер сохраняет результат (P/L, направление, причина входа).

2. Инструмент анализирует историю:

средний результат по типу сигнала,

где чаще всего фиксируется убыток,

какие часы/дни дают лучший результат.

3. Генерируется отчёт: «95% убыточных сделок совершаются в азиатскую сессию», или «перезаход после стопа всегда минусовой».

import pandas as pd

# История сделок трейдера
data = [
    {"time": "2025-09-01 12:00", "signal": "breakout", "result": -50},
    {"time": "2025-09-01 15:00", "signal": "retest", "result": +120},
    {"time": "2025-09-02 10:00", "signal": "impulse", "result": -30},
    {"time": "2025-09-02 18:00", "signal": "breakout", "result": -40},
    {"time": "2025-09-03 14:00", "signal": "retest", "result": +90},
]

df = pd.DataFrame(data)
df['time'] = pd.to_datetime(df['time'])
df['day'] = df['time'].dt.day_name()
df['hour'] = df['time'].dt.hour

# Анализ по сигналам
signal_stats = df.groupby('signal')['result'].agg(['count','mean','sum'])

# Анализ по времени суток
hour_stats = df.groupby('hour')['result'].mean()

# Общий вывод
print("📊 Ошибки и паттерны трейдера:\n")
print("По сигналам:\n", signal_stats)
print("\nПо часам:\n", hour_stats)

#инструмент

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

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