(self, data_instances, w, n_iter_, batch_index, loss_norm=None)
Compute hetero-lr loss for:
        loss = (1/N)*∑(log2 - 1/2*ywx + 1/8*(wx)^2), where y is label, w is model weight and x is features
        where (wx)^2 = (Wg * Xg + Wh * Xh)^2 = (Wg*Xg)^2 + (Wh*Xh)^2 + 2 * Wg*Xg * Wh*Xh

        Then loss = log2 - (1/N)*0.5*∑ywx + (1/N)*0.125*[∑(Wg*Xg)^2 + ∑(Wh*Xh)^2 + 2 * ∑(Wg*Xg * Wh*Xh)]

        where Wh*Xh is a table obtain from host and ∑(Wh*Xh)^2 is a sum number get from host.
Compute hetero-lr loss for:
        loss = (1/N)*∑(log2 - 1/2*ywx + 1/8*(wx)^2), where y is label, w is model weight and x is features
        where (wx)^2 = (Wg * Xg + Wh * Xh)^2 = (Wg*Xg)^2 + (Wh*Xh)^2 + 2 * Wg*Xg * Wh*Xh
Compute hetero - lr loss for : loss = ( 1 / N ) * ∑ ( log2 - 1 / 2 * ywx + 1 / 8 * ( wx ) ^2 ) where y is label w is model weight and x is features where ( wx ) ^2 = ( Wg * Xg + Wh * Xh ) ^2 = ( Wg * Xg ) ^2 + ( Wh * Xh ) ^2 + 2 * Wg * Xg * Wh * Xh
def compute_loss(self, data_instances, w, n_iter_, batch_index, loss_norm=None):
        """
        Compute hetero-lr loss for:
        loss = (1/N)*∑(log2 - 1/2*ywx + 1/8*(wx)^2), where y is label, w is model weight and x is features
        where (wx)^2 = (Wg * Xg + Wh * Xh)^2 = (Wg*Xg)^2 + (Wh*Xh)^2 + 2 * Wg*Xg * Wh*Xh

        Then loss = log2 - (1/N)*0.5*∑ywx + (1/N)*0.125*[∑(Wg*Xg)^2 + ∑(Wh*Xh)^2 + 2 * ∑(Wg*Xg * Wh*Xh)]

        where Wh*Xh is a table obtain from host and ∑(Wh*Xh)^2 is a sum number get from host.
        """
        current_suffix = (n_iter_, batch_index)
        n = data_instances.count()

        host_wx_y = self.host_forwards[0].join(data_instances, lambda x, y: (x, y.label))
        self_wx_y = self.half_d.join(data_instances, lambda x, y: (x, y.label))

        def _sum_ywx(wx_y):
            sum1, sum2 = 0, 0
            for _, (x, y) in wx_y:
                if y == 1:
                    sum1 += x
                else:
                    sum2 -= x
            return sum1 + sum2

        ywx = host_wx_y.applyPartitions(_sum_ywx).reduce(reduce_add) + \
              self_wx_y.applyPartitions(_sum_ywx).reduce(reduce_add)
        ywx = ywx * 4 + 2 * n

        # quarter_wx = self.host_forwards[0].join(self.half_d, lambda x, y: x + y)
        # ywx = quarter_wx.join(data_instances, lambda wx, d: wx * (4 * d.label) + 2).reduce(reduce_add)

        self_wx_square = data_instances.mapValues(
            lambda v: np.square(vec_dot(v.features, w.coef_) + w.intercept_)).reduce(reduce_add)
        half_wx = data_instances.mapValues(
            lambda v: vec_dot(v.features, w.coef_) + w.intercept_)

        loss_list = []
        wx_squares = self.get_host_loss_intermediate(suffix=current_suffix)

        if loss_norm is not None:
            host_loss_regular = self.get_host_loss_regular(suffix=current_suffix)
        else:
            host_loss_regular = []

        # for host_idx, host_forward in enumerate(self.host_forwards):
        if len(self.host_forwards) > 1:
            LOGGER.info("More than one host exist, loss is not available")
        else:
            host_forward = self.host_forwards[0]
            wx_square = wx_squares[0]
            wxg_wxh = half_wx.join(host_forward, lambda wxg, wxh: wxg * wxh).reduce(reduce_add)
            loss = np.log(2) - 0.5 * (1 / n) * ywx + 0.125 * (1 / n) * \
                   (self_wx_square + wx_square + 2 * wxg_wxh)
            if loss_norm is not None:
                loss += loss_norm
                loss += host_loss_regular[0]
            loss_list.append(loss)
        LOGGER.debug("In compute_loss, loss list are: {}".format(loss_list))
        self.sync_loss_info(loss_list, suffix=current_suffix)
def compute_loss ( self , data_instances , w , n_iter_ , batch_index , loss_norm = None ) : current_suffix = ( n_iter_ , batch_index ) n = data_instances . count ( ) host_wx_y = self . host_forwards [ 0 ] . join ( data_instances , lambda x , y : ( x , y . label ) ) self_wx_y = self . half_d . join ( data_instances , lambda x , y : ( x , y . label ) ) def _sum_ywx ( wx_y ) : sum1 , sum2 = 0 , 0 for _ , ( x , y ) in wx_y : if y == 1 : sum1 += x else : sum2 -= x return sum1 + sum2 ywx = host_wx_y . applyPartitions ( _sum_ywx ) . reduce ( reduce_add ) + self_wx_y . applyPartitions ( _sum_ywx ) . reduce ( reduce_add ) ywx = ywx * 4 + 2 * n # quarter_wx = self.host_forwards[0].join(self.half_d, lambda x, y: x + y) # ywx = quarter_wx.join(data_instances, lambda wx, d: wx * (4 * d.label) + 2).reduce(reduce_add) self_wx_square = data_instances . mapValues ( lambda v : np . square ( vec_dot ( v . features , w . coef_ ) + w . intercept_ ) ) . reduce ( reduce_add ) half_wx = data_instances . mapValues ( lambda v : vec_dot ( v . features , w . coef_ ) + w . intercept_ ) loss_list = [ ] wx_squares = self . get_host_loss_intermediate ( suffix = current_suffix ) if loss_norm is not None : host_loss_regular = self . get_host_loss_regular ( suffix = current_suffix ) else : host_loss_regular = [ ] # for host_idx, host_forward in enumerate(self.host_forwards): if len ( self . host_forwards ) > 1 : LOGGER . info ( "More than one host exist, loss is not available" ) else : host_forward = self . host_forwards [ 0 ] wx_square = wx_squares [ 0 ] wxg_wxh = half_wx . join ( host_forward , lambda wxg , wxh : wxg * wxh ) . reduce ( reduce_add ) loss = np . log ( 2 ) - 0.5 * ( 1 / n ) * ywx + 0.125 * ( 1 / n ) * ( self_wx_square + wx_square + 2 * wxg_wxh ) if loss_norm is not None : loss += loss_norm loss += host_loss_regular [ 0 ] loss_list . append ( loss ) LOGGER . debug ( "In compute_loss, loss list are: {}" . format ( loss_list ) ) self . sync_loss_info ( loss_list , suffix = current_suffix )