Résoudre le problème de répartition de marché avec le ParityQC Parity Twine Optimizer

Estimation d'utilisation : 10 secondes sur un processeur Nighthawk r2. (REMARQUE : il s'agit seulement d'une estimation. Ton temps d'exécution peut varier.)

Objectifs d'apprentissage

• Obtenir et formater le problème de répartition de marché à partir de la QOBLIB - Quantum Optimization Benchmarking Library.

• Configurer et utiliser le Parity Twine Optimizer pour résoudre une instance de répartition de marché.

• Apprendre à choisir les options du Parity Twine Optimizer et quels résultats sont produits en sortie.

Contexte

Ce tutoriel montre comment résoudre le problème de répartition de marché à l'aide du ParityQC Parity Twine Optimizer.

L'instance du problème est obtenue à partir de la QOBLIB - Quantum Optimization Benchmarking Library.

Problème de répartition de marché

Le problème de répartition de marché est un problème réel d'allocation de ressources NP-difficile, devenu un banc d'essai pour les algorithmes d'optimisation quantique. Il représente un défi logistique à fort enjeu : comment répartir un paysage complexe de clients et de produits en territoires gérables et équilibrés.

L'objectif est de diviser $n$ marchés en deux régions de vente équilibrées, de sorte que chaque région reçoive exactement la moitié de la demande totale pour $m$ produits. La solution est la configuration spécifique qui permet d'obtenir la répartition la plus uniforme possible de la demande de produits, permettant à une entreprise de mettre en œuvre une stratégie logistique et de personnel où les deux régions sont équilibrées, minimisant les risques tels que les pénuries localisées de produits ou les débordements d'entrepôt.

À mesure que le nombre de marchés et de produits augmente, le nombre de permutations possibles croît de manière exponentielle, ce qui rend difficile la recherche de la meilleure répartition à l'aide de recherches exhaustives traditionnelles.

Formulation mathématique

Soit $A$ une matrice $m \times n$ représentant la demande de produits à travers les marchés, où $A_{ij}$ est la demande du produit $i$ sur le marché $j$.

Un vecteur d'affectation binaire, $x = [x_1, x_2, \dots, x_n]^T \in \{0, 1\}^n$, est défini de la manière suivante :

• $x_j = 1$ affecte le marché $j$ à la région A.
• $x_j = 0$ affecte le marché $j$ à la région B.

Soit $d = [d_1, d_2, \dots, d_m]^T$ le vecteur de demande totale pour chaque produit, calculé comme $d = A \cdot \mathbf{1}$. Le volume de ventes cible par région pour le produit $i$ est exactement $\frac{d_i}{2}$.

La contrainte d'optimisation ou de faisabilité exige que les ventes totales allouées à la région A correspondent exactement à la moitié de la demande totale pour chaque produit :

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

En pratique, comme une division exacte est rarement possible, le problème est formulé de manière à minimiser la violation quadratique de la contrainte (la fonction de coût) :

$\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.$

En développant cette expression, on obtient une forme équivalente à un problème d'optimisation binaire quadratique sans contrainte (QUBO).

Une fois résolu, le vecteur solution $x$ détermine à quelle région le marché est affecté. Il s'agit de la configuration qui permet d'obtenir la répartition la plus équilibrée possible de la demande de produits.

Prérequis

Avant de commencer ce tutoriel, assure-toi que les éléments suivants sont installés :

• Qiskit Functions Catalog IBM Client (pip install qiskit-ibm-catalog)

• Qiskit addon Optimization Mapper (pip install qiskit_addon_opt_mapper)

• NumPy (pip install numpy)

Tu as également besoin d'une autorisation pour accéder à la fonction ParityQC Twine Optimizer. Pour demander l'accès, remplis ce formulaire.

Configuration

(Ce code suppose que tu as déjà enregistré ton compte dans ton environnement local.)

Commence par importer tous les paquets requis pour ce tutoriel.

# 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

Charge le Parity Twine Optimizer depuis le catalogue Qiskit Functions :

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

Étape 1 : définir le problème comme une fonction objectif

Obtiens une instance du problème de répartition de marché (market split) à partir de la QOBLIB - Quantum Optimization Benchmarking Library comme suit.

La fonction load_market_split_problem récupère un problème donné depuis QOBLIB et le convertit en un problème 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)

La fonction load_market_split_problem nécessite les fonctions d'analyse suivantes pour récupérer et traiter les données du problème de répartition de marché depuis 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)

Une fois définie, load_marketsplit_problem peut être utilisée pour charger une instance spécifique de problème depuis la bibliothèque :

ms_instance = "ms_04_050_001"

ms_problem = load_market_split_problem(ms_instance)

Étape 2 : convertir au format JSON

À la première étape, tu as obtenu la forme QUBO du problème. Convertis-la maintenant au format JSON pour la fonction d'optimisation :

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

L'instance QUBO du problème de répartition de marché est maintenant convertie au format JSON comme suit :

json_ms_problem = optimization_problem_to_json(ms_problem)

Étape 3 : résoudre le problème avec le Parity Twine Optimizer

Maintenant que tu as obtenu le problème de répartition de marché et que tu l'as converti dans le bon format, tu peux trouver une solution en utilisant le Twine Optimizer et un backend IBM® de ton choix.

Pour exécuter la fonction, choisis un backend adapté ; par exemple, ibm_phoenix.

Tu peux utiliser options pour un contrôle supplémentaire (facultatif) sur la soumission :

options = {
    "shots": 100000,
    "postprocessing_level": 1,
    "transpile_only": False,
    "job_tags": ["market_split"],
}

où shots est un entier qui spécifie le nombre d'exécutions du circuit, postprocessing_level détermine si un post-traitement est appliqué au résultat, transpile_only spécifie si le problème est uniquement transpilé en circuit (et non résolu), et job_tags est l'étiquette utilisée pour identifier le job sur IBM Quantum® Platform.

Exécute l'optimiseur :

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}")

Vérifie le statut du job :

# Monitor the job status
function_job.status()

Récupère les résultats :

# Retrieve the job result if the status is DONE
result = function_job.result()

result

Le résultat est de la forme :

{
    '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},
        },
    }
}

où le dictionnaire solution correspond aux qubits définis dans le problème et donne leurs valeurs de spin optimisées. metadata fournit des informations sur la transpilation (nombre/profondeur des portes à deux qubits, portes utilisées, qubits actifs) et divers temps d'exécution.

Dans le contexte du problème de répartition de marché, la chaîne de bits de solution représente un vecteur d'affectation binaire utilisé pour répartir les marchés en deux régions distinctes. Une valeur de 1 affecte ce marché spécifique à la région A, tandis qu'une valeur de 0 l'affecte à la région B. Pour la solution optimale, la combinaison équilibre la répartition, ce qui signifie que les deux régions reçoivent exactement la moitié de la demande totale de l'entreprise pour chaque produit.

Étapes suivantes

Recommandations

• Demande l'accès à la fonction en remplissant ce formulaire.
• Consulte la référence API pour cette Qiskit Function.
• Lis le guide.
• Essaie le tutoriel pour appliquer le Parity Twine Optimizer au modèle de Sherrington-Kirkpatrick.
• Consulte le préprint ArXiv Connectivity-aware Synthesis of Quantum Algorithms, Drier et al. (2025).