أسعار الأسهم في الوقت الفعلي في Excel باستخدام Python

الأساس في تداول البورصة إنك تبقى على علم بآخر مستجدات الأسهم اللي بتتداولها، مع Excel Stock Price Calculator Template، تقدر تتابع سعر الأسهم اللي اشتريتها وتشوف أداء محفظتك بشكل مباشر، بس لو عايز تتعمّق أكتر، اتفرج على المعلومات المتاحة عن الأسهم اللي بتتداولها، المتداولين بيستخدموا الأسعار المحدثة في الوقت الحقيقي عشان يختاروا الوقت المناسب للشراء والبيع، والبيانات دي ممكن تتوفر بشكل مباشر في Excel.

الوقت هو أكتر حاجة مهمة في تحقيق النجاح في الاستثمار، فاستثمر في الشركات الكويسة قبل ما تتعدل الحالة.
بيتر لينش

يعني في التدوينة دي هنشرحلك إزاي تقدر تجيب الأسعار المحدثة في الوقت الحقيقي في Excel باستخدام شوية من كود Python، بدل ما تتعب نفسك وتدوس على زرار التحديث كل شوية. Python دلوقتي لغة برمجة مشهورة جدًا في صناعة المال، فلو كنت بدور على طريقة تساعدك في تحليل بيانات التداول بشكل أفضل، يُستحسن تتعلم شوية Python جنب Excel.

بايثون ولا إكسل؟ استخدم الاتنين مع بعض!

ناس كتير بتتكلم عن استخدام Python بدل Excel، بس الحقيقة إن Python مش بديل للشيت الإلكتروني! Excel هو أداة قوية جدًا في كل صناعة، ولسبب وجيه، Python لغة برمجة كويسة، بس بتركّز أكتر على المعالجة والتحليل الكبير للبيانات والبرمجة المتقدمة، لكن Excel لغة برمجة مدمجة فيها، بتساعدك تعمل حسابات سريعة وتقارير مختلفة وتنظيم بياناتك بشكل منظم.

في مهام معينة، خصوصًا اللي بتحتاج لحسابات معقدة، Python بيكون الخيار الأفضل، ولكن ممكن تستخدم Python و Excel مع بعض، لحسن الحظ مفيش حاجة تخلينا نختار بين الاتنين. ممكن نستدعي دوال بايثون من داخل إكسل باستخدام PyXLL، وهو اضافة بايثون لإكسل، دمج إكسل و بايثون عن طريق استخدام إكسل كواجهة أمامية (أو واجهة المستخدم) و بايثون كقوة دافعة للمهام المعقدة جدًا هو تطابق مثالي.

نستخدم بايثون عشان ناخد أسعار الأسهم الحقيقية بالوقت الحقيقي

أسعار الأسهم في الوقت الفعلي في Excel باستخدام Python — نستخدم بايثون عشان ناخد أسعار الأسهم الحقيقية بالوقت الحقيقي

اللي جاي ده بيفترض إنك عارف بايثون شوية، ولو مش عارف، متقلقش لو مفهمتش كل حاجة دلوقتي، ممكن تتابع معانا عادي. المصدر اللي هنستخدمه عشان نعرف أسعار الأسهم هو iextrading.com. عندهم API بإسم Socket.IO، هنستخدمها عشان ناخد تحديثات بالوقت الحقيقي. Socket.IO دي مكتبة لتطبيقات الويب اللي بتشتغل في الوقت الحقيقي، وبنقدر نوصل لـ API بتاعت IEX اللي بتستخدم Socket.IO باستخدام بايثون.

أول حاجة لازم نعملها هي تنزيل باكدج Socket.IO عشان نقدر نتكلم مع IEX، هننزلها باستخدام برنامج إدارة الحزم في بايثون Pip، من خلال الـ Command Prompt:

pip install python-socketio

