test_consistency.py

# coding: utf-8
from pathlib import Path

import numpy as np
from sklearn.datasets import load_svmlight_file

import lightgbm as lgb

EXAMPLES_DIR = Path(__file__).absolute().parents[2] / 'examples'


class FileLoader:

    def __init__(self, directory, prefix, config_file='train.conf'):
        self.directory = directory
        self.prefix = prefix
        self.params = {'gpu_use_dp': True}
        with open(self.directory / config_file, 'r') as f:
            for line in f.readlines():
                line = line.strip()
                if line and not line.startswith('#'):
                    key, value = [token.strip() for token in line.split('=')]
                    if 'early_stopping' not in key:  # disable early_stopping
                        self.params[key] = value if key != 'num_trees' else int(value)

    def load_dataset(self, suffix, is_sparse=False):
        filename = str(self.path(suffix))
        if is_sparse:
            X, Y = load_svmlight_file(filename, dtype=np.float64, zero_based=True)
            return X, Y, filename
        else:
            mat = np.loadtxt(filename, dtype=np.float64)
            return mat[:, 1:], mat[:, 0], filename

    def load_field(self, suffix):
        return np.loadtxt(str(self.directory / f'{self.prefix}{suffix}'))

    def load_cpp_result(self, result_file='LightGBM_predict_result.txt'):
        return np.loadtxt(str(self.directory / result_file))

    def train_predict_check(self, lgb_train, X_test, X_test_fn, sk_pred):
        params = dict(self.params)
        params['force_row_wise'] = True
        gbm = lgb.train(params, lgb_train)
        y_pred = gbm.predict(X_test)
        cpp_pred = gbm.predict(X_test_fn)
        np.testing.assert_allclose(y_pred, cpp_pred)
        np.testing.assert_allclose(y_pred, sk_pred)

    def file_load_check(self, lgb_train, name):
        lgb_train_f = lgb.Dataset(self.path(name), params=self.params).construct()
        for f in ('num_data', 'num_feature', 'get_label', 'get_weight', 'get_init_score', 'get_group'):
            a = getattr(lgb_train, f)()
            b = getattr(lgb_train_f, f)()
            if a is None and b is None:
                pass
            elif a is None:
                assert np.all(b == 1), f
            elif isinstance(b, (list, np.ndarray)):
                np.testing.assert_allclose(a, b)
            else:
                assert a == b, f

    def path(self, suffix):
        return self.directory / f'{self.prefix}{suffix}'


def test_binary():
    fd = FileLoader(EXAMPLES_DIR / 'binary_classification', 'binary')
    X_train, y_train, _ = fd.load_dataset('.train')
    X_test, _, X_test_fn = fd.load_dataset('.test')
    weight_train = fd.load_field('.train.weight')
    lgb_train = lgb.Dataset(X_train, y_train, params=fd.params, weight=weight_train)
    gbm = lgb.LGBMClassifier(**fd.params)
    gbm.fit(X_train, y_train, sample_weight=weight_train)
    sk_pred = gbm.predict_proba(X_test)[:, 1]
    fd.train_predict_check(lgb_train, X_test, X_test_fn, sk_pred)
    fd.file_load_check(lgb_train, '.train')


def test_binary_linear():
    fd = FileLoader(EXAMPLES_DIR / 'binary_classification', 'binary', 'train_linear.conf')
    X_train, y_train, _ = fd.load_dataset('.train')
    X_test, _, X_test_fn = fd.load_dataset('.test')
    weight_train = fd.load_field('.train.weight')
    lgb_train = lgb.Dataset(X_train, y_train, params=fd.params, weight=weight_train)
    gbm = lgb.LGBMClassifier(**fd.params)
    gbm.fit(X_train, y_train, sample_weight=weight_train)
    sk_pred = gbm.predict_proba(X_test)[:, 1]
    fd.train_predict_check(lgb_train, X_test, X_test_fn, sk_pred)
    fd.file_load_check(lgb_train, '.train')


def test_multiclass():
    fd = FileLoader(EXAMPLES_DIR / 'multiclass_classification', 'multiclass')
    X_train, y_train, _ = fd.load_dataset('.train')
    X_test, _, X_test_fn = fd.load_dataset('.test')
    lgb_train = lgb.Dataset(X_train, y_train)
    gbm = lgb.LGBMClassifier(**fd.params)
    gbm.fit(X_train, y_train)
    sk_pred = gbm.predict_proba(X_test)
    fd.train_predict_check(lgb_train, X_test, X_test_fn, sk_pred)
    fd.file_load_check(lgb_train, '.train')


def test_regression():
    fd = FileLoader(EXAMPLES_DIR / 'regression', 'regression')
    X_train, y_train, _ = fd.load_dataset('.train')
    X_test, _, X_test_fn = fd.load_dataset('.test')
    init_score_train = fd.load_field('.train.init')
    lgb_train = lgb.Dataset(X_train, y_train, init_score=init_score_train)
    gbm = lgb.LGBMRegressor(**fd.params)
    gbm.fit(X_train, y_train, init_score=init_score_train)
    sk_pred = gbm.predict(X_test)
    fd.train_predict_check(lgb_train, X_test, X_test_fn, sk_pred)
    fd.file_load_check(lgb_train, '.train')


def test_lambdarank():
    fd = FileLoader(EXAMPLES_DIR / 'lambdarank', 'rank')
    X_train, y_train, _ = fd.load_dataset('.train', is_sparse=True)
    X_test, _, X_test_fn = fd.load_dataset('.test', is_sparse=True)
    group_train = fd.load_field('.train.query')
    lgb_train = lgb.Dataset(X_train, y_train, group=group_train)
    params = dict(fd.params)
    params['force_col_wise'] = True
    gbm = lgb.LGBMRanker(**params)
    gbm.fit(X_train, y_train, group=group_train)
    sk_pred = gbm.predict(X_test)
    fd.train_predict_check(lgb_train, X_test, X_test_fn, sk_pred)
    fd.file_load_check(lgb_train, '.train')


def test_xendcg():
    fd = FileLoader(EXAMPLES_DIR / 'xendcg', 'rank')
    X_train, y_train, _ = fd.load_dataset('.train', is_sparse=True)
    X_test, _, X_test_fn = fd.load_dataset('.test', is_sparse=True)
    group_train = fd.load_field('.train.query')
    lgb_train = lgb.Dataset(X_train, y_train, group=group_train)
    gbm = lgb.LGBMRanker(**fd.params)
    gbm.fit(X_train, y_train, group=group_train)
    sk_pred = gbm.predict(X_test)
    fd.train_predict_check(lgb_train, X_test, X_test_fn, sk_pred)
    fd.file_load_check(lgb_train, '.train')