ফাউস্টের পটভূমির কাজ, পার্ট II: এজেন্ট এবং দল

বিষয়বস্তু সূচি

1. পর্ব I: ভূমিকা

2. পার্ট II: এজেন্ট এবং দল

আমরা এখানে কি করছি?

সুতরাং, তাই, দ্বিতীয় অংশ. যেমনটি আগে লেখা হয়েছে, এতে আমরা নিম্নলিখিতগুলি করব:

1. আমাদের প্রয়োজনীয় শেষ পয়েন্টগুলির জন্য অনুরোধ সহ aiohttp-এ alphavantage-এর জন্য একটি ছোট ক্লায়েন্ট লিখি।

2. আসুন একটি এজেন্ট তৈরি করি যারা সিকিউরিটিজ এবং তাদের উপর মেটা তথ্য সংগ্রহ করবে।

কিন্তু, আমরা এই প্রকল্পের জন্যই এটি করব, এবং ফাস্ট গবেষণার পরিপ্রেক্ষিতে, আমরা শিখব কীভাবে এজেন্ট লিখতে হয় যা কাফকা থেকে ইভেন্টগুলি প্রসেস করে, সেইসাথে কীভাবে কমান্ড লিখতে হয় (ক্লিক র্যাপার), আমাদের ক্ষেত্রে - এজেন্ট যে বিষয়ে নজরদারি করছে সেই বিষয়ে ম্যানুয়াল পুশ বার্তার জন্য।

প্রশিক্ষণ

আলফাভ্যান্টেজ ক্লায়েন্ট

প্রথমে, অ্যালফাভান্টেজের অনুরোধের জন্য একটি ছোট aiohttp ক্লায়েন্ট লিখি।

alphavantage.py

ভক্ষক

import urllib.parse as urlparse
from io import StringIO
from typing import Any, Dict, List, Union

import aiohttp
import pandas as pd
import stringcase
from loguru import logger

from horton.config import API_ENDPOINT


class AlphaVantageClient:
    def __init__(
        self,
        session: aiohttp.ClientSession,
        api_key: str,
        api_endpoint: str = API_ENDPOINT,
    ):
        self._query_params = {"datatype": "json", "apikey": api_key}
        self._api_endpoint = api_endpoint
        self._session = session

    @logger.catch
    def _format_fields(self, data: Dict[str, Any]) -> Dict[str, Any]:
        formatted_data = {}

        for field, item in data.items():
            formatted_data[stringcase.snakecase(field)] = item

        return formatted_data

    @logger.catch
    async def _construct_query(
        self, function: str, to_json: bool = True, **kwargs
    ) -> Union[Dict[str, Any], str]:
        path = "query/"

        async with self._session.get(
            urlparse.urljoin(self._api_endpoint, path),
            params={"function": function, **kwargs, **self._query_params},
        ) as response:
            data = (await response.json()) if to_json else (await response.text())

            if to_json:
                data = self._format_fields(data)

        return data

    @logger.catch
    async def get_securities(self, state: str = "active") -> List[Dict[str, str]]:
        data = await self._construct_query("LISTING_STATUS", state=state, to_json=False)

        data = pd.read_csv(StringIO(data))

        securities = data.to_dict("records")

        for index, security in enumerate(securities):
            security = self._format_fields(security)
            security["_type"] = "physical"

            securities[index] = security

        return securities

    @logger.catch
    async def get_security_overview(self, symbol: str) -> Dict[str, str]:
        return await self._construct_query("OVERVIEW", symbol=symbol)

    @logger.catch
    async def get_historical_data(self, symbol: str) -> Dict[str, Any]:
        return await self._construct_query(
            "TIME_SERIES_DAILY_ADJUSTED", symbol=symbol, outputsize="full"
        )

    @logger.catch
    async def get_last_price_data(self, symbol: str) -> Dict[str, Any]:
        return await self._construct_query("GLOBAL_QUOTE", symbol=symbol)

    @logger.catch
    async def get_indicator_data(
        self, symbol: str, indicator: str, **indicator_options
    ) -> Dict[str, Any]:
        return await self._construct_query(
            indicator, symbol=symbol, **indicator_options
        )

