如何使用 class 方法参数 (?) 作为代码输入 (IBKR / IB API / IB TWS) - Python
How to use class method parameter (?) as input in code (IBKR / IB API / IB TWS) - Python
(抱歉,如果 terminology/explanation 不正确)
我想打印“我的美元账户余额是:xxx”。目的是了解如何使用 class 方法参数(不确定这是否是正确的术语)作为代码输入
然而,结果是:“我的美元账户余额是:None”。已打印帐户余额,但在单独的一行上。
from ibapi.wrapper import EWrapper # handles incoming messages
from ibapi.contract import Contract
from ibapi.order import *
import threading
import time
class IBapi(EWrapper, EClient):
def __init__(self):
EClient.__init__(self, self)
self.contract_details = {}
self.bardata = {} # initialise directory to store bar data
self.USD_cash_balance = 0
def nextValidId(self, orderId: int):
super().nextValidId(orderId)
self.nextorderId = orderId
print('The next valid order id is: ', self.nextorderId)
def accountSummary(self, reqId: int, account: str, tag: str, value: str,
currency: str):
if tag == "CashBalance":
print(value)
IBapi.USD_cash_balance = value
if reqId == 131:
return value
def run_loop():
app.run() # starts communication with TWS
app = IBapi()
app.nextorderId = None
app.connect('127.0.0.1', 7497, 123)
# Start the socket in a thread
api_thread = threading.Thread(target=run_loop, daemon=True) # Algo doesn't have "daemon=True"
api_thread.start()
# Check if the API is connected via orderid
while True:
if isinstance(app.nextorderId,
int): # the IB API sends out the next available order ID as soon as connection is made
# The isinstance() function returns True if the specified object is of the specified type, otherwise False.
print('connected')
break
else:
print('waiting for connection')
time.sleep(2)
print("My account balance in USD is: " + str(app.reqAccountSummary(131, "All", "$LEDGER:USD")))
首先,您不将 USD_cash_balance 分配给 class,而是分配给 class 实例(使用 self 而不是 class 名称)。
其次,您必须等待回复,有很多方法可以做到这一点。一种是等待值并在其上休眠,如有必要,您可以通过实现 timeout/retry 来扩展它。您还可以使用队列。
另外,不要忘记在完成后断开会话。
from ibapi.client import EClient
from ibapi.wrapper import EWrapper # handles incoming messages
from ibapi.contract import Contract
from ibapi.order import *
import threading
import time
class IBapi(EWrapper, EClient):
def __init__(self):
EClient.__init__(self, self)
self.contract_details = {}
self.bardata = {} # initialise directory to store bar data
self.USD_cash_balance = 0
def nextValidId(self, orderId: int):
super().nextValidId(orderId)
self.nextorderId = orderId
print('The next valid order id is: ', self.nextorderId)
def accountSummary(self, reqId: int, account: str, tag: str, value: str,
currency: str):
if tag == "CashBalance" and reqId == 131:
self.USD_cash_balance = value
def run_loop():
app.run() # starts communication with TWS
app = IBapi()
app.nextorderId = None
app.connect('127.0.0.1', 7600, 123)
# Start the socket in a thread
api_thread = threading.Thread(target=run_loop, daemon=True) # Algo doesn't have "daemon=True"
api_thread.start()
# Check if the API is connected via orderid
while True:
if isinstance(app.nextorderId,
int): # the IB API sends out the next available order ID as soon as connection is made
# The isinstance() function returns True if the specified object is of the specified type, otherwise False.
print('connected')
break
else:
print('waiting for connection')
time.sleep(2)
while not getattr(app, 'USD_cash_balance', None):
app.reqAccountSummary(131, "All", "$LEDGER:USD")
time.sleep(0.5)
print("My account balance in USD is: " + app.USD_cash_balance)
app.disconnect()
(抱歉,如果 terminology/explanation 不正确) 我想打印“我的美元账户余额是:xxx”。目的是了解如何使用 class 方法参数(不确定这是否是正确的术语)作为代码输入 然而,结果是:“我的美元账户余额是:None”。已打印帐户余额,但在单独的一行上。
from ibapi.wrapper import EWrapper # handles incoming messages
from ibapi.contract import Contract
from ibapi.order import *
import threading
import time
class IBapi(EWrapper, EClient):
def __init__(self):
EClient.__init__(self, self)
self.contract_details = {}
self.bardata = {} # initialise directory to store bar data
self.USD_cash_balance = 0
def nextValidId(self, orderId: int):
super().nextValidId(orderId)
self.nextorderId = orderId
print('The next valid order id is: ', self.nextorderId)
def accountSummary(self, reqId: int, account: str, tag: str, value: str,
currency: str):
if tag == "CashBalance":
print(value)
IBapi.USD_cash_balance = value
if reqId == 131:
return value
def run_loop():
app.run() # starts communication with TWS
app = IBapi()
app.nextorderId = None
app.connect('127.0.0.1', 7497, 123)
# Start the socket in a thread
api_thread = threading.Thread(target=run_loop, daemon=True) # Algo doesn't have "daemon=True"
api_thread.start()
# Check if the API is connected via orderid
while True:
if isinstance(app.nextorderId,
int): # the IB API sends out the next available order ID as soon as connection is made
# The isinstance() function returns True if the specified object is of the specified type, otherwise False.
print('connected')
break
else:
print('waiting for connection')
time.sleep(2)
print("My account balance in USD is: " + str(app.reqAccountSummary(131, "All", "$LEDGER:USD")))
首先,您不将 USD_cash_balance 分配给 class,而是分配给 class 实例(使用 self 而不是 class 名称)。 其次,您必须等待回复,有很多方法可以做到这一点。一种是等待值并在其上休眠,如有必要,您可以通过实现 timeout/retry 来扩展它。您还可以使用队列。 另外,不要忘记在完成后断开会话。
from ibapi.client import EClient
from ibapi.wrapper import EWrapper # handles incoming messages
from ibapi.contract import Contract
from ibapi.order import *
import threading
import time
class IBapi(EWrapper, EClient):
def __init__(self):
EClient.__init__(self, self)
self.contract_details = {}
self.bardata = {} # initialise directory to store bar data
self.USD_cash_balance = 0
def nextValidId(self, orderId: int):
super().nextValidId(orderId)
self.nextorderId = orderId
print('The next valid order id is: ', self.nextorderId)
def accountSummary(self, reqId: int, account: str, tag: str, value: str,
currency: str):
if tag == "CashBalance" and reqId == 131:
self.USD_cash_balance = value
def run_loop():
app.run() # starts communication with TWS
app = IBapi()
app.nextorderId = None
app.connect('127.0.0.1', 7600, 123)
# Start the socket in a thread
api_thread = threading.Thread(target=run_loop, daemon=True) # Algo doesn't have "daemon=True"
api_thread.start()
# Check if the API is connected via orderid
while True:
if isinstance(app.nextorderId,
int): # the IB API sends out the next available order ID as soon as connection is made
# The isinstance() function returns True if the specified object is of the specified type, otherwise False.
print('connected')
break
else:
print('waiting for connection')
time.sleep(2)
while not getattr(app, 'USD_cash_balance', None):
app.reqAccountSummary(131, "All", "$LEDGER:USD")
time.sleep(0.5)
print("My account balance in USD is: " + app.USD_cash_balance)
app.disconnect()