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)