Faust मा पृष्ठभूमि कार्यहरू, भाग II: एजेन्ट र टोलीहरू

सामग्रीको तालिका

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

2. भाग II: एजेन्ट र टोलीहरू

हामी यहाँ के गर्दैछौं?

त्यसैले, दोस्रो भाग। पहिले लेखिए जस्तै, यसमा हामी निम्न गर्नेछौं:

1. हामीलाई चाहिने अन्तिम बिन्दुहरूका लागि अनुरोधहरूको साथ aiohttp मा alphavantage को लागि एउटा सानो ग्राहक लेखौं।

2. एक एजेन्ट सिर्जना गरौं जसले धितोमा डाटा र तिनीहरूमा मेटा जानकारी सङ्कलन गर्नेछ।

तर, यो हामी परियोजनाको लागि आफैं गर्नेछौं, र फस्ट अनुसन्धानको सन्दर्भमा, हामी काफ्काबाट घटनाहरू स्ट्रिम गर्ने एजेन्टहरू कसरी लेख्ने, साथै कसरी आदेशहरू लेख्ने (क्लिक रैपर), हाम्रो मामलामा - एजेन्टले निगरानी गरिरहेको विषयमा म्यानुअल पुश सन्देशहरूका लागि।

प्रशिक्षण

AlphaVantage ग्राहक

पहिले, 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 सजावट।

PS स्थानीय रूपमा config.yml मा alphavantage टोकन थप्न नबिर्सनुहोस्, वा वातावरण चर निर्यात गर्नुहोस्। 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. प्यारामिटर आन्तरिक, 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

PS सुविधाहरू वेब कम्पोनेन्ट म लेखहरूमा फास्टलाई विचार गर्दिन, त्यसैले हामीले उपयुक्त झण्डा सेट गर्यौं।

हाम्रो लन्च कमाण्डमा, हामीले फास्टलाई जानकारी लग आउटपुट स्तरको साथ एप्लिकेसन वस्तु कहाँ खोज्ने र यसको साथ के गर्ने (कार्यकर्ता सुरू गर्नुहोस्) भन्यौं। हामीले निम्न आउटपुट पाउँछौं:

बिघ्नकर्ता

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

PS प्रयोग गर्दै @ हामीले देखाउँछौं कि हामीले "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 मा योजनाको साथ नयाँ प्यारामिटर पास गर्छौं। यसबाहेक, सबै कुरा एउटै योजना पछ्याउँछ, त्यसैले म अरू कुनै कुरामा बस्ने कुनै बिन्दु देख्दिन।

ठीक छ, अन्तिम टच भनेको मेटा जानकारी सङ्कलन एजेन्टलाई कल_सेक्युराइटाइटहरूमा कल थप्नु हो:

....
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. सोध्नुहोस् - परिणामको लागि पर्खनुहोस्। तपाईले एउटा एजेन्टलाई शीर्षकमा निर्दिष्ट गर्न सक्नुहुन्छ जसमा परिणाम जान्छ।

त्यसोभए, आजको लागि एजेन्टहरूसँग त्यो सबै हो!

सपनाको टोली

मैले यस भागमा लेख्ने वाचा गरेको अन्तिम कुरा भनेको आदेशहरू हो। पहिले उल्लेख गरिएझैं, फास्टमा आदेशहरू क्लिकको वरिपरि र्यापर हुन्। वास्तवमा, faust ले -A कुञ्जी निर्दिष्ट गर्दा हाम्रो अनुकूलन आदेशलाई यसको इन्टरफेसमा संलग्न गर्दछ

घोषणा पछि एजेन्टहरू मा agents.py डेकोरेटरसँग एउटा प्रकार्य थप्नुहोस् app.commandविधि कल गर्दै जात у सङ्कलन_securitites:

@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

अब के हुने हो ?

अर्को भागमा, उदाहरणको रूपमा बाँकी एजेन्टहरू प्रयोग गर्दै, हामी वर्षको लागि व्यापारको बन्द मूल्य र एजेन्टहरूको क्रोन प्रक्षेपणमा चरमको खोजी गर्न सिंक संयन्त्रलाई विचार गर्नेछौं।

आजको लागि यति हो! पढ्नु भएकोमा धन्यवाद :)

यस भागको लागि कोड

PS अन्तिम भाग अन्तर्गत मलाई फास्ट र कन्फ्लुएन्ट काफ्काको बारेमा सोधिएको थियो (संगममा के सुविधाहरू छन्?)। यस्तो देखिन्छ कि confluent धेरै तरिकामा अधिक कार्यात्मक छ, तर तथ्य यो हो कि faust संग confluent को लागी पूर्ण ग्राहक समर्थन छैन - यो निम्न बाट छ कागजातमा ग्राहक प्रतिबन्धहरूको विवरण.

स्रोत: www.habr.com