Optimización dinámica de carteras con el Portfolio Optimizer de Global Data Quantum

Estimación de uso: Aproximadamente 55 minutos en un procesador Heron r2. (NOTA: Esto es sólo una estimación. El tiempo de ejecución real puede variar)

En segundo plano

El problema de la optimización dinámica de carteras tiene como objetivo encontrar la estrategia de inversión óptima a lo largo de múltiples periodos de tiempo para maximizar la rentabilidad esperada de la cartera y minimizar los riesgos, a menudo bajo ciertas restricciones como el presupuesto, los costes de transacción o la aversión al riesgo. A diferencia de la optimización de carteras estándar, que considera un único momento para reequilibrar la cartera, la versión dinámica tiene en cuenta la naturaleza evolutiva de los activos y adapta las inversiones en función de los cambios en el rendimiento de los activos a lo largo del tiempo.

Este tutorial muestra cómo realizar una optimización dinámica de carteras utilizando la función Qiskit de Quantum Portfolio Optimizer. En concreto, ilustramos cómo utilizar esta función de aplicación para resolver un problema de asignación de inversiones en múltiples pasos temporales.

El enfoque consiste en formular la optimización de la cartera como un problema multiobjetivo de optimización binaria cuadrática sin restricciones (QUBO). En concreto, formulamos la función QUBO $O$ para optimizar simultáneamente cuatro objetivos diferentes:

• Maximizar la función de retorno $F$
• Minimizar el riesgo de la inversión $R$
• Minimizar los costes de transacción $C$
• Cumplir las restricciones de inversión, formuladas en un plazo adicional para minimizar $P$.

En resumen, para abordar estos objetivos formulamos la función QUBO como $O = -F + \frac{\gamma}{2} R + C + \rho P,$ donde $\gamma$ es el coeficiente de aversión al riesgo y $\rho$ es el coeficiente de refuerzo de las restricciones (multiplicador de Lagrange). La formulación explícita se encuentra en la Ec. (15) de nuestro manuscrito [1].

Resolvemos utilizando un método híbrido cuántico-clásico basado en el Eigensolver Cuántico Variacional (VQE). En esta configuración, el circuito cuántico estima la función de coste, mientras que la optimización clásica se realiza mediante el algoritmo de Evolución Diferencial, lo que permite una navegación eficiente por el paisaje de soluciones. El número de qubits necesarios depende de tres factores principales: el número de activos na, el número de periodos de tiempo nt, y la resolución de bits utilizada para representar la inversión nq. En concreto, el número mínimo de qubits en nuestro problema es na*nt*nq.

En este tutorial, nos centraremos en la optimización de una cartera regional basada en el índice español IBEX 35. En concreto, utilizamos una cartera de siete activos, como se indica en el cuadro siguiente:

Reequilibramos nuestra cartera en cuatro etapas, cada una separada por un intervalo de 30 días a partir del 1 de noviembre de 2022. Cada variable de inversión se codifica mediante dos bits. El resultado es un problema que requiere 56 qubits para resolverse.

Utilizamos el ansatz de Amplitudes Reales Optimizadas, una adaptación personalizada y eficiente desde el punto de vista del hardware del ansatz de Amplitudes Reales estándar, diseñado específicamente para mejorar el rendimiento de este tipo de problema de optimización financiera.

La ejecución cuántica se realiza en el backend ibm_torino . Para una explicación detallada de la formulación del problema, la metodología y la evaluación del rendimiento, consulte el manuscrito publicado [1].

Requisitos

!pip install qiskit-ibm-catalog
!pip install pandas
!pip install matplotlib
!pip install yfinance

Configuración

Para utilizar el Optimizador de Cartera Quantum, seleccione la función a través del Catálogo de Funciones Qiskit. Necesita una cuenta IBM Quantum Premium Plan o Flex Plan con una licencia de Global Data Quantum para ejecutar esta función.

En primer lugar, inicia sesión con tu clave API. A continuación, carga la función deseada desde el archivo « Qiskit Functions Catalog ». Aquí estás accediendo a la quantum_portfolio_optimizer función del catálogo mediante la QiskitFunctionsCatalog clase. Esta función nos permite utilizar el solucionador Quantum Portfolio Optimization predefinido.

from qiskit_ibm_catalog import QiskitFunctionsCatalog

catalog = QiskitFunctionsCatalog(
    channel="ibm_quantum_platform",
    instance="INSTANCE_CRN",
    # For `token`, use the 44-character API_KEY you created
    # and saved from the IBM Quantum Platform Home dashboard
    token="YOUR_API_KEY",
)

# Access function
dpo_solver = catalog.load("global-data-quantum/quantum-portfolio-optimizer")

Paso 1: Lea la cartera de entrada

