Crypto Python
818 subscribers
448 photos
26 files
455 links
Алготрейдинг 🤖 , работа с API бирж и агрегаторов 🌐 , автоматизации в крипто сфере📈 🚀
Ваши предложения📝 @binance_de
Download Telegram
📌 Powered Trading Bot – Бот с машинным обучением (XGBoost) 🧑‍💻

Смысл стратегии:

Используем XGBoost – одну из самых мощных библиотек машинного обучения для анализа ценовых данных.
Модель обучается на исторических данных, используя RSI, MACD, Bollinger Bands, Volume, ATR и другие индикаторы для предсказания направлений рынка (лонг / шорт).

Этапы работы:

1. Сбор данных (Binance, CCXT).

2. Формирование признаков (features) – технические индикаторы.

3. Обучение модели XGBoost на исторических данных.

4. Применение модели на новых данных для прогнозирования сигналов.

⚠️ Важно: Модель нуждается в постоянном обновлении данных и тестировании перед торговлей! 🚀

import ccxt
import pandas as pd
import talib
import time
import xgboost as xgb
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
TRAIN_SIZE = 200 # Количество свечей для обучения
FEATURES = ["rsi", "macd", "boll_upper", "boll_lower", "volume", "atr"]

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение исторических данных
def fetch_ohlcv(symbol, limit=TRAIN_SIZE + 50):
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=TIMEFRAME, limit=limit)
df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
return df

# Генерация признаков
def generate_features(df):
df["rsi"] = talib.RSI(df["close"], timeperiod=14)
macd, signal, _ = talib.MACD(df["close"], fastperiod=12, slowperiod=26, signalperiod=9)
df["macd"] = macd - signal
upper, _, lower = talib.BBANDS(df["close"], timeperiod=20, nbdevup=2, nbdevdn=2)
df["boll_upper"] = upper
df["boll_lower"] = lower
df["atr"] = talib.ATR(df["high"], df["low"], df["close"], timeperiod=14)
df.dropna(inplace=True)
return df

# Создание метки для предсказания
def generate_labels(df):
df["target"] = np.where(df["close"].shift(-1) > df["close"], 1, 0) # 1 = рост, 0 = падение
return df

# Обучение модели
def train_model(df):
X = df[FEATURES]
y = df["target"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, shuffle=False)

model = xgb.XGBClassifier(use_label_encoder=False, eval_metric="logloss")
model.fit(X_train, y_train)

y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)
print(f"🔹 Точность модели: {accuracy:.2f}")

return model

# Логика прогнозирования
def ai_trading_bot(model):
df = fetch_ohlcv(PAIR)
df = generate_features(df)
X_live = df[FEATURES].iloc[-1:].values
prediction = model.predict(X_live)[0]

print(f"🔹 Прогноз: {'📈 Лонг' if prediction == 1 else '📉 Шорт'}")

# Запуск
if __name__ == "__main__":
df = fetch_ohlcv(PAIR)
df = generate_features(df)
df = generate_labels(df)
model = train_model(df)

while True:
ai_trading_bot(model)
time.sleep(60) # Проверка раз в минуту

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

📌 Подпишись  Crypto Python❗️
🔥14👍3
📌 OBV Divergence Detector – Поиск дивергенций OBV

Смысл стратегии:

OBV (On-Balance Volume) – индикатор, анализирующий объем, чтобы предсказать будущие движения цены.

Бычья дивергенция → цена падает, но OBV растет → сигнал на покупку (лонг).

Медвежья дивергенция → цена растет, но OBV падает → сигнал на продажу (шорт).

🔹 Применение:

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

Работает лучше в сочетании с трендовыми индикаторами (например, MACD, RSI).

import ccxt
import pandas as pd
import talib
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
LOOKBACK_PERIOD = 10 # Количество свечей для анализа дивергенции
CHECK_INTERVAL = 60 # Проверка раз в минуту

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение данных
def fetch_ohlcv(symbol):
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=TIMEFRAME, limit=LOOKBACK_PERIOD + 5)
df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
return df

# Рассчет OBV
def calculate_obv(df):
obv = talib.OBV(df["close"], df["volume"])
return obv

# Проверка дивергенций
def detect_obv_divergence():
df = fetch_ohlcv(PAIR)
df["obv"] = calculate_obv(df)

# Анализ цен и OBV на последних 10 свечах
price_min = df["close"].iloc[-LOOKBACK_PERIOD:].min()
price_max = df["close"].iloc[-LOOKBACK_PERIOD:].max()
obv_min = df["obv"].iloc[-LOOKBACK_PERIOD:].min()
obv_max = df["obv"].iloc[-LOOKBACK_PERIOD:].max()
current_price = df["close"].iloc[-1]
current_obv = df["obv"].iloc[-1]

print(f"Цена: {current_price:.2f}, OBV: {current_obv:.2f}")

if price_min < current_price and obv_min > current_obv:
print("📈 Бычья дивергенция! Возможен рост (лонг).")
elif price_max > current_price and obv_max < current_obv:
print("📉 Медвежья дивергенция! Возможен спад (шорт).")
else:
print(" Нет четкого сигнала.")

# Запуск
if __name__ == "__main__":
while True:
detect_obv_divergence()
time.sleep(CHECK_INTERVAL)

#индикатор

📌 Подпишись  Crypto Python❗️
🔥101👍1
📌 Trailing Stop Order Manager – Автоматическое управление стоп-лоссами

