This page contains the auto-generated API reference documentation.

Preprocessing Config

class sequifier.config.preprocess_config.PreprocessorModel(*, project_root: str, preprocessing_data_path: str, read_format: str = 'csv', write_format: str = 'parquet', merge_output: bool = True, allow_sequence_splitting: bool = False, selected_columns: list[str] | None = None, column_data_types: dict[str, str] | None = None, normalize_real_columns: bool = True, split_ratios: list[float], split_method: str = 'within_sequence', stored_context_width: int, max_target_offset: int = 1, stride_by_split: list[int] | None = None, max_rows: int | None = None, seed: int = 1010, n_cores: int | None = None, batches_per_file: int = 1024, process_by_file: bool = True, continue_preprocessing: bool = False, subsequence_start_mode: str = 'distribute', use_precomputed_maps: list[str] | None = None, metadata_config_path: str | None = None, mask_column: str | None = None)

Top-level preprocessing config.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Training Config

class sequifier.config.train_config.LoadedTrainConfig(config: SequifierConfig, resolved: ResolvedSequifierConfig, metadata: DatasetMetadata)

A validated authored config together with its resolved runtime config.

property metadata_values: dict[str, Any]

Compatibility view of validated preprocessing metadata.

property model: ResolvedSequifierConfig

Compatibility alias for callers that previously consumed model.

property source_values: dict[str, Any]

Compatibility view used by the legacy partial-search compiler.

class sequifier.config.train_config.ModelSpecModel(*, ingestion_spec: ~sequifier.config.train_config.DirectEmbedIngestionConfig | ~sequifier.config.train_config.PassThroughIngestionConfig | ~sequifier.config.train_config.FeaturePoolIngestionConfig | ~sequifier.config.train_config.GroupedIngestionConfig | ~sequifier.config.train_config.SiameseIngestionConfig | ~sequifier.config.train_config.TemporalConvIngestionConfig | ~sequifier.config.train_config.StructuredIngestionConfig | dict[str, typing.Annotated[typing.Union[sequifier.config.train_config.DirectEmbedIngestionConfig, sequifier.config.train_config.PassThroughIngestionConfig, sequifier.config.train_config.FeaturePoolIngestionConfig, sequifier.config.train_config.GroupedIngestionConfig, sequifier.config.train_config.SiameseIngestionConfig, sequifier.config.train_config.TemporalConvIngestionConfig, sequifier.config.train_config.StructuredIngestionConfig], FieldInfo(annotation=NoneType, required=True, discriminator='type')]] | None = None, ingestion_merge: ~sequifier.config.train_config.IngestionMergeConfig | None = None, allow_shared_ingestion_columns: bool = False, allow_unused_input_columns: bool = False, auxiliary_input_columns: list[str] = <factory>, initialization: ~sequifier.config.initialization_config.ModelInitializationConfig = <factory>, dim_model: int, n_head: int, dim_feedforward: int, num_layers: int, activation_fn: str = 'swiglu', normalization: str = 'rmsnorm', positional_encoding: str = 'learned', positional_encoding_scope: str = 'per_feature', attention_type: str = 'mha', attention_output_projection: bool = True, norm_first: bool = True, shared_layer_groups: list[list[int]] = <factory>, n_kv_heads: int | None = None, rope_theta: float = 10000.0, prediction_length: int, decoding_support: int = 1, decoding_spec: ~sequifier.config.train_config.LinearDecodingConfig | ~sequifier.config.train_config.MLPDecodingConfig | dict[str, typing.Annotated[typing.Union[sequifier.config.train_config.LinearDecodingConfig, sequifier.config.train_config.MLPDecodingConfig], FieldInfo(annotation=NoneType, required=True, discriminator='type')]] | None = None)

Transformer architecture settings.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sequifier.config.train_config.ResolvedSequifierConfig(*, project_root: str, preprocessing_data_path: str | None = None, metadata_config_path: str, model_name: str, training_objective: str, device: str, data_path: str, validation_data_path: str, read_format: str = 'parquet', input_columns: list[str], column_data_types: dict[str, str], target_columns: list[str], target_column_types: dict[str, str], categorical_decoder_special_tokens: dict[str, list[typing.Literal['unknown', 'other', 'mask']]] = <factory>, context_length: int, target_offset: int = 1, model_window_stride: int | None = None, inference_batch_size: int, seed: int = 1010, export_generative_model: bool, export_embedding_model: bool, export_onnx: bool = True, export_pt: bool = False, export_with_dropout: bool = False, feature_layout: ~sequifier.config.train_config.FeatureLayoutRegistryModel | None = None, model_spec: ~sequifier.config.train_config.ModelSpecModel, training_spec: ~sequifier.config.train_config.TrainingSpecModel, categorical_columns: list[str], real_columns: list[str], id_maps: dict[str, dict[str | int, int]], special_token_ids: dict[str, int] = <factory>, storage_layout: ~sequifier.helpers.StoredWindowLayout, window_view: ~sequifier.helpers.ModelWindowView, n_classes: dict[str, int])

Internal training config after preprocessing metadata has been resolved.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sequifier.config.train_config.SequifierConfig(*, project_root: str, preprocessing_data_path: str | None = None, metadata_config_path: str | None = None, model_name: str, training_objective: str, device: str, data_path: str | None = None, validation_data_path: str | None = None, read_format: str = 'parquet', input_columns: list[str] | None, column_data_types: dict[str, str] | None = None, target_columns: list[str], target_column_types: dict[str, str] | None = None, categorical_decoder_special_tokens: dict[str, list[typing.Literal['unknown', 'other', 'mask']]] = <factory>, context_length: int, target_offset: int = 1, model_window_stride: int | None = None, inference_batch_size: int, seed: int = 1010, export_generative_model: bool, export_embedding_model: bool, export_onnx: bool = True, export_pt: bool = False, export_with_dropout: bool = False, feature_layout: ~sequifier.config.train_config.FeatureLayoutRegistryModel | None = None, model_spec: ~sequifier.config.train_config.ModelSpecModel, training_spec: ~sequifier.config.train_config.TrainingSpecModel)

