diff --git a/pyqpanda-algorithm/pyqpanda_alg/QKmeans/QuantumKmeans.py b/pyqpanda-algorithm/pyqpanda_alg/QKmeans/QuantumKmeans.py index 83d75d5f..281fec09 100644 --- a/pyqpanda-algorithm/pyqpanda_alg/QKmeans/QuantumKmeans.py +++ b/pyqpanda-algorithm/pyqpanda_alg/QKmeans/QuantumKmeans.py @@ -13,13 +13,28 @@ import numpy as np from copy import deepcopy from numpy import pi -from pyqpanda3.core import CPUQVM, QCircuit, QProg, SWAP, U3, H, measure, draw_qprog +from pyqpanda3.core import CPUQVM, QCircuit, QProg, SWAP, U3, H -def _QuantumKmeansCircuit(theta0, phi0, theta, phi): - machine = CPUQVM() +def _QuantumKmeansCircuit(theta0, phi0, theta, phi, machine=None): + """Swap-test circuit returning the ancilla probabilities ``{'0': .., '1': ..}``. + + The probability of the ancilla being ``1`` is the normalised squared + distance between the two encoded states. + + Two issues made this helper unnecessarily expensive, since ``fit`` calls it + once per (sample, centroid) pair on every iteration: + + * a fresh ``CPUQVM`` was built on every call (now optional and shareable); + * ``draw_qprog(prog)`` was called unconditionally, which *builds a text + diagram of the circuit and throws it away* -- its return value was never + used. It has been removed from the hot path. + + The program is also simulated without a measurement instruction, so the + ancilla probability is exact instead of a 1024-shot estimate. + """ prog = QProg(3) qlist = prog.qubits() cir = QCircuit() @@ -29,22 +44,21 @@ def _QuantumKmeansCircuit(theta0, phi0, theta, phi): cir << SWAP(qlist[0], qlist[1]).control(qlist[2]) cir << H(qlist[2]) prog << cir - prog << measure(qlist[2], qlist[2]) - - draw_qprog(prog) - machine.run(prog, 1024) + if machine is None: + machine = CPUQVM() + machine.run(prog, 1) result = machine.result().get_prob_dict([qlist[2]]) return result -def _point_centroid_distances(point, centroids, k): +def _point_centroid_distances(point, centroids, k, machine=None): xval = [point[0]] for i in range(k): - xval.append(centroids[i][0]) + xval.append(centroids[i][0]) yval = [point[1]] for i in range(k): - yval.append(centroids[i][1]) + yval.append(centroids[i][1]) theta_t = [((x + 1) * pi / 2) for x in xval] theta_c = [((x + 1) * pi / 2) for x in yval] @@ -52,7 +66,7 @@ def _point_centroid_distances(point, centroids, k): results_list = [] for i in range(1, k + 1): - result = _QuantumKmeansCircuit(theta_c[0], theta_t[0], theta_c[i], theta_t[i]) + result = _QuantumKmeansCircuit(theta_c[0], theta_t[0], theta_c[i], theta_t[i], machine) results_list.append(result['1'] if '1' in result else 0) return results_list @@ -175,10 +189,13 @@ def fit(self, data): iter_counter = 1 clusters = None + # One simulator reused across every distance evaluation in every + # iteration (was: a fresh CPUQVM per sample/centroid pair). + machine = CPUQVM() while abs(error - upper_error) > self.tol: centers = centers_new - distances = np.array(list(map(lambda x: _point_centroid_distances(x, centers, self.K), data))) + distances = np.array(list(map(lambda x: _point_centroid_distances(x, centers, self.K, machine), data))) clusters = np.argmin(distances, axis=1) diff --git a/pyqpanda-algorithm/pyqpanda_alg/QSVM/quantum_kernel_svm.py b/pyqpanda-algorithm/pyqpanda_alg/QSVM/quantum_kernel_svm.py index 562976b1..81cc6db5 100644 --- a/pyqpanda-algorithm/pyqpanda_alg/QSVM/quantum_kernel_svm.py +++ b/pyqpanda-algorithm/pyqpanda_alg/QSVM/quantum_kernel_svm.py @@ -11,7 +11,7 @@ # limitations under the License. import numpy as np -from pyqpanda3.core import CPUQVM, QCircuit, QProg, CNOT, U1, U2, measure +from pyqpanda3.core import CPUQVM, QCircuit, QProg, CNOT, U1, U2 from ..plugin import * @@ -21,6 +21,21 @@ def _build_circuit(qlist, n_qbits, weights_x, weights_y): + """Build the kernel circuit for the ``n_qbits``-qubit feature map. + + .. note:: + + The entangling blocks below are hard-coded to a single ``CNOT`` between + ``qlist[0]`` and ``qlist[1]``, so only ``n_qbits == 2`` is meaningful. + Previously any other value silently produced a wrong kernel matrix; + now it fails loudly. + """ + if n_qbits != 2: + raise ValueError( + f"_build_circuit only supports n_qbits=2, got {n_qbits!r}. " + "The entangling block is hard-coded to CNOT(qlist[0], qlist[1])." + ) + circuit = QCircuit() for i in range(n_qbits): circuit << U2(qlist[i], 0, np.pi) @@ -50,17 +65,31 @@ def _build_circuit(qlist, n_qbits, weights_x, weights_y): return circuit -def _run_circuit(n_qbits, weights_x, weights_y): - machine = CPUQVM() +def _run_circuit(n_qbits, weights_x, weights_y, machine=None): + """Evaluate the kernel circuit and return the *exact* output probabilities. + + Performance notes + ----------------- + Two things used to make this helper the hot spot when building an + ``N x N`` kernel matrix: + + 1. A brand new ``CPUQVM`` was constructed on every call (``N*N`` times). + 2. The program contained a ``measure`` instruction, so the probabilities + were **shot-sampled** (1024 shots -> ~3% relative noise) rather than + computed exactly. The kernel matrix therefore changed between runs. + + The program is now simulated without any measurement instruction, so the + state-vector simulator returns the exact amplitudes/probabilities, and a + caller-supplied ``machine`` can be reused across the whole matrix. + """ prog = QProg(n_qbits) qubits = prog.qubits() - circuit = QCircuit() - circuit << _build_circuit(qubits, n_qbits, weights_x, weights_y) - prog << circuit - prog << measure(qubits, qubits) - machine.run(prog, 1024) - result = machine.result().get_counts() - return result + prog << _build_circuit(qubits, n_qbits, weights_x, weights_y) + + if machine is None: + machine = CPUQVM() + machine.run(prog, 1) + return machine.result().get_prob_dict(qubits) class QuantumKernel_vqnet: @@ -306,6 +335,12 @@ def qsvm_classification(): qsvm_classification() """ + if self._n_qbits != 2: + raise ValueError( + f"QuantumKernel_vqnet only supports n_qbits=2, got {self._n_qbits!r}. " + "Pass n_qbits explicitly, e.g. QuantumKernel_vqnet(n_qbits=2)." + ) + if not isinstance(x_vec, np.ndarray): x_vec = np.asarray(x_vec) if y_vec is not None and not isinstance(y_vec, np.ndarray): @@ -350,10 +385,11 @@ def qsvm_classification(): mus = np.asarray(mus.flat) nus = np.asarray(nus.flat) - is_statevector_sim = False - measurement = not is_statevector_sim measurement_basis = "0" * self._n_qbits + # One simulator reused for the whole matrix (was: one per matrix entry). + machine = CPUQVM() + for idx in range(0, len(mus), self._batch_size): to_be_computed_data_pair = [] to_be_computed_index = [] @@ -362,20 +398,22 @@ def qsvm_classification(): j = nus[sub_idx] x_i = x_vec[i] y_j = y_vec[j] - if not np.all(x_i == y_j): + if np.all(x_i == y_j): + # Identical feature vectors encode to the same state, so the + # fidelity is exactly 1. The symmetric branch pre-fills the + # diagonal, but the asymmetric branch used to leave these + # entries at 0 -- record them explicitly for both cases. + kernel[i, j] = 1.0 + if is_symmetric: + kernel[j, i] = 1.0 + else: to_be_computed_data_pair.append((x_i, y_j)) to_be_computed_index.append((i, j)) matrix_elements = [] for x, y in to_be_computed_data_pair: - result = _run_circuit(self._n_qbits, x, y) - try: - counts = result[measurement_basis] - states = np.sum(list(result.values())) - probability = counts / states - except: - probability = 0.0001 - matrix_elements.append(probability) + probabilities = _run_circuit(self._n_qbits, x, y, machine=machine) + matrix_elements.append(float(probabilities.get(measurement_basis, 0.0))) for (i, j), value in zip(to_be_computed_index, matrix_elements): kernel[i, j] = value @@ -383,7 +421,16 @@ def qsvm_classification(): kernel[j, i] = kernel[i, j] if is_symmetric: - D, U = np.linalg.eig(kernel) - kernel = U @ np.diag(np.maximum(0, D)) @ U.transpose() + # ``kernel`` is real symmetric: use ``eigh`` (real spectrum, ascending) + # instead of ``eig``, which could return a complex dtype. + D, U = np.linalg.eigh(kernel) + kernel = U @ np.diag(np.maximum(0.0, D)) @ U.T + kernel = np.real_if_close(kernel) + + # Exact fidelities live in [0, 1]; clip away floating-point overshoot and + # restore the unit diagonal that a quantum kernel must satisfy. + kernel = np.clip(kernel, 0.0, 1.0) + if is_symmetric: + np.fill_diagonal(kernel, 1.0) return kernel diff --git a/pyqpanda-algorithm/pyqpanda_alg/QSVR/QSVR.py b/pyqpanda-algorithm/pyqpanda_alg/QSVR/QSVR.py index 4ee3c4da..adbf3041 100644 --- a/pyqpanda-algorithm/pyqpanda_alg/QSVR/QSVR.py +++ b/pyqpanda-algorithm/pyqpanda_alg/QSVR/QSVR.py @@ -12,10 +12,8 @@ from pyqpanda3.core import CPUQVM, QCircuit, QProg, RX, RY, CZ, H import numpy as np -from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler from sklearn.svm import SVR -from sklearn.metrics import mean_squared_error, r2_score import matplotlib.pyplot as plt from sklearn.decomposition import PCA @@ -108,34 +106,72 @@ def cir_real(qb, cs): Qcir << RX(qb[i], cs[n - 1 - i]) return Qcir - def dist(self, x, y): - machine = CPUQVM() + def dist(self, x, y, machine=None): + """Fidelity ``||^2`` between the two encoded states. + + The circuit carries **no measurement instruction**, so the state-vector + simulator returns the exact probability instead of a shot-sampled + frequency. A ``machine`` may be supplied so that an entire kernel + matrix is built with a single simulator instead of one per entry. + """ + if machine is None: + machine = CPUQVM() prog = QProg(2) qv = prog.qubits() prog << self.cir_real(qv, x) << self.cir_real(qv, y).dagger() - machine.run(prog, 1000) + machine.run(prog, 1) re = machine.result().get_prob_dict(qv) re = parse_quantum_result_dict(re, qv, select_max=-1)['0' * 2] return re def k_kernel(self, X, Y): - matrix = np.zeros((len(X), len(Y))) - for i in range(len(X)): - for j in range(len(Y)): - matrix[i][j] = self.dist(X[i], Y[j]) + """Build the quantum kernel matrix between ``X`` and ``Y``. + + When ``X`` and ``Y`` hold the same samples the matrix is symmetric, so + only the upper triangle is simulated and mirrored. This is the case + scikit-learn uses during ``fit``, so roughly half of the circuit + simulations are avoided there. + """ + X = np.asarray(X) + Y = np.asarray(Y) + n, m = len(X), len(Y) + matrix = np.zeros((n, m)) + + # Single simulator shared by every entry of the matrix. + machine = CPUQVM() + + if n == m and np.array_equal(X, Y): + for i in range(n): + matrix[i, i] = 1.0 + for j in range(i + 1, m): + value = self.dist(X[i], Y[j], machine) + matrix[i, j] = value + matrix[j, i] = value + else: + for i in range(n): + for j in range(m): + matrix[i, j] = self.dist(X[i], Y[j], machine) + return matrix - def get_res(self): + def _fit(self): + """Fit the SVR on the training data (shared by ``get_res``/``show_res``).""" svr = SVR(kernel=self.k_kernel, gamma=0.1) svr.fit(self.x, self.y) + return svr + + def get_res(self): + svr = self._fit() y_ppp = svr.predict(self.x) return y_ppp, self.y def show_res(self): - svr = SVR(kernel=self.k_kernel, gamma=0.1) - svr.fit(self.x, self.y) - x0_test = np.linspace(min(self.x[:, 0]), max(self.x[:, 1]), 30) - x1_test = np.linspace(min(self.x[:, 0]), max(self.x[:, 1]), 30) + svr = self._fit() + # NOTE: each axis must be sampled over its own range. The original code + # used ``max(self.x[:, 1])`` for *both* axes, so the plotted surface was + # computed on the wrong grid. + x0_test = np.linspace(min(self.x[:, 0]), max(self.x[:, 0]), 30) + x1_test = np.linspace(min(self.x[:, 1]), max(self.x[:, 1]), 30) X0_test, X1_test = np.meshgrid(x0_test, x1_test) X_test = np.c_[X0_test.ravel(), X1_test.ravel()] y_pred = svr.predict(X_test).reshape(X0_test.shape) diff --git a/test/QAlgBase/Test_class_qsvr_Quantum_SVR.py b/test/QAlgBase/Test_class_qsvr_Quantum_SVR.py index 810a2af3..173bcab3 100644 --- a/test/QAlgBase/Test_class_qsvr_Quantum_SVR.py +++ b/test/QAlgBase/Test_class_qsvr_Quantum_SVR.py @@ -1,9 +1,28 @@ -import pytest -import numpy as np +# -*-coding:utf-8-*- +"""QSVR 单元测试。 + +除原有的接口冒烟测试外，这里补充核矩阵的数值性质与实现约束的回归用例： + +* 核矩阵对称、对角线为 1、取值落在 [0, 1]； +* 同一输入可完全复现； +* 整个核矩阵只构造 1 个 ``CPUQVM``（原实现每个矩阵元新建一个）； +* ``show_res`` 的绘图网格取自各自维度的取值范围。 +""" import os -from pyqpanda_alg.QSVR import Quantum_SVR +import sys import warnings -import os +from pathlib import Path + +import matplotlib + +matplotlib.use("Agg") # 无显示环境下也能运行 + +import numpy as np +import pytest + +sys.path.append((Path.cwd().parent.parent).__str__()) + +from pyqpanda_alg.QSVR import Quantum_SVR class Test_class_qsvr_Quantum_SVR: @@ -39,6 +58,96 @@ def test_interface11_show_res_basic(self): pytest.fail(f"show_res()方法执行失败: {e}") -if __name__ == "__main__": - # 运行测试 - pytest.main([__file__, "-v", "-s"]) \ No newline at end of file +class TestQuantumSVRKernel: + """量子核矩阵的数值性质。""" + + @staticmethod + def _data(n_samples=8, seed=0): + rng = np.random.default_rng(seed) + x = rng.random((n_samples, 2)) * 10 + y = 2 * np.sin(x[:, 0]) + 1.5 * np.cos(x[:, 1]) + return Quantum_SVR(x, y) + + def test_kernel_is_symmetric_with_unit_diagonal(self): + qsvr = self._data() + kernel = qsvr.k_kernel(qsvr.x, qsvr.x) + + assert kernel.shape == (len(qsvr.x), len(qsvr.x)) + assert np.array_equal(kernel, kernel.T) + assert np.array_equal(np.diag(kernel), np.ones(len(qsvr.x))) + + def test_kernel_is_deterministic(self): + """同一输入两次构建必须给出完全相同的核矩阵。""" + qsvr = self._data() + first = qsvr.k_kernel(qsvr.x, qsvr.x) + second = qsvr.k_kernel(qsvr.x, qsvr.x) + + assert np.array_equal(first, second) + + def test_kernel_values_are_bounded(self): + qsvr = self._data() + kernel = qsvr.k_kernel(qsvr.x, qsvr.x) + + assert kernel.min() >= 0.0 + assert kernel.max() <= 1.0 + + def test_dist_is_symmetric_and_self_fidelity_is_one(self): + qsvr = self._data() + a, b = qsvr.x[0], qsvr.x[1] + + assert qsvr.dist(a, b) == pytest.approx(qsvr.dist(b, a), abs=1e-12) + assert qsvr.dist(a, a) == pytest.approx(1.0, abs=1e-12) + + def test_single_simulator_per_kernel_matrix(self, monkeypatch): + """回归：整个核矩阵只应构造 1 个 CPUQVM，而不是每个矩阵元一个。""" + import sys as _sys + + import pyqpanda3.core as core + + module = _sys.modules["pyqpanda_alg.QSVR.QSVR"] + created = [] + + class _CountingCPUQVM(core.CPUQVM): + def __init__(self, *args, **kwargs): + created.append(1) + super().__init__(*args, **kwargs) + + monkeypatch.setattr(module, "CPUQVM", _CountingCPUQVM) + qsvr = self._data() + qsvr.k_kernel(qsvr.x, qsvr.x) + assert len(created) == 1 + + def test_get_res_returns_prediction_for_every_sample(self): + qsvr = self._data() + predicted, actual = qsvr.get_res() + + assert predicted.shape == actual.shape + assert np.all(np.isfinite(predicted)) + + def test_show_res_grid_covers_each_axis_range(self, monkeypatch): + """回归：绘图网格两个维度应各自取自本维度的取值范围。 + + 原实现两个 ``np.linspace`` 都用 ``min(x[:, 0])`` 与 ``max(x[:, 1])``， + 导致第 0 维的上界取自第 1 维。 + """ + import matplotlib.pyplot as plt + from mpl_toolkits.mplot3d.axes3d import Axes3D + + captured = {} + + def _fake_plot_surface(self, x_grid, y_grid, z, *args, **kwargs): + captured["x"] = np.asarray(x_grid) + captured["y"] = np.asarray(y_grid) + return None + + monkeypatch.setattr(Axes3D, "plot_surface", _fake_plot_surface) + monkeypatch.setattr(plt, "show", lambda *a, **k: None) + + qsvr = self._data(n_samples=6) + qsvr.show_res() + + assert captured["x"].min() == pytest.approx(min(qsvr.x[:, 0])) + assert captured["x"].max() == pytest.approx(max(qsvr.x[:, 0])) + assert captured["y"].min() == pytest.approx(min(qsvr.x[:, 1])) + assert captured["y"].max() == pytest.approx(max(qsvr.x[:, 1])) + diff --git a/test/QKmeans/Test_qkmeans.py b/test/QKmeans/Test_qkmeans.py new file mode 100644 index 00000000..54359660 --- /dev/null +++ b/test/QKmeans/Test_qkmeans.py @@ -0,0 +1,142 @@ +# -*-coding:utf-8-*- +"""QKmeans 单元测试。 + +本模块此前没有任何单元测试。覆盖点： + +* 距离电路返回归一化的 ancilla 概率； +* 去掉 ``measure`` 之后结果**精确且可复现**（原实现对 1024 次采样取值）； +* 显式传入的模拟器被复用，``fit`` 全程只构造 1 个 ``CPUQVM`` + （原实现每个"样本 × 质心"组合都新建一个）； +* ``fit`` 的聚类输出形状、簇数正确，固定随机种子下可复现。 + +.. note:: + + ``pyqpanda_alg.QKmeans.QuantumKmeans`` 这个属性被 ``__init__.py`` 导出的 + 同名**类**遮蔽了，因此这里通过 ``sys.modules`` 取真正的子模块。 +""" +import contextlib +import io +import sys +from pathlib import Path + +import numpy as np +import pytest + +sys.path.append((Path.cwd().parent.parent).__str__()) + +import pyqpanda_alg.QKmeans # noqa: F401 确保子模块已被导入 +from pyqpanda_alg.QKmeans import QuantumKmeans + +_qk_module = sys.modules["pyqpanda_alg.QKmeans.QuantumKmeans"] + + +def _make_two_blobs(seed=0, per_cluster=4): + """构造两团线性可分的数据，避免聚类过程中出现空簇。""" + rng = np.random.default_rng(seed) + return np.vstack( + [ + rng.normal(-1.0, 0.1, size=(per_cluster, 2)), + rng.normal(1.0, 0.1, size=(per_cluster, 2)), + ] + ) + + +class TestQuantumKmeansCircuit: + """底层 swap-test 距离电路。""" + + def test_ancilla_probabilities_are_normalised(self): + """ancilla 的 0/1 概率之和应为 1。""" + result = _qk_module._QuantumKmeansCircuit(1.0, 1.0, 2.0, 2.0) + assert set(result.keys()) == {"0", "1"} + assert sum(result.values()) == pytest.approx(1.0, abs=1e-9) + + def test_probabilities_are_exact_not_sampled(self): + """同一输入多次求值必须完全一致。 + + 原实现带 ``measure`` + 1024 shots，返回的是 ``k/1024`` 形式的采样值， + 边界样本上的簇归属会因此抖动。 + """ + first = _qk_module._QuantumKmeansCircuit(1.0, 1.0, 2.0, 2.0) + for _ in range(3): + assert _qk_module._QuantumKmeansCircuit(1.0, 1.0, 2.0, 2.0) == first + + def test_identical_encoded_states_have_zero_ancilla_probability(self): + """两个参数完全相同的态，测量到 ancilla=1 的概率应为 0。""" + result = _qk_module._QuantumKmeansCircuit(1.3, 0.4, 1.3, 0.4) + assert result["1"] == pytest.approx(0.0, abs=1e-9) + + def test_supplied_machine_is_reused(self, monkeypatch): + """传入模拟器后不应再新建 ``CPUQVM``。""" + import pyqpanda3.core as core + + created = [] + + class _CountingCPUQVM(core.CPUQVM): + def __init__(self, *args, **kwargs): + created.append(1) + super().__init__(*args, **kwargs) + + monkeypatch.setattr(_qk_module, "CPUQVM", _CountingCPUQVM) + machine = core.CPUQVM() + _qk_module._QuantumKmeansCircuit(1.0, 1.0, 2.0, 2.0, machine) + assert created == [] + + +class TestPointCentroidDistances: + + def test_one_distance_per_centroid(self): + centroids = np.array([[0.5, 0.5], [-0.5, -0.5]]) + values = _qk_module._point_centroid_distances(np.array([0.2, 0.3]), centroids, 2) + assert len(values) == 2 + assert all(0.0 <= v <= 1.0 for v in values) + + def test_reproducible(self): + centroids = np.array([[0.5, 0.5], [-0.5, -0.5]]) + first = _qk_module._point_centroid_distances(np.array([0.2, 0.3]), centroids, 2) + second = _qk_module._point_centroid_distances(np.array([0.2, 0.3]), centroids, 2) + assert first == second + + +class TestQuantumKmeansFit: + + def test_fit_two_blobs(self): + data = _make_two_blobs() + np.random.seed(7) + centers, clusters = QuantumKmeans(k=2).fit(data) + + assert np.asarray(centers).shape == (2, 2) + assert len(clusters) == len(data) + assert set(np.unique(clusters).tolist()).issubset({0, 1}) + assert np.all(np.isfinite(centers)) + + def test_fit_is_reproducible_for_fixed_seed(self): + data = _make_two_blobs() + + np.random.seed(7) + centers_a, clusters_a = QuantumKmeans(k=2).fit(data) + np.random.seed(7) + centers_b, clusters_b = QuantumKmeans(k=2).fit(data) + + assert np.array_equal(clusters_a, clusters_b) + assert np.allclose(centers_a, centers_b, atol=1e-12) + + def test_fit_uses_a_single_simulator(self, monkeypatch): + """整个 ``fit`` 只应构造 1 个模拟器。""" + import pyqpanda3.core as core + + created = [] + + class _CountingCPUQVM(core.CPUQVM): + def __init__(self, *args, **kwargs): + created.append(1) + super().__init__(*args, **kwargs) + + monkeypatch.setattr(_qk_module, "CPUQVM", _CountingCPUQVM) + np.random.seed(7) + with contextlib.redirect_stdout(io.StringIO()): + QuantumKmeans(k=2).fit(_make_two_blobs()) + assert len(created) == 1 + + +if __name__ == "__main__": + pytest.main([__file__, "-v", "-s"]) diff --git a/test/QSVM/Test_qsvm.py b/test/QSVM/Test_qsvm.py index d1e37444..099aaa5e 100644 --- a/test/QSVM/Test_qsvm.py +++ b/test/QSVM/Test_qsvm.py @@ -37,3 +37,91 @@ def test_data_loading(): 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0.]) assert np.all(test_labels == [1., 1., 1., 1., 1., 0., 0., 0., 0., 0.]) + +class TestQuantumKernelVqnet: + """量子核矩阵的数值性质与实现约束。 + + 这些用例针对三处原有缺陷建立回归保护： + + 1. 核矩阵带 1024 次采样的随机噪声（两次调用结果不同）； + 2. 对称矩阵经 ``np.linalg.eig`` 投影后出现复数 dtype、对角线不再为 1， + 甚至出现大于 1 的"概率"； + 3. 非对称调用（``y_vec`` 指定）时，若某测试点与训练点完全相同，该处的 + 保真度会被跳过而留下 0，而不是 1。 + """ + + @staticmethod + def _samples(n=8, seed=0): + rng = np.random.default_rng(seed) + return rng.random((n, 2)) * 2 * np.pi + + def test_kernel_is_symmetric_with_unit_diagonal(self): + x = self._samples() + kernel = QuantumKernel_vqnet(n_qbits=2).evaluate(x_vec=x) + + assert kernel.shape == (len(x), len(x)) + assert np.allclose(kernel, kernel.T, atol=1e-12) + assert np.allclose(np.diag(kernel), 1.0, atol=1e-12) + + def test_kernel_is_deterministic(self): + """回归：去掉 measure 采样后，同一输入必须给出完全相同的核矩阵。""" + x = self._samples() + kernel_oracle = QuantumKernel_vqnet(n_qbits=2) + + first = kernel_oracle.evaluate(x_vec=x) + second = kernel_oracle.evaluate(x_vec=x) + + assert np.array_equal(first, second) + + def test_kernel_values_are_real_and_bounded(self): + """保真度必须是 [0, 1] 内的实数。""" + x = self._samples() + kernel = QuantumKernel_vqnet(n_qbits=2).evaluate(x_vec=x) + + assert not np.iscomplexobj(kernel) + assert kernel.min() >= 0.0 + assert kernel.max() <= 1.0 + + def test_asymmetric_kernel_matches_identical_rows(self): + """回归：非对称分支下相同的样本对必须得到保真度 1。""" + x = self._samples() + kernel = QuantumKernel_vqnet(n_qbits=2).evaluate(x_vec=x[:4], y_vec=x[:6]) + + assert kernel.shape == (4, 6) + for i in range(4): + assert kernel[i, i] == pytest.approx(1.0, abs=1e-12) + + def test_unsupported_n_qbits_raises(self): + """回归：n_qbits != 2 曾静默给出错误的核矩阵，现在必须显式报错。""" + x = self._samples(3) + with pytest.raises(ValueError): + QuantumKernel_vqnet(n_qbits=3).evaluate(x_vec=x) + with pytest.raises(ValueError): + QuantumKernel_vqnet().evaluate(x_vec=x) + + def test_single_simulator_per_matrix(self, monkeypatch): + """回归：整个核矩阵只应构造 1 个 CPUQVM，而不是每个矩阵元一个。""" + import pyqpanda3.core as core + + module = sys.modules["pyqpanda_alg.QSVM.quantum_kernel_svm"] + created = [] + + class _CountingCPUQVM(core.CPUQVM): + def __init__(self, *args, **kwargs): + created.append(1) + super().__init__(*args, **kwargs) + + monkeypatch.setattr(module, "CPUQVM", _CountingCPUQVM) + QuantumKernel_vqnet(n_qbits=2).evaluate(x_vec=self._samples()) + assert len(created) == 1 + + def test_usable_as_sklearn_kernel(self): + """核矩阵仍可直接作为 ``sklearn.svm.SVC`` 的 kernel 回调使用。""" + from sklearn.svm import SVC + + train_features, test_features, train_labels, test_labels, _ = _read_vqc_qsvm_data(data_path) + svc = SVC(kernel=QuantumKernel_vqnet(n_qbits=2).evaluate) + svc.fit(train_features, train_labels) + + assert svc.score(test_features, test_labels) >= 0.5 + diff --git a/test/pytest.ini b/test/pytest.ini index 2887f64b..0fa1b337 100644 --- a/test/pytest.ini +++ b/test/pytest.ini @@ -2,9 +2,10 @@ testpaths = QAlgBase QAOA - QRAM + QARM QPCA QSVM + QKmeans python_files =