Смысл стратегии:

Trailing Stop (скользящий стоп-лосс) автоматически перемещает стоп-лосс за ценой, фиксируя прибыль, если цена идет в благоприятном направлении.

При росте цены → стоп-лосс поднимается.

При падении цены → стоп-лосс остается (пока не сработает)🧑‍💻

import ccxt
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TRAILING_PERCENT = 2 # На сколько % ниже максимальной цены ставится стоп-лосс
CHECK_INTERVAL = 10 # Проверка раз в 10 секунд

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Текущее состояние
max_price = None
stop_loss = None

# Получение текущей цены
def get_current_price(symbol):
ticker = exchange.fetch_ticker(symbol)
return ticker["last"]

# Логика Trailing Stop
def trailing_stop_manager():
global max_price, stop_loss
current_price = get_current_price(PAIR)

# Обновляем максимум цены
if max_price is None or current_price > max_price:
max_price = current_price
stop_loss = max_price * (1 - TRAILING_PERCENT / 100)
print(f"🔼 Новый максимум: {max_price:.2f}, стоп-лосс обновлен: {stop_loss:.2f}")

# Проверяем, сработал ли стоп-лосс
if stop_loss and current_price <= stop_loss:
print(f" Цена {current_price:.2f} достигла стоп-лосса {stop_loss:.2f} – сигнал на выход.")
max_price, stop_loss = None, None # Сбрасываем после срабатывания

# Запуск
if name == "main":
while True:
trailing_stop_manager()
time.sleep(CHECK_INTERVAL)

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

📌 Подпишись  Crypto Python❗️
7🔥5👍1
📌 OCO Orders Script–Установка связанных ордеров (Take Profit/Stop Loss)🧑‍💻

Смысл стратегии:

OCO (One Cancels the Other) – это метод установки двух связанных ордеров:

1. Take Profit (фиксируем прибыль).

2. Stop Loss (ограничиваем убытки).
Когда один из ордеров срабатывает, второй автоматически отменяется.


⚠️ Важно: OCO ордера доступны не на всех биржах – проверьте API биржи перед использованием! 🚀

import ccxt

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TRADE_AMOUNT = 0.001 # Количество актива для торговли
ENTRY_PRICE = 50000 # Цена покупки (можно заменить на реальную рыночную)
TAKE_PROFIT = 52000 # Цель по прибыли
STOP_LOSS = 48500 # Уровень стоп-лосса

# API-ключи (замените своими)
API_KEY = "your_api_key"
API_SECRET = "your_api_secret"

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

# Установка OCO ордера
def place_oco_order():
try:
# Размещение OCO ордера (Take Profit + Stop Loss)
order = exchange.create_order(
symbol=PAIR,
type="OCO",
side="sell",
amount=TRADE_AMOUNT,
params={
"stopPrice": STOP_LOSS, # Уровень Stop Loss
"price": TAKE_PROFIT, # Уровень Take Profit
"stopLimitPrice": STOP_LOSS - 50, # Stop Limit (чуть ниже Stop Loss)
"stopLimitTimeInForce": "GTC" # Ордер действует до исполнения
}
)
print(" OCO ордер успешно установлен:", order)
except Exception as e:
print(" Ошибка установки OCO ордера:", e)

# Запуск
if __name__ == "__main__":
place_oco_order()

#ордера

📌 Подпишись  Crypto Python❗️
👍7
📌 VWAP Trend Analyzer – Анализ трендов с учетом VWAP

Смысл стратегии:

VWAP (Volume Weighted Average Price) – это индикатор, который показывает среднюю цену сделки с учетом объема.

Используется для оценки тренда:

Цена выше VWAP → бычий тренд (поиск покупок).

Цена ниже VWAP → медвежий тренд (поиск продаж).

VWAP в боковом движении → рынок во флэте.

🔹 Применение:

- Подтверждение тренда → если цена выше VWAP, лучше искать покупки.

- Оценка откатов → если цена падает к VWAP в бычьем тренде, можно искать вход.

- Анализ краткосрочного тренда для скальпинга.

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "5m" # Анализ краткосрочного тренда
VWAP_PERIOD = 20 # Количество свечей для расчета VWAP
CHECK_INTERVAL = 60 # Проверка раз в минуту

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение данных
def fetch_ohlcv(symbol):
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=TIMEFRAME, limit=VWAP_PERIOD + 5)
df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
return df

# Рассчет VWAP
def calculate_vwap(df):
typical_price = (df["high"] + df["low"] + df["close"]) / 3
vwap = (typical_price * df["volume"]).cumsum() / df["volume"].cumsum()
return vwap.iloc[-1]

# Анализ тренда с VWAP
def vwap_trend_analyzer():
df = fetch_ohlcv(PAIR)
vwap = calculate_vwap(df)
close_price = df["close"].iloc[-1]

print(f"Цена: {close_price:.2f}, VWAP: {vwap:.2f}")

if close_price > vwap:
print("📈 Цена выше VWAP → бычий тренд (лонг при откатах).")
elif close_price < vwap:
print("📉 Цена ниже VWAP → медвежий тренд (шорт при откатах).")
else:
print(" Цена у VWAP, возможный флэт.")

# Запуск
if __name__ == "__main__":
while True:
vwap_trend_analyzer()
time.sleep(CHECK_INTERVAL)

#индикатор

