(self, stack_types, n_blocks,
n_time_in, n_time_out,
n_x, n_x_hidden, n_s, n_s_hidden,
n_layers, n_mlp_units,
n_pool_kernel_size, n_freq_downsample, pooling_mode, interpolation_mode,
batch_normalization, dropout_prob_theta,
activation, shared_weights, initialization)
| 434 | self.blocks = t.nn.ModuleList(blocks) |
| 435 | |
| 436 | def create_stack(self, stack_types, n_blocks, |
| 437 | n_time_in, n_time_out, |
| 438 | n_x, n_x_hidden, n_s, n_s_hidden, |
| 439 | n_layers, n_mlp_units, |
| 440 | n_pool_kernel_size, n_freq_downsample, pooling_mode, interpolation_mode, |
| 441 | batch_normalization, dropout_prob_theta, |
| 442 | activation, shared_weights, initialization): |
| 443 | block_list = [] |
| 444 | for i in range(len(stack_types)): |
| 445 | assert stack_types[i] in ['identity', 'exogenous', 'exogenous_tcn', 'exogenous_wavenet'], 'f Invalid stack type {stack_types[i]}' |
| 446 | for block_id in range(n_blocks[i]): |
| 447 | |
| 448 | # Batch norm only on first block |
| 449 | if (len(block_list)==0) and (batch_normalization): |
| 450 | batch_normalization_block = True |
| 451 | else: |
| 452 | batch_normalization_block = False |
| 453 | |
| 454 | # Shared weights |
| 455 | if shared_weights and block_id>0: |
| 456 | nbeats_block = block_list[-1] |
| 457 | else: |
| 458 | if stack_types[i] == 'identity': |
| 459 | n_theta = (self.n_time_in + max(self.n_time_out//n_freq_downsample[i], 1) ) |
| 460 | basis = _IdentityBasis(backcast_size=n_time_in, |
| 461 | forecast_size=n_time_out, |
| 462 | interpolation_mode=interpolation_mode) |
| 463 | |
| 464 | elif stack_types[i] == 'exogenous': |
| 465 | n_theta = 2 * n_x |
| 466 | basis = _ExogenousBasisInterpretable() |
| 467 | |
| 468 | elif stack_types[i] == 'exogenous_tcn': |
| 469 | n_theta = 2 * n_x_hidden |
| 470 | basis = _ExogenousBasisTCN(n_x_hidden, n_x) |
| 471 | |
| 472 | elif stack_types[i] == 'exogenous_wavenet': |
| 473 | n_theta = 2 * n_x_hidden |
| 474 | basis = _ExogenousBasisWavenet(n_x_hidden, n_x) |
| 475 | |
| 476 | nbeats_block = _NHITSBlock(n_time_in=self.n_time_in, |
| 477 | n_time_out=self.n_time_out, |
| 478 | n_x=n_x, |
| 479 | n_s=n_s, |
| 480 | n_s_hidden=n_s_hidden, |
| 481 | n_theta=n_theta, |
| 482 | n_mlp_units=n_mlp_units[i], |
| 483 | n_pool_kernel_size=n_pool_kernel_size[i], |
| 484 | pooling_mode=pooling_mode, |
| 485 | basis=basis, |
| 486 | n_layers=n_layers[i], |
| 487 | batch_normalization=batch_normalization_block, |
| 488 | dropout_prob=dropout_prob_theta, |
| 489 | activation=activation) |
| 490 | |
| 491 | # Select type of evaluation and apply it to all layers of block |
| 492 | init_function = partial(_init_weights, initialization=initialization) |
| 493 | nbeats_block.layers.apply(init_function) |
no test coverage detected