加密貨幣是近年來很熱門的課題，我們這次就來看看如何用簡單實作量化交易的整個流程，在這個教學裡面會一次展示如何爬取資料，算出一大堆技術指標(目前大約150個)，然後設計一個交易策略，在歷史data以及線上交易上做回測，用這個策略自動交易。
基本設定
首先，安裝和設定環境，大家可以在colab上嘗試:
pip install cryptota -U接下來做一些基本設定，我們這次用ADA作為我們的範例
# Fetch data setting
CRYPTO = "ADAUSDT"
START = '7 day ago UTC'
END = 'now UTC'
INTERVAL = '1m'
# trading strategy parameter
PARAMETER = { "initial_state": 1, "delay": 500, "initial_money": 100,"max_buy":10, "max_sell":10 }
# binance api key and secret
APIKEY = ""
APISECRET = "環境設定
import cryptota
import vectorbt as vbt
import numpy as np
from binance import Client, ThreadedWebsocketManager, ThreadedDepthCacheManager
import matplotlib.pyplot as plt
import time
from datetime import timedelta
client = Client(APIKEY,APISECRET)
UNITS = {"s":"seconds", "m":"minutes", "h":"hours", "d":"days", "w":"weeks"}
def convert_to_seconds(s):
count = int(s[:-1])
unit = UNITS[ s[-1] ]
td = timedelta(**{unit: count})
return td.seconds + 60 * 60 * 24 * td.days資料取得
把資料拿下來，算指標
binance_data = vbt.BinanceData.download(
CRYPTO,
start=START,
end=END,
interval=INTERVAL
)
price = binance_data.get()
ta = cryptota.TA_Features()
df_full = ta.get_all_indicators(price.copy())
直接看看得到的指標和資料
有資料和指標，我們就可以設計自己的策略
def buy_stock(
real_movement,
delay = 5,
initial_state = 1,
initial_money = 10000,
max_buy = 1,
max_sell = 1,
print_log=True
):
"""
real_movement = actual movement in the real world
delay = how much interval you want to delay to change our decision from buy to sell, vice versa
initial_state = 1 is buy, 0 is sell
initial_money = 1000, ignore what kind of currency
max_buy = max quantity for share to buy
max_sell = max quantity for share to sell
"""
starting_money = initial_money
delay_change_decision = delay
current_decision = 0
state = initial_state
current_val = real_movement[0]
states_sell = []
states_buy = []
states_entry = []
states_exit = []
current_inventory = 0
def buy(i, initial_money, current_inventory):
shares = initial_money // real_movement[i]
if shares < 1:
if print_log:
print(
'day %d: total balances %f, not enough money to buy a unit price %f'
% (i, initial_money, real_movement[i])
)
else:
if shares > max_buy:
buy_units = max_buy
else:
buy_units = shares
initial_money -= buy_units * real_movement[i]
current_inventory += buy_units
if print_log:
print(
'day %d: buy %d units at price %f, total balance %f'
% (i, buy_units, buy_units * real_movement[i], initial_money)
)
states_buy.append(0)
return initial_money, current_inventory
if state == 1:
initial_money, current_inventory = buy(
0, initial_money, current_inventory
)
for i in range(0, real_movement.shape[0], 1):
sentry = False
sexit = False
if real_movement[i] < current_val and state == 0:
if current_decision < delay_change_decision:
current_decision += 1
else:
state = 1
initial_money, current_inventory = buy(
i, initial_money, current_inventory
)
current_decision = 0
states_buy.append(i)
sentry = True
if real_movement[i] > current_val and state == 1:
if current_decision < delay_change_decision:
current_decision += 1
else:
state = 0
if current_inventory == 0:
if print_log:
print('day %d: cannot sell anything, inventory 0' % (i))
else:
if current_inventory > max_sell:
sell_units = max_sell
else:
sell_units = current_inventory
current_inventory -= sell_units
total_sell = sell_units * real_movement[i]
initial_money += total_sell
try:
invest = (
(real_movement[i] - real_movement[states_buy[-1]])
/ real_movement[states_buy[-1]]
) * 100
except:
invest = 0
if print_log:
print(
'day %d, sell %d units at price %f, investment %f %%, total balance %f,'
% (i, sell_units, total_sell, invest, initial_money)
)
current_decision = 0
states_sell.append(i)
sexit = True
states_entry.append(sentry)
states_exit.append(sexit)
current_val = real_movement[i]
invest = ((initial_money - starting_money) / starting_money) * 100
total_gains = initial_money - starting_money
return states_buy, states_sell,states_entry,states_exit, total_gains, invest這個策略的成效如何呢，我們拿歷史data測試看看
states_buy, states_sell, states_entry, states_exit, total_gains, invest = buy_stock(df_full.close,**PARAMETER)
畫成圖表來看看
close = df_full['close']
fig = plt.figure(figsize = (15,5))
plt.plot(close, color='r', lw=2.)
plt.plot(close, '^', markersize=10, color='m', label = 'buying signal', markevery = states_buy)
plt.plot(close, 'v', markersize=10, color='k', label = 'selling signal', markevery = states_sell)
plt.legend()
plt.show()
交易其實還有手續費，高頻交易時也要考慮看看
fees = 0.001
try:
fees = client.get_trade_fee(symbol=CRYPTO)[0]['makerCommission']
except:
pass然後我們再看看更詳細的回測成效
portfolio_kwargs = dict(size=np.inf, fees=float(fees), freq=INTERVAL)
portfolio = vbt.Portfolio.from_signals(df_full['close'], states_entry, states_exit, **portfolio_kwargs)
portfolio.plot().show()
portfolio.stats()Start 2022-03-19 07:57:00+00:00
End 2022-03-26 07:56:00+00:00
Period 7 days 00:00:00
Start Value 100.0
End Value 119.015983
Total Return [%] 19.015983
Benchmark Return [%] 32.366589
Max Gross Exposure [%] 100.0
Total Fees Paid 0.546932
Max Drawdown [%] 6.113537
Max Drawdown Duration 1 days 03:33:00
Total Trades 3
Total Closed Trades 2
Total Open Trades 1
Open Trade PnL 1.864827
Win Rate [%] 100.0
Best Trade [%] 10.15132
Worst Trade [%] 6.370903
Avg Winning Trade [%] 8.261111
Avg Losing Trade [%] NaN
Avg Winning Trade Duration 0 days 23:04:00
Avg Losing Trade Duration NaT
Profit Factor inf
Expectancy 8.575578
Sharpe Ratio 12.557528
Calmar Ratio 143196.901926
Omega Ratio 1.079802
Sortino Ratio 18.805026
Name: close, dtype: object一旦找到合適的策略，我們就可以將他上線了
先確定看看資訊
info = client.get_symbol_info(CRYPTO)
info{'allowTrailingStop': False,
'baseAsset': 'ADA',
'baseAssetPrecision': 8,
'baseCommissionPrecision': 8,
'filters': [{'filterType': 'PRICE_FILTER',
'maxPrice': '1000.00000000',
'minPrice': '0.00100000',
'tickSize': '0.00100000'},
{'avgPriceMins': 5,
'filterType': 'PERCENT_PRICE',
'multiplierDown': '0.2',
'multiplierUp': '5'},
{'filterType': 'LOT_SIZE',
'maxQty': '900000.00000000',
'minQty': '0.10000000',
'stepSize': '0.10000000'},
{'applyToMarket': True,
'avgPriceMins': 5,
'filterType': 'MIN_NOTIONAL',
'minNotional': '10.00000000'},
{'filterType': 'ICEBERG_PARTS', 'limit': 10},
{'filterType': 'MARKET_LOT_SIZE',
'maxQty': '8005213.92147324',
'minQty': '0.00000000',
'stepSize': '0.00000000'},
{'filterType': 'MAX_NUM_ORDERS', 'maxNumOrders': 200},
{'filterType': 'MAX_NUM_ALGO_ORDERS', 'maxNumAlgoOrders': 5}],
'icebergAllowed': True,
'isMarginTradingAllowed': True,
'isSpotTradingAllowed': True,
'ocoAllowed': True,
'orderTypes': ['LIMIT',
'LIMIT_MAKER',
'MARKET',
'STOP_LOSS_LIMIT',
'TAKE_PROFIT_LIMIT'],
'permissions': ['SPOT', 'MARGIN'],
'quoteAsset': 'USDT',
'quoteAssetPrecision': 8,
'quoteCommissionPrecision': 8,
'quoteOrderQtyMarketAllowed': True,
'quotePrecision': 8,
'status': 'TRADING',
'symbol': 'ADAUSDT'}這裡就用一個最簡單的loop自動交易, 我們的策略會按照當前的data判斷是否買賣，需要的話就自動買賣。
while True:
binance_data = binance_data.update()
price = binance_data.get()
states_buy, states_sell, states_entry, states_exit, total_gains, invest = buy_stock(price.Close,
initial_state = 1,
delay = 10,
initial_money = 1,
max_buy=1,
max_sell=1,
print_log=False)
states_entry[-1],states_exit[-1]
if not (states_entry[-1] or states_exit[-1]):
print("doing_noting")
if states_entry[-1]:
order = client.create_test_order( ## use test_order for real~
symbol='ADAUSDT',
side=Client.SIDE_BUY,
type=Client.ORDER_TYPE_MARKET,
quantity=8)
print("buy",order)
if states_exit[-1]:
order = client.create_test_order( ## use test_order for real~
symbol='ADAUSDT',
side=Client.SIDE_BUY,
type=Client.ORDER_TYPE_MARKET,
quantity=8)
print("sell",order)
time.sleep(convert_to_seconds(INTERVAL))
完整程式碼: https://voidful.dev/jupyter/2021/02/20/cryptotaipynb.html
