Crypto Python
818 subscribers
448 photos
26 files
455 links
Алготрейдинг 🤖 , работа с API бирж и агрегаторов 🌐 , автоматизации в крипто сфере📈 🚀
Ваши предложения📝 @binance_de
Download Telegram
📌"Поведенческое зеркало толпы"-
Торговля против эмоциональных паттернов розничных трейдеров🧑‍💻

📌 Идея:🧨

Поведение толпы на рынке часто подвержено эмоциям — паника в просадке и жадность при росте. Эта стратегия использует метрики соцсигналов и объёмов в сочетании с волатильностью и RSI, чтобы идти против эмоционального перегрева.

🔍 Логика:🛠️

Индикатор перегретости толпы = spike в упоминаниях актива + spike в объёмах + RSI > 70 (или RSI < 30 для паники)

При перегреве — открывается контртрендовая позиция:

- если рост на эмоциях — открытие шорта

- если паника на дне — открытие лонга

💡 Используемые данные:

📢 Соцсигналы: число упоминаний в Twitter, Reddit, Telegram (через API, например LunarCrush, Santiment, CrowdTangle)

📊 Объёмы: рост объёма спота/фьючей

📉 RSI: стандартный индикатор

🔄 Волатильность: ATR, либо std dev цены за период

import ccxt
import time
from datetime import datetime
import random # Имитация соц.сигналов

exchange = ccxt.binance()
symbol = 'BTC/USDT'
rsi_threshold_high = 70
rsi_threshold_low = 30

def get_mock_social_hype_score():
# Здесь должен быть реальный парсинг с API
return random.randint(0, 100)

def calculate_rsi(prices, period=14):
deltas = [prices[i+1]-prices[i] for i in range(len(prices)-1)]
ups = [delta for delta in deltas if delta > 0]
downs = [-delta for delta in deltas if delta < 0]
avg_gain = sum(ups[-period:])/period if ups else 0.001
avg_loss = sum(downs[-period:])/period if downs else 0.001
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
return rsi

while True:
candles = exchange.fetch_ohlcv(symbol, timeframe='5m', limit=100)
close_prices = [c[4] for c in candles]
rsi = calculate_rsi(close_prices)
social_score = get_mock_social_hype_score()

if rsi > rsi_threshold_high and social_score > 80:
print(f"[{datetime.now()}] ⚠️ ПЕРЕГРЕВ! RSI={rsi:.2f}, Hype={social_score} — сигнал на ШОРТ")
elif rsi < rsi_threshold_low and social_score > 80:
print(f"[{datetime.now()}] 😱 ПАНИКА! RSI={rsi:.2f}, Hype={social_score} — сигнал на ЛОНГ")
else:
print(f"[{datetime.now()}] 🕒 RSI={rsi:.2f}, Hype={social_score} — нет сигнала")

time.sleep(60)

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

📌 Подпишись  Crypto Python❗️
🔥4👍3
📌"Fakeout Hunter" (Охотник за ложными пробоями)🧑‍💻

💡 Идея:

Многие трейдеры теряют деньги на пробоях, которые оказываются ложными. Стратегия “Fakeout Hunter” строится на поиске и входе в рынок в момент, когда цена возвращается обратно после ложного пробоя уровня.

🧠 Логика:🛠️

1. Находим локальные экстремумы (уровни поддержки/сопротивления).

2. Отслеживаем пробой уровня, за которым не следует импульс.

3. Ждём возврата под уровень (или над уровень, если это ложный пробой вниз).

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

5. Фильтруем по объёму и ATR — нужна высокая волатильность.

📊 Пример условий входа:💸

- Цена пробила уровень сопротивления, но закрытие свечи ниже уровня → ложный пробой

- Объём выше среднего → подтверждение интереса

- RSI в зоне перекупленности → усиливает сигнал на шорт

import ccxt
import pandas as pd
import time

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

def fetch_ohlcv():
data = exchange.fetch_ohlcv(symbol, timeframe, limit=100)
df = pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
return df

def detect_fakeout(df):
last = df.iloc[-1]
prev = df.iloc[-2]

resistance = df['high'][-20:-1].max()
support = df['low'][-20:-1].min()

fakeout_up = prev['high'] > resistance and last['close'] < resistance
fakeout_down = prev['low'] < support and last['close'] > support

if fakeout_up:
print(f"[{last['timestamp']}] 🚨 Fakeout вверх! Потенциал шорта: {symbol}")
elif fakeout_down:
print(f"[{last['timestamp']}] 🚨 Fakeout вниз! Потенциал лонга: {symbol}")

while True:
df = fetch_ohlcv()
detect_fakeout(df)
time.sleep(60)

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

📌 Подпишись  Crypto Python❗️
👍32
📌Risk Manager CLI — Консольный помощник по управлению рисками🧑‍💻

📌 Что это?

Это консольное приложение на Python, которое помогает трейдеру автоматически рассчитывать размер позиции, расставлять цели и показывать риск/профит-профиль до входа в сделку.

Работает с клавиатуры — вводишь цену входа, стоп и риск в %, а инструмент показывает, сколько нужно купить, на какой объём, где тейк и каков RR (risk/reward).

🧠 Для чего нужен:🚀

- Избавляет от ручных расчётов при открытии позиции

- Учит дисциплине и риск-менеджменту

- Особенно полезен при мануальной торговле и для тех, кто торгует от уровней

