forked from nique_372/AiDataGenByLeo
420 lines
No EOL
18 KiB
Python
420 lines
No EOL
18 KiB
Python
# Copyright 2026, Niquel Mendoza | Leo.
|
|
# https://www.mql5.com/es/users/nique_372
|
|
# trainer_regression.py
|
|
|
|
#+------------------------------------------------------------------+
|
|
#| Imports |
|
|
#+------------------------------------------------------------------+
|
|
import matplotlib
|
|
matplotlib.use('Agg')
|
|
|
|
import os
|
|
|
|
# Clases
|
|
from comunicator import CMqlComunication
|
|
from trainer import CModelTrainer
|
|
from regresion_trainer import CModelTrainerRegression
|
|
|
|
# Funciones
|
|
import inspect
|
|
# Copias profundas
|
|
from copy import deepcopy
|
|
|
|
# Archvios
|
|
from pathlib import Path
|
|
from natsort import natsorted
|
|
|
|
# Fix onnx
|
|
import onnx
|
|
from onnx import helper
|
|
from out_fix import fix_onnx_output_shape
|
|
|
|
# Funciones y logger base
|
|
from trainer import CLoggerBase
|
|
from trainer import Funciones
|
|
|
|
# Typing
|
|
from typing import Dict, Any
|
|
|
|
|
|
#+------------------------------------------------------------------+
|
|
#| Clase base para entrenar todos los modelos de un folder simbolo |
|
|
#+------------------------------------------------------------------+
|
|
class CPipelineTraining(CLoggerBase):
|
|
|
|
def __init__(self, general_config : dict, config_regresion : dict,
|
|
config_clasificacion : dict, initial_log_flags : int, comunicador : CMqlComunication):
|
|
super().__init__()
|
|
|
|
# Iniciamos banderas
|
|
self.AddLogFlags(initial_log_flags)
|
|
|
|
|
|
# Comunicador
|
|
self.m_comunicador : CMqlComunication = comunicador
|
|
self.m_comunicador.AddLogFlags(self.LogFlags())
|
|
|
|
# Features init
|
|
self.m_features_pred : list[str] = []
|
|
self.m_features_tp : list[str] = []
|
|
self.m_features_sl : list[str] = []
|
|
|
|
# Rutas relativas (constantes, nunca cambian)
|
|
self.m_features_pred_relativo : str = general_config.get('features_pred_file', '')
|
|
self.m_features_tp_relativo : str = general_config.get('features_tp_file', '')
|
|
self.m_features_sl_relativo : str = general_config.get('features_sl_file', '')
|
|
|
|
# Nombres de archvios
|
|
self.m_filename_idx : str = general_config.get("file_name_idx", "")
|
|
self.m_filename_features_ptr : str = general_config.get("file_name_features_ptr", "")
|
|
|
|
# filenames
|
|
self.m_path_features_pred : str = ""
|
|
self.m_path_features_tp : str = ""
|
|
self.m_path_features_sl : str = ""
|
|
|
|
# init
|
|
self.m_current_timeframe_folder_idx : int = -1
|
|
self.m_current_strategy_folder_idx : int = -1
|
|
|
|
# Seteamos config
|
|
self.m_config_regresion : dict = deepcopy(config_regresion)
|
|
self.m_config_clasificacion : dict = deepcopy(config_clasificacion)
|
|
|
|
# Ruta de la carpeta base para entrenar modelos (normalmente nombre de un simbolo)
|
|
self.m_ruta_base : str = general_config.get('data_path')
|
|
self.m_ruta_entrenmiento : str = os.path.join(self.m_ruta_base, general_config.get('path_project'))
|
|
self.m_simbolo = Path(self.m_ruta_entrenmiento).name
|
|
|
|
# file_data
|
|
self.m_archivo_desc : Path = Path(os.path.join(self.m_ruta_entrenmiento,"temp.txt"))
|
|
|
|
if self.m_archivo_desc.exists():
|
|
self.LogInfo("Archivo init existe temp.txt")
|
|
cadena : str = self.m_archivo_desc.read_text('utf-16-le').lstrip('\ufeff').strip()
|
|
cadenas : list[str] = cadena.split("_")
|
|
|
|
# solo si el tamaño es de 2 asingamos
|
|
if len(cadenas) == 2:
|
|
self.m_current_timeframe_folder_idx = int(cadenas[0])
|
|
self.m_current_strategy_folder_idx = int(cadenas[1])
|
|
|
|
self.LogInfo(f"Empezando desde timeframe folder index = {self.m_current_timeframe_folder_idx} y strategy index = {self.m_current_strategy_folder_idx}")
|
|
else:
|
|
self.LogInfo("Archvio temp.txt de init no existe empzando desde inicio")
|
|
|
|
# ahora listamos todos los folders
|
|
self.m_folders : list[str] = []
|
|
|
|
# obteemos todos lo foldes del folder simbolo
|
|
for f in os.listdir(self.m_ruta_entrenmiento):
|
|
full_path : str = os.path.join(self.m_ruta_entrenmiento, f)
|
|
if(os.path.isdir(full_path)):
|
|
self.m_folders.append(full_path)
|
|
|
|
# check
|
|
if(len(self.m_folders) < 1):
|
|
self.LogError("El folder no tiene elementos")
|
|
return
|
|
|
|
# ordenamos
|
|
self.m_folders = natsorted(self.m_folders)
|
|
|
|
# imprimimos los folders
|
|
if(self.IsInfoLogEnabled()):
|
|
self.FastLog(inspect.currentframe().f_code.co_name,CLoggerBase.INFO_TEXT, f"Folders encontrados en folder con simbolo {self.m_simbolo}: ")
|
|
print(self.m_folders)
|
|
|
|
#+--------------------------------------------------------------------------+
|
|
#| Leer archivo de features, simple read y split de comma |
|
|
#+--------------------------------------------------------------------------+
|
|
def _LeerFeaturesFile(self, path: str) -> list[str] | None:
|
|
try:
|
|
contenido = Path(path).read_text(encoding='utf-16-le').lstrip('\ufeff').strip()
|
|
features = [f.strip() for f in contenido.split(',')]
|
|
return features if features else None
|
|
except Exception as e:
|
|
self.LogError(f"Fallo leer features: {path} → {e}")
|
|
return None
|
|
|
|
#+--------------------------------------------------------------------------+
|
|
#| Intenta sobreescribir cada array si encuentra el archivo en este nivel |
|
|
#+--------------------------------------------------------------------------+
|
|
def _IntentarCargarFeatures(self, carpeta_nivel: str) -> None:
|
|
|
|
rutas : Dict[str, str] = {
|
|
'pred': os.path.join(carpeta_nivel, self.m_features_pred_relativo),
|
|
'tp': os.path.join(carpeta_nivel, self.m_features_tp_relativo),
|
|
'sl': os.path.join(carpeta_nivel, self.m_features_sl_relativo),
|
|
}
|
|
|
|
# Iteracion por todo el dict
|
|
# La idea es ir viendo si se existe el archivo relativo y si existe sobreecribir el valor
|
|
# de la varialbe miembro correspondiente
|
|
for tipo, path in rutas.items():
|
|
if os.path.exists(path):
|
|
features = self._LeerFeaturesFile(path)
|
|
if features is not None:
|
|
setattr(self, f"m_path_features_{tipo}", path.replace(self.m_ruta_base + os.sep, ""))
|
|
setattr(self, f"m_features_{tipo}", features)
|
|
self.LogInfo(f"Override features [{tipo}] desde: {path}")
|
|
|
|
|
|
#+--------------------------------------------------------------------------+
|
|
#| Check antes del entremiento de cada modelo (por ahora solo features) |
|
|
#+--------------------------------------------------------------------------+
|
|
def _CheckAntesEntrenamiento(self) -> None:
|
|
if (len(self.m_features_sl) < 1 or
|
|
len(self.m_features_pred) < 1 or
|
|
len(self.m_features_tp) < 1):
|
|
self.LogFatalError("No se han logrado cargar las features para el entremiento")
|
|
Funciones.Remover(1) # Salimos fallo fatal
|
|
|
|
|
|
|
|
#+------------------------------------------------------------------+
|
|
#| Funcion base para procesar un string array |
|
|
#| La idea es obtener los indices del modelo (feautres elegidas) |
|
|
#+------------------------------------------------------------------+
|
|
def _ProcesarStringArray(self, array_features : list[str], array_base_feautres : list[str]) -> list[int]:
|
|
idx_arr : list[int] = []
|
|
|
|
# Iteracion
|
|
for feature_str in array_features:
|
|
# declracion inciial
|
|
idx : int = -1
|
|
|
|
# tratamos de obtener su indice
|
|
try:
|
|
idx = array_base_feautres.index(feature_str)
|
|
except ValueError:
|
|
if(feature_str != " tipo de operacion"):
|
|
self.LogCriticalError(f"Error al obtener indice de features, features '{feature_str}' invalida")
|
|
self.LogInfo(f' Base features:\n{Funciones.array_to_string(array_base_feautres,"|","[","]")}')
|
|
self.LogInfo(f' Selected features:\n{Funciones.array_to_string(array_features,"|","[","]")}')
|
|
Funciones.Remover(1)
|
|
return None
|
|
|
|
if(idx == -1):
|
|
self.LogInfo(f"Omitiendo features {feature_str} dado que no se encontro en indice, puede ser normal si noe s tipo de operacion")
|
|
continue
|
|
else:
|
|
idx_arr.append(idx) # encontramos el indice de la feautres
|
|
|
|
|
|
return idx_arr # retonramos el array con los idncies encontrados
|
|
|
|
|
|
#+------------------------------------------------------------------+
|
|
#| Funcion base para procesar un string array |
|
|
#| La idea es obtener los indices del modelo (feautres elegidas) |
|
|
#+------------------------------------------------------------------+
|
|
def _ProcesarTimeframeFolder(self, path_folder_timeframe: str) -> bool:
|
|
|
|
# Intentemoa cargar la confix de feautures de este nivel
|
|
self._IntentarCargarFeatures(path_folder_timeframe)
|
|
|
|
# Iteramos sobre todos los folsers [Estrategias] de este nivel
|
|
folders : list[str] = self._GetSortedSubFolders(path_folder_timeframe)
|
|
|
|
for index, folder_final in enumerate(folders):
|
|
if index <= self.m_current_strategy_folder_idx:
|
|
continue
|
|
|
|
if not self._ProcesarStrategyFolder(index, folder_final):
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
#+----------------------------------------------------------------------------+
|
|
#| Funcion para procesar un folder de estratega (esta si tiene los modelos) |
|
|
#+----------------------------------------------------------------------------+
|
|
def _ProcesarStrategyFolder(self, index: int, folder: str) -> bool:
|
|
# Intentamos cargar las features que pueda exisitir en este nivel
|
|
self._IntentarCargarFeatures(folder)
|
|
|
|
# Check de features
|
|
self._CheckAntesEntrenamiento()
|
|
|
|
# Info
|
|
self.LogInfo(f"Procesando folder: {folder}")
|
|
|
|
contenido_idx : str = ""
|
|
|
|
# Clasificacion
|
|
idx = self._EntrenarClasificacion(folder)
|
|
if idx is None:
|
|
return False
|
|
contenido_idx += self._FormatIdx("modelo_pred_idx", idx)
|
|
|
|
# Regresion TP
|
|
idx = self._EntrenarRegresion(folder, "tp")
|
|
if idx is None:
|
|
return False
|
|
contenido_idx += self._FormatIdx("modelo_pred_tp_idx", idx)
|
|
|
|
# Regresión SL
|
|
idx = self._EntrenarRegresion(folder, "sl")
|
|
if idx is None:
|
|
return False
|
|
contenido_idx += self._FormatIdx("modelo_pred_sl_idx", idx)
|
|
|
|
# Escritura y fix
|
|
if not self._EscribirIdx(folder, contenido_idx):
|
|
return False
|
|
if not self._EscriirFileFeaturesPtr(folder):
|
|
return False
|
|
|
|
self._FixOnnxModelos(folder)
|
|
self._GuardarCheckpoint(index)
|
|
|
|
return True
|
|
|
|
#+----------------------------------------------------------------------------+
|
|
#| Funcion para entnrenar a los modelos de clasificacion |
|
|
#+----------------------------------------------------------------------------+
|
|
def _EntrenarClasificacion(self, folder: str) -> list[int] | None:
|
|
self.m_config_clasificacion['csv_file'] = os.path.join(folder, self.m_config_clasificacion["data_csv_file"])
|
|
self.m_config_clasificacion['output_folder'] = folder
|
|
|
|
# configracuinl del modelo y propagacion de flags
|
|
modelo = CModelTrainer(self.m_config_clasificacion)
|
|
modelo.AddLogFlags(self.LogFlags())
|
|
|
|
if not modelo.Execute():
|
|
self.LogCriticalError(f"Fallo clasificacion, data = : {self.m_config_clasificacion['csv_file']}")
|
|
return None
|
|
else:
|
|
self.m_comunicador.SendClasificacion(folder, modelo.GetMetrics())
|
|
|
|
return self._ProcesarStringArray(modelo.GetSelectedFeatures(), self.m_features_pred)
|
|
|
|
#+----------------------------------------------------------------------------+
|
|
#| Funcion para entnrenar a los modelos de regresion |
|
|
#+----------------------------------------------------------------------------+
|
|
def _EntrenarRegresion(self, folder: str, tipo: str) -> list[int] | None:
|
|
# tipo = "tp" o "sl"
|
|
# configuracion inicial del modelo de regresion
|
|
self.m_config_regresion['csv_file'] = os.path.join(folder, self.m_config_regresion[f"data_csv_file_{tipo}"])
|
|
self.m_config_regresion['output_folder'] = folder
|
|
self.m_config_regresion['model_name'] = self.m_config_regresion[f'model_name_{tipo}']
|
|
|
|
# Creamos el modelo y propagamos sus flags
|
|
modelo = CModelTrainerRegression(self.m_config_regresion)
|
|
modelo.AddLogFlags(self.LogFlags())
|
|
|
|
if not modelo.Execute():
|
|
self.LogCriticalError(f"Fallo regresion {tipo}, archivo = : {self.m_config_regresion['csv_file']}")
|
|
return None
|
|
else:
|
|
self.m_comunicador.SendRegresion(folder, modelo.GetMetrics())
|
|
|
|
return self._ProcesarStringArray(modelo.GetSelectedFeatures(), getattr(self, f"m_features_{tipo}"))
|
|
|
|
#+----------------------------------------------------------------------------+
|
|
#| Funcion para arreglar la salida ouptu de los modelos para hacerlo 1 |
|
|
#+----------------------------------------------------------------------------+
|
|
def _FixOnnxModelos(self, folder: str) -> None:
|
|
for tipo in ["tp", "sl"]:
|
|
nombre = self.m_config_regresion[f'model_name_{tipo}']
|
|
src = os.path.join(folder, f"{nombre}.onnx")
|
|
dest = os.path.join(folder, f"{nombre}_f.onnx")
|
|
fix_onnx_output_shape(src, dest)
|
|
|
|
#+----------------------------------------------------------------------------+
|
|
#| Escribir todos los indices de feaures en el archivo idx luego del training |
|
|
#+----------------------------------------------------------------------------+
|
|
def _EscribirIdx(self, folder: str, contenido: str) -> bool:
|
|
|
|
# Indices de modelos
|
|
path : Path = Path(os.path.join(folder, self.m_filename_idx))
|
|
try:
|
|
path.write_text(contenido, encoding='utf-16-le')
|
|
return True
|
|
except Exception as e:
|
|
self.LogError(f"Fallo escribir en {path.name}, err: {e}")
|
|
return False
|
|
|
|
#+----------------------------------------------------------------------------------------------+
|
|
#| Escribir el archivo pointer que da las rutas de donde se ubican los archivos de features |
|
|
#+----------------------------------------------------------------------------------------------+
|
|
def _EscriirFileFeaturesPtr(self, folder : str) -> bool:
|
|
# Path
|
|
path : Path = Path(os.path.join(folder, self.m_filename_features_ptr))
|
|
data : str = ""
|
|
|
|
# contenido
|
|
data += self.m_path_features_pred + "\n"
|
|
data += self.m_path_features_sl + "\n"
|
|
data += self.m_path_features_tp + "\n"
|
|
|
|
# intentamos la escritura
|
|
try:
|
|
path.write_text(data, encoding='utf-16-le')
|
|
return True
|
|
except Exception as e:
|
|
self.LogError(f"Fallo escribir en {path.name}, err: {e}")
|
|
return False
|
|
|
|
|
|
|
|
#+----------------------------------------------------------------------------+
|
|
#| Guardar el progreso |
|
|
#+----------------------------------------------------------------------------+
|
|
def _GuardarCheckpoint(self, index: int) -> None:
|
|
self.m_current_strategy_folder_idx = index
|
|
cadena = f"{self.m_current_timeframe_folder_idx}_{index}"
|
|
self.m_archivo_desc.write_text(cadena, encoding='utf-16-le')
|
|
|
|
#+----------------------------------------------------------------------------+
|
|
#| Format idx para obtener una cadena |
|
|
#+----------------------------------------------------------------------------+
|
|
def _FormatIdx(self, nombre: str, indices: list[int]) -> str:
|
|
return nombre + ": " + Funciones.array_to_string(indices, ",", "[", "]") + "\n"
|
|
|
|
|
|
# Sort de los folders
|
|
def _GetSortedSubFolders(self, path: str) -> list[str]:
|
|
folders = [
|
|
os.path.join(path, f)
|
|
for f in os.listdir(path)
|
|
if os.path.isdir(os.path.join(path, f))
|
|
]
|
|
return natsorted(folders)
|
|
|
|
#+---------------------------------------------------------------------+
|
|
#| Esta funcion es la principal aqui iteramos sobre todos los folders |
|
|
#| Timeframe y llamaamos a ProcesarTimeframeFolder |
|
|
#| Para que procese todos los modelos |
|
|
#+---------------------------------------------------------------------+
|
|
def Execute(self) -> bool:
|
|
# Check point inicial
|
|
self._GuardarCheckpoint(self.m_current_strategy_folder_idx)
|
|
|
|
# Intenteamos cargar feautres del root
|
|
self._IntentarCargarFeatures(self.m_ruta_entrenmiento)
|
|
|
|
# Iteracion principal
|
|
for index, folder in enumerate(self.m_folders):
|
|
|
|
# Si el indice es menor al inicio previsto omitmimos esta iteracion
|
|
if(index<=self.m_current_timeframe_folder_idx):
|
|
continue
|
|
|
|
# Procesamos archivo
|
|
if(not self._ProcesarTimeframeFolder(folder)):
|
|
self.LogError(f"Fallo al procesar el folder timeframe:\n{folder}")
|
|
Funciones.Remover(1)
|
|
return False
|
|
else: # Exito al procesar
|
|
self.m_current_strategy_folder_idx = -1 # indice de strategy
|
|
self.m_current_timeframe_folder_idx = index
|
|
self._GuardarCheckpoint(self.m_current_strategy_folder_idx)
|
|
|
|
|
|
# Fin del ckecpoint
|
|
self.m_archivo_desc.write_text("finalizado",encoding='utf-16-le')
|
|
|
|
# Retornamos exito
|
|
return True # Exito |