682 lines
18 KiB
Python
682 lines
18 KiB
Python
# piker: trading gear for hackers
|
|
# Copyright (C) Tyler Goodlet (in stewardship for piker0)
|
|
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
|
|
"""
|
|
Position info and display
|
|
|
|
"""
|
|
from __future__ import annotations
|
|
from dataclasses import dataclass
|
|
from functools import partial
|
|
from math import floor, copysign
|
|
from typing import Optional
|
|
|
|
|
|
from PyQt5.QtWidgets import QStyle
|
|
from PyQt5.QtGui import QIcon
|
|
from pyqtgraph import functions as fn
|
|
|
|
from ._annotate import LevelMarker
|
|
from ._anchors import (
|
|
pp_tight_and_right, # wanna keep it straight in the long run
|
|
gpath_pin,
|
|
)
|
|
from ..calc import humanize, pnl
|
|
from ..clearing._allocate import Allocator, Position
|
|
from ..data._normalize import iterticks
|
|
from ..data.feed import Feed
|
|
from ._label import Label
|
|
from ._lines import LevelLine, order_line
|
|
from ._style import _font
|
|
from ._forms import FieldsForm, FillStatusBar, QLabel
|
|
from ..log import get_logger
|
|
|
|
log = get_logger(__name__)
|
|
_pnl_tasks: dict[str, bool] = {}
|
|
|
|
|
|
async def display_pnl(
|
|
|
|
feed: Feed,
|
|
order_mode: OrderMode, # noqa
|
|
|
|
) -> None:
|
|
'''Real-time display the current pp's PnL in the appropriate label.
|
|
|
|
``ValueError`` if this task is spawned where there is a net-zero pp.
|
|
|
|
'''
|
|
global _pnl_tasks
|
|
|
|
pp = order_mode.current_pp
|
|
live = pp.live_pp
|
|
key = live.symbol.key
|
|
|
|
if live.size < 0:
|
|
types = ('ask', 'last', 'last', 'utrade')
|
|
|
|
elif live.size > 0:
|
|
types = ('bid', 'last', 'last', 'utrade')
|
|
|
|
else:
|
|
raise RuntimeError('No pp?!?!')
|
|
|
|
# real-time update pnl on the status pane
|
|
try:
|
|
async with feed.stream.subscribe() as bstream:
|
|
# last_tick = time.time()
|
|
async for quotes in bstream:
|
|
|
|
# now = time.time()
|
|
# period = now - last_tick
|
|
|
|
for sym, quote in quotes.items():
|
|
|
|
for tick in iterticks(quote, types):
|
|
# print(f'{1/period} Hz')
|
|
|
|
size = order_mode.current_pp.live_pp.size
|
|
if size == 0:
|
|
# terminate this update task since we're
|
|
# no longer in a pp
|
|
order_mode.pane.pnl_label.format(pnl=0)
|
|
return
|
|
|
|
else:
|
|
# compute and display pnl status
|
|
order_mode.pane.pnl_label.format(
|
|
pnl=copysign(1, size) * pnl(
|
|
# live.avg_price,
|
|
order_mode.current_pp.live_pp.avg_price,
|
|
tick['price'],
|
|
),
|
|
)
|
|
|
|
# last_tick = time.time()
|
|
finally:
|
|
assert _pnl_tasks[key]
|
|
assert _pnl_tasks.pop(key)
|
|
|
|
|
|
# https://www.pythonguis.com/faq/built-in-qicons-pyqt/
|
|
# account status icons taken from built-in set
|
|
_icon_names: dict[str, str] = {
|
|
# 'connected': 'SP_DialogYesButton',
|
|
'unavailable': 'SP_DockWidgetCloseButton',
|
|
'long_pp': 'SP_TitleBarShadeButton',
|
|
'short_pp': 'SP_TitleBarUnshadeButton',
|
|
'ready': 'SP_MediaPlay',
|
|
}
|
|
_icons: dict[str, QIcon] = {}
|
|
|
|
|
|
|
|
@dataclass
|
|
class SettingsPane:
|
|
'''Composite set of widgets plus an allocator model for configuring
|
|
order entry sizes and position limits per tradable instrument.
|
|
|
|
'''
|
|
# input fields
|
|
form: FieldsForm
|
|
|
|
# output fill status and labels
|
|
fill_bar: FillStatusBar
|
|
|
|
step_label: QLabel
|
|
pnl_label: QLabel
|
|
limit_label: QLabel
|
|
|
|
# encompasing high level namespace
|
|
order_mode: Optional['OrderMode'] = None # typing: ignore # noqa
|
|
|
|
def update_accounts_icon(
|
|
self,
|
|
status: str, # one of the values in ``_icons`` above
|
|
keys: Optional[list[str]] = None,
|
|
|
|
) -> None:
|
|
|
|
form = self.form
|
|
|
|
global _icons, _icon_name
|
|
if not _icons:
|
|
# load account selection using current style
|
|
for name, icon_name in _icon_names.items():
|
|
pixmap = getattr(QStyle, icon_name)
|
|
|
|
# TODO: set color to our default pp color
|
|
# https://stackoverflow.com/questions/13350631/simple-color-fill-qicons-in-qt
|
|
# mask = pixmap.createMaskFromColor(
|
|
# QColor('transparent'),
|
|
# Qt.MaskOutColor
|
|
# )
|
|
# pixmap.fill((QColor(hcolor('default_light'))))
|
|
# pixmap.setMask(mask)
|
|
|
|
_icons[name] = form.style().standardIcon(pixmap)
|
|
|
|
acct_select = form.fields['account']
|
|
keys = list(keys or acct_select._items.keys())
|
|
|
|
# breakpoint()
|
|
for key in keys:
|
|
i = acct_select._items[key]
|
|
icon = _icons[status]
|
|
acct_select.setItemIcon(i, icon)
|
|
|
|
def on_selection_change(
|
|
self,
|
|
|
|
text: str,
|
|
key: str,
|
|
|
|
) -> None:
|
|
'''Called on any order pane drop down selection change.
|
|
|
|
'''
|
|
log.info(f'selection input: {text}')
|
|
self.on_ui_settings_change(key, text)
|
|
|
|
def on_ui_settings_change(
|
|
self,
|
|
|
|
key: str,
|
|
value: str,
|
|
|
|
) -> bool:
|
|
'''Called on any order pane edit field value change.
|
|
|
|
'''
|
|
mode = self.order_mode
|
|
|
|
# an account switch request
|
|
if key == 'account':
|
|
|
|
# hide details on the old selection
|
|
old_tracker = mode.current_pp
|
|
old_tracker.hide_info()
|
|
|
|
# re-assign the order mode tracker
|
|
account_name = value
|
|
tracker = mode.trackers.get(account_name)
|
|
|
|
# if selection can't be found (likely never discovered with
|
|
# a ``brokerd`) then error and switch back to the last
|
|
# selection.
|
|
if tracker is None:
|
|
sym = old_tracker.chart.linked.symbol.key
|
|
log.error(
|
|
f'Account `{account_name}` can not be set for {sym}'
|
|
)
|
|
self.form.fields['account'].setCurrentText(
|
|
old_tracker.alloc.account_name())
|
|
return
|
|
|
|
self.order_mode.current_pp = tracker
|
|
assert tracker.alloc.account_name() == account_name
|
|
self.form.fields['account'].setCurrentText(account_name)
|
|
tracker.show()
|
|
tracker.hide_info()
|
|
|
|
self.display_pnl(tracker)
|
|
|
|
# load the new account's allocator
|
|
alloc = tracker.alloc
|
|
|
|
else:
|
|
tracker = mode.current_pp
|
|
alloc = tracker.alloc
|
|
|
|
size_unit = alloc.size_unit
|
|
|
|
# WRITE any settings to current pp's allocator
|
|
if key == 'limit':
|
|
if size_unit == 'currency':
|
|
alloc.currency_limit = float(value)
|
|
else:
|
|
alloc.units_limit = float(value)
|
|
|
|
elif key == 'slots':
|
|
alloc.slots = int(value)
|
|
|
|
elif key == 'size_unit':
|
|
# TODO: if there's a limit size unit change re-compute
|
|
# the current settings in the new units
|
|
alloc.size_unit = value
|
|
|
|
elif key != 'account':
|
|
raise ValueError(f'Unknown setting {key}')
|
|
|
|
# READ out settings and update UI
|
|
log.info(f'settings change: {key}: {value}')
|
|
|
|
suffix = {'currency': ' $', 'units': ' u'}[size_unit]
|
|
limit = alloc.limit()
|
|
|
|
# TODO: a reverse look up from the position to the equivalent
|
|
# account(s), if none then look to user config for default?
|
|
self.update_status_ui(pp=tracker)
|
|
|
|
step_size, currency_per_slot = alloc.step_sizes()
|
|
|
|
if size_unit == 'currency':
|
|
step_size = currency_per_slot
|
|
|
|
self.step_label.format(
|
|
step_size=str(humanize(step_size)) + suffix
|
|
)
|
|
self.limit_label.format(
|
|
limit=str(humanize(limit)) + suffix
|
|
)
|
|
|
|
# update size unit in UI
|
|
self.form.fields['size_unit'].setCurrentText(
|
|
alloc._size_units[alloc.size_unit]
|
|
)
|
|
self.form.fields['slots'].setText(str(alloc.slots))
|
|
self.form.fields['limit'].setText(str(limit))
|
|
|
|
# TODO: maybe return a diff of settings so if we can an error we
|
|
# can have general input handling code to report it through the
|
|
# UI in some way?
|
|
return True
|
|
|
|
def update_status_ui(
|
|
self,
|
|
|
|
pp: PositionTracker,
|
|
|
|
) -> None:
|
|
|
|
alloc = pp.alloc
|
|
slots = alloc.slots
|
|
used = alloc.slots_used(pp.live_pp)
|
|
|
|
# calculate proportion of position size limit
|
|
# that exists and display in fill bar
|
|
# TODO: what should we do for fractional slot pps?
|
|
self.fill_bar.set_slots(
|
|
slots,
|
|
|
|
# TODO: how to show "partial" slots?
|
|
# min(round(prop * slots), slots)
|
|
min(used, slots)
|
|
)
|
|
|
|
def display_pnl(
|
|
self,
|
|
tracker: PositionTracker,
|
|
|
|
) -> bool:
|
|
'''Display the PnL for the current symbol and personal positioning (pp).
|
|
|
|
If a position is open start a background task which will
|
|
real-time update the pnl label in the settings pane.
|
|
|
|
'''
|
|
mode = self.order_mode
|
|
sym = mode.chart.linked.symbol
|
|
size = tracker.live_pp.size
|
|
feed = mode.quote_feed
|
|
global _pnl_tasks
|
|
|
|
if (
|
|
size and
|
|
sym.key not in _pnl_tasks
|
|
):
|
|
_pnl_tasks[sym.key] = True
|
|
|
|
# immediately compute and display pnl status from last quote
|
|
self.pnl_label.format(
|
|
pnl=copysign(1, size) * pnl(
|
|
tracker.live_pp.avg_price,
|
|
# last historical close price
|
|
feed.shm.array[-1][['close']][0],
|
|
),
|
|
)
|
|
|
|
log.info(
|
|
f'Starting pnl display for {tracker.alloc.account_name()}')
|
|
self.order_mode.nursery.start_soon(
|
|
display_pnl,
|
|
feed,
|
|
mode,
|
|
)
|
|
return True
|
|
|
|
else:
|
|
# set 0% pnl
|
|
self.pnl_label.format(pnl=0)
|
|
return False
|
|
|
|
|
|
def position_line(
|
|
|
|
chart: 'ChartPlotWidget', # noqa
|
|
size: float,
|
|
level: float,
|
|
color: str,
|
|
|
|
orient_v: str = 'bottom',
|
|
marker: Optional[LevelMarker] = None,
|
|
|
|
) -> LevelLine:
|
|
'''Convenience routine to create a line graphic representing a "pp"
|
|
aka the acro for a,
|
|
"{piker, private, personal, puny, <place your p-word here>} position".
|
|
|
|
If ``marker`` is provided it will be configured appropriately for
|
|
the "direction" of the position.
|
|
|
|
'''
|
|
line = order_line(
|
|
chart,
|
|
level,
|
|
|
|
# TODO: could we maybe add a ``action=None`` which
|
|
# would be a mechanism to check a marker was passed in?
|
|
|
|
color=color,
|
|
highlight_on_hover=False,
|
|
movable=False,
|
|
hide_xhair_on_hover=False,
|
|
only_show_markers_on_hover=False,
|
|
always_show_labels=False,
|
|
|
|
# explicitly disable ``order_line()`` factory's creation
|
|
# of a level marker since we do it in this tracer thing.
|
|
show_markers=False,
|
|
)
|
|
|
|
if marker:
|
|
# configure marker to position data
|
|
|
|
if size > 0: # long
|
|
style = '|<' # point "up to" the line
|
|
elif size < 0: # short
|
|
style = '>|' # point "down to" the line
|
|
|
|
marker.style = style
|
|
|
|
# set marker color to same as line
|
|
marker.setPen(line.currentPen)
|
|
marker.setBrush(fn.mkBrush(line.currentPen.color()))
|
|
marker.level = level
|
|
marker.update()
|
|
marker.show()
|
|
|
|
# show position marker on view "edge" when out of view
|
|
vb = line.getViewBox()
|
|
vb.sigRangeChanged.connect(marker.position_in_view)
|
|
|
|
line.set_level(level)
|
|
|
|
return line
|
|
|
|
|
|
class PositionTracker:
|
|
'''Track and display real-time positions for a single symbol
|
|
over multiple accounts on a single chart.
|
|
|
|
Graphically composed of a level line and marker as well as labels
|
|
for indcating current position information. Updates are made to the
|
|
corresponding "settings pane" for the chart's "order mode" UX.
|
|
|
|
'''
|
|
# inputs
|
|
chart: 'ChartPlotWidget' # noqa
|
|
|
|
alloc: Allocator
|
|
startup_pp: Position
|
|
live_pp: Position
|
|
|
|
# allocated
|
|
pp_label: Label
|
|
size_label: Label
|
|
line: Optional[LevelLine] = None
|
|
|
|
_color: str = 'default_lightest'
|
|
|
|
def __init__(
|
|
self,
|
|
chart: 'ChartPlotWidget', # noqa
|
|
alloc: Allocator,
|
|
startup_pp: Position,
|
|
|
|
) -> None:
|
|
|
|
self.chart = chart
|
|
|
|
self.alloc = alloc
|
|
self.startup_pp = startup_pp
|
|
self.live_pp = startup_pp.copy()
|
|
|
|
view = chart.getViewBox()
|
|
|
|
# literally the 'pp' (pee pee) label that's always in view
|
|
self.pp_label = pp_label = Label(
|
|
view=view,
|
|
fmt_str='pp',
|
|
color=self._color,
|
|
update_on_range_change=False,
|
|
)
|
|
|
|
# create placeholder 'up' level arrow
|
|
self._level_marker = None
|
|
self._level_marker = self.level_marker(size=1)
|
|
|
|
pp_label.scene_anchor = partial(
|
|
gpath_pin,
|
|
gpath=self._level_marker,
|
|
label=pp_label,
|
|
)
|
|
pp_label.render()
|
|
|
|
self.size_label = size_label = Label(
|
|
view=view,
|
|
color=self._color,
|
|
|
|
# this is "static" label
|
|
# update_on_range_change=False,
|
|
fmt_str='\n'.join((
|
|
':{slots_used:.1f}x',
|
|
)),
|
|
|
|
fields={
|
|
'slots_used': 0,
|
|
},
|
|
)
|
|
size_label.render()
|
|
|
|
size_label.scene_anchor = partial(
|
|
pp_tight_and_right,
|
|
label=self.pp_label,
|
|
)
|
|
|
|
@property
|
|
def pane(self) -> FieldsForm:
|
|
'''Return handle to pp side pane form.
|
|
|
|
'''
|
|
return self.chart.linked.godwidget.pp_pane
|
|
|
|
def update_graphics(
|
|
self,
|
|
marker: LevelMarker
|
|
|
|
) -> None:
|
|
'''Update all labels.
|
|
|
|
Meant to be called from the maker ``.paint()``
|
|
for immediate, lag free label draws.
|
|
|
|
'''
|
|
self.pp_label.update()
|
|
self.size_label.update()
|
|
|
|
def update_from_pp(
|
|
self,
|
|
position: Optional[Position] = None,
|
|
|
|
) -> None:
|
|
'''Update graphics and data from average price and size passed in our
|
|
EMS ``BrokerdPosition`` msg.
|
|
|
|
'''
|
|
# live pp updates
|
|
pp = position or self.live_pp
|
|
|
|
self.update_line(
|
|
pp.avg_price,
|
|
pp.size,
|
|
self.chart.linked.symbol.lot_size_digits,
|
|
)
|
|
|
|
# label updates
|
|
self.size_label.fields['slots_used'] = round(
|
|
self.alloc.slots_used(pp), ndigits=1)
|
|
self.size_label.render()
|
|
|
|
if pp.size == 0:
|
|
self.hide()
|
|
|
|
else:
|
|
self._level_marker.level = pp.avg_price
|
|
|
|
# these updates are critical to avoid lag on view/scene changes
|
|
self._level_marker.update() # trigger paint
|
|
self.pp_label.update()
|
|
self.size_label.update()
|
|
|
|
self.show()
|
|
|
|
# don't show side and status widgets unless
|
|
# order mode is "engaged" (which done via input controls)
|
|
self.hide_info()
|
|
|
|
def level(self) -> float:
|
|
if self.line:
|
|
return self.line.value()
|
|
else:
|
|
return 0
|
|
|
|
def show(self) -> None:
|
|
if self.live_pp.size:
|
|
self.line.show()
|
|
self.line.show_labels()
|
|
|
|
self._level_marker.show()
|
|
self.pp_label.show()
|
|
self.size_label.show()
|
|
|
|
def hide(self) -> None:
|
|
self.pp_label.hide()
|
|
self._level_marker.hide()
|
|
self.size_label.hide()
|
|
if self.line:
|
|
self.line.hide()
|
|
|
|
def hide_info(self) -> None:
|
|
'''Hide details (right now just size label?) of position.
|
|
|
|
'''
|
|
self.size_label.hide()
|
|
if self.line:
|
|
self.line.hide_labels()
|
|
|
|
# TODO: move into annoate module
|
|
def level_marker(
|
|
self,
|
|
size: float,
|
|
|
|
) -> LevelMarker:
|
|
|
|
if self._level_marker:
|
|
self._level_marker.delete()
|
|
|
|
# arrow marker
|
|
# scale marker size with dpi-aware font size
|
|
font_size = _font.font.pixelSize()
|
|
|
|
# scale marker size with dpi-aware font size
|
|
arrow_size = floor(1.375 * font_size)
|
|
|
|
if size > 0:
|
|
style = '|<'
|
|
|
|
elif size < 0:
|
|
style = '>|'
|
|
|
|
arrow = LevelMarker(
|
|
chart=self.chart,
|
|
style=style,
|
|
get_level=self.level,
|
|
size=arrow_size,
|
|
on_paint=self.update_graphics,
|
|
)
|
|
|
|
self.chart.getViewBox().scene().addItem(arrow)
|
|
arrow.show()
|
|
|
|
return arrow
|
|
|
|
def update_line(
|
|
self,
|
|
price: float,
|
|
size: float,
|
|
size_digits: int,
|
|
|
|
) -> None:
|
|
'''Update personal position level line.
|
|
|
|
'''
|
|
# do line update
|
|
line = self.line
|
|
|
|
if size:
|
|
if line is None:
|
|
|
|
# create and show a pp line
|
|
line = self.line = position_line(
|
|
chart=self.chart,
|
|
level=price,
|
|
size=size,
|
|
color=self._color,
|
|
marker=self._level_marker,
|
|
)
|
|
|
|
else:
|
|
|
|
line.set_level(price)
|
|
self._level_marker.level = price
|
|
self._level_marker.update()
|
|
|
|
# update LHS sizing label
|
|
line.update_labels({
|
|
'size': size,
|
|
'size_digits': size_digits,
|
|
'fiat_size': round(price * size, ndigits=2),
|
|
|
|
# TODO: per account lines on a single (or very related) symbol
|
|
'account': self.alloc.account_name(),
|
|
})
|
|
line.show()
|
|
|
|
elif line: # remove pp line from view if it exists on a net-zero pp
|
|
line.delete()
|
|
self.line = None
|