📌 Подпишись  Crypto Python❗️
🔥8
📌 Smart Order Execution – Умное исполнение ордеров (разбитие на части)🧑‍💻

Смысл стратегии:

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

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

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

Оптимизирует исполнение за счет анализа ликвидности.

import ccxt
import time
import random

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TOTAL_AMOUNT = 1 # Общий объем покупки (BTC)
SPLIT_ORDERS = 10 # Количество частей
MIN_WAIT = 5 # Минимальная задержка между ордерами (сек)
MAX_WAIT = 15 # Максимальная задержка между ордерами (сек)

# API-ключи (замените своими)
API_KEY = "your_api_key"
API_SECRET = "your_api_secret"

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

# Получение текущей цены
def get_current_price(symbol):
ticker = exchange.fetch_ticker(symbol)
return ticker["last"]

# Умное исполнение ордера
def smart_order_execution():
part_size = TOTAL_AMOUNT / SPLIT_ORDERS # Размер одной сделки

for i in range(SPLIT_ORDERS):
price = get_current_price(PAIR)
print(f"📥 Размещение ордера {i+1}/{SPLIT_ORDERS}: Покупка {part_size:.4f} BTC по {price:.2f} USDT")

try:
order = exchange.create_market_buy_order(PAIR, part_size)
print(f" Ордер исполнен: {order}")
except Exception as e:
print(f" Ошибка при исполнении ордера: {e}")

wait_time = random.randint(MIN_WAIT, MAX_WAIT)
print(f" Ждем {wait_time} секунд перед следующим ордером...")
time.sleep(wait_time)

# Запуск
if __name__ == "__main__":
smart_order_execution()

#ордера

📌 Подпишись  Crypto Python❗️
👍7
📌 Liquidity Pool Scanner – Поиск ликвидных активов🧑‍💻

Смысл стратегии:

Ликвидность – это ключевой фактор успешной торговли. Скрипт анализирует объем торгов и глубину стакана, чтобы найти самые ликвидные активы.

Используется для:

📍Поиска монет с высоким объемом и узким спредом.

📍Выбора лучших активов для скальпинга или арбитража.

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

import ccxt
import pandas as pd

# Параметры
EXCHANGE_NAME = "binance"
TOP_N = 10 # Количество лучших активов по ликвидности

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()
exchange.load_markets()

# Получение ликвидных активов
def scan_liquidity():
tickers = exchange.fetch_tickers()
liquidity_data = []

for symbol, data in tickers.items():
if "/USDT" in symbol and "quoteVolume" in data:
bid = data.get("bid", None)
ask = data.get("ask", None)
spread = (ask - bid) / ask * 100 if bid and ask else None

liquidity_data.append({
"pair": symbol,
"volume_24h": data["quoteVolume"],
"spread_%": spread
})

df = pd.DataFrame(liquidity_data)
df = df.dropna().sort_values(by=["volume_24h"], ascending=False).head(TOP_N)

print("📊 Топ ликвидных активов:")
print(df.to_string(index=False))

# Запуск
if __name__ == "__main__":
scan_liquidity()

#индикатор

📌 Подпишись  Crypto Python❗️
👍6🔥4
📌 Market Impact Calculator – Оценка влияния крупных сделок на цену 🧑‍💻

Смысл стратегии:

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

🔹 Метрики влияния:

Объем стакана – сколько ликвидности доступно.

Проскальзывание (Slippage) – насколько изменится цена после сделки.

Рассчет на основе Order Book Depth.

import ccxt

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
ORDER_SIZE = 1 # Объем сделки (в BTC)

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение стакана ордеров
def get_order_book(symbol):
order_book = exchange.fetch_order_book(symbol, limit=50)
return order_book

# Оценка влияния сделки
def calculate_market_impact():
order_book = get_order_book(PAIR)
bids = order_book["bids"] # Покупки
asks = order_book["asks"] # Продажи

# Расчет проскальзывания на покупку
buy_impact = 0
sell_impact = 0
buy_volume = ORDER_SIZE
sell_volume = ORDER_SIZE

for price, volume in asks:
if buy_volume > volume:
buy_impact += volume * price
buy_volume -= volume
else:
buy_impact += buy_volume * price
break

for price, volume in bids:
if sell_volume > volume:
sell_impact += volume * price
sell_volume -= volume
else:
sell_impact += sell_volume * price
break

avg_buy_price = buy_impact / ORDER_SIZE
avg_sell_price = sell_impact / ORDER_SIZE
mid_price = (asks[0][0] + bids[0][0]) / 2

buy_slippage = ((avg_buy_price - mid_price) / mid_price) * 100
sell_slippage = ((mid_price - avg_sell_price) / mid_price) * 100

print(f"📊 Средняя цена покупки: {avg_buy_price:.2f}, проскальзывание: {buy_slippage:.4f}%")
print(f"📉 Средняя цена продажи: {avg_sell_price:.2f}, проскальзывание: {sell_slippage:.4f}%")

# Запуск
if __name__ == "__main__":
calculate_market_impact()

#ордера

📌 Подпишись  Crypto Python❗️
🔥7👍2❤‍🔥1
📌 Delta Volume Analyzer – Анализ дельты объемов для фьючерсов🧑‍💻

Смысл стратегии:

Дельта объема – это разница между объемами покупок по рынку (market buy) и продаж по рынку (market sell).

Положительная дельта (ΔV > 0) → больше рыночных покупок → сильный спрос.

