diff --git a/backend_api_python/app/services/backtest/metrics.py b/backend_api_python/app/services/backtest/metrics.py index 60be10d32..235e14839 100644 --- a/backend_api_python/app/services/backtest/metrics.py +++ b/backend_api_python/app/services/backtest/metrics.py @@ -13,10 +13,13 @@ __all__ = [ "DEFAULT_INFORMATION_RATIO_BANDS", "benchmark_level_curve", + "calculate_batting_average", "calculate_benchmark_relative_metrics", "calculate_beta_and_alpha", "calculate_capture_ratios", + "calculate_dual_beta", "calculate_information_ratio", + "calculate_m_squared", ] @@ -303,16 +306,28 @@ def calculate_benchmark_relative_metrics( "correlationStatus": "insufficient_history", "treynorStatus": "insufficient_history", "captureStatus": "insufficient_history", + "mSquaredStatus": "insufficient_history", + "dualBetaStatus": "insufficient_history", + "battingStatus": "insufficient_history", "beta": None, "alphaAnnualized": None, "treynorRatio": None, "correlation": None, "rSquared": None, + "mSquared": None, + "mSquaredAlpha": None, "upCapture": None, "downCapture": None, "captureRatio": None, "upMarketPeriods": 0, "downMarketPeriods": 0, + "upBeta": None, + "downBeta": None, + "betaSpread": None, + "battingAverage": None, + "outperformancePeriods": 0, + "underperformancePeriods": 0, + "outperformancePayoffRatio": None, "benchmark": benchmark, "observations": observations, "frequency": normalized_frequency, @@ -332,6 +347,8 @@ def calculate_benchmark_relative_metrics( }) result.update(_calculate_capture_metrics(returns)) + result.update(_calculate_dual_beta_metrics(returns)) + result.update(_calculate_batting_metrics(returns)) if observations < 2: return result @@ -346,6 +363,11 @@ def calculate_benchmark_relative_metrics( annualization_factor=factor, risk_free_rate=rf, )) + result.update(_calculate_m_squared_metrics( + returns, + annualization_factor=factor, + risk_free_rate=rf, + )) if math.isclose(periodic_tracking_error, 0.0, rel_tol=0.0, abs_tol=1e-15): result["status"] = "zero_tracking_error" @@ -602,3 +624,276 @@ def _classify_information_ratio( if value < upper_bound: return label return bands[-1][1] + + + +def calculate_m_squared( + portfolio_curve: Iterable[dict[str, Any]], + benchmark_curve: Iterable[dict[str, Any]], + *, + benchmark: str | None = None, + frequency: str = "1d", + annualization_factor: float = 252.0, + risk_free_rate: float = 0.0, + market: str = "", +) -> dict[str, Any]: + """Calculate the Modigliani-Modigliani (M^2) measure and M^2 Alpha. + + M^2 leverages or deleverages the portfolio to match benchmark total volatility, + converting the Sharpe ratio into an annualized percentage return directly + comparable to the benchmark: + M^2 = R_f + Sharpe_p * Vol_b_ann + M^2_alpha = M^2 - R_b_ann + """ + factor = float(annualization_factor) + if not math.isfinite(factor) or factor <= 0: + raise ValueError("annualization_factor must be a positive finite number") + + rf = float(risk_free_rate) + if not math.isfinite(rf): + raise ValueError("risk_free_rate must be a finite number") + + returns, normalized_frequency = _align_curves_and_returns( + portfolio_curve, benchmark_curve, frequency, market=market + ) + observations = int(len(returns.index)) + result: dict[str, Any] = { + "status": "insufficient_history", + "portfolioReturnAnnualized": None, + "benchmarkReturnAnnualized": None, + "riskFreeRate": rf, + "mSquared": None, + "mSquaredAlpha": None, + "benchmark": benchmark, + "observations": observations, + "frequency": normalized_frequency, + "annualizationFactor": factor, + } + if observations < 2: + return result + + result["portfolioReturnAnnualized"] = float(returns["portfolio"].mean()) * factor + result["benchmarkReturnAnnualized"] = float(returns["benchmark"].mean()) * factor + m2 = _calculate_m_squared_metrics( + returns, + annualization_factor=factor, + risk_free_rate=rf, + ) + result.update({ + "status": m2["mSquaredStatus"], + "mSquared": m2["mSquared"], + "mSquaredAlpha": m2["mSquaredAlpha"], + }) + return result + + +def calculate_dual_beta( + portfolio_curve: Iterable[dict[str, Any]], + benchmark_curve: Iterable[dict[str, Any]], + *, + benchmark: str | None = None, + frequency: str = "1d", + market: str = "", +) -> dict[str, Any]: + """Calculate Up-Market Beta (beta+), Down-Market Beta (beta-), and Beta Spread. + + Up-market beta measures sensitivity during benchmark positive periods (R_b > 0). + Down-market beta measures sensitivity during benchmark negative periods (R_b < 0). + Beta spread = beta+ - beta-. A positive spread indicates convex upside participation. + """ + returns, normalized_frequency = _align_curves_and_returns( + portfolio_curve, benchmark_curve, frequency, market=market + ) + observations = int(len(returns.index)) + result: dict[str, Any] = { + "status": "insufficient_history", + "upBeta": None, + "downBeta": None, + "betaSpread": None, + "upMarketPeriods": 0, + "downMarketPeriods": 0, + "benchmark": benchmark, + "observations": observations, + "frequency": normalized_frequency, + } + if observations < 2: + return result + + dual = _calculate_dual_beta_metrics(returns) + result.update({ + "status": dual["dualBetaStatus"], + "upBeta": dual["upBeta"], + "downBeta": dual["downBeta"], + "betaSpread": dual["betaSpread"], + "upMarketPeriods": int(len(returns.loc[returns["benchmark"] > 0].index)), + "downMarketPeriods": int(len(returns.loc[returns["benchmark"] < 0].index)), + }) + return result + + +def calculate_batting_average( + portfolio_curve: Iterable[dict[str, Any]], + benchmark_curve: Iterable[dict[str, Any]], + *, + benchmark: str | None = None, + frequency: str = "1d", + market: str = "", +) -> dict[str, Any]: + """Calculate Batting Average and Outperformance Payoff Ratio relative to benchmark. + + Batting average is the percentage of observation periods where portfolio return + exceeds benchmark return: Sum(I(R_p > R_b)) / T. + Outperformance payoff ratio is average excess return in outperforming periods divided + by average underperformance in underperforming periods. + """ + returns, normalized_frequency = _align_curves_and_returns( + portfolio_curve, benchmark_curve, frequency, market=market + ) + observations = int(len(returns.index)) + result: dict[str, Any] = { + "status": "insufficient_history", + "battingAverage": None, + "outperformancePeriods": 0, + "underperformancePeriods": 0, + "outperformancePayoffRatio": None, + "benchmark": benchmark, + "observations": observations, + "frequency": normalized_frequency, + } + if observations == 0: + return result + + batting = _calculate_batting_metrics(returns) + result.update({ + "status": batting["battingStatus"], + "battingAverage": batting["battingAverage"], + "outperformancePeriods": batting["outperformancePeriods"], + "underperformancePeriods": batting["underperformancePeriods"], + "outperformancePayoffRatio": batting["outperformancePayoffRatio"], + }) + return result + + +def _calculate_m_squared_metrics( + returns: pd.DataFrame, + *, + annualization_factor: float, + risk_free_rate: float, +) -> dict[str, Any]: + result: dict[str, Any] = { + "mSquaredStatus": "insufficient_history", + "mSquared": None, + "mSquaredAlpha": None, + } + if len(returns.index) < 2: + return result + + bench_var = max(0.0, float(returns["benchmark"].var(ddof=1))) + port_var = max(0.0, float(returns["portfolio"].var(ddof=1))) + port_vol_ann = math.sqrt(port_var * annualization_factor) + bench_vol_ann = math.sqrt(bench_var * annualization_factor) + + portfolio_annualized = float(returns["portfolio"].mean()) * annualization_factor + benchmark_annualized = float(returns["benchmark"].mean()) * annualization_factor + + if port_vol_ann <= 1e-15: + result.update({ + "mSquaredStatus": "zero_portfolio_variance", + "mSquared": None, + "mSquaredAlpha": None, + }) + return result + + if bench_vol_ann <= 1e-15: + result.update({ + "mSquaredStatus": "zero_benchmark_variance", + "mSquared": risk_free_rate, + "mSquaredAlpha": risk_free_rate - benchmark_annualized, + }) + return result + + sharpe_p = (portfolio_annualized - risk_free_rate) / port_vol_ann + m_squared = risk_free_rate + sharpe_p * bench_vol_ann + m_squared_alpha = m_squared - benchmark_annualized + + result.update({ + "mSquaredStatus": "available", + "mSquared": m_squared, + "mSquaredAlpha": m_squared_alpha, + }) + return result + + +def _calculate_dual_beta_metrics(returns: pd.DataFrame) -> dict[str, Any]: + result: dict[str, Any] = { + "dualBetaStatus": "insufficient_history", + "upBeta": None, + "downBeta": None, + "betaSpread": None, + } + if len(returns.index) < 2: + return result + + up_returns = returns.loc[returns["benchmark"] > 0] + down_returns = returns.loc[returns["benchmark"] < 0] + + up_beta: float | None = None + if len(up_returns.index) >= 2: + up_bench_var = max(0.0, float(up_returns["benchmark"].var(ddof=1))) + if up_bench_var > 1e-15: + up_covar = float(up_returns["portfolio"].cov(up_returns["benchmark"])) + up_beta = up_covar / up_bench_var + + down_beta: float | None = None + if len(down_returns.index) >= 2: + down_bench_var = max(0.0, float(down_returns["benchmark"].var(ddof=1))) + if down_bench_var > 1e-15: + down_covar = float(down_returns["portfolio"].cov(down_returns["benchmark"])) + down_beta = down_covar / down_bench_var + + beta_spread = (up_beta - down_beta) if (up_beta is not None and down_beta is not None) else None + status = "available" if (up_beta is not None or down_beta is not None) else "insufficient_regimes" + + result.update({ + "dualBetaStatus": status, + "upBeta": up_beta, + "downBeta": down_beta, + "betaSpread": beta_spread, + }) + return result + + +def _calculate_batting_metrics(returns: pd.DataFrame) -> dict[str, Any]: + result: dict[str, Any] = { + "battingStatus": "insufficient_history", + "battingAverage": None, + "outperformancePeriods": 0, + "underperformancePeriods": 0, + "outperformancePayoffRatio": None, + } + if returns.empty: + return result + + active = returns["portfolio"] - returns["benchmark"] + outperformed = active > 0 + underperformed = active < 0 + + out_count = int(outperformed.sum()) + under_count = int(underperformed.sum()) + batting_avg = float(out_count / len(returns.index)) + + avg_out: float | None = float(active.loc[outperformed].mean()) if out_count > 0 else None + avg_under: float | None = float(abs(active.loc[underperformed].mean())) if under_count > 0 else None + + payoff: float | None = None + if avg_out is not None and avg_under is not None and avg_under > 1e-15: + payoff = avg_out / avg_under + + result.update({ + "battingStatus": "available", + "battingAverage": batting_avg, + "outperformancePeriods": out_count, + "underperformancePeriods": under_count, + "outperformancePayoffRatio": payoff, + }) + return result diff --git a/backend_api_python/tests/test_backtest_metrics.py b/backend_api_python/tests/test_backtest_metrics.py index c1b5c8038..38d419133 100644 --- a/backend_api_python/tests/test_backtest_metrics.py +++ b/backend_api_python/tests/test_backtest_metrics.py @@ -5,10 +5,13 @@ from app.services.backtest.metrics import ( benchmark_level_curve, + calculate_batting_average, calculate_benchmark_relative_metrics, calculate_beta_and_alpha, calculate_capture_ratios, + calculate_dual_beta, calculate_information_ratio, + calculate_m_squared, ) @@ -577,3 +580,182 @@ def test_calculate_benchmark_relative_metrics_zero_tracking_error(): assert result["upCapture"] == pytest.approx(1.0) assert result["downCapture"] == pytest.approx(1.0) assert result["captureRatio"] == pytest.approx(1.0) + + + +def test_calculate_m_squared_known_values(): + port_ret = [0.02, 0.04, -0.01, 0.03] + bench_ret = [0.01, 0.02, -0.005, 0.015] + portfolio = _curve(port_ret) + benchmark = _curve(bench_ret) + + result = calculate_m_squared( + portfolio, + benchmark, + benchmark="USStock:SPY", + frequency="1d", + annualization_factor=252.0, + risk_free_rate=0.02, + market="USStock", + ) + + port_s = pd.Series(port_ret) + bench_s = pd.Series(bench_ret) + factor = 252.0 + rf = 0.02 + + port_ann = float(port_s.mean()) * factor + bench_ann = float(bench_s.mean()) * factor + port_vol = float(port_s.std(ddof=1)) * math.sqrt(factor) + bench_vol = float(bench_s.std(ddof=1)) * math.sqrt(factor) + + expected_sharpe = (port_ann - rf) / port_vol + expected_m2 = rf + expected_sharpe * bench_vol + expected_m2_alpha = expected_m2 - bench_ann + + assert result["status"] == "available" + assert result["mSquared"] == pytest.approx(expected_m2) + assert result["mSquaredAlpha"] == pytest.approx(expected_m2_alpha) + assert result["portfolioReturnAnnualized"] == pytest.approx(port_ann) + assert result["benchmarkReturnAnnualized"] == pytest.approx(bench_ann) + + +def test_calculate_m_squared_zero_variances_and_validation(): + # Flat portfolio: zero portfolio variance + flat_portfolio = _curve([0.01, 0.01, 0.01]) + varying_bench = _curve([0.01, 0.02, -0.005]) + + res_flat_p = calculate_m_squared(flat_portfolio, varying_bench) + assert res_flat_p["status"] == "zero_portfolio_variance" + assert res_flat_p["mSquared"] is None + assert res_flat_p["mSquaredAlpha"] is None + + # Flat benchmark: zero benchmark variance + varying_portfolio = _curve([0.01, 0.02, -0.005]) + flat_bench = _curve([0.01, 0.01, 0.01]) + + res_flat_b = calculate_m_squared(varying_portfolio, flat_bench, risk_free_rate=0.03) + assert res_flat_b["status"] == "zero_benchmark_variance" + assert res_flat_b["mSquared"] == pytest.approx(0.03) + assert res_flat_b["mSquaredAlpha"] == pytest.approx(0.03 - 252.0 * 0.01) + + # Input validation + with pytest.raises(ValueError, match="annualization_factor"): + calculate_m_squared(varying_portfolio, varying_bench, annualization_factor=-1.0) + with pytest.raises(ValueError, match="risk_free_rate"): + calculate_m_squared(varying_portfolio, varying_bench, risk_free_rate=float("nan")) + + +def test_calculate_dual_beta_known_values(): + # Up-market periods: indices 0, 1, 3 (bench: 0.01, 0.02, 0.015) + # Down-market periods: indices 2, 4 (bench: -0.01, -0.02) + port_ret = [0.02, 0.03, -0.015, 0.025, -0.035] + bench_ret = [0.01, 0.02, -0.01, 0.015, -0.02] + portfolio = _curve(port_ret) + benchmark = _curve(bench_ret) + + result = calculate_dual_beta(portfolio, benchmark) + + assert result["status"] == "available" + assert result["upMarketPeriods"] == 3 + assert result["downMarketPeriods"] == 2 + + # Up beta: + up_p = [0.02, 0.03, 0.025] + up_b = [0.01, 0.02, 0.015] + up_cov = float(pd.Series(up_p).cov(pd.Series(up_b))) + up_var = float(pd.Series(up_b).var(ddof=1)) + expected_up_beta = up_cov / up_var + + # Down beta: + down_p = [-0.015, -0.035] + down_b = [-0.01, -0.02] + down_cov = float(pd.Series(down_p).cov(pd.Series(down_b))) + down_var = float(pd.Series(down_b).var(ddof=1)) + expected_down_beta = down_cov / down_var + + assert result["upBeta"] == pytest.approx(expected_up_beta) + assert result["downBeta"] == pytest.approx(expected_down_beta) + assert result["betaSpread"] == pytest.approx(expected_up_beta - expected_down_beta) + + +def test_calculate_dual_beta_insufficient_regimes(): + # Only up-market + portfolio = _curve([0.02, 0.03, 0.025]) + benchmark = _curve([0.01, 0.02, 0.015]) + + result = calculate_dual_beta(portfolio, benchmark) + assert result["status"] == "available" + assert result["upBeta"] is not None + assert result["downBeta"] is None + assert result["betaSpread"] is None + assert result["upMarketPeriods"] == 3 + assert result["downMarketPeriods"] == 0 + + +def test_calculate_batting_average_and_payoff_ratio(): + # 4 observations: + # 0: port 0.02, bench 0.01 -> diff +0.01 (win) + # 1: port 0.03, bench 0.01 -> diff +0.02 (win) + # 2: port -0.01, bench 0.005 -> diff -0.015 (loss) + # 3: port 0.015, bench 0.01 -> diff +0.005 (win) + # Batting average = 3 / 4 = 0.75 + # Outperformance average = (0.01 + 0.02 + 0.005) / 3 = 0.035 / 3 = 0.01166667 + # Underperformance average = 0.015 + # Payoff ratio = (0.035 / 3) / 0.015 = 0.035 / 0.045 = 0.7777777777777778 + port_ret = [0.02, 0.03, -0.01, 0.015] + bench_ret = [0.01, 0.01, 0.005, 0.01] + portfolio = _curve(port_ret) + benchmark = _curve(bench_ret) + + result = calculate_batting_average(portfolio, benchmark) + + assert result["status"] == "available" + assert result["battingAverage"] == pytest.approx(0.75) + assert result["outperformancePeriods"] == 3 + assert result["underperformancePeriods"] == 1 + expected_payoff = (0.035 / 3.0) / 0.015 + assert result["outperformancePayoffRatio"] == pytest.approx(expected_payoff) + + +def test_calculate_batting_average_all_wins_no_losses(): + portfolio = _curve([0.02, 0.03]) + benchmark = _curve([0.01, 0.01]) + + result = calculate_batting_average(portfolio, benchmark) + assert result["battingAverage"] == pytest.approx(1.0) + assert result["outperformancePeriods"] == 2 + assert result["underperformancePeriods"] == 0 + assert result["outperformancePayoffRatio"] is None + + +def test_calculate_benchmark_relative_metrics_includes_m2_dual_beta_and_batting(): + # Integrate into full calculate_benchmark_relative_metrics + port_ret = [0.02, 0.03, -0.015, 0.025, -0.035] + bench_ret = [0.01, 0.02, -0.01, 0.015, -0.02] + portfolio = _curve(port_ret) + benchmark = _curve(bench_ret) + + result = calculate_benchmark_relative_metrics( + portfolio, + benchmark, + benchmark="USStock:SPY", + frequency="1d", + annualization_factor=252.0, + risk_free_rate=0.01, + market="USStock", + ) + + # Verify that new Idea 2 fields are properly integrated + assert result["mSquaredStatus"] == "available" + assert result["mSquared"] is not None + assert result["mSquaredAlpha"] is not None + + assert result["dualBetaStatus"] == "available" + assert result["upBeta"] is not None + assert result["downBeta"] is not None + assert result["betaSpread"] is not None + + assert result["battingStatus"] == "available" + assert result["battingAverage"] is not None + assert result["outperformancePeriods"] > 0 diff --git a/docs/product/BACKTEST_CENTER.md b/docs/product/BACKTEST_CENTER.md index f2461f1e1..9a85e3361 100644 --- a/docs/product/BACKTEST_CENTER.md +++ b/docs/product/BACKTEST_CENTER.md @@ -58,9 +58,17 @@ In addition to the Information Ratio, `benchmarkRelativeMetrics` provides: - `upCapture`: Geometric-mean portfolio return divided by geometric-mean benchmark return across periods where $R_b > 0$. - `downCapture`: Geometric-mean portfolio return divided by geometric-mean benchmark return across periods where $R_b < 0$. - `captureRatio`: Ratio of upside capture to downside capture (`upCapture / downCapture`). A value $> 1.0$ indicates greater participation in up markets than down markets. +- **Modigliani–Modigliani ($M^2$) & $M^2 ext{ Alpha}$:** Scales portfolio volatility to match benchmark volatility, converting the Sharpe ratio into an annualized percentage return: $M^2 = R_f + ext{Sharpe}_p imes \sigma_{b, ext{ann}}$ and $M^2_{ ext{alpha}} = M^2 - R_{b, ext{ann}}$. +- **Dual Beta (Up-Market & Down-Market Beta):** + - `upBeta` ($eta^+$): Systematic sensitivity evaluated strictly over benchmark up-market periods ($R_b > 0$). + - `downBeta` ($eta^-$): Systematic sensitivity evaluated strictly over benchmark down-market periods ($R_b < 0$). + - `betaSpread`: Difference between upside and downside beta ($eta^+ - eta^-$). A positive spread indicates convex upside participation with defensive downside insulation. +- **Batting Average & Outperformance Payoff:** + - `battingAverage`: Percentage of observation periods where portfolio return exceeds benchmark return ($\sum \mathbb{I}(R_{p,t} > R_{b,t}) / T$). + - `outperformancePayoffRatio`: Ratio of average active return during outperforming periods to average active loss during underperforming periods. The top-level `status` remains the Information Ratio status for compatibility. -`betaStatus`, `correlationStatus`, `treynorStatus`, and `captureStatus` +`betaStatus`, `correlationStatus`, `treynorStatus`, `captureStatus`, `mSquaredStatus`, `dualBetaStatus`, and `battingStatus` describe metric-specific availability. In particular, a flat benchmark makes Beta unavailable but does not suppress a valid Information Ratio. Correlation and $R^2$ remain unset when either return series is constant because Pearson diff --git a/docs/product/BACKTEST_CENTER_CN.md b/docs/product/BACKTEST_CENTER_CN.md index 7e5515bd9..a46db0d50 100644 --- a/docs/product/BACKTEST_CENTER_CN.md +++ b/docs/product/BACKTEST_CENTER_CN.md @@ -40,9 +40,17 @@ - **相关系数和 $R^2$**：策略与基准收益的 Pearson 相关性及其解释比例。任一收益序列恒定时，这两个值保持为空，因为此时相关系数没有数学定义。 - **上涨/下跌捕获率**：分别在基准上涨和下跌周期内，用策略几何平均收益除以基准几何平均收益；`1.0` 表示 100% 捕获。 - **捕获比率**：上涨捕获率除以下跌捕获率。 +- **Modigliani–Modigliani ($M^2$) 及 $M^2$ Alpha**：将策略杠杆调整至与基准波动率一致后的年化收益率：$M^2 = R_f + \text{Sharpe}_p \times \sigma_{b,\text{ann}}$，以及相对基准的超额 $M^2_{\text{alpha}} = M^2 - R_{b,\text{ann}}$。 +- **双重 Beta（上涨 Beta 与下跌 Beta）**： + - `upBeta`（$\beta^+$）：仅在基准上涨周期（$R_b > 0$）内计算的系统性敏感度。 + - `downBeta`（$\beta^-$）：仅在基准下跌周期（$R_b < 0$）内计算的系统性敏感度。 + - `betaSpread`：上涨 Beta 与下跌 Beta 之差（$\beta^+ - \beta^-$）。正差值体现上行参与度高、下行防御性强的凸性收益特征。 +- **胜率（Batting Average）与超额盈亏比**： + - `battingAverage`：策略收益跑赢基准的周期占比（$\sum \mathbb{I}(R_{p,t} > R_{b,t}) / T$）。 + - `outperformancePayoffRatio`：跑赢周期平均超额收益与跑输周期平均超额亏损之比。 为保持兼容，顶层 `status` 继续表示信息比率状态；`betaStatus`、 -`correlationStatus`、`treynorStatus` 和 `captureStatus` 分别说明对应指标是否 +`correlationStatus`、`treynorStatus`、`captureStatus`、`mSquaredStatus`、`dualBetaStatus` 和 `battingStatus` 分别说明对应指标是否 可用。基准收益恒定只会使 Beta 不可用，不会清空仍然可以计算的信息比率。 读取指标前应先检查相应状态和有效观察数。