Bakgrundsuppgifter om Faust, del II: Agenter och team

innehållsförteckning

1. Del I: Introduktion

2. Del II: Agenter och team

Vad gör vi här?

Så, så, andra delen. Som skrivits tidigare kommer vi att göra följande i den:

1. Låt oss skriva en liten klient för alphavantage på aiohttp med förfrågningar om de slutpunkter vi behöver.

2. Låt oss skapa en agent som kommer att samla in data om värdepapper och metainformation om dem.

Men det här är vad vi kommer att göra för själva projektet, och när det gäller snabbforskning, kommer vi att lära oss hur man skriver agenter som bearbetar strömningshändelser från kafka, såväl som hur man skriver kommandon (click wrapper), i vårt fall - för manuella push-meddelanden till ämnet som agenten övervakar.

Utbildning

AlphaVantage-klient

Låt oss först skriva en liten aiohttp-klient för förfrågningar till alphavantage.

alphavantage.py

Spoiler

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
        )

Egentligen är allt tydligt från det:

1. AlphaVantage API är ganska enkelt och vackert designat, så jag bestämde mig för att göra alla förfrågningar genom metoden construct_query där det i sin tur finns ett http-anrop.

2. Jag tar med alla fält till snake_case för komfort.

3. Jo, logger.catch-dekorationen för vacker och informativ spårning.

PS Glöm inte att lägga till alphavantage-token lokalt till config.yml, eller exportera miljövariabeln HORTON_SERVICE_APIKEY. Vi får en token här.

CRUD klass

Vi kommer att ha en värdepapperssamling för att lagra metainformation om värdepapper.

database/security.py

Enligt min mening finns det inget behov av att förklara någonting här, och själva basklassen är ganska enkel.

get_app()

Låt oss lägga till en funktion för att skapa ett applikationsobjekt i app.py

Spoiler

import faust

from horton.config import KAFKA_BROKERS


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

För nu kommer vi att ha den enklaste applikationsskapandet, lite senare kommer vi att utöka det, dock för att inte låta dig vänta, här referenser till App-klass. Jag råder dig också att ta en titt på inställningsklassen, eftersom den är ansvarig för de flesta inställningarna.

Huvuddelen

Agent för insamling och underhåll av en lista över värdepapper

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

Så först får vi faust-applikationsobjektet - det är ganska enkelt. Därefter deklarerar vi uttryckligen ett ämne för vår agent... Här är det värt att nämna vad det är, vad den interna parametern är och hur detta kan ordnas annorlunda.

1. Ämnen i kafka, om vi vill veta den exakta definitionen är det bättre att läsa av. dokumentera, eller så kan du läsa kompendium på Habré på ryska, där allt också reflekteras ganska exakt :)

2. Parameter intern, som beskrivs ganska bra i faust-dokumentet, tillåter oss att konfigurera ämnet direkt i koden, naturligtvis betyder detta parametrarna som tillhandahålls av faust-utvecklarna, till exempel: retention, retention policy (som standard radera, men du kan ställa in kompakt), antal partitioner per ämne (poängatt göra till exempel mindre än global betydelse applikationer faust).

3. I allmänhet kan agenten skapa ett hanterat ämne med globala värden, men jag gillar att deklarera allt explicit. Dessutom kan vissa parametrar (till exempel antalet partitioner eller lagringspolicy) för ämnet i agentannonsen inte konfigureras.

   Så här kan det se ut utan att manuellt definiera ämnet:

app = get_app()

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

Nåväl, låt oss nu beskriva vad vår agent kommer att göra :)

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

Så i början av agenten öppnar vi en aiohttp-session för förfrågningar genom vår klient. Sålunda, när en arbetare startar, när vår agent startas, kommer en session omedelbart att öppnas - en, under hela tiden arbetaren är igång (eller flera, om du ändrar parametern samtidighet från en agent med en standardenhet).

Därefter följer vi strömmen (vi placerar meddelandet i _, eftersom vi, i denna agent, inte bryr oss om innehållet) av meddelanden från vårt ämne, om de finns vid den aktuella offset, annars kommer vår cykel att vänta på deras ankomst. Tja, inne i vår loop loggar vi mottagandet av meddelandet, får en lista över aktiva (get_securities returnerar endast aktiva som standard, se kundkod) värdepapper och sparar det i databasen, kontrollerar om det finns ett värdepapper med samma ticker och utbyte i databasen, om det finns, kommer det (papperet) helt enkelt att uppdateras.

Låt oss lansera vår skapelse!

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

PS funktioner webbkomponent Jag kommer inte att överväga faust i artiklarna, så vi sätter rätt flagga.

