gp_binance_demo/kappa/data_processing.py

import json
from decimal import Decimal, getcontext
import pandas as pd
import threading
from collections import deque
from scipy.integrate import trapz
import numpy as np
from scipy.optimize import minimize
from logger_config import logger

# 设置全局精度
getcontext().prec = 28

# 假设我们有一个数据流，订单簿和成交数据
order_book_snapshots = deque(maxlen=600)        # 存储过去600个订单簿快照
spread_delta_snapshots = deque(maxlen=6000)     # 存储过去600个价差数据(最小变动价格的倍数)
trade_snapshots = deque(maxlen=6000)            # 存储过去6000个成交数据

stop_event = threading.Event()

# 初始参数
k_initial = 6
A_initial = 140

# 定义参数范围
bounds = [(10, 1000.0),             # A 的范围
          (0.01, 100.0)]            # k 的范围

# 假设S0是初始的参考价格
S0 = -1


def on_message(_ws, message):
    json_message = json.loads(message)

    if json_message["channel"] == "futures.order_book" and json_message["event"] == "all":
        on_message_depth(json_message)
    elif json_message["channel"] == "futures.trades" and json_message["event"] == "update":
        on_message_trade(json_message)


def get_tick_size_from_prices(ask_price, bid_price):
    # 获取价格的小数位数
    ask_decimal_places = len(str(ask_price).split('.')[1])
    bid_decimal_places = len(str(bid_price).split('.')[1])

    # 确定最小变动单元
    tick_size = 10 ** -max(ask_decimal_places, bid_decimal_places)

    return tick_size


def on_message_trade(json_message):
    global trade_snapshots

    # 解析 Gate.io 的成交信息
    for trade_origin in json_message['result']:
        price = trade_origin['price']
        size = trade_origin['size']
        timestamp = pd.to_datetime(trade_origin['create_time_ms'], unit='ms')

        # 转换为需要的格式
        trade = {
            'price': float(price),
            'qty': float(size),
            'timestamp': timestamp
        }

        # 买卖方向判断
        if trade["qty"] > 0:
            trade["side"] = "buy"
        else:
            trade["side"] = "sell"

        trade_snapshots.append(trade)
        process_depth_data()


def on_message_depth(json_message):
    global order_book_snapshots, spread_delta_snapshots

    # 解析 Gate.io 的深度信息
    result = json_message['result']
    bids = result['bids']
    asks = result['asks']
    timestamp = pd.to_datetime(result['t'], unit='ms')

    # 转换为 Binance 兼容的格式
    asks_converted = [[float(ask['p']), float(ask['s'])] for ask in asks[:10]]
    bids_converted = [[float(bid['p']), float(bid['s'])] for bid in bids[:10]]

    depth = {
        'bids': bids_converted,
        'asks': asks_converted,
        'timestamp': timestamp
    }
    order_book_snapshots.append(depth)

    # 求价差
    ask_price = Decimal(str(asks_converted[0][0]))
    bid_price = Decimal(str(bids_converted[0][0]))
    tick_size = get_tick_size_from_prices(ask_price, bid_price)
    spread = float(ask_price - bid_price)
    spread_delta = int(spread / tick_size)
    spread_delta_snapshots.append(spread_delta)

    process_depth_data()


def calculate_phi(prices, k, S0):
    """
    计算 φ(k, ξ) 的值
    :param prices: 时间序列的价格数据
    :param k: 参数 k
    :param S0: 初始价格
    :return: φ(k, ξ) 的值
    """
    price_changes = np.array(prices) - S0
    exponentials = np.exp(k * price_changes)
    phi = np.mean(exponentials)
    return phi


def calculate_integral_phi(prices, k, S0, time_points):
    """
    计算 ∫ φ(k, ξ) dξ 的值，越大说明波动率也越大（价格波动）。
    :param prices: 时间序列的价格数据
    :param k: 参数 k
    :param S0: 初始价格
    :param time_points: 时间点数组
    :return: ∫ φ(k, ξ) dξ 的值
    """
    # 计算每个时间点的 φ(k, ξ) 值
    phi_values = [calculate_phi(prices[:i + 1], k, S0) for i in range(len(prices))]

    # 使用梯形法计算积分
    integral_phi = trapz(phi_values, time_points)

    return integral_phi


