Skip to content

Function Composition & Piping

Overview

Function composition combines simple functions into complex workflows: - Composition: f(g(x)) - apply functions in sequence - Piping: x |> g |> f - data-first style - Builder pattern: Chain methods fluently - Workflow automation: Clear data transformations

Function Composition

Basic Composition

# Simple functions
double = lambda x: x * 2
add_one = lambda x: x + 1
square = lambda x: x ** 2

# Compose manually
result = square(add_one(double(5)))  # ((5*2)+1)² = 121

# Composition utility
def compose(*functions):
    """Compose functions (right to left)."""
    def composed(x):
        result = x
        for func in reversed(functions):
            result = func(result)
        return result
    return composed

# Now compose elegantly
pipeline = compose(square, add_one, double)
print(pipeline(5))  # 121

# Order matters!
pipeline2 = compose(double, add_one, square)
print(pipeline2(5))  # (5²+1)*2 = 52

Piping Style

# Pipe function (left-to-right, opposite of compose)
def pipe(*functions):
    """Apply functions left to right."""
    def piped(x):
        result = x
        for func in functions:
            result = func(result)
        return result
    return piped

# More intuitive for data flow
normalize = lambda x: (x - x.mean()) / x.std()
scale = lambda x: x * 255
quantize = lambda x: x.astype(int)

preprocess = pipe(normalize, scale, quantize)
# Data flows: normalize → scale → quantize

Composition Operators

Operator Overloading for Composition

class Composable:
    """Function wrapper that supports composition."""

    def __init__(self, func):
        self.func = func

    def __call__(self, *args, **kwargs):
        return self.func(*args, **kwargs)

    def __rshift__(self, other):
        """f >> g means g(f(x))."""
        return Composable(lambda x: other(self.func(x)))

    def __lshift__(self, other):
        """f << g means f(g(x))."""
        return Composable(lambda x: self.func(other(x)))

# Use it
square = Composable(lambda x: x ** 2)
add_one = Composable(lambda x: x + 1)
double = Composable(lambda x: x * 2)

# Compose with >> operator
pipeline = double >> add_one >> square
print(pipeline(5))  # ((5*2)+1)² = 121

Data Transformation Pipelines

ML Data Pipeline

import numpy as np

def create_pipeline(*steps):
    """Create data transformation pipeline."""
    def process(data):
        result = data
        for step in steps:
            result = step(result)
        return result
    return process

# Define transformations
def load_data(filepath):
    """Load numpy array."""
    return np.random.randn(100, 10)  # Placeholder

def remove_outliers(data):
    """Remove outliers (3-sigma)."""
    mean = data.mean(axis=0)
    std = data.std(axis=0)
    mask = np.abs(data - mean) < 3 * std
    return data[mask.all(axis=1)]

def normalize(data):
    """Normalize to mean=0, std=1."""
    return (data - data.mean(axis=0)) / data.std(axis=0)

def pca_reduce(data, components=5):
    """Reduce with PCA."""
    # Simplified PCA
    return data[:, :components]

# Create pipeline
data_pipeline = create_pipeline(
    load_data,
    remove_outliers,
    normalize,
    lambda x: pca_reduce(x, 5)
)

# Use it
processed_data = data_pipeline("path/to/data.csv")
print(processed_data.shape)  # (100, 5)

Inference Pipeline

# Model inference pipeline
class Model:
    def forward(self, x):
        # Simplified model
        return x * 2

def create_inference_pipeline(model):
    """Create model inference pipeline."""

    def preprocess(raw_input):
        """Convert raw input to tensor."""
        return raw_input

    def inference(preprocessed):
        """Run model."""
        return model.forward(preprocessed)

    def postprocess(output):
        """Convert output to result."""
        return {"prediction": output, "confidence": 0.95}

    # Compose into pipeline
    return pipe(preprocess, inference, postprocess)