def risk_manager():
balance = float(input("💰 Введите сумму депозита (USDT): "))
risk_percent = float(input("📉 Риск на сделку (%) например 1: ")) / 100
entry_price = float(input("📥 Цена входа: "))
stop_price = float(input("🛑 Стоп-лосс: "))
take_profit = float(input("🎯 Тейк-профит: "))

risk_amount = balance * risk_percent
stop_distance = abs(entry_price - stop_price)
position_size = risk_amount / stop_distance
position_value = position_size * entry_price
potential_profit = abs(take_profit - entry_price) * position_size
rr_ratio = potential_profit / risk_amount

print("\n📊 Результаты:")
print(f"🔹 Риск на сделку: {risk_amount:.2f} USDT")
print(f"🔹 Размер позиции: {position_size:.4f} BTC")
print(f"🔹 Объём позиции: {position_value:.2f} USDT")
print(f"🔹 Потенциальная прибыль: {potential_profit:.2f} USDT")
print(f"🔹 Соотношение риск/прибыль: 1:{rr_ratio:.2f}")

if __name__ == "__main__":
risk_manager()

#инструмент

📌 Подпишись  Crypto Python❗️
👍43
📌"Алгоритм охоты за ликвидностью" (Liquidity Sweep Sniper)🧑‍💻

🧠 Идея:🚀

Стратегия ориентирована на поведение крупных участников (смарт-денег), которые охотятся за стопами розницы. Мы ищем зоны ложного пробоя локальных экстремумов (high/low), где рынок "съедает" ликвидность, и заходим против направления прокола после подтверждения.

📊 Сценарий входа (лонг):💸

1. Цена пробивает локальный минимум → забирает стопы → формируется хвост (пинбар, молот).

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

3. Вход в лонг по закрытию подтверждающей свечи.

4. Стоп — под ложный прокол, Тейк — в область предыдущей консолидации или по R:R 1:2+.

🤖 Особенности:💻

- Работает на любом таймфрейме от 5m до 4H, особенно хорошо в азиатскую сессию.

- Не требует индикаторов, только цена и объёмы (если есть).

- Подходит как для мануального, так и алготрейдинга.

- Можно расширить до автоопределения "стопохантов".

Преимущества:🧨

- Заходит в момент смены дисбаланса (где много ликвидности)

- Позволяет точно выставлять стоп

- Даёт высокое соотношение риск/прибыль (1:2 — 1:5)

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '5m'

def get_candles(symbol, tf='5m', limit=50):
ohlcv = exchange.fetch_ohlcv(symbol, tf, limit=limit)
return [{'time': c[0], 'open': c[1], 'high': c[2], 'low': c[3], 'close': c[4]} for c in ohlcv]

def detect_liquidity_sweep(candles):
last = candles[-1]
prev = candles[-2]

# Условие: был ложный прокол минимума
if last['low'] < prev['low'] and last['close'] > prev['low']:
print("🟢 ВОЗМОЖЕН ЛОНГ (поглощение ликвидности вниз)")
print(f"Цена: {last['close']}")
elif last['high'] > prev['high'] and last['close'] < prev['high']:
print("🔴 ВОЗМОЖЕН ШОРТ (поглощение ликвидности вверх)")
print(f"Цена: {last['close']}")

def run_monitor():
print("🔍 Мониторинг ликвидности...")
while True:
try:
candles = get_candles(symbol, timeframe)
detect_liquidity_sweep(candles)
time.sleep(60) # каждую минуту
except Exception as e:
print(f"Ошибка: {e}")
time.sleep(60)

if __name__ == "__main__":
run_monitor()

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

📌 Подпишись  Crypto Python❗️
3👍2
📌Универсальный калькулятор риска и позиции с журналом сделок🧑‍💻

📌 Назначение:🛠️

Позволяет в режиме консоли:

- рассчитывать объём позиции и стоп-лосс в $ и %

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

- логировать каждую сделку в CSV

- собирать статистику: winrate, avg R, expectancy и др.

- быстро переключаться между биржами/активами

import csv
import datetime

def calculate_position(balance, risk_pct, entry_price, stop_price, leverage=1):
risk_usd = balance * (risk_pct / 100)
stop_loss_per_unit = abs(entry_price - stop_price)
position_size = (risk_usd / stop_loss_per_unit) * leverage
return round(position_size, 4), round(risk_usd, 2)

def log_trade(symbol, direction, entry, stop, tp, size, risk_usd, result=None):
with open('trade_journal.csv', 'a', newline='') as f:
writer = csv.writer(f)
writer.writerow([
datetime.datetime.now().strftime('%Y-%m-%d %H:%M'),
symbol, direction, entry, stop, tp, size, risk_usd, result
])

def main():
print("📊 Расчёт позиции и логгер сделок")
symbol = input("Тикер (например, BTC/USDT): ")
direction = input("Направление (long/short): ")
balance = float(input("Баланс аккаунта ($): "))
risk_pct = float(input("Риск на сделку (%): "))
entry = float(input("Цена входа: "))
stop = float(input("Цена стоп-лосса: "))
tp = float(input("Тейк-профит (по желанию, можно 0): "))
leverage = float(input("Плечо (по умолчанию 1): ") or 1)

size, risk_usd = calculate_position(balance, risk_pct, entry, stop, leverage)
print(f"👉 Объём позиции: {size} единиц актива\n📉 Риск в долларах: ${risk_usd}")

log_trade(symbol, direction, entry, stop, tp, size, risk_usd)
print(" Сделка записана в журнал (trade_journal.csv)")

if __name__ == "__main__":
main()

#инструмент

