algotrader/portfolio.py

"""Matrix portfolio engine.

The single-asset engine treats turnover as ``|target[t] - target[t-1]|``. That
is wrong the moment weights are fractional, because a position you did not
touch still *drifts*: hold 50% of your book in a name that doubles and you are
now at 67% without trading. Charging costs against the previous target instead
of the previous *actual* weight understates the cost of doing nothing and
overstates the cost of rebalancing.

This module models the drift explicitly and closes the gap:

    w_start[t] = target[t - lag]                     what we want to hold
    r_p[t]     = sum(w_start[t] * R[t])              portfolio return that bar
    w_end[t]   = w_start[t] * (1 + R[t]) / (1 + r_p[t])   drifted by the bar
    turnover[t] = sum |w_start[t] - w_end[t - 1]|    what we actually traded

Every step is a function of the current bar and the one before, so the whole
thing stays vectorised -- no Python loop over time.
"""

from __future__ import annotations

from typing import Optional

import numpy as np
import pandas as pd

from .metrics import compute_metrics, infer_periods_per_year
from .panel import Panel
from .types import BacktestResult, CostModel

__all__ = ["run_portfolio_backtest", "PortfolioResult", "normalise_weights"]


class PortfolioResult(BacktestResult):
    """A :class:`BacktestResult` that also keeps the per-asset weight history."""

    def __init__(self, *args, weights: pd.DataFrame, held: pd.DataFrame, panel: Panel, **kwargs):
        super().__init__(*args, **kwargs)
        self.weights = weights  # requested, post-constraint
        self.held = held  # actually held during each bar
        self.panel = panel

    def attribution(self) -> pd.Series:
        """Total return contribution per symbol, largest first."""
        contrib = (self.held * self.panel.returns().fillna(0.0)).sum(axis=0)
        return contrib.sort_values(ascending=False)


def normalise_weights(
    weights: pd.DataFrame,
    listed: pd.DataFrame,
    gross_leverage: float = 1.0,
    max_weight: Optional[float] = None,
    allow_short: bool = True,
) -> pd.DataFrame:
    """Apply the constraints a real book has, in the order a real book applies them.

    ``listed`` is "did this name have a price when the decision was made" --
    deliberately not the stricter "is a return defined over this bar" mask. A
    decision taken at Monday's close only needs Monday's price to exist; whether
    the position can actually be carried is enforced after the lag shift.
    """
    w = weights.reindex(index=listed.index, columns=listed.columns).astype(float).fillna(0.0)

    # You cannot ask for exposure to something that is not listed yet.
    w = w.where(listed, 0.0)

    if not allow_short:
        w = w.clip(lower=0.0)
    if max_weight is not None:
        w = w.clip(-abs(max_weight), abs(max_weight))

    # Scale down (never up) so gross exposure respects the leverage cap.
    gross = w.abs().sum(axis=1)
    scale = np.minimum(1.0, gross_leverage / gross.replace(0.0, np.nan))
    return w.mul(scale.fillna(1.0), axis=0)


def run_portfolio_backtest(
    panel: Panel,
    weights: pd.DataFrame,
    costs: Optional[CostModel] = None,
    lag: int = 1,
    gross_leverage: float = 1.0,
    max_weight: Optional[float] = None,
    allow_short: bool = True,
    initial_capital: float = 100_000.0,
    periods_per_year: Optional[int] = None,
    rf: float = 0.0,
    benchmark: Optional[pd.Series] = None,
    rebalance_on: Optional[pd.Series] = None,
    meta: Optional[dict] = None,
) -> PortfolioResult:
    """Backtest a ``T x N`` weight matrix against a panel.

    ``weights[t]`` is the exposure decided using information up to the close of
    bar ``t``; it is held from bar ``t + lag``. The default benchmark is the
    equal-weight universe, which is a far more honest comparison for a
    cross-sectional strategy than any single ticker.
    """
    if len(panel) < 2:
        raise ValueError("Cannot backtest a panel with fewer than two bars")
    if lag < 1:
        raise ValueError("lag must be >= 1; lag=0 would trade on unavailable information")

    costs = costs or CostModel()
    ppy = periods_per_year or infer_periods_per_year(panel.index)

    asset_returns = panel.returns().fillna(0.0)
    tradable = panel.tradable()
    listed = panel.close.notna()

    target = normalise_weights(weights, listed, gross_leverage, max_weight, allow_short)
    held = target.shift(lag).fillna(0.0)
    # Re-apply tradability after the shift: a name can delist between the
    # decision and the fill, and we must not be holding it when it does.
    held = held.where(tradable, 0.0)

    if rebalance_on is not None:
        held = _apply_rebalance_schedule(held, asset_returns, tradable, rebalance_on)

    gross_return = (held * asset_returns).sum(axis=1)

    # Weights after the bar's move, renormalised to the new portfolio value.
    growth = (1.0 + gross_return).replace(0.0, np.nan)
    drifted = (held * (1.0 + asset_returns)).div(growth, axis=0).fillna(0.0)
    previous = drifted.shift(1).fillna(0.0)

    traded = (held - previous).abs().sum(axis=1)
    trade_cost = traded * (costs.one_way_bps / 1e4)
    borrow_cost = held.clip(upper=0.0).abs().sum(axis=1) * (costs.short_borrow_bps / 1e4) / ppy
    total_cost = trade_cost + borrow_cost

    net = gross_return - total_cost
    equity = initial_capital * (1.0 + net).cumprod()

    if benchmark is None:
        # Equal weight across whatever was tradable on each bar.
        counts = tradable.sum(axis=1).replace(0, np.nan)
        equal = tradable.astype(float).div(counts, axis=0).fillna(0.0)
        benchmark = (equal.shift(lag).fillna(0.0) * asset_returns).sum(axis=1)
    benchmark = benchmark.reindex(panel.index).fillna(0.0)
    benchmark_equity = initial_capital * (1.0 + benchmark).cumprod()

    exposure = held.abs().sum(axis=1)
    metrics = compute_metrics(net, equity, exposure, ppy, rf)
    metrics.update(_portfolio_metrics(held, traded, metrics.get("years", 1.0)))

    survivorship = panel.survivorship()

    result = PortfolioResult(
        equity=equity,
        returns=net,
        gross_returns=gross_return,
        position=exposure,
        target=target.abs().sum(axis=1),
        costs=total_cost,
        benchmark_equity=benchmark_equity,
        metrics=metrics,
        benchmark_metrics=compute_metrics(benchmark, benchmark_equity, None, ppy, rf),
        meta={
            "lag": lag,
            "gross_leverage": gross_leverage,
            "max_weight": max_weight,
            "allow_short": allow_short,
            "initial_capital": initial_capital,
            "periods_per_year": ppy,
            "n_symbols": len(panel.symbols),
            "survivorship": survivorship,
            **(meta or {}),
        },
        weights=target,
        held=held,
        panel=panel,
    )
    return result


