Python API

Simplest Usage

from open_mlpipe import run

# One line — full pipeline
ctx = run("dataset.csv", target="price")

Import Methods

# Method 1: Import run function
from open_mlpipe import run

# Method 2: Import run_config for YAML
from open_mlpipe import run_config

# Method 3: Import classes for full control
from open_mlpipe import PipelineConfig, PipelineRunner

# Method 4: Import specific components
from open_mlpipe.config.schema import PipelineConfig, DataConfig
from open_mlpipe.core.pipeline import PipelineRunner
from open_mlpipe.utils.io import load_data

Level 1: Zero-Touch

from open_mlpipe import run

# Auto-detect everything
ctx = run("data.csv")

Level 2: Specify Target

from open_mlpipe import run

ctx = run("data.csv", target="price")

# Access results
print(f"Best model: {ctx.best_model_name}")
print(f"Test R²: {ctx.metrics['test_r2']:.4f}")

Level 3: Custom Configuration

from open_mlpipe import PipelineConfig, PipelineRunner
from open_mlpipe.config.resolver import build_level1_config

# Build config from data
config = build_level1_config("data.csv", target="price")

# Customize
config.project = "my-project"
config.tuning.enabled = True
config.tuning.n_trials = 50

# Run pipeline
runner = PipelineRunner(config)
ctx = runner.run()

Level 4: Full Manual Control

from open_mlpipe.config.schema import (
    PipelineConfig, DataConfig, ModelSelectionConfig,
    TuningConfig, FeatureSelectionConfig, EvaluationConfig,
)

config = PipelineConfig(
    project="custom-project",
    task="auto",
    data=DataConfig(path="data.csv", target="price"),
    model_selection=ModelSelectionConfig(
        candidates=["lightgbm", "xgboost"],
        scoring=["r2"],
    ),
    tuning=TuningConfig(enabled=True, n_trials=30),
    feature_selection=FeatureSelectionConfig(enabled=True),
    evaluation=EvaluationConfig(explainability=True),
)

runner = PipelineRunner(config)
ctx = runner.run()

Level 5: Load and Predict

import joblib
import pandas as pd

# Load saved model
model = joblib.load("artifacts/model_v1.joblib")

# Predict
new_data = pd.DataFrame({"feature1": [1.0], "feature2": ["a"]})
predictions = model.predict(new_data)

API Reference

open_mlpipe.run

run(data: str, target: str | None = None, project: str = 'open-mlpipe')

Run the full ML pipeline on a dataset.

PARAMETER	DESCRIPTION
data	Path to CSV/Parquet/Excel file TYPE: str
target	Target column name (auto-detected if None) TYPE: str | None DEFAULT: None
project	Project name for tracking TYPE: str DEFAULT: 'open-mlpipe'

RETURNS	DESCRIPTION
	PipelineContext with all results, metrics, and the trained model

Source code in src\open_mlpipe\__init__.py

def run(data: str, target: str | None = None, project: str = "open-mlpipe"):
    """Run the full ML pipeline on a dataset.

    Args:
        data: Path to CSV/Parquet/Excel file
        target: Target column name (auto-detected if None)
        project: Project name for tracking

    Returns:
        PipelineContext with all results, metrics, and the trained model
    """
    from open_mlpipe.config.resolver import build_level1_config
    config = build_level1_config(data, target)
    config.project = project
    runner = PipelineRunner(config)
    return runner.run()

open_mlpipe.run_config

run_config(config_path: str, project: str | None = None)

Run the full ML pipeline from a YAML config file.

PARAMETER	DESCRIPTION
config_path	Path to YAML config file TYPE: str
project	Override project name (optional) TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
	PipelineContext with all results, metrics, and the trained model

Source code in src\open_mlpipe\__init__.py

def run_config(config_path: str, project: str | None = None):
    """Run the full ML pipeline from a YAML config file.

    Args:
        config_path: Path to YAML config file
        project: Override project name (optional)

    Returns:
        PipelineContext with all results, metrics, and the trained model
    """
    from open_mlpipe.config.resolver import load_config, resolve_config
    from open_mlpipe.utils.io import load_data
    config = load_config(config_path)
    if project:
        config.project = project
    df = load_data(config.data.path)
    config = resolve_config(config, df)
    runner = PipelineRunner(config)
    return runner.run()

open_mlpipe.PipelineConfig

Bases: BaseModel

Full pipeline configuration.

open_mlpipe.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