Python для трейдинга: как создать прибыльного торгового робота

# Простая функция для расчета скользящей средней
def calculate_sma(prices, window):
if len(prices) < window:
return None
return sum(prices[-window:]) / window

# Пример использования
closing_prices = [100, 102, 104, 103, 105]
sma_5 = calculate_sma(closing_prices, 5)
print(f"5-дневная SMA: {sma_5}")

pip install pandas numpy matplotlib plotly ta-lib backtrader ccxt alpaca-trade-api

import pandas as pd
import numpy as np
import yfinance as yf

# Загрузка данных акций Tesla за последний год
tesla = yf.download('TSLA', period='1y')

# Базовый анализ данных
print(f"Форма данных: {tesla.shape}")
print(f"Последние 5 записей:\n{tesla.tail()}")

# Расчет дневной доходности
tesla['Returns'] = tesla['Close'].pct_change()
print(f"Средняя дневная доходность: {tesla['Returns'].mean()*100:.2f}%")
print(f"Волатильность: {tesla['Returns'].std()*100:.2f}%")

# Векторизация расчетов с NumPy
# В 100-1000 раз быстрее, чем циклы Python

# Расчет 20-дневной экспоненциальной скользящей средней (EMA)
tesla['EMA20'] = tesla['Close'].ewm(span=20, adjust=False).mean()

# Расчет стандартного отклонения цен закрытия за 20 дней
tesla['StdDev'] = tesla['Close'].rolling(window=20).std()

# Расчет полос Боллинджера
tesla['UpperBand'] = tesla['EMA20'] + (tesla['StdDev'] * 2)
tesla['LowerBand'] = tesla['EMA20'] – (tesla['StdDev'] * 2)

import matplotlib.pyplot as plt
from mplfinance import make_marketcolors, make_mpf_style, plot

# Создание стиля для графика свечей
mc = make_marketcolors(up='green', down='red', 
edge='inherit', wick='black')
s = make_mpf_style(marketcolors=mc)

# Построение графика свечей
plot(tesla[-60:], type='candle', style=s, 
title='TSLA – Последние 60 дней',
volume=True, figsize=(12, 7))

import matplotlib.pyplot as plt

# Создание графика с несколькими индикаторами
fig, (ax1, ax2, ax3) = plt.subplots(3, 1, figsize=(12, 10), sharex=True)

# График цен и скользящих средних
ax1.plot(tesla.index, tesla['Close'], label='Цена закрытия')
ax1.plot(tesla.index, tesla['EMA20'], label='EMA 20', color='orange')
ax1.fill_between(tesla.index, tesla['UpperBand'], tesla['LowerBand'], alpha=0.2, color='gray')
ax1.set_title('TSLA с полосами Боллинджера')
ax1.legend()

# График объема
ax2.bar(tesla.index, tesla['Volume'], color='blue', alpha=0.5)
ax2.set_title('Объем торгов')

# График RSI
tesla['RSI'] = calculate_rsi(tesla['Close'], 14) # Предполагаем, что функция calculate_rsi определена
ax3.plot(tesla.index, tesla['RSI'], color='purple')
ax3.axhline(70, color='red', linestyle='--', alpha=0.5)
ax3.axhline(30, color='green', linestyle='--', alpha=0.5)
ax3.set_title('RSI (14)')
ax3.set_ylim(0, 100)

plt.tight_layout()
plt.show()

import plotly.graph_objects as go
from plotly.subplots import make_subplots

# Создание интерактивного графика с Plotly
fig = make_subplots(rows=2, cols=1, shared_xaxes=True, 
vertical_spacing=0.1, row_heights=[0\.7, 0.3])

# Добавление свечей
fig.add_trace(
go.Candlestick(x=tesla.index, open=tesla['Open'], high=tesla['High'],
low=tesla['Low'], close=tesla['Close'], name='TSLA'),
row=1, col=1
)

# Добавление линий скользящих средних
fig.add_trace(
go.Scatter(x=tesla.index, y=tesla['EMA20'], name='EMA 20', line=dict(color='orange')),
row=1, col=1
)

# Добавление объема
fig.add_trace(
go.Bar(x=tesla.index, y=tesla['Volume'], name='Volume', marker_color='blue'),
row=2, col=1
)

# Настройка внешнего вида
fig.update_layout(
title='TSLA Interactive Chart',
yaxis_title='Цена (USD)',
xaxis_rangeslider_visible=False,
height=800
)

fig.show()

# Библиотека TA-Lib значительно упрощает расчет индикаторов
import talib as ta