Отрицательная дельта (ΔV < 0) → больше рыночных продаж → сильное предложение.

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


🔹 Применение:

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

Если цена падает, но дельта объема растет → возможный разворот.

Используется в скальпинге и фьючерсной торговле.

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "1m" # Используем низкие таймфреймы для анализа потоков объема
CHECK_INTERVAL = 10 # Обновление каждые 10 секунд

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение данных по ленте сделок
def fetch_trades(symbol):
trades = exchange.fetch_trades(symbol, limit=100)
df = pd.DataFrame(trades)
return df

# Рассчет дельты объемов
def calculate_delta_volume():
df = fetch_trades(PAIR)

# Фильтруем рыночные покупки и продажи
buy_volume = df[df["side"] == "buy"]["amount"].sum()
sell_volume = df[df["side"] == "sell"]["amount"].sum()

delta_volume = buy_volume - sell_volume # Разница между рыночными покупками и продажами

print(f"📊 Дельта объема за последние 100 сделок: {delta_volume:.4f} BTC")
print(f"📈 Покупки: {buy_volume:.4f} BTC, 📉 Продажи: {sell_volume:.4f} BTC")

if delta_volume > 0:
print("🔼 Преобладают рыночные покупки → возможный бычий импульс.")
elif delta_volume < 0:
print("🔽 Преобладают рыночные продажи → возможное снижение.")
else:
print(" Баланс спроса и предложения.")

# Запуск
if __name__ == "__main__":
while True:
calculate_delta_volume()
time.sleep(CHECK_INTERVAL)

#индикатор

📌 Подпишись  Crypto Python❗️
🔥8👍3❤‍🔥1
📌 Dynamic Position Sizing – Адаптивный размер позиций на основе ATR 🧑‍💻

Смысл стратегии:

Adaptive Position Sizing помогает избежать излишней волатильности, регулируя размер позиции в зависимости от ATR (Average True Range).

Высокий ATR → рынок волатильный → уменьшаем размер позиции.

Низкий ATR → рынок спокойный → увеличиваем размер позиции.

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

import ccxt
import pandas as pd
import talib
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
ATR_PERIOD = 14 # Период ATR
RISK_PER_TRADE = 100 # Максимальный риск на сделку (в $)
RISK_MULTIPLIER = 1.5 # Коэффициент риска (K)

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение данных
def fetch_ohlcv(symbol):
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=TIMEFRAME, limit=ATR_PERIOD + 5)
df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
return df

# Рассчет ATR и размера позиции
def calculate_position_size():
df = fetch_ohlcv(PAIR)
atr = talib.ATR(df["high"], df["low"], df["close"], timeperiod=ATR_PERIOD).iloc[-1]
current_price = df["close"].iloc[-1]

# Рассчитываем размер позиции в BTC
position_size = RISK_PER_TRADE / (atr * RISK_MULTIPLIER)

print(f"📊 Цена: {current_price:.2f} USDT, ATR: {atr:.2f}")
print(f"🔹 Оптимальный размер позиции: {position_size:.4f} BTC")

# Запуск
if __name__ == "__main__":
while True:
calculate_position_size()
time.sleep(60) # Проверка раз в минуту

#ордера

📌 Подпишись  Crypto Python❗️
🔥9👍3❤‍🔥1
📌 Order Book Imbalance Detector – Выявление дисбаланса заявок 🧑‍💻

Смысл стратегии:

Дисбаланс стакана заявок – это ситуация, когда в стакане больше лимитных ордеров на покупку (бидов) или продажу (асков).

Дисбаланс в сторону бидов → рынок может пойти вверх (сильный спрос).

Дисбаланс в сторону асков → рынок может упасть (сильное предложение).

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

import ccxt
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
ORDER_BOOK_DEPTH = 20 # Глубина анализа (количество уровней в стакане)
CHECK_INTERVAL = 5 # Проверка раз в 5 секунд

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение стакана заявок
def get_order_book(symbol):
order_book = exchange.fetch_order_book(symbol, limit=ORDER_BOOK_DEPTH)
return order_book

# Анализ дисбаланса
def detect_order_book_imbalance():
order_book = get_order_book(PAIR)

bids = order_book["bids"] # Лимитные покупки (цена, объем)
asks = order_book["asks"] # Лимитные продажи (цена, объем)

# Рассчитываем общий объем заявок
bid_volume = sum([volume for price, volume in bids])
ask_volume = sum([volume for price, volume in asks])

# Дисбаланс стакана
imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume)

print(f"📊 Дисбаланс заявок: {imbalance:.4f} | Биды: {bid_volume:.2f} | Аски: {ask_volume:.2f}")

if imbalance > 0.1:
print("📈 Преобладание бидов → возможный рост.")
elif imbalance < -0.1:
print("📉 Преобладание асков → возможное падение.")
else:
print(" Баланс спроса и предложения.")

# Запуск
if __name__ == "__main__":
while True:
detect_order_book_imbalance()
time.sleep(CHECK_INTERVAL)

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

📌 Подпишись  Crypto Python❗️
👍10
📌 Slippage Estimator – Расчет проскальзывания для разных бирж🧑‍💻

Смысл стратегии:

Проскальзывание (Slippage) – это разница между ожидаемой ценой сделки и фактической ценой исполнения. Оно возникает из-за недостаточной ликвидности или высокой волатильности.

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

import ccxt
import time

