Как завершить app.run() после завершения запроса reqHistoricalTicks или других?

Question

Я хотел бы иметь программу на Python для запроса исторических данных о тиках из IBapi, выполнить некоторые вычисления после получения всех запрошенных тиков и выйти из программы.

Тем не менее, IBapi EClient.run() является бесконечным циклом, как я могу существовать цикл после получения всех исторических данных тиков?

Спасибо.

Answer 1

Вы не показали никакого кода, так что вот отправная точка. Это просит contractDetails и закрывается, когда вызывается метод end wrapper.

/questions/4797338/kak-poluchit-informatsiyu-o-kontrakte-iz-api-interactive-brokers/4797349#4797349

Что вы должны сделать, это запросить исторические данные один раз nextValidId Получено, то есть вы знаете, что соединение установлено и готово к передаче данных.

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

Если есть проблема, напишите свой код и добавьте комментарий для меня.

Answer 2

IB не умеют документировать это. Они показывают пример кода без должного объяснения того, что вам нужно использовать потоки. Ниже приведен образец кода, который должен решить вашу проблему и открыть совершенно новые возможности. Важные заметки:

app.run() фактически выполняется в новой функции и выполняется как отдельный поток.

Это означает, что после вызова для запуска потока ваш код переходит как обычно (то есть за пределы app.run), и вы можете нормально кодировать и выполнять вызовы TWS API. Ваши ответы будут подняты в отдельной ветке, где происходит обратный звонок. Вы можете объявлять переменные в классе обратного вызова EWrapper в функции init, я объявил для вас 3 примера. Затем они могут быть ссылками вне класса EWrapper в вашем основном коде, ссылаясь на них, например, как app.data или app.example_flag и т. Д.

Это означает, что вы можете заполнить результаты обратных вызовов, такие как reqHistoricalData, в ваших собственных переменных и использовать их в основном теле. Вот как это должно быть сделано, но документы IB очень плохи и в основном этого не показывают. Они пропускают поток, и это не работает, вы можете сделать только один звонок, и тогда вы застряли!

Маркус.

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
import threading
import time

class IBapi(EWrapper, EClient):

    def __init__(self):
        EClient.__init__(self, self)
        # Below I have created three sample vars. These can be accessed in the main
        # code outside this app / class by replacing self with app.varname 
        self.data = []
        self.other_example_array = []
        self.example_flag = 0

    def historicalData(self, reqId, bar):
        print(f'Time: {bar.date} Close: {bar.close}')

def run_loop():
    app.run()

app = IBapi()
app.connect('127.0.0.1', 7497, 123)

#Start the socket in a thread

api_thread = threading.Thread(target=run_loop, daemon=True)
api_thread.start()

time.sleep(1) #Sleep interval to allow time for connection to server

#Create contract object
eurusd_contract = Contract()
eurusd_contract.symbol = 'EUR'
eurusd_contract.secType = 'CASH'
eurusd_contract.exchange = 'IDEALPRO'
eurusd_contract.currency = 'USD'

#Request historical candles
app.reqHistoricalData(1, eurusd_contract, '', '2 D', '1 hour', 'BID', 0, 2, False, [])

time.sleep(5) #sleep to allow enough time for data to be returned
app.disconnect()

Answer 3

Спасибо вам обоим, я в самом начале пути обучения. Уже есть запрос на загрузку исторических данных. Требуются исторические данные в CSV-файле. Я хочу, чтобы этот процесс выполнялся в фоновом режиме, обновляя свои исторические данные в формате CSV каждую минуту. Так что его можно использовать для учебы. Я бы хотел, чтобы он работал непрерывно на заднем плане, пока не был остановлен. Нет, мне нужно обновлять его вручную, поэтому я хотел бы автоматизировать это, чтобы освободить свой разум и руки.

Искренне спасибо ...

      from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
from threading import Timer

def print_to_file(*args):
    with open('mnq.CsV', 'a') as fh:
        fh.write(' '.join(map(str,args)))
        fh.write('\n')
        fh.close()
print = print_to_file


class TestApp(EWrapper, EClient):
    def __init__(self):
      EClient.__init__(self, self)

      open('mnq.csv', 'w')
     # Layout = "{!s:1} {!s:2} {!s:3} {!s:4}}  "
     # print(Layout.format("DateTime", "High;", "Close;", "Volume "))

    def historicalData(self, reqId, bar):
          print( bar.date.replace(' ', ''),";", bar.high,";", bar.low,";", bar.volume)

    def stop(self):
      self.done = True
      self.disconnect()

def main():
    app = TestApp()
   # app. netxtOrderIdetxtOrderId =0
    app.connect("127.0.0.1", 7497, 0)

    # define MNQ
    contract = Contract()
    contract.symbol = "MNQ"
    contract.secType = "FUT"
    contract.exchange = "GLOBEX"
    contract.currency = "USD"
    contract.lastTradeDateOrContractMonth = "202103"

    app.reqHistoricalData(1, contract, "", "86400 S", "1 min", "TRADES", 0, 1, False, [])
    Timer(4, app.stop).start()
    app.run()

if __name__ == "__main__":
    main()