Quantum Hypercore Case Study

Quantum Hypercore Case Study#

Introduction#

This case study demonstrates a complete hyper-causal DAG (Directed Acyclic Graph) experiment that integrates counterfactual projection, finite-difference training, and post-training evaluation metrics. It simulates a dynamic system with injected anomalies to assess the stability, consistency, and predictive robustness of a multi-node hyper-causal architecture.

Main objectives:

  1. Build a synthetic dynamic environment with controlled anomalies.

  2. Train a three-node hyper-causal DAG using counterfactual backends.

  3. Evaluate robustness through quantitative metrics such as MAPE, Settling Time, and ROC-AUC.

—

General Flow Structure#

The experiment combines deterministic simulation with causal inference and finite-difference optimization. Each backend models a dynamic node connected through directed edges forming a DAG, where causal dependencies propagate forward in time.

  • SyntheticScenario: generates a temporal dataset with sinusoidal base signals and anomaly injections.

  • CFBackend: a parametric backend with recursive \(\tanh\) transformation and counterfactual anticipation.

  • HCGraph: connects nodes via directed edges to form a fully causal structure.

  • Loss functions: combine MSE, consistency, and coherence components.

  • Finite-difference optimization: updates parameters numerically per epoch.

  • DepthScheduler: progressively increases recursion depth through epochs.

—

How to Run#

# From the project root
python -m examples.ex_quantum_hypercore_case_study

# Or directly
python examples/ex_quantum_hypercore_case_study.py

—

Relevant Code Snippets#

Definition of CFBackend (parametric backend with counterfactual anticipation)#
  1        # Base smooth signals (3 channels)
  2        base = np.stack([
  3            0.35 * np.sin(0.25 * t + 0.10),
  4            0.20 * np.sin(0.25 * t + 0.75),
  5            0.15 * np.cos(0.25 * t + 0.40),
  6        ], axis=1)
  7
  8        # Small Gaussian noise
  9        noise = 0.02 * rng.standard_normal(size=base.shape)
 10
 11        x_seq = base + noise
 12        y_target = np.zeros_like(x_seq)
 13
 14        # Inject anomalies into channel 0
 15        y_anom = np.zeros(self.T, dtype=float)
 16        for at in self.anomaly_times:
 17            if 0 <= at < self.T:
 18                y_anom[at] = 1.0
 19                x_seq[at:, 0] += self.anomaly_mag
 20
 21        return x_seq, y_target, y_anom
 22
 23
 24# ============================================================================
 25# 2) Parametric backend with counterfactual anticipator
 26# ============================================================================
 27def shock_up(center: Array) -> Array:
 28    """What-if scenario: channel 0 increases slightly."""
 29    v = center.copy()
 30    v[0] += 0.10
 31    return v
 32
 33
 34def shock_down(center: Array) -> Array:
 35    """What-if scenario: channel 0 decreases slightly."""
 36    v = center.copy()
 37    v[0] -= 0.10
 38    return v
 39
 40
 41def lateral_shift(center: Array) -> Array:
 42    """What-if scenario: small lateral shift in channel 1."""
 43    v = center.copy()
 44    v[1] += 0.06
 45    return v
 46
 47
 48class CFBackend(BaseBackend):
 49    """
 50    Parametric backend with counterfactual anticipation.
 51
 52    - ``run()``: applies a tanh transformation depth times.
 53    - ``project_future()``: linear projection around s_t with counterfactual perturbations.
 54    """
 55
 56    def __init__(
 57        self, config: BackendConfig, w: float = 0.95, b: float = 0.02,
 58        proj_span: float = 0.25,
 59        perts: Sequence[Callable[[Array], Array]] | None = None,
 60        symmetric: bool = True,
 61    ):
 62        super().__init__(config)
 63        self.w = float(w)
 64        self.b = float(b)
 65        self.depth = 1
 66        self._base_span = float(proj_span)
 67        self._projector = LinearProjector(weight=1.0, bias=0.0, span=self._base_span)
 68        self._perts = list(perts or [])
 69        self._anticipator = ContrafactualAnticipator(
 70            projector=self._projector,
 71            perturbations=self._perts,
 72            config=None,
 73        )
 74        self._symmetric = bool(symmetric)
 75
 76    def get_params(self):
 77        """Return current parameters as arrays."""
 78        return {"w": np.array([self.w], dtype=float),
 79                "b": np.array([self.b], dtype=float)}
 80
 81    def set_params(self, params: dict):
 82        """Set backend parameters."""
 83        if "w" in params:
 84            self.w = float(np.asarray(params["w"]).reshape(()))
 85        if "b" in params:
 86            self.b = float(np.asarray(params["b"]).reshape(()))
 87
 88    def run(self, params: dict | None = None) -> Array:
 89        """Apply recursive tanh transformation."""
 90        if params:
 91            self.set_params(params)
 92        x = self._require_input()
 93        s = x.astype(float)
 94        for _ in range(max(1, int(self.depth))):
 95            s = np.tanh(self.w * s + self.b)
 96        return self._validate_state(s)
 97
 98    def project_future(self, s_t: Array, branches: int = 2) -> Array:
 99        """Generate future states using updated projector span and perturbations."""