# Параметры
EXCHANGES = ["binance", "kucoin"] # Биржи для сравнения
PAIR = "BTC/USDT"
ORDER_SIZE = 1 # Объем сделки (в BTC)
ORDER_BOOK_DEPTH = 20 # Глубина анализа стакана

# Инициализация бирж
exchanges = {name: getattr(ccxt, name)() for name in EXCHANGES}

# Получение стакана заявок
def get_order_book(exchange, symbol):
try:
return exchange.fetch_order_book(symbol, limit=ORDER_BOOK_DEPTH)
except Exception as e:
print(f" Ошибка получения данных с {exchange.id}: {e}")
return None

# Расчет проскальзывания
def calculate_slippage():
results = []

for name, exchange in exchanges.items():
order_book = get_order_book(exchange, PAIR)
if not order_book:
continue

bids = order_book["bids"] # Лимитные покупки (цена, объем)
asks = order_book["asks"] # Лимитные продажи (цена, объем)

# Средняя цена покупки (Market Buy)
buy_impact, sell_impact = 0, 0
buy_volume, sell_volume = ORDER_SIZE, ORDER_SIZE

for price, volume in asks:
if buy_volume > volume:
buy_impact += volume * price
buy_volume -= volume
else:
buy_impact += buy_volume * price
break

for price, volume in bids:
if sell_volume > volume:
sell_impact += volume * price
sell_volume -= volume
else:
sell_impact += sell_volume * price
break

avg_buy_price = buy_impact / ORDER_SIZE
avg_sell_price = sell_impact / ORDER_SIZE
mid_price = (asks[0][0] + bids[0][0]) / 2

buy_slippage = ((avg_buy_price - mid_price) / mid_price) * 100
sell_slippage = ((mid_price - avg_sell_price) / mid_price) * 100

results.append({
"Биржа": name,
"Средняя цена покупки": round(avg_buy_price, 2),
"Проскальзывание покупки %": round(buy_slippage, 4),
"Средняя цена продажи": round(avg_sell_price, 2),
"Проскальзывание продажи %": round(sell_slippage, 4),
})

# Вывод результатов
print("\n📊 Сравнение проскальзывания на биржах:")
for res in results:
print(f"{res['Биржа'].capitalize()}: Покупка {res['Средняя цена покупки']} USDT ({res['Проскальзывание покупки %']}%), "
f"Продажа {res['Средняя цена продажи']} USDT ({res['Проскальзывание продажи %']}%)")

# Запуск
if __name__ == "__main__":
while True:
calculate_slippage()
time.sleep(10) # Проверка каждые 10 секунд

#ордера

📌 Подпишись  Crypto Python❗️
👍15
📌 Volume Spike Detector – Выявление резких всплесков объемов 🧑‍💻

Смысл стратегии:

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

Резкий рост объема → возможное продолжение тренда.

Объем растет, но цена не меняется → возможен разворот.

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

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "1m"
LOOKBACK_PERIOD = 20 # Количество свечей для расчета среднего объема
SPIKE_THRESHOLD = 2 # Минимальный коэффициент всплеска
CHECK_INTERVAL = 10 # Проверка раз в 10 секунд

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение данных
def fetch_ohlcv(symbol):
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=TIMEFRAME, limit=LOOKBACK_PERIOD + 5)
df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
return df

# Анализ всплесков объема
def detect_volume_spike():
df = fetch_ohlcv(PAIR)

avg_volume = df["volume"].iloc[-LOOKBACK_PERIOD:-1].mean() # Средний объем за N свечей
current_volume = df["volume"].iloc[-1] # Объем текущей свечи
spike_ratio = current_volume / avg_volume if avg_volume > 0 else 0

print(f"📊 Текущий объем: {current_volume:.2f}, Средний объем: {avg_volume:.2f}, Коэффициент всплеска: {spike_ratio:.2f}")

if spike_ratio > SPIKE_THRESHOLD:
print("🔥 Всплеск объема! Возможное сильное движение.")
else:
print(" Объем в норме.")

# Запуск
if __name__ == "__main__":
while True:
detect_volume_spike()
time.sleep(CHECK_INTERVAL)

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

📌 Подпишись  Crypto Python❗️
🔥14👍2
📌 Market Depth Analyzer – Анализ глубины рынка 🧑‍💻

Смысл стратегии:

Глубина рынка показывает, сколько ликвидности доступно на разных уровнях стакана.

Глубокий рынок → меньше проскальзывания, стабильные цены.

Мелкий рынок → высокое проскальзывание, возможны резкие движения.

Анализ глубины стакана помогает выявлять уровни поддержки и сопротивления🚀

import ccxt
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
ORDER_BOOK_DEPTH = 20 # Количество уровней в стакане
CHECK_INTERVAL = 5 # Проверка раз в 5 секунд

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение стакана заявок
def get_order_book(symbol):
try:
return exchange.fetch_order_book(symbol, limit=ORDER_BOOK_DEPTH)
except Exception as e:
print(f" Ошибка получения стакана с {exchange.id}: {e}")
return None

# Анализ глубины рынка
def analyze_market_depth():
order_book = get_order_book(PAIR)
if not order_book:
return

bids = order_book["bids"] # Лимитные покупки (цена, объем)
asks = order_book["asks"] # Лимитные продажи (цена, объем)

# Рассчитываем общий объем ликвидности
bid_depth = sum([volume for price, volume in bids])
ask_depth = sum([volume for price, volume in asks])

