📌Whale Trace🧑‍💻 - (отслеживание «китов» через аномальные сделки и кластеры ликвидности)

🔍 Идея: 🚀

Крупные игроки («киты») двигают рынок. Их следы видно в:

аномальных объёмах на свече,

кластерных ордерах (ряд свечей с выше-нормативным объёмом),

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

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

Если после зоны цена возвращается → это «уровень удержания».

Если цена пробивает → может начаться сильное движение.

⚙️ Польза: 💸

- Помогает трейдеру видеть где реально был интерес денег, а не просто «линии на графике».

- Можно использовать как динамическую поддержку/сопротивление.

- Отлично работает в крипте, где стопы выбивают именно за счёт крупных ордеров

import ccxt
import pandas as pd
import numpy as np
from datetime import datetime, timedelta, timezone
import time

SYMBOL = "BTC/USDT"
TF = "15m"
DAYS = 10

def fetch_ohlcv(ex, symbol, tf, since):
    data = []
    while True:
        batch = ex.fetch_ohlcv(symbol, tf, since, 1000)
        if not batch:
            break
        data += batch
        if len(batch) < 1000:
            break
        since = batch[-1][0] + 1
        time.sleep(ex.rateLimit / 1000.0)
    df = pd.DataFrame(data, columns=["ts","open","high","low","close","volume"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms", utc=True)
    return df.set_index("dt")

def detect_whale_zones(df, vol_mult=2.0, cluster_len=3):
    df["vol_avg"] = df["volume"].rolling(50).mean()
    whale_zones = []
    for i in range(len(df)):
        if df["volume"].iloc[i] > vol_mult * df["vol_avg"].iloc[i]:
            # берем кластер из нескольких свечей
            zone = df.iloc[max(0,i-cluster_len):i+1]
            price_min = zone["low"].min()
            price_max = zone["high"].max()
            whale_zones.append((df.index[i], price_min, price_max))
    return whale_zones

if __name__ == "__main__":
    ex = ccxt.binance({"enableRateLimit": True})
    since = int((datetime.now(timezone.utc) - timedelta(days=DAYS)).timestamp()*1000)

    df = fetch_ohlcv(ex, SYMBOL, TF, since)
    zones = detect_whale_zones(df)

    print(f"Whale Zones for {SYMBOL}:")
    for z in zones[-10:]:
        print(f"{z[0]} | Zone: {z[1]:.2f} - {z[2]:.2f}")

#инструмент

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

📌Liquidity Spiral🧑‍💻

(ловим «спираль ликвидности» — резкие выбросы цены после сжатия диапазона)

🔍 Идея:🧨

На крипторынке часто бывает так:

1. Цена сжимается в узком диапазоне (низкая волатильность, кластеры стопов копятся).

2. Резкий выброс ликвидности в одну сторону «срывает» стопы.

3. Часто после этого рынок делает короткий возврат (reversion).

Мы ловим именно возврат после выброса:

ждём фазу «сжатия» (низкий ATR / Bollinger Bands очень узкие),

если свеча пробивает диапазон с большим объёмом → входим в противоположную сторону (играем на возврат).

⚙️ Правила:🛠️

1. Считаем Bollinger Bands (20, 2σ).

2. Ищем периоды, когда ширина BB < 1.5% от цены (сжатие).

3. Если появляется свеча:

закрытие выше верхней BB + объём > 1.5× среднего → открываем шорт.

закрытие ниже нижней BB + объём > 1.5× среднего → открываем лонг.

4. Стоп: 0.5× ширины диапазона за экстремумом.

5. Тейк: середина диапазона (reversion).

import ccxt
import pandas as pd
import numpy as np
from datetime import datetime, timedelta, timezone
import time

SYMBOL = "BTC/USDT"
TF = "15m"
HISTORY_DAYS = 30

def fetch_ohlcv(ex, symbol, tf, since):
    data = []
    while True:
        batch = ex.fetch_ohlcv(symbol, tf, since, 1000)
        if not batch:
            break
        data += batch
        if len(batch) < 1000:
            break
        since = batch[-1][0] + 1
        time.sleep(ex.rateLimit / 1000.0)
    df = pd.DataFrame(data, columns=["ts","open","high","low","close","volume"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms", utc=True)
    return df.set_index("dt")

def bollinger(series, n=20, k=2):
    ma = series.rolling(n).mean()
    std = series.rolling(n).std()
    upper = ma + k*std
    lower = ma - k*std
    return ma, upper, lower

if __name__ == "__main__":
    ex = ccxt.binance({"enableRateLimit": True})
    since = int((datetime.now(timezone.utc) - timedelta(days=HISTORY_DAYS)).timestamp()*1000)

    df = fetch_ohlcv(ex, SYMBOL, TF, since)

    df["ma"], df["bb_up"], df["bb_dn"] = bollinger(df["close"])
    df["bb_width"] = (df["bb_up"] - df["bb_dn"]) / df["close"] * 100
    df["vol_avg"] = df["volume"].rolling(20).mean()

    signals = []
    for t, row in df.iterrows():
        if pd.isna(row["bb_up"]): 
            continue
        # условие сжатия
        if row["bb_width"] < 1.5:
            # пробой вверх
            if row["close"] > row["bb_up"] and row["volume"] > 1.5*row["vol_avg"]:
                stop = row["high"] + (row["bb_width"]/200)*row["close"]
                target = row["ma"]
                signals.append((t, "SHORT", row["close"], stop, target))
            # пробой вниз
            elif row["close"] < row["bb_dn"] and row["volume"] > 1.5*row["vol_avg"]:
                stop = row["low"] - (row["bb_width"]/200)*row["close"]
                target = row["ma"]
                signals.append((t, "LONG", row["close"], stop, target))

    print("Signals:")
    for s in signals[-10:]:
        print(s)

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

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

📌Strategy Fitness Tracker 🧑‍💻 - («фитнес-трекер» для торговых стратегий в реальном времени)

🔍 Идея: 🧨

Обычно трейдеры тестируют стратегию в бэктесте → запускают в реал → потом сложно понять: а стратегия жива или уже «сломалась»?

Этот инструмент решает проблему:

- В реальном времени сравнивает эффективность стратегии с её же исторической моделью.

- Если стратегия отклоняется от нормы (например, просадка или winrate ниже среднего) → подаёт сигнал «пора пересмотреть/выключить».

- Работает как пульсометр для алгоритмов.

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

1. Подключается к API и получает сигналы/сделки твоей стратегии.

2. Считает rolling-метрики (winrate, средняя доходность сделки, средний риск/профит).

3. Сравнивает с историческими нормами (например, последние 100 сделок).

4. Если отклонение > Xσ → предупреждает («стратегия в зоне риска»).

import pandas as pd
import numpy as np

class StrategyFitness:
    def __init__(self, window=100):
        self.window = window
        self.trades = []

    def add_trade(self, pnl):
        """Добавляем сделку (pnl в % или $)"""
        self.trades.append(pnl)

    def check_fitness(self):
        if len(self.trades) < self.window:
            return "Недостаточно данных"
        
        df = pd.Series(self.trades[-self.window:])
        mean = df.mean()
        std = df.std()
        winrate = (df > 0).mean()

        alerts = []
        if mean < 0:
            alerts.append("⚠️ Средний PnL отрицательный")
        if winrate < 0.4:
            alerts.append("⚠️ Winrate падает")
        if std > abs(mean)*3:
            alerts.append("⚠️ Риск/волатильность слишком высоки")

        return alerts if alerts else ["✅ Стратегия здорова"]

# пример использования
tracker = StrategyFitness(window=50)

# симулируем сделки
for trade in np.random.normal(0.1, 1, 200):  
    tracker.add_trade(trade)

print(tracker.check_fitness())

#инструмент

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

📌DC-OR: Directional Change Overshoot Reversal🧑‍💻 - (разворот после «переразмаха»Directional Change)

Идея:🧨

Модель Directional Change (DC) рассматривает не время, а события: тренд меняется тогда, когда цена отклоняется от последнего экстремума больше порога λ. После смены направления часто возникает overshoot — «переразмах» за базовый порог. Наша логика:

Отслеживаем события DC при пороге λ (например, 0.6%).

Когда переразмах (overshoot) становится аномально большим (отн. ATR), ждём свечу «возврата» внутрь — и входим против движения.

Тейк — к цене, на которой зафиксировалась смена DC (или 1.5× стоп), стоп — за экстремумом переразмаха.

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

Экстремальные переразмахи часто вызваны выносом ликвидности/стопов; возврат к цене DC — естественный «магнит».

Правила:🛠

Таймфрейм: 5m/15m/1h (пример — 15m).

Порог DC: λ = 0.6% (настроечно).

Сигнал на шорт: текущий режим — «up DC», overshoot ≥ k×ATR (например, 1.0×), и свеча закрылась ниже экстремума overshoot → шорт.
Цель: цена DC (уровень смены), стоп: максимум overshoot + buffer.

Сигнал на лонг: зеркально для «down DC».

Фильтр волатильности: ATR/Close ≥ 0.5%.

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

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

📌Volatility Coil 🧑‍💻 - (катушка волатильности — играем на резком выбросе после «сжатия»)

🔍 Идея: 🚀

Рынок не бывает долго одинаково активным:

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

направление выбираем с помощью фильтров тренда (EMA или объёмы).

👉 Мы ищем зоны, где рынок «сжимается в катушку» → вход на пробой.

⚙️ Логика входа: 💸

1. Считаем ATR (волатильность).

2. Берём скользящее среднее ATR → находим моменты, когда текущая вола сильно ниже среднего (например, < 0.7 * SMA(ATR)).

3. Это сигнал «сжатия».

4. Входим в сделку на пробой High/Low диапазона последних N свечей.

Если цена выше EMA → лонг на пробой.

Если ниже EMA → шорт.

import ccxt
import pandas as pd
import numpy as np
from datetime import datetime, timedelta, timezone

SYMBOL = "BTC/USDT"
TF = "1h"
DAYS = 60

def fetch_ohlcv():
    ex = ccxt.binance({"enableRateLimit": True})
    since = int((datetime.now(timezone.utc) - timedelta(days=DAYS)).timestamp() * 1000)
    data = ex.fetch_ohlcv(SYMBOL, TF, since)
    df = pd.DataFrame(data, columns=["ts","open","high","low","close","volume"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    return df

def atr(df, period=14):
    hl = df["high"] - df["low"]
    hc = abs(df["high"] - df["close"].shift())
    lc = abs(df["low"] - df["close"].shift())
    tr = pd.concat([hl,hc,lc], axis=1).max(axis=1)
    return tr.rolling(period).mean()

def generate_signals(df):
    df["ATR"] = atr(df)
    df["ATR_SMA"] = df["ATR"].rolling(50).mean()
    df["EMA"] = df["close"].ewm(span=50).mean()

    signals = []
    for i in range(50, len(df)):
        if df["ATR"].iloc[i] < 0.7 * df["ATR_SMA"].iloc[i]:  # сжатие
            rng_high = df["high"].iloc[i-10:i].max()
            rng_low = df["low"].iloc[i-10:i].min()
            price = df["close"].iloc[i]
            ema = df["EMA"].iloc[i]

            if price > ema:  # тренд вверх
                signals.append((df["dt"].iloc[i], "LONG", rng_high))
            elif price < ema:
                signals.append((df["dt"].iloc[i], "SHORT", rng_low))
    return signals

if __name__ == "__main__":
    df = fetch_ohlcv()
    signals = generate_signals(df)
    for s in signals[-10:]:
        print(s)

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

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

📌Market Stress Monitor 🧑‍💻 - (монитор стресса рынка — индикатор перегрева и паники)

🔍 Идея: 🧨

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

👉 Этот инструмент агрегирует метрики перегрева рынка и показывает трейдеру в реальном времени уровень «стресса»:

📈 Funding rate (перегретый лонг/шорт).

🔥 Открытый интерес (OI) — резкие скачки → перегруз позиций.

💧 Ликвидации (массовые стоп-ауты).

📊 Скорость изменения цены (аномально быстрое движение).

На основе этих сигналов строится индекс стресса рынка (0–100).

⚙️ Польза: 💸

- Если стресс высокий (например, >70) → рынок опасен, лучше не открывать новые сделки.

- Если стресс низкий (<30) → рынок «спокоен», хорошее время для аккуратного входа.

- Можно использовать как фильтр к любым стратегиям.

import ccxt
import pandas as pd
import numpy as np
import time

exchange = ccxt.binance({
    "enableRateLimit": True,
    "options": {"defaultType": "future"}
})

SYMBOL = "BTC/USDT"

def get_funding_rate():
    data = exchange.fapiPublic_get_fundingrate({"symbol": "BTCUSDT", "limit": 1})
    return float(data[0]["fundingRate"])

def get_open_interest():
    data = exchange.fapiPublic_get_openinterest({"symbol": "BTCUSDT"})
    return float(data["openInterest"])

def get_ticker():
    t = exchange.fetch_ticker(SYMBOL)
    return t["last"], t["quoteVolume"]

def stress_index():
    funding = abs(get_funding_rate()) * 10000   # в bps
    oi = get_open_interest()
    price, volume = get_ticker()

    # нормализация (примитивный пример)
    funding_score = min(funding, 50)
    oi_score = np.log(oi) % 100
    vol_score = np.log(volume) % 100

    score = (funding_score*0.3 + oi_score*0.3 + vol_score*0.4)
    return round(score, 2)

while True:
    print("Market Stress:", stress_index())
    time.sleep(60)

#инструмент

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

📌Position Sizer Pro - (умный калькулятор объёма позиции с адаптивным риском) 🧑‍💻

🔍 Идея: 🧨

Обычный калькулятор позиции скучный: трейдер сам вводит риск и плечо.
👉 Здесь мы делаем умный авто-калькулятор, который:

сам подбирает размер позиции на основе волатильности инструмента (ATR),

учитывает баланс трейдера,

автоматически корректирует риск, если рынок в «стрессе».

📌 Польза: 💸

Каждый день можно быстро понять: сколько лотов брать, чтобы не словить маржин-колл.

Работает как универсальный помощник, который адаптирует риск под рынок.

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

import ccxt
import pandas as pd
import numpy as np
from datetime import datetime, timedelta, timezone

SYMBOL = "BTC/USDT"
TF = "1h"
BALANCE = 1000  # USDT
MAX_RISK = 0.02  # 2% от депо

def fetch_data():
    ex = ccxt.binance({"enableRateLimit": True})
    since = int((datetime.now(timezone.utc) - timedelta(days=30)).timestamp() * 1000)
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, since)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
    return df

def atr(df, period=14):
    hl = df["high"] - df["low"]
    hc = abs(df["high"] - df["close"].shift())
    lc = abs(df["low"] - df["close"].shift())
    tr = pd.concat([hl,hc,lc], axis=1).max(axis=1)
    return tr.rolling(period).mean()

def position_size(df, balance, max_risk):
    df["ATR"] = atr(df)
    atr_val = df["ATR"].iloc[-1]
    price = df["close"].iloc[-1]

    # риск на сделку
    risk_amount = balance * max_risk

    # стоп = ATR
    stop_dist = atr_val

    # размер позиции
    qty = risk_amount / stop_dist
    contracts = qty / price * balance
    return round(contracts, 4), price, atr_val

if __name__ == "__main__":
    df = fetch_data()
    size, price, atr_val = position_size(df, BALANCE, MAX_RISK)
    print(f"Цена: {price:.2f} | ATR: {atr_val:.2f}")
    print(f"Рекомендуемый размер позиции: {size} контрактов")

#инструмент

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

📌Echo Levels - (эхо-уровни — торговля на повторных отскоках скрытых уровней) 🧑‍💻

🔍 Идея: 🧨

Большинство стратегий ищут пробои или тренды.
Мы используем эффект памяти рынка:

Когда цена резко отскакивает от какого-то уровня (например, на сильном объёме),

Этот уровень остаётся «акустическим эхом» → рынок часто повторно реагирует на него спустя часы/дни.

👉 Мы строим «эхо-уровни» на основе сильных отскоков и ловим повторные касания.

⚙️ Логика: 🛠️

1. Ищем свечи с аномальным объёмом + длинным хвостом (шипом).
→ это значит, что там «защищали уровень».

2. Запоминаем цену отскока (например, минимум хвоста).

3. Если в будущем цена снова касается этой зоны ±0.2% → входим в сторону отскока.

Лонг при повторном касании «эхо-дна».

Шорт при повторном касании «эхо-верха».

4. Стоп → чуть за уровень (например, 0.5% ниже).

import ccxt
import pandas as pd
import numpy as np
from datetime import datetime, timedelta, timezone

SYMBOL = "BTC/USDT"
TF = "1h"
DAYS = 30

def fetch_ohlcv():
    ex = ccxt.binance({"enableRateLimit": True})
    since = int((datetime.now(timezone.utc) - timedelta(days=DAYS)).timestamp() * 1000)
    data = ex.fetch_ohlcv(SYMBOL, TF, since)
    df = pd.DataFrame(data, columns=["ts","open","high","low","close","volume"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    return df

def find_echo_levels(df, vol_mult=2.5):
    levels = []
    vol_mean = df["volume"].rolling(20).mean()
    for i in range(20, len(df)):
        if df["volume"].iloc[i] > vol_mult * vol_mean.iloc[i]:
            body = abs(df["close"].iloc[i] - df["open"].iloc[i])
            wick_low = df["low"].iloc[i]
            wick_high = df["high"].iloc[i]
            # если свеча с хвостом вниз
            if wick_low < df["close"].iloc[i] - body:
                levels.append(("LONG", wick_low))
            # если свеча с хвостом вверх
            if wick_high > df["close"].iloc[i] + body:
                levels.append(("SHORT", wick_high))
    return levels

def check_signals(df, levels, tolerance=0.2):
    signals = []
    for _, row in df.iterrows():
        price = row["close"]
        for side, lvl in levels:
            if side == "LONG" and abs(price - lvl) / lvl < tolerance/100:
                signals.append((row["dt"], "LONG", lvl))
            if side == "SHORT" and abs(price - lvl) / lvl < tolerance/100:
                signals.append((row["dt"], "SHORT", lvl))
    return signals

if __name__ == "__main__":
    df = fetch_ohlcv()
    levels = find_echo_levels(df)
    signals = check_signals(df, levels)
    for s in signals[-10:]:
        print(s)

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

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

📌Correlation Radar - (онлайн-радар корреляций между монетами)🧑‍💻

🔍 Идея: 🧨

Трейдеры часто смотрят на один актив (например, BTC или ETH), но забывают, что альткоины и даже разные сектора рынка двигаются синхронно или расходятся.
👉 Инструмент в реальном времени считает корреляции между монетами и показывает:

кто сейчас ходит «в унисон» с BTC,

кто отклеился и может дать уникальный трейд,

кто образует «кластер» синхронных движений.

📌 Польза: 💸

Легко находить альткоины, которые опаздывают за BTC → точка входа.

Видеть «аномально» независимые монеты → диверсификация.

Использовать как фильтр: торговать только то, что слабо связано с BTC (для хеджей).

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

exchange = ccxt.binance({"enableRateLimit": True})

SYMBOLS = ["BTC/USDT", "ETH/USDT", "SOL/USDT", "BNB/USDT", "XRP/USDT"]
TF = "1h"
DAYS = 7

def fetch_data(symbol):
    since = int((datetime.utcnow() - timedelta(days=DAYS)).timestamp() * 1000)
    ohlcv = exchange.fetch_ohlcv(symbol, TF, since)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    df.set_index("dt", inplace=True)
    return df["close"]

def correlation_matrix(symbols):
    data = {}
    for s in symbols:
        data[s.split("/")[0]] = fetch_data(s)
    df = pd.DataFrame(data)
    corr = df.pct_change().corr()
    return corr

if __name__ == "__main__":
    corr = correlation_matrix(SYMBOLS)
    print("Корреляции (за последние 7 дней):")
    print(corr)

#инструмент

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

📌Volatility Harvest- (сборщик волатильности — зарабатываем на чередовании «спокойных» и «бурных» фаз рынка) 🧑‍💻

🔍 Идея: 🚀

Рынок двигается циклами:

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

за ними почти всегда идут взрывы волатильности (пробой, импульс).

👉 Задача стратегии — войти в самом конце «тишины» и поймать «бурю».

⚙ Логика: 🛠️

1. Рассчитываем индекс сжатия рынка:

берем ATR (средний диапазон свечей),

сравниваем его с историческим средним.

2. Если ATR сильно ниже нормы (рынок «уснул») → ставим ловушки на пробой в обе стороны.

3. Вход:

Buy Stop чуть выше максимума последних N свечей,

Sell Stop чуть ниже минимума.

4. Когда одна из ловушек срабатывает → вторую отменяем.

5. Стоп = противоположный край диапазона, тейк = 2–3× стоп.

import ccxt
import pandas as pd
import numpy as np
from datetime import datetime, timedelta, timezone

SYMBOL = "BTC/USDT"
TF = "1h"
DAYS = 60

def fetch_data():
    ex = ccxt.binance({"enableRateLimit": True})
    since = int((datetime.now(timezone.utc) - timedelta(days=DAYS)).timestamp() * 1000)
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, since)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    return df

def atr(df, period=14):
    hl = df["high"] - df["low"]
    hc = abs(df["high"] - df["close"].shift())
    lc = abs(df["low"] - df["close"].shift())
    tr = pd.concat([hl,hc,lc], axis=1).max(axis=1)
    return tr.rolling(period).mean()

def find_signals(df, atr_period=14, window=20, threshold=0.5):
    df["ATR"] = atr(df, atr_period)
    signals = []
    for i in range(window, len(df)):
        atr_now = df["ATR"].iloc[i]
        atr_mean = df["ATR"].iloc[i-window:i].mean()

        # рынок "уснул"
        if atr_now < threshold * atr_mean:
            rng_high = df["high"].iloc[i-window:i].max()
            rng_low = df["low"].iloc[i-window:i].min()
            signals.append((df["dt"].iloc[i], "SET_TRAP", rng_low, rng_high))
    return signals

if __name__ == "__main__":
    df = fetch_data()
    signals = find_signals(df)
    for s in signals[-5:]:
        print(s)

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

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

📌Emotion Index - (индекс эмоций рынка в реальном времени) 🧑‍💻

🔍 Идея: 🧨

Рынок движется эмоциями: страхом и жадностью. У нас есть «индекс страха и жадности» в интернете, но он статичный и обновляется раз в сутки.
👉 Этот инструмент строит свой локальный индекс эмоций из данных:

волатильность (чем выше → тем больше паники/ажиотажа),

соотношение красных и зелёных свечей,

объёмные всплески,

перекупленность/перепроданность (RSI).

📌 Польза: 💸

Можно быстро видеть, в каком «психологическом режиме» сейчас рынок.

Использовать как фильтр:

если паника → лучше искать лонг (контртренд),

если эйфория → лучше искать шорт.

import ccxt
import pandas as pd
import numpy as np
from datetime import datetime, timedelta, timezone

SYMBOL = "BTC/USDT"
TF = "1h"
DAYS = 7

def fetch_data():
    ex = ccxt.binance({"enableRateLimit": True})
    since = int((datetime.now(timezone.utc) - timedelta(days=DAYS)).timestamp() * 1000)
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, since)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    return df

def rsi(series, period=14):
    delta = series.diff()
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
    avg_gain = gain.rolling(period).mean()
    avg_loss = loss.rolling(period).mean()
    rs = avg_gain / avg_loss
    return 100 - (100 / (1 + rs))

def emotion_index(df):
    # компоненты
    vol = df["close"].pct_change().rolling(24).std() * 100
    trend = (df["close"] > df["open"]).rolling(24).mean() * 100
    rsi_val = rsi(df["close"])
    
    # нормализация
    vol_norm = (vol - vol.min()) / (vol.max() - vol.min())
    trend_norm = trend / 100
    rsi_norm = (rsi_val - 30) / 40  # 0 при RSI=30, 1 при RSI=70
    
    # интегральный индекс
    emotion = (vol_norm + trend_norm + rsi_norm) / 3
    df["EmotionIndex"] = emotion
    return df

if __name__ == "__main__":
    df = fetch_data()
    df = emotion_index(df)
    print(df[["dt", "close", "EmotionIndex"]].tail(10))

#инструмент

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

📌Liquidity Vacuum - (играем на “пылесосе ликвидности” после ложного пробоя)🧑‍💻

🔍 Идея: 🧨

Большие игроки часто двигают цену так:

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

2. Забирают стопы толпы.

3. Возвращают цену обратно → начинается импульс в противоположную сторону.

👉 Стратегия ловит именно этот момент — когда «всосали ликвидность» и цена разворачивается.

⚙ Логика: 🛠️

1. Определяем локальный диапазон (например, high/low за последние 24 свечи).

2. Если цена пробивает high/low и возвращается внутрь диапазона → сигнал в противоположную сторону.

3. Стоп: за экстремум пробоя.

4. Тейк: 2–3× стоп или середина диапазона.

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

SYMBOL = "BTC/USDT"
TF = "15m"
DAYS = 5

def fetch_data():
    ex = ccxt.binance({"enableRateLimit": True})
    since = int((datetime.now(timezone.utc) - timedelta(days=DAYS)).timestamp() * 1000)
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, since)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    return df

def liquidity_vacuum(df, window=24):
    signals = []
    for i in range(window, len(df)):
        rng_high = df["high"].iloc[i-window:i].max()
        rng_low = df["low"].iloc[i-window:i].min()
        price = df["close"].iloc[i]

        # ложный пробой high
        if df["high"].iloc[i] > rng_high and price < rng_high:
            signals.append((df["dt"].iloc[i], "SHORT", rng_high, price))

        # ложный пробой low
        if df["low"].iloc[i] < rng_low and price > rng_low:
            signals.append((df["dt"].iloc[i], "LONG", rng_low, price))

    return signals

if __name__ == "__main__":
    df = fetch_data()
    sigs = liquidity_vacuum(df)
    for s in sigs[-5:]:
        print(s)

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

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

📌Stop-Loss Heatmap - (тепловая карта вероятных стопов толпы) 🧑‍💻

🔍 Идея: 🧨

Большинство трейдеров ставят стопы «по учебнику»:

чуть ниже локального минимума,

чуть выше локального максимума,

за круглым уровнем (например, 30 000, 25 000).

👉 Этот инструмент строит тепловую карту зон, где скапливаются вероятные стопы.
Когда цена подходит к такой зоне — велика вероятность:

или резкого пробоя (снос стопов),

или ложного прокола и возврата (идеальная точка входа).

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

1. Берём историю свечей (например, 14 дней).

2. Ищем локальные минимумы/максимумы + круглые уровни.

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

4. Рисуем тепловую карту уровней.

import ccxt
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta, timezone

SYMBOL = "BTC/USDT"
TF = "1h"
DAYS = 10

def fetch_data():
    ex = ccxt.binance({"enableRateLimit": True})
    since = int((datetime.now(timezone.utc) - timedelta(days=DAYS)).timestamp() * 1000)
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, since)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    return df

def stop_loss_heatmap(df, round_step=500):
    levels = []

    # локальные экстремумы
    for i in range(2, len(df)-2):
        if df["low"].iloc[i] < df["low"].iloc[i-1] and df["low"].iloc[i] < df["low"].iloc[i+1]:
            levels.append(df["low"].iloc[i])
        if df["high"].iloc[i] > df["high"].iloc[i-1] and df["high"].iloc[i] > df["high"].iloc[i+1]:
            levels.append(df["high"].iloc[i])

    # круглые уровни
    price_min, price_max = df["low"].min(), df["high"].max()
    round_levels = np.arange(int(price_min), int(price_max), round_step)
    levels.extend(round_levels)

    # тепловая карта (гистограмма)
    hist, bins = np.histogram(levels, bins=50)

    plt.figure(figsize=(10,6))
    plt.hist(levels, bins=50, color="red", alpha=0.6)
    plt.title("Stop-Loss Heatmap")
    plt.xlabel("Price Level")
    plt.ylabel("Stop Density")
    plt.show()

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

#инструмент

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

📌Time-Shift Arbitrage - (арбитраж времени между таймфреймами) 🧑‍💻

🔍 Идея: 🚀

Большинство стратегий смотрят только один таймфрейм. Но рынок часто даёт асинхронные сигналы:

На младшем ТФ (5m–15m) формируется локальный импульс.

На старшем ТФ (4h–1d) тренд ещё не развернулся.

👉 Если входить в сделку тогда, когда младший ТФ уже даёт сигнал, а старший ещё “не в курсе”, можно поймать движение на ранней стадии. Это и есть «арбитраж времени».

⚙ Логика: 🛠️

1. Считаем RSI или MACD на двух разных ТФ: например, 15m и 4h.

2. Если на младшем ТФ появляется сигнал (например, RSI < 30 = перепроданность), а на старшем ещё нет — входим заранее.

3. Выход → когда старший ТФ подтверждает тот же сигнал.

4. Это даёт преимущество «форвард-сигнала» в несколько часов.

import ccxt
import pandas as pd
import numpy as np
from datetime import datetime, timedelta, timezone

SYMBOL = "BTC/USDT"

def fetch_data(tf="15m", days=7):
    ex = ccxt.binance({"enableRateLimit": True})
    since = int((datetime.now(timezone.utc) - timedelta(days=days)).timestamp() * 1000)
    ohlcv = ex.fetch_ohlcv(SYMBOL, tf, since)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
    df["dt"] = pd.to_datetime(df["ts"], unit="ms")
    return df

def rsi(series, period=14):
    delta = series.diff()
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
    avg_gain = gain.rolling(period).mean()
    avg_loss = loss.rolling(period).mean()
    rs = avg_gain / avg_loss
    return 100 - (100 / (1 + rs))

def time_shift_arbitrage(df_fast, df_slow):
    df_fast["RSI"] = rsi(df_fast["close"])
    df_slow["RSI"] = rsi(df_slow["close"])
    signals = []

    for i in range(len(df_fast)):
        if df_fast["RSI"].iloc[i] < 30:  # перепроданность на младшем
            # проверим старший
            last_slow = df_slow[df_slow["dt"] <= df_fast["dt"].iloc[i]].iloc[-1]
            if last_slow["RSI"] > 40:  # старший ещё не перепродан
                signals.append((df_fast["dt"].iloc[i], "LONG", df_fast["close"].iloc[i]))

        if df_fast["RSI"].iloc[i] > 70:  # перекупленность на младшем
            last_slow = df_slow[df_slow["dt"] <= df_fast["dt"].iloc[i]].iloc[-1]
            if last_slow["RSI"] < 60:  # старший ещё не перекуплен
                signals.append((df_fast["dt"].iloc[i], "SHORT", df_fast["close"].iloc[i]))

    return signals

if __name__ == "__main__":
    fast = fetch_data("15m", days=5)
    slow = fetch_data("4h", days=30)
    sigs = time_shift_arbitrage(fast, slow)
    for s in sigs[-5:]:
        print(s)

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

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

📌Position Stress Tester - (стресс-тестер позиции в реальном времени) 🧑‍💻

🔍 Идея : 🚀

Большинство трейдеров видят только текущую цену и PnL. Но гораздо полезнее знать:

что будет с позицией, если цена упадёт на −5% или вырастет на +5%;

при каком сценарии тебя ликвиднёт;

сколько реально потеряешь/заработаешь при изменении волатильности.

👉 Инструмент моделирует несколько сценариев движения цены и показывает стресс-карту позиции.

📌 Польза: 💸

- Можно заранее понять риск при любом движении рынка.

- Помогает решать: усилить позицию, захеджировать или выйти.

- Отлично подходит для риск-менеджмента.

import numpy as np
import matplotlib.pyplot as plt

# параметры позиции
entry_price = 27000
position_size = 0.5  # BTC
leverage = 5
margin = (entry_price * position_size) / leverage

# диапазон сценариев: -15% ... +15%
scenarios = np.linspace(-0.15, 0.15, 50)
prices = entry_price * (1 + scenarios)

# расчёт PnL
pnl = (prices - entry_price) * position_size * leverage

# точка ликвидации (приблизительно)
liq_price = entry_price - margin / (position_size / entry_price)

plt.figure(figsize=(10,5))
plt.plot(prices, pnl, label="PnL")
plt.axvline(entry_price, color="blue", linestyle="--", label="Entry")
plt.axvline(liq_price, color="red", linestyle="--", label="Liq Price")
plt.axhline(0, color="black", linewidth=0.7)

plt.title("Stress Test позиции")
plt.xlabel("Цена BTC")
plt.ylabel("PnL (USDT)")
plt.legend()
plt.grid(True)
plt.show()

#инструмент

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

📌Liquidity Sweep Reversal - (разворот после сноса ликвидности) 🧑‍💻

🔍 Идея: 🧨

Крупные игроки часто собирают ликвидность:

цена выбивает стопы за локальными минимумами/максимумами,

после чего идёт быстрый возврат в диапазон.

👉 Стратегия зарабатывает на том, чтобы ловить разворот именно после сноса стопов.

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

1. Определяем локальный экстремум (high/low последних N свечей).

2. Если цена пробивает уровень и через 1–2 свечи возвращается обратно — это сигнал.

3. Входим в лонг, если снесли low и вернулись выше уровня.
Входим в шорт, если снесли high и вернулись ниже уровня.

4. Стоп — чуть дальше экстремума.

5. Тейк — 2–3х от риска.

import ccxt
import pandas as pd

SYMBOL = "BTC/USDT"
TF = "15m"
LOOKBACK = 20

ex = ccxt.binance({"enableRateLimit": True})

def fetch_data():
    ohlcv = ex.fetch_ohlcv(SYMBOL, TF, limit=500)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","volume"])
    return df

def liquidity_sweep_strategy(df):
    signals = []
    for i in range(LOOKBACK, len(df)):
        high_level = df["high"].iloc[i-LOOKBACK:i].max()
        low_level  = df["low"].iloc[i-LOOKBACK:i].min()
        c, o = df["close"].iloc[i], df["open"].iloc[i]

        # Лонг после сноса минимума
        if df["low"].iloc[i] < low_level and c > low_level:
            signals.append((df["ts"].iloc[i], "LONG", c, low_level))

        # Шорт после сноса максимума
        if df["high"].iloc[i] > high_level and c < high_level:
            signals.append((df["ts"].iloc[i], "SHORT", c, high_level))

    return signals

if __name__ == "__main__":
    df = fetch_data()
    sigs = liquidity_sweep_strategy(df)
    for s in sigs[-5:]:
        print(s)

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

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

📌Fractal Pattern Scanner - (сканер самоподобных паттернов) 🧑‍💻

🔍 Идея: 🚀

Рынок часто повторяет фрактальные структуры: маленькие паттерны свечей копируют крупные (например, мини-треугольник внутри большого).
👉 Инструмент ищет на истории повторяющиеся самоподобные комбинации свечей и отмечает, когда они возникают снова в реальном времени.

📌 Польза : 💸

- Можно находить скрытые закономерности поведения цены.

- Дает ранние сигналы о том, что рынок повторяет старый сценарий.

- Универсален: работает и на минутках, и на дневках.

import ccxt
import pandas as pd
import numpy as np
from scipy.spatial.distance import euclidean

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

def fetch_ohlcv(limit=300):
    ohlcv = ex.fetch_ohlcv(symbol, "5m", limit=limit)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","vol"])
    return df

def pattern_vector(df, i, n=5):
    """создаём вектор паттерна из n свечей"""
    window = df.iloc[i:i+n]
    return (window["close"].values - window["open"].values) / window["open"].values

def find_fractals(df, n=5, threshold=0.1):
    patterns = []
    for i in range(len(df)-n):
        v1 = pattern_vector(df, i, n)
        for j in range(i+n, len(df)-n):
            v2 = pattern_vector(df, j, n)
            if euclidean(v1, v2) < threshold:
                patterns.append((i, j))
    return patterns

df = fetch_ohlcv()
fractals = find_fractals(df)

print("🔎 Найдено фрактальных повторов:", len(fractals))

#инструмент

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

📌Order Imbalance Sniper - (снайпер дисбаланса ордербука) 🧑‍💻

🔍 Идея: 🧨

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

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

⚙ Логика: 🛠️

1. Загружаем ордербук (например, 20 уровней).

2. Считаем сумму заявок на buy и sell.

3. Если buy_volume / sell_volume > K → идём в лонг.

4. Если наоборот sell_volume / buy_volume > K → идём в шорт.

5. Стоп и тейк короткие (это стратегия скальпинга).

import ccxt
import time

ex = ccxt.binance({"enableRateLimit": True})
symbol = "BTC/USDT"
threshold = 2.0  # во сколько раз должен быть дисбаланс

def check_imbalance():
    orderbook = ex.fetch_order_book(symbol, limit=20)
    bids = sum([b[1] for b in orderbook["bids"]])
    asks = sum([a[1] for a in orderbook["asks"]])

    if bids / asks > threshold:
        return "LONG", bids/asks
    elif asks / bids > threshold:
        return "SHORT", asks/bids
    else:
        return None, 1.0

while True:
    signal, ratio = check_imbalance()
    if signal:
        print(f"📊 Сигнал: {signal} | Дисбаланс = {ratio:.2f}")
    time.sleep(3)

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

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

📌"Volatility Pocket Breakout" - (прорыв кармана волатильности) 🧑‍💻

🔍 Идея: 🧨

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

👉 Стратегия ищет такие карманы (низкая ATR и сжатие Боллинджера) и заходит в сторону пробоя.

⚙ Логика: 🛠️

1. Считаем ATR (волатильность) и ширину Bollinger Bands.

2. Если и ATR, и BB ниже среднего уровня → рынок в кармане.

3. Ждём пробоя вверх/вниз (закрытие свечи за границей BB).

4. Входим в сторону пробоя, стоп — за противоположной границей, тейк = 2× стоп

import ccxt
import pandas as pd
import numpy as np

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

def fetch_data(limit=200):
    ohlcv = ex.fetch_ohlcv(symbol, "1h", limit=limit)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","vol"])
    df["ret"] = df["close"].pct_change()
    return df

def bollinger_bands(series, window=20, mult=2):
    ma = series.rolling(window).mean()
    std = series.rolling(window).std()
    upper = ma + mult * std
    lower = ma - mult * std
    return upper, lower

def atr(df, window=14):
    high_low = df["high"] - df["low"]
    high_close = np.abs(df["high"] - df["close"].shift())
    low_close = np.abs(df["low"] - df["close"].shift())
    tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
    return tr.rolling(window).mean()

df = fetch_data()
df["ATR"] = atr(df)
df["BB_upper"], df["BB_lower"] = bollinger_bands(df["close"])

# сигналы
df["in_pocket"] = (df["ATR"] < df["ATR"].rolling(100).mean()*0.7) & \
                  ((df["BB_upper"]-df["BB_lower"]) < (df["BB_upper"]-df["BB_lower"]).rolling(100).mean()*0.7)

df["long_signal"] = (df["in_pocket"]) & (df["close"] > df["BB_upper"])
df["short_signal"] = (df["in_pocket"]) & (df["close"] < df["BB_lower"])

print(df[["ts","close","long_signal","short_signal"]].tail(20))

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

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

📌Market Symmetry Detector - (детектор рыночной симметрии) 🧑‍💻

🔍 Идея: 🚀

Рынок часто строит симметричные движения:

рост = падение по времени и амплитуде,

зеркальные паттерны на графике (левое «плечо» ≈ правому),

циклы повторяются через одинаковые интервалы.

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

📌 Польза: 💸

- Можно заранее ожидать «отзеркаливание» цены.

- Помогает в определении точек разворота.

- Работает как фильтр для стратегий на паттернах (H&S, W, M).

import ccxt
import pandas as pd
import numpy as np

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

def fetch_data(limit=300):
    ohlcv = ex.fetch_ohlcv(symbol, "1h", limit=limit)
    df = pd.DataFrame(ohlcv, columns=["ts","open","high","low","close","vol"])
    return df

def symmetry_score(segment1, segment2):
    """сравнение двух отрезков графика (по форме движения)"""
    s1 = (segment1 - segment1[0]) / segment1[0]
    s2 = (segment2 - segment2[-1]) / segment2[-1]  # реверс
    return np.corrcoef(s1, np.flip(s2))[0,1]

df = fetch_data()
window = 20
scores = []

for i in range(len(df)-2*window):
    seg1 = df["close"].iloc[i:i+window].values
    seg2 = df["close"].iloc[i+window:i+2*window].values
    score = symmetry_score(seg1, seg2)
    scores.append((i, score))

signals = [(i, s) for i, s in scores if s > 0.8]
print("🔍 Найдены симметричные паттерны:", signals[:5])

#инструмент

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