প্রকৃতপক্ষে, এটি থেকে সবকিছু পরিষ্কার:

1. AlphaVantage API বেশ সহজ এবং সুন্দরভাবে ডিজাইন করা হয়েছে, তাই আমি পদ্ধতির মাধ্যমে সমস্ত অনুরোধ করার সিদ্ধান্ত নিয়েছি construct_query যেখানে ঘুরে একটি http কল আছে।

2. আমি সব ক্ষেত্র আনা snake_case আরামের জন্য

3. ভাল, সুন্দর এবং তথ্যপূর্ণ ট্রেসব্যাক আউটপুট জন্য logger.catch সজ্জা.

PS config.yml-এ স্থানীয়ভাবে alphavantage টোকেন যোগ করতে ভুলবেন না, অথবা এনভায়রনমেন্ট ভেরিয়েবল এক্সপোর্ট করতে ভুলবেন না HORTON_SERVICE_APIKEY. আমরা একটি টোকেন গ্রহণ করি এখানে.

CRUD ক্লাস

সিকিউরিটিজ সম্পর্কে মেটা তথ্য সংরক্ষণ করার জন্য আমাদের একটি সিকিউরিটিজ সংগ্রহ থাকবে।

database/security.py

আমার মতে, এখানে কিছু ব্যাখ্যা করার প্রয়োজন নেই, এবং বেস ক্লাস নিজেই বেশ সহজ।

অ্যাপ টি নিন()

একটি অ্যাপ্লিকেশন বস্তু তৈরি করার জন্য একটি ফাংশন যোগ করা যাক app.py

ভক্ষক

import faust

from horton.config import KAFKA_BROKERS


def get_app():
    return faust.App("horton", broker=KAFKA_BROKERS)

আপাতত আমাদের কাছে সহজতম অ্যাপ্লিকেশন তৈরি হবে, একটু পরে আমরা এটিকে প্রসারিত করব, তবে, আপনাকে অপেক্ষা না করার জন্য, এখানে তথ্যসূত্র অ্যাপ-ক্লাসে। আমি আপনাকে সেটিংস ক্লাসটি একবার দেখার পরামর্শ দিচ্ছি, যেহেতু এটি বেশিরভাগ সেটিংসের জন্য দায়ী।

প্রধান অংশ

সিকিউরিটিজের তালিকা সংগ্রহ ও রক্ষণাবেক্ষণের জন্য এজেন্ট

app = get_app()

collect_securities_topic = app.topic("collect_securities", internal=True)

@app.agent(collect_securities_topic)
async def collect_securities(stream: StreamT[None]) -> AsyncIterable[bool]:
	pass

সুতরাং, প্রথমে আমরা ফাস্ট অ্যাপ্লিকেশন অবজেক্টটি পাই - এটি বেশ সহজ। এর পরে, আমরা স্পষ্টভাবে আমাদের এজেন্টের জন্য একটি বিষয় ঘোষণা করি... এখানে এটি কী, অভ্যন্তরীণ প্যারামিটার কী এবং এটিকে কীভাবে আলাদাভাবে সাজানো যেতে পারে তা উল্লেখ করার মতো।

1. কাফকার টপিকস, যদি আমরা সঠিক সংজ্ঞা জানতে চাই, তাহলে পড়া ভালো বন্ধ নথি, অথবা আপনি পড়তে পারেন বিমূর্ত রাশিয়ান ভাষায় হাব্রেতে, যেখানে সবকিছু বেশ সঠিকভাবে প্রতিফলিত হয় :)

