Risolvi il problema della frammentazione del mercato con l’ottimizzatore Parity Twine di ParityQC

Stima del tempo di esecuzione: 10 secondi su un processore Nighthawk r2. (NOTA: Si tratta solo di una stima. (La durata potrebbe variare.)

Risultati di apprendimento

• Recuperare e formattare il problema di suddivisione del mercato dalla libreria QOBLIB (Quantum Optimization Benchmarking Library).
• Configurare e utilizzare Parity Twine Optimizer per risolvere un caso di split di mercato.
• Scopri come scegliere le opzioni per il Parity Twine Optimizer e quali risultati vengono generati.

Sfondo

Questo tutorial illustra come risolvere il problema della suddivisione del mercato utilizzando l’ottimizzatore Parity Twine di ParityQC.

L'esempio del problema è tratto dalla QOBLIB (Quantum Optimization Benchmarking Library).

Problema di frammentazione del mercato

Il problema della suddivisione del mercato è un problema reale di allocazione delle risorse di complessità NP-difficile ed è diventato un punto di riferimento per gli algoritmi di ottimizzazione quantistica. Si tratta di una sfida logistica di grande importanza: come suddividere un panorama complesso di clienti e prodotti in territori gestibili e di dimensioni equivalenti.

L'obiettivo è quello di suddividere i mercati di $n$ in due regioni di vendita equilibrate, in modo tale che ciascuna regione riceva esattamente la metà della domanda totale dei prodotti di $m$. La soluzione consiste nella configurazione specifica che garantisce la distribuzione più uniforme possibile della domanda di prodotti, consentendo all’azienda di attuare una strategia logistica e di gestione del personale che garantisca l’equilibrio tra le due regioni, riducendo al minimo i rischi quali carenze localizzate di prodotti o sovraffollamento dei magazzini.

Con l'aumentare del numero di mercati e prodotti, il numero di possibili combinazioni cresce in modo esponenziale, rendendo difficile individuare la suddivisione ottimale ricorrendo alle tradizionali ricerche esaustive.

Formulazione matematica

Sia $A$ una matrice di tipo $m \times n$ che rappresenta la domanda dei prodotti nei vari mercati, dove $A_{ij}$ è la domanda del prodotto $i$ nel mercato $j$.

Si definisce un vettore di assegnazione binario, $x = [x_1, x_2, \dots, x_n]^T \in \{0, 1\}^n$, per cui:

• $x_j = 1$ assegna l' $j$ e di mercato alla Regione A.
• $x_j = 0$ assegna l’ $j$ e di mercato alla Regione B.

Sia $d = [d_1, d_2, \dots, d_m]^T$ il vettore della domanda totale per ciascun prodotto, calcolato come $d = A \cdot \mathbf{1}$. Il volume di vendite target per regione del prodotto $i$ è esattamente $\frac{d_i}{2}$.

Il vincolo di ottimizzazione o di fattibilità richiede che il totale delle vendite assegnate alla Regione A corrisponda esattamente alla metà della domanda totale per ciascun prodotto:

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

In pratica, poiché raramente è possibile effettuare una divisione esatta, il problema viene formulato in modo da minimizzare il quadrato della violazione dei vincoli (la funzione di costo):

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

Espandendo questa espressione si ottiene una forma equivalente a un problema di ottimizzazione binaria quadratica senza vincoli (QUBO).

Una volta risolto il problema, il vettore di soluzione $x$ determina a quale regione viene assegnato il mercato. Questa è la configurazione che consente di ottenere la distribuzione più equilibrata possibile della domanda di prodotto.

Requisiti

Prima di iniziare questo tutorial, assicurati che siano installati i seguenti componenti:

• Qiskit Functions Catalog IBM Cliente (pip install qiskit-ibm-catalog)
• Componente aggiuntivo di Qiskit: Optimization Mapper (pip install qiskit_addon_opt_mapper)
• NumPy (pip install numpy)

È inoltre necessario disporre dell'autorizzazione per accedere alla funzione "Twine Optimizer" di ParityQC. Per richiedere l'accesso, compila questo modulo.

Configura

(Questo codice presuppone che tu abbia già salvato il tuo account nell'ambiente locale.)

Per prima cosa, importa tutti i pacchetti necessari per questo tutorial.

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

Carica il “Parity Twine Optimizer” dal catalogo “ Qiskit Functions ”:

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

Fase 1: Definire il problema come funzione obiettivo

È possibile ottenere un esempio di problema di suddivisione del mercato dalla libreria QOBLIB (Quantum Optimization Benchmarking Library) come indicato di seguito.

La funzione load_market_split_problem recupera un problema specificato da QOBLIB e lo converte in un problema 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 funzione load_market_split_problem richiede le seguenti funzioni di analisi sintattica per recuperare ed elaborare i dati relativi al problema della suddivisione del mercato da 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)

Una volta definito, load_marketsplit_problem può essere utilizzato per caricare un'istanza specifica del problema dalla libreria:

ms_instance = "ms_04_050_001"

ms_problem = load_market_split_problem(ms_instance)

Fase 2: Conversione in formato JSON

Nella prima fase, hai ottenuto la rappresentazione QUBO del problema. Ora, convertilo in formato JSON per la funzione di ottimizzazione:

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'istanza QUBO del problema della suddivisione del mercato è ora convertita in formato JSON come segue:

json_ms_problem = optimization_problem_to_json(ms_problem)

Fase 3: Risolvi il problema utilizzando il Parity Twine Optimizer

Ora che hai ottenuto il problema della suddivisione del mercato e lo hai convertito nella forma corretta, puoi trovare una soluzione utilizzando Twine Optimizer e un backend IBM® a tua scelta.

Per eseguire la funzione, selezionare un dispositivo backend adeguato; ad esempio, ibm_phoenix.

È possibile utilizzare options per un ulteriore controllo (facoltativo) sull'invio:

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

dove shots è un numero intero che specifica il numero di esecuzioni del circuito, postprocessing_level determina se al risultato viene applicata una post-elaborazione, transpile_onlyspecifica se il problema viene solo transpilato in un circuito (e non risolto), e job_tags è l'etichetta utilizzata per identificare il job su IBM Quantum® Platform.

Esegui l'ottimizzatore:

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

Verifica lo stato dell'attività:

# Monitor the job status
function_job.status()

Recupera i risultati:

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

result

Il risultato ha la forma seguente:

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

dove il dizionario solution corrisponde ai qubit definiti nel problema e fornisce i loro valori di spin ottimizzati. metadata fornisce informazioni sulla transpilazione (numero di porte a due qubit/profondità, porte utilizzate, qubit attivi) e sui vari tempi di esecuzione.

Nel contesto del problema della suddivisione dei mercati, la stringa di bit della soluzione rappresenta un vettore di assegnazione binario utilizzato per suddividere i mercati in due regioni distinte. Un valore pari a 1 assegna quel specifico mercato alla Regione A, mentre un valore pari a 0 lo assegna alla Regione B. Nella soluzione ottimale, la combinazione bilancia la ripartizione, il che significa che entrambe le regioni ricevono esattamente la metà della domanda totale dell’azienda per ogni prodotto.

Passi successivi