Core

PipelineRunner

open_mlpipe.core.pipeline.PipelineRunner

PipelineRunner(config: PipelineConfig, registry: StageRegistry | None = None)

Executes a pipeline from config through all stages.

Source code in src\open_mlpipe\core\pipeline.py

def __init__(self, config: PipelineConfig, registry: StageRegistry | None = None) -> None:
    self.config = config
    self.registry = registry or StageRegistry()
    self.ctx = PipelineContext(config=config)

Methods:

run

run() -> PipelineContext

Execute the full pipeline.

Source code in src\open_mlpipe\core\pipeline.py

def run(self) -> PipelineContext:
    """Execute the full pipeline."""
    from open_mlpipe.stages import (
        CleanStage,
        CompareStage,
        DataLoaderStage,
        DeployStage,
        EDALoaderStage,
        EvaluateStage,
        ExplainStage,
        FeatureEngStage,
        PreprocessStage,
        SaveStage,
        SelectStage,
        SplitStage,
        TuneStage,
    )

    # Register all stages
    stages: list[Stage] = [
        DataLoaderStage(),
        EDALoaderStage(),
        CleanStage(),
        FeatureEngStage(),
        SplitStage(),
        PreprocessStage(),
        CompareStage(),
        TuneStage(),
        SelectStage(),
        EvaluateStage(),
        ExplainStage(),
        SaveStage(),
    ]

    if self.config.deployment.enabled:
        stages.append(DeployStage())

    for s in stages:
        self.registry.register(s)

    # Print header
    self._print_header()

    # Execute stages
    for stage in self.registry.default_order():
        if not self.registry.has(stage):
            continue
        s = self.registry.get(stage)
        if s.should_skip(self.ctx):
            console.print(f"  [dim]skip {stage}[/dim]")
            continue

        self._run_stage(s)

        # Resolve config after data is loaded (need raw_data for auto-detection)
        if stage == "load" and self.ctx.raw_data is not None:
            self.config = resolve_config(self.config, self.ctx.raw_data)
            self.ctx.config = self.config

    # Print summary
    self._print_summary()

    return self.ctx

PipelineContext

open_mlpipe.core.context.PipelineContext dataclass

PipelineContext(
    raw_data: DataFrame | None = None,
    clean_data: DataFrame | None = None,
    featured_data: DataFrame | None = None,
    preprocessed_data: DataFrame | None = None,
    X_train: DataFrame | None = None,
    X_test: DataFrame | None = None,
    y_train: Series | None = None,
    y_test: Series | None = None,
    baseline_models: dict[str, Any] = dict(),
    best_model_name: str | None = None,
    best_model: Any | None = None,
    tuned_model: Any | None = None,
    final_model: Any | None = None,
    config: Any | None = None,
    task_type: TaskType | None = None,
    target_column: str | None = None,
    raw_feature_columns: list[str] = list(),
    column_types: dict[str, ColumnType] = dict(),
    numeric_columns: list[str] = list(),
    categorical_columns: list[str] = list(),
    datetime_columns: list[str] = list(),
    skewed_columns: list[str] = list(),
    normal_columns: list[str] = list(),
    high_cardinality_columns: list[str] = list(),
    low_cardinality_columns: list[str] = list(),
    columns_to_drop: list[str] = list(),
    preprocessor: Any = None,
    stage_history: list[StageMetadata] = list(),
    metrics: dict[str, float] = dict(),
    eda_report: dict[str, Any] | None = None,
    feature_importance: DataFrame | None = None,
    correlation_matrix: DataFrame | None = None,
    vif_scores: DataFrame | None = None,
    statistical_tests: DataFrame | None = None,
    experiment_id: str | None = None,
    run_id: str | None = None,
    artifact_dir: str | None = None,
    reports: dict[str, str] = dict(),
)

Shared state passed between stages.

Methods:

add_stage

add_stage(meta: StageMetadata) -> None

Record a stage execution.

Source code in src\open_mlpipe\core\context.py

def add_stage(self, meta: StageMetadata) -> None:
    """Record a stage execution."""
    self.stage_history.append(meta)

summary

summary() -> str

Human-readable summary of pipeline state.

Source code in src\open_mlpipe\core\context.py

def summary(self) -> str:
    """Human-readable summary of pipeline state."""
    lines = [
        f"Task: {self.task_type}",
        f"Target: {self.target_column}",
        f"Columns: {len(self.column_types)}",
        f"Numeric: {len(self.numeric_columns)}",
        f"Categorical: {len(self.categorical_columns)}",
        f"Stages run: {len(self.stage_history)}",
    ]
    if self.final_model:
        lines.append(f"Model: {self.best_model_name}")
    return "\n".join(lines)

StageMetadata

open_mlpipe.core.context.StageMetadata dataclass

StageMetadata(
    stage_name: str,
    stage_version: str = "1.0",
    input_rows: int = 0,
    input_cols: int = 0,
    output_rows: int = 0,
    output_cols: int = 0,
    parameters: dict[str, Any] = dict(),
    metrics: dict[str, float] = dict(),
    artifacts: dict[str, str] = dict(),
    duration_seconds: float = 0.0,
    warnings: list[str] = list(),
)

Immutable record of what a stage did.

Stage

open_mlpipe.core.stage.Stage

Bases: ABC

Base class for all pipeline stages.

Attributes

name abstractmethod property

name: str

Stage name for logging.

Methods:

execute abstractmethod

execute(ctx: PipelineContext) -> PipelineContext

Run this stage. Read from ctx, modify ctx, return ctx.

Source code in src\open_mlpipe\core\stage.py

@abstractmethod
def execute(self, ctx: PipelineContext) -> PipelineContext:
    """Run this stage. Read from ctx, modify ctx, return ctx."""

should_skip

should_skip(ctx: PipelineContext) -> bool

Override to conditionally skip this stage.

Source code in src\open_mlpipe\core\stage.py

def should_skip(self, ctx: PipelineContext) -> bool:
    """Override to conditionally skip this stage."""
    return False

StageRegistry

open_mlpipe.core.registry.StageRegistry

StageRegistry()

Registry for pipeline stages.

Source code in src\open_mlpipe\core\registry.py

def __init__(self) -> None:
    self._stages: dict[str, Stage] = {}

Methods:

default_order

default_order() -> list[str]

Return stages in canonical execution order.

Source code in src\open_mlpipe\core\registry.py

def default_order(self) -> list[str]:
    """Return stages in canonical execution order."""
    return [
        "load",
        "eda",
        "clean",
        "feature_eng",
        "split",
        "preprocess",
        "compare",
        "tune",
        "select",
        "evaluate",
        "explain",
        "save",
        "deploy",
    ]