📌"Market Memory Index" - (индекс рыночной памяти) 🧑‍💻

🔍 Идея: 🧨

Цена не всегда «помнит» прошлое одинаково.

В трендовых режимах каждый следующий бар коррелирует с предыдущими (есть инерция).

Во флэте рынок «забывает» направление (корреляция быстро уходит в ноль).

👉 Инструмент измеряет автокорреляцию доходностей на разных лагах и строит индекс «рыночной памяти».

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

Берёт последние N баров.

Считает лаговую корреляцию (например, 1–10 шагов).

Выводит индекс 0…1:

Близко к 1 → рынок «помнит» (инерционный режим → тренд).

Близко к 0 → рынок забывчивый (хаос → флэт).

import ccxt
import pandas as pd
import numpy as np

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"])
    df["ret"] = np.log(df["c"]).diff()
    return df.dropna()

def market_memory_index(df, max_lag=10):
    rets = df["ret"].values
    corrs = []
    for lag in range(1, max_lag+1):
        x = rets[:-lag]
        y = rets[lag:]
        corr = np.corrcoef(x, y)[0,1]
        corrs.append(corr)
    return np.mean([abs(c) for c in corrs])

if __name__ == "__main__":
    df = fetch_data()
    mmi = market_memory_index(df)
    if mmi > 0.5:
        print(f"🟢 Рынок трендовый (MMI={mmi:.2f})")
    elif mmi < 0.2:
        print(f"🔴 Рынок хаотичный (MMI={mmi:.2f})")
    else:
        print(f"🟡 Нейтральный рынок (MMI={mmi:.2f})")

#инструмент

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

📌"Импульсные ловушки"🧑‍💻

🔍 Идея: 🧨

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

Если рынок сделал быстрый импульс вверх, но через несколько свечей не продолжил рост → открываем шорт.

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

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

1. Берём короткий таймфрейм (1–5 мин).

2. Считаем относительное изменение цены за последние k свечей.

3. Если движение > X% (импульс), то ждём подтверждения:

если цена не обновляет хай/лоу импульса в течение N свечей → это «ловушка».

4. Входим против импульса.

5. Тейк-профит = середина диапазона импульса, стоп = за его экстремумом.

import ccxt
import pandas as pd
import numpy as np

SYMBOL = "BTC/USDT"
TF = "1m"
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 impulse_trap(df, k=5, threshold=0.004, hold=3):
    signals = []
    for i in range(k, len(df)-hold):
        change = (df["c"].iloc[i] - df["c"].iloc[i-k]) / df["c"].iloc[i-k]
        if abs(change) > threshold:
            if change > 0:  # импульс вверх
                high = df["h"].iloc[i]
                if all(df["h"].iloc[i+1:i+1+hold] < high):
                    signals.append((df["ts"].iloc[i], "SELL", df["c"].iloc[i]))
            else:  # импульс вниз
                low = df["l"].iloc[i]
                if all(df["l"].iloc[i+1:i+1+hold] > low):
                    signals.append((df["ts"].iloc[i], "BUY", df["c"].iloc[i]))
    return signals

if __name__ == "__main__":
    df = fetch_data()
    traps = impulse_trap(df)
    for t, side, price in traps:
        print(f"{pd.to_datetime(t, unit='ms')} → {side} @ {price}")

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

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

📌"Market Microstructure Lag Detector" - (детектор микроструктурного лага) 🧑‍💻

🔍 Идея: 🚀

На биржах иногда возникает лаг между движением цены и реакцией объёма.

В норме: рост цены сопровождается ростом объёмов.

В аномальных ситуациях: цена движется, но объём запаздывает (или наоборот).

👉 Этот инструмент измеряет сдвиг во времени между изменением цены и объёмом → показывает, есть ли «запаздывание реакции рынка».

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

1. Берёт последние N свечей.

2. Строит кросс-корреляцию между:

изменением цены,

изменением объёма.

3. Находит лаг (в свечах), где корреляция максимальна.

4. Выводит результат:

Лаг = 0 → рынок синхронен.

Лаг > 0 → объём опаздывает за ценой.

Лаг < 0 → объём идёт впереди (предупреждает движение).

Польза:💸

- Можно выявлять режимы, когда объём становится опережающим сигналом.

- Подходит для поиска «умных денег», которые входят заранее.

- Отличный фильтр для стратегий на пробои и импульсы.

import ccxt
import pandas as pd
import numpy as np

