Python-algo-learning/import MetaTrader5 as mt5.py at main · Cptnstbn/Python-algo-learning · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
import MetaTrader5 as mt5
import pandas as pd
import numpy as np
from datetime import datetime
from ta.volume import MFIIndicator
from sklearn.cluster import KMeans
from scipy.optimize import minimize
import matplotlib.pyplot as plt

# Initialize MT5 connection
mt5.initialize()

# Define the symbol and timeframe
symbol = "XAUUSD"
timeframe = mt5.TIMEFRAME_M15
start = datetime(2023, 1, 1)
end = datetime(2024, 1, 1)

# Fetch historical data
rates = mt5.copy_rates_range(symbol, timeframe, start, end)
data = pd.DataFrame(rates)
data['time'] = pd.to_datetime(data['time'], unit='s')

mt5.shutdown()

# Alligator Indicator
def alligator(data, jaw_length=13, teeth_length=8, lips_length=5, jaw_offset=8, teeth_offset=5, lips_offset=3):
    data['jaw'] = data['close'].rolling(window=jaw_length).mean().shift(jaw_offset)
    data['teeth'] = data['close'].rolling(window=teeth_length).mean().shift(teeth_offset)
    data['lips'] = data['close'].rolling(window=lips_length).mean().shift(lips_offset)
    return data

# Stochastic Oscillator
def stochastic(data, k_period=14, d_period=3):
    data['low_k'] = data['low'].rolling(window=k_period).min()
    data['high_k'] = data['high'].rolling(window=k_period).max()
    data['%K'] = 100 * (data['close'] - data['low_k']) / (data['high_k'] - data['low_k'])
    data['%D'] = data['%K'].rolling(window=d_period).mean()
    return data

# ATR Indicator
def atr(data, atr_period=14):
    data['hl'] = data['high'] - data['low']
    data['hc'] = abs(data['high'] - data['close'].shift(1))
    data['lc'] = abs(data['low'] - data['close'].shift(1))
    data['tr'] = data[['hl', 'hc', 'lc']].max(axis=1)
    data['atr'] = data['tr'].rolling(window=atr_period).mean()
    return data

# Fractals
def fractals(data, n=2):
    data['fractal_up'] = data['high'][(data['high'] > data['high'].shift(1)) & (data['high'] > data['high'].shift(-1))]
    data['fractal_down'] = data['low'][(data['low'] < data['low'].shift(1)) & (data['low'] < data['low'].shift(-1))]
    return data

# Machine Learning Enhanced MFI
def ml_mfi(data, mfi_length=14, train_length=300, iterations=5):
    mfi = MFIIndicator(high=data['high'], low=data['low'], close=data['close'], volume=data['tick_volume'], window=mfi_length).money_flow_index()
    data['mfi'] = mfi

    def kmeans_adjust(data, iterations, train_length):
        kmeans = KMeans(n_clusters=3)
        for i in range(iterations):
            sample = data['mfi'].iloc[-train_length:]
            kmeans.fit(sample.values.reshape(-1, 1))
            centers = sorted(kmeans.cluster_centers_.flatten())
            data['overbought'], data['neutral'], data['oversold'] = centers[2], centers[1], centers[0]
            data['mfi'] = data['mfi'].apply(lambda x: (x - centers[0]) / (centers[2] - centers[0]) * 100)

        return data

    data = kmeans_adjust(data, iterations, train_length)
    return data

# Apply Indicators and Strategy Logic
def apply_strategy(data, params):
    jaw_length, teeth_length, lips_length, jaw_offset, teeth_offset, lips_offset, k_period, d_period, atr_period, atr_multiplier = params

    # Apply indicators
    data = alligator(data, jaw_length=int(jaw_length), teeth_length=int(teeth_length), lips_length=int(lips_length), jaw_offset=int(jaw_offset), teeth_offset=int(teeth_offset), lips_offset=int(lips_offset))
    data = stochastic(data, k_period=int(k_period), d_period=int(d_period))
    data = atr(data, atr_period=int(atr_period))
    data = fractals(data)
    data = ml_mfi(data)

    # Define strategy conditions
    data['buy_signal'] = (data['close'] > data['lips']) & (data['mfi'] < 20) & (data['%K'] > data['%D'])
    data['sell_signal'] = (data['close'] < data['lips']) & (data['mfi'] > 80) & (data['%K'] < data['%D'])

    # ATR trailing stop loss
    data['trail_stop'] = data['atr'] * atr_multiplier

    return data

# Backtesting
def backtest(data, initial_balance=10000):
    balance = initial_balance
    position = 0
    entry_price = 0

    for i in range(len(data)):
        if data['buy_signal'].iloc[i] and position == 0:
            position = 1
            entry_price = data['close'].iloc[i]
            stop_loss = entry_price - data['trail_stop'].iloc[i]

        elif data['sell_signal'].iloc[i] and position == 1:
            balance += (data['close'].iloc[i] - entry_price)
            position = 0

        elif position == 1 and data['close'].iloc[i] < stop_loss:
            balance += (data['close'].iloc[i] - entry_price)
            position = 0

    return balance

# Define the objective function for optimization
def objective_function(params):
    data_copy = data.copy()
    data_copy = apply_strategy(data_copy, params)
    final_balance = backtest(data_copy)
    return -final_balance  # We minimize the negative final balance to maximize it

# Initial parameter guesses
initial_params = [13, 8, 5, 8, 5, 3, 14, 3, 14, 1.75]

# Bounds for each parameter (jaw_length, teeth_length, lips_length, jaw_offset, teeth_offset, lips_offset, k_period, d_period, atr_period, atr_multiplier)
bounds = [(5, 21), (5, 21), (3, 10), (3, 10), (3, 10), (1, 5), (5, 21), (1, 5), (5, 21), (1.0, 3.0)]

# Optimize the parameters
result = minimize(objective_function, initial_params, bounds=bounds, method='L-BFGS-B')
optimized_params = result.x

print(f"Optimized Parameters: {optimized_params}")