Алгоритмический трейдинг: Создание, тестирование и запуск роботов на рынке Форекс

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("Ошибка загрузки данных!")