ファウストのバックグラウンド タスク、パート II: エージェントとチーム

目次

1. パート I: はじめに

2. パート II: エージェントとチーム

私たちはここで何をしているのでしょうか？

それで、それで、第二部。 前に書いたように、その中で次のことを行います。

1. 必要なエンドポイントに対するリクエストを含む aiohttp 上の alphavantage 用の小さなクライアントを作成してみましょう。

2. 証券に関するデータと証券に関するメタ情報を収集するエージェントを作成しましょう。

しかし、これはプロジェクト自体に対して行うことであり、ファウストの研究の観点からは、kafka からのストリーム イベントを処理するエージェントの作成方法と、コマンド (クリック ラッパー) の作成方法を学びます。エージェントが監視しているトピックへの手動プッシュ メッセージの場合。

訓練

AlphaVantage クライアント

まず、alphavantage へのリクエスト用の小さな aiohttp クライアントを作成しましょう。

アルファバンテージ.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クラス

証券に関するメタ情報を保存するための証券コレクションを用意します。

データベース/セキュリティ.py

私の意見では、ここで何も説明する必要はなく、基底クラス自体は非常に単純です。

アプリを取得（）

にアプリケーションオブジェクトを作成する関数を追加しましょう。 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

したがって、最初に faust アプリケーション オブジェクトを取得します。これは非常に単純です。 次に、エージェントのトピックを明示的に宣言します。ここでは、トピックが何であるか、内部パラメータが何であるか、およびこれをどのように変更できるかについて言及する価値があります。

1. Kafka のトピック。正確な定義を知りたい場合は、読むことをお勧めします。 オフ。 書類、または読むことができます 概要 ロシア語のハブレでは、すべてが非常に正確に反映されています:)

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 セッションを開きます。 したがって、ワーカーを開始するとき、エージェントが起動されると、セッションがすぐに開かれます - ワーカーが実行されている間ずっと XNUMX つ (またはパラメーターを変更した場合は複数) 並行性 デフォルトのユニットを使用するエージェントから）。

次に、ストリームに従います (メッセージを _、このエージェントでは、現在のオフセットにメッセージが存在する場合、トピックからのメッセージの内容を気にしないため、そうでない場合、サイクルはメッセージの到着を待ちます。 さて、私たちのループ内では、メッセージの受信をログに記録し、アクティブな証券のリストを取得して (get_securities はデフォルトでアクティブな証券のみを返します。クライアント コードを参照)、それをデータベースに保存し、同じティッカーを持つ証券があるかどうかを確認し、データベース内に交換があれば、それ (論文) が更新されるだけです。

私たちの作品を立ち上げましょう！

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

PSの特徴 ウェブコンポーネント 記事内ではファウストは考慮しないので適当なフラグを立てます。

起動コマンドでは、情報ログの出力レベルを使用して、アプリケーション オブジェクトをどこで探すか、それをどう処理するか (ワーカーを開始する) を faust に指示しました。 次の出力が得られます。

スポイラー

┌ƒ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 を使用して) mongo を調べて、証券がデータベース内にあることを確認することもできます。

私は億万長者ではないので、最初の視聴オプションに満足しています。

幸せと喜び - 最初のエージェントの準備ができました:)

エージェントの準備は完了です。新しいエージェント万歳！

はい、皆さん、この記事で用意した手順の 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

ご想像のとおり、faust は Python 型アノテーションを使用してメッセージ スキーマを記述します。そのため、ライブラリでサポートされる最小バージョンは次のとおりです。 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_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"])
    )
....

メッセージには以前に発表されたスキームを使用します。 この場合、エージェントからの結果を待つ必要がないため、.cast メソッドを使用しましたが、言及する価値があります。 方法 トピックにメッセージを送信します。

1. Cast - 結果を期待していないためブロックしません。 結果をメッセージとして別のトピックに送信することはできません。

2. send - 結果を期待していないためブロックしません。 結果が送信されるトピック内でエージェントを指定できます。

3. ask - 結果を待ちます。 結果が送信されるトピック内でエージェントを指定できます。

それでは、今日のエージェントについてはここまでです。

最高のチーム

このパートで最後に書くと約束したのはコマンドです。 前述したように、faust のコマンドはクリックに関するラッパーです。 実際、faust は、-A キーを指定するときにカスタム コマンドをインターフェイスにアタッチするだけです。

発表されたエージェントの後、 エージェント.py デコレータを使って関数を追加する アプリコマンドメソッドの呼び出し キャスト у 収集_セキュリティ:

@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

次に何が起こるのだろうか？

次のパートでは、残りのエージェントを例として、その年の取引の終値の極値を探すためのシンクメカニズムとエージェントの cron 起動について検討します。

それが今日のすべてです！ 読んでくれてありがとう ：）

この部分のコード

PS 最後の部分で、ファウストと合流カフカについて質問されました (Confluentにはどのような機能がありますか?）。 confluent は多くの点でより機能的であるように見えますが、実際のところ、faust は confluent を完全にクライアントでサポートしていません。これは次のとおりです。 ドキュメント内のクライアント制限の説明.

出所： habr.com