Commit 1ee31355 authored by Gaetano Scandariato's avatar Gaetano Scandariato
Browse files

feat: Refactor runner.py for improved performance and code clarity; update analysis documentation

parent ee04206c
Loading
Loading
Loading
Loading
Loading
+82 −0
Changes for .kilo/runner_refactor_plan.md: 82 added lines, 0 removed lines.
Original line number Diff line number Diff line
# PyAETNA `runner.py` Refactor Plan

## Goal
Create a focused refactor of `PyAETNA/sampling/engine/runner.py` to:
1. remove redundant or duplicate code,
2. remove dead variables,
3. remove useless comments,
4. fix incorrect logic,
5. improve performance.

---

## 1. Remove Redundant / Duplicate Code

- Eliminate duplicated config reads and overriding assignments:
  - Remove the second `checkpoint_interval = mcmc_config.get('checkpoint_interval', 100)`.
  - Remove the second `max_tree_depth = mcmc_config.get('max_tree_depth', 6)`.
  - Remove the second `target_accept_prob = mcmc_config.get('target_accept_prob', 0.9)`.
- Keep the one validated assignment for each value and drop the later fallback read.
- Replace the early `num_chains_if_nuts = mcmc_config_peek.get('num_chains_if_nuts', 4)` with a separate hint variable, e.g. `num_chains_if_nuts_hint`, and use it only for early device-count estimation.
- Remove the unnecessary intermediate `ensemble_mcmc` object in direct ESS/AIES production mode. Use `_unified_ensemble_state` directly for `batch_mcmc.post_warmup_state` and `batch_mcmc._warmup_state`.
- Move `dense_mass = mcmc_config.get('dense_mass', False)` into the validated config-read block for consistency.

## 2. Remove Dead Variables

- Replace all unused `_jit_time` unpackings from `run_batched(...)` with `_`.
- Replace `warmup_energy = np.zeros((num_chains, num_warmup))` with a smaller representation, because only first and last warmup energies are used.
- Eliminate the redundant `samples_grouped = {}` reset when a checkpoint is incompatible.
- Inline `num_cores` into its log statement, since it is only used once.
- Remove `ensemble_mcmc.last_state` assignment in direct ensemble mode if it is not guaranteed to be populated.
- Replace `_pilot_elapsed = _unified_jit_elapsed if _unified_jit_elapsed else 0.0` with a direct expression where used.

## 3. Remove Useless Comments

- Remove overly verbose or redundant comments such as:
  - configuration file path resolution notes that merely restate the code
  - `reset was already determined above`
  - `Just return the existing data`
  - `Reverting the failing line.`
  - repeated explanatory blocks about constrained-space samples and postprocess behavior
  - `Note: Raw sample summary is now printed in reports.py` unless it is needed for developer context
- Preserve only comments that document non-obvious behavior and control-flow decisions.

## 4. Fix Incorrect Logic

- Fix the `NameError` bug in the dense mass logging block: replace `axis1= unfolded` with `axis1=1`.
- Fix the warmup SKIP branch so it runs a single pilot `ensemble_mcmc.run(...)` before calling `ensemble_mcmc.get_samples(group_by_chain=True)`.
- Fix direct ensemble production so `batch_mcmc.post_warmup_state` uses `_unified_ensemble_state` directly instead of `ensemble_mcmc.last_state`.
- Ensure `checkpoint_interval`, `max_tree_depth`, and `target_accept_prob` are not silently overwritten after validation.
- Adjust warmup pilot RNG handling so it does not reuse the same key on resumed warmup runs.

## 5. Improve Performance

- Replace repeated warmup `np.concatenate` growth with batch accumulation lists and a single final `np.concatenate`.
- Consolidate repeated NUTS adaptation logging into a helper function to remove duplication and keep logging bug fixes in one place.
- Consider batching the Phase 2 energy computation with `jax.vmap` if `_compute_log_joint` can be vectorized, avoiding a Python loop over chains.
- Remove unnecessary repeated late imports or move them to the module-level import block where safe.
- Document the JAX device-count over-provision strategy so future maintainers understand the pre-initialization constraint.

