def _make_append_bias()

in core/src/main/python/synapse/ml/cyber/anomaly/collaborative_filtering.py [0:0]

• 26 lines of code
• 21 McCabe index (conditional complexity)

    def _make_append_bias(self, user_col: str, res_col: str, col_name: str, target_col_name: str, rank: int) -> f.udf:
        assert col_name == user_col or col_name == res_col
        assert target_col_name == user_col or target_col_name == res_col

        def value_at(bias: float, value: float, i: int) -> float:
            if col_name != target_col_name or i < rank:
                res = value
            elif col_name == user_col and i == rank:
                res = bias + value
            elif col_name == res_col and i == (rank + 1):
                res = bias + value
            else:
                res = value

            assert res == 1.0 if i == rank and col_name == res_col else True
            assert res == 1.0 if i == (rank + 1) and col_name == user_col else True

            return res

        @f.udf(t.ArrayType(t.DoubleType()))
        def append_bias(v: List[float], bias: float, coeff: float = 1.0) -> List[float]:
            assert len(v) == rank or len(v) == rank + 2

            if len(v) == rank:
                # increase vector size to adjust for bias
                fix_value = bias if col_name == target_col_name else 0.0
                u = [fix_value, 1.0] if col_name == user_col else [1.0, fix_value]
                return [float(coeff * value) for value in np.append(np.array(v), np.array(u))]
            else:
                # fix enhanced vector to adjust for another bias
                assert len(v) == rank + 2
                return [coeff * value_at(bias, v[i], i) for i in range(len(v))]

        return append_bias