Crypto Python
817 subscribers
448 photos
26 files
455 links
Алготрейдинг 🤖 , работа с API бирж и агрегаторов 🌐 , автоматизации в крипто сфере📈 🚀
Ваши предложения📝 @binance_de
Download Telegram
📌Genetic Strategy Generator – Генерация торговых стратегий с помощью генетических алгоритмов🧑‍💻

Суть стратегии:🚀

- Генетический алгоритм (GA) используется для поиска оптимальных правил входа/выхода в рынок, основываясь на:

- технических индикаторах,

- логических операторах (AND, OR, >, <),

- целевой функции (например, максимизация доходности или Sharpe Ratio).

Такой подход имитирует естественный отбор:
→ лучшие "особи" (стратегии) выживают и улучшаются из поколения в поколение.

Что делает Genetic Strategy Generator:🛠️

1. Генерирует случайные стратегии из набора правил и индикаторов

2. Применяет их к историческим данным

3. Оценивает результат (fitness) — прибыль, стабильность, drawdown

4. Оставляет лучшие стратегии и мутирует их

5. Повторяет до достижения лучших результатов

Пример Python-структуры: стратегия на базе RSI и MACD
import numpy as np
import pandas as pd
import talib
import ccxt
import random

# Получение данных
exchange = ccxt.binance()
PAIR = "BTC/USDT"
TIMEFRAME = "1h"
LOOKBACK = 300

def fetch_data():
ohlcv = exchange.fetch_ohlcv(PAIR, timeframe=TIMEFRAME, limit=LOOKBACK)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
df["rsi"] = talib.RSI(df["close"], timeperiod=14)
macd, signal, _ = talib.MACD(df["close"])
df["macd"] = macd
df["signal"] = signal
df.dropna(inplace=True)
return df

# Генерация случайной стратегии
def generate_strategy():
return {
"rsi_threshold": random.randint(20, 80),
"macd_relation": random.choice([">", "<"])
}

# Применение стратегии
def apply_strategy(df, strategy):
signals = []
for i in range(len(df)):
rsi = df["rsi"].iloc[i]
macd = df["macd"].iloc[i]
signal = df["signal"].iloc[i]

cond1 = rsi < strategy["rsi_threshold"]
cond2 = (macd > signal) if strategy["macd_relation"] == ">" else (macd < signal)

if cond1 and cond2:
signals.append(1) # BUY
else:
signals.append(0) # NO ACTION
df["signal"] = signals
return df

# Подсчёт PnL (доходности)
def evaluate(df):
df["returns"] = df["close"].pct_change().fillna(0)
df["strategy_return"] = df["returns"] * df["signal"].shift(1)
total_return = (df["strategy_return"] + 1).prod() - 1
return total_return

# Генетический цикл
def genetic_optimizer(pop_size=20, generations=10):
population = [generate_strategy() for _ in range(pop_size)]
df = fetch_data()

for gen in range(generations):
scores = []
for strat in population:
test_df = apply_strategy(df.copy(), strat)
score = evaluate(test_df)
scores.append((score, strat))

scores.sort(reverse=True, key=lambda x: x[0])
print(f"🧬 Поколение {gen}: лучшая доходность = {scores[0][0]:.2%}")

