Skip to content

动态联合

该文件是TPOT库的一部分。

当前版本的TPOT是由以下人员在Cedars-Sinai开发的： - Pedro Henrique Ribeiro (https://github.com/perib, https://www.linkedin.com/in/pedro-ribeiro/) - Anil Saini (anil.saini@cshs.org) - Jose Hernandez (jgh9094@gmail.com) - Jay Moran (jay.moran@cshs.org) - Nicholas Matsumoto (nicholas.matsumoto@cshs.org) - Hyunjun Choi (hyunjun.choi@cshs.org) - Miguel E. Hernandez (miguel.e.hernandez@cshs.org) - Jason Moore (moorejh28@gmail.com)

TPOT的原始版本主要由宾夕法尼亚大学的以下人员开发： - Randal S. Olson (rso@randalolson.com) - Weixuan Fu (weixuanf@upenn.edu) - Daniel Angell (dpa34@drexel.edu) - Jason Moore (moorejh28@gmail.com) - 以及许多慷慨的开源贡献者

TPOT 是免费软件：您可以根据自由软件基金会发布的 GNU 宽通用公共许可证的条款重新分发和/或修改它，无论是许可证的第 3 版，还是（根据您的选择）任何以后的版本。

TPOT 的发布是希望它能够有用， 但没有任何保证；甚至没有对适销性或特定用途适用性的暗示保证。更多详情请参阅 GNU 较宽松通用公共许可证。

您应该已经收到了一份GNU较宽松通用公共许可证的副本，随TPOT一起提供。如果没有，请参见http://www.gnu.org/licenses/。

DynamicUnionPipeline

基类：SearchSpace

Source code in tpot2/search_spaces/pipelines/dynamicunion.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
class DynamicUnionPipeline(SearchSpace):
    def __init__(self, search_space : SearchSpace, max_estimators=None, allow_repeats=False ) -> None:
        """
        Takes in a list of search spaces. will produce a pipeline of Sequential length. Each step in the pipeline will correspond to the the search space provided in the same index.
        """

        self.search_space = search_space
        self.max_estimators = max_estimators
        self.allow_repeats = allow_repeats

    def generate(self, rng=None):
        rng = np.random.default_rng(rng)
        return DynamicUnionPipelineIndividual(self.search_space, max_estimators=self.max_estimators, allow_repeats=self.allow_repeats, rng=rng)

__init__(search_space, max_estimators=None, allow_repeats=False)

接收一个搜索空间列表。将生成一个顺序长度的管道。管道中的每一步将对应于相同索引中提供的搜索空间。

Source code in tpot2/search_spaces/pipelines/dynamicunion.py 
187
188
189
190
191
192
193
194
def __init__(self, search_space : SearchSpace, max_estimators=None, allow_repeats=False ) -> None:
    """
    Takes in a list of search spaces. will produce a pipeline of Sequential length. Each step in the pipeline will correspond to the the search space provided in the same index.
    """

    self.search_space = search_space
    self.max_estimators = max_estimators
    self.allow_repeats = allow_repeats

DynamicUnionPipelineIndividual

基类：SklearnIndividual

接受一个搜索空间。 将生成一个最多包含 max_estimators 数量的步骤的 FeatureUnion。 FeatureUnion 的输出将是所有步骤连接在一起的结果。

Source code in tpot2/search_spaces/pipelines/dynamicunion.py 
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
class DynamicUnionPipelineIndividual(SklearnIndividual):
    """
    Takes in one search space.
    Will produce a FeatureUnion of up to max_estimators number of steps.
    The output of the FeatureUnion will the all of the steps concatenated together.

    """

    def __init__(self, search_space : SearchSpace, max_estimators=None, allow_repeats=False, rng=None) -> None:
        super().__init__()
        self.search_space = search_space

        if max_estimators is None:
            self.max_estimators = np.inf
        else:
            self.max_estimators = max_estimators

        self.allow_repeats = allow_repeats

        self.union_dict = {}

        if self.max_estimators == np.inf:
            init_max = 3
        else:
            init_max = self.max_estimators

        rng = np.random.default_rng(rng)

        for _ in range(rng.integers(1, init_max)):
            self._mutate_add_step(rng)


    def mutate(self, rng=None):
        rng = np.random.default_rng(rng)
        mutation_funcs = [self._mutate_add_step, self._mutate_remove_step, self._mutate_replace_step, self._mutate_note]
        rng.shuffle(mutation_funcs)
        for mutation_func in mutation_funcs:
            if mutation_func(rng):
                return True

    def _mutate_add_step(self, rng):
        rng = np.random.default_rng(rng)
        max_attempts = 10
        if len(self.union_dict) < self.max_estimators:
            for _ in range(max_attempts):
                new_step = self.search_space.generate(rng)
                if new_step.unique_id() not in self.union_dict:
                    self.union_dict[new_step.unique_id()] = new_step
                    return True
        return False

    def _mutate_remove_step(self, rng):
        rng = np.random.default_rng(rng)
        if len(self.union_dict) > 1:
            self.union_dict.pop( rng.choice(list(self.union_dict.keys())))  
            return True
        return False

    def _mutate_replace_step(self, rng):
        rng = np.random.default_rng(rng)        
        changed = self._mutate_remove_step(rng) or self._mutate_add_step(rng)
        return changed

    #TODO mutate one step or multiple?
    def _mutate_note(self, rng):
        rng = np.random.default_rng(rng)
        changed = False
        values = list(self.union_dict.values())
        for step in values:
            if rng.random() < 0.5:
                changed = step.mutate(rng) or changed

        self.union_dict = {step.unique_id(): step for step in values}

        return changed


    def crossover(self, other, rng=None):
        rng = np.random.default_rng(rng)

        cx_funcs = [self._crossover_swap_multiple_nodes, self._crossover_node]
        rng.shuffle(cx_funcs)
        for cx_func in cx_funcs:
            if cx_func(other, rng):
                return True

        return False


    def _crossover_swap_multiple_nodes(self, other, rng):
        rng = np.random.default_rng(rng)
        self_values = list(self.union_dict.values())
        other_values = list(other.union_dict.values())

        rng.shuffle(self_values)
        rng.shuffle(other_values)

        self_idx = rng.integers(0,len(self_values))
        other_idx = rng.integers(0,len(other_values))

        #Note that this is not one-point-crossover since the sequence doesn't matter. this is just a quick way to swap multiple random items
        self_values[:self_idx], other_values[:other_idx] = other_values[:other_idx], self_values[:self_idx]

        self.union_dict = {step.unique_id(): step for step in self_values}
        other.union_dict = {step.unique_id(): step for step in other_values}

        return True


    def _crossover_node(self, other, rng):
        rng = np.random.default_rng(rng)

        changed = False
        self_values = list(self.union_dict.values())
        other_values = list(other.union_dict.values())

        rng.shuffle(self_values)
        rng.shuffle(other_values)

        for self_step, other_step in zip(self_values, other_values):
            if rng.random() < 0.5:
                changed = self_step.crossover(other_step, rng) or changed

        self.union_dict = {step.unique_id(): step for step in self_values}
        other.union_dict = {step.unique_id(): step for step in other_values}

        return changed

    def export_pipeline(self, **kwargs):
        values = list(self.union_dict.values())
        return sklearn.pipeline.make_union(*[step.export_pipeline(**kwargs) for step in values])

    def unique_id(self):
        values = list(self.union_dict.values())
        l = [step.unique_id() for step in values]
        # if all items are strings, then sort them
        if all([isinstance(x, str) for x in l]):
            l.sort()
        l = ["FeatureUnion"] + l
        return TupleIndex(frozenset(l))