Python WebSocket技术疑问：如何获取Ticks历史数据？

嘿，我懂你的困扰——你已经搞定了实时WebSocket ticks的获取，现在想在同一个活跃连接里发送历史数据请求对吧？其实这个操作很直观，只需要在现有代码里加几个小改动就行，我给你捋清楚怎么做：

1. 封装历史数据请求函数 #

先写一个专门的函数来构造并发送历史数据请求，复用起来更方便：

def send_ticks_history(ws):
    # 把你需要的请求负载封装成字典
    history_request = {
        "ticks_history": "R_50",
        "end": "latest",
        "start": 1,
        "style": "ticks",
        "adjust_start_time": 1,
        "count": 10
    }
    ws.send(json.dumps(history_request))

2. 区分处理不同类型的返回消息 #

原来的on_message只是打印所有消息，现在我们可以解析返回的JSON，区分实时ticks和历史数据响应，输出会更清晰：

def on_message(ws, message):
    message_data = json.loads(message)
    # 根据返回的关键字段区分消息类型
    if 'tick' in message_data:
        print('实时ticks更新:\n', json.dumps(message_data['tick'], indent=2))
    elif 'history' in message_data:
        print('历史数据返回:\n', json.dumps(message_data['history'], indent=2))
    else:
        print('系统消息:\n', message)

3. 在运行中的应用里触发请求 #

因为WebSocket是长连接，我们需要一个方式在运行时手动触发请求——比如加一个后台线程监听用户输入，这样你可以随时发送请求，还不会阻塞实时数据的接收：

if __name__ == "__main__":
    apiUrl = "wss://ws.binaryws.com/websockets/v3?app_id=1089"
    ws = websocket.WebSocketApp(apiUrl, on_message = on_message, on_open = on_open)
    
    # 启动后台线程处理用户输入
    import threading
    def input_handler():
        while True:
            cmd = input("\n输入 'history' 即可获取R_50的历史数据: ").strip().lower()
            if cmd == 'history':
                send_ticks_history(ws)
    
    threading.Thread(target=input_handler, daemon=True).start()
    
    ws.run_forever()

完整可运行代码 #

把所有部分整合起来，就是这个完整的版本：

import websocket
import json
import threading

def on_open(ws):
    # 连接建立后先发送实时ticks请求
    realtime_request = json.dumps({'ticks':'R_100'})
    ws.send(realtime_request)
    print("WebSocket连接已建立，输入 'history' 获取历史数据")

def on_message(ws, message):
    message_data = json.loads(message)
    # 区分不同类型的返回消息
    if 'tick' in message_data:
        print('实时ticks更新:\n', json.dumps(message_data['tick'], indent=2))
    elif 'history' in message_data:
        print('历史数据返回:\n', json.dumps(message_data['history'], indent=2))
    else:
        print('系统消息:\n', message)

def send_ticks_history(ws):
    # 构造历史数据请求负载
    history_request = {
        "ticks_history": "R_50",
        "end": "latest",
        "start": 1,
        "style": "ticks",
        "adjust_start_time": 1,
        "count": 10
    }
    ws.send(json.dumps(history_request))

if __name__ == "__main__":
    apiUrl = "wss://ws.binaryws.com/websockets/v3?app_id=1089"
    ws = websocket.WebSocketApp(apiUrl, on_message = on_message, on_open = on_open)
    
    # 后台线程处理用户输入，避免阻塞WebSocket主线程
    def input_thread():
        while True:
            user_input = input().strip().lower()
            if user_input == 'history':
                send_ticks_history(ws)
    
    threading.Thread(target=input_thread, daemon=True).start()
    
    ws.run_forever()

使用说明 #

运行代码后，控制台会提示连接已建立，这时候你输入history并回车，就能收到R_50的最近10条历史ticks数据，同时实时的R_100 ticks还会继续更新，互不影响。

如果你想在特定时机自动发送历史请求（比如连接建立后立刻发送），直接在on_open函数里调用send_ticks_history(ws)就行，不用那个输入线程也可以。

内容的提问来源于stack exchange，提问作者Jaffer Wilson