ഫൗസ്റ്റിലെ പശ്ചാത്തല ജോലികൾ, ഭാഗം II: ഏജന്റുമാരും ടീമുകളും

ഉള്ളടക്ക പട്ടിക

1. ഭാഗം I: ആമുഖം

2. ഭാഗം II: ഏജന്റുമാരും ടീമുകളും

നമ്മൾ ഇവിടെ എന്താണ് ചെയ്യുന്നത്?

അതിനാൽ, രണ്ടാം ഭാഗം. നേരത്തെ എഴുതിയതുപോലെ, അതിൽ ഞങ്ങൾ ഇനിപ്പറയുന്നവ ചെയ്യും:

1. നമുക്ക് ആവശ്യമുള്ള എൻഡ് പോയിന്റുകൾക്കായുള്ള അഭ്യർത്ഥനകൾക്കൊപ്പം aiohttp-യിൽ ആൽഫവാന്റേജിനായി ഒരു ചെറിയ ക്ലയന്റ് എഴുതാം.

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-ലേക്ക് പ്രാദേശികമായി ആൽഫവാന്റേജ് ടോക്കൺ ചേർക്കാൻ മറക്കരുത്, അല്ലെങ്കിൽ പരിസ്ഥിതി വേരിയബിൾ കയറ്റുമതി ചെയ്യുക HORTON_SERVICE_APIKEY. ഞങ്ങൾക്ക് ഒരു ടോക്കൺ ലഭിക്കുന്നു ഇവിടെ.

CRUD ക്ലാസ്

സെക്യൂരിറ്റികളെക്കുറിച്ചുള്ള മെറ്റാ വിവരങ്ങൾ സംഭരിക്കുന്നതിന് ഞങ്ങൾക്ക് ഒരു സെക്യൂരിറ്റീസ് ശേഖരം ഉണ്ടായിരിക്കും.

ഡാറ്റാബേസ്/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. കാഫ്കയിലെ വിഷയങ്ങൾ, കൃത്യമായ നിർവചനം അറിയണമെങ്കിൽ, വായിക്കുന്നതാണ് നല്ലത് ഓഫ്. പ്രമാണം, അല്ലെങ്കിൽ നിങ്ങൾക്ക് വായിക്കാം സംഗ്രഹം റഷ്യൻ ഭാഷയിൽ ഹബ്രെയിൽ, എല്ലാം വളരെ കൃത്യമായി പ്രതിഫലിപ്പിക്കുന്നു :)

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

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, എടുത്തത് വിഷയം_പാർട്ടീഷനുകൾ - ആപ്ലിക്കേഷൻ ഒബ്‌ജക്റ്റ് പാരാമീറ്റർ), ഞങ്ങളുടെ വിഷയത്തിനായി ഞങ്ങൾ ഒരു വ്യക്തിഗത മൂല്യം വ്യക്തമാക്കിയിട്ടില്ലാത്തതിനാൽ (പാർട്ടീഷനുകൾ വഴി). വർക്കറിലെ ലോഞ്ച് ചെയ്ത ഏജന്റിന് എല്ലാ 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-ലേക്ക് പോയി - kafdrop on-ലേക്ക് പോയി നിങ്ങൾക്ക് ഇത് പരിശോധിക്കാം localhost:9000

ഞങ്ങളുടെ തൊഴിലാളിയുമായി ടെർമിനൽ വിൻഡോയിലേക്ക് പോകുമ്പോൾ, ലോഗുരു ഉപയോഗിച്ച് അയച്ച സന്തോഷകരമായ സന്ദേശം ഞങ്ങൾ കാണും:

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. ചോദിക്കുക - ഒരു ഫലത്തിനായി കാത്തിരിക്കുന്നു. ഫലം പോകുന്ന വിഷയത്തിൽ നിങ്ങൾക്ക് ഒരു ഏജന്റിനെ വ്യക്തമാക്കാൻ കഴിയും.

അതിനാൽ, ഇന്നത്തെ ഏജന്റുമാരുമായി അത്രമാത്രം!

സ്വപ്ന ടീം

ഈ ഭാഗത്ത് എഴുതാമെന്ന് ഞാൻ വാഗ്ദാനം ചെയ്ത അവസാന കാര്യം കമാൻഡുകൾ ആണ്. നേരത്തെ സൂചിപ്പിച്ചതുപോലെ, ഫാസ്റ്റിലെ കമാൻഡുകൾ ക്ലിക്കിന് ചുറ്റുമുള്ള ഒരു റാപ്പറാണ്. വാസ്തവത്തിൽ, -A കീ വ്യക്തമാക്കുമ്പോൾ, അതിന്റെ ഇന്റർഫേസിലേക്ക് faust നമ്മുടെ ഇഷ്‌ടാനുസൃത കമാൻഡ് അറ്റാച്ചുചെയ്യുന്നു

പ്രഖ്യാപിച്ച ഏജന്റുമാർക്ക് ശേഷം 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

അടുത്തതായി എന്ത് സംഭവിക്കും?

അടുത്ത ഭാഗത്ത്, ശേഷിക്കുന്ന ഏജന്റുമാരെ ഉദാഹരണമായി ഉപയോഗിച്ച്, വർഷത്തേക്കുള്ള ട്രേഡിംഗിന്റെ അവസാന വിലയിലും ഏജന്റുമാരുടെ ക്രോൺ ലോഞ്ചിലും അതിരുകടന്നതിനായി തിരയുന്നതിനുള്ള സിങ്ക് മെക്കാനിസം ഞങ്ങൾ പരിഗണിക്കും.

ഇന്നത്തേക്ക് അത്രമാത്രം! വായിച്ചതിന് നന്ദി :)

ഈ ഭാഗത്തിനുള്ള കോഡ്

PS അവസാന ഭാഗത്തിന് കീഴിൽ, ഫൗസ്റ്റും സംഗമവും ആയ കാഫ്കയെക്കുറിച്ച് എന്നോട് ചോദിച്ചു (സംഗമത്തിന് എന്ത് സവിശേഷതകൾ ഉണ്ട്?). കൺ‌ഫ്ലൂയന്റ് പല തരത്തിൽ കൂടുതൽ പ്രവർത്തനക്ഷമമാണെന്ന് തോന്നുന്നു, പക്ഷേ ഫൗസ്റ്റിന് കൺ‌ഫ്ലൂയന്റിന് പൂർണ്ണമായ ക്ലയന്റ് പിന്തുണയില്ല എന്നതാണ് വസ്തുത - ഇത് ഇനിപ്പറയുന്നതിൽ നിന്ന് പിന്തുടരുന്നു പ്രമാണത്തിലെ ക്ലയന്റ് നിയന്ത്രണങ്ങളുടെ വിവരണങ്ങൾ.

അവലംബം: www.habr.com