# Добавим основные индикаторы к нашим данным
tesla['SMA20'] = ta.SMA(tesla['Close'], timeperiod=20)
tesla['SMA50'] = ta.SMA(tesla['Close'], timeperiod=50)
tesla['RSI'] = ta.RSI(tesla['Close'], timeperiod=14)
tesla['MACD'], tesla['MACD_Signal'], tesla['MACD_Hist'] = ta.MACD(
tesla['Close'], fastperiod=12, slowperiod=26, signalperiod=9)
tesla['ATR'] = ta.ATR(tesla['High'], tesla['Low'], tesla['Close'], timeperiod=14)
tesla['Bollinger_Upper'], tesla['Bollinger_Middle'], tesla['Bollinger_Lower'] = ta.BBANDS(
tesla['Close'], timeperiod=20, nbdevup=2, nbdevdn=2)

# Создание сигналов на основе пересечения скользящих средних
tesla['Signal'] = 0 # 0 = нет позиции, 1 = покупка, -1 = продажа

# Сигнал на покупку: SMA20 пересекает SMA50 снизу вверх
buy_signal = (tesla['SMA20'].shift(1) < tesla['SMA50'].shift(1)) & (tesla['SMA20'] > tesla['SMA50'])
tesla.loc[buy_signal, 'Signal'] = 1

# Сигнал на продажу: SMA20 пересекает SMA50 сверху вниз
sell_signal = (tesla['SMA20'].shift(1) > tesla['SMA50'].shift(1)) & (tesla['SMA20'] < tesla['SMA50'])
tesla.loc[sell_signal, 'Signal'] = -1

# Подсчет числа сигналов
print(f"Количество сигналов на покупку: {len(tesla[tesla['Signal'] == 1])}")
print(f"Количество сигналов на продажу: {len(tesla[tesla['Signal'] == -1])}")

# Расширенная стратегия: пересечение MA + подтверждение RSI + объем
tesla['Advanced_Signal'] = 0

# Условия для покупки
buy_conditions = (
(tesla['SMA20'] > tesla['SMA50']) & # Тренд вверх
(tesla['RSI'] > 50) & # Сила тренда
(tesla['RSI'] < 70) & # Не перекуплен
(tesla['Volume'] > tesla['Volume'].rolling(20).mean()) # Повышенный объем
)
tesla.loc[buy_conditions, 'Advanced_Signal'] = 1

# Условия для продажи
sell_conditions = (
(tesla['SMA20'] < tesla['SMA50']) | # Тренд вниз
(tesla['RSI'] > 70) | # Перекуплен
(tesla['Close'] < tesla['Bollinger_Lower']) # Пробой нижней полосы Боллинджера
)
tesla.loc[sell_conditions & (tesla['Advanced_Signal'].shift(1) == 1), 'Advanced_Signal'] = -1

# Результаты стратегии
print(f"Улучшенных сигналов на покупку: {len(tesla[tesla['Advanced_Signal'] == 1])}")
print(f"Улучшенных сигналов на продажу: {len(tesla[tesla['Advanced_Signal'] == -1])}")

# Моделирование результатов стратегии
tesla['Strategy_Returns'] = tesla['Signal'].shift(1) * tesla['Returns']
tesla['Cumulative_Strategy_Returns'] = (1 + tesla['Strategy_Returns']).cumprod()
tesla['Cumulative_Market_Returns'] = (1 + tesla['Returns']).cumprod()

# Сравнение результатов
print(f"Общая доходность стратегии: {(tesla['Cumulative_Strategy_Returns'].iloc[-1] – 1) * 100:.2f}%")
print(f"Общая доходность рынка: {(tesla['Cumulative_Market_Returns'].iloc[-1] – 1) * 100:.2f}%")

import backtrader as bt
import datetime as dt

# Определяем нашу стратегию
class SMACrossStrategy(bt.Strategy):
params = (
('fast_period', 20),
('slow_period', 50),
)

def __init__(self):
# Сохраняем ссылки на SMA
self.fast_sma = bt.indicators.SMA(period=self.params.fast_period)
self.slow_sma = bt.indicators.SMA(period=self.params.slow_period)
self.crossover = bt.indicators.CrossOver(self.fast_sma, self.slow_sma)

# Для отслеживания заказов и позиций
self.order = None

def next(self):
# Если уже есть активный ордер – пропускаем
if self.order:
return

# Если нет открытой позиции
if not self.position:
# Покупаем при пересечении снизу вверх
if self.crossover > 0:
self.order = self.buy()
print(f'Покупка: {self.data.datetime.date(0)} | Цена: {self.data.close[0]:.2f}')

