199 lines
8.3 KiB
Python
199 lines
8.3 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
"""Trading Bot Simulation
|
||
|
|
|
||
|
|
Automatically generated by Colab.
|
||
|
|
|
||
|
|
Original file is located at
|
||
|
|
https://colab.research.google.com/drive/1xQA5F3HvFfWWO3Xe2PKXEM68FFx_s-rc
|
||
|
|
"""
|
||
|
|
|
||
|
|
import numpy as np
|
||
|
|
import time
|
||
|
|
# NOTE: In a real environment, you must install the MetaTrader5 package:
|
||
|
|
# pip install MetaTrader5
|
||
|
|
# import MetaTrader5 as mt5
|
||
|
|
|
||
|
|
# --- MOCK MT5 LIBRARY & CONFIGURATION ---
|
||
|
|
# The functions below are STUBS (placeholders) that mimic the real MT5 library behavior.
|
||
|
|
# Replace the comments with your actual MT5 connection logic and credentials.
|
||
|
|
|
||
|
|
# Strategy parameters
|
||
|
|
SYMBOL = "EURUSD"
|
||
|
|
TIMEFRAME = "M15" # E.g., 15-minute bars
|
||
|
|
SHORT_WINDOW = 5
|
||
|
|
LONG_WINDOW = 15
|
||
|
|
VOLUME = 0.1 # Trading volume (lot size)
|
||
|
|
|
||
|
|
def mt5_initialize_connection(login, password, server):
|
||
|
|
"""Initializes connection to MT5 terminal."""
|
||
|
|
print("Connecting to MT5 terminal...")
|
||
|
|
# if not mt5.initialize(login=login, password=password, server=server):
|
||
|
|
# print(f"Failed to initialize MT5: {mt5.last_error()}")
|
||
|
|
# return False
|
||
|
|
# print("MT5 connection successful.")
|
||
|
|
return True
|
||
|
|
|
||
|
|
def mt5_fetch_data(symbol, period_size, count):
|
||
|
|
"""Fetches the last 'count' bars of data from MT5."""
|
||
|
|
# This mock data simulates fetching the last 30 closing prices
|
||
|
|
print(f"Fetching {count} historical bars for {symbol}...")
|
||
|
|
|
||
|
|
# In a real bot, you'd use:
|
||
|
|
# rates = mt5.copy_rates_from_pos(symbol, period_size, 0, count)
|
||
|
|
# df = pd.DataFrame(rates)
|
||
|
|
# df['time'] = pd.to_datetime(df['time'], unit='s')
|
||
|
|
# return df['close'].to_numpy() # We only need the closing prices for SMA
|
||
|
|
|
||
|
|
# Using the simulation data for demonstration:
|
||
|
|
PRICE_DATA = [
|
||
|
|
100.00, 101.50, 102.10, 103.00, 105.50, 104.90, 106.30, 107.80, 108.20, 107.50,
|
||
|
|
110.00, 112.40, 115.80, 114.30, 113.10, 116.70, 118.90, 120.00, 122.50, 121.10,
|
||
|
|
125.00, 126.90, 124.70, 128.00, 130.40, 127.80, 126.10, 125.50, 129.00, 132.00
|
||
|
|
]
|
||
|
|
return np.array(PRICE_DATA)
|
||
|
|
|
||
|
|
def mt5_send_order(symbol, direction, volume, price_limit=None):
|
||
|
|
"""Mocks sending a market or limit order to MT5."""
|
||
|
|
order_type = "BUY" if direction == 1 else "SELL"
|
||
|
|
# This is where your actual order execution code goes using mt5.order_send()
|
||
|
|
|
||
|
|
# In a real bot, you'd calculate price, deviation, and create a request dictionary:
|
||
|
|
# request = {
|
||
|
|
# "action": mt5.TRADE_ACTION_DEAL,
|
||
|
|
# "symbol": symbol,
|
||
|
|
# "volume": volume,
|
||
|
|
# "type": mt5.ORDER_TYPE_BUY if direction == 1 else mt5.ORDER_TYPE_SELL,
|
||
|
|
# "price": mt5.symbol_info_tick(symbol).ask if direction == 1 else mt5.symbol_info_tick(symbol).bid,
|
||
|
|
# "deviation": 20, # Max price slippage
|
||
|
|
# "magic": 202409,
|
||
|
|
# "comment": "SMA Crossover Bot",
|
||
|
|
# "type_time": mt5.ORDER_TIME_GTC,
|
||
|
|
# "type_filling": mt5.ORDER_FILLING_FOC,
|
||
|
|
# }
|
||
|
|
# result = mt5.order_send(request)
|
||
|
|
|
||
|
|
print(f"\n[ORDER SENT] --- {order_type} {volume} lots of {symbol}")
|
||
|
|
# print(f"MT5 API Response: {result}")
|
||
|
|
|
||
|
|
return True
|
||
|
|
|
||
|
|
# --- CORE LOGIC FUNCTIONS (RETAINED FOR TECHNICAL ANALYSIS) ---
|
||
|
|
|
||
|
|
def calculate_sma(data, window):
|
||
|
|
"""Calculates the Simple Moving Average (SMA) for a given window."""
|
||
|
|
weights = np.repeat(1.0, window) / window
|
||
|
|
smas = np.convolve(data, weights, 'valid')
|
||
|
|
padding = np.full(window - 1, np.nan)
|
||
|
|
return np.concatenate((padding, smas))
|
||
|
|
|
||
|
|
def generate_signals(prices, short_window, long_window):
|
||
|
|
"""Generates buy (1) and sell (-1) signals based on SMA Crossover."""
|
||
|
|
# We only need the last two values of the SMAs to detect a recent crossover
|
||
|
|
prices_array = np.array(prices)
|
||
|
|
short_sma = calculate_sma(prices_array, short_window)
|
||
|
|
long_sma = calculate_sma(prices_array, long_window)
|
||
|
|
|
||
|
|
# Check for non-NaN values
|
||
|
|
if np.isnan(short_sma[-1]) or np.isnan(long_sma[-1]) or \
|
||
|
|
np.isnan(short_sma[-2]) or np.isnan(long_sma[-2]):
|
||
|
|
return 0, short_sma, long_sma # No signal if data is insufficient
|
||
|
|
|
||
|
|
# Bullish Crossover (BUY signal)
|
||
|
|
if short_sma[-1] > long_sma[-1] and short_sma[-2] <= long_sma[-2]:
|
||
|
|
return 1, short_sma, long_sma
|
||
|
|
|
||
|
|
# Bearish Crossover (SELL signal)
|
||
|
|
elif short_sma[-1] < long_sma[-1] and short_sma[-2] >= long_sma[-2]:
|
||
|
|
return -1, short_sma, long_sma
|
||
|
|
|
||
|
|
# No Crossover (HOLD signal)
|
||
|
|
return 0, short_sma, long_sma
|
||
|
|
|
||
|
|
# --- LIVE TRADING LOOP ---
|
||
|
|
|
||
|
|
def live_trading_main(symbol, timeframe, short_window, long_window, volume, interval_seconds=60):
|
||
|
|
"""
|
||
|
|
Main loop for live trading on MT5.
|
||
|
|
NOTE: Replace the mock connection details with your real MT5 credentials.
|
||
|
|
"""
|
||
|
|
# Replace with your actual credentials
|
||
|
|
LOGIN = 12345678
|
||
|
|
PASSWORD = "your_password"
|
||
|
|
SERVER = "your_server_name"
|
||
|
|
|
||
|
|
if not mt5_initialize_connection(LOGIN, PASSWORD, SERVER):
|
||
|
|
return
|
||
|
|
|
||
|
|
# The loop runs forever, constantly checking for new data and signals
|
||
|
|
print(f"\nStarting live trading loop for {symbol}. Checking every {interval_seconds} seconds...")
|
||
|
|
while True:
|
||
|
|
try:
|
||
|
|
# 1. Fetch data required for the longest moving average (LONG_WINDOW)
|
||
|
|
# We need at least LONG_WINDOW bars + 1 previous bar to detect a cross
|
||
|
|
DATA_POINTS_TO_FETCH = long_window + 1
|
||
|
|
|
||
|
|
# Fetch real data (mocked here)
|
||
|
|
close_prices = mt5_fetch_data(symbol, timeframe, DATA_POINTS_TO_FETCH)
|
||
|
|
|
||
|
|
# 2. Generate signal based on the latest data
|
||
|
|
signal, short_sma, long_sma = generate_signals(
|
||
|
|
close_prices, short_window, long_window
|
||
|
|
)
|
||
|
|
|
||
|
|
current_price = close_prices[-1]
|
||
|
|
latest_short_ma = short_sma[-1] if not np.isnan(short_sma[-1]) else 'N/A'
|
||
|
|
latest_long_ma = long_sma[-1] if not np.isnan(long_sma[-1]) else 'N/A'
|
||
|
|
|
||
|
|
print(f"\n[{time.strftime('%H:%M:%S')}] Price: ${current_price:.4f} | SMA({short_window}): {latest_short_ma} | SMA({long_window}): {latest_long_ma} | Signal: {signal}")
|
||
|
|
|
||
|
|
# 3. Execute Trade based on signal
|
||
|
|
if signal == 1:
|
||
|
|
# Check for existing open positions before sending BUY order (crucial for MT5)
|
||
|
|
# If there are open SELL positions, you'd close them first.
|
||
|
|
print(">>> BULLISH CROSSOVER DETECTED. ATTEMPTING TO BUY.")
|
||
|
|
mt5_send_order(symbol, 1, volume) # 1 for BUY
|
||
|
|
|
||
|
|
elif signal == -1:
|
||
|
|
# Check for existing open positions before sending SELL order
|
||
|
|
# If there are open BUY positions, you'd close them first.
|
||
|
|
print(">>> BEARISH CROSSOVER DETECTED. ATTEMPTING TO SELL.")
|
||
|
|
mt5_send_order(symbol, -1, volume) # -1 for SELL
|
||
|
|
|
||
|
|
else:
|
||
|
|
print("No signal, holding position.")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"An error occurred: {e}")
|
||
|
|
# mt5.shutdown() # Uncomment this in a real app to cleanly close the connection
|
||
|
|
break # Exit loop on critical error
|
||
|
|
|
||
|
|
# Wait for the next check cycle
|
||
|
|
time.sleep(interval_seconds)
|
||
|
|
|
||
|
|
|
||
|
|
# --- 4. EXECUTION ---
|
||
|
|
if __name__ == "__main__":
|
||
|
|
# The backtesting functions are now separated from the live trading execution.
|
||
|
|
# To run the live trading loop, uncomment the line below.
|
||
|
|
# IMPORTANT: You must install the MetaTrader5 library (pip install MetaTrader5)
|
||
|
|
# and run this script from a Python environment that is allowed to connect to
|
||
|
|
# your running MT5 terminal.
|
||
|
|
|
||
|
|
print("--- TRADING BOT DRAFT FOR MT5 INTEGRATION ---")
|
||
|
|
print("This file contains the SMA logic and the structure for MT5 integration.")
|
||
|
|
print("The code below simulates the backtesting process using the mock data.")
|
||
|
|
|
||
|
|
# Simulated Backtesting (using the old function structure)
|
||
|
|
price_history = mt5_fetch_data(SYMBOL, TIMEFRAME, 30)
|
||
|
|
signals, short_sma, long_sma = generate_signals(price_history, SHORT_WINDOW, LONG_WINDOW)
|
||
|
|
|
||
|
|
print("\n--- Backtesting Results (Simulated Data) ---")
|
||
|
|
print("Index | Price | Short MA | Long MA | Signal")
|
||
|
|
print("--------------------------------------------------")
|
||
|
|
for i in range(len(price_history)):
|
||
|
|
short_val = f"{short_sma[i]:.2f}" if not np.isnan(short_sma[i]) else 'N/A'
|
||
|
|
long_val = f"{long_sma[i]:.2f}" if not np.isnan(long_sma[i]) else 'N/A'
|
||
|
|
print(f"{i:5} | {price_history[i]:5.2f} | {short_val:8} | {long_val:7} | {signals[i]:.0f}")
|
||
|
|
|
||
|
|
# To run the live loop, uncomment this line (after installing MT5):
|
||
|
|
# live_trading_main(SYMBOL, TIMEFRAME, SHORT_WINDOW, LONG_WINDOW, VOLUME)
|