📌 Подпишись  Crypto Python❗️
🔥5
📌Индикаторная флуктуация + импульсный фильтр на объёме🧑‍💻

🎯 Идея:🧨

Комбинируется флуктуация RSI в нейтральной зоне (45–55) с импульсным скачком объёма, что говорит о потенциальной скорой смене направления тренда. Это контр-интуитивная стратегия, основанная не на перекупленности/перепроданности, а на “вялой зоне”, из которой рынок выходит резко.

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

1. RSI(14) пересекает снизу вверх уровень 50

2. RSI находится в диапазоне 45–55 не менее 20 свечей (накапливается энергия)

3. Происходит скачок объёма (Volume > SMA(20) * 1.8)

4. Подтверждение: цена пробивает локальный high последних 10 свечей

💥 Вход на пробой high
📍 SL — под low накопления
🎯 TP1 — R=1.5, TP2 — трейлинг или exit по RSI>70

import ccxt
import pandas as pd
import ta

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '1h'
limit = 100

ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['rsi'] = ta.momentum.RSIIndicator(close=df['close']).rsi()
df['vol_sma'] = df['volume'].rolling(20).mean()

# Условия
rsi_range = (df['rsi'] > 45) & (df['rsi'] < 55)
df['in_range'] = rsi_range.rolling(20).apply(lambda x: x.all(), raw=True)
df['volume_spike'] = df['volume'] > df['vol_sma'] * 1.8
df['signal'] = (df['rsi'].shift(1) < 50) & (df['rsi'] > 50) & df['in_range'] & df['volume_spike']

signals = df[df['signal'] == True]
print(signals[['timestamp', 'close', 'rsi', 'volume']].tail())

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

📌 Подпишись  Crypto Python❗️
🔥7
"Тепловая карта стакана (Order Book Heatmap) в терминале"🧑‍💻

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

Показывает визуализацию лимитных ордеров в стакане — где стоят крупные заявки, на каких ценовых уровнях есть кластеры. Это помогает видеть уровни сопротивления/поддержки, спуфинг, а также реакции на крупные лимитные ордера.

import ccxt
import time
import numpy as np
import os

exchange = ccxt.binance()
symbol = 'BTC/USDT'
depth_limit = 50  # Глубина стакана

def fetch_orderbook():
    order_book = exchange.fetch_order_book(symbol, limit=depth_limit)
    bids = order_book['bids']
    asks = order_book['asks']
    return bids, asks

def render_heatmap(bids, asks):
    os.system('cls' if os.name == 'nt' else 'clear')
    print(f"🔥 Order Book Heatmap: {symbol}\n")

    # Конвертация объема в "яркость" символов
    def format_line(side, color):
        for price, volume in side:
            intensity = int(min(volume, 5) * 10)  # Ограничить яркость
            bar = "#" * intensity
            print(f"{color}{price:>12.2f} {bar:<50} ({volume:.3f})\033[0m")

    print("🟥 Asks:")
    format_line(reversed(asks[:depth_limit]), "\033[91m")  # Красный

    print("\n🟩 Bids:")
    format_line(bids[:depth_limit], "\033[92m")  # Зелёный

while True:
    bids, asks = fetch_orderbook()
    render_heatmap(bids, asks)
    time.sleep(3)

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

#инструмент
5
📌"Противоимпульсный выброс (Antipump Trap)"🧑‍💻

📌 Идея:🚀

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

📊 Условия входа в шорт:

1. Свеча (1m или 5m) с телом > 2× среднего тела за последние 20 свечей.

2. Объём на свече > 2× среднего объёма.

3. Следующая свеча закрывается ниже 50% тела предыдущей (не подтверждён импульс).

4. Открытие шорт-позиции со стопом чуть выше хая импульсной свечи.

🎯 TP/SL:

TP: 1.5× риск

SL: выше хая импульсной свечи

Дополнительный выход при росте объёма на падении (фиксация прибыли)

import ccxt
import pandas as pd
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
tf = '5m'

def fetch_ohlcv():
data = exchange.fetch_ohlcv(symbol, timeframe=tf, limit=100)
df = pd.DataFrame(data, columns=['time', 'open', 'high', 'low', 'close', 'volume'])
df['body'] = abs(df['close'] - df['open'])
return df

def detect_antipump(df):
avg_body = df['body'][:-2].tail(20).mean()
avg_vol = df['volume'][:-2].tail(20).mean()

last = df.iloc[-2]
confirm = df.iloc[-1]

if last['body'] > 2 * avg_body and last['volume'] > 2 * avg_vol:
body_mid = (last['open'] + last['close']) / 2
if confirm['close'] < body_mid:
print(f"🔻 Antipump сигнал на шорт ({symbol}) @ {confirm['close']}")
print(f" SL: {last['high']}, TP: ~{confirm['close'] - (last['high'] - confirm['close'])*1.5}")

while True:
df = fetch_ohlcv()
detect_antipump(df)
time.sleep(15)

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

📌 Подпишись  Crypto Python❗️
🔥4
📌Автоматический Дневной Отчёт по Портфелю и Рынку🧑‍💻

Назначение:🧨

Каждое утро (или в выбранное время) скрипт автоматически собирает данные по рынку, активам, новостям, ключевым уровням и составляет краткий отчёт в консоль или в Telegram. Это личный дайджест трейдера — как утренний брифинг на десктопе.

Функции:🛠️

- Портфель и изменение за сутки (% и $)

- Текущие уровни поддержки/сопротивления (по ATR/каналу/SR)

