Global Data Quantum의 포트폴리오 최적화기로 동적 포트폴리오 최적화 수행하기

참고

Qiskit 함수는 IBM Quantum® 프리미엄 플랜, 플렉스 플랜, 온프레미스(IBM Quantum Platform API 경유) 플랜 사용자에게만 제공되는 실험적 기능입니다. 현재 미리보기 출시 상태이며 변경될 수 있습니다.

사용 시간 추정: Heron r2 프로세서에서 약 55분. (참고: 이는 추정치이며, 실제 실행 시간은 다를 수 있습니다.)

배경

동적 포트폴리오 최적화 문제는 여러 기간에 걸쳐 예상 포트폴리오 수익을 극대화하고 위험을 최소화하기 위한 최적의 투자 전략을 찾는 것을 목표로 합니다. 이는 주로 예산, 거래 비용, 위험 회피 등의 제약 조건 하에서 수행됩니다. 단일 시점에 포트폴리오를 재조정하는 표준 포트폴리오 최적화와 달리, 동적 버전은 자산의 변화하는 특성을 고려하고 시간에 따른 자산 성과 변화에 따라 투자를 조정합니다.

이 튜토리얼에서는 양자 포트폴리오 최적화기 Qiskit 함수를 사용하여 동적 포트폴리오 최적화를 수행하는 방법을 설명합니다. 구체적으로, 이 애플리케이션 함수를 사용하여 여러 시간 단계에 걸친 투자 배분 문제를 해결하는 방법을 살펴봐요.

이 접근 방식은 포트폴리오 최적화를 다목적 이차 비제약 이진 최적화(QUBO) 문제로 공식화하는 것을 포함합니다. 구체적으로, 네 가지 목표를 동시에 최적화하기 위해 QUBO 함수 $O$를 공식화합니다:

• 수익 함수 $F$ 극대화
• 투자 위험 $R$ 최소화
• 거래 비용 $C$ 최소화
• 추가 항으로 공식화된 투자 제한 조건 $P$ 준수 및 최소화.

요약하면, 이러한 목표를 달성하기 위해 QUBO 함수를 다음과 같이 공식화합니다: $O = -F + \frac{\gamma}{2} R + C + \rho P,$ 여기서 $\gamma$는 위험 회피 계수이고 $\rho$는 제약 강화 계수(라그랑주 승수)입니다. 명시적인 공식은 논문 [1]의 수식 (15)에서 확인할 수 있습니다.

변분 양자 고유값 분해기(VQE)를 기반으로 한 하이브리드 양자-고전 방법을 사용하여 문제를 풉니다. 이 설정에서 양자 Circuit은 비용 함수를 추정하고, 고전적 최적화는 차등 진화 알고리즘을 사용하여 수행되어 솔루션 공간을 효율적으로 탐색할 수 있습니다. 필요한 Qubit 수는 세 가지 주요 요소에 따라 달라집니다: 자산 수 na, 기간 수 nt, 그리고 투자를 나타내는 데 사용되는 비트 해상도 nq. 구체적으로, 문제에서 필요한 최소 Qubit 수는 na*nt*nq입니다.

이 튜토리얼에서는 스페인 IBEX 35 지수를 기반으로 한 지역 포트폴리오 최적화에 집중합니다. 구체적으로, 아래 표에 나타난 7개 자산 포트폴리오를 사용합니다:

IBEX 35 포트폴리오	ACS.MC	ITX.MC	FER.MC	ELE.MC	SCYR.MC	AENA.MC	AMS.MC

2022년 11월 1일부터 시작하여 30일 간격으로 구분된 네 번의 시간 단계에서 포트폴리오를 재조정합니다. 각 투자 변수는 2비트를 사용하여 인코딩됩니다. 이로 인해 56개의 Qubit이 필요한 문제가 됩니다.

최적화된 실수 진폭(Optimized Real Amplitudes) 앤사츠를 사용합니다. 이는 이러한 유형의 금융 최적화 문제에 대한 성능을 향상시키기 위해 특별히 맞춤화된 표준 실수 진폭 앤사츠의 하드웨어 효율적인 적응형 버전입니다.

양자 실행은 ibm_torino Backend에서 수행됩니다. 문제 공식화, 방법론 및 성능 평가에 대한 자세한 설명은 게재된 논문 [1]을 참조하세요.

요구 사항

# Added by doQumentation — required packages for this notebook
!pip install -q numpy

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

설정

양자 포트폴리오 최적화기를 사용하려면 Qiskit 함수 카탈로그를 통해 함수를 선택하세요. 이 함수를 실행하려면 IBM Quantum 프리미엄 플랜 또는 플렉스 플랜 계정과 Global Data Quantum의 라이선스가 필요합니다.

먼저 API 키로 인증하세요. 그런 다음 Qiskit 함수 카탈로그에서 원하는 함수를 로드합니다. 여기서는 QiskitFunctionsCatalog 클래스를 사용하여 카탈로그에서 quantum_portfolio_optimizer 함수에 접근합니다. 이 함수를 통해 사전 정의된 양자 포트폴리오 최적화 솔버를 사용할 수 있습니다.

from qiskit_ibm_catalog import QiskitFunctionsCatalog

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

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

1단계: 입력 포트폴리오 읽기

이 단계에서는 IBEX 35 지수에서 선택한 7개 자산의 역사적 데이터를 2022년 11월 1일부터 2023년 4월 1일까지 로드합니다.

