NetworkXのグラフをpyvisのグラフに変換して可視化する

pyvisはグラフを手軽に可視化できますし、pyvis自体のメソッドでノードやエッジを追加してグラフを構築することもできるので基本的な処理はこれだけで完結させることももちろんできます。しかし、グラフデータの分析をしていると、各種アルゴリズムが充実しているNetworkXでいろんな分析を行って、それを最後にpyvisで可視化したい、ということはよくある話です。

こういう場合、pyvisのネットワークオブジェクトが持っている、from_nxってメソッドが使えます。
参考: Documentation — pyvis 0.1.3.1 documentation

この記事の主題は「from_nxが便利だよ」で終わりなのですが、それだけではあんまりなので細かい話をいろいろ書いていきます。

まず、基本的な使い方ですが、from_nxは pyvisモジュールから直接呼び出せるメソッドではなく、pyvisのpyvis.network.Network オブジェクトに実装されているメソッドなので、まずそのインスタンスを生成します。以前の記事でも書いていますが、可視化するときのキャンパスサイズとか背景色などを指定して生成するやつですね。

具体的に適当なネットワークでやってみると次の様になります。

import networkx as nx
from pyvis.network import Network


# サンプルとして、NetworkXのグラフを生成
nx_graph = nx.Graph()
nx_graph.add_node("a")
nx_graph.add_node("b")
nx_graph.add_node("c")
nx_graph.add_edge("a", "b")
nx_graph.add_edge("b", "c")
nx_graph.add_edge("c", "a")


# ネットワークのインスタンス生成
network = Network(
    height="500px",
    width="500px",
    notebook=True,
    bgcolor='#ffffff',
    directed=False, 
)

# pyvisのネットワークに、NetworkXのグラフのノードやエッジ情報を取り込む。
network.from_nx(nx_graph)
# 可視化
network.show("sample.html")

結果は省略しますが、これで、a,b,cの3個のノードを持ったネットワークが可視化されます。

NetworkXで設定されていなかったがpyvisで可視化するときに必要な種類の情報はデフォルト値で補完されています。

# NetworkXのノードの情報
print(dict(nx_graph.nodes))
# {'a': {'size': 10}, 'b': {'size': 10}, 'c': {'size': 10}}

# pyvisに取り込んだときに、colorやshape、sizeのデフォルト値が設定されている。
print(network.nodes)
"""
[{'color': '#97c2fc', 'size': 10, 'id': 'a', 'label': 'a', 'shape': 'dot'},
 {'color': '#97c2fc', 'size': 10, 'id': 'b', 'label': 'b', 'shape': 'dot'},
 {'color': '#97c2fc', 'size': 10, 'id': 'c', 'label': 'c', 'shape': 'dot'}]
"""

# NetworkXのエッジの情報
print(dict(nx_graph.edges))
# {('a', 'b'): {'width': 1}, ('a', 'c'): {'width': 1}, ('b', 'c'): {'width': 1}}

# エッジはほぼそのまま取り込まれている。
print(network.edges)
# [{'width': 1, 'from': 'a', 'to': 'b'}, {'width': 1, 'from': 'a', 'to': 'c'}, {'width': 1, 'from': 'b', 'to': 'c'}]

ノードのサイズやエッジの太さは、デフォルト値をそれぞれ、default_node_size, default_edge_weight という引数で指定することもできるので、有効に使っていきましょう。例えばWebサイトのページ遷移データのネットワーク等で、NetworkX時点ではPVなどの大きい値をsizeとして計算していたら、それを可視化時のサイズとして使ってしまうとえらいことになります。

また、node_size_transf , edge_weight_transf という引数で、関数を渡しておくことで、それぞれの値を変換することもできます。元々の値が非常に大きい、または小さい場合にこれを使って補正することができますね。
例えば、
node_size_transf = (lambda x: x/10) とすると、ノードのサイズを1/10にできます。

ここで注意というか、version 3.2時点の pyvisにはバグがあって、エッジを持たない孤立ノードには node_size_transf が適用されません。

該当部分のソースコードがこちらです。

        if len(edges) > 0:
            for e in edges:
                if 'size' not in nodes[e[0]].keys():
                    nodes[e[0]]['size']=default_node_size
                nodes[e[0]]['size']=int(node_size_transf(nodes[e[0]]['size']))
                if 'size' not in nodes[e[1]].keys():
                    nodes[e[1]]['size']=default_node_size
                nodes[e[1]]['size']=int(node_size_transf(nodes[e[1]]['size']))
                self.add_node(e[0], **nodes[e[0]])
                self.add_node(e[1], **nodes[e[1]])

                # if user does not pass a 'weight' argument
                if "value" not in e[2] or "width" not in e[2]:
                    if edge_scaling:
                        width_type = 'value'
                    else:
                        width_type = 'width'
                    if "weight" not in e[2].keys():
                        e[2]["weight"] = default_edge_weight
                    e[2][width_type] = edge_weight_transf(e[2]["weight"])
                    # replace provided weight value and pass to 'value' or 'width'
                    e[2][width_type] = e[2].pop("weight")
                self.add_edge(e[0], e[1], **e[2])

        for node in nx.isolates(nx_graph):
            if 'size' not in nodes[node].keys():
                nodes[node]['size'] = default_node_size
            self.add_node(node, **nodes[node])

エッジが存在するノードの情報を取り込むときは、node_size_transfしてるのに、その後の孤立ノードの取り込みでは元のsizeとデフォルトノードサイズしか考慮してませんね。

これは将来のバージョンで修正されると思うのですが、こういうバグもあるので、サイズを変換したい場合はnode_size_transfではなく、自分で元のデータを修正してform_nxに渡した方が良いでしょう。

さらに、便利な機能なのですがノードのサイズやエッジの太さ以外の属性については、一通り全部コピーしてくれます。これを使って、可視化するときに設定したい情報などをNetowrkXのグラフオブジェクトの時点で設定しておくことも可能です。これはもちろん、pyvisのネットワークに変換してから付与してももちろん大丈夫なのですが。

ただ、例えばノードのクラスタリング結果を可視化時の色に反映させたい等の、何かしらのアルゴリズムの結果を可視化に使いたい場合は、NetworkXの時点で設定する方がやりやすいことが多いです。ただ、可視化時点でしか必要のない情報をNetworkXのオブジェクトに付与していくことに抵抗がある人もいるかもしれないので好みの問題だと思います。

基本的な話なのですが、以下の様にして属性を付与していけます。ノードを1つ、赤い三角形にしてみたり、エッジの一つを黒く塗ってラベルつけたりしています。

# 事前にグラフにいろいろ属性を設定できる。
nx_graph.nodes["a"]["color"] = "red"
nx_graph.nodes["a"]["shape"] = "triangle"

nx_graph.edges[("b", "c")]["color"] = "black"
nx_graph.edges[("b", "c")]["label"] = "hogehoge"

# 以降の Networkオブジェクトを作って from_nxするところは同じ。

以上が from_nx の説明です。とても便利なのでぜひ使ってみてください。

