Portfolio API

asrquant.optimization

Portfolio construction and risk decomposition without mandatory solvers.

random_frontier

random_frontier(expected_returns: Series | ndarray, covariance: DataFrame | ndarray, n_portfolios: int = 5000, risk_free_rate: float = 0.0, random_state: int | None = 0) -> pd.DataFrame

Monte Carlo long-only portfolio cloud.

Source code in src/asrquant/optimization.py

def random_frontier(
    expected_returns: pd.Series | np.ndarray,
    covariance: pd.DataFrame | np.ndarray,
    n_portfolios: int = 5_000,
    risk_free_rate: float = 0.0,
    random_state: int | None = 0,
) -> pd.DataFrame:
    """Monte Carlo long-only portfolio cloud."""
    mu = np.asarray(expected_returns, dtype=float)
    cov = np.asarray(covariance, dtype=float)
    rng = np.random.default_rng(random_state)
    weights = rng.dirichlet(np.ones(len(mu)), size=n_portfolios)
    rets = weights @ mu
    vols = np.sqrt(np.einsum("ij,jk,ik->i", weights, cov, weights))
    sharpes = np.divide(rets - risk_free_rate, vols, out=np.full_like(rets, np.nan), where=vols > 0)
    frame = pd.DataFrame({"return": rets, "volatility": vols, "sharpe": sharpes})
    for i in range(weights.shape[1]):
        frame[f"w{i}"] = weights[:, i]
    return frame

estimate_covariance

estimate_covariance(returns: DataFrame, method: str = 'sample', *, annualization: int = 252, span: int = 60) -> pd.DataFrame

Estimate annualized covariance using sample, EWMA, Ledoit-Wolf, or OAS.

Source code in src/asrquant/optimization.py

def estimate_covariance(
    returns: pd.DataFrame,
    method: str = "sample",
    *,
    annualization: int = 252,
    span: int = 60,
) -> pd.DataFrame:
    """Estimate annualized covariance using sample, EWMA, Ledoit-Wolf, or OAS."""
    frame = pd.DataFrame(returns, dtype=float).dropna()
    key = method.lower().replace("-", "_")
    if key == "sample":
        cov = frame.cov().to_numpy() * annualization
    elif key == "ewma":
        cov = frame.ewm(span=span, adjust=False).cov().groupby(level=1).tail(1).droplevel(0).reindex(index=frame.columns, columns=frame.columns).to_numpy() * annualization
    elif key in {"ledoit_wolf", "ledoitwolf"}:
        from sklearn.covariance import LedoitWolf
        cov = LedoitWolf().fit(frame).covariance_ * annualization
    elif key == "oas":
        from sklearn.covariance import OAS
        cov = OAS().fit(frame).covariance_ * annualization
    else:
        raise ValueError("method must be sample, ewma, ledoit_wolf, or oas")
    return pd.DataFrame(cov, index=frame.columns, columns=frame.columns)

maximum_diversification

maximum_diversification(covariance: DataFrame | ndarray, long_only: bool = True) -> np.ndarray

Maximize the diversification ratio w' sigma / sqrt(w' Sigma w).

Source code in src/asrquant/optimization.py

def maximum_diversification(covariance: pd.DataFrame | np.ndarray, long_only: bool = True) -> np.ndarray:
    """Maximize the diversification ratio w' sigma / sqrt(w' Sigma w)."""
    cov = np.asarray(covariance, dtype=float)
    asset_vol = np.sqrt(np.diag(cov)); n = len(asset_vol)
    bounds = [(0.0, 1.0)] * n if long_only else [(-1.0, 1.0)] * n
    def objective(w):
        vol = portfolio_volatility(w, cov)
        return -(w @ asset_vol) / vol if vol > 0 else 1e6
    result = minimize(objective, np.repeat(1/n, n), method="SLSQP", bounds=bounds, constraints={"type": "eq", "fun": lambda w: w.sum()-1})
    if not result.success:
        raise RuntimeError(f"optimization failed: {result.message}")
    return result.x

efficient_frontier

efficient_frontier(expected_returns: Series | ndarray, covariance: DataFrame | ndarray, points: int = 50, long_only: bool = True) -> pd.DataFrame

Compute minimum-volatility portfolios across a return grid.

Source code in src/asrquant/optimization.py

