Skip to content

ntsa.data

Data-driven front end: run the equation-free half of ntsa on a raw measured series — no model equations required. DataSeries wraps a scalar record \(x(t)\) sampled at fixed dt, then embeds (optimal_lag, false_nearest_neighbours), computes the correlation dimension \(D_2\), estimates the leading Lyapunov exponent from the data itself (lyapunov.rosenstein_lyapunov), classifies the regime, and draws the same 8-panel diagnostic row as ntsa.characterize. Only the methods that must re-integrate a model are bypassed: the full Lyapunov spectrum and bifurcation sweeps.

from ntsa.data import DataSeries

ds = DataSeries(x, dt=1e-3, label='hot-wire probe')
res = ds.analyze()            # zeta, dim, D2, regime, evidence, stats, MDS
ds.characterize('figs/probe.pdf')

8-panel characterization of Lorenz63 from measurements only

Lorenz63 characterized from its measurements (and state snapshots for the MDS panel) only: \(\lambda_1 = 0.97 \pm 0.06\) (true 0.906), classified chaotic.

analyze(lam1='auto') (the default) runs the Rosenstein estimator, so measured chaotic data reaches the chaotic label with no model at all; the estimator's fit guards return nan on non-chaotic data rather than a spurious slope. Pass a float for an external estimate, or None to skip — see the regime classification section.

Full reference

ntsa.data

Data-driven front end: the equation-free half of ntsa on a raw measured series.

DataSeries wraps a scalar record x(t) sampled at a fixed dt and runs everything that does not need model equations: embedding diagnostics, correlation dimension, regime classification, signal statistics, MDS, and the 8-panel diagnostic row. Lyapunov exponents and bifurcation sweeps need an integrable model and are bypassed; an externally estimated lam1 can be passed through to the classifier.

DataSeries dataclass

A measured scalar time series and its equation-free ntsa characterization.

Parameters:

Name Type Description Default
x ndarray

Scalar observable, shape (Nt,).

 required
dt float

Sampling time.

 required
label str

Case label used in figure titles.

 'data'
Y ndarray

Simultaneous multivariate record (Nt, N) — e.g. the full measured state — used for the MDS panel; the delay embedding of x is used when omitted.

 None
trim bool

Trim any residual transient detected by stationary_start on construction (printed note when nonzero), mirroring run_long's guard for model runs.

 True