User-authored configuration for one concrete training run.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

sequifier.config.train_config.TrainModel

alias of ResolvedSequifierConfig

class sequifier.config.train_config.TrainingSpecModel(*, device_max_concat_length: int = 12, epochs: int, log_interval: int = 10, class_share_log_columns: list[str] = <factory>, early_stopping_epochs: int | None = None, save_interval_epochs: int, save_latest_interval_minutes: float | None = None, save_interval_minutes: float | None = None, save_interval_batches: int | None = None, save_interval_val_loss: bool = True, calculate_validation_loss_on_initialization: bool = True, batch_size: int, learning_rate: float, criterion: dict[str, str], class_weights: dict[str, list[float]] | None = None, accumulation_steps: int | None = None, gradient_clip: float | None = None, dropout: float = 0.0, loss_weights: dict[str, float] | None = None, optimizer: ~sequifier.config.train_config.DotDict = <factory>, scheduler: ~sequifier.config.train_config.DotDict = <factory>, scheduler_step_on: str = 'epoch', bert_spec: ~sequifier.config.train_config.BERTSpecModel | None = None, next_occurrence_config: ~sequifier.config.train_config.NextOccurrenceConfigModel | None = None, continue_training: bool = True, enforce_determinism: bool = False, distributed: bool = False, load_full_data_to_ram: bool = True, max_ram_gb: int | float = 16, world_size: int = 1, num_workers: int = 0, backend: str = 'nccl', layer_type_dtypes: dict[str, str] | None = None, layer_autocast: bool | None = False, data_parallelism: str | None = None, fsdp_cpu_offload: bool | None = None, torch_compile: str = 'outer', float32_matmul_precision: str = 'highest')

Training loop, optimization, precision, and distribution settings.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

sequifier.config.train_config.load_train_config_with_source(config_path: str, args_config: dict[str, Any], skip_metadata: bool) → LoadedTrainConfig

Compose and validate authored YAML, then resolve preprocessing metadata.

sequifier.config.train_config.resolve_sequifier_config(config: SequifierConfig, metadata: DatasetMetadata) → ResolvedSequifierConfig

Resolve dataset metadata without mutating the validated authored config.

Inference Config

class sequifier.config.infer_config.InferenceConfig(*, project_root: str, preprocessing_data_path: str | None = None, metadata_config_path: str | None = None, model_path: str | list[str], model_type: str, training_objective: str, data_path: str | None = None, training_config_path: str | None = 'configs/train.yaml', read_format: str = 'parquet', write_format: str = 'csv', input_columns: list[str] | None, target_columns: list[str], column_data_types: dict[str, str] | None = None, target_column_types: dict[str, str] | None = None, enforce_deterministic_inference: bool = False, output_probabilities: bool = False, map_to_id: bool = True, seed: int = 1010, device: str, context_length: int, target_offset: int = 1, model_window_stride: int | None = None, prediction_length: int | None = None, inference_batch_size: int, sample_from_distribution_columns: list[str] | None = None, infer_with_dropout: bool = False, autoregression: bool = False, autoregression_total_steps: int | None = None)

User-authored configuration for one inference run.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

sequifier.config.infer_config.InfererModel

alias of ResolvedInferenceConfig

class sequifier.config.infer_config.ResolvedInferenceConfig(*, project_root: str, preprocessing_data_path: str | None = None, metadata_config_path: str, model_path: str | list[str], model_type: str, training_objective: str, data_path: str, training_config_path: str | None = 'configs/train.yaml', read_format: str = 'parquet', write_format: str = 'csv', input_columns: list[str], target_columns: list[str], column_data_types: dict[str, str], target_column_types: dict[str, str], enforce_deterministic_inference: bool = False, output_probabilities: bool = False, map_to_id: bool = True, seed: int = 1010, device: str, context_length: int, target_offset: int = 1, model_window_stride: int | None = None, prediction_length: int | None = None, inference_batch_size: int, sample_from_distribution_columns: list[str] | None = None, infer_with_dropout: bool = False, autoregression: bool = False, autoregression_total_steps: int | None = None, categorical_columns: list[str], real_columns: list[str], storage_layout: StoredWindowLayout, window_view: ModelWindowView, dataset_metadata: DatasetMetadata | None = None)

Internal inference config after dataset metadata has been resolved.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

sequifier.config.infer_config.resolve_inference_config(config: InferenceConfig, metadata: DatasetMetadata) → ResolvedInferenceConfig

Return an inference config with all metadata-derived values populated.

Config Composition and Metadata

Composition helpers for user-authored configuration fragments.

sequifier.config.composition.deep_merge_config(base: Mapping[str, Any], override: Mapping[str, Any], *, atomic_paths: frozenset[tuple[str, ...]] = frozenset({('model_spec', 'decoding_spec'), ('model_spec', 'ingestion_merge'), ('model_spec', 'ingestion_spec'), ('training_spec', 'bert_spec', 'span_masking'), ('training_spec', 'optimizer'), ('training_spec', 'scheduler')})) → dict[str, Any]

Return a deep merge of two authored mappings.

Dictionaries merge recursively. Lists and scalar values are replaced, and an explicit None clears the inherited value. Neither input is mutated.

sequifier.config.composition.load_composed_yaml_config(config_path: str, *, atomic_paths: frozenset[tuple[str, ...]] = frozenset({('model_spec', 'decoding_spec'), ('model_spec', 'ingestion_merge'), ('model_spec', 'ingestion_spec'), ('training_spec', 'bert_spec', 'span_masking'), ('training_spec', 'optimizer'), ('training_spec', 'scheduler')})) → dict[str, Any]

Load one YAML config and its direct, complementary fragments.

