Maghdam
417 lines
16 KiB
Python
417 lines
16 KiB
Python
import sys
|
|
import os
|
|
import warnings
|
|
from pathlib import Path
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1) SET PROJECT ROOT AND UPDATE PATH/WORKING DIRECTORY
|
|
# ---------------------------------------------------------------------------
|
|
project_root = Path.cwd().parent.parent # Adjust if your notebook is in notebooks/time_series
|
|
sys.path.append(str(project_root))
|
|
os.chdir(str(project_root))
|
|
warnings.filterwarnings("ignore")
|
|
import warnings
|
|
warnings.filterwarnings("ignore")
|
|
|
|
import MetaTrader5 as mt5
|
|
import pandas as pd
|
|
import numpy as np
|
|
import ta
|
|
from datetime import datetime, timedelta
|
|
from sklearn.model_selection import train_test_split
|
|
from sklearn.preprocessing import StandardScaler
|
|
from sklearn.ensemble import RandomForestRegressor
|
|
from sklearn.feature_selection import SelectFromModel
|
|
import time
|
|
import logging
|
|
import joblib
|
|
|
|
# Setup logging
|
|
logging.basicConfig(
|
|
filename='models/saved_models/trading_app.log',
|
|
level=logging.INFO,
|
|
format='%(asctime)s %(levelname)s:%(message)s',
|
|
datefmt='%Y-%m-%d %H:%M:%S'
|
|
)
|
|
|
|
def log_and_print(message, is_error=False):
|
|
"""
|
|
Logs and prints a message.
|
|
If is_error=True, logs at the ERROR level; otherwise logs at INFO level.
|
|
"""
|
|
if is_error:
|
|
logging.error(message)
|
|
else:
|
|
logging.info(message)
|
|
print(message)
|
|
|
|
# Update the login credentials and server information accordingly
|
|
name = 52868686
|
|
key = 'kkk7s$zzz6'
|
|
serv = 'ICMarketsSC-Demo'
|
|
|
|
# Global variables
|
|
SYMBOL = "EURUSD"
|
|
LOT_SIZE = 0.01
|
|
TIMEFRAME = mt5.TIMEFRAME_D1
|
|
N_BARS = 50000
|
|
MAGIC_NUMBER = 234003
|
|
SLEEP_TIME = 86400 # 4 hours in seconds
|
|
COMMENT_ML = "regression return"
|
|
|
|
def select_features_rf_reg(X, y, estimator, max_features=20):
|
|
"""
|
|
Use a RandomForest (or similar) to select top 'max_features' features.
|
|
"""
|
|
selector = SelectFromModel(estimator=estimator, threshold=-np.inf, max_features=max_features).fit(X, y)
|
|
X_transformed = selector.transform(X)
|
|
selected_features_mask = selector.get_support()
|
|
return X_transformed, selected_features_mask
|
|
|
|
class TradingApp:
|
|
def __init__(self, symbol, lot_size, magic_number):
|
|
self.symbol = symbol
|
|
self.lot_size = lot_size
|
|
self.magic_number = magic_number
|
|
self.pipeline = None # We'll store the loaded pipeline here
|
|
self.last_retrain_time = None
|
|
|
|
def get_data(self, symbol, n, timeframe):
|
|
"""
|
|
Fetch 'n' bars of historical data for the given symbol and timeframe.
|
|
"""
|
|
rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, n)
|
|
rates_frame = pd.DataFrame(rates)
|
|
rates_frame['time'] = pd.to_datetime(rates_frame['time'], unit='s')
|
|
rates_frame.set_index('time', inplace=True)
|
|
return rates_frame
|
|
|
|
def add_all_ta_features(self, df):
|
|
"""
|
|
Add technical analysis features to the DataFrame using the 'ta' library.
|
|
"""
|
|
df = ta.add_all_ta_features(
|
|
df,
|
|
open="open",
|
|
high="high",
|
|
low="low",
|
|
close="close",
|
|
volume="tick_volume",
|
|
fillna=True
|
|
)
|
|
return df
|
|
|
|
def load_pipeline(self, pipeline_path):
|
|
"""
|
|
Load a pre-trained pipeline (scaler + model + possibly feature selection)
|
|
from disk, e.g. 'best_rf_pipeline.pkl'.
|
|
"""
|
|
self.pipeline = joblib.load(pipeline_path)
|
|
logging.info(f"Loaded pipeline from {pipeline_path}")
|
|
|
|
def ml_signal_generation(self, symbol, n_bars, timeframe):
|
|
"""
|
|
Generate buy/sell signals using the loaded pipeline.
|
|
Make sure the pipeline expects the same features as we create below.
|
|
"""
|
|
if self.pipeline is None:
|
|
logging.error("No pipeline loaded. Call load_pipeline(...) first.")
|
|
return False, False, True, True
|
|
|
|
# 1) Fetch new data
|
|
df = self.get_data(symbol, n_bars, timeframe)
|
|
# 2) Add TA features (if your pipeline doesn't handle feature eng, do it here)
|
|
df = self.add_all_ta_features(df)
|
|
df.fillna(method='ffill', inplace=True)
|
|
|
|
# 3) Prepare the features (the pipeline will do scaling/selection if included)
|
|
X_new = df # If your pipeline expects specific columns, subset accordingly.
|
|
|
|
# 4) Predict with the pipeline
|
|
predictions = self.pipeline.predict(X_new)
|
|
latest_pred = predictions[-1] # Get the most recent bar's prediction
|
|
|
|
buy_signal = latest_pred > 0
|
|
sell_signal = latest_pred < 0
|
|
|
|
return buy_signal, sell_signal, not buy_signal, not sell_signal
|
|
|
|
def calculate_future_returns(self, df):
|
|
"""
|
|
(Optional) Example function to calculate future returns for labeling.
|
|
"""
|
|
df["future_returns"] = df["close"].pct_change().shift(-1)
|
|
return df.dropna()
|
|
|
|
def orders(self, symbol, lot, is_buy=True, id_position=None, sl=None, tp=None):
|
|
"""
|
|
Place an order (BUY or SELL) for the specified symbol and lot size.
|
|
"""
|
|
symbol_info = mt5.symbol_info(symbol)
|
|
if symbol_info is None:
|
|
log_and_print(f"Symbol {symbol} not found, can't place order.", is_error=True)
|
|
return "Symbol not found"
|
|
|
|
# Make sure symbol is selected/visible
|
|
if not symbol_info.visible:
|
|
if not mt5.symbol_select(symbol, True):
|
|
log_and_print(f"Failed to select symbol {symbol}", is_error=True)
|
|
return "Symbol not visible or could not be selected."
|
|
|
|
tick_info = mt5.symbol_info_tick(symbol)
|
|
if tick_info is None:
|
|
log_and_print(f"Could not get tick info for {symbol}.", is_error=True)
|
|
return "Tick info unavailable"
|
|
|
|
# Check for valid bid/ask
|
|
if tick_info.bid <= 0 or tick_info.ask <= 0:
|
|
log_and_print(
|
|
f"Zero or invalid bid/ask for {symbol}: bid={tick_info.bid}, ask={tick_info.ask}",
|
|
is_error=True
|
|
)
|
|
return "Invalid prices"
|
|
|
|
# ----------- LOT SIZE VALIDATION -----------
|
|
lot = max(lot, symbol_info.volume_min)
|
|
step = symbol_info.volume_step
|
|
if step > 0:
|
|
remainder = lot % step
|
|
if remainder != 0:
|
|
lot = lot - remainder + step
|
|
if lot > symbol_info.volume_max:
|
|
lot = symbol_info.volume_max
|
|
|
|
log_and_print(
|
|
f"Adjusted lot size to {lot} (min={symbol_info.volume_min}, "
|
|
f"step={symbol_info.volume_step}, max={symbol_info.volume_max})"
|
|
)
|
|
|
|
# ----------- FORCE ORDER_FILLING_IOC -----------
|
|
filling_mode = 1 # ORDER_FILLING_IOC
|
|
|
|
order_type = mt5.ORDER_TYPE_BUY if is_buy else mt5.ORDER_TYPE_SELL
|
|
deviation = 20
|
|
|
|
request = {
|
|
"action": mt5.TRADE_ACTION_DEAL,
|
|
"symbol": symbol,
|
|
"volume": lot,
|
|
"type": order_type,
|
|
"deviation": deviation,
|
|
"magic": self.magic_number,
|
|
"comment": COMMENT_ML,
|
|
"type_time": mt5.ORDER_TIME_GTC,
|
|
"type_filling": filling_mode,
|
|
}
|
|
|
|
if sl is not None:
|
|
request["sl"] = sl
|
|
if tp is not None:
|
|
request["tp"] = tp
|
|
if id_position is not None:
|
|
request["position"] = id_position
|
|
|
|
log_and_print(f"Sending order request: {request}")
|
|
result = mt5.order_send(request)
|
|
|
|
order_type_str = "BUY" if is_buy else "SELL"
|
|
if result is None or result.retcode != mt5.TRADE_RETCODE_DONE:
|
|
error_message = f"Order failed for {symbol}"
|
|
if result:
|
|
error_message += f", retcode={result.retcode}, comment={result.comment}"
|
|
additional_info = (
|
|
f"Date/Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
|
|
f"Order Type: {order_type_str}\n"
|
|
f"Lot Size: {lot}\n"
|
|
f"SL: {sl if sl else 'None'}\n"
|
|
f"TP: {tp if tp else 'None'}\n"
|
|
f"Comment: {COMMENT_ML}\n"
|
|
f"Request: {request}\n"
|
|
f"Result: {result}"
|
|
)
|
|
log_and_print(f"Order failed details: {additional_info}", is_error=True)
|
|
else:
|
|
success_message = f"Order successful for {symbol}, comment={result.comment}"
|
|
additional_info = (
|
|
f"Date/Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
|
|
f"Order Type: {order_type_str}\n"
|
|
f"Lot Size: {lot}\n"
|
|
f"SL: {sl if sl else 'None'}\n"
|
|
f"TP: {tp if tp else 'None'}\n"
|
|
f"Comment: {COMMENT_ML}"
|
|
)
|
|
log_and_print(success_message)
|
|
|
|
def get_positions_by_magic(self, symbol, magic_number):
|
|
"""
|
|
Retrieve open positions for the specified symbol and magic number.
|
|
"""
|
|
all_positions = mt5.positions_get(symbol=symbol)
|
|
if not all_positions:
|
|
log_and_print("No positions found.", is_error=False)
|
|
return []
|
|
return [pos for pos in all_positions if pos.magic == magic_number]
|
|
|
|
def run_strategy(self, symbol, lot, buy_signal, sell_signal):
|
|
"""
|
|
Based on buy/sell signals, decide whether to open or close positions.
|
|
"""
|
|
log_and_print("------------------------------------------------------------------")
|
|
log_and_print(
|
|
f"Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}, "
|
|
f"SYMBOL: {symbol}, BUY SIGNAL: {buy_signal}, SELL SIGNAL: {sell_signal}"
|
|
)
|
|
|
|
# Retrieve positions based on the magic number to manage trades specific to this instance
|
|
positions = self.get_positions_by_magic(symbol, self.magic_number)
|
|
has_buy = any(pos.type == mt5.POSITION_TYPE_BUY for pos in positions)
|
|
has_sell = any(pos.type == mt5.POSITION_TYPE_SELL for pos in positions)
|
|
|
|
# Decision making based on current signals and existing positions
|
|
if buy_signal and not has_buy:
|
|
if has_sell:
|
|
log_and_print("Existing sell positions found. Attempting to close...")
|
|
if self.close_position(symbol, is_buy=True):
|
|
log_and_print("Sell positions closed. Placing new buy order.")
|
|
self.orders(symbol, lot, is_buy=True)
|
|
else:
|
|
log_and_print("Failed to close sell positions.")
|
|
else:
|
|
self.orders(symbol, lot, is_buy=True)
|
|
elif sell_signal and not has_sell:
|
|
if has_buy:
|
|
log_and_print("Existing buy positions found. Attempting to close...")
|
|
if self.close_position(symbol, is_buy=False):
|
|
log_and_print("Buy positions closed. Placing new sell order.")
|
|
self.orders(symbol, lot, is_buy=False)
|
|
else:
|
|
log_and_print("Failed to close buy positions.")
|
|
else:
|
|
self.orders(symbol, lot, is_buy=False)
|
|
else:
|
|
log_and_print("Appropriate position already exists or no signal to act on.")
|
|
|
|
def close_position(self, symbol, is_buy):
|
|
"""
|
|
Closes positions of the opposite type (BUY/SELL) for this app's magic number.
|
|
"""
|
|
positions = mt5.positions_get(symbol=symbol)
|
|
if not positions:
|
|
log_and_print(f"No positions to close for symbol: {symbol}")
|
|
return False
|
|
|
|
initial_balance = mt5.account_info().balance
|
|
closed_any = False
|
|
|
|
for position in positions:
|
|
# Close positions of the opposite type with the same magic number
|
|
if position.magic == self.magic_number and (
|
|
(is_buy and position.type == mt5.POSITION_TYPE_SELL) or
|
|
(not is_buy and position.type == mt5.POSITION_TYPE_BUY)
|
|
):
|
|
close_request = {
|
|
"action": mt5.TRADE_ACTION_DEAL,
|
|
"symbol": symbol,
|
|
"volume": position.volume,
|
|
"type": mt5.ORDER_TYPE_BUY if position.type == mt5.POSITION_TYPE_SELL else mt5.ORDER_TYPE_SELL,
|
|
"position": position.ticket,
|
|
"deviation": 20,
|
|
"magic": self.magic_number,
|
|
"comment": COMMENT_ML,
|
|
"type_time": mt5.ORDER_TIME_GTC,
|
|
"type_filling": mt5.ORDER_FILLING_RETURN,
|
|
}
|
|
result = mt5.order_send(close_request)
|
|
if result.retcode != mt5.TRADE_RETCODE_DONE:
|
|
error_message = (
|
|
f"Failed to close position {position.ticket} for {symbol}: {result.retcode}"
|
|
)
|
|
log_and_print(error_message, is_error=True)
|
|
else:
|
|
log_and_print(f"Successfully closed position {position.ticket} for {symbol}")
|
|
closed_any = True
|
|
|
|
if closed_any:
|
|
final_balance = mt5.account_info().balance
|
|
profit = final_balance - initial_balance
|
|
success_message = f"Closed positions successfully, Profit: {profit}"
|
|
log_and_print(success_message)
|
|
return True
|
|
|
|
return False
|
|
|
|
def check_and_execute_trades(self):
|
|
"""
|
|
Convenience method to perform the entire flow:
|
|
generate signals, run strategy, and deselect symbol.
|
|
"""
|
|
mt5.symbol_select(self.symbol, True)
|
|
buy, sell, _, _ = self.ml_signal_generation(self.symbol, N_BARS, TIMEFRAME)
|
|
self.run_strategy(self.symbol, self.lot_size, buy, sell)
|
|
mt5.symbol_select(self.symbol, False)
|
|
log_and_print("Waiting for new signals...")
|
|
|
|
def is_market_open():
|
|
"""
|
|
Check if the current time is within typical Forex trading session hours (CET/CEST).
|
|
- Closes: Friday 10:00 PM CET
|
|
- Opens: Sunday 11:00 PM CET
|
|
- Closed all day Saturday
|
|
"""
|
|
current_time_utc = datetime.utcnow()
|
|
# Adjust for CET (UTC+1) or CEST (UTC+2)
|
|
current_time_cet = current_time_utc + timedelta(hours=2) if time.localtime().tm_isdst else current_time_utc + timedelta(hours=1)
|
|
|
|
# Friday after 10 PM CET
|
|
if current_time_cet.weekday() == 4 and current_time_cet.hour >= 22:
|
|
return False
|
|
# Sunday before 11 PM CET
|
|
elif current_time_cet.weekday() == 6 and current_time_cet.hour < 23:
|
|
return False
|
|
# All day Saturday
|
|
elif current_time_cet.weekday() == 5:
|
|
return False
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
if not mt5.initialize(login=name, server=serv, password=key):
|
|
log_and_print("Failed to initialize MetaTrader 5", is_error=True)
|
|
exit()
|
|
|
|
app = TradingApp(symbol=SYMBOL, lot_size=LOT_SIZE, magic_number=MAGIC_NUMBER)
|
|
|
|
# Load a previously trained pipeline (scaler + model, etc.)
|
|
pipeline_path = "models/saved_models/best_rf_pipeline.pkl"
|
|
app.load_pipeline(pipeline_path)
|
|
log_and_print(f"Loaded final pipeline for {app.symbol}")
|
|
|
|
while True:
|
|
log_and_print("Checking market status...")
|
|
if is_market_open():
|
|
log_and_print("Market is open. Executing trades...")
|
|
|
|
# Generate signals using the loaded pipeline
|
|
buy_signal, sell_signal, _, _ = app.ml_signal_generation(
|
|
symbol=app.symbol,
|
|
n_bars=N_BARS,
|
|
timeframe=TIMEFRAME
|
|
)
|
|
|
|
# Run strategy
|
|
app.run_strategy(app.symbol, app.lot_size, buy_signal, sell_signal)
|
|
else:
|
|
log_and_print("Market is closed. No actions performed.")
|
|
|
|
# Sleep for the configured interval (e.g., 4 hours)
|
|
time.sleep(SLEEP_TIME)
|
|
|
|
except KeyboardInterrupt:
|
|
log_and_print("Shutdown signal received.")
|
|
except Exception as e:
|
|
error_message = f"An error occurred: {e}"
|
|
log_and_print(error_message, is_error=True)
|
|
finally:
|
|
mt5.shutdown()
|
|
log_and_print("MetaTrader 5 shutdown completed.")
|