271 lines
10 KiB
Python
271 lines
10 KiB
Python
"""
| |
This module handles the collection of market data from MetaTrader 5 (MT5).
| |
It defines tasks for connecting to MT5, requesting historical kline data,
| |
and storing it in the database. It also includes health checks for the MT5 connection.
| |
"""
| |
| |
import time
| |
from datetime import datetime, timedelta
| |
| |
import pandas as pd
| |
import asyncio
| |
import MetaTrader5 as mt5
| |
import pytz
| |
| |
from common.utils import mt5_freq_from_pandas, get_timedelta_for_mt5_timeframe
| |
from service.App import *
| |
from common.utils import *
| |
from service.analyzer import *
| |
from service.mt5 import connect_mt5
| |
| |
import logging
| |
| |
log = logging.getLogger('collector')
| |
| |
| |
async def main_collector_task() -> int:
| |
"""
| |
The main collector task that is executed periodically.
| |
It checks the MT5 connection, synchronizes data, and handles any errors.
| |
| |
Returns:
| |
int: 0 if the task completed successfully, 1 otherwise.
| |
"""
| |
| |
symbol = App.config["symbol"]
| |
pandas_freq = App.config["freq"]
| |
start_ts, end_ts = pandas_get_interval(pandas_freq)
| |
now_ts = now_timestamp()
| |
| |
log.info(f"===> Start collector task. Timestamp {now_ts}. Interval [{start_ts},{end_ts}].")
| |
| |
#
| |
# 0. Check MT5 connection
| |
#
| |
if data_provider_problems_exist():
| |
await data_provider_health_check()
| |
if data_provider_problems_exist():
| |
log.error(f"Problems with the data provider server found. No signaling, no trade. Will try next time.")
| |
return 1
| |
| |
#
| |
# 1. Ensure that we are up-to-date with klines
| |
#
| |
res = await sync_data_collector_task()
| |
| |
if res > 0:
| |
log.error(f"Problem getting data from the server. No signaling, no trade. Will try next time.")
| |
return 1
| |
| |
log.info(f"<=== End collector task.")
| |
return 0
| |
| |
| |
#
| |
# Request/update market data
| |
#
| |
| |
async def sync_data_collector_task() -> int:
| |
"""
| |
Synchronizes the local data state with the latest data from MT5.
| |
This task retrieves the most recent kline data for specified symbols and
| |
stores it in the database.
| |
| |
Returns:
| |
int: 0 if the data was synchronized successfully, 1 otherwise.
| |
| |
Raises:
| |
TimeoutError: If the data request times out.
| |
Exception: If any other error occurs during data retrieval.
| |
"""
| |
| |
CHUNK_SIZE = 10000 # How many bars worth of duration to request in each chunk
| |
RATE_LIMIT_DELAY = 0.1 # Small delay between requests (seconds)
| |
| |
data_sources = App.config.get("data_sources", [])
| |
symbols = [x.get("folder") for x in data_sources]
| |
pandas_freq = App.config["freq"]
| |
mt5_timeframe = mt5_freq_from_pandas(pandas_freq)
| |
| |
if not symbols:
| |
symbols = [App.config["symbol"]]
| |
| |
# Connect to trading account (same as before)
| |
mt5_account_id = App.config.get("mt5_account_id")
| |
mt5_password = App.config.get("mt5_password")
| |
mt5_server = App.config.get("mt5_server")
| |
if mt5_account_id and mt5_password and mt5_server:
| |
authorized = connect_mt5(int(mt5_account_id), password=str(mt5_password), server=str(mt5_server))
| |
if not authorized:
| |
log.error(f"MT5 Login failed for account #{mt5_account_id}, error code: {mt5.last_error()}")
| |
return 1
| |
| |
| |
# How many records are missing (and to be requested) for each symbol (not used here)
| |
# missing_klines_counts = [App.analyzer.get_missing_klines_count(sym) for sym in symbols]
| |
| |
# Create a list of tasks for retrieving data
| |
tasks = [asyncio.create_task(request_klines(s, pandas_freq, mt5_timeframe, CHUNK_SIZE, RATE_LIMIT_DELAY)) for s in symbols]
| |
| |
results = {}
| |
timeout = 10 # Seconds to wait for the result
| |
| |
# Process responses in the order of arrival
| |
for fut in asyncio.as_completed(tasks, timeout=timeout):
| |
# Get the results
| |
res = None
| |
try:
| |
res = await fut
| |
except TimeoutError as te:
| |
log.warning(f"Timeout {timeout} seconds when requesting kline data.")
| |
return 1
| |
except Exception as e:
| |
log.warning(f"Exception when requesting kline data.")
| |
return 1
| |
| |
# Add to the database (will overwrite existing klines if any)
| |
if res and res.keys():
| |
# res is dict for symbol, which is a list of record lists of 12 fields
| |
# ==============================
| |
# TODO: We need to check these fields for validity (presence, non-null)
| |
# TODO: We can load maximum 999 latest klines, so if more 1600, then some other method
| |
# TODO: Print somewhere diagnostics about how many lines are in history buffer of db, and if nans are found
| |
results.update(res)
| |
try:
| |
added_count = App.analyzer.store_klines(res)
| |
except Exception as e:
| |
log.error(f"Error storing kline result in the database. Exception: {e}")
| |
return 1
| |
else:
| |
log.error("Received empty or wrong result from klines request.")
| |
return 1
| |
| |
# --- Shutdown MT5 (same as before) ---
| |
log.info("\nShutting down MetaTrader 5 connection...")
| |
mt5.shutdown()
| |
| |
return 0
| |
| |
| |
async def request_klines(symbol: str, pandas_freq: str, mt5_timeframe: int, CHUNK_SIZE: int, RATE_LIMIT_DELAY: float) -> dict:
| |
"""
| |
Requests kline data from MT5 for a given symbol.
| |
It fetches data in chunks to avoid overloading the server and handles
| |
potential errors during the data retrieval process.
| |
| |
Args:
| |
symbol (str): The trading symbol (e.g., "EURUSD").
| |
pandas_freq (str): The pandas frequency string (e.g., "1min", "5min").
| |
mt5_timeframe (int): The MT5 timeframe constant (e.g., mt5.TIMEFRAME_M1).
| |
CHUNK_SIZE (int): The number of bars to request in each chunk.
| |
RATE_LIMIT_DELAY (float): The delay in seconds between requests.
| |
| |
Returns:
| |
dict: A dictionary with the symbol as the key and a list of klines as the value.
| |
Each kline is a list of data points.
| |
| |
Raises:
| |
ValueError: If an invalid MT5 timeframe is provided.
| |
Exception: If any other error occurs during data retrieval.
| |
"""
| |
| |
time_column = App.config["time_column"]
| |
timezone = pytz.timezone("Etc/UTC")
| |
| |
# Define end point for download (now)
| |
end_dt = datetime.now(timezone)
| |
| |
# Get the last kline from the database
| |
last_kline_dt = App.analyzer.get_last_kline_time(symbol)
| |
if last_kline_dt:
| |
current_start_dt = last_kline_dt
| |
log.info(f"Existing data found. Will download data starting from {current_start_dt.strftime('%Y-%m-%d %H:%M:%S %Z')}")
| |
else:
| |
# Define a default historical start if no file exists | 2017 | 2024
| |
current_start_dt = datetime(2014, 1, 1, tzinfo=timezone) # Or get from config if needed
| |
log.info(f"No existing data found. Starting download from {current_start_dt.strftime('%Y-%m-%d %H:%M:%S %Z')}.")
| |
| |
all_klines_list = []
| |
| |
try:
| |
while current_start_dt < end_dt:
| |
try:
| |
# Calculate the duration for CHUNK_SIZE bars
| |
chunk_duration = get_timedelta_for_mt5_timeframe(mt5_timeframe, CHUNK_SIZE)
| |
except ValueError as e:
| |
log.error(f"Error calculating duration: {e}. Stopping download for {symbol}.")
| |
break
| |
| |
# Calculate the temporary end date for this chunk request
| |
temp_end_dt = current_start_dt + chunk_duration
| |
| |
# Ensure the temporary end date doesn't go beyond the overall end date
| |
temp_end_dt = min(temp_end_dt, end_dt)
| |
| |
# Add a small buffer (e.g., 1 second) to start_dt for the request
| |
# to definitively exclude the current_start_dt bar itself in the range request.
| |
request_start_dt = current_start_dt + timedelta(seconds=1)
| |
| |
# Avoid making a request if the adjusted start is already >= temp_end
| |
if request_start_dt >= temp_end_dt:
| |
log.info(f" Skipping request: Calculated range is empty or invalid ({request_start_dt.strftime('%Y-%m-%d %H:%M:%S %Z')} to {temp_end_dt.strftime('%Y-%m-%d %H:%M:%S %Z')}).")
| |
break # Likely means we are caught up
| |
| |
log.info(f" Fetching range from {request_start_dt.strftime('%Y-%m-%d %H:%M:%S %Z')} to {temp_end_dt.strftime('%Y-%m-%d %H:%M:%S %Z')}...")
| |
| |
# Use copy_rates_range, since copy_rates_from seems to fetch data using backward lookback(present to past)
| |
rates = mt5.copy_rates_range(symbol, mt5_timeframe, request_start_dt, temp_end_dt)
| |
| |
if rates is None:
| |
log.error(f" mt5.copy_rates_range returned None. Error: {mt5.last_error()}. Stopping download for {symbol}.")
| |
break
| |
if len(rates) == 0:
| |
log.info(" No data returned in this range. Download may be complete or data gap.")
| |
# If no data, advance start time past this chunk's end to avoid getting stuck
| |
current_start_dt = temp_end_dt
| |
if current_start_dt >= end_dt:
| |
log.info(" Reached end date after empty range.")
| |
break
| |
else:
| |
log.info(f" Advancing start time to {current_start_dt.strftime('%Y-%m-%d %H:%M:%S %Z')} and continuing.")
| |
time.sleep(RATE_LIMIT_DELAY) # Still pause slightly
| |
continue # Try the next chunk
| |
| |
chunk_df = pd.DataFrame(rates)
| |
# Convert 'time' (Unix seconds) to datetime objects (UTC)
| |
chunk_df[time_column] = pd.to_datetime(chunk_df['time'], unit='s', utc=True)
| |
all_klines_list.append(chunk_df)
| |
last_bar_time_in_chunk = chunk_df[time_column].iloc[-1]
| |
log.info(f" Fetched {len(chunk_df)} bars. Last timestamp in chunk: {last_bar_time_in_chunk.strftime('%Y-%m-%d %H:%M:%S %Z')}")
| |
# Update the start time for the next chunk request to be the end time of THIS chunk's last bar
| |
current_start_dt = last_bar_time_in_chunk
| |
# Small delay before next request
| |
time.sleep(RATE_LIMIT_DELAY)
| |
| |
except Exception as e:
| |
log.error(f"Exception while requesting klines: {e}")
| |
return {}
| |
| |
# Return all received klines with the symbol as a key
| |
return {symbol: all_klines_list}
| |
| |
#
| |
# Server and account info
| |
#
| |
| |
| |
async def data_provider_health_check() -> int:
| |
"""
| |
Performs a health check on the MT5 connection.
| |
It verifies if the MT5 terminal is accessible and returns the status.
| |
| |
Returns:
| |
int: 0 if the MT5 connection is healthy, 1 otherwise.
| |
| |
"""
| |
| |
# Check MT5 connection
| |
if not mt5.terminal_info():
| |
log.error(f"MT5 terminal not found.")
| |
return 1
| |
return 0
|