Crypto Python
818 subscribers
448 photos
26 files
455 links
Алготрейдинг 🤖 , работа с API бирж и агрегаторов 🌐 , автоматизации в крипто сфере📈 🚀
Ваши предложения📝 @binance_de
Download Telegram
📌"Стратегия сжимающегося диапазона (Volatility Squeeze Breakout)".🧑‍💻

📌 Идея стратегии:🚀

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

⚙️ Логика:🛠️

1. Считаем Bollinger Bands и Keltner Channels:

Когда полосы Боллинджера сжимаются внутрь каналов Кельтнера → рынок в сжатии.

2. Ждём выхода из диапазона:

Цена закрывается выше верхней границы Боллинджера → сигнал на покупку.

Цена закрывается ниже нижней границы Боллинджера → сигнал на продажу.


3. Дополнительно проверяем объём: пробой должен сопровождаться повышенным объёмом.

4. Сигналы выводим в консоль.

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

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

def fetch_data():
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=["time","open","high","low","close","volume"])
return df

def bollinger_keltner_strategy(df):
df["MA20"] = df["close"].rolling(20).mean()
df["STD20"] = df["close"].rolling(20).std()

df["UpperBB"] = df["MA20"] + 2 * df["STD20"]
df["LowerBB"] = df["MA20"] - 2 * df["STD20"]

df["ATR"] = (df["high"] - df["low"]).rolling(20).mean()
df["UpperKC"] = df["MA20"] + 1.5 * df["ATR"]
df["LowerKC"] = df["MA20"] - 1.5 * df["ATR"]

df["squeeze_on"] = (df["UpperBB"] < df["UpperKC"]) & (df["LowerBB"] > df["LowerKC"])
df["squeeze_off"] = (df["UpperBB"] > df["UpperKC"]) & (df["LowerBB"] < df["LowerKC"])

return df

while True:
try:
df = fetch_data()
df = bollinger_keltner_strategy(df)

last = df.iloc[-1]

print(f"\n=== {symbol} {timeframe} ===")
if last["squeeze_on"]:
print("🔵 Рынок в сжатии (ожидается пробой)")
elif last["squeeze_off"]:
if last["close"] > last["UpperBB"] and last["volume"] > df["volume"].rolling(20).mean().iloc[-1]:
print("🟢 Сигнал BUY — пробой вверх с объёмом")
elif last["close"] < last["LowerBB"] and last["volume"] > df["volume"].rolling(20).mean().iloc[-1]:
print("🔴 Сигнал SELL — пробой вниз с объёмом")
else:
print(" Нет подтверждения пробоя")
else:
print(" Рынок без особых сигналов")

time.sleep(20)

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

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

📌 Подпишись  Crypto Python❗️
👍8
trs.py
10.5 KB
📌Trade Readiness Score (TRS) Scanner — «сканер готовности к сделке».🧑‍💻

Он до запуска любой стратегии показывает, насколько “торгуем” сейчас конкретный инструмент, складывая в один балл ключевые микро- и макро-метрики рынка.

Что делает:🧨

- Снимает по REST (ccxt/Binance): спред, ликвидность в стакане (глубина в bps), дисбаланс заявок, волатильность (ATR/Close), объём vs средний, “колючесть” свечей.

- Считает сводный балл TRS (0–100) и печатает причины: что хорошо/плохо.

Помогает:💸

- отсеять пары/моменты с тонким стаканом, завышенным спредом, бушующей волой;

- выбрать лучший тикер для входа среди списка;

- ставить фильтр «не торговать ниже N баллов».

#инструмент

📌 Подпишись  Crypto Python❗️
🔥31
📌"Liquidity Vacuum" (вакуум ликвидности) 🧑‍💻

Идея:🧨

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

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

1. Находим диапазон за последние N свечей.

2. Ищем свечу, которая пробивает границы диапазона с аномально низким объёмом (то есть пробой в "пустоту").

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

4. Выход: на середине диапазона или по фиксированному соотношению риск/прибыль (например 1:2).

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

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

- Это смесь false breakout и анализа ликвидности.

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

# Конфиг
SYMBOL = "BTC/USDT"
TIMEFRAME = "15m"
BARS = 200
RANGE_LOOKBACK = 48 # размер диапазона (например сутки на 15m)
VOLUME_FACTOR = 0.7 # пробой считается слабым, если объём < 70% среднего
SLEEP = 30 # пауза между обновлениями

ex = ccxt.binance()