- Волатильность по ключевым монетам

- Горячие монеты дня (по объёму и импульсу)

- Важные экономические события (например, CPI, FOMC — можно через календарь событий)

- Сводка по рынку: BTC, ETH, фондовый индекс, золото

import ccxt
import datetime

exchange = ccxt.binance()
symbol_list = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT']

def fetch_summary(symbol):
ohlcv = exchange.fetch_ohlcv(symbol, timeframe='1d', limit=2)
today = ohlcv[-1]
yesterday = ohlcv[-2]

change = (today[4] - yesterday[4]) / yesterday[4] * 100
return {
'pair': symbol,
'close': today[4],
'change': round(change, 2)
}

def daily_report():
print(f"📅 Daily Market Summary: {datetime.date.today()}")
for sym in symbol_list:
summary = fetch_summary(sym)
print(f"🔹 {summary['pair']}: ${summary['close']} ({summary['change']}%)")

if __name__ == "__main__":
daily_report()

💡 Расширения:

- Добавить новости через API (например, CryptoPanic)

- Интеграция с Telegram Bot API

- Хранение истории в SQLite/CSV

- Графическое отображение динамики портфеля (через matplotlib или plotly)

- Оповещение, если дневное изменение > 5%

#инструмент

📌 Подпишись  Crypto Python❗️
🔥32👌2
📌Обратная Точка Баланса” (Reverse Balance Point)🧑‍💻

💡 Идея:🚀

Внутридневные и краткосрочные трейдеры часто используют уровни VWAP или среднюю цену как "баланс". Эта стратегия делает противоположное — ищет моменты, когда цена слишком долго избегает баланса, и ловит возможность возврата к средней цене.

⚙️ Условия входа (LONG):

1. Цена открылась и удерживается выше VWAP как минимум 2 часа.

2. После сильного роста (>1.5% от VWAP) цена резко уходит ниже VWAP на повышенном объёме.

3. Вход в LONG на первом отскоке от нижнего стандартного отклонения VWAP (или уровней -1/-2 SD).

4. Цель — возврат к VWAP, Stop — под предыдущий минимум.

📉 SHORT:

То же самое, но в зеркальном отражении (сначала удержание под VWAP, затем пробой вверх и возврат).

import ccxt
import pandas as pd
import numpy as np

exchange = ccxt.binance()
symbol = 'BTC/USDT'
timeframe = '15m'
limit = 120 # 30 часов

ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['time', 'open', 'high', 'low', 'close', 'volume'])
df['time'] = pd.to_datetime(df['time'], unit='ms')

# VWAP
df['typical_price'] = (df['high'] + df['low'] + df['close']) / 3
df['cum_tp_vol'] = (df['typical_price'] * df['volume']).cumsum()
df['cum_vol'] = df['volume'].cumsum()
df['vwap'] = df['cum_tp_vol'] / df['cum_vol']

# Отношение к VWAP
df['distance'] = (df['close'] - df['vwap']) / df['vwap'] * 100

# Сигнал на возврат к балансу
last = df.iloc[-1]
prev = df.iloc[-2]

if prev['distance'] > 1.5 and last['distance'] < 0:
print("🔁 Возможен возврат к VWAP: Рассмотреть LONG")

elif prev['distance'] < -1.5 and last['distance'] > 0:
print("🔁 Возможен возврат к VWAP: Рассмотреть SHORT")
else:
print(" Нет сигнала. Ждём аномалии")

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

📌 Подпишись  Crypto Python❗️
👍3🔥1
📌Маркет-сканер волатильности с кластеризацией аномалий🧑‍💻

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

Этот инструмент в реальном времени анализирует десятки монет (например, с Binance через ccxt) и ищет аномальные ценовые или объёмные всплески, применяя кластеризацию (например, DBSCAN или KMeans).

Он не просто говорит: "Цена выросла на 4%", а выявляет группы активов, которые ведут себя одинаково, но резко отличаются от других.

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

- Находить монеты в начале пампа.

- Выявлять нелогичные движения (арбитражные окна).

- Искать сигналы толпы / инсайдерского спроса.

import ccxt
import pandas as pd
from sklearn.cluster import DBSCAN
from sklearn.preprocessing import StandardScaler
import numpy as np

exchange = ccxt.binance()
symbols = [s['symbol'] for s in exchange.load_markets().values() if '/USDT' in s['symbol'] and s['active']]

data = []

for symbol in symbols[:50]:
try:
ohlcv = exchange.fetch_ohlcv(symbol, '5m')[-3:]
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
pct_change = (df['close'].iloc[-1] - df['close'].iloc[0]) / df['close'].iloc[0]
vol_change = df['volume'].iloc[-1] - df['volume'].iloc[0]
data.append([symbol, pct_change, vol_change])
except:
continue

df = pd.DataFrame(data, columns=['symbol', 'pct_change', 'vol_change'])

# Кластеризация
X = StandardScaler().fit_transform(df[['pct_change', 'vol_change']])
clustering = DBSCAN(eps=0.8, min_samples=3).fit(X)
df['cluster'] = clustering.labels_

# Выводим аномалии
anomalies = df[df['cluster'] == -1]
print("📊 Обнаружены аномалии:\n", anomalies)

#инструмент

📌 Подпишись  Crypto Python❗️
🔥71
📌Алгокарта кросс-корреляций в реальном времени🧑‍💻

📌 Что это?🧨