En este paso, cargamos datos históricos para los siete activos seleccionados del índice IBEX 35, concretamente desde el 1 de noviembre de 2022 hasta el 1 de abril de 2023.

Obtenemos los datos utilizando la API de Yahoo Finanzas, centrándonos en los precios de cierre. A continuación, los datos se procesan para garantizar que todos los activos tengan el mismo número de días con datos. Cualquier dato que falte (días no hábiles) se trata adecuadamente, garantizando que todos los activos estén alineados en las mismas fechas.

Los datos se estructuran en DataFrame con un formato coherente en todos los activos.

import yfinance as yf
import pandas as pd

# List of IBEX 35 symbols
symbols = [
    "ACS.MC",
    "ITX.MC",
    "FER.MC",
    "ELE.MC",
    "SCYR.MC",
    "AENA.MC",
    "AMS.MC",
]

start_date = "2022-11-01"
end_date = "2023-4-01"

series_list = []
symbol_names = [symbol.replace(".", "_") for symbol in symbols]

# Create a full date index including weekends
full_index = pd.date_range(start=start_date, end=end_date, freq="D")

for symbol, name in zip(symbols, symbol_names):
    print(f"Downloading data for {symbol}...")
    data = yf.download(symbol, start=start_date, end=end_date)["Close"]
    data.name = name

    # Reindex to include weekends
    data = data.reindex(full_index)

    # Fill missing values (for example, weekends or holidays)
    # by forward/backward fill
    data.ffill(inplace=True)
    data.bfill(inplace=True)

    series_list.append(data)

# Combine all series into a single DataFrame
assets = pd.concat(series_list, axis=1)

# Convert index to string for consistency
assets.index = assets.index.astype(str)

Output:

[*********************100%***********************]  1 of 1 completed

[*********************100%***********************]  1 of 1 completed
[*********************100%***********************]  1 of 1 completed
[*********************100%***********************]  1 of 1 completed
[*********************100%***********************]  1 of 1 completed
[*********************100%***********************]  1 of 1 completed
[*********************100%***********************]  1 of 1 completed

Downloading data for ACS.MC...
Downloading data for ITX.MC...
Downloading data for FER.MC...
Downloading data for ELE.MC...
Downloading data for SCYR.MC...
Downloading data for AENA.MC...
Downloading data for AMS.MC...

if (
    assets.isna().values.any()
):  # If has nan values, load the data from the fallback online database
    print(
        "WARNING: Asset data format is invalid or incomplete. "
        "Loading data from the remote backup database."
    )

    url = "https://raw.githubusercontent.com/Global-Data-Quantum/PortfolioData/1f3ebb95fdef245014a04d4273f688a2951e0061/data/qpo_tutorial_ibex.csv"
    assets = pd.read_csv(url, index_col=0)

Paso 2: Definir los datos de entrada del problema

Los parámetros necesarios para definir el problema QUBO se configuran en el diccionario qubo_settings . Definimos el número de pasos temporales (nt), el número de bits para la especificación de la inversión (nq) y la ventana temporal para cada paso temporal (dt). Además, fijamos la inversión máxima por activo, el coeficiente de aversión al riesgo, la comisión de transacción y el coeficiente de restricción (véase nuestro documento para más detalles sobre la formulación del problema). Estos ajustes nos permiten adaptar el problema QUBO al escenario de inversión específico.

qubo_settings = {
    "nt": 4,
    "nq": 2,
    "dt": 30,
    # maximum investment per asset is 2**nq/max_investment = 80%
    "max_investment": 5,
    "risk_aversion": 1000.0,
    "transaction_fee": 0.01,
    "restriction_coeff": 1.0,
}

El optimizer_settings diccionario configura el proceso de optimización, incluyendo parámetros como num_generations para el número de iteraciones y population_size para el número de soluciones candidatas por generación. Otros parámetros controlan aspectos como la tasa de recombinación, los trabajos en paralelo, el tamaño de los lotes y el rango de mutaciones. Además, los parámetros primitivos, como estimator_shots, estimator_precision, y sampler_shots, definen las configuraciones del estimador y el muestreador cuánticos para el proceso de optimización.

optimizer_settings = {
    "de_optimizer_settings": {
        "num_generations": 20,
        "population_size": 40,
        "recombination": 0.4,
        "max_parallel_jobs": 5,
        "max_batchsize": 4,
        "mutation_range": [0.0, 0.25],
    },
    "optimizer": "differential_evolution",
    "primitive_settings": {
        "estimator_shots": 25_000,
        "estimator_precision": None,
        "sampler_shots": 100_000,
    },
}