model = Model()
pipeline = create_inference_pipeline(model)

# Use end-to-end
result = pipeline(5)
print(result)  # {'prediction': 10, 'confidence': 0.95}

Builder Pattern

Fluent Interface

class DataProcessing:
    """Fluent data processing builder."""

    def __init__(self, data):
        self.data = data

    def normalize(self, mean=None, std=None):
        """Normalize data."""
        data = self.data
        mean = mean or data.mean(axis=0)
        std = std or data.std(axis=0)
        self.data = (data - mean) / std
        return self  # Return self for chaining

    def scale(self, factor):
        """Scale data."""
        self.data = self.data * factor
        return self

    def filter_outliers(self, std_threshold=3):
        """Remove outliers."""
        data = self.data
        mask = np.abs(data) < std_threshold * np.std(data)
        self.data = data[mask.all(axis=1)]
        return self

    def get(self):
        """Get processed data."""
        return self.data

# Use fluent interface
import numpy as np

data = np.random.randn(100, 10)
processed = (DataProcessing(data)
    .normalize()
    .scale(255)
    .filter_outliers(3)
    .get())

print(processed.shape)  # Transformed data shape

Advanced Composition

Monadic Composition

# Composition with error handling
class Result:
    """Represents success or failure."""

    def __init__(self, value=None, error=None):
        self.value = value
        self.error = error

    def is_success(self):
        return self.error is None

    def flat_map(self, func):
        """Apply function if successful."""
        if self.is_success():
            try:
                return func(self.value)
            except Exception as e:
                return Result(error=str(e))
        return self

    def map(self, func):
        """Transform value if successful."""
        return self.flat_map(lambda x: Result(func(x)))

    def get_or_default(self, default):
        """Get value or default if failed."""
        return self.value if self.is_success() else default

# Use monadic composition
def safe_divide(x, y):
    if y == 0:
        return Result(error="Division by zero")
    return Result(x / y)

def safe_sqrt(x):
    if x < 0:
        return Result(error="Negative number")
    return Result(x ** 0.5)

# Compose operations that can fail
result = (Result(16)
    .flat_map(lambda x: safe_sqrt(x))    # sqrt(16) = 4
    .flat_map(lambda x: safe_divide(8, x))  # 8/4 = 2
    .get_or_default(0))

print(result)  # 2.0

Real-World ML Pipelines

Training Pipeline

class TrainingPipeline:
    """Chainable training pipeline."""

    def __init__(self, model):
        self.model = model
        self.optimizer = None
        self.loss_fn = None
        self.metrics = []

    def with_optimizer(self, optimizer):
        self.optimizer = optimizer
        return self

    def with_loss(self, loss_fn):
        self.loss_fn = loss_fn
        return self

    def track_metric(self, metric_name):
        self.metrics.append(metric_name)
        return self

    def compile(self):
        """Verify all components are set."""
        if not self.optimizer:
            raise ValueError("Optimizer required")
        if not self.loss_fn:
            raise ValueError("Loss function required")
        return self

    def train(self, data, epochs):
        """Run training."""
        print(f"Training {self.model} for {epochs} epochs")
        print(f"Optimizer: {self.optimizer}")
        print(f"Loss: {self.loss_fn}")
        print(f"Tracking: {self.metrics}")
        # Actual training code here

# Use it
import torch
model = torch.nn.Linear(10, 5)

(TrainingPipeline(model)
    .with_optimizer('adam')
    .with_loss('mse')
    .track_metric('accuracy')
    .track_metric('loss')
    .compile()
    .train(data, epochs=10))

Summary: Composition Patterns

Pattern Use Case Example
Function composition Chain operations compose(f, g, h)
Piping Left-to-right data flow pipe(f, g, h)(x)
Builder Fluent configuration obj.set_a().set_b().build()
Monadic Handle errors in chain result.flat_map(f)
Pipeline Multi-step workflow preprocess → model → postprocess