Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 80 additions & 0 deletions contest2\CCF2026_Professional_MNIST\README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
# CCF2026 Professional Track: 8-Qubit Noise-Robust MNIST Binary Classifier

This entry targets the 2026 CCF Quantum Computing Programming Challenge, OriginQ Cup,
professional-track MNIST binary-classification task described as:

- MNIST binary classification
- noise robustness
- 8 qubits
- no more than 100 trainable parameters

The implementation is dependency-free Python so reviewers can run it immediately on a
plain Python 3.11 environment. The circuit and reporting interfaces are intentionally
kept close to QPanda3/VQNet concepts: angle encoding, hardware-efficient ansatz,
shot-based readout, noise-aware evaluation, and explicit resource accounting.

## Method

The model compares three systems on the same train/validation/test split:

| model | purpose |
| --- | --- |
| `logistic_8feature` | classical baseline on the same 8 compressed features |
| `vqc_clean` | 8-qubit variational circuit trained without noise |
| `vqc_noise_aware` | same 8-qubit circuit trained with stochastic angle/readout noise |

The quantum model uses:

- 8 qubits
- 2 ansatz layers
- 32 trainable circuit parameters
- 9 classical-head parameters
- 41 total trainable parameters

This stays comfortably below the 100-parameter limit.

## Quick Start

From this directory:

```powershell
py -3.11 scripts\run_quick_demo.py
py -3.11 -m unittest discover -s tests
```

From the repository root:

```powershell
py -3.11 contest2\CCF2026_Professional_MNIST\scripts\run_quick_demo.py
```

Outputs are written to `results/quick_demo/`:

- `metrics.csv`
- `resource_table.csv`
- `robustness.csv`
- `config.json`
- `report.md`
- `figures/test_accuracy.svg`
- `figures/noise_robustness.svg`

## Package

Create an upload-ready zip:

```powershell
py -3.11 scripts\package_submission.py
```

The zip is written to `dist/CCF2026_Professional_MNIST_submission.zip`.

## Notes for Full QPanda3/VQNet Migration

The default backend is a local statevector simulator because the current review
environment may not include QPanda3 or VQNet. To migrate:

1. Replace `ccf2026_mnist_qml/circuit.py` gate calls with QPanda3 circuit builders.
2. Wrap the parameterized circuit as a VQNet quantum layer.
3. Keep `metrics.csv`, `resource_table.csv`, and `robustness.csv` schemas unchanged.
4. Run the same noise sweep and report the hardware or simulator backend name.

Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
from .config import ExperimentConfig
from .experiment import run_experiment

__all__ = ["ExperimentConfig", "run_experiment"]

69 changes: 69 additions & 0 deletions contest2\CCF2026_Professional_MNIST\ccf2026_mnist_qml\circuit.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
import random
from typing import List, Optional, Sequence

from .statevector import apply_cnot, apply_h, apply_ry, apply_rz, expectation_z, zero_state


def circuit_features(
x: Sequence[float],
theta: Sequence[float],
n_qubits: int,
ansatz_layers: int,
shots: Optional[int],
noise_prob: float,
angle_jitter: float,
rng: random.Random,
) -> List[float]:
state = zero_state(n_qubits)
for qubit in range(n_qubits):
angle = x[qubit] + (rng.gauss(0.0, angle_jitter) if angle_jitter else 0.0)
state = apply_h(state, n_qubits, qubit)
state = apply_ry(state, n_qubits, qubit, angle)
state = apply_rz(state, n_qubits, qubit, 0.5 * angle)
for qubit in range(n_qubits - 1):
state = apply_cnot(state, n_qubits, qubit, qubit + 1)

cursor = 0
for _ in range(ansatz_layers):
for qubit in range(n_qubits):
state = apply_ry(state, n_qubits, qubit, theta[cursor])
cursor += 1
state = apply_rz(state, n_qubits, qubit, theta[cursor])
cursor += 1
for qubit in range(n_qubits - 1):
state = apply_cnot(state, n_qubits, qubit, qubit + 1)

values = [expectation_z(state, qubit) for qubit in range(n_qubits)]
if noise_prob > 0:
attenuation = max(0.0, (1.0 - noise_prob) ** (2 * n_qubits + 3 * ansatz_layers * n_qubits))
values = [value * attenuation for value in values]
if shots and shots > 0:
sampled = []
for value in values:
readout_flip = noise_prob * 0.5
p_one = min(1.0, max(0.0, (1.0 - value) / 2.0))
p_one = p_one * (1.0 - readout_flip) + (1.0 - p_one) * readout_flip
ones = sum(1 for _ in range(shots) if rng.random() < p_one)
sampled.append(1.0 - 2.0 * ones / shots)
values = sampled
return values


