Visual Studio Codeを起動してプログラムファイルを作成する
Pythonの開発環境の準備が完了したら、 VS Codeを起動して新規のPythonファイル(*.py)を作成します。
ここで作成したPythonのファイルには「リスト１」のコードをコピペします。
リスト１：Article139.py
# Article139.py
### Import the libraries
import os
import math
import numpy as np
import pandas as pd
import requests
import datetime as dt
from datetime import timedelta
from time import sleep
from dash import Dash, dash_table, html, Output, Input, State, dcc, ctx
from dash.dash_table import DataTable, FormatTemplate
from dash.dash_table.Format import Format, Scheme, Align, Trim
import dash_bootstrap_components as dbc
from lib.com_lib import get_data, calculate_cum_log_return
from lib.gmo_api import get_crypto, get_crypto_symbols
import warnings
warnings.simplefilter('ignore')
#################################
# Main
#################################
### Load the data from gmo coin
# get all symbols from gmo coin
df = get_crypto_symbols()
symbols = df['symbol'].values.tolist()
interval = '1day'
date_list = ['2020','2021','2022','2023']
symbol_list = []
cum_log_return_list = []
cum_log_return_pct_list = []
for symbol in symbols:
csv_file = f"datasets/csv/gmo_crypto_2020_2023({symbol})_{interval}.csv"
isFile = os.path.isfile(csv_file)
if not isFile:
for date in date_list: # '2020','2021','2022','2023'
df = get_crypto(symbol, interval, date) # get n rows from starting date
if not df.empty:
df.to_csv(csv_file, index=True)
# end of if not isFile:
isFile = os.path.isfile(csv_file)
if isFile:
df = get_data(csv_file)
if not df.empty:
df = calculate_cum_log_return(df)
df.replace([np.inf, -np.inf], np.nan).dropna(axis=1, inplace=True)
cum_log_return = df.iloc[-1]['cum_log_return']
cum_log_return_pct = df.iloc[-1]['cum_log_return_pct']
symbol_list.append(symbol)
cum_log_return_list.append(cum_log_return)
cum_log_return_pct_list.append(cum_log_return_pct)
# end of for symbol in symbols:
### Create DataFrame from dict
data = {
'symbol': symbol_list,
'cum_log_return': cum_log_return_list,
'cum_log_return_pct': cum_log_return_pct_list
}
raw_df = pd.DataFrame(data)
if raw_df.empty:
print(f"Quit the program due to raw_df is empty: {raw_df.empty=}")
quit()
### Replace np.inf or -np.inf (positive or negative infinity) with np.nan(Not A Number)
df = raw_df.replace([np.inf, -np.inf], np.nan)
df.isnull().sum()
### Drop rows if np.nan (Not A Number)
df.dropna(axis=0, inplace=True)
df.isnull().sum()
### Define Dash DataTable info
dt_dict = df.to_dict('records')
dt_cols = [{"name": i, "id": i} for i in df.columns]
### Instantiate Dash Class
app = Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP])
### Layout the Web Page
app.layout = dbc.Container(
[
dbc.Row(
html.H3('GMO Coin: Cryptocurrencies by Cumulative Log Return'),
),
dbc.Row(html.Br()),
dbc.Row(
dash_table.DataTable(
id='dt',
data=dt_dict,
columns=dt_cols,
),
),
dbc.Row(html.Br()),
], fluid=False,
)
### Run the server
if __name__ == '__main__':
app.run_server(debug=True)
図2
図2は、VS Codeの編集画面にプログラムのソースコードが表示されている画面です。
本記事で解説している全てのコードを掲載
ここでは、本記事で解説している全てのコードを掲載しています。
ここに掲載している「Article139.py」は最終版ですから、全ての機能が含まれています。
なお、Article139.pyでインポートしている「gmo_api.py, com_lib.py」のファイルは、「lib」フォルダに格納してください。
リスト２: Article139.py
# Article.py
### Import the libraries
import os
import math
import numpy as np
import pandas as pd
import requests
import datetime as dt
from datetime import timedelta
from time import sleep
from dash import Dash, dash_table, html, Output, Input, State, dcc, ctx
from dash.dash_table import DataTable, FormatTemplate
from dash.dash_table.Format import Format, Scheme, Align, Trim
import dash_bootstrap_components as dbc
from lib.com_lib import get_data, calculate_cum_log_return
from lib.gmo_api import get_crypto, get_crypto_symbols
import warnings
warnings.simplefilter('ignore')
#################################
# Main
#################################
### Load the data from gmo coin
# get all symbols from gmo coin
df = get_crypto_symbols()
symbols = df['symbol'].values.tolist()
interval = '1day'
date_list = ['2020','2021','2022','2023']
symbol_list = []
cum_log_return_list = []
cum_log_return_pct_list = []
for symbol in symbols:
csv_file = f"datasets/csv/gmo_crypto_2020_2023({symbol})_{interval}.csv"
isFile = os.path.isfile(csv_file)
if not isFile:
for date in date_list: # '2020','2021','2022','2023'
df = get_crypto(symbol, interval, date) # get n rows from starting date
if not df.empty:
df.to_csv(csv_file, index=True)
# end of if not isFile:
isFile = os.path.isfile(csv_file)
if isFile:
df = get_data(csv_file)
if not df.empty:
df = calculate_cum_log_return(df)
df.replace([np.inf, -np.inf], np.nan).dropna(axis=1, inplace=True)
cum_log_return = df.iloc[-1]['cum_log_return']
cum_log_return_pct = df.iloc[-1]['cum_log_return_pct']
symbol_list.append(symbol)
cum_log_return_list.append(cum_log_return)
cum_log_return_pct_list.append(cum_log_return_pct)
# end of for symbol in symbols:
### Create DataFrame from dict
data = {
'symbol': symbol_list,
'cum_log_return': cum_log_return_list,
'cum_log_return_pct': cum_log_return_pct_list
}
raw_df = pd.DataFrame(data)
if raw_df.empty:
print(f"Quit the program due to raw_df is empty: {raw_df.empty=}")
quit()
### Replace np.inf or -np.inf (positive or negative infinity) with np.nan(Not A Number)
df = raw_df.replace([np.inf, -np.inf], np.nan)
df.isnull().sum()
### Drop rows if np.nan (Not A Number)
df.dropna(axis=0, inplace=True)
df.isnull().sum()
### Define Dash DataTable info
df['cum_log_return_pct'] = df['cum_log_return']
dt_dict = df.to_dict('records')
# dt_cols = [{"name": i, "id": i} for i in df.columns]
log_return = FormatTemplate.Format(precision=6, scheme=Scheme.fixed)
log_return_pct = FormatTemplate.percentage(2)
dt_cols = [
dict(id='symbol', name='symbol', type='text'),
dict(id='cum_log_return', name='cum_log_return', type='numeric', deletable=True, format=log_return),
dict(id='cum_log_return_pct', name='cum_log_return_pct', type='numeric', format=log_return_pct)
]
tip_header_cols = {
'symbol': 'crypto symbol',
'cum_log_return': 'cumulative log return',
'cum_log_return_pct': 'cumulative log return percentage (%)',
}
style_header = {
'backgroundColor': 'rgb(210, 210, 210)',
'color': 'black',
'fontWeight': 'bold'
}
style_cell_conditional= [{
'if': {'column_id': 'symbol'},
'textAlign': 'left'
}]
style_data_conditional = [{
'if': {'row_index': 'odd'},
'backgroundColor': 'rgb(220, 220, 220)',
}]
### Instantiate Dash Class
app = Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP])
### Layout the Web Page
app.layout = dbc.Container(
[
dbc.Row(
html.H3('GMO Coin: Cryptocurrencies by Cumulative Log Return'),
),
dbc.Row(html.Br()),
dbc.Row(
dash_table.DataTable(
id='dt',
data=dt_dict,
columns=dt_cols,
tooltip_header=tip_header_cols,
tooltip_delay=0,
tooltip_duration=None,
sort_action='native',
sort_mode='multi',
style_header=style_header,
style_cell_conditional=style_cell_conditional,
style_data_conditional=style_data_conditional,
),
),
dbc.Row(html.Br()),
], fluid=False,
)
### Run the server
if __name__ == '__main__':
app.run_server(debug=True)
リスト３: /lib/gmo_api.py
# gmo_api.py
import math
import numpy as np
import pandas as pd
import requests
import datetime as dt
from datetime import timedelta
from time import sleep
import warnings
warnings.simplefilter('ignore')
####################################################
def get_crypto_symbols(symbol=None) -> pd.DataFrame:
# Define GMO Coin API endpoint
endpoint = 'https://api.coin.z.com/public/v1/ticker'
params = {}
if symbol is not None:
# add a new element(symbol) to the params dictionary
params['symbol'] = symbol
res_dict = requests.get(endpoint, params=params)
dict = res_dict.json()
status = dict.get('status')
# no error ?
if status == 0:
data_list = dict.get('data')
df = pd.DataFrame(data_list) # ask, bid, high, last, low, symbol, timestamp, volume
return df
else:
print(f"get_crypto_symbols() error => {status=}")
return pd.DataFrame()
############################################################ Load data from GMO Coin
def get_crypto(symbol='BTC', interval='1day', start='2018'):
'''
interval=1min 5min 10min 15min 30min 1hour for YYYYMMDD
interval=4hour 8hour 12hour 1day 1week 1month YYYY
start='2018'
'''
url = 'https://api.coin.z.com/public/v1/klines'
params = {
'symbol': symbol,
'interval': interval,
'date': start
}
try:
res_dict = requests.get(url, params=params)
dict = res_dict.json()
status = dict.get('status')
# no error ?
if status == 0:
data_list = dict.get('data')
df = pd.DataFrame(data_list)
df.columns = ['date', 'open', 'high', 'low', 'close', 'volume']
df['date'] = pd.to_datetime(df['date'], unit='ms')
df.set_index('date', inplace=True)
df = df.astype(float)
df['symbol'] = symbol
# df.reset_index(inplace=True)
# print(f"get_crypto({symbol=}, {interval=}, {start=}) => {df.shape[0]=}")
return df
else:
print(f"get_crypto({symbol=}, {interval=}, {start=}) error => {status=}")
return pd.DataFrame()
except requests.exceptions.HTTPError as e:
print(f"get_crypto({symbol=}, {interval=}, {start=}) HTTP error: {e}")
except Exception as e:
print(f"get_crypto({symbol=}, {interval=}, {start=}) exception error: {e}")
return pd.DataFrame()
リスト４: /lib/com_lib.py
# com_lib.py
import math
import numpy as np
import pandas as pd
import requests
import datetime as dt
from datetime import timedelta
from time import sleep
import warnings
warnings.simplefilter('ignore')
############################################
def get_data(csv_file: str) -> pd.DataFrame:
# print(f"Loading data: {csv_file} ")
df = pd.read_csv(csv_file)
# date,open,high,low,close,adj close,volume,symbol
df['date'] = pd.to_datetime(df['date'])
df.set_index(['date'], inplace=True)
return df
###############################################################
def calculate_cum_log_return(df: pd.DataFrame) -> pd.DataFrame:
# Calculate log return
df['log_return'] = np.log(df['close'] / df['close'].shift(1))
# Calculate cumulative log return
df['cum_log_return'] = np.exp(df['log_return'].cumsum()) - 1
df['cum_log_return_pct'] = df['cum_log_return'] * 100
# Preview the resulting dataframe
# print(f"Cumulative Log Return for {df.iloc[-1]['symbol']} = {df.iloc[-1]['cum_log_return_pct']:.2%}")
return df