Portfolio

Portfolio is Fundcloud's shared post-simulation object. Every entry path — Simulator.run_strategy, run_weights, run_signals, run_orders, and the skfolio round-trip via from_skfolio / to_skfolio — produces one. Metrics (sharpe, max_drawdown, turnover, attribution), the full summary() bundle, and the tear-sheet renderers all read from the same object, which keeps notebook exploration and production reporting numerically identical. Population holds a set of portfolios for cross-strategy comparison; Position is the per-asset record.

fundcloud.portfolio

Unified position + analytics container.

One :class:Portfolio class handles both live simulation state (via :meth:Portfolio.apply / :meth:Portfolio.mark_to_market) and post-run analytics (Sharpe, drawdowns, CVaR, attribution, …). :class:Population compares several portfolios side-by-side.

Portfolio

Portfolio(
    *,
    returns: DataFrame | Series | None = None,
    weights: DataFrame | Series | None = None,
    benchmark: Series | None = None,
    cash: float = 0.0,
    positions: dict[str, float] | None = None,
    name: str | None = None,
)

Unified position + analytics container.

Source code in python/fundcloud/portfolio/portfolio.py

def __init__(
    self,
    *,
    returns: pd.DataFrame | pd.Series | None = None,
    weights: pd.DataFrame | pd.Series | None = None,
    benchmark: pd.Series | None = None,
    cash: float = 0.0,
    positions: dict[str, float] | None = None,
    name: str | None = None,
) -> None:
    self._name = name
    self._benchmark = benchmark
    self._returns: pd.Series | None = None
    self._weights_frame: pd.DataFrame | None = None

    if returns is not None:
        self._returns = _coerce_returns(returns)
    if weights is not None:
        self._weights_frame = _coerce_weights(weights)

    # Live state — only populated for live-mode portfolios.
    self._live = _LiveState(cash=float(cash))
    if positions:
        for asset, qty in positions.items():
            self._live.positions[asset] = Position(qty=float(qty))

equity_curve property

equity_curve: Series

Running equity. For analytics-mode portfolios, cumulates returns.

positions property

positions: Series

Current open positions as a Series keyed by asset.

apply

apply(trade: Any) -> None

Apply a Trade (from :mod:fundcloud.sim) to live state.

The method only depends on the trade's duck-typed attributes (asset, qty, price, fee) so this module doesn't have to import the simulator package.

Source code in python/fundcloud/portfolio/portfolio.py

def apply(self, trade: Any) -> None:
    """Apply a ``Trade`` (from :mod:`fundcloud.sim`) to live state.

    The method only depends on the trade's duck-typed attributes
    (``asset``, ``qty``, ``price``, ``fee``) so this module doesn't have to
    import the simulator package.
    """
    asset = str(trade.asset)
    qty = float(trade.qty)
    price = float(trade.price)
    fee = float(getattr(trade, "fee", 0.0))
    pos = self.position(asset)
    notional = qty * price
    self._live.cash -= notional + fee
    # Weighted-average cost update for adds; leave avg_cost alone on closes.
    if pos.qty == 0 or (pos.qty > 0) == (qty > 0):
        total = pos.qty + qty
        if total != 0:
            pos.avg_cost = (pos.qty * pos.avg_cost + qty * price) / total
    pos.qty += qty
    self._live.trade_log.append(trade)

attribution

attribution() -> pd.DataFrame

Asset-level return contribution = weights × returns (shifted).

Requires a weights frame. Uses the current-bar weight × current-bar asset return, which is the standard backward-looking decomposition.

Source code in python/fundcloud/portfolio/portfolio.py

def attribution(self) -> pd.DataFrame:
    """Asset-level return contribution = weights × returns (shifted).

    Requires a weights frame. Uses the current-bar weight × current-bar
    asset return, which is the standard backward-looking decomposition.
    """
    w = self._weights_frame
    if w is None:
        return pd.DataFrame()
    if self._returns is None or self._returns.empty:
        return pd.DataFrame(columns=w.columns)
    # If returns is a total-portfolio series (no per-asset info), attribution
    # is undefined beyond `weights * total_return`.
    contrib = w.reindex(self._returns.index).mul(self._returns, axis=0)
    return contrib

