Skip to content

MALA

Metropolis-adjusted Langevin Algorithm — gradient-biased proposals for faster mixing.

Requires grad_log_likelihood (and optionally grad_log_prior) on the Problem.

import mcmckit as mc

sampler = mc.MALA(step_size=0.3, n_samples=10_000)
result  = sampler.run(problem_with_grad, x0=[0.0, 0.0])

Optimal acceptance rate ≈ 57.4% (tune step_size accordingly).

MALA

Bases: BaseSampler

Metropolis-Adjusted Langevin Algorithm (MALA).

Uses the gradient of the log-posterior to bias proposals toward higher probability regions, giving faster mixing than vanilla MH.

Proposal: x* = x + (ε²/2) * ∇log p(x|y) + ε * noise, noise ~ N(0, I)

The MH accept/reject step corrects for the asymmetric proposal, so the chain targets the exact posterior.

Parameters:

Name Type Description Default
step_size float

Langevin step size ε. Smaller = higher acceptance, slower mixing. Larger = lower acceptance, faster exploration when accepted. Typical starting point: tune so acceptance rate is ~0.57 (optimal for MALA in high dimensions, vs ~0.23 for MH).

 required
n_samples int

Number of samples to collect when calling run().

 required
Notes

Requires problem.has_grad == True, i.e. both grad_log_likelihood and grad_log_prior must be provided to the Problem.

Examples:

Full run

sampler = MALA(step_size=0.1, n_samples=5000) result = sampler.run(problem, x0=[0.0, 0.0])

Step-by-step

sampler = MALA(step_size=0.1, n_samples=5000) sampler.initialize(problem, x0=[0.0, 0.0]) for i in range(5000): sampler.step() result = sampler.get_result()

Source code in mcmckit/samplers/mala.py 
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class MALA(BaseSampler):
    """Metropolis-Adjusted Langevin Algorithm (MALA).

    Uses the gradient of the log-posterior to bias proposals toward higher
    probability regions, giving faster mixing than vanilla MH.

    Proposal:
        x* = x + (ε²/2) * ∇log p(x|y) + ε * noise,   noise ~ N(0, I)

    The MH accept/reject step corrects for the asymmetric proposal, so the
    chain targets the exact posterior.

    Parameters
    ----------
    step_size : float
        Langevin step size ε. Smaller = higher acceptance, slower mixing.
        Larger = lower acceptance, faster exploration when accepted.
        Typical starting point: tune so acceptance rate is ~0.57 (optimal
        for MALA in high dimensions, vs ~0.23 for MH).
    n_samples : int
        Number of samples to collect when calling run().

    Notes
    -----
    Requires ``problem.has_grad == True``, i.e. both ``grad_log_likelihood``
    and ``grad_log_prior`` must be provided to the Problem.

    Examples
    --------
    # Full run
    sampler = MALA(step_size=0.1, n_samples=5000)
    result = sampler.run(problem, x0=[0.0, 0.0])

    # Step-by-step
    sampler = MALA(step_size=0.1, n_samples=5000)
    sampler.initialize(problem, x0=[0.0, 0.0])
    for i in range(5000):
        sampler.step()
    result = sampler.get_result()
    """

    def __init__(self, step_size, n_samples):
        self.step_size = step_size
        self.n_samples = n_samples
        self._initialized = False

    # ------------------------------------------------------------------
    # Setup
    # ------------------------------------------------------------------

    def initialize(self, problem, x0):
        if not problem.has_grad:
            raise RuntimeError(
                "MALA requires grad_log_likelihood and grad_log_prior "
                "to be provided to Problem."
            )
        x0 = np.asarray(x0, dtype=float)
        self._problem = problem
        self.current = x0.copy()
        self.current_logp, self.current_grad = problem.log_posterior_and_grad(self.current)

        self._samples: list[np.ndarray] = []
        self._log_posteriors: list[float] = []
        self._n_accepted = 0
        self._n_steps = 0
        self._initialized = True

    # ------------------------------------------------------------------
    # Core interface
    # ------------------------------------------------------------------

    def step(self):
        """Perform one MALA step, updating internal state."""
        if not self._initialized:
            raise RuntimeError("Call initialize(problem, x0) before step().")

        eps = self.step_size
        d = len(self.current)

        # Forward proposal: drift toward higher probability + noise
        mean_fwd = self.current + 0.5 * eps**2 * self.current_grad
        proposal = mean_fwd + eps * np.random.randn(d)

        logp_prop, grad_prop = self._problem.log_posterior_and_grad(proposal)

        # MH correction for asymmetric proposal
        log_q_fwd = self._log_proposal_density(self.current, proposal, self.current_grad)
        log_q_rev = self._log_proposal_density(proposal, self.current, grad_prop)

        log_alpha = logp_prop + log_q_rev - self.current_logp - log_q_fwd

        if np.log(np.random.rand()) < log_alpha:
            self.current = proposal
            self.current_logp = logp_prop
            self.current_grad = grad_prop
            self._n_accepted += 1

        self._samples.append(self.current.copy())
        self._log_posteriors.append(self.current_logp)
        self._n_steps += 1

    def run(self, problem, x0):
        """Initialize and run for n_samples steps, returning a Result."""
        self.initialize(problem, x0)
        for _ in range(self.n_samples):
            self.step()
        return self.get_result()

    def get_result(self):
        """Return a Result from all samples collected so far."""
        if not self._samples:
            raise RuntimeError("No samples collected yet.")
        return Result(
            samples=np.array(self._samples),
            log_posteriors=np.array(self._log_posteriors),
            param_names=self._problem.param_names,
            acceptance_rate=self._n_accepted / self._n_steps,
        )

    # ------------------------------------------------------------------
    # Helpers
    # ------------------------------------------------------------------

    def _log_proposal_density(self, x_from, x_to, grad_from):
        """Log density of proposing x_to given x_from and its gradient."""
        eps = self.step_size
        mean = x_from + 0.5 * eps**2 * grad_from
        diff = x_to - mean
        return -0.5 / (eps**2) * np.dot(diff, diff)

    # ------------------------------------------------------------------
    # Convenience properties
    # ------------------------------------------------------------------

    @property
    def acceptance_rate(self):
        if self._n_steps == 0:
            return None
        return self._n_accepted / self._n_steps

    @property
    def n_steps(self):
        return self._n_steps