SYMBOL = "BTC/USDT"
TF = "1m"
LOOKBACK = 500

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"])
    df["ret"] = df["c"].pct_change().fillna(0)
    df["vol_chg"] = df["v"].pct_change().fillna(0)
    return df

def micro_lag(df, max_lag=10):
    x = df["ret"].values
    y = df["vol_chg"].values
    lags = range(-max_lag, max_lag+1)
    corrs = []
    for lag in lags:
        if lag < 0:
            corr = np.corrcoef(x[:lag], y[-lag:])[0,1]
        elif lag > 0:
            corr = np.corrcoef(x[lag:], y[:-lag])[0,1]
        else:
            corr = np.corrcoef(x, y)[0,1]
        corrs.append(corr)
    best_idx = int(np.argmax(np.abs(corrs)))
    return lags, corrs, lags[best_idx], corrs[best_idx]

if __name__ == "__main__":
    df = fetch_data()
    lags, corrs, best_lag, best_corr = micro_lag(df)
    print(f"📊 Лучший лаг: {best_lag} свечей | Корреляция={best_corr:.2f}")
    if best_lag > 0:
        print("⚡ Объём запаздывает за ценой")
    elif best_lag < 0:
        print("🔮 Объём предсказывает цену")
    else:
        print("⚖ Цена и объём синхронны")

#инструмент

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

📌"Noise-to-Signal Ratio" - (коэффициент шума к сигналу) 🧑‍💻

🔍 Идея: 🚀

Не всегда рынок двигается «осмысленно». Иногда цена просто «шумит» в узком диапазоне, и любые сигналы стратегий становятся ложными.

👉 Этот инструмент измеряет отношение шумовых колебаний к направленному движению.

Если шум >> тренд → лучше не торговать (рынок «грязный»).

Если тренд >> шум → высокая вероятность эффективных сделок.

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

1. Берёт окно последних N свечей.

2. Считает:

Общее движение;

Сумму всех колебаний;

3. Рассчитывает коэффициент NSR

NSR ≈ 1 → движение чистое (трендовое).

NSR >> 1 → рынок шумный, нет смысла торговать пробои.

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 noise_signal_ratio(df, window=50):
    closes = df["c"].iloc[-window:].values
    total_move = abs(closes[-1] - closes[0])
    swings = np.sum(abs(np.diff(closes)))
    if total_move == 0:
        return float("inf")
    return swings / total_move

if __name__ == "__main__":
    df = fetch_data()
    nsr = noise_signal_ratio(df)
    print(f"📊 Noise-to-Signal Ratio: {nsr:.2f}")
    if nsr < 2:
        print("🔥 Чистый трендовый рынок — пробойные стратегии в приоритете")
    elif nsr < 5:
        print("⚖ Смешанный режим — подойдёт скальпинг")
    else:
        print("🌪 Шумный рынок — лучше не торговать пробои")

#инструмент

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

📌"Импульсные ритмы" - (торговля по ритмике микродвижений) 🧑‍💻

🔍 Идея: 🧨

Рынок двигается «рывками» — серии маленьких импульсов в одном направлении часто означают, что алгоритмы двигают цену.

Если появляется серия импульсов одной стороны (например, несколько свечей подряд с телом > X%), это значит, что активные участники толкают рынок.

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

Если серия усиливается → продолжаем в сторону движения.

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

1. Находим последовательности из k свечей с телами больше threshold.

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

3. Если каждая следующая больше → входим в сторону импульса.

4. Выход: по противоположному импульсу или при смене «ритма».

import ccxt
import pandas as pd

SYMBOL = "BTC/USDT"
TF = "1m"
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"])
    df["body"] = abs(df["c"] - df["o"])
    df["dir"] = df["c"] > df["o"]
    return df

def impulse_rhythm(df, k=3, threshold=0.001):
    signals = []
    for i in range(k, len(df)):
        seq = df.iloc[i-k+1:i+1]
        if all(seq["body"] > seq["c"].iloc[-1] * threshold):
            if seq["dir"].all():  # все вверх
                if seq["body"].is_monotonic_increasing:
                    signals.append((df["ts"].iloc[i], "BUY", df["c"].iloc[i]))
                elif seq["body"].is_monotonic_decreasing:
                    signals.append((df["ts"].iloc[i], "SELL", df["c"].iloc[i]))
            elif (~seq["dir"]).all():  # все вниз
                if seq["body"].is_monotonic_increasing:
                    signals.append((df["ts"].iloc[i], "SELL", df["c"].iloc[i]))
                elif seq["body"].is_monotonic_decreasing:
                    signals.append((df["ts"].iloc[i], "BUY", df["c"].iloc[i]))
    return signals

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

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

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

