code: def fit_single_model(
      self, ctx: Context, train_data: DataFrame, valid_data: DataFrame
  ) -> None:
      self.header = train_data.schema.columns.to_list()
      rank_a, rank_b = ctx.hosts[0].rank, ctx.guest.rank
      if ctx.is_on_host:
          self.init_param["fit_intercept"] = False
      if self.w is None:
          initialize_func = get_initialize_func(**self.init_param)
      else:
          initialize_func = lambda x: self.w
      if self.init_param.get("fit_intercept"):
          train_data["intercept"] = 1.0
      train_data_n = train_data.shape[0]
      layer = SSHELogisticRegressionLayer(
          ctx,
          in_features_a=ctx.mpc.option_call(
              lambda: train_data.shape[1], dst=rank_a
          ),
          in_features_b=ctx.mpc.option_call(
              lambda: train_data.shape[1], dst=rank_b
          ),
          out_features=1,
          rank_a=rank_a,
          rank_b=rank_b,
          wa_init_fn=initialize_func,
          wb_init_fn=initialize_func,
      )
      loss_fn = SSHELogisticRegressionLossLayer(
          ctx, rank_a=rank_a, rank_b=rank_b
      )
      optimizer = SSHEOptimizerSGD(ctx, layer.parameters(), lr=self.lr)
      wa = layer.wa
      wb = layer.wb
      if ctx.is_on_guest:
          batch_loader = dataframe.DataLoader(
              train_data,
              ctx=ctx,
              batch_size=self.batch_size,
              mode="hetero",
              role="guest",
              sync_arbiter=False,
          )
      else:
          batch_loader = dataframe.DataLoader(
              train_data,
              ctx=ctx,
              batch_size=self.batch_size,
              mode="hetero",
              role="host",
          )
      # if self.reveal_every_epoch:
      if self.early_stop == "weight_diff":
          wa_p = wa.get_plain_text(dst=rank_a)
          wb_p = wb.get_plain_text(dst=rank_b)
          if ctx.is_on_guest:
              self.converge_func.set_pre_weight(wb_p)
          else:
              self.converge_func.set_pre_weight(wa_p)
      for i, epoch_ctx in ctx.on_iterations.ctxs_range(self.epochs):
          epoch_loss = None
          logger.info(f"enter {i}th epoch")
          for batch_ctx, batch_data in epoch_ctx.on_batches.ctxs_zip(
              batch_loader
          ):
              h = batch_data.x
              y = batch_ctx.mpc.cond_call(
                  lambda: batch_data.label, lambda: None, dst=rank_b
              )
              # if self.reveal_every_epoch:
              #    z = batch_ctx.mpc.cond_call(lambda: torch.matmul(h, wa_p.detach()),
              #                                lambda: torch.matmul(h, wb_p.detach()), dst=rank_a)
              # else:
              z = layer(h)
              loss = loss_fn(z, y)
              if i % self.reveal_loss_freq == 0:
                  if epoch_loss is None:
                      epoch_loss = loss.get(dst=rank_b)
                      if epoch_loss:
                          epoch_loss = epoch_loss * h.shape[0]
                  else:
                      batch_loss = loss.get(dst=rank_b)
                      if batch_loss:
                          epoch_loss += batch_loss * h.shape[0]
              loss.backward()
              optimizer.step()
          if epoch_loss is not None and ctx.is_on_guest:
              epoch_loss = epoch_loss / train_data_n
              epoch_ctx.metrics.log_loss("lr_loss", epoch_loss.tolist())
          # if self.reveal_every_epoch:
          #    wa_p = wa.get_plain_text(dst=rank_a)
          #    wb_p = wb.get_plain_text(dst=rank_b)
          if ctx.is_on_guest:
              if self.early_stop == "weight_diff":
                  """if self.reveal_every_epoch:
                      wb_p_delta = self.converge_func.compute_weight_diff(wb_p - self.converge_func.pre_weight)
                      w_diff = wb_p_delta + epoch_ctx.hosts.get("wa_p_delta")[0]
                      self.converge_func.set_pre_weight(wb_p)
                      if w_diff < self.tol:
                          self.is_converged = True
                  else:
                      raise ValueError(f"early stop {self.early_stop} is not supported when "
                                       f"reveal_every_epoch is False")
                  """
                  wa_p = wa.get_plain_text(dst=rank_a)
                  wb_p = wb.get_plain_text(dst=rank_b)
                  wb_p_delta = self.converge_func.compute_weight_diff(
                      wb_p - self.converge_func.pre_weight
                  )
                  w_diff = (
                      wb_p_delta + epoch_ctx.hosts.get("wa_p_delta")[0]
                  )
                  self.converge_func.set_pre_weight(wb_p)
                  if w_diff < self.tol:
                      self.is_converged = True
              else:
                  if i % self.reveal_loss_freq == 0:
                      self.is_converged = (
                          self.converge_func.is_converge(epoch_loss)
                      )
              epoch_ctx.hosts.put("converge_flag", self.is_converged)
          else:
              if self.early_stop == "weight_diff":
                  # if self.reveal_every_epoch:
                  wa_p = wa.get_plain_text(dst=rank_a)
                  wb_p = wb.get_plain_text(dst=rank_b)
                  wa_p_delta = self.converge_func.compute_weight_diff(
                      wa_p - self.converge_func.pre_weight
                  )
                  epoch_ctx.guest.put("wa_p_delta", wa_p_delta)
                  self.converge_func.set_pre_weight(wa_p)
              self.is_converged = epoch_ctx.guest.get("converge_flag")
          if self.is_converged:
              self.end_epoch = i
              break
      if not self.is_converged:
          self.end_epoch = self.epochs
      wa_p = wa.get_plain_text(dst=rank_a)
      wb_p = wb.get_plain_text(dst=rank_b)
      if ctx.is_on_host:
          self.w = wa_p
      else:
          self.w = wb_p
  
imports: import logging
  import torch
  from fate.arch import Context, dataframe
  from fate.arch.dataframe import DataFrame
  from fate.arch.protocol.mpc.nn.sshe.lr_layer import SSHELogisticRegressionLayer, SSHELogisticRegressionLossLayer, SSHEOptimizerSGD
  from fate.ml.abc.module import Module, HeteroModule
  from fate.ml.utils import predict_tools
  from fate.ml.utils._convergence import converge_func_factory
  from fate.ml.utils._model_param import get_initialize_func
  from fate.ml.utils._model_param import serialize_param, deserialize_param
label_desc: Use image node_id 2 for calling the SSHELREstimator obj's underlying member method code with example usage: obj.fit_single_model(ctx, train_data, valid_data) without return types