逆に、pyvisのNetworkをNetworkXのグラフに変換するメソッドは無いのかな、とも思ったのですが、専用のものはなさそうですね。まぁ、それぞれのライブラリの用途を考えれば必要もなさそうですし、どうしてもやりたければノードとエッジの情報をそれぞれ取り出してNetworkXのグラフを構築すればいいだけの話なのでそんなに難しくもなさそうです。

NetworkXでグラフが平面グラフかどうか判定する

諸事情ありグラフが平面グラフかどうか手軽に判定する方法が必要になったので、その方法を調べました。(正確には、僕が調べたかったのは平面的グラフかどうかですが。)

平面グラフの定義についてはWikipediaがわかりやすいです。
参考: 平面グラフ – Wikipedia

簡単にいうと、グラフを2次元の図に可視化したときに、辺が交差しない様に書いたものが平面グラフです。そして、グラフによっては描き方によって辺が交差したりしなかったりするのですが、平面グラフと同型なグラフを平面的グラフと呼びます。
どうにか頑張れば辺が交差しない様に頂点を配置できるグラフが平面的グラフ、どうやってもどこかで交差してしまうグラフが平面的で無いグラフ、と考えた方がわかりやすいでしょう。

これをどうやって判定しようかなと考えていたのですが、networkxに専用のメソッドが用意されていました。
check_planarity
is_planar

is_planar の方がシンプルで、NetworkXのグラフオブジェクトを渡すとそれが平面的だったかどうかでTrue/Falseを返してくれます。
check_planarity の方は、平面的だったかどうかの情報だけでなく、埋め込みの情報も取得できます。これは、ノードをどの様に配置したら平面的になるかを示すものです。

is_planar の方が結果がシンプルなので速度面で早いとか何かメリットあるのかなと思っていたのですが、NetworkXのソースコードを見ると、内部でcheck_planarityを呼び出して、一個目の戻り値だけを返すという作りだったのでそういうメリットはなさそうです。

とりあえず、結果がシンプルなis_planarの方を使ってみます。

import networkx as nx


# 頂点4個の完全グラフは平面的
K4 = nx.complete_graph(4)
print(nx.is_planar(K4))
# True

# 頂点5個の完全グラフは平面的では無い
K5 = nx.complete_graph(5)
print(nx.is_planar(K5))
# False

簡単ですね。

check_planarityの方を使う場合は、戻り値として判定結果と埋め込み情報が返ってくるので、それぞれ個別に受け取ります。

is_planar, P = nx.check_planarity(K4)

print(is_planar)
# True
print(P)
# PlanarEmbedding with 4 nodes and 12 edges
# これは、ノードをキーにして辞書の様にして使うと中身を見れる。
print(P[0])
# {1: {'cw': 3, 'ccw': 2}, 2: {'cw': 1, 'ccw': 3}, 3: {'cw': 2, 'ccw': 1}}
# 上記の結果で、ノード0に、1, 2, 3と繋がるエッジがあることがわかる。
print(P[0][1])
# {'cw': 3, 'ccw': 2}
# これは、ノード0に他のノードがどの様な順番で繋がっているかを示し、
# ノード1の時計回りの隣が3,反時計回りの隣が2であることを示す。

注意しないといけないのは、平面的グラフだからといって描写したら平面グラフで描かれるとは限らないという点です。というか、平面グラフで描かれる方が稀です。さっきの例で挙げた頂点4個の完全フラフでさえ、普通に力学モデルで配置したらエッジが交差します。

そのため、NetworkXでは平面グラフ描写用のメソッドも持っています。
参考: planar_layout — NetworkX 3.1 documentation

spring_layoutと比較してみましょう。なぜか、with_labels引数のデフォルト値が違うのでTrueつけないとノードのidを表示してくれません。

import matplotlib.pyplot as plt


fig = plt.figure(figsize=(12, 6), facecolor="w")
ax = fig.add_subplot(121, title="spring_layout")
ax.axis("off")
nx.draw_networkx(K4, ax=ax)

ax = fig.add_subplot(122, title="planar_layout")
nx.draw_planar(K4, ax=ax, with_labels=True)

出てくる図がこちら。

draw_planar を使わなくても、他のアルゴリズム同様にlayout系のメソッドで座標を取得してそれで描写することもできます。
参考: planar_layout — NetworkX 3.1 documentation

pos = nx.layout.planar_layout(K4)
print(pos)
"""
{0: array([-1.   , -0.375]),
 1: array([ 1.   , -0.375]),
 2: array([0.   , 0.125]),
 3: array([0.   , 0.625])}
"""

nx.draw_networkx(K4, pos=pos)
# 結果略

Pythonですでに宣言されている変数名や関数名などを取得、確認する

ある変数がすでに宣言されているかどうかで処理を分けるにはどうしたら良いか調べたので記事にしておきます。また、変数以外にも組み込み関数名の一覧の確認方法とか、あるオブジェクトが持ってるプロパティの取得方法とか似てる話題をまとめておきます。

コードを雑に書いてると、if文の制御の中で変数を定義することがあります。そうなるとその以降の処理でその変数が宣言されているかどうかが不明になりますね。例えば、どこからかデータの取得を試みて、成功したらpandasのDataFrameにして、dfとかそれらしい名前の変数に格納するけど、取得失敗したらdf変数が宣言されていない状態になるみたいなケースです。

ここから紹介する内容を否定する様なことを先に書いておきますが、この様な状況では、df = pd.DataFrame() みたいに空っぽのデータフレームか何か作って確実に変数が宣言されている状態にして、その後データが取れたらdfを置き換えて、以降はlen(df)が0か1以上かで処理を分けるみたいな回避策をとった方が確実にバグが少なく可読性の高いコードが書けそうです。

せっかく調べたので記事は書きますが、豆知識くらいに思って実際は別の回避策を検討するのがおすすめです。

それではやっていきましょう。結論として、Pythonの組み込み関数の中に、宣言されている変数の一覧(シンボルテーブル)を取得するメソッドがあります。グローバルレベルで結果を返してくれるものと、関数内等で使えばローカル変数だけ返してくれるもの(モジュールレベルで使うとグローバルと同じ結果)の二つがあります。

参考: 組み込み関数 — Python 3.10.11 ドキュメント の globals()とlocals()を参照。

イメージを掴むには使ってみると早いです。辞書型で結果を返してくれます。

$ python
>>> print(globals())
{'__name__': '__main__', '__doc__': None, '__package__': None, '__loader__': <class '_frozen_importlib.BuiltinImporter'>, '__spec__': None, '__annotations__': {}, '__builtins__': <module 'builtins' (built-in)>}
# 変数を一つ宣言してみる。
>>> foo = 4
# 最後に含まれている。
>>> print(globals())
{'__name__': '__main__', '__doc__': None, '__package__': None, '__loader__': <class '_frozen_importlib.BuiltinImporter'>, '__spec__': None, '__annotations__': {}, '__builtins__': <module 'builtins' (built-in)>, 'foo': 4}

# この様にして 変数 foo がすでに宣言されていることがわかる。
>>> "foo" in globals()
True

# locals() はスコープ内のローカル変数だけを持っている。
>>> def dummy_func():
...     func_var = 1
...     print(locals())
...
>>> dummy_func()
{'func_var': 1}