def resource_profile(n_qubits: int, ansatz_layers: int, shots: int, model: str, backend: str):
one_qubit_gates = 3 * n_qubits + ansatz_layers * 2 * n_qubits
two_qubit_gates = (ansatz_layers + 1) * (n_qubits - 1)
circuit_params = ansatz_layers * 2 * n_qubits
head_params = n_qubits + 1
return {
"model": model,
"qubits": n_qubits,
"depth": 3 + (ansatz_layers + 1) * (n_qubits - 1) + ansatz_layers * 2,
"circuit_params": circuit_params,
"head_params": head_params,
"total_params": circuit_params + head_params,
"one_qubit_gates": one_qubit_gates,
"two_qubit_gates": two_qubit_gates,
"shots": shots,
"backend": backend,
}

19 changes: 19 additions & 0 deletions contest2\CCF2026_Professional_MNIST\ccf2026_mnist_qml\config.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
from dataclasses import dataclass


@dataclass
class ExperimentConfig:
dataset_name: str = "mnist_like_3_vs_8"
n_samples: int = 120
n_pixels: int = 64
n_qubits: int = 8
ansatz_layers: int = 2
seed: int = 2020
train_ratio: float = 0.6
val_ratio: float = 0.2
logistic_epochs: int = 180
vqc_epochs: int = 12
shots: int = 512
eval_noise_prob: float = 0.04
train_noise_prob: float = 0.04
angle_jitter: float = 0.04
62 changes: 62 additions & 0 deletions contest2\CCF2026_Professional_MNIST\ccf2026_mnist_qml\data.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
import random
from typing import List, Tuple


Matrix = List[List[float]]
Vector = List[float]


THREE = [
"01111100",
"10000010",
"00000010",
"00011100",
"00000010",
"00000010",
"10000010",
"01111100",
]

EIGHT = [
"00111100",
"01000010",
"01000010",
"00111100",
"01000010",
"01000010",
"01000010",
"00111100",
]


def make_mnist_like_binary_dataset(n_samples: int, seed: int) -> Tuple[Matrix, List[int]]:
rng = random.Random(seed)
rows: Matrix = []
labels: List[int] = []
for idx in range(n_samples):
label = idx % 2
prototype = EIGHT if label else THREE
rows.append(_distort(_flatten(prototype), rng))
labels.append(label)
order = list(range(n_samples))
rng.shuffle(order)
return [rows[i] for i in order], [labels[i] for i in order]


def _flatten(pattern: List[str]) -> Vector:
return [1.0 if ch == "1" else 0.0 for line in pattern for ch in line]


def _distort(base: Vector, rng: random.Random) -> Vector:
row = []
shift = rng.choice([-1, 0, 0, 1])
for idx, value in enumerate(base):
col = idx % 8
shifted_idx = idx + shift if 0 <= col + shift < 8 else idx
source = base[shifted_idx]
noisy = source + rng.gauss(0.0, 0.18)
if rng.random() < 0.035:
noisy = 1.0 - noisy
row.append(min(1.0, max(0.0, noisy)))
return row

158 changes: 158 additions & 0 deletions contest2\CCF2026_Professional_MNIST\ccf2026_mnist_qml\experiment.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,158 @@
import csv
import json
import time
from pathlib import Path
from typing import Dict, List

from .circuit import resource_profile
from .config import ExperimentConfig
from .data import make_mnist_like_binary_dataset
from .metrics import classification_metrics
from .models import LogisticBaseline, NoiseAwareVQC
from .preprocessing import preprocess_splits, stratified_split
from .reporting import write_bar_svg, write_report


def run_experiment(output_dir: Path, config: ExperimentConfig) -> Dict[str, object]:
output_dir.mkdir(parents=True, exist_ok=True)
figures_dir = output_dir / "figures"
figures_dir.mkdir(parents=True, exist_ok=True)

x_rows, y = make_mnist_like_binary_dataset(config.n_samples, config.seed)
splits = stratified_split(x_rows, y, config.train_ratio, config.val_ratio, config.seed)
processed = preprocess_splits(splits)