# Отбор + мутация
best = [x[1] for x in scores[:pop_size // 2]]
new_generation = []
for s in best:
new = s.copy()
if random.random() < 0.5:
new["rsi_threshold"] += random.randint(-5, 5)
new["rsi_threshold"] = max(10, min(new["rsi_threshold"], 90))
if random.random() < 0.3:
new["macd_relation"] = ">" if s["macd_relation"] == "<" else "<"
new_generation.append(new)
population = best + new_generation

return scores[0][1]

if __name__ == "__main__":
best_strategy = genetic_optimizer()
print(f"\n🏆 Лучшая стратегия: {best_strategy}")

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

📌 Подпишись  Crypto Python❗️
🔥11👍3❤‍🔥1
📌Reinforcement Learning Agent – Стратегия на основе самообучающегося агента🧑‍💻

Суть подхода:🚀

Reinforcement Learning (RL) — обучение агента принимать оптимальные решения путём взаимодействия со средой и получения награды за результат.
В контексте трейдинга агент:

Наблюдает состояние рынка (индикаторы, цены, объёмы),

Принимает действия: купить, продать, удерживать,

Получает награду (например, PnL),

Обучается максимизировать общую прибыль.
import gym
import numpy as np
import pandas as pd
import ccxt
import ta
from gym import spaces
from stable_baselines3 import PPO

# Создаём кастомную среду
class TradingEnv(gym.Env):
def __init__(self, df):
super().__init__()
self.df = df.reset_index(drop=True)
self.current_step = 0
self.start_balance = 10000
self.balance = self.start_balance
self.position = 0 # 0: нет, 1: лонг
self.entry_price = 0

self.action_space = spaces.Discrete(3) # hold, buy, sell
self.observation_space = spaces.Box(low=-np.inf, high=np.inf, shape=(4,), dtype=np.float32)

def _get_obs(self):
row = self.df.iloc[self.current_step]
return np.array([row['close'], row['rsi'], row['macd'], self.position], dtype=np.float32)

def _get_reward(self, price):
if self.position == 1:
return price - self.entry_price
return 0

def step(self, action):
done = False
price = self.df.iloc[self.current_step]['close']

reward = 0
if action == 1 and self.position == 0: # BUY
self.position = 1
self.entry_price = price
elif action == 2 and self.position == 1: # SELL
reward = price - self.entry_price
self.balance += reward
self.position = 0

self.current_step += 1
if self.current_step >= len(self.df) - 1:
done = True

obs = self._get_obs()
return obs, reward, done, {}

def reset(self):
self.current_step = 0
self.balance = self.start_balance
self.position = 0
self.entry_price = 0
return self._get_obs()

# Подготовка данных
def fetch_data():
exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv("BTC/USDT", timeframe="1h", limit=500)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
df["rsi"] = ta.momentum.RSIIndicator(df["close"]).rsi()
macd = ta.trend.MACD(df["close"])
df["macd"] = macd.macd()
return df.dropna()

# Обучение агента
df = fetch_data()
env = TradingEnv(df)
model = PPO("MlpPolicy", env, verbose=1)
model.learn(total_timesteps=10000)

# Тестирование агента
obs = env.reset()
done = False
while not done:
action, _ = model.predict(obs)
obs, reward, done, _ = env.step(action)
print(f"Action: {action}, Reward: {reward:.2f}")

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

📌 Подпишись  Crypto Python❗️
🔥15👍2❤‍🔥11
📌AutoFeature Engine for Strategies – Автоматическая генерация признаков для оптимизации торговых правил🧑‍💻

Суть стратегии:🚀

AutoFeature Engine — это компонент, который:

- создаёт множество технических признаков (features) на основе OHLCV-данных,

- помогает отбирать наиболее значимые признаки для стратегии,

- используется в связке с машинным обучением или генерацией правил (в том числе для XGBoost, RL, GA и т.д.).

Типы автоматически создаваемых признаков:🛠️

1. Технические индикаторы: RSI, MACD, ATR, Bollinger, EMA и др.

2. Статистические признаки: скользящее среднее, std, z-score, rate of change

3. Комбинированные признаки: кроссы, отклонения, сигнальные уровни

4. Целевые признаки (label): будущий доход / направление движения

import ccxt
import pandas as pd
import ta # pip install ta
import numpy as np

# Загрузка данных
exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv("BTC/USDT", timeframe="1h", limit=500)
df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"])
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")

# Генерация признаков
def generate_features(df):
df["return_1h"] = df["close"].pct_change()
df["sma_10"] = ta.trend.sma_indicator(df["close"], window=10)
df["sma_50"] = ta.trend.sma_indicator(df["close"], window=50)
df["rsi"] = ta.momentum.RSIIndicator(df["close"]).rsi()
macd = ta.trend.MACD(df["close"])
df["macd"] = macd.macd()
df["macd_signal"] = macd.macd_signal()
df["bollinger_h"] = ta.volatility.BollingerBands(df["close"]).bollinger_hband()
df["bollinger_l"] = ta.volatility.BollingerBands(df["close"]).bollinger_lband()
df["atr"] = ta.volatility.AverageTrueRange(df["high"], df["low"], df["close"]).average_true_range()
df["roc"] = ta.momentum.ROCIndicator(df["close"]).roc()
df["zscore"] = (df["close"] - df["close"].rolling(20).mean()) / df["close"].rolling(20).std()
return df.dropna()

# Целевая переменная (напр. рост цены через 3 часа)
def generate_target(df, forward=3):
df["future_return"] = df["close"].shift(-forward) / df["close"] - 1
df["target"] = (df["future_return"] > 0).astype(int) # 1 - BUY, 0 - SELL
return df.dropna()

# Основной процесс
df = generate_features(df)
df = generate_target(df)

print(" Сгенерированы признаки:")
print(df[["return_1h", "sma_10", "sma_50", "rsi", "macd", "zscore", "target"]].tail())

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

📌 Подпишись  Crypto Python❗️
🔥6👍3❤‍🔥1🤔1
📌Sentiment-Based Trading – Интеграция с API новостей и Twitter для анализа настроений🧑‍💻

Суть стратегии:🚀

Sentiment-Based Trading опирается на анализ настроений из внешних источников, таких как:

- Новости (новостные API)

- Twitter / X (социальные сигналы)

- Reddit, Telegram, YouTube, Discord и др.

→ Алгоритм оценивает, позитивно или негативно настроен рынок,

→ На основе этого принимает торговые решения (например, лонг на позитиве).

import tweepy
import pandas as pd
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
import time

# Настройки Twitter API (получить на https://developer.twitter.com/)
api_key = "YOUR_API_KEY"
api_secret = "YOUR_API_SECRET"
access_token = "YOUR_ACCESS_TOKEN"
access_token_secret = "YOUR_ACCESS_SECRET"

# Авторизация
auth = tweepy.OAuth1UserHandler(api_key, api_secret, access_token, access_token_secret)
api = tweepy.API(auth)

# Анализатор тональности
analyzer = SentimentIntensityAnalyzer()

# Настройки мониторинга
query = "bitcoin OR btc"
limit = 50

def fetch_sentiment():
tweets = api.search_tweets(q=query, lang="en", count=limit, result_type="recent")
data = []
for tweet in tweets:
score = analyzer.polarity_scores(tweet.text)
data.append(score['compound'])
return data

def run_sentiment_signal():
sentiments = fetch_sentiment()
avg_sentiment = sum(sentiments) / len(sentiments)
print(f"\nСреднее настроение по Twitter: {avg_sentiment:.3f}")

if avg_sentiment > 0.2:
print("📈 Позитивное настроение → сигнал на покупку")
elif avg_sentiment < -0.2:
print("📉 Негативное настроение → сигнал на продажу")
else:
print(" Нейтральный фон, сигналов нет")

# Основной цикл
if __name__ == "__main__":
while True:
run_sentiment_signal()
time.sleep(60 * 10)

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

📌 Подпишись  Crypto Python❗️
🔥10❤‍🔥1
📌Zero-Lag MACD Strategy – Стратегия MACD без задержки🧑‍💻

Суть стратегии:🚀

Стандартный MACD (Moving Average Convergence Divergence) основан на экспоненциальных скользящих средних (EMA), которые по своей природе запаздывают.
Zero-Lag MACD — модификация, которая уменьшает задержку сигнала и позволяет раньше входить в позицию.

Основная идея:📝

Используется двойное EMA сглаживание (или фильтр Тилсона / Т3), чтобы сократить лаг.

Принцип остался тот же:🛠️

Линия MACD = ZeroLag(EMA_fast) − ZeroLag(EMA_slow)

Сигнальная линия = ZeroLag(EMA_MACD)

Сигналы:💸

- Пересечение MACD и сигнальной линии

- Пересечение нуля

- Дивергенции
import ccxt
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

# Zero-Lag EMA (ZLEMA)
def zlema(series, period):
lag = (period - 1) // 2
ema = series + (series - series.shift(lag))
return ema.ewm(span=period, adjust=False).mean()

# Получение данных
exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv("BTC/USDT", timeframe="1h", limit=200)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")

# Расчёт Zero-Lag MACD
fast = zlema(df["close"], 12)
slow = zlema(df["close"], 26)
df["zmacd"] = fast - slow
df["zsignal"] = zlema(df["zmacd"], 9)

# Сигналы
df["signal"] = 0
df.loc[df["zmacd"] > df["zsignal"], "signal"] = 1
df.loc[df["zmacd"] < df["zsignal"], "signal"] = -1

# Визуализация
plt.figure(figsize=(12, 6))
plt.plot(df["ts"], df["zmacd"], label="Zero-Lag MACD")
plt.plot(df["ts"], df["zsignal"], label="Signal Line", linestyle='--')
plt.axhline(0, color='gray', linestyle=':')
plt.legend()
plt.title("Zero-Lag MACD Strategy")
plt.show()

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

📌 Подпишись  Crypto Python❗️
👍8❤‍🔥1🔥1
📌Correlation Matrix Monitor – Отслеживание взаимосвязей между активами🧑‍💻

Суть стратегии:🚀

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

Correlation Matrix Monitor позволяет:👇

- Обнаружить переоценённые или отстающие активы

- Снизить риски путём диверсификации портфеля

- Построить парные стратегии или рыночно-нейтральные позиции

Типы корреляций:📝

- Положительная: активы растут/падают вместе (BTC и ETH)

- Отрицательная: один растёт — другой падает (например, BTC и USDT.D)

- Слабая/нулевая: движения не связаны
import ccxt
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

# Настройки
symbols = ["BTC/USDT", "ETH/USDT", "BNB/USDT", "XRP/USDT", "SOL/USDT"]
exchange = ccxt.binance()
timeframe = "1h"
limit = 200

def fetch_closes():
closes = {}
for symbol in symbols:
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
closes[symbol] = df["close"]
return pd.DataFrame(closes)

# Получение данных и расчёт корреляции
data = fetch_closes()
returns = data.pct_change().dropna()
correlation_matrix = returns.corr()

# Визуализация
plt.figure(figsize=(8, 6))
sns.heatmap(correlation_matrix, annot=True, cmap="coolwarm", vmin=-1, vmax=1)
plt.title("Матрица корреляции между активами")
plt.show()

#индикатор

📌 Подпишись  Crypto Python❗️
👍8❤‍🔥2🔥1
📌 Keltner Channel Breakout – Стратегия пробоя канала Келтнера 🧑‍💻

Суть стратегии:🚀

Keltner Channel — это волатильностный индикатор, основанный на экспоненциальной скользящей средней (EMA) и ATR (среднем истинном диапазоне).

Стратегия пробоя канала Келтнера подразумевает:🛠️

→ Покупку при пробое верхней границы канала,

→ Продажу при пробое нижней границы,

→ Фильтрация ложных сигналов — через объём, RSI, тренд.

import ccxt
import pandas as pd
import ta
import matplotlib.pyplot as plt

# Получаем данные
exchange = ccxt.binance()
symbol = "BTC/USDT"
tf = "1h"
limit = 200

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

# Индикаторы
ema = ta.trend.ema_indicator(df["close"], window=20)
atr = ta.volatility.average_true_range(df["high"], df["low"], df["close"], window=14)
mult = 2.0

df["kc_middle"] = ema
df["kc_upper"] = ema + mult * atr
df["kc_lower"] = ema - mult * atr

# Сигналы
df["signal"] = 0
df.loc[df["close"] > df["kc_upper"], "signal"] = 1 # BUY
df.loc[df["close"] < df["kc_lower"], "signal"] = -1 # SELL

# Визуализация
plt.figure(figsize=(12, 6))
plt.plot(df["ts"], df["close"], label="Close")
plt.plot(df["ts"], df["kc_middle"], label="EMA (20)")
plt.plot(df["ts"], df["kc_upper"], label="Keltner Upper", linestyle='--')
plt.plot(df["ts"], df["kc_lower"], label="Keltner Lower", linestyle='--')
plt.fill_between(df["ts"], df["kc_lower"], df["kc_upper"], color='gray', alpha=0.1)
plt.legend()
plt.title("Keltner Channel Breakout")
plt.show()

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

📌 Подпишись  Crypto Python❗️
🔥5🤔3👍2❤‍🔥1
📌 Neural ODEs for Financial Series – Непрерывные модели временных рядов 🧑‍💻

Суть идеи:🚀

Neural ODEs (Neural Ordinary Differential Equations) — это подход, который моделирует непрерывную эволюцию признаков с помощью нейросетей, решая дифференциальные уравнения вместо дискретных шагов (как в RNN или LSTM).

Когда использовать Neural ODE в трейдинге:🛠️

- Для предсказания кривых цен / доходностей

- Для моделирования динамики риска, ликвидности

- Для сложных алгоритмических стратегий с обучением от среды

import torch
import torch.nn as nn
from torchdiffeq import odeint
import yfinance as yf
import numpy as np
import matplotlib.pyplot as plt

# Загрузка данных
data = yf.download("BTC-USD", period="1mo", interval="1h")
prices = data["Close"].dropna().values
x = np.arange(len(prices))
y = (prices - np.mean(prices)) / np.std(prices)

# Подготовка данных
x_tensor = torch.tensor(x / x.max(), dtype=torch.float32)
y_tensor = torch.tensor(y, dtype=torch.float32)

# ODE модель
class ODEFunc(nn.Module):
def __init__(self):
super().__init__()
self.linear = nn.Sequential(
nn.Linear(1, 50),
nn.Tanh(),
nn.Linear(50, 1)
)

def forward(self, t, y):
return self.linear(y)

# Инициализация
func = ODEFunc()
y0 = y_tensor[0].unsqueeze(0)
t = x_tensor.unsqueeze(1)
optimizer = torch.optim.Adam(func.parameters(), lr=1e-3)

# Обучение
for i in range(1000):
pred_y = odeint(func, y0, t).squeeze()
loss = torch.mean((pred_y - y_tensor) ** 2)
optimizer.zero_grad()
loss.backward()
optimizer.step()
if i % 100 == 0:
print(f"Step {i} | Loss: {loss.item():.6f}")

# Визуализация
with torch.no_grad():
pred_y = odeint(func, y0, t).squeeze().numpy()

plt.plot(y, label="Real")
plt.plot(pred_y, label="Neural ODE Prediction")
plt.legend()
plt.title("Neural ODE for BTC-USD (hourly)")
plt.show()

#индикатор

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

Суть идеи:🛠️

Dynamic Position Sizing — это подход, при котором размер позиции автоматически адаптируется под:🚀

- волатильность рынка (например, через ATR),

- или размер текущего капитала (equity).

Цель — поддерживать стабильный риск на сделку, независимо от условий.💸

import ccxt
import pandas as pd
import talib

# Настройки
exchange = ccxt.binance()
symbol = "BTC/USDT"
timeframe = "1h"
account_equity = 10000 # текущий капитал в USDT
risk_per_trade_pct = 0.01 # риск на сделку (1%)
atr_period = 14
lookback = 100

# Получение данных
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=lookback)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
atr = talib.ATR(df["high"], df["low"], df["close"], timeperiod=atr_period)

# Расчёт объёма позиции
risk_amount = account_equity * risk_per_trade_pct
atr_last = atr.iloc[-1]
position_size = risk_amount / atr_last # в USDT
entry_price = df["close"].iloc[-1]
qty = position_size / entry_price # в BTC

print(f"ATR: {atr_last:.2f} | Risk: {risk_amount} USDT")
print(f"Размер позиции: {qty:.6f} BTC (эквивалентно {position_size:.2f} USDT)")

#индикатор

📌 Подпишись  Crypto Python❗️
🔥9❤‍🔥2
📌 Detrended Price Oscillator (DPO) – Циклические сигналы без влияния тренда🧑‍💻

Суть индикатора:🚀

DPO (Detrended Price Oscillator) — осциллятор, созданный для выявления циклических колебаний цены, игнорируя долгосрочный тренд.

В отличие от большинства осцилляторов, DPO не используется для генерации сигналов тренда, а фокусируется на короткосрочной перекупленности/перепроданности.

import ccxt
import pandas as pd

# Настройки
exchange = ccxt.binance()
symbol = "BTC/USDT"
timeframe = "1h"
period = 14

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

# Расчёт DPO
shift = int(period / 2 + 1)
df["sma"] = df["close"].rolling(window=period).mean()
df["dpo"] = df["close"].shift(shift) - df["sma"]

# Пример последних значений
print(df[["ts", "close", "dpo"]].tail(10))

#индикатор

📌 Подпишись  Crypto Python❗️
👍7❤‍🔥1🔥1
📌 Whale Wallet Tracking Strategy – Реакция на перемещения с крупных кошельков🧑‍💻

Суть стратегии:🚀

Крупные кошельки ("whales") часто принадлежат биржам, фондам, институционалам или ранним инвесторам.

Мониторинг их активности позволяет:

- Предугадывать возможную продажу или покупку,

- Реагировать на переводы на биржи (продажа) или с выводом с бирж (накопление),

- Использовать в стратегиях как фильтр или триггер.

Источники данных:💸

Whale Alert API

Glassnode – аналитика перемещений

Arkham Intelligence – адреса по именам

Onchain API (например, через etherscan, blockchair, bitquery, chainbase)
import requests
import time

API_KEY = "YOUR_WHALE_ALERT_API_KEY"
MIN_VALUE = 500000 # USD
INTERVAL = 60 # сек

def fetch_transfers():
url = "https://api.whale-alert.io/v1/transactions"
params = {
"api_key": API_KEY,
"min_value": MIN_VALUE,
"currency": "btc",
"limit": 10
}
r = requests.get(url, params=params)
data = r.json()
return data.get("transactions", [])

def parse_transfer(tx):
direction = "→ EXCHANGE" if tx["to"]["owner_type"] == "exchange" else "← FROM EXCHANGE"
print(f"[{tx['timestamp']}] {tx['amount']} BTC {direction} | From: {tx['from']['owner']} → {tx['to']['owner']}")

if __name__ == "__main__":
while True:
txs = fetch_transfers()
for tx in txs:
parse_transfer(tx)
time.sleep(INTERVAL)

#индикатор

📌 Подпишись  Crypto Python❗️
👍8🔥4👏2❤‍🔥1
📌 Volatility Regime Classifier – Классификация рынка на флет и тренд🧑‍💻

Суть идеи:🚀

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

Тренд — направленное движение (вверх или вниз),

Флет — боковик, консолидация, низкая волатильность.

Цель стратегии:🛠️
автоматически определить текущий режим рынка, чтобы:

- применять подходящую стратегию (например, трендовая/флетовая),

- фильтровать ложные сигналы,

- управлять рисками.
import ccxt
import pandas as pd
import talib

# Параметры
exchange = ccxt.binance()
symbol = "BTC/USDT"
timeframe = "1h"
atr_period = 14
lookback = 150
volatility_threshold = 100 # например, 100 USDT по BTC

# Получение данных
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=lookback)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")

