Python

こんにちは。ナナオです。 アービトラージに興味があり、開発してみたいな～と思いつつなかなか手が出せなかったのですが、この度重い腰をあげて開発してみました。 使用技術 監視にはPrometheus + Grafanaを使用しました。 実装はPythonを使い、HTTPリクエストにrequestsを使っています。 実装 早速実装です。 メインになる実装は以下の通りです。 ここの実装が大体100行くらいです。 import time import logging import threading from itertools import combinations # リクエストに使うAPI（自作） from vc_bot import exchange_api from prometheus_client import start_http_server, Gauge, Counter # 必要に応じてログの設定（これは標準エラー出力に出す設定です） logging.basicConfig(level=logging.ERROR) logger = logging.getLogger(__name__) PROFIT_GAUGE = Gauge("profit", "利益", ["exchange", "symbol"]) PROFIT_RATE_GAUGE = Gauge("profit_rate", "利益率", ["exchange", "symbol"]) REQUEST_ERROR_GAUGE = Counter("request_error", "リクエストエラー", ["exchange"]) def worker(): # 対象の取引所 exchange_pair = { "BTC_JPY": ["gmo", "coincheck", "binance", "bitflyer", "zaif", "bitbank", "okcoin"], "ETH_JPY": ["gmo", "coincheck", "binance", "bitflyer", "zaif", "bitbank", "okcoin"], "XRP_JPY": ['gmo', 'coincheck', 'binance', 'bitflyer', 'bitbank', 'okcoin'], "MONA_JPY": ['coincheck', 'bitflyer', 'zaif', 'bitbank'], } while True: for pair, exchanges in exchange_pair.items(): prices = {} # リクエストが成功した取引所のみ格納 success_exchanges = [] for exchange in exchanges: try: api = getattr(exchange_api, exchange)() prices.update({exchange: api.fetch_ticker(pair=pair)}) success_exchanges += [exchange] except Exception: logger.exception("リクエスト中にエラーが発生しました") REQUEST_ERROR_GAUGE.labels( exchange=exchange, ).inc() continue # 各取引所の比較結果を格納するGaugeオブジェクトを初期化 exchange_combination = list(combinations(success_exchanges, 2)) for exchange1, exchange2 in exchange_combination: ex1 = prices[exchange1] ex2 = prices[exchange2] # パターンA: Ex1で買って(Ask)、Ex2で売る(Bid) profit_a = ex2["bid"] - ex1["ask"] profit_rate_a = (profit_a / ex1["ask"]) * 100 print(f"{pair} {exchange1}-{exchange2} profit : {profit_a}") print(f"{pair} {exchange1}-{exchange2} profit rate: {profit_rate_a}") PROFIT_GAUGE.labels( exchange=f"{exchange1}_{exchange2}", symbol=pair ).set(profit_a) PROFIT_RATE_GAUGE.labels( exchange=f"{exchange1}_{exchange2}", symbol=pair ).set(profit_rate_a) # パターンB: Ex2で買って(Ask)、Ex1で売る(Bid) profit_b = ex1["bid"] - ex2["ask"] profit_rate_b = (profit_a / ex2["ask"]) * 100 print(f"{pair} {exchange2}-{exchange1} profit : {profit_b}") print(f"{pair} {exchange2}-{exchange1} profit rate: {profit_rate_b}") PROFIT_GAUGE.labels( exchange=f"{exchange2}_{exchange1}", symbol=pair ).set(profit_b) PROFIT_RATE_GAUGE.labels( exchange=f"{exchange2}_{exchange1}", symbol=pair ).set(profit_rate_b) def main(): # Prometheus ExporterのHTTPサーバーをポート8000で起動 start_http_server(8000) # メトリクス更新ワーカーを別スレッドで実行 worker_thread = threading.Thread(target=worker, daemon=True) worker_thread.start() print("Prometheus metrics server running on port 8000") # メインスレッドを維持 try: while True: time.sleep(1) except KeyboardInterrupt: print("Exiting.") if __name__ == "__main__": main() exchange_apiの実装は以下の通りです。 ...