📌"Ценовой маятник" 🧑‍💻

🔍 Идея: 🚀

Цена на коротких интервалах ведёт себя как маятник:

резкий рывок → накопление «кинетической энергии»,

затем возврат к равновесию.

Мы ловим эти отклонения от равновесной точки и зарабатываем на возврате.

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

1. Считаем «центр равновесия» цены = скользящее среднее по N свечам.

2. Измеряем отклонение от центра в процентах.

3. Если отклонение > X%, открываем позицию в обратную сторону (ждём возврата маятника).

4. Выход — возврат цены к центру (тейк), стоп — за пределами расширенного диапазона.

import ccxt
import pandas as pd
import numpy as np

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 pendulum_strategy(df, n=50, threshold=0.02):
    df["center"] = df["c"].rolling(n).mean()
    df = df.dropna()
    signals = []
    for i in range(len(df)):
        price = df["c"].iloc[i]
        center = df["center"].iloc[i]
        deviation = (price - center) / center
        if deviation > threshold:
            signals.append((df["ts"].iloc[i], "SELL", price))
        elif deviation < -threshold:
            signals.append((df["ts"].iloc[i], "BUY", price))
    return signals

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

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

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

📌"Радар крупных ордеров" 🧑‍💻

📌 Идея: 🧨

На биржах часто выставляют айсберг-ордера или «стены» в стакане, которые могут:

- удерживать цену,

- создавать иллюзию спроса/предложения,

- сигнализировать о возможном развороте.


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

import ccxt
import time

SYMBOL = "BTC/USDT"
DEPTH = 20
THRESHOLD = 50  # минимальный размер ордера в BTC для "аномалии"

ex = ccxt.binance()

def radar():
    orderbook = ex.fetch_order_book(SYMBOL, limit=DEPTH)
    bids = orderbook["bids"]
    asks = orderbook["asks"]

    big_bids = [(p, v) for p, v in bids if v > THRESHOLD]
    big_asks = [(p, v) for p, v in asks if v > THRESHOLD]

    if big_bids:
        print("🟢 Крупные покупки:")
        for p, v in big_bids:
            print(f"   {v:.2f} BTC @ {p}")

    if big_asks:
        print("🔴 Крупные продажи:")
        for p, v in big_asks:
            print(f"   {v:.2f} BTC @ {p}")

if __name__ == "__main__":
    while True:
        radar()
        time.sleep(3)

#инструмент

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

📌"Барометр рынка"🧑‍💻

📌 Идея: 🧨

Вместо того, чтобы смотреть на RSI/МАСD, мы можем оценивать:

сколько свечей подряд закрываются зелёными,

сколько — красными,

и какова их средняя амплитуда.

👉 Это создаёт «барометр»: преобладает ли давление покупателей или продавцов.

🎯 Польза: 💸

- Быстро видно, кто контролирует рынок — покупатели или продавцы.

- Можно использовать как фильтр для стратегий (входить только, если «барометр» за тебя).

- Помогает оценивать моментум без стандартных индикаторов.

import ccxt
import pandas as pd

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

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 market_barometer(df, streak_limit=5):
    df["color"] = df.apply(lambda x: "green" if x["c"] > x["o"] else "red", axis=1)
    df["body"] = abs(df["c"] - df["o"])

    streaks = []
    streak, prev = 0, None

    for i, row in df.iterrows():
        if row["color"] == prev:
            streak += 1
        else:
            streak = 1
        prev = row["color"]
        if streak >= streak_limit:
            streaks.append((row["ts"], row["color"], streak, row["body"]))

    return streaks

if __name__ == "__main__":
    df = fetch_data()
    sigs = market_barometer(df)
    for t, col, s, b in sigs[-5:]:
        print(f"{pd.to_datetime(t, unit='ms')} → {col.upper()} streak x{s}, avg body={b:.2f}")

#инструмент

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

📌"Импульс толпы" 🧑‍💻

🔍 Идея: 🚀

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

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

📌 Логика: 🧨

1. Считаем скорость движения цены = разница закрытия за последние k свечей.

2. Если скорость > порога и сопровождается повышенным объёмом → сигнал.

3. Входим в сторону импульса, держим позицию коротко (2–5 свечей).

4. Стоп — за началом импульса, тейк — фиксированный или по обратному импульсу.

import ccxt
import pandas as pd
import numpy as np