def efficient_frontier(
    expected_returns: pd.Series | np.ndarray,
    covariance: pd.DataFrame | np.ndarray,
    points: int = 50,
    long_only: bool = True,
) -> pd.DataFrame:
    """Compute minimum-volatility portfolios across a return grid."""
    mu = np.asarray(expected_returns, dtype=float); cov = np.asarray(covariance, dtype=float); n = len(mu)
    bounds = [(0.0, 1.0)] * n if long_only else [(-1.0, 1.0)] * n
    targets = np.linspace(mu.min(), mu.max(), points)
    rows = []
    for target in targets:
        result = minimize(lambda w: w@cov@w, np.repeat(1/n,n), method="SLSQP", bounds=bounds, constraints=[{"type":"eq","fun":lambda w:w.sum()-1},{"type":"eq","fun":lambda w,t=target:w@mu-t}])
        if result.success:
            row={"return": result.x@mu, "volatility": np.sqrt(result.x@cov@result.x)}
            row.update({f"w{i}": value for i,value in enumerate(result.x)}); rows.append(row)
    return pd.DataFrame(rows)

black_litterman

black_litterman(covariance: DataFrame | ndarray, market_weights: ndarray, risk_aversion: float, views: ndarray | None = None, pick_matrix: ndarray | None = None, view_covariance: ndarray | None = None, tau: float = 0.05) -> tuple[np.ndarray, np.ndarray]

Return Black-Litterman posterior mean and covariance.

Source code in src/asrquant/optimization.py

def black_litterman(
    covariance: pd.DataFrame | np.ndarray,
    market_weights: np.ndarray,
    risk_aversion: float,
    views: np.ndarray | None = None,
    pick_matrix: np.ndarray | None = None,
    view_covariance: np.ndarray | None = None,
    tau: float = 0.05,
) -> tuple[np.ndarray, np.ndarray]:
    """Return Black-Litterman posterior mean and covariance."""
    cov = np.asarray(covariance, dtype=float); w = np.asarray(market_weights, dtype=float)
    pi = risk_aversion * cov @ w
    if views is None or pick_matrix is None:
        return pi, cov + tau*cov
    q = np.asarray(views, dtype=float); p = np.asarray(pick_matrix, dtype=float)
    omega = np.asarray(view_covariance, dtype=float) if view_covariance is not None else np.diag(np.diag(p @ (tau*cov) @ p.T))
    tau_inv = np.linalg.inv(tau*cov); omega_inv = np.linalg.inv(omega)
    posterior_cov = np.linalg.inv(tau_inv + p.T@omega_inv@p)
    posterior_mean = posterior_cov @ (tau_inv@pi + p.T@omega_inv@q)
    return posterior_mean, cov + posterior_cov

hierarchical_risk_parity

hierarchical_risk_parity(returns: DataFrame) -> pd.Series

Hierarchical risk parity using correlation clustering and recursive bisection.

Source code in src/asrquant/optimization.py

def hierarchical_risk_parity(returns: pd.DataFrame) -> pd.Series:
    """Hierarchical risk parity using correlation clustering and recursive bisection."""
    from scipy.cluster.hierarchy import leaves_list, linkage
    from scipy.spatial.distance import squareform
    frame = pd.DataFrame(returns, dtype=float).dropna()
    cov = frame.cov(); corr = frame.corr().clip(-1,1)
    distance = np.sqrt((1-corr)/2)
    order = leaves_list(linkage(squareform(distance.to_numpy(), checks=False), method="single"))
    ordered = list(frame.columns[order])
    weights = pd.Series(1.0, index=ordered)
    clusters = [ordered]
    def cluster_var(items):
        sub = cov.loc[items, items]
        ivp = 1/np.diag(sub); ivp = ivp/ivp.sum()
        return float(ivp @ sub.to_numpy() @ ivp)
    while clusters:
        next_clusters=[]
        for cluster in clusters:
            if len(cluster)<=1: continue
            split=len(cluster)//2; left=cluster[:split]; right=cluster[split:]
            v_left=cluster_var(left); v_right=cluster_var(right); alpha=1-v_left/(v_left+v_right)
            weights[left]*=alpha; weights[right]*=1-alpha
            next_clusters.extend([left,right])
        clusters=next_clusters
    return weights.reindex(frame.columns)