# Copyright 2025, MetaQuotes Ltd. # https://www.mql5.com from datetime import datetime, timezone import json import sys import os import MetaTrader5 as mt5 import numpy as np import pandas as pd from arch import arch_model from arch.univariate.base import ARCHModelResult MQL5_FILES_FOLDER = "MQL5\\FILES" MQL5_COMMON_FOLDER = "FILES" def arch_to_json(fitted_arch_model: ARCHModelResult, file_path: str): """ Serializes a fitted_arch_model ARMA-GARCH model to a JSON file compatible with MQL5 arch library. Args: fitted_arch_model: The arch_model result object (e.g., from arch_model().fit()). file_path (str): The path and filename to save the JSON file to. """ if not isinstance(fitted_arch_model,ARCHModelResult): raise TypeError(f'invalid type supplied, "fitted_arch_model" should be an instance of ARCHModelResult') if len(file_path) < 1 or not isinstance(file_path,str): raise TypeError(f'invalid filename supplied') print(f"-> Attempting to save model parameters to: {file_path}...") try: params_dict = fitted_arch_model.params.to_dict() mean_model_spec = { 'name':fitted_arch_model.model.name, 'num_params':fitted_arch_model.model.num_params, 'parameter_names':fitted_arch_model.model.parameter_names(), 'lags':np.asarray(fitted_arch_model.model.lags).tolist(), 'hold_back':fitted_arch_model.model.hold_back, 'rescale':str(fitted_arch_model.model.rescale), 'scale':fitted_arch_model.model.scale, 'constant':str(fitted_arch_model.model.constant), 'use_rotated':str(fitted_arch_model.model.use_rotated) } volatility_model_spec = { 'name': fitted_arch_model.model.volatility.name, 'num_params':fitted_arch_model.model.volatility.num_params, 'parameter_names':fitted_arch_model.model.volatility.parameter_names(), 'start':fitted_arch_model.model.volatility.start, 'stop':fitted_arch_model.model.volatility.stop, 'closed_form':str(fitted_arch_model.model.volatility.closed_form), 'updateable':str(fitted_arch_model.model.volatility.updateable), 'p':getattr(fitted_arch_model.model.volatility, "p", 0), 'o':getattr(fitted_arch_model.model.volatility, "o", 0), 'q':getattr(fitted_arch_model.model.volatility, "q", 0), 'power':getattr(fitted_arch_model.model.volatility,"power",0) } distribution_model_spec = { 'name':fitted_arch_model.model.distribution.name, 'num_params':fitted_arch_model.model.distribution.num_params, 'parameter_names':fitted_arch_model.model.distribution.parameter_names() } model_spec = { 'arch_model':mean_model_spec, 'volatility_process':volatility_model_spec, 'distribution':distribution_model_spec } # 3. Combine Data and Save to JSON opt = getattr(fitted_arch_model,"optimization_result",None) data_to_save = { 'parameters': params_dict, 'model_spec': model_spec, 'fun': getattr(opt,"fun",None), 'jac': list(getattr(opt,"jac","")), 'x' : list(getattr(opt,"x","")) } with open(file_path, 'w') as file: json.dump(data_to_save, file, indent=None,separators=(',', ':')) except Exception as e: print(f"!!! Error saving model to JSON: {e}") sys.exit(1) def parse_lags(lag_string): """Helper to convert a comma-separated string of integers into a list.""" if not lag_string: return 0 try: if "," in lag_string: return [int(x.strip()) for x in lag_string.split(",") if x.strip()] return int(lag_string) except Exception: sys.exit(1) raise ArgumentTypeError("Lags must be an integer or a comma-separated list of integers.") def main(): file_name = str(sys.argv[1]) dtime = str(sys.argv[2]) symbol = str(sys.argv[3]) timeframe = str(sys.argv[4]) scale = float(sys.argv[5]) count = int(sys.argv[6]) _p = int(sys.argv[7]) _o = int(sys.argv[8]) _q = int(sys.argv[9]) _mean = str(sys.argv[10]) _include_constant = bool(sys.argv[11]) _vol = str(sys.argv[12]) _dist = str(sys.argv[13]) _lags = str(sys.argv[14]) vol_mapping = { "constant": "constant", "arch": "arch", "garch": "garch", "gjr-garch": "garch", # GJR-GARCH is a GARCH model with asymmetric lags (o > 0) "tarch": "garch", # TARCH is GARCH with absolute values (power=1.0) "avarch": "garch", # Absolute value ARCH "avgarch": "garch", # Absolute value GARCH } dist_mapping = { "normal": "normal", "student": "studentsut", "skewed-student": "skewt", "gen-error": "ged" } chosen_vol = vol_mapping[_vol] chosen_dist = dist_mapping[_dist] power_val = 2.0 if _vol in ["tarch", "avarch", "avgarch"]: power_val = 1.0 if not mt5.initialize(): print(f"Error: MT5 initialization failed.") sys.exit(1) return tinfo = mt5.terminal_info() if tinfo != None: tdict = mt5.terminal_info()._asdict() file_name = os.path.join(tdict['commondata_path'], MQL5_COMMON_FOLDER, file_name) print(f"file_name {file_name}") else: print("Terminal info failure") mt5.shutdown() sys.exit(1) return timeframe_mapping = { "M1": mt5.TIMEFRAME_M1, "M5": mt5.TIMEFRAME_M5, "M15": mt5.TIMEFRAME_M15, "M30": mt5.TIMEFRAME_M30, "H1": mt5.TIMEFRAME_H1, "H4": mt5.TIMEFRAME_H4, "D1": mt5.TIMEFRAME_D1 } try: tz = timezone.utc dt = datetime.strptime(dtime, "%Y.%m.%d_%H:%M") rates = mt5.copy_rates_from(symbol, timeframe_mapping[timeframe], dt.astimezone(tz), count) df = pd.DataFrame(rates) returns = np.log(df["close"] / df["close"].shift(1)).dropna() * scale except Exception as e: print(f"Error parsing model tuning parameters: {str(e)}") sys.exit(1) finally: mt5.shutdown() try: model = arch_model( returns, mean=_mean, lags=parse_lags(_lags), vol=chosen_vol, p=_p, o=_o, q=_q, power=power_val, dist=chosen_dist ) res = model.fit(disp="off") arch_to_json(res,file_name) except Exception as e: print(f"Model Configuration/Fitting Error: {str(e)}") sys.exit(1) """ out = f"Dependent Variable Summary (Scaled x{args.scale})\n" out += f"Mean Model : {_mean.upper()} (Constant: {_include_constant}, Lags: {_lags})\n" out += f"Volatility Model : {_vol.upper()} (p={_p}, o={_o}, q={_q})\n" out += f"Distribution : {_dist.upper()}\n" out += f"Minimum Negative Log-Likelihood: {-res.loglikelihood:.6f}\n" out += f"Estimated Parameters:\n" out += f"{res}" try: gradients = res.model.score(res.params) for param_name, grad_val in zip(res.params.index, gradients): print(f"{param_name:<12} : {grad_val:.10f}") except Exception: print("Gradient calculation unsupported or failed for this specification.") print("==================================================") """ if __name__ == "__main__": main()