def fetch_data():
ohlcv = ex.fetch_ohlcv(SYMBOL, TIMEFRAME, limit=BARS)
df = pd.DataFrame(ohlcv, columns=["time","open","high","low","close","volume"])
df["time"] = pd.to_datetime(df["time"], unit="ms")
return df

def check_signal(df):
# диапазон последних N свечей (без текущей)
recent = df.iloc[-RANGE_LOOKBACK-2:-2]
rng_high, rng_low = recent["high"].max(), recent["low"].min()
avg_vol = recent["volume"].mean()

# пробойная свеча (предыдущая)
br = df.iloc[-2]
# подтверждающая свеча (последняя)
cf = df.iloc[-1]

# проверка на пробой вверх
if br["close"] > rng_high and br["volume"] < avg_vol*VOLUME_FACTOR:
if cf["close"] < rng_high: # возврат внутрь диапазона
print(f"[{cf['time']}] SELL signal ⚠️ fake breakout up → {SYMBOL}")
return

# пробой вниз
if br["close"] < rng_low and br["volume"] < avg_vol*VOLUME_FACTOR:
if cf["close"] > rng_low: # возврат внутрь
print(f"[{cf['time']}] BUY signal fake breakout down → {SYMBOL}")
return

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

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

📌 Подпишись  Crypto Python❗️
👍7
ews.py
6.7 KB
📌Regime Shift Early Warning (EWS)🧑‍💻

Сенсор раннего предупреждения о смене рыночного режима: считает набор устойчивых метрик (волатильность, автокорреляция AR(1), херст H, “тяжёлые хвосты”, всплеск кросс-корреляций) и даёт сводный балл EWS. При превышении порога печатает алерт по тикеру.

Что даёт:💸

- Ранние сигналы «критического замедления» → возможная смена тренда/режима.

- Фильтр для включения/выключения стратегий (не торговать при высоком EWS).

- Кросс-секционный радар: синхронные сдвиги по рынку.

#инструмент

📌 Подпишись  Crypto Python❗️
🔥63
📌Market Microstructure Entropy (MME)🧑‍💻

📌 Суть:🚀

Это алгоритм, который в реальном времени вычисляет энтропию рыночных тиков — насколько рынок хаотичен или упорядочен.

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

1. Через ccxt или прямой WebSocket получаем поток сделок (trades).

2. Для каждого окна (например, 1000 последних сделок) строим распределение направлений и размеров сделок.

- 70% покупок и все одинакового размера → низкая энтропия (рынок «упорядочен», скорее идёт тренд).

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

3. Вычисляем Shannon entropy для распределения.

4. На основе этого подаём сигнал:

🔴 Низкая энтропия → лучше торговать по тренду.

🟢 Высокая энтропия → лучше использовать mean reversion (возврат к среднему).

5. Можно строить индикатор "энтропия → стратегия".
import time
import math
from collections import Counter, deque
import numpy as np
import ccxt

class MarketMicrostructureEntropy:
def __init__(self, window_size=1000, size_bins=None, low_thresh=0.4, high_thresh=0.75, ema_alpha=0.1):
self.window_size = window_size
self.trades = deque(maxlen=window_size)
self.size_bins = size_bins or [0.5, 1.0, 2.0, 5.0]
self.low_thresh = low_thresh
self.high_thresh = high_thresh
self.ema_alpha = ema_alpha
self.ema_entropy = None

def add_trade(self, size, side):
self.trades.append({"size": float(size), "side": side})

def _size_ratio(self):
sizes = np.array([t["size"] for t in self.trades], dtype=float)
if len(sizes) == 0:
return np.array([])
median = np.median(sizes) or 1e-9
return sizes / median

def _size_category(self, ratio):
for i, b in enumerate(self.size_bins):
if ratio <= b:
return i
return len(self.size_bins)

def compute_entropy(self, base=2.0):
n = len(self.trades)
if n == 0:
return 0.0, 0.0

ratios = self._size_ratio()
categories = []
for i, t in enumerate(self.trades):
cat_size = self._size_category(ratios[i])
categories.append((t["side"], cat_size))

cnt = Counter(categories)
probs = np.array(list(cnt.values()), dtype=float) / n
with np.errstate(divide="ignore", invalid="ignore"):
logs = np.log(probs) / np.log(base)
H = -np.sum(probs * logs)

M = 2 * (len(self.size_bins) + 1)
maxH = math.log(M) / math.log(base)
norm = H / maxH if maxH > 0 else 0.0