どうも、ナナオです。 最近PythonからRustを呼び出す実装をすることがありまして、pyo3にお世話になることがありました。 pyo3のビルドにはmaturinというCLIを使うのですが、これをRustのワークスペース機能と併用できるのかどうか気になったので、検証してみたいと思います。 準備 とりあえず適当にpyo3を使用したライブラリを作ります。 ryeを使っていれば以下のコマンドでmaturinをビルダーに指定したプロジェクトを作成できます。 rye init maturin-workspace-playground --build-system maturin maturinをインストールしていなかったので、以下のコマンドでインストールしておきます。 rye install maturin 作ったプロジェクトに移動して、ワークスペースのメンバーになるプロジェクトを作成しておきます。 cd maturin-workspace-playground mkdir rust && cd rust cargo init --lib app cargo init --lib python-api 作成したプロジェクトをワークスペースのメンバーになるように設定をしていきます。 ルートディレクトリのCargo.tomlを以下のように編集します。 [workspace.package] name = "maturin-workspace-playground" version = "0.1.0" edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [workspace] members = ["rust/python-api", "rust/app"] rust/appのCargo.tomlは以下のように編集します。 [package] name = "app" version.workspace = true edition.workspace = true # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] rust/python-apiのCargo.tomlは以下のように編集します。 ...

前回、prisma-client-pythonに入門しました。 しかし、どうやってNodeが動いているのかちょっとわからなかったので調べてみました。 Nodeの実態は？ 現状、グローバルにprismaが使える状態ではないため、普通にprismaコマンドを実行しても実行できません。 % prisma zsh: command not found: prisma しかし、前回作ったパッケージ上であればprismaコマンドは実行できます。 % poetry run prisma This command is only intended to be invoked internally. Please run the following instead: prisma <command> e.g. prisma generate 説明を見るかぎり、ここで実行されているprismaは本家のCLIのラッパーとして実装されているようです。 Prisma Client Python exposes a CLI interface which wraps the Prisma CLI. This works by downloading a Node binary, if you don’t already have Node installed on your machine, installing the CLI with npm and running the CLI using Node. ...

今、自分の中でORMツールとしてアツいのがprismaです。 Prisma | Instant Postgres plus an ORM for simpler db workflows モデル定義もマイグレーションも、ドキュメント読んだり実装してみたりすればするほどかなり使える書き方ができるので、すごく可能性を感じています。 そんなprismaですが、python用のクライアントがあるということで、勉強がてら触ってみたいと思います。 Prisma Client Python pythonプロジェクトの初期化 まずはpoetryでプロジェクトを初期化します。 poetry new prisma-client-python-playground 作成されたプロジェクトに依存性を追加します。 poetry add -D prisma さて、これでいつものprisma CLIが使えるようになりました。 早速Prismaの初期化をしていきましょう。 今回もsqliteを使用していきます。 % poetry run prisma init --datasource-provider sqlite ✔ Your Prisma schema was created at prisma/schema.prisma You can now open it in your favorite editor. Next steps: 1. Set the DATABASE_URL in the .env file to point to your existing database. If your database has no tables yet, read https://pris.ly/d/getting-started 2. Run prisma db pull to turn your database schema into a Prisma schema. 3. Run prisma generate to generate the Prisma Client. You can then start querying your database. More information in our documentation: https://pris.ly/d/getting-started 初期化できました。 ...

markdownをplain textにしたうえで、その文章をginzaでリスト化しました。 必要なライブラリをインストールします。 pip install Markdown beautifulsoup4 ginza ja-ginza 以下のような実装を行いました。 from bs4 import BeautifulSoup from markdown import markdown import re import spacy def markdown_to_text(markdown_string): """マークダウンを平文に変換します""" # md -> html -> text since BeautifulSoup can extract text cleanly html = markdown(markdown_string) # remove code snippets html = re.sub(r'<pre>(.*?)</pre>', ' ', html) html = re.sub(r'<code>(.*?)</code >', ' ', html) # extract text soup = BeautifulSoup(html, "html.parser") text = ''.join(soup.find_all(string=True)).replace("\n", "　") return text def split_sentences(text): nlp = spacy.load('ja_ginza') doc = nlp(text) sentences = [sent.text for sent in doc.sents] return sentences if __name__ == "__main__": markdown_string = """# タイトル ## サブタイトル これは段落です。 これは別の段落です。 """ text = markdown_to_text(markdown_string) print(text) sentences = split_sentences(text) print(sentences) 出力は以下のとおりです。 ...

いつも通り開発していたのに、poetry installを行ったら急に全く動かなくなりました。 そんな場合の対処法について発見したので共有します。 対処法 poetry cacheコマンドを利用してキャッシュの削除を行います。 まず、poetry cache listでキャッシュの一覧を表示します。 次に、先程表示したキャッシュ一覧の名前を使用してpoetry cache clear --all [キャッシュ名]とタイプします。 すべてのキャッシュ一覧に対してこの操作を実行後、poetry installを実施したらうまく動きました。 めでたしめでたし。