数据科学与机器学习（第43篇）：使用潜在高斯混合模型（LGMM）识别指标数据中的隐藏模式·进阶篇

class="kw">import pandas as pd
class="kw">import numpy as np
class="kw">import MetaTrader5 as mt5
class="kw">import os
from Trade.TerminalInfo class="kw">import CTerminalInfo
class="kw">import matplotlib.pyplot as plt
class="kw">import seaborn
class="kw">import warnings
warnings.filterwarnings("ignore")
seaborn.set_style("darkgrid")
if not mt5.initialize():
    print("Failed to Initialize MetaTrade5, Error = ",mt5.last_error())
    mt5.shutdown()
    
terminal = CTerminalInfo() # similarly to CTerminalInfo from MQL5. 用于获取 MetaTrader class="num">5 应用程序的信息
common_path = os.path.join(terminal.common_data_path(), "Files")
symbol = "XAUUSD"
timeframe = "PERIOD_D1"
df = pd.read_csv(os.path.join(common_path, f"Oscillators.{symbol}.{timeframe}.csv")) # the same naming pattern as the one used in the MQL5 script
# Identify max class="type">float value
max_float = np.finfo(class="type">float).max
# Replace all max class="type">float (class="type">class="kw">double) values with NaN produced by preliminary indicator calculations
df = df.replace(max_float, np.nan)
df.dropna(inplace=True)
df["Time"] = pd.to_datetime(df["Time"], unit="s")
df.head()

df["future_close"] = df["Close"].shift(-lookahead)
new_df = df.dropna()
new_df["Direction"] = np.where(new_df["future_close"]>new_df["Close"], class="num">1, -class="num">1) # if a the close value in the next bar(s)=lookahead is above the current close price, thats a class="type">long signal otherwise that&class="macro">#x27;s a class="type">short signal
from sklearn.model_selection class="kw">import train_test_split
X = new_df.drop(columns=[
    "Time",
    "Open",
    "High",
    "Low",
    "Close",
    "future_close",
    "Direction"
])
y = new_df["Direction"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=class="num">0.2, shuffle=True, random_state=class="num">42)
X_train.head()

from sklearn.mixture class="kw">import GaussianMixture
from skl2onnx class="kw">import convert_sklearn
from skl2onnx.common.data_types class="kw">import FloatTensorType
components = class="num">3
gmm = GaussianMixture(n_components=components, covariance_type="full", random_state=class="num">42)
gmm.fit(X_train)
latent_features_train = gmm.predict_proba(X_train)
latent_features_test = gmm.predict_proba(X_test)
latent_features_train
array([[class="num">9.48947877e-13, class="num">1.08107288e-62, class="num">1.00000000e+00],
       [class="num">9.71935407e-01, class="num">2.80542130e-02, class="num">1.03801388e-05],
       [class="num">5.35722226e-03, class="num">9.94642667e-01, class="num">1.10916653e-07],
       ...,
       [class="num">7.72441751e-08, class="num">8.80712550e-41, class="num">9.99999923e-01],
       [class="num">9.99975623e-01, class="num">1.07924534e-33, class="num">2.43771745e-05],
       [class="num">1.91968188e-01, class="num">8.08030586e-01, class="num">1.22621110e-06]], shape=(class="num">3760, class="num">3))

# Define input type(shape should match your training data)
initial_type = [("float_input", FloatTensorType([None, X_train.shape[class="num">1]]))]
# Convert the pipeline to ONNX format
onnx_model = convert_sklearn(gmm, initial_types=initial_type)
# Save the model to a file
with open(os.path.join(common_path, f"LGMM.{symbol}.{timeframe}.onnx"), "wb") as f:
    f.write(onnx_model.SerializeToString())
