mirror of
https://github.com/asavinov/intelligent-trading-bot.git
synced 2026-05-04 16:26:44 +00:00
353 lines
16 KiB
Python
353 lines
16 KiB
Python
import time
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
import os
|
|
|
|
import pandas as pd
|
|
import click
|
|
import MetaTrader5 as mt5
|
|
import pytz
|
|
|
|
from common.utils import mt5_freq_from_pandas, get_timedelta_for_mt5_timeframe
|
|
from service.App import App, load_config
|
|
from service.mt5 import connect_mt5
|
|
|
|
|
|
print("MetaTrader5 package author: ", mt5.__author__)
|
|
print("MetaTrader5 package version: ", mt5.__version__)
|
|
|
|
# --- Configuration ---
|
|
DEFAULT_BAR_CHUNK_SIZE = 10000 # How many bars worth of duration to request in each chunk
|
|
DEFAULT_TICK_CHUNK_SIZE = 5 # How many ticks worth of duration to request in each chunk
|
|
RATE_LIMIT_DELAY = 0.1 # Small delay between requests (seconds)
|
|
# ---------------------
|
|
|
|
|
|
# -------------------------------------------------
|
|
|
|
@click.command()
|
|
@click.option('--config_file', '-c', type=click.Path(), default='', help='Configuration file name')
|
|
def main(config_file):
|
|
"""
|
|
Retrieving historic klines from MetaTrader5 server incrementally using copy_rates_range
|
|
with calculated chunk durations. Downloads data from the last record in the existing file
|
|
(or a historical start date) up to the current time, fetching in duration-based chunks.
|
|
"""
|
|
load_config(config_file)
|
|
|
|
data_sources = App.config["data_sources"]
|
|
time_column = App.config["time_column"]
|
|
data_path = Path(App.config["data_folder"])
|
|
download_max_rows = App.config.get("download_max_rows", 0)
|
|
mt5_account_id = App.config.get("mt5_account_id")
|
|
mt5_password = App.config.get("mt5_password")
|
|
mt5_server = App.config.get("mt5_server")
|
|
|
|
script_start_time = datetime.now()
|
|
|
|
pandas_freq = App.config["freq"]
|
|
print(f"Pandas frequency: {pandas_freq}")
|
|
|
|
mt5_timeframe = mt5_freq_from_pandas(pandas_freq)
|
|
# Use timeframe_description for clearer output if available
|
|
try:
|
|
tf_description = mt5.timeframe_description(mt5_timeframe)
|
|
print(f"MetaTrader5 frequency: {tf_description} ({mt5_timeframe})")
|
|
except AttributeError: # Handle older MT5 versions potentially lacking this func
|
|
print(f"MetaTrader5 frequency: {mt5_timeframe}")
|
|
|
|
|
|
# Define the timezone for MT5 (usually UTC)
|
|
timezone = pytz.timezone("Etc/UTC")
|
|
# Define a default historical start if no file exists => 2014 | 2017 | 2024
|
|
historical_start_date = datetime(2017, 1, 1, tzinfo=timezone) # Or get from config if needed
|
|
|
|
# Connect to trading account
|
|
if mt5_account_id and mt5_password and mt5_server:
|
|
authorized = connect_mt5(mt5_account_id, password=str(mt5_password), server=str(mt5_server))
|
|
if authorized:
|
|
print("MT5 Login successful.")
|
|
account_info = mt5.account_info()
|
|
if account_info:
|
|
print(f"Account Info: Login={account_info.login}, Server={account_info.server}, Balance={account_info.balance}")
|
|
else:
|
|
print(f"Could not retrieve account info. Error: {mt5.last_error()}")
|
|
else:
|
|
print(f"MT5 Login failed for account #{mt5_account_id}, error code: {mt5.last_error()}")
|
|
mt5.shutdown()
|
|
return
|
|
else:
|
|
print("MT5 credentials not fully provided in config. Proceeding without login (might affect available symbols/data).")
|
|
|
|
print(f"Terminal Info: {mt5.terminal_info()}")
|
|
|
|
|
|
# --- Loop through data sources ---
|
|
|
|
processed_symbols = []
|
|
|
|
for ds in data_sources:
|
|
quote = str(ds.get("folder"))
|
|
file_type = str(ds.get("file")).lower()
|
|
|
|
if not quote:
|
|
print("ERROR: Folder (symbol) is not specified in data_sources.")
|
|
continue
|
|
|
|
|
|
print(f"\n--- Processing symbol: {quote} ---")
|
|
|
|
file_path = data_path / quote
|
|
file_path.mkdir(parents=True, exist_ok=True)
|
|
file_name = (file_path / "klines").with_suffix(".csv")
|
|
chunk_size = int(ds.get("chunk_size", DEFAULT_BAR_CHUNK_SIZE))
|
|
|
|
|
|
if file_type == "ticks":
|
|
file_name = (file_path / "ticks").with_suffix(".csv")
|
|
chunk_size = int(ds.get("chunk_size", DEFAULT_TICK_CHUNK_SIZE))
|
|
|
|
|
|
existing_df = pd.DataFrame()
|
|
start_dt = historical_start_date
|
|
|
|
|
|
# Check if file exists and load data
|
|
if file_name.is_file():
|
|
try:
|
|
print(f"Loading existing data from: {file_name}")
|
|
# Specify date format for potentially faster parsing if consistent
|
|
existing_df = pd.read_csv(file_name, parse_dates=[time_column], date_format='ISO8601')
|
|
# Ensure timezone is set correctly after parsing
|
|
if pd.api.types.is_datetime64_any_dtype(existing_df[time_column]) and existing_df[time_column].dt.tz is None:
|
|
existing_df[time_column] = existing_df[time_column].dt.tz_localize('UTC')
|
|
elif pd.api.types.is_datetime64_any_dtype(existing_df[time_column]):
|
|
existing_df[time_column] = existing_df[time_column].dt.tz_convert('UTC')
|
|
else: # Fallback if parsing failed or column is not datetime
|
|
print(f"Warning: Column '{time_column}' not parsed as datetime. Attempting conversion.")
|
|
existing_df[time_column] = pd.to_datetime(existing_df[time_column], errors='coerce', utc=True)
|
|
|
|
existing_df = existing_df.dropna(subset=[time_column]) # Drop rows where conversion failed
|
|
|
|
if not existing_df.empty:
|
|
# Sort just in case file wasn't sorted
|
|
existing_df = existing_df.sort_values(by=time_column)
|
|
# Start downloading from the timestamp of the last record
|
|
start_dt = existing_df[time_column].iloc[-1]
|
|
print(f"Existing file found. Will download data starting from {start_dt.strftime('%Y-%m-%d %H:%M:%S %Z')}")
|
|
else:
|
|
print("Existing file was empty or had invalid dates after loading. Starting from historical date.")
|
|
existing_df = pd.DataFrame() # Reset to empty
|
|
except Exception as e:
|
|
print(f"Error loading existing file {file_name}: {e}. Starting from historical date.")
|
|
existing_df = pd.DataFrame() # Reset to empty
|
|
else:
|
|
print(f"File not found. Starting download from {historical_start_date.strftime('%Y-%m-%d %H:%M:%S %Z')}.")
|
|
start_dt = historical_start_date # Ensure start_dt is set
|
|
|
|
|
|
# Define end point for download (now)
|
|
end_dt = datetime.now(timezone)
|
|
|
|
# Check if symbol is available
|
|
symbol_info = mt5.symbol_info(quote)
|
|
if not symbol_info:
|
|
print(f"Symbol {quote} not found or not available in MT5 terminal. Skipping. Error: {mt5.last_error()}")
|
|
continue
|
|
if file_type == "ticks" and not symbol_info.trade_tick_size:
|
|
print(f"Ticks data is not available for {quote}. Skipping. Error: {mt5.last_error()}")
|
|
os.remove(file_name)
|
|
continue
|
|
print(f"Symbol {quote} found in MT5.")
|
|
|
|
|
|
all_klines_list = []
|
|
current_start_dt = start_dt
|
|
|
|
print(f"Starting download loop for {quote} from {current_start_dt.strftime('%Y-%m-%d %H:%M:%S %Z')} to {end_dt.strftime('%Y-%m-%d %H:%M:%S %Z')}...")
|
|
|
|
# --- Download Loop using copy_rates_range or copy_ticks_range with calculated duration ---
|
|
while current_start_dt < end_dt:
|
|
try:
|
|
# Calculate the duration for chunk_size bars or ticks
|
|
|
|
|
|
chunk_duration = get_timedelta_for_mt5_timeframe(mt5_timeframe, chunk_size)
|
|
except ValueError as e:
|
|
print(f"Error calculating duration: {e}. Stopping download for {quote}.")
|
|
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:
|
|
print(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
|
|
|
|
print(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 or copy_ticks_range
|
|
if file_type == "ticks":
|
|
rates = mt5.copy_ticks_range(quote, request_start_dt, temp_end_dt, mt5.COPY_TICKS_ALL)
|
|
if rates is None:
|
|
print(f" mt5.copy_ticks_range returned None. Error: {mt5.last_error()}. Stopping download for {quote}.")
|
|
break
|
|
else:
|
|
rates = mt5.copy_rates_range(quote, mt5_timeframe, request_start_dt, temp_end_dt)
|
|
if rates is None:
|
|
print(f" mt5.copy_rates_range returned None. Error: {mt5.last_error()}. Stopping download for {quote}.")
|
|
break
|
|
|
|
if len(rates) == 0:
|
|
print(" 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:
|
|
print(" Reached end date after empty range.")
|
|
break
|
|
else:
|
|
print(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)
|
|
if file_type == "ticks":
|
|
# Convert 'time_msc' (Unix milliseconds) to datetime objects (UTC)
|
|
chunk_df[time_column] = pd.to_datetime(chunk_df['time_msc'], unit='ms', utc=True)
|
|
else:
|
|
# Convert 'time' (Unix seconds) to datetime objects (UTC)
|
|
chunk_df[time_column] = pd.to_datetime(chunk_df['time'], unit='s', utc=True)
|
|
|
|
# --- IMPORTANT: Filtering is no longer needed here ---
|
|
# Since we requested data *starting after* current_start_dt using request_start_dt,
|
|
# the check `chunk_df = chunk_df[chunk_df[time_column] > current_start_dt]`
|
|
# is redundant and can be removed.
|
|
|
|
# if chunk_df.empty: # This check is effectively handled by len(rates) == 0 now
|
|
# print(" No new bars found in the fetched chunk (after filtering). Stopping.")
|
|
# break
|
|
|
|
all_klines_list.append(chunk_df)
|
|
last_bar_time_in_chunk = chunk_df[time_column].iloc[-1]
|
|
print(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
|
|
|
|
# Check if we've downloaded past the intended end time (redundant check, loop condition handles it)
|
|
# if current_start_dt >= end_dt:
|
|
# print(" Reached or passed target end time. Download complete.")
|
|
# break
|
|
|
|
# Small delay before next request
|
|
time.sleep(RATE_LIMIT_DELAY)
|
|
|
|
# --- Combine and Process Data ---
|
|
if not all_klines_list:
|
|
print(f"No new data downloaded for {quote}.")
|
|
if existing_df.empty:
|
|
print(f"No existing or new data for {quote}. Skipping save.")
|
|
continue
|
|
else:
|
|
print("Saving existing data only (no updates).")
|
|
final_df = existing_df # Use existing data if no new data was fetched
|
|
else:
|
|
print("Combining downloaded data...")
|
|
new_df = pd.concat(all_klines_list, ignore_index=True)
|
|
|
|
# Combine existing and new data
|
|
if not existing_df.empty:
|
|
final_df = pd.concat([existing_df, new_df], ignore_index=True)
|
|
else:
|
|
final_df = new_df
|
|
|
|
print("Processing combined data (duplicates, sorting, columns)...")
|
|
if file_type == "ticks":
|
|
# Standardize columns (assuming MT5 names)
|
|
final_df.rename(columns={
|
|
'time_msc': 'time',
|
|
'flags': 'flags',
|
|
'bid': 'bid',
|
|
'ask': 'ask',
|
|
'last': 'last',
|
|
'volume': 'volume',
|
|
}, inplace=True, errors='ignore') # Added errors='ignore'
|
|
else:
|
|
final_df.rename(columns={
|
|
'tick_volume': 'volume', # Use tick_volume as 'volume'
|
|
}, inplace=True, errors='ignore') # Added errors='ignore'
|
|
|
|
|
|
# Ensure time column is the primary datetime column and drop time column if exist
|
|
if 'time' in final_df.columns and time_column != 'time':
|
|
final_df = final_df.drop('time', axis=1)
|
|
|
|
# Select desired columns (ensure time_column is first)
|
|
required_columns = [time_column, 'open', 'high', 'low', 'close', 'volume']
|
|
# Keep only columns that actually exist in the dataframe
|
|
final_df = final_df[[col for col in required_columns if col in final_df.columns]]
|
|
|
|
# Remove duplicates based on timestamp, keeping the latest entry
|
|
initial_rows = len(final_df)
|
|
final_df = final_df.drop_duplicates(subset=[time_column], keep='last')
|
|
if initial_rows > len(final_df):
|
|
print(f" Removed {initial_rows - len(final_df)} duplicate rows based on '{time_column}'.")
|
|
|
|
# Sort by timestamp
|
|
final_df = final_df.sort_values(by=time_column)
|
|
|
|
# Remove the last row *if* it represents the current, incomplete bar.
|
|
if not final_df.empty:
|
|
# Check if the last bar's time is too close to the script end time
|
|
# A simple heuristic: if the last bar's time is after the loop's end_dt minus one interval, it might be incomplete.
|
|
# Or just always drop the last row after bulk download.
|
|
print("Removing potentially incomplete last bar.")
|
|
final_df = final_df.iloc[:-1]
|
|
|
|
|
|
# Apply max rows limit if specified (same as before)
|
|
if download_max_rows and len(final_df) > download_max_rows:
|
|
print(f"Applying download_max_rows limit: {download_max_rows}")
|
|
final_df = final_df.tail(download_max_rows)
|
|
|
|
# Final check if DataFrame is valid before saving (same as before)
|
|
if final_df.empty:
|
|
print(f"Final dataframe for {quote} is empty after processing. Skipping save.")
|
|
continue
|
|
|
|
# Reset index before saving (same as before)
|
|
final_df = final_df.reset_index(drop=True)
|
|
|
|
# --- Save Data (same as before) ---
|
|
|
|
try:
|
|
if file_type == "ticks":
|
|
final_df = final_df.drop(['time'], axis=1)
|
|
|
|
print(f"Saving {len(final_df)} rows to {file_name}...")
|
|
final_df.to_csv(file_name, index=False, date_format='%Y-%m-%dT%H:%M:%SZ')
|
|
print(f"Finished saving '{quote}'.")
|
|
processed_symbols.append(quote)
|
|
except Exception as e:
|
|
print(f"Error saving file {file_name}: {e}")
|
|
|
|
|
|
# --- Shutdown MT5 (same as before) ---
|
|
print("\nShutting down MetaTrader 5 connection...")
|
|
mt5.shutdown()
|
|
|
|
elapsed = datetime.now() - script_start_time
|
|
print(f"\nFinished downloading data for symbols: {', '.join(processed_symbols) if processed_symbols else 'None'}")
|
|
print(f"Total time: {str(elapsed).split('.')[0]}")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|