用 Python 篩選收益優(yōu)秀的加密貨幣
Python中文社區(qū) (ID:python-china)
在市值排名前 10 的加密貨幣中,從純粹的經(jīng)濟角度來看,你認為自 2017 年以來表現(xiàn)最好的加密貨幣是哪一種?
不管你信不信,幣安自己的 BNB 實際上遠遠超過了其他所有加密貨幣。我編寫了一個腳本來幫助我了解幾種加密貨幣的歷史表現(xiàn),當我決定只加入前 10 名加密貨幣并看看表現(xiàn)最好的貨幣是哪個。
在運行腳本之前,我很確定它可能將是 DOGE。所以我坐在這里,等待歷史數(shù)據(jù)下載,以便我的腳本可以繪制一些加密圖表。
腳本運行完畢,結(jié)果出來了,感謝中本聰,這不是 DOGE。哦,等等,這更有趣——它是 BNB。
自 2017 年以來,BNB 已上漲超過 20,000%。
程序能夠為你下載歷史數(shù)據(jù),并分析任意數(shù)量的幣種。如果您想對任意數(shù)量的加密貨幣的收益百分比進行快速比較分析,這很方便。您所需要的只是一些 Python 知識。
編寫加密貨幣分析工具
該代碼也可在 GitHub 上找到。
- https://github.com/CyberPunkMetalHead/crypto-performance-tracker
首先創(chuàng)建一個文本文件并將其命名為coins.txt。在此文本文件中,放入一些您想要分析的幣種名稱。它們需要包含配對符號,并且每行必須是 1 個貨幣,不能有逗號:
- BTCUSDT
- ETHUSDT
- BNBUSDT
創(chuàng)建一個 binancedata.py 文件。我們將使用此文件輪詢 Binance API 以獲取我們需要的金融數(shù)據(jù)。由于我們使用的是開放端口,因此不需要 API 密鑰和密碼。
讓我們導(dǎo)入一些依賴項并定義一個空的 Binance 客戶端:
- # needed for the binance API and websockets
- from binance.client import Client
- import csv
- import os
- import time
- from datetime import date, datetime
- client = Client()
現(xiàn)在讓我們編寫一個函數(shù)來從我們的coins.txt文件中打開和讀取貨幣:
- def get_coins():
- with open('coins.txt', 'r') as f:
- coins = f.readlines()
- coins = [coin.strip('\n') for coin in coins]
- return coins
此文件中的最后一個函數(shù)將為我們獲取歷史數(shù)據(jù)并以 CSV 格式保存:
- def get_historical_data(coin, since, kline_interval):
- """
- Args example:
- coin = 'BTCUSDT'
- since = '1 Jan 2021'
- kline_interval = Client.KLINE_INTERVAL_1MINUTE
- """
- if os.path.isfile(f'data/{coin}_{since}.csv'):
- print('Datafile already exists, loading file...')
- else:
- print(f'Fetching historical data for {coin}, this may take a few minutes...')
- start_time = time.perf_counter()
- data = client.get_historical_klines(coin, kline_interval, since)
- data = [item[0:5] for item in data]
- # field names
- fields = ['timstamp', 'high', 'low', 'open', 'close']
- # save the data
- with open(f'data/{coin}_{since}.csv', 'w', newline='') as f:
- # using csv.writer method from CSV package
- write = csv.writer(f)
- write.writerow(fields)
- write.writerows(data)
- end_time = time.perf_counter()
- # calculate how long it took to produce the file
- time_elapsed = round(end_time - start_time)
- print(f'Historical data for {coin} saved as {coin}_{since}.csv. Time elapsed: {time_elapsed} seconds')
- return f'{coin}_{since}.csv'
此函數(shù)還將檢查文件是否已經(jīng)存在,如果存在它不會再次下載。該函數(shù)接受 3 個參數(shù):coin、since 和 kline_interval。檢查函數(shù)下方的注釋,了解我們將傳遞給這些參數(shù)的正確格式。
保存文件,現(xiàn)在是創(chuàng)建我們的主要執(zhí)行文件的時候了,我們將把這個文件的內(nèi)容導(dǎo)入到其中。
繼續(xù)創(chuàng)建一個 main.py 文件并安裝以下依賴項:
- from binancedata import *
- import threading
- import matplotlib.pyplot as plt
- import matplotlib.cbook as cbook
- import numpy as np
- import pandas as pd
- # needed for the binance API and websockets
- from binance.client import Client
- import csv
- import os
- import time
- from datetime import datetime, date
讓我們開始一些線程。該腳本是為了一次下載多個數(shù)據(jù)文件,所以為了避免等待一次下載每個歷史數(shù)據(jù)文件,我們將使用線程并下載這些文件,如下所示:
- threads = []
- coins = get_coins()
- for coin in coins:
- t = threading.Thread(target=get_historical_data, args=(coin, '1 Jan 2017', Client.KLINE_INTERVAL_1DAY) ) #'get_historical_data('ETHUSDT', '1 Jan 2021', Client.KLINE_INTERVAL_1MINUTE)
- t.start()
- threads.append(t)
- [thread.join() for thread in threads]
現(xiàn)在我們需要一個函數(shù)來返回我們下載的所有數(shù)據(jù)文件的文件名:
- def get_all_filenames():
- return [get_historical_data(coin, '1 Jan 2017', Client.KLINE_INTERVAL_1DAY) for coin in coins]
最后,我們將定義主要函數(shù),我們將在其中繪制這些數(shù)據(jù)并運行腳本:
- def main():
- historical_data = get_all_filenames()
- for file in historical_data:
- data = pd.read_csv(f'data/{file}')
- rolling_percentage = data['close']
- rolling_percentage = [(item - rolling_percentage[0]) / rolling_percentage[0]*100 for item in rolling_percentage ]
- timestamp = data['timstamp']
- timestamp = [datetime.fromtimestamp(item/1000) for item in timestamp]
- plt.legend()
- plt.plot(timestamp, rolling_percentage, label=file)
- plt.xlabel("Date")
- plt.ylabel("% gain")
- plt.show()
- if __name__ == "__main__":
- main()
現(xiàn)在剩下要做的就是在腳本目錄中創(chuàng)建一個空文件夾并將其命名為 data。大功告成,您現(xiàn)在可以分析您想要的所有代幣的歷史收益。