Скрипт, строящий динамическую тепловую карту кросс-корреляций между несколькими криптоактивами (например, BTC, ETH, SOL, BNB и т.д.) в реальном времени — на малых таймфреймах (1m/5m), и обновляющий её в терминале или графически.

🧠 Как помогает трейдеру/алготрейдеру?💸

Находит активы, которые ведут себя синхронно или наоборот расходятся, что важно:

- Для парного трейдинга

- Для арбитражных стратегий

- Для фильтрации сигналов

- Для контроля риска по скоррелированным активам

import ccxt
import pandas as pd
import time
from colorama import Fore, Style, init

init(autoreset=True)

symbols = ['BTC/USDT', 'ETH/USDT', 'BNB/USDT', 'SOL/USDT', 'XRP/USDT']
exchange = ccxt.binance()
timeframe = '1m'
limit = 100

def fetch_ohlcv(symbol):
try:
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
return df['close']
except Exception as e:
print(f"[{symbol}] Ошибка: {e}")
return pd.Series(dtype=float)

def print_correlation_matrix(data):
df = pd.DataFrame(data)
corr_matrix = df.corr()

print("\nКросс-корреляция:")
headers = corr_matrix.columns
print(" " + " ".join([f"{h[:4]}" for h in headers]))

for i, row in corr_matrix.iterrows():
row_str = f"{i[:4]} "
for val in row:
if val > 0.75:
row_str += Fore.GREEN + f"{val:.2f}" + Style.RESET_ALL + " "
elif val < -0.75:
row_str += Fore.RED + f"{val:.2f}" + Style.RESET_ALL + " "
else:
row_str += f"{val:.2f} "
print(row_str)

if __name__ == "__main__":
data = {}
for symbol in symbols:
closes = fetch_ohlcv(symbol)
if not closes.empty:
data[symbol] = closes
time.sleep(0.5)

if len(data) >= 2:
print_correlation_matrix(data)
else:
print("Недостаточно данных.")

#инструмент

📌 Подпишись  Crypto Python❗️
3👍2
📌Обратный эхолот ордербука🧑‍💻

🧠 Идея:🚀

Большие лимитные заявки в ордербуке часто используются не для исполнения, а для манипуляции рынком — так называемые "спуф-ордера". Эта стратегия выявляет такие аномалии в ордербуке и торгует в противоположную сторону манипуляции, когда "эхо" исчезает.

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

1. Сканируем стакан (order book) по выбранной паре с определённой частотой (например, каждые 5 секунд).

2. Ищем крупные заявки (например, 5x выше среднего).

3. Проверяем, исчезают ли они внезапно — это может быть признаком попытки манипулировать.

4. При исчезновении крупной заявки — открываем позицию против этого "давления":

Если исчез крупный buy — открываем short.

Если исчез крупный sell — открываем long.

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

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

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
threshold_multiplier = 5

def fetch_order_book(symbol):
ob = exchange.fetch_order_book(symbol, limit=100)
return ob['bids'], ob['asks']

def detect_spoof_disappearance(prev, curr):
def large_orders_diff(old, new):
large = []
avg_size = sum([x[1] for x in old]) / len(old)
for price, size in old:
if size > avg_size * threshold_multiplier and not any(abs(price - p[0]) < 1e-4 for p in new):
large.append((price, size))
return large
disappeared_bids = large_orders_diff(prev['bids'], curr['bids'])
disappeared_asks = large_orders_diff(prev['asks'], curr['asks'])
return disappeared_bids, disappeared_asks

last_ob = {'bids': [], 'asks': []}
print(" Мониторинг ордербука...")

while True:
try:
bids, asks = fetch_order_book(symbol)
if last_ob['bids'] and last_ob['asks']:
lost_bids, lost_asks = detect_spoof_disappearance(last_ob, {'bids': bids, 'asks': asks})
if lost_bids:
print(f"⚠️ Исчезли крупные заявки BUY: {lost_bids} → сигнал на SHORT")
if lost_asks:
print(f"⚠️ Исчезли крупные заявки SELL: {lost_asks} → сигнал на LONG")
last_ob = {'bids': bids, 'asks': asks}
except Exception as e:
print("Ошибка:", e)
time.sleep(5)

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

📌 Подпишись  Crypto Python❗️
3🔥2👍1
📌"Гиперволатильный ретест"

(авторская идея, направлена на поиск резких выбросов цены и обратного возврата к уровню — со встроенным фильтром на тренд)🧑‍💻

Суть стратегии:🛠️

1. Вход:

Когда происходит резкий импульс вверх или вниз (например, > 1.5% за 1 минуту), а через 1-3 свечи происходит откат обратно к предыдущему уровню (ретест) — открывается сделка в направлении возврата (на откат).

2. Фильтр тренда:

Стратегия работает только при наличии сильного направленного движения за последние 30 минут (например, изменение > 2.5%). То есть, торгует только в условиях тренда.

3. Индикаторы (опционально):

- EMA(20) и EMA(50) для фильтра тренда.

- ATR для оценки силы импульса.

- Отслеживание свечных паттернов "Pin Bar" или "Engulfing" на ретесте — как дополнительное подтверждение.

import ccxt
import pandas as pd
import time

symbol = 'BTC/USDT'
timeframe = '1m'
limit = 100

exchange = ccxt.binance()

def fetch_ohlcv():
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['returns'] = df['close'].pct_change()
df['ema20'] = df['close'].ewm(span=20).mean()
df['ema50'] = df['close'].ewm(span=50).mean()
df['atr'] = df['high'] - df['low']
return df