# スコープの外では参照できないので、globals()の結果には含まれていない。
>>> "func_var" in globals()
False

Jupyter等で長い長い作業を行ってたら、新しい変数を使うときにその時点で使ってないかの確認に使ったりとかできるかなぁとも思ったのですが、正直これを使わずにすむ様な回避策を探った方が良いと思います。僕自身、6年以上Python書いてますが、これまでglobals()やlocals()が必須な状況にはなった事ありませんし。

さて、以上で自分で宣言した変数やメソッド名の一覧(元々存在している__name__など含む)の取得方法がわかりました。

ここから先は元の主題から外れますが興味があったので調べた内容のメモです。

Pythonに限らずプログラミング言語では、あらかじめ予約語として抑えられていて使えない単語が複数あります。ifとかforとかfromとかですね。これらの名前は先ほどのglobals()の結果では出てきませんでした。

これらの予約語については、確認するための専用ライブラリが標準で用意されています。
参考: keyword — Python キーワードチェック — Python 3.11.3 ドキュメント

キーワードと、3個だけですがソフトキーワドの2種類あります。

import keyword


print(keyword.kwlist)
"""
['False', 'None', 'True', 'and', 'as', 'assert', 'async', 'await',
 'break', 'class', 'continue', 'def', 'del', 'elif', 'else', 'except',
 'finally', 'for', 'from', 'global', 'if', 'import', 'in', 'is',
 'lambda', 'nonlocal', 'not', 'or', 'pass', 'raise', 'return', 'try',
 'while', 'with', 'yield']
"""
print(keyword.softkwlist)
"""
['_', 'case', 'match']
"""

意外と少ないですね。

今の時点で、 sum や len など普段よく使っている組み込み関数たちが登場してないので、これらの情報がどこかで取得できないかも調べました。結果わかったのは、__builtins__ ってモジュールの配下に定義されているってことです。dirメソッドで確認できます。組み込みエラーとかもここにあるんですね。

dir(__builtins__)
['ArithmeticError',
 'AssertionError',
 'AttributeError',
 'BaseException',
 'BaseExceptionGroup',
# 中略
 'abs',
 'aiter',
 'all',
 'anext',
 'any',
 'ascii',
 'bin',
 'bool',
# 中略
 'enumerate',
 'eval',
 'exec',
 'execfile',
 'filter',
 'float',
 'format',
# 中略
 'str',
 'sum',
 'super',
 'tuple',
 'type',
 'vars',
 'zip']

sumとかabsとかstrとかお馴染みさんたちがいましたね。

ここで使いましたが、dir()は引数で渡したオブジェクトやモジュールが持っている属性やメソッドを羅列してくれる組み込み関数です。これもメソッド探しで使うことがあります。
参考: dir([object])

__builtins__ は省略してもメソッドにアクセスできるので通常は使うことはないし、これを使わなきゃいけない様な状況も作るべきでは無いと思いますが、無理やり活用するとしたら組み込み変数名を上書きしちゃったときでも元の機能が呼び出せます。

# 元々は足し算
sum([1, 2, 3])
# 6

# 予約後と違って組み込み変数は上書き出てきてしまう。
sum = sum([1, 2, 3])
# これでsumの中身が数値6になったので、ただのsumは関数として使えなくなった。
print(sum)
# 6

# __builtins__.sum は元通り足し算として機能する
__builtins__.sum([1, 2, 3])
# 6

まぁ、常識的に考えてこんな使い方をしなくてすむ様にコードを書くべきだと思いますね。和の変数名をsumにしたり、最大値の変数名をmaxにしたりして組み込み関数を上書きしてしまった経験は初心者時代にありますが。

boto3のAPIエラー発生時のリトライ回数の設定を変更する

先日、boto3でAmazon Comprehendを使っていたら、大量のデータを処理する中で次のエラーが頻発する様になってしまいました。

ClientError: An error occurred (ThrottlingException) when calling the BatchDetectSentiment operation (reached max retries: 4): Rate exceeded

短時間に実行しすぎてレートの上限に引っかかり、リトライも規定回数失敗してしまった様です。

Comprehend の使い方自体は過去の記事で書いてます。
参考: Amazon Comprehend でテキストのセンチメント分析

軽く紹介しておくとこんな感じですね。

import boto3


comprehend = boto3.client("comprehend")
comprehend_result = comprehend.batch_detect_sentiment(
        TextList=text_list,  # ここで判定したいテキストを渡す。
        LanguageCode="ja"
    )

応急処置的な対応としては、エラーをキャッチして自分でリトライを実装するという手もあります。
参考: 失敗しやすい処理にリトライをスクラッチで実装する

ただ、boto3はどうやら標準機能でリトライ設定の回数を設定できる様なのです。batch_detect_sentiment メソッドのドキュメントをいくら読んでもそれらしい引数がないのでできないと思ってました。
この設定は、その一歩手前、クライアントインスタンスを生成する時点で設定しておく必要があります。

ドキュメントはこちら。
参考: Retries – Boto3 1.26.142 documentation

ドキュメントでは ec2 のAPIに適用していますが、comprehendでも同じ様に使えます。以下の様にするとリトライ回数の上限を10回に上げられる様です。

import boto3
from botocore.config import Config

config = Config(
   retries = {
      'max_attempts': 10,
      'mode': 'standard'
   }
)

comprehend = boto3.client('comprehend', config=config)

mode は、 legacy, standard, adaptive の3種類があります。とりあえず、standardを選んだら良いのではないでしょうか。対応しているエラーがlegacyより多く、以下のエラーに対してリトライしてくれます。

# Transient errors/exceptions
RequestTimeout
RequestTimeoutException
PriorRequestNotComplete
ConnectionError
HTTPClientError

# Service-side throttling/limit errors and exceptions
Throttling
ThrottlingException
ThrottledException
RequestThrottledException
TooManyRequestsException
ProvisionedThroughputExceededException
TransactionInProgressException
RequestLimitExceeded
BandwidthLimitExceeded
LimitExceededException
RequestThrottled
SlowDown
EC2ThrottledException

max_attempts の方が、リトライ回数の設定ですが、正直、何回に設定したら成功する様になるのかは状況による部分が多く、僕もまだ良い策定方法がわかっていません。(というより、これまでこんなに失敗することがなかった)。一応、responseのメタデータの中に、RetryAttemptsって項目があって、何回目の再実行で成功したかとか見れるのですが、ほとんど成功するので大体これ0なんですよね。ここを監視してたら参考指標になるのかもしれませんが、これを監視できる実装にするのは面倒です。基本的には成功するものですから。

もし、上に列挙したエラーのどれかが発生してboto3の実行が止まってしまうよ、という状況が発生したら、リトライ回数の設定変更を検討に入れてみてください。

ただ、リトライというのはあくまでも同じ処理を繰り返すだけなので、百発百中で失敗する様な処理をリトライしても無意味です。ほとんど確実に成功するのに超低確率で失敗する事象が起きてしまう場合に、その失敗確率をもっと下げるという用途でのみ有効です。

Louvain法のmodularityの式について検証してみる