100        s = self._validate_state(s_t)
101        k = max(2, int(branches))
102        span = max(0.05, self._base_span / (1.0 + 0.25 * (self.depth - 1)))
103        self._projector = LinearProjector(weight=1.0, bias=0.0, span=span)
104        self._anticipator = ContrafactualAnticipator(
105            projector=self._projector,
106            perturbations=self._perts,
107            config=None,
108        )
109        fut = self._anticipator.generate(s)
110        if fut.shape[0] >= k:
111            fut = fut[:k]
112        else:
113            rep = np.repeat(fut[-1][None, :], k - fut.shape[0], axis=0)
114            fut = np.concatenate([fut, rep], axis=0)
115        return self._validate_branches(fut)
116
117
118# ============================================================================
119# 3) Building the hyper-causal DAG and training utilities
120# ============================================================================
121def build_hc_dag(D=3, seed=7):
122    """Construct a minimal 3-node hyper-causal DAG."""
123    cfg = BackendConfig(output_dim=D, seed=seed)
124    bA = CFBackend(cfg, w=0.95, b=0.03, proj_span=0.22, perts=[shock_up, lateral_shift])
125    bB = CFBackend(cfg, w=1.00, b=0.01, proj_span=0.25, perts=[shock_down])
126    bC = CFBackend(cfg, w=1.05, b=0.00, proj_span=0.28, perts=[lateral_shift])
127
128    def l2_risk(branch: Array) -> float:
129        return float(np.sqrt(np.sum(branch ** 2)))
Main experiment function quantum_hypercore_case_study()#
 1            coh_vals = []
 2            for key in ("A", "B", "C"):
 3                branches = info_map[key].get("branches", None)
 4                if isinstance(branches, np.ndarray) and branches.ndim == 2:
 5                    coh_vals.append(loss_coh(branches))
 6            if coh_vals:
 7                total_coh += float(np.mean(coh_vals))
 8            s_prev = s_map
 9            callbacks.on_step_end(t_idx, {
10                "step": int(t_idx),
11                "norm_sC": float(np.linalg.norm(sC)),
12                "task_sC": float(loss_task(sC, y_target[t_idx])),
13            })
14        task, cons, coh = total_task / T, total_cons / max(1, T - 1), total_coh / T
15        total = task + 0.5 * cons + 0.3 * coh
16        return total, {"task": task, "cons": cons, "coh": coh, "total": total}, np.asarray(y_pred_seq), np.asarray(scores)
17
18    best = None
19    for epoch in range(EPOCHS):
20        for be in backends:
21            depth_cb.on_epoch_begin(epoch, {"backend": be})
22        callbacks.on_epoch_begin(epoch, {"epoch": int(epoch)})
23        total0, det0, _, _ = forward_sequence()
24
25        def loss_wrapper():
26            l, _, _, _ = forward_sequence()
27            return l
28
29        grads = finite_diff_grads(loss_wrapper, params, apply_params_fn)
30        params, opt_state = opt.step(params, grads, opt_state)
31        apply_params_fn(params)
32        total1, det1, _, _ = forward_sequence()
33        callbacks.on_epoch_end(epoch, {"epoch": int(epoch), "loss_before": det0, "loss_after": det1})
34
35        if best is None or det1["total"] < best["total"]:
36            best = {"epoch": int(epoch), **{k: float(v) for k, v in det1.items()}}
37
38        print(f"[Epoch {epoch}] total_before={det0['total']:.6f} total_after={det1['total']:.6f} depth={[int(be.depth) for be in backends]}")
39
40    _, _, y_pred, scores = forward_sequence()
41
42    # Evaluation
43    mape_val = mape(y_target[:, 0], y_pred[:, 0])
44    over_val = overshoot(y_target[:, 0], y_pred[:, 0])
45    setl_val = settling_time(y_target[:, 0], y_pred[:, 0], tol=0.05)
46    rob_val = robustness(y_target[:, 0], y_pred[:, 0])
47    auc_early = early_roc_auc(y_true=y_anom, scores=scores, horizon=3)
48    thr = np.percentile(scores, 80)
49    rec_lag = recall_at_lag(y_true=y_anom, y_pred=(scores > thr).astype(float), lag=3)
50
51    print("\n=== Final Metrics ===")
52    print(f"MAPE (%):          {mape_val:.6f}")
53    print(f"Overshoot:         {over_val:.6f}")
54    print(f"Settling Time:     {int(setl_val)} samples")
55    print(f"Robustness:        {rob_val:.6f}")
56    print(f"Early ROC-AUC@H=3: {auc_early:.6f}")
57    print(f"Recall@Lag=3:      {rec_lag:.6f}")
58    print("\nBest epoch snapshot:", best)
59
60    if LOG_PATH.exists():
61        print("\nTelemetry JSONL →", LOG_PATH.resolve())
62
63    return {
64        "best": best,
65        "metrics": {
66            "mape": float(mape_val),
67            "overshoot": float(over_val),
68            "settling_time": int(setl_val),
69            "robustness": float(rob_val),
70            "early_auc_h3": float(auc_early),
71            "recall_lag3": float(rec_lag),
72        },
73    }
74
75
76# ============================================================================
77# 5) Entry point
78# ============================================================================
79if __name__ == "__main__":
80    summary = quantum_hypercore_case_study()
81    print("\nSummary:")
82    print(json.dumps(summary, indent=2))