contribution

contribution() -> pd.Series

Average per-asset contribution to total return.

Source code in python/fundcloud/portfolio/portfolio.py

def contribution(self) -> pd.Series:
    """Average per-asset contribution to total return."""
    attr = self.attribution()
    if attr.empty:
        return pd.Series(dtype=float)
    return attr.mean()

drawdown_details

drawdown_details() -> pd.DataFrame

One row per drawdown episode: start / valley / recovery + durations.

See :func:fundcloud.metrics.drawdown_details for the column definitions.

Source code in python/fundcloud/portfolio/portfolio.py

def drawdown_details(self) -> pd.DataFrame:
    """One row per drawdown episode: start / valley / recovery + durations.

    See :func:`fundcloud.metrics.drawdown_details` for the column
    definitions.
    """
    return _metrics.drawdown_details(self.returns)

from_skfolio classmethod

from_skfolio(
    portfolio: Any, *, benchmark: Series | None = None
) -> Portfolio

Lift a skfolio Portfolio into a Fundcloud Portfolio.

Copies the returns series and the (per-period) weight vector if one is exposed. Compatible with skfolio >= 0.6.

Source code in python/fundcloud/portfolio/portfolio.py

@classmethod
def from_skfolio(cls, portfolio: Any, *, benchmark: pd.Series | None = None) -> Portfolio:
    """Lift a skfolio ``Portfolio`` into a Fundcloud ``Portfolio``.

    Copies the returns series and the (per-period) weight vector if one is
    exposed. Compatible with skfolio >= 0.6.
    """
    returns = _safe_skfolio_returns(portfolio)
    weights = _safe_skfolio_weights(portfolio)
    name = getattr(portfolio, "name", None) or type(portfolio).__name__
    return cls(
        returns=returns,
        weights=weights,
        benchmark=benchmark,
        name=name,
    )

mark_to_market

mark_to_market(prices: Series, ts: Timestamp) -> float

Compute and record equity at timestamp ts.

prices is a Series of asset → price at that timestamp. When a price is missing (NaN) — typical for a mixed-frequency panel on a weekend / market holiday — the portfolio falls back to the last known price for that asset so a held position keeps its value across the closed bar. Cash-only positions (qty == 0) are skipped.

Source code in python/fundcloud/portfolio/portfolio.py

def mark_to_market(
    self,
    prices: pd.Series,
    ts: pd.Timestamp,
) -> float:
    """Compute and record equity at timestamp ``ts``.

    ``prices`` is a Series of asset → price at that timestamp. When a
    price is missing (``NaN``) — typical for a mixed-frequency panel
    on a weekend / market holiday — the portfolio falls back to the
    last known price for that asset so a held position keeps its
    value across the closed bar. Cash-only positions (``qty == 0``)
    are skipped.
    """
    # Refresh the last-known price cache from this bar's quotes.
    for asset, raw in prices.items():
        px = float(raw)
        if np.isfinite(px) and px > 0:
            self._live.last_prices[str(asset)] = px

    equity = self._live.cash
    per_asset_value: dict[str, float] = {}
    for asset, pos in self._live.positions.items():
        if pos.qty == 0:
            continue
        # Prefer the current bar's price; fall back to the last known
        # price for that asset; final fallback is the position's average
        # cost (covers the unusual case where we buy and immediately
        # need to mark-to-market on a NaN bar).
        raw_px = prices.get(asset, np.nan)
        px = float(raw_px) if raw_px is not None else float("nan")
        if not np.isfinite(px) or px <= 0:
            px = self._live.last_prices.get(
                asset, pos.avg_cost if pos.avg_cost > 0 else float("nan")
            )
        if not np.isfinite(px) or px <= 0:
            continue
        value = pos.qty * px
        equity += value
        per_asset_value[asset] = value
    self._live.equity_history.append((ts, equity))
    if equity != 0:
        weights = {a: v / equity for a, v in per_asset_value.items()}
    else:
        weights = {a: 0.0 for a in per_asset_value}
    self._live.weights_history.append((ts, weights))
    return equity

metrics