def detect_signal(df):
last = df.iloc[-1]
prev = df.iloc[-2]

# Тренд фильтр: EMA20 > EMA50
in_trend = last['ema20'] > last['ema50']
recent_movement = (df['close'].iloc[-1] - df['close'].iloc[-30]) / df['close'].iloc[-30] * 100

if in_trend and recent_movement > 2.5:
# Импульс был 2 свечи назад
impulse = df['returns'].iloc[-3]
retest = df['returns'].iloc[-1]

if impulse > 0.015 and retest < -0.01:
print("🔽 Short signal detected (Impulse up → Retest down)")

elif impulse < -0.015 and retest > 0.01:
print("🔼 Long signal detected (Impulse down → Retest up)")

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

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

📌 Подпишись  Crypto Python❗️
👍6
📌"AlgoMarketMap" — Карта активности и плотности алгоритмической торговли в реальном времени🧑‍💻

🔍 Суть инструмента:🚀

Визуализатор и анализатор мест концентрации алготорговой активности (ботов) по:

- частоте обновлений ордербука;

- всплескам объёма на уровнях;

- микродвижениям цены без объёма;

- аномально точным лимиткам (в миллисекунды);

- частым сменам позиции в стакане.

💡 Зачем это нужно:💸

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

- Позволяет избегать ложных пробоев, созданных маркетмейкерами-ботами.

- Используется для построения контр-алгоботов и пассивного арбитража.

import asyncio
import websockets
import json
import time
from collections import deque

symbol = "btcusdt"
depth_url = f"wss://stream.binance.com:9443/ws/{symbol}@depth"

event_times = deque(maxlen=100)

async def monitor_depth():
async with websockets.connect(depth_url) as websocket:
while True:
try:
msg = await websocket.recv()
data = json.loads(msg)

now = time.time()
event_times.append(now)

# Оцениваем активность обновлений
if len(event_times) > 10:
freq = len(event_times) / (event_times[-1] - event_times[0])
if freq > 30:
print(f"⚠️ Алготрейдерская активность обнаружена! Частота: {freq:.2f} обновлений/сек")

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

asyncio.run(monitor_depth())

#инструмент

📌 Подпишись  Crypto Python❗️
🔥4
📌"Delta Drift Reversal" — Интегральный разворот по дельте объёма и микроимпульсам🧑‍💻

💡 Идея стратегии:🛠️

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

📌 Используется только рыночная информация: стакан, лента, объём и цена.

🔬 Условия входа (LONG):💸

1. На кластерной ленте видно:

Сильное агрессивное давление продавцов (дельта < -X);

Но цена не пробивает минимум, остаётся в диапазоне → затухание импульса.

2. В течение последних N минут:

Был хотя бы 1 мощный толчок вниз, но объём при повторном тесте ниже среднего.

3. Формируется "антиимпульс" — небольшое поглощение (delta становится положительной, цена идёт вверх).

4. Цена выше скользящего VWAP или среднего баланса за сессию.

🟢 → Вход в LONG

🔴 → Аналогично работает и в SHORT при обратной конфигурации.

📊 Фильтры:💻

- Не торгуется в низковолатильные сессии (определяется по ATR).

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

- Выход по Take Profit = x2 от ближайшего скопления ликвидности (по дельте)
или после формирования противоположного "антиимпульса".

import ccxt
import time

exchange = ccxt.binance()
symbol = 'BTC/USDT'
window = 5 # в минутах

def fetch_volume_delta(symbol, minutes=5):
ohlcv = exchange.fetch_ohlcv(symbol, '1m', limit=minutes)
delta = 0
for candle in ohlcv:
open_, high, low, close, volume = candle[1:6]
direction = 1 if close > open_ else -1
delta += volume * direction
return delta

while True:
delta = fetch_volume_delta(symbol, window)
print(f"Δ объёма за {window} минут: {delta:.2f}")

if delta < -1500:
print("⚠️ Мощная распродажа! Следим за затуханием импульса...")
elif delta > 1500:
print("⚠️ Активация покупателей! Возможен отскок.")
else:
print("Нейтральное давление...")

time.sleep(60)

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

📌 Подпишись  Crypto Python❗️
🔥5🤔1
📌SmartVol Scanner — это инструмент, который отслеживает кластерные накопления низкой волатильности (волатильные "зажимы") и прогнозирует вероятное направление разрядки на основе🧑‍💻:

- поведения тикового объёма,

- сужения диапазона (волатильность),

- направления агрессии (дельта),

- времени суток (часы активности),

- соотношения лимитных и рыночных ордеров.

⚙️ Что делает инструмент:🛠️

1. Находит “волатильные зажимы” — периоды, когда диапазон сужается (low volatility).

2. Отслеживает накопление объёма в этих узких диапазонах (squeeze zones).

3. Пытается предсказать, в какую сторону будет "выброс" (breakout):

- по агрессии дельты,

- по направлению объёма,

- по плотностям в стакане (опционально).

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

5. Идеален для алготрейдинговых моделей на пробой/отбой и построения сеток.

import ccxt
import time
import datetime
import statistics

symbol = 'BTC/USDT'
exchange = ccxt.binance({'enableRateLimit': True})
timeframe = '1m'
lookback_candles = 30
squeeze_threshold = 0.3 # < 0.3% средний диапазон — зажим
volume_growth_threshold = 1.5 # объем вырос в 1.5x
delay_sec = 10

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

