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
Output
Last updated