Блог

Я создаю панель мониторинга для живых тикеров

Я использую zerodha Kiteconnect для получения текущих данных по опционам и фьючерсам SE для этого … Клиент, для которого я создаю панель мониторинга, предоставил мне ключ api, а также ежедневно предоставляет мне токен доступа …. Ситуация такова, что он использует те же учетные данные в своей собственной программе (которую он запускает отдельно на своем ноутбуке) для получения данных о реальном времени

Когда я использую KiteConnect для извлечения дампа инструмента, он работает, приведенный ниже код успешно выполняется,

   
   from kiteconnect import KiteTicker, KiteConnect

   access_token = '*********' # changes every day
   api_key = '*********'

   kite = KiteConnect(api_key=key_secret, access_token = access_token)
   instrument_list = kite.instruments(exchange=kite.EXCHANGE_NFO)

 

но когда я использую KiteTicker (потоковая передача WebSocket) с теми же учетными данными, как показано в приведенном ниже коде, это приводит к ошибке подключения 1006:

 kws = KiteTicker(api_key, access_token=kite.access_token)

####### define the callbacks ############

def on_connect(ws, response):
    # Callback on successful connect.
    # Subscribe to a list of instrument_tokens (RELIANCE and ACC here).
    ws.subscribe(instrument_tokens)

    # Set tick in `full` mode.
    ws.set_mode(ws.MODE_FULL, instrument_tokens)


def on_ticks(ws, ticks):
    # Callback to receive ticks.
    # logging.debug("Ticks: {}".format(ticks))
    print(ticks)
    ticks_list.append(ticks)

    # close the connection after some time
    if (time.time() - begin_time) > 120: # run for 2 minutes
        write_json(ticks_list)
        print("close called")
        ws.close()


def on_close(ws, code, reason):
    # On connection close stop the event loop.
    # Reconnection will not happen after executing `ws.stop()`
    print("-------- Stopping ------------")
    ws.stop()
    print("--------- Stopped ----------")


### define
ticks_list = [] # will hold list of JSON objects
##### Assign the callbacks.
kws.on_ticks = on_ticks
kws.on_connect = on_connect
kws.on_close = on_close

begin_time = time.time()

# Infinite loop on the main thread. Nothing after this will run.
# You have to use the pre-defined callbacks to manage subscriptions.
kws.connect()

 

Точная ошибка, вызванная:

Ошибка подключения: 1006 — соединение было закрыто нечисто (я сбросил TCP-соединение WebSocket: причина закрытия без кода закрытия)

Не могли бы вы, пожалуйста, объяснить мне, почему это происходит? Также возможно ли использовать одни и те же учетные данные параллельно двумя разными программами с разных IP-адресов для получения данных о тиках в реальном времени…

Спасибо

Вы размещаете заказ внутри метода on_tick? вы не должны вкладывать какую-либо логику в поток on_tick.Вам нужно асинхронно передать галочку в другом методе, не блокируя поток on_tick.

Существует два способа передачи данных о тиках из потока on_tick для выполнения любой операции без блокировки основного потока on_ticks.
1> Вы можете отправлять данные о тиках в очередь(использовать сельдерей,rq и т. Д.) В качестве диспетчера очередей задач и использовать другой метод,который считывает эти данные очереди и выполняет задачи. например:

 def on_ticks(ws,ticks):
    #Pass the tick data to Queue using celery,rq,etc
    #Using celery delay method here to call helper_method task
    helper_method.delay(ticks)

def helper_method(ticks):
    #Perform require operation using tick data
 

2>Создайте другой поток и выполните необходимую операцию во 2-м потоке, используя threaded=True
P. S: Не забудьте назначить обратный вызов тикера для нового потока.

 import logging
from kiteconnect import KiteTicker

logging.basicConfig(level=logging.DEBUG)

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

def on_ticks(ws, ticks):
    logging.debug("Ticks: {}".format(ticks))

def on_connect(ws, response):
    ws.subscribe([738561, 5633])
    ws.set_mode(ws.MODE_FULL, [738561])

def on_close(ws, code, reason): 
    ws.stop()

kws.on_ticks = on_ticks
kws.on_connect = on_connect
kws.on_close = on_close
kws.connect(threaded=True)

while True:
   #Perform required data operation using tick data
   def on_ticks(ws, ticks):
       ..........
       helper_method(ticks)
       .........

   def helper_method(ticks):
       .........
       Perform computation here
       ........
  #Assign callback
  kws.on_ticks=on_t