# Дисбаланс ликвидности
imbalance = (bid_depth - ask_depth) / (bid_depth + ask_depth)

print(f"📊 Глубина рынка для {PAIR}:")
print(f"🔹 Общий объем BID (покупки): {bid_depth:.2f} BTC")
print(f"🔸 Общий объем ASK (продажи): {ask_depth:.2f} BTC")
print(f"📉 Дисбаланс: {imbalance:.4f}")

if imbalance > 0.1:
print("📈 Преобладает спрос → возможный рост цены.")
elif imbalance < -0.1:
print("📉 Преобладает предложение → возможное снижение цены.")
else:
print(" Баланс спроса и предложения.")

# Запуск
if __name__ == "__main__":
while True:
analyze_market_depth()
time.sleep(CHECK_INTERVAL)

#индикатор

📌 Подпишись  Crypto Python❗️
🔥10👍4
📌 Fear and Greed Index Analyzer – Торговля по индексу страха и жадности🧑‍💻

Смысл стратегии:

Индекс страха и жадности (Fear & Greed Index) анализирует настроение рынка:

0-25 → Страх (Fear) → возможная перепроданность → поиск покупок.

25-50 → Нейтральный рынок → нет четкого направления.

50-75 → Жадность (Greed) → рынок перегрет, но тренд может продолжаться.

75-100 → Крайняя жадность (Extreme Greed) → возможная перекупленность → поиск продаж.


⚠️ Важно: Fear & Greed Index лучше работает на среднесрочных таймфреймах (4H, 1D) для определения глобального настроения рынка. 🚀

import ccxt
import requests
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
CHECK_INTERVAL = 3600 # Проверка раз в час
FEAR_GREED_API = "https://api.alternative.me/fng/" # Источник данных

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение текущего индекса страха и жадности
def get_fear_greed_index():
try:
response = requests.get(FEAR_GREED_API)
data = response.json()
index_value = int(data["data"][0]["value"])
return index_value
except Exception as e:
print(f" Ошибка получения индекса: {e}")
return None

# Получение текущей цены BTC
def get_current_price():
ticker = exchange.fetch_ticker(PAIR)
return ticker["last"]

# Логика торговли по индексу
def fear_greed_trading():
index_value = get_fear_greed_index()
if index_value is None:
return

price = get_current_price()

print(f"📊 Индекс страха и жадности: {index_value}")
print(f"💰 Текущая цена {PAIR}: {price:.2f} USDT")

if index_value <= 25:
print("📈 Рынок в страхе → Возможная точка входа (покупка).")
elif index_value >= 75:
print("📉 Рынок в жадности → Возможная точка выхода (продажа).")
else:
print(" Рынок нейтральный, ждем сигнал.")

# Запуск
if __name__ == "__main__":
while True:
fear_greed_trading()
time.sleep(CHECK_INTERVAL)

#индикатор

📌 Подпишись  Crypto Python❗️
🔥12👍3
📌 Open Interest Analyzer – Анализ открытого интереса на фьючерсах🧑‍💻

Смысл стратегии:

Открытый интерес (Open Interest, OI) – это общее количество открытых контрактов на фьючерсном рынке.

Рост OI → новые деньги входят в рынок → усиление текущего тренда.

Падение OI → закрытие позиций → возможный разворот тренда.

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

import ccxt
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
CHECK_INTERVAL = 60 # Проверка раз в минуту

# Инициализация биржи
exchange = ccxt.binance({
"options": {"defaultType": "future"} # Работаем с фьючерсами
})

# Получение открытого интереса
def get_open_interest(symbol):
try:
oi_data = exchange.fapiPublicGetOpenInterest({"symbol": symbol.replace("/", "")})
return float(oi_data["openInterest"])
except Exception as e:
print(f" Ошибка получения OI: {e}")
return None

# Получение текущей цены
def get_current_price(symbol):
ticker = exchange.fetch_ticker(symbol)
return ticker["last"]

# Логика анализа OI
def open_interest_analyzer():
prev_oi = None

while True:
oi = get_open_interest(PAIR)
price = get_current_price(PAIR)

if oi is None:
time.sleep(CHECK_INTERVAL)
continue

print(f"📊 Цена: {price:.2f} USDT | Открытый интерес: {oi:.2f}")

if prev_oi is not None:
delta_oi = oi - prev_oi
print(f"🔍 Изменение OI: {delta_oi:.2f}")

if delta_oi > 0:
print("📈 OI растет → В рынок заходят новые деньги, тренд усиливается.")
elif delta_oi < 0:
print("📉 OI падает → Закрытие позиций, возможный разворот.")

prev_oi = oi
time.sleep(CHECK_INTERVAL)

# Запуск
if __name__ == "__main__":
open_interest_analyzer()

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

📌 Подпишись  Crypto Python❗️
🔥13
📌 Hidden Order Placement – Алгоритм скрытых ордеров для больших объемов🧑‍💻

Смысл стратегии:

Когда крупный ордер размещается в стакане, он может влиять на цену и привлекать внимание других трейдеров.

Этот алгоритм разбивает ордер на мелкие части и размещает их случайно в течение времени, имитируя естественные рыночные покупки/продажи.

import ccxt
import time
import random

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TOTAL_AMOUNT = 5 # Общий объем ордера (BTC)
SPLIT_ORDERS = 20 # Количество частей
MIN_WAIT = 10 # Минимальная задержка между ордерами (сек)
MAX_WAIT = 60 # Максимальная задержка между ордерами (сек)

