import MetaTrader5 as mt5 import pandas as pd import numpy as np import matplotlib.pyplot as plt from datetime import datetime, timedelta import pytz # Путь к MetaTrader 5 терминалу terminal_path = "C:/Program Files/ForexBroker - MetaTrader 5/Arima/terminal64.exe" def remove_duplicate_indices(df): """Удаляет дублирующиеся индексы, сохраняя только первую строку с уникальным индексом.""" return df[~df.index.duplicated(keep='first')] def get_historical_data(start_date, end_date, terminal_path): if not mt5.initialize(path=terminal_path): print(f"Failed to connect to MetaTrader 5 terminal at {terminal_path}") return None symbols = ["AUDUSD", "AUDJPY", "CADJPY", "AUDCHF", "AUDNZD", "USDCAD", "USDCHF", "USDJPY", "NZDUSD", "GBPUSD", "EURUSD", "CADCHF", "CHFJPY", "NZDCAD", "NZDCHF", "NZDJPY", "GBPCAD", "GBPCHF", "GBPJPY", "GBPNZD", "EURCAD", "EURCHF", "EURGBP", "EURJPY", "EURNZD"] historical_data = {} for symbol in symbols: timeframe = mt5.TIMEFRAME_M1 rates = mt5.copy_rates_range(symbol, timeframe, start_date, end_date) if rates is not None and len(rates) > 0: df = pd.DataFrame(rates) df['time'] = pd.to_datetime(df['time'], unit='s') df.set_index('time', inplace=True) df = df[['open', 'high', 'low', 'close']] df['bid'] = df['close'] # Упрощение: используем 'close' как 'bid' df['ask'] = df['close'] + 0.000001 # Упрощение: добавляем спред historical_data[symbol] = df mt5.shutdown() return historical_data def calculate_synthetic_prices(data): synthetic_prices = {} pairs = [('AUDUSD', 'USDCHF'), ('AUDUSD', 'NZDUSD'), ('AUDUSD', 'USDJPY'), ('USDCHF', 'USDCAD'), ('USDCHF', 'NZDCHF'), ('USDCHF', 'CHFJPY'), ('USDJPY', 'USDCAD'), ('USDJPY', 'NZDJPY'), ('USDJPY', 'GBPJPY'), ('NZDUSD', 'NZDCAD'), ('NZDUSD', 'NZDCHF'), ('NZDUSD', 'NZDJPY'), ('GBPUSD', 'GBPCAD'), ('GBPUSD', 'GBPCHF'), ('GBPUSD', 'GBPJPY'), ('EURUSD', 'EURCAD'), ('EURUSD', 'EURCHF'), ('EURUSD', 'EURJPY'), ('CADCHF', 'CADJPY'), ('CADCHF', 'GBPCAD'), ('CADCHF', 'EURCAD'), ('CHFJPY', 'GBPCHF'), ('CHFJPY', 'EURCHF'), ('CHFJPY', 'NZDCHF'), ('NZDCAD', 'NZDJPY'), ('NZDCAD', 'GBPNZD'), ('NZDCAD', 'EURNZD'), ('NZDCHF', 'NZDJPY'), ('NZDCHF', 'GBPNZD'), ('NZDCHF', 'EURNZD'), ('NZDJPY', 'GBPNZD'), ('NZDJPY', 'EURNZD')] for pair1, pair2 in pairs: if pair1 in data and pair2 in data: synthetic_prices[f'{pair1}_{pair2}_1'] = data[pair1]['bid'] / data[pair2]['ask'] synthetic_prices[f'{pair1}_{pair2}_2'] = data[pair1]['bid'] / data[pair2]['bid'] return pd.DataFrame(synthetic_prices) def analyze_arbitrage(data, synthetic_prices): spreads = {} for pair in data.keys(): for synth_pair in synthetic_prices.columns: if pair in synth_pair: spreads[synth_pair] = data[pair]['bid'] - synthetic_prices[synth_pair] arbitrage_opportunities = pd.DataFrame(spreads) > 0.00008 return arbitrage_opportunities def simulate_trade(data, direction, entry_price, take_profit, stop_loss): for i, row in data.iterrows(): current_price = row['bid'] if direction == "BUY" else row['ask'] if direction == "BUY": if current_price >= entry_price + take_profit: return {'profit': take_profit * 800, 'duration': i} elif current_price <= entry_price - stop_loss: return {'profit': -stop_loss * 400, 'duration': i} else: # SELL if current_price <= entry_price - take_profit: return {'profit': take_profit * 800, 'duration': i} elif current_price >= entry_price + stop_loss: return {'profit': -stop_loss * 400, 'duration': i} # If the loop completes without hitting TP or SL, close at the last price last_price = data['bid'].iloc[-1] if direction == "BUY" else data['ask'].iloc[-1] profit = (last_price - entry_price) * 100000 if direction == "BUY" else (entry_price - last_price) * 100000 return {'profit': profit, 'duration': len(data)} def backtest_arbitrage_system(historical_data, start_date, end_date): equity_curve = [10000] # Starting with $10,000 trades = [] dates = pd.date_range(start=start_date, end=end_date, freq='D') for current_date in dates: print(f"Backtesting for date: {current_date.date()}") # Get data for the current day data = {symbol: df[df.index.date == current_date.date()] for symbol, df in historical_data.items()} # Skip if no data for the current day if all(df.empty for df in data.values()): continue synthetic_prices = calculate_synthetic_prices(data) arbitrage_opportunities = analyze_arbitrage(data, synthetic_prices) # Simulate trades based on arbitrage opportunities for symbol in arbitrage_opportunities.columns: if arbitrage_opportunities[symbol].any(): direction = "BUY" if arbitrage_opportunities[symbol].iloc[0] else "SELL" base_symbol = symbol.split('_')[0] if base_symbol in data and not data[base_symbol].empty: price = data[base_symbol]['bid'].iloc[-1] if direction == "BUY" else data[base_symbol]['ask'].iloc[-1] take_profit = 800 * 0.00001 # Convert to price stop_loss = 400 * 0.00001 # Convert to price # Simulate trade trade_result = simulate_trade(data[base_symbol], direction, price, take_profit, stop_loss) trades.append(trade_result) # Update equity curve equity_curve.append(equity_curve[-1] + trade_result['profit']) return equity_curve, trades def main(): start_date = datetime(2024, 1, 1, tzinfo=pytz.UTC) end_date = datetime(2024, 8, 31, tzinfo=pytz.UTC) # Бэктест за январь 2023 print("Fetching historical data...") historical_data = get_historical_data(start_date, end_date, terminal_path) if historical_data is None: print("Failed to fetch historical data. Exiting.") return print("Starting backtest...") equity_curve, trades = backtest_arbitrage_system(historical_data, start_date, end_date) total_profit = sum(trade['profit'] for trade in trades) win_rate = sum(1 for trade in trades if trade['profit'] > 0) / len(trades) if trades else 0 print(f"Backtest completed. Results:") print(f"Total Profit: ${total_profit:.2f}") print(f"Win Rate: {win_rate:.2%}") print(f"Final Equity: ${equity_curve[-1]:.2f}") # Plot equity curve plt.figure(figsize=(15, 10)) plt.plot(equity_curve) plt.title('Equity Curve: Backtest Results') plt.xlabel('Trade Number') plt.ylabel('Account Balance ($)') plt.savefig('equity_curve.png') plt.close() print("Equity curve saved as 'equity_curve.png'.") if __name__ == "__main__": main()