# Расчёт ATR (волатильности)
df["atr"] = talib.ATR(df["high"], df["low"], df["close"], timeperiod=atr_period)

# Классификация режима
df["regime"] = "flat"
df.loc[df["atr"] > volatility_threshold, "regime"] = "trend"

# Последнее состояние
last = df.iloc[-1]
print(f"Последний режим рынка: {last['regime'].upper()} (ATR = {last['atr']:.2f})")

#индикатор

📌 Подпишись  Crypto Python❗️
👍7🔥4❤‍🔥1
📌 Trade Execution Optimizer – Оптимизация входа с учётом комиссий и проскальзывания🧑‍💻

Суть идеи:🚀

Цель Trade Execution Optimizer — минимизировать издержки при входе/выходе в позицию за счёт:

- правильного выбора типа ордера (market / limit / TWAP / VWAP),

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

- адаптации к ликвидности и объёму актива,

- разбиения крупных ордеров на части (execution slicing).

Факторы, которые учитываются:🛠️

1. Тип ордера:

Market – быстро, но с проскальзыванием

Limit – без проскальзывания, но не факт что исполнится

TWAP/VWAP – равномерное исполнение во времени/объёму

2. Комиссия биржи (taker/maker fees)

3. Глубина стакана (Order Book)

- влияет на проскальзывание при рыночных ордерах