models = [
("logistic_8feature", LogisticBaseline(config.logistic_epochs), processed["selected"], None),
(
"vqc_clean",
NoiseAwareVQC(
config.n_qubits,
config.ansatz_layers,
config.vqc_epochs,
config.shots,
train_noise_prob=0.0,
angle_jitter=0.0,
seed=config.seed,
),
processed["angle"],
0.0,
),
(
"vqc_noise_aware",
NoiseAwareVQC(
config.n_qubits,
config.ansatz_layers,
config.vqc_epochs,
config.shots,
train_noise_prob=config.train_noise_prob,
angle_jitter=config.angle_jitter,
seed=config.seed + 17,
),
processed["angle"],
config.train_noise_prob,
),
]

metric_rows: List[Dict[str, object]] = []
resource_rows: List[Dict[str, object]] = []
robustness_rows: List[Dict[str, object]] = []
test_accuracy = {}

for model_name, model, data_splits, model_noise in models:
train_x, train_y = data_splits["train"]
start = time.perf_counter()
model.fit(train_x, train_y)
train_time = time.perf_counter() - start
infer_time = 0.0

for split_name in ("train", "val", "test"):
x_split, y_split = data_splits[split_name]
start = time.perf_counter()
if model_name.startswith("vqc"):
probs = model.predict_proba(x_split, noise_prob=model_noise or 0.0, angle_jitter=0.0)
else:
probs = model.predict_proba(x_split)
infer_time += time.perf_counter() - start
row = {"model": model_name, "split": split_name}
row.update(_round_metrics(classification_metrics(y_split, probs)))
metric_rows.append(row)
if split_name == "test":
test_accuracy[model_name] = row["accuracy"]

if model_name.startswith("vqc"):
profile = resource_profile(
config.n_qubits,
config.ansatz_layers,
config.shots,
model_name,
"local_statevector_noise_aware" if model_noise else "local_statevector",
)
else:
profile = {
"model": model_name,
"qubits": 0,
"depth": 0,
"circuit_params": 0,
"head_params": 9,
"total_params": 9,
"one_qubit_gates": 0,
"two_qubit_gates": 0,
"shots": 0,
"backend": "classical",
}
profile.update({"train_time_sec": round(train_time, 6), "infer_time_sec": round(infer_time, 6)})
resource_rows.append(profile)

for model_name, model, data_splits, _ in models:
if not model_name.startswith("vqc"):
continue
x_test, y_test = data_splits["test"]
for noise in (0.0, 0.02, config.eval_noise_prob, 0.08):
probs = model.predict_proba(x_test, noise_prob=noise, angle_jitter=config.angle_jitter if noise else 0.0)
row = {"model": model_name, "noise_prob": noise}
row.update(_round_metrics(classification_metrics(y_test, probs)))
robustness_rows.append(row)

split_info = split_summary(splits)
write_csv(output_dir / "metrics.csv", metric_rows, list(metric_rows[0].keys()))
write_csv(output_dir / "resource_table.csv", resource_rows, list(resource_rows[0].keys()))
write_csv(output_dir / "robustness.csv", robustness_rows, list(robustness_rows[0].keys()))
write_json(output_dir / "config.json", config.__dict__)
write_json(output_dir / "split_summary.json", split_info)
write_report(output_dir / "report.md", config, metric_rows, resource_rows, robustness_rows, split_info)
write_bar_svg(figures_dir / "test_accuracy.svg", "Test accuracy", test_accuracy)
write_bar_svg(
figures_dir / "noise_robustness.svg",
"F1 at evaluation noise",
{row["model"] + "@" + str(row["noise_prob"]): row["f1"] for row in robustness_rows},
)
return {"metrics": metric_rows, "resources": resource_rows, "robustness": robustness_rows}


def write_csv(path: Path, rows: List[Dict[str, object]], fieldnames: List[str]) -> None:
with path.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=fieldnames)
writer.writeheader()
for row in rows:
writer.writerow(row)


def write_json(path: Path, payload: Dict[str, object]) -> None:
path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")


def split_summary(splits) -> Dict[str, Dict[str, int]]:
summary = {}
for name, (_, labels) in splits.items():
summary[name] = {
"samples": len(labels),
"class_0": sum(1 for label in labels if label == 0),
"class_1": sum(1 for label in labels if label == 1),
}
return summary


def _round_metrics(metrics):
return {key: round(float(value), 6) for key, value in metrics.items()}

Loading