if self.ema_entropy is None:
self.ema_entropy = norm
else:
self.ema_entropy = (1 - self.ema_alpha) * self.ema_entropy + self.ema_alpha * norm

return H, norm

def get_signal(self):
if self.ema_entropy is None:
return "neutral"
if self.ema_entropy < self.low_thresh:
return "trend"
elif self.ema_entropy > self.high_thresh:
return "mean_reversion"
return "neutral"

def run_polling(exchange_id="binance", symbol="BTC/USDT", poll_interval=2.0, window_size=1000):
ex = getattr(ccxt, exchange_id)()
mme = MarketMicrostructureEntropy(window_size=window_size)

print(f"Polling {exchange_id} {symbol} every {poll_interval}s...")

while True:
try:
trades = ex.fetch_trades(symbol, limit=200)
for t in trades:
side = t.get("side", "buy")
size = t.get("amount", 0.0)
mme.add_trade(size=size, side=side)

H, norm = mme.compute_entropy()
sig = mme.get_signal()
print(f"[{time.strftime('%H:%M:%S')}] trades={len(mme.trades)} H={H:.4f} norm={norm:.4f} ema={mme.ema_entropy:.4f} signal={sig}")

except Exception as e:
print("Error:", e)

time.sleep(poll_interval)


if __name__ == "__main__":
run_polling(exchange_id="binance", symbol="BTC/USDT", poll_interval=2.0, window_size=1000)

#инструмент

📌 Подпишись  Crypto Python❗️
👍82
arr.py
7.4 KB
📌"Асимметричный риск-реверс" (Asymmetric Risk-Reversal)🧑‍💻

🔑 Суть💸

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

- Падение обычно резкое и короткое.

- Рост более плавный и долгий.

Мы строим стратегию, где правила входа/выхода разные для лонга и для шорта.

📐 Логика:🚀

1. Лонг (основная часть стратегии):

Вход: после локальной консолидации (N свечей с низкой волатильностью).

Фильтр: тренд по EMA(100) вверх.

Выход: по тейку в 1.5–2 ATR.

2. Шорт (агрессивный, но короткий):

Вход: при всплеске волатильности (цена падает >2σ за свечу).

Выход: быстрый тейк = 0.5–1 ATR.

Стоп короткий (ниже последнего high).

3. Таким образом:

В лонгах мы "сидим на росте".

В шортах — "снимаем сливки с паники".

🎯 Почему работает:🧨

- Использует структурную асимметрию рынка.

- Даёт меньше убытков на ложных пробоях вниз.

- Хорошо комбинируется с фьючерсами и плечом.

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

📌 Подпишись  Crypto Python❗️
👍9
📌"Анализ уникальности торгового сигнала"🧑‍💻

Большинство стратегий в алготрейдинге страдают от переоптимизации (overfitting). Сигналы часто повторяются в "шуме", и бот гоняет туда-сюда без реального edge.

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

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

- Отсечь повторяющиеся сигналы без смысла.

- Искать реально редкие паттерны, где вероятность отработки выше.

- Добавить "фильтр качества" к любой стратегии.

🔧 Описание логики:🛠️

1. Собираем исторические данные OHLCV.

2. Вычисляем вектор признаков свечи / паттерна (волатильность, позиция цены в диапазоне, ATR, RSI, объём и т.д.).

3. Сравниваем текущий сигнал с историческими через метрику расстояния (cosine similarity / евклидово расстояние).

4. Если сигнал слишком похож на уже бывшие убыточные, игнорируем его.

5. Если сигнал уникален (редкий паттерн) — помечаем как "высокого качества".

import ccxt
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.metrics.pairwise import cosine_similarity

# ========== Параметры ==========
exchange = ccxt.binance()
symbol = "BTC/USDT"
timeframe = "1h"
limit = 500

# === Шаг 1: загрузка данных ===
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=["time", "open", "high", "low", "close", "volume"])
df["time"] = pd.to_datetime(df["time"], unit="ms")

# === Шаг 2: генерация признаков ===
df["return"] = df["close"].pct_change()
df["range"] = (df["high"] - df["low"]) / df["close"]
df["pos_in_range"] = (df["close"] - df["low"]) / (df["high"] - df["low"] + 1e-9)
df["volatility"] = df["return"].rolling(10).std()
df["volume_z"] = (df["volume"] - df["volume"].rolling(20).mean()) / df["volume"].rolling(20).std()

