Source code for tradingstrategy.analysis.portfolioanalyzer

"""Analyse the portfolio development over the time.

.. warning ::

    This module is deprecated and replaced by tradeexecutor analysis modules.

- Portfolio situation at the start of the each tick
- Currently held assets
- Net asset value (NAV)
- Asset valuation change

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 import ExchangeUniverse
from tradingstrategy.pair import DEXPair, LegacyPairUniverse
from tradingstrategy.types import USDollarAmount, PrimaryKey

[docs]@dataclass class AssetSnapshot: """Asset holdings at a specific timepoint.""" #: How many tokens of the asset we had quantity: float #: Daily market value of the tokens market_value: USDollarAmount realised_pnl: USDollarAmount unrealised_pnl: USDollarAmount total_pnl: USDollarAmount # def __post_init__(self): assert self.quantity > 0, f"Bad quantity {self.quantity}" assert self.market_value > 0, f"Bad market value {self.market_value}" assert type(self.total_pnl) == float assert type(self.realised_pnl) == float assert type(self.unrealised_pnl) == float
[docs]@dataclass class PortfolioSnapshot: """Represents the portfolio status at the start of the day/candle snapshot""" #: A running counter where the first backtest simulated event is tick 1, the next one tick 2 tick: int #: What reserve currencies we have. #: E.g. USD: 10_000 #: Resever currenecies expressed as strings for the backwards compatibiltiy. cash_balances: Dict[str, float] #: What reserve currencies we have. #: E.g. USD: 10_000 #: Resever currenecies expressed as strings for the backwards compatibiltiy. asset_snapshots: Dict[PrimaryKey, AssetSnapshot] #: 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
[docs] def get_ordered_assets_stable(self) -> List[Tuple[PrimaryKey, AssetSnapshot]]: """Return asset snapshots in a stable order between days. """ assets = [(pair_id, s) for pair_id, s in self.asset_snapshots.items()] assets.sort(key=lambda a: a[0]) return assets
[docs] def get_ordered_assets_by_weight(self) -> List[Tuple[PrimaryKey, AssetSnapshot]]: """Return asset snapshots in an order where the heaviest asset is first. """ assets = [(pair_id, s) for pair_id, s in self.asset_snapshots.items()] assets.sort(key=lambda a: a[1].market_value, reverse=True) return assets
[docs]@dataclass class PortfolioAnalyzer: """Represents the portfolio analysis over the backtest period.""" snapshots: Dict[pd.Timestamp, PortfolioSnapshot]
[docs] def get_max_assets_held_once(self) -> int: """Find out what was the max number of assets the strategy was holding at the same time.""" max_hold = 0 for s in self.snapshots.values(): max_hold = max(max_hold, len(s.asset_snapshots)) return max_hold
[docs]def expand_snapshot_to_row( exchange_universe: ExchangeUniverse, pair_universe: LegacyPairUniverse, ts: pd.Timestamp, snapshot: PortfolioSnapshot, max_assets: int, timestamp_format: str) -> dict: """Create DataFrame rows from each portfolio snapshot.""" # timestamp = # ??? assert max_assets r = { # "timestamp": timestamp, "Id": snapshot.tick, "Holdings at": ts.strftime(timestamp_format), "NAV USD": 0, "Cash USD": f"{snapshot.cash_balances['USD']:,.0f}", } assets = snapshot.get_ordered_assets_by_weight() # Initialize empty asset columns for i in range(max_assets): idx = i + 1 r[f"#{idx} asset"] = pd.NA r[f"#{idx} value"] = pd.NA r[f"#{idx} weight %"] = pd.NA r[f"#{idx} PnL"] = pd.NA r[f"#{idx} PnL raw"] = 0 total_asset_value = 0 for i, asset_tuple in enumerate(assets): idx = i + 1 pair_id, asset_snapshot = asset_tuple pair = pair_universe.get_pair_by_id(pair_id) r[f"#{idx} asset"] = pair.base_token_symbol[0:8] # Cut long ticker names r[f"#{idx} value"] = f"{asset_snapshot.market_value:,.0f}" r[f"#{idx} PnL"] = f"{asset_snapshot.total_pnl:,.2f}" r[f"#{idx} PnL raw"] = asset_snapshot.total_pnl assert type(asset_snapshot.total_pnl) == float total_asset_value += asset_snapshot.market_value for i, asset_tuple in enumerate(assets): idx = i + 1 pair_id, asset_snapshot = asset_tuple value = r[f"#{idx} value"] if value: r[f"#{idx} weight %"] = f"{asset_snapshot.market_value / total_asset_value * 100:.0f}" r["NAV USD"] = f"{snapshot.cash_balances['USD'] + total_asset_value:,.2f}" return r
[docs]def expand_timeline( exchange_universe: ExchangeUniverse, pair_universe: LegacyPairUniverse, analyzer: PortfolioAnalyzer, create_html_styles=True, vmin=-0.3, vmax=0.2, timestamp_format="%Y-%m-%d", ) -> pd.DataFrame: """Console output for the portfolio development over the time. Each row presents the portfolio status at the end of the day/candle. The outputted data frame is intented to be human readable and not for programmatic manipulation. :param create_html_styles: Create a formatter function that can be applied to hide and recolour columns. :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()` :return: pd.Dataframe rendering the portfolio development over the time """ asset_column_count = analyzer.get_max_assets_held_once() raw_output = [expand_snapshot_to_row(exchange_universe, pair_universe, ts, s, asset_column_count, timestamp_format) for ts, s in analyzer.snapshots.items()] applied_df = pd.DataFrame(raw_output) # timeline.apply(expander, axis='columns', result_type='expand') # Sort portfoli snapshots by backtest tick events # 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 = hidden_columns = [] asset_colums = [] for i in range(asset_column_count): # Add a background color for each asset column group idx = i + 1 styles = styles.background_gradient( axis=0, subset=[f"#{idx} asset", f"#{idx} value", f"#{idx} PnL", f"#{idx} weight %"], gmap=applied_df[f"#{idx} 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 # Add more border between assets # # hidden_columns.append(f"#{idx} PnL raw") asset_colums.append(f"#{idx} asset") # Build table col styles styles_dict = { # Don't break timestamp value to multiple lines "Holdings at": [{'selector': 'td', 'props': [('white-space', 'nowrap')]}], } # Format asset column groups for col in asset_colums: styles_dict[col] = [{'selector': 'td', 'props': [('border-left', '3px solid #888'), ("font-weight", "bold"), ("text-align", "left")]}] styles = styles.set_table_styles(styles_dict) styles = styles.hide_columns(hidden_columns) return styles if create_html_styles: return applied_df, apply_styles else: # Format for console for i in range(asset_column_count): idx = i + 1 del applied_df[f"#{idx} PnL raw"] return applied_df, None