Source code in ntsa/data.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
@dataclass
class DataSeries:
    """A measured scalar time series and its equation-free ntsa characterization.

    Parameters
    ----------
    x : np.ndarray
        Scalar observable, shape (Nt,).
    dt : float
        Sampling time.
    label : str
        Case label used in figure titles.
    Y : np.ndarray, optional
        Simultaneous multivariate record (Nt, N) — e.g. the full measured state —
        used for the MDS panel; the delay embedding of `x` is used when omitted.
    trim : bool
        Trim any residual transient detected by `stationary_start` on construction
        (printed note when nonzero), mirroring `run_long`'s guard for model runs.
    """

    x: np.ndarray
    dt: float
    label: str = 'data'
    Y: np.ndarray = None
    trim: bool = True
    results: dict = field(default=None, init=False, repr=False)

    def __post_init__(self):
        self.x = np.asarray(self.x, dtype=float).ravel()
        if self.Y is not None:
            self.Y = np.asarray(self.Y, dtype=float)
            if len(self.Y) != len(self.x):
                raise ValueError(f'len(Y)={len(self.Y)} != len(x)={len(self.x)}')
        if self.trim:
            i0 = ntsa_tools.stationary_start(self.x)
            if i0:
                print(f'[DataSeries] trimmed {i0 * self.dt:.3g} t.u. of transient '
                      f'({100 * i0 // len(self.x)}% of the record)')
                self.x = self.x[i0:]
                if self.Y is not None:
                    self.Y = self.Y[i0:]

    @property
    def t(self):
        return np.arange(len(self.x)) * self.dt

    def analyze(self, lam1='auto', lam1_std=0.0, mds=True):
        """Run the equation-free pipeline and cache the results dict.

        Same keys as one `ntsa.characterize.characterize` case, plus `D2`/`d2_fit`
        (Grassberger-Procaccia dimension of the delay embedding — the data-only
        substitute for the Kaplan-Yorke dimension). `lam1='auto'` (default)
        estimates the leading Lyapunov exponent from the data itself with
        `lyapunov.rosenstein_lyapunov` (nan when its fit guards reject, so
        non-chaotic data never gets a spurious exponent); pass a float to use an
        external estimate instead, or None to skip. `spectrum` stays None — a
        full spectrum needs model equations.
        """
        x, dt = self.x, self.dt
        zeta = ntsa_tools.optimal_lag(x)
        dim, fnn = ntsa_tools.false_nearest_neighbours(x, zeta)
        E = ntsa_tools.delay_embed(x, dim, zeta)
        D2, d2_fit = ntsa_tools.correlation_dimension(E)
        lyap_fit = None
        if isinstance(lam1, str):  # 'auto'
            try:
                lam1, lam1_std, lyap_fit = lyap.rosenstein_lyapunov(x, dt, zeta=zeta, dim=dim)
            except ValueError as err:
                print(f'[DataSeries] rosenstein_lyapunov skipped: {err}')
                lam1, lam1_std = None, 0.0
        regime, evidence = classify_regime(x, dt, lam1=lam1, lam1_std=lam1_std,
                                           t_total=len(x) * dt)
        gamma = t_gamma = None
        if mds:
            gamma, idx = ntsa_tools.classical_mds(self.Y if self.Y is not None else E)
            t_gamma = self.t[idx]
        self.results = {'label': self.label, 'zeta': zeta, 'dim': dim, 'fnn': fnn,
                        'D2': D2, 'd2_fit': d2_fit,
                        'lambda1': lam1, 'lambda1_std': lam1_std, 'lyap_fit': lyap_fit,
                        'spectrum': None, 'regime': regime, 'evidence': evidence,
                        'stats': ntsa_tools.signal_stats(x, dt),
                        'gamma': gamma, 't_gamma': t_gamma, 't': self.t, 'x': x}
        return self.results

    def characterize(self, pdf_name='figs/ntsa_data.pdf', **analyze_kwargs):
        """Draw the 8-panel diagnostic row (PDF + same-name PNG) via `plot_row`.

        Every panel is computed from the data alone; the Lyapunov panel shows
        the Rosenstein (or external) `lam1` marker, and the Rosenstein
        log-divergence fit is appended as a second PDF page when available.
        """
        res = self.analyze(**analyze_kwargs) if self.results is None or analyze_kwargs else self.results
        f1 = res['evidence'].get('f1')
        t_CR = 1.0 / f1 if f1 else 0.01 * len(self.x) * self.dt  # plot-window fallback only
        lam_ok = res['lambda1'] is not None and np.isfinite(res['lambda1'])
        lam_txt = f', $\\lambda_1$={res["lambda1"]:.3f}' if lam_ok else ''
        title = (f'{res["label"]}\n{res["regime"]}  $\\zeta$={res["zeta"]}, d={res["dim"]}, '
                 f'$D_2$={res["D2"]:.2f}{lam_txt}')
        fig = plt.figure(figsize=(19, 2.7), layout='constrained')
        _tight(fig)
        gs = fig.add_gridspec(1, 1)
        plot_row(fig, gs[0], res['t'], res['x'], self.dt, res['zeta'], res['dim'], t_CR,
                 title=title, evidence=res['evidence'], gamma=res['gamma'],
                 t_gamma=res['t_gamma'], lam1=res['lambda1'], lam1_std=res['lambda1_std'])
        figs = [fig]
        if res['lyap_fit'] is not None:
            figs.append(plot_lyapunov_fit(res['lyap_fit']))
        out_dir = os.path.dirname(pdf_name)
        if out_dir:
            os.makedirs(out_dir, exist_ok=True)
        fig.savefig(os.path.splitext(pdf_name)[0] + '.png', dpi=150, bbox_inches='tight')
        save_figs_pdf_tight(pdf_name, figs)
        print(f'Saved figures --> {pdf_name}')
        return res

analyze(lam1='auto', lam1_std=0.0, mds=True)

Run the equation-free pipeline and cache the results dict.

Same keys as one ntsa.characterize.characterize case, plus D2/d2_fit (Grassberger-Procaccia dimension of the delay embedding — the data-only substitute for the Kaplan-Yorke dimension). lam1='auto' (default) estimates the leading Lyapunov exponent from the data itself with lyapunov.rosenstein_lyapunov (nan when its fit guards reject, so non-chaotic data never gets a spurious exponent); pass a float to use an external estimate instead, or None to skip. spectrum stays None — a full spectrum needs model equations.

