初めまして、ブロックチェーンベンチャーで暗号資産周辺の会計処理、内部統制の構築に関連する業務を行なっている公認会計士のyuk_tです。
個人的にも4年ほど前から暗号資産で遊んでいます。
Pythonでよく使うチェーンのトランザクション内容を複数ウォレット分まとめて出力するコードを書いたので公開してみます。
非エンジニアなので至らない点も多々あるでしょうがご容赦ください。
このコードで出来ること
google colaboratoryで動かすことを想定したものです。
指定したアドレスのEthereum、polygon、Gnosis chainでの取引をほぼ全て出力することができます。APIで取得できるものだけなので残念ながら一部の取引は対象外となります。
（出力したトランザクション累計で算出した残高と、ノードより取得した残高が一致していることの確認によって把握するべきトランザクションの漏れが無いかチェックできます）
発生したトランザクションを一覧できるだけでなく、自分のウォレット間で行われた取引の判別や、消費したGas代の確認もできるようになっています。
アドレスごとにExcelファイルが作成され、個々のExcelファイル内ではトークン別に取引が記載されたシートが作成されます。
なお、シート1枚目（"sheet"というシート名）は抽出できたそのウォレットでの取引が全て記載されているサマリーシートのようなものになっています。
主な用途
企業で暗号資産周辺の経理業務をする場合、特に便利に使っていただけると思います。
個人の資産の管理や確定申告にも役立てていただけるかもしれません。
少し応用していただければ同じ形式での他チェーンの取引出力や、取引所アドレスとの取引判別などもできると思います。
使い方
- EtherscanとpolygonscanのAPIキーを取得し、"mykey" "polkey" にそれぞれ代入します。（APIキー取得方法はここでは省略させていただきます）
- myaddsの[ ]内にデータが欲しいウォレットのアドレスを入力します。アドレスは "" で囲んでくださいね。複数ある場合は , で区切ってください。
- alladdsの[ ]内に自分のウォレットアドレスすべてを入力してください。myaddsと同様、アドレスは "" で囲んで複数ある場合は , で区切ってください。自分のウォレット間取引を判別するものなので、迷ったらmyaddsと同じ内容で大丈夫です。
- bridgeには判別したいブリッジ用コントラクトがあれば入力してください。こちらは任意です。
出力された内容についての説明
hash：トランザクションハッシュです
timeStamp：取引日時（日本標準時間）です
from：送信元アドレス
to：送信先アドレス
tokenSymbol：ETH、wBTCなどのトークンシンボルです。
value：取引枚数です。tokendecimalは処理済です。
_in：トークンが入ってきた枚数です
_out：トークンが出ていった枚数です
TransactionFee:自分のウォレットで発生したgas代です。native tokenのトランザクションと紐付けて処理しています。
wallet：トランザクションが発生したウォレットが記入されます
chain：トランザクションが発生したチェーンが記入されます
notes：自分のウォレット間取引であればここに"Both are my wallets"と書き込まれます（本当はinternalにしたかったけど紛らわしかったので・・・）。リスト:bridgeにブリッジコントラクトを入力している場合、ブリッジのトランザクションについては"bridge"と書き込まれます。
import requests
import json
import pandas as pd
import datetime
import openpyxl as opx
import time
import re
pd.set_option('display.max_columns', 30)
mykey = "" # etherscanのAPIキー
polkey = "" # polygonscanのAPIキー
def all_txns(myadd, alladdress, bridge_address):
network = ["ether", "polygon", "xdai"]
native_token = ["ETH", "MATIC", "DAI"]
#ここからnormal
normal_url = []
normal_url.append(
f'https://api.etherscan.io/api?module=account&action=txlist&address={myadd}&startblock=0&endblock=99999999&page=1&offset=999&sort=asc&apikey={mykey}')
normal_url.append(
f'https://api.polygonscan.com/api?module=account&action=txlist&address={myadd}&startblock=1&endblock=99999999&sort=asc&apikey={polkey}')
normal_url.append(
f'https://blockscout.com/xdai/mainnet/api?module=account&action=txlist&address={myadd}')
normal_response_data = []
for i, u in enumerate(normal_url):
while True:
normal_data = requests.get(u)
time.sleep(1)
if normal_data.ok:
normal_response_data.append(json.loads(normal_data.text))
break
df_normal_txns = pd.DataFrame()
for i, normal_data_sample in enumerate(normal_response_data):
if normal_data_sample["status"] == "0": # 取引なしのウォレットでエラーが出ないように分岐させる
print(network[i]+"でのnormal_txnなし:"+myadd)
else:
normal_df = pd.DataFrame(normal_data_sample['result'])
normal_df = normal_df.loc[:, ['hash', 'timeStamp', 'from', 'to', 'value',"gasPrice","gasUsed"]]
normal_df.insert(loc = 4, column= 'tokenSymbol', value= native_token[i])
normal_df['value'] = normal_df['value'].astype('float128')/pow(10,18)
normal_df = normal_df.assign(_in=0, _out=0, TransactionFee=0, wallet=myadd,
chain=network[i], check1=0, check2=0, notes=0)
normal_df['TransactionFee'] = normal_df["gasPrice"].astype(int) * normal_df["gasUsed"].astype(int)
normal_df['TransactionFee'] = normal_df['TransactionFee'].astype('float128')/pow(10,18)
df_normal_txns = pd.concat([df_normal_txns, normal_df])
#ここからERC20
ERC20_url = []
ERC20_url.append(
f'https://api.etherscan.io/api?module=account&action=tokentx&address={myadd}&startblock=0&endblock=99999999&page=1&offset=999&sort=asc&apikey={mykey}')
ERC20_url.append(
f'https://api.polygonscan.com/api?module=account&action=tokentx&address={myadd}&startblock=0&endblock=99999999&sort=asc&apikey={polkey}')
ERC20_url.append(
f'https://blockscout.com/xdai/mainnet/api?module=account&action=tokentx&address={myadd}')
ERC20_response_data = []
for i, v in enumerate(ERC20_url):
while True:
data = requests.get(v)
time.sleep(1)
if data.ok:
ERC20_response_data.append(json.loads(data.text))
break
df_ERC20_txns = pd.DataFrame()
for i, data_sample in enumerate(ERC20_response_data):
if data_sample["status"] == "0": # 取引なしのウォレットでエラーが出ないように分岐させる
print(network[i]+"でのERC20_txnsなし:"+myadd)
else:
ERC20＿df = pd.DataFrame(data_sample['result'])
dicimals = ERC20＿df['tokenDecimal']
ERC20＿df = ERC20＿df.loc[:, ['hash', 'timeStamp', 'from', 'to', 'tokenSymbol','value',"gasPrice","gasUsed"]]
ERC20＿df['value'] = ERC20＿df['value'].astype('float128')/pow(10, dicimals.astype('float128'))
ERC20＿df = ERC20＿df.assign(_in=0, _out=0, TransactionFee=0, wallet=myadd,
chain=network[i], check1=0, check2=0, notes=0)
df_ERC20_txns = pd.concat([df_ERC20_txns, ERC20＿df])
#ここからinternal（xDAIは取れない）
internal_url = []
internal_url.append(
f'https://api.etherscan.io/api?module=account&action=txlistinternal&address={myadd}&startblock=0&endblock=99999999&page=1&offset=999&sort=asc&apikey={mykey}')
internal_url.append(
f'https://api.polygonscan.com/api?module=account&action=txlistinternal&address={myadd}&startblock=0&endblock=99999999&sort=asc&apikey={polkey}')
internal_response_data = []
for i, w in enumerate(internal_url):
while True:
data = requests.get(w)
time.sleep(1)
if data.ok:
internal_response_data.append(json.loads(data.text))
break
df_internal_txns = pd.DataFrame()
for i, data_sample in enumerate(internal_response_data):
if data_sample["status"] == "0":
print(network[i]+"でのinternal_txnsなし:"+myadd)
else:
internal_df = pd.DataFrame(data_sample['result'])
internal_df = internal_df.loc[:, ['hash', 'timeStamp', 'from', 'to','value']]
internal_df.insert(loc = 4, column= 'tokenSymbol', value= native_token[i])
internal_df['value'] = internal_df['value'].astype('float128')/pow(10,18)
internal_df = internal_df.assign(gasPrice=0,gasUsed=0,_in=0, _out=0, TransactionFee=0, wallet=myadd,
chain=network[i], check1=0, check2=0, notes=0)
df_internal_txns = pd.concat([df_internal_txns, internal_df])
df_all_txns = pd.concat([df_ERC20_txns, df_normal_txns,df_internal_txns])
df_token_list = []
token_name = []
#ここから全体的な処理
if df_all_txns.empty == False :
df_all_txns['timeStamp'] = pd.to_datetime(df_all_txns['timeStamp'].astype(int)+9*60*60, unit='s')
df_all_txns = df_all_txns.sort_values(by='timeStamp',ascending=True)
out_loc = df_all_txns['from'] == myadd.lower()
df_all_txns.loc[out_loc, '_out'] = df_all_txns.loc[out_loc, 'value']
in_loc = df_all_txns['to'] == myadd.lower()
df_all_txns.loc[in_loc, '_in'] = df_all_txns.loc[in_loc, 'value']
gas_loc = df_all_txns['to'] == myadd.lower()
df_all_txns.loc[gas_loc, 'TransactionFee']=0
def myfunc_in(x): return x in alladdress # fromが自分のウォレットか判定
tmp_in = df_all_txns['from'].apply(myfunc_in)
df_all_txns.loc[tmp_in, 'check1'] = "mine"
def myfunc_out(y): return y in alladdress # toが自分のウォレットか判定
tmp_out = df_all_txns['to'].apply(myfunc_out)
df_all_txns.loc[tmp_out, 'check2'] = "mine"
# fromもtoも自分のウォレットであれば、notes列に "JPYC's" 記載
notes_loc = df_all_txns['check1'] == df_all_txns['check2']
df_all_txns.loc[notes_loc, 'notes'] = "Both are my wallets"
# toがブリッジ用コントラクトアドレスか判定
def myfunc_brg(b): return b in bridge_address
tmp_brg = df_all_txns['to'].apply(myfunc_brg)
df_all_txns.loc[tmp_brg, 'notes'] = "bridge"
df_all_txns = df_all_txns.drop(['gasPrice','gasUsed','check1','check2'], axis=1)#check列、gas計算列を消す
def okikae(k):
#tokensymbolが空欄（機種依存文字的なものが空欄扱いになってる？）の物にはとりあえず何か入れる
if not k :
k ='_'
return k
#tokensymbolにシート名に使えない文字が入っている場合があるので、アルファベット以外は使える記号に置き換える。
else:
t=re.sub(r"[^a-zA-Z]", "_", k)
return t
df_all_txns['tokenSymbol'] = df_all_txns['tokenSymbol'].apply(okikae)
#tokensymbolをリスト化する
tokens=list(set(df_all_txns["tokenSymbol"].tolist()))
for token in tokens:
token_list=df_all_txns[df_all_txns['tokenSymbol'] == token]
df_token_list.append(token_list)
token_name.append(token)
return df_token_list, token_name,df_all_txns
else:
return df_token_list, token_name,df_all_txns
#ここで関数の定義終わり
# 情報が欲しいアドレスを入れる
myadds = [""]
# 自分のウォレットのアドレスを全部入れる（自分のウォレット間取引判定用）
alladds = [""]
alladds = list(map(lambda z: z.lower(), alladds))
# ブリッジコントラクトとのトランザクション特定用
bridge = []
bridge = list(map(lambda br: br.lower(), bridge))
for myadd in myadds:
wb = opx.Workbook()
wb.save(myadd+ ".xlsx")
returned_data = all_txns(myadd, alladds, bridge)
EXL=pd.ExcelWriter(myadd+ ".xlsx", engine='openpyxl')
EXL.sheets=dict((ws.title, ws) for ws in EXL.book.worksheets)
returned_data[2].to_excel(EXL, sheet_name="Sheet", index=False)
for i, token_list2 in enumerate(returned_data[0]):
token_list2.to_excel(EXL, sheet_name=returned_data[1][i], index=False)
EXL.save()