前回に引き続き、Louvain法の話です。前回はPythonで使う方法を紹介しましたが、今回はこの手法が利用しているモジュラリティ(modularity)という指標について検証していきます。そもそもなぜこの式でグラフのノードのクラスタを評価できるのかってのを確認しましょう。
(ちなみに、モジュラリティを最大化するアルゴリズムについては今回の記事では取り上げません。あくまでもモジュラリティの定義について見ていきます。)

前回の記事でも書いてるのですが今回の記事では主役なので定義式を再掲します。

$$Q=\frac{1}{2m}\sum_{ij}\left[A_{ij}\ -\ \frac{k_ik_j}{2m} \right]\delta(c_i, c_j).$$

$A_{ij}$: ノード$i$, ノード$j$間のエッジの重み。
$k_i, k_j$: ノード$i$, ノード$j$それぞれに接続されたエッジの重みの合計。
$m$: グラフの全てのエッジの重みの合計。
$c_i, c_j$: ノード$i$, ノード$j$それぞれが所属しているコミュニティ。
$\delta$: クロネッカーのデルタ。二つの引数が等しければ1でそれ以外は0を返す関数。

一見しただけだと、これでクラスタを評価できるってわかりにくいですよね。

順番に見ていきましょう。

まず、$\delta(c_i, c_j)$の部分です。これ、$c_i$と$c_j$が別のクラスタだったら値が$0$なので、$\sum$のそれらの項は消滅し、さらに同じクラスタだったら値が$1$になるので、それらの項はそのままの値で残ります。

なのでこの式はこうなります。

$$Q=\frac{1}{2m}\sum_{c_iとc_jが同じクラスタとなるi,j}\left[A_{ij}\ -\ \frac{k_ik_j}{2m} \right].$$

そして、$A_{ij}$はエッジの重みなので、和の中身の前半部分だけに着目すると、次の様に言えます。

$$\frac{1}{2m}\sum_{c_iとc_jが同じクラスタとなるi,j}A_{ij} = \frac{1}{2m}\{クラスタ内部のエッジの重みの総和\}$$

$m$はクラスタに関係なく、グラフによって定まってる定数なので無視して良いですから、この部分は確かに同じクラスタの中にたくさんエッジがあると値が大きくなり評価指標としての意味を持っている様に見えます。

ただし、この項しかないと、全部のノードを1個のクラスタとした場合に$Q$が最大となっしまうので、それに対応する必要があります。そこで登場するのが残りの項です。

もし、ノード$i$とノード$j$の間にエッジがなかったりあってもウェイトが非常に小さかったりすると、負の数を足すことによってモジュラリティが下がる様になってるのです。たとえば、エッジがない場合、$A_{ij}$が0ですから、$A_{ij}\ -\ \frac{k_ik_j}{2m} = -\frac{k_ik_j}{2m} $です。

そして、$k_i$、$k_j$はそれぞれのノードに接続されたエッジの重みの和ですから、重みの大きなエッジをたくさん持っているノード間についてはこの値の絶対値は大きくなります。

つまり、エッジがたくさんあるノード同士にも関わらずそれらのノード間にエッジがなかったら強くペナルティを課すよ、ってのがこの項の意味なのです。
分母の$2m$とかにもちゃんと意味があるのですが細かい話になるので割愛します。これらの細かい工夫により、-0.5から1の値を取る指標となっています。計算量とのバランスもとりながら非常によく考えられた指標だと思いました。

最後に、前回の記事でも少しだけ触れているのですが、この$\sum_{ij}$が取る和の$ij$部分について確認したのでその話書いておきます。

これは$i$と$j$がそれぞれ独立に全てのパターンを取ります。$(i, j)$と$(j, i)$は個別に足されるし、$(i, i)$も考慮されるってことですね。ライブラリの結果と、自分で実装した結果を並べてみて確認しました。とりあえず復習兼ねて前回のサンプルのグラフでやります。まずライブラリ利用。

import numpy as np
import networkx as nx
import community

# 前回の記事で生成したのと同じグラフを再現する。
# 30個のノード
node_list = list(range(30))

# 同じクラスタ内は0.5, 別クラスタは0.02の確率でエッジを生成する
edge_list = []
np.random.seed(1)  # シード固定
for i in range(30):
    for j in range(i+1, 30):
        if i // 10 == j // 10:
            if np.random.rand() < 0.5:
                edge_list.append((i, j))
        else:
            if np.random.rand() < 0.02:
                edge_list.append((i, j))

# グラフ生成
G = nx.Graph()
G.add_nodes_from(node_list)
G.add_edges_from(edge_list)

# コミュニティーの検出
partition = community.best_partition(G)

# ライブラリによるモジュラリティの計算
print(community.modularity(partition, G))
# 0.5316751700680272

同じ値を自分で計算して出してみましょう。エッジのウェイトが全部1なので、$m$はただのエッジの本数になるし、$k_i$, $k_j$はそれぞれのノードの次数に等しいことに注意してください。

Q = 0
m = G.size()
# iと jは全部の組み合わせをとる。
for i in range(len(G)):
    for j in range(len(G)):
        if partition[i] == partition[j]:
            # ノードi と ノードjの間にエッジがあったら1(ウェイト)を足す
            if (i, j) in G.edges:
                Q += 1 
            
            # Σの後半部分
            Q -= G.degree(i) * G.degree(j) / (2*m)
# 全体を2mで割る
Q/=2*m
print(Q)
# 0.5316751700680273

一致してますね。

ここから余談ですが、Qは -0.5から 1の値を取りますが、どんなグラフについてもQが-0.5を取ったり1を取ったりする分割が存在するわけではなさそうです。上記のサンプルでも0.53程度に留まっていますしね。

-0.5 の方は、完全2部グラフを構築して、さらにその2部をそれぞれコミュニティとすることで、同じクラスタ内部には1本もエッジがないのに別クラスタのノード間には必ずエッジが存在するという状態にすると発生します。

また、逆にQが大きくなるのは非連結な多数の完全グラフからなるグラフをそれぞれの完全グラフを一つのクラスタとすると、非連結なグラフが増えるにつれて1に近づいていくのを確認できています。ただ、近づくだけで1になることはないんじゃないでしょうか。(証明はできていませんが。)

便利なのでなんとなく使っていた手法でしたが、改めて指標を検証し複数のグラフや分割について自分で計算してみたことで理解を深めることができました。

モジュラリティは定義中に記号多いですしぱっと見難しそうに見えるのですが、落ち着いて見れば四則演算だけで構成されている非常にシンプルな指標なので、興味があれば皆さんもいろいろ試して遊んでみてください。

Louvain法を用いてグラフのコミュニティーを検出する

昔、グラフ関係の記事を書いてた頃に書いてたと思ってたら、まだ書いてなかったので、グラフのコミュニティーを検出する方法を書きます。

グラフのコミュニティ検出というのは、グラフのノードを複数のグループに分け、同じグループ内のノード同士ではエッジが密で、異なるグループのノード間ではエッジが疎になる様にすることを言います。人の集まりを仲良しグループでいくつかの組に分けるのに似ていますね。この記事の最後にも結果の例の画像をつけていますのでそれをみていただくとイメージがつきやすいと思います。

複数のアルゴリズムが存在するのですが、僕が一番気に入っていてよく使っているのはLouvain法による方法なのでこれを紹介していきます。
参考: Louvain method – Wikipedia