SYMBOL = "BTC/USDT"
TF = "1m"
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 crowd_impulse_strategy(df, k=3, vol_mult=2, speed_thr=0.003):
    df["ret"] = df["c"].pct_change(k)
    avg_vol = df["v"].rolling(50).mean()
    signals = []
    for i in range(len(df)):
        if i < k: 
            continue
        speed = df["ret"].iloc[i]
        vol = df["v"].iloc[i]
        if speed > speed_thr and vol > avg_vol.iloc[i] * vol_mult:
            signals.append((df["ts"].iloc[i], "BUY", df["c"].iloc[i]))
        elif speed < -speed_thr and vol > avg_vol.iloc[i] * vol_mult:
            signals.append((df["ts"].iloc[i], "SELL", df["c"].iloc[i]))
    return signals

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

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

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

📌"Радар резких движений"🧑‍💻

📌 Идея: 💸

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

- войти в импульс,

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


👉 Этот инструмент отслеживает мгновенные всплески волатильности и сигналит прямо в консоль.

import ccxt
import pandas as pd
import time

SYMBOL = "BTC/USDT"
TF = "1m"
LOOKBACK = 50
THRESHOLD = 0.003  # 0.3% за минуту

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 radar(df):
    df["ret"] = df["c"].pct_change()
    last = df.iloc[-1]
    if abs(last["ret"]) > THRESHOLD:
        direction = "📈 ВСПЛЕСК ВВЕРХ" if last["ret"] > 0 else "📉 ВСПЛЕСК ВНИЗ"
        print(f"{pd.to_datetime(last['ts'], unit='ms')} — {direction} {last['ret']*100:.2f}%")

if __name__ == "__main__":
    while True:
        try:
            df = fetch_data()
            radar(df)
            time.sleep(10)
        except Exception as e:
            print("Ошибка:", e)
            time.sleep(5)

#инструмент

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

📌"Поглощение агрессора" 🧑‍💻

🔍 Идея: 🚀

Когда рынок двигается в одну сторону, всегда есть «агрессор» — те, кто толкает цену ордерами по рынку. Но иногда лимитные заявки полностью поглощают их напор: свеча идёт в одну сторону, но закрывается в другую.

👉 Это момент, когда инициатива переходит к противоположной стороне.


📌 Логика: 🧨

1. Ищем свечи, у которых:

- тело большое,

- направление противоположно предыдущей свече,

- а объём выше среднего.

2. Входим в сторону свечи-поглотителя.

3. Стоп — за максимум/минимум предыдущей свечи.

4. Тейк — фиксированный R:R (например, 2:1).

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 absorption_strategy(df, vol_mult=1.5):
    avg_vol = df["v"].rolling(50).mean()
    signals = []

    for i in range(1, len(df)):
        prev = df.iloc[i-1]
        curr = df.iloc[i]

        # свеча поглощает предыдущую
        if curr["c"] > curr["o"] and prev["c"] < prev["o"] and curr["c"] >= prev["o"] and curr["v"] > avg_vol.iloc[i]*vol_mult:
            signals.append((curr["ts"], "BUY", curr["c"]))

        elif curr["c"] < curr["o"] and prev["c"] > prev["o"] and curr["c"] <= prev["o"] and curr["v"] > avg_vol.iloc[i]*vol_mult:
            signals.append((curr["ts"], "SELL", curr["c"]))

    return signals

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

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

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

📌"Алерт уровней"💸

📌 Идея: 🧨

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

👉 Этот инструмент будет в реальном времени следить за ценой и сигналить в консоль, когда уровень достигнут или пробит.

Польза: 💸

Следит за важными зонами без постоянного взгляда на график.

Работает как «сторожевой пёс» — сразу сообщает, когда цена рядом с ключевым уровнем.

Можно использовать для:

- уровней Фибоначчи,

- сильных хай/лоу,

- круглых чисел (60000, 65000 и т.д.).

import ccxt
import time

SYMBOL = "BTC/USDT"
LEVELS = [60000, 59000, 60500]  # отслеживаемые уровни
THRESHOLD = 50  # допустимое отклонение в $

ex = ccxt.binance()

def check_levels():
    ticker = ex.fetch_ticker(SYMBOL)
    price = ticker["last"]
    for lvl in LEVELS:
        if abs(price - lvl) <= THRESHOLD:
            print(f"🚨 Цена {price} близка к уровню {lvl} (±{THRESHOLD})")

if __name__ == "__main__":
    while True:
        try:
            check_levels()
            time.sleep(5)
        except Exception as e:
            print("Ошибка:", e)
            time.sleep(5)

#инструмент

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

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

🔍 Идея: 🚀

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

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

📌 Логика: 🧨

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