2. অভ্যন্তরীণ প্যারামিটার, faust ডক-এ বেশ ভালভাবে বর্ণনা করা হয়েছে, আমাদেরকে কোডে সরাসরি বিষয় কনফিগার করতে দেয়, অবশ্যই, এর অর্থ ফাস্ট ডেভেলপারদের দ্বারা প্রদত্ত পরামিতিগুলি, উদাহরণস্বরূপ: ধারণ, ধারণ নীতি (ডিফল্টরূপে মুছে ফেলুন, তবে আপনি সেট করতে পারেন নিচ্ছিদ্র), বিষয় প্রতি পার্টিশনের সংখ্যা (পার্টিশনকরতে, উদাহরণস্বরূপ, এর চেয়ে কম বিশ্বব্যাপী তাৎপর্য অ্যাপ্লিকেশন ফাস্ট)।

3. সাধারণভাবে, এজেন্ট বিশ্বব্যাপী মান সহ একটি পরিচালিত বিষয় তৈরি করতে পারে, তবে, আমি স্পষ্টভাবে সবকিছু ঘোষণা করতে চাই। উপরন্তু, এজেন্ট বিজ্ঞাপনে বিষয়ের কিছু প্যারামিটার (উদাহরণস্বরূপ, পার্টিশনের সংখ্যা বা ধরে রাখার নীতি) কনফিগার করা যাবে না।

   ম্যানুয়ালি বিষয়টি সংজ্ঞায়িত না করে এটি দেখতে কেমন হতে পারে তা এখানে:

app = get_app()

@app.agent()
async def collect_securities(stream: StreamT[None]) -> AsyncIterable[bool]:
	pass

আচ্ছা, এখন আমাদের এজেন্ট কী করবে তা বর্ণনা করি :)

app = get_app()

collect_securities_topic = app.topic("collect_securities", internal=True)

@app.agent(collect_securities_topic)
async def collect_securities(stream: StreamT[None]) -> AsyncIterable[bool]:
    async with aiohttp.ClientSession() as session:
        async for _ in stream:
            logger.info("Start collect securities")

            client = AlphaVantageClient(session, API_KEY)

            securities = await client.get_securities()

            for security in securities:
                await SecurityCRUD.update_one(
                    {"symbol": security["symbol"], "exchange": security["exchange"]}, security, upsert=True
                )

            yield True

সুতরাং, এজেন্টের শুরুতে, আমরা আমাদের ক্লায়েন্টের মাধ্যমে অনুরোধের জন্য একটি aiohttp সেশন খুলি। এইভাবে, একজন কর্মী শুরু করার সময়, যখন আমাদের এজেন্ট চালু করা হয়, তখনই একটি সেশন খোলা হবে - এক, পুরো সময় কর্মী চলছে (অথবা একাধিক, যদি আপনি প্যারামিটার পরিবর্তন করেন) সম্পাতবিন্দু একটি ডিফল্ট ইউনিট সহ একটি এজেন্ট থেকে)।

