حل مسألة تقسيم السوق باستخدام مُحسِّن ParityQC Parity Twine

التقدير التقريبي للاستخدام: 10 ثوانٍ على معالج Nighthawk r2. (ملاحظة: هذا تقدير فقط. قد يختلف وقت التشغيل الفعلي لديك.)

نتائج التعلّم

• الحصول على مسألة تقسيم السوق وتنسيقها من QOBLIB - مكتبة معايير التحسين الكمي.

• إعداد واستخدام مُحسِّن Parity Twine لحل مثيل تقسيم السوق.

• تعلّم كيفية اختيار خيارات مُحسِّن Parity Twine وما هي النتائج التي يتم إخراجها.

الخلفية

يوضح هذا الدرس التعليمي كيفية حل مسألة تقسيم السوق باستخدام مُحسِّن ParityQC Parity Twine.

تم الحصول على مثيل المسألة من QOBLIB - مكتبة معايير التحسين الكمي.

مسألة تقسيم السوق

مسألة تقسيم السوق هي مسألة تخصيص موارد واقعية من فئة NP-hard، وقد أصبحت معيارًا لخوارزميات التحسين الكمي. إنها تمثل تحديًا لوجستيًا عالي المخاطر: كيفية تقسيم مشهد معقد من العملاء والمنتجات إلى أقاليم قابلة للإدارة ومتوازنة.

الهدف هو تقسيم $n$ من الأسواق إلى منطقتي مبيعات متوازنتين بحيث تحصل كل منطقة على بالضبط نصف الطلب الإجمالي على $m$ من المنتجات. الحل هو التكوين المحدد الذي يحقق أكثر توزيع ممكن تكافؤًا للطلب على المنتجات، مما يتيح للشركة تطبيق استراتيجية لوجستية وتوظيفية تكون فيها المنطقتان متوازنتين، مما يقلل من مخاطر مثل نقص المنتجات الموضعي أو فيضان المستودعات.

مع تزايد عدد الأسواق والمنتجات، ينمو عدد التباديل الممكنة بشكل أسي، مما يجعل من الصعب إيجاد أفضل تقسيم باستخدام عمليات البحث الشامل التقليدية.

الصياغة الرياضية

لتكن $A$ مصفوفة بأبعاد $m \times n$ تمثل الطلب على المنتجات عبر الأسواق، حيث $A_{ij}$ هو الطلب على المنتج $i$ في السوق $j$.

يُعرَّف متجه تخصيص ثنائي، $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$ هو بالضبط $\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$ المنطقة التي يُخصَّص إليها كل سوق. وهذا هو التكوين الذي يحقق أكثر توزيع ممكن توازنًا للطلب على المنتجات.

المتطلبات

قبل البدء في هذا الدرس التعليمي، تأكد من تثبيت ما يلي:

• عميل IBM لـ Qiskit Functions Catalog (pip install qiskit-ibm-catalog)

• مُلحق Qiskit لمخطط التحسين (pip install qiskit_addon_opt_mapper)

• NumPy (pip install numpy)

أنت أيضًا بحاجة إلى إذن للوصول إلى دالة مُحسِّن ParityQC Twine. لطلب الوصول، أكمل هذا النموذج.

الإعداد

(يفترض هذا الكود أنك قمت بالفعل بـحفظ حسابك في بيئتك المحلية.)

أولاً، استورد جميع الحزم المطلوبة لهذا الدرس التعليمي.

# 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

حمّل مُحسِّن Parity Twine من كتالوج Qiskit Functions:

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

الخطوة 1: تعريف المسألة كدالة هدف

احصل على مثيل مسألة تقسيم السوق من QOBLIB - مكتبة معايير التحسين الكمي على النحو التالي.

تسترجع دالة 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 دوال التحليل (parser) التالية لاسترجاع بيانات مسألة تقسيم السوق ومعالجتها من 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 لدالة المُحسِّن (optimizer):

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

بعد أن حصلت على مسألة تقسيم السوق وحوّلتها إلى الصيغة الصحيحة، يمكنك إيجاد حل باستخدام مُحسِّن Twine وخلفية (backend) من IBM® تختارها.

لتشغيل الدالة، اختر جهاز خلفية (backend) مناسبًا؛ على سبيل المثال، ibm_phoenix.

يمكنك استخدام options للتحكم الإضافي (الاختياري) في الإرسال:

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

حيث shots هو عدد صحيح يحدد عدد مرات تنفيذ الدائرة، وpostprocessing_level يحدد ما إذا كانت المعالجة اللاحقة تُطبَّق على النتيجة، وtranspile_only يحدد ما إذا كانت المسألة تُحوَّل (transpile) فقط إلى دائرة (ولا يتم حلها)، وjob_tags هو التصنيف المستخدم لتحديد المهمة على منصة IBM Quantum®.

شغّل المُحسِّن:

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 مع الكيوبتات المعرّفة في المسألة ويعطي قيم السبين (spin) المُحسَّنة الخاصة بها. metadata يعطي معلومات حول التحويل (transpilation) (عدد بوابات الكيوبتين وعمقها، والبوابات المستخدمة، والكيوبتات النشطة)، وأوقات تشغيل مختلفة.

في سياق مسألة تقسيم السوق، تمثل سلسلة البتات (bitstring) للحل متجه تخصيص ثنائي يُستخدم لتقسيم الأسواق إلى منطقتين منفصلتين. القيمة 1 تخصص ذلك السوق المحدد إلى المنطقة A بينما القيمة 0 تخصصه إلى المنطقة B. في الحل الأمثل، يكون التقسيم متوازنًا، أي أن كلتا المنطقتين تحصلان على نصف إجمالي طلب الشركة تمامًا لكل منتج.

الخطوات التالية

توصيات

• اطلب الوصول إلى الدالة عبر تعبئة هذا النموذج.
• قم بزيارة مرجع API الخاص بدالة Qiskit هذه.
• اقرأ الدليل.
• جرّب الدرس التعليمي لتطبيق مُحسِّن Parity Twine على نموذج شيرينغتون-كيركباتريك.
• راجع المقالة الأولية Connectivity-aware Synthesis of Quantum Algorithms, Drier et al. (2025) على ArXiv.