4. Объём сделки относительно ликвидности

- если объём слишком большой → разбивать на части

import ccxt

# Параметры
exchange = ccxt.binance()
symbol = "BTC/USDT"
trade_amount_usdt = 10000
maker_fee = 0.0001
taker_fee = 0.0004

# Получение стакана
order_book = exchange.fetch_order_book(symbol, limit=20)

def estimate_market_execution_cost(order_book, amount_usdt):
total_cost = 0
remaining = amount_usdt
for price, volume in order_book['asks']: # покупка по market
vol_usdt = price * volume
if vol_usdt >= remaining:
total_cost += price * (remaining / price)
break
else:
total_cost += vol_usdt
remaining -= vol_usdt
avg_price = total_cost / amount_usdt
return avg_price

def estimate_limit_execution_cost(price, amount_usdt):
return price # если исполнится – цена фиксированная

# Расчёты
market_price = estimate_market_execution_cost(order_book, trade_amount_usdt)
limit_price = order_book['asks'][0][0] # best ask

cost_market = market_price * (1 + taker_fee)
cost_limit = limit_price * (1 + maker_fee)

print(f"Market исполнение: {market_price:.2f} USDT (с учётом комиссии: {cost_market:.2f})")
print(f"Limit исполнение: {limit_price:.2f} USDT (с учётом комиссии: {cost_limit:.2f})")

