PythonでSubscan APIを使用してステーキング報酬履歴を取得してクリプタクトのカスタムファイル(csv)を作成する #Python

背景

以前こちらの記事でGASでSubscan(DOT)のステーキング履歴(csv)からクリプタクトのカスタムファイルを作成する方法を紹介しましたが、以下の点で使いづらさがあったのでSubscan APIとPythonを使用して取得する方法を検討しました。

csvファイルのDLが必要
GASの処理に時間がかかる

作成したもの

PythonのRequestsモジュールとSubscan APIを使用してステーキング報酬履歴を取得し、クリプタクトのカスタムファイルフォーマットのcsvファイルをローカルに保存するソースコードを作成しました。

前提条件

Subscan APIのAPI Keysを取得済みであること
Pandas,Requestsモジュールのインストール

仕様

取得するデータはreward-slashを対象する
取得件数はコマンドラインから入力する
取得するデータはPythonのRequestsモジュールを使用して、Subscan APIドキュメントに従い設定したAPI Endpoint情報でHTTPのPOSTメソッドで送信し、Responseオブジェクトのstatus_codeがHTTP Status Codes: 200の場合に受信したデータをJSONオブジェクトとして保存する
HTTP Status Codes: 200以外の値の場合はstatus_codeを表示し、プログラムを終了する
取得したデータはカスタムファイルのフォーマット仕様で整形し、指定したパスにcsvファイルとして保存する
ファイルの書き出し方法として新規作成と既存ファイルに追記する2つのケースに対応し、それぞれソースコードを作成する

注意事項

Subscanの仕様やクリプタクトのデータフォーマットは変わることがありますので、利用する際は自己責任でお願いします
データのチェックとして重複は考慮していますが、不足しているデータのチェックまではしていません
Subscan APIで動作確認したNetworkはPolkadotのみ確認しています
特に作成したcsvファイルのデータについては目的のデータを作成できていること、
トランザクションデータを参照して、差異がないことや誤りがないことも確認してください

ソースコード

使い方はREADME.mdを参照してください。
以下では処理の詳細、利用する上での注意点を記載します。

Subscan APIの仕様について

reward-slashから取得する件数はrowとpageで指定できます
Subscan APIを使用して受信データを確認したところ、取得件数はrow,データの開始要素数（listオブジェクト）はrow * pageで決まるようです。¹（正直、Pageという名前が分かりづらいです）。
今回作成するソースコードではステーキング報酬量を最新のデータ(listオブジェクトの要素数:0)から取得するためrowは可変とし、pageは0固定とします（下表参照）
[2022/06/05追記]
Subscan APIドキュメントに記載はありませんが、動作確認したところ1pageで取得できる件数は100件までとなっていました。本コードでは101件以上の値の入力はサポートしておらず、入力する400 Bad Requestが返るためquit()関数でプログラムは終了します。
101件以上の履歴を取得したい場合は分割処理に対応したこちらの記事を参考にしてください。

表：row,pageに対する取得件数とデータの開始要素数の関係

row	page	データの開始要素	取得件数
30	0	0	30
10	1	10	10
10	2	20	10
21	1	21	21

API情報について

RequestsのPOSTメソッド送信するAPI情報はsettings.pyで定義する定数を指定します

getSubscanStakingRewardsInputNewCreateCsvData.py

# Subscan APIを使用しrequestsモジュールでPOSTリクエストする
# header情報
headers_dict = {
      'Content-Type': 'application/json',
      'X-API-Key': API_KEY,
}
# data-raw情報
# rowの指定のみでlistの情報は取得できるためpageは0固定とする
data_dict = { 'row': 0, 'page': 0, 'address': ADDRESS }
      
