Support Deephaven 0.10.0 (#31)

* Change DynamicTableWriter from using logRow to logRowPermissive.

* Added missing error code.

* Use int64 instead of int32 for column types.

* Debug logging for deadlocks

* Rerequest IDs, in case they have not been sent.

* Debug logging for deadlocks

* Rerequest IDs, in case they have not been sent.

* Added thread names.

* Add stack traces for all threads.

* Generalizing the strategy for obtaining order ids.

* Make the strategy for getting order IDs selectable.

* Fixing bad import

* Fixing bad import

* Fixing bad import

* Fix broken enum

* Cleaned up debugging output

* Update docs

* Update error messages

* Add type annotation

* Added space to error message.

* Fixed thread name.

Author: chipkent

README.md

@@ -144,7 +144,7 @@ See [Access your file system with Docker data volumes](https://deephaven.io/core
 
 Follow these steps to run a [Deephaven](https://deephaven.io) plus [Interactive Brokers](https://interactivebrokers.com) system. 
 
-`<deephaven_version>` is the version of [Deephaven](https://deephaven.io) to run (e.g., `0.9.0`).  A list of available versions 
+`<deephaven_version>` is the version of [Deephaven](https://deephaven.io) to run (e.g., `0.10.0`).  A list of available versions 
 can be found on the [Deephaven Releases GitHub page](https://github.com/deephaven/deephaven-core/releases).
 
 **Windows users need to run the commands in WSL.**
@@ -219,6 +219,11 @@ of Docker, `host` should be set to `host.docker.internal`.
 communicates on.  This value can be found in the [IB Trader Workstation (TWS)](https://www.interactivebrokers.com/en/trading/tws.php)
 settings.  By default, production trading uses port 7496, and paper trading uses port 7497.  See [Setup](#setup) and [TWS Initial Setup](https://interactivebrokers.github.io/tws-api/initial_setup.html) for more details.
 
+`order_id_strategy` is the strategy used for obtaining new order ids.
+* `OrderIdStrategy.RETRY` (default) - Request a new order ID from TWS every time one is needed.  Retry if TWS does not respond quickly.  This usually avoids a TWS bug where it does not always respond.
+* `OrderIdStrategy.BASIC` - Request a new order ID from TWS every time one is needed.  Does not retry, so it may deadlock if TWS does not respond.
+* `OrderIdStrategy.INCREMENT` - Use the initial order ID sent by TWS and increment the value upon every request.  This is fast, but it may fail for multiple, concurrent sessions connected to TWS.
+
 ```python
 import deephaven_ib as dhib
 

src/deephaven_ib/__init__.py

@@ -8,10 +8,11 @@
 
 from ._query_inputs import *
 from ._tws import IbTwsClient
+from ._tws.order_id_queue import OrderIdStrategy
 from .time import dh_to_ib_datetime
 
-__all__ = ["MarketDataType", "TickDataType", "BarDataType", "BarSize", "Duration", "Request", "RegisteredContract",
-           "IbSessionTws"]
+__all__ = ["MarketDataType", "TickDataType", "BarDataType", "BarSize", "Duration", "OrderIdStrategy",
+           "Request", "RegisteredContract", "IbSessionTws"]
 
 
 class MarketDataType(Enum):
@@ -317,6 +318,8 @@ class IbSessionTws:
 
             **NOTE: Each client MUST connect with a unique clientId.**
         download_short_rates (bool): True to download a short rates table.
+        order_id_strategy (OrderIdStrategy): strategy for obtaining new order ids.
+
 
     Tables:
         ####
@@ -390,11 +393,11 @@ class IbSessionTws:
     _tables_raw: Dict[str, Any]  # TODO: should be Dict[str, Table] with deephaven v2
     _tables: Dict[str, Any]  # TODO: should be Dict[str, Table] with deephaven v2
 
-    def __init__(self, host: str = "", port: int = 7497, client_id: int = 0, download_short_rates=True):
+    def __init__(self, host: str = "", port: int = 7497, client_id: int = 0, download_short_rates: bool = True, order_id_strategy: OrderIdStrategy = OrderIdStrategy.RETRY):
         self._host = host
         self._port = port
         self._client_id = client_id
-        self._client = IbTwsClient(download_short_rates=download_short_rates)
+        self._client = IbTwsClient(download_short_rates=download_short_rates, order_id_strategy=order_id_strategy)
         self._tables_raw = {f"raw_{k}": v for k, v in self._client.tables.items()}
         self._tables = dict(sorted(IbSessionTws._make_tables(self._tables_raw).items()))
 

src/deephaven_ib/_internal/error_codes.py

@@ -34,6 +34,7 @@ def load_error_codes() -> Tuple[Dict[int, str], Dict[int, str]]:
         502: "Couldn't connect to TWS. Confirm that 'Enable ActiveX and Socket EClients' is enabled and connection port is the same as 'Socket Port' on the TWS 'Edit->Global Configuration...->API->Settings' menu. Live Trading ports: TWS: 7496; IB Gateway: 4001. Simulated Trading ports for new installations of version 954.1 or newer:  TWS: 7497; IB Gateway: 4002",
         2113: "The order size for Bonds (Bills) is entered as a nominal par value of the order, and must be a multiple",
         10089: "Requested market data requires additional subscription for API.See link in 'Market Data Connections' dialog for more details.",
+        10172: "Failed to request news article: No data available",
         10187: "Failed to request historical ticks",
         10189: "Failed to request tick-by-tick data",
     }

src/deephaven_ib/_internal/tablewriter.py

@@ -51,7 +51,7 @@ def _check_logged_value_types(self, values: List) -> None:
                 continue
 
             if (t is dht.string and not isinstance(v, str)) or \
-                    (t is dht.int32 and not isinstance(v, int)) or \
+                    (t is dht.int64 and not isinstance(v, int)) or \
                     (t is dht.float64 and not isinstance(v, float)):
                 logging.error(
                     f"TableWriter column type and value type are mismatched: column_name={n} column_type={t} value_type={type(v)} value={v}\n{trace_str()}\n-----")
@@ -73,7 +73,7 @@ def write_row(self, values: List) -> None:
             if values[i] == "":
                 values[i] = None
 
-        self._dtw.logRow(values)
+        self._dtw.logRowPermissive(values)
 
 
 ArrayStringSet = jpy.get_type("io.deephaven.stringset.ArrayStringSet")

src/deephaven_ib/_internal/threading.py

@@ -24,7 +24,7 @@ def __init__(self, timeout_sec: float, sleep_sec: float):
         self.locks = {}
         self._lock = threading.Lock()
 
-        self._thread = threading.Thread(target=self._run, daemon=True)
+        self._thread = threading.Thread(name="DeadlockMonitor", target=self._run, daemon=True)
         self._thread.start()
         setattr(self, "deadlock_monitor_thread", self._thread)
 

src/deephaven_ib/_internal/trace.py

@@ -1,8 +1,27 @@
 """Functionality for working with stack traces."""
 
 import traceback
-
+import threading
+import sys
 
 def trace_str() -> str:
     """Gets a string of the current stacktrace."""
     return "".join(traceback.format_stack())
+
+
+def trace_thread_str(thread:threading.Thread) -> str:
+    """Gets a string of the stacktrace of a thread."""
+    return "".join(traceback.format_stack(sys._current_frames()[thread.ident]))
+
+
+def trace_all_threads_str() -> str:
+    """Get the stacktraces for all threads as a string."""
+
+    rst = "Stack Traces:\n"
+
+    for th in threading.enumerate():
+        rst += str(th)
+        rst += trace_thread_str(th)
+        rst += "\n"
+
+    return rst

src/deephaven_ib/_tws/ib_type_logger.py

@@ -81,7 +81,7 @@ def map_right(right: str) -> Union[str, None]:
         return right
 
     return [
-        ("ContractId", dht.int32, lambda contract: contract.conId),
+        ("ContractId", dht.int64, lambda contract: contract.conId),
         ("SecId", dht.string, lambda contract: contract.secId),
         ("SecIdType", dht.string, lambda contract: contract.secIdType),
         ("SecType", dht.string, lambda contract: contract.secType),
@@ -130,8 +130,8 @@ def map_sec_id_list(value):
         ("MinTick", dht.float64, lambda cd: cd.minTick),
         ("OrderTypes", dht.stringset, lambda cd: to_string_set(cd.orderTypes.split(","))),
         ("ValidExchanges", dht.stringset, lambda cd: to_string_set(cd.validExchanges.split(","))),
-        ("PriceMagnifier", dht.int32, lambda cd: cd.priceMagnifier),
-        ("UnderConId", dht.int32, lambda cd: cd.underConId),
+        ("PriceMagnifier", dht.int64, lambda cd: cd.priceMagnifier),
+        ("UnderConId", dht.int64, lambda cd: cd.underConId),
         ("LongName", dht.string, lambda cd: cd.longName),
         ("ContractMonth", dht.string, lambda cd: cd.contractMonth),
         ("Industry", dht.string, lambda cd: cd.industry),
@@ -141,9 +141,9 @@ def map_sec_id_list(value):
         ("TradingHours", dht.stringset, lambda cd: to_string_set(cd.tradingHours.split(";"))),
         ("LiquidHours", dht.stringset, lambda cd: to_string_set(cd.liquidHours.split(";"))),
         ("EvRule", dht.string, lambda cd: cd.evRule),
-        ("EvMultiplier", dht.int32, lambda cd: cd.evMultiplier),
-        ("MdSizeMultiplier", dht.int32, lambda cd: cd.mdSizeMultiplier),
-        ("AggGroup", dht.int32, lambda cd: map_null_int(cd.aggGroup)),
+        ("EvMultiplier", dht.int64, lambda cd: cd.evMultiplier),
+        ("MdSizeMultiplier", dht.int64, lambda cd: cd.mdSizeMultiplier),
+        ("AggGroup", dht.int64, lambda cd: map_null_int(cd.aggGroup)),
         ("UnderSymbol", dht.string, lambda cd: cd.underSymbol),
         ("UnderSecType", dht.string, lambda cd: cd.underSecType),
         ("MarketRuleIds", dht.stringset, lambda cd: to_string_set(cd.marketRuleIds.split(","))),
@@ -159,7 +159,7 @@ def map_sec_id_list(value):
         ("CouponType", dht.string, lambda cd: cd.couponType),
         ("Callable", dht.bool_, lambda cd: cd.callable),
         ("Putable", dht.bool_, lambda cd: cd.putable),
-        ("Coupon", dht.int32, lambda cd: cd.coupon),
+        ("Coupon", dht.int64, lambda cd: cd.coupon),
         ("Convertible", dht.bool_, lambda cd: cd.convertible),
         ("Maturity", dht.string, lambda cd: cd.maturity),
         # TODO: convert date time?  Values are not provided in TWS, and the format is not documented. (https://github.com/deephaven-examples/deephaven-ib/issues/10)
@@ -207,8 +207,8 @@ def map_null(val):
         ("High", dht.float64, lambda bd: bd.high),
         ("Low", dht.float64, lambda bd: bd.low),
         ("Close", dht.float64, lambda bd: bd.close),
-        ("Volume", dht.int32, lambda bd: map_null(bd.volume)),
-        ("BarCount", dht.int32, lambda bd: map_null(bd.barCount)),
+        ("Volume", dht.int64, lambda bd: map_null(bd.volume)),
+        ("BarCount", dht.int64, lambda bd: map_null(bd.barCount)),
         ("Average", dht.float64, lambda bd: map_null(bd.average)),
     ]
 
@@ -234,9 +234,9 @@ def map_null(val):
         ("High", dht.float64, lambda bd: bd.high),
         ("Low", dht.float64, lambda bd: bd.low),
         ("Close", dht.float64, lambda bd: bd.close),
-        ("Volume", dht.int32, lambda bd: map_null(bd.volume)),
+        ("Volume", dht.int64, lambda bd: map_null(bd.volume)),
         ("WAP", dht.float64, lambda bd: map_null(bd.wap)),
-        ("Count", dht.int32, lambda bd: map_null(bd.count)),
+        ("Count", dht.int64, lambda bd: map_null(bd.count)),
     ]
 
 
@@ -319,7 +319,7 @@ def map_special_conditions(special_conditions: str) -> Any:
     return [
         ("Timestamp", dht.datetime, lambda t: unix_sec_to_dh_datetime(t.time)),
         ("Price", dht.float64, lambda t: t.price),
-        ("Size", dht.int32, lambda t: t.size),
+        ("Size", dht.int64, lambda t: t.size),
         *_include_details(_details_tick_attrib_last(), lambda t: t.tickAttribLast),
         ("Exchange", dht.string, lambda t: t.exchange),
         ("SpecialConditions", dht.stringset, lambda t: map_special_conditions(t.specialConditions))
@@ -352,8 +352,8 @@ def _details_historical_tick_bid_ask() -> List[Tuple]:
         ("Timestamp", dht.datetime, lambda t: unix_sec_to_dh_datetime(t.time)),
         ("BidPrice", dht.float64, lambda t: t.priceBid),
         ("AskPrice", dht.float64, lambda t: t.priceAsk),
-        ("BidSize", dht.int32, lambda t: t.sizeBid),
-        ("AskSize", dht.int32, lambda t: t.sizeAsk),
+        ("BidSize", dht.int64, lambda t: t.sizeBid),
+        ("AskSize", dht.int64, lambda t: t.sizeAsk),
         *_include_details(_details_tick_attrib_bid_ask(), lambda t: t.tickAttribBidAsk),
     ]
 
@@ -384,9 +384,9 @@ def _details_order() -> List[Tuple]:
     return [
 
         # order identifier
-        ("OrderId", dht.int32, lambda o: o.orderId),
-        ("ClientId", dht.int32, lambda o: o.clientId),
-        ("PermId", dht.int32, lambda o: o.permId),
+        ("OrderId", dht.int64, lambda o: o.orderId),
+        ("ClientId", dht.int64, lambda o: o.clientId),
+        ("PermId", dht.int64, lambda o: o.permId),
 
         # main order fields
         ("Action", dht.string, lambda o: o.action),
@@ -403,18 +403,18 @@ def _details_order() -> List[Tuple]:
         ("OcaType", dht.string, lambda o: map_values(o.ocaType, oca_types)),
         ("OrderRef", dht.string, lambda o: o.orderRef),
         ("Transmit", dht.bool_, lambda o: o.transmit),
-        ("ParentId", dht.int32, lambda o: o.parentId),
+        ("ParentId", dht.int64, lambda o: o.parentId),
         ("BlockOrder", dht.bool_, lambda o: o.blockOrder),
         ("SweepToFill", dht.bool_, lambda o: o.sweepToFill),
-        ("DisplaySize", dht.int32, lambda o: o.displaySize),
+        ("DisplaySize", dht.int64, lambda o: o.displaySize),
         ("TriggerMethod", dht.string, lambda o: map_values(o.triggerMethod, trigger_methods)),
         ("OutsideRth", dht.bool_, lambda o: o.outsideRth),
         ("Hidden", dht.bool_, lambda o: o.hidden),
         ("GoodAfterTime", dht.string, lambda o: o.goodAfterTime),
         ("GoodTillDate", dht.string, lambda o: o.goodTillDate),
         ("Rule80A", dht.string, lambda o: map_values(o.rule80A, rule80_values)),
         ("AllOrNone", dht.bool_, lambda o: o.allOrNone),
-        ("MinQty", dht.int32, lambda o: o.minQty),
+        ("MinQty", dht.int64, lambda o: o.minQty),
         ("PercentOffset", dht.float64, lambda o: o.percentOffset),
         ("OverridePercentageConstraints", dht.bool_, lambda o: o.overridePercentageConstraints),
         ("TrailStopPrice", dht.float64, lambda o: o.trailStopPrice),
@@ -431,7 +431,7 @@ def _details_order() -> List[Tuple]:
         ("OpenClose", dht.string, lambda o: map_values(o.openClose, open_close_values)),
         ("Origin", dht.string, lambda o: map_values(o.origin, origin_values)),
         ("ShortSaleSlot", dht.string, lambda o: map_values(o.shortSaleSlot, short_sale_slot_values)),
-        ("ExemptCode", dht.int32, lambda o: o.exemptCode),
+        ("ExemptCode", dht.int64, lambda o: o.exemptCode),
 
         # SMART routing only
         ("DiscretionaryAmt", dht.float64, lambda o: o.discretionaryAmt),
@@ -458,31 +458,31 @@ def _details_order() -> List[Tuple]:
         ("VolatilityType", dht.string, lambda o: map_values(o.volatilityType, volatility_type)),
         ("DeltaNeutralOrderType", dht.string, lambda o: o.deltaNeutralOrderType),
         ("DeltaNeutralAuxPrice", dht.float64, lambda o: o.deltaNeutralAuxPrice),
-        ("DeltaNeutralConId", dht.int32, lambda o: o.deltaNeutralConId),
+        ("DeltaNeutralConId", dht.int64, lambda o: o.deltaNeutralConId),
         ("DeltaNeutralSettlingFirm", dht.string, lambda o: o.deltaNeutralSettlingFirm),
         ("DeltaNeutralClearingAccount", dht.string, lambda o: o.deltaNeutralClearingAccount),
         ("DeltaNeutralClearingIntent", dht.string, lambda o: o.deltaNeutralClearingIntent),
         ("DeltaNeutralOpenClose", dht.string, lambda o: o.deltaNeutralOpenClose),
         ("DeltaNeutralShortSale", dht.bool_, lambda o: o.deltaNeutralShortSale),
-        ("DeltaNeutralShortSaleSlot", dht.int32, lambda o: o.deltaNeutralShortSaleSlot),
+        ("DeltaNeutralShortSaleSlot", dht.int64, lambda o: o.deltaNeutralShortSaleSlot),
         ("DeltaNeutralDesignatedLocation", dht.string, lambda o: o.deltaNeutralDesignatedLocation),
         ("ContinuousUpdate", dht.bool_, lambda o: o.continuousUpdate),
         ("ReferencePriceType", dht.string, lambda o: map_values(o.referencePriceType, reference_price_type)),
 
         # COMBO ORDERS ONLY
         ("BasisPoints", dht.float64, lambda o: o.basisPoints),
-        ("BasisPointsType", dht.int32, lambda o: o.basisPointsType),
+        ("BasisPointsType", dht.int64, lambda o: o.basisPointsType),
 
         # SCALE ORDERS ONLY
-        ("ScaleInitLevelSize", dht.int32, lambda o: o.scaleInitLevelSize),
-        ("ScaleSubsLevelSize", dht.int32, lambda o: o.scaleSubsLevelSize),
+        ("ScaleInitLevelSize", dht.int64, lambda o: o.scaleInitLevelSize),
+        ("ScaleSubsLevelSize", dht.int64, lambda o: o.scaleSubsLevelSize),
         ("ScalePriceIncrement", dht.float64, lambda o: o.scalePriceIncrement),
         ("ScalePriceAdjustValue", dht.float64, lambda o: o.scalePriceAdjustValue),
-        ("ScalePriceAdjustInterval", dht.int32, lambda o: o.scalePriceAdjustInterval),
+        ("ScalePriceAdjustInterval", dht.int64, lambda o: o.scalePriceAdjustInterval),
         ("ScaleProfitOffset", dht.float64, lambda o: o.scaleProfitOffset),
         ("ScaleAutoReset", dht.bool_, lambda o: o.scaleAutoReset),
-        ("ScaleInitPosition", dht.int32, lambda o: o.scaleInitPosition),
-        ("ScaleInitFillQty", dht.int32, lambda o: o.scaleInitFillQty),
+        ("ScaleInitPosition", dht.int64, lambda o: o.scaleInitPosition),
+        ("ScaleInitFillQty", dht.int64, lambda o: o.scaleInitFillQty),
         ("ScaleRandomPercent", dht.bool_, lambda o: o.scaleRandomPercent),
         ("ScaleTable", dht.string, lambda o: o.scaleTable),
 
@@ -521,7 +521,7 @@ def _details_order() -> List[Tuple]:
         ("OrderMiscOptions", dht.stringset, lambda o: to_string_set(o.orderMiscOptions)),
 
         # VER PEG2BENCH fields:
-        ("ReferenceContractId", dht.int32, lambda o: o.referenceContractId),
+        ("ReferenceContractId", dht.int64, lambda o: o.referenceContractId),
         ("PeggedChangeAmount", dht.float64, lambda o: o.peggedChangeAmount),
         ("IsPeggedChangeAmountDecrease", dht.bool_, lambda o: o.isPeggedChangeAmountDecrease),
         ("ReferenceChangeAmount", dht.float64, lambda o: o.referenceChangeAmount),
@@ -532,7 +532,7 @@ def _details_order() -> List[Tuple]:
         ("AdjustedStopPrice", dht.float64, lambda o: o.adjustedStopPrice),
         ("AdjustedStopLimitPrice", dht.float64, lambda o: o.adjustedStopLimitPrice),
         ("AdjustedTrailingAmount", dht.float64, lambda o: o.adjustedTrailingAmount),
-        ("AdjustableTrailingUnit", dht.int32, lambda o: o.adjustableTrailingUnit),
+        ("AdjustableTrailingUnit", dht.int64, lambda o: o.adjustableTrailingUnit),
         ("LmtPriceOffset", dht.float64, lambda o: o.lmtPriceOffset),
 
         ("Conditions", dht.stringset, lambda o: to_string_set(o.conditions)),
@@ -558,12 +558,12 @@ def _details_order() -> List[Tuple]:
 
         ("AutoCancelDate", dht.string, lambda o: o.autoCancelDate),
         ("FilledQuantity", dht.float64, lambda o: o.filledQuantity),
-        ("RefFuturesConId", dht.int32, lambda o: o.refFuturesConId),
+        ("RefFuturesConId", dht.int64, lambda o: o.refFuturesConId),
         ("AutoCancelParent", dht.bool_, lambda o: o.autoCancelParent),
         ("Shareholder", dht.string, lambda o: o.shareholder),
         ("ImbalanceOnly", dht.bool_, lambda o: o.imbalanceOnly),
         ("RouteMarketableToBbo", dht.bool_, lambda o: o.routeMarketableToBbo),
-        ("ParentPermId", dht.int32, lambda o: o.parentPermId),
+        ("ParentPermId", dht.int64, lambda o: o.parentPermId),
 
         ("UsePriceMgmtAlgo", dht.bool_, lambda o: o.usePriceMgmtAlgo),
 
@@ -619,17 +619,17 @@ def _details_execution() -> List[Tuple]:
         ("Side", dht.string, lambda e: e.side),
         ("Shares", dht.float64, lambda e: e.shares),
         ("Price", dht.float64, lambda e: e.price),
-        ("PermId", dht.int32, lambda e: e.permId),
-        ("ClientId", dht.int32, lambda e: e.clientId),
-        ("OrderId", dht.int32, lambda e: e.orderId),
-        ("Liquidation", dht.int32, lambda e: e.liquidation),
+        ("PermId", dht.int64, lambda e: e.permId),
+        ("ClientId", dht.int64, lambda e: e.clientId),
+        ("OrderId", dht.int64, lambda e: e.orderId),
+        ("Liquidation", dht.int64, lambda e: e.liquidation),
         ("CumQty", dht.float64, lambda e: e.cumQty),
         ("AvgPrice", dht.float64, lambda e: e.avgPrice),
         ("OrderRef", dht.string, lambda e: e.orderRef),
         ("EvRule", dht.string, lambda e: e.evRule),
         ("EvMultiplier", dht.float64, lambda e: e.evMultiplier),
         ("ModelCode", dht.string, lambda e: e.modelCode),
-        ("LastLiquidity", dht.int32, lambda e: e.lastLiquidity),
+        ("LastLiquidity", dht.int64, lambda e: e.lastLiquidity),
     ]