Relative fragment paths are resolved against the entry config’s project_root. Fragments cannot include further fragments. Duplicate authored fields are rejected before the resulting mapping reaches the command-specific Pydantic model.

sequifier.config.composition.merge_complementary_config_fragments(fragments: Iterable[tuple[str, collections.abc.Mapping[str, Any]]], *, atomic_paths: frozenset[tuple[str, ...]] = frozenset({('model_spec', 'decoding_spec'), ('model_spec', 'ingestion_merge'), ('model_spec', 'ingestion_spec'), ('training_spec', 'bert_spec', 'span_masking'), ('training_spec', 'optimizer'), ('training_spec', 'scheduler')})) → dict[str, Any]

Merge sourced fragments while rejecting duplicate authored fields.

sequifier.config.composition.merge_config_fragments(fragments: Iterable[Mapping[str, Any]], *, atomic_paths: frozenset[tuple[str, ...]] = frozenset({('model_spec', 'decoding_spec'), ('model_spec', 'ingestion_merge'), ('model_spec', 'ingestion_spec'), ('training_spec', 'bert_spec', 'span_masking'), ('training_spec', 'optimizer'), ('training_spec', 'scheduler')})) → dict[str, Any]

Merge authored fragments in order, with later fragments taking priority.

Typed preprocessing metadata used during config resolution.

class sequifier.config.metadata.DatasetMetadata(*, split_paths: list[str] = <factory>, column_data_types: dict[str, str] = <factory>, n_classes: dict[str, int] = <factory>, id_maps: dict[str, dict[str | int, int]] = <factory>, special_token_ids: dict[str, int] = <factory>, selected_columns_statistics: dict[str, dict[str, float]] = <factory>, normalize_real_columns: bool = True, stored_context_width: int, max_target_offset: int = 1, stored_window_layout_version: int = 2, **extra_data: ~typing.Any)

The stable subset of preprocessing metadata consumed by other commands.

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

sequifier.config.metadata.extract_inline_metadata(values: dict[str, Any]) → tuple[dict[str, Any], dict[str, Any] | None]

Split the historical skip_metadata representation into two mappings.

sequifier.config.metadata.load_dataset_metadata(path: str) → DatasetMetadata

Load and validate one preprocessing metadata JSON file.

Hyperparameter Search Config

class sequifier.config.hyperparameter_search_config.HyperparameterSearchConfig(*, project_root: str, metadata_config_path: str, hp_search_name: str, search_strategy: str = 'bayesian', global_seed: int | None = None, seed: list[int] | None = None, n_samples: int | None = None, prune_trials: bool | None = True, pruning_warmup_epochs: int | None = None, pruning_warmup_batches: int | None = None, model_config_write_path: str, data_path: str, validation_data_path: str, read_format: str = 'parquet', input_columns: list[list[str]], column_data_types: list[dict[str, str]], categorical_columns: list[list[str]], real_columns: list[list[str]], target_columns: list[str], target_column_types: dict[str, str], id_maps: dict[str, dict[str | int, int]], special_token_ids: dict[str, int] = <factory>, categorical_decoder_special_tokens: dict[str, list[str]] = <factory>, context_length: list[int], target_offset: int = 1, storage_layout: ~sequifier.helpers.StoredWindowLayout, model_window_stride: int | None = None, n_classes: dict[str, int], inference_batch_size: int, export_generative_model: bool, export_embedding_model: bool, export_onnx: bool = True, export_pt: bool = False, export_with_dropout: bool = False, feature_layout: ~sequifier.config.train_config.FeatureLayoutRegistryModel | None = None, evaluation_inference_config: str | None = None, evaluation_script: str | None = None, evaluation_metric_directions: list[str] | None = None, evaluation_metrics: list[str] | None = None, model_hyperparameter_sampling: ~sequifier.config.hyperparameter_search_config.ModelSpecHyperparameterSampling, training_hyperparameter_sampling: ~sequifier.config.hyperparameter_search_config.TrainingSpecHyperparameterSampling, override_input: bool = False)

Top-level Optuna search config.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

sample_trial(trial: Any, run_index: int) → SequifierConfig

Sample and validate one concrete authored training config.

validate_grid_n_trials_matches_combinations()

Require an explicit grid trial limit to cover the complete grid.