دلوقتي في ملف بايثون، هنقدر نستدعي المكتبة ونربط مع السيرفر اللي في IEX، عشان يكون عندنا تحديثات أسعار الأسهم، هنحتاج نضيف بعض الدوال اللي العميل هيستدعيها في رد الفعل لأحداث معينة، وعشان تحقق ده، نقدر نستخدم الكود ده:

import socketio

endpoint = "https://ws-api.iextrading.com/1.0/tops"
symbols = [
    "AAPL",
    "MSFT",
    "SNAP"
]

@client.on("connect", namespace=namespace)
async def on_connect():
    for symbol in symbols:
        await client.emit("subscribe", symbol, namespace=namespace)
        print(f"Subscribed to '{symbol}'")

أول حاجة، لما العميل يوصل للسيرفر، بنرجعله بعض الأوامر عشان يشترك في بعض الأسهم (اللي بنسميها تيكرز عادة). لما العميل يوصل، الكول باك اللي بيتنادى “on_connect” هيتنادى مرة واحدة، لما يشترك في كل تيكر، بيتم طباعة رسالة في الـ output الخاص بالبايثون.

كول باك تانية هتحتاج تستقبل الداتا لما تتحدث الأسعار. كل ما الداتا بتتحدث، API socket.io بتاع IEX هتستدعي كول باك “message callback” مع الداتا بتاعتها مشفرة كـ JSON string. احنا بنستخدم مكتبة json في بايثون عشان نقرأ الـ JSON string ده ونحوله لـ Python dictionary object ونطبعه.

import json

@client.on("message", namespace=namespace)
def on_message(message):
    data = json.loads(message)
    print(data)

ممكن تكون لاحظت إن الكود اللي فوق بيستخدم async و await كلمات مفتاحية في بايثون. في Python 3.5 في المجال البرمجي الأسينكروني ده بيُستخدم كتير. هنا بنحاول نتجنب إن البرنامج كله يستني عملية معينة تخلص، زي انتظار الرد من سيرفر الويب البعيد. علي العكس Python بتستخدم حلقة الأحداث (Event Loop) عشان تسجل كل وظيفة، ولو وظيفة محتاجة تستني حاجة، Python بيعلق تنفيذ الوظيفة دي ويبدأ ينفذ أي حاجة تانية جاهزة.

عشان نكمل الكود ده، لازم نجيب Python asyncio event loop، ونجدول المهمة عشان ننشئ العميل اللي قبل كدا، ونبدأ event loop.

import asyncio

loop = asyncio.get_event_loop()
loop.create_task(task)
loop.run_forever()

لما نشغل الكود ده، الـ output اللي بيطلع بيبان كده، وبيتحدث عشان الأسعار بتتغير في الوقت الحقيقي.

Subscribed to 'AAPL'
Subscribed to 'MSFT'
Subscribed to 'SNAP'
{'symbol': 'AAPL', 'sector': 'technologyhardwareequipment', 'securityType': 'commonstock', 'bidPrice': 181.1, 'bidSize': 100, 'askPrice': 182.03, 'askSize': 100, 'lastUpdated': 1552491513626, 'lastSalePrice': 182.04, 'lastSaleSize': 100, 'lastSaleTime': 1552491513046, 'volume': 355975, 'marketPercent': 0.02927, 'seq': 84605}

تدفق البيانات من Python إلى Excel

في الجزء اللي فات، اتعلمنا إزاي نجيب أسعار الأسهم الحية من API Socket.IO من IEX. في الجزء ده هنشوف ازاي نستخدم البيانات دي عشان نعمل جدول في إكسل بيتحدث في الوقت الحقيقي.

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