class CGaussianMixture
  {
class="kw">protected:
   class="type">bool initialized;
   class="type">long onnx_handle;
   class="type">void PrintTypeInfo(const class="type">long num,const class="type">class="kw">string layer,const OnnxTypeInfo& type_info);

   class="type">ulong inputs[]; class=class="str">"cmt">//Inputs of a model in dimensions [nxn]
   class="kw">struct outputs_struct
    {
      class="type">ulong outputs[];
    } model_output_structure[];  class=class="str">"cmt">//Outputs of the model structure array
class="type">bool CGaussianMixture::OnnxLoad(class="type">long &handle)
{
class=class="str">"cmt">//--- since not all sizes defined in the input tensor we must set them explicitly
class=class="str">"cmt">//--- first index - batch size, second index - series size, third index - number of series(only Close)

  OnnxTypeInfo type_info; class=class="str">"cmt">//Getting onnx information for Reference In case you forgot what the loaded ONNX is all about
  class="type">long input_count=OnnxGetInputCount(handle);
  if (MQLInfoInteger(MQL_DEBUG))
     Print("model has ",input_count," input(s)");

  for(class="type">long i=class="num">0; i<input_count; i++)
   {
     class="type">class="kw">string input_name=OnnxGetInputName(handle,i);
     if (MQLInfoInteger(MQL_DEBUG))
        Print(i," input name is ",input_name);

     if(OnnxGetInputTypeInfo(handle,i,type_info))
      {
       if (MQLInfoInteger(MQL_DEBUG))
          PrintTypeInfo(i,"input",type_info);
       ArrayCopy(inputs, type_info.tensor.dimensions);
      }
   }
  class="type">long output_count=OnnxGetOutputCount(handle);
  if (MQLInfoInteger(MQL_DEBUG))
     Print("model has ",output_count," output(s)");

  ArrayResize(model_output_structure, (class="type">int)output_count);

  for(class="type">long i=class="num">0; i<output_count; i++)
   {
     class="type">class="kw">string output_name=OnnxGetOutputName(handle,i);
     if (MQLInfoInteger(MQL_DEBUG))
        Print(i," output name is ",output_name);

if(OnnxGetOutputTypeInfo(handle,i,type_info))
  {
    if (MQLInfoInteger(MQL_DEBUG))
      PrintTypeInfo(i,"output",type_info);
    
    ArrayCopy(model_output_structure[i].outputs, type_info.tensor.dimensions);
  }
  
  class=class="str">"cmt">//--- Set the output shape
  
    replace(model_output_structure);
    if(!OnnxSetOutputShape(handle, i, model_output_structure[i].outputs))
    {
      if (MQLInfoInteger(MQL_DEBUG))
        {
          printf("Failed to set the Output[%d] shape Err=%d",i,GetLastError());
          DebugBreak();
        }
      
      class="kw">return false;
    }
  }

class=class="str">"cmt">//---
  
  replace(inputs);
  
class=class="str">"cmt">//--- Setting the input size
  for (class="type">long i=class="num">0; i<input_count; i++)  
    if (!OnnxSetInputShape(handle, i, inputs)) class=class="str">"cmt">//Giving the Onnx handle the input shape
    {
      if (MQLInfoInteger(MQL_DEBUG))
        printf("Failed to set the input shape Err=%d",GetLastError());
      DebugBreak();
      class="kw">return false;
    }

  
  initialized = true;
  if (MQLInfoInteger(MQL_DEBUG))
    Print("ONNX model Initialized");
    
  class="kw">return true;
}
class=class="str">"cmt">//+------------------------------------------------------------------+
class=class="str">"cmt">//|                                                                  |
class=class="str">"cmt">//+------------------------------------------------------------------+
class="type">bool CGaussianMixture::Init(class="type">class="kw">string onnx_filename, class="type">uint flags=ONNX_DEFAULT)
{  
  onnx_handle = OnnxCreate(onnx_filename, flags);
  
  if (onnx_handle == INVALID_HANDLE)
    class="kw">return false;
  
  class="kw">return OnnxLoad(onnx_handle);
}
class=class="str">"cmt">//+------------------------------------------------------------------+
class=class="str">"cmt">//|                                                                  |
class=class="str">"cmt">//+------------------------------------------------------------------+