class sequifier.config.hyperparameter_search_config.ModelSpecHyperparameterSampling(*, dim_model: list[int], ingestion_spec: ~sequifier.config.train_config.DirectEmbedIngestionConfig | ~sequifier.config.train_config.PassThroughIngestionConfig | ~sequifier.config.train_config.FeaturePoolIngestionConfig | ~sequifier.config.train_config.GroupedIngestionConfig | ~sequifier.config.train_config.SiameseIngestionConfig | ~sequifier.config.train_config.TemporalConvIngestionConfig | ~sequifier.config.train_config.StructuredIngestionConfig | dict[str, typing.Annotated[typing.Union[sequifier.config.train_config.DirectEmbedIngestionConfig, sequifier.config.train_config.PassThroughIngestionConfig, sequifier.config.train_config.FeaturePoolIngestionConfig, sequifier.config.train_config.GroupedIngestionConfig, sequifier.config.train_config.SiameseIngestionConfig, sequifier.config.train_config.TemporalConvIngestionConfig, sequifier.config.train_config.StructuredIngestionConfig], FieldInfo(annotation=NoneType, required=True, discriminator='type')]] | list[typing.Union[typing.Annotated[typing.Union[sequifier.config.train_config.DirectEmbedIngestionConfig, sequifier.config.train_config.PassThroughIngestionConfig, sequifier.config.train_config.FeaturePoolIngestionConfig, sequifier.config.train_config.GroupedIngestionConfig, sequifier.config.train_config.SiameseIngestionConfig, sequifier.config.train_config.TemporalConvIngestionConfig, sequifier.config.train_config.StructuredIngestionConfig], FieldInfo(annotation=NoneType, required=True, discriminator='type')], dict[str, typing.Annotated[typing.Union[sequifier.config.train_config.DirectEmbedIngestionConfig, sequifier.config.train_config.PassThroughIngestionConfig, sequifier.config.train_config.FeaturePoolIngestionConfig, sequifier.config.train_config.GroupedIngestionConfig, sequifier.config.train_config.SiameseIngestionConfig, sequifier.config.train_config.TemporalConvIngestionConfig, sequifier.config.train_config.StructuredIngestionConfig], FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]] | None = None, ingestion_merge: ~sequifier.config.train_config.IngestionMergeConfig | list[sequifier.config.train_config.IngestionMergeConfig] | None = None, allow_shared_ingestion_columns: bool = False, allow_unused_input_columns: bool = False, auxiliary_input_columns: list[str] = <factory>, initialization: ~sequifier.config.initialization_config.ModelInitializationSamplingConfig = <factory>, n_head: list[int], dim_feedforward: list[int] | ~sequifier.config.hyperparameter_search_config.IntDistribution, num_layers: list[int] | ~sequifier.config.hyperparameter_search_config.IntDistribution, prediction_length: int, decoding_support: int | list[int] | ~sequifier.config.hyperparameter_search_config.IntDistribution = 1, decoding_spec: ~sequifier.config.train_config.LinearDecodingConfig | ~sequifier.config.train_config.MLPDecodingConfig | dict[str, typing.Annotated[typing.Union[sequifier.config.train_config.LinearDecodingConfig, sequifier.config.train_config.MLPDecodingConfig], FieldInfo(annotation=NoneType, required=True, discriminator='type')]] | list[typing.Union[typing.Annotated[typing.Union[sequifier.config.train_config.LinearDecodingConfig, sequifier.config.train_config.MLPDecodingConfig], FieldInfo(annotation=NoneType, required=True, discriminator='type')], dict[str, typing.Annotated[typing.Union[sequifier.config.train_config.LinearDecodingConfig, sequifier.config.train_config.MLPDecodingConfig], FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]] | None = None, activation_fn: list[str], normalization: list[str], positional_encoding: list[str], positional_encoding_scope: list[str] = <factory>, attention_type: list[str], attention_output_projection: list[bool] = <factory>, norm_first: list[bool], shared_layer_groups: list[list[int]] = <factory>, n_kv_heads: list[typing.Optional[int]], rope_theta: list[float] | ~sequifier.config.hyperparameter_search_config.FloatDistribution)

Model-architecture search space with paired width choices.

grid_size() → int

Return the number of model-spec grid combinations.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

sample_trial(trial: Any) → ModelSpecModel

Sample architecture hyperparameters for one Optuna trial.

validation_model(*, width_index: int = 0, dim_feedforward: int | None = None, num_layers: int | None = None, decoding_support: int | None = None, decoding_spec_index: int = 0, activation_fn: str | None = None, normalization: str | None = None, positional_encoding: str | None = None, positional_encoding_scope: str | None = None, attention_type: str | None = None, attention_output_projection: bool | None = None, norm_first: bool | None = None, n_kv_heads: object = <object object>, rope_theta: float | None = None) → ModelSpecModel

Build a representative concrete model specification.

class sequifier.config.hyperparameter_search_config.TrainingSpecHyperparameterSampling(*, device: str, epochs: list[int], log_interval: int = 10, class_share_log_columns: list[str] = <factory>, early_stopping_epochs: int | None = None, save_interval_epochs: int, save_latest_interval_minutes: float | None = None, save_interval_minutes: float | None = None, save_interval_val_loss: bool = True, save_interval_batches: int | None = None, calculate_validation_loss_on_initialization: bool = False, training_objective: list[str] = <factory>, batch_size: list[int] | ~sequifier.config.hyperparameter_search_config.IntDistribution, learning_rate: list[float], bert_spec: ~sequifier.config.hyperparameter_search_config.BERTSpecHyperparameterSampling | None = None, next_occurrence_config: ~sequifier.config.train_config.NextOccurrenceConfigModel | None = None, criterion: dict[str, str], class_weights: dict[str, list[float]] | None = None, accumulation_steps: list[typing.Optional[int]] | ~sequifier.config.hyperparameter_search_config.IntDistribution, gradient_clip: float | list[typing.Optional[float]] | ~sequifier.config.hyperparameter_search_config.FloatDistribution | None = None, dropout: list[float] | ~sequifier.config.hyperparameter_search_config.FloatDistribution = [0.0], loss_weights: dict[str, float] | None = None, optimizer: list[sequifier.config.train_config.DotDict] = <factory>, scheduler: list[sequifier.config.train_config.DotDict] = <factory>, continue_training: bool, scheduler_step_on: str = 'epoch', distributed: bool = False, load_full_data_to_ram: bool = True, max_ram_gb: int | float = 16, device_max_concat_length: int = 12, world_size: int = 1, num_workers: int = 0, backend: str = 'nccl', layer_type_dtypes: dict[str, str] | None = None, layer_autocast: bool | None = False, data_parallelism: str | None = None, fsdp_cpu_offload: bool | None = None, torch_compile: str = 'outer', float32_matmul_precision: str = 'highest')

Training-spec search space with paired LR/scheduler candidates.

__init__(**kwargs)

Normalize optimizer/scheduler dicts before Pydantic validation.

grid_size() → int

Return the number of training-spec grid combinations.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

sample_trial(trial: Any) → SampledTrainingConfig

Sample training hyperparameters for one Optuna trial.

validation_model(*, schedule_index: int = 0, optimizer_index: int = 0, training_objective: str | None = None) → SampledTrainingConfig

Build a representative concrete training specification.

Non-standard Optimizers

class sequifier.optimizers.ademamix.AdEMAMix(params={}, lr=0.001, betas=(0.9, 0.999, 0.9999), eps=1e-08, weight_decay=0, alpha=5.0, T_alpha_beta3=None)

AdEMAMix optimizer.

step(closure=None)

Run one optimizer step.

Internals

sequifier.sequifier.build_args_config(args: Any) → dict[str, Any]

Build config overrides from parsed CLI args.

sequifier.sequifier.main() → None

Main function to run the Sequifier CLI.

sequifier.sequifier.setup_parser() → ArgumentParser

Build the sequifier CLI parser.

class sequifier.preprocess.Preprocessor(project_root: str, continue_preprocessing: bool, preprocessing_data_path: str, read_format: str, write_format: str, merge_output: bool, allow_sequence_splitting: bool, selected_columns: list[str] | None, split_ratios: list[float], stored_context_width: int, stride_by_split: list[int], max_rows: int | None, seed: int, n_cores: int | None, batches_per_file: int, process_by_file: bool, subsequence_start_mode: str, use_precomputed_maps: list[str] | None, metadata_config_path: str | None, max_target_offset: int = 1, mask_column: str | None = None, column_data_types: dict[str, str] | None = None, split_method: str = 'within_sequence', normalize_real_columns: bool = True)

Stateful preprocessing pipeline for single-file or folder inputs.

__init__(project_root: str, continue_preprocessing: bool, preprocessing_data_path: str, read_format: str, write_format: str, merge_output: bool, allow_sequence_splitting: bool, selected_columns: list[str] | None, split_ratios: list[float], stored_context_width: int, stride_by_split: list[int], max_rows: int | None, seed: int, n_cores: int | None, batches_per_file: int, process_by_file: bool, subsequence_start_mode: str, use_precomputed_maps: list[str] | None, metadata_config_path: str | None, max_target_offset: int = 1, mask_column: str | None = None, column_data_types: dict[str, str] | None = None, split_method: str = 'within_sequence', normalize_real_columns: bool = True)

Initialize and run preprocessing from validated config fields.

sequifier.preprocess.cast_columns_to_string(data: DataFrame) → DataFrame

Cast Polars column names to strings.

sequifier.preprocess.combine_maps(map1: dict[Union[str, int], int], map2: dict[Union[str, int], int]) → dict[Union[str, int], int]

Merge maps and reassign user IDs after reserved tokens.

sequifier.preprocess.combine_multiprocessing_outputs(project_root: str, target_dir: str, n_splits: int, input_files: dict[int, list[str]], dataset_name: str, write_format: str, in_target_dir: bool = False, pre_split_str: str | None = None, post_split_str: str | None = None) → None

Combine per-split intermediate files.

sequifier.preprocess.combine_parquet_files(files: list[str], out_path: str) → None

Stream-concatenate Parquet files with the first file schema.

sequifier.preprocess.create_file_paths_for_multiple_files1(project_root: str, target_dir: str, n_splits: int, n_batches: int, process_id: int, file_index_str: str, dataset_name: str, write_format: str) → dict[int, list[str]]

Return per-split temp paths for one multi-file shard.

sequifier.preprocess.create_file_paths_for_multiple_files2(project_root: str, target_dir: str, n_splits: int, n_processes: int, n_files: dict[int, int], dataset_name: str, write_format: str) → dict[int, list[str]]

Return per-split intermediate paths for multi-file merge.

sequifier.preprocess.create_file_paths_for_single_file(project_root: str, target_dir: str, n_splits: int, n_batches: int, dataset_name: str, write_format: str) → dict[int, list[str]]

Return per-split temp paths for one single-file run.

sequifier.preprocess.create_id_map(data: DataFrame, column: str) → dict[Union[str, int], int]

Map sorted user values to IDs after reserved tokens.

sequifier.preprocess.delete_files(files: list[str] | dict[int, list[str]]) → None

Delete paths from a list or split-indexed dict.

sequifier.preprocess.extract_sequences(data: DataFrame, schema: Any, layout: StoredWindowLayout, stride_for_split: int, columns: list[str], subsequence_start_mode: str) → DataFrame

Extract long-format windows from grouped sequences.

sequifier.preprocess.extract_subsequences(in_seq: dict[str, list], stored_context_width: int, stride_for_split: int, columns: list[str], subsequence_start_mode: str) → tuple[dict[str, list[list[Union[float, int]]]], list[int], numpy.ndarray]

Extract padded windows plus left-pad lengths from one sequence.

sequifier.preprocess.get_batch_limits(data: DataFrame, n_batches: int, allow_sequence_splitting: bool) → list[tuple[int, int]]

Split rows into batches without crossing sequenceId boundaries unless allowed.

sequifier.preprocess.get_combined_statistics(n1: int, mean1: float, std1: float, n2: int, mean2: float, std2: float) → tuple[float, float]

Combine two mean/std summaries.

sequifier.preprocess.get_group_bounds(data_subset: DataFrame, split_ratios: list[float])

Return per-split row bounds for one sequence.

sequifier.preprocess.get_subsequence_starts(in_context_length: int, stored_context_width: int, stride_for_split: int, subsequence_start_mode: str) → ndarray

Return window start indices for distribute/exact modes.

sequifier.preprocess.insert_top_folder(path: str, folder_name: str) → str

Insert folder_name before the basename.

sequifier.preprocess.load_precomputed_id_maps(project_root: str, data_columns: list[str] | None, required_maps: list[str] | None = None) → dict[str, dict[Union[str, int], int]]

Load and validate precomputed ID maps.

sequifier.preprocess.preprocess(args: Any, args_config: dict[str, Any]) → None

Load preprocessing config and run preprocessing.

sequifier.preprocess.preprocess_batch(project_root: str, data_name_root: str, process_id: int, batch: DataFrame, schema: Any, split_paths: list[str], layout: StoredWindowLayout, stride_by_split: list[int], data_columns: list[str], col_types: dict[str, str], split_ratios: list[float], target_dir: str, write_format: str, batches_per_file: int, subsequence_start_mode: str, merge_output: bool, split_method: str = 'within_sequence', seed: int = 1010) → None

Extract and write all split windows for one batch.

sequifier.preprocess.process_and_write_data_pt(data: DataFrame, stored_context_width: int, path: str, column_data_types: dict[str, str])

Write long-format sequences as packed PT tensors.

class sequifier.train.TransformerEmbeddingModel(transformer_model: TransformerModel)

Embedding-only wrapper for TransformerModel.

forward(src: dict[str, torch.Tensor], metadata: dict[str, torch.Tensor])

Return embedding output from the wrapped model.

class sequifier.train.TransformerModel(hparams: Any, rank: int | None = None, local_rank: int | None = None)

Sequifier transformer plus train/eval/export routines.

__init__(hparams: Any, rank: int | None = None, local_rank: int | None = None)

Build model modules and training state from config.

apply_softmax(target_column: str, output: Tensor) → Tensor

Apply LogSoftmax only for categorical targets.

decode(target_column: str, output: Tensor) → Tensor

Project hidden states through one target decoder.

forward(src: dict[str, torch.Tensor], metadata: dict[str, torch.Tensor], return_logits: bool | Tensor = False) → dict[str, torch.Tensor]

Return final-step logits or predictions for inference/eval.

forward_embed(src: dict[str, torch.Tensor], metadata: dict[str, torch.Tensor]) → Tensor

Return final-step embeddings.

forward_inner(src: dict[str, torch.Tensor], metadata: dict[str, torch.Tensor]) → Tensor

Encode inputs into contextual hidden states.

forward_train(src: dict[str, torch.Tensor], metadata: dict[str, torch.Tensor]) → dict[str, torch.Tensor]

Return raw decoded outputs for all target columns.

initialize_optimizer(params: Any = None) → None

Create optimizer and scheduler from training config.

train_model(train_loader: DataLoader, valid_loader: DataLoader, ddp_model: Module | None = None) → None

Run epochs, validation, checkpointing, export, and interruption cleanup.

sequifier.train.accumulate_class_counts(counts: dict[str, torch.Tensor], output: dict[str, torch.Tensor], valid_mask: Tensor, n_classes: dict[str, int]) → None

Accumulates predicted class counts over valid evaluation tokens.

sequifier.train.cleanup()

Destroy the active distributed process group.

sequifier.train.infer_with_embedding_model(model: Module, x: list[dict[str, numpy.ndarray]], device: str, size: int, target_columns: list[str], metadata: list[dict[str, numpy.ndarray]], column_data_types: dict[str, torch.dtype]) → ndarray

Run batched embedding inference and concatenate CPU outputs.

sequifier.train.infer_with_generative_model(model: Module, x: list[dict[str, numpy.ndarray]], device: str, size: int, target_columns: list[str], metadata: list[dict[str, numpy.ndarray]], column_data_types: dict[str, torch.dtype]) → dict[str, numpy.ndarray]

Run batched generative inference and trim CPU outputs.

sequifier.train.load_inference_model(model_type: str, model_path: str, training_config_path: str, args_config: dict[str, Any], device: str, infer_with_dropout: bool) → Module

Load a PT checkpoint as a generative or embedding inference module.

sequifier.train.train(args: Any, args_config: dict[str, Any]) → None

Load train config and launch local or distributed training.

sequifier.train.train_worker(local_rank: int, world_size: int, config: ResolvedSequifierConfig, from_folder: bool, global_rank: int, torch_compile: str)

Run one local distributed-training worker.

class sequifier.infer.Inferer(model_type: str, model_path: str, project_root: str, id_maps: dict[str, dict[str | int, int]] | None, selected_columns_statistics: dict[str, dict[str, float]], map_to_id: bool, categorical_columns: list[str], real_columns: list[str], input_columns: list[str] | None, target_columns: list[str], target_column_types: dict[str, str], sample_from_distribution_columns: list[str] | None, infer_with_dropout: bool, prediction_length: int, inference_batch_size: int, device: str, args_config: dict[str, Any], training_config_path: str | None, training_objective: str | None = None, normalize_real_columns: bool = True, target_decoder_ids: dict[str, list[int]] | None = None)

Inference runtime for PT/ONNX sequifier models.

__init__(model_type: str, model_path: str, project_root: str, id_maps: dict[str, dict[str | int, int]] | None, selected_columns_statistics: dict[str, dict[str, float]], map_to_id: bool, categorical_columns: list[str], real_columns: list[str], input_columns: list[str] | None, target_columns: list[str], target_column_types: dict[str, str], sample_from_distribution_columns: list[str] | None, infer_with_dropout: bool, prediction_length: int, inference_batch_size: int, device: str, args_config: dict[str, Any], training_config_path: str | None, training_objective: str | None = None, normalize_real_columns: bool = True, target_decoder_ids: dict[str, list[int]] | None = None)

Load a PT or ONNX backend and postprocessing state.

adjust_and_infer_embedding(x: dict[str, numpy.ndarray], size: int, metadata: dict[str, numpy.ndarray], column_data_types: dict[str, torch.dtype])

Batch embedding inference across the active backend.

adjust_and_infer_generative(x: dict[str, numpy.ndarray], size: int, metadata: dict[str, numpy.ndarray], column_data_types: dict[str, torch.dtype])

Batch generative inference across the active backend.

expand_to_batch_size(x: ndarray) → ndarray

Repeat leading samples until the ONNX batch size is met.

infer_embedding(x: dict[str, numpy.ndarray], metadata: dict[str, numpy.ndarray], column_data_types: dict[str, torch.dtype]) → ndarray

Return embeddings for a feature-array batch.

infer_generative(x: dict[str, numpy.ndarray] | None, metadata: dict[str, numpy.ndarray], probs: dict[str, numpy.ndarray] | None = None, return_probs: bool = False, column_data_types: dict[str, torch.dtype] | None = None) → dict[str, numpy.ndarray]

Return target probabilities or decoded predictions.

infer_pure(x: dict[str, numpy.ndarray], metadata: dict[str, numpy.ndarray]) → list[numpy.ndarray]

Run one ONNX batch and flatten sequence-major outputs.

invert_normalization(values: ndarray, target_column: str) → ndarray

Invert target-column Z-score normalization.

prepare_inference_batches(x: dict[str, numpy.ndarray], pad_to_batch_size: bool) → list[dict[str, numpy.ndarray]]

Split feature arrays into backend-sized batches.

class sequifier.infer.WindowedInferenceBatch(inputs: dict[str, torch.Tensor], metadata: dict[str, torch.Tensor], sequence_ids: Tensor, subsequence_ids: Tensor, model_start_positions: Tensor, window_start_offsets: Tensor)

Model-facing windows plus physical identities and adjusted starts.

sequifier.infer.apply_inference_column_types(data: DataFrame, config: ResolvedInferenceConfig) → DataFrame

Cast loaded long-format sequence values to the configured unified dtype.

sequifier.infer.apply_inference_tensor_types(sequences_dict: dict[str, torch.Tensor], column_data_types: dict[str, torch.dtype]) → dict[str, torch.Tensor]

Cast loaded PT feature tensors to the configured per-column dtype.

sequifier.infer.calculate_item_positions(start_positions: ndarray, context_length: int, prediction_length: int, training_objective: str, target_offset: int = 1) → ndarray

Return flattened absolute item positions for inference outputs.

sequifier.infer.fill_number(number: int | float, max_length: int) → str

Left-pad a number for sortable string keys.

sequifier.infer.get_embeddings(config: Any, inferer: Inferer, data: DataFrame, column_data_types: dict[str, torch.dtype]) → ndarray

Infer embeddings from a Polars chunk.

sequifier.infer.get_embeddings_pt(config: Any, inferer: Inferer, data: dict[str, torch.Tensor], metadata: dict[str, torch.Tensor], column_data_types: dict[str, torch.dtype]) → ndarray

Infer embeddings from PT tensors.

sequifier.infer.get_probs_preds_autoregression(config: Any, inferer: Inferer, data: DataFrame, column_data_types: dict[str, torch.dtype], context_length: int) → tuple[Optional[dict[str, numpy.ndarray]], dict[str, numpy.ndarray], numpy.ndarray, numpy.ndarray, numpy.ndarray]

Infer autoregressive predictions with sequence IDs, positions, and mask.

sequifier.infer.get_probs_preds_from_df(config: Any, inferer: Inferer, data: DataFrame, column_data_types: dict[str, torch.dtype]) → tuple[Optional[dict[str, numpy.ndarray]], dict[str, numpy.ndarray]]

Infer non-autoregressive predictions from a Polars chunk.

sequifier.infer.get_probs_preds_from_dict(config: Any, inferer: Inferer, data: dict[str, torch.Tensor], metadata: dict[str, torch.Tensor], column_data_types: dict[str, torch.dtype], total_steps: int = 1) → tuple[Optional[dict[str, numpy.ndarray]], dict[str, numpy.ndarray]]

Infer PT predictions, flattened sample-major across autoregressive steps.

sequifier.infer.infer(args: Any, args_config: dict[str, Any]) → None

Load inference config and dispatch the worker.

sequifier.infer.infer_embedding(config: ResolvedInferenceConfig, inferer: Inferer, model_id: str, dataset: list[Any] | Iterator[Any], column_data_types: dict[str, torch.dtype]) → None

Write embeddings for each dataset chunk.

sequifier.infer.infer_generative(config: ResolvedInferenceConfig, inferer: Inferer, model_id: str, dataset: list[Any] | Iterator[Any], column_data_types: dict[str, torch.dtype])

Write generative predictions/probabilities for each dataset chunk.

sequifier.infer.infer_worker(config: Any, args_config: dict[str, Any], id_maps: dict[str, dict[str | int, int]] | None, selected_columns_statistics: dict[str, dict[str, float]], percentage_limits: tuple[float, float] | None, normalize_real_columns: bool)

Load data, instantiate models, and run the configured inference mode.

sequifier.infer.load_onnx_target_decoder_ids(session: InferenceSession, target_columns: list[str], target_column_types: dict[str, str], model_type: str) → dict[str, list[int]]

Load and validate categorical decoder-index mappings from ONNX metadata.

sequifier.infer.load_parquet_folder_dataset(data_path: str, start_pct: float, end_pct: float) → Iterator[Any]

Yield a percentage slice of sorted top-level Parquet files.

sequifier.infer.load_pt_dataset(data_path: str, start_pct: float, end_pct: float) → Iterator[Any]

Yield a percentage slice of sorted top-level PT files.

sequifier.infer.normalize(outs: dict[str, numpy.ndarray]) → dict[str, numpy.ndarray]

Softmax logits by target column.

sequifier.infer.sample_with_cumsum(probs: ndarray, is_log_probs: bool = True) → ndarray

Sample class indices from log-probabilities or probabilities.

sequifier.infer.verify_variable_order(data: DataFrame) → None

Require sequenceId order and in-sequence subsequenceId order.

sequifier.make.make(args)

Create a sequifier project scaffold.

sequifier.hyperparameter_search.hyperparameter_search(config_path: str, skip_metadata: bool) → None

Load config, create Optuna study, and optimize trials.

sequifier.hyperparameter_search.objective(trial: Trial, config) → float | tuple[float, ...]

Run one Optuna trial through the CLI trainer and metrics JSONL.

sequifier.hyperparameter_search.set_pdeathsig()

Ask Linux to SIGTERM children when this parent dies.

class sequifier.helpers.ModelWindowSamplingPlan(resolved_view: ResolvedWindowView, stride: int | None = None)

Resolve logical model windows contained in one stored window.

build_masks(left_pad_lengths: Tensor, input_starts: Tensor) → dict[str, torch.Tensor]

Build masks for model windows with different positions in storage.

property candidate_input_starts: Tensor

Return chronological starts, anchored to include the rightmost view.

first_eligible_start_indices(left_pad_lengths: Tensor) → Tensor

Return the first candidate with at least one valid target position.

gather(tensor: Tensor, stored_row_indices: Tensor, input_starts: Tensor, *, target: bool = False) → Tensor

Gather input or target windows without materializing all overlaps.

sample_counts(left_pad_lengths: Tensor) → Tensor

Return the number of usable logical samples in each stored row.

class sequifier.helpers.ModelWindowView(context_length: int, objective: str, target_offset: int)

class sequifier.helpers.ResolvedWindowView(storage: sequifier.helpers.StoredWindowLayout, view: sequifier.helpers.ModelWindowView, required_width: int, input_slice: slice, target_slice: slice)

build_masks(left_pad_lengths: Tensor) → dict[str, torch.Tensor]

Build explicit input-attention and target-validity masks for this view.

class sequifier.helpers.StoredWindowLayout(stored_context_width: int, max_target_offset: int, version: int)

class sequifier.helpers.WindowSampleIndex(plan: ModelWindowSamplingPlan, left_pad_lengths: Tensor)

Compact logical-index mapping for variable per-row window counts.

sequifier.helpers.assign_sequence_to_split(sequence_id: int, split_ratios: list[float], seed: int) → int

Deterministically assign one sequenceId to a split index.

sequifier.helpers.build_valid_mask(left_pad_lengths: Tensor, full_length: int, view_slice: slice) → Tensor

Boolean mask from left-padding metadata.

sequifier.helpers.configure_determinism(seed: int, strict: bool = False) → None

Enforces deterministic execution for reproducibility.

sequifier.helpers.configure_logger(project_root: str, model_name: str, rank: int | None = 0)

Configure console plus rank-scoped debug/info log files.

sequifier.helpers.configured_model_window_stride(config: Any) → int | None

Read the optional stride from validated configs or legacy test doubles.

sequifier.helpers.construct_index_maps(id_maps: dict[str, dict[Union[str, int], int]] | None, target_columns_index_map: list[str], map_to_id: bool | None) → dict[str, dict[int, Union[str, int]]]

Build index-to-ID maps, including reserved token labels.

sequifier.helpers.derive_target_column_types(target_columns: list[str], column_data_types: dict[str, str]) → dict[str, str]

Derive categorical/real target kinds from configured physical dtypes.

sequifier.helpers.get_best_model_path(project_root: str, run_name: str, model_type: str) → tuple[str, int]

Return the highest-epoch exported best model path.

sequifier.helpers.get_last_training_batch_timedelta(model_name: str, rank: int, project_root: str = '.') → float

Return seconds between the last two mid-epoch train log entries.

sequifier.helpers.get_left_pad_lengths_from_preprocessed_data(data: DataFrame) → Tensor

One leftPadLength per long-format subsequence.

sequifier.helpers.get_torch_dtype(dtype_str: str) → dtype

String-to-torch dtype mapping.

sequifier.helpers.metadata_config_path_from_preprocessing_data_path(preprocessing_data_path: str) → str

Return the metadata path generated for a preprocessing input path.

sequifier.helpers.normalize_path(path: str, project_root: str) → str

Return path rooted under project_root.

sequifier.helpers.numpy_storage_to_pytorch(data: DataFrame, column_data_types: dict[str, torch.dtype], all_columns: list[str], stored_context_width: int, sort_rows: bool = True) → tuple[dict[str, torch.Tensor], torch.Tensor]

Convert complete stored windows to tensors for virtual window sampling.

sequifier.helpers.numpy_to_pytorch(data: DataFrame, column_data_types: dict[str, torch.dtype], all_columns: list[str], resolved_view: ResolvedWindowView) → tuple[dict[str, torch.Tensor], dict[str, torch.Tensor]]

Convert long-format Polars windows to tensors plus masks.

sequifier.helpers.read_data(path: str, read_format: str, columns: list[str] | None = None) → DataFrame

Read CSV/Parquet into Polars.

sequifier.helpers.resolve_unified_polars_numeric_dtype(column_data_types: dict[str, str]) → Any

Resolve one Polars dtype for long-format numeric sequence columns.

sequifier.helpers.subset_to_input_columns(data: DataFrame | LazyFrame, input_columns: list[str]) → DataFrame | LazyFrame

Keep long-format rows whose inputCol is selected.

sequifier.helpers.write_data(data: DataFrame, path: str, write_format: str, **kwargs) → None

Write Polars/Pandas data as CSV or Parquet.

class sequifier.io.yaml.TrainModelDumper(stream, default_style=None, default_flow_style=False, canonical=None, indent=None, width=None, allow_unicode=None, line_break=None, encoding=None, explicit_start=None, explicit_end=None, version=None, tags=None, sort_keys=True)

YAML dumper for sequifier config objects.

increase_indent(flow=False, indentless=False)

Indent block sequences.

sequifier.io.yaml.represent_dot_dict(dumper, data)

Represent DotDict as a plain YAML mapping.

sequifier.io.yaml.represent_numpy_float(dumper, data)

Represent NumPy floats as YAML floats.

sequifier.io.yaml.represent_numpy_int(dumper, data)

Represent NumPy integers as YAML integers.

sequifier.io.yaml.represent_sequifier_object(dumper, data)

Represent sequifier config objects as plain YAML mappings.

class sequifier.io.sequifier_dataset_from_file.SequifierDatasetFromFile(data_path: str, config: ResolvedSequifierConfig, shuffle: bool = True)

Eager single-file dataset yielding pre-collated batches.

set_epoch(epoch: int)

Set the shuffle epoch.

set_start_batch(start_batch: int)

Set the first global batch to yield on the next iteration.

sequifier.optimizers.optimizers.get_optimizer_class(optimizer_name: str) → Optimizer

Resolve a custom, torch-optimizer, or torch optimizer class.

```