Skip to content
All library documents

Calculating Portfolio Risk from Optimized and Buffered Positions

Code pysystemtrade

Summary

This risk stage calculates portfolio risk for several position representations. It can pass optimized portfolio weights directly to the portfolio stage, or estimate risk from original positions after buffering and rounding. For the latter path, it gathers each instrument’s buffered, rounded position, fills missing observations forward, and combines positions with each instrument’s value as a proportion of capital to form portfolio weights.

The resulting weights are then passed to a portfolio-level risk calculation. The module also exposes original unrounded position risk for comparison and obtains optimized weights from a dynamic optimization stage. If that stage is absent, the optimized risk accessor raises an exception explaining that the measure is inappropriate without dynamic optimization. The code delegates the risk formula to another component, so it does not specify the risk model, assumptions, or validation evidence itself.

Key ideas

  • Portfolio risk can be calculated from optimized weights or from original positions after buffering and rounding.
  • Buffered positions are gathered by instrument and missing values are carried forward.
  • Position counts are combined with per-contract value as a proportion of capital to construct portfolio weights.
  • Risk estimation is delegated to the portfolio stage, and optimized-position risk requires a dynamic optimization stage.

Tags

Full text
# risk.py


```py
import pandas as pd

from systems.stage import SystemStage
from systems.portfolio import Portfolios
from systems.provided.dynamic_small_system_optimise.optimised_positions_stage import (
    optimisedPositions,
)
from systems.system_cache import diagnostic

from syscore.pandas.list_of_df import listOfDataFrames

from sysquant.optimisation.weights import seriesOfPortfolioWeights


class Risk(SystemStage):
    @property
    def name(self):
        return "risk"

    @diagnostic()
    def get_portfolio_risk_for_optimised_positions(self) -> pd.Series:
        weights = self.get_optimised_weights_df()
        return self._get_portfolio_risk_given_weights(weights)

    @diagnostic()
    def get_portfolio_risk_for_original_positions_rounded_buffered(self) -> pd.Series:
        positions = self.get_original_buffered_rounded_positions_df()
        positions = positions.round()
        return self._get_portfolio_risk_given_positions(positions)

    @diagnostic()
    def get_portfolio_risk_for_original_positions(self) -> pd.Series:
        return self.portfolio_stage.get_portfolio_risk_for_original_positions()

    @diagnostic()
    def get_original_buffered_rounded_positions_df(self) -> pd.DataFrame:
        instrument_list = self.instrument_list()
        positions_dict = dict(
            [
                (
                    instrument_code,
                    self.get_original_buffered_rounded_position_for_instrument(
                        instrument_code
                    ),
                )
                for instrument_code in instrument_list
            ]
        )

        positions = pd.DataFrame(positions_dict)
        positions = positions.ffill()

        return positions

    @diagnostic()
    def get_original_buffered_rounded_position_for_instrument(
        self, instrument_code: str
    ) -> pd.Series:
        return self.accounts_stage.get_buffered_position(
            instrument_code, roundpositions=True
        )

    def _get_portfolio_risk_given_positions(self, positions: pd.DataFrame) -> pd.Series:
        weight_per_position = (
            self.portfolio_stage.get_per_contract_value_as_proportion_of_capital_df()
        )
        portfolio_weights = listOfDataFrames(
            [weight_per_position, positions]
        ).fill_and_multipy()

        portfolio_weights = seriesOfPortfolioWeights(portfolio_weights)

        return self._get_portfolio_risk_given_weights(portfolio_weights)

    def _get_portfolio_risk_given_weights(
        self, portfolio_weights: seriesOfPortfolioWeights
    ) -> pd.Series:
        return self.portfolio_stage.get_portfolio_risk_given_weights(portfolio_weights)

    def get_optimised_weights_df(self) -> seriesOfPortfolioWeights:
        return self.optimised_stage.get_optimised_weights_df()

    @property
    def optimised_stage(self) -> optimisedPositions:
        try:
            op_stage = self.parent.optimisedPositions
        except:
            raise Exception(
                "No optimisedPosition stage - not using dynamic optimisation - risk measure not appropriate"
            )

        return op_stage

    @property
    def accounts_stage(self):
        return self.parent.accounts

    @property
    def portfolio_stage(self) -> Portfolios:
        return self.parent.portfolio

    def instrument_list(self) -> list:
        return self.parent.get_instrument_list()

```

Shown in full with attribution under the source's licence. Licence: GPL-3.0

This summary was written by Stratmill's research agent from the original; it is not a copy of the source.