El ansatz_settings diccionario configura el ansatz del circuito cuántico. El ansatz parámetro especifica el uso del "optimized_real_amplitudes" enfoque, que es un ansatz eficiente en términos de hardware diseñado para problemas de optimización financiera. La multiple_passmanager opción, que permite el uso de múltiples gestores de pases (incluido el gestor de pases local predeterminado de Qiskit y el servicio de transpilador basado en IA de Qiskit) para mejorar el rendimiento y la eficiencia de la optimización de circuitos, está actualmente desactivada porque esta función está temporalmente deshabilitada.

ansatz_settings = {
    "ansatz": "optimized_real_amplitudes",
    "multiple_passmanager": False,
}

Por último, ejecutamos la optimización ejecutando la función dpo_solver.run() , pasando las entradas preparadas. Estos incluyen el diccionario de datos de activos (assets), la configuración de QUBO (qubo_settings), los parámetros de optimización (optimizer_settings), y los ajustes del ansatz del circuito cuántico (ansatz_settings). Además, especificamos los detalles de ejecución, como el backend, y si se debe aplicar post-procesamiento a los resultados. Esto inicia el proceso de optimización dinámica de la cartera en el backend cuántico seleccionado.

dpo_job = dpo_solver.run(
    assets=assets.to_dict(),  # Insert assets as a dictionary
    qubo_settings=qubo_settings,
    optimizer_settings=optimizer_settings,
    ansatz_settings=ansatz_settings,
    backend_name="ibm_torino",
    previous_session_id=[],
    apply_postprocess=True,
)

Paso 3: Analizar los resultados de la optimización

En esta sección, extraemos y mostramos la solución con el menor coste objetivo a partir de los resultados de la optimización. Junto con el coste objetivo mínimo, también presentamos métricas clave asociadas a la solución asociada, incluida la desviación de restricción, el ratio de Sharpe y el rendimiento de la inversión.

# Get the results of the job
dpo_result = dpo_job.result()

# Show the solution strategy
dpo_result["result"]

Output:

{'time_step_0': {'ACS.MC': 0.11764705882352941,
  'ITX.MC': 0.20588235294117646,
  'FER.MC': 0.38235294117647056,
  'ELE.MC': 0.058823529411764705,
  'SCYR.MC': 0.0,
  'AENA.MC': 0.058823529411764705,
  'AMS.MC': 0.17647058823529413},
 'time_step_1': {'ACS.MC': 0.11428571428571428,
  'ITX.MC': 0.14285714285714285,
  'FER.MC': 0.2,
  'ELE.MC': 0.02857142857142857,
  'SCYR.MC': 0.42857142857142855,
  'AENA.MC': 0.0,
  'AMS.MC': 0.08571428571428572},
 'time_step_2': {'ACS.MC': 0.0,
  'ITX.MC': 0.09375,
  'FER.MC': 0.3125,
  'ELE.MC': 0.34375,
  'SCYR.MC': 0.0,
  'AENA.MC': 0.0,
  'AMS.MC': 0.25},
 'time_step_3': {'ACS.MC': 0.3939393939393939,
  'ITX.MC': 0.09090909090909091,
  'FER.MC': 0.12121212121212122,
  'ELE.MC': 0.18181818181818182,
  'SCYR.MC': 0.0,
  'AENA.MC': 0.0,
  'AMS.MC': 0.21212121212121213}}

import pandas as pd

# Get results from the job
dpo_result = dpo_job.result()

# Convert metadata to a DataFrame, excluding 'session_id'
df = pd.DataFrame(dpo_result["metadata"]["all_samples_metrics"])

# Find the minimum objective cost
min_cost = df["objective_costs"].min()
print(f"Minimum Objective Cost Found: {min_cost:.2f}")

# Extract the row with the lowest cost
best_row = df[df["objective_costs"] == min_cost].iloc[0]

# Display the results associated with the best solution
print("Best Solution:")
print(f"  - Restriction Deviation: {best_row['rest_breaches']}%")
print(f"  - Sharpe Ratio: {best_row['sharpe_ratios']:.2f}")
print(f"  - Return: {best_row['returns']:.2f}")

Output:

Minimum Objective Cost Found: -3.67
Best Solution:
  - Restriction Deviation: 40.0%
  - Sharpe Ratio: 14.54
  - Return: 0.28

El siguiente código muestra cómo visualizar y comparar la distribución de costes de un algoritmo de optimización con una distribución de muestreo aleatorio. Del mismo modo, exploramos el panorama de la función objetivo QUBO (que se puede cargar desde la salida de la función) evaluándola con inversiones aleatorias. Trazamos ambas distribuciones normalizadas en amplitud para facilitar la comparación de cómo el proceso de optimización difiere del muestreo aleatorio en términos de coste. Además, el resultado obtenido con DOCPlex se incluye como una línea de referencia vertical discontinua que sirve como referencia clásica. Utilizamos la versión gratuita de DOCPlex, la IBM® biblioteca de código abierto para la optimización matemática en Python, para resolver el mismo problema de forma clásica.