このLouvain法では、グラフとコミュニティに対してモジュラリティ(modularity)という指標を定義し、このモジュラリティが最大になる様なコミュニティの分け方を探すことでグラフのノードを分割します。モジュラリティの定義式はWikipediaにもありますが以下の通りです。

$$Q=\frac{1}{2m}\sum_{ij}\left[A_{ij}\ -\ \frac{k_ik_j}{2m} \right]\delta(c_i, c_j).$$

ここで、各記号の意味は以下の通りです。
$A_{ij}$: ノード$i$, ノード$j$間のエッジの重み。
$k_i, k_j$: ノード$i$, ノード$j$それぞれに接続されたエッジの重みの合計。
$m$: グラフの全てのエッジの重みの合計。
$c_i, c_j$: ノード$i$, ノード$j$それぞれが所属しているコミュニティ。
$\delta$: クロネッカーのデルタ。二つの引数が等しければ1でそれ以外は0を返す関数。

元論文を読めていないのですが、ライブラリの挙動を確認した結果によると、和の$ij$は$i$, $j$の全てのペアを渡る和で$i,j$の組み合わせとそれを入れ替えた$j, i$の組み合わせを両方取り、さらに$i=j$となる自己ループなエッジも考慮している様です。

さて、実際にやってみましょう。Pythonでは、communityというライブラリがあり、Louvain法が実装されています。(networkx自体にも他のコミュニティ検出のアルゴリズムが実装されているのですが、Louvain法は別ライブラリになっています。)

pipyでの登録名がpython-louvainなので、インストール時のコマンドが以下であることに注意してください。Pythonでimportするときと名前が違います。

$ pip install python-louvain

ドキュメントはこちら
参考: Community detection for NetworkX’s documentation — Community detection for NetworkX 2 documentation

早速使ってみましょう。例としてコミュニティがわかりやすいグラフを乱数で生成します。
0~29のノードを持ち、0~9, 10~19, 20〜29が同じコミュニティで、同コミュニティー内では50%の確率でエッジが貼られていて異なるミュにティだと2%の確率でしかエッジがないという設置です。

import numpy as np
import networkx as nx
import community

# 30個のノード
node_list = list(range(30))

# 同じクラスタ内は0.5, 別クラスタは0.02の確率でエッジを生成する
edge_list = []
np.random.seed(1)  # シード固定
for i in range(30):
    for j in range(i+1, 30):
        if i // 10 == j // 10:
            if np.random.rand() < 0.5:
                edge_list.append((i, j))
        else:
            if np.random.rand() < 0.02:
                edge_list.append((i, j))

# グラフ生成
G = nx.Graph()
G.add_nodes_from(node_list)
G.add_edges_from(edge_list)

さて、グラフができたのでコミュニティ検出をやっていきます。これものすごく簡単で、best_partitionというメソッドにnetworkxのグラフオブジェクトを渡すとノードと所属するグループの辞書として結果が帰ってきます。

注意点ですが、アルゴリズムの高速化のための工夫の弊害により絶対にベストな分割が得られるというわけではなく少々確率的な振る舞いをします。今回のサンプルコードでは非常にわかりやすい例を用意したので一発で成功していますが、通常の利用では何度か試して結果を比較してみるのが良いでしょう。

ではやっていきます。

partition = community.best_partition(G)

print(partition)
"""
{0: 2, 1: 2, 2: 2, 3: 2, 4: 2, 5: 2, 6: 2, 7: 2, 8: 2, 9: 2,
10: 0, 11: 0, 12: 0, 13: 0, 14: 0, 15: 0, 16: 0, 17: 0, 18: 0, 19: 0,
20: 1, 21: 1, 22: 1, 23: 1, 24: 1, 25: 1, 26: 1, 27: 1, 28: 1, 29: 1}
"""

0〜9は2で、10〜19は0で、20〜29は1と、綺麗に分類できてますね。

画像に表示してみましょう。ドキュメントのサンプルコードだとlist(partition.values())と辞書の値をそのまま可視化メソッドに渡していますが、辞書型のデータなので順序は保証されておらずこれはリスキーな気がします。実はそれでもうまくいくのですが念のため、ノードと順番が揃う様に配列として取り出してそれを使う様にしました。

# ノードの並びと同じ順でグループのリストを配列として取得する。
partition_list = [partition[n] for n in G.nodes]
# 配置決め
pos = nx.spring_layout(G, seed=3)
# 描写
nx.draw_networkx(G, node_color=partition_list, cmap="tab10", pos=pos)

結果がこちらです。

綺麗に3グループに分かれていますね。

ちなみに、このグループ分けのモジュラリティを取得するメソッドもあります。グループ分けとグラフ本体のデータをそれぞれ渡すことで使えます。

print(community.modularity(partition, G))
# 0.5316751700680272

結構高いですね。

自分のChromeブラウザの履歴をPythonで集計する方法

以前、SQLite3の使い方を紹介しましたが、その応用です。

自分がどんなサイトにアクセスしているかを自分で集計したくなったのでその方法を調べました。(firefoxなどもそうらしいのですが) Chromeはアクセス履歴をSQLiteのDB(ファイル)に保存しています。そのため、該当ファイルに接続すればSQLで集計ができます。

Mac の場合、以下のパスに履歴ファイルがあります。拡張子がないので紛らわしいですが、Historyがファイル名で、Defaultまでがディレクトリパスです。
/Users/{ユーザー名}/Library/Application\ Support/Google/Chrome/Default/History

人によってはDefault 以外のディレクトリにあることもあるそうなので、もしそれらしいファイルがなかったら付近のディレクトリの中を探してみてください。例えば、Default/History ではなく、System Profile/Historyや、Guest Profile/Historyに存在する環境も見たことがあります。

ロックがかかったりするとよくないので、このファイルに直接繋ぐのではなく、どこかにコピーを取って、その中を確認しましょう。

# カレントディレクトリにコピーする
$ cp /Users/{ユーザー名}/Library/Application\ Support/Google/Chrome/Default/History .

早速接続してみましょう。とりあえずどんなテーブルがあるのか見てみます。

import sqlite3
import pandas as pd


# データベースに接続
con = sqlite3.connect('History')
con.row_factory = sqlite3.Row
cursor = con.cursor()

cursor.execute("SELECT name FROM sqlite_master")
pd.DataFrame(
    cursor.fetchall(),
    columns=[row[0] for row in cursor.description],
)

"""
name
0	meta
1	sqlite_autoindex_meta_1
2	urls
3	sqlite_sequence
4	visits
5	visit_source
6	visits_url_index
7	visits_from_index
8	visits_time_index
9	visits_originator_id_index
10	keyword_search_terms
11	keyword_search_terms_index1
12	keyword_search_terms_index2
13	keyword_search_terms_index3
14	downloads
15	downloads_url_chains
# 多いので以下略。自分の環境では合計35テーブルありました。
"""

目当てのアクセス履歴が保存されているテーブルは、visitsです。ただ、このテーブルはアクセスしたURLのパスそのものは含んでおらず、URLはurlsテーブルに入っていて、visitsテーブルにはurlsテーブルのidが入ってます。