def analyze(ohlcv):
ranges = [(c[2] - c[3]) / c[1] * 100 for c in ohlcv if c[1] > 0]
avg_range = statistics.mean(ranges)
volumes = [c[5] for c in ohlcv]
avg_volume = statistics.mean(volumes)
latest_volume = volumes[-1]
return avg_range, avg_volume, latest_volume

def detect(avg_range, avg_volume, latest_volume):
if avg_range < squeeze_threshold:
if latest_volume > avg_volume * volume_growth_threshold:
return " Всплеск объёма после зажима: потенциальный импульс!"
return "🕵️ Зажим волатильности. Объём пока обычный."
return " Волатильность нормальная. Сигналов нет."

print("📡 SmartVol Scanner запущен...")

while True:
try:
candles = fetch_candles()
avg_range, avg_volume, last_volume = analyze(candles)
signal = detect(avg_range, avg_volume, last_volume)
print(f"[{datetime.datetime.now()}] Диапазон: {avg_range:.2f}% | Объём: {last_volume:.2f} | {signal}")
except Exception as e:
print(f"Ошибка: {e}")
time.sleep(delay_sec)

#инструмент

📌 Подпишись  Crypto Python❗️
2👍2
📌"Order Book Mismatch Pressure"🧑‍💻

Идея:🛠️

Вместо технических индикаторов используется дисбаланс в лимитных заявках на покупку/продажу. Стратегия выявляет, когда маркет-участники массово "переставляют" стакан в сторону покупки или продажи — но цена ещё не среагировала. Это момент опережающего входа.

🔍 Условия входа в лонг:🚀

1. В стакане:

суммарный объём bid (покупка) > 3× объёма ask (продажа).

2. Последние 3 свечи — flat или слегка падающие (ещё нет роста).

3. Объёмы в стакане стабильны (нет спуфинга).

→ Вход в лонг по рынку или лимитке.

🛑 Выход:

Тейк: +0.5–1% (можно использовать трейлинг).

Стоп: ниже последнего локального минимума или фиксированный -0.3%.

Почему это эффективно:💸

- Стратегия не догоняет рынок, а опережает его, используя поведение других участников.

- Аналог предсказания импульса по изменению поведения стакана.

import ccxt
import time

# Конфигурация
SYMBOL = 'BTC/USDT'
THRESHOLD_MULTIPLIER = 3 # BID в 3 раза больше ASK = сигнал на лонг
SLEEP_INTERVAL = 5 # интервал между проверками в секундах

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

def fetch_order_book(symbol):
try:
book = exchange.fetch_order_book(symbol)
bid_volume = sum([bid[1] for bid in book['bids'][:10]])
ask_volume = sum([ask[1] for ask in book['asks'][:10]])
return bid_volume, ask_volume
except Exception as e:
print(f"[!] Ошибка получения стакана: {e}")
return None, None

def monitor_order_book(symbol, threshold):
while True:
bid_volume, ask_volume = fetch_order_book(symbol)
if bid_volume is None:
time.sleep(SLEEP_INTERVAL)
continue

if bid_volume > ask_volume * threshold:
print(f"[+] Сигнал на ЛОНГ: BID {bid_volume:.2f} > {threshold}× ASK {ask_volume:.2f}")
elif ask_volume > bid_volume * threshold:
print(f"[-] Сигнал на ШОРТ: ASK {ask_volume:.2f} > {threshold}× BID {bid_volume:.2f}")
else:
print(f"[=] Нет сигнала. BID: {bid_volume:.2f}, ASK: {ask_volume:.2f}")
time.sleep(SLEEP_INTERVAL)

monitor_order_book(SYMBOL, THRESHOLD_MULTIPLIER)

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

📌 Подпишись  Crypto Python❗️
🔥51
📌"Volatility Clustering Analyzer"
(Анализ кластеризации волатильности с сигналами на повышенную вероятность движения цены)🧑‍💻

📌 Описание идеи:💻

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

Смысл:🚀

- Выделяются «кластерные окна» резких движений.

- Когда появляется "затишье", модель отслеживает вероятность наступления нового импульса.

- Это можно использовать для входа в момент будущего движения цены (breakout или mean-reversion).

⚙️ Возможности инструмента:🛠️

- Расчёт текущего уровня кластеризации волатильности на основе ATR/стандартного отклонения.

- Автоматическое выявление фазы "сжатия" и сигнал о возможном выходе из неё.

- Генерация сигнала (в консоль или файл) при выходе за порог волатильности.

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

# Инициализация биржи
exchange = ccxt.binance({
'enableRateLimit': True,
})

symbol = 'BTC/USDT'
timeframe = '5m'
lookback_candles = 120
volatility_threshold = 1.5 # Множитель стандартного отклонения

def fetch_ohlcv(symbol, timeframe, limit=120):
try:
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
return df
except Exception as e:
print(f"Ошибка при получении данных: {e}")
return None

def calculate_volatility(df):
df['log_return'] = np.log(df['close'] / df['close'].shift(1))
df.dropna(inplace=True)
return df['log_return'].std() * np.sqrt(len(df)) * 100 # годовая волатильность в %

def main_loop():
print("🔁 Старт анализа волатильности... (Ctrl+C для остановки)\n")
prev_vol = None

while True:
df = fetch_ohlcv(symbol, timeframe, limit=lookback_candles)
if df is None:
time.sleep(10)
continue

current_vol = calculate_volatility(df)
ts = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')