এরপরে, আমরা স্ট্রীম অনুসরণ করি (আমরা বার্তাটি রাখি _, যেহেতু আমরা, এই এজেন্টে, আমাদের বিষয় থেকে বার্তাগুলির বিষয়বস্তু সম্পর্কে চিন্তা করি না, যদি সেগুলি বর্তমান অফসেটে বিদ্যমান থাকে, অন্যথায় আমাদের চক্র তাদের আগমনের জন্য অপেক্ষা করবে। ঠিক আছে, আমাদের লুপের ভিতরে, আমরা বার্তার রসিদ লগ করি, সক্রিয়ের একটি তালিকা পাই (get_securities শুধুমাত্র ডিফল্টরূপে সক্রিয় হয়, ক্লায়েন্ট কোড দেখুন) সিকিউরিটিজ এবং ডাটাবেসে সংরক্ষণ করি, একই টিকারের সাথে একটি নিরাপত্তা আছে কিনা তা পরীক্ষা করে এবং ডাটাবেসে বিনিময়, যদি থাকে, তবে এটি (কাগজ) কেবল আপডেট করা হবে।

আমাদের সৃষ্টি চালু করা যাক!

> docker-compose up -d
... Запуск контейнеров ...
> faust -A horton.agents worker --without-web -l info

পিএস বৈশিষ্ট্য ওয়েব উপাদান আমি নিবন্ধগুলিতে ফাস্ট বিবেচনা করব না, তাই আমরা উপযুক্ত পতাকা সেট করেছি।

আমাদের লঞ্চ কমান্ডে, আমরা তথ্য লগ আউটপুট স্তরের সাথে অ্যাপ্লিকেশন অবজেক্টটি কোথায় খুঁজতে হবে এবং এটির সাথে কী করতে হবে (একজন কর্মী লঞ্চ করুন) বলেছিলাম। আমরা নিম্নলিখিত আউটপুট পেতে:

ভক্ষক

┌ƒaµS† v1.10.4┬───────────────────────────────────────────────────┐
│ id          │ horton                                            │
│ transport   │ [URL('kafka://localhost:9092')]                   │
│ store       │ memory:                                           │
│ log         │ -stderr- (info)                                   │
│ pid         │ 1271262                                           │
│ hostname    │ host-name                                         │
│ platform    │ CPython 3.8.2 (Linux x86_64)                      │
│ drivers     │                                                   │
│   transport │ aiokafka=1.1.6                                    │
│   web       │ aiohttp=3.6.2                                     │
│ datadir     │ /path/to/project/horton-data                      │
│ appdir      │ /path/to/project/horton-data/v1                   │
└─────────────┴───────────────────────────────────────────────────┘
... логи, логи, логи ...

┌Topic Partition Set─────────┬────────────┐
│ topic                      │ partitions │
├────────────────────────────┼────────────┤
│ collect_securities         │ {0-7}      │
│ horton-__assignor-__leader │ {0}        │
└────────────────────────────┴────────────┘ 

এটা জীবিত!!!

আসুন পার্টিশন সেটটি দেখি। আমরা দেখতে পাচ্ছি, একটি বিষয় তৈরি করা হয়েছিল যে নামটি আমরা কোডে মনোনীত করেছি, পার্টিশনের ডিফল্ট সংখ্যা (8, থেকে নেওয়া topic_partitions - অ্যাপ্লিকেশন অবজেক্ট প্যারামিটার), যেহেতু আমরা আমাদের বিষয়ের জন্য একটি পৃথক মান নির্দিষ্ট করিনি (পার্টিশনের মাধ্যমে)। ওয়ার্কারে লঞ্চ করা এজেন্টকে সমস্ত 8টি পার্টিশন বরাদ্দ করা হয়েছে, যেহেতু এটি একমাত্র, তবে এটি ক্লাস্টারিং সম্পর্কে অংশে আরও বিশদে আলোচনা করা হবে।

ঠিক আছে, এখন আমরা অন্য টার্মিনাল উইন্ডোতে যেতে পারি এবং আমাদের বিষয়ে একটি খালি বার্তা পাঠাতে পারি:

> faust -A horton.agents send @collect_securities
{"topic": "collect_securities", "partition": 6, "topic_partition": ["collect_securities", 6], "offset": 0, "timestamp": ..., "timestamp_type": 0}

পিএস ব্যবহার করে @ আমরা দেখাই যে আমরা "collect_securities" নামের একটি বিষয়ে একটি বার্তা পাঠাচ্ছি।

এই ক্ষেত্রে, বার্তাটি পার্টিশন 6-এ গিয়েছিল - আপনি ক্যাফড্রপ অন-এ গিয়ে এটি পরীক্ষা করতে পারেন localhost:9000

আমাদের কর্মীর সাথে টার্মিনাল উইন্ডোতে গিয়ে, আমরা loguru ব্যবহার করে পাঠানো একটি খুশির বার্তা দেখতে পাব:

2020-09-23 00:26:37.304 | INFO     | horton.agents:collect_securities:40 - Start collect securities

আমরা মঙ্গো (Robo3T বা Studio3T ব্যবহার করে) দেখতে পারি এবং দেখতে পারি যে সিকিউরিটিগুলি ডাটাবেসে রয়েছে:

আমি বিলিয়নেয়ার নই, এবং তাই আমরা প্রথম দেখার বিকল্পে সন্তুষ্ট।

সুখ এবং আনন্দ - প্রথম এজেন্ট প্রস্তুত :)

