फॉस्ट, भाग II वर पार्श्वभूमी कार्ये: एजंट आणि संघ

सामग्री सारणी

1. भाग I: परिचय

2. भाग II: एजंट आणि संघ

आपण इथे काय करत आहोत?

तर, म्हणून, दुसरा भाग. आधी लिहिल्याप्रमाणे, त्यात आपण पुढील गोष्टी करू:

1. आम्हाला आवश्यक असलेल्या एंडपॉइंट्सच्या विनंतीसह aiohttp वर alphavantage साठी एक लहान क्लायंट लिहूया.

2. चला एक एजंट तयार करू जो सिक्युरिटीजवरील डेटा आणि त्यावरील मेटा माहिती गोळा करेल.

परंतु, आम्ही प्रकल्पासाठी हेच करणार आहोत आणि फास्ट रिसर्चच्या संदर्भात, आम्ही काफ्कामधील घटनांवर प्रक्रिया करणारे एजंट कसे लिहायचे, तसेच आमच्या बाबतीत कमांड्स (क्लिक रॅपर) कसे लिहायचे ते शिकू - एजंट निरीक्षण करत असलेल्या विषयावर मॅन्युअल पुश संदेशांसाठी.

प्रशिक्षण

AlphaVantage क्लायंट

प्रथम, अल्फाव्हंटेजच्या विनंतीसाठी एक लहान 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 सजावट.

P.S. config.yml वर स्थानिकरित्या अल्फाव्हंटेज टोकन जोडण्यास विसरू नका किंवा पर्यावरण व्हेरिएबल निर्यात करू नका HORTON_SERVICE_APIKEY. आम्हाला एक टोकन मिळते येथे.

CRUD वर्ग

सिक्युरिटीजबद्दल मेटा माहिती साठवण्यासाठी आमच्याकडे सिक्युरिटीज संग्रह असेल.

database/security.py

माझ्या मते, येथे काहीही स्पष्ट करण्याची गरज नाही आणि बेस क्लास स्वतःच अगदी सोपा आहे.

get_app()

मध्ये ऍप्लिकेशन ऑब्जेक्ट तयार करण्यासाठी फंक्शन जोडू 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. काफ्कामधील विषय, जर आपल्याला अचूक व्याख्या जाणून घ्यायची असेल तर ते वाचणे चांगले आहे बंद. दस्तऐवज, किंवा तुम्ही वाचू शकता संकलन रशियन भाषेत Habré वर, जिथे सर्वकाही अगदी अचूकपणे प्रतिबिंबित होते :)

2. अंतर्गत पॅरामीटर, फॉस्ट डॉकमध्ये चांगले वर्णन केले आहे, आम्हाला थेट कोडमध्ये विषय कॉन्फिगर करण्याची परवानगी देते, अर्थातच, याचा अर्थ फॉस्ट डेव्हलपरद्वारे प्रदान केलेले पॅरामीटर्स, उदाहरणार्थ: धारणा, धारणा धोरण (डीफॉल्टनुसार हटवा, परंतु आपण सेट करू शकता संक्षिप्त), प्रति विषय विभाजनांची संख्या (विभाजनेकरण्यासाठी, उदाहरणार्थ, पेक्षा कमी जागतिक महत्त्व अनुप्रयोग फॉस्ट).

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

P.S. शक्यता वेब घटक मी लेखांमध्ये फॉस्टचा विचार करणार नाही, म्हणून आम्ही योग्य ध्वज सेट केला.

आमच्या लॉन्च कमांडमध्ये, आम्ही माहिती लॉग आउटपुट लेव्हलसह ऍप्लिकेशन ऑब्जेक्ट कुठे शोधायचे आणि त्याचे काय करायचे (वर्कर लाँच करा) सांगितले. आम्हाला खालील आउटपुट मिळते:

स्पायलेटर

┌ƒ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}

P.S. वापरून @ आम्ही दाखवतो की आम्ही “collect_securities” नावाच्या विषयाला संदेश पाठवत आहोत.

या प्रकरणात, संदेश विभाजन 6 वर गेला - तुम्ही kafdrop on वर जाऊन हे तपासू शकता 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

तुम्ही अंदाज केला असेल, फॉस्ट मेसेज स्कीमाचे वर्णन करण्यासाठी पायथन प्रकार भाष्य वापरते, म्हणूनच लायब्ररीद्वारे समर्थित किमान आवृत्ती आहे 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. कास्ट - अवरोधित करत नाही कारण ते परिणामाची अपेक्षा करत नाही. तुम्ही निकाल दुसऱ्या विषयावर संदेश म्हणून पाठवू शकत नाही.

2. पाठवा - अवरोधित करत नाही कारण ते परिणामाची अपेक्षा करत नाही. निकाल ज्या विषयावर जाईल त्या विषयामध्ये तुम्ही एजंट निर्दिष्ट करू शकता.

3. विचारा - निकालाची वाट पहा. निकाल ज्या विषयावर जाईल त्या विषयामध्ये तुम्ही एजंट निर्दिष्ट करू शकता.

तर, आज एजंट्ससाठी एवढेच!

ड्रीम टीम

मी या भागात लिहिण्याचे वचन दिलेली शेवटची गोष्ट म्हणजे आज्ञा. आधी सांगितल्याप्रमाणे, फॉस्ट मधील कमांड क्लिकच्या आसपास रॅपर असतात. खरं तर, -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

पुढे काय होणार?

पुढील भागात, उर्वरित एजंट्सचा उदाहरण म्हणून वापर करून, आम्ही वर्षभरातील ट्रेडिंगच्या बंद किंमती आणि एजंट्सच्या क्रॉन लॉन्चमध्ये टोकाचा शोध घेण्यासाठी सिंक यंत्रणा विचारात घेऊ.

आजसाठी एवढेच! वाचल्याबद्दल धन्यवाद :)

या भागासाठी कोड

P.S. शेवटच्या भागाखाली मला फास्ट आणि कंफ्लुएंट काफ्का (संगमात कोणती वैशिष्ट्ये आहेत?). असे दिसते की confluent अनेक मार्गांनी अधिक कार्यशील आहे, परंतु वस्तुस्थिती अशी आहे की फॉस्टला confluent साठी पूर्ण क्लायंट समर्थन नाही - हे खालील प्रमाणे आहे दस्तऐवजातील क्लायंट निर्बंधांचे वर्णन.

स्त्रोत: www.habr.com