Methods:

step 

step()

Perform one MALA step, updating internal state.

Source code in mcmckit/samplers/mala.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def step(self):
    """Perform one MALA step, updating internal state."""
    if not self._initialized:
        raise RuntimeError("Call initialize(problem, x0) before step().")

    eps = self.step_size
    d = len(self.current)

    # Forward proposal: drift toward higher probability + noise
    mean_fwd = self.current + 0.5 * eps**2 * self.current_grad
    proposal = mean_fwd + eps * np.random.randn(d)

    logp_prop, grad_prop = self._problem.log_posterior_and_grad(proposal)

    # MH correction for asymmetric proposal
    log_q_fwd = self._log_proposal_density(self.current, proposal, self.current_grad)
    log_q_rev = self._log_proposal_density(proposal, self.current, grad_prop)

    log_alpha = logp_prop + log_q_rev - self.current_logp - log_q_fwd

    if np.log(np.random.rand()) < log_alpha:
        self.current = proposal
        self.current_logp = logp_prop
        self.current_grad = grad_prop
        self._n_accepted += 1

    self._samples.append(self.current.copy())
    self._log_posteriors.append(self.current_logp)
    self._n_steps += 1

run 

run(problem, x0)

Initialize and run for n_samples steps, returning a Result.

Source code in mcmckit/samplers/mala.py 
108
109
110
111
112
113
def run(self, problem, x0):
    """Initialize and run for n_samples steps, returning a Result."""
    self.initialize(problem, x0)
    for _ in range(self.n_samples):
        self.step()
    return self.get_result()

get_result 

get_result()

Return a Result from all samples collected so far.

Source code in mcmckit/samplers/mala.py 
115
116
117
118
119
120
121
122
123
124
def get_result(self):
    """Return a Result from all samples collected so far."""
    if not self._samples:
        raise RuntimeError("No samples collected yet.")
    return Result(
        samples=np.array(self._samples),
        log_posteriors=np.array(self._log_posteriors),
        param_names=self._problem.param_names,
        acceptance_rate=self._n_accepted / self._n_steps,
    )