How to build a crypto market scanner in python?
Crypto market scanner in python
The Why:
Why would I want to build a scanner? First there’s edge in knowing which assets are going up and which assets are not. Some of my systems depend on assets trending for the past few days in order to start applying capital to them.
Another reason is to gauge sentiment of the market. You might want to know how the global market is doing without having to go chart by chart checking prices against their most commonly analysed metrics.
So my goal is to always automate this boring, repetitive tasks. In this article we’ll go through from the stage zero to having a fully developed scanner that with one click you get the visual representation of the assets above a certain metric.
Lets build the code:
First we need to import the packages we’ll need for this project.
import pandas as pd
import numpy as np
import datetime
from dateutil.relativedelta import relativedelta
from binance import Client, ThreadedWebsocketManager, ThreadedDepthCacheManager
from ta.trend import SMAIndicator
import matplotlib.pyplot as plt
import warnings
import timeAfter we’ve imported our packages we need to connect to the exchange we use to trade. In my case I use Binance.
def exchange_connection():
apikey = ''
secretkey = ''
client = Client(apikey, secretkey)
return clientIn this function you connect with your exchange. I have left empty the “apikey” and “secretkey” so that I don’t disclose my keys obviously. All you have to do is go to Binance, generate your own keys, and input there.
After you’ve done that, we need to extract the list of assets that we want to keep track off.
We could manually input the assets we wish to track into a list, but that’s not the goal of this article as we aim for full automation. In the code below we will connect to Binance and extract the name of the various assets tradable on Binance.
You can do that with the function “get_all_tickers()” from the Binance package. After that we do the conversions we need to workable data, remove the stable coins, and return a list of assets to keep track off.
def get_assets(exchange_connection):
# Binance Assets
tickers = exchange_connection.get_all_tickers()
tickers = pd.DataFrame(tickers)
tickers['collateral'] = [x.strip()[-4:] for x in tickers['symbol']]
tickers = tickers.drop(tickers[(tickers['collateral'] != 'USDT')].index)
tickers = tickers.reset_index(drop=True)
tickers["price"] = pd.to_numeric(tickers["price"])
list_of_stables = ["EURUSDT", "DAIUSDT", "TUSDUSDT", "USDCUSDT", "BUSDUSDT", "USDTPUSDT", "USTUSDT", "USDSBUSDT", "USDSUSDT",
"SUSDUSDT", "USDPUSDT", "PAXUSDT"]
tickers = tickers[~tickers['symbol'].isin(list_of_stables)]
tickers = tickers.reset_index(drop=True)
tickers = tickers['symbol'].tolist()
return tickers
The next function we will build is to actually extract the data for the list of assets. We need the data in order to calculate our metrics that we want to measure right?
def get_data(days_range, timeframe, list_of_tickers, exchange_connection):
#Function Time Tracker
start = time.time()
#Extract Data for Calculations
df_list = {}
for ticker in list_of_tickers:
try:
print("Fetching " + str(days_range) + " Days of Data For: " + str(ticker))
ticker_for_data = ticker
today_date = datetime.date.today()
trade_date_data = today_date - relativedelta(days=days_range)
historical = exchange_connection.get_historical_klines(ticker, timeframe, str(trade_date_data))
hist_df = pd.DataFrame(historical)
hist_df.columns = ['Open Time', 'Open', 'High', 'Low', 'Close', 'Volume', 'Close Time', 'Quote Asset Volume',
'Number of Trades', 'TB Base Volume', 'TB Quote Volume', 'Ignore']
hist_df["Ticker"] = ticker_for_data
hist_df['Open Time'] = pd.to_datetime(hist_df['Open Time']/1000, unit='s')
hist_df['Close Time'] = pd.to_datetime(hist_df['Close Time']/1000, unit='s')
numeric_columns = ['Open', 'High', 'Low', 'Close', 'Volume', 'Quote Asset Volume', 'TB Base Volume', 'TB Quote Volume']
hist_df[numeric_columns] = hist_df[numeric_columns].apply(pd.to_numeric, axis=1)
df = hist_df[["Ticker", "Open Time", "Open", "High", "Low", "Close", "Volume"]]
df['RowN'] = np.arange(len(df))
df = df.rename(columns={'Open Time': 'date'})
df.columns = [x.lower() for x in df.columns]
df = df[["ticker", 'rown', 'date', 'open', 'high', 'low', "close", "volume"]]
df_list[ticker] = {}
df_list[ticker] = df
print("Finished Fetching " + str(days_range) + " Days of Data For: " + str(ticker))
except:
print("Error Fetching Data for Ticker: "+ str(ticker))
end = time.time()
print("Time Elapsed On get_data Function:" + str(end - start))
return df_listIn this function we extract the data of each individual asset with a for loop, and we append it to a python dictionary. That way we can keep accurate track of the data that we extract.
For the above function to work we have to give it the following variables:
days_range: The total amount of days for analysis (Usually I set it to 3k days as it gives me the maximum amount of days available. Feel free to set what suits you best).
timeframe: The timeframe of our analysis. For this example we’ll set it at 1d.
list_of_tickers: The list of tickers we have extracted from the previous function called “get_assets”.
exchange_connection: The function that connects us to the broker module above.
Right now with the functions above we have:
Our connection to the exchange
Our list of assets
Data for each asset in the list of assets
Now all we have to do is build the scanner in order to extract the charts we are looking for.
def scanner(list_of_data):
#Fonts and Presentation
font1 = {'family':'serif','color':'black','size':12}
font2 = {'family':'serif','color':'black','size':10}
assets_data = {}
for key, df in list_of_data.items():
assets_data[key] = {}
#Define Criteria for Scanner
indicator_step1 = SMAIndicator(close=df["close"], window=50)
df["50_sma"] = indicator_step1.sma_indicator()
df["close_above_50_sma"] = np.where((df["close"] > df["50_sma"]), 1, 0)
assets_data[key] = df
#Get a List of Assets Above 50SMA
merged_dfs_50sma = pd.concat(assets_data, ignore_index=True)
merged_dfs_50sma = merged_dfs_50sma[(merged_dfs_50sma.close_above_50_sma == 1)]
merged_dfs_50sma = merged_dfs_50sma.sort_values('date')
merged_dfs_50sma = merged_dfs_50sma.reset_index(drop=True)
merged_dfs_50sma = merged_dfs_50sma[["date", "ticker", "close_above_50_sma"]]
#Get a List of the Number of Assets Above the 50SMA
unique_dates_50sma = pd.DataFrame(merged_dfs_50sma.date.dropna().unique(), columns=['date'])
unique_dates_50sma["assets_above_50sma"] = unique_dates_50sma.date.apply(lambda y: merged_dfs_50sma[merged_dfs_50sma.date == y].close_above_50_sma.sum()).replace(0, np.nan)
#Chart the Results
plt.bar(unique_dates_50sma["date"], unique_dates_50sma["assets_above_50sma"], color="blue")
plt.gcf().autofmt_xdate()
plt.title('Total # of Crypto Assets Above Their 50 Simple Moving Average', fontdict = font1)
plt.xlabel('Dates', fontdict = font2)
plt.ylabel('# of Crypto Assets', fontdict = font2)
plt.show()
return merged_dfs_50smaWith the function above you “feed” it the list of assets and then it proceeds on the following manner:
For each asset it calculates its metric for analysis. In this case it will calculate the 50SMA using the TA module from python.
After calculating the metrics, it will append the dataframes into a new dictionary that will contain all this data.
We merge all the dataframes from this dictionary into one massive dataframe.
We extract the unique dates on this dataframe and then we can calculate how many assets were above their 50SMA.
We plot the results using matplotlib from python.
Conclusion:
In this article I’ve shared the the full code ready for implementation if you want to use.
My goal is to share my research as I go here on substack to keep track of everything I am doing. If you like this sort of content about systematic trading and coding, feel free to subscribe as I am planning to share a lot more.
Have a great week!