{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Accompanying notebook\n",
    "\n",
    "Reference implementation for *Understanding Portfolio Performance: Contribution and\n",
    "Attribution*. It contains only the code required to produce the six figures in the\n",
    "article, with fixed seeds.\n",
    "\n",
    "Run top to bottom. The final cell writes every figure to `figs/` as a PDF, under the\n",
    "file names used by the article source.\n",
    "\n",
    "Requires `numpy` and `matplotlib`. LaTeX typesetting is used when available and falls\n",
    "back to mathtext otherwise."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1  Style and palette"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "\n",
    "from abc import ABC, abstractmethod\n",
    "from dataclasses import dataclass, field\n",
    "from pathlib import Path\n",
    "\n",
    "import matplotlib as mpl\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "from matplotlib.patches import Rectangle\n",
    "\n",
    "INK = \"#16202B\"\n",
    "MUTED = \"#69747F\"\n",
    "RULE = \"#D5DBDF\"\n",
    "BLUE = \"#2B4C7E\"\n",
    "GREY = \"#9AA3AB\"\n",
    "TEAL = \"#356F7E\"\n",
    "RUST = \"#A8452A\"\n",
    "OCHRE = \"#8A6D1F\"\n",
    "MOSS = \"#41684E\"\n",
    "EFFECT_COLOURS = (TEAL, RUST, OCHRE)\n",
    "EFFECT_NAMES = (\"Allocation\", \"Selection\", \"Interaction\")\n",
    "\n",
    "mpl.rcParams.update(\n",
    "    {\n",
    "        \"text.usetex\": True,\n",
    "        \"text.latex.preamble\": r\"\\usepackage[T1]{fontenc}\"\n",
    "                               r\"\\usepackage{lmodern}\"\n",
    "                               r\"\\usepackage{textcomp}\",\n",
    "        \"font.family\": \"serif\",\n",
    "        \"font.size\": 12.5,\n",
    "        \"text.color\": INK,\n",
    "        \"axes.labelcolor\": INK,\n",
    "        \"axes.edgecolor\": RULE,\n",
    "        \"xtick.color\": MUTED,\n",
    "        \"ytick.color\": MUTED,\n",
    "        \"axes.linewidth\": 0.9,\n",
    "        \"xtick.major.size\": 0,\n",
    "        \"ytick.major.size\": 3,\n",
    "        \"figure.facecolor\": \"none\",\n",
    "        \"axes.facecolor\": \"none\",\n",
    "        \"savefig.facecolor\": \"none\",\n",
    "    }\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.1  Typesetting fallback\n",
    "\n",
    "`text.usetex` needs a working LaTeX toolchain with `type1ec.sty`. Where that is not\n",
    "installed the notebook would otherwise fail at the first `savefig`, so the setting is\n",
    "probed once and downgraded to mathtext if it does not work. Figures still render; only\n",
    "the glyphs differ."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def _usetex_available() -> bool:\n",
    "    import shutil\n",
    "    if not shutil.which(\"latex\"):\n",
    "        return False\n",
    "    try:\n",
    "        fig = plt.figure()\n",
    "        fig.text(0.5, 0.5, r\"$R_P$\")\n",
    "        fig.canvas.draw()\n",
    "        plt.close(fig)\n",
    "        return True\n",
    "    except Exception:\n",
    "        plt.close(\"all\")\n",
    "        return False\n",
    "\n",
    "\n",
    "if not _usetex_available():\n",
    "    mpl.rcParams.update({\n",
    "        \"text.usetex\": False,\n",
    "        \"mathtext.fontset\": \"cm\",\n",
    "        \"font.serif\": [\"Latin Modern Roman\", \"CMU Serif\", \"DejaVu Serif\"],\n",
    "    })\n",
    "    print(\"LaTeX unavailable; using mathtext fallback. Figures render, glyphs differ slightly.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2  Panel\n",
    "\n",
    "`SegmentPanel` holds portfolio and benchmark weights and returns as period-by-segment\n",
    "matrices and derives everything else. A single period is a panel with one row."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@dataclass(frozen=True)\n",
    "class SegmentPanel:\n",
    "\n",
    "    portfolio_weights: np.ndarray\n",
    "    benchmark_weights: np.ndarray\n",
    "    portfolio_segment_returns: np.ndarray\n",
    "    benchmark_segment_returns: np.ndarray\n",
    "    segment_names: tuple[str, ...] = field(default=())\n",
    "\n",
    "    FIELDS = (\n",
    "        \"portfolio_weights\",\n",
    "        \"benchmark_weights\",\n",
    "        \"portfolio_segment_returns\",\n",
    "        \"benchmark_segment_returns\",\n",
    "    )\n",
    "\n",
    "    @staticmethod\n",
    "    def _as_matrix(values) -> np.ndarray:\n",
    "        array = np.asarray(values, dtype=float)\n",
    "        return array[None, :] if array.ndim == 1 else array\n",
    "\n",
    "    def __post_init__(self) -> None:\n",
    "        for name in self.FIELDS:\n",
    "            object.__setattr__(self, name, self._as_matrix(getattr(self, name)))\n",
    "        names = tuple(self.segment_names) or tuple(\n",
    "            f\"Segment {i + 1}\" for i in range(self.portfolio_weights.shape[1])\n",
    "        )\n",
    "        object.__setattr__(self, \"segment_names\", names)\n",
    "\n",
    "    @property\n",
    "    def periods(self) -> int:\n",
    "        return int(self.portfolio_weights.shape[0])\n",
    "\n",
    "    @property\n",
    "    def weight_differences(self) -> np.ndarray:\n",
    "        return self.portfolio_weights - self.benchmark_weights\n",
    "\n",
    "    @property\n",
    "    def return_differences(self) -> np.ndarray:\n",
    "        return self.portfolio_segment_returns - self.benchmark_segment_returns\n",
    "\n",
    "    @property\n",
    "    def portfolio_returns(self) -> np.ndarray:\n",
    "        return np.sum(self.portfolio_weights * self.portfolio_segment_returns, axis=1)\n",
    "\n",
    "    @property\n",
    "    def benchmark_returns(self) -> np.ndarray:\n",
    "        return np.sum(self.benchmark_weights * self.benchmark_segment_returns, axis=1)\n",
    "\n",
    "    @property\n",
    "    def allocation_notional_returns(self) -> np.ndarray:\n",
    "        return np.sum(self.portfolio_weights * self.benchmark_segment_returns, axis=1)\n",
    "\n",
    "    @property\n",
    "    def selection_notional_returns(self) -> np.ndarray:\n",
    "        return np.sum(self.benchmark_weights * self.portfolio_segment_returns, axis=1)\n",
    "\n",
    "    @property\n",
    "    def active_returns(self) -> np.ndarray:\n",
    "        return self.portfolio_returns - self.benchmark_returns\n",
    "\n",
    "    @property\n",
    "    def growth_before(self) -> np.ndarray:\n",
    "        growth = np.cumprod(1.0 + self.portfolio_returns)\n",
    "        return np.concatenate(([1.0], growth[:-1]))\n",
    "\n",
    "    @property\n",
    "    def compounded_portfolio_return(self) -> float:\n",
    "        return float(np.prod(1.0 + self.portfolio_returns) - 1.0)\n",
    "\n",
    "    @property\n",
    "    def compounded_benchmark_return(self) -> float:\n",
    "        return float(np.prod(1.0 + self.benchmark_returns) - 1.0)\n",
    "\n",
    "    @property\n",
    "    def compounded_active_return(self) -> float:\n",
    "        return self.compounded_portfolio_return - self.compounded_benchmark_return"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3  Single period attribution\n",
    "\n",
    "Brinson and Fachler (1985). Allocation measures the weight difference against the\n",
    "benchmark segment's return relative to the whole benchmark, selection holds the weight\n",
    "at the benchmark weight, and interaction is what remains."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@dataclass(frozen=True)\n",
    "class SinglePeriodResult:\n",
    "\n",
    "    allocation: np.ndarray\n",
    "    selection: np.ndarray\n",
    "    interaction: np.ndarray\n",
    "    panel: SegmentPanel\n",
    "\n",
    "    @property\n",
    "    def sector_totals(self) -> np.ndarray:\n",
    "        return self.allocation + self.selection + self.interaction\n",
    "\n",
    "    @property\n",
    "    def period_effects(self) -> np.ndarray:\n",
    "        return np.column_stack(\n",
    "            (\n",
    "                self.allocation.sum(axis=1),\n",
    "                self.selection.sum(axis=1),\n",
    "                self.interaction.sum(axis=1),\n",
    "            )\n",
    "        )\n",
    "\n",
    "\n",
    "class BrinsonFachler1985:\n",
    "\n",
    "    def allocation_effect(self, panel: SegmentPanel) -> np.ndarray:\n",
    "        relative = panel.benchmark_segment_returns - panel.benchmark_returns[:, None]\n",
    "        return panel.weight_differences * relative\n",
    "\n",
    "    def selection_effect(self, panel: SegmentPanel) -> np.ndarray:\n",
    "        return panel.benchmark_weights * panel.return_differences\n",
    "\n",
    "    def interaction_effect(self, panel: SegmentPanel) -> np.ndarray:\n",
    "        return panel.weight_differences * panel.return_differences\n",
    "\n",
    "    def decompose(self, panel: SegmentPanel) -> SinglePeriodResult:\n",
    "        return SinglePeriodResult(\n",
    "            allocation=self.allocation_effect(panel),\n",
    "            selection=self.selection_effect(panel),\n",
    "            interaction=self.interaction_effect(panel),\n",
    "            panel=panel,\n",
    "        )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4  Linking\n",
    "\n",
    "Five methods, each reconciling the aggregate effects with the compounded active return\n",
    "\u2014 `Linker.link` raises otherwise \u2014 while distributing the compounding adjustment\n",
    "differently, which is the point of the linking figure. `Menchero2000` uses the\n",
    "limiting scale $M=(1+R_P^{(1:T)})^{(T-1)/T}$ when the compounded portfolio and\n",
    "benchmark returns coincide, and zeroes the period corrections only when every period\n",
    "active return is zero, not merely when they sum to zero."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Linker(ABC):\n",
    "\n",
    "    name: str\n",
    "\n",
    "    def link(self, result: SinglePeriodResult) -> np.ndarray:\n",
    "        totals = self._adjust(result).sum(axis=0)\n",
    "\n",
    "        if not np.isclose(\n",
    "            totals.sum(),\n",
    "            result.panel.compounded_active_return,\n",
    "            atol=1e-10,\n",
    "        ):\n",
    "            raise ValueError(\"Linked effects do not reconcile.\")\n",
    "\n",
    "        return totals\n",
    "\n",
    "    @abstractmethod\n",
    "    def _adjust(self, result: SinglePeriodResult) -> np.ndarray:\n",
    "        ...\n",
    "\n",
    "\n",
    "class Carino1999(Linker):\n",
    "\n",
    "    name = \"Cari\\u00f1o (1999)\"\n",
    "\n",
    "    @staticmethod\n",
    "    def _coefficient(portfolio, benchmark) -> np.ndarray:\n",
    "        portfolio = np.atleast_1d(np.asarray(portfolio, dtype=float))\n",
    "        benchmark = np.atleast_1d(np.asarray(benchmark, dtype=float))\n",
    "        gap = portfolio - benchmark\n",
    "        limit = 1.0 / (1.0 + portfolio)\n",
    "        with np.errstate(divide=\"ignore\", invalid=\"ignore\"):\n",
    "            ratio = (np.log1p(portfolio) - np.log1p(benchmark)) / gap\n",
    "        return np.where(np.abs(gap) < 1e-12, limit, ratio)\n",
    "\n",
    "    def _adjust(self, result: SinglePeriodResult) -> np.ndarray:\n",
    "        panel = result.panel\n",
    "        period_k = self._coefficient(panel.portfolio_returns, panel.benchmark_returns)\n",
    "        horizon_k = float(\n",
    "            self._coefficient(\n",
    "                [panel.compounded_portfolio_return],\n",
    "                [panel.compounded_benchmark_return],\n",
    "            )[0]\n",
    "        )\n",
    "        return result.period_effects * (period_k / horizon_k)[:, None]\n",
    "\n",
    "\n",
    "class Menchero2000(Linker):\n",
    "\n",
    "    name = \"Menchero (2000)\"\n",
    "\n",
    "    def _adjust(self, result: SinglePeriodResult) -> np.ndarray:\n",
    "        panel = result.panel\n",
    "        periods = panel.periods\n",
    "        total_portfolio = panel.compounded_portfolio_return\n",
    "        total_benchmark = panel.compounded_benchmark_return\n",
    "        active = panel.active_returns\n",
    "\n",
    "        root_gap = (1.0 + total_portfolio) ** (1.0 / periods) - (\n",
    "            1.0 + total_benchmark\n",
    "        ) ** (1.0 / periods)\n",
    "\n",
    "        if abs(root_gap) < 1e-14:\n",
    "            # Removable singularity at equal compounded returns.\n",
    "            scale = (1.0 + total_portfolio) ** ((periods - 1.0) / periods)\n",
    "        else:\n",
    "            scale = ((total_portfolio - total_benchmark) / periods) / root_gap\n",
    "\n",
    "        shortfall = (total_portfolio - total_benchmark) - scale * float(active.sum())\n",
    "        squared = float(np.sum(active**2))\n",
    "\n",
    "        # A zero cumulative active return does not imply zero corrections; the\n",
    "        # corrections vanish only when every period active return is zero.\n",
    "        if squared < 1e-24:\n",
    "            correction = np.zeros_like(active)\n",
    "        else:\n",
    "            correction = shortfall * active / squared\n",
    "\n",
    "        return result.period_effects * (scale + correction)[:, None]\n",
    "\n",
    "\n",
    "class GRAP1997(Linker):\n",
    "\n",
    "    name = \"GRAP (1997)\"\n",
    "\n",
    "    def _adjust(self, result: SinglePeriodResult) -> np.ndarray:\n",
    "        panel = result.panel\n",
    "        benchmark_growth = np.cumprod((1.0 + panel.benchmark_returns)[::-1])[::-1]\n",
    "        after = np.concatenate((benchmark_growth[1:], [1.0]))\n",
    "        return result.period_effects * (panel.growth_before * after)[:, None]\n",
    "\n",
    "\n",
    "class Frongello2002(Linker):\n",
    "\n",
    "    name = \"Frongello (2002)\"\n",
    "\n",
    "    def _adjust(self, result: SinglePeriodResult) -> np.ndarray:\n",
    "        panel = result.panel\n",
    "        effects = result.period_effects\n",
    "        adjusted = np.zeros_like(effects)\n",
    "        running = np.zeros(effects.shape[1])\n",
    "        for t in range(panel.periods):\n",
    "            adjusted[t] = effects[t] * panel.growth_before[t] + panel.benchmark_returns[t] * running\n",
    "            running = running + adjusted[t]\n",
    "        return adjusted\n",
    "\n",
    "\n",
    "class DaviesLaker2001(Linker):\n",
    "\n",
    "    name = \"Davies and Laker (2001)\"\n",
    "\n",
    "    def horizon_effects(self, result: SinglePeriodResult) -> np.ndarray:\n",
    "        panel = result.panel\n",
    "        benchmark = float(np.prod(1.0 + panel.benchmark_returns))\n",
    "        allocation_path = float(np.prod(1.0 + panel.allocation_notional_returns))\n",
    "        selection_path = float(np.prod(1.0 + panel.selection_notional_returns))\n",
    "        portfolio = float(np.prod(1.0 + panel.portfolio_returns))\n",
    "        return np.array(\n",
    "            [\n",
    "                allocation_path - benchmark,\n",
    "                selection_path - benchmark,\n",
    "                portfolio - selection_path - allocation_path + benchmark,\n",
    "            ]\n",
    "        )\n",
    "\n",
    "    def _adjust(self, result: SinglePeriodResult) -> np.ndarray:\n",
    "        totals = self.horizon_effects(result)\n",
    "        raw = result.period_effects\n",
    "        column_sums = raw.sum(axis=0)\n",
    "        scale = np.divide(\n",
    "            totals, column_sums, out=np.zeros_like(totals), where=np.abs(column_sums) > 1e-15\n",
    "        )\n",
    "        adjusted = raw * scale\n",
    "        adjusted[-1] = adjusted[-1] + (totals - adjusted.sum(axis=0))\n",
    "        return adjusted\n",
    "\n",
    "\n",
    "ARITHMETIC_LINKERS: tuple[Linker, ...] = (\n",
    "    Carino1999(),\n",
    "    Menchero2000(),\n",
    "    GRAP1997(),\n",
    "    Frongello2002(),\n",
    "    DaviesLaker2001(),\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5  Combining contributions across periods\n",
    "\n",
    "Method A uses the initial weight and the compounded segment return, the direct sum adds\n",
    "the period contributions unchanged, and Method B places each period contribution on a\n",
    "common capital base. Only Method B reconciles in general."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@dataclass(frozen=True)\n",
    "class ContributionResult:\n",
    "\n",
    "    method: str\n",
    "    segment_contributions: np.ndarray\n",
    "    panel: SegmentPanel\n",
    "\n",
    "    @property\n",
    "    def total(self) -> float:\n",
    "        return float(self.segment_contributions.sum())\n",
    "\n",
    "    @property\n",
    "    def gap(self) -> float:\n",
    "        return self.total - self.panel.compounded_portfolio_return\n",
    "\n",
    "    @property\n",
    "    def reconciles(self) -> bool:\n",
    "        return bool(abs(self.gap) < 1e-12)\n",
    "\n",
    "\n",
    "class ContributionMethod(ABC):\n",
    "\n",
    "    name: str\n",
    "\n",
    "    def combine(self, panel: SegmentPanel) -> ContributionResult:\n",
    "        return ContributionResult(self.name, self._contributions(panel), panel)\n",
    "\n",
    "    @abstractmethod\n",
    "    def _contributions(self, panel: SegmentPanel) -> np.ndarray:\n",
    "        ...\n",
    "\n",
    "\n",
    "class InitialWeightMethodA(ContributionMethod):\n",
    "\n",
    "    name = \"A, initial weight and compounded segment return\"\n",
    "\n",
    "    def _contributions(self, panel: SegmentPanel) -> np.ndarray:\n",
    "        compounded = np.prod(1.0 + panel.portfolio_segment_returns, axis=0) - 1.0\n",
    "        return panel.portfolio_weights[0] * compounded\n",
    "\n",
    "\n",
    "class DirectSum(ContributionMethod):\n",
    "\n",
    "    name = \"Direct sum of period contributions\"\n",
    "\n",
    "    def _contributions(self, panel: SegmentPanel) -> np.ndarray:\n",
    "        return np.sum(panel.portfolio_weights * panel.portfolio_segment_returns, axis=0)\n",
    "\n",
    "\n",
    "class CommonCapitalBaseMethodB(ContributionMethod):\n",
    "\n",
    "    name = \"B, period contributions on a common capital base\"\n",
    "\n",
    "    def _contributions(self, panel: SegmentPanel) -> np.ndarray:\n",
    "        period = panel.portfolio_weights * panel.portfolio_segment_returns\n",
    "        return np.sum(period * panel.growth_before[:, None], axis=0)\n",
    "\n",
    "\n",
    "CONTRIBUTION_METHODS: tuple[ContributionMethod, ...] = (\n",
    "    InitialWeightMethodA(),\n",
    "    DirectSum(),\n",
    "    CommonCapitalBaseMethodB(),\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.1  Two asset test book\n",
    "\n",
    "`TwoAssetPath` runs the same twelve monthly segment returns two ways, rebalanced to the\n",
    "target weights at each month end and left to drift, which isolates Method A's error as a\n",
    "function of trading. Contribution needs no benchmark, so the panel carries a benchmark of\n",
    "zero weights and zero returns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def contribution_panel(weights, returns, names=()) -> SegmentPanel:\n",
    "    \"\"\"A panel with no benchmark, used where only contribution is required.\"\"\"\n",
    "    weights = SegmentPanel._as_matrix(weights)\n",
    "    returns = SegmentPanel._as_matrix(returns)\n",
    "    return SegmentPanel(\n",
    "        portfolio_weights=weights,\n",
    "        benchmark_weights=np.zeros_like(weights),\n",
    "        portfolio_segment_returns=returns,\n",
    "        benchmark_segment_returns=np.zeros_like(returns),\n",
    "        segment_names=names or tuple(f\"Segment {i+1}\" for i in range(weights.shape[1])),\n",
    "    )\n",
    "\n",
    "\n",
    "@dataclass(frozen=True)\n",
    "class TwoAssetPath:\n",
    "    \"\"\"Twelve monthly periods for a two segment book, run two ways.\"\"\"\n",
    "\n",
    "    segment_returns: np.ndarray\n",
    "    target_weights: np.ndarray\n",
    "    names: tuple[str, ...] = (\"Equity\", \"Bonds\")\n",
    "\n",
    "    @classmethod\n",
    "    def from_annual(\n",
    "        cls, annual_returns, volatilities, target_weights, periods=12, seed=11\n",
    "    ) -> \"TwoAssetPath\":\n",
    "        generator = np.random.default_rng(seed)\n",
    "        columns = []\n",
    "        for total, volatility in zip(annual_returns, volatilities):\n",
    "            shocks = np.exp(generator.normal(0.0, volatility, periods))\n",
    "            shocks = shocks / shocks.mean()\n",
    "            scale = ((1.0 + total) / np.prod(shocks)) ** (1.0 / periods)\n",
    "            columns.append(shocks * scale - 1.0)\n",
    "        return cls(np.column_stack(columns), np.asarray(target_weights, dtype=float))\n",
    "\n",
    "    def panel(self, rebalance: bool) -> SegmentPanel:\n",
    "        weights = np.zeros_like(self.segment_returns)\n",
    "        current = self.target_weights.copy()\n",
    "        for t, returns in enumerate(self.segment_returns):\n",
    "            weights[t] = current\n",
    "            period_return = float(current @ returns)\n",
    "            current = current * (1.0 + returns) / (1.0 + period_return)\n",
    "            if rebalance:\n",
    "                current = self.target_weights.copy()\n",
    "        return contribution_panel(weights, self.segment_returns, self.names)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 6  Extensions\n",
    "\n",
    "Weighted duration (Van Breukelen, 2000), multicurrency in additive form, and a\n",
    "returns based active factor model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@dataclass(frozen=True)\n",
    "class WeightedDurationInput:\n",
    "\n",
    "    portfolio_weights: np.ndarray\n",
    "    benchmark_weights: np.ndarray\n",
    "    portfolio_durations: np.ndarray\n",
    "    benchmark_durations: np.ndarray\n",
    "    portfolio_yield_changes: np.ndarray\n",
    "    benchmark_yield_changes: np.ndarray\n",
    "\n",
    "    FIELDS = (\n",
    "        \"portfolio_weights\",\n",
    "        \"benchmark_weights\",\n",
    "        \"portfolio_durations\",\n",
    "        \"benchmark_durations\",\n",
    "        \"portfolio_yield_changes\",\n",
    "        \"benchmark_yield_changes\",\n",
    "    )\n",
    "\n",
    "    def __post_init__(self) -> None:\n",
    "        for name in self.FIELDS:\n",
    "            object.__setattr__(\n",
    "                self, name, np.asarray(getattr(self, name), dtype=float).ravel()\n",
    "            )\n",
    "\n",
    "    @property\n",
    "    def portfolio_duration(self) -> float:\n",
    "        return float(np.sum(self.portfolio_weights * self.portfolio_durations))\n",
    "\n",
    "    @property\n",
    "    def benchmark_duration(self) -> float:\n",
    "        return float(np.sum(self.benchmark_weights * self.benchmark_durations))\n",
    "\n",
    "    @property\n",
    "    def portfolio_segment_returns(self) -> np.ndarray:\n",
    "        return -self.portfolio_durations * self.portfolio_yield_changes\n",
    "\n",
    "    @property\n",
    "    def benchmark_segment_returns(self) -> np.ndarray:\n",
    "        return -self.benchmark_durations * self.benchmark_yield_changes\n",
    "\n",
    "    @property\n",
    "    def portfolio_return(self) -> float:\n",
    "        return float(np.sum(self.portfolio_weights * self.portfolio_segment_returns))\n",
    "\n",
    "    @property\n",
    "    def benchmark_return(self) -> float:\n",
    "        return float(np.sum(self.benchmark_weights * self.benchmark_segment_returns))\n",
    "\n",
    "    @property\n",
    "    def active_return(self) -> float:\n",
    "        return self.portfolio_return - self.benchmark_return\n",
    "\n",
    "\n",
    "@dataclass(frozen=True)\n",
    "class WeightedDurationResult:\n",
    "\n",
    "    overall_duration: float\n",
    "    market_allocation_effects: np.ndarray\n",
    "    issue_selection_effects: np.ndarray\n",
    "\n",
    "    def as_dict(self) -> dict[str, float]:\n",
    "        return {\n",
    "            \"Overall duration\": self.overall_duration,\n",
    "            \"Market allocation\": float(self.market_allocation_effects.sum()),\n",
    "            \"Issue selection\": float(self.issue_selection_effects.sum()),\n",
    "        }\n",
    "\n",
    "\n",
    "class WeightedDurationAttribution:\n",
    "\n",
    "    def decompose(self, data: WeightedDurationInput) -> WeightedDurationResult:\n",
    "        benchmark_duration = data.benchmark_duration\n",
    "        portfolio_duration = data.portfolio_duration\n",
    "\n",
    "        average_yield_change = float(\n",
    "            np.sum(\n",
    "                data.benchmark_weights\n",
    "                * data.benchmark_durations\n",
    "                * data.benchmark_yield_changes\n",
    "            )\n",
    "            / benchmark_duration\n",
    "        )\n",
    "\n",
    "        portfolio_duration_weights = (\n",
    "            data.portfolio_weights * data.portfolio_durations / portfolio_duration\n",
    "        )\n",
    "        benchmark_duration_weights = (\n",
    "            data.benchmark_weights * data.benchmark_durations / benchmark_duration\n",
    "        )\n",
    "\n",
    "        return WeightedDurationResult(\n",
    "            overall_duration=float(\n",
    "                (portfolio_duration - benchmark_duration) * (-average_yield_change)\n",
    "            ),\n",
    "            market_allocation_effects=(\n",
    "                portfolio_duration\n",
    "                * (portfolio_duration_weights - benchmark_duration_weights)\n",
    "                * (-data.benchmark_yield_changes + average_yield_change)\n",
    "            ),\n",
    "            issue_selection_effects=(\n",
    "                data.portfolio_weights\n",
    "                * data.portfolio_durations\n",
    "                * (-data.portfolio_yield_changes + data.benchmark_yield_changes)\n",
    "            ),\n",
    "        )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@dataclass(frozen=True)\n",
    "class CurrencyInput:\n",
    "\n",
    "    portfolio_weights: np.ndarray\n",
    "    benchmark_weights: np.ndarray\n",
    "    portfolio_local_returns: np.ndarray\n",
    "    benchmark_local_returns: np.ndarray\n",
    "    currency_returns: np.ndarray\n",
    "\n",
    "    FIELDS = (\n",
    "        \"portfolio_weights\",\n",
    "        \"benchmark_weights\",\n",
    "        \"portfolio_local_returns\",\n",
    "        \"benchmark_local_returns\",\n",
    "        \"currency_returns\",\n",
    "    )\n",
    "\n",
    "    def __post_init__(self) -> None:\n",
    "        for name in self.FIELDS:\n",
    "            object.__setattr__(\n",
    "                self, name, np.asarray(getattr(self, name), dtype=float).ravel()\n",
    "            )\n",
    "\n",
    "    @property\n",
    "    def portfolio_base_returns(self) -> np.ndarray:\n",
    "        return self.portfolio_local_returns + self.currency_returns\n",
    "\n",
    "    @property\n",
    "    def benchmark_base_returns(self) -> np.ndarray:\n",
    "        return self.benchmark_local_returns + self.currency_returns\n",
    "\n",
    "    @property\n",
    "    def benchmark_local_return(self) -> float:\n",
    "        return float(np.sum(self.benchmark_weights * self.benchmark_local_returns))\n",
    "\n",
    "    @property\n",
    "    def portfolio_base_return(self) -> float:\n",
    "        return float(np.sum(self.portfolio_weights * self.portfolio_base_returns))\n",
    "\n",
    "    @property\n",
    "    def benchmark_base_return(self) -> float:\n",
    "        return float(np.sum(self.benchmark_weights * self.benchmark_base_returns))\n",
    "\n",
    "    @property\n",
    "    def active_base_return(self) -> float:\n",
    "        return self.portfolio_base_return - self.benchmark_base_return\n",
    "\n",
    "\n",
    "@dataclass(frozen=True)\n",
    "class CurrencyResult:\n",
    "\n",
    "    market_allocation: float\n",
    "    security_selection: float\n",
    "    currency_allocation: float\n",
    "    interaction: float\n",
    "\n",
    "    def as_dict(self) -> dict[str, float]:\n",
    "        return {\n",
    "            \"Market allocation\": self.market_allocation,\n",
    "            \"Security selection\": self.security_selection,\n",
    "            \"Currency allocation\": self.currency_allocation,\n",
    "            \"Interaction\": self.interaction,\n",
    "        }\n",
    "\n",
    "\n",
    "class MulticurrencyAttribution:\n",
    "    \"\"\"Additive separation of market and currency effects.\"\"\"\n",
    "\n",
    "    def decompose(self, data: CurrencyInput) -> CurrencyResult:\n",
    "        weight_gap = data.portfolio_weights - data.benchmark_weights\n",
    "        return_gap = data.portfolio_local_returns - data.benchmark_local_returns\n",
    "        benchmark_currency = float(\n",
    "            np.sum(data.benchmark_weights * data.currency_returns)\n",
    "        )\n",
    "        return CurrencyResult(\n",
    "            market_allocation=float(\n",
    "                np.sum(\n",
    "                    weight_gap\n",
    "                    * (data.benchmark_local_returns - data.benchmark_local_return)\n",
    "                )\n",
    "            ),\n",
    "            security_selection=float(np.sum(data.benchmark_weights * return_gap)),\n",
    "            currency_allocation=float(\n",
    "                np.sum(weight_gap * (data.currency_returns - benchmark_currency))\n",
    "            ),\n",
    "            interaction=float(np.sum(weight_gap * return_gap)),\n",
    "        )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@dataclass(frozen=True)\n",
    "class ActiveFactorResult:\n",
    "\n",
    "    alpha: float\n",
    "    portfolio_betas: np.ndarray\n",
    "    benchmark_betas: np.ndarray\n",
    "    factor_names: tuple[str, ...]\n",
    "    factor_returns: np.ndarray\n",
    "    active_returns: np.ndarray\n",
    "\n",
    "    @property\n",
    "    def active_betas(self) -> np.ndarray:\n",
    "        return self.portfolio_betas - self.benchmark_betas\n",
    "\n",
    "    @property\n",
    "    def factor_contributions(self) -> np.ndarray:\n",
    "        return self.active_betas * self.factor_returns.mean(axis=0)\n",
    "\n",
    "    @property\n",
    "    def mean_active_return(self) -> float:\n",
    "        return float(self.active_returns.mean())\n",
    "\n",
    "\n",
    "class ActiveFactorAttribution:\n",
    "\n",
    "    @staticmethod\n",
    "    def _fit_one(returns: np.ndarray, design: np.ndarray) -> np.ndarray:\n",
    "        coefficients, *_ = np.linalg.lstsq(design, returns, rcond=None)\n",
    "        return coefficients\n",
    "\n",
    "    def fit(\n",
    "        self, portfolio_returns, benchmark_returns, factor_returns, factor_names=()\n",
    "    ) -> ActiveFactorResult:\n",
    "        portfolio = np.asarray(portfolio_returns, dtype=float).ravel()\n",
    "        benchmark = np.asarray(benchmark_returns, dtype=float).ravel()\n",
    "        factors = np.atleast_2d(np.asarray(factor_returns, dtype=float))\n",
    "\n",
    "        design = np.column_stack((np.ones(portfolio.size), factors))\n",
    "        portfolio_coefficients = self._fit_one(portfolio, design)\n",
    "        benchmark_coefficients = self._fit_one(benchmark, design)\n",
    "\n",
    "        names = tuple(factor_names) or tuple(\n",
    "            f\"Factor {i + 1}\" for i in range(factors.shape[1])\n",
    "        )\n",
    "        return ActiveFactorResult(\n",
    "            alpha=float(portfolio_coefficients[0] - benchmark_coefficients[0]),\n",
    "            portfolio_betas=portfolio_coefficients[1:],\n",
    "            benchmark_betas=benchmark_coefficients[1:],\n",
    "            factor_names=names,\n",
    "            factor_returns=factors,\n",
    "            active_returns=portfolio - benchmark,\n",
    "        )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 7  Data\n",
    "\n",
    "The worked quarter, the two asset book, the eight quarter linking panel, and the inputs\n",
    "for each extension. The linking panel's weights are renormalised so each row sums to one."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "QUARTER = SegmentPanel(\n",
    "    portfolio_weights=[0.40, 0.30, 0.30],\n",
    "    benchmark_weights=[0.25, 0.35, 0.40],\n",
    "    portfolio_segment_returns=[0.12, 0.04, 0.02],\n",
    "    benchmark_segment_returns=[0.10, 0.05, 0.03],\n",
    "    segment_names=(\"Technology\", \"Healthcare\", \"Energy\"),\n",
    ")\n",
    "QUARTER_RESULT = BrinsonFachler1985().decompose(QUARTER)\n",
    "\n",
    "BOOK = TwoAssetPath.from_annual(\n",
    "    annual_returns=(0.369, -0.056), volatilities=(0.035, 0.012),\n",
    "    target_weights=(0.60, 0.40),\n",
    ")\n",
    "REBALANCED = BOOK.panel(rebalance=True)\n",
    "BUY_AND_HOLD = BOOK.panel(rebalance=False)\n",
    "\n",
    "LINK_WP = np.array(\n",
    "    [\n",
    "        [0.2832, 0.2643, 0.4525],\n",
    "        [0.3325, 0.3118, 0.3558],\n",
    "        [0.2224, 0.2554, 0.5223],\n",
    "        [0.3240, 0.1887, 0.4873],\n",
    "        [0.3364, 0.3775, 0.2861],\n",
    "        [0.3012, 0.3552, 0.3436],\n",
    "        [0.2219, 0.2830, 0.4951],\n",
    "        [0.2076, 0.2721, 0.5203],\n",
    "    ]\n",
    ")\n",
    "LINK_WP = LINK_WP / LINK_WP.sum(axis=1, keepdims=True)\n",
    "LINK_WB = np.tile([0.25, 0.35, 0.40], (8, 1))\n",
    "LINK_RB = np.array(\n",
    "    [\n",
    "        [0.0382, 0.0828, -0.0063],\n",
    "        [0.1005, 0.0689, 0.0050],\n",
    "        [0.0614, -0.0368, 0.0559],\n",
    "        [0.1063, 0.0716, 0.0992],\n",
    "        [0.0340, 0.0378, -0.1272],\n",
    "        [0.0290, 0.0104, -0.0795],\n",
    "        [-0.2646, -0.0028, 0.1002],\n",
    "        [-0.0166, 0.0308, -0.0512],\n",
    "    ]\n",
    ")\n",
    "LINK_RP = np.array(\n",
    "    [\n",
    "        [0.0427, 0.1305, -0.0003],\n",
    "        [0.0912, 0.0924, 0.0374],\n",
    "        [0.0999, -0.0436, 0.0344],\n",
    "        [0.1289, 0.0672, 0.1183],\n",
    "        [0.0543, 0.0320, -0.1537],\n",
    "        [0.0893, -0.0007, -0.0571],\n",
    "        [-0.2967, 0.0046, 0.0887],\n",
    "        [-0.0130, -0.0198, -0.0759],\n",
    "    ]\n",
    ")\n",
    "LINK_PANEL = SegmentPanel(LINK_WP, LINK_WB, LINK_RP, LINK_RB, QUARTER.segment_names)\n",
    "LINK_RESULT = BrinsonFachler1985().decompose(LINK_PANEL)\n",
    "\n",
    "BONDS = WeightedDurationInput(\n",
    "    portfolio_weights=[0.50, 0.50],\n",
    "    benchmark_weights=[0.55, 0.45],\n",
    "    portfolio_durations=[6.5, 2.8],\n",
    "    benchmark_durations=[4.8, 3.2],\n",
    "    portfolio_yield_changes=[-0.0035, 0.0020],\n",
    "    benchmark_yield_changes=[-0.0030, 0.0015],\n",
    ")\n",
    "FIXED_INCOME = WeightedDurationAttribution().decompose(BONDS)\n",
    "\n",
    "GLOBAL_PORTFOLIO = CurrencyInput(\n",
    "    portfolio_weights=[0.60, 0.40],\n",
    "    benchmark_weights=[0.50, 0.50],\n",
    "    portfolio_local_returns=[0.055, 0.018],\n",
    "    benchmark_local_returns=[0.040, 0.030],\n",
    "    currency_returns=[0.030, -0.020],\n",
    ")\n",
    "CURRENCY = MulticurrencyAttribution().decompose(GLOBAL_PORTFOLIO)\n",
    "\n",
    "_generator = np.random.default_rng(11)\n",
    "FACTOR_RETURNS = _generator.normal([0.005, 0.002], 0.02, (72, 2))\n",
    "PORTFOLIO_RETURNS = (\n",
    "    0.0018 + FACTOR_RETURNS @ np.array([0.92, 0.38]) + _generator.normal(0.0, 0.004, 72)\n",
    ")\n",
    "BENCHMARK_RETURNS = (\n",
    "    0.0002 + FACTOR_RETURNS @ np.array([1.00, 0.05]) + _generator.normal(0.0, 0.002, 72)\n",
    ")\n",
    "FACTORS = ActiveFactorAttribution().fit(\n",
    "    PORTFOLIO_RETURNS, BENCHMARK_RETURNS, FACTOR_RETURNS, (\"Market\", \"Value\")\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 8  Article figures\n",
    "\n",
    "Each class writes one file under `figs/`, named as the article's `\\articlefigure`\n",
    "command expects."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class ArticleFigure(ABC):\n",
    "\n",
    "    name: str = \"figure\"\n",
    "    output: Path = Path(\"figs\")\n",
    "\n",
    "    @abstractmethod\n",
    "    def draw(self) -> plt.Figure:\n",
    "        ...\n",
    "\n",
    "    def save(self) -> None:\n",
    "        figure = self.draw()\n",
    "        self.output.mkdir(exist_ok=True)\n",
    "        figure.savefig(\n",
    "            self.output / f\"{self.name}.pdf\", format=\"pdf\",\n",
    "            bbox_inches=\"tight\", pad_inches=0.14, transparent=True,\n",
    "        )\n",
    "        plt.close(figure)\n",
    "\n",
    "    @staticmethod\n",
    "    def tidy(ax, left=True, bottom=True):\n",
    "        for side in (\"top\", \"right\"):\n",
    "            ax.spines[side].set_visible(False)\n",
    "        ax.spines[\"left\"].set_visible(left)\n",
    "        ax.spines[\"bottom\"].set_visible(bottom)\n",
    "        ax.tick_params(labelsize=11.5)\n",
    "        return ax\n",
    "\n",
    "    @staticmethod\n",
    "    def panel_title(ax, text, pad=12):\n",
    "        ax.set_title(text, loc=\"left\", fontsize=12.8, color=INK, pad=pad)\n",
    "\n",
    "    @staticmethod\n",
    "    def signed(value, digits=2) -> str:\n",
    "        return f\"${value:+.{digits}f}$\"\n",
    "\n",
    "    @staticmethod\n",
    "    def plain(value, digits=2) -> str:\n",
    "        return f\"${value:.{digits}f}$\"\n",
    "\n",
    "    @staticmethod\n",
    "    def effect_legend(fig, y=0.935) -> None:\n",
    "        handles = [Rectangle((0, 0), 1, 1, color=colour) for colour in EFFECT_COLOURS]\n",
    "        fig.legend(\n",
    "            handles, EFFECT_NAMES,\n",
    "            loc=\"lower left\", bbox_to_anchor=(0.012, y), ncol=3, frameon=False,\n",
    "            fontsize=12, handlelength=1.25, handleheight=0.95, columnspacing=2.0,\n",
    "            handletextpad=0.55,\n",
    "        )\n",
    "\n",
    "    @staticmethod\n",
    "    def sector_panel(ax, values, colours, title, ylabel, digits=2):\n",
    "        ArticleFigure.tidy(ax)\n",
    "        x = np.arange(len(values), dtype=float)\n",
    "        ax.axvspan(0.52, 1.48, color=GREY, alpha=0.10, zorder=0)\n",
    "        ax.bar(x, values, 0.5, color=colours, zorder=3)\n",
    "        for centre, value in zip(x, values):\n",
    "            above = value >= 0\n",
    "            ax.text(centre, value + (0.13 if above else -0.13),\n",
    "                    ArticleFigure.signed(value, digits), ha=\"center\",\n",
    "                    va=\"bottom\" if above else \"top\", fontsize=11.8, color=INK)\n",
    "        ax.axhline(0, color=INK, lw=1.0, zorder=4)\n",
    "        ax.set_xticks(x)\n",
    "        ax.set_xticklabels(QUARTER.segment_names, fontsize=12.2, color=INK)\n",
    "        ax.set_ylim(-1.15, 5.75)\n",
    "        ax.set_ylabel(ylabel, fontsize=11.6, color=MUTED, linespacing=1.4)\n",
    "        ax.grid(axis=\"y\", color=RULE, lw=0.7, zorder=0)\n",
    "        ax.set_axisbelow(True)\n",
    "        ArticleFigure.panel_title(ax, title)\n",
    "\n",
    "    @staticmethod\n",
    "    def component_panel(ax, entries, total_label, total_value):\n",
    "        ArticleFigure.tidy(ax, left=False)\n",
    "        labels = [label for label, _, _ in entries] + [total_label]\n",
    "        amounts = [value for _, value, _ in entries] + [total_value]\n",
    "        colours = [colour for _, _, colour in entries] + [INK]\n",
    "        y = np.arange(len(labels))[::-1].astype(float)\n",
    "        y[-1] -= 0.45\n",
    "\n",
    "        ax.barh(y, amounts, 0.56, color=colours, zorder=3)\n",
    "        span = max(abs(min(amounts)), abs(max(amounts)), 1.0)\n",
    "        for position, amount, colour in zip(y, amounts, colours):\n",
    "            forward = amount >= 0\n",
    "            ax.text(amount + (0.028 if forward else -0.028) * span, position,\n",
    "                    ArticleFigure.signed(amount, 1), va=\"center\",\n",
    "                    ha=\"left\" if forward else \"right\", fontsize=11.2, color=colour)\n",
    "\n",
    "        low = min(0.0, min(amounts))\n",
    "        high = max(0.0, max(amounts))\n",
    "        reach = high - low\n",
    "        limits = (low - 0.10 * reach, high + 0.20 * reach)\n",
    "\n",
    "        ax.axvline(0, color=INK, lw=1.0, zorder=4)\n",
    "        ax.hlines(y[-1] + 0.72, limits[0], limits[1], color=RULE, lw=0.9, zorder=2)\n",
    "        ax.set_yticks(y)\n",
    "        ax.set_yticklabels(labels, fontsize=11.8, color=INK)\n",
    "        ax.tick_params(axis=\"y\", length=0)\n",
    "        ax.set_xlim(*limits)\n",
    "        ax.set_xlabel(\"Basis points\", fontsize=11.5, color=MUTED, labelpad=6)\n",
    "        ax.grid(axis=\"x\", color=RULE, lw=0.7)\n",
    "        ax.set_axisbelow(True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class ContributionFigure(ArticleFigure):\n",
    "    \"\"\"Panel (a): sector contributions over the worked quarter, summing to R_P.\"\"\"\n",
    "\n",
    "    name = \"contribution_panel_a\"\n",
    "\n",
    "    def draw(self) -> plt.Figure:\n",
    "        contribution = (QUARTER.portfolio_weights * QUARTER.portfolio_segment_returns)[0] * 100\n",
    "\n",
    "        fig, ax = plt.subplots(figsize=(7.8, 4.4))\n",
    "        self.sector_panel(\n",
    "            ax, contribution, [TEAL] * len(contribution),\n",
    "            r\"(a) Contribution, summing to $R_P$\",\n",
    "            \"Contribution to portfolio return,\\npercentage points\",\n",
    "        )\n",
    "\n",
    "        fig.subplots_adjust(left=0.135, right=0.99, bottom=0.125, top=0.875)\n",
    "        return fig"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class AttributionFigure(ArticleFigure):\n",
    "    \"\"\"Panel (b): total sector effects, summing to the arithmetic active return.\"\"\"\n",
    "\n",
    "    name = \"attribution_panel_b\"\n",
    "\n",
    "    def draw(self) -> plt.Figure:\n",
    "        attribution = QUARTER_RESULT.sector_totals[0] * 100\n",
    "\n",
    "        fig, ax = plt.subplots(figsize=(7.8, 4.4))\n",
    "        self.sector_panel(\n",
    "            ax, attribution, [TEAL if value > 0 else RUST for value in attribution],\n",
    "            r\"(b) Attribution, summing to $R_P-R_B$\",\n",
    "            \"Total effect on active return,\\npercentage points\",\n",
    "        )\n",
    "\n",
    "        ax.annotate(\"positive contribution,\\nnegative attribution\",\n",
    "                    xy=(1, attribution[1]), xytext=(1, 2.95), ha=\"center\",\n",
    "                    fontsize=11.4, color=RUST,\n",
    "                    arrowprops=dict(arrowstyle=\"-\", color=RUST, lw=1.0, shrinkB=6))\n",
    "\n",
    "        fig.subplots_adjust(left=0.135, right=0.99, bottom=0.125, top=0.875)\n",
    "        return fig"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class ContributionMethodsFigure(ArticleFigure):\n",
    "    \"\"\"Method A, the direct sum, and Method B under two portfolio paths.\"\"\"\n",
    "\n",
    "    name = \"contribution_methods_panel_a\"\n",
    "\n",
    "    CASES = ((\"Monthly rebalancing\", REBALANCED), (\"No trading\", BUY_AND_HOLD))\n",
    "    SHORT = (\"A\", \"direct sum\", \"B\")\n",
    "\n",
    "    def draw(self) -> plt.Figure:\n",
    "        fig, ax = plt.subplots(figsize=(8.8, 4.8))\n",
    "        self.tidy(ax, bottom=False)\n",
    "\n",
    "        colours = (RUST, GREY, TEAL)\n",
    "        for group, (label, panel) in enumerate(self.CASES):\n",
    "            base = group * 4.4\n",
    "            true = panel.compounded_portfolio_return * 100\n",
    "            for offset, method in enumerate(CONTRIBUTION_METHODS):\n",
    "                outcome = method.combine(panel)\n",
    "                ax.bar(base + offset, outcome.total * 100, 0.72,\n",
    "                       color=colours[offset], zorder=3)\n",
    "                note = \"exact\" if outcome.reconciles else f\"${outcome.gap * 1e4:+.0f}$ bp\"\n",
    "                ax.text(base + offset, outcome.total * 50, note, ha=\"center\", va=\"center\",\n",
    "                        rotation=90, fontsize=11.2, color=\"white\", zorder=5)\n",
    "                ax.text(base + offset, -0.7, self.SHORT[offset], ha=\"center\", va=\"top\",\n",
    "                        fontsize=11.4, color=MUTED)\n",
    "            ax.plot([base - 0.6, base + 2.9], [true, true], color=INK, lw=1.1, zorder=6)\n",
    "            ax.text(base + 2.98, true, self.plain(true), ha=\"left\", va=\"center\",\n",
    "                    fontsize=11.6, color=INK)\n",
    "            ax.text(base + 1, -2.1, label, ha=\"center\", va=\"top\", fontsize=12.2, color=INK)\n",
    "\n",
    "        ax.set_xlim(-1.0, 8.9)\n",
    "        ax.set_ylim(0, 23)\n",
    "        ax.set_xticks([])\n",
    "        ax.set_ylabel(\"Sum of segment contributions, percent\",\n",
    "                      fontsize=11.8, color=MUTED)\n",
    "        ax.grid(axis=\"y\", color=RULE, lw=0.7, zorder=0)\n",
    "        ax.set_axisbelow(True)\n",
    "        self.panel_title(ax, \"Three ways of combining contributions\")\n",
    "\n",
    "        fig.subplots_adjust(left=0.105, right=0.985, bottom=0.145, top=0.9)\n",
    "        return fig"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class LinkersFigure(ArticleFigure):\n",
    "    \"\"\"Unlinked and linked effects over the eight quarter panel.\"\"\"\n",
    "\n",
    "    name = \"linkers\"\n",
    "\n",
    "    @staticmethod\n",
    "    def _caption(name: str) -> str:\n",
    "        head, _, tail = name.partition(\" (\")\n",
    "        if len(head) > 12 and \" \" in head:\n",
    "            first, _, last = head.rpartition(\" \")\n",
    "            head = f\"{first}\\n{last}\"\n",
    "        return f\"{head}\\n({tail}\" if tail else head\n",
    "\n",
    "    def draw(self) -> plt.Figure:\n",
    "        rows = [(\"Unlinked sum\", LINK_RESULT.period_effects.sum(axis=0) * 100)]\n",
    "        for linker in ARITHMETIC_LINKERS:\n",
    "            rows.append((linker.name, linker.link(LINK_RESULT) * 100))\n",
    "\n",
    "        values = np.array([total for _, total in rows])\n",
    "        totals = values.sum(axis=1)\n",
    "        target = LINK_PANEL.compounded_active_return * 100\n",
    "        gap = target - totals[0]\n",
    "\n",
    "        captions = [self._caption(name) for name, _ in rows]\n",
    "        x = np.arange(len(rows), dtype=float)\n",
    "\n",
    "        fig, ax = plt.subplots(figsize=(10.4, 6.1))\n",
    "        self.tidy(ax)\n",
    "\n",
    "        ax.add_patch(\n",
    "            Rectangle((-1.32, totals[0]), len(rows) + 1.0, gap, color=GREY,\n",
    "                      alpha=0.11, linewidth=0, zorder=0)\n",
    "        )\n",
    "\n",
    "        bottoms = np.zeros(len(rows))\n",
    "        for column, colour in enumerate(EFFECT_COLOURS):\n",
    "            heights = values[:, column]\n",
    "            for k in range(len(rows)):\n",
    "                ax.bar(x[k], heights[k], 0.58, bottom=bottoms[k], color=colour,\n",
    "                       alpha=0.36 if k == 0 else 1.0, linewidth=0, zorder=3)\n",
    "            bottoms = bottoms + heights\n",
    "\n",
    "        for k in range(len(rows)):\n",
    "            base = 0.0\n",
    "            for column in range(3):\n",
    "                value = values[k, column]\n",
    "                if value > 1.5:\n",
    "                    ax.text(x[k], base + value / 2, self.plain(value), ha=\"center\",\n",
    "                            va=\"center\", fontsize=11.5, color=\"white\", zorder=4)\n",
    "                base += value\n",
    "            if k:\n",
    "                ax.text(x[k], totals[k] + 0.16, self.plain(totals[k]), ha=\"center\",\n",
    "                        va=\"bottom\", fontsize=12.5, color=INK, zorder=4)\n",
    "\n",
    "        ax.axhline(target, color=INK, lw=1.1, zorder=5)\n",
    "        ax.axhline(totals[0], color=MUTED, lw=1.0, ls=(0, (2.6, 2.6)), zorder=5)\n",
    "        ax.annotate(\n",
    "            \"\", xy=(-1.06, target), xytext=(-1.06, totals[0]),\n",
    "            arrowprops=dict(arrowstyle=\"<->\", color=MUTED, lw=0.9, shrinkA=0, shrinkB=0),\n",
    "            zorder=6,\n",
    "        )\n",
    "        ax.text(-0.98, (target + totals[0]) / 2, self.plain(gap), ha=\"left\",\n",
    "                va=\"center\", fontsize=10.8, color=MUTED, zorder=6)\n",
    "\n",
    "        ax.set_xlim(-1.32, len(rows) - 0.32)\n",
    "        ax.set_ylim(0, 9.15)\n",
    "        ax.set_yticks([0, 2, 4, 6, totals[0], target])\n",
    "        ax.set_yticklabels([\"$0$\", \"$2$\", \"$4$\", \"$6$\", self.plain(totals[0]),\n",
    "                            self.plain(target)], fontsize=11.5)\n",
    "        for label, colour in zip(ax.get_yticklabels(), [MUTED] * 5 + [INK]):\n",
    "            label.set_color(colour)\n",
    "        ax.set_ylabel(\"Active return over the eight quarters,\\npercentage points\",\n",
    "                      fontsize=12, color=MUTED, labelpad=10, linespacing=1.4)\n",
    "        ax.set_xticks(x)\n",
    "        ax.set_xticklabels(captions, fontsize=11.8, color=INK, linespacing=1.35)\n",
    "        ax.tick_params(axis=\"x\", length=0, pad=8)\n",
    "\n",
    "        self.effect_legend(fig, y=0.955)\n",
    "        fig.subplots_adjust(left=0.105, right=0.99, bottom=0.115, top=0.925)\n",
    "        return fig"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class WeightsThroughTimeFigure(ArticleFigure):\n",
    "    \"\"\"Three weight histories that end at the same active weight.\"\"\"\n",
    "\n",
    "    name = \"weights_through_time\"\n",
    "\n",
    "    HISTORIES = (\n",
    "        (\"Held at $55\\\\%$ throughout\", (55.0, 55.0, 55.0, 55.0), BLUE),\n",
    "        (\"Bought up to $55\\\\%$ in quarter 3\", (50.0, 50.0, 55.0, 55.0), RUST),\n",
    "        (\"Drifted to $55\\\\%$ as prices moved\", (50.0, 51.5, 53.3, 55.0), MOSS),\n",
    "    )\n",
    "\n",
    "    def draw(self) -> plt.Figure:\n",
    "        fig, ax = plt.subplots(figsize=(8.2, 4.8))\n",
    "        self.tidy(ax)\n",
    "\n",
    "        quarters = np.arange(1, 5)\n",
    "        ax.plot(quarters, np.full(4, 50.0), color=MUTED, lw=1.2, ls=(0, (2, 2.4)),\n",
    "                label=\"Benchmark weight $50\\\\%$\", zorder=2)\n",
    "        for label, path, colour in self.HISTORIES:\n",
    "            ax.plot(quarters, np.array(path), color=colour, lw=1.9, marker=\"o\", ms=5,\n",
    "                    label=label, zorder=3)\n",
    "\n",
    "        ax.set_xticks(quarters)\n",
    "        ax.set_xticklabels([f\"Quarter {q}\" for q in quarters], fontsize=11.8, color=INK)\n",
    "        ax.set_ylabel(\"Weight in percent\", fontsize=11.8, color=MUTED)\n",
    "        ax.set_ylim(43.5, 59.5)\n",
    "        ax.set_yticks([45, 50, 55])\n",
    "        ax.grid(axis=\"y\", color=RULE, lw=0.7)\n",
    "        ax.set_axisbelow(True)\n",
    "        ax.legend(loc=\"lower left\", fontsize=11.2, frameon=False, handlelength=1.9,\n",
    "                  labelspacing=0.55, borderaxespad=0.4)\n",
    "        self.panel_title(ax, \"Three histories that end at the same weight\")\n",
    "\n",
    "        fig.subplots_adjust(left=0.115, right=0.985, bottom=0.115, top=0.9)\n",
    "        return fig"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class ExtensionsFigure(ArticleFigure):\n",
    "    \"\"\"Fixed income, multicurrency, and factor based attribution.\"\"\"\n",
    "\n",
    "    name = \"extensions\"\n",
    "\n",
    "    def draw(self) -> plt.Figure:\n",
    "        fig, axes = plt.subplots(\n",
    "            3, 1, figsize=(10.2, 8.8),\n",
    "            gridspec_kw=dict(hspace=0.66, height_ratios=[1.0, 1.25, 1.0]),\n",
    "        )\n",
    "\n",
    "        self.component_panel(\n",
    "            axes[0],\n",
    "            [(label, value * 1e4, colour)\n",
    "             for (label, value), colour in zip(FIXED_INCOME.as_dict().items(),\n",
    "                                               EFFECT_COLOURS)],\n",
    "            \"Active return\", BONDS.active_return * 1e4,\n",
    "        )\n",
    "        self.panel_title(\n",
    "            axes[0],\n",
    "            \"(a) Fixed income, Van Breukelen (2000): duration \"\n",
    "            f\"{BONDS.portfolio_duration:.2f} years against \"\n",
    "            f\"{BONDS.benchmark_duration:.2f}\",\n",
    "            pad=10,\n",
    "        )\n",
    "\n",
    "        palette = (TEAL, RUST, MOSS, OCHRE)\n",
    "        self.component_panel(\n",
    "            axes[1],\n",
    "            [(label, value * 1e4, colour)\n",
    "             for (label, value), colour in zip(CURRENCY.as_dict().items(), palette)],\n",
    "            \"Active return\", GLOBAL_PORTFOLIO.active_base_return * 1e4,\n",
    "        )\n",
    "        self.panel_title(\n",
    "            axes[1],\n",
    "            \"(b) Multicurrency, market and currency effects in additive form\",\n",
    "            pad=10,\n",
    "        )\n",
    "\n",
    "        factor_entries = [(\"Active intercept\", FACTORS.alpha * 1e4, TEAL)]\n",
    "        for name, beta, contribution in zip(FACTORS.factor_names, FACTORS.active_betas,\n",
    "                                            FACTORS.factor_contributions):\n",
    "            factor_entries.append(\n",
    "                (f\"{name} factor, $\\\\beta_P-\\\\beta_B={beta:+.2f}$\",\n",
    "                 contribution * 1e4, RUST if contribution >= 0 else OCHRE)\n",
    "            )\n",
    "        self.component_panel(\n",
    "            axes[2], factor_entries, \"Mean active return\",\n",
    "            FACTORS.mean_active_return * 1e4,\n",
    "        )\n",
    "        self.panel_title(\n",
    "            axes[2],\n",
    "            \"(c) Returns based factor attribution of active return, \"\n",
    "            \"72 monthly observations\",\n",
    "            pad=10,\n",
    "        )\n",
    "\n",
    "        fig.subplots_adjust(left=0.245, right=0.985, bottom=0.055, top=0.965)\n",
    "        return fig"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ARTICLE_FIGURES: tuple[ArticleFigure, ...] = (\n",
    "    ContributionFigure(),\n",
    "    ContributionMethodsFigure(),\n",
    "    AttributionFigure(),\n",
    "    LinkersFigure(),\n",
    "    WeightsThroughTimeFigure(),\n",
    "    ExtensionsFigure(),\n",
    ")\n",
    "\n",
    "for figure in ARTICLE_FIGURES:\n",
    "    figure.save()\n",
    "    print(f\"wrote {figure.output / figure.name}.pdf\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "name": "python",
   "version": "3.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}