text: #     Copyright 2016-present CERN – European Organization for Nuclear Research
  #
  #     Licensed under the Apache License, Version 2.0 (the "License");
  #     you may not use this file except in compliance with the License.
  #     You may obtain a copy of the License at
  #
  #         http://www.apache.org/licenses/LICENSE-2.0
  #
  #     Unless required by applicable law or agreed to in writing, software
  #     distributed under the License is distributed on an "AS IS" BASIS,
  #     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  #     See the License for the specific language governing permissions and
  #     limitations under the License.
  from datetime import datetime
  
  import matplotlib as plt
  import pandas as pd
  from pandas import Timedelta
  from pandas.tseries.frequencies import to_offset
  
  from qf_lib.analysis.common.abstract_document import AbstractDocument
  from qf_lib.common.utils.error_handling import ErrorHandling
  from qf_lib.analysis.trade_analysis.trades_generator import TradesGenerator
  from qf_lib.backtesting.contract.contract import Contract
  from qf_lib.backtesting.monitoring.backtest_result import BacktestResult
  from qf_lib.backtesting.portfolio.backtest_position import BacktestPositionSummary
  from qf_lib.common.enums.frequency import Frequency
  from qf_lib.containers.dataframe.qf_dataframe import QFDataFrame
  from qf_lib.containers.series.qf_series import QFSeries
  from qf_lib.documents_utils.document_exporting.element.chart import ChartElement
  from qf_lib.documents_utils.document_exporting.element.df_table import DFTable
  from qf_lib.documents_utils.document_exporting.element.heading import HeadingElement
  from qf_lib.documents_utils.document_exporting.element.new_page import NewPageElement
  from qf_lib.documents_utils.document_exporting.element.paragraph import ParagraphElement
  from qf_lib.documents_utils.document_exporting.pdf_exporter import PDFExporter
  from qf_lib.plotting.charts.line_chart import LineChart
  from qf_lib.plotting.decorators.axes_label_decorator import AxesLabelDecorator
  from qf_lib.plotting.decorators.axes_position_decorator import AxesPositionDecorator
  from qf_lib.plotting.decorators.data_element_decorator import DataElementDecorator
  from qf_lib.plotting.decorators.legend_decorator import LegendDecorator
  from qf_lib.plotting.decorators.title_decorator import TitleDecorator
  from qf_lib.settings import Settings
  
  
  @ErrorHandling.class_error_logging()
  class PortfolioAnalysisSheet(AbstractDocument):
      """
      Creates a PDF containing a visual representation of portfolio changes over time.
  
      Parameters
      -----------
      settings: Settings
          necessary settings
      pdf_exporter: PDFExporter
          used to export the document to PDF
      backtest_result: BacktestResult
          used to access all trading records
      title: str
          title of the document, will be a part of the filename. Do not use special characters.
      generate_pnl_chart_per_ticker: bool
          If true, adds pnl chart for each asset traded in the portfolio
      """
  
      def __init__(self, settings: Settings, pdf_exporter: PDFExporter, backtest_result: BacktestResult,
                   title: str = "Portfolio Analysis Sheet", generate_pnl_chart_per_ticker: bool = False):
          super().__init__(settings, pdf_exporter, title=title)
          self.backtest_result = backtest_result
          self.full_image_axis_position = (0.08, 0.1, 0.915, 0.80)  # (left, bottom, width, height)
          self.trades_generator = TradesGenerator()
          self.generate_pnl_chart_per_ticker = generate_pnl_chart_per_ticker
  
      def build_document(self):
          self._add_header()
          self.document.add_element(ParagraphElement("\n"))
          self._add_leverage_chart()
          self._add_assets_number_in_portfolio_chart()
          self._add_concentration_of_portfolio_chart((1, 5,))
          self._add_gini_coefficient_chart()
          self._add_number_of_transactions_chart('D', 'Number of transactions per day')
          self._add_number_of_transactions_chart('30D', 'Number of transactions per month')
          self._add_number_of_transactions_chart('365D', 'Number of transactions per year')
          self._add_volume_traded()
  
          self._add_avg_time_in_the_market_per_ticker()
          self._add_performance_statistics()
  
      def _add_leverage_chart(self):
          lev_chart = self._get_leverage_chart(self.backtest_result.portfolio.leverage_series())
  
          position_decorator = AxesPositionDecorator(*self.full_image_axis_position)
          lev_chart.add_decorator(position_decorator)
  
          # Add y label
          label_decorator = AxesLabelDecorator(y_label="Leverage")
          lev_chart.add_decorator(label_decorator)
  
          self.document.add_element(ChartElement(lev_chart, figsize=self.full_image_size, dpi=self.dpi))
  
      def _add_assets_number_in_portfolio_chart(self):
          chart = LineChart(rotate_x_axis=False)
          legend = LegendDecorator(key="legend_decorator")
  
          position_decorator = AxesPositionDecorator(*self.full_image_axis_position)
          chart.add_decorator(position_decorator)
  
          positions_history = self.backtest_result.portfolio.positions_history()
  
          # Find all not NaN values (not NaN values indicate that the position was open for this contract at that time)
          # and count their number for each row (for each of the dates)
          number_of_contracts = positions_history.notnull().sum(axis=1)
          number_of_contracts_decorator = DataElementDecorator(number_of_contracts)
  
          chart.add_decorator(number_of_contracts_decorator)
          legend.add_entry(number_of_contracts_decorator, "Contracts")
  
          # Count number of assets in the portfolio (if on two contracts from same future family exist in the same time,
          # e.g. during rolling day, in the portfolio, they are counted as one asset)
          def contract_to_ticker(c: Contract):
              return self.backtest_result.portfolio.contract_ticker_mapper. \
                  contract_to_ticker(c, strictly_to_specific_ticker=False)
  
          assets_history = positions_history.rename(columns=contract_to_ticker)
          assets_history = assets_history.groupby(level=0, axis=1).apply(func=(
              # For each asset, group all of the corresponding columns (each of which corresponds to one contract),
              # and check if at any given timestamp the "value" of any of the contracts was different than None - this
              # indicates that at this point of time a position concerning the given asset was open in the portfolio
              # (which in the resulting series will be denoted as 1, otherwise - 0, so that it will be possible to
              # sum positions of all open assets at any given point of time)
              lambda x: x.notna().any(axis=1).astype(int)
          ))
  
          number_of_assets = assets_history.sum(axis=1)
          number_of_assets_decorator = DataElementDecorator(number_of_assets)
          chart.add_decorator(number_of_assets_decorator)
          legend.add_entry(number_of_assets_decorator, "Assets")
  
          # Add title
          title_decorator = TitleDecorator("Number of assets in the portfolio")
          chart.add_decorator(title_decorator)
  
          # Add legend
          chart.add_decorator(legend)
  
          # Add y label
          label_decorator = AxesLabelDecorator(y_label="Number of contracts / assets")
          chart.add_decorator(label_decorator)
  
          self.document.add_element(ChartElement(chart, figsize=self.full_image_size, dpi=self.dpi))
  
      def _add_line_chart_element(self, series, title):
          chart = self._get_line_chart(series, title)
  
          position_decorator = AxesPositionDecorator(*self.full_image_axis_position)
          chart.add_decorator(position_decorator)
  
          self.document.add_element(ChartElement(chart, figsize=self.full_image_size, dpi=self.dpi))
  
      def _add_concentration_of_portfolio_chart(self, top_assets_numbers: tuple = (1, 5)):
          chart = LineChart(rotate_x_axis=False)
  
          position_decorator = AxesPositionDecorator(*self.full_image_axis_position)
          chart.add_decorator(position_decorator)
  
          # Define a legend
          legend = LegendDecorator(key="legend_decorator")
  
          # Add y label
          label_decorator = AxesLabelDecorator(y_label="Mean total exposure of top assets")
          chart.add_decorator(label_decorator)
  
          # Add top asset contribution
          positions_history = self.backtest_result.portfolio.positions_history()
          if positions_history.empty:
              raise ValueError("No positions found in positions history")
  
          positions_history = positions_history.applymap(
              lambda x: x.total_exposure if isinstance(x, BacktestPositionSummary) else 0)
  
          # Map all the single contracts onto tickers (including future tickers) and take the maximal total exposure for
          # each of the groups - in case if two contracts for a single asset will be included in the open positions in
          # the portfolio at any point of time, only one (with higher total exposure) will be considered while generating
          # the top assets plot
          def contract_to_ticker(c: Contract):
              return self.backtest_result.portfolio.contract_ticker_mapper. \
                  contract_to_ticker(c, strictly_to_specific_ticker=False)
  
          assets_history = positions_history.rename(columns=contract_to_ticker)
          assets_history = assets_history.groupby(level=0, axis=1).apply(func=(
              lambda x: x.abs().max(axis=1).astype(float)
          ))
  
          for assets_number in top_assets_numbers:
              # For each date (row), find the top_assets largest assets and compute the mean value of their market value
              top_assets_mean_values = assets_history.stack(dropna=False).groupby(level=0).apply(
                  lambda group: group.nlargest(assets_number).mean()
              ).resample('D').last()
              # Divide the computed mean values by the portfolio value, for each of the dates
              top_assets_percentage_value = top_assets_mean_values / self.backtest_result.portfolio.portfolio_eod_series()
  
              concentration_top_asset = DataElementDecorator(top_assets_percentage_value)
              chart.add_decorator(concentration_top_asset)
  
              # Add to legend
              legend.add_entry(concentration_top_asset, "TOP {} assets".format(assets_number))
  
          # Add title
          title_decorator = TitleDecorator("Concentration of assets")
          chart.add_decorator(title_decorator)
  
          # Add legend
          chart.add_decorator(legend)
  
          self.document.add_element(ChartElement(chart, figsize=self.full_image_size, dpi=self.dpi))
  
      def _add_number_of_transactions_chart(self, pandas_freq: str, title: str):
          transactions = self.backtest_result.transactions
          transactions_series = QFSeries(data=transactions, index=(t.time for t in transactions))
  
          if transactions_series.empty:
              raise ValueError("Transactions series is empty")
  
          # Compute the number of transactions per day
          transactions_series = transactions_series.resample(Frequency.DAILY.to_pandas_freq()).count()
  
          # Aggregate the transactions using the given frequency
          if to_offset(pandas_freq) > to_offset('D'):
              transactions_series = transactions_series.rolling(pandas_freq).sum()
  
              # Cut the non complete beginning of the outputs (e.g. in case of 30 days window, cut the first 30 days)
              start_date = transactions_series.index[0]
              transactions_series = transactions_series.loc[start_date + Timedelta(pandas_freq):]
              if transactions_series.empty:
                  # The available time period is too short to compute the statistics with the provided frequency
                  return
  
          elif to_offset(pandas_freq) < to_offset('D'):
              raise ValueError("The provided pandas frequency can not be higher than the daily frequency")
  
          self._add_line_chart_element(transactions_series, title)
  
      def _add_volume_traded(self):
          transactions = self.backtest_result.transactions
          transactions_series = QFSeries(data=transactions, index=(t.time for t in transactions))
          if transactions_series.empty:
              raise ValueError("Transactions series is empty")
  
          # Add the chart containing the volume traded in terms of quantity
          quantities = [abs(t.quantity) for t in transactions_series]
          quantities_series = QFSeries(data=quantities, index=transactions_series.index)
  
          # Aggregate the quantities for each day
          quantities_series = quantities_series.resample(Frequency.DAILY.to_pandas_freq()).sum()
  
          # Generate chart and add it to the document
          self._add_line_chart_element(quantities_series, "Volume traded per day [in contracts]")
  
          # Add the chart containing the exposure of the traded assets
          total_exposures = [abs(t.quantity) * t.price * t.contract.contract_size for t in transactions_series]
          total_exposures_series = QFSeries(data=total_exposures, index=transactions_series.index)
          total_exposures_series = total_exposures_series.resample(Frequency.DAILY.to_pandas_freq()).sum()
  
          # Generate chart and add it to the document
          self._add_line_chart_element(total_exposures_series, "Volume traded per day [notional in currency units]")
  
      def _add_gini_coefficient_chart(self):
          chart = LineChart(rotate_x_axis=False)
  
          position_decorator = AxesPositionDecorator(*self.full_image_axis_position)
          chart.add_decorator(position_decorator)
  
          # Get the assets history
          assets_history = self.backtest_result.portfolio.positions_history()
          assets_history = assets_history.applymap(
              lambda x: x.total_exposure if isinstance(x, BacktestPositionSummary) else 0)
  
          def gini(group):
              # Function computing the Gini coefficients for each row
              group = group.abs()
              ranks = group.rank(ascending=True)
              mean_value = group.mean()
              samples = group.size
              group = group * (2 * ranks - samples - 1)
              return group.sum() / (samples * samples * mean_value)
  
          assets_history = assets_history.stack(dropna=False).groupby(level=0).apply(gini).to_frame(name="Gini "
                                                                                                         "coefficient")
  
          assets_history_decorator = DataElementDecorator(assets_history)
          chart.add_decorator(assets_history_decorator)
  
          # Add title
          title_decorator = TitleDecorator("Concentration of assets - Gini coefficient")
          chart.add_decorator(title_decorator)
  
          self.document.add_element(ChartElement(chart, figsize=self.full_image_size, dpi=self.dpi))
  
      def _ticker_name(self, contract: Contract) -> str:
          ticker = self.backtest_result.portfolio.contract_ticker_mapper.contract_to_ticker(contract, False)
          return ticker.name
  
      def _add_avg_time_in_the_market_per_ticker(self):
          """
          Compute the total time in the market per ticker (separately for long and short positions) in minutes
          and divide it by the total duration of the backtest in minutes.
          """
          self.document.add_element(NewPageElement())
          self.document.add_element(HeadingElement(level=2, text="Average time in the market per asset"))
  
          start_time = self.backtest_result.start_date
          end_time = self.backtest_result.portfolio.timer.now()
          backtest_duration = pd.Timedelta(end_time - start_time) / pd.Timedelta(minutes=1)  # backtest duration in min
  
          closed_positions_time = [
              (self._ticker_name(position.contract()), position.start_time, position.end_time, position.direction())
              for position in self.backtest_result.portfolio.closed_positions()
          ]
          open_positions_time = [
              (self._ticker_name(c), position.start_time, end_time, position.direction())
              for c, position in self.backtest_result.portfolio.open_positions_dict.items()
          ]
  
          positions = QFDataFrame(data=closed_positions_time + open_positions_time,
                                  columns=["Tickers name", "Start time", "End time", "Position direction"])
  
          def compute_duration(grouped_rows):
              return pd.DatetimeIndex([]).union_many(
                  [pd.date_range(row["Start time"], row["End time"], freq='T', closed='left')
                   for _, row in grouped_rows.iterrows()]).size
  
          positions = positions.groupby(by=["Tickers name", "Position direction"]).apply(compute_duration) \
              .rename("Duration (minutes)").reset_index()
          positions["Duration"] = positions["Duration (minutes)"] / backtest_duration
          positions = positions.pivot_table(index="Tickers name", columns="Position direction",
                                            values="Duration").reset_index()
          positions = positions.rename(columns={-1: "Short", 1: "Long"})
  
          # Add default 0 column in case if only short / long positions occurred in the backtest
          for column in ["Short", "Long"]:
              if column not in positions.columns:
                  positions[column] = 0.0
  
          positions["Out"] = 1.0 - positions["Long"] - positions["Short"]
          positions[["Long", "Short", "Out"]] = positions[["Long", "Short", "Out"]].applymap(lambda x: '{:.2%}'.format(x))
          positions = positions.fillna(0.0)
  
          table = DFTable(positions, css_classes=['table', 'left-align'])
          table.add_columns_classes(["Tickers name"], 'wide-column')
          self.document.add_element(table)
  
      def _add_performance_statistics(self):
          """
          For each ticker computes its overall performance (PnL of short positions, PnL of long positions, total PnL).
          It generates a table containing final PnL values for each of the ticker nad optionally plots the performance
          throughout the backtest.
          """
          closed_positions = self.backtest_result.portfolio.closed_positions()
          closed_positions_pnl = QFDataFrame.from_records(
              data=[(self._ticker_name(p.contract()), p.end_time, p.direction(), p.total_pnl) for p in closed_positions],
              columns=["Tickers name", "Time", "Direction", "Realised PnL"]
          )
          closed_positions_pnl = closed_positions_pnl.sort_values(by="Time")
  
          # Get all open positions history
          open_positions_history = self.backtest_result.portfolio.positions_history()
          open_positions_history = open_positions_history.reset_index().melt(
              id_vars='index', value_vars=open_positions_history.columns, var_name='Contract',
              value_name='Position summary')
          open_positions_pnl = QFDataFrame(data={
              "Tickers name": open_positions_history["Contract"].apply(lambda contract: self._ticker_name(contract)),
              "Time": open_positions_history["index"],
              "Direction": open_positions_history["Position summary"].apply(
                  lambda p: p.direction if isinstance(p, BacktestPositionSummary) else 0),
              "Total PnL of open position": open_positions_history["Position summary"].apply(
                  lambda p: p.total_pnl if isinstance(p, BacktestPositionSummary) else 0)
          })
  
          all_positions_pnl = pd.concat([closed_positions_pnl, open_positions_pnl], sort=False)
  
          performance_dicts_series = all_positions_pnl.groupby(by=["Tickers name"]).apply(
              self._performance_series_for_ticker)
          performance_df = QFDataFrame(performance_dicts_series.tolist(), index=performance_dicts_series.index)
  
          self.document.add_element(NewPageElement())
          self.document.add_element(HeadingElement(level=2, text="Performance of each asset"))
          final_performance = performance_df. \
              applymap(lambda pnl_series: pnl_series.iloc[-1] if not pnl_series.empty else 0.0). \
              sort_values(by="Overall performance", ascending=False). \
              applymap(lambda p: '{:,.2f}'.format(p)). \
              reset_index()
          table = DFTable(final_performance, css_classes=['table', 'left-align'])
          table.add_columns_classes(["Tickers name"], 'wide-column')
          self.document.add_element(table)
  
          # Add performance plots
          if self.generate_pnl_chart_per_ticker:
              self.document.add_element(NewPageElement())
              self.document.add_element(
                  HeadingElement(level=2, text="Performance of each asset during the whole backtest"))
              for ticker_name, performance in performance_df.iterrows():
                  self._plot_ticker_performance(ticker_name, performance)
  
      def _performance_series_for_ticker(self, df: QFDataFrame):
          df = df.sort_values(by="Time")
          time_index = pd.DatetimeIndex(df[df["Total PnL of open position"].notnull()]["Time"].drop_duplicates())
  
          plot_lines = [
              ("Long positions", [1]),
              ("Short positions", [-1]),
              ("Overall performance", [0, 1, -1])
          ]
  
          return_data = {}
          for title, directions in plot_lines:
              open_positions_pnl = df[["Total PnL of open position", "Time", "Direction"]].dropna()
              open_positions_pnl.loc[~open_positions_pnl["Direction"].isin(directions), 'Total PnL of open position'] = 0
              open_positions_pnl = open_positions_pnl.groupby("Time").agg({'Total PnL of open position': 'sum'})
  
              closed_positions_pnl = df[["Realised PnL", "Time", "Direction"]]
              closed_positions_pnl.loc[~closed_positions_pnl["Direction"].isin(directions), 'Realised PnL'] = 0
              closed_positions_pnl = closed_positions_pnl.groupby("Time").agg({'Realised PnL': 'sum'})
              closed_positions_pnl["Realised PnL"] = closed_positions_pnl["Realised PnL"].cumsum()
  
              pnl_series = QFSeries(
                  data=closed_positions_pnl["Realised PnL"] + open_positions_pnl['Total PnL of open position'],
                  index=time_index).fillna(method='ffill').fillna(0.0)
              return_data[title] = pnl_series
  
          return return_data
  
      def _plot_ticker_performance(self, ticker_name: str, performance):
          self.document.add_element(HeadingElement(level=2, text="PnL of {}".format(ticker_name)))
  
          chart = LineChart()
          legend = LegendDecorator(key="legend_decorator")
          line_colors = iter(("#add8e6", "#000000", "#fa8072"))
  
          for title, pnl_series in performance.iteritems():
              # Plot series only in case if it consist anything else then 0
              if (pnl_series != 0).any():
                  data_series = DataElementDecorator(pnl_series, **{"color": next(line_colors)})
                  legend.add_entry(data_series, title)
                  chart.add_decorator(data_series)
  
          chart.add_decorator(legend)
          self.document.add_element(ChartElement(chart, figsize=self.full_image_size, dpi=self.dpi))
  
      def save(self, report_dir: str = ""):
          # Set the style for the report
          plt.style.use(['tearsheet'])
  
          filename = "%Y_%m_%d-%H%M Portfolio Analysis Sheet.pdf"
          filename = datetime.now().strftime(filename)
          return self.pdf_exporter.generate([self.document], report_dir, filename)
  