# оставляем чистые признаки
features = df[["return", "range", "pos_in_range", "volatility", "volume_z"]].dropna()

# === Шаг 3: нормализация ===
scaler = StandardScaler()
X = scaler.fit_transform(features)

# === Шаг 4: функция уникальности ===
def signal_uniqueness(current_idx, lookback=100):
"""
Возвращает "уникальность сигнала" от 0 до 1.
0 = полностью копия старых, 1 = абсолютно новый паттерн
"""
if current_idx < lookback:
return 0.5 # мало истории

current_vec = X[current_idx].reshape(1, -1)
history_vecs = X[current_idx - lookback: current_idx]

sims = cosine_similarity(current_vec, history_vecs)[0]
uniqueness = 1 - sims.max() # чем меньше сходство — тем больше уникальность
return float(uniqueness)

# === Шаг 5: применяем ===
df = df.iloc[len(df) - len(X):].copy()
df["uniqueness"] = [signal_uniqueness(i) for i in range(len(X))]

# === Пример: фильтр сигналов ===
threshold = 0.3 # ниже этого — сигнал игнорируется
df["signal"] = np.where(df["uniqueness"] > threshold, "TRADE", "IGNORE")

print(df[["time", "close", "uniqueness", "signal"]].tail(20))

#инструмент

📌 Подпишись  Crypto Python❗️
🔥73🥰2
mrs_mme_vwap.py
14.3 KB
📌MME VWAP Regime Switch (MRS)🧑‍💻

Ключ: сочетать микроструктурную энтропию (MME) из тиков с поведенческими паттернами дня (Opening Range) и средним притяжения (VWAP).

Режимы по энтропии:🧨

- Низкая энтропия (упорядоченность) → торгуем пробой/продолжение:

- После формирования Opening Range (первые OR_MIN минут) берём пробой ORH/ORL в сторону относительно VWAP.

- Высокая энтропия (хаос) → торгуем mean-reversion:

Когда цена отклоняется от VWAP больше, чем K * ATR(1m), играем возврат к VWAP.

Фильтры & выходы:💸

- В трендовом режиме — вход только если цена по сторону VWAP (для лонга выше, для шорта ниже).

- В MR-режиме — вход против отклонения, выход у VWAP или по стопу.

- Переключение режима (из-за MME) закрывает позицию.

- Стоп-лосс — кратный ATR.

- Торгуем 1 инструмент (по умолчанию BTC/USDT), сигналы только в консоль (без выставления ордеров).

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

📌 Подпишись  Crypto Python❗️
🔥5👍42
📌Индекс агрессивности ликвидности (Liquidity Aggression Index, LAI)🧑‍💻

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

🔧 Логика:🛠️

1. Берём свечу (OHLCV).

2. Считаем:

Объём покупок (свеча закрылась выше открытия → часть объёма считаем "покупки").

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

3. Строим индекс агрессивности: LAI

LAI → +1 = агрессивные покупки (бычий импульс)

LAI → -1 = агрессивные продажи (медвежий импульс)

LAI около 0 = рынок сбалансирован

4. Фильтруем сигналы стратегий: торгуем только в сторону агрессии ликвидности.

import ccxt
import pandas as pd
import numpy as np

# === Параметры ===
exchange = ccxt.binance()
symbol = "BTC/USDT"
timeframe = "15m"
limit = 500

# === Загрузка данных ===
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=["time", "open", "high", "low", "close", "volume"])
df["time"] = pd.to_datetime(df["time"], unit="ms")

# === Индекс агрессивности ===
def calc_lai(row):
if row["close"] > row["open"]:
buy_vol = row["volume"] * (row["close"] - row["open"]) / (row["high"] - row["low"] + 1e-9)
sell_vol = row["volume"] - buy_vol
else:
sell_vol = row["volume"] * (row["open"] - row["close"]) / (row["high"] - row["low"] + 1e-9)
buy_vol = row["volume"] - sell_vol

return (buy_vol - sell_vol) / (buy_vol + sell_vol + 1e-9)

df["LAI"] = df.apply(calc_lai, axis=1)

# === Генерация сигналов ===
threshold = 0.5 # сила агрессии
df["signal"] = np.where(df["LAI"] > threshold, "BUY",
np.where(df["LAI"] < -threshold, "SELL", "HOLD"))

print(df[["time", "close", "volume", "LAI", "signal"]].tail(20))

📊 Что даёт 💸

