# 僕の個人的なクセと基本的なapiの処理
## 一番初めの状態
たとえばbinanceでUSDT先物のkline/candlestickデータ(open,high,low,close,volumeとかのohlcデータ)を取る関数だったらbinanceのapiドキュメント
https://binance-docs.github.io/apidocs/futures/en/#kline-candlestick-data
を参照して
```python=
import requests
import pandas as pd
class binance_api:
def ohlc_data(self):
symbol = "BTCUSDT" #今回はビットコインのデータをとる（例えばイーサリアムがほしければETHUSDTとか）
interval = "4h" #4時間足のデータをとる（enum形式の記述についてはbinanceのapiドキュメント参照）
response = requests.get(f"https://fapi.binance.com/fapi/v1/klines?symbol={symbol}&interval={interval}") #BTCUSDT、4時間足データ
response = response.json()
#responseの二次元配列から必要なデータのみをpandasで扱いやすい形に直す関数
def two_dim_array_to_dic(res):
dic = {
"symbol" : symbol,
"date" : res[0],
"open" : res[1],
"high" : res[2],
"low" : res[3],
"close" : res[4],
"volume" : res[5],
}
return dic
df = map(two_dim_array_to_dic, response)
df = pd.DataFrame(df)
df["date"] = df["date"].apply(lambda x: pd.to_datetime(x,unit="ms")) #timestampだと分かりにくいから日付型に直す、binanceのtimestampは12ケタのmsオーダー
df[["open","high","low","close"]] = df[["open","high","low","close"]].applymap(float) #レスポンスは全部strで返されてだるいから全部floatに直しとく
print(df)
return
if __name__ == "__main__":
ba = binance_api()
ba.ohlc_data()
```
```
symbol date open high low close volume
0 BTCUSDT 2022-04-20 12:00:00 42070.70 42160.00 41037.30 41247.10 84586.898
1 BTCUSDT 2022-04-20 16:00:00 41247.00 41444.00 40800.00 41177.60 51498.554
2 BTCUSDT 2022-04-20 20:00:00 41177.50 41660.00 41118.20 41327.70 26596.361
3 BTCUSDT 2022-04-21 00:00:00 41326.00 41768.00 41292.00 41673.90 22093.892
4 BTCUSDT 2022-04-21 04:00:00 41673.90 41940.00 41398.00 41833.70 26747.921
.. ... ... ... ... ... ... ...
495 BTCUSDT 2022-07-12 00:00:00 19954.30 19994.80 19780.00 19934.30 75346.940
496 BTCUSDT 2022-07-12 04:00:00 19934.20 20048.90 19618.20 19722.90 105992.070
497 BTCUSDT 2022-07-12 08:00:00 19722.90 19843.00 19552.40 19765.10 106267.951
498 BTCUSDT 2022-07-12 12:00:00 19765.00 19999.50 19680.80 19876.90 132675.241
499 BTCUSDT 2022-07-12 16:00:00 19876.90 19993.00 19749.10 19756.20 49451.165
[500 rows x 7 columns]
```
みたいに僕は書く。
pandas大好き人間だから結構特殊（？）な書き方かも。
## try-exceptでエラー処理
でも上のコードのままだと、例えば8行目で通信エラーが起きたときの処理を一切考慮していない。
だからとりあえずtry-exceptでエラー処理を実装したい。
当然本当は各行の処理でエラー内容に応じて細かく分岐させるのが良いに決まってるけど、横着して僕は関数全体の処理をtryでくくって、Exceptionに全部エラーぶん投げるとかいう力技くそコードで書いてる。
そのコードで例えばこんな風に存在しない銘柄のapiを呼ぼうとしたときに発生したエラーに対してはこんな風に分岐できる。
```python=
import requests
import pandas as pd
import traceback
class binance_api:
def ohlc_data(self):
try:
symbol = "BTUSDT" #今回はビットコインのデータをとる（例えばイーサリアムがほしければETHUSDTとか）
interval = "4h" #4時間足のデータをとる（enum形式の記述についてはbinanceのapiドキュメント参照）
response = requests.get(f"https://fapi.binance.com/fapi/v1/klines?symbol={symbol}&interval={interval}") #BTCUSDT、4時間足データ
response = response.json()
#responseの二次元配列から必要なデータのみをpandasで扱いやすい形に直す関数
def two_dim_array_to_dic(res):
dic = {
"symbol" : symbol,
"date" : res[0],
"open" : res[1],
"high" : res[2],
"low" : res[3],
"close" : res[4],
"volume" : res[5],
}
return dic
df = map(two_dim_array_to_dic, response)
df = pd.DataFrame(df)
df["date"] = df["date"].apply(lambda x: pd.to_datetime(x,unit="ms")) #timestampだと分かりにくいから日付型に直す、binance
df[["open","high","low","close"]] = df[["open","high","low","close"]].applymap(float) #レスポンスは全部strで返されてだるいから全部floatに直しとく
print(df)
return
except Exception as e:
print("\n___エラー発生___\n")
traceback.print_exc()
return
if __name__ == "__main__":
ba = binance_api()
ba.ohlc_data()
```
```
___エラー発生___
Traceback (most recent call last):
File "c:/Users/nagoy/OneDrive/デスクトップ/save/to_kishimoto/test.py", line 27, in ohlc_data
df = pd.DataFrame(df)
File "C:\Users\nagoy\anaconda3\lib\site-packages\pandas\core\frame.py", line 502, in __init__
data = list(data)
File "c:/Users/nagoy/OneDrive/デスクトップ/save/to_kishimoto/test.py", line 21, in two_dim_array_to_dic
"close" : res[4],
IndexError: string index out of range
```
このエラー内容を読んでも原因が探りにくいし正直くそみたいな設計だとは思うけど、とりま最低限上手く行ってないことだけは分かるしまあこれでいったんヨシとする。
## whileでエラー時リトライ、閾値超えたら強制終了
エラーの時は同じ処理を何回か繰り返してみて、ある一定回数以上試してもだめなら処理を強制終了とすれば実運用として最低限使えるものにはなりそう。
だからtryの上からさらにwhileでくくってみる。
```python=
import requests
import pandas as pd
import traceback
import time
import sys
class binance_api:
def ohlc_data(self):
retry_time = 10
retry_th = 10240
while retry_time<=retry_th:
try:
symbol = "BTUSDT" #今回はビットコインのデータをとる（例えばイーサリアムがほしければETHUSDTとか）
interval = "4h" #4時間足のデータをとる（enum形式の記述についてはbinanceのapiドキュメント参照）
response = requests.get(f"https://fapi.binance.com/fapi/v1/klines?symbol={symbol}&interval={interval}") #BTCUSDT、4時間足データ
response = response.json()
#responseの二次元配列から必要なデータのみをpandasで扱いやすい形に直す関数
def two_dim_array_to_dic(res):
dic = {
"symbol" : symbol,
"date" : res[0],
"open" : res[1],
"high" : res[2],
"low" : res[3],
"close" : res[4],
"volume" : res[5],
}
return dic
df = map(two_dim_array_to_dic, response)
df = pd.DataFrame(df)
df["date"] = df["date"].apply(lambda x: pd.to_datetime(x,unit="ms")) #timestampだと分かりにくいから日付型に直す、binance
df[["open","high","low","close"]] = df[["open","high","low","close"]].applymap(float) #レスポンスは全部strで返されてだるいから全部floatに直しとく
print(df)
return
except Exception as e:
print("\n___エラー発生___\n")
traceback.print_exc()
print(f"{retry_time} 秒後にリトライします")
time.sleep(retry_time)
retry_time *= 2
print("リトライ回数の閾値を超えたためbotを強制終了します")
sys.exit()
if __name__ == "__main__":
ba = binance_api()
ba.ohlc_data()
```
```
___エラー発生___
Traceback (most recent call last):
File "c:/Users/nagoy/OneDrive/デスクトップ/save/to_kishimoto/test.py", line 32, in ohlc_data
df = pd.DataFrame(df)
File "C:\Users\nagoy\anaconda3\lib\site-packages\pandas\core\frame.py", line 502, in __init__
data = list(data)
File "c:/Users/nagoy/OneDrive/デスクトップ/save/to_kishimoto/test.py", line 26, in two_dim_array_to_dic
"close" : res[4],
IndexError: string index out of range
10 秒後にリトライします
...
リトライ回数の閾値を超えたためbotを強制終了します
```
こんな感じ。
ftxapimの全関数をこの考え方を基本にして書いてる。
でもさすがにこの書き方だと保守管理しにくすぎて発狂しそうだからいつか根本的に変えると思う。
今のところはこれが独学素人の限界。
Sign in with Wallet
Connect another wallet