Merge pull request #421 from pikers/ib_contract_updates

`ib` futes contract consolidation fixes

piker/brokers/ib/api.py

@@ -167,35 +167,35 @@ _futes_venues = (
 _adhoc_futes_set = {
 
     # equities
-    'nq.globex',
-    'mnq.globex',  # micro
+    'nq.cme',
+    'mnq.cme',  # micro
 
-    'es.globex',
-    'mes.globex',  # micro
+    'es.cme',
+    'mes.cme',  # micro
 
     # cypto$
-    'brr.cmecrypto',
-    'ethusdrr.cmecrypto',
+    'brr.cme',
+    'ethusdrr.cme',
 
     # agriculture
-    'he.nymex',  # lean hogs
-    'le.nymex',  # live cattle (geezers)
-    'gf.nymex',  # feeder cattle (younguns)
+    'he.comex',  # lean hogs
+    'le.comex',  # live cattle (geezers)
+    'gf.comex',  # feeder cattle (younguns)
 
     # raw
-    'lb.nymex',  # random len lumber
+    'lb.comex',  # random len lumber
 
     # metals
     # https://misc.interactivebrokers.com/cstools/contract_info/v3.10/index.php?action=Conid%20Info&wlId=IB&conid=69067924
     'xauusd.cmdty',  # london gold spot ^
-    'gc.nymex',
-    'mgc.nymex',  # micro
+    'gc.comex',
+    'mgc.comex',  # micro
 
     # oil & gas
-    'cl.nymex',
+    'cl.comex',
 
     'xagusd.cmdty',  # silver spot
-    'ni.nymex',  # silver futes
+    'ni.comex',  # silver futes
     'qi.comex',  # mini-silver futes
 }
 
@@ -1031,7 +1031,7 @@ def con2fqsn(
         case ibis.Forex() | ibis.Contract(secType='CASH'):
             dst, src = con.localSymbol.split('.')
             symbol = ''.join([dst, src])
-            suffix = con.exchange
+            suffix = con.exchange or 'idealpro'
 
             # no real volume on forex feeds..
             calc_price = True
@@ -1053,7 +1053,10 @@ def con2fqsn(
     if expiry:
         suffix += f'.{expiry}'
 
-    fqsn_key = '.'.join((symbol, suffix)).lower()
+    fqsn_key = symbol.lower()
+    if suffix:
+        fqsn_key = '.'.join((fqsn_key, suffix)).lower()
+
     _cache[con.conId] = fqsn_key, calc_price
     return fqsn_key, calc_price
 

piker/brokers/ib/feed.py

@@ -130,6 +130,11 @@ async def open_history_client(
         mean: float = 0
         count: int = 0
 
+        head_dt: None | datetime = None
+        if (
+            # fx cons seem to not provide this endpoint?
+            'idealpro' not in fqsn
+        ):
             head_dt = await proxy.get_head_time(fqsn=fqsn)
 
         async def get_hist(
@@ -170,7 +175,9 @@ async def open_history_client(
                 )
 
             if (
-                end_dt and end_dt <= head_dt
+                end_dt
+                and head_dt
+                and end_dt <= head_dt
             ):
                 raise DataUnavailable(f'First timestamp is {head_dt}')
 
@@ -895,7 +902,9 @@ async def stream_quotes(
                         # last = time.time()
                         async for ticker in stream:
                             quote = normalize(ticker)
-                            await send_chan.send({quote['fqsn']: quote})
+                            fqsn = quote['fqsn']
+                            # print(f'sending {fqsn}:\n{quote}')
+                            await send_chan.send({fqsn: quote})
 
                             # ugh, clear ticks since we've consumed them
                             ticker.ticks = []