Source code for tradingstrategy.analysis.tradeanalyzer

"""Analyze the trade performance of algorithm

.. warning ::

    This module is deprecated and replaced by tradeexecutor analysis modules.

- Trade summary
- Won and lost trades
- Trade won/lost distribution graph
- Trade timeline and analysis of individual trades made

import datetime
from dataclasses import dataclass, field
from typing import List, Dict, Iterable, Optional, Tuple, Callable

import numpy as np
import pandas as pd
from import Styler

from tradingstrategy.analysis.tradehint import TradeHint, TradeHintType
from import ExchangeUniverse
from tradingstrategy.pair import LegacyPairUniverse, PandasPairUniverse
from tradingstrategy.types import PrimaryKey, USDollarAmount
from tradingstrategy.utils.format import format_value, format_percent, format_price, format_duration_days_hours_mins, \
from tradingstrategy.utils.summarydataframe import as_dollar, as_integer, create_summary_table, as_percent, as_missing

[docs]@dataclass class SpotTrade: """Track spot trades to construct position performance. For sells, quantity is negative. """ #: Internal running counter to uniquely label all trades in trade analysis trade_id: PrimaryKey #: Trading pair for this trade pair_id: PrimaryKey #: When this trade was made, the backtes simulation thick timestamp: pd.Timestamp #: Asset price at buy in price: USDollarAmount #: How much we bought the asset. Negative value for sells. quantity: float #: How much fees we paid to the exchange commission: USDollarAmount #: How much we lost against the midprice due to the slippage slippage: USDollarAmount #: Any hints applied for this trade why it was performed hint: Optional[TradeHint] = None #: Internal state dump of the algorithm when this trade was made. #: This is mostly useful when doing the trade analysis try to understand #: why some trades were made. #: It also allows you to reconstruct the portfolio state over the time. state_details: Optional[Dict] = None def is_buy(self): return self.quantity > 0 def is_sell(self): return self.quantity < 0 @property def value(self) -> USDollarAmount: return abs(self.price * self.quantity)
[docs]@dataclass class TradePosition: """How a particular asset traded. Each asset can have multiple entries (buys) and exits (sells) For a simple strategies there can be only one or two trades per position. * Enter (buy) * Exit (sell optionally) """ #: List of all trades done for this position trades: List[SpotTrade] = field(default_factory=list) #: Closing the position could be deducted from the trades themselves, #: but we cache it by hand to speed up processing opened_at: Optional[pd.Timestamp] = None #: Closing the position could be deducted from the trades themselves, #: but we cache it by hand to speed up processing closed_at: Optional[pd.Timestamp] = None def __eq__(self, other: "TradePosition"): """Trade positions are unique by opening timestamp and pair id.] We assume there cannot be a position opened for the same asset at the same time twice. """ return self.position_id == other.position_id def __hash__(self): """Allows easily create index (hash map) of all positions""" return hash((self.position_id)) @property def position_id(self) -> PrimaryKey: """Position id is the same as the opening trade id.""" return self.trades[0].trade_id @property def pair_id(self) -> PrimaryKey: """Position id is the same as the opening trade id.""" return self.trades[0].pair_id @property def duration(self) -> Optional[datetime.timedelta]: """How long this position was held. :return: None if the position is still open """ if not self.is_closed(): return None return self.closed_at - self.opened_at def is_open(self): return self.closed_at is None def is_closed(self): return not self.is_open() @property def open_quantity(self) -> float: return sum([t.quantity for t in self.trades]) @property def open_value(self) -> float: """The current value of this open position, with the price at the time of opening.""" assert self.is_open() return sum([t.value for t in self.trades]) @property def open_price(self) -> float: """At what price we opened this position. Supports only simple enter/exit positions. """ return self.get_first_entry_price()
[docs] def get_first_entry_price(self) -> float: """What was the price when the first entry buy for this position was made. """ buys = list(self.buys) return buys[0].price
[docs] def get_last_exit_price(self) -> float: """What was the time when the last sell for this position was executd. """ sells = list(self.sells) return sells[-1].price assert len(sells) == 1
@property def close_price(self) -> float: """At what price we exited this position. Supports only simple enter/exit positions. """ return self.get_last_exit_price() @property def buys(self) -> Iterable[SpotTrade]: return [t for t in self.trades if t.is_buy()] @property def sells(self) -> Iterable[SpotTrade]: return [t for t in self.trades if t.is_sell()] @property def buy_value(self) -> USDollarAmount: return sum([t.value - t.commission for t in self.trades if t.is_buy()]) @property def sell_value(self) -> USDollarAmount: return sum([t.value - t.commission for t in self.trades if t.is_sell()]) @property def realised_profit(self) -> USDollarAmount: """Calculated life-time profit over this position.""" assert not self.is_open() return -sum([t.quantity * t.price - t.commission for t in self.trades]) @property def realised_profit_percent(self) -> float: """Calculated life-time profit over this position.""" assert not self.is_open() buy_value = self.buy_value sell_value = self.sell_value return sell_value / buy_value - 1
[docs] def is_win(self): """Did we win this trade.""" assert not self.is_open() return self.realised_profit > 0
def is_lose(self): assert not self.is_open() return self.realised_profit < 0
[docs] def is_stop_loss(self) -> bool: """Was stop loss triggered for this position""" for t in self.trades: if t.hint: if t.hint.type == TradeHintType.stop_loss_triggered: return True return False
def add_trade(self, t: SpotTrade): if self.trades: last_trade = self.trades[-1] assert t.timestamp > last_trade.timestamp, f"Tried to do trades in wrong order. Last: {last_trade}, got {t}" self.trades.append(t) def can_trade_close_position(self, t: SpotTrade): assert self.is_open() if not t.is_sell(): return False open_quantity = self.open_quantity closing_quantity = -t.quantity assert closing_quantity <= open_quantity, "Cannot sell more than we have in balance sheet" return closing_quantity == open_quantity
[docs] def get_max_size(self) -> USDollarAmount: """Get the largest size of this position over the time""" cur_size = 0 max_size = 0 for t in self.trades: cur_size += t.value max_size = max(cur_size, max_size) return max_size
[docs] def get_trade_count(self) -> int: """How many individual trades was done to manage this position.""" return len(self.trades)
[docs]@dataclass class AssetTradeHistory: """How a particular asset traded. Each position can have increments or decrements. When position is decreased to zero, it is considered closed, and a new buy open a new position. """ positions: List[TradePosition] = field(default_factory=list) def get_first_opened_at(self) -> Optional[pd.Timestamp]: if self.positions: return self.positions[0].opened_at return None def get_last_closed_at(self) -> Optional[pd.Timestamp]: for position in reversed(self.positions): if not position.is_open(): return position.closed_at return None
[docs] def add_trade(self, t: SpotTrade): """Adds a new trade to the asset history. If there is an open position the trade is added against this, otherwise a new position is opened for tracking. """ current_position = None if self.positions: if self.positions[-1].is_open(): current_position = self.positions[-1] if current_position: if current_position.can_trade_close_position(t): # Close the existing position current_position.closed_at = t.timestamp current_position.add_trade(t) assert current_position.open_quantity == 0 else: # Add to the existing position current_position.add_trade(t) else: # Open new position new_position = TradePosition(opened_at=t.timestamp) new_position.add_trade(t) self.positions.append(new_position)
[docs]@dataclass class TradeSummary: """Some generic statistics over all the trades""" won: int lost: int zero_loss: int stop_losses: int undecided: int realised_profit: USDollarAmount open_value: USDollarAmount uninvested_cash: USDollarAmount initial_cash: USDollarAmount extra_return: USDollarAmount
[docs] def to_dataframe(self) -> pd.DataFrame: """Creates a human-readable Pandas dataframe table from the object.""" total_trades = self.won + self.lost human_data = { "Return %": as_percent(self.realised_profit / self.initial_cash), "Cash at start": as_dollar(self.initial_cash), "Value at end": as_dollar(self.open_value + self.uninvested_cash), "Trade win percent": as_percent(self.won / total_trades) if total_trades else as_missing(), "Total trades done": as_integer(self.won + self.lost + self.zero_loss), "Won trades": as_integer(self.won), "Lost trades": as_integer(self.lost), "Stop losses triggered": as_integer(self.stop_losses), "Stop loss % of all": as_percent(self.stop_losses / total_trades) if total_trades else as_missing(), "Stop loss % of lost": as_percent(self.stop_losses / self.lost) if self.lost else as_missing(), "Zero profit trades": as_integer(self.zero_loss), "Positions open at the end": as_integer(self.undecided), "Realised profit and loss": as_dollar(self.realised_profit), "Portfolio unrealised value": as_dollar(self.open_value), "Extra returns on lending pool interest": as_dollar(self.extra_return), "Cash left at the end": as_dollar(self.uninvested_cash), } return create_summary_table(human_data)
[docs]@dataclass class TradeAnalyzer: """Analysis of trades in a portfolio.""" #: How a particular asset traded. Asset id -> Asset history mapping asset_histories: Dict[object, AssetTradeHistory] = field(default_factory=dict) def get_first_opened_at(self) -> Optional[pd.Timestamp]: def all_opens(): for history in self.asset_histories.values(): yield history.get_first_opened_at() return min(all_opens()) def get_last_closed_at(self) -> Optional[pd.Timestamp]: def all_closes(): for history in self.asset_histories.values(): closed = history.get_last_closed_at() if closed: yield closed return max(all_closes())
[docs] def get_all_positions(self) -> Iterable[Tuple[PrimaryKey, TradePosition]]: """Return open and closed positions over all traded assets.""" for pair_id, history in self.asset_histories.items(): for position in history.positions: yield pair_id, position
[docs] def get_open_positions(self) -> Iterable[Tuple[PrimaryKey, TradePosition]]: """Return open and closed positions over all traded assets.""" for pair_id, history in self.asset_histories.items(): for position in history.positions: if position.is_open(): yield pair_id, position
[docs] def calculate_summary_statistics(self, initial_cash, uninvested_cash, extra_return=0) -> TradeSummary: """Calculate some statistics how our trades went.""" won = lost = zero_loss = stop_losses = undecided = 0 open_value: USDollarAmount = 0 profit: USDollarAmount = 0 for pair_id, position in self.get_all_positions(): if position.is_open(): open_value += position.open_value undecided += 1 continue if position.is_stop_loss(): stop_losses += 1 if position.is_win(): won += 1 elif position.is_lose(): lost += 1 else: # Any profit exactly balances out loss in slippage and commission zero_loss += 1 profit += position.realised_profit return TradeSummary( won=won, lost=lost, zero_loss=zero_loss, stop_losses=stop_losses, undecided=undecided, realised_profit=profit + extra_return, open_value=open_value, uninvested_cash=uninvested_cash, initial_cash=initial_cash, extra_return=extra_return, )
[docs] def create_timeline(self) -> pd.DataFrame: """Create a timeline feed how we traded over a course of time. Note: We assume each position has only one enter and exit event, not position increases over the lifetime. :return: DataFrame with timestamp and timeline_event columns """ def gen_events(): for pair_id, position in self.get_all_positions(): yield (position.position_id, position) df = pd.DataFrame(gen_events(), columns=["position_id", "position"]) return df
[docs]def expand_timeline( exchange_universe: ExchangeUniverse, pair_universe: PandasPairUniverse, timeline: pd.DataFrame, vmin=-0.3, vmax=0.2, timestamp_format="%Y-%m-%d", hidden_columns=["Id", "PnL % raw"]) -> Tuple[pd.DataFrame, Callable]: """Expand trade history timeline to human readable table. This will the outputting much easier in Python Notebooks. Currently does not incrementing/decreasing positions gradually. Instaqd of applying styles or returning a styled dataframe, we return a callable that applies the styles. This is because of Pandas issue - hidden indexes, columns, etc. are not exported. :param vmax: Trade success % to have the extreme green color. :param vmin: The % of lost capital on the trade to have the extreme red color. :param timestamp_format: How to format Opened at column, as passed to `strftime()` :param hidden_columns: Hide columns in the output table :return: DataFrame with human readable position win/loss information, having DF indexed by timestamps and a styler function """ # def expander(row): position: TradePosition = row["position"] # timestamp = # ??? pair_id = position.pair_id pair_info = pair_universe.get_pair_by_id(pair_id) exchange = exchange_universe.get_by_id(pair_info.exchange_id) remarks = "SL" if position.is_stop_loss() else "" r = { # "timestamp": timestamp, "Id": position.position_id, "Remarks": remarks, "Opened at": position.opened_at.strftime(timestamp_format), "Duration": format_duration_days_hours_mins(position.duration) if position.duration else np.nan, "Exchange":, "Base asset": pair_info.base_token_symbol, "Quote asset": pair_info.quote_token_symbol, "Position max size": format_value(position.get_max_size()), "PnL USD": format_value(position.realised_profit) if position.is_closed() else np.nan, "PnL %": format_percent_2_decimals(position.realised_profit_percent) if position.is_closed() else np.nan, "PnL % raw": position.realised_profit_percent if position.is_closed() else 0, "Open price USD": format_price(position.open_price), "Close price USD": format_price(position.close_price) if position.is_closed() else np.nan, "Trade count": position.get_trade_count(), } return r applied_df = timeline.apply(expander, axis='columns', result_type='expand') # applied_df\ .sort_values(by=['Id'], ascending=[True], inplace=True) # Get rid of NaN labels # applied_df.fillna('', inplace=True) def apply_styles(df: pd.DataFrame): # Create a Pandas Styler with multiple styling options applied # Dynamically color the background of trade outcome coluns # styles =\ .hide_index()\ .hide_columns(hidden_columns)\ .background_gradient( axis=0, gmap=applied_df['PnL % raw'], cmap='RdYlGn', vmin=vmin, # We can only lose 100% of our money on position vmax=vmax) # 50% profit is 21.5 position. Assume this is the max success color we can hit over # Don't let the text inside a cell to wrap styles = styles.set_table_styles({ "Opened at": [{'selector': 'td', 'props': [('white-space', 'nowrap')]}], "Exchange": [{'selector': 'td', 'props': [('white-space', 'nowrap')]}], }) return styles return applied_df, apply_styles