Initial commit: OptimizR - High-performance optimization algorithms in Rust with Python bindings
This commit is contained in:
Melvin Avarez
@@ -0,0 +1,29 @@
|
||||
"""
|
||||
OptimizR - High-Performance Optimization Algorithms
|
||||
===================================================
|
||||
|
||||
Fast, reliable implementations of advanced optimization and statistical
|
||||
inference algorithms with Rust acceleration and pure Python fallbacks.
|
||||
|
||||
.. moduleauthor:: OptimizR Contributors
|
||||
|
||||
"""
|
||||
|
||||
from optimizr.hmm import HMM
|
||||
from optimizr.core import (
|
||||
mcmc_sample,
|
||||
differential_evolution,
|
||||
grid_search,
|
||||
mutual_information,
|
||||
shannon_entropy,
|
||||
)
|
||||
|
||||
__version__ = "0.1.0"
|
||||
__all__ = [
|
||||
"HMM",
|
||||
"mcmc_sample",
|
||||
"differential_evolution",
|
||||
"grid_search",
|
||||
"mutual_information",
|
||||
"shannon_entropy",
|
||||
]
|
||||
@@ -0,0 +1,367 @@
|
||||
"""
|
||||
Core optimization functions with Rust acceleration
|
||||
"""
|
||||
|
||||
import warnings
|
||||
from typing import Callable, List, Tuple, Optional
|
||||
import numpy as np
|
||||
|
||||
# Try to import Rust backend
|
||||
try:
|
||||
from optimizr._core import (
|
||||
mcmc_sample as _rust_mcmc_sample,
|
||||
differential_evolution as _rust_differential_evolution,
|
||||
grid_search as _rust_grid_search,
|
||||
mutual_information as _rust_mutual_information,
|
||||
shannon_entropy as _rust_shannon_entropy,
|
||||
)
|
||||
RUST_AVAILABLE = True
|
||||
except ImportError:
|
||||
RUST_AVAILABLE = False
|
||||
warnings.warn(
|
||||
"Rust backend not available. Using pure Python fallbacks. "
|
||||
"Install with 'pip install optimizr' to enable Rust acceleration.",
|
||||
RuntimeWarning
|
||||
)
|
||||
|
||||
|
||||
def mcmc_sample(
|
||||
log_likelihood_fn: Callable[[List[float], List[float]], float],
|
||||
data: np.ndarray,
|
||||
initial_params: np.ndarray,
|
||||
param_bounds: List[Tuple[float, float]],
|
||||
n_samples: int = 10000,
|
||||
burn_in: int = 1000,
|
||||
proposal_std: float = 0.1,
|
||||
) -> np.ndarray:
|
||||
"""
|
||||
MCMC Metropolis-Hastings sampler.
|
||||
|
||||
Generates samples from a target distribution using the Metropolis-Hastings
|
||||
algorithm with Gaussian random walk proposals.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
log_likelihood_fn : callable
|
||||
Function that computes log P(data | params). Should accept
|
||||
(params: list, data: list) and return float.
|
||||
data : np.ndarray
|
||||
Observed data (passed to log_likelihood_fn)
|
||||
initial_params : np.ndarray
|
||||
Starting parameter values
|
||||
param_bounds : list of (float, float)
|
||||
[(min, max), ...] bounds for each parameter
|
||||
n_samples : int, default=10000
|
||||
Number of samples to generate (after burn-in)
|
||||
burn_in : int, default=1000
|
||||
Number of initial samples to discard
|
||||
proposal_std : float, default=0.1
|
||||
Standard deviation of Gaussian proposals
|
||||
|
||||
Returns
|
||||
-------
|
||||
samples : np.ndarray
|
||||
Array of shape (n_samples, n_params) with parameter samples
|
||||
|
||||
Examples
|
||||
--------
|
||||
>>> def log_likelihood(params, data):
|
||||
... mu, sigma = params
|
||||
... residuals = (data - mu) / sigma
|
||||
... return -0.5 * np.sum(residuals**2) - len(data) * np.log(sigma)
|
||||
>>> data = np.random.randn(100) + 2.0
|
||||
>>> samples = mcmc_sample(
|
||||
... log_likelihood_fn=log_likelihood,
|
||||
... data=data,
|
||||
... initial_params=np.array([0.0, 1.0]),
|
||||
... param_bounds=[(-10, 10), (0.1, 10)],
|
||||
... n_samples=10000,
|
||||
... burn_in=1000
|
||||
... )
|
||||
>>> print(f"Posterior mean: {np.mean(samples[:, 0]):.2f}")
|
||||
"""
|
||||
if RUST_AVAILABLE:
|
||||
samples = _rust_mcmc_sample(
|
||||
log_likelihood_fn=log_likelihood_fn,
|
||||
data=data.tolist(),
|
||||
initial_params=initial_params.tolist(),
|
||||
param_bounds=param_bounds,
|
||||
n_samples=n_samples,
|
||||
burn_in=burn_in,
|
||||
proposal_std=proposal_std,
|
||||
)
|
||||
return np.array(samples)
|
||||
else:
|
||||
# Pure Python fallback
|
||||
return _mcmc_sample_python(
|
||||
log_likelihood_fn, data, initial_params, param_bounds,
|
||||
n_samples, burn_in, proposal_std
|
||||
)
|
||||
|
||||
|
||||
def differential_evolution(
|
||||
objective_fn: Callable[[np.ndarray], float],
|
||||
bounds: List[Tuple[float, float]],
|
||||
popsize: int = 15,
|
||||
maxiter: int = 1000,
|
||||
f: float = 0.8,
|
||||
cr: float = 0.7,
|
||||
) -> Tuple[np.ndarray, float]:
|
||||
"""
|
||||
Differential Evolution global optimizer.
|
||||
|
||||
Population-based stochastic optimization effective for non-convex,
|
||||
multimodal objective functions.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
objective_fn : callable
|
||||
Function to minimize: f(x) -> float where x is np.ndarray
|
||||
bounds : list of (float, float)
|
||||
[(min, max), ...] bounds for each parameter
|
||||
popsize : int, default=15
|
||||
Population size multiplier (total size = popsize × n_params)
|
||||
maxiter : int, default=1000
|
||||
Maximum number of generations
|
||||
f : float, default=0.8
|
||||
Mutation factor (typically 0.5-2.0)
|
||||
cr : float, default=0.7
|
||||
Crossover probability (typically 0.1-0.9)
|
||||
|
||||
Returns
|
||||
-------
|
||||
x : np.ndarray
|
||||
Best parameters found
|
||||
fun : float
|
||||
Best objective value (minimum)
|
||||
|
||||
Examples
|
||||
--------
|
||||
>>> def rosenbrock(x):
|
||||
... return sum(100*(x[i+1] - x[i]**2)**2 + (1-x[i])**2
|
||||
... for i in range(len(x)-1))
|
||||
>>> result = differential_evolution(
|
||||
... objective_fn=rosenbrock,
|
||||
... bounds=[(-5, 5)] * 10,
|
||||
... popsize=15,
|
||||
... maxiter=1000
|
||||
... )
|
||||
>>> print(f"Minimum: {result[1]:.6f} at {result[0]}")
|
||||
"""
|
||||
if RUST_AVAILABLE:
|
||||
result = _rust_differential_evolution(
|
||||
objective_fn=objective_fn,
|
||||
bounds=bounds,
|
||||
popsize=popsize,
|
||||
maxiter=maxiter,
|
||||
f=f,
|
||||
cr=cr,
|
||||
)
|
||||
return np.array(result.x), result.fun
|
||||
else:
|
||||
# Pure Python fallback (scipy)
|
||||
try:
|
||||
from scipy.optimize import differential_evolution as scipy_de
|
||||
result = scipy_de(objective_fn, bounds=bounds, maxiter=maxiter,
|
||||
popsize=popsize, mutation=f, recombination=cr)
|
||||
return result.x, result.fun
|
||||
except ImportError:
|
||||
raise ImportError(
|
||||
"Rust backend not available and scipy not installed. "
|
||||
"Install scipy or build OptimizR with Rust support."
|
||||
)
|
||||
|
||||
|
||||
def grid_search(
|
||||
objective_fn: Callable[[np.ndarray], float],
|
||||
bounds: List[Tuple[float, float]],
|
||||
n_points: int = 10,
|
||||
) -> Tuple[np.ndarray, float]:
|
||||
"""
|
||||
Grid search optimizer.
|
||||
|
||||
Exhaustively evaluates objective function at all points on a regular grid.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
objective_fn : callable
|
||||
Function to maximize: f(x) -> float where x is np.ndarray
|
||||
bounds : list of (float, float)
|
||||
[(min, max), ...] bounds for each parameter
|
||||
n_points : int, default=10
|
||||
Number of grid points per dimension
|
||||
|
||||
Returns
|
||||
-------
|
||||
x : np.ndarray
|
||||
Best parameters found
|
||||
fun : float
|
||||
Best objective value (maximum)
|
||||
|
||||
Examples
|
||||
--------
|
||||
>>> def objective(x):
|
||||
... return -(x[0]**2 + x[1]**2) # Peak at (0, 0)
|
||||
>>> result = grid_search(
|
||||
... objective_fn=objective,
|
||||
... bounds=[(-5, 5), (-5, 5)],
|
||||
... n_points=50
|
||||
... )
|
||||
>>> print(f"Maximum: {result[1]:.6f} at {result[0]}")
|
||||
"""
|
||||
if RUST_AVAILABLE:
|
||||
result = _rust_grid_search(
|
||||
objective_fn=objective_fn,
|
||||
bounds=bounds,
|
||||
n_points=n_points,
|
||||
)
|
||||
return np.array(result.x), result.fun
|
||||
else:
|
||||
# Pure Python fallback
|
||||
return _grid_search_python(objective_fn, bounds, n_points)
|
||||
|
||||
|
||||
def mutual_information(
|
||||
x: np.ndarray,
|
||||
y: np.ndarray,
|
||||
n_bins: int = 10,
|
||||
) -> float:
|
||||
"""
|
||||
Compute mutual information between two variables.
|
||||
|
||||
I(X;Y) = H(X) + H(Y) - H(X,Y)
|
||||
|
||||
Measures how much knowing one variable reduces uncertainty about the other.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
x : np.ndarray
|
||||
Sample values from first variable
|
||||
y : np.ndarray
|
||||
Sample values from second variable (must be same length as x)
|
||||
n_bins : int, default=10
|
||||
Number of bins for histogram estimation
|
||||
|
||||
Returns
|
||||
-------
|
||||
mi : float
|
||||
Mutual information in nats (multiply by 1/ln(2) for bits)
|
||||
|
||||
Examples
|
||||
--------
|
||||
>>> x = np.random.randn(10000)
|
||||
>>> y = 2 * x + np.random.randn(10000) * 0.5
|
||||
>>> mi = mutual_information(x, y, n_bins=20)
|
||||
>>> print(f"MI: {mi:.4f} nats")
|
||||
"""
|
||||
if RUST_AVAILABLE:
|
||||
return _rust_mutual_information(x.tolist(), y.tolist(), n_bins=n_bins)
|
||||
else:
|
||||
# Pure Python fallback
|
||||
return _mutual_information_python(x, y, n_bins)
|
||||
|
||||
|
||||
def shannon_entropy(
|
||||
x: np.ndarray,
|
||||
n_bins: int = 10,
|
||||
) -> float:
|
||||
"""
|
||||
Compute Shannon entropy of a variable.
|
||||
|
||||
H(X) = -Σ p(x) log(p(x))
|
||||
|
||||
Quantifies the uncertainty/information content of a random variable.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
x : np.ndarray
|
||||
Sample values from the variable
|
||||
n_bins : int, default=10
|
||||
Number of bins for histogram estimation
|
||||
|
||||
Returns
|
||||
-------
|
||||
entropy : float
|
||||
Shannon entropy in nats (multiply by 1/ln(2) for bits)
|
||||
|
||||
Examples
|
||||
--------
|
||||
>>> x_uniform = np.random.uniform(0, 1, 10000)
|
||||
>>> h_uniform = shannon_entropy(x_uniform, n_bins=20)
|
||||
>>> x_peaked = np.random.normal(0, 0.1, 10000)
|
||||
>>> h_peaked = shannon_entropy(x_peaked, n_bins=20)
|
||||
>>> print(f"Uniform: {h_uniform:.4f}, Peaked: {h_peaked:.4f}")
|
||||
"""
|
||||
if RUST_AVAILABLE:
|
||||
return _rust_shannon_entropy(x.tolist(), n_bins=n_bins)
|
||||
else:
|
||||
# Pure Python fallback
|
||||
return _shannon_entropy_python(x, n_bins)
|
||||
|
||||
|
||||
# Pure Python fallback implementations
|
||||
def _mcmc_sample_python(log_likelihood_fn, data, initial_params, param_bounds,
|
||||
n_samples, burn_in, proposal_std):
|
||||
"""Pure Python MCMC implementation"""
|
||||
current_params = initial_params.copy()
|
||||
samples = []
|
||||
current_ll = log_likelihood_fn(current_params.tolist(), data.tolist())
|
||||
|
||||
for _ in range(n_samples + burn_in):
|
||||
# Propose
|
||||
proposed = current_params + np.random.randn(len(current_params)) * proposal_std
|
||||
for i, (low, high) in enumerate(param_bounds):
|
||||
proposed[i] = np.clip(proposed[i], low, high)
|
||||
|
||||
# Accept/reject
|
||||
proposed_ll = log_likelihood_fn(proposed.tolist(), data.tolist())
|
||||
if np.log(np.random.rand()) < proposed_ll - current_ll:
|
||||
current_params = proposed
|
||||
current_ll = proposed_ll
|
||||
|
||||
if len(samples) >= burn_in:
|
||||
samples.append(current_params.copy())
|
||||
|
||||
return np.array(samples)
|
||||
|
||||
|
||||
def _grid_search_python(objective_fn, bounds, n_points):
|
||||
"""Pure Python grid search implementation"""
|
||||
n_params = len(bounds)
|
||||
grids = [np.linspace(low, high, n_points) for low, high in bounds]
|
||||
|
||||
best_params = None
|
||||
best_score = float('-inf')
|
||||
|
||||
import itertools
|
||||
for point in itertools.product(*grids):
|
||||
score = objective_fn(np.array(point))
|
||||
if score > best_score:
|
||||
best_score = score
|
||||
best_params = np.array(point)
|
||||
|
||||
return best_params, best_score
|
||||
|
||||
|
||||
def _mutual_information_python(x, y, n_bins):
|
||||
"""Pure Python MI implementation"""
|
||||
hist_2d, x_edges, y_edges = np.histogram2d(x, y, bins=n_bins)
|
||||
|
||||
pxy = hist_2d / np.sum(hist_2d)
|
||||
px = np.sum(pxy, axis=1)
|
||||
py = np.sum(pxy, axis=0)
|
||||
|
||||
px_py = px[:, None] * py[None, :]
|
||||
|
||||
# Only compute where both are nonzero
|
||||
nonzero = (pxy > 0) & (px_py > 0)
|
||||
mi = np.sum(pxy[nonzero] * np.log(pxy[nonzero] / px_py[nonzero]))
|
||||
|
||||
return max(0.0, mi)
|
||||
|
||||
|
||||
def _shannon_entropy_python(x, n_bins):
|
||||
"""Pure Python entropy implementation"""
|
||||
hist, _ = np.histogram(x, bins=n_bins)
|
||||
probs = hist[hist > 0] / np.sum(hist)
|
||||
return -np.sum(probs * np.log(probs))
|
||||
@@ -0,0 +1,304 @@
|
||||
"""
|
||||
Hidden Markov Model implementation
|
||||
"""
|
||||
|
||||
import warnings
|
||||
from typing import Optional
|
||||
import numpy as np
|
||||
|
||||
# Try to import Rust backend
|
||||
try:
|
||||
from optimizr._core import fit_hmm as _rust_fit_hmm, viterbi_decode as _rust_viterbi
|
||||
RUST_AVAILABLE = True
|
||||
except ImportError:
|
||||
RUST_AVAILABLE = False
|
||||
|
||||
|
||||
class HMM:
|
||||
"""
|
||||
Hidden Markov Model for regime detection and sequence analysis.
|
||||
|
||||
Uses the Baum-Welch algorithm (EM) for parameter estimation and
|
||||
Viterbi algorithm for finding the most likely state sequence.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
n_states : int
|
||||
Number of hidden states
|
||||
|
||||
Attributes
|
||||
----------
|
||||
transition_matrix_ : np.ndarray or None
|
||||
Learned transition probabilities (n_states × n_states)
|
||||
emission_means_ : np.ndarray or None
|
||||
Mean of Gaussian emission for each state
|
||||
emission_stds_ : np.ndarray or None
|
||||
Std dev of Gaussian emission for each state
|
||||
|
||||
Examples
|
||||
--------
|
||||
>>> import numpy as np
|
||||
>>> from optimizr import HMM
|
||||
>>>
|
||||
>>> # Generate data with regime changes
|
||||
>>> returns = np.concatenate([
|
||||
... np.random.normal(0.01, 0.02, 500), # Bull market
|
||||
... np.random.normal(-0.01, 0.03, 500), # Bear market
|
||||
... ])
|
||||
>>>
|
||||
>>> # Fit HMM
|
||||
>>> hmm = HMM(n_states=2)
|
||||
>>> hmm.fit(returns, n_iterations=100)
|
||||
>>>
|
||||
>>> # Decode states
|
||||
>>> states = hmm.predict(returns)
|
||||
>>> print(f"Detected states: {np.unique(states)}")
|
||||
"""
|
||||
|
||||
def __init__(self, n_states: int = 2):
|
||||
if n_states < 2:
|
||||
raise ValueError("n_states must be at least 2")
|
||||
|
||||
self.n_states = n_states
|
||||
self.transition_matrix_: np.ndarray = np.zeros((n_states, n_states))
|
||||
self.emission_means_: np.ndarray = np.zeros(n_states)
|
||||
self.emission_stds_: np.ndarray = np.ones(n_states)
|
||||
self._params = None
|
||||
|
||||
def fit(self, X: np.ndarray, n_iterations: int = 100, tolerance: float = 1e-6) -> 'HMM':
|
||||
"""
|
||||
Fit HMM parameters using Baum-Welch algorithm.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
X : np.ndarray
|
||||
Time series observations (1D array)
|
||||
n_iterations : int, default=100
|
||||
Maximum number of EM iterations
|
||||
tolerance : float, default=1e-6
|
||||
Convergence threshold for log-likelihood change
|
||||
|
||||
Returns
|
||||
-------
|
||||
self : HMM
|
||||
Fitted model
|
||||
"""
|
||||
X = np.asarray(X).flatten()
|
||||
|
||||
if len(X) == 0:
|
||||
raise ValueError("X cannot be empty")
|
||||
|
||||
if RUST_AVAILABLE:
|
||||
# Use Rust implementation
|
||||
self._params = _rust_fit_hmm(
|
||||
observations=X.tolist(),
|
||||
n_states=self.n_states,
|
||||
n_iterations=n_iterations,
|
||||
tolerance=tolerance
|
||||
)
|
||||
|
||||
self.transition_matrix_ = np.array(self._params.transition_matrix)
|
||||
self.emission_means_ = np.array(self._params.emission_means)
|
||||
self.emission_stds_ = np.array(self._params.emission_stds)
|
||||
else:
|
||||
# Pure Python fallback
|
||||
warnings.warn(
|
||||
"Rust backend not available. Using slower Python implementation.",
|
||||
RuntimeWarning
|
||||
)
|
||||
self._fit_python(X, n_iterations, tolerance)
|
||||
|
||||
return self
|
||||
|
||||
def predict(self, X: np.ndarray) -> np.ndarray:
|
||||
"""
|
||||
Predict most likely state sequence using Viterbi algorithm.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
X : np.ndarray
|
||||
Time series observations (1D array)
|
||||
|
||||
Returns
|
||||
-------
|
||||
states : np.ndarray
|
||||
Most likely state at each time step
|
||||
"""
|
||||
if self.transition_matrix_ is None:
|
||||
raise ValueError("Model must be fitted before prediction")
|
||||
|
||||
X = np.asarray(X).flatten()
|
||||
|
||||
if RUST_AVAILABLE and self._params is not None:
|
||||
states = _rust_viterbi(X.tolist(), self._params)
|
||||
return np.array(states)
|
||||
else:
|
||||
return self._viterbi_python(X)
|
||||
|
||||
def score(self, X: np.ndarray) -> float:
|
||||
"""
|
||||
Compute log-likelihood of observations.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
X : np.ndarray
|
||||
Time series observations
|
||||
|
||||
Returns
|
||||
-------
|
||||
log_likelihood : float
|
||||
Log P(X | model)
|
||||
"""
|
||||
if self.transition_matrix_ is None:
|
||||
raise ValueError("Model must be fitted before scoring")
|
||||
|
||||
X = np.asarray(X).flatten()
|
||||
alpha = self._forward_python(X)
|
||||
return np.log(np.sum(alpha[-1]))
|
||||
|
||||
def _fit_python(self, X: np.ndarray, n_iterations: int, tolerance: float):
|
||||
"""Pure Python implementation of Baum-Welch"""
|
||||
n_obs = len(X)
|
||||
|
||||
# Initialize parameters
|
||||
self.transition_matrix_ = np.ones((self.n_states, self.n_states)) / self.n_states
|
||||
|
||||
# Initialize emissions based on quantiles
|
||||
quantiles = np.linspace(0, 1, self.n_states + 1)
|
||||
self.emission_means_ = np.zeros(self.n_states)
|
||||
self.emission_stds_ = np.ones(self.n_states)
|
||||
|
||||
for i in range(self.n_states):
|
||||
mask = (X >= np.quantile(X, quantiles[i])) & (X < np.quantile(X, quantiles[i+1]))
|
||||
if np.any(mask):
|
||||
self.emission_means_[i] = np.mean(X[mask])
|
||||
self.emission_stds_[i] = max(np.std(X[mask]), 1e-6)
|
||||
|
||||
prev_ll = float('-inf')
|
||||
|
||||
# EM iterations
|
||||
for _ in range(n_iterations):
|
||||
# E-step
|
||||
alpha = self._forward_python(X)
|
||||
beta = self._backward_python(X)
|
||||
gamma = self._compute_gamma_python(alpha, beta)
|
||||
xi = self._compute_xi_python(X, alpha, beta)
|
||||
|
||||
# M-step
|
||||
self._update_parameters_python(X, gamma, xi)
|
||||
|
||||
# Check convergence
|
||||
ll = np.log(np.sum(alpha[-1]))
|
||||
if abs(ll - prev_ll) < tolerance:
|
||||
break
|
||||
prev_ll = ll
|
||||
|
||||
def _forward_python(self, X: np.ndarray) -> np.ndarray:
|
||||
"""Forward algorithm"""
|
||||
n_obs = len(X)
|
||||
alpha = np.zeros((n_obs, self.n_states))
|
||||
|
||||
# Initialize
|
||||
for s in range(self.n_states):
|
||||
alpha[0, s] = (1.0 / self.n_states) * self._emission_prob(X[0], s)
|
||||
|
||||
alpha[0] /= np.sum(alpha[0])
|
||||
|
||||
# Recursion
|
||||
for t in range(1, n_obs):
|
||||
for s in range(self.n_states):
|
||||
alpha[t, s] = np.sum(alpha[t-1] * self.transition_matrix_[:, s]) * self._emission_prob(X[t], s)
|
||||
alpha[t] /= max(np.sum(alpha[t]), 1e-10)
|
||||
|
||||
return alpha
|
||||
|
||||
def _backward_python(self, X: np.ndarray) -> np.ndarray:
|
||||
"""Backward algorithm"""
|
||||
n_obs = len(X)
|
||||
beta = np.zeros((n_obs, self.n_states))
|
||||
beta[-1] = 1.0
|
||||
|
||||
for t in range(n_obs - 2, -1, -1):
|
||||
for s in range(self.n_states):
|
||||
beta[t, s] = np.sum(
|
||||
self.transition_matrix_[s] *
|
||||
np.array([self._emission_prob(X[t+1], s2) for s2 in range(self.n_states)]) *
|
||||
beta[t+1]
|
||||
)
|
||||
beta[t] /= max(np.sum(beta[t]), 1e-10)
|
||||
|
||||
return beta
|
||||
|
||||
def _compute_gamma_python(self, alpha: np.ndarray, beta: np.ndarray) -> np.ndarray:
|
||||
"""Compute state occupation probabilities"""
|
||||
gamma = alpha * beta
|
||||
gamma /= np.sum(gamma, axis=1, keepdims=True)
|
||||
return gamma
|
||||
|
||||
def _compute_xi_python(self, X: np.ndarray, alpha: np.ndarray, beta: np.ndarray) -> np.ndarray:
|
||||
"""Compute transition probabilities"""
|
||||
n_obs = len(X)
|
||||
xi = np.zeros((n_obs - 1, self.n_states, self.n_states))
|
||||
|
||||
for t in range(n_obs - 1):
|
||||
for i in range(self.n_states):
|
||||
for j in range(self.n_states):
|
||||
xi[t, i, j] = (alpha[t, i] * self.transition_matrix_[i, j] *
|
||||
self._emission_prob(X[t+1], j) * beta[t+1, j])
|
||||
xi[t] /= max(np.sum(xi[t]), 1e-10)
|
||||
|
||||
return xi
|
||||
|
||||
def _update_parameters_python(self, X: np.ndarray, gamma: np.ndarray, xi: np.ndarray):
|
||||
"""M-step: update parameters"""
|
||||
# Update transitions
|
||||
for i in range(self.n_states):
|
||||
denom = np.sum(gamma[:-1, i])
|
||||
for j in range(self.n_states):
|
||||
numer = np.sum(xi[:, i, j])
|
||||
self.transition_matrix_[i, j] = numer / max(denom, 1e-10)
|
||||
|
||||
# Update emissions
|
||||
for s in range(self.n_states):
|
||||
weights = gamma[:, s]
|
||||
sum_weights = np.sum(weights)
|
||||
|
||||
if sum_weights > 1e-10:
|
||||
self.emission_means_[s] = np.sum(weights * X) / sum_weights
|
||||
self.emission_stds_[s] = max(
|
||||
np.sqrt(np.sum(weights * (X - self.emission_means_[s])**2) / sum_weights),
|
||||
1e-6
|
||||
)
|
||||
|
||||
def _emission_prob(self, obs: float, state: int) -> float:
|
||||
"""Gaussian emission probability"""
|
||||
mean = self.emission_means_[state]
|
||||
std = self.emission_stds_[state]
|
||||
z = (obs - mean) / std
|
||||
return max(np.exp(-0.5 * z**2) / (std * np.sqrt(2 * np.pi)), 1e-10)
|
||||
|
||||
def _viterbi_python(self, X: np.ndarray) -> np.ndarray:
|
||||
"""Viterbi decoding"""
|
||||
n_obs = len(X)
|
||||
delta = np.full((n_obs, self.n_states), float('-inf'))
|
||||
psi = np.zeros((n_obs, self.n_states), dtype=int)
|
||||
|
||||
# Initialize
|
||||
for s in range(self.n_states):
|
||||
delta[0, s] = np.log(1.0 / self.n_states) + np.log(self._emission_prob(X[0], s))
|
||||
|
||||
# Recursion
|
||||
for t in range(1, n_obs):
|
||||
for s in range(self.n_states):
|
||||
trans_probs = delta[t-1] + np.log(self.transition_matrix_[:, s] + 1e-10)
|
||||
psi[t, s] = np.argmax(trans_probs)
|
||||
delta[t, s] = trans_probs[psi[t, s]] + np.log(self._emission_prob(X[t], s))
|
||||
|
||||
# Backtrack
|
||||
path = np.zeros(n_obs, dtype=int)
|
||||
path[-1] = np.argmax(delta[-1])
|
||||
|
||||
for t in range(n_obs - 2, -1, -1):
|
||||
path[t] = psi[t + 1, path[t + 1]]
|
||||
|
||||
return path
|
||||
Reference in New Issue
Block a user