if prev_vol is not None:
ratio = current_vol / prev_vol
if ratio > volatility_threshold:
print(f"[{ts}] 🚨 Повышение волатильности! (x{ratio:.2f}) Текущая волатильность: {current_vol:.2f}%")
else:
print(f"[{ts}] 📊 Волатильность стабильна: {current_vol:.2f}%")
else:
print(f"[{ts}] 🔹 Начальное измерение волатильности: {current_vol:.2f}%")

prev_vol = current_vol
time.sleep(60)

if __name__ == "__main__":
main_loop()

#инструмент

📌 Подпишись  Crypto Python❗️
👍41
📌Импульс с фильтром тренда” — твой фильтр шума рынка🧑‍💻

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

⚙️ Фильтры стратегии:🛠️

– EMA 50/200 → определяем тренд

– ADX > 25 → только сильное движение

– Пробой high/low на объёме

– RSI не в зоне перекупленности/перепроданности

Вход — строго по направлению тренда

Нет тренда? Нет сделки.

📊 Работает на крипте, фьючах, акциях.

🔁 Бэктест — положительный.

import ccxt
import pandas as pd
import ta
import time

# --- Параметры стратегии ---
symbol = 'BTC/USDT'
timeframe = '1h'
limit = 300

# --- Подключение к бирже ---
exchange = ccxt.binance()
exchange.load_markets()

# --- Получение исторических свечей ---
def get_ohlcv():
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
return df

# --- Расчёт индикаторов ---
def add_indicators(df):
df['ema50'] = ta.trend.ema_indicator(df['close'], window=50).ema_indicator()
df['ema200'] = ta.trend.ema_indicator(df['close'], window=200).ema_indicator()
df['adx'] = ta.trend.adx(df['high'], df['low'], df['close'], window=14).adx()
df['rsi'] = ta.momentum.RSIIndicator(df['close'], window=14).rsi()
df['atr'] = ta.volatility.AverageTrueRange(df['high'], df['low'], df['close'], window=14).average_true_range()
df['volume_ma'] = df['volume'].rolling(20).mean()
return df

# --- Логика сигналов ---
def get_signal(df):
latest = df.iloc[-1]
previous = df.iloc[-2]

if (
latest['ema50'] > latest['ema200'] and
latest['adx'] > 25 and
latest['close'] > previous['high'] and
latest['volume'] > latest['volume_ma'] and
latest['rsi'] < 70
):
return "🔼 LONG SIGNAL"

if (
latest['ema50'] < latest['ema200'] and
latest['adx'] > 25 and
latest['close'] < previous['low'] and
latest['volume'] > latest['volume_ma'] and
latest['rsi'] > 30
):
return "🔽 SHORT SIGNAL"

return None

# --- Основной цикл ---
def run():
while True:
try:
df = get_ohlcv()
df = add_indicators(df)
signal = get_signal(df)

last_time = df['timestamp'].iloc[-1]

if signal:
print(f"[{last_time}] {signal}")
else:
print(f"[{last_time}] Нет сигнала")

time.sleep(60 * 60) # Пауза на 60 минут

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

if __name__ == "__main__":
run()

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

📌 Подпишись  Crypto Python❗️
👍73
📌Smart Position Recovery Simulator (SPRS)🧑‍💻
Симулятор восстановления убыточных позиций по стратегиям усреднения и реверса с расчётом оптимальных шагов и оценки риска.

💡 Идея🚀

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

📈 Что делает инструмент:🛠️

1. Загружает исторические данные по цене (через ccxt или локально).

2. Позволяет задать начальный убыточный вход.

3. Строит сценарии восстановления позиции:

- шаговое усреднение,

- агрессивное удвоение,

- реверс с последующим возвратом.

import ccxt
import time
from datetime import datetime, timedelta

# === Настройки ===
api_key = 'YOUR_API_KEY'
api_secret = 'YOUR_SECRET_KEY'
symbol = 'BTC/USDT'
amount = 0.001
price_drop_threshold = 1.5 # %
price_rise_target = 2.0 # %
check_interval = 60 # сек

# === Инициализация биржи ===
exchange = ccxt.binance({
'apiKey': api_key,
'secret': api_secret,
'enableRateLimit': True,
})

# === История входов ===
entry_price = None

def get_recent_prices(symbol, minutes=5):
since = exchange.milliseconds() - minutes * 60 * 1000
candles = exchange.fetch_ohlcv(symbol, timeframe='1m', since=since)
return [c[4] for c in candles] # Закрытия

while True:
try:
print(f'\n[{datetime.now().strftime("%H:%M:%S")}] Проверка рынка...')
prices = get_recent_prices(symbol)
if len(prices) < 2:
print("Недостаточно данных.")
time.sleep(check_interval)
continue

current_price = prices[-1]
max_recent = max(prices)
price_drop = (max_recent - current_price) / max_recent * 100

if entry_price:
profit = (current_price - entry_price) / entry_price * 100
print(f"Актуальная цена: {current_price:.2f} | Прибыль: {profit:.2f}%")

if profit >= price_rise_target:
print("🎯 Цель достигнута. Продаем...")
exchange.create_market_sell_order(symbol, amount)
print(" Продажа выполнена.")
entry_price = None
else:
print(f"Актуальная цена: {current_price:.2f} | Падение за 5м: {price_drop:.2f}%")
if price_drop >= price_drop_threshold:
print("📉 Условие входа выполнено. Покупаем...")
exchange.create_market_buy_order(symbol, amount)
entry_price = current_price
print(f" Покупка по цене {entry_price:.2f}")

except Exception as e:
print(f"[Ошибка] {e}")

time.sleep(check_interval)

#инструмент

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