Streaming distributed optical Monte-Carlo with converging sketches and bounded output

[Areustle]

Summary

Adds a streaming, distributed optical Monte-Carlo path beside the existing one-shot compute(), exposed as a new nuspacesim stream command. Events are thrown in batches across a dask cluster; the MC integral accumulates as an exact additive sketch with live statistics, the run stops when it reaches a target precision, and only a bounded weighted sample of events is written to disk.

Motivated by three limits of the fixed-N path: memory grows with thrown_events (every valid event is retained), the integral's precision is only known after the run completes, and a long run gives no feedback while in flight.

nuspacesim stream --rel-unc 0.01 --reservoir-size 5000 sample_input_file.toml

compute() and the radio/ToO paths are untouched; this is purely additive.

Why the integral streams exactly

Every optical quantity RegionGeom.mcintegral reports is a linear functional of a per-event contribution scaled by config-only constants: np.sum, np.var, np.count_nonzero over the events, divided by total thrown. mcnorm is config-derived, and both spec_norm and sum_spec_weights are closed-form constants (the latter despite its name — it is not a sum over the drawn sample). There is therefore no hidden global normalization coupling batches, and the moment accumulators (n_thrown, n_valid, s1, s2, s1_geo, n_pass) merge additively with no loss of precision versus a one-shot run.

Validated: streaming vs compute() optical integral agree within 1.22σ at 200k thrown (ratio 1.017 — pure MC noise).

What's in it

Module	Role
simulation/streaming/sketch.py	MomentSketch — exact additive integral accumulator
simulation/streaming/reservoir.py	WeightedReservoir — Efraimidis–Spirakis bottom-k sample
simulation/streaming/batch.py	run_batch — the per-batch worker unit
simulation/streaming/driver.py	compute_streaming — sliding-window dask loop, live panel, stop criteria
geometry/region_geometry.py	mc_contribution() extracted (shared with the one-shot integral)

Stop criteria (first to trip): relative-uncertainty target with a min_thrown warmup, thrown-count backstop (count, default infinity), wall-clock limit, or a spacebar press — which ends the run early but gracefully, finalizing the partial result rather than aborting like Ctrl-C.

Batch sizing adapts to the throughput knee by default; --batch-size N fixes it for a reproducible schedule.

Notes for review

• Log-domain sampling key. The Efraimidis–Spirakis key is computed as log(u)/w, not the textbook u**(1/w). Optical contributions are ~1e-5, so 1/w is ~1e5 and the direct form underflows to 0.0 for nearly every event, collapsing the sample. The log form is order-equivalent, underflow-free, and invariant to global weight rescaling. This was a real bug caught by the end-to-end test.
• Parallelism moved up a level. EAS.__call__ gains serial=True so the optical kernel runs serially inside a worker (a nested cluster there would be pathological); the cluster uses threads_per_worker=1 so the per-batch seeding of the legacy global np.random is race-free.
• Reproducibility. Per-batch RNG derives from SeedSequence(entropy, spawn_key=(batch_idx,)), so batch i yields the same events and keys regardless of scheduling. The materialized reservoir is reproducible; only the float-sum order in the sketch varies (harmlessly). Adaptive batch sizing is timing-dependent and therefore not bit-reproducible — use --batch-size when that matters.
• mc_contribution extraction is regression-locked: its test oracle is a verbatim copy of the pre-refactor mcintegral body, asserting bit-exact equality.
• Scope: optical only. Radio and Target-of-Opportunity are deferred (RegionGeomToO divides by a fixed time grid — a different streaming mode). Threading explicit RNG generators through the physics stages is also deferred.

Verification

• 208 tests pass (35 new); black/isort/flake8/mypy/bandit/vulture/codespell clean on the new code.
• Sketch equality vs a real RegionGeom.mcintegral is bit-exact on the linear terms; reservoir merge is order-independent, uniform-weight marginals pass a KS test.
• End-to-end CLI runs verified, including the spacebar interrupt and multi-million-throw runs.

🤖 Generated with Claude Code

[Areustle]

Regression check against current main (post #124 + #127)

Rebased onto main @ a745b13 (clean, no conflicts). Since the branch's previous base, main changed only CI/packaging files (.github/workflows/*, .pre-commit-config.yaml, pyproject.toml, setup.cfg); this branch touches none of them — zero file overlap, so the merge cannot revert anything on main.

Behavioral A/B of the existing one-shot path. Ran a seeded compute() (np.random.seed(20260910), 100k thrown, main's own sample_input_file.toml — no [simulation.streaming] block, optical + radio enabled) on a clean worktree of main and on this branch, then compared the full result table:

• All 16 columns bit-identical across 99,205 events — including the radio EFields (99205, 27) matrix, since radio shares RegionGeom.mcintegral whose per-event math this branch extracts into mc_contribution().
• All 8 scalar integrals identical: OMCINT, OMCINTGO, OMCINTUN, ONEVPASS, RMCINT, RMCINTGO, RMCINTUN, RNEVPASS.
• Loading main's config file unchanged also confirms the new optional [simulation.streaming] section is genuinely backward compatible.

Tests: 208 pass on the rebased branch; main's existing tests are modified only additively (0 deleted lines — test_config.py +52, test_cli.py +20); pre-commit all green.

Net: the streaming work is purely additive to the existing pipeline — compute() output is byte-for-byte unchanged.