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

ib: be symcache compat by using bypass attr 

Since there's no easy way to support it yet, we bypass symbology caching
in for now and instead allow the `ib.ledger` routines to fill in
`MktPair` and `Asset` entries ad-hoc for the purposes of txn ledger
processing.
account_tests
Tyler Goodlet 2023-07-17 17:31:34 -04:00
parent a05a82486d
commit 9e87b6515b
3 changed files with 263 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
piker/brokers/ib/__init__.py
View File

@ -30,18 +30,27 @@ from .api import (
)
from .feed import (
open_history_client,
open_symbol_search,
stream_quotes,
get_mkt_info,
open_symbol_search,
)
from .broker import (
open_trade_dialog,
)
from .ledger import (
norm_trade,
norm_trade_records,
)
# TODO:
# from .symbols import (
# get_mkt_info,
# open_symbol_search,
# )
__all__ = [
'get_client',
'get_mkt_info',
'norm_trade',
'norm_trade_records',
'open_trade_dialog',
'open_history_client',
@ -75,3 +84,8 @@ _spawn_kwargs = {
# know if ``brokerd`` should be spawned with
# ``tractor``'s aio mode.
_infect_asyncio: bool = True
# XXX NOTE: for now we disable symcache with this backend since
# there is no clearly simple nor practical way to download "all
# symbology info" for all supported venues..
_no_symcache: bool = True

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

@ -60,9 +60,13 @@ from piker.accounting import (
open_trade_ledger,
TransactionLedger,
iter_by_dt,
open_pps,
open_account,
Account,
)
from piker.data._symcache import (
open_symcache,
SymbologyCache,
)
from piker.clearing._messages import (
Order,
Status,
@ -295,6 +299,10 @@ async def update_ledger_from_api_trades(
client: Union[Client, MethodProxy],
accounts_def_inv: bidict[str, str],
# provided for ad-hoc insertions "as transactions are
# processed"
symcache: SymbologyCache | None = None,
) -> tuple[
dict[str, Transaction],
dict[str, dict],
@ -325,7 +333,7 @@ async def update_ledger_from_api_trades(
# pack in the ``Contract.secType``
entry['asset_type'] = condict['secType']
entries = api_trades_to_ledger_entries(
entries: dict[str, dict] = api_trades_to_ledger_entries(
accounts_def_inv,
trade_entries,
)
@ -334,7 +342,10 @@ async def update_ledger_from_api_trades(
for acctid, trades_by_id in entries.items():
# normalize to transaction form
trans_by_acct[acctid] = norm_trade_records(trades_by_id)
trans_by_acct[acctid] = norm_trade_records(
trades_by_id,
symcache=symcache,
)
return trans_by_acct, entries
@ -547,11 +558,11 @@ async def open_trade_dialog(
) -> AsyncIterator[dict[str, Any]]:
# from piker.brokers import (
# get_brokermod,
# )
accounts_def = config.load_accounts(['ib'])
# TODO: do this as part of `open_account()`!?
from piker.data._symcache import open_symcache
global _client_cache
# deliver positions to subscriber before anything else
@ -565,12 +576,14 @@ async def open_trade_dialog(
proxies,
aioclients,
),
# TODO: do this as part of `open_account()`!?
open_symcache('ib', only_from_memcache=True) as symcache,
):
# 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] = {}
ledgers: dict[str, TransactionLedger] = {}
tables: dict[str, Account] = {}
order_msgs: list[Status] = []
conf = get_config()
@ -617,7 +630,7 @@ async def open_trade_dialog(
# positions reported by ib's sys that may not yet be in
# piker's ``pps.toml`` state-file.
tables[acctid] = lstack.enter_context(
open_pps(
open_account(
'ib',
acctid,
write_on_exit=True,
@ -640,7 +653,10 @@ async def open_trade_dialog(
# update position table with latest ledger from all
# gathered transactions: ledger file + api records.
trans: dict[str, Transaction] = norm_trade_records(ledger)
trans: dict[str, Transaction] = norm_trade_records(
ledger,
symcache=symcache,
)
# update trades ledgers for all accounts from connected
# api clients which report trades for **this session**.
@ -655,6 +671,7 @@ async def open_trade_dialog(
api_trades,
proxy,
accounts_def_inv,
symcache=symcache,
)
# if new api_trades are detected from the API, prepare
@ -797,7 +814,11 @@ async def emit_pp_update(
acnts: dict[str, Account],
) -> None:
'''
Extract trade record from an API event, convert it into a `Transaction`,
update the backing ledger and finally emit a position update to the EMS.
'''
accounts_def_inv: bidict[str, str] = accounts_def.inverse
accnum: str = trade_entry['execution']['acctNumber']
fq_acctid: str = accounts_def_inv[accnum]

77
piker/brokers/ib/ledger.py
View File

@ -28,6 +28,10 @@ from typing import (
from bidict import bidict
import pendulum
from piker.data import (
Struct,
SymbologyCache,
)
from piker.accounting import (
Asset,
dec_digits,
@ -39,20 +43,19 @@ from ._flex_reports import parse_flex_dt
from ._util import log
def norm_trade_records(
ledger: dict[str, Any],
def norm_trade(
tid: str,
record: dict[str, Any],
) -> dict[str, Transaction]:
'''
Normalize (xml) flex-report or (recent) API trade records into
our ledger format with parsing for `MktPair` and `Asset`
extraction to fill in the `Transaction.sys: MktPair` field.
# this is the dict that was returned from
# `Client.get_mkt_pairs()` and when running offline ledger
# processing from `.accounting`, this will be the table loaded
# into `SymbologyCache.pairs`.
pairs: dict[str, Struct],
symcache: SymbologyCache | None = None,
'''
# select: list[transactions] = []
records: list[Transaction] = []
) -> Transaction | None:
for tid, record in ledger.items():
conid = record.get('conId') or record['conid']
comms = record.get('commission')
if comms is None:
@ -90,7 +93,7 @@ def norm_trade_records(
# strike = tail[7:]
print(f'skipping opts contract {symbol}')
continue
return None
# timestamping is way different in API records
dtstr = record.get('datetime')
@ -188,7 +191,7 @@ def norm_trade_records(
# NOTE: can't serlialize `tomlkit.String` so cast to native
atype: str = str(dst.atype)
pair = MktPair(
mkt = MktPair(
bs_mktid=str(conid),
dst=dst,
@ -210,7 +213,14 @@ def norm_trade_records(
_fqme_without_src=(atype != 'fiat'),
)
fqme: str = pair.fqme
fqme: str = mkt.fqme
# XXX: if passed in, we fill out the symcache ad-hoc in order
# to make downstream accounting work..
if symcache:
symcache.mktmaps[fqme] = mkt
symcache.assets[src.name] = src
symcache.assets[dst.name] = dst
# NOTE: for flex records the normal fields for defining an fqme
# sometimes won't be available so we rely on two approaches for
@ -222,7 +232,7 @@ def norm_trade_records(
# should already have entries if the pps are still open, in
# which case, we can pull the fqme from that table (see
# `trades_dialogue()` above).
trans = Transaction(
return Transaction(
fqme=fqme,
tid=tid,
size=size,
@ -232,9 +242,40 @@ def norm_trade_records(
expiry=expiry,
bs_mktid=str(conid),
)
def norm_trade_records(
ledger: dict[str, Any],
symcache: SymbologyCache | None = None,
) -> dict[str, Transaction]:
'''
Normalize (xml) flex-report or (recent) API trade records into
our ledger format with parsing for `MktPair` and `Asset`
extraction to fill in the `Transaction.sys: MktPair` field.
'''
# select: list[transactions] = []
records: list[Transaction] = []
for tid, record in ledger.items():
txn = norm_trade(
tid,
record,
# NOTE: currently no symcache support
pairs={},
symcache=symcache,
)
if txn is None:
continue
insort(
records,
trans,
txn,
key=lambda t: t.dt
)
@ -258,14 +299,14 @@ def api_trades_to_ledger_entries(
# instead of pre-casting to dicts?
trade_entries: list[dict],
) -> dict:
) -> dict[str, dict]:
'''
Convert API execution objects entry objects into ``dict`` form,
pretty much straight up without modification except add
a `pydatetime` field from the parsed timestamp.
'''
trades_by_account = {}
trades_by_account: dict[str, dict] = {}
for t in trade_entries:
# NOTE: example of schema we pull from the API client.
# {