gavindiaz/Neural-Network-MT5-Trading-Bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Made improvements to overall functionality

Browse Source
This commit is contained in:
Mike
2023-12-01 15:49:45 +02:00
parent f53fcb7dc3
commit 9f3d469f14
7 changed files with 300 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+55 -22
View File
@@ -7,28 +7,32 @@ Author: Mike Kiwalabye
"""
import time
from flask import Flask, render_template, request, redirect, globals
from flask import Flask, render_template, request, redirect, json, Response
from src.connectors import mt5_connector
from src.models import neural_network_model
from src.strategies.trading_strategy import get_historical_data, calculate_indicators_and_detect_patterns, generate_trade_signals, execute_trade
from src.utils.visualization import plot_trade_signals
import threading
import pandas as pd
app = Flask(__name__)
# Define input shape for the neural network
input_shape = (10,) # Adjust the input shape based on your features and data
input_shape = (11,) # Adjust the input shape based on your features and data
# Create the neural network model
neural_network_model = neural_network_model.create_neural_network_model(input_shape)
# Global state to track whether MT5 is initialized
globals.mt5_initialized = False
mt5_initialized = False
latest_trade_signals = []
# Web Interface Routes
@app.route('/')
def index():
"""Render the main page with login form."""
"""Render the main page with the login form."""
return render_template('index.html')
@app.route('/login', methods=['POST'])
@@ -41,6 +45,7 @@ def login():
Returns:
- str: HTML response.
"""
global mt5_initialized
if request.method == 'POST':
credentials = {
'username': request.form['username'],
@@ -50,7 +55,7 @@ def login():
}
if mt5_connector.connect_to_mt5(credentials):
# Set MT5 initialization state to True
globals.mt5_initialized = True
mt5_initialized = True
# Redirect to the main dashboard or another page
return redirect('/dashboard')
else:
@@ -73,45 +78,68 @@ def dashboard():
def start_ml_bot():
"""
Handle the request to start the ML bot.
"""
# Start the ML bot
run_trading_bot_web_interface()
threading.Thread(target=run_trading_bot_web_interface).start()
# Return an empty response
return Response(status=200)
# Main Trading Bot Logic
@app.route("/stop_ml_bot", methods=['GET'])
def stop_ml_bot():
mt5_connector.stop_mt5_ml_bot()
redirect('/dashboard')
def map_signal_priority(signal_priority):
# Define a mapping for string values to integers
signal_mapping = {
'Both': 1,
'Pattern': 2,
'RSI': 3
# Add more mappings as needed
}
# Use the mapping, default to 0 if not found
return signal_mapping.get(signal_priority, 0)
# Main Trading Bot Logic
def run_trading_bot_web_interface():
"""
Run the trading bot using MetaTrader 5 credentials from the web interface.
"""
global latest_trade_signals
historical_data_df = pd.DataFrame()
while True:
try:
symbol = 'EURUSD'
lot_size = 0.01
stop_loss = 100
take_profit = 150
take_profit = 200
# Get the latest historical data
latest_data = get_historical_data(symbol).iloc[-1:]
historical_data_df = get_historical_data(symbol, historical_data_df)
# Calculate indicators and detect patterns for the latest data
df = calculate_indicators_and_detect_patterns(latest_data)
df = calculate_indicators_and_detect_patterns(historical_data_df)
# Generate trade signals for the latest data
df = generate_trade_signals(df)
print(df)
# Execute trades
for i in range(len(latest_data)):
signal = df['signal'].iloc[i]
if signal != 'None':
print(df['signal'].array)
execute_trade(signal, df, symbol, lot_size, stop_loss, take_profit)
df.to_csv('your_file.csv', sep='\t', index=False)
# Visualize data
# plot_trade_signals(df)
# Inside the run_trading_bot_web_interface function
latest_trade_signals = df.replace({pd.NA: 'null'}).to_json(orient='records')
# Execute trades
for i in range(len(df)):
signal_priority = df['signal'].iloc[i] # Replace with your actual value
mapped_priority = map_signal_priority(signal_priority)
if mapped_priority != 0:
execute_trade(mapped_priority, df, symbol, lot_size, stop_loss, take_profit)
except Exception as e:
print(f"Error running trading bot: {str(e)}")
@@ -119,6 +147,11 @@ def run_trading_bot_web_interface():
# Wait for the next iteration
time.sleep(60) # Adjust the time interval as needed
@app.route('/get_latest_trade_signals', methods=['GET'])
def get_latest_trade_signals():
global latest_trade_signals
return json.dumps(latest_trade_signals)
# Start the Flask app
if __name__ == '__main__':
app.run(debug=True)
src/connectors/mt5_connector.py
+6 -1
View File
@@ -70,4 +70,9 @@ def get_account_info():
# Fetch account information
account_info = mt5.account_info()
return account_info
return account_info
def stop_mt5_ml_bot():
mt5.shutdown()
return "Disconnected from MetaTrader 5"
src/models/neural_network_model.py
+1 -1
View File
@@ -96,7 +96,7 @@ def update_neural_network_model(trade_outcome: dict, dataset_path: str) -> None:
model = load_model('model_weights.h5')
except (OSError, ValueError):
# If loading fails, create a new model
input_shape = (4,) # Replace with the actual input shape
input_shape = (5,) # Replace with the actual input shape
model = create_neural_network_model(input_shape)
compile_neural_network_model(model, learning_rate=0.001)
src/strategies/trading_strategy.py
+113 -52
View File
@@ -12,31 +12,69 @@ import MetaTrader5 as mt5
import pandas as pd
import numpy as np
from sklearn.preprocessing import MinMaxScaler
import talib
from talib import abstract
from src.models import neural_network_model
def get_historical_data(symbol: str) -> pd.DataFrame:
def get_historical_data(symbol: str, existing_data: pd.DataFrame = None) -> pd.DataFrame:
"""
Retrieve historical data for a given symbol and timeframe from MetaTrader 5.
Parameters:
- symbol (str): The financial instrument symbol (e.g., 'EURUSD').
- existing_data (pd.DataFrame): Existing historical data DataFrame.
Returns:
- pd.DataFrame: DataFrame containing historical data with columns: ['time', 'open', 'high', 'low', 'close', 'tick_volume', 'spread', 'real_volume'].
"""
# Retrieve historical data
rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_M1, 0, 1000)
print(len(existing_data))
if len(existing_data) == 0:
# If no existing data, fetch the last 2500 bars
rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_M1, 0, 2500)
df = pd.DataFrame(rates)
else:
# If existing data is provided, fetch only the latest bar
rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_M1, 0, 1)
new_data = pd.DataFrame(rates)
# Concatenate the new data to the existing data
df = pd.concat([existing_data, new_data])
# Convert data to DataFrame
df = pd.DataFrame(rates)
# Convert the 'time' column to datetime
df['time'] = pd.to_datetime(df['time'], unit='s')
# Set the 'time' column as the index
df.set_index('time', inplace=True)
return df
def calculate_patterns(df: pd.DataFrame) -> pd.DataFrame:
"""
Calculate common trade patterns such as double tops & bottoms, pennants, wedges, and bull and bear flags.
Parameters:
- df (pd.DataFrame): The DataFrame containing price and indicator information.
Returns:
- pd.DataFrame: The DataFrame with added columns for detected patterns.
"""
# Detect Double Tops & Bottoms
df['double_top'] = np.where((df['high'].shift(1) > df['high']) & (df['high'].shift(1) > df['high'].shift(2)), 'Double Top', 'None')
df['double_bottom'] = np.where((df['low'].shift(1) < df['low']) & (df['low'].shift(1) < df['low'].shift(2)), 'Double Bottom', 'None')
# Detect Bull and Bear Flags
df['bull_flag'] = np.where((df['close'] > abstract.BBANDS(df['close'], timeperiod=5, nbdevup=2.0, nbdevdn=2.0)[0]) & (df['close'].shift(1) < abstract.BBANDS(df['close'].shift(1), timeperiod=5, nbdevup=2.0, nbdevdn=2.0)[0]), 'Bull Flag', 'None')
df['bear_flag'] = np.where((df['close'] < abstract.BBANDS(df['close'], timeperiod=5, nbdevup=2.0, nbdevdn=2.0)[2]) & (df['close'].shift(1) > abstract.BBANDS(df['close'].shift(1), timeperiod=5, nbdevup=2.0, nbdevdn=2.0)[2]), 'Bear Flag', 'None')
# Assign patterns based on conditions
df['pattern'] = 'None'
conditions = [
(df['double_top'] != 'None'),
(df['double_bottom'] != 'None'),
(df['bull_flag'] != 'None'),
(df['bear_flag'] != 'None')
]
choices = ['Double Top', 'Double Bottom', 'Bull Flag', 'Bear Flag']
df['pattern'] = np.select(conditions, choices, default='None')
return df
def calculate_indicators_and_detect_patterns(df: pd.DataFrame) -> pd.DataFrame:
@@ -53,17 +91,25 @@ def calculate_indicators_and_detect_patterns(df: pd.DataFrame) -> pd.DataFrame:
# Add your indicator calculation and pattern detection logic here
# Example: Calculate RSI
df['rsi'] = talib.RSI(df['close'], timeperiod=14)
df['rsi'] = abstract.RSI(df['close'], timeperiod=14)
# print(df['close'].values)
# Example: Detect RSI divergence
df['rsi_divergence'] = (df['rsi'] > 70) & (df['close'] < df['close'].shift())
# Example: Detect MACD divergence
# Example: Detect TREND signal
df['trend_signal'] = 'None'
df['short_ma'] = df['close'].rolling(window=50).mean()
df['long_ma'] = df['close'].rolling(window=200).mean()
df.loc[df['short_ma'] > df['long_ma'], 'trend_signal'] = 'Uptrend'
df.loc[df['short_ma'] < df['long_ma'], 'trend_signal'] = 'Downtrend'
# Add your MACD divergence detection logic here
# Example: Detect patterns
df['pattern'] = 'None'
# Add your pattern detection logic here
df = calculate_patterns(df)
return df
@@ -89,13 +135,22 @@ def generate_trade_signals(df: pd.DataFrame) -> pd.DataFrame:
# Placeholder for 'resistance' calculation - replace this with your actual logic
(df['close'] > df['close'].rolling(window=10).max()),
(df['close'] < df['close'].rolling(window=10).min()),
# Placeholder for 'trend_200' calculation - replace this with your actual logic
(df['close'] > df['close'].rolling(window=200).mean()),
(df['close'] < df['close'].rolling(window=200).mean())
# Use the calculated 'trend_signal' column for trend condition
(df['trend_signal'] == 'Uptrend'),
(df['trend_signal'] == 'Downtrend'),
# Additional condition to check if the pattern is valid
(df['pattern'] != 'None'),
]
choices = ['Divergence', 'Resistance', 'Support', 'Uptrend', 'Downtrend']
df['support_resistance_signal'] = np.select(conditions, choices, default='None')
choices = ['Divergence', 'Resistance', 'Support', 'Uptrend', 'Downtrend', 'Pattern']
# Ensure that the lengths of conditions and choices are the same
if len(conditions) == len(choices):
df['support_resistance_signal'] = np.select(conditions, choices, default='None')
else:
# Handle the case where lengths do not match (print an error message for debugging)
print("Error: Lengths of conditions and choices do not match.")
df['support_resistance_signal'] = 'None'
# Iterate over the data points
for i in range(1, len(df)):
@@ -124,17 +179,15 @@ def execute_trade(signal_priority, df, symbol, lot_size, stop_loss, take_profit)
- stop_loss (float): The stop-loss level.
- take_profit (float): The take-profit level.
"""
# Initialize outcome and request
outcome = None
request = {}
for index, row in df.iterrows():
# Calculate risk and position size based on lot size, stop loss, and take profit
risk_multiplier = 1.2 if 'RSI' in df['strongest_divergence_signal'].iloc[-1] else 1.5
risk = lot_size * stop_loss * risk_multiplier
position_size = risk / (take_profit - stop_loss)
try:
if signal_priority == 3:
# Additional conditions for Buy trade
if (
(signal_priority == 3 and row['rsi_divergence'] and row['rsi_value'] < 30 and row['trend_signal'] == 'Downtrend') or
(signal_priority == 2 and row['pattern'] == 'Double Bottom' and row['trend_signal'] == 'Downtrend') or
(signal_priority == 1 and 40 <= row['rsi_value'] <= 60 and row['pattern'] == 'Bull Flag' and row['trend_signal'] == 'Uptrend') or
(signal_priority == 0 and row['rsi_value'] < 30 and row['rsi_divergence'] and row['pattern'] == 'Bull')
):
# Place a buy trade
request = {
'action': mt5.TRADE_ACTION_DEAL,
@@ -148,9 +201,14 @@ def execute_trade(signal_priority, df, symbol, lot_size, stop_loss, take_profit)
'magic': 123456,
'comment': "Buy trade",
'type_time': mt5.ORDER_TIME_GTC,
'type_filling': mt5.ORDER_FILLING_RETURN,
'type_filling': mt5.ORDER_FILLING_IOC,
}
elif signal_priority == 2:
elif (
(signal_priority == 3 and row['rsi_divergence'] and row['rsi_value'] > 70 and row['trend_signal'] == 'Uptrend') or
(signal_priority == 2 and row['pattern'] == 'Double Top' and row['trend_signal'] == 'Uptrend') or
(signal_priority == 1 and 40 <= row['rsi_value'] <= 60 and row['pattern'] == 'Bear Flag' and row['trend_signal'] == 'Downtrend') or
(signal_priority == 0 and row['rsi_value'] > 70 and row['rsi_divergence'] and row['pattern'] == 'Bear')
):
# Place a sell trade
request = {
'action': mt5.TRADE_ACTION_DEAL,
@@ -164,30 +222,33 @@ def execute_trade(signal_priority, df, symbol, lot_size, stop_loss, take_profit)
'magic': 123456,
'comment': "Sell trade",
'type_time': mt5.ORDER_TIME_GTC,
'type_filling': mt5.ORDER_FILLING_RETURN,
'type_filling': mt5.ORDER_FILLING_IOC,
}
result = mt5.order_send(request)
outcome = 'Win' if result.retcode == mt5.TRADE_RETCODE_DONE else 'Loss'
try:
if signal_priority != 0:
result = mt5.order_send(request)
print(result)
outcome = 'Win' if result.retcode == mt5.TRADE_RETCODE_DONE else 'Loss'
# Example trade outcome information
trade_outcome = {
'pattern': df['pattern'].iloc[-1],
'divergence_strength': df['strongest_divergence_signal'].iloc[-1],
'time': df.index[-1],
'trend_direction': df['trend_signal'].iloc[-1],
'indicator_used': df['strongest_divergence_signal'].iloc[-1],
'outcome': outcome
}
# Example trade outcome information
trade_outcome = {
'pattern': row['pattern'],
'divergence_strength': row['strongest_divergence_signal'],
'time': pd.Timestamp.now(),
'trend_direction': row['trend_signal'],
'indicator_used': row['strongest_divergence_signal'],
'outcome': outcome
}
# Update TensorFlow neural network model with trade outcome
neural_network_model.update_neural_network_model(trade_outcome, 'tradedata.csv')
# Update TensorFlow neural network model with trade outcome
neural_network_model.update_neural_network_model(trade_outcome, 'tradedata.csv')
# Example print statements for debugging
print(
f"Executed trade with signal priority: {signal_priority}, position size: {lot_size}")
print(f"Trade outcome: {trade_outcome}")
# Example print statements for debugging
print(
f"Executed trade with signal priority: {signal_priority}, position size: {position_size}")
print(f"Trade outcome: {trade_outcome}")
# Additional logic for trade management, monitoring, etc.
except Exception as e:
print(f"Error executing trade: {str(e)}")
# Additional logic for trade management, monitoring, etc.
except Exception as e:
print(f"Error executing trade: {str(e)}")
src/utils/visualization.py
+36 -24
View File
@@ -21,17 +21,24 @@ def plot_price_data(df: pd.DataFrame, title: str = 'Price Chart') -> None:
Returns:
- None
"""
plt.figure(figsize=(10, 6))
plt.plot(df.index, df['close'], label='Close')
# Add visualizations for other indicators, levels, and patterns
# (Add more visualizations as needed)
try:
plt.figure(figsize=(10, 6))
plt.plot(df.index, df['close'], label='Close')
# Add visualizations for other indicators, levels, and patterns
# (Add more visualizations as needed)
plt.title(title)
plt.xlabel('Time')
plt.ylabel('Price')
plt.legend()
plt.show()
plt.title(title)
plt.xlabel('Time')
plt.ylabel('Price')
plt.legend()
# Use plt.show(block=True) to make the plot blocking
plt.show(block=True)
except Exception as e:
print(f"Error plotting trade signals: {str(e)}")
# ...
def plot_trade_signals(df: pd.DataFrame, title: str = 'Trade Signals') -> None:
"""
@@ -44,18 +51,23 @@ def plot_trade_signals(df: pd.DataFrame, title: str = 'Trade Signals') -> None:
Returns:
- None
"""
plt.figure(figsize=(10, 6))
plt.plot(df.index, df['close'], label='Close')
# Plot trade signals
buy_signals = df[df['signal'] == 'Buy']
sell_signals = df[df['signal'] == 'Sell']
try:
plt.figure(figsize=(10, 6))
plt.plot(df.index, df['close'], label='Close')
# Plot trade signals
buy_signals = df[df['signal'] == 'Buy']
sell_signals = df[df['signal'] == 'Sell']
plt.scatter(buy_signals.index, buy_signals['close'], color='green', marker='^', label='Buy Signal')
plt.scatter(sell_signals.index, sell_signals['close'], color='red', marker='v', label='Sell Signal')
plt.title(title)
plt.xlabel('Time')
plt.ylabel('Price')
plt.legend()
plt.show()
plt.scatter(buy_signals.index, buy_signals['close'], color='green', marker='^', label='Buy Signal')
plt.scatter(sell_signals.index, sell_signals['close'], color='red', marker='v', label='Sell Signal')
plt.title(title)
plt.xlabel('Time')
plt.ylabel('Price')
plt.legend()
# Use plt.show(block=True) to make the plot blocking
plt.show(block=True)
except Exception as e:
print(f"Error plotting trade signals: {str(e)}")
templates/dashboard.html
+89 -5
View File
@@ -6,6 +6,9 @@
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Trading Dashboard</title>
<!-- Include Chart.js from a CDN -->
<script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
</head>
<body>
<h1>Trading Dashboard</h1>
@@ -15,12 +18,93 @@
<p><strong>Account Balance:</strong> {{ account_balance }} <small>{{ account_currency }}</small></p>
</div>
<form action="/logout" method="post">
<button type="submit">Logout</button>
<!-- Add a canvas element for the chart -->
<canvas id="tradeChart" width="800" height="400"></canvas>
<form action="/stop_ml_bot" method="get">
<button type="submit">Stop ML Bot</button>
</form>
<form action="/start_ml_bot" method="post">
<button type="submit">Start ML Bot</button>
</form>
<!-- Use a button without a form to start the ML Bot -->
<button onclick="startBot()">Start ML Bot</button>
<script>
// Function to update the chart with new trade signals
function updateChart(tradeSignals) {
// Parse the JSON-formatted string to an object
const parsedTradeSignals = JSON.parse(tradeSignals);
// Extract relevant data for the chart (modify as needed)
const timestamps = parsedTradeSignals.map(signal => signal.time);
const prices = parsedTradeSignals.map(signal => signal.close);
// Get the canvas element
const ctx = document.getElementById('tradeChart');
// Destroy existing chart if it exists
if (ctx.chart) {
ctx.chart.destroy();
}
// Initialize the chart
const myChart = new Chart(ctx, {
type: 'line',
data: {
labels: timestamps,
datasets: [{
label: 'Close Price',
data: prices,
borderColor: 'rgba(75, 192, 192, 1)',
borderWidth: 1,
fill: false
}]
},
options: {
scales: {
x: {
type: 'time',
time: {
unit: 'minute' // Adjust as needed
}
},
y: {
beginAtZero: false
}
}
}
});
}
// Function to periodically update the chart
function periodicallyUpdateChart() {
// Fetch the latest trade signals from the server
fetch('/get_latest_trade_signals')
.then(response => response.json())
.then(tradeSignals => {
// Update the chart with the new trade signals
updateChart(tradeSignals);
// Schedule the next update
setTimeout(periodicallyUpdateChart, 5000); // Update every 5 seconds
})
.catch(error => {
console.error('Error fetching trade signals:', error);
// Retry the update after an interval
setTimeout(periodicallyUpdateChart, 5000); // Retry after 5 seconds
});
}
// Start the initial chart update
periodicallyUpdateChart();
function startBot() {
// Send an asynchronous request to start the bot
fetch('/start_ml_bot', { method: 'POST' });
// Optionally, you can add logic here to update the UI or provide feedback to the user
console.log('Bot started!');
}
</script>
</body>
</html>
your_file.csv
View File