ਫੌਸਟ, ਭਾਗ II 'ਤੇ ਪਿਛੋਕੜ ਕਾਰਜ: ਏਜੰਟ ਅਤੇ ਟੀਮਾਂ

ਵਿਸ਼ਾ-ਸੂਚੀ

1. ਭਾਗ I: ਜਾਣ-ਪਛਾਣ

2. ਭਾਗ II: ਏਜੰਟ ਅਤੇ ਟੀਮਾਂ

ਅਸੀਂ ਇੱਥੇ ਕੀ ਕਰ ਰਹੇ ਹਾਂ?

ਇਸ ਲਈ, ਇਸ ਲਈ, ਦੂਜਾ ਭਾਗ. ਜਿਵੇਂ ਕਿ ਪਹਿਲਾਂ ਲਿਖਿਆ ਗਿਆ ਸੀ, ਇਸ ਵਿੱਚ ਅਸੀਂ ਹੇਠਾਂ ਦਿੱਤੇ ਕੰਮ ਕਰਾਂਗੇ:

1. ਆਉ ਅਸੀਂ ਲੋੜੀਂਦੇ ਅੰਤਮ ਬਿੰਦੂਆਂ ਲਈ ਬੇਨਤੀਆਂ ਦੇ ਨਾਲ aiohttp 'ਤੇ ਅਲਫਾਵਾਂਟੇਜ ਲਈ ਇੱਕ ਛੋਟਾ ਕਲਾਇੰਟ ਲਿਖੀਏ।

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 ਸਜਾਵਟ।

PS ਸਥਾਨਕ ਤੌਰ 'ਤੇ 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. ਕਾਫਕਾ ਵਿੱਚ ਵਿਸ਼ੇ, ਜੇਕਰ ਅਸੀਂ ਸਹੀ ਪਰਿਭਾਸ਼ਾ ਜਾਣਨਾ ਚਾਹੁੰਦੇ ਹਾਂ, ਤਾਂ ਪੜ੍ਹਨਾ ਬਿਹਤਰ ਹੈ ਬੰਦ ਦਸਤਾਵੇਜ਼, ਜਾਂ ਤੁਸੀਂ ਪੜ੍ਹ ਸਕਦੇ ਹੋ ਸੰਗ੍ਰਹਿ ਰੂਸੀ ਵਿੱਚ ਹੈਬਰੇ 'ਤੇ, ਜਿੱਥੇ ਹਰ ਚੀਜ਼ ਬਿਲਕੁਲ ਸਹੀ ਰੂਪ ਵਿੱਚ ਪ੍ਰਤੀਬਿੰਬਿਤ ਹੁੰਦੀ ਹੈ :)

