Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 62 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
name: CI

on:
push:
branches: [main]
pull_request:
branches: [main]

permissions:
contents: read

jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.9", "3.10", "3.11", "3.12"]

steps:
- uses: actions/checkout@v4

- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}

- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e ".[dev]"

- name: Run tests
run: pytest tests/ -v

build:
runs-on: ubuntu-latest
needs: test

steps:
- uses: actions/checkout@v4

- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.12"

- name: Install build tools
run: python -m pip install --upgrade pip build

- name: Build distributions
run: python -m build

- name: Verify wheel installs cleanly
run: |
pip install dist/*.whl
python -c "from engine import JumpDiffusionEngine; print('import OK')"

- name: Upload distributions
uses: actions/upload-artifact@v4
with:
name: dist
path: dist/
23 changes: 23 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
# Build / distribution artifacts
*.egg-info/
build/
dist/

# Python cache
__pycache__/
*.py[cod]
*.pyo

# Test / coverage artifacts
.pytest_cache/
.coverage
htmlcov/

# Virtual environments
.venv/
venv/
env/

# Editor / OS
.DS_Store
*.swp
1 change: 1 addition & 0 deletions CITATION.cff
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
cff-version: 1.2.0
message: "If you use this software, please cite it as below."
title: "Jump Diffusion Engine"
version: "0.1.0"
abstract: "A universal framework for multistable stochastic control."
type: software
authors:
Expand Down
79 changes: 60 additions & 19 deletions README.md
Original file line number Diff line number Diff line change
@@ -1,47 +1,88 @@
# Jump-Diffusion-Engine – Perfectly Balanced Stochastic Control
**One set of inputs. Continuous, self-regulating outputs.**
# Jump-Diffusion-Engine – Stochastic Stability Analysis
**Simulate, analyse, and steer noisy nonlinear systems.**

A universal stability port for any dynamic system with a Source (Λ(t)), a Medium (Δ), and a Sink (f(Δ)).
A simulation framework for systems with a Source (Λ(t)), a Medium (Δ), and a nonlinear Sink (f(Δ))
subject to continuous diffusion and discrete jump noise.

## Core Equation

`dΔ = [Λ(t) − f(Δ)] dt + σ dW + J dN`

- `Λ(t) = ε₀ + A·sin(ωt)` — the source, steady and singing
- `f(Δ) = kΔ + gΔ²/(K²+Δ²)` — the sink, linear then saturating
- `Δ* : Λ = f(Δ), f′(Δ*) > 0` — the bowl, the stable held place
- `Λ(t) = ε₀ + A·sin(ωt)` — the source (constant or time-varying)
- `f(Δ) = kΔ + gΔ²/(K²+Δ²)` — the nonlinear sink (linear + saturating)
- `Δ* : Λ = f(Δ), f′(Δ*) > 0` — a stable equilibrium (basin centre)

Use `engine.py` to **force** any stochastic system into its stable basin, then **verify** its resilience:
Use `engine.py` to **analyse** stochastic systems and **steer** trajectories toward stable basins:

| # | Action | Method | What it does |
|:-|:---|:---|:---|
| 1 | **FORCE** it into the bowl | `seat_and_release()` | Applies a transient push, then **releases control to zero**. The basin holds it forever. |
| 2 | **BREATHE** with the noise | `adaptive_k()` | Dynamically stiffens when calm, relaxes when volatile. Prevents numerical blow-up while maximizing rejection. |
| 3 | **MAP** where the bowls are | `find_fixed_points()` | Finds all stable (`f′>0`) and unstable equilibria before you force it. |
| 4 | **MEASURE** how deep the bowl is | `basin_depth()` | Quantifies the energy barrier—how hard you'd have to push to knock it out. |
| 5 | **PREDICT** escape risk | `escape_probability()` | Empirical escape rate over Monte Carlo trials. Should be near-zero after seating. |
| 6 | **VISUALIZE** the long-term cloud | `stationary_density()` | The normalized PDF \( p(\Delta) \propto e^{-2V/\sigma^2} \)—where it lives once forced. |
| 1 | **Seat** into a stable basin | `seat_and_release()` | Applies a transient corrective push, then releases control. Basin strength determines how well it holds. |
| 2 | **Adapt** to volatility | `adaptive_k()` | Updates the reversion coefficient based on recent variance. |
| 3 | **Map** equilibria | `find_fixed_points()` | Finds stable (`f′>0`) and unstable fixed points for a given Λ. |
| 4 | **Measure** basin depth | `basin_depth()` | Quantifies the potential-energy barrier around each basin. |
| 5 | **Estimate** escape risk | `escape_probability()` | Empirical escape rate via Monte Carlo trials. |
| 6 | **Visualise** the stationary distribution | `stationary_density()` | Computes the Boltzmann-weighted PDF p(Δ) ∝ e^{2V/σ²}. |

## Installation

This project requires Python 3 and the libraries used by `engine.py`:
**Standard install (recommended)**

- `numpy`
- `scipy`
- `matplotlib`
```bash
pip install -e .
```

This installs the package and all dependencies from the root of the repository.

Install them with pip:
**Manual dependency install**

If you prefer not to use the package install, install dependencies directly:

```bash
pip install numpy scipy matplotlib
```

Then import the engine in your Python code:
Then add the repository root to your Python path and import:

```python
from engine import JumpDiffusionEngine
```

> **Note:** A legacy `packaging/pyproject.toml` is also present in the repository.
> For standard `pip install -e .` use the root `pyproject.toml`; the `packaging/`
> directory is kept for historical reference and may be removed in a future release.

## Quick Start

```python
import numpy as np
from engine import JumpDiffusionEngine

# Define a source: constant with a small oscillation
def lambda_func(t):
return 0.5 + 0.1 * np.sin(2 * np.pi * 0.1 * t)

eng = JumpDiffusionEngine(lambda_func, sigma=0.3, jump_rate=0.05, seed=42)

# 1. Find stable equilibria
fps = eng.find_fixed_points(lambda_val=0.5)
print("Fixed points:", fps)

# 2. Simulate a few trajectories
results = eng.simulate(t_max=20.0, x0=0.0, n_realizations=3)
eng.plot_trajectories(results)

# 3. Steer into a stable basin, then release control
result = eng.seat_and_release(t_max=15.0, x0=3.0, lambda_val=0.5)
print(f"Released at step {result['release_idx']}, seated at Δ* ≈ {result['x_star']:.3f}")

# 4. Estimate escape probability from the basin
p_escape = eng.escape_probability(threshold=2.0, t_max=10.0, n_trials=200)
print(f"Empirical escape probability: {p_escape:.3f}")

# 5. Stationary density
x, p = eng.stationary_density(lambda_val=0.5)
```

## Use Cases

Use this on any system that needs to maintain a steady state while being bombarded by both constant noise and sudden, unpredictable shocks.
Expand Down
Binary file added __pycache__/engine.cpython-312.pyc
Binary file not shown.
48 changes: 44 additions & 4 deletions engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -469,7 +469,10 @@ def seat_and_release(self, t_max: float, x0: float = 0.0,
def stationary_density(self, lambda_val: float, x_range: Tuple[float, float] = (-10, 10), n_points: int = 2000):
x = np.linspace(x_range[0], x_range[1], n_points)
V = self.potential(x, lambda_val)
p = np.exp(-2 * V / (self.sigma ** 2 + 1e-12))
# Shift V by its minimum before exponentiation to avoid numerical overflow.
# The constant shift cancels exactly in the normalisation.
V_shifted = V - np.min(V)
p = np.exp(-2 * V_shifted / (self.sigma ** 2 + 1e-12))
norm = np.trapezoid(p, x) if hasattr(np, 'trapezoid') else np.trapz(p, x)
p /= (norm + 1e-12)
return x, p
Expand Down Expand Up @@ -567,10 +570,47 @@ def _plot_sweep(self, results: List[Dict], param_name: str, metrics: List[str]):
plt.tight_layout()
plt.show()

# Plotting helpers (plot_trajectories, plot_potential, basin_analysis) unchanged
def plot_trajectories(self, results: List[Dict], title: str = "Jump-Diffusion Trajectories", show_energy: bool = False):
# ... (standard implementation as before)
pass # Replace with full from previous if needed
"""Plot simulated trajectories from simulate().

Parameters
----------
results : list of dicts returned by simulate()
title : figure title
show_energy : if True and energy was recorded, add a second panel
"""
has_energy = show_energy and any('energy' in r for r in results)
n_panels = 2 if has_energy else 1
fig, axes = plt.subplots(n_panels, 1, figsize=(10, 4 * n_panels), squeeze=False)
ax_traj = axes[0, 0]

for i, r in enumerate(results):
label = f"Run {i + 1}" if len(results) > 1 else None
ax_traj.plot(r['t'], r['x'], lw=0.8, alpha=0.7, label=label)

ax_traj.set_xlabel("Time")
ax_traj.set_ylabel("Δ")
ax_traj.set_title(title)
ax_traj.grid(True, alpha=0.3)
if len(results) > 1:
ax_traj.legend(fontsize=8)

if has_energy:
ax_en = axes[1, 0]
for i, r in enumerate(results):
if 'energy' in r:
label = f"Run {i + 1}" if len(results) > 1 else None
ax_en.plot(r['t'], r['energy'], lw=0.8, alpha=0.7, label=label)
ax_en.set_xlabel("Time")
ax_en.set_ylabel("Energy V(Δ)")
ax_en.set_title("Potential Energy")
ax_en.grid(True, alpha=0.3)
if len(results) > 1:
ax_en.legend(fontsize=8)

plt.tight_layout()
plt.show()
return fig

# Quick test / usage
if __name__ == "__main__":
Expand Down
Loading
Loading