tests/test_iworkflow/optimize_iworkflow.py

0001 #!/usr/bin/env python
0002 #
0003 # Licensed under the Apache License, Version 2.0 (the "License");
0004 # You may not use this file except in compliance with the License.
0005 # You may obtain a copy of the License at
0006 # http://www.apache.org/licenses/LICENSE-2.0OA
0007 #
0008 # Authors:
0009 # - Wen Guan, <wen.guan@cern.ch>, 2024
0010
0011
0012 """
0013 Test workflow.
0014 """
0015
0016 import inspect     # noqa F401
0017 import logging
0018 import os          # noqa F401
0019 import shutil      # noqa F401
0020 import sys         # noqa F401
0021
0022 # from nose.tools import assert_equal
0023 # from idds.common.imports import get_func_name
0024 from idds.common.utils import setup_logging, get_unique_id_for_dict
0025
0026 from idds.iworkflow.workflow import Workflow       # workflow
0027 from idds.iworkflow.work import work
0028
0029
0030 setup_logging(__name__)
0031
0032
0033 def get_initial_parameter():
0034     opt_params = {
0035         'colsample_bytree': (0.1, 1),
0036         'scale_pos_weight': (0, 10),
0037         'max_delta_step': (0, 10),
0038         'seed': (1, 50),
0039         'min_child_weight': (0, 100),
0040         'subsample': (0.1, 1),
0041         'eta': (0, 0.1),
0042         'alpha': (0, 1),
0043         # 'lambda': (0, 100),
0044         'max_depth': (0, 50),
0045         'gamma': (0, 1),
0046         # 'num_boost_round': (100000, 1000000)
0047     }
0048     return opt_params
0049
0050
0051 @work(map_results=True)
0052 def optimize_work(opt_params, retMethod=None, hist=True, saveModel=False, input_weight=None, **kwargs):
0053     from optimize import evaluate_bdt, load_data
0054
0055     data, label = load_data()
0056     train, val = data
0057     y_train_cat, y_val_cat = label
0058     input_x = [train, val]
0059     input_y = [y_train_cat, y_val_cat]
0060
0061     ret = evaluate_bdt(input_x=input_x, input_y=input_y, opt_params=opt_params, retMethod=retMethod, hist=hist,
0062                        saveModel=saveModel, input_weight=input_weight, **kwargs)
0063     return ret
0064
0065
0066 def optimize_workflow():
0067     from optimize import evaluate_bdt, get_bayesian_optimizer_and_util
0068
0069     opt_method = 'auc'
0070     params = {'num_boost_round': 1000}
0071
0072     opt_params = get_initial_parameter()
0073
0074     print("To optimize with parameters: %s" % opt_params)
0075
0076     optFunc = lambda **z: evaluate_bdt(input_x=None, input_y=None,
0077                                        opt_params=params, opt_method=opt_method,
0078                                        hist=True, saveModel=False, input_weight=None, **z)   # noqa F731
0079     bayesopt, util = get_bayesian_optimizer_and_util(optFunc, opt_params)
0080
0081     n_iterations, n_points_per_iteration = 10, 20
0082     for i in range(n_iterations):
0083         print("Iteration %s" % i)
0084         points = {}
0085         multi_jobs_kwargs_list = []
0086         for j in range(n_points_per_iteration):
0087             x_probe = bayesopt.suggest(util)
0088             u_id = get_unique_id_for_dict(x_probe)
0089             print('x_probe (%s): %s' % (u_id, x_probe))
0090             points[u_id] = {'kwargs': x_probe}
0091             multi_jobs_kwargs_list.append(x_probe)
0092
0093         results = optimize_work(opt_params=params, opt_method=opt_method, hist=True, saveModel=False, input_weight=None,
0094                                 retMethod=opt_method, multi_jobs_kwargs_list=multi_jobs_kwargs_list)
0095         print("points: %s" % str(points))
0096
0097         for u_id in points:
0098             points[u_id]['ret'] = results.get_result(name=None, args=points[u_id]['kwargs'])
0099             print('ret :%s, kwargs: %s' % (points[u_id]['ret'], points[u_id]['kwargs']))
0100             bayesopt.register(points[u_id]['kwargs'], points[u_id]['ret'])
0101
0102     print(bayesopt.res)
0103     p = bayesopt.max
0104     print("Best params: %s" % p)
0105     return p
0106
0107
0108 if __name__ == '__main__':
0109     logging.info("start")
0110     os.chdir(os.path.dirname(os.path.realpath(__file__)))
0111     # wf = Workflow(func=test_workflow, service='idds', distributed=False)
0112     # init_env = 'singularity exec /eos/user/w/wguan/idds_ml/singularity/idds_ml_al9.simg '
0113     init_env = 'singularity exec /afs/cern.ch/user/w/wguan/workdisk/iDDS/test/eic/idds_ml_al9.simg '
0114     wf = Workflow(func=optimize_workflow, service='idds', init_env=init_env)
0115
0116     # wf.queue = 'BNL_OSG_2'
0117     wf.queue = 'FUNCX_TEST'
0118     wf.cloud = 'US'
0119
0120     logging.info("prepare workflow")
0121     wf.prepare()
0122     logging.info("prepared workflow")
0123
0124     wf.submit()