إجراء تحسين ديناميكي للمحافظ الاستثمارية باستخدام محسِّن المحافظ من Global Data Quantum
دوال Qiskit ميزة تجريبية متاحة فقط لمستخدمي IBM Quantum® Premium Plan وFlex Plan وOn-Prem (عبر IBM Quantum Platform API) Plan. وهي في حالة إصدار معاينة وقابلة للتغيير.
تقدير الاستخدام: ما يقارب 55 دقيقة على معالج Heron r2. (ملاحظة: هذا تقدير فحسب. قد يختلف وقت التشغيل الفعلي.)
الخلفية
تهدف مسألة تحسين المحافظ الاستثمارية الديناميكية إلى إيجاد الاستراتيجية الاستثمارية المثلى عبر فترات زمنية متعددة لتعظيم العائد المتوقع للمحفظة وتقليل المخاطر، وغالباً في ظل قيود معينة كالميزانية وتكاليف المعاملات أو درجة تجنب المخاطر. خلافاً لتحسين المحافظ الاستثمارية المعتاد الذي يأخذ في الاعتبار لحظة زمنية واحدة لإعادة توازن المحفظة، يراعي الإصدار الديناميكي الطبيعة المتطورة للأصول ويُكيِّف الاستثمارات بناءً على التغيرات في أداء الأصول عبر الزمن.
يوضح هذا البرنامج التعليمي كيفية إجراء تحسين ديناميكي للمحافظ الاستثمارية باستخدام دالة Qiskit الخاصة بمحسِّن المحافظ الكمومية. تحديداً، نوضح كيفية استخدام دالة التطبيق هذه لحل مسألة تخصيص الاستثمار عبر خطوات زمنية متعددة.
يتضمن النهج المتبع صياغة تحسين المحافظ الاستثمارية على هيئة مسألة تحسين ثنائي تربيعي غير مقيد متعدد الأهداف (QUBO). تحديداً، نصيغ دالة QUBO المسماة لتحسين أربعة أهداف مختلفة في آنٍ واحد:
- تعظيم دالة العائد
- تقليل مخاطر الاستثمار
- تقليل تكاليف المعاملات
- الامتثال للقيود الاستثمارية، المصاغة في حد إضافي للتقليل .
خلاصة القول، ولمعالجة هذه الأهداف نصيغ دالة QUBO على النحو الآتي: حيث هو معامل تجنب المخاطر و هو معامل تعزيز القيود (مُضاعِف لاغرانج). يمكن الاطلاع على الصياغة التفصيلية في المعادلة (15) من ورقتنا البحثية [1].
نحل المسألة باستخدام طريقة هجينة كمومية-كلاسيكية مبنية على حل القيم الذاتية الكمومي التغايري (VQE). في هذا الإطار، تقدِّر الدائرة الكمومية دالة التكلفة، فيما يُنفَّذ التحسين الكلاسيكي باستخدام خوارزمية التطور التفاضلي، مما يُتيح التنقل الفعّال في فضاء ال حلول. يعتمد عدد البتات الكمومية المطلوبة على ثلاثة عوامل رئيسية: عدد الأصول na، وعدد الفترات الزمنية nt، ودقة البت المستخدمة لتمثيل الاستثمار nq. تحديداً، الحد الأدنى لعدد البتات الكمومية في مسألتنا هو na*nt*nq.
في هذا البرنامج التعليمي، نركز على تحسين محفظة إقليمية مبنية على مؤشر IBEX 35 الإسباني. تحديداً، نستخدم محفظة من سبعة أصول كما هو موضح في الجدول أدناه:
| محفظة IBEX 35 | ACS.MC | ITX.MC | FER.MC | ELE.MC | SCYR.MC | AENA.MC | AMS.MC |
|---|
نعيد توازن محفظتنا في أربع خطوات زمنية، تفصل بين كل منها فترة 30 يوماً بدءاً من 1 نوفمبر 2022. يُرمَّز كل متغير استثماري باستخدام بتين. ينتج عن ذلك مسألة تتطلب 56 بتاً كمومياً لحلها.
نستخدم ansatz المسمى Optimized Real Amplitudes، وهو تكيُّف مخصص وفعّال من حيث الأجهزة لـ ansatz Real Amplitudes المعتاد، مصمم خصيصاً لتحسين الأداء في هذا النوع من مسائل التحسين المالي.
يُنفَّذ التشغيل الكمومي على خلفية ibm_torino. للاطلاع على شرح مفصل لصياغة المسألة والمنهجية وتقييم الأداء، راجع الورقة البحثية المنشورة [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 Functions. تحتاج إلى حساب IBM Quantum Premium Plan أو Flex Plan مع ترخيص من Global Data Quantum لتشغيل هذه الدالة.
أولاً، قم بالمصادقة باستخدام مفتاح API الخاص بك. ثم، قم بتحميل الدالة المطلوبة من كتالوج Qiskit Functions. هنا، تصل إلى دالة quantum_portfolio_optimizer من الكتالوج باستخدام الفئة QiskitFunctionsCatalog. تتيح لنا هذه الدالة استخدام حل تحسين المحافظ الكمومية المحدد مسبقاً.
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")
الخطوة الأولى: قراءة المحفظة الاستثمارية المدخلة
في هذه الخطوة، نحمّل البيانات التاريخية للأصول السبعة المختارة من مؤشر IBEX 35، تحديداً من 1 نوفمبر 2022 إلى 1 أبريل 2023.
نجلب البيانات باستخدام واجهة برمجة تطبيقات Yahoo Finance، مع التركيز على أسعار الإغلاق. تُعالَج البيانات بعد ذلك للتأكد من أن جميع الأصول لها نفس عدد الأيام التي تتوفر فيها بيانات. تُعالَج أي بيانات مفقودة (أيام غير تداولية) بشكل مناسب، مما يضمن محاذاة جميع الأصول على نفس التواريخ.
تُهيكَل البيانات في 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...
الخطوة الثانية: تحديد مدخلات المسألة
تُهيَّأ المعاملات اللازمة لتعريف مسألة 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، تهيئات المُقدِّر الكمومي والمُعيِّن لعملية التحسين.
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"، وهو ansatz فعّال من حيث الأجهزة مصمم لمسائل التحسين المالي. بالإضافة إلى ذلك، يُفعَّل إعداد multiple_passmanager للسماح بمديري مرور متعددين (بما في ذلك مدير المرور المحلي الافتراضي لـ Qiskit وخدمة المترجِم المدعومة بالذكاء الاصطناعي من Qiskit) أثناء عملية التحسين، مما يُحسِّن الأداء الإجمالي وكفاءة تنفيذ الدائرة.
ansatz_settings = {
"ansatz": "optimized_real_amplitudes",
"multiple_passmanager": False,
}
أخيراً، ننفذ التحسين بتشغيل دالة dpo_solver.run()، مع تمرير المدخلات المُعدَّة. تشمل هذه المدخلات قاموس بيانات الأصول (assets)، وتهيئة QUBO (qubo_settings)، ومعاملات التحسين (optimizer_settings)، وإعدادات ansatz الدائرة الكمومية (ansatz_settings). بالإضافة إلى ذلك، نحدد تفاصيل التنفيذ مثل الخلفية، وما إذا كنا نريد تطبيق المعالجة اللاحقة على النتائج. يُطلق هذا عملية تحسين المحافظ الاستثمارية الديناميكية على الخلفية الكمومية المختارة.
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,
)
الخطوة الثالثة: تحليل نتائج التحسين
في هذا القسم، نستخرج الحل ذا أدنى تكلفة هدفية من نتائج التحسين ونعرضه. إلى جانب الحد الأدنى لتكلفة الهدف، نعرض أيضاً المقاييس الرئيسية المرتبطة بالحل المقابل، بما في ذلك انحراف القيود، ونسبة Sharpe، وعائد الاستثمار.
# 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 كخط مرجعي منقط عمودي ليكون بمثابة معيار مقارنة كلاسيكي. نستخدم الإصدار المجاني من DOCPlex — مكتبة IBM® مفتوحة المصدر للتحسين الرياضي في Python — لحل نفس المسألة بالطريقة الكلاسيكية.
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)
يوضح الرسم البياني كيف يُعيد محسِّن المحافظ الكمومية باستمرار استراتيجيات استثمار مُحسَّنة.
المراجع
استطلاع البرنامج التعليمي
يُرجى تخصيص دقيقة لتقديم ملاحظاتك حول هذا البرنامج التعليمي. ستساعدنا آراؤك في تحسين محتوى عروضنا وتجربة المستخدم. رابط الاستطلاع