# API-ключи (замените своими)
API_KEY = "your_api_key"
API_SECRET = "your_api_secret"

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

# Получение текущей цены
def get_current_price(symbol):
ticker = exchange.fetch_ticker(symbol)
return ticker["last"]

# Размещение скрытых ордеров
def hidden_order_execution():
part_size = TOTAL_AMOUNT / SPLIT_ORDERS # Размер одной сделки

for i in range(SPLIT_ORDERS):
price = get_current_price(PAIR)
print(f"📥 Размещение ордера {i+1}/{SPLIT_ORDERS}: Покупка {part_size:.4f} BTC по {price:.2f} USDT")

try:
order = exchange.create_limit_buy_order(PAIR, part_size, price)
print(f" Ордер размещен: {order}")
except Exception as e:
print(f" Ошибка при размещении ордера: {e}")

wait_time = random.randint(MIN_WAIT, MAX_WAIT)
print(f" Ждем {wait_time} секунд перед следующим ордером...")
time.sleep(wait_time)

# Запуск
if __name__ == "__main__":
hidden_order_execution()

#ордера

📌 Подпишись  Crypto Python❗️
1👍6🐳4
📌 Cumulative Delta Strategy – Стратегия на основе кумулятивной дельты объема🧑‍💻

Смысл стратегии:

Кумулятивная дельта объема (CVD, Cumulative Volume Delta) — это накопленная разница между объемом рыночных покупок и продаж.

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

CVD падает → доминируют рыночные продавцы (медвежий сигнал).

Дивергенция между ценой и CVD → возможный разворот.


🔹 Применение стратегии:🚀

CVD + цена растут вместе → подтверждение восходящего тренда (лонг).

CVD падает, а цена растёт → дивергенция, возможный разворот вниз.

CVD растёт, а цена падает → скрытый спрос, возможный разворот вверх.

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TIMEFRAME = "1m"
CVD_LENGTH = 100 # Количество последних трейдов
CHECK_INTERVAL = 10 # Проверка каждые 10 секунд

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение последних сделок
def fetch_trades(symbol):
trades = exchange.fetch_trades(symbol, limit=CVD_LENGTH)
df = pd.DataFrame(trades)
return df

# Расчет кумулятивной дельты
def calculate_cvd(trades_df):
trades_df['delta'] = trades_df.apply(
lambda row: row['amount'] if row['side'] == 'buy' else -row['amount'], axis=1
)
trades_df['cvd'] = trades_df['delta'].cumsum()
return trades_df['cvd'].iloc[-1], trades_df['cvd']

# Получение текущей цены
def get_price(symbol):
return exchange.fetch_ticker(symbol)['last']

# Основной цикл стратегии
def cumulative_delta_strategy():
prev_cvd = None
while True:
trades_df = fetch_trades(PAIR)
current_price = get_price(PAIR)
cvd_value, cvd_series = calculate_cvd(trades_df)

print(f"Цена: {current_price:.2f} USDT | CVD: {cvd_value:.4f}")

# Примитивная логика входа на основе CVD
if prev_cvd is not None:
delta_cvd = cvd_value - prev_cvd

if delta_cvd > 0:
print("📈 Кумулятивная дельта растёт → доминируют покупатели (лонг сигнал).")
elif delta_cvd < 0:
print("📉 Кумулятивная дельта падает → доминируют продавцы (шорт сигнал).")
else:
print("⚖️ Баланс между покупателями и продавцами.")

prev_cvd = cvd_value
time.sleep(CHECK_INTERVAL)

# Запуск
if __name__ == "__main__":
cumulative_delta_strategy()

#индикатор

📌 Подпишись  Crypto Python❗️
👍9🔥7
📌 Выявление скрытых крупных заявок (Iceberg Orders)🧑‍💻

Что это и зачем?

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

🔥 Как это использовать?

📌 Если Iceberg в BID → крупный игрок накапливает позицию, рост вероятен

📌 Если Iceberg в ASK → крупный игрок разгружает позицию, может быть падение

import requests
import time

# API Bybit (глубина стакана)
ORDER_BOOK_URL = "https://api.bybit.com/v5/market/orderbook"
SYMBOL = "BTCUSDT"
DEPTH = 50  # Берём топ-50 заявок

def get_order_book():
    """Получает стакан заявок"""
    params = {"category": "spot", "symbol": SYMBOL, "limit": DEPTH}
    response = requests.get(ORDER_BOOK_URL, params=params)
    if response.status_code == 200:
        data = response.json().get("result", {})
        bids = [(float(price), float(size)) for price, size in data.get("b", [])]
        asks = [(float(price), float(size)) for price, size in data.get("a", [])]
        return bids, asks
    else:
        print(f"Ошибка API: {response.status_code}")
        return None, None

