Source code for tradeexecutor.strategy.output

"""Helpers for outputting strategy execution information to Python logging and Discord."""

from io import StringIO
from typing import List, Iterable

from tradeexecutor.state.portfolio import Portfolio
from tradeexecutor.state.position import TradingPosition
from import TradeExecution

#: See setup_discord_logging()

[docs]def format_trade(portfolio: Portfolio, trade: TradeExecution) -> List[str]: """Write a trade status line to logs. :return: List of log lines """ pair = trade.pair if pair.info_url: link = pair.info_url else: link = "" if trade.is_buy(): trade_type = "Buy" else: trade_type = "Sell" existing_position = portfolio.get_existing_open_position_by_trading_pair(trade.pair) if existing_position: # Quantity returns the total balance of unexecuted trades existing_balance = existing_position.get_quantity() amount = abs(trade.planned_quantity / existing_balance) existing_text = f", {amount*100:,.2f}% of existing position #{existing_position.position_id}" else: existing_text = "" existing_balance = 0 lines = [ f"{trade_type:5} #{trade.trade_id} {pair.get_human_description()} ${trade.get_planned_value():,.2f} ({abs(trade.get_position_quantity())} {pair.base.token_symbol}){existing_text}", ] # Add existing balance if existing_position: lines += [ f"Exiting position: {existing_position}, with pre-trade balance: {existing_balance}" ] else: lines += [ f"This trade will open a new position" ] if link: lines.append(f"Trading pair link: {link}") return lines
[docs]def format_position(position: TradingPosition, up_symbol="🌲", down_symbol="🔻") -> List[str]: """Write a position status line to logs. Position can be open/closed. :return: List of log lines """ symbol = up_symbol if position.get_total_profit_percent() >= 0 else down_symbol if position.pair.info_url: link = position.pair.info_url else: link = "" base_token_ticker = position.pair.base.token_symbol position_kind = position.pair.kind.value lines =[ f"{symbol} #{position.position_id} {position.pair.get_human_description()} {position_kind} value: ${position.get_value():,.2f}, size: {abs(position.get_quantity()):,.4f}, {base_token_ticker} profit: {(position.get_total_profit_percent()*100):.2f}% ({position.get_total_profit_usd():,.4f} USD)" ] if position.has_executed_trades(): price_diff = position.get_current_price() - position.get_opening_price() lines.append(f" current price: ${position.get_current_price():,.8f}, opened: ${position.get_opening_price():,.8f}, diff: {price_diff:,.8f} USD") lines.append(f" last tx: {position.get_last_tx_hash()}") if position.is_frozen(): last_trade = "buy" if position.get_last_trade().is_buy() else "sell" lines.append(f" last trade: {last_trade}, freeze reason: {position.get_freeze_reason()}") if link: lines.append(f" link: {link}") return lines
[docs]def output_positions(positions: Iterable[TradingPosition], buf: StringIO, empty_message="No positions", break_after=4): """Write info on multiple trading positions formatted for Python logging system. :break_after: Insert Discord message break after this many positions to avoid chopped Discord messages. :return: A plain text string as a log message, suitable for Discord logging """ positions = list(positions) if len(positions) > 0: position: TradingPosition for idx, position in enumerate(positions): for line in format_position(position): print(" " + line, file=buf) if (idx + 1) % break_after == 0: # Discord message break print(DISCORD_BREAK_CHAR, file=buf) else: # Line break print("", file=buf) else: print(f" {empty_message}", file=buf) return buf.getvalue()
[docs]def output_trades(trades: List[TradeExecution], portfolio: Portfolio, buf: StringIO): """Write trades to the output logs.""" for t in trades: for line in format_trade(portfolio, t): print(" " + line, file=buf) print("", file=buf)