# 件数で上書き
data_dict['row'] = input_num
# Staking API / rewards-slash指定
response = requests.post(API_HOST+REQUEST_URL, headers=headers_dict, data=json.dumps(dat
# HTTPステータスコード確認
if response.status_code == 200:
      print(' -- HTTP Status Codes:',response.status_code)
else:
      print(' -- HTTP Status Codes:',response.status_code)
      print(' -- Check Subscan API Documents: ' + SUBSCAN_API_DOC)
      print(' -- quit()')
      quit()

csvファイル書き出し用データ作成処理

レスポンスデータはJSONで1件ずつ取得し、block_timestamp,amount,event_indexと各固定値を合わせたリストを作成します
block_timestampはそのままではUNIX時間のためfromtimestamp()でローカル時間に変換します
日時情報はクリプタクトの指定に合わせるためフォーマットを指定して文字列に変換します
amountはそのままでは実際の報酬量と一致しないためADJUST_VALUEを掛けて調整します

getSubscanStakingRewardsInputNewCreateCsvData.py

# レスポンスデータ(JSON形式)
response_json = response.json()

# 1件数ずつ処理する
for i in range(input_num):
      # block_timestamp取得(csv:Timestamp)
      block_timestamp = response_json['data']['list'][i]['block_timestamp']
      # UTC(日本時間)以外にする場合はtimedeltaオブジェクトで調整すること。
      dt_block_timestamp = datetime.datetime.fromtimestamp(block_timestamp)
      # フォーマット変更
      dt_block_timestamp_fmt = dt_block_timestamp.strftime('%Y/%m/%d %H:%M:%S')
      # 0埋め表記を有効にするためシングルコーテーションを付ける
      dt_block_timestamp_fmt = "'" + dt_block_timestamp_fmt
      # ステーキング報酬取得(csv:Volume)
      amount = response_json['data']['list'][i]['amount']
      # 報酬量調整
      amount_float = float(amount) * ADJUST_VALUE
      # event_index取得(csv:Comment)
      event_index = response_json['data']['list'][i]['event_index']
      # csvファイル書き用データを作成
      value = [dt_block_timestamp_fmt,ACTION,SOURCE,BASE,amount_float,PRICE,COUNTER,FEE,FEECCY,event_index]

カスタムファイル用データ整形処理

カスタムファイル用のデータはpandas.DataFrameで作成します（1件ずつ1行分追加）
ステーキング報酬量はpandas.DataFrameのobject型で処理する場合、桁数が変わることがあるためfloat型で数値データに変換してround()で桁数を調整します
Feeは単位が円で数値データとして保存するためint型に変換します

getSubscanStakingRewardsInputNewCreateCsvData.py

# DataFrame作成
# 行データを追加する
df_add.loc[i, :] = value
# Volumeの表示桁数調整
df_add['Volume'] = df_add['Volume'].astype(float)
df_add = df_add.round({'Volume':10})
# Feeの型変換
df_add['Fee'] = df_add['Fee'].astype(int)

ファイル書き出し処理(新規作成)

データはTimestamp列で昇順にソートします
csvファイルはパスとファイル名+コード実行時の日付で保存します

getSubscanStakingRewardsInputNewCreateCsvData.py

# ファイル書き出し処理
# Timestamp列で昇順にソート
df_csv = df_new.sort_values('Timestamp')
# プログラム実行時の日付を取得
dt_today = datetime.date.today()
# ファイル名を日付指定で保存
filename = FILE_NAME + '_' + str(dt_today) + '.csv'
# index指定なしでファイル書き出し
df_csv.to_csv(PATH + filename,index = False)
print(' -- Save Location: ['+ PATH +  filename + ']\n')

ファイル書き出し処理(既存ファイルに追記)

データは既存ファイルのデータと結合します
データに重複がある場合はdrop_duplicates()完全一致の重複行を削除します

getSubscanStakingRewardsInputAddCsvData.py

# データを結合する
df_cc = pd.concat([df_base,df_add])
# 重複行を削除する
df_cc = df_cc.drop_duplicates()

最後に

PythonからSubscan APIを使用してステーキング報酬履歴を取得してクリプタクトのカスタムファイルフォーマットでcsvファイルに保存する方法を紹介しました。
Subscan APIでは他の情報も取得できるので今後はDOT以外のNetworkや他のRequest情報を取得する方法についても試してみようかと思います。
処理についてはGASよりも速いですが、最終的なカスタムファイルは他の履歴も含めるとスプレッドシートで管理できた方が楽なのでこちらの記事のようにGoogleのAPIを使うことや、GASでSubscan APIを使用して定期実行するなど、改良点はあるので引き続き検討してみようと思います。

Subscan APIドキュメントには記載がなく正式な情報ではありません。誤りや仕様に関して情報があればコメントしていただけると助かります。 ↩