moshferatu/ibkr
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Trivial change to introduce more type hints

Browse Source
This commit is contained in:
moshferatu 2024-02-18 06:09:37 -08:00
parent d95269cf4f
commit 46ca62ac73
1 changed files with 11 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
ibkr/client.py
View File

@ -1,7 +1,7 @@
import pandas as pd import pandas as pd
from datetime import datetime from datetime import datetime
from ib_insync import ComboLeg, Contract, IB, Index, Option, Order, Trade from ib_insync import ComboLeg, Contract, IB, Index, Option, Order, Ticker, Trade
from ib_insync.util import isNan from ib_insync.util import isNan
from random import randint from random import randint
from typing import Callable, List from typing import Callable, List
@ -14,24 +14,24 @@ from .order_action import BUY, SELL
class Client: class Client:
def __init__(self, host: str = '127.0.0.1', port: int = 7497, client_id = randint(1, 10000)) -> None: def __init__(self, host: str = '127.0.0.1', port: int = 7497, client_id: int = randint(1, 10000)) -> None:
self.ib = IB() self.ib = IB()
self.ib.connect(host, port, clientId = client_id) self.ib.connect(host, port, clientId = client_id)
self.ib.reqMarketDataType(LIVE) self.ib.reqMarketDataType(LIVE)
def get_ticker(self, symbol: str, exchange: str): def get_ticker(self, symbol: str, exchange: str) -> Ticker:
underlying = Index(symbol, exchange) underlying = Index(symbol, exchange)
self.ib.qualifyContracts(underlying) self.ib.qualifyContracts(underlying)
return self.ib.reqTickers(underlying)[0] return self.ib.reqTickers(underlying)[0]
def get_market_data(self, contract: Contract, streaming: bool = False): def get_market_data(self, contract: Contract, streaming: bool = False) -> Ticker:
market_data = self.ib.reqMktData(contract, '', market_data = self.ib.reqMktData(contract, '',
snapshot = not streaming, regulatorySnapshot = False) snapshot = not streaming, regulatorySnapshot = False)
while isNan(market_data.bid) or isNan(market_data.ask): while isNan(market_data.bid) or isNan(market_data.ask):
# TODO: Add a timeout. # TODO: Add a timeout.
self.ib.sleep() self.ib.sleep()
return market_data return market_data
def get_option_chain(self, symbol: str, expiration: datetime, sub_symbol: str = None, def get_option_chain(self, symbol: str, expiration: datetime, sub_symbol: str = None,
contract_filter: Callable = None) -> pd.DataFrame: contract_filter: Callable = None) -> pd.DataFrame:
expiration_date = expiration.strftime('%Y%m%d') expiration_date = expiration.strftime('%Y%m%d')
@ -78,8 +78,8 @@ class Client:
option_contract.tradingClass = option_leg.sub_symbol option_contract.tradingClass = option_leg.sub_symbol
self.ib.qualifyContracts(option_contract) self.ib.qualifyContracts(option_contract)
return option_contract return option_contract
def submit_option_order(self, leg: OptionLeg, quantity: int): def submit_option_order(self, leg: OptionLeg, quantity: int) -> Trade:
option_contract = self.get_option_contract(leg) option_contract = self.get_option_contract(leg)
contract = Contract() contract = Contract()
contract.conId = option_contract.conId contract.conId = option_contract.conId
@ -124,7 +124,7 @@ class Client:
combo_order.transmit = True combo_order.transmit = True
return self.ib.placeOrder(combo_contract, combo_order) return self.ib.placeOrder(combo_contract, combo_order)
def submit_stop_loss_order(self, trade: Trade, stop_loss_price: float, limit_price: float = None) -> Trade: def submit_stop_loss_order(self, trade: Trade, stop_loss_price: float, limit_price: float = None) -> Trade:
stop_loss_order = Order() stop_loss_order = Order()
stop_loss_order.action = SELL if trade.order.action == BUY else BUY stop_loss_order.action = SELL if trade.order.action == BUY else BUY
@ -139,6 +139,6 @@ class Client:
stop_loss_order.totalQuantity = trade.order.totalQuantity stop_loss_order.totalQuantity = trade.order.totalQuantity
stop_loss_order.transmit = True stop_loss_order.transmit = True
return self.ib.placeOrder(trade.contract, stop_loss_order) return self.ib.placeOrder(trade.contract, stop_loss_order)
def run_event_loop(self): def run_event_loop(self):
self.ib.run() self.ib.run()