ફોસ્ટ, ભાગ 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 શણગાર.

પી.એસ. 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. આંતરિક પરિમાણ, ફૉસ્ટ ડૉકમાં ખૂબ સારી રીતે વર્ણવેલ છે, જે અમને કોડમાં સીધા જ વિષયને ગોઠવવાની મંજૂરી આપે છે, અલબત્ત, આનો અર્થ ફૉસ્ટ ડેવલપર્સ દ્વારા પ્રદાન કરવામાં આવેલા પરિમાણો છે, ઉદાહરણ તરીકે: રીટેન્શન, રીટેન્શન પૉલિસી (ડિફૉલ્ટ રૂપે કાઢી નાખો, પરંતુ તમે સેટ કરી શકો છો. કોમ્પેક્ટ), વિષય દીઠ પાર્ટીશનોની સંખ્યા (પાર્ટીશનોકરવા માટે, ઉદાહરણ તરીકે, કરતાં ઓછું વૈશ્વિક મહત્વ એપ્લિકેશન્સ ફોસ્ટ).

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 પર ગયો હતો - તમે 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

જેમ તમે જોઈ શકો છો, અમે વિષયની શરૂઆતની પદ્ધતિ - મૂલ્ય_પ્રકાર માટે સ્કીમ સાથે એક નવું પરિમાણ પસાર કરીએ છીએ. આગળ, બધું સમાન યોજનાને અનુસરે છે, તેથી મને અન્ય કંઈપણ પર ધ્યાન આપવાનો કોઈ મુદ્દો દેખાતો નથી.

ઠીક છે, અંતિમ સ્પર્શ એ મેટા માહિતી સંગ્રહ એજન્ટને એકત્રિત_સિક્યોરિટાઇટ્સ માટે કૉલ ઉમેરવાનો છે:

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

આગળ શું થશે?

આગળના ભાગમાં, ઉદાહરણ તરીકે બાકીના એજન્ટોનો ઉપયોગ કરીને, અમે વર્ષના ટ્રેડિંગના બંધ ભાવ અને એજન્ટોના ક્રોન લોંચમાં ચરમસીમા શોધવા માટેની સિંક મિકેનિઝમ પર વિચાર કરીશું.

આજ માટે આટલું જ! વાંચવા બદલ આભાર :)

આ ભાગ માટે કોડ

પી.એસ. છેલ્લા ભાગ હેઠળ મને ફોસ્ટ અને કન્ફ્લુઅન્ટ કાફકા વિશે પૂછવામાં આવ્યું હતું (સંગમમાં શું લક્ષણો છે?). એવું લાગે છે કે સંગમ ઘણી રીતે વધુ કાર્યાત્મક છે, પરંતુ હકીકત એ છે કે ફૉસ્ટને સંગમ માટે સંપૂર્ણ ક્લાયંટ સપોર્ટ નથી - આ નીચે મુજબ છે દસ્તાવેજમાં ક્લાયંટ પ્રતિબંધોનું વર્ણન.

સોર્સ: www.habr.com