def estimate_lambda(waiting_times, T):
    """
    通过等待时间估计 λ(δ)
    :param waiting_times: 等待时间的数组
    :param T: 时间窗口的大小
    :return: λ(δ) 的估计值
    """
    # 将 waiting_times 转换为 NumPy 数组
    waiting_times = np.array(waiting_times)

    sum_indicator = np.sum(waiting_times < T)
    sum_waiting_times = int(np.sum(waiting_times).total_seconds() * 1000)
    lambda_hat = sum_indicator / sum_waiting_times
    return lambda_hat


def objective_function(params, delta_max, log_lambda_hat_value, log_integral_phi_value):
    """
    目标函数 r(A, k)
    :param params: 包含 A 和 k 的数组
    :param delta_max: 最大的价格偏移
    :param log_lambda_hat_value: log(λ(δ)) 的值
    :param log_integral_phi_value: log(∫ φ(k, ξ) dξ) 的值
    :return: 目标函数值
    """
    A, k = params

    residuals = []
    for delta in range(1, delta_max + 1):
        rst = (log_lambda_hat_value + k * delta - np.log(A) - log_integral_phi_value)
        residual = rst ** 2
        residuals.append(residual)
    return np.sum(residuals)


def calculate_delta_sum(gamma, sigma_squared, T_minus_t, k):
    term1 = gamma * sigma_squared * T_minus_t
    term2 = (2 / gamma) * np.log(1 + (gamma / k))
    delta_sum = term1 + term2
    return delta_sum


def calculate_sigma_squared(prices, timestamps):
    """
    计算 σ^2 的值
    :param prices: 时间序列的价格数据
    :param timestamps: 时间戳数组
    :return: σ^2 的值
    """
    n = len(prices)
    if n < 2:
        return 0.0

    time_diff = int(int((timestamps[-1] - timestamps[0]).total_seconds() * 1000) / 100)
    price_diff_squared = [(prices[i] - prices[i - 1]) ** 2 for i in range(1, n)]
    sigma_squared = np.sum(price_diff_squared) / time_diff

    return sigma_squared


def process_depth_data():
    global order_book_snapshots, trade_snapshots, spread_delta_snapshots
    global k_initial, A_initial, S0

    # 数据预热，至少3条深度数据以及10条成交数据才能用于计算
    if len(order_book_snapshots) < 3 or len(trade_snapshots) < 10:
        return

    S_values = [((snapshot['bids'][0][0] + snapshot['asks'][0][0]) / 2) for snapshot in order_book_snapshots]

    if S0 < 0:
        S0 = S_values[0]

    # ========================= 计算 log(∫ φ(k, ξ) dξ) ==================
    # 提取时间戳并计算时间间隔
    order_book_timestamps = [snapshot['timestamp'] for snapshot in order_book_snapshots]
    order_book_time_points = [(timestamp - order_book_timestamps[0]).total_seconds() for timestamp in order_book_timestamps]

    # 计算 ∫ φ(k, ξ) dξ
    integral_phi_value = calculate_integral_phi(S_values, k_initial, S0, order_book_time_points)

    # 计算 log(∫ φ(k, ξ) dξ)
    log_integral_phi_value = np.log(integral_phi_value)

    # ========================== 估计 λ(δ) ==============================
    # 计算等待时间
    trade_timestamps = [snapshot['timestamp'] for snapshot in trade_snapshots]
    waiting_times = [trade_timestamps[i] - trade_timestamps[i - 1] for i in range(1, len(trade_timestamps))]
    # 时间窗口的大小
    T = pd.to_datetime(100, unit='ms') - pd.to_datetime(0, unit='ms')

    # 计算 λ(δ) 的估计值
    lambda_hat = estimate_lambda(waiting_times, T)

    # 计算 log(λ(δ))
    log_lambda_hat_value = np.log(lambda_hat)

    # ========================== 校准 A 和 k =============================
    delta_max = np.max(spread_delta_snapshots)

    # 优化目标函数以找到最优的 A 和 k
    result = minimize(objective_function, np.array([A_initial, k_initial]),
                      args=(delta_max, log_lambda_hat_value, log_integral_phi_value),
                      bounds=bounds)

    if result.success:
        A_initial, k_initial = result.x
        # logger.info(f"Optimal k: {k_initial}, delta_max: {delta_max}")
    else:
        logger.error(result)

    # ========================== 计算 σ^2 ==========================
    sigma_squared = calculate_sigma_squared(S_values, order_book_timestamps)

    # ========================== 计算 δ^a + δ^b ==========================
    gamma = 1.0
    T_minus_t = 1.0
    delta_sum = calculate_delta_sum(gamma, sigma_squared, T_minus_t, k_initial)
    logger.info(f"δ^a + δ^b: {delta_sum}")