এজেন্ট প্রস্তুত, নতুন এজেন্ট দীর্ঘজীবী হোক!

হ্যাঁ, ভদ্রলোক, আমরা এই নিবন্ধটি দ্বারা প্রস্তুত পথের মাত্র 1/3 অংশ কভার করেছি, তবে নিরুৎসাহিত হবেন না, কারণ এখন এটি আরও সহজ হবে।

সুতরাং এখন আমাদের এমন একজন এজেন্টের প্রয়োজন যা মেটা তথ্য সংগ্রহ করে এবং একটি সংগ্রহ নথিতে রাখে:

collect_security_overview_topic = app.topic("collect_security_overview", internal=True)


@app.agent(collect_security_overview_topic)
async def collect_security_overview(
    stream: StreamT[?],
) -> AsyncIterable[bool]:
    async with aiohttp.ClientSession() as session:
        async for event in stream:
            ...

যেহেতু এই এজেন্ট একটি নির্দিষ্ট নিরাপত্তা সম্পর্কে তথ্য প্রক্রিয়া করবে, তাই আমাদের বার্তায় এই নিরাপত্তার টিকার (প্রতীক) নির্দেশ করতে হবে। এই উদ্দেশ্যে ফাউস্টে আছে রেকর্ডস — ক্লাস যেগুলি এজেন্ট বিষয়ে বার্তা স্কিম ঘোষণা করে।

এই ক্ষেত্রে, এর যান records.py এবং এই বিষয়ের বার্তাটি কেমন হওয়া উচিত তা বর্ণনা করুন:

import faust


class CollectSecurityOverview(faust.Record):
    symbol: str
    exchange: str

আপনি অনুমান করতে পারেন, faust বার্তা স্কিমা বর্ণনা করতে পাইথন টাইপ টীকা ব্যবহার করে, যার কারণে লাইব্রেরি দ্বারা সমর্থিত সর্বনিম্ন সংস্করণ 3.6.

আসুন এজেন্টে ফিরে যাই, প্রকারগুলি সেট করুন এবং এটি যোগ করুন:

collect_security_overview_topic = app.topic(
    "collect_security_overview", internal=True, value_type=CollectSecurityOverview
)


@app.agent(collect_security_overview_topic)
async def collect_security_overview(
    stream: StreamT[CollectSecurityOverview],
) -> AsyncIterable[bool]:
    async with aiohttp.ClientSession() as session:
        async for event in stream:
            logger.info(
                "Start collect security [{symbol}] overview", symbol=event.symbol
            )

            client = AlphaVantageClient(session, API_KEY)

            security_overview = await client.get_security_overview(event.symbol)

            await SecurityCRUD.update_one({"symbol": event.symbol, "exchange": event.exchange}, security_overview)

            yield True

আপনি দেখতে পাচ্ছেন, আমরা টপিক প্রারম্ভিক পদ্ধতিতে একটি স্কিম সহ একটি নতুন প্যারামিটার পাস করি - value_type। আরও, সবকিছু একই স্কিম অনুসরণ করে, তাই আমি অন্য কিছুতে থাকার কোন বিন্দু দেখতে পাচ্ছি না।

ঠিক আছে, চূড়ান্ত স্পর্শ হল মেটা তথ্য সংগ্রহকারী এজেন্টকে collect_securitites করার জন্য একটি কল যোগ করা:

....
for security in securities:
    await SecurityCRUD.update_one({
            "symbol": security["symbol"],
            "exchange": security["exchange"]
        },
        security,
        upsert = True,
    )

    await collect_security_overview.cast(
        CollectSecurityOverview(symbol = security["symbol"], exchange = security["exchange"])
    )
....