metrics(
    *,
    benchmark: Series | None = None,
    risk_free: float | None = None,
    periods_per_year: int | None = None,
    cvar_alpha: float = 0.95,
) -> pd.Series

Full portfolio-metrics bundle.

Delegates to :func:fundcloud.metrics.metrics. When this Portfolio was constructed with benchmark=, that benchmark is used by default; pass an explicit benchmark= to override.

Source code in python/fundcloud/portfolio/portfolio.py

def metrics(
    self,
    *,
    benchmark: pd.Series | None = None,
    risk_free: float | None = None,
    periods_per_year: int | None = None,
    cvar_alpha: float = 0.95,
) -> pd.Series:
    """Full portfolio-metrics bundle.

    Delegates to :func:`fundcloud.metrics.metrics`. When this Portfolio
    was constructed with ``benchmark=``, that benchmark is used by
    default; pass an explicit ``benchmark=`` to override.
    """
    r = self.returns
    bench = benchmark if benchmark is not None else self.benchmark
    return _metrics.metrics(
        r,
        benchmark=bench,
        risk_free=risk_free,
        periods_per_year=periods_per_year,
        cvar_alpha=cvar_alpha,
    ).rename(self.name)

period_returns

period_returns(
    *,
    benchmark: Series | None = None,
    periods_per_year: int | None = None,
) -> pd.Series | pd.DataFrame

MTD / 3M / 6M / YTD / 1Y / 3Y / 5Y / 10Y / All-time bundle.

When a benchmark is not passed and :attr:benchmark was set on construction, it's used as the default. See :func:fundcloud.metrics.period_returns.

Source code in python/fundcloud/portfolio/portfolio.py

def period_returns(
    self,
    *,
    benchmark: pd.Series | None = None,
    periods_per_year: int | None = None,
) -> pd.Series | pd.DataFrame:
    """MTD / 3M / 6M / YTD / 1Y / 3Y / 5Y / 10Y / All-time bundle.

    When a benchmark is not passed and :attr:`benchmark` was set on
    construction, it's used as the default. See
    :func:`fundcloud.metrics.period_returns`.
    """
    bench = benchmark if benchmark is not None else self.benchmark
    return _metrics.period_returns(
        self.returns,
        benchmark=bench,
        periods_per_year=periods_per_year,
    )

runup_details

runup_details() -> pd.DataFrame

One row per runup (rally) episode between drawdowns.

See :func:fundcloud.metrics.runup_details for the column definitions.

Source code in python/fundcloud/portfolio/portfolio.py

def runup_details(self) -> pd.DataFrame:
    """One row per runup (rally) episode between drawdowns.

    See :func:`fundcloud.metrics.runup_details` for the column
    definitions.
    """
    return _metrics.runup_details(self.returns)

snapshot

snapshot() -> Portfolio

Freeze live state into an analytics-mode copy.

Builds returns from the equity curve (if there is one) and weights from the recorded weights history, then detaches live state so the returned instance behaves immutably for analytics.

Source code in python/fundcloud/portfolio/portfolio.py

def snapshot(self) -> Portfolio:
    """Freeze live state into an analytics-mode copy.

    Builds ``returns`` from the equity curve (if there is one) and
    ``weights`` from the recorded weights history, then detaches live
    state so the returned instance behaves immutably for analytics.
    """
    equity = pd.Series(
        {ts: val for ts, val in self._live.equity_history},
        dtype=float,
    ).sort_index()
    returns = equity.pct_change().dropna() if len(equity) > 1 else pd.Series([], dtype=float)

    weights_frame: pd.DataFrame | None
    if self._live.weights_history:
        raw = pd.DataFrame.from_dict(
            {ts: w for ts, w in self._live.weights_history}, orient="index"
        ).sort_index()
        weights_frame = raw.fillna(0.0)
    else:
        weights_frame = None

    snap = Portfolio(
        returns=returns,
        weights=weights_frame,
        benchmark=self._benchmark,
        name=self._name,
    )
    return snap

summary

summary(
    *,
    risk_free: float | None = None,
    periods_per_year: int | None = None,
    cvar_alpha: float = 0.95,
) -> pd.Series

Single-column summary of key metrics (rows = metric names).

