gp_binance_demo/binance_gp_demo.py

import threading

import websocket
import json
import pandas as pd
import numpy as np
from scipy.optimize import minimize
import time
import warnings
import logging
import colorlog

# 忽略 FutureWarning
warnings.simplefilter(action='ignore', category=FutureWarning)

# 配置日志
handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter(
    "%(log_color)s%(asctime)s - %(name)s - %(levelname)s \n %(message)s",
    datefmt=None,
    reset=True,
    log_colors={
        'DEBUG': 'cyan',
        'INFO': 'blue',
        'WARNING': 'yellow',
        'ERROR': 'red',
        'CRITICAL': 'bold_red',
    }
))
logger = logging.getLogger("binance_gp_demo")
logger.setLevel(logging.INFO)
logger.addHandler(handler)

# 步骤二：订阅Binance的成交数据和订单簿数据
# Binance WebSocket API URL
SOCKET_TRADE = "wss://stream.binance.com:9443/ws/btcusdt@trade"
SOCKET_DEPTH = "wss://stream.binance.com:9443/ws/btcusdt@depth20@100ms"

# Initialize the DataFrame
df_trades = pd.DataFrame(columns=['price', 'qty', 'timestamp'])
df_order_book = pd.DataFrame(columns=['bid_price', 'bid_qty', 'ask_price', 'ask_qty'])


def on_message_trade(_ws, message):
    global df_trades
    json_message = json.loads(message)
    trade = {
        'price': float(json_message['p']),
        'qty': float(json_message['q']),
        'timestamp': pd.to_datetime(json_message['T'], unit='ms')
    }
    trade_df = pd.DataFrame([trade])
    if not trade_df.empty and not trade_df.isna().all().all():
        df_trades = pd.concat([df_trades, trade_df], ignore_index=True)


# Function to handle order book messages
def on_message_depth(_ws, message):
    global df_order_book
    json_message = json.loads(message)
    bids = json_message['bids'][:10]  # Top 10 bids
    asks = json_message['asks'][:10]  # Top 10 asks
    order_book = {
        'bid_price': [float(bid[0]) for bid in bids],
        'bid_qty': [float(bid[1]) for bid in bids],
        'ask_price': [float(ask[0]) for ask in asks],
        'ask_qty': [float(ask[1]) for ask in asks]
    }
    df_order_book = pd.DataFrame([order_book])


def on_error(_ws, error):
    logger.error(error)


def on_open(_ws):
    print("### opened ###")


# Create a WebSocket app
ws_trade = websocket.WebSocketApp(SOCKET_TRADE, on_message=on_message_trade, on_error=on_error)
ws_depth = websocket.WebSocketApp(SOCKET_DEPTH, on_message=on_message_depth, on_error=on_error)

# 定义要传递给 run_forever 的参数
http_proxy_host = "127.0.0.1"
http_proxy_port = 7890
proxy_type = "http"
# Run the WebSocket with proxy settings
trade_thread = threading.Thread(target=ws_trade.run_forever, kwargs={
    'http_proxy_host': http_proxy_host,
    'http_proxy_port': http_proxy_port,
    'proxy_type': proxy_type
})
depth_thread = threading.Thread(target=ws_depth.run_forever, kwargs={
    'http_proxy_host': http_proxy_host,
    'http_proxy_port': http_proxy_port,
    'proxy_type': proxy_type
})
trade_thread.start()
depth_thread.start()

# 步骤三：价差随机过程建模
# Define the states and transition matrix
states = ['tight', 'normal', 'wide']
transition_matrix = np.zeros((3, 3))


# 定义函数来更新转移矩阵
def update_transition_matrix(df):
    global transition_matrix
    for i in range(len(df) - 1):
        current_state = df['state'].iloc[i]
        next_state = df['state'].iloc[i + 1]
        transition_matrix[states.index(current_state), states.index(next_state)] += 1

    row_sums = transition_matrix.sum(axis=1, keepdims=True)
    row_sums[row_sums == 0] = 1
    transition_matrix = transition_matrix / row_sums

