数据科学和机器学习（第 32 部分）：保持您的 AI 模型更新，在线学习（基础篇）

class="kw">import pandas as pd
class="kw">import numpy as np
class="kw">import MetaTrader5 as mt5
from class="type">class="kw">datetime class="kw">import class="type">class="kw">datetime
if not mt5.initialize(): # Initialize the MetaTrader class="num">5 platform
    print("initialize() failed")
    mt5.shutdown()
def getData(start = class="num">1, bars = class="num">1000):
    rates = mt5.copy_rates_from_pos("EURUSD", mt5.TIMEFRAME_H1, start, bars)

  if len(rates) < bars: # if the received information is less than specified
        print("Failed to copy rates from MetaTrader class="num">5, error = ",mt5.last_error())
    # create a pnadas DataFrame out of the obtained data
    df_rates = pd.DataFrame(rates)
                        class="kw">return df_rates
print("Trading info:\n" ,getData(class="num">1, class="num">100)) # get class="num">100 bars starting at the recent closed bar
                      time     open     high      low     close  tick_volume  spread  real_volume
class="num">0   class="num">1731351600  class="num">1.06520  class="num">1.06564  class="num">1.06451  class="num">1.06491          class="num">1688       class="num">0            class="num">0
class="num">1   class="num">1731355200  class="num">1.06491  class="num">1.06519  class="num">1.06460  class="num">1.06505          class="num">1607       class="num">0            class="num">0
class="num">2   class="num">1731358800  class="num">1.06505  class="num">1.06573  class="num">1.06495  class="num">1.06512          class="num">1157       class="num">0            class="num">0
class="num">3   class="num">1731362400  class="num">1.06512  class="num">1.06564  class="num">1.06512  class="num">1.06557          class="num">1112       class="num">0            class="num">0
class="num">4   class="num">1731366000  class="num">1.06557  class="num">1.06579  class="num">1.06553  class="num">1.06557           class="num">776       class="num">0            class="num">0
..          ...      ...      ...      ...      ...           ...     ...          ...
class="num">95  class="num">1731693600  class="num">1.05354  class="num">1.05516  class="num">1.05333  class="num">1.05513          class="num">5125       class="num">0            class="num">0

from catboost class="kw">import CatBoostClassifier
from sklearn.metrics class="kw">import accuracy_score
from onnx.helper class="kw">import get_attribute_value
from skl2onnx class="kw">import convert_sklearn, update_registered_converter
from sklearn.pipeline class="kw">import Pipeline
from skl2onnx.common.shape_calculator class="kw">import (
    calculate_linear_classifier_output_shapes,
)  # noqa
from skl2onnx.common.data_types class="kw">import (
    FloatTensorType,
    Int64TensorType,
    guess_tensor_type,
)
from skl2onnx._parse class="kw">import _apply_zipmap, _get_sklearn_operator_name
from catboost.utils class="kw">import convert_to_onnx_object
# Example initial data(X_initial, y_initial are your initial feature matrix and target)
class CatBoostClassifierModel():
    def __init__(self, X_train, X_test, y_train, y_test):
        self.X_train = X_train
        self.X_test = X_test
        self.y_train = y_train
        self.y_test = y_test
        self.model = None
    def train(self, iterations=class="num">100, depth=class="num">6, learning_rate=class="num">0.1, loss_function="CrossEntropy", use_best_model=True):
        # Initialize the CatBoost model
        params = {
            "iterations": iterations,
            "depth": depth,
            "learning_rate": learning_rate,
            "loss_function": loss_function,
            "use_best_model": use_best_model
        }
        self.model = Pipeline([ # wrap a catboost classifier in sklearn pipeline | good practice(not necessary tho :))
            ("catboost", CatBoostClassifier(**params))
        ])
        # Testing the model
        
        self.model.fit(X=self.X_train, y=self.y_train, catboost__eval_set=(self.X_test, self.y_test))
        y_pred = self.model.predict(self.X_test)
        print("Model&class="macro">#x27;s accuracy on out-of-sample data = ",accuracy_score(self.y_test, y_pred))
    # a function for saving the trained CatBoost model to ONNX format
    def to_onnx(self, model_name):
        update_registered_converter(
            CatBoostClassifier,
            "CatBoostCatBoostClassifier",
            calculate_linear_classifier_output_shapes,
            self.skl2onnx_convert_catboost,
            parser=self.skl2onnx_parser_castboost_classifier,

options={"nocl": [True, False], "zipmap": [True, False, "columns"]},
)
model_onnx = convert_sklearn(
    self.model,
    "pipeline_catboost",
    [("input", FloatTensorType([None, self.X_train.shape[class="num">1]]))],
    target_opset={"": class="num">12, "ai.onnx.ml": class="num">2},
)
# And save.
with open(model_name, "wb") as f:
    f.write(model_onnx.SerializeToString())
data = getData(start=class="num">1, bars=class="num">1000)
# Preparing the target variable
data["future_open"] = data["open"].shift(-class="num">1) # shift one bar into the future
data["future_close"] = data["close"].shift(-class="num">1)
target = []
for row in range(data.shape[class="num">0]):
    if data["future_close"].iloc[row] > data["future_open"].iloc[row]: # bullish signal
        target.append(class="num">1)
    else: # bearish signal
        target.append(class="num">0)
data["target"] = target # add the target variable to the dataframe
data = data.dropna() # drop empty rows
X = data.drop(columns = ["spread","real_volume","future_close","future_open","target"])
y = data["target"]
X_train, X_test, y_train, y_test = train_test_split(X, y, train_size=class="num">0.7, random_state=class="num">42)
catboost_model = catboost_models.CatBoostClassifierModel(X_train, X_test, y_train, y_test)
catboost_model.train()
terminal_info_dict = mt5.terminal_info()._asdict()
common_path = terminal_info_dict["commondata_path"]
# Save models in a specific location under the common parent folder
models_path = os.path.join(common_path, "Files")
if not os.path.exists(models_path): class="macro">#if the folder exists
    os.makedirs(models_path) # Create the folder if it doesn&class="macro">#x27;t exist
catboost_model.to_onnx(model_name=os.path.join(models_path, "catboost.H1.onnx"))
def trainAndSaveCatBoost():
    data = getData(start=class="num">1, bars=class="num">1000)
    # Check if we were able to receive some data
    if (len(data)<=class="num">0):
        print("Failed to obtain data from Metatrader5, error = ",mt5.last_error())
        mt5.shutdown()
    # Preparing the target variable
    data["future_open"] = data["open"].shift(-class="num">1) # shift one bar into the future
    data["future_close"] = data["close"].shift(-class="num">1)
    target = []
    for row in range(data.shape[class="num">0]):
        if data["future_close"].iloc[row] > data["future_open"].iloc[row]: # bullish signal
            target.append(class="num">1)
        else: # bearish signal
            target.append(class="num">0)