Yahoo Finance API를 사용하여 종가에 집중하여 데이터를 가져옵니다. 그런 다음 모든 자산이 동일한 수의 데이터 날짜를 갖도록 데이터를 처리합니다. 누락된 데이터(비거래일)는 적절히 처리되어 모든 자산이 동일한 날짜에 정렬되도록 합니다.

데이터는 모든 자산에 걸쳐 일관된 형식으로 DataFrame에 구조화됩니다.

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
df = pd.concat(series_list, axis=1)

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

# Convert DataFrame to dictionary
assets = df.to_dict()

[*********************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...

Step 2: 문제 입력값 정의

QUBO 문제를 정의하는 데 필요한 파라미터는 qubo_settings 딕셔너리에 설정됩니다. 시간 단계 수(nt), 투자 명세를 위한 비트 수(nq), 각 시간 단계의 시간 창(dt)을 정의합니다. 또한 자산당 최대 투자액, 위험 회피 계수, 거래 수수료, 제약 계수를 설정합니다(문제 공식화에 대한 자세한 내용은 논문을 참고하세요). 이러한 설정을 통해 특정 투자 시나리오에 맞게 QUBO 문제를 조정할 수 있습니다.

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

optimizer_settings 딕셔너리는 최적화 프로세스를 구성하며, num_generations(반복 횟수)와 population_size(세대당 후보 솔루션 수) 등의 파라미터를 포함합니다. 그 외의 설정은 재조합 비율, 병렬 작업, 배치 크기, 변이 범위 등을 제어합니다. 또한 estimator_shots, estimator_precision, sampler_shots와 같은 프리미티브 설정은 최적화 프로세스에서 사용하는 양자 Estimator 및 Sampler 구성을 정의합니다.

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

참고

회로의 총 수는 optimizer_settings 파라미터에 따라 달라지며, (num_generations + 1) * population_size로 계산됩니다.

ansatz_settings 딕셔너리는 양자 회로 앤사츠(ansatz)를 구성합니다. ansatz 파라미터는 금융 최적화 문제를 위해 설계된 하드웨어 효율적 앤사츠인 "optimized_real_amplitudes" 방식의 사용을 지정합니다. 또한 multiple_passmanager 설정을 활성화하면 최적화 프로세스 중 여러 패스 매니저(기본 로컬 Qiskit 패스 매니저 및 Qiskit AI 기반 Transpiler 서비스 포함)를 허용하여 회로 실행의 전반적인 성능과 효율성을 향상시킵니다.

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

마지막으로, 준비된 입력값을 dpo_solver.run() 함수에 전달하여 최적화를 실행합니다. 입력값에는 자산 데이터 딕셔너리(assets), QUBO 구성(qubo_settings), 최적화 파라미터(optimizer_settings), 양자 회로 앤사츠 설정(ansatz_settings)이 포함됩니다. 또한 Backend, 결과에 후처리를 적용할지 여부 등 실행 세부 사항도 지정합니다. 이를 통해 선택한 양자 Backend에서 동적 포트폴리오 최적화 프로세스가 시작됩니다.

dpo_job = dpo_solver.run(
    assets=assets,
    qubo_settings=qubo_settings,
    optimizer_settings=optimizer_settings,
    ansatz_settings=ansatz_settings,
    backend_name="ibm_torino",
    previous_session_id=[],
    apply_postprocess=True,
)

Step 3: 최적화 결과 분석

이 섹션에서는 최적화 결과에서 목적 비용이 가장 낮은 솔루션을 추출하여 표시합니다. 최소 목적 비용과 함께, 해당 솔루션에 관련된 핵심 지표인 제약 편차, 샤프 비율, 투자 수익률도 함께 제시합니다.

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

# Show the solution strategy
dpo_result["result"]

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

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

다음 코드는 최적화 알고리즘의 비용 분포와 무작위 샘플링 분포를 시각화하고 비교하는 방법을 보여줍니다. 마찬가지로, 무작위 투자로 평가하여 QUBO 목적 함수의 경관(함수 출력에서 로드 가능)을 탐색합니다. 최적화 프로세스가 비용 측면에서 무작위 샘플링과 어떻게 다른지 쉽게 비교할 수 있도록 두 분포를 진폭 기준으로 정규화하여 플롯합니다. 또한 고전적 벤치마크로 활용하기 위해 DOCPlex를 사용한 결과를 수직 점선 참조선으로 포함시킵니다. 동일한 문제를 고전적으로 풀기 위해 Python의 수학 최적화를 위한 IBM® 오픈소스 라이브러리인 DOCPlex 무료 버전을 사용합니다.

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"][
    "objective_costs"
]  # List of how many times each solution occurred
cost_list = dpo_result["metadata"]["all_samples_metrics"][
    "counts"
]  # 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)

그래프는 양자 포트폴리오 최적화 도구가 일관되게 최적화된 투자 전략을 반환하는 방식을 보여줍니다.

참고 문헌

[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. "Scaling the Variational Quantum Eigensolver for Dynamic Portfolio Optimization." arXiv preprint arXiv:2412.19150 (2024).

튜토리얼 설문

이 튜토리얼에 대한 피드백을 남겨 주세요. 여러분의 의견은 콘텐츠 품질과 사용자 경험을 개선하는 데 큰 도움이 됩니다. 설문 링크

Note: This survey is provided by IBM Quantum and relates to the original English content. To give feedback on doQumentation's website, translations, or code execution, please open a GitHub issue.