ParityQC Parity Twine Optimizer로 마켓 스플릿 문제 해결하기
사용량 추정치: Nighthawk r2 프로세서에서 10초. (참고: 이는 추정치일 뿐이며, 실제 실행 시간은 다를 수 있습니다.)
학습 목표
-
QOBLIB - Quantum Optimization Benchmarking Library에서 마켓 스플릿(Market Split) 문제를 가져와 형식화합니다.
-
마켓 스플릿 인스턴스를 풀기 위해 Parity Twine Optimizer를 설정하고 사용합니다.
-
Parity Twine Optimizer의 옵션을 선택하는 방법과 어떤 결과가 출력되는지 배웁니다.
배경
이 튜토리얼에서는 ParityQC Parity Twine Optimizer를 사용하여 마켓 스플릿 문제를 해결하는 방법을 보여줘요.
문제 인스턴스는 QOBLIB - Quantum Optimization Benchmarking Library에서 가져옵니다.
마켓 스플릿 문제
마켓 스플릿 문제는 실제 세계의 NP-난해한 자원 할당 문제이며, 양자 최적화 알고리즘의 벤치마크로 자리 잡았습니다. 이는 고위험 물류 과제를 나타냅니다: 복잡한 고객 및 제품 환경을 관리 가능하고 균등한 구역으로 어떻게 분할할 것인가.
목표는 개의 시장을 두 개의 균형 잡힌 판매 구역으로 나누어, 각 구역이 개 제품에 대한 총수요의 정확히 절반씩을 받도록 하는 것입니다. 해는 제품 수요의 가능한 가장 균등한 분배를 달성하는 특정 구성으로, 회사가 두 구역이 균형을 이루는 물류 및 인력 배치 전략을 실행할 수 있도록 하며, 국지적인 제품 부족이나 창고 과잉 재고와 같은 위험을 최소화합니다.
시장과 제품의 수가 증가함에 따라 가능한 순열의 수가 기하급수적으로 늘어나므로, 전통적인 전수 탐색 방식으로 최선의 분할을 찾는 것이 어려워집니다.
수학적 정식화
를 시장 간 제품 수요를 나타내는 행렬이라고 하고, 는 시장 에서 제품 에 대한 수요라고 합시다.
이진 할당 벡터 가 다음과 같이 정의됩니다.
- 은 시장 를 구역 A에 배정합니다.
- 은 시장 를 구역 B에 배정합니다.
를 각 제품에 대한 총수요 벡터라고 하며, 이는 로 계산됩니다. 제품 에 대해 구역별 목표 판매량은 정확히 입니다.
최적화 또는 실현 가능성 제약 조건은 구역 A에 할당된 총 판매량이 모든 제품에 대해 총수요의 정확히 절반과 일치해야 함을 요구합니다.
실제로는 정확한 분할이 거의 불가능하므로, 이 문제는 제약 위반의 제곱을 최소화하는 방식(비용 함수)으로 정식화됩니다.
이를 전개하면 이차 무제약 이진 최적화(quadratic unconstrained binary optimization, QUBO) 문제와 동등한 형태가 됩니다.
풀이가 끝나면, 해 벡터 는 각 시장이 어느 구역에 배정되는지를 결정합니다. 이는 제품 수요의 가능한 가장 균형 잡힌 분배를 달성하는 구성입니다.
요구 사항
이 튜토리얼을 시작하기 전에, 다음이 설치되어 있는지 확인하세요.
-
Qiskit Functions Catalog IBM Client (
pip install qiskit-ibm-catalog) -
Qiskit addon Optimization Mapper (
pip install qiskit_addon_opt_mapper) -
NumPy (
pip install numpy)
또한 ParityQC Twine Optimizer 함수에 접근할 권한이 필요합니다. 접근 권한을 요청하려면 이 양식을 작성하세요.
설정
(이 코드는 로컬 환경에 이미 계정을 저장했다고 가정해요.)
먼저 이 튜토리얼에 필요한 모든 패키지를 가져옵니다.
# Added by doQumentation — installs the packages this notebook needs if they are missing
import importlib.util
_needed = {"numpy": "numpy", "qiskit_addon_opt_mapper": "qiskit-addon-opt-mapper", "qiskit_ibm_catalog": "qiskit-ibm-catalog", "requests": "requests"}
_missing = [pip for module, pip in _needed.items()
if importlib.util.find_spec(module) is None]
# One at a time, so a package that fails to install does not block the others
for _pip in _missing:
%pip install -q {_pip}
if not _missing:
print("\u2713 All required packages are installed")
import tempfile
from collections.abc import Callable
from pathlib import Path
import numpy as np
import requests
from qiskit_addon_opt_mapper import OptimizationProblem
from qiskit_addon_opt_mapper.converters import OptimizationProblemToQubo
from qiskit_ibm_catalog import QiskitFunctionsCatalog
Qiskit Functions 카탈로그에서 Parity Twine Optimizer를 로드합니다.
catalog = QiskitFunctionsCatalog(channel="ibm_quantum_platform")
function = catalog.load("parityqc/parity-twine-optimizer")
1단계: 문제를 목적 함수로 정의하기
다음과 같이 QOBLIB - Quantum Optimization Benchmarking Library에서 마켓 스플릿(market split) 문제 인스턴스를 가져와요.
load_market_split_problem 함수는 QOBLIB에서 주어진 문제를 가져와 QUBO 문제로 변환합니다.
def load_market_split_problem(instance_name: str) -> OptimizationProblem:
"""Load and formulate a market split optimization problem from an QOBLIB instance.
The QOBLIB library can be found here:
https://github.com/ZIB-AOPT/QOBLIB.
Args:
instance_name: Name of the market split instance to load as specified by the .dat file
in the QOBLIB repo.
Returns:
The output OptimizationProblem containing the loaded market split problem.
"""
problem_matrix, problem_vector = fetch_and_parse(
instance_name, "01-marketsplit", parse_marketsplit_dat
)
# Create optimization problem
optimization_problem = OptimizationProblem(instance_name)
# Add binary variables (one for each market)
optimization_problem.binary_var_list(problem_matrix.shape[1])
# Add equality constraints (one for each product)
for idx, rhs in enumerate(problem_vector):
optimization_problem.linear_constraint(
problem_matrix[idx, :], sense="==", rhs=rhs
)
# Convert to QUBO with penalty parameter
return OptimizationProblemToQubo(penalty=1).convert(optimization_problem)
load_market_split_problem 함수는 QOBLIB에서 시장 분할 문제 데이터를 가져오고 처리하기 위해 다음 파서 함수가 필요합니다.
def fetch_and_parse(instance_name: str, problem: str, parse_func: Callable):
"""Generic function to fetch and parse data from QOBLIB repository.
Args:
instance_name: Name of the instance to fetch.
problem: Category of the problem (e.g., '01-marketsplit', '07-independentset').
parse_func: Function used to parse the downloaded file
(e.g., parse_marketsplit_dat, parse_gph_file).
Returns:
Result of `parse_func` - either (np.ndarray, np.ndarray) for marketsplit
or nx.Graph for MIS.
"""
base_url = (
"https://raw.githubusercontent.com/ZIB-AOPT/QOBLIB/refs/heads/main/"
)
url = (
base_url
+ problem
+ "/instances/"
+ instance_name
+ (".dat" if problem == "01-marketsplit" else ".gph")
)
try:
response = requests.get(url, timeout=30)
response.raise_for_status()
with tempfile.NamedTemporaryFile(
mode="w",
suffix=".dat" if problem == "01-marketsplit" else ".gph",
delete=False,
encoding="utf-8",
) as temp_file:
temp_file.write(response.text)
temp_file_path = temp_file.name
try:
return parse_func(temp_file_path)
finally:
Path(temp_file_path).unlink(missing_ok=True)
except requests.RequestException as e:
print(f"Error fetching data from repository: {e}")
except (ValueError, OSError) as e:
print(f"Error processing data: {e}")
return None
def parse_marketsplit_dat(filename: str) -> tuple[np.ndarray, np.ndarray]:
"""Parse a market split problem from a .dat file format.
Args:
filename: Path to the .dat file.
Returns:
Tuple of (A, b) where:
- A: (m, n) array of coefficients.
- b: (m,) array of target values.
Raises:
ValueError: If file format is invalid or file is empty.
"""
with Path(filename).open(encoding="utf-8") as f:
lines = [
line.strip()
for line in f
if line.strip() and not line.startswith("#")
]
if not lines:
raise ValueError("Empty or invalid .dat file")
# First line: m n (number of products and markets)
try:
m, n = map(int, lines[0].split())
except (ValueError, IndexError) as e:
raise ValueError(
"Invalid file format: first line must contain 'm n' integers"
) from e
if len(lines) < m + 1:
raise ValueError(
f"File contains {len(lines)} lines but expected {m + 1} lines"
)
# Next m lines: each row of A followed by corresponding element of b
mat_a = []
vec_b = []
for i in range(1, m + 1):
try:
values = list(map(int, lines[i].split()))
except ValueError as e:
raise ValueError(f"Invalid integer values in line {i + 1}") from e
if len(values) != n + 1:
raise ValueError(
f"Line {i + 1} contains {len(values)} values but expected {n + 1}"
)
mat_a.append(values[:-1]) # First n values: product sales per market
vec_b.append(values[-1]) # Last value: target sales for this product
return np.array(mat_a), np.array(vec_b)
정의되면, load_marketsplit_problem을 사용하여 라이브러리에서 특정 문제 인스턴스를 로드할 수 있습니다.
ms_instance = "ms_04_050_001"
ms_problem = load_market_split_problem(ms_instance)
2단계: JSON 형식으로 변환하기
첫 번째 단계에서 문제의 QUBO 형태를 얻었습니다. 이제 옵티마이저 함수를 위해 이를 JSON 형식으로 변환합니다.
def optimization_problem_to_json(
problem: OptimizationProblem,
) -> dict[str, float]:
"""
Converts an unconstrained quadratic OptimizationProblem in terms of binary or spin variables
to the JSON input format of the Parity Twine Qiskit Function.
Args:
problem: The optimization problem to convert to JSON.
Returns:
The JSON input format of the given problem.
"""
ising, constant = problem.to_ising()
output = {"()": float(constant)}
for op, coefficient in zip(ising.paulis, ising.coeffs, strict=True):
# Invert the label strings because Qiskit has opposite convention
qubits = tuple(
num
for num, pauli in enumerate(op.to_label()[::-1])
if pauli == "Z"
)
output[str(qubits)] = float(coefficient)
return output
이제 시장 분할 문제의 QUBO 인스턴스가 다음과 같이 JSON 형식으로 변환되었습니다.
json_ms_problem = optimization_problem_to_json(ms_problem)
3단계: Parity Twine Optimizer를 사용해 문제 해결하기
시장 분할 문제를 얻어 올바른 형태로 변환했으므로, 이제 Twine Optimizer와 선택한 IBM® 백엔드를 사용하여 해답을 찾을 수 있습니다.
함수를 실행하려면 ibm_phoenix 같은 적합한 백엔드 장치를 선택하세요.
제출에 대한 (선택적) 추가 제어를 위해 options를 사용할 수 있습니다.
options = {
"shots": 100000,
"postprocessing_level": 1,
"transpile_only": False,
"job_tags": ["market_split"],
}
여기서 shots는 회로 실행 횟수를 지정하는 정수이고, postprocessing_level은 결과에 후처리를 적용할지 여부를 결정하며, transpile_only
는 문제를 회로로 트랜스파일만 할지(해결하지 않을지) 여부를 지정하고, job_tags는 IBM Quantum® Platform에서 작업을 식별하는 데 사용되는 레이블입니다.
옵티마이저를 실행합니다.
function_job = function.run(
problem=json_ms_problem,
variable_type="spin",
backend_name="ibm_phoenix",
options=options,
)
print(f"Job ID: {function_job.job_id}")
작업 상태를 확인합니다.
# Monitor the job status
function_job.status()
결과를 가져옵니다.
# Retrieve the job result if the status is DONE
result = function_job.result()
result
결과는 다음과 같은 형태입니다.
{
'solution': {'0': 1, '1': -1, '10': 1, ... },
'objective_value': 1.0,
'solution_bitstring': '010000011101111011001001110010',
'metadata': {
'circuit_metrics': {
'depth': 309,
'gate_count': 3880,
'two_qubit_gate_depth': 116,
'two_qubit_gate_count': 899,
'num_qubits': 30,
'operations': {'sx': 1244, 'rz': 1227, 'cz': 899, 'delay': 473, 'measure': 30, 'x': 7},
},
'solver_info': {
'variable_mapping': {'0': 0, '1': 1, '10': 2, ... },
'bitstring_distributions': {
'before_postprocessing': {'011101110010110111001110011000': 1, ...},
'after_postprocessing': {'011011110000110101001111011000': 1, ...}
},
'best_parameters': {
'beta': [-0.18054534155552715],
'gamma': [1.4141236348317905]
}
},
'resource_usage': {
'RUNNING: MAPPING': {'CPU_TIME': 172.936},
'RUNNING: OPTIMIZING_FOR_HARDWARE': {'CPU_TIME': 0.272},
'RUNNING: WAITING_FOR_QPU': {'CPU_TIME': 7.798},
'RUNNING: EXECUTING_QPU': {'QPU_TIME': 30.0},
'RUNNING: POST_PROCESSING': {'CPU_TIME': 31.613},
},
}
}
여기서 solution 딕셔너리는 문제에서 정의된 큐비트에 대응하며 이들의 최적화된 스핀 값을 제공합니다.
metadata는 트랜스파일에 대한 정보(2큐비트 게이트 수/깊이, 사용된 게이트, 활성 큐비트)와 다양한 실행 시간을 제공합니다.
마켓 스플릿 문제의 맥락에서, 해 비트열은 시장을 두 개의 별도 구역으로 분할하는 데 사용되는 이진 배정 벡터를 나타냅니다. 값이 1이면 해당 시장을 구역 A에 배정하고, 값이 0이면 구역 B에 배정합니다. 최적의 해에서는 이 조합이 분할의 균형을 맞추어, 두 구역이 모든 제품에 대해 회사 전체 수요의 정확히 절반씩을 받도록 합니다.
다음 단계
- 이 양식을 작성하여 함수에 대한 액세스를 요청하세요.
- 이 Qiskit Function에 대한 API 참조를 방문하세요.
- 가이드를 읽어보세요.
- Sherrington-Kirkpatrick 모델에 Parity Twine Optimizer를 적용하는 튜토리얼을 시도해보세요.
- Connectivity-aware Synthesis of Quantum Algorithms, Drier et al. (2025) ArXiv 프리프린트를 검토하세요.