Compact 11-metric view. For the full ~55-metric bundle use :meth:metrics.

Source code in python/fundcloud/portfolio/portfolio.py

def summary(
    self,
    *,
    risk_free: float | None = None,
    periods_per_year: int | None = None,
    cvar_alpha: float = 0.95,
) -> pd.Series:
    """Single-column summary of key metrics (rows = metric names).

    Compact 11-metric view. For the full ~55-metric bundle use
    :meth:`metrics`.
    """
    r = self.returns
    stats = _metrics.returns_stats(
        r,
        risk_free=risk_free,
        periods_per_year=periods_per_year,
        cvar_alpha=cvar_alpha,
    )
    return stats.iloc[:, 0].rename(self.name)

to_skfolio

to_skfolio() -> Any

Build a skfolio Portfolio mirror of this object.

Requires the [pf] extra. The resulting object is an instance of :class:skfolio.Portfolio.

Source code in python/fundcloud/portfolio/portfolio.py

def to_skfolio(self) -> Any:
    """Build a skfolio ``Portfolio`` mirror of this object.

    Requires the ``[pf]`` extra. The resulting object is an instance of
    :class:`skfolio.Portfolio`.
    """
    try:
        from skfolio import Portfolio as SkPortfolio  # type: ignore[import-not-found]
    except ImportError as e:
        msg = "to_skfolio() requires skfolio; install with: uv add 'fundcloud[pf]'"
        raise ImportError(msg) from e
    # skfolio's Portfolio constructor expects an X/returns and a weights
    # vector. We pass a per-period weights frame when available.
    return SkPortfolio(
        X=self.returns.to_frame() if isinstance(self.returns, pd.Series) else self.returns,
        weights=self._weights_frame.iloc[-1].to_numpy()
        if self._weights_frame is not None and len(self._weights_frame) > 0
        else None,
        name=self.name,
    )

turnover

turnover() -> float

Average one-way turnover across rebalance boundaries.

Returns 0.0 when weights are constant or unknown.

Source code in python/fundcloud/portfolio/portfolio.py

def turnover(self) -> float:
    """Average one-way turnover across rebalance boundaries.

    Returns ``0.0`` when weights are constant or unknown.
    """
    w = self._weights_frame
    if w is None or len(w) < 2:
        return 0.0
    return float(w.diff().abs().sum(axis=1).iloc[1:].mean() / 2.0)

worst_drawdowns

worst_drawdowns(top: int = 10) -> pd.DataFrame

Top-top drawdown episodes, display-formatted.

Columns: Started / Recovered / Drawdown / Days. Episodes are sorted by depth (worst first).

Source code in python/fundcloud/portfolio/portfolio.py

def worst_drawdowns(self, top: int = 10) -> pd.DataFrame:
    """Top-``top`` drawdown episodes, display-formatted.

    Columns: ``Started`` / ``Recovered`` / ``Drawdown`` / ``Days``.
    Episodes are sorted by depth (worst first).
    """
    dd = _metrics.drawdown_details(self.returns)
    if dd.empty:
        return pd.DataFrame(columns=["Started", "Recovered", "Drawdown", "Days"])
    view = (
        dd
        .head(top)[["start", "recovery", "max_drawdown", "duration_days"]]
        .rename(
            columns={
                "start": "Started",
                "recovery": "Recovered",
                "max_drawdown": "Drawdown",
                "duration_days": "Days",
            }
        )
        .reset_index(drop=True)
    )
    return view

worst_runups

worst_runups(top: int = 10) -> pd.DataFrame

Top-top runup episodes, display-formatted.

Columns: Started / Peaked / Runup / Days. Episodes are sorted by magnitude (biggest first).

Source code in python/fundcloud/portfolio/portfolio.py

def worst_runups(self, top: int = 10) -> pd.DataFrame:
    """Top-``top`` runup episodes, display-formatted.

    Columns: ``Started`` / ``Peaked`` / ``Runup`` / ``Days``.
    Episodes are sorted by magnitude (biggest first).
    """
    ru = _metrics.runup_details(self.returns)
    if ru.empty:
        return pd.DataFrame(columns=["Started", "Peaked", "Runup", "Days"])
    view = (
        ru
        .head(top)[["start", "peak", "max_runup", "duration_days"]]
        .rename(
            columns={
                "start": "Started",
                "peak": "Peaked",
                "max_runup": "Runup",
                "duration_days": "Days",
            }
        )
        .reset_index(drop=True)
    )
    return view