def detect_iceberg_orders():
    """Выявляет скрытые Iceberg-заявки"""
    prev_bids, prev_asks = get_order_book()
    if not prev_bids or not prev_asks:
        print(" Ошибка получения стакана.")
        return

    print("📡 Начинаем мониторинг Iceberg-заявок...")
    while True:
        time.sleep(2)  # Ждём обновления стакана
        bids, asks = get_order_book()
        if not bids or not asks:
            continue

        # Анализ заявок (BID - покупатели)
        for i in range(min(len(bids), len(prev_bids))):
            if bids[i][0] == prev_bids[i][0] and bids[i][1] > prev_bids[i][1] * 1.5:
                print(f"❄️ Iceberg-заявка в BID: {bids[i][0]} - новый объём {bids[i][1]} BTC (было {prev_bids[i][1]} BTC)")

        # Анализ заявок (ASK - продавцы)
        for i in range(min(len(asks), len(prev_asks))):
            if asks[i][0] == prev_asks[i][0] and asks[i][1] > prev_asks[i][1] * 1.5:
                print(f"❄️ Iceberg-заявка в ASK: {asks[i][0]} - новый объём {asks[i][1]} BTC (было {prev_asks[i][1]} BTC)")

        prev_bids, prev_asks = bids, asks  # Обновляем стакан

if __name__ == "__main__":
    detect_iceberg_orders()

#индикатор

📌 Подпишись  Crypto Python❗️
👍132
📌 Whale Tracker – Отслеживание крупных сделок и ордеров в стакане🧑‍💻

Смысл стратегии:

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

Крупные рыночные сделки → сильное движение, импульс.

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

Используется для скальпинга, интрадей и подтверждения трендов🚀

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
MIN_TRADE_SIZE = 10 # Минимальный объем сделки (в BTC), чтобы считать её «китовой»
MIN_ORDER_VOLUME = 50 # Объем заявки в стакане для отслеживания
CHECK_INTERVAL = 5 # Проверка раз в 5 секунд

# Инициализация биржи
exchange = getattr(ccxt, EXCHANGE_NAME)()

# Получение последних сделок
def fetch_large_trades():
trades = exchange.fetch_trades(PAIR, limit=100)
df = pd.DataFrame(trades)
large_trades = df[df['amount'] >= MIN_TRADE_SIZE]

for _, row in large_trades.iterrows():
side = row['side'].upper()
print(f"🐋 Крупная {side} сделка: {row['amount']} BTC по {row['price']} USDT")

# Анализ стакана ордеров
def check_order_book():
order_book = exchange.fetch_order_book(PAIR, limit=50)
bids = order_book["bids"]
asks = order_book["asks"]

print("🔍 Поиск крупных заявок в стакане...")
for price, volume in bids[:10]:
if volume >= MIN_ORDER_VOLUME:
print(f"⬇️ BID: {volume:.2f} BTC на {price:.2f} USDT")

for price, volume in asks[:10]:
if volume >= MIN_ORDER_VOLUME:
print(f"⬆️ ASK: {volume:.2f} BTC на {price:.2f} USDT")

# Основной цикл
def whale_tracker():
while True:
print("\n--- Обновление ---")
fetch_large_trades()
check_order_book()
time.sleep(CHECK_INTERVAL)

# Запуск
if __name__ == "__main__":
whale_tracker()

#индикатор

📌 Подпишись  Crypto Python❗️
👍11
📌 VWAP Order Execution – Исполнение крупных ордеров по VWAP🧑‍💻

Смысл стратегии:

VWAP (Volume Weighted Average Price) – это средняя цена, по которой актив торговался с учётом объёма.

Цель стратегии:

Исполнить большой ордер максимально близко к VWAP,

Избежать проскальзывания,

Не влиять на рынок, как при рыночной сделке.

🔹 VWAP Execution = Разбиение большого ордера на части, распределённых во времени и ориентированных на текущую VWAP-цену.

import ccxt
import pandas as pd
import time

# Параметры
EXCHANGE_NAME = "binance"
PAIR = "BTC/USDT"
TOTAL_AMOUNT = 1 # Всего купить 1 BTC
SPLIT = 10 # Разбить на 10 частей
TIMEFRAME = "1m"
VWAP_PERIOD = 20
INTERVAL = 60 # Интервал между проверками (сек)

# API-ключи (если нужно реальное исполнение)
API_KEY = "your_api_key"
API_SECRET = "your_api_secret"

# Биржа
exchange = ccxt.binance({
"apiKey": API_KEY,
"secret": API_SECRET,
"enableRateLimit": True,
})

# Получение данных
def fetch_ohlcv(symbol):
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=TIMEFRAME, limit=VWAP_PERIOD + 5)
df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
return df

# Расчет VWAP
def calculate_vwap(df):
tp = (df["high"] + df["low"] + df["close"]) / 3
vwap = (tp * df["volume"]).cumsum() / df["volume"].cumsum()
return vwap.iloc[-1]

# Получение текущей цены
def get_price(symbol):
return exchange.fetch_ticker(symbol)["last"]

# Основной цикл
def vwap_order_execution():
part_size = TOTAL_AMOUNT / SPLIT
executed = 0

while executed < SPLIT:
df = fetch_ohlcv(PAIR)
vwap = calculate_vwap(df)
price = get_price(PAIR)

print(f"Цена: {price:.2f} | VWAP: {vwap:.2f}")

if price <= vwap * 1.002: # Допускаем отклонение 0.2% вверх
print(f"📥 Покупка части {executed + 1}/{SPLIT} по цене {price:.2f}")
try:
order = exchange.create_limit_buy_order(PAIR, part_size, price)
print(" Ордер отправлен:", order["id"])
executed += 1
except Exception as e:
print(" Ошибка размещения ордера:", e)
else:
print(" Цена выше VWAP, ждем...")

time.sleep(INTERVAL)

# Запуск
if __name__ == "__main__":
vwap_order_execution()

#ордера

📌 Подпишись  Crypto Python❗️
🔥111👍1