Sample codes and details.
Sample Code:
import requests
import pandas as pd
import os
import websocket
import json
import ssl
import threading
import datetime
# Set Binance API Key as an environment variable
api_key = os.getenv('BINANCE_API_KEY')
latest_funding_rates = {}
def on_message(ws, message):
global latest_funding_rates
data = json.loads(message)
if 'stream' in data:
symbol = data['stream'].split('@')[0].upper() # Extract symbol from stream name in uppercase
latest_funding_rates[symbol] = float(data['data']['r']) # Store the rate
def on_error(ws, error):
print("Error:", error)
def on_close(ws, code, reason):
print(f"WebSocket closed with code {code}, reason {reason}")
def on_open(ws):
global streams
symbols = get_futures_symbols()
streams = [f"{symbol.lower()}@markPrice" for symbol in symbols]
subscribe_message = json.dumps({
"method": "SUBSCRIBE",
"params": streams,
"id": 1
})
ws.send(subscribe_message)
def connect_websocket():
websocket.enableTrace(False)
ws = websocket.WebSocketApp("wss://dstream.binance.com/stream",
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close)
def run_ws():
ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
thread = threading.Thread(target=run_ws)
thread.start()
return ws, thread
def get_futures_symbols():
url = 'https://dapi.binance.com/dapi/v1/exchangeInfo'
headers = {'X-MBX-APIKEY': api_key}
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
symbols = [item['symbol'] for item in data['symbols'] if item['contractType'] == 'PERPETUAL']
return symbols
return []
# Global weights for APR calculation
WEIGHTS = [W3, W7, W30, W_prev, W_next]
def calculate_apr_score(group, symbol):
symbol = symbol.upper() # Ensure the symbol is in uppercase
aprs = [
group.head(9)['fundingRate'].mean() * 3 * 360 * 100,
group.head(21)['fundingRate'].mean() * 3 * 360 * 100,
group.head(90)['fundingRate'].mean() * 3 * 360 * 100,
group.iloc[1]['fundingRate'] * 3 * 360 * 100,
latest_funding_rates.get(symbol, group.iloc[0]['fundingRate']) * 3 * 360 * 100 # Use real-time rate if available
]
apr_score = sum(apr * weight for apr, weight in zip(aprs, WEIGHTS))
return apr_score
def get_funding_rate_history(symbol, display_days=30):
url = 'https://dapi.binance.com/dapi/v1/fundingRate'
params = {'symbol': symbol, 'limit': display_days * 3}
headers = {'X-MBX-APIKEY': api_key}
response = requests.get(url, params=params, headers=headers)
if response.status_code == 200:
data = response.json()
df = pd.DataFrame(data)
df['fundingRate'] = pd.to_numeric(df['fundingRate'], errors='coerce')
df['Time'] = pd.to_datetime(df['fundingTime'], unit='ms', utc=True).dt.tz_convert(None)
return df.sort_values('Time', ascending=False).head(display_days)
return pd.DataFrame()
def process_data():
ws, thread = connect_websocket()
thread.join(timeout=30) # Wait for the WebSocket to collect data
symbols = get_futures_symbols()
all_data_frames = []
for symbol in symbols:
df = get_funding_rate_history(symbol, 60)
if not df.empty:
df['symbol'] = symbol
all_data_frames.append(df)
final_data_list = []
for data in all_data_frames:
symbol = data['symbol'].iloc[0]
apr_score = calculate_apr_score(data, symbol)
latest_data = data.sort_values('Time', ascending=False).iloc[0].to_dict()
latest_data['Previous Funding Rate'] = data.iloc[1]['fundingRate'] * 360 * 3 * 100
latest_data['Next Funding Rate'] = latest_funding_rates.get(symbol, data.iloc[0]['fundingRate']) * 360 * 3 * 100
latest_data['3 Day Cum Funding APR'] = data.head(9)['fundingRate'].mean() * 360 * 3 * 100
latest_data['7 Day Cum Funding APR'] = data.head(21)['fundingRate'].mean() * 360 * 3 * 100
latest_data['30 Day Cum Funding APR'] = data.head(90)['fundingRate'].mean() * 360 * 3 * 100
latest_data['APR Score'] = f"{apr_score:.3f}%"
# Select only the desired columns
final_data_list.append({
'symbol': latest_data['symbol'],
'Time': latest_data['Time'],
'Previous Funding Rate': f"{latest_data['Previous Funding Rate']:.3f}%",
'Next Funding Rate': f"{latest_data['Next Funding Rate']:.3f}%",
'3 Day Cum Funding APR': f"{latest_data['3 Day Cum Funding APR']:.3f}%",
'7 Day Cum Funding APR': f"{latest_data['7 Day Cum Funding APR']:.3f}%",
'30 Day Cum Funding APR': f"{latest_data['30 Day Cum Funding APR']:.3f}%",
'APR Score': latest_data['APR Score']
})
final_data = pd.DataFrame(final_data_list)
final_data.sort_values('APR Score', ascending=False, inplace=True)
# Add timestamp to file name
timestamp = datetime.datetime.utcnow().strftime('%Y-%m-%d_%H-%M-%S_UTC')
file_name = f'APR_results_{timestamp}.csv'
final_data.to_csv(file_name, index=False)
print(f"Data saved to '{file_name}'")
if ws.sock and ws.sock.connected:
ws.close()
if __name__ == "__main__":
process_data()Math formula
1. Average Funding Rate Calculation:
Average Funding Rate_n = (Sum of Funding Rates over the last n periods) / n
Here, n represents the number of periods, such as 3, 7, or 30 days.
2. Annualization of the Funding Rate:
Annualized Funding APR_n = Average Funding Rate_n x 360 x 3 x 100
This converts the average funding rate for n days into an annual percentage rate by multiplying by the number of days in a year (365) and then converting it to a percentage by multiplying by 100.
3. APR Score Calculation with Weights:
APR Score = (Annualized Funding APR_3 x W_3) + (Annualized Funding APR_7 x W_7) + (Annualized Funding APR_30 x W_30) + (Annualized Funding APR_next x W_next) + (Annualized Funding APR_prev x W_prev)
The APR Score is calculated by taking the weighted average of several Annualized Funding APRs for different periods. This is done by multiplying each Annualized Funding APR by its respective weight and then summing up these products.Output
Last updated
