ფონური ამოცანები ფაუსტზე, ნაწილი II: აგენტები და გუნდები

სარჩევი

1. ნაწილი I: შესავალი

2. ნაწილი II: აგენტები და გუნდები

რას ვაკეთებთ აქ?

მაშ ასე, მეორე ნაწილი. როგორც ადრე დავწერე, მასში ჩვენ გავაკეთებთ შემდეგს:

1. მოდით დავწეროთ პატარა კლიენტი alphavantage-ისთვის aiohttp-ზე ჩვენთვის საჭირო ბოლო წერტილების მოთხოვნით.

2. მოდით შევქმნათ აგენტი, რომელიც შეაგროვებს მონაცემებს ფასიანი ქაღალდების შესახებ და მათზე მეტ ინფორმაციას.

მაგრამ, ეს არის ის, რასაც ჩვენ გავაკეთებთ თავად პროექტისთვის და ფაუსტის კვლევის თვალსაზრისით, ჩვენ ვისწავლით როგორ დავწეროთ აგენტები, რომლებიც ამუშავებენ მოვლენებს კაფკადან, ასევე როგორ დავწეროთ ბრძანებები (click wrapper), ჩვენს შემთხვევაში - ხელით წასვლის შეტყობინებებისთვის იმ თემაზე, რომელსაც აგენტი აკონტროლებს.

სასწავლო

AlphaVantage კლიენტი

პირველი, მოდით დავწეროთ პატარა aiohttp კლიენტი alphavantage-ის მოთხოვნებისთვის.

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 არ დაგავიწყდეთ alphavantage ჟეტონის ადგილობრივად დამატება 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. პარამეტრი შიდა, რომელიც საკმაოდ კარგად არის აღწერილი 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, აღებული თემა_ტიხრები - განაცხადის ობიექტის პარამეტრი), რადგან ჩვენ არ დავაკონკრეტეთ ინდივიდუალური მნიშვნელობა ჩვენი თემისთვის (ტიხრების მეშვეობით). მუშაკში გაშვებულ აგენტს ენიჭება 8-ვე დანაყოფი, რადგან ის ერთადერთია, მაგრამ ეს უფრო დეტალურად იქნება განხილული კლასტერიზაციის შესახებ ნაწილში.

ახლა ჩვენ შეგვიძლია გადავიდეთ სხვა ტერმინალის ფანჯარაში და გავაგზავნოთ ცარიელი შეტყობინება ჩვენს თემაში:

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

PS გამოყენებით @ ჩვენ ვაჩვენებთ, რომ ჩვენ ვაგზავნით შეტყობინებას თემაზე სახელწოდებით "collect_securities".

ამ შემთხვევაში, შეტყობინება გადავიდა მე-6 განყოფილებაში - ამის შემოწმება შეგიძლიათ kafdrop-ზე გადასვლით localhost:9000

ჩვენს თანამშრომელთან ერთად ტერმინალის ფანჯარაში გადასვლისას დავინახავთ ბედნიერ შეტყობინებას, რომელიც გაგზავნილია loguru-ს გამოყენებით:

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

ვინაიდან ეს აგენტი დაამუშავებს ინფორმაციას კონკრეტული უსაფრთხოების შესახებ, ჩვენ უნდა მივუთითოთ ამ უსაფრთხოების ტიკერი (სიმბოლო) შეტყობინებაში. ამ მიზნით ფაუსტში არსებობს ჩანაწერები — კლასები, რომლებიც აცხადებენ შეტყობინებების სქემას აგენტის თემაში.

ამ შემთხვევაში, მოდით წავიდეთ ჩანაწერები.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. გარდა ამისა, ყველაფერი ერთი და იგივე სქემით მიჰყვება, ასე რომ, მე ვერ ვხედავ აზრს სხვა რამეზე ფიქრში.

ისე, საბოლოო შეხება არის ზარის დამატება მეტა ინფორმაციის შეგროვების აგენტთან collect_securites-ისთვის:

....
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 კლავიშს

მას შემდეგ, რაც გამოცხადდა აგენტები აგენტები.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 Worker და დავიწყოთ ფასიანი ქაღალდების სრულფასოვანი კოლექცია:

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

რა მოხდება შემდეგ?

შემდეგ ნაწილში, დარჩენილი აგენტების მაგალითის გამოყენებით, ჩვენ განვიხილავთ საძირკვლის მექანიზმს ვაჭრობის დახურვის ფასებში ექსტრემების მოსაძებნად წლის განმავლობაში და აგენტების გაშვება.

დღეისთვის სულ ესაა! Მადლობა წაკითხვისთვის :)

კოდი ამ ნაწილისთვის

PS ბოლო ნაწილში მკითხეს ფაუსტისა და შერწყმის კაფკაზე (რა თვისებები აქვს კონფლენტს?). როგორც ჩანს, კონფლუენტი მრავალი თვალსაზრისით უფრო ფუნქციონალურია, მაგრამ ფაქტია, რომ ფაუსტს არ აქვს სრული კლიენტის მხარდაჭერა კონფუენტისთვის - ეს გამომდინარეობს იქიდან. კლიენტის შეზღუდვების აღწერა დოკუმენტში.

წყარო: www.habr.com