ืืืฉ ืคืื ืืื ืืึทืื
-
ืืืื ืื: ืึทืืขื ืฅ ืืื ืืืื
ืืืืก ืืืขื ืืืจ ืื?
ืึทืืื, ืึทืืื, ืืขืจ ืฆืืืืืืขืจ ืืืื. ืืื ืืขืฉืจืืื ืคืจืืขืจ, ืืื ืขืก ืืืจ ืืืขืื ืืึธื ืื ืคืืืืขื ืืข:
-
ืืึธืืืจ ืฉืจืืึทืื ืึท ืงืืืื ืงืืืขื ื ืคึฟืึทืจ ืึทืืคืึทืืืึทื ืืืืืฉ ืืืืฃ aiohttp ืืื ืจืืงืืืขืก ืคึฟืึทืจ ืื ืขื ืืคึผืืื ืืก ืืืึธืก ืืืจ ืืึทืจืคึฟื.
-
ืืึธืืืจ ืฉืึทืคึฟื ืึทื ืึทืืขื ื ืืืึธืก ืืืขื ืงืืืึทืื ืืึทืื ืืืขืื ืกืืงืืืจืึทืืื ืืื ืืขืืึท ืืื ืคึฟืึธืจืืึทืฆืืข ืืืืฃ ืืื.
ืึธืืขืจ, ืืึธืก ืืื ืืืึธืก ืืืจ ืืืขืื ืืึธื ืคึฟืึทืจ ืื ืคึผืจืืืขืงื ืืื, ืืื ืืื ืืขืจืืื ืขื ืคืื ืคืึทืกื ืคืึธืจืฉืื ื, ืืืจ ืืืขืื ืืขืจื ืขื ืืื ืฆื ืฉืจืืึทืื ืึทืืขื ืฅ ืืืึธืก ืคึผืจืึทืกืขืกื ืกืืจืืืื ื ืืขืฉืขืขื ืืฉื ืคืื ืงืึทืคืงืึท, ืืื ืืืื ืืื ืฆื ืฉืจืืึทืื ืงืึทืืึทื ืื (ืืื ืจืึทืคึผืขืจ), ืืื ืืื ืืืขืจ ืคืึทื - ืคึฟืึทืจ ืืึทื ืืึทื ืฉืืืคึผื ืึทืจืืืงืืขื ืฆื ืืขืจ ืืขืืข ืืืึธืก ืืขืจ ืึทืืขื ื ืืื ืืึธื ืืืึธืจืื ื.
ืืจืึทืื ืื ื
AlphaVantage Client
ืขืจืฉืืขืจ, ืืึธืื ืืื ืื ืฉืจืืึทืื ืึท ืงืืืื ืึทืืึธืืืืคึผ ืงืืืขื ื ืคึฟืึทืจ ืจืืงืืืขืก ืฆื ืึทืืคืึทืืืึทื ืืืืืฉ.
ืกืคึผืึธืืืขืจ
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
)
ืืื ืคืึทืงื, ืึทืืฅ ืืื ืงืืึธืจ ืคืื ืืื:
-
ืื AlphaVantage API ืืื ืืึทื ืฅ ืคึผืฉืื ืืื ืืืืืึทืคืื ืืืืืื ื, ืึทืืื ืืื ืืึทืฉืืึธืกื ืฆื ืืึทืื ืึทืืข ืจืืงืืืขืก ืืืจื ืืขื ืืืคึฟื
construct_queryืืื ืืื ืงืขืจ ืขืก ืืื ืึท ืืืืคึผ ืจืืคื. -
ืืื ืืจืขื ื ืืืข ืคืขืืืขืจ ืฆื
snake_caseืคึฟืึทืจ ืืจืืืกื. -
ื ื, ืื logger.catch ืืึทืคึผืืฆืื ื ืคึฟืึทืจ ืฉืืื ืืื ืื ืคืึธืจืืึทืืืื ืืจืืืกืืึทืง ืจืขืืืืืึทื.
ืคึผืก ืื ืืืืกื ื ืืฉื ืคืึทืจืืขืกื ืฆื ืืืืื ืื ืึทืืคืึทืืืึทื ืืึทืืข ืกืืืขื ืืึธืืงืึทืื ืฆื config.yml, ืึธืืขืจ ืึทืจืืืกืคืืจื ืื ืกืืืืืืข ืืืึทืืขืืืืืง HORTON_SERVICE_APIKEY. ืืืจ ืืึทืงืืืขื ืึท ืกืืื .
CRUD ืงืืึทืก
ืืืจ ืืืขืื ืืึธืื ืึท ืกืืงืืืจืึทืืื ืืึทืืืื ื ืฆื ืงืจืึธื ืืขืืึท ืืื ืคึฟืึธืจืืึทืฆืืข ืืืขืื ืกืืงืืืจืึทืืื.
ืืื ืืืื ืืืื ืื ื, ืขืก ืืื ื ืื ืืึทืจืคึฟื ืฆื ืืขืจืงืืขืจื ืขืคึผืขืก ืืึธ, ืืื ืื ืืึทืืข ืงืืึทืก ืืื ืืื ืืึทื ืฅ ืคึผืฉืื.
ืืึทืงืืืขื_ืึทืคึผ ()
ืืึธืืืจ ืืืืื ืึท ืคึฟืื ืงืฆืืข ืคึฟืึทืจ ืงืจืืืืืื ื ืึท ืึทืคึผืืึทืงืืืฉืึทื ืืืืคืขืฅ ืืื
ืกืคึผืึธืืืขืจ
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ืึทืืื, ืขืจืฉืืขืจ ืืืจ ืืึทืงืืืขื ืื ืคืึทืกื ืึทืคึผืืึทืงืืืฉืึทื ืืืืคืขืฅ - ืืึธืก ืืื ืืึทื ืฅ ืคึผืฉืื. ืืืืึทืืขืจ, ืืืจ ืืขืจืงืืขืจื ืืคืืจืืฉ ืึท ืืขืืข ืคึฟืึทืจ ืืื ืืืขืจ ืึทืืขื ื ... ืืึธ ืขืก ืืื ืืืขืจื ืืขืจืืึธื ืขื ืืืึธืก ืขืก ืืื, ืืืึธืก ืืื ืื ืื ืขืจืืขื ืคึผืึทืจืึทืืขืืขืจ ืืื ืืื ืืึธืก ืงืขื ืขื ืืืื ืขืจืืื ืืืฉื ืึทื ืืขืจืฉ.
-
ืืขืืขืก ืืื ืงืึทืคืงืึท, ืืืื ืืืจ ืืืืื ืฆื ืืืืกื ืื ืคึผืื ืืืขื ืืขืคึฟืื ืืฆืืข, ืขืก ืืื ืืขืกืขืจ ืฆื ืืืืขื ืขื , ืึธืืขืจ ืืืจ ืงืขื ืขื ืืืืขื ืขื ืืืืฃ Habrรฉ ืืื ืจืืกืืฉ, ืืื ืึทืืฅ ืืื ืืืื ืฉืคืืืื ืืึทื ืฅ ืึทืงืืขืจืึทืืื :)
-
, ืืืกืงืจืืืื ืืึทื ืฅ ืืื ืืื ืื ืคืึทืืกื ืืึธืง, ืึทืืึทืื ืืื ืื ืฆื ืงืึทื ืคืืืืขืจ ืื ืืขืืข ืืืืื ืืื ืื ืงืึธื, ืืึธื, ืืึธืก ืืืื ืื ืคึผืึทืจืึทืืขืืขืจืก ืฆืืืขืฉืืขืื ืืืจื ืื ืคืึทืืกื ืืขืืืขืืึธืคึผืขืจืก, ืืืฉื: ืจืืืขื ืฉืึทื, ืจืืืขื ืฉืึทื ืคึผืึธืืืืืง (ืืืจื ืคืขืืืงืืึทื ืืืกืืขืงื, ืึธืืขืจ ืืืจ ืงืขื ืขื ืฉืืขืื ), ื ืืืขืจ ืคืื ืคึผืึทืจืืืฉืึทื ื ืคึผืขืจ ืืขืืข (ืฆื ืืึธื, ืืืฉื, ืืืืื ืืงืขืจ ืืื ืคืึทืกื ืึทืคึผืืึทืงืืืฉืึทื ื).
-
ืืื ืึทืืืขืืืื, ืืขืจ ืึทืืขื ื ืงืขื ืขื ืืึทืื ืึท ืืขืจืืื ืืขืืข ืืื ืืืืืืืข ืืืึทืืืขืก, ืึธืืขืจ ืืื ืืื ืฆื ืืขืจืงืืขืจื ืึทืืฅ ืืคืืจืืฉ. ืืื ืึทืืืฉืึทื, ืขืืืขืืข ืคึผืึทืจืึทืืขืืขืจืก (ืืืฉื, ืื ื ืืืขืจ ืคืื ืคึผืึทืจืืืฉืึทื ื ืึธืืขืจ ืจืืืขื ืฉืึทื ืคึผืึธืืืืืง) ืคืื ืื ืืขืืข ืืื ืื ืึทืืขื ื ืึทืืืืขืจืืืืืืึทื ื ืงืขื ืขื ื ืื ืืืื ืงืึทื ืคืืืืขืจื.
ืืึธ ืก ืืืึธืก ืขืก ืงืขื ืงืืงื ืืื ืึธื ืืึทื ืืืึทืื ืืืคืืื ืื ื ืื ืืขืืข:
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 ืกืขืกืืข ืคึฟืึทืจ ืจืืงืืืขืก ืืืจื ืืื ืืืขืจ ืงืืืขื ื. ืืืื, ืืืขื ืืืจ ืึธื ืืืืื ืึท ืึทืจืืขืืขืจ, ืืืขื ืืื ืืืขืจ ืึทืืขื ื ืืื ืืึธื ืืฉื, ืึท ืกืขืกืืข ืืืขื ืืืืื ืขืคึฟืขื ืขื - ืืืื ืขืจ, ืคึฟืึทืจ ืื ืืื ืฆืข ืฆืืื ืืืึธืก ืืขืจ ืึทืจืืขืืขืจ ืืื ืคืืืกื ืืืง (ืึธืืขืจ ืขืืืขืืข, ืืืื ืืืจ ืืืืฉื ืืขื ืคึผืึทืจืึทืืขืืขืจ ืคืื ืึทื ืึทืืขื ื ืืื ืึท ืคืขืืืงืืึทื ืึทืคึผืึทืจืึทื).
ืืืืึทืืขืจ, ืืืจ ื ืึธืืืืื ืื ืืืึทื (ืืืจ ืฉืืขืื ืืขื ืึธื ืืึธื ืืื _, ืืื ื ืืืจ, ืืื ืืขื ืึทืืขื ื, ืืึธื ื ืื ืืึธืจืื ืืืขืื ืื ืืื ืืึทืื) ืคืื ืึทืจืืืงืืขื ืคืื ืืื ืืืขืจ ืืขืืข, ืืืื ืืื ืขืงืกืืกืืืจื ืืื ืืขื ืงืจืึทื ื ืคืึธืืึธ, ืึทื ืืขืจืฉ ืืื ืืืขืจ ืฆืืงื ืืืขื ืืืึทืจืื ืคึฟืึทืจ ืืืืขืจ ืึธื ืงืืืขื. ื ื, ืืื ืืื ืืืขืจ ืฉืืืืฃ, ืืืจ ืงืืึธืฅ ืื ืงืึทืืึธืืข ืคืื โโืืขื ืึธื ืืึธื, ืืึทืงืืืขื ืึท ืจืฉืืื ืคืื ืึทืงืืืื (ืืขื_ืกืขืงืืจืืืืขืก ืงืขืจื ืืืืื ืึทืงืืืื ืืืจื ืคืขืืืงืืึทื, ืืขื ืงืืืขื ื ืงืึธื) ืกืืงืืืจืึทืืื ืืื ืจืึทืืขืืืขื ืขืก ืฆื ืื ืืึทืืึทืืืืก, ืงืึธื ืืจืึธืืืจื ืืืื ืขืก ืืื ืึท ืืืืขืจืืืื ืืื ืืขืจ ืืขืืืืงืขืจ ืืืงืขืจ ืืื ืืืขืงืกื ืืื ืื ืืึทืืึทืืืืก , ืืืื ืขืก ืืื, ืขืก (ืื ืคึผืึทืคึผืืจ) ืืืขื ืคืฉืื ืืืื ืืขืจืืืึทื ืืืงื.
ืืื ืก ืงืึทืืขืจ ืืื ืืืขืจ ืฉืึทืคืื ื!
> 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, ืืขื ืืืขื ืคึฟืื - ืึทืคึผืืึทืงืืืฉืึทื ืืืืคืขืฅ ืคึผืึทืจืึทืืขืืขืจ), ืืื ื ืืืจ ืืึธืื ื ืืฉื ืกืคึผืขืฆืืคืืฆืืจื ืึท ืืืื ืืืขืจื ืคึฟืึทืจ ืืื ืืืขืจ ืืขืืข (ืืืจื ืคึผืึทืจืืืฉืึทื ื). ืื ืืึธื ืืฉื ืึทืืขื ื ืืื ืืขืจ ืึทืจืืขืืขืจ ืืื ืึทืกืืื ื ืึทืืข 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 - ืืืจ ืงืขื ืขื ืงืึธื ืืจืึธืืืจื ืืขื ืืืจื ืืืื ืฆื ืงืึทืคืืจืึธืคึผ ืืืืฃ 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:
...ืืื ื ืืขืจ ืึทืืขื ื ืืืขื ืคึผืจืึธืฆืขืก ืืื ืคึฟืึธืจืืึทืฆืืข ืืืขืื ืึท ืกืคึผืขืฆืืคืืฉ ืืืืขืจืืืื, ืืืจ ืืึทืจืคึฟื ืฆื ืึธื ืืืืึทืื ืื ืืืงืขืจ (ืกืืืืึธื) ืคืื ืืขื ืืืืขืจืืืื ืืื ืืขื ืึธื ืืึธื. ืคึฟืึทืจ ืืขื ืฆืืืขืง ืืื ืคืึทืืกื ืขืก ืืขื ืขื - ืงืืืกื ืืืึธืก ืืขืจืงืืขืจื ืื ืึธื ืืึธื ืกืืขืืข ืืื ืื ืึทืืขื ื ืืขืืข.
ืืื ืืขื ืคืึทื, ืืึธืื ืืื ืื ืืืื ืฆื ืืื ืืึทืฉืจืืึทืื ืืื ืืขืจ ืึธื ืืึธื ืคึฟืึทืจ ืืขื ืืขืืข ืืึธื ืงืืงื ืืื:
import faust
class CollectSecurityOverview(faust.Record):
symbol: str
exchange: str
ืืื ืืืจ ืงืขื ืืึธืื ืืขืกื, Faust ื ืืฆื ืื ืคึผืืืืึธื ืืืคึผ ืึทื ืึธืืึทืฆืืข ืฆื ืืึทืฉืจืืึทืื ืื ืึธื ืืึธื ืกืืขืืข, ืืืึธืก ืืื ืืืึธืก ืื ืืื ืืืื ืืืขืจืกืืข ืืขืฉืืืฆื ืืืจื ืื ืืืืืืึธืืขืง ืืื .
ืืึธืืืจ ืฆืืจืืงืงืืืขื ืฆื ืื ืึทืืขื ื, ืฉืืขืื ืื ืืืืคึผืก ืืื ืืืืื ืขืก:
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"])
)
....ืืืจ ื ืืฆื ืื ืคืจืืขืจ ืืืืืข ืกืืขืืข ืคึฟืึทืจ ืืขื ืึธื ืืึธื. ืืื ืืขื ืคืึทื, ืืื ืืขืืืืื ื ืื .ืงืึทืกื ืืืคึฟื ืืื ื ืืืจ ืืึธื ื ืื ืืึทืจืคึฟื ืฆื ืืืึทืจืื ืคึฟืึทืจ ืื ืจืขืืืืืึทื ืคืื ืื ืึทืืขื ื, ืึธืืขืจ ืขืก ืืื ืืืขืจื ืืขืจืืึธื ืขื ืึทื ืฉืืงื ืึท ืึธื ืืึธื ืฆื ืื ืืขืืข:
-
ืืืึทืจืคื - ืืื ื ืืฉื ืคืึทืจืฉืคึผืึทืจื ืืืืึทื ืขืก ืืื ื ืืฉื ืืขืจืืืึทืจืื ืึท ืจืขืืืืืึทื. ืืืจ ืงืขื ืขื ื ืืฉื ืฉืืงื ืื ืจืขืืืืืึทื ืฆื ืื ืื ืืขืจ ืืขืืข ืืื ืึท ืึธื ืืึธื.
-
ืฉืืงื - ืืื ื ืืฉื ืคืึทืจืฉืคึผืึทืจื ืืืืึทื ืขืก ืืื ื ืืฉื ืืขืจืืืึทืจืื ืึท ืจืขืืืืืึทื. ืืืจ ืงืขื ืขื ืกืคึผืขืฆืืคืืฆืืจื ืึทื ืึทืืขื ื ืืื ืืขืจ ืืขืืข ืฆื ืืืึธืก ืืขืจ ืจืขืืืืืึทื ืืืขื ืืืื.
-
ืคืจืขืื - ืืืืืฅ ืคึฟืึทืจ ืึท ืจืขืืืืืึทื. ืืืจ ืงืขื ืขื ืกืคึผืขืฆืืคืืฆืืจื ืึทื ืึทืืขื ื ืืื ืืขืจ ืืขืืข ืฆื ืืืึธืก ืืขืจ ืจืขืืืืืึทื ืืืขื ืืืื.
ืึทืืื, ืืึธืก ืืื ืึทืืข ืืื ืืืขื ืื ืคึฟืึทืจ ืืืึทื ื!
ืืขืจ ืืืื ืืึทื ืฉืึทืคึฟื
ืื ืืขืฆืืข ืืึทื ืืื ืฆืืืขืืืื ืฆื ืฉืจืืึทืื ืืื ืืขื ืืืื ืืื ืงืึทืืึทื ืื. ืืื ืืขืจืืื ื ืคืจืืขืจ, ืงืึทืืึทื ืื ืืื ืคืึทืืกื ืืขื ืขื ืึท ืจืึทืคึผืขืจ ืึทืจืื ืืื. ืืื ืคืึทืงื, Faust ืคืฉืื ืึทืืึทืืฉืื ืืื ืืืขืจ ืื ืื ืืึทืคึฟืขื ืฆื ืืืื ืฆืืืื ื ืืืขื ืกืคึผืขืฆืืคืืฆืืจื ืื -A ืฉืืืกื
ื ืึธื ืื ืืืืืข ืืืขื ืื ืืื ืืืืื ืึท ืคึฟืื ืงืฆืืข ืืื ืึท ืืขืงืขืจืืืืขืจ app.commandืจืืคื ืืขื ืืืคึฟื ืืืก ั collect_securitites:
@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