Websocket Threading

Question

Ниже приведен код для получения живых тиков с помощью WebSocket.При каждом получении тика вызывается функция обратного вызова on_ticks (), которая будет печатать тики.

Могу ли я создать один поток в функции on_ticks () и вызвать функцию store_ticks () для сохранения тиков в базе данных?если да, может кто-нибудь помочь и показать, как это можно сделать?Или есть какой-то другой способ вызвать функцию store_ticks () и сохранять тики при каждом получении тиков?

from kiteconnect import KiteTicker

kws = KiteTicker("your_api_key", "your_access_token")

def on_ticks(ws, ticks):
    print(ticks)

def on_connect(ws, response):
    # Callback on successful connect.
    # Subscribe to a list of instrument_tokens 
    ws.subscribe([738561, 5633])

def store_ticks():
    # Store ticks here

def on_close(ws, code, reason):
    # On connection close stop the main loop
    # Reconnection will not happen after executing `ws.stop()`
    ws.stop()

# Assign the callbacks.
kws.on_ticks = on_ticks
kws.on_connect = on_connect
kws.on_close = on_close


kws.connect()

Answer 1

Если причина, по которой вы хотите создать новый поток, состоит в том, чтобы избежать задержек, я бы сказал, не беспокойтесь.

Я использовал mysql-client (MySQLDB-коннектор) с сервером MariaDB,подписался на 100+ инструментов в полном режиме, в течение последних 2 месяцев, и не было никаких задержек при записи тиков в БД.

Кроме того, мы не знаем, когда и сколько тиков мы получим один размы запускаем тикер. Это затрудняет время / счет и закрывает поток и соединение с БД.Может закончиться тем, что исчерпал лимит соединения и резьбы # очень быстро.(Пул соединений с БД в данном случае является излишним)

Причина, по которой я использую коннектор MySQLDB, а не pymysql - я видел увеличение времени записи примерно на 20% при использовании pymsql.Это не было бы очевидно в живых тиках.Я клонировал БД среднего размера (1 мельница + строки), выгружал ее в Dataframe в python, а затем записывал ее построчно в другую БД, и стенд отмечал результат в течение 10 итераций.

Причина, по которой я использую MariaDB- все возможности корпоративной версии MySQL без суеты Oracle.

Просто убедитесь, что вы установили достаточный объем памяти для используемого вами сервера БД.Это создает передышку для буфера БД на всякий случай.Избегание удаленного сервера и использование локального сервера также очень помогает.Если вы хотите выполнить резервное копирование данных из локальной резервной копии в облачную, вы можете настроить ежедневное задание для выгрузки в локальную папку, экспорта в облако и загрузки в базу данных там

Если вы ищете полное развертывание, это page уже имеет пример, а также кодовый фрагмент видео.

Answer 2

Вы можете изменить вашу store_ticks() функцию на

def store_ticks(ticks):
    # code to store tick into database

, а затем изменить вашу on_ticks функцию на:

def on_ticks(ws, ticks):
    print(ticks)
    store_ticks(ticks)

Что входит в store_ticks(ticks), зависит от того, чтобазу данных, которую вы хотите использовать, и какую именно информацию вы хотите хранить там.

РЕДАКТИРОВАТЬ:Чтобы создать новый поток для store_ticks (), используйте модуль _thread:

import _thread

def store_ticks(ticks):
    # code to store tick into database

def on_ticks(ticks):
    print(ticks)
    try:
        _thread.start_new_thread(store_ticks, (ticks,))
    except:
        # unable to start the thread, probably want some logging here