yearly_returns

yearly_returns(
    *, benchmark: Series | None = None
) -> pd.Series | pd.DataFrame

End-of-year returns.

Returns a :class:pd.Series when no benchmark is available, or a two-column :class:pd.DataFrame (benchmark, strategy) when one is supplied (or set on construction).

Source code in python/fundcloud/portfolio/portfolio.py

def yearly_returns(self, *, benchmark: pd.Series | None = None) -> pd.Series | pd.DataFrame:
    """End-of-year returns.

    Returns a :class:`pd.Series` when no benchmark is available, or a
    two-column :class:`pd.DataFrame` (``benchmark``, ``strategy``)
    when one is supplied (or set on construction).
    """
    bench = benchmark if benchmark is not None else self.benchmark
    strategy = _metrics.yearly_returns(self.returns).rename(self.name)
    if bench is None:
        return strategy
    bench_yearly = _metrics.yearly_returns(bench).rename(
        str(bench.name) if bench.name is not None else "benchmark"
    )
    return pd.concat([bench_yearly, strategy], axis=1)

Population

Population(portfolios: Sequence[Portfolio])

A named bag of :class:Portfolio objects.

Source code in python/fundcloud/portfolio/population.py

def __init__(self, portfolios: Sequence[Portfolio]) -> None:
    self._portfolios = list(portfolios)
    # Disambiguate name collisions so `summary` produces unique columns.
    counts: dict[str, int] = {}
    for p in self._portfolios:
        counts[p.name] = counts.get(p.name, 0) + 1
    seen: dict[str, int] = {}
    for p in self._portfolios:
        base = p.name
        if counts[base] > 1:
            seen[base] = seen.get(base, 0) + 1
            p.rename(f"{base}_{seen[base]}")

composition

composition() -> pd.DataFrame

Latest weights per portfolio, as rows-per-portfolio × asset columns.

Source code in python/fundcloud/portfolio/population.py

def composition(self) -> pd.DataFrame:
    """Latest weights per portfolio, as rows-per-portfolio × asset columns."""
    rows: dict[str, pd.Series] = {}
    for p in self._portfolios:
        w = p.weights
        if w is None or len(w) == 0:
            continue
        rows[p.name] = w.iloc[-1]
    if not rows:
        return pd.DataFrame()
    return pd.DataFrame(rows).T.fillna(0.0)

cumulative_returns

cumulative_returns() -> pd.DataFrame

Wide frame of cumulative (compounded) returns per portfolio.

Source code in python/fundcloud/portfolio/population.py

def cumulative_returns(self) -> pd.DataFrame:
    """Wide frame of cumulative (compounded) returns per portfolio."""
    if not self._portfolios:
        return pd.DataFrame()
    series = {}
    for p in self._portfolios:
        try:
            r = p.returns
        except ValueError:
            continue
        series[p.name] = (1.0 + r).cumprod()
    return pd.DataFrame(series)

summary

summary(
    *,
    risk_free: float | None = None,
    periods_per_year: int | None = None,
    cvar_alpha: float = 0.95,
) -> pd.DataFrame

Metric-by-portfolio comparison table (rows = metrics, cols = portfolios).

Source code in python/fundcloud/portfolio/population.py

def summary(
    self,
    *,
    risk_free: float | None = None,
    periods_per_year: int | None = None,
    cvar_alpha: float = 0.95,
) -> pd.DataFrame:
    """Metric-by-portfolio comparison table (rows = metrics, cols = portfolios)."""
    cols = [
        p.summary(
            risk_free=risk_free,
            periods_per_year=periods_per_year,
            cvar_alpha=cvar_alpha,
        )
        for p in self._portfolios
    ]
    if not cols:
        return pd.DataFrame()
    return pd.concat(cols, axis=1)

Position dataclass

Position(qty: float = 0.0, avg_cost: float = 0.0)

Live position for a single asset.