import matplotlib.pyplot as plt
from matplotlib.ticker import MultipleLocator
import matplotlib.patheffects as patheffects


def plot_normalized(dpo_x, dpo_y_normalized, random_x, random_y_normalized):
    """
    Plots normalized results for two sampling results.

    Parameters:
        dpo_x (array-like): X-values for the VQE Post-processed curve.
        dpo_y_normalized (array-like): Y-values (normalized) for the
        VQE Post-processed curve.
        random_x (array-like): X-values for the Noise (Random) curve.
        random_y_normalized (array-like): Y-values (normalized) for the
        Noise (Random) curve.
    """
    plt.figure(figsize=(6, 3))
    plt.tick_params(axis="both", which="major", labelsize=12)

    # Define custom colors
    colors = ["#4823E8", "#9AA4AD"]

    # Plot DPO results
    (line1,) = plt.plot(
        dpo_x, dpo_y_normalized, label="VQE Postprocessed", color=colors[0]
    )
    line1.set_path_effects(
        [patheffects.withStroke(linewidth=3, foreground="white")]
    )

    # Plot Random results
    (line2,) = plt.plot(
        random_x, random_y_normalized, label="Noise (Random)", color=colors[1]
    )
    line2.set_path_effects(
        [patheffects.withStroke(linewidth=3, foreground="white")]
    )

    # Set X-axis ticks to increment by 5 units
    plt.gca().xaxis.set_major_locator(MultipleLocator(5))

    # Axis labels and legend
    plt.xlabel("Objective cost", fontsize=14)
    plt.ylabel("Normalized Counts", fontsize=14)

    # Add DOCPLEX reference line
    plt.axvline(
        x=-4.11, color="black", linestyle="--", linewidth=1, label="DOCPlex"
    )  # DOCPlex value
    plt.ylim(bottom=0)

    plt.legend()

    # Adjust layout
    plt.tight_layout()
    plt.show()

import numpy as np
from collections import defaultdict

# ================================
# STEP 1: DPO COST DISTRIBUTION
# ================================

# Extract data from DPO results
counts_list = dpo_result["metadata"]["all_samples_metrics"][
    "counts"
]  # List of how many times each solution occurred
cost_list = dpo_result["metadata"]["all_samples_metrics"][
    "objective_costs"
]  # List of corresponding objective function values (costs)

# Round costs to one decimal and accumulate counts for each unique cost
dpo_counter = defaultdict(int)
for cost, count in zip(cost_list, counts_list):
    rounded_cost = round(cost, 1)
    dpo_counter[rounded_cost] += count

# Prepare data for plotting
dpo_x = sorted(dpo_counter.keys())  # Sorted list of cost values
dpo_y = [dpo_counter[c] for c in dpo_x]  # Corresponding counts

# Normalize the counts to the range [0, 1] for better comparison
dpo_min = min(dpo_y)
dpo_max = max(dpo_y)
dpo_y_normalized = [
    (count - dpo_min) / (dpo_max - dpo_min) for count in dpo_y
]

# ================================
# STEP 2: RANDOM COST DISTRIBUTION
# ================================

# Read the QUBO matrix
qubo = np.array(dpo_result["metadata"]["qubo"])

bitstring_length = qubo.shape[0]
num_random_samples = 100_000  # Number of random samples to generate
random_cost_counter = defaultdict(int)

# Generate random bitstrings and calculate their cost
for _ in range(num_random_samples):
    x = np.random.randint(0, 2, size=bitstring_length)
    cost = float(x @ qubo @ x.T)
    rounded_cost = round(cost, 1)
    random_cost_counter[rounded_cost] += 1

# Prepare random data for plotting
random_x = sorted(random_cost_counter.keys())
random_y = [random_cost_counter[c] for c in random_x]

# Normalize the random cost distribution
random_min = min(random_y)
random_max = max(random_y)
random_y_normalized = [
    (count - random_min) / (random_max - random_min) for count in random_y
]

# ================================
# STEP 3: PLOTTING
# ================================

plot_normalized(dpo_x, dpo_y_normalized, random_x, random_y_normalized)

Output:

El gráfico muestra cómo el optimizador cuántico de carteras devuelve sistemáticamente estrategias de inversión optimizadas.

Referencias

[1] Nodar, Álvaro, Irene De León, Danel Arias, Ernesto Mamedaliev, María Esperanza Molina, Manuel Martín-Cordero, Senaida Hernández-Santana et al. «Ampliación del solucionador variacional de valores propios cuánticos para el « Portfolio Optimization » dinámico» arXiv Preimpresión arXiv:2412.19150 (2024).

Encuesta tutorial

Tómese un minuto para comentar este tutorial. Su opinión nos ayudará a mejorar nuestra oferta de contenidos y la experiencia de los usuarios. Enlace a la encuesta