if cost_market < cost_limit:
print(" Выгоднее использовать MARKET ордер")
else:
print(" Выгоднее использовать LIMIT ордер")

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

📌 Подпишись  Crypto Python❗️
🔥6👍41❤‍🔥1
📌 Donchian Channel Reversal – Стратегия разворота от границ канала Дончиана🧑‍💻

Суть стратегии:🚀

Канал Дончиана (Donchian Channel) строится по максимуму и минимуму за определённое количество баров (обычно 20).
Вместо пробойной логики, в этой стратегии используется отскок от границ канала, что подходит для флетового рынка или контртрендовой торговли.

Правила стратегии:💸

- Покупка, когда цена коснулась нижней границы и начинает отскакивать

- Продажа, когда цена достигла верхней границы и начинает разворачиваться вниз

import ccxt
import pandas as pd

# Настройки
exchange = ccxt.binance()
symbol = "BTC/USDT"
timeframe = "1h"
lookback = 100
channel_period = 20

# Получение данных
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=lookback)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")

# Канал Дончиана
df["donchian_high"] = df["high"].rolling(window=channel_period).max()
df["donchian_low"] = df["low"].rolling(window=channel_period).min()

# Сигналы разворота
df["signal"] = 0
df.loc[
(df["low"] <= df["donchian_low"]) & (df["close"] > df["open"]),
"signal"
] = 1 # Покупка после отскока от нижней границы