Source code in ntsa/data.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def analyze(self, lam1='auto', lam1_std=0.0, mds=True):
    """Run the equation-free pipeline and cache the results dict.

    Same keys as one `ntsa.characterize.characterize` case, plus `D2`/`d2_fit`
    (Grassberger-Procaccia dimension of the delay embedding — the data-only
    substitute for the Kaplan-Yorke dimension). `lam1='auto'` (default)
    estimates the leading Lyapunov exponent from the data itself with
    `lyapunov.rosenstein_lyapunov` (nan when its fit guards reject, so
    non-chaotic data never gets a spurious exponent); pass a float to use an
    external estimate instead, or None to skip. `spectrum` stays None — a
    full spectrum needs model equations.
    """
    x, dt = self.x, self.dt
    zeta = ntsa_tools.optimal_lag(x)
    dim, fnn = ntsa_tools.false_nearest_neighbours(x, zeta)
    E = ntsa_tools.delay_embed(x, dim, zeta)
    D2, d2_fit = ntsa_tools.correlation_dimension(E)
    lyap_fit = None
    if isinstance(lam1, str):  # 'auto'
        try:
            lam1, lam1_std, lyap_fit = lyap.rosenstein_lyapunov(x, dt, zeta=zeta, dim=dim)
        except ValueError as err:
            print(f'[DataSeries] rosenstein_lyapunov skipped: {err}')
            lam1, lam1_std = None, 0.0
    regime, evidence = classify_regime(x, dt, lam1=lam1, lam1_std=lam1_std,
                                       t_total=len(x) * dt)
    gamma = t_gamma = None
    if mds:
        gamma, idx = ntsa_tools.classical_mds(self.Y if self.Y is not None else E)
        t_gamma = self.t[idx]
    self.results = {'label': self.label, 'zeta': zeta, 'dim': dim, 'fnn': fnn,
                    'D2': D2, 'd2_fit': d2_fit,
                    'lambda1': lam1, 'lambda1_std': lam1_std, 'lyap_fit': lyap_fit,
                    'spectrum': None, 'regime': regime, 'evidence': evidence,
                    'stats': ntsa_tools.signal_stats(x, dt),
                    'gamma': gamma, 't_gamma': t_gamma, 't': self.t, 'x': x}
    return self.results

characterize(pdf_name='figs/ntsa_data.pdf', **analyze_kwargs)

Draw the 8-panel diagnostic row (PDF + same-name PNG) via plot_row.

Every panel is computed from the data alone; the Lyapunov panel shows the Rosenstein (or external) lam1 marker, and the Rosenstein log-divergence fit is appended as a second PDF page when available.

Source code in ntsa/data.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def characterize(self, pdf_name='figs/ntsa_data.pdf', **analyze_kwargs):
    """Draw the 8-panel diagnostic row (PDF + same-name PNG) via `plot_row`.

    Every panel is computed from the data alone; the Lyapunov panel shows
    the Rosenstein (or external) `lam1` marker, and the Rosenstein
    log-divergence fit is appended as a second PDF page when available.
    """
    res = self.analyze(**analyze_kwargs) if self.results is None or analyze_kwargs else self.results
    f1 = res['evidence'].get('f1')
    t_CR = 1.0 / f1 if f1 else 0.01 * len(self.x) * self.dt  # plot-window fallback only
    lam_ok = res['lambda1'] is not None and np.isfinite(res['lambda1'])
    lam_txt = f', $\\lambda_1$={res["lambda1"]:.3f}' if lam_ok else ''
    title = (f'{res["label"]}\n{res["regime"]}  $\\zeta$={res["zeta"]}, d={res["dim"]}, '
             f'$D_2$={res["D2"]:.2f}{lam_txt}')
    fig = plt.figure(figsize=(19, 2.7), layout='constrained')
    _tight(fig)
    gs = fig.add_gridspec(1, 1)
    plot_row(fig, gs[0], res['t'], res['x'], self.dt, res['zeta'], res['dim'], t_CR,
             title=title, evidence=res['evidence'], gamma=res['gamma'],
             t_gamma=res['t_gamma'], lam1=res['lambda1'], lam1_std=res['lambda1_std'])
    figs = [fig]
    if res['lyap_fit'] is not None:
        figs.append(plot_lyapunov_fit(res['lyap_fit']))
    out_dir = os.path.dirname(pdf_name)
    if out_dir:
        os.makedirs(out_dir, exist_ok=True)
    fig.savefig(os.path.splitext(pdf_name)[0] + '.png', dpi=150, bbox_inches='tight')
    save_figs_pdf_tight(pdf_name, figs)
    print(f'Saved figures --> {pdf_name}')
    return res