Create advanced_inference.py

advanced_inference.py

@@ -0,0 +1,73 @@
+"""
+Hamiltonian Monte Carlo (NUTS) for complex pattern discovery.
+Uses Pyro with NUTS and ArviZ for visualization.
+"""
+import logging
+import pyro
+import pyro.distributions as dist
+from pyro.infer import MCMC, NUTS
+import torch
+import numpy as np
+import pandas as pd
+from typing import Dict, Any, Optional
+
+logger = logging.getLogger(__name__)
+
+class HMCAnalyzer:
+    """Runs HMC on a simple regression model to demonstrate advanced inference."""
+
+    def __init__(self):
+        self.mcmc = None
+        self.trace = None
+
+    def _model(self, x, y=None):
+        # Linear regression with unknown noise
+        alpha = pyro.sample("alpha", dist.Normal(0, 10))
+        beta = pyro.sample("beta", dist.Normal(0, 1))
+        sigma = pyro.sample("sigma", dist.HalfNormal(1))
+        mu = alpha + beta * x
+        with pyro.plate("data", len(x)):
+            pyro.sample("obs", dist.Normal(mu, sigma), obs=y)
+
+    def run_inference(self, data: Optional[pd.DataFrame] = None, num_samples: int = 500, warmup: int = 200):
+        """
+        Run HMC on synthetic or provided data.
+        If no data, generate synthetic trend data.
+        """
+        if data is None:
+            # Create synthetic data: a linear trend with noise
+            x = torch.linspace(0, 10, 50)
+            true_alpha, true_beta, true_sigma = 2.0, -0.3, 0.5
+            y = true_alpha + true_beta * x + torch.randn(50) * true_sigma
+        else:
+            # Assume data has columns 'x' and 'y'
+            x = torch.tensor(data['x'].values, dtype=torch.float32)
+            y = torch.tensor(data['y'].values, dtype=torch.float32)
+
+        nuts_kernel = NUTS(self._model)
+        self.mcmc = MCMC(nuts_kernel, num_samples=num_samples, warmup_steps=warmup)
+        self.mcmc.run(x, y)
+
+        self.trace = self.mcmc.get_samples()
+        return self._summary()
+
+    def _summary(self) -> Dict[str, Any]:
+        """Return summary statistics of posterior samples."""
+        if self.trace is None:
+            return {}
+        summary = {}
+        for key in ['alpha', 'beta', 'sigma']:
+            samples = self.trace[key].numpy()
+            summary[key] = {
+                'mean': float(samples.mean()),
+                'std': float(samples.std()),
+                'hpd_5': float(np.percentile(samples, 5)),
+                'hpd_95': float(np.percentile(samples, 95))
+            }
+        return summary
+
+    def get_trace_data(self) -> Dict[str, np.ndarray]:
+        """Return posterior samples for plotting."""
+        if self.trace is None:
+            return {}
+        return {k: v.numpy() for k, v in self.trace.items()}