df.loc[
(df["high"] >= df["donchian_high"]) & (df["close"] < df["open"]),
"signal"
] = -1 # Продажа после отскока от верхней границы

# Последний сигнал
last = df.iloc[-1]
print(f"Последняя свеча: {last['ts']}")
print(f"Donchian High: {last['donchian_high']:.2f} | Low: {last['donchian_low']:.2f}")
print(f"Цена закрытия: {last['close']:.2f}")
print(f"Сигнал: {'BUY' if last['signal'] == 1 else 'SELL' if last['signal'] == -1 else 'NONE'}")

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

📌 Подпишись  Crypto Python❗️
👍9🔥1
📌 Risk-Reward Ratio Analyzer – Анализ соотношения риска к прибыли (R:R) по открытым сделкам🧑‍💻

Суть идеи:🛠️

Risk-Reward Ratio (R:R) — это ключевой показатель в управлении капиталом.
Он показывает, сколько потенциальной прибыли вы можете получить на каждый единичный риск (стоп-лосс).

Для чего нужен Risk-Reward Analyzer:🛠️

- Оценить качество сделок, а не только результат

- Отфильтровывать сигналы с плохим соотношением (например, R:R < 1.5)

- Помогать трейдеру принимать осознанные входы

# Пример параметров сделки
entry_price = 27000 # точка входа
stop_loss = 26500 # стоп-лосс
take_profit = 28000 # тейк-профит
position_side = "long" # или "short"

# Расчёт риска и прибыли
if position_side == "long":
risk = entry_price - stop_loss
reward = take_profit - entry_price
else:
risk = stop_loss - entry_price
reward = entry_price - take_profit