---

## Implementation Notes

- Prefer smaller, atomic changes in the refactor:
  1. fix config duplicate-read logic,
  2. repair warmup SKIP/run state handling,
  3. remove dead variables and comments,
  4. consolidate repeated NUTS logging,
  5. optimize warmup sample accumulation.
- Keep the existing structure of the auto/NUTS/ensemble pipeline intact while removing the redundant blocks that cause bugs and confusion.
- Validate the file with `python3 -m py_compile PyAETNA/sampling/engine/runner.py` after each major change.

---

## Success Criteria

- `runner.py` contains no duplicate validated config reads.
- No unused local variables remain in the hot code path.
- The warmup SKIP/resume paths no longer fail on `get_samples()` with `NoneType` state.
- Dense mass logging no longer throws `NameError`.
- The direct ensemble production path preserves the JIT-initialized state correctly.
- Warmup accumulation is more efficient and avoids O(N²) concatenation overhead.
+2 −2
Changes for PyAETNA/kernels/factory.py: 2 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -442,7 +442,7 @@ def _build_spectral_mixture_params(params, spec, inst_suffix, section_name, per_
    for p in param_names:
        if re.match(r'(log_)?(gp_)?sigma', p) or re.match(r'(log_)?sigma', p):
            sigma_param_names.append(p)
        elif re.match(r'(log_)?(gp_)?lengthscale', p) or re.match(r'(log_)?lengthscale', p):
        elif re.match(r'(log_)?(gp_)?(lengthscale|scale)', p) or re.match(r'(log_)?(lengthscale|scale)', p):
            ls_param_names.append(p)

    # Process parameters with specific logic for SpectralMixture
@@ -586,7 +586,7 @@ def _count_spectral_mixture_components(params, spec, inst_suffix, section_name,
    for p in param_names:
        if re.match(r'(log_)?(gp_)?sigma', p) or re.match(r'(log_)?sigma', p):
            sigma_param_names.append(p)
        elif re.match(r'(log_)?(gp_)?lengthscale', p) or re.match(r'(log_)?lengthscale', p):
        elif re.match(r'(log_)?(gp_)?(lengthscale|scale)', p) or re.match(r'(log_)?(lengthscale|scale)', p):
            ls_param_names.append(p)
    
    # Count sigmas and lengthscales in the params dict
+31 −25
Changes for PyAETNA/sampling/engine/runner.py: 31 added lines, 25 removed lines.
Original line number Diff line number Diff line
@@ -275,8 +275,7 @@ def runSampler_NumPyro(configFile, dataFile, logFile, num_chains=None, outDir='.
                "Example: mcmc: { walkers_per_free_parameter: 3 }  # → 3 × n_params walkers"
            )
    
    num_chains_if_nuts = mcmc_config_peek.get('num_chains_if_nuts', 4)
    num_cores = max(cpu_count() - 2, 1)
    num_chains_if_nuts_hint = mcmc_config_peek.get('num_chains_if_nuts', 4)
    
    # Determine num_chains (= ensemble walker count for fresh runs, or pickle value for resume).
    # For fresh starts the actual walker count is computed later (after n_params is known)
@@ -299,7 +298,7 @@ def runSampler_NumPyro(configFile, dataFile, logFile, num_chains=None, outDir='.
        _est_max_walkers = int(_math.ceil(walkers_coefficient * 200))
    else:
        _est_max_walkers = num_chains if num_chains is not None else 128
    numpyro.set_host_device_count(max(_est_max_walkers, num_chains_if_nuts))
    numpyro.set_host_device_count(max(_est_max_walkers, num_chains_if_nuts_hint))
    
    # GPU/CPU detection and chain method
    # In JAX ≤0.6, CUDA device_kind is 'gpu'; in JAX ≥0.9, it's 'cuda'.
@@ -318,22 +317,17 @@ def runSampler_NumPyro(configFile, dataFile, logFile, num_chains=None, outDir='.
        device_type = "gpu"
        chain_method = "vectorized"
        if _walkers_from_coefficient:
            _log(logFile, f"Walkers: {walkers_coefficient}×n_params (TBD), NUTS chains: {num_chains_if_nuts} — GPU vectorized mode.")
            _log(logFile, f"Walkers: {walkers_coefficient}×n_params (TBD), NUTS chains: {num_chains_if_nuts_hint} — GPU vectorized mode.")
        else:
            _log(logFile, f"Ensemble walkers: {num_chains}, NUTS chains (if unimodal): {num_chains_if_nuts} — GPU vectorized mode.")
            _log(logFile, f"Ensemble walkers: {num_chains}, NUTS chains (if unimodal): {num_chains_if_nuts_hint} — GPU vectorized mode.")
    else:
        device_type = "cpu"
        chain_method = "parallel"
        if _walkers_from_coefficient:
            _log(logFile, f"No GPU detected. Walkers: {walkers_coefficient}×n_free_params (TBD), NUTS chains: {num_chains_if_nuts} — CPU parallel mode.")
            _log(logFile, f"No GPU detected. Walkers: {walkers_coefficient}×n_free_params (TBD), NUTS chains: {num_chains_if_nuts_hint} — CPU parallel mode.")
        else:
            _log(logFile, f"No GPU detected. Ensemble walkers: {num_chains}, NUTS chains: {num_chains_if_nuts} — CPU parallel mode.")
        _log(logFile, f"Available CPU cores: {num_cores}")
    
    # Configuration file path resolution.
    # User MUST provide the config file with .yaml extension (e.g., 'examples/config.yaml' or '/path/to/config.yaml').
    # No backward compatibility guessing - strict requirement on input format.
    # (Already loaded above for reset flag, so config_peek is available)
        _log(logFile, f"Available CPU cores: {max(cpu_count() - 2, 1)}")
    
    # Load checkpoint and configure precision
    from .checkpoint import load_checkpoint
@@ -592,14 +586,10 @@ def runSampler_NumPyro(configFile, dataFile, logFile, num_chains=None, outDir='.
    # Calculate how many new samples we need
    if existing_samples is not None and existing_num_samples >= target_samples:
        _log(plModel, f"Already have {existing_num_samples} samples >= requested {target_samples}. Nothing to do.\n")
        # Just return the existing data
        return None, plModel, plData, config
    
    new_samples_needed = target_samples - existing_num_samples if existing_samples is not None else target_samples
    
    # Get checkpoint interval (save every N samples, default 100)
    checkpoint_interval = mcmc_config.get('checkpoint_interval', 100)
    
    # Sampler type string ('nuts', 'aies', 'ess') — used in pickle metadata
    sampler_type = saved_sampler_type  # Will be overridden if ensemble is triggered
    
@@ -649,9 +639,13 @@ def runSampler_NumPyro(configFile, dataFile, logFile, num_chains=None, outDir='.
    _log(plModel, f"Free params ({n_params}): {list(param_config['free_params'].keys())}")
    
    # NUTS tuning parameters (used in both PATH A resume and PATH B fresh start)
    max_tree_depth = mcmc_config.get('max_tree_depth', 6)
    max_tree_depth = _get_required_mcmc('max_tree_depth', int)
    if max_tree_depth < 5:
        raise ValueError(f"mcmc.max_tree_depth should be >= 5, got: {max_tree_depth}")
    dense_mass = mcmc_config.get('dense_mass', False)
    target_accept_prob = mcmc_config.get('target_accept_prob', 0.9)
    target_accept_prob = _get_required_mcmc('target_accept_prob', (int, float))
    if not (0.5 <= target_accept_prob <= 0.99):
        raise ValueError(f"mcmc.target_accept_prob must be in [0.5, 0.99], got: {target_accept_prob}")
    
    # ==================================================================
    # PATH A: Extending existing chains (reset=False, pickle found)
@@ -689,7 +683,6 @@ def runSampler_NumPyro(configFile, dataFile, logFile, num_chains=None, outDir='.
        if not _checkpoint_ok:
            existing_samples = None
            existing_num_samples = 0
            samples_grouped = {}

    # -----------------------------------------------------------------
    # PATH A (cont.): resume from a COMPATIBLE existing checkpoint
@@ -742,7 +735,7 @@ def runSampler_NumPyro(configFile, dataFile, logFile, num_chains=None, outDir='.
            mcmc.post_warmup_state = mcmc.last_state
            mcmc._warmup_state = mcmc.last_state
        
        samples_grouped, energy_grouped, last_state, sampling_elapsed, _jit_time = run_batched(
        samples_grouped, energy_grouped, last_state, sampling_elapsed, _ = run_batched(
            mcmc, target_samples, num_chains,
            start_offset=existing_num_samples,
            existing_grouped=samples_grouped,
@@ -790,6 +783,7 @@ def runSampler_NumPyro(configFile, dataFile, logFile, num_chains=None, outDir='.
        _unified_jit_elapsed = _pilot_elapsed
      else:
        _unified_ensemble_state = None
        _unified_jit_elapsed = 0.0

      if production_sampler == 'auto':
        # ==============================================================
@@ -872,6 +866,20 @@ def runSampler_NumPyro(configFile, dataFile, logFile, num_chains=None, outDir='.
            _warmup_jit_elapsed = _unified_jit_elapsed if _unified_jit_elapsed else 0.0
            _log(plModel, f"  JIT compilation (unified): {_warmup_jit_elapsed:.1f}s")

            # Run a single pilot warmup sample on this warmup MCMC object to
            # populate last_state and sample storage before the batched loop.
            _log(plModel, f"  Running 1 pilot warmup sample to populate state…")
            _pilot_start = ttt.time()
            ensemble_mcmc.run(jax.random.PRNGKey(random_seed + _warmup_start_iter),
                              pl_data=plData, pl_model=plModel)
            jax.block_until_ready(ensemble_mcmc.last_state)
            _pilot_elapsed = ttt.time() - _pilot_start
            if _unified_ensemble_state is None:
                _unified_ensemble_state = ensemble_mcmc.last_state
                _unified_jit_elapsed = _pilot_elapsed
                _warmup_jit_elapsed = _pilot_elapsed
                _log(plModel, f"  JIT compilation + pilot warmup: {_warmup_jit_elapsed:.1f}s")

            # Accumulate pilot sample; shape: {name: (num_chains, 1)}
            warmup_samples = {k: v.copy()
                              for k, v in ensemble_mcmc.get_samples(group_by_chain=True).items()}
@@ -1184,7 +1192,7 @@ def runSampler_NumPyro(configFile, dataFile, logFile, num_chains=None, outDir='.
                batch_mcmc.post_warmup_state = ensemble_mcmc.last_state
                batch_mcmc._warmup_state = ensemble_mcmc.last_state
            
                samples_grouped, energy_grouped, last_state, sampling_elapsed, _jit_time = run_batched(
                samples_grouped, energy_grouped, last_state, sampling_elapsed, _ = run_batched(
                    batch_mcmc, num_samples_if_ensemble, num_chains,
                track_energy=False, label="ensemble", skip_jit_pilot=True)
                potential_energy = None
@@ -1398,7 +1406,7 @@ def runSampler_NumPyro(configFile, dataFile, logFile, num_chains=None, outDir='.
        _log(plModel, f"NUTS production: {num_samples_if_nuts} samples, "
             f"{num_chains_if_nuts} chains")
        
        samples_grouped, energy_grouped, last_state, sampling_elapsed, _jit_time = run_batched(
        samples_grouped, energy_grouped, last_state, sampling_elapsed, _ = run_batched(
            nuts_mcmc, num_samples_if_nuts, num_chains_if_nuts,
            track_energy=True, label="NUTS", skip_jit_pilot=True)
        
@@ -1441,7 +1449,7 @@ def runSampler_NumPyro(configFile, dataFile, logFile, num_chains=None, outDir='.
        batch_mcmc.post_warmup_state = ensemble_mcmc.last_state
        batch_mcmc._warmup_state = ensemble_mcmc.last_state
        
        samples_grouped, energy_grouped, last_state, sampling_elapsed, _jit_time = run_batched(
        samples_grouped, _, last_state, sampling_elapsed, _ = run_batched(
            batch_mcmc, num_samples_if_ensemble, num_chains,
            track_energy=False, label=_ens_label, skip_jit_pilot=True)
        
@@ -1457,8 +1465,6 @@ def runSampler_NumPyro(configFile, dataFile, logFile, num_chains=None, outDir='.
    # ==================================================================
    param_order = param_config.get('param_order', list(samples.keys()))
    
    # Note: Raw sample summary is now printed in reports.py when loading results,
    # to support early sampling interruption and checkpoint inspection via reports.py
    
    from numpyro.diagnostics import effective_sample_size, split_gelman_rubin
    diagnostics = {}
+7 −2
Changes for docs/rajpaul_bis_analysis.md: 7 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -400,10 +400,15 @@ Given that you are working with CHEOPS data and your own RV+ancillary-indicator
**Step 1: Quick correlation-based screening (before full model fit)**

1. Compute the measured Pearson correlation $\rho$ between your RV and BIS time series.
2. From a preliminary GP fit or spectral analysis, estimate the realized variances $\sigma_g^2$ and $\sigma_{\dot g}^2$ in your baseline. Check that both activity timescales (rotation modulation and spot evolution) are well-represented — if one is barely visible, flag this as data-driven rank deficiency.
2. Estimate the realized variances $\sigma_g^2$ (rotation modulation) and $\sigma_{\dot g}^2$ (spot evolution) in your baseline. Since these represent the power in the two basis functions, you can obtain them via:
   - **Preliminary Single-Indicator GP Fit (Recommended):** Run a quick GP fit on RV alone using a Quasi-Periodic kernel. While this fits the observable rather than the latent process, you can use the resulting GP variance as a proxy for $\sigma_g^2$. (Note: The inferred GP amplitude is actually the product of the coupling coefficient and the latent variance, i.e., $\text{Var}(\text{RV}_{\text{GP}}) \approx V_c^2 \sigma_g^2$. However, because the determinant $\Delta$ also scales proportionally with these coupling coefficients, the absolute scales cancel out when evaluating the system's conditioning relative to the total observable variance. This makes the screening result scale-invariant and valid for a first-order estimate). To estimate $\sigma_{\dot g}^2$, use the "unit projection" approximation: $\sigma_{\dot g}^2 \approx \sigma_g^2 / \ell^2$ (where $\ell$ is the decay lengthscale). This assumes that the observable is dominated by the rotation modulation term and that the derivative term's power scales inversely with the square of the evolution timescale, providing a sufficient order-of-magnitude estimate for screening.
   - **Spectral Analysis:** Use a Lomb-Scargle periodogram. Integrate power under the main rotation peak for $\sigma_g^2$ and the broadband/wing power for $\sigma_{\dot g}^2$.
   - **Stellar Priors:** Use known $v \sin i$, $R_*$, and typical spot lifetimes for the stellar type as a first-order approximation.
   
   *Check that both activity timescales are well-represented — if one is barely visible (e.g., $\ell$ hits the prior boundary in a GP fit), flag this as data-driven rank deficiency.*
3. Apply the formula from Section 2.6.1:
   $$|\Delta| \approx \frac{\sqrt{\text{Var}(\text{RV})} \cdot \sqrt{\text{Var}(\text{BIS})}}{\sigma_g\sigma_{\dot g}}\sqrt{2(1-\rho)}$$
4. If $|\Delta|$ is small, proceed cautiously — you have evidence of near-degeneracy. If $|\Delta|$ is large, RV and BIS are accessing different directions; proceed with confidence.
4. If $|\Delta|$ is small compared to the scale set by the product $\sigma_g\sigma_{\dot g}$ or its posterior uncertainty, proceed cautiously — you have evidence of near-degeneracy. If $|\Delta|$ is large on that scale, RV and BIS are accessing different directions; proceed with confidence.

**Step 2: Full posterior-based validation (after joint model fit)**