بمجرد ما تحمل الإضافة PyXLL، هتتبع الإرشادات عشان تثبتها. في ملف الضبط pyxll.cfg، هتلاقي فيه إعدادات زي pythonpath وmodules. الـ pythonpath بتكون قايمة بالمجلدات اللي البايثون هيدور فيها عشان يحمل الموديولات اللي معرفينها في modules. الموديول ده مجرد ملف نصي فيه كود بايثون وبينتهي اسمه بـ “.py”.
هنبتدي بعمل موديول بايثون جديد (ملف نصي بينتهي باسم “.py”) ونحفظه في مكان معين (مثال: C:/Projects/ExcelStockPrices/iex.py). نضيف المجلد اللي احنا حفظنا فيه الملف ده لقيمة الـ pythonpath في ملف pyxll.cfg (مثال: C:/Projects/ExcelStockPrices)، ونضيف اسم الموديول في قائمة modules (مثال: iex). الـ module name مش بيشمل امتداد الملف “.py”.
لما تبدأ إكسل، أو تعيد تحميل الإضافة PyXLL، الـ module الجديد “iex” هيحمل.
بعد كدا هنكتب دالة في بايثون ممكن تستدعيها من إكسل. بنعمل كدا باستخدام الـ decorator @xl_func من PyXLL. عشان الدالة بترجع بيانات في الوقت الحقيقي، ومش بس قيمة واحدة، بنرجع نوع خاص اسمه RTD object. في بايثون هنعمل كلاس جديد من RTD class في PyXLL.

from pyxll import xl_func, RTD


class IEXClient(RTD):
    def __init__(self, symbols):
        super().__init__(self, value=None)
        self.symbols = symbols


@xl_func("str[] symbols: rtd<object>")
def iex(symbols):
    return IEXClient(symbols)

الـ function iex بتاخد قايمة من الـ strings وبترجع RTD object. الـ string اللي بيتم تمريره للـ decorator @xl_func في الأعلى هو توقيع الدالة. PyXLL بيستخدم الـ signature دي عشان يحول القيم اللي بيتم تمريرها من إكسل لبايثون، وعشان يحول قيمة الرجوع من بايثون لإكسل. الـ signature دي معناها إن symbols بتكون قائمة من الـ strings، والـ function هيرجع RTD object اللي قيمته بتكون Python object. القيمة اللي في الـ RTD object هي اللي هتظهر في إكسل.

لما بنستدعي الـ function iex بقائمة من الـ symbols، النتيجة في إكسل بتكون مؤشر على الـ Python object” None”. لو كان لينا نوع تاني لرجوع الدالة (مثال: “rtd”)، هيكون ممكن نرجع أنواع تانية لإكسل بدل مؤشر على Python object.

الـ RTD objects فيها دالتين، connect و disconnect. دول بيتم استدعاؤهم وقت ما إكسل يكون مستعد يبدأ يستقبل القيم، ولما إكسل يكون محتاج يوقف التحديث عن القيم. مثال: connect هيتم استدعاؤها لما المستخدم يدخل الصيغة، وdisconnect هيتم استدعائها لو المستخدم حذف الصيغة.

عشان نرسل قيمة جديدة لإكسل من الـ RTD object، كل اللي علينا نعمله هو إننا نعيد الـ value property على الـ RTD object. هنقدر ناخد الكود اللي كتبناه في الجزء اللي فات ونضيفه لـ IEXClient class. هنضيفها للـ connect method عشان نوصل لـ IEX server باستخدام Socket.IO client لما إكسل يوصل إلى الـ RTD object بتاعتنا. بدل ما نطبع البيانات بس، هنحفظها في Python dictionary عشان نقدر نوصل ليها بعد كده.

class IEXClient(RTD):

    endpoint = "https://ws-api.iextrading.com/1.0/tops"
    namespace = "/1.0/tops"

    def __init__(self, symbols):
        super().__init__(value={})
        self.symbols = symbols
        self.client = socketio.AsyncClient()
        self.data = {}

    async def connect(self):
        """Connect AsyncClient and subscribe to updates."""
        @self.client.on("connect", namespace=self.namespace)
        async def on_connect():
            for symbol in self.symbols:
                await self.client.emit("subscribe",
                                       symbol,
                                       namespace=self.namespace)

        @self.client.on("message", namespace=self.namespace)
        def on_message(message):
            latest = json.loads(message)
            symbol = latest["symbol"]

            previous = self.data.setdefault(symbol, {})
            previous.update(latest)

            # Notify Excel that there's been an update
            self.value = self.data

        await self.client.connect(self.endpoint,
                                  namespaces=[self.namespace])

    async def disconnect(self):
        await self.client.disconnect()