# Проверка на ноль
if risk <= 0 or reward <= 0:
print("Некорректные значения (TP и SL должны быть на разных сторонах от входа)")
else:
rr_ratio = reward / risk
print(f"Сделка: {position_side.upper()} | R:R = {rr_ratio:.2f}")
if rr_ratio >= 2:
print(" Отличное соотношение")
elif rr_ratio >= 1:
print("⚠️ Приемлемо")
else:
print(" Низкое соотношение (нежелательная сделка)")

#инструмент

📌 Подпишись  Crypto Python❗️
👍9
📌 Ehler’s Fisher Transform – Определение экстремумов через нормализацию🧑‍💻

Суть индикатора:🛠️

Fisher Transform от Джона Эйлерса — это математическая трансформация, которая переводит данные в псевдо-нормальное распределение, усиливая крайние значения (экстремумы).
Он помогает точнее выявлять разворотные точки, особенно в сочетании с другими осцилляторами.

Типовые сигналы:💸

- Пересечение нуля: возможный разворот

- Выход за уровни ±1.5 или ±2: перекупленность/перепроданность

import ccxt
import pandas as pd
import numpy as np

# Получение данных
exchange = ccxt.binance()
symbol = "BTC/USDT"
tf = "1h"
limit = 300

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

# Fisher Transform
length = 10
price = (df["high"] + df["low"]) / 2
min_low = price.rolling(length).min()
max_high = price.rolling(length).max()
value = 2 * ((price - min_low) / (max_high - min_low + 1e-9) - 0.5)

# Ограничим значения от -0.999 до 0.999
value = value.clip(-0.999, 0.999)

# Применим трансформацию
fisher = 0.5 * np.log((1 + value) / (1 - value))
df["fisher"] = fisher.rolling(3).mean()

# Последние значения
print(df[["ts", "close", "fisher"]].tail(10))

#индикатор

📌 Подпишись  Crypto Python❗️
👍7🔥3
📌Synthetic Asset Arbitrage – Арбитраж между синтетическими и реальными активами🧑‍💻

Суть стратегии:🚀

Синтетические активы (например, sBTC, sETH, cETH, ibBTC) — это токены, которые отражают цену другого актива, но торгуются на других платформах (DeFi-протоколах, мостах, деривативах).

Synthetic Arbitrage предполагает:🛠️

- Покупку актива там, где он дешевле,

- Продажу эквивалентного реального или синтетического актива там, где он дороже,

- Получение прибыли из-за временных дисбалансов в цене между синтетическим и базовым активом.

На что обращать внимание:

- Комиссии за своп, газ, мосты

- Время подтверждения (можно проскочить окно арбитража)

- Лимит ликвидности в пуле

- Возможность обратного обмена (exit liquidity)
import ccxt
import requests

def get_binance_btc_price():
binance = ccxt.binance()
ticker = binance.fetch_ticker("BTC/USDT")
return ticker['last']

def get_synthetix_sbtc_price():
# Пример: через API 1inch или Coingecko (тут демонстрация через Coingecko)
url = "https://api.coingecko.com/api/v3/simple/price"
params = {"ids": "sbtc", "vs_currencies": "usd"}
r = requests.get(url, params=params)
return r.json()["sbtc"]["usd"]

real_btc = get_binance_btc_price()
synthetic_btc = get_synthetix_sbtc_price()

spread = synthetic_btc - real_btc
spread_pct = spread / real_btc * 100

print(f"BTC (Binance): ${real_btc:.2f}")
print(f"sBTC (Synthetix): ${synthetic_btc:.2f}")
print(f"Спред: ${spread:.2f} ({spread_pct:.2f}%)")

if abs(spread_pct) > 0.5:
print("⚠️ Обнаружена возможность арбитража")

#арбитраж

📌 Подпишись  Crypto Python❗️
👍4🔥2
📌 Fractal Adaptive Moving Average (FRAMA) – Адаптивная скользящая средняя на фрактальной основе🧑‍💻

Суть индикатора:🛠️

FRAMA — это усовершенствованная скользящая средняя, предложенная Джоном Эйлерсом, которая:

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

- становится гладкой во флете и быстрой в тренде,

- основана на оценке фрактальной размерности цены (измерение "неровности").
import ccxt
import pandas as pd
import numpy as np

# Настройки
symbol = "BTC/USDT"
timeframe = "1h"
n = 16 # длина окна FRAMA
w = 5 # размер сегмента внутри окна

exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=300)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")

def frama(close, n=16, w=5):
alpha_min = 0.01
alpha_max = 1.0
frama = [close[0]]

for i in range(1, len(close)):
if i < n:
frama.append(close[i])
continue