- Фильтр сделок: бот входит только если есть явный перевес агрессии.

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

- Можно комбинировать с любой стратегией (например, сигналы по ATR, но вход только если LAI согласуется).

#инструмент

📌 Подпишись  Crypto Python❗️
👍101❤‍🔥1
slsr.py
7.1 KB
📌«Squeeze + Liquidity Sweep Reversal»🧑‍💻

Идея: 🚀

Ищем моменты, когда рынок сжат (низкая волатильность), после чего цена делает «съём ликвидности» за пределами вчерашнего экстремума и возвращается обратно.

Вход 🚀 — в сторону возврата, подтверждённый локальной агрессией (LAI).

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

После сжатия участники копят позиции; прокол вчерашнего high/low собирает стопы и ликвидность. Возврат внутрь диапазона + агрессивный перекос объёма часто приводит к импульсу против прокола.

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

📌 Подпишись  Crypto Python❗️
👍8
atr_ladder.py
7.9 KB
📌ATR-Ladder: адаптивная лесенка лимиток 🧑‍💻

Идея:🚀

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

- Размер (геометрическое распределение риска, чтобы суммарный риск ≤ заданного процента депозита).

- TP и SL (в ATR-единицах), плюс оценка ожидаемого R.

- Экспортируем план в CSV/JSON

#инструмент

📌 Подпишись  Crypto Python❗️
👍5
sls.py
10.8 KB
📌Session Liquidity Sweep → VWAP Reversion🧑‍💻

Идея:🧨

Рынок часто «снимает ликвидность» за пределами максимумов/минимумов предыдущего дня (стоп-ханты), после чего цена возвращается к справедливой стоимости внутри дня — VWAP. Используем это: ждём прокола вчерашнего High/Low («sweep») и возврата внутрь диапазона → входим в сторону возврата, цель — интрадей VWAP.

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

За пределами High/Low вчерашнего дня скапливаются стоп-ордера — уклонение ликвидности часто даёт импульс в обратную сторону.

VWAP выступает «магнитом» справедливой цены в рамках сессии.

Правила (коротко)📝

Таймфрейм сигналов: 5m (можно 1m–15m).

Фильтр волатильности: 14-дневный ATR/Close ≥ 1.0%.

Short: текущая свеча сделала High выше вчерашнего High на ≥ τ (например, 0.05%), но закрылась ниже вчерашнего High → вход по close этой свечи, цель — VWAP дня, стоп — максимум этой свечи + buffer (например, 0.02%).

Long: зеркально при проколе вчерашнего Low и закрытии выше него.

Риск-менеджмент: доля капитала на сделку = min(max_risk, Kelly-лайт по скользящей винрейту). По умолчанию 0.5%–1%.

Выход: тейк по VWAP (частичный или полный), стоп по свече-сигналу. Опционально трейлинг по ATR.

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

📌 Подпишись  Crypto Python❗️
🔥3👍1
📌Volatility Compass 🧑‍💻

🔍 Идея: 🛠️

Вместо десятков индикаторов и чартов можно иметь один «компас», который показывает:

- направление (бычье/медвежье смещение)

- силу (на основе нормализованной волатильности)

- сразу для нескольких таймфреймов: 5m, 1h, 1d.

📌 Трейдеру достаточно взглянуть на «стрелку», чтобы понять:

- рынок синхронно смотрит вверх (конфлюэнс → тренд),

- или ТФ конфликтуют (осторожность / флэт).

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

1. Для каждого ТФ считаем:

σ (ATR%) — относительная волатильность.

EMA-баланс: close - EMA(20).

2. Комбинируем в вектор:

направление = знак(close - EMA).

сила = σ (нормализованная).

3. Рисуем компактный вывод:

↑↑↑ — все ТФ бычьи.

↓↓↑ — конфликт, короткосрок против старшего ТФ.

числовая «сила импульса» (0–100).

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

SYMBOL = "BTC/USDT"
TFS = ["5m", "1h", "1d"]
HISTORY_DAYS = 30
ATR_LEN = 14

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 ema(series, n=20):
return series.ewm(span=n, adjust=False).mean()

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

def analyze_tf(df, tf):
df["ema"] = ema(df["close"], 20)
df["atr"] = atr(df, ATR_LEN)
row = df.iloc[-1]
direction = 1 if row["close"] > row["ema"] else -1
strength = (row["atr"]/row["close"]) * 100 # ATR%
return {"tf": tf, "dir": direction, "strength": strength}

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