# Если есть открытая позиция
else:
# Продаем при пересечении сверху вниз
if self.crossover < 0:
self.order = self.sell()
print(f'Продажа: {self.data.datetime.date(0)} | Цена: {self.data.close[0]:.2f}')

def notify_order(self, order):
if order.status in [order.Completed]:
self.order = None

# Создаем движок для бэктестинга
cerebro = bt.Cerebro()

# Добавляем данные
data = bt.feeds.PandasData(dataname=tesla, 
fromdate=dt.datetime(2021, 1, 1),
todate=dt.datetime(2022, 1, 1))
cerebro.adddata(data)

# Добавляем стратегию
cerebro.addstrategy(SMACrossStrategy)

# Устанавливаем начальный капитал
cerebro.broker.setcash(10000.0)

# Устанавливаем комиссию 0.1%
cerebro.broker.setcommission(commission=0.001)

# Запускаем бэктест
print(f'Начальный капитал: {cerebro.broker.getvalue():.2f}')
cerebro.run()
print(f'Конечный капитал: {cerebro.broker.getvalue():.2f}')

# Визуализация результатов
cerebro.plot(style='candle', barup='green', bardown='red')

# Для расчета дополнительных метрик эффективности
from backtrader.analyzers import SharpeRatio, DrawDown, Returns, TradeAnalyzer

# Добавляем анализаторы
cerebro.addanalyzer(SharpeRatio, _name='sharpe')
cerebro.addanalyzer(DrawDown, _name='drawdown')
cerebro.addanalyzer(Returns, _name='returns')
cerebro.addanalyzer(TradeAnalyzer, _name='trades')

# Запускаем с анализаторами
results = cerebro.run()
strategy = results[0]

# Выводим результаты анализа
print(f"Коэффициент Шарпа: {strategy.analyzers.sharpe.get_analysis()['sharperatio']:.3f}")
print(f"Максимальная просадка: {strategy.analyzers.drawdown.get_analysis()['max']['drawdown']:.2f}%")
print(f"Годовая доходность: {strategy.analyzers.returns.get_analysis()['rnorm100']:.2f}%")

trade_analysis = strategy.analyzers.trades.get_analysis()
print(f"Всего сделок: {trade_analysis['total']['total']}")
print(f"Выигрышных сделок: {trade_analysis['won']['total']} ({trade_analysis['won']['total']/trade_analysis['total']['total']*100:.1f}%)")
print(f"Проигрышных сделок: {trade_analysis['lost']['total']} ({trade_analysis['lost']['total']/trade_analysis['total']['total']*100:.1f}%)")

# Оптимизация параметров стратегии
cerebro = bt.Cerebro(optreturn=False) # не сохраняем полные результаты для экономии памяти
cerebro.adddata(data)
cerebro.optstrategy(
SMACrossStrategy,
fast_period=range(10, 31, 5), # 10, 15, 20, 25, 30
slow_period=range(40, 61, 5) # 40, 45, 50, 55, 60
)
cerebro.broker.setcash(10000.0)
cerebro.broker.setcommission(commission=0.001)
cerebro.addanalyzer(Returns, _name='returns')

# Запускаем оптимизацию
opt_results = cerebro.run()

# Анализируем результаты
for run in opt_results:
params = run[0].params
returns = run[0].analyzers.returns.get_analysis()
annualized_return = returns.get('rnorm100', 0)

print(f"SMA Fast: {params.fast_period}, SMA Slow: {params.slow_period}, Return: {annualized_return:.2f}%")

from alpaca_trade_api.rest import REST, TimeFrame
import backtrader as bt

# Создаем класс для доступа к данным Alpaca
class AlpacaStore(bt.stores.AlpacaStore):
def __init__(self):
super().__init__(
key_id='YOUR_API_KEY',
secret_key='YOUR_SECRET_KEY',
paper=True # True для бумажной торговли, False для реальной
)

# Инициализируем Backtrader с живой торговлей
cerebro = bt.Cerebro()
store = AlpacaStore()

# Подключаем брокера и данные
broker = store.getbroker()
cerebro.setbroker(broker)

# Добавляем данные (TSLA с 1-минутными свечами)
data = store.getdata(dataname='TSLA', timeframe=bt.TimeFrame.Minutes)
cerebro.adddata(data)

# Добавляем нашу оптимизированную стратегию
cerebro.addstrategy(SMACrossStrategy, fast_period=15, slow_period=50)

# Запускаем торговлю
cerebro.run()