Botで仮想通貨取引すると思わぬバグ等で暴走・損失するケースがあるのでテスト用プログラムを作成してみます。
Pythonを使った仮想通貨取引のテストプログラムです。データベースが必要になりますがsqliteを使います。Botにデータベース接続、挿入、変更のプログラムを組み込みます。
Pythonに必要なモジュール
import time
import os
import re
import sqlite3
from contextlib import closing, suppress
テーブルは「id, pos, unixtime, price, stop, target, flag」です。
テスト用なのでidは便宜的にunixtimeを割り当てます。posはポジションで「S」か「L」を割り当てます。priceはオーダ時のBTC価格、stopの設定は今回は必要ありません。targetはlimitに相当します(データベースのクエリにlimitを入れるとエラーになります)。flagは取引状態を記録します。今回はオーダー時は「T」、利確時は「Z」になります。
データベースの作成・登録
# database
dbname = 'database.db'
with closing(sqlite3.connect(dbname)) as conn:
c = conn.cursor()
create_table = '''CREATE TABLE IF NOT EXISTS transactions (id INT, pos VARCHAR(1),
unixtime INT, price REAL, stop REAL, target REAL, flag VARCHAR(1))'''
c.execute(create_table)
conn.commit()
オーダの登録
def test_db_pos_insert(d_id, position, unixtime, btc_close, stop, target, flag = ''):
if position == 'short':
position_cap = 'S'
elif position == 'long':
position_cap = 'L'
with closing(sqlite3.connect(dbname)) as conn:
c = conn.cursor()
sql = 'insert into transactions (id, pos, unixtime, price, stop, target, flag) values (?,?,?,?,?,?,?)'
dat = (d_id, position_cap, unixtime, btc_close, stop, target, flag)
c.execute(sql, dat)
conn.commit()
レコードの取得
def test_db_read(st=''):
with closing(sqlite3.connect(dbname)) as conn:
c = conn.cursor()
select_sql = 'select * from transactions ' + st
c.execute(select_sql)
rows = c.fetchall()
return rows
レコードのアップデート
def test_db_update(st=''):
with closing(sqlite3.connect(dbname)) as conn:
c = conn.cursor()
update_sql = 'update transactions ' + st
c.execute(update_sql)
conn.commit()
利確
def test_profit(d_id, btc_close):
st = 'set flag = "Z", target = ' + str(btc_close) + ' where id = ' + str(d_id)
test_db_update(st)
ボットにデータベースプログラムを埋め込んだら次にレポートプログラムを作成します。現時点から遡って1日、1週間、1ヶ月のレポートが可能です。(デフォルトでは1日)
#!/home/user/environments/test_env/bin/python3
import os
import sys
#import subprocess
import re
import time
import sqlite3
from contextlib import closing
dbname = 'database.db'
#start_hour = 0
#start_day = 1
#start_month = 'jan'
#start_week = 'sun'
timezone = 'JST-9' # timezone: Japan
unit = 'daily' # default
c_trade_cost = 0.075 # percent (total transaction costs; transaction fee, spread, swap and so on)
os.environ['TZ'] = timezone
time.tzset()
unixtime = int(time.time())
if len(sys.argv) > 1:
arg = sys.argv[1]
num = int()
unit = str()
matchObj = re.match( r'(\d*)(\w*)', arg, re.M|re.I)
if matchObj:
if matchObj.group(1):
num = int(matchObj.group(1))
unit = matchObj.group(2)
if re.match( r'daily$', unit, re.M|re.I):
unit = 'daily'
elif re.match( r'weekly$', unit, re.M|re.I):
unit = 'weekly'
elif re.match( r'monthly$', unit, re.M|re.I):
unit = 'monthly'
# elif re.match( r'd$|days?$', unit, re.M|re.I) and num > 0:
# unit = 'd'
# elif re.match( r'w$|weeks?$', unit, re.M|re.I) and num > 0:
# unit = 'w'
# elif re.match( r'm$|months?$', unit, re.M|re.I) and num > 0:
# unit = 'm'
# elif re.match( r'y$|years?$|yrs?$', unit, re.M|re.I) and num > 0:
# unit = 'y'
# elif re.match( r'yearly$', unit, re.M|re.I):
# unit = 'yearly'
else:
print('ERROR: wrong argument!')
exit(1)
def db_read(args):
with closing(sqlite3.connect(dbname)) as conn:
c = conn.cursor()
select_sql = 'select * from transactions where ' + str(args)
c.execute(select_sql)
rows = c.fetchall()
return rows
def db_read_one(args):
with closing(sqlite3.connect(dbname)) as conn:
c = conn.cursor()
select_sql = str(args)
c.execute(select_sql)
row = c.fetchone()
return row
# TABLE: id, pos, unixtime, price, stop, target, flag
def get_data(t_start, t_end):
profit = float()
price_high = float()
price_low = float()
fee = float()
st = 'unixtime > ' + str(t_start) + ' and unixtime < ' + str(t_end) + ' and flag = "Z"'
rows = db_read(st)
_a = db_read_one('select max(price) from transactions where ' + st)
price_high = _a[0]
_a = db_read_one('select min(price) from transactions where ' + st)
price_low = _a[0]
_a = db_read_one('select count(*) from transactions where ' + st)
count = _a[0]
for d in rows:
if d[1] == 'L':
profit = profit + (d[5] - d[3])
elif d[1] == 'S':
profit = profit + (d[3] - d[5])
fee = fee + d[3] * c_trade_cost / 100
return profit, price_high, price_low, count, fee
def calc(unit):
if unit == 'daily':
t_start = unixtime - (1 * 24 * 60 * 60)
t_end = unixtime
elif unit == 'weekly':
t_start = unixtime - (1 * 24 * 60 * 60 * 7)
t_end = unixtime
elif unit == 'monthly':
t_end = unixtime
t_end_customized = time.localtime(t_end)
t_m = str(time.strftime("%m", t_end_customized))
t_y = str(time.strftime("%y", t_end_customized))
t_y_leap = int(t_y) % 4
if re.match( r'^1$|^5$|^7$|^8$|^10$|^12$', t_m):
t_d = 30
elif re.match( r'^2$|^4$|^6$|^9$|^11$', t_m):
t_d = 31
elif re.match( r'^3$', t_m) and t_y_leap != 0:
t_d = 28
elif re.match( r'^3$', t_m) and t_y_leap == 0:
t_d = 29
t_start = unixtime - (1 * 24 * 60 * 60 * t_d)
return t_start, t_end
def report(unit):
t_start, t_end = calc(unit)
profit, price1, price2, count, fee = get_data(t_start, t_end)
print("\n")
print(' Action :', unit, 'profit')
print(' From :', time.strftime('%X %x %Z', time.localtime(int(t_start))))
print(' TO :', time.strftime('%X %x %Z', time.localtime(int(t_end))))
print(' Transactions :', '{:>5}'.format(count))
print(' Price(high) :', '{:>8.2f}'.format(price1))
print(' Price(low) :', '{:>8.2f}'.format(price2))
print(' Fee :', '{:>8.2f}'.format(fee))
print(' Profit :', '{:>8.2f}'.format(profit - fee))
print("\n")
report(unit)

価格差が200ドル開いていて36ドルの利益ではアルゴリズムを改良する必要があるのがわかります。
[追記 1]

手数料を追加で計算に入れてみると取引が多くなるほど利益が著しく減少します。