compass = []
for tf in TFS:
df = fetch_ohlcv(ex, SYMBOL, tf, since)
res = analyze_tf(df, tf)
compass.append(res)

# Вывод «компаса»
arrows = "".join(["↑" if r["dir"]>0 else "↓" for r in compass])
avg_strength = np.mean([r["strength"] for r in compass])
print(f"\n=== Volatility Compass for {SYMBOL} ===")
for r in compass:
print(f"{r['tf']:>3}: {'↑' if r['dir']>0 else '↓'} | strength={r['strength']:.2f}%")
print(f"\nCompass: {arrows} | Avg strength={avg_strength:.2f}%")

#инструмент

📌 Подпишись  Crypto Python❗️
🔥7
📌Time-Shift Momentum🧑‍💻

🔍 Идея🧨

Большинство стратегий берёт индикаторы на одном ТФ. Но если сдвинуть сигналы с более «медленного» таймфрейма (например, 1h) и использовать их как опережающий фильтр для более «быстрого» (например, 5m), можно ловить хорошие импульсы.

Пример:🚀

Берём RSI(14) на 1h и смотрим его состояние (перекуплен / перепродан).

Торгуем 5m свечи, но разрешаем сигналы только в сторону, согласованную с «будущим» направлением RSI.

То есть используем старший ТФ как фазовый фильтр, но на младшем — ловим вход.

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

1. Рассчитываем RSI(14) на 1h.

Если RSI < 35 → искать лонги.

Если RSI > 65 → искать шорты.

2. На 5m ищем импульс свечи: close > EMA(20) и рост объёма → вход в лонг (если старший RSI разрешает).
Аналогично для шорта.

3. Стоп: ниже минимума сигнальной свечи (лонг) / выше максимума (шорт).

4. Тейк: 2× стоп или трейлинг по EMA(20).

5. Фильтр волатильности: пропускаем сигналы, если ATR(14)/close < 0.5%.

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

SYMBOL = "BTC/USDT"
TF_FAST = "5m"
TF_SLOW = "1h"
HISTORY_DAYS = 30

def fetch_ohlcv(exchange, symbol, tf, since):
data = []
while True:
batch = exchange.fetch_ohlcv(symbol, tf, since, 1000)
if not batch:
break
data += batch
if len(batch) < 1000:
break
since = batch[-1][0] + 1
time.sleep(exchange.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 rsi(series, n=14):
delta = series.diff()
up = delta.clip(lower=0)
down = -delta.clip(upper=0)
roll_up = up.ewm(alpha=1/n, adjust=False).mean()
roll_down = down.ewm(alpha=1/n, adjust=False).mean()
rs = roll_up / (roll_down + 1e-9)
return 100 - (100 / (1 + rs))

def ema(series, n=20):
return series.ewm(span=n, adjust=False).mean()

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

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

df_fast = fetch_ohlcv(ex, SYMBOL, TF_FAST, since)
df_slow = fetch_ohlcv(ex, SYMBOL, TF_SLOW, since)

df_slow["rsi"] = rsi(df_slow["close"], 14)
df_fast["ema20"] = ema(df_fast["close"], 20)
df_fast["atr"] = atr(df_fast, 14)

# мэппинг RSI с 1h в 5m
df_fast["rsi_slow"] = df_slow["rsi"].reindex(df_fast.index, method="ffill")

signals = []
for t, row in df_fast.iterrows():
if pd.isna(row["rsi_slow"]) or pd.isna(row["ema20"]) or pd.isna(row["atr"]):
continue
vol_ok = row["atr"]/row["close"] >= 0.005 # 0.5% волатильность
if not vol_ok:
continue

# правила входа
if row["close"] > row["ema20"] and row["volume"] > df_fast["volume"].rolling(20).mean().loc[t]:
if row["rsi_slow"] < 35:
signals.append((t, "LONG", row["close"]))
if row["close"] < row["ema20"] and row["volume"] > df_fast["volume"].rolling(20).mean().loc[t]:
if row["rsi_slow"] > 65:
signals.append((t, "SHORT", row["close"]))

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

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

📌 Подпишись  Crypto Python❗️
🔥51👍1
📌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❗️
4👍3
📌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❗️
👍4🔥3
📌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❗️
2🔥2
dcor.py
9.8 KB
📌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❗️
👍4
📌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❗️
👍51
📌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❗️
🔥6
📌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❗️
👍7