—

Functional Explanation#

This experiment evaluates a hyper-causal DAG under perturbative dynamics and counterfactual projections. It integrates predictive, causal, and diagnostic components in a single computational framework.

  1. Synthetic Dynamic Environment

    The SyntheticScenario generates temporal signals with injected anomalies:

    \[x_t = \sin(\omega t + \phi) + \epsilon_t, \qquad x_{t,\text{anom}} = x_t + \delta \mathbf{1}_{t \in \mathcal{A}}\]

    where \(\epsilon_t\) represents Gaussian noise and \(\mathcal{A}\) marks anomaly indices. This allows controlled evaluation of the model’s ability to adapt to irregular perturbations.

  2. Counterfactual Backend

    Each backend implements recursive dynamics:

    \[S_t = \tanh(w S_{t-1} + b)\]

    and generates counterfactual future branches:

    \[S_{t+1}^{(k)} = S_t + \mathcal{P}_k(S_t)\]

    where \(\mathcal{P}_k\) are perturbation operators encoding “what-if” transformations. This enables simultaneous evaluation of nominal and hypothetical state trajectories.

  3. Hyper-Causal Graph Structure

    The DAG is defined as:

    \[\mathcal{G} = (\mathcal{V}, \mathcal{E}), \quad \mathcal{V} = \{A, B, C\}, \quad \mathcal{E} = \{A \to B, A \to C, B \to C\}\]

    Each node receives upstream states, applies its transformation, and emits a projected state to downstream nodes. This simulates hierarchical causal propagation within a directed structure.

  4. Loss Composition

    The total loss function integrates predictive, temporal, and coherence constraints:

    \[\mathcal{L}_{total} = \mathcal{L}_{task} + 0.5\,\mathcal{L}_{consistency} + 0.3\,\mathcal{L}_{coherence}\]

    • Task loss: prediction accuracy.

    • Consistency loss: stability of consecutive states.

    • Coherence loss: alignment across projected branches.

    Each loss term contributes to maintaining causal smoothness and coherent evolution.

  5. Training via Finite-Difference Gradients

    Instead of backpropagation, gradients are computed numerically:

    \[g_i = \frac{\mathcal{L}(\theta_i + \epsilon) - \mathcal{L}(\theta_i - \epsilon)}{2\epsilon}\]

    Parameters are updated using simple gradient descent with adaptive learning rate control.

  6. Evaluation Metrics

    After training, several metrics are computed to quantify system stability and sensitivity:

    • MAPE – Mean Absolute Percentage Error.

    • Overshoot – amplitude deviation beyond target.

    • Settling Time – number of samples to reach stability within tolerance.

    • Robustness – correlation between predicted and target signals.

    • Early ROC-AUC@H=3 – anomaly discrimination within a 3-step horizon.

    • Recall@Lag=3 – sensitivity to delayed anomaly detection.

    These indicators evaluate both predictive accuracy and control stability under dynamic perturbations.

—

Exact Output#

[Epoch 0] total_before=0.098290 total_after=0.092576 depth=[1, 1, 1]
[Epoch 1] total_before=0.069731 total_after=0.062847 depth=[2, 2, 2]
[Epoch 2] total_before=0.062847 total_after=0.056539 depth=[2, 2, 2]
[Epoch 3] total_before=0.042616 total_after=0.041622 depth=[3, 3, 3]
[Epoch 4] total_before=0.041622 total_after=0.039550 depth=[3, 3, 3]
[Epoch 5] total_before=0.030317 total_after=0.029850 depth=[4, 4, 4]

=== Final Metrics ===
MAPE (%):          19094053004148.468750
Overshoot:         0.000000
Settling Time:     60 samples
Robustness:        0.957343
Early ROC-AUC@H=3: 0.224138
Recall@Lag=3:      0.000000

Best epoch snapshot: {'epoch': 5, 'task': 0.019989936715902698, 'cons': 0.015169491076939604, 'coh': 0.007584946879521136, 'total': 0.02985016631822884}

Telemetry JSONL → runs/quantum_hypercore_case_telemetry.jsonl

Summary:
{
  "best": {
    "epoch": 5,
    "task": 0.019989936715902698,
    "cons": 0.015169491076939604,
    "coh": 0.007584946879521136,
    "total": 0.02985016631822884
  },
  "metrics": {
    "mape": 19094053004148.47,
    "overshoot": 0.0,
    "settling_time": 60,
    "robustness": 0.9573428590759466,
    "early_auc_h3": 0.22413793103448276,
    "recall_lag3": 0.0
  }
}
Contents