আমরা বার্তার জন্য পূর্বে ঘোষিত স্কিম ব্যবহার করি। এই ক্ষেত্রে, আমি .cast পদ্ধতি ব্যবহার করেছি যেহেতু আমাদের এজেন্টের ফলাফলের জন্য অপেক্ষা করতে হবে না, তবে এটি উল্লেখ করার মতো উপায় বিষয়ে একটি বার্তা পাঠান:

1. cast - ব্লক করে না কারণ এটি ফলাফল আশা করে না। আপনি একটি বার্তা হিসাবে অন্য বিষয়ে ফলাফল পাঠাতে পারবেন না.

2. send - ব্লক করে না কারণ এটি ফলাফল আশা করে না। আপনি একটি এজেন্ট উল্লেখ করতে পারেন যে বিষয়ে ফলাফল যাবে।

3. জিজ্ঞাসা করুন - ফলাফলের জন্য অপেক্ষা করুন। আপনি একটি এজেন্ট উল্লেখ করতে পারেন যে বিষয়ে ফলাফল যাবে।

সুতরাং, আজকের জন্য এজেন্টদের সাথে যে সব!

স্বপ্নের দল

শেষ জিনিসটি আমি এই অংশে লিখতে প্রতিশ্রুতি দিয়েছি তা হল কমান্ড। আগেই উল্লেখ করা হয়েছে, faust-এ কমান্ডগুলি ক্লিকের চারপাশে একটি মোড়ক। আসলে, faust -A কী নির্দিষ্ট করার সময় আমাদের কাস্টম কমান্ডকে এর ইন্টারফেসে সংযুক্ত করে

ঘোষিত এজেন্টদের পরে agents.py ডেকোরেটরের সাথে একটি ফাংশন যোগ করুন app.commandপদ্ধতি কলিং নিক্ষেপ у সিকিউরিটিস সংগ্রহ করুন:

@app.command()
async def start_collect_securities():
    """Collect securities and overview."""

    await collect_securities.cast()

সুতরাং, যদি আমরা কমান্ডের তালিকা কল করি, আমাদের নতুন কমান্ড এতে থাকবে:

> faust -A horton.agents --help

....
Commands:
  agents                    List agents.
  clean-versions            Delete old version directories.
  completion                Output shell completion to be evaluated by the...
  livecheck                 Manage LiveCheck instances.
  model                     Show model detail.
  models                    List all available models as a tabulated list.
  reset                     Delete local table state.
  send                      Send message to agent/topic.
  start-collect-securities  Collect securities and overview.
  tables                    List available tables.
  worker                    Start worker instance for given app.

আমরা এটিকে অন্য কারও মতো ব্যবহার করতে পারি, তাই আসুন ফাস্ট ওয়ার্কারটি পুনরায় চালু করি এবং সিকিউরিটিজের একটি সম্পূর্ণ সংগ্রহ শুরু করি:

> faust -A horton.agents start-collect-securities

পরবর্তীতে কী হবে?

পরবর্তী অংশে, একটি উদাহরণ হিসাবে অবশিষ্ট এজেন্ট ব্যবহার করে, আমরা বছরের ট্রেডিং এবং ক্রন লঞ্চের এজেন্টের ক্লোজিং প্রাইসের চরমতা অনুসন্ধানের জন্য সিঙ্ক প্রক্রিয়া বিবেচনা করব।

আজ যে জন্য সব! পড়ার জন্য ধন্যবাদ :)

এই অংশের জন্য কোড

পিএস শেষ অংশের অধীনে আমাকে ফাস্ট এবং সঙ্গম কাফকা সম্পর্কে জিজ্ঞাসা করা হয়েছিল (সঙ্গমে কি বৈশিষ্ট্য আছে?) দেখে মনে হচ্ছে সঙ্গমটি অনেক উপায়ে আরও কার্যকরী, তবে সত্যটি হল যে ফাস্টের সঙ্গমের জন্য সম্পূর্ণ ক্লায়েন্ট সমর্থন নেই - এটি অনুসরণ করে নথিতে ক্লায়েন্ট সীমাবদ্ধতার বর্ণনা.

উত্স: www.habr.com