ملحوظة: إن connect و disconnect methods معلمين بـ async. PyXLL بيفهم الـ async methods وبيخلي الـ Python event loop يعملهم لينا.

دلوقتي لما نعيد تحميل إضافة PyXLL ويتم استدعاء الـ function تاني، النتيجة هي مؤشر على Python dictionary. ولما يحصل تحديث في الأسعار، الـ dictionary بيتحدث والقيمة المعروضة في إكسل بتتحدث.

لوحدها دي مش بتفيدنا كتير! لازم نلاقي طريقة لإخراج القيم من الـ Python dictionary ده للجدول الخاص بينا. الحل بسيط وهو هنكتب دالة تانية بتأخد الـ dictionary، symbol و key، وبترجع القيمة. توقيع الدالة دي بيقول إن الـ parameter الأول هو Python object، وبعد كدا الـ symbol و key كـ strings، والدالة بترجع قيمة من نوع متغير (يعني هتبقى string، رقم، أو نوع تاني).

@xl_func("object data, str symbols, str key: var")
def iex_unpack(data, symbol, key):
    # get the values for the symbol
    values = data.get(symbol, None)
    if not values:
        return "#NoData"

    # return the value for the specified key
    return values.get(key, "#NoData")

الـ function iex_unpack دي بيتم استدعائها بالمؤشر على الـ dictionary كـ argument الأول، والـ symbol والـ key، وبترجع القيمة من الـ dictionary. ولما يحصل تحديث في القيم والـ dictionary يتحدث، القيم بتتحدث في الجدول بردو.

حسابات المحفظة مع أسعار الأسهم في الوقت الحقيقي

دلوقتي عندنا كل اللي نحتاجه عشان نضيف الأسعار في الوقت الحقيقي لـ Portfolio Calculator! وده اذا كان موجود عندك أو تقدر تبحث عنه، لو بس بتحدث الورقة زي ما هي عندك، ولما تضغط على زرار “Calculate”، الخلايا اللي اتغيرت هتتغير تلقائي.

ممكن تجيب بيانات تانية غير الأسعار اللي في الوقت الحقيقي من IEX’s REST API. الكود اللي تحت ده بيوضح إزاي تجيب قيمة واحدة باستخدام الـ Python package اللي اسمه aiohttp.

from pyxll import xl_func
import aiohttp

endpoint = "https://api.iextrading.com/1.0/"

@xl_func
async def iex_fetch(symbol, key):
    """returns a value for a symbol from iextrading.com"""
    url = endpoint + f"stock/{symbol}/batch?types=quote"
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            assert response.status == 200
            data = await response.read()

    data = json.loads(data)["quote"]
    return data.get(key, "#NoData")

الـ function اللي فوق دي بتكون asynchronous. ده معناه إنك ممكن تستدعيها أكتر من مرة، وكل request لـ IEX REST endpoint هيشتغل بتوازن مع بعض، وده هيخلي الجدول يستجيب بشكل أسرع. عشان تجيب open price لأي سهم، هتستدعي الـ function اللي فوق في إكسل كدا:

=iex_fetch("AAPL", "open")

ممكن تلاقي معلومات أكتر عن IEX API على موقعهم الرسمي.

مهم: هل ممكن الاكسل يستخدم بايثون وإزاي؟

في النهاية

ده كان شرح كامل عن أسعار الأسهم في الوقت الفعلي في Excel باستخدام Python وكيفية استخدام Python لعرض أسعار الأسهم، وممكن كمان تطور الكود وتضيف وظائف تانية علي حسب احتياجاتك، زي عرض مؤشرات فنية أو رسم البيانات بشكل بصري.