2. ਅੰਦਰੂਨੀ ਪੈਰਾਮੀਟਰ, faust doc ਵਿੱਚ ਬਹੁਤ ਵਧੀਆ ਢੰਗ ਨਾਲ ਵਰਣਨ ਕੀਤਾ ਗਿਆ ਹੈ, ਸਾਨੂੰ ਕੋਡ ਵਿੱਚ ਸਿੱਧੇ ਵਿਸ਼ੇ ਨੂੰ ਕੌਂਫਿਗਰ ਕਰਨ ਦੀ ਇਜਾਜ਼ਤ ਦਿੰਦਾ ਹੈ, ਬੇਸ਼ਕ, ਇਸਦਾ ਮਤਲਬ ਹੈ ਫੌਸਟ ਡਿਵੈਲਪਰਾਂ ਦੁਆਰਾ ਪ੍ਰਦਾਨ ਕੀਤੇ ਗਏ ਮਾਪਦੰਡ, ਉਦਾਹਰਨ ਲਈ: ਧਾਰਨ, ਧਾਰਨ ਨੀਤੀ (ਡਿਫੌਲਟ ਮਿਟਾਓ, ਪਰ ਤੁਸੀਂ ਸੈੱਟ ਕਰ ਸਕਦੇ ਹੋ ਸੰਖੇਪਪ੍ਰਤੀ ਵਿਸ਼ਾ ਭਾਗਾਂ ਦੀ ਗਿਣਤੀ (ਭਾਗਕਰਨ ਲਈ, ਉਦਾਹਰਨ ਲਈ, ਘੱਟ ਗਲੋਬਲ ਮਹੱਤਤਾ ਐਪਲੀਕੇਸ਼ਨ 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}

ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹੋਏ ਪੀ.ਐਸ @ ਅਸੀਂ ਦਿਖਾਉਂਦੇ ਹਾਂ ਕਿ ਅਸੀਂ “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:
            ...

ਕਿਉਂਕਿ ਇਹ ਏਜੰਟ ਇੱਕ ਖਾਸ ਸੁਰੱਖਿਆ ਬਾਰੇ ਜਾਣਕਾਰੀ ਦੀ ਪ੍ਰਕਿਰਿਆ ਕਰੇਗਾ, ਸਾਨੂੰ ਸੰਦੇਸ਼ ਵਿੱਚ ਇਸ ਸੁਰੱਖਿਆ ਦੇ ਟਿਕਰ (ਪ੍ਰਤੀਕ) ਨੂੰ ਦਰਸਾਉਣ ਦੀ ਲੋੜ ਹੈ। faust ਵਿੱਚ ਇਸ ਮਕਸਦ ਲਈ ਹਨ ਰਿਕਾਰਡ - ਉਹ ਕਲਾਸਾਂ ਜੋ ਏਜੰਟ ਵਿਸ਼ੇ ਵਿੱਚ ਸੰਦੇਸ਼ ਸਕੀਮ ਦਾ ਐਲਾਨ ਕਰਦੀਆਂ ਹਨ।

ਇਸ ਮਾਮਲੇ ਵਿੱਚ, ਆਓ ਜਾਣੀਏ 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

ਜਿਵੇਂ ਕਿ ਤੁਸੀਂ ਵੇਖ ਸਕਦੇ ਹੋ, ਅਸੀਂ ਵਿਸ਼ੇ ਦੀ ਸ਼ੁਰੂਆਤੀ ਵਿਧੀ - ਮੁੱਲ_ਟਾਈਪ ਲਈ ਇੱਕ ਸਕੀਮ ਦੇ ਨਾਲ ਇੱਕ ਨਵਾਂ ਪੈਰਾਮੀਟਰ ਪਾਸ ਕਰਦੇ ਹਾਂ। ਇਸ ਤੋਂ ਇਲਾਵਾ, ਸਭ ਕੁਝ ਉਸੇ ਸਕੀਮ ਦੀ ਪਾਲਣਾ ਕਰਦਾ ਹੈ, ਇਸਲਈ ਮੈਨੂੰ ਕਿਸੇ ਹੋਰ ਚੀਜ਼ 'ਤੇ ਧਿਆਨ ਦੇਣ ਦਾ ਕੋਈ ਬਿੰਦੂ ਨਜ਼ਰ ਨਹੀਂ ਆਉਂਦਾ।

ਖੈਰ, ਅੰਤਿਮ ਛੋਹ ਮੈਟਾ ਜਾਣਕਾਰੀ ਇਕੱਤਰ ਕਰਨ ਵਾਲੇ ਏਜੰਟ ਨੂੰ ਇਕੱਠਾ_ਸੁਰੱਖਿਆ ਕਰਨ ਲਈ ਇੱਕ ਕਾਲ ਜੋੜਨਾ ਹੈ:

....
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

ਅੱਗੇ ਕੀ ਹੋਵੇਗਾ?

ਅਗਲੇ ਹਿੱਸੇ ਵਿੱਚ, ਇੱਕ ਉਦਾਹਰਨ ਦੇ ਤੌਰ 'ਤੇ ਬਾਕੀ ਬਚੇ ਏਜੰਟਾਂ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹੋਏ, ਅਸੀਂ ਸਾਲ ਲਈ ਵਪਾਰ ਦੀਆਂ ਸਮਾਪਤੀ ਕੀਮਤਾਂ ਅਤੇ ਏਜੰਟਾਂ ਦੇ ਕਰੌਨ ਲਾਂਚ ਵਿੱਚ ਅਤਿਅੰਤਤਾ ਦੀ ਖੋਜ ਲਈ ਸਿੰਕ ਵਿਧੀ ਨੂੰ ਦੇਖਾਂਗੇ।

ਇਹ ਸਭ ਅੱਜ ਲਈ ਹੈ! ਪੜ੍ਹਨ ਲਈ ਧੰਨਵਾਦ :)

ਇਸ ਹਿੱਸੇ ਲਈ ਕੋਡ

PS ਪਿਛਲੇ ਭਾਗ ਦੇ ਤਹਿਤ ਮੈਨੂੰ ਫੌਸਟ ਅਤੇ ਕੰਫਲੂਐਂਟ ਕਾਫਕਾ ਬਾਰੇ ਪੁੱਛਿਆ ਗਿਆ ਸੀ (ਸੰਗਮ ਵਿੱਚ ਕਿਹੜੀਆਂ ਵਿਸ਼ੇਸ਼ਤਾਵਾਂ ਹਨ?). ਅਜਿਹਾ ਲਗਦਾ ਹੈ ਕਿ ਕੰਫਲੂਐਂਟ ਬਹੁਤ ਸਾਰੇ ਤਰੀਕਿਆਂ ਨਾਲ ਵਧੇਰੇ ਕਾਰਜਸ਼ੀਲ ਹੈ, ਪਰ ਤੱਥ ਇਹ ਹੈ ਕਿ ਫੌਸਟ ਕੋਲ ਕੰਫਲੂਐਂਟ ਲਈ ਪੂਰਾ ਗਾਹਕ ਸਮਰਥਨ ਨਹੀਂ ਹੈ - ਇਹ ਇਸ ਤੋਂ ਬਾਅਦ ਹੈ ਦਸਤਾਵੇਜ਼ ਵਿੱਚ ਕਲਾਇੰਟ ਪਾਬੰਦੀਆਂ ਦਾ ਵੇਰਵਾ.

ਸਰੋਤ: www.habr.com