hl = close[i - n + 1:i + 1]
hl1 = close[i - n + 1:i - n//2 + 1]
hl2 = close[i - n//2 + 1:i + 1]

n1 = (hl1.max() - hl1.min()) / (n // 2)
n2 = (hl2.max() - hl2.min()) / (n // 2)
n3 = (hl.max() - hl.min()) / n

if n1 > 0 and n2 > 0 and n3 > 0:
D = (np.log(n1 + n2) - np.log(n3)) / np.log(2)
alpha = np.exp(-4.6 * (D - 1))
alpha = min(max(alpha, alpha_min), alpha_max)
else:
alpha = alpha_min

frama_val = alpha * close[i] + (1 - alpha) * frama[-1]
frama.append(frama_val)

return pd.Series(frama, index=close.index)

# Применение
df["frama"] = frama(df["close"])

# Вывод последних значений
print(df[["ts", "close", "frama"]].tail(10))

#индикатор

📌 Подпишись  Crypto Python❗️
👍6🔥4
📌 Order Book Depth Shift Detector – Реакция на сдвиг в плотности стакана🧑‍💻

Суть стратегии:🚀

Стакан заявок (Order Book) отражает текущие уровни спроса и предложения.
Depth Shift Detector отслеживает внезапные изменения в плотности ордеров — например:

- Появление/исчезновение крупных стен (bid/ask walls)

- Смещение объёмов ближе/дальше от текущей цены

- Неожиданное перевешивание стороны (ask >> bid или наоборот)

Зачем отслеживать:💸

- Это может быть сигналом к движению цены в сторону "разгрузки"

- Крупные лимитные ордера влияют на ликвидность и исполнение

- В High Frequency Trading и скальпинге — ключевая информация

import ccxt
import time

exchange = ccxt.binance()
symbol = "BTC/USDT"
depth_limit = 20
poll_interval = 3 # секунд

def get_depth_snapshot():
ob = exchange.fetch_order_book(symbol, limit=depth_limit)
bids = {round(price, 2): size for price, size in ob['bids']}
asks = {round(price, 2): size for price, size in ob['asks']}
return bids, asks

prev_bids, prev_asks = get_depth_snapshot()

while True:
time.sleep(poll_interval)
bids, asks = get_depth_snapshot()

for price in bids:
prev_size = prev_bids.get(price, 0)
curr_size = bids[price]
if curr_size > prev_size * 2 and curr_size > 10:
print(f"🔵 Увеличение BID на {price}: {prev_size:.2f} → {curr_size:.2f}")
elif curr_size < prev_size * 0.5 and prev_size > 10:
print(f"⚠️ Снятие BID на {price}: {prev_size:.2f} → {curr_size:.2f}")

for price in asks:
prev_size = prev_asks.get(price, 0)
curr_size = asks[price]
if curr_size > prev_size * 2 and curr_size > 10:
print(f"🔴 Увеличение ASK на {price}: {prev_size:.2f} → {curr_size:.2f}")
elif curr_size < prev_size * 0.5 and prev_size > 10:
print(f"⚠️ Снятие ASK на {price}: {prev_size:.2f} → {curr_size:.2f}")

prev_bids, prev_asks = bids, asks

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

📌 Подпишись  Crypto Python❗️
🔥10❤‍🔥2
📌 Hull Moving Average Crossover – Cтратегия пересечения HMA🧑‍💻

Суть индикатора:🚀

Hull Moving Average (HMA) — это гладкая и быстро реагирующая скользящая средняя, разработанная Аланом Халлом.
Она снижает запаздывание обычных MA (например, EMA) и при этом остаётся плавной.

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

Сигналы стратегии:🛠️

- Buy: HMA_fast пересекает HMA_slow снизу вверх

- Sell: HMA_fast пересекает HMA_slow сверху вниз

import ccxt
import pandas as pd
import numpy as np

def wma(series, period):
weights = np.arange(1, period + 1)
return series.rolling(period).apply(lambda prices: np.dot(prices, weights)/weights.sum(), raw=True)

def hma(series, period):
half = int(period / 2)
sqrt_p = int(np.sqrt(period))

wma_half = wma(series, half)
wma_full = wma(series, period)
raw = 2 * wma_half - wma_full
return wma(raw, sqrt_p)

# Данные с Binance
exchange = ccxt.binance()
symbol = "BTC/USDT"
tf = "1h"
ohlcv = exchange.fetch_ohlcv(symbol, timeframe=tf, limit=200)
df = pd.DataFrame(ohlcv, columns=["ts", "open", "high", "low", "close", "volume"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")

# HMA
df["hma_fast"] = hma(df["close"], 14)
df["hma_slow"] = hma(df["close"], 55)

# Сигналы пересечения
df["signal"] = 0
df.loc[(df["hma_fast"] > df["hma_slow"]) & (df["hma_fast"].shift() <= df["hma_slow"].shift()), "signal"] = 1 # BUY
df.loc[(df["hma_fast"] < df["hma_slow"]) & (df["hma_fast"].shift() >= df["hma_slow"].shift()), "signal"] = -1 # SELL

# Вывод последних сигналов
print(df[["ts", "close", "hma_fast", "hma_slow", "signal"]].tail(10))

#индикатор

📌 Подпишись  Crypto Python❗️
❤‍🔥6🔥4👍3