pikers
/
piker
3
0
Fork 0
Code Issues 4 Pull Requests 6 Packages Projects Releases Wiki Activity

Port `ib` broker machinery to new ctx mngr pp api 

This drops the use of `pp.update_pps_conf()` (and friends) and instead
moves to using the context style `open_trade_ledger()` and `open_pps()`
managers for faster pp msg gen due to delayed file writing (which was
the main source update latency).

In order to make this work with potentially multiple accounts this also
uses an exit stack which loads each ledger / `pps.toml` into an account
id mapped `dict`; a POC for likely how we should implement some higher
level position manager api.
ib_native_data_hack
Tyler Goodlet 2022-07-21 10:06:18 -04:00
parent b9fbbeb44e
commit a741ed3161
1 changed files with 195 additions and 199 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

236
piker/brokers/ib/broker.py
View File

@ -18,6 +18,7 @@ Order and trades endpoints for use with ``piker``'s EMS.
""" """
from __future__ import annotations from __future__ import annotations
from contextlib import ExitStack
from dataclasses import asdict from dataclasses import asdict
from functools import partial from functools import partial
from pprint import pformat from pprint import pformat
@ -35,8 +36,8 @@ from trio_typing import TaskStatus
import tractor import tractor
from ib_insync.contract import ( from ib_insync.contract import (
Contract, Contract,
Option, # Option,
Forex, # Forex,
) )
from ib_insync.order import ( from ib_insync.order import (
Trade, Trade,
@ -47,11 +48,18 @@ from ib_insync.objects import (
Execution, Execution,
CommissionReport, CommissionReport,
) )
from ib_insync.objects import Position from ib_insync.objects import Position as IbPosition
import pendulum import pendulum
from piker import config from piker import config
from piker import pp from piker.pp import (
# update_pps_conf,
Position,
Transaction,
open_trade_ledger,
open_pps,
PpTable,
)
from piker.log import get_console_log from piker.log import get_console_log
from piker.clearing._messages import ( from piker.clearing._messages import (
BrokerdOrder, BrokerdOrder,
@ -66,7 +74,8 @@ from piker.data._source import Symbol
from .api import ( from .api import (
_accounts2clients, _accounts2clients,
# _adhoc_futes_set, # _adhoc_futes_set,
_adhoc_symbol_map, con2fqsn,
# _adhoc_symbol_map,
log, log,
get_config, get_config,
open_client_proxies, open_client_proxies,
@ -76,49 +85,12 @@ from .api import (
def pack_position( def pack_position(
pos: Position pos: IbPosition
) -> dict[str, Any]: ) -> dict[str, Any]:
con = pos.contract con = pos.contract
fqsn, calc_price = con2fqsn(con)
if isinstance(con, Option):
# TODO: option symbol parsing and sane display:
symbol = con.localSymbol.replace(' ', '')
else:
# TODO: lookup fqsn even for derivs.
symbol = con.symbol.lower()
# TODO: probably write a mofo exchange mapper routine since ib
# can't get it's shit together like, ever.
# try our best to figure out the exchange / venue
exch = (con.primaryExchange or con.exchange).lower()
if not exch:
if isinstance(con, Forex):
# bc apparently it's not in the contract obj?
exch = 'idealfx'
else:
# for wtv cucked reason some futes don't show their
# exchange (like CL.NYMEX) ...
entry = _adhoc_symbol_map.get(
con.symbol or con.localSymbol
)
if entry:
meta, kwargs = entry
cid = meta.get('conId')
if cid:
assert con.conId == meta['conId']
exch = meta['exchange']
assert exch, f'No clue:\n {con}'
fqsn = '.'.join((symbol, exch))
expiry = con.lastTradeDateOrContractMonth
if expiry:
fqsn += f'.{expiry}'
# TODO: options contracts into a sane format.. # TODO: options contracts into a sane format..
return ( return (
@ -305,12 +277,10 @@ async def update_ledger_from_api_trades(
client: Union[Client, MethodProxy], client: Union[Client, MethodProxy],
) -> tuple[ ) -> tuple[
dict[str, pp.Transaction], dict[str, Transaction],
dict[str, dict], dict[str, dict],
]: ]:
conf = get_config()
# XXX; ERRGGG.. # XXX; ERRGGG..
# pack in the "primary/listing exchange" value from a # pack in the "primary/listing exchange" value from a
# contract lookup since it seems this isn't available by # contract lookup since it seems this isn't available by
@ -331,39 +301,34 @@ async def update_ledger_from_api_trades(
entry['listingExchange'] = pexch entry['listingExchange'] = pexch
conf = get_config()
entries = trades_to_ledger_entries( entries = trades_to_ledger_entries(
conf['accounts'].inverse, conf['accounts'].inverse,
trade_entries, trade_entries,
) )
# write recent session's trades to the user's (local) ledger file. # normalize recent session's trades to the `Transaction` type
records: dict[str, pp.Transactions] = {} trans_by_acct: dict[str, dict[str, Transaction]] = {}
for acctid, trades_by_id in entries.items(): for acctid, trades_by_id in entries.items():
# normalize to transaction form # normalize to transaction form
records[acctid] = norm_trade_records(trades_by_id) trans_by_acct[acctid] = norm_trade_records(trades_by_id)
return records, entries return trans_by_acct, entries
async def update_and_audit_msgs( async def update_and_audit_msgs(
acctid: str, # no `ib.` prefix is required! acctid: str, # no `ib.` prefix is required!
pps: list[pp.Position], pps: list[Position],
cids2pps: dict[tuple[str, int], BrokerdPosition], cids2pps: dict[tuple[str, int], BrokerdPosition],
validate: bool = False, validate: bool = False,
) -> list[BrokerdPosition]: ) -> list[BrokerdPosition]:
msgs: list[BrokerdPosition] = [] msgs: list[BrokerdPosition] = []
# pps: dict[int, pp.Position] = {}
for p in pps: for p in pps:
bsuid = p.bsuid bsuid = p.bsuid
# build trade-session-actor local table
# of pps from unique symbol ids.
# pps[bsuid] = p
# retreive equivalent ib reported position message # retreive equivalent ib reported position message
# for comparison/audit versus the piker equivalent # for comparison/audit versus the piker equivalent
# breakeven pp calcs. # breakeven pp calcs.
@ -479,7 +444,7 @@ async def trades_dialogue(
client = aioclients[account] client = aioclients[account]
async def open_stream( async def open_trade_event_stream(
task_status: TaskStatus[ task_status: TaskStatus[
trio.abc.ReceiveChannel trio.abc.ReceiveChannel
] = trio.TASK_STATUS_IGNORED, ] = trio.TASK_STATUS_IGNORED,
@ -493,7 +458,7 @@ async def trades_dialogue(
task_status.started(trade_event_stream) task_status.started(trade_event_stream)
await trio.sleep_forever() await trio.sleep_forever()
trade_event_stream = await nurse.start(open_stream) trade_event_stream = await nurse.start(open_trade_event_stream)
clients.append((client, trade_event_stream)) clients.append((client, trade_event_stream))
@ -501,49 +466,73 @@ async def trades_dialogue(
accounts.add(account) accounts.add(account)
cids2pps: dict[str, BrokerdPosition] = {} cids2pps: dict[str, BrokerdPosition] = {}
update_records: dict[str, bidict] = {}
# Open a trade ledgers stack for appending trade records over
# multiple accounts.
# TODO: we probably want to generalize this into a "ledgers" api..
ledgers: dict[str, dict] = {}
tables: dict[str, PpTable] = {}
with (
ExitStack() as lstack,
):
# process pp value reported from ib's system. we only use these # process pp value reported from ib's system. we only use these
# to cross-check sizing since average pricing on their end uses # to cross-check sizing since average pricing on their end uses
# the so called (bs) "FIFO" style which more or less results in # the so called (bs) "FIFO" style which more or less results in
# a price that's not useful for traders who want to not lose # a price that's not useful for traders who want to not lose
# money.. xb # money.. xb
acctids = set()
for client in aioclients.values(): for client in aioclients.values():
for pos in client.positions(): for pos in client.positions():
cid, msg = pack_position(pos)
acctid = msg.account = accounts_def.inverse[msg.account]
acctid = acctid.strip('ib.')
cids2pps[(acctid, cid)] = msg
assert msg.account in accounts, (
f'Position for unknown account: {msg.account}')
# collect all ib-pp reported positions so that we can be # collect all ib-pp reported positions so that we can be
# sure know which positions to update from the ledger if # sure know which positions to update from the ledger if
# any are missing from the ``pps.toml`` # any are missing from the ``pps.toml``
update_records.setdefault(acctid, bidict())[cid] = msg.symbol cid, msg = pack_position(pos)
acctid = msg.account = accounts_def.inverse[msg.account]
acctid = acctid.strip('ib.')
acctids.add(acctid)
cids2pps[(acctid, cid)] = msg
assert msg.account in accounts, (
f'Position for unknown account: {msg.account}')
for acctid in acctids:
# open ledger and pptable wrapper for each
# detected account.
ledgers[acctid] = lstack.enter_context(
open_trade_ledger('ib', acctid)
)
tables[acctid] = lstack.enter_context(
open_pps('ib', acctid)
)
# update trades ledgers for all accounts from # update trades ledgers for all accounts from
# connected api clients which report trades for **this session**. # connected api clients which report trades for **this session**.
new_trades = {}
for account, proxy in proxies.items(): for account, proxy in proxies.items():
trades = await proxy.trades() trades = await proxy.trades()
( (
records_by_acct, trans_by_acct,
ledger_entries, ready_for_ledger_entries,
) = await update_ledger_from_api_trades( ) = await update_ledger_from_api_trades(
trades, trades,
proxy, proxy,
) )
new_trades.update(records_by_acct) acctid = acctid.strip('ib.')
ledgers[acctid].update(ready_for_ledger_entries)
for acctid, trans in new_trades.items(): # WTF, yet again this key error is getting ignored?!?!
for t in trans: # tables[acctid].update_from_trans(trans_by_acct[account])
bsuid = t.bsuid
if bsuid in update_records: # this causes a hang..
assert update_records[bsuid] == t.fqsn # - marketstored tries to kill container, cant,
else: # - ctrl-c makes pikerd get stuck...
update_records.setdefault(acctid, bidict())[bsuid] = t.fqsn # assert 0
trans = trans_by_acct.get(acctid)
if trans:
tables[acctid].update_from_trans(trans)
# load all positions from `pps.toml`, cross check with ib's # load all positions from `pps.toml`, cross check with ib's
# positions data, and relay re-formatted pps as msgs to the ems. # positions data, and relay re-formatted pps as msgs to the ems.
@ -553,15 +542,13 @@ async def trades_dialogue(
# - no new trades yet but we want to reload and audit any # - no new trades yet but we want to reload and audit any
# positions reported by ib's sys that may not yet be in # positions reported by ib's sys that may not yet be in
# piker's ``pps.toml`` state-file. # piker's ``pps.toml`` state-file.
for acctid, to_update in update_records.items(): for acctid in acctids:
trans = new_trades.get(acctid)
active, closed = pp.update_pps_conf( table = tables[acctid]
'ib', _, closed_pps = table.dump_active('ib')
acctid, active_pps = table.pps
trade_records=trans,
ledger_reload=to_update, for pps in [active_pps, closed_pps]:
)
for pps in [active, closed]:
msgs = await update_and_audit_msgs( msgs = await update_and_audit_msgs(
acctid, acctid,
pps.values(), pps.values(),
@ -593,15 +580,15 @@ async def trades_dialogue(
# write ledger with all new trades **AFTER** we've updated the # write ledger with all new trades **AFTER** we've updated the
# `pps.toml` from the original ledger state! # `pps.toml` from the original ledger state!
for acctid, trades_by_id in ledger_entries.items(): for acctid, trades_by_id in ready_for_ledger_entries.items():
with pp.open_trade_ledger('ib', acctid) as ledger: ledgers[acctid].update(trades_by_id)
ledger.update(trades_by_id)
async with ( async with (
ctx.open_stream() as ems_stream, ctx.open_stream() as ems_stream,
trio.open_nursery() as n, trio.open_nursery() as n,
): ):
# start order request handler **before** local trades event loop # start order request handler **before** local trades
# event loop
n.start_soon(handle_order_requests, ems_stream, accounts_def) n.start_soon(handle_order_requests, ems_stream, accounts_def)
# allocate event relay tasks for each client connection # allocate event relay tasks for each client connection
@ -613,6 +600,9 @@ async def trades_dialogue(
accounts_def, accounts_def,
cids2pps, cids2pps,
proxies, proxies,
ledgers,
tables,
) )
# block until cancelled # block until cancelled
@ -626,44 +616,44 @@ async def emit_pp_update(
proxies: dict, proxies: dict,
cids2pps: dict, cids2pps: dict,
ledgers,
tables,
) -> None: ) -> None:
# compute and relay incrementally updated piker pp # compute and relay incrementally updated piker pp
acctid = accounts_def.inverse[trade_entry['execution']['acctNumber']] acctid = accounts_def.inverse[trade_entry['execution']['acctNumber']]
proxy = proxies[acctid] proxy = proxies[acctid]
acctname = acctid.strip('ib.') acctid = acctid.strip('ib.')
records_by_acct, ledger_entries = await update_ledger_from_api_trades( (
records_by_acct,
ready_for_ledger_entries,
) = await update_ledger_from_api_trades(
[trade_entry], [trade_entry],
proxy, proxy,
) )
records = records_by_acct[acctname] trans = records_by_acct[acctid]
r = records[0] r = list(trans.values())[0]
# update and load all positions from `pps.toml`, cross check with table = tables[acctid]
# ib's positions data, and relay re-formatted pps as msgs to the table.update_from_trans(trans)
# ems. we report both the open and closed updates in one map since _, closed = table.dump_active('ib')
# for incremental update we may have just fully closed a pp and need active = table.pps
# to relay that msg as well!
active, closed = pp.update_pps_conf(
'ib',
acctname,
trade_records=records,
ledger_reload={r.bsuid: r.fqsn},
)
# NOTE: write ledger with all new trades **AFTER** we've updated the # NOTE: update ledger with all new trades
# `pps.toml` from the original ledger state! for acctid, trades_by_id in ready_for_ledger_entries.items():
for acctid, trades_by_id in ledger_entries.items(): ledger = ledgers[acctid]
with pp.open_trade_ledger('ib', acctid) as ledger:
ledger.update(trades_by_id) ledger.update(trades_by_id)
# generate pp msgs and cross check with ib's positions data, relay
# re-formatted pps as msgs to the ems.
for pos in filter( for pos in filter(
bool, bool,
[active.get(r.bsuid), closed.get(r.bsuid)] [active.get(r.bsuid), closed.get(r.bsuid)]
): ):
msgs = await update_and_audit_msgs( msgs = await update_and_audit_msgs(
acctname, acctid,
[pos], [pos],
cids2pps, cids2pps,
@ -685,6 +675,9 @@ async def deliver_trade_events(
cids2pps: dict[tuple[str, str], BrokerdPosition], cids2pps: dict[tuple[str, str], BrokerdPosition],
proxies: dict[str, MethodProxy], proxies: dict[str, MethodProxy],
ledgers,
tables,
) -> None: ) -> None:
''' '''
Format and relay all trade events for a given client to emsd. Format and relay all trade events for a given client to emsd.
@ -834,6 +827,8 @@ async def deliver_trade_events(
accounts_def, accounts_def,
proxies, proxies,
cids2pps, cids2pps,
ledgers,
tables,
) )
case 'cost': case 'cost':
@ -866,6 +861,8 @@ async def deliver_trade_events(
accounts_def, accounts_def,
proxies, proxies,
cids2pps, cids2pps,
ledgers,
tables,
) )
case 'error': case 'error':
@ -916,14 +913,13 @@ async def deliver_trade_events(
def norm_trade_records( def norm_trade_records(
ledger: dict[str, Any], ledger: dict[str, Any],
) -> list[pp.Transaction]: ) -> list[Transaction]:
''' '''
Normalize a flex report or API retrieved executions Normalize a flex report or API retrieved executions
ledger into our standard record format. ledger into our standard record format.
''' '''
records: list[pp.Transaction] = [] records: dict[str, Transaction] = {}
for tid, record in ledger.items(): for tid, record in ledger.items():
conid = record.get('conId') or record['conid'] conid = record.get('conId') or record['conid']
@ -1001,7 +997,7 @@ def norm_trade_records(
# which case, we can pull the fqsn from that table (see # which case, we can pull the fqsn from that table (see
# `trades_dialogue()` above). # `trades_dialogue()` above).
records.append(pp.Transaction( records[tid] = Transaction(
fqsn=fqsn, fqsn=fqsn,
tid=tid, tid=tid,
size=size, size=size,
@ -1010,7 +1006,7 @@ def norm_trade_records(
dt=dt, dt=dt,
expiry=expiry, expiry=expiry,
bsuid=conid, bsuid=conid,
)) )
return records return records
@ -1151,7 +1147,7 @@ def load_flex_trades(
ledgers = {} ledgers = {}
for acctid, trades_by_id in trades_by_account.items(): for acctid, trades_by_id in trades_by_account.items():
with pp.open_trade_ledger('ib', acctid) as ledger: with open_trade_ledger('ib', acctid) as ledger:
ledger.update(trades_by_id) ledger.update(trades_by_id)
ledgers[acctid] = ledger ledgers[acctid] = ledger