Читать книгу Алгоритмический трейдинг: Создание, тестирование и запуск роботов на рынке Форекс - Иван Алексеевич Евдокимов - Страница 4
Часть 2: Галерея стратегий: 10 роботов для Форекса
ОглавлениеРобот 1: «Следователь за трендом» (Trend Follower) – Python версия для Backtesting
Вот полная реализация стратегии Trend Follower на Python с использованием библиотеки backtrader для бэктестинга и анализа:
```python
# trend_follower_backtest.py
import backtrader as bt
import backtrader.analyzers as btanalyzers
import pandas as pd
import yfinance as yf
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
# =====================================================================
# 1. СОЗДАНИЕ СТРАТЕГИИ
# =====================================================================
class TrendFollowerStrategy(bt.Strategy):
"""
Стратегия следования за трендом
Правила:
1. Покупка: когда быстрая MA (50) пересекает медленную (200) снизу вверх И ADX > 25
2. Закрытие: при обратном пересечении MA или по стоп-лоссу/тейк-профиту
"""
params = (
('fast_ma', 50), # Период быстрой скользящей средней
('slow_ma', 200), # Период медленной скользящей средней
('adx_period', 14), # Период ADX
('adx_threshold', 25),# Пороговое значение ADX для силы тренда
('stop_loss', 0.02), # Стоп-лосс 2%
('take_profit', 0.04),# Тейк-профит 4%
('risk_per_trade', 0.01), # Риск на сделку 1% от капитала
('use_atr_filter', True), # Использовать фильтр ATR
('atr_period', 14), # Период ATR
('atr_multiplier', 2),# Множитель ATR для стоп-лосса
('printlog', False), # Вывод логов
)
def __init__(self):
"""Инициализация индикаторов"""
# Создаем индикаторы
self.fast_ma = bt.indicators.SimpleMovingAverage(
self.data.close, period=self.params.fast_ma
)
self.slow_ma = bt.indicators.SimpleMovingAverage(
self.data.close, period=self.params.slow_ma
)
# ADX индикатор (направленность тренда)
self.adx = bt.indicators.AverageDirectionalMovementIndex(
self.data, period=self.params.adx_period
)
# ATR для фильтра волатильности и расчета стоп-лосса
self.atr = bt.indicators.AverageTrueRange(
self.data, period=self.params.atr_period
)
# Переменные для отслеживания состояния
self.order = None
self.buyprice = None
self.buycomm = None
# Для логирования
self.trades = []
self.current_trade = {}
def log(self, txt, dt=None, doprint=False):
"""Метод для логирования"""
if self.params.printlog or doprint:
dt = dt or self.datas[0].datetime.date(0)
print(f'{dt.isoformat()} {txt}')
def notify_order(self, order):
"""Обработка изменения статуса ордера"""
if order.status in [order.Submitted, order.Accepted]:
# Ордер отправлен/принят – ничего не делаем
return
if order.status in [order.Completed]:
if order.isbuy():
self.log(
f'ПОКУПКА ИСПОЛНЕНА, Цена: {order.executed.price:.2f}, '
f'Стоимость: {order.executed.value:.2f}, '
f'Комиссия: {order.executed.comm:.2f}'
)
self.buyprice = order.executed.price
self.buycomm = order.executed.comm
# Устанавливаем стоп-лосс и тейк-профит
if self.params.use_atr_filter:
sl = self.buyprice – self.atr[0] * self.params.atr_multiplier
tp = self.buyprice + self.atr[0] * self.params.atr_multiplier * 2
else:
sl = self.buyprice * (1 – self.params.stop_loss)
tp = self.buyprice * (1 + self.params.take_profit)
self.sell(exectype=bt.Order.StopTrail, price=sl)
self.sell(exectype=bt.Order.Limit, price=tp)
# Сохраняем информацию о сделке
self.current_trade = {
'entry_date': self.datas[0].datetime.date(0),
'entry_price': order.executed.price,
'stop_loss': sl,
'take_profit': tp,
'size': order.executed.size
}
elif order.issell():
self.log(
f'ПРОДАЖА ИСПОЛНЕНА, Цена: {order.executed.price:.2f}, '
f'Стоимость: {order.executed.value:.2f}, '
f'Комиссия: {order.executed.comm:.2f}'
)
# Фиксируем завершение сделки
if self.current_trade:
self.current_trade['exit_date'] = self.datas[0].datetime.date(0)
self.current_trade['exit_price'] = order.executed.price
self.current_trade['pnl'] = (
(order.executed.price – self.current_trade['entry_price']) *
self.current_trade['size']
)
self.current_trade['pnl_percent'] = (
(order.executed.price / self.current_trade['entry_price'] – 1) * 100
)
self.trades.append(self.current_trade.copy())
self.current_trade = {}
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log(f'Ордер отменен/отклонен: {order.getstatusname()}')
self.order = None
def notify_trade(self, trade):
"""Обработка изменения статуса сделки"""
if not trade.isclosed:
return
self.log(f'СДЕЛКА ЗАКРЫТА, Прибыль: {trade.pnl:.2f}, Чистая прибыль: {trade.pnlcomm:.2f}')
def next(self):
"""Основная логика на каждом новом баре"""
# Пропускаем если ордер уже открыт
if self.order:
return
# Проверяем, есть ли открытая позиция
if not self.position:
# УСЛОВИЕ ДЛЯ ПОКУПКИ:
# 1. Быстрая MA пересекает медленную снизу вверх
# 2. ADX > порогового значения (сила тренда)
crossover_up = (self.fast_ma[-1] < self.slow_ma[-1] and
self.fast_ma[0] > self.slow_ma[0])
strong_trend = self.adx[0] > self.params.adx_threshold
if crossover_up and strong_trend:
# Рассчитываем размер позиции на основе риска
capital = self.broker.getvalue()
risk_amount = capital * self.params.risk_per_trade
if self.params.use_atr_filter:
stop_distance = self.atr[0] * self.params.atr_multiplier
else:
stop_distance = self.data.close[0] * self.params.stop_loss
if stop_distance > 0:
size = risk_amount / stop_distance
size = min(size, capital * 0.1 / self.data.close[0]) # Не более 10% капитала
self.log(f'СИГНАЛ НА ПОКУПКУ: цена={self.data.close[0]:.2f}, '
f'ADX={self.adx[0]:.2f}, размер={size:.2f}')
self.order = self.buy(size=size)
else:
# УСЛОВИЕ ДЛЯ ЗАКРЫТИЯ ПОЗИЦИИ:
# Быстрая MA пересекает медленную сверху вниз
crossover_down = (self.fast_ma[-1] > self.slow_ma[-1] and
self.fast_ma[0] < self.slow_ma[0])
if crossover_down:
self.log(f'СИГНАЛ НА ЗАКРЫТИЕ: цена={self.data.close[0]:.2f}')
self.order = self.close()
# =====================================================================
# 2. КЛАСС ДЛЯ ОПТИМИЗАЦИИ ПАРАМЕТРОВ
# =====================================================================
class TrendFollowerOptimizer(bt.Strategy):
"""Версия стратегии для оптимизации параметров"""
params = (
('fast_ma', 50),
('slow_ma', 200),
('adx_threshold', 25),
('risk_per_trade', 0.01),
)
def __init__(self):
self.fast_ma = bt.indicators.SMA(self.data.close, period=self.p.fast_ma)
self.slow_ma = bt.indicators.SMA(self.data.close, period=self.p.slow_ma)
self.adx = bt.indicators.ADX(self.data, period=14)
self.order = None
def next(self):
if self.order:
return
if not self.position:
if (self.fast_ma[-1] < self.slow_ma[-1] and
self.fast_ma[0] > self.slow_ma[0] and
self.adx[0] > self.p.adx_threshold):
size = self.broker.getvalue() * self.p.risk_per_trade / self.data.close[0]
self.order = self.buy(size=size)
else:
if self.fast_ma[-1] > self.slow_ma[-1] and self.fast_ma[0] < self.slow_ma[0]:
self.order = self.close()
# =====================================================================
# 3. ФУНКЦИИ ДЛЯ ТЕСТИРОВАНИЯ И АНАЛИЗА
# =====================================================================
def download_data(symbol='EURUSD=X', start_date='2020-01-01', end_date='2023-12-31'):
"""Загрузка данных с Yahoo Finance"""
print(f"Загрузка данных для {symbol}…")
data = yf.download(symbol, start=start_date, end=end_date, progress=False)
# Преобразуем индексы для backtrader
data.index = pd.to_datetime(data.index)
data = data[['Open', 'High', 'Low', 'Close', 'Volume']]
return data
def run_backtest(data, initial_cash=10000, commission=0.001, **strategy_params):
"""Запуск бэктеста"""
cerebro = bt.Cerebro()
# Добавляем данные
data_feed = bt.feeds.PandasData(dataname=data)
cerebro.adddata(data_feed)
# Добавляем стратегию
cerebro.addstrategy(TrendFollowerStrategy, **strategy_params)
# Настройки брокера
cerebro.broker.setcash(initial_cash)
cerebro.broker.setcommission(commission=commission)
# Добавляем анализаторы
cerebro.addanalyzer(btanalyzers.Returns, _name='returns')
cerebro.addanalyzer(btanalyzers.SharpeRatio, _name='sharpe', riskfreerate=0.0)
cerebro.addanalyzer(btanalyzers.DrawDown, _name='drawdown')
cerebro.addanalyzer(btanalyzers.TradeAnalyzer, _name='trades')
cerebro.addanalyzer(btanalyzers.SQN, _name='sqn')
print(f'Начальный капитал: {initial_cash:.2f}')
print(f'Комиссия: {commission*100:.2f}%')
# Запуск
results = cerebro.run()
strat = results[0]
# Вывод результатов
print('\n' + '='*50)
print('РЕЗУЛЬТАТЫ БЭКТЕСТА')
print('='*50)
final_value = cerebro.broker.getvalue()
total_return = (final_value / initial_cash – 1) * 100
print(f'Конечный капитал: {final_value:.2f}')
print(f'Общая доходность: {total_return:.2f}%')
# Анализ результатов
if hasattr(strat, 'analyzers'):
returns = strat.analyzers.returns.get_analysis()
sharpe = strat.analyzers.sharpe.get_analysis()
drawdown = strat.analyzers.drawdown.get_analysis()
trades = strat.analyzers.trades.get_analysis()
sqn = strat.analyzers.sqn.get_analysis()
print(f'\nАНАЛИТИКА:')
print(f'Sharpe Ratio: {sharpe.get("sharperatio", 0):.3f}')
print(f'Максимальная просадка: {drawdown.max.drawdown:.2f}%')
print(f'Продолжительность просадки: {drawdown.max.len} дней')
if trades.total.total:
print(f'\nСТАТИСТИКА СДЕЛОК:')
print(f'Всего сделок: {trades.total.total}')
print(f'Прибыльных: {trades.won.total} ({trades.won.total/trades.total.total*100:.1f}%)')
print(f'Убыточных: {trades.lost.total} ({trades.lost.total/trades.total.total*100:.1f}%)')
print(f'Profit Factor: {trades.won.pnl.total/abs(trades.lost.pnl.total):.2f}')
print(f'Средняя прибыль: {trades.won.pnl.average:.2f}')
print(f'Средний убыток: {trades.lost.pnl.average:.2f}')
print(f'SQN: {sqn.sqn:.2f}')
# Возвращаем детали сделок для дальнейшего анализа
trades_details = getattr(strat, 'trades', [])
return cerebro, strat, trades_details
def optimize_parameters(data, parameter_ranges):
"""Оптимизация параметров стратегии"""
cerebro = bt.Cerebro(optreturn=False)
# Добавляем данные
data_feed = bt.feeds.PandasData(dataname=data)
cerebro.adddata(data_feed)
# Добавляем стратегию с параметрами для оптимизации
cerebro.optstrategy(
TrendFollowerOptimizer,
fast_ma=parameter_ranges.get('fast_ma', range(10, 101, 10)),
slow_ma=parameter_ranges.get('slow_ma', range(50, 301, 50)),
adx_threshold=parameter_ranges.get('adx_threshold', range(20, 41, 5)),
risk_per_trade=parameter_ranges.get('risk_per_trade', [0.005, 0.01, 0.02])
)
# Настройки
cerebro.broker.setcash(10000)
cerebro.broker.setcommission(commission=0.001)
# Оптимизация
print("Запуск оптимизации параметров…")
opt_results = cerebro.run(maxcpus=1)
# Анализ результатов оптимизации
results = []
for run in opt_results:
for strat in run:
# Собираем статистику для каждого набора параметров
value = cerebro.broker.getvalue()
returns = (value / 10000 – 1) * 100
results.append({
'fast_ma': strat.params.fast_ma,
'slow_ma': strat.params.slow_ma,
'adx_threshold': strat.params.adx_threshold,
'risk_per_trade': strat.params.risk_per_trade,
'final_value': value,
'returns': returns
})
# Создаем DataFrame с результатами
results_df = pd.DataFrame(results)
results_df = results_df.sort_values('final_value', ascending=False)
return results_df
def plot_results(cerebro, save_path='trend_follower_results.png'):
"""Визуализация результатов"""
# График 1: Кривая баланса
fig = cerebro.plot(style='candlestick', iplot=False)[0][0]
fig.set_size_inches(14, 8)
fig.suptitle('Trend Follower Strategy – Equity Curve', fontsize=16)
# Сохраняем график
plt.tight_layout()
plt.savefig(save_path, dpi=100, bbox_inches='tight')
plt.show()
return fig
def generate_report(strategy_params, trades_details, filename='trend_follower_report.txt'):
"""Генерация детального отчета"""
with open(filename, 'w', encoding='utf-8') as f:
f.write("="*60 + "\n")
f.write("ОТЧЕТ ПО СТРАТЕГИИ TREND FOLLOWER\n")
f.write("="*60 + "\n\n")
f.write("ПАРАМЕТРЫ СТРАТЕГИИ:\n")
f.write("-"*40 + "\n")
for key, value in strategy_params.items():
f.write(f"{key}: {value}\n")
f.write("\n" + "="*60 + "\n\n")
if trades_details:
f.write("ДЕТАЛИ СДЕЛОК:\n")
f.write("-"*40 + "\n")
total_pnl = 0
winning_trades = 0
for i, trade in enumerate(trades_details, 1):
pnl = trade.get('pnl', 0)
pnl_percent = trade.get('pnl_percent', 0)
f.write(f"\nСделка #{i}:\n")
f.write(f" Дата входа: {trade.get('entry_date')}\n")
f.write(f" Цена входа: {trade.get('entry_price', 0):.4f}\n")
f.write(f" Дата выхода: {trade.get('exit_date')}\n")
f.write(f" Цена выхода: {trade.get('exit_price', 0):.4f}\n")
f.write(f" P&L: {pnl:.2f} ({pnl_percent:.2f}%)\n")
total_pnl += pnl
if pnl > 0:
winning_trades += 1
f.write("\n" + "="*60 + "\n")
f.write(f"ИТОГИ:\n")
f.write(f"Всего сделок: {len(trades_details)}\n")
f.write(f"Прибыльных сделок: {winning_trades} ({winning_trades/len(trades_details)*100:.1f}%)\n")
f.write(f"Общий P&L: {total_pnl:.2f}\n")
print(f"Отчет сохранен в файл: {filename}")
# =====================================================================
# 4. ОСНОВНОЙ СКРИПТ ДЛЯ ЗАПУСКА
# =====================================================================
def main():
"""Основная функция для запуска бэктеста"""
print("="*60)
print("TREND FOLLOWER STRATEGY BACKTEST")
print("="*60)
# 1. Загружаем данные
symbol = 'EURUSD=X' # EUR/USD
data = download_data(
symbol=symbol,
start_date='2020-01-01',
end_date='2023-12-31'
)
if data.empty:
print("Ошибка загрузки данных!")
