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 tradeexecutor.state.trade import TradeExecution

#: See setup_discord_logging()
DISCORD_BREAK_CHAR = "…"


[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: amount = abs(trade.planned_quantity / existing_position.get_net_quantity()) existing_text = f", {amount*100:,.2f}% of existing position" else: existing_text = "" 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}", ] if link: lines.append(f" 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 = "" lines =[ f"{symbol} #{position.position_id} {position.pair.get_human_description()} size:${position.get_value():,.2f}, profit:{position.get_total_profit_percent():.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}, open price:${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)