urlsテーブルから1行だけ取り出してみます。(転置してprintしました。)

cursor.execute("""
    SELECT
        *
    FROM
        urls
    WHERE
        url = 'https://analytics-note.xyz/'
""")

print(pd.DataFrame(
    cursor.fetchall(),
    columns=[row[0] for row in cursor.description],
).T)
"""
                                           0
id                                        53
url              https://analytics-note.xyz/
title                                  分析ノート
visit_count                               43
typed_count                                0
last_visit_time            13327924159724978
hidden                                     0
"""

idが53らしいですね。URLとページタイトルが入っています。あと、visit_countとして訪問回数入ってます。(思ったより少ないです。他のPCやスマホで見てることが多いからでしょうか。)

次は、visitsテーブルも見てみます。urlという列にさっきのurlテーブルのidが入ってる(列名をurls_idとかにして欲しかった)ので、それで絞り込んでいます。

cursor.execute("""
    SELECT
        *
    FROM
        visits
    WHERE
        url = 53
    ORDER BY
        visit_time DESC
    LIMIT
        1
""")

print(pd.DataFrame(
    cursor.fetchall(),
    columns=[row[0] for row in cursor.description],
).T)
"""
                                                 0
id                                            8774
url                                             53
visit_time                       13327924159724978
from_visit                                       0
transition                               805306370
segment_id                                       9
visit_duration                                   0
incremented_omnibox_typed_score                  0
opener_visit                                     0
originator_cache_guid                             
originator_visit_id                              0
originator_from_visit                            0
originator_opener_visit                          0
is_known_to_sync                                 0
"""

visit_timeでアクセスした時刻がわかる様ですね。

ちなみに、UNIXタイムスタンプ等に比べて非常にでっかい数字が入っていますが、これはChromeの仕様で、1601年1月1日(UTC)からの経過時間をマイクロ秒(μ秒、100万分の1秒)単位で計測したものだそうです。(この情報は他サイトで入手したのですが、できればChromeの公式ドキュメントか何かで確認したいところです。概ね正しそうなのですが、自分の環境だと数分ずれていて違和感あります。)

さて、テーブルの構造がわかったので具体的なSQL構築していきます。たとえば、直近1000アクセス分の情報を見る場合は以下の様なクエリで良いのではないでしょうか。

sql = """
    SELECT
        visits.visit_time,
        urls.url,
        urls.title
    FROM
        visits
    LEFT OUTER JOIN
        urls
    ON
        visits.url = urls.id
    ORDER BY
        visits.visit_time DESC
    LIMIT 1000;
"""

cursor.execute(sql)

df = pd.DataFrame(
    cursor.fetchall(),
    columns=[row[0] for row in cursor.description],
)

仕上げに、visit_time がこのままだと使いにくすぎるので、これを人が読めるように変換します。手順は以下の通りです。

  1. 10^6で割ってマイクロ秒を秒に変換する。
  2. 1601年1月1日のタイムスタンプを足して現在時刻のタイムスタンプに変換する。
  3. 時刻型に戻す。
  4. 9時間足して日本時間にする。

順番にやっていくとこんなコードになります。

from datetime import datetime
from datetime import timedelta


df["アクセス時刻"] = df["visit_time"].apply(
    lambda x: x/10**6
).apply(
    lambda x: x+datetime(1601, 1, 1).timestamp()
).apply(
    datetime.fromtimestamp
).apply(
    lambda x: x+timedelta(hours=9)
)

もっといいやり方はいくらでもある気がする(1601/1/1のタイムスタンプとか9時間分の秒数とか事前に定数として持っておいてマイクロを取った時に足してしまうなど)のですが、可読性重視でいけばこの書き方で良いと思います。

dfコマンドとduコマンドを用いたディスク容量使用状況の調査方法

はじめに

何度か必要になったことがあるのですがその度に使い方を忘れてしまっているdfコマンドとduコマンドの使い方のメモです。

それぞれ一言で言ってしまえば次の様になります。
– dfコマンドはディスクの使用済み容量と空き容量を確認できるコマンドです。
– duコマンドは指定したディレクトリの使用容量を調べるコマンドです。

どっちがどっちだったかよく忘れるのでメモっておこうという記事の目的がこれで終わってしまったのですが、あんまりなのでもう少し続けます。

このブログが動いてるサーバーもそうですし、投資や各種技術検証で使っているEC2インスタンスなども費用が自己負担なのでディスクサイズが最低限の状態で動いており、時々ディスクの空き具合を気にしてあげる必要があります。

また、職場で利用しているデータ分析関係のワークフローを構築しているサーバーも本番サービスのサーバーと違って最低限構成なので何度かディスクのパンクによる障害を経験しています。

そんな時に調査するのにdfコマンドとdfコマンドを使います。

dfコマンドについて

まず、dfコマンドからです。

先ほどざっと書いた通りで、このコマンドを打つとマウントされているディスクの一覧が表示されてそれぞれの容量と使用量、そして丁寧なことに使用率なども表示してくれます。

ディスク容量のパンクが疑われる場合、まずこのコマンドで状況を確認にすることになると思います。(本来は、パンクが疑われなくても時々監視しておくべきなのだと思いますが。)

自分が持ってるとあるEC2インスタンスで実行した結果が以下です。(EC2で実行したらヘッダーが日本語になりました。)

$ df
ファイルシス   1K-ブロック    使用  使用可 使用% マウント位置
devtmpfs            485340       0  485340    0% /dev
tmpfs               494336       0  494336    0% /dev/shm
tmpfs               494336     352  493984    1% /run
tmpfs               494336       0  494336    0% /sys/fs/cgroup
/dev/xvda1         8376300 3768544 4607756   45% /

-h オプション(“Human-readable”の頭文字らしい)をつけると、人間が読みやすいキロメガギガ単位で表示してくれます。

$ df -H
ファイルシス   サイズ  使用  残り 使用% マウント位置
devtmpfs         497M     0  497M    0% /dev
tmpfs            507M     0  507M    0% /dev/shm
tmpfs            507M  361k  506M    1% /run
tmpfs            507M     0  507M    0% /sys/fs/cgroup
/dev/xvda1       8.6G  3.9G  4.8G   45% /

一番下の /dev/xvda1がメインのファイルシステムですね。現時点で45%使っていることがわかります。まだ余裕。

Mac (zsh)だと出力の形がちょっと違います。

% df -h
Filesystem       Size   Used  Avail Capacity iused      ifree %iused  Mounted on
/dev/disk3s1s1  228Gi  8.5Gi  192Gi     5%  356093 2018439760    0%   /
devfs           199Ki  199Ki    0Bi   100%     690          0  100%   /dev
/dev/disk3s6    228Gi   20Ki  192Gi     1%       0 2018439760    0%   /System/Volumes/VM
/dev/disk3s2    228Gi  4.4Gi  192Gi     3%     844 2018439760    0%   /System/Volumes/Preboot
/dev/disk3s4    228Gi  4.8Mi  192Gi     1%      43 2018439760    0%   /System/Volumes/Update
/dev/disk1s2    500Mi  6.0Mi  482Mi     2%       1    4936000    0%   /System/Volumes/xarts
/dev/disk1s1    500Mi  6.2Mi  482Mi     2%      30    4936000    0%   /System/Volumes/iSCPreboot
/dev/disk1s3    500Mi  1.0Mi  482Mi     1%      62    4936000    0%   /System/Volumes/Hardware
/dev/disk3s5    228Gi   22Gi  192Gi    11%  460541 2018439760    0%   /System/Volumes/Data
map auto_home     0Bi    0Bi    0Bi   100%       0          0  100%   /System/Volumes/Data/home