# 定义函数来分类价差状态
def classify_spread(spread):
    if spread < 0.01:
        return 'tight'
    elif spread < 0.02:
        return 'normal'
    else:
        return 'wide'

# 定义函数来计算价差并进行分类
def calculate_and_classify_spread():
    global df_trades
    df_trades['spread'] = df_trades['price'].diff().abs()
    df_trades['state'] = df_trades['spread'].apply(classify_spread)

# 定义周期性更新转移矩阵的函数
stop_event = threading.Event()

def update_transition_matrix_periodically():
    while not stop_event.is_set():
        calculate_and_classify_spread()
        update_transition_matrix(df_trades)
        current_state = df_trades['state'].iloc[-1] if not df_trades.empty else 'unknown'
        # logger.info("Current State: %s\nTransition Matrix:\n%s\n", current_state, transition_matrix)
        stop_event.wait(5)  # 每5秒更新一次

# 创建并启动线程
transition_matrix_update_thread = threading.Thread(target=update_transition_matrix_periodically)
transition_matrix_update_thread.start()


# 四、参数估计，我们将通过订单簿数据估计成交密度。
# 用于存储前一个订单簿的全局变量
previous_order_book = None
# Function to estimate fill probabilities and order flow based on order book data and trades
def estimate_fill_probabilities_and_order_flow(order_book, trades):
    global previous_order_book
    fill_probabilities = {}
    order_flow = {'new_orders': 0, 'cancellations': 0, 'executions': 0}

    if previous_order_book is not None:
        for level in range(10):  # 取前10档数据
            bid_price = order_book['bid_price'].iloc[0][level]
            ask_price = order_book['ask_price'].iloc[0][level]
            bid_qty = order_book['bid_qty'].iloc[0][level]
            ask_qty = order_book['ask_qty'].iloc[0][level]

            prev_bid_qty = previous_order_book['bid_qty'].iloc[0][level]
            prev_ask_qty = previous_order_book['ask_qty'].iloc[0][level]

            # 计算成交量占订单量的比例
            fill_bid = trades[(trades['price'] <= bid_price) & (trades['price'] > bid_price - 0.01)]
            fill_ask = trades[(trades['price'] >= ask_price) & (trades['price'] < ask_price + 0.01)]

            fill_probabilities[f'bid_{level+1}'] = fill_bid['qty'].sum() / bid_qty if bid_qty > 0 else 0
            fill_probabilities[f'ask_{level+1}'] = fill_ask['qty'].sum() / ask_qty if ask_qty > 0 else 0

            # 计算订单流
            new_bid_orders = max(0, bid_qty - prev_bid_qty)
            new_ask_orders = max(0, ask_qty - prev_ask_qty)
            cancellations_bid = max(0, prev_bid_qty - bid_qty)
            cancellations_ask = max(0, prev_ask_qty - ask_qty)

            order_flow['new_orders'] += new_bid_orders + new_ask_orders
            order_flow['cancellations'] += cancellations_bid + cancellations_ask
            order_flow['executions'] += fill_bid['qty'].sum() + fill_ask['qty'].sum()

    # 更新 previous_order_book
    previous_order_book = order_book.copy()

    return fill_probabilities, order_flow

# Estimate fill probabilities and order flow periodically
def estimate_fill_probabilities_and_order_flow_periodically():
    global df_order_book
    while not stop_event.is_set():
        if not df_order_book.empty and not df_trades.empty:
            fill_probabilities, order_flow = estimate_fill_probabilities_and_order_flow(df_order_book, df_trades)
            logger.info("Fill Probabilities:\n%s", fill_probabilities)
            logger.info("Order Flow:\n%s\n", order_flow)
        stop_event.wait(5)  # 每10秒更新一次

# 创建并启动线程
fill_probabilities_thread = threading.Thread(target=estimate_fill_probabilities_and_order_flow_periodically)
fill_probabilities_thread.start()