def _apply_rebalance_schedule(
    held: pd.DataFrame,
    asset_returns: pd.DataFrame,
    tradable: pd.DataFrame,
    rebalance_on: pd.Series,
) -> pd.DataFrame:
    """Trade only on rebalance bars; let the book drift in between.

    Without this, a monthly strategy whose target is constant between
    rebalances gets charged turnover every single bar for holding still --
    the model would be paying to *prevent* drift that a real book simply lets
    happen. This is the one genuinely recursive step in the engine: today's
    holding depends on yesterday's drifted holding.
    """
    schedule = rebalance_on.reindex(held.index).fillna(False).to_numpy(dtype=bool)
    target = held.to_numpy(dtype=float)
    returns = asset_returns.to_numpy(dtype=float)
    can_hold = tradable.to_numpy(dtype=bool)

    n_bars, n_assets = target.shape
    out = np.zeros((n_bars, n_assets), dtype=float)
    carried = np.zeros(n_assets, dtype=float)

    for t in range(n_bars):
        current = target[t] if schedule[t] else carried
        current = np.where(can_hold[t], current, 0.0)
        out[t] = current
        # Drift into the next bar, renormalised to the new portfolio value.
        portfolio_return = float(current @ returns[t])
        growth = 1.0 + portfolio_return
        carried = current * (1.0 + returns[t]) / growth if abs(growth) > 1e-12 else current

    return pd.DataFrame(out, index=held.index, columns=held.columns)


def rebalance_schedule(index: pd.DatetimeIndex, frequency: str = "M") -> pd.Series:
    """Boolean per-bar mask marking rebalance dates.

    ``frequency`` is ``D`` (every bar), ``W``, ``M``, ``Q``, or an integer
    number of bars as a string.
    """
    frequency = str(frequency).upper().strip()
    if frequency in ("D", "1", "B", ""):
        return pd.Series(True, index=index)
    if frequency.isdigit():
        step = max(1, int(frequency))
        mask = np.zeros(len(index), dtype=bool)
        mask[::step] = True
        return pd.Series(mask, index=index)

    periods = {"W": index.to_period("W"), "M": index.to_period("M"), "Q": index.to_period("Q")}
    if frequency not in periods:
        raise ValueError(f"Unknown rebalance frequency '{frequency}'")
    period = periods[frequency]
    # First bar of each period -- known at the time, unlike the last bar.
    return pd.Series(period != pd.Series(period, index=index).shift(1).to_numpy(), index=index)


def _portfolio_metrics(held: pd.DataFrame, traded: pd.Series, years: float) -> dict:
    """Book-level statistics a portfolio manager will look for first."""
    absolute = held.abs()
    gross = absolute.sum(axis=1)
    active = (absolute > 1e-9).sum(axis=1)

    # Herfindahl on the gross book: 1.0 is everything in one name, 1/n is even.
    shares = absolute.div(gross.replace(0.0, np.nan), axis=0)
    hhi = (shares**2).sum(axis=1)

    return {
        "gross_exposure": float(gross.mean()),
        "net_exposure": float(held.sum(axis=1).mean()),
        "max_gross_exposure": float(gross.max()),
        "avg_positions": float(active.mean()),
        "max_positions": float(active.max()),
        "concentration_hhi": float(hhi.mean(skipna=True)) if hhi.notna().any() else float("nan"),
        "turnover_ann": float(traded.sum() / years) if years > 0 else 0.0,
        "n_trades": float((traded > 1e-9).sum()),
    }