ParityQC Parity Twine Optimizerでマーケットスプリット問題を解く

使用量の目安: Nighthawk r2プロセッサーで10秒。(注: これはあくまで目安です。実際の実行時間は変動する場合があります。)

学習成果

• QOBLIB - Quantum Optimization Benchmarking Libraryからマーケットスプリット問題を取得しフォーマットする

• Parity Twine Optimizerをセットアップして使用し、マーケットスプリットのインスタンスを解く

• Parity Twine Optimizerのオプションの選び方と、出力される結果について学ぶ

背景

このチュートリアルでは、ParityQC Parity Twine Optimizerを使用してマーケットスプリット問題を解く方法を示します。

この問題インスタンスは、QOBLIB - Quantum Optimization Benchmarking Libraryから取得されます。

マーケットスプリット問題

マーケットスプリット問題は、現実世界のNP困難な資源配分問題であり、量子最適化アルゴリズムのベンチマークとなっています。 これは、顧客と製品の複雑な状況を、管理可能で均等化されたテリトリーに分割するという、リスクの高いロジスティクス上の課題を表しています。

目標は、$n$個の市場を、各地域が$m$個の製品に対する総需要のちょうど半分を受け取るように、2つのバランスの取れた販売地域に分割することです。 この解は、製品需要の分布を可能な限り均等にする具体的な構成であり、企業は両地域の均衡が取れたロジスティクスと人員配置の戦略を実施でき、 局所的な製品不足や倉庫のオーバーフローといったリスクを最小化できます。

市場と製品の数が増加するにつれて、可能な順列の数は指数関数的に増加するため、従来の網羅的探索を用いて最良の分割を見つけることが困難になります。

数学的定式化

$A$を、市場全体における製品の需要を表す$m \times n$行列とし、$A_{ij}$を市場$j$における製品$i$の需要とします。

二値割り当てベクトル$x = [x_1, x_2, \dots, x_n]^T \in \{0, 1\}^n$を次のように定義します。

• $x_j = 1$は市場$j$を地域Aに割り当てます。
• $x_j = 0$は市場$j$を地域Bに割り当てます。

$d = [d_1, d_2, \dots, d_m]^T$を、各製品の総需要ベクトルとし、$d = A \cdot \mathbf{1}$として計算されるとします。製品$i$に対する1地域あたりの目標販売量は、ちょうど$\frac{d_i}{2}$です。

最適化または実行可能性の制約は、地域Aに割り当てられた総販売量が、すべての製品について総需要の半分と完全に一致することを要求します。

$A x = \frac{1}{2} A \mathbf{1} = b.$

実際には、厳密な分割はほとんど不可能であるため、この問題は制約違反の二乗(コスト関数)を最小化するように定式化されます。

$\min_{x} \left\Vert{} A x - b \right\Vert{}^2 = \sum_{i=1}^{m} \left( \sum_{j=1}^{n} A_{ij} x_j - b\right)^2.$

これを展開すると、二次制約なし二値最適化(QUBO)問題と等価な形式が得られます。

解を求めると、解ベクトル$x$によって各市場がどちらの地域に割り当てられるかが決まります。これは、製品需要の分布を可能な限りバランスよくする構成です。

要件

このチュートリアルを開始する前に、以下がインストールされていることを確認してください。

• 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 Catalog から Parity Twine Optimizer を読み込みます。

catalog = QiskitFunctionsCatalog(channel="ibm_quantum_platform")
function = catalog.load("parityqc/parity-twine-optimizer")

ステップ1: 問題を目的関数として定義する

次のようにして、QOBLIB - Quantum Optimization Benchmarking Library からマーケットスプリット問題のインスタンスを取得します。

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量子ビットゲートの数/深さ、使用されたゲート、アクティブな量子ビット)や各種の実行時間を示します。

市場分割問題の文脈では、解のビット列は市場を2つの異なる地域に分割するために使用される二値割り当てベクトルを表します。値が1の場合はその 特定の市場を地域Aに割り当て、値が0の場合は地域Bに割り当てます。最適解では、この組み合わせが分割のバランスを取り、両方の地域がすべての製品について会社全体の 需要のちょうど半分を受け取ることを意味します。

次のステップ

おすすめ

• このフォームに記入して、この関数へのアクセスをリクエストしてください。
• このQiskit FunctionについてはAPIリファレンスをご覧ください。
• ガイドをお読みください。
• Sherrington-Kirkpatrickモデルに Parity Twine Optimizer を適用するチュートリアルをお試しください。
• Connectivity-aware Synthesis of Quantum Algorithms, Drier et al. (2025) のArXivプレプリントをご覧ください。