I vårt startkommando berättade vi för faust var man skulle leta efter applikationsobjektet och vad man skulle göra med det (starta en arbetare) med informationsloggutdatanivån. Vi får följande utdata:

Spoiler

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

Det är levande!!!

Låt oss titta på partitionsuppsättningen. Som vi kan se skapades ett ämne med namnet som vi angav i koden, standardantalet partitioner (8, hämtat från topic_partitions - application object parameter), eftersom vi inte angav ett individuellt värde för vårt ämne (via partitioner). Den startade agenten i arbetaren tilldelas alla 8 partitioner, eftersom det är den enda, men detta kommer att diskuteras mer detaljerat i delen om klustring.

Nåväl, nu kan vi gå till ett annat terminalfönster och skicka ett tomt meddelande till vårt ämne:

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

PS använder @ vi visar att vi skickar ett meddelande till ett ämne som heter "collect_securities".

I det här fallet gick meddelandet till partition 6 - du kan kontrollera detta genom att gå till kafdrop on localhost:9000

När vi går till terminalfönstret med vår arbetare kommer vi att se ett glatt meddelande skickat med loguru:

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

Vi kan också titta på mongo (med Robo3T eller Studio3T) och se att värdepapperen finns i databasen:

Jag är ingen miljardär, och därför nöjer vi oss med det första visningsalternativet.

Lycka och glädje - den första agenten är redo :)

Agent redo, länge leve den nya agenten!

Ja, mina herrar, vi har bara täckt 1/3 av vägen som utarbetats av den här artikeln, men var inte avskräckta, för nu kommer det att bli lättare.

Så nu behöver vi en agent som samlar in metainformation och lägger in den i ett insamlingsdokument:

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

Eftersom denna agent kommer att bearbeta information om ett specifikt värdepapper, måste vi ange tickern (symbolen) för detta värdepapper i meddelandet. För detta ändamål i faust finns det Register — klasser som deklarerar meddelandeschemat i agentämnet.

I det här fallet, låt oss gå till records.py och beskriv hur meddelandet för detta ämne ska se ut:

import faust


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

Som du kanske har gissat använder faust annoteringen av pythontyp för att beskriva meddelandeschemat, varför den minsta versionen som stöds av biblioteket är 3.6.

Låt oss återgå till agenten, ställa in typerna och lägga till den:

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

Som du kan se skickar vi en ny parameter med ett schema till ämnesinitieringsmetoden - value_type. Dessutom följer allt samma schema, så jag ser ingen mening med att uppehålla mig vid något annat.

Tja, sista handen är att lägga till ett samtal till agenten för insamling av metainformation till collect_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"])
    )
....

Vi använder det tidigare aviserade schemat för meddelandet. I det här fallet använde jag .cast-metoden eftersom vi inte behöver vänta på resultatet från agenten, men det är värt att nämna att sätt skicka ett meddelande till ämnet:

1. cast - blockerar inte eftersom den inte förväntar sig ett resultat. Du kan inte skicka resultatet till ett annat ämne som ett meddelande.

2. skicka - blockerar inte eftersom det inte förväntar sig ett resultat. Du kan ange en agent i ämnet som resultatet ska gå till.

3. fråga - väntar på ett resultat. Du kan ange en agent i ämnet som resultatet ska gå till.

Så, det är allt med agenter för idag!

Dröm teamet

Det sista jag lovade att skriva i den här delen är kommandon. Som nämnts tidigare är kommandon i faust ett omslag runt klick. Faktum är att faust helt enkelt kopplar vårt anpassade kommando till dess gränssnitt när du anger -A-tangenten

Efter de annonserade agenterna in agents.py lägg till en funktion med en dekoratör app.commandkallar metoden gjutas у collect_securitites:

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

    await collect_securities.cast()

Således, om vi anropar listan med kommandon, kommer vårt nya kommando att finnas i den:

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

Vi kan använda det som alla andra, så låt oss starta om faust-arbetaren och påbörja en fullfjädrad samling av värdepapper:

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

Vad kommer hända härnäst?

I nästa del, med de återstående agenterna som exempel, kommer vi att överväga sjunkmekanismen för att söka efter ytterligheter i stängningspriserna för handel för året och kronlanseringen av agenter.

Det är allt för idag! Tack för att du läser :)

Kod för denna del

PS Under den sista delen fick jag frågan om faust och confluent kafka (vilka egenskaper har confluent?). Det verkar som att confluent är mer funktionellt på många sätt, men faktum är att faust inte har fullt klientstöd för confluent - detta följer av beskrivningar av klientrestriktioner i dokumentet.

Källa: will.com