仮想通貨取引のテストプログラム

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]

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