Fona uzdevumi par Faustu, II daļa: Aģenti un komandas

Satura

1. I daļa: Ievads

2. II daļa: Aģenti un komandas

Ko mēs šeit darām?

Tātad, tā, otrā daļa. Kā rakstīts iepriekš, tajā mēs darīsim sekojošo:

1. Uzrakstīsim nelielu klientu alfavantage vietnei aiohttp ar pieprasījumiem par mums nepieciešamajiem galapunktiem.

2. Izveidosim aģentu, kas apkopos datus par vērtspapīriem un meta informāciju par tiem.

Bet tas ir tas, ko mēs darīsim paša projekta labā, un, runājot par fausta izpēti, mēs iemācīsimies rakstīt aģentus, kas apstrādā straumēšanas notikumus no kafka, kā arī kā rakstīt komandas (klikšķa iesaiņojums), mūsu gadījumā - manuāliem push ziņojumiem uz tēmu, kuru aģents uzrauga.

Treniņš

AlphaVantage klients

Vispirms uzrakstīsim nelielu aiohttp klientu alfavantage pieprasījumiem.

alfavantage.py

Spoileris

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
        )

Patiesībā no tā viss ir skaidrs:

1. AlphaVantage API ir diezgan vienkārši un skaisti izstrādāta, tāpēc es nolēmu veikt visus pieprasījumus, izmantojot šo metodi construct_query kur savukārt ir http izsaukums.

2. Es vedu visus laukus uz snake_case komfortam.

3. Labi, logger.catch dekorācija skaistai un informatīvai izsekošanai.

PS Neaizmirstiet vietēji pievienot alfavantage marķieri vietnei config.yml vai eksportēt vides mainīgo HORTON_SERVICE_APIKEY. Mēs saņemam žetonu šeit.

CRUD klase

Mums būs vērtspapīru kolekcija, lai saglabātu metainformāciju par vērtspapīriem.

datubāze/security.py

Manuprāt, šeit nekas nav jāskaidro, un pati bāzes klase ir diezgan vienkārša.

get_app()

Pievienosim funkciju lietojumprogrammas objekta izveidei app.py

Spoileris

import faust

from horton.config import KAFKA_BROKERS


def get_app():
    return faust.App("horton", broker=KAFKA_BROKERS)

Pagaidām mums būs visvienkāršākā aplikāciju izveide, nedaudz vēlāk to paplašināsim, tomēr, lai neliktu gaidīt, šeit atsauces uz App-class. Iesaku arī apskatīt iestatījumu klasi, jo tā ir atbildīga par lielāko daļu iestatījumu.

Galvenā daļa

Aģents vērtspapīru saraksta vākšanai un uzturēšanai

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

Tātad, vispirms mēs iegūstam faust lietojumprogrammas objektu - tas ir diezgan vienkārši. Tālāk mēs skaidri deklarējam tēmu savam aģentam... Šeit ir vērts pieminēt, kas tas ir, kāds ir iekšējais parametrs un kā to var sakārtot citādi.

1. Kafka tēmas, ja vēlamies uzzināt precīzu definīciju, labāk izlasīt izslēgts. dokumentu, vai arī varat lasīt apkopojums uz Habré krievu valodā, kur arī viss ir diezgan precīzi atspoguļots :)

2. Iekšējais parametrs, diezgan labi aprakstīts faust dokumentā, ļauj konfigurēt tēmu tieši kodā, protams, tas nozīmē faust izstrādātāju sniegtos parametrus, piemēram: saglabāšana, saglabāšanas politika (pēc noklusējuma dzēst, bet var iestatīt kompakts), nodalījumu skaits vienā tēmā (rādītājidarīt, piemēram, mazāk nekā globāla nozīme pieteikumi faust).

3. Kopumā aģents var izveidot pārvaldītu tēmu ar globālām vērtībām, tomēr man patīk visu skaidri deklarēt. Turklāt dažus aģenta sludinājuma tēmas parametrus (piemēram, nodalījumu skaitu vai saglabāšanas politiku) nevar konfigurēt.

   Lūk, kā tas varētu izskatīties, manuāli nedefinējot tēmu:

app = get_app()

@app.agent()
async def collect_securities(stream: StreamT[None]) -> AsyncIterable[bool]:
	pass

Nu, tagad aprakstīsim, ko darīs mūsu aģents :)

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

Tātad aģenta darbības sākumā mēs atveram aiohttp sesiju pieprasījumiem, izmantojot mūsu klientu. Tādējādi, startējot strādnieku, kad tiek palaists mūsu aģents, nekavējoties tiks atvērta sesija - viena, uz visu laiku, kad darbinieks darbojas (vai vairākas, ja mainīsit parametru vienlaicība no aģenta ar noklusējuma vienību).

Tālāk mēs sekojam straumei (ievietojam ziņojumu _, jo mums, šajā aģentā, nerūp mūsu tēmas ziņojumu saturs, ja tie pastāv pašreizējā nobīdē, pretējā gadījumā mūsu cikls gaidīs to ierašanos. Mūsu cilpas ietvaros mēs reģistrējam ziņojuma saņemšanu, iegūstam aktīvo (get_securities pēc noklusējuma atgriež tikai aktīvus, skatiet klienta kodu) vērtspapīru sarakstu un saglabājam to datu bāzē, pārbaudot, vai ir vērtspapīrs ar tādu pašu atzīmi un apmaiņa datubāzē , ja ir, tad tas (papīrs) vienkārši tiks atjaunināts.

Sāksim savu radīšanu!

> docker-compose up -d
... Запуск контейнеров ...
> faust -A horton.agents worker --without-web -l info

PS funkcijas tīmekļa komponents Rakstos Faustu neuzskatīšu, tāpēc uzlikām atbilstošu karogu.

Savā palaišanas komandā mēs teicām faustam, kur meklēt lietojumprogrammas objektu un ko ar to darīt (palaist darbinieku) ar informācijas žurnāla izvades līmeni. Mēs iegūstam šādu izvadi:

Spoileris

┌ƒ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}        │
└────────────────────────────┴────────────┘ 

Tas ir dzīvs!!!

Apskatīsim nodalījumu komplektu. Kā redzam, tika izveidota tēma ar nosaukumu, ko norādījām kodā, noklusēto nodalījumu skaitu (8, ņemts no topic_partitions - lietojumprogrammas objekta parametrs), jo mēs nenorādījām savai tēmai atsevišķu vērtību (izmantojot nodalījumus). Darbinieka palaistajam aģentam tiek piešķirti visi 8 nodalījumi, jo tas ir vienīgais, taču par to sīkāk tiks runāts sadaļā par klasterizāciju.

Nu, tagad mēs varam pāriet uz citu termināļa logu un nosūtīt tukšu ziņojumu mūsu tēmai:

> faust -A horton.agents send @collect_securities
{"topic": "collect_securities", "partition": 6, "topic_partition": ["collect_securities", 6], "offset": 0, "timestamp": ..., "timestamp_type": 0}

PS izmanto @ mēs parādām, ka mēs sūtām ziņojumu uz tēmu ar nosaukumu “collect_securities”.

Šajā gadījumā ziņojums tika nosūtīts uz 6. nodalījumu — to varat pārbaudīt, dodoties uz kafdrop on localhost:9000

Dodoties uz termināļa logu ar mūsu darbinieku, mēs redzēsim priecīgu ziņojumu, kas nosūtīts, izmantojot loguru:

2020-09-23 00:26:37.304 | INFO     | horton.agents:collect_securities:40 - Start collect securities

Mēs varam arī izpētīt mongo (izmantojot Robo3T vai Studio3T) un redzēt, ka vērtspapīri atrodas datu bāzē:

Es neesmu miljardieris, un tāpēc mēs esam apmierināti ar pirmo skatīšanās iespēju.

Laime un prieks - pirmais aģents gatavs :)

Aģents gatavs, lai dzīvo jaunais aģents!

Jā, kungi, esam veikuši tikai 1/3 no šī raksta sagatavotā ceļa, bet nekautrējieties, jo tagad būs vieglāk.

Tātad tagad mums ir nepieciešams aģents, kas apkopo meta informāciju un ievieto to savākšanas dokumentā:

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

Tā kā šis aģents apstrādās informāciju par konkrētu vērtspapīru, mums ziņojumā ir jānorāda šī vērtspapīra svārsts (simbols). Šim nolūkam faustā ir Ieraksti — klases, kas deklarē ziņojuma shēmu aģenta tēmā.

Šajā gadījumā pāriesim pie records.py un aprakstiet, kā vajadzētu izskatīties vēstījumam par šo tēmu:

import faust


class CollectSecurityOverview(faust.Record):
    symbol: str
    exchange: str

Kā jau varēja nojaust, fausts izmanto python tipa anotāciju, lai aprakstītu ziņojuma shēmu, tāpēc bibliotēkas atbalstītā minimālā versija ir 3.6.

Atgriezīsimies pie aģenta, iestatīsim veidus un pievienosim to:

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

Kā redzat, tēmas inicializācijas metodei mēs nododam jaunu parametru ar shēmu - value_type. Turklāt viss notiek pēc tās pašas shēmas, tāpēc es neredzu jēgu kavēties pie kaut kā cita.

Pēdējais pieskāriens ir pievienot metainformācijas vākšanas aģentam izsaukumu collection_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"])
    )
....

Ziņojumam izmantojam iepriekš izziņoto shēmu. Šajā gadījumā es izmantoju .cast metodi, jo mums nav jāgaida aģenta rezultāts, taču ir vērts pieminēt, ka veidus nosūtīt ziņu tēmai:

1. cast - nebloķē, jo negaida rezultātu. Jūs nevarat nosūtīt rezultātu uz citu tēmu kā ziņojumu.

2. sūtīt - nebloķē, jo negaida rezultātu. Tēmā varat norādīt aģentu, uz kuru tiks nosūtīts rezultāts.

3. jautāt - gaida rezultātu. Tēmā varat norādīt aģentu, uz kuru tiks nosūtīts rezultāts.

Tātad ar aģentiem šodien viss!

Sapņu komanda

Pēdējais, ko solījos uzrakstīt šajā daļā, ir komandas. Kā minēts iepriekš, komandas programmā Faust ir iesaiņojums ap klikšķi. Faktiski Faust vienkārši pievieno mūsu pielāgoto komandu savai saskarnei, norādot taustiņu -A

Pēc izsludināto aģentu ienākšanas agents.py pievienot funkciju ar dekoratoru app.commandsaucot metodi mest у vākt_vērtspapīrus:

@app.command()
async def start_collect_securities():
    """Collect securities and overview."""

    await collect_securities.cast()

Tādējādi, ja mēs izsauksim komandu sarakstu, mūsu jaunā komanda būs tajā:

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

Mēs to varam izmantot tāpat kā jebkurš cits, tāpēc restartēsim faust worker un sāksim pilnvērtīgu vērtspapīru kolekciju:

> faust -A horton.agents start-collect-securities

Kas notiks tālāk?

Nākamajā daļā, izmantojot atlikušos aģentus kā piemēru, aplūkosim grimšanas mehānismu galējību meklēšanai gada tirdzniecības slēgšanas cenās un aģentu kronu palaišanu.

Tas šodienai viss! Paldies, ka izlasīji :)

Šīs daļas kods

PS Zem pēdējās daļas man jautāja par faust and confluent kafka (kādas īpašības ir konfluentam?). Šķiet, ka konfluents daudzējādā ziņā ir funkcionālāks, taču fakts ir tāds, ka Faustam nav pilnīgs klientu atbalsts konfluentam - tas izriet no klientu ierobežojumu apraksti dok.

Avots: www.habr.com