iノードの利用状況も一緒に出してくれている様です。この端末にはそんな大きなファイルを溜めてないのでスカスカですね。

duコマンドについて

この記事用にdfコマンドを試したサーバーや端末は問題なかったのですが、もしディスク容量が逼迫していることが判明したら原因を探すことになります。

その時は当然ディレクトリごとの使用量を調べて、原因となっているディレクトリを探すのですが、その際に使うのがduコマンドです。

du -sh {対象ディレクトリ}
の構文で使うことが多いです。-hオプションはdfの時と同じで人間が読みやすい単位で表示するものです。-sは対象のディレクトリの合計だけを表示するものです。これを指定しないと配下のディレクトリを再起的に全部表示するので場所によっては出力が膨大になります。

対象ディレクトリを省略するとカレントディレクトリが対象になります。

$ du -sh
1.2G    .

上記の様に、1.2G使ってるって結果が得られます。もう少しだけ内訳みたい、て場合は、-sの代わりに、-d {深さ} オプションで何層下まで表示するかを決められます。 -s は -d 0 と同じです。

$ du -h -d 0
1.2G    .
$ du -h -d 1
4.0K    ./.ssh
973M    ./.pyenv
180M    ./.cache
3.0M    ./.local
9.0M    ./.ipython
104K    ./.jupyter
0       ./.config
1.2G    .

あとは、 -d 1 はオプションでやらず、ディレクトリの指定を ./* みたいにワイルドカードで指定しても似たことができますね。(-s は付けてください)

注意ですが、 ルートディレクトリなどで(/) でこのコマンド打つとかなり重い処理になる可能性があります。注意して実行しましょう。

現実的には、いろんなメディアファイルが配置されそうなパスとかログファイルが出力されている先とかそういうパスに当たりをつけてその範囲で調査するコマンドと考えた方が安全です。

os.walk()メソッドでファイル走査

最近、ちょっとしたツールや簡易的なスクリプトの構築でChatGPTを使っているのですが、ファイルの整理系のPythonスクリプトを書いてもらうときにしばしos.walkというメソッドを使っていて、それを自分が知らなかったのでドキュメントを読んでみたのでそのまとめです。

基本的には、あるディレクトリ配下のファイルやディレクトリを一通りリストアップするメソッドになります。
参考: os — 雑多なオペレーティングシステムインターフェース — Python 3.11.3 ドキュメント

同様の目的だと僕はこれまでglobを使ってました。以下の記事にしています。
参考: globでサブフォルダを含めて再帰的にファイルを探索する

まだ、自分で書くときはglob使ってますがChatGPTの出力を見てるとwalkの方が使い勝手が良さそうな気がしています。

使い方試すためにこんな感じて適当にファイルやディレクトリ(いくつかは空ディレクトリ)を用意しました。

$ find targetdir | sort
targetdir
targetdir/001.txt
targetdir/002.txt
targetdir/subdir1
targetdir/subdir1/101.txt
targetdir/subdir1/102.txt
targetdir/subdir1/subdir1_1
targetdir/subdir1/subdir1_1/111.txt
targetdir/subdir1/subdir1_1/subdir1_1_1
targetdir/subdir2
targetdir/subdir2/201.txt
targetdir/subdir3

使いたですが、上にリンクを貼ったドキュメントの様に、走査したいトップディレクトリを指定して渡してあげると、その配下の各ディレクトリごとに、(dirpath, dirnames, filenames) というタプルを返すジェネレーターを作ってくれます。

dirpath に最初に指定したディレクトリを含む配下のディレクトリたちが順に格納され、difnamesにその直下のディレクトリのディレクトリ名のリストが入り、filenamesにdirpathのディレクトリ直下のファイル名がリストで入ります。

ジェネレーターなのでfor文で回して結果をprintしてみます。

for dirpath, dirnames, filenames in os.walk("targetdir"):
    print("dirpath:", dirpath)
    print("dirnames:", dirnames)
    print("filenames:", filenames)
    print("- "*10)

"""
dirpath: targetdir
dirnames: ['subdir3', 'subdir2', 'subdir1']
filenames: ['002.txt', '001.txt']
- - - - - - - - - -
dirpath: targetdir/subdir3
dirnames: []
filenames: []
- - - - - - - - - -
dirpath: targetdir/subdir2
dirnames: []
filenames: ['201.txt']
- - - - - - - - - -
dirpath: targetdir/subdir1
dirnames: ['subdir1_1']
filenames: ['101.txt', '102.txt']
- - - - - - - - - -
dirpath: targetdir/subdir1/subdir1_1
dirnames: ['subdir1_1_1']
filenames: ['111.txt']
- - - - - - - - - -
dirpath: targetdir/subdir1/subdir1_1/subdir1_1_1
dirnames: []
filenames: []
- - - - - - - - - -
"""

簡単ですね。

globでファイルを探索した時は最初からファイル名ではなくファイルパスの形で得られていましたが、os.walkではディレクトリパスとファイル名が個別に得られるのでファイルパスが欲しい場合は連結する必要があります。せっかくosライブラリをimportしてるので専用メソッド使ってやりましょう。次がos.walkを使って配下のファイル一覧を得る方法です。

for dirpath, dirnames, filenames in os.walk("targetdir"):
    for filename in filenames:
        print(os.path.join(dirpath, filename))
"""
targetdir/002.txt
targetdir/001.txt
targetdir/subdir2/201.txt
targetdir/subdir1/101.txt
targetdir/subdir1/102.txt
targetdir/subdir1/subdir1_1/111.txt
"""

簡単ですね。ディレクトリでも同じ様にできると思いますし、連結してしまえばただの文字列なので、拡張子(末尾の文字列)で絞り込むと言った応用も簡単にできると思います。

J-Quants API の基本的な使い方

J-Quants APIの紹介

個人的に株式投資を行なっており、その他にも時系列データ分析の学習などの用途でも株価データを頻繁に利用しています。元々、それらのデータをどうやって用意するかが課題で証券会社のページ等々から引っ張ってきていたのですが、この度、日本取引所グループが公式にJ-Quants API という非常に便利なAPIを提供してくれるようになりました。

参考: J-Quants

ベータ版が登場した時から即座に登録して使っていたのですが、この度正式にリリースされましたので使い方とか紹介していこうと思います。

会員登録しないと使えないので、興味のある人は上記のリンクから登録してください。プラン表にある通り、Freeプランもあるので使い勝手を見るには無料で試せます。(Freeプランは12週間遅延のデータなので運用には使いにくいですが、時系列データ分析のサンプルデータ取得等の用途では十分活用できます。)

ちなみに僕はライトプランを使っているでこの記事ではそれを前提とします。(将来的にスタンダード以上のプランに上げるかも。)

肝心の取得できる情報なのですが、東証各市場の上場銘柄の一覧や、株価の四本値、財務情報や投資部門別情報などの情報を取得することができます。物によっては他の手段での収集がかなり手間なものもあり、とても貴重な情報源になると思います。

基本的な使い方

APIは認証周り以外はかなりシンプルな設計になっていて、API仕様にそって、認証情報とパラメーターを指定して規定のURLを叩けばJSONで結果を得られます。
参考: API仕様 J-Quants API

認証だけちょっと珍しい形式をとっていて、以下の手順を踏みます。

  1. メールアドレスとパスワードを用いてWeb画面かAPIでリフレッシュトークンを取得する。
  2. リフレッシュトークンを用いて専用APIでIDトークンを取得する。
  3. IDトークンを用いて、各種情報を取得するAPIを利用する。

ベータ版の初期はリフレッシュトークンの取得がWebでログインしてコピーする以外になかったのでそこもAPI化されたことで大変便利になりました。

なお、リフレッシュトークンの有効期限は1週間、IDトークンの有効期限は24時間です。

Pythonでトークンの取得

ではここからやっていきましょう。実は専用クライアントライブラリ等も開発が進んでるのですが、requests でも十分使っていけるのでrequestsで説明します。

まず、リフレッシュトークンの取得です。
参考: リフレッシュトークン取得(/token/auth_user) – J-Quants API

指定のURLにメールアドレスとパスワードをPOSTし、戻ってきたJSONから取り出します。

import json
import requests
import pandas as pd


# POSTするデータを作る。
email = "{メールアドレス}"
password = "{パスワード}"
account_data = json.dumps({
        "mailaddress": email,
        "password": password,
    })

auth_user_url = "https://api.jquants.com/v1/token/auth_user"

auth_result = requests.post(auth_user_url, data=account_data)
refresh_token = auth_result.json()["refreshToken"]

これでリフレッシュトークンが取得できたので、次は IDトークンを取得します。
参考: IDトークン取得(/token/auth_refresh) – J-Quants API

これはクエリパラメーターでURL内にリフレッシュトークンを指定して使います。(ここは少し設計の意図が不明ですね。何もPOSTしてないのでGETが自然な気がしますが、GETだとエラーになります。)

auth_refresh_url=f"https://api.jquants.com/v1/token/auth_refresh?refreshtoken={refresh_token}"
refresh_result = requests.post(auth_refresh_url)
id_token = refresh_result.json()["idToken"]

これでIDトークンも取得できました。

ちなみに両トークンとも1000文字以上の結構長い文字列です。

各種情報の取得

IDトークンが手に入ったらいよいよこれを使って必要な情報を取得します。APIの仕様書はかなりわかりやすいので、あまり迷わずに使えると思います。
たとえば、株価4本値が必要だとしましょう。

参考: 株価四本値(/prices/daily_quotes) – J-Quants API

仕様書にある通り、証券コード(code) や日付(dateもしくはfrom/to)をURLで指定し、ヘッダーにさっきの認証情報をつけて、GETメソッドでAPIを叩きます。

たとえば、トヨタ自動車(7203)の昨年1年間の株価を取得してみましょう。必須ではありませんが、得られたデータはDataFrameに変換しやすい形で帰ってくるのでDataFrameにしました。

import pandas as pd


code = "7203" # 4桁のコードでも5桁のコード72030でもよい。
from_ = "2022-01-01"
to_ = "2022-12-31"

daily_quotes_url = f"https://api.jquants.com/v1/prices/daily_quotes?code={code}&from={from_}&to={to_}"
# idトークンはヘッダーにセットする
headers = {"Authorization": f"Bearer {id_token}"}
daily_quotes_result = requests.get(daily_quotes_url, headers=headers)
# DataFrameにすると使いやすい
daily_quotes_df = pd.DataFrame(daily_quotes_result.json()["daily_quotes"])

取れたデータを表示してみましょう。ブログレイアウトの都合ですが3レコードほどを縦横転置してprintしたのが以下です。

print(daily_quotes_df.head(3).T)
"""
                              0               1              2
Date                 2022-01-04      2022-01-05     2022-01-06
Code                      72030           72030          72030
Open                     2158.0          2330.0         2284.5
High                     2237.5          2341.0         2327.0
Low                      2154.5          2259.0         2277.0
Close                    2234.5          2292.0         2284.5
Volume               43072600.0      53536300.0     37579600.0
TurnoverValue     94795110250.0  123423876550.0  86533079050.0
AdjustmentFactor            1.0             1.0            1.0
AdjustmentOpen           2158.0          2330.0         2284.5
AdjustmentHigh           2237.5          2341.0         2327.0
AdjustmentLow            2154.5          2259.0         2277.0
AdjustmentClose          2234.5          2292.0         2284.5
AdjustmentVolume     43072600.0      53536300.0     37579600.0
"""

バッチリデータ取れてますね。

もう一つ、例として財務情報も取ってみましょう。

参考: 財務情報(/fins/statements) – J-Quants API

取れる情報の種類がめっちゃ多いので結果は一部絞ってprintしてます。

code = "7203"

statements_url = f"https://api.jquants.com/v1/fins/statements?code={code}"

headers = {"Authorization": f"Bearer {id_token}"}
statements_result = requests.get(statements_url, headers=headers)
statements_df = pd.DataFrame(statements_result.json()["statements"])

# 100列以上取れるので、1レコードを20列だけprint
print(statements_df.head(1).T.head(20))
"""
DisclosedDate                                          2018-05-09
DisclosedTime                                            13:25:00
LocalCode                                                   72030
DisclosureNumber                                   20180312488206
TypeOfDocument              FYFinancialStatements_Consolidated_US
TypeOfCurrentPeriod                                            FY
CurrentPeriodStartDate                                 2017-04-01
CurrentPeriodEndDate                                   2018-03-31
CurrentFiscalYearStartDate                             2017-04-01
CurrentFiscalYearEndDate                               2018-03-31
NextFiscalYearStartDate                                2018-04-01
NextFiscalYearEndDate                                  2019-03-31
NetSales                                           29379510000000
OperatingProfit                                     2399862000000
OrdinaryProfit                                                   
Profit                                              2493983000000
EarningsPerShare                                            842.0
DilutedEarningsPerShare                                    832.78
TotalAssets                                        50308249000000
Equity                                             19922076000000
"""

以上が、J-Quants APIの基本的な使い方になります。

クライアントライブラリについての紹介

ここまで、requestsライブラリを使ってAPIを直接叩いてましたが、実はもっと手軽に使える様にするための専用ライブラリの開発も進んでおりすでにリリースされています。

参考: jquants-api-client · PyPI

ライブラリも有志の方々が頑張って開発していただけていますが、なにせAPI本体もまだリリースされたばかりですし、まだバグフィックスなどもされている最中の様なので今時点では僕はこちらのライブラリは使っていません。

API本体のアップデートとかも今後多そうですしね。

ただ、将来的にはこちらのライブラリが非常に便利な物になると期待しているので、API/ライブラリ双方の開発が落ち着いてきたら移行したいなと思ってます。