From c534ddeb3a88425c766f20f73a750bd6da572fe1 Mon Sep 17 00:00:00 2001 From: huuan26 Date: Fri, 23 Oct 2026 10:25:38 +0700 Subject: [PATCH 1/2] feat(backtest): expand benchmark-relative metrics with Beta, Jensen's Alpha, Treynor, and Capture Ratios - Add _align_curves_and_returns helper in app/services/backtest/metrics.py to share common benchmark curve alignment and interval validation. - Implement calculate_beta_and_alpha for systematic risk (Beta), CAPM excess return (Jensen's Alpha), Treynor Ratio, Pearson Correlation, and R-squared. - Implement calculate_capture_ratios for compounded Up-Market Capture, Down-Market Capture, and Overall Capture Ratio. - Implement calculate_benchmark_relative_metrics combining Information Ratio, Beta/Alpha, and Capture metrics while maintaining backward compatibility with calculate_information_ratio. - Update Strategy API V2 backtest service to compute comprehensive benchmark-relative metrics. - Document metrics and formulas in docs/product/BACKTEST_CENTER.md. - Add comprehensive unit test coverage in tests/test_backtest_metrics.py. --- .../app/services/backtest/metrics.py | 383 +++++++++++++++++- .../app/services/strategy_v2/service.py | 8 +- .../tests/test_backtest_metrics.py | 247 ++++++++++- docs/product/BACKTEST_CENTER.md | 11 + 4 files changed, 629 insertions(+), 20 deletions(-) diff --git a/backend_api_python/app/services/backtest/metrics.py b/backend_api_python/app/services/backtest/metrics.py index f4025175c..09e271cc6 100644 --- a/backend_api_python/app/services/backtest/metrics.py +++ b/backend_api_python/app/services/backtest/metrics.py @@ -10,6 +10,15 @@ from app.services.market_schedule import CALENDAR_BY_MARKET, equity_bar_session_date +__all__ = [ + "DEFAULT_INFORMATION_RATIO_BANDS", + "benchmark_level_curve", + "calculate_benchmark_relative_metrics", + "calculate_beta_and_alpha", + "calculate_capture_ratios", + "calculate_information_ratio", +] + _FREQUENCY_SECONDS = { "1m": 60, @@ -33,6 +42,37 @@ ) +def _align_curves_and_returns( + portfolio_curve: Iterable[dict[str, Any]], + benchmark_curve: Iterable[dict[str, Any]], + frequency: str, + market: str = "", +) -> tuple[pd.DataFrame, str]: + """Align portfolio and benchmark curves to native benchmark timestamps. + + Returns the aligned periodic returns DataFrame with columns 'portfolio' and + 'benchmark', along with the normalized frequency. Non-continuous gaps and + non-consecutive equity sessions are excluded. + """ + normalized_frequency = _normalize_frequency(frequency) + levels = pd.concat( + ( + _curve_levels(portfolio_curve).rename("portfolio"), + _curve_levels(benchmark_curve).rename("benchmark"), + ), + axis=1, + join="inner", + ).dropna() + returns = levels.pct_change(fill_method=None).dropna() + if not returns.empty: + valid_intervals = _valid_interval_mask(levels.index, normalized_frequency, market) + returns = returns.loc[valid_intervals] + returns = returns.loc[ + returns.apply(lambda row: all(math.isfinite(float(value)) for value in row), axis=1) + ] + return returns, normalized_frequency + + def calculate_information_ratio( portfolio_curve: Iterable[dict[str, Any]], benchmark_curve: Iterable[dict[str, Any]], @@ -51,30 +91,17 @@ def calculate_information_ratio( frequency than the strategy. Gaps that are not valid consecutive market periods are excluded instead of being annualized as a single period. """ - normalized_frequency = _normalize_frequency(frequency) factor = float(annualization_factor) if not math.isfinite(factor) or factor <= 0: raise ValueError("annualization_factor must be a positive finite number") bands = _validate_classification_bands(classification_bands) - levels = pd.concat( - ( - _curve_levels(portfolio_curve).rename("portfolio"), - _curve_levels(benchmark_curve).rename("benchmark"), - ), - axis=1, - join="inner", - ).dropna() - returns = levels.pct_change(fill_method=None).dropna() - if not returns.empty: - valid_intervals = _valid_interval_mask(levels.index, normalized_frequency, market) - returns = returns.loc[valid_intervals] - returns = returns.loc[ - returns.apply(lambda row: all(math.isfinite(float(value)) for value in row), axis=1) - ] + returns, normalized_frequency = _align_curves_and_returns( + portfolio_curve, benchmark_curve, frequency, market=market + ) observations = int(len(returns.index)) - result = { + result: dict[str, Any] = { "status": "insufficient_history", "portfolioReturnAnnualized": None, "benchmarkReturnAnnualized": None, @@ -118,6 +145,328 @@ def calculate_information_ratio( return result +def calculate_beta_and_alpha( + 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 Beta, Jensen's Alpha, Treynor Ratio, Correlation, and R-squared. + + Beta measures systematic risk relative to the benchmark: + beta = Cov(R_p, R_b) / Var(R_b). + Jensen's Alpha measures the annualized risk-adjusted excess return over CAPM: + alpha_ann = (R_p_ann - R_f) - beta * (R_b_ann - R_f). + Treynor Ratio measures excess return per unit of systematic risk: + (R_p_ann - R_f) / beta. + """ + 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, + "beta": None, + "alphaAnnualized": None, + "treynorRatio": None, + "correlation": None, + "rSquared": None, + "benchmark": benchmark, + "observations": observations, + "frequency": normalized_frequency, + "annualizationFactor": factor, + } + if observations == 0: + return result + + portfolio_annualized = float(returns["portfolio"].mean()) * factor + benchmark_annualized = float(returns["benchmark"].mean()) * factor + result.update({ + "portfolioReturnAnnualized": portfolio_annualized, + "benchmarkReturnAnnualized": benchmark_annualized, + }) + if observations < 2: + return result + + bench_var = float(returns["benchmark"].var(ddof=1)) + port_var = float(returns["portfolio"].var(ddof=1)) + covar = float(returns["portfolio"].cov(returns["benchmark"])) + + if bench_var <= 1e-15: + result["status"] = "zero_benchmark_variance" + return result + + beta = covar / bench_var + alpha_annualized = (portfolio_annualized - rf) - beta * (benchmark_annualized - rf) + treynor_ratio = (portfolio_annualized - rf) / beta if abs(beta) > 1e-15 else None + + bench_std = math.sqrt(bench_var) + port_std = math.sqrt(port_var) + if port_std > 1e-15 and bench_std > 1e-15: + raw_corr = covar / (port_std * bench_std) + corr = max(-1.0, min(1.0, raw_corr)) + r_squared = max(0.0, min(1.0, corr**2)) + elif port_std <= 1e-15: + corr = 0.0 + r_squared = 0.0 + else: + corr = None + r_squared = None + + result.update({ + "status": "available", + "beta": beta, + "alphaAnnualized": alpha_annualized, + "treynorRatio": treynor_ratio, + "correlation": corr, + "rSquared": r_squared, + }) + return result + + +def calculate_capture_ratios( + 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 Capture, Down-Market Capture, and Capture Ratio. + + Up-market periods are intervals where the benchmark return is strictly positive (> 0). + Down-market periods are intervals where the benchmark return is strictly negative (< 0). + Capture ratios are computed as the compounded return of the portfolio divided by + the compounded return of the benchmark in those respective periods: + Up Capture = (Prod(1 + R_p_up) - 1) / (Prod(1 + R_b_up) - 1) + Down Capture = (Prod(1 + R_p_down) - 1) / (Prod(1 + R_b_down) - 1) + Capture Ratio = Up Capture / Down Capture. + """ + 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", + "upCapture": None, + "downCapture": None, + "captureRatio": None, + "upMarketPeriods": 0, + "downMarketPeriods": 0, + "benchmark": benchmark, + "observations": observations, + "frequency": normalized_frequency, + } + if observations == 0: + return result + + up_returns = returns.loc[returns["benchmark"] > 0] + down_returns = returns.loc[returns["benchmark"] < 0] + up_periods = int(len(up_returns)) + down_periods = int(len(down_returns)) + result["upMarketPeriods"] = up_periods + result["downMarketPeriods"] = down_periods + + up_capture = None + if up_periods > 0: + bench_up_comp = float((1.0 + up_returns["benchmark"]).prod() - 1.0) + port_up_comp = float((1.0 + up_returns["portfolio"]).prod() - 1.0) + if abs(bench_up_comp) > 1e-15: + up_capture = port_up_comp / bench_up_comp + + down_capture = None + if down_periods > 0: + bench_down_comp = float((1.0 + down_returns["benchmark"]).prod() - 1.0) + port_down_comp = float((1.0 + down_returns["portfolio"]).prod() - 1.0) + if abs(bench_down_comp) > 1e-15: + down_capture = port_down_comp / bench_down_comp + + capture_ratio = None + if up_capture is not None and down_capture is not None and abs(down_capture) > 1e-15: + capture_ratio = up_capture / down_capture + + status = ( + "available" + if (up_capture is not None or down_capture is not None) + else "insufficient_regimes" + ) + result.update({ + "status": status, + "upCapture": up_capture, + "downCapture": down_capture, + "captureRatio": capture_ratio, + }) + return result + + +def calculate_benchmark_relative_metrics( + portfolio_curve: Iterable[dict[str, Any]], + benchmark_curve: Iterable[dict[str, Any]], + *, + benchmark: str | None, + frequency: str, + annualization_factor: float, + risk_free_rate: float = 0.0, + market: str = "", + classification_bands: Sequence[tuple[float, str]] = DEFAULT_INFORMATION_RATIO_BANDS, +) -> dict[str, Any]: + """Calculate the complete suite of benchmark-relative performance metrics. + + Includes: + - Information Ratio, Active Return, Tracking Error, and Classification + - Beta, Jensen's Alpha, and Treynor Ratio + - Correlation and R-Squared + - Up-Market Capture, Down-Market Capture, and Capture Ratio + """ + 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") + + bands = _validate_classification_bands(classification_bands) + 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, + "activeReturnAnnualized": None, + "trackingErrorAnnualized": None, + "informationRatio": None, + "classification": None, + "riskFreeRate": rf, + "beta": None, + "alphaAnnualized": None, + "treynorRatio": None, + "correlation": None, + "rSquared": None, + "upCapture": None, + "downCapture": None, + "captureRatio": None, + "upMarketPeriods": 0, + "downMarketPeriods": 0, + "benchmark": benchmark, + "observations": observations, + "frequency": normalized_frequency, + "annualizationFactor": factor, + } + if observations == 0: + return result + + active_returns = returns["portfolio"] - returns["benchmark"] + portfolio_annualized = float(returns["portfolio"].mean()) * factor + benchmark_annualized = float(returns["benchmark"].mean()) * factor + active_annualized = float(active_returns.mean()) * factor + result.update({ + "portfolioReturnAnnualized": portfolio_annualized, + "benchmarkReturnAnnualized": benchmark_annualized, + "activeReturnAnnualized": active_annualized, + }) + + # Capture ratios + up_returns = returns.loc[returns["benchmark"] > 0] + down_returns = returns.loc[returns["benchmark"] < 0] + up_periods = int(len(up_returns)) + down_periods = int(len(down_returns)) + result["upMarketPeriods"] = up_periods + result["downMarketPeriods"] = down_periods + + up_capture = None + if up_periods > 0: + bench_up_comp = float((1.0 + up_returns["benchmark"]).prod() - 1.0) + port_up_comp = float((1.0 + up_returns["portfolio"]).prod() - 1.0) + if abs(bench_up_comp) > 1e-15: + up_capture = port_up_comp / bench_up_comp + + down_capture = None + if down_periods > 0: + bench_down_comp = float((1.0 + down_returns["benchmark"]).prod() - 1.0) + port_down_comp = float((1.0 + down_returns["portfolio"]).prod() - 1.0) + if abs(bench_down_comp) > 1e-15: + down_capture = port_down_comp / bench_down_comp + + capture_ratio = None + if up_capture is not None and down_capture is not None and abs(down_capture) > 1e-15: + capture_ratio = up_capture / down_capture + + result.update({ + "upCapture": up_capture, + "downCapture": down_capture, + "captureRatio": capture_ratio, + }) + + if observations < 2: + return result + + # Tracking error + periodic_tracking_error = float(active_returns.std(ddof=1)) + tracking_error_annualized = periodic_tracking_error * math.sqrt(factor) + result["trackingErrorAnnualized"] = tracking_error_annualized + + bench_var = float(returns["benchmark"].var(ddof=1)) + port_var = float(returns["portfolio"].var(ddof=1)) + covar = float(returns["portfolio"].cov(returns["benchmark"])) + + if bench_var > 1e-15: + beta = covar / bench_var + alpha_annualized = (portfolio_annualized - rf) - beta * (benchmark_annualized - rf) + treynor_ratio = (portfolio_annualized - rf) / beta if abs(beta) > 1e-15 else None + result.update({ + "beta": beta, + "alphaAnnualized": alpha_annualized, + "treynorRatio": treynor_ratio, + }) + + bench_std = math.sqrt(bench_var) + port_std = math.sqrt(port_var) + if port_std > 1e-15 and bench_std > 1e-15: + raw_corr = covar / (port_std * bench_std) + corr = max(-1.0, min(1.0, raw_corr)) + r_squared = max(0.0, min(1.0, corr**2)) + result["correlation"] = corr + result["rSquared"] = r_squared + elif port_std <= 1e-15 and bench_std > 1e-15: + result["correlation"] = 0.0 + result["rSquared"] = 0.0 + + if math.isclose(periodic_tracking_error, 0.0, rel_tol=0.0, abs_tol=1e-15): + result["status"] = "zero_tracking_error" + return result + + if bench_var <= 1e-15: + result["status"] = "zero_benchmark_variance" + return result + + information_ratio = active_annualized / tracking_error_annualized + result.update({ + "status": "available", + "informationRatio": information_ratio, + "classification": _classify_information_ratio(information_ratio, bands), + }) + return result + + def benchmark_level_curve(frame: pd.DataFrame | None) -> list[dict[str, Any]]: """Return valid native-frequency benchmark levels without forward filling.""" if frame is None or frame.empty or "close" not in frame.columns: diff --git a/backend_api_python/app/services/strategy_v2/service.py b/backend_api_python/app/services/strategy_v2/service.py index 383f9c761..ba27a8a09 100644 --- a/backend_api_python/app/services/strategy_v2/service.py +++ b/backend_api_python/app/services/strategy_v2/service.py @@ -13,7 +13,11 @@ import pandas as pd from app.data_sources.errors import MarketDataUnavailableError -from app.services.backtest.metrics import benchmark_level_curve, calculate_information_ratio +from app.services.backtest.metrics import ( + benchmark_level_curve, + calculate_benchmark_relative_metrics, + calculate_information_ratio, +) from app.services.backtest_limits import ( BacktestRangeLimitError, backtest_warmup_calendar_days, @@ -379,7 +383,7 @@ def resolve_universe(reference: str, timestamp: pd.Timestamp) -> list[str]: result["excessReturn"] = float(result.get("totalReturn") or 0.0) - float(result.get("benchmarkTotalReturn") or 0.0) normalized_benchmark_frequency = normalize_frequency(benchmark_frequency) benchmark_markets = (benchmark_spec.market,) if benchmark_spec is not None else manifest.markets - result["benchmarkRelativeMetrics"] = calculate_information_ratio( + result["benchmarkRelativeMetrics"] = calculate_benchmark_relative_metrics( result.get("equityCurve") or [], benchmark_level_curve(benchmark_frame), benchmark=benchmark_spec.key if benchmark_spec is not None else None, diff --git a/backend_api_python/tests/test_backtest_metrics.py b/backend_api_python/tests/test_backtest_metrics.py index 996832388..81e033b56 100644 --- a/backend_api_python/tests/test_backtest_metrics.py +++ b/backend_api_python/tests/test_backtest_metrics.py @@ -3,7 +3,13 @@ import pandas as pd import pytest -from app.services.backtest.metrics import benchmark_level_curve, calculate_information_ratio +from app.services.backtest.metrics import ( + benchmark_level_curve, + calculate_benchmark_relative_metrics, + calculate_beta_and_alpha, + calculate_capture_ratios, + calculate_information_ratio, +) def _curve(returns, *, frequency="D"): @@ -276,3 +282,242 @@ def test_metric_normalizes_monthly_aliases(): market="Crypto", ) assert result["frequency"] == "1mo" + + + +def test_calculate_beta_and_alpha_known_values(): + result = calculate_beta_and_alpha( + _curve([0.02, 0.04, -0.01, 0.03]), + _curve([0.01, 0.02, -0.005, 0.015]), + benchmark="USStock:SPY", + frequency="1d", + annualization_factor=252.0, + risk_free_rate=0.0, + market="USStock", + ) + + assert result["status"] == "available" + assert result["observations"] == 4 + assert result["frequency"] == "1d" + assert result["annualizationFactor"] == 252.0 + assert result["portfolioReturnAnnualized"] == pytest.approx(5.04) + assert result["benchmarkReturnAnnualized"] == pytest.approx(2.52) + assert result["beta"] == pytest.approx(2.0) + assert result["alphaAnnualized"] == pytest.approx(0.0) + assert result["treynorRatio"] == pytest.approx(2.52) + assert result["correlation"] == pytest.approx(1.0) + assert result["rSquared"] == pytest.approx(1.0) + + +def test_calculate_beta_and_alpha_with_positive_risk_free_rate(): + result = calculate_beta_and_alpha( + _curve([0.02, 0.04, -0.01, 0.03]), + _curve([0.01, 0.02, -0.005, 0.015]), + benchmark="USStock:SPY", + frequency="1d", + annualization_factor=252.0, + risk_free_rate=0.04, + market="USStock", + ) + + assert result["status"] == "available" + assert result["riskFreeRate"] == pytest.approx(0.04) + assert result["beta"] == pytest.approx(2.0) + # alpha = (5.04 - 0.04) - 2.0 * (2.52 - 0.04) = 5.00 - 4.96 = 0.04 + assert result["alphaAnnualized"] == pytest.approx(0.04) + # treynor = (5.04 - 0.04) / 2.0 = 2.50 + assert result["treynorRatio"] == pytest.approx(2.50) + + +def test_calculate_capture_ratios_known_values(): + result = calculate_capture_ratios( + _curve([0.06, -0.02, 0.04, -0.01]), + _curve([0.05, -0.04, 0.03, -0.02]), + benchmark="USStock:SPY", + frequency="1d", + market="USStock", + ) + + assert result["status"] == "available" + assert result["observations"] == 4 + assert result["upMarketPeriods"] == 2 + assert result["downMarketPeriods"] == 2 + # Up market: benchmark = [0.05, 0.03], port = [0.06, 0.04] + # bench_up = 1.05 * 1.03 - 1 = 0.0815 + # port_up = 1.06 * 1.04 - 1 = 0.1024 + expected_up_capture = 0.1024 / 0.0815 + assert result["upCapture"] == pytest.approx(expected_up_capture) + # Down market: benchmark = [-0.04, -0.02], port = [-0.02, -0.01] + # bench_down = 0.96 * 0.98 - 1 = -0.0592 + # port_down = 0.98 * 0.99 - 1 = -0.0298 + expected_down_capture = -0.0298 / -0.0592 + assert result["downCapture"] == pytest.approx(expected_down_capture) + assert result["captureRatio"] == pytest.approx(expected_up_capture / expected_down_capture) + + +def test_capture_ratios_only_up_market(): + result = calculate_capture_ratios( + _curve([0.02, 0.03, 0.01]), + _curve([0.01, 0.02, 0.01]), + frequency="1d", + market="USStock", + ) + + assert result["status"] == "available" + assert result["upMarketPeriods"] == 3 + assert result["downMarketPeriods"] == 0 + assert result["upCapture"] is not None + assert result["downCapture"] is None + assert result["captureRatio"] is None + + +def test_capture_ratios_only_down_market(): + result = calculate_capture_ratios( + _curve([-0.01, -0.02, -0.005]), + _curve([-0.02, -0.03, -0.01]), + frequency="1d", + market="USStock", + ) + + assert result["status"] == "available" + assert result["upMarketPeriods"] == 0 + assert result["downMarketPeriods"] == 3 + assert result["upCapture"] is None + assert result["downCapture"] is not None + assert result["captureRatio"] is None + + +def test_zero_benchmark_variance(): + result = calculate_beta_and_alpha( + _curve([0.01, 0.02, 0.03]), + _curve([0.0, 0.0, 0.0]), + frequency="1d", + annualization_factor=252.0, + market="USStock", + ) + + assert result["status"] == "zero_benchmark_variance" + assert result["beta"] is None + assert result["alphaAnnualized"] is None + assert result["treynorRatio"] is None + + +def test_zero_beta_treynor_is_none(): + result = calculate_beta_and_alpha( + _curve([0.02, 0.02, -0.02, -0.02]), + _curve([0.01, -0.01, 0.01, -0.01]), + frequency="1d", + annualization_factor=252.0, + market="USStock", + ) + + assert result["status"] == "available" + assert result["beta"] == pytest.approx(0.0, abs=1e-12) + assert result["treynorRatio"] is None + + +def test_benchmark_metrics_insufficient_history(): + result_zero = calculate_benchmark_relative_metrics( + [], + [], + benchmark="USStock:SPY", + frequency="1d", + annualization_factor=252.0, + market="USStock", + ) + assert result_zero["status"] == "insufficient_history" + assert result_zero["observations"] == 0 + assert result_zero["portfolioReturnAnnualized"] is None + assert result_zero["beta"] is None + + result_one = calculate_benchmark_relative_metrics( + [{"time": "2026-01-01T00:00:00Z", "value": 100.0}, {"time": "2026-01-02T00:00:00Z", "value": 102.0}], + [{"time": "2026-01-01T00:00:00Z", "value": 100.0}, {"time": "2026-01-02T00:00:00Z", "value": 101.0}], + benchmark="USStock:SPY", + frequency="1d", + annualization_factor=252.0, + market="USStock", + ) + assert result_one["status"] == "insufficient_history" + assert result_one["observations"] == 1 + assert result_one["portfolioReturnAnnualized"] == pytest.approx(252.0 * 0.02) + assert result_one["benchmarkReturnAnnualized"] == pytest.approx(252.0 * 0.01) + assert result_one["beta"] is None + assert result_one["informationRatio"] is None + + +def test_benchmark_metrics_input_validation(): + with pytest.raises(ValueError, match="annualization_factor"): + calculate_benchmark_relative_metrics( + [], [], benchmark="USStock:SPY", frequency="1d", annualization_factor=-1.0 + ) + + with pytest.raises(ValueError, match="annualization_factor"): + calculate_benchmark_relative_metrics( + [], [], benchmark="USStock:SPY", frequency="1d", annualization_factor=float("nan") + ) + + with pytest.raises(ValueError, match="risk_free_rate"): + calculate_benchmark_relative_metrics( + [], [], benchmark="USStock:SPY", frequency="1d", annualization_factor=252.0, risk_free_rate=float("nan") + ) + + +def test_calculate_benchmark_relative_metrics_full_integration(): + portfolio = _curve([0.01, 0.02, -0.01, 0.005]) + benchmark = _curve([0.005, 0.01, -0.005, 0.002]) + + result = calculate_benchmark_relative_metrics( + portfolio, + benchmark, + benchmark="USStock:SPY", + frequency="1d", + annualization_factor=252.0, + risk_free_rate=0.02, + market="USStock", + ) + + assert result["status"] == "available" + assert result["benchmark"] == "USStock:SPY" + assert result["frequency"] == "1d" + assert result["annualizationFactor"] == 252.0 + assert result["observations"] == 4 + assert result["portfolioReturnAnnualized"] == pytest.approx(1.575) + assert result["benchmarkReturnAnnualized"] == pytest.approx(0.756) + assert result["activeReturnAnnualized"] == pytest.approx(0.819) + assert result["informationRatio"] == pytest.approx(8.270196225621152) + assert result["classification"] == "exceptional" + assert result["riskFreeRate"] == pytest.approx(0.02) + assert result["beta"] == pytest.approx(1.991525423728826) + assert result["alphaAnnualized"] == pytest.approx(0.08923728813558407) + assert result["treynorRatio"] == pytest.approx(0.7808085106382866) + assert result["correlation"] == pytest.approx(0.99920872649114) + assert result["rSquared"] == pytest.approx(0.9984180790960459) + assert result["upMarketPeriods"] == 3 + assert result["downMarketPeriods"] == 1 + assert result["upCapture"] == pytest.approx(2.0697185613667264) + assert result["downCapture"] == pytest.approx(2.0) + assert result["captureRatio"] == pytest.approx(1.0348592806833516) + + +def test_calculate_benchmark_relative_metrics_zero_tracking_error(): + curve = _curve([0.01, 0.02, -0.005, 0.015]) + result = calculate_benchmark_relative_metrics( + curve, + curve, + benchmark="Crypto:BTC/USDT", + frequency="1d", + annualization_factor=365.25, + market="Crypto", + ) + + assert result["status"] == "zero_tracking_error" + assert result["trackingErrorAnnualized"] == pytest.approx(0.0, abs=1e-12) + assert result["informationRatio"] is None + assert result["beta"] == pytest.approx(1.0) + assert result["alphaAnnualized"] == pytest.approx(0.0, abs=1e-12) + assert result["correlation"] == pytest.approx(1.0) + assert result["rSquared"] == pytest.approx(1.0) + assert result["upCapture"] == pytest.approx(1.0) + assert result["downCapture"] == pytest.approx(1.0) + assert result["captureRatio"] == pytest.approx(1.0) diff --git a/docs/product/BACKTEST_CENTER.md b/docs/product/BACKTEST_CENTER.md index 0acb7b2e6..959ba470f 100644 --- a/docs/product/BACKTEST_CENTER.md +++ b/docs/product/BACKTEST_CENTER.md @@ -48,6 +48,17 @@ The default interpretation bands (`weak`, `acceptable`, `good`, and `exceptional`) are operational labels rather than a universal market standard. The calculation accepts alternative bands for mandates that require them. +In addition to the Information Ratio, `benchmarkRelativeMetrics` provides: + +- **Beta ($\beta$):** Systematic risk relative to the benchmark ($\text{Cov}(R_p, R_b) / \text{Var}(R_b)$). Zero benchmark variance yields `zero_benchmark_variance` status. +- **Jensen's Alpha ($\alpha_{\text{ann}}$):** Annualized risk-adjusted excess return over CAPM expected return: $(R_{p,\text{ann}} - R_f) - \beta (R_{b,\text{ann}} - R_f)$. +- **Treynor Ratio:** Annualized excess return per unit of systematic risk: $(R_{p,\text{ann}} - R_f) / \beta$. +- **Correlation & $R^2$:** Pearson correlation coefficient ($r \in [-1, 1]$) and coefficient of determination ($R^2 \in [0, 1]$). +- **Capture Ratios:** + - `upCapture`: Compounded portfolio return divided by compounded benchmark return across periods where $R_b > 0$. + - `downCapture`: Compounded portfolio return divided by compounded 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. + Benchmark choice remains part of the research hypothesis. CDI can be suitable for Brazilian cash-like and low-duration fixed-income strategies, but it is not an automatic default for inflation-linked, longer-duration, or credit-risk From fb15bce991d9fd6d5cd2c4d0d86f5f3c302630da Mon Sep 17 00:00:00 2001 From: TIANHE Date: Fri, 23 Oct 2026 13:27:06 +0800 Subject: [PATCH 2/2] fix(backtest): harden benchmark-relative metrics --- .../app/services/backtest/metrics.py | 277 +++++++++--------- .../tests/test_backtest_metrics.py | 80 ++++- docs/product/BACKTEST_CENTER.md | 14 +- docs/product/BACKTEST_CENTER_CN.md | 23 ++ 4 files changed, 243 insertions(+), 151 deletions(-) diff --git a/backend_api_python/app/services/backtest/metrics.py b/backend_api_python/app/services/backtest/metrics.py index 09e271cc6..60be10d32 100644 --- a/backend_api_python/app/services/backtest/metrics.py +++ b/backend_api_python/app/services/backtest/metrics.py @@ -178,6 +178,8 @@ def calculate_beta_and_alpha( observations = int(len(returns.index)) result: dict[str, Any] = { "status": "insufficient_history", + "correlationStatus": "insufficient_history", + "treynorStatus": "insufficient_history", "portfolioReturnAnnualized": None, "benchmarkReturnAnnualized": None, "riskFreeRate": rf, @@ -203,39 +205,14 @@ def calculate_beta_and_alpha( if observations < 2: return result - bench_var = float(returns["benchmark"].var(ddof=1)) - port_var = float(returns["portfolio"].var(ddof=1)) - covar = float(returns["portfolio"].cov(returns["benchmark"])) - - if bench_var <= 1e-15: - result["status"] = "zero_benchmark_variance" - return result - - beta = covar / bench_var - alpha_annualized = (portfolio_annualized - rf) - beta * (benchmark_annualized - rf) - treynor_ratio = (portfolio_annualized - rf) / beta if abs(beta) > 1e-15 else None - - bench_std = math.sqrt(bench_var) - port_std = math.sqrt(port_var) - if port_std > 1e-15 and bench_std > 1e-15: - raw_corr = covar / (port_std * bench_std) - corr = max(-1.0, min(1.0, raw_corr)) - r_squared = max(0.0, min(1.0, corr**2)) - elif port_std <= 1e-15: - corr = 0.0 - r_squared = 0.0 - else: - corr = None - r_squared = None - - result.update({ - "status": "available", - "beta": beta, - "alphaAnnualized": alpha_annualized, - "treynorRatio": treynor_ratio, - "correlation": corr, - "rSquared": r_squared, - }) + systematic = _calculate_systematic_risk_metrics( + returns, + annualization_factor=factor, + risk_free_rate=rf, + ) + result.update(systematic) + result["status"] = systematic["betaStatus"] + result.pop("betaStatus", None) return result @@ -251,10 +228,10 @@ def calculate_capture_ratios( Up-market periods are intervals where the benchmark return is strictly positive (> 0). Down-market periods are intervals where the benchmark return is strictly negative (< 0). - Capture ratios are computed as the compounded return of the portfolio divided by - the compounded return of the benchmark in those respective periods: - Up Capture = (Prod(1 + R_p_up) - 1) / (Prod(1 + R_b_up) - 1) - Down Capture = (Prod(1 + R_p_down) - 1) / (Prod(1 + R_b_down) - 1) + Capture ratios compare the geometric mean portfolio and benchmark returns + within each regime. Results use ratio units, where 1.0 means 100% capture: + Up Capture = GeoMean(R_p_up) / GeoMean(R_b_up) + Down Capture = GeoMean(R_p_down) / GeoMean(R_b_down) Capture Ratio = Up Capture / Down Capture. """ returns, normalized_frequency = _align_curves_and_returns( @@ -272,45 +249,10 @@ def calculate_capture_ratios( "observations": observations, "frequency": normalized_frequency, } - if observations == 0: - return result - - up_returns = returns.loc[returns["benchmark"] > 0] - down_returns = returns.loc[returns["benchmark"] < 0] - up_periods = int(len(up_returns)) - down_periods = int(len(down_returns)) - result["upMarketPeriods"] = up_periods - result["downMarketPeriods"] = down_periods - - up_capture = None - if up_periods > 0: - bench_up_comp = float((1.0 + up_returns["benchmark"]).prod() - 1.0) - port_up_comp = float((1.0 + up_returns["portfolio"]).prod() - 1.0) - if abs(bench_up_comp) > 1e-15: - up_capture = port_up_comp / bench_up_comp - - down_capture = None - if down_periods > 0: - bench_down_comp = float((1.0 + down_returns["benchmark"]).prod() - 1.0) - port_down_comp = float((1.0 + down_returns["portfolio"]).prod() - 1.0) - if abs(bench_down_comp) > 1e-15: - down_capture = port_down_comp / bench_down_comp - - capture_ratio = None - if up_capture is not None and down_capture is not None and abs(down_capture) > 1e-15: - capture_ratio = up_capture / down_capture - - status = ( - "available" - if (up_capture is not None or down_capture is not None) - else "insufficient_regimes" - ) - result.update({ - "status": status, - "upCapture": up_capture, - "downCapture": down_capture, - "captureRatio": capture_ratio, - }) + capture = _calculate_capture_metrics(returns) + result.update(capture) + result["status"] = capture["captureStatus"] + result.pop("captureStatus", None) return result @@ -356,6 +298,11 @@ def calculate_benchmark_relative_metrics( "informationRatio": None, "classification": None, "riskFreeRate": rf, + "riskFreeRateSource": "configured" if rf else "default_zero", + "betaStatus": "insufficient_history", + "correlationStatus": "insufficient_history", + "treynorStatus": "insufficient_history", + "captureStatus": "insufficient_history", "beta": None, "alphaAnnualized": None, "treynorRatio": None, @@ -384,37 +331,7 @@ def calculate_benchmark_relative_metrics( "activeReturnAnnualized": active_annualized, }) - # Capture ratios - up_returns = returns.loc[returns["benchmark"] > 0] - down_returns = returns.loc[returns["benchmark"] < 0] - up_periods = int(len(up_returns)) - down_periods = int(len(down_returns)) - result["upMarketPeriods"] = up_periods - result["downMarketPeriods"] = down_periods - - up_capture = None - if up_periods > 0: - bench_up_comp = float((1.0 + up_returns["benchmark"]).prod() - 1.0) - port_up_comp = float((1.0 + up_returns["portfolio"]).prod() - 1.0) - if abs(bench_up_comp) > 1e-15: - up_capture = port_up_comp / bench_up_comp - - down_capture = None - if down_periods > 0: - bench_down_comp = float((1.0 + down_returns["benchmark"]).prod() - 1.0) - port_down_comp = float((1.0 + down_returns["portfolio"]).prod() - 1.0) - if abs(bench_down_comp) > 1e-15: - down_capture = port_down_comp / bench_down_comp - - capture_ratio = None - if up_capture is not None and down_capture is not None and abs(down_capture) > 1e-15: - capture_ratio = up_capture / down_capture - - result.update({ - "upCapture": up_capture, - "downCapture": down_capture, - "captureRatio": capture_ratio, - }) + result.update(_calculate_capture_metrics(returns)) if observations < 2: return result @@ -424,40 +341,16 @@ def calculate_benchmark_relative_metrics( tracking_error_annualized = periodic_tracking_error * math.sqrt(factor) result["trackingErrorAnnualized"] = tracking_error_annualized - bench_var = float(returns["benchmark"].var(ddof=1)) - port_var = float(returns["portfolio"].var(ddof=1)) - covar = float(returns["portfolio"].cov(returns["benchmark"])) - - if bench_var > 1e-15: - beta = covar / bench_var - alpha_annualized = (portfolio_annualized - rf) - beta * (benchmark_annualized - rf) - treynor_ratio = (portfolio_annualized - rf) / beta if abs(beta) > 1e-15 else None - result.update({ - "beta": beta, - "alphaAnnualized": alpha_annualized, - "treynorRatio": treynor_ratio, - }) - - bench_std = math.sqrt(bench_var) - port_std = math.sqrt(port_var) - if port_std > 1e-15 and bench_std > 1e-15: - raw_corr = covar / (port_std * bench_std) - corr = max(-1.0, min(1.0, raw_corr)) - r_squared = max(0.0, min(1.0, corr**2)) - result["correlation"] = corr - result["rSquared"] = r_squared - elif port_std <= 1e-15 and bench_std > 1e-15: - result["correlation"] = 0.0 - result["rSquared"] = 0.0 + result.update(_calculate_systematic_risk_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" return result - if bench_var <= 1e-15: - result["status"] = "zero_benchmark_variance" - return result - information_ratio = active_annualized / tracking_error_annualized result.update({ "status": "available", @@ -467,6 +360,118 @@ def calculate_benchmark_relative_metrics( return result +def _calculate_systematic_risk_metrics( + returns: pd.DataFrame, + *, + annualization_factor: float, + risk_free_rate: float, +) -> dict[str, Any]: + result: dict[str, Any] = { + "betaStatus": "insufficient_history", + "correlationStatus": "insufficient_history", + "treynorStatus": "insufficient_history", + "beta": None, + "alphaAnnualized": None, + "treynorRatio": None, + "correlation": None, + "rSquared": 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))) + covar = float(returns["portfolio"].cov(returns["benchmark"])) + if bench_var <= 1e-15: + result.update({ + "betaStatus": "zero_benchmark_variance", + "correlationStatus": "undefined_constant_benchmark", + "treynorStatus": "zero_benchmark_variance", + }) + return result + + portfolio_annualized = float(returns["portfolio"].mean()) * annualization_factor + benchmark_annualized = float(returns["benchmark"].mean()) * annualization_factor + beta = covar / bench_var + alpha_annualized = ( + (portfolio_annualized - risk_free_rate) + - beta * (benchmark_annualized - risk_free_rate) + ) + treynor_ratio = None + treynor_status = "zero_beta" + if abs(beta) > 1e-15: + treynor_ratio = (portfolio_annualized - risk_free_rate) / beta + treynor_status = "available" + + result.update({ + "betaStatus": "available", + "treynorStatus": treynor_status, + "beta": beta, + "alphaAnnualized": alpha_annualized, + "treynorRatio": treynor_ratio, + }) + if port_var <= 1e-15: + result["correlationStatus"] = "undefined_constant_portfolio" + return result + + raw_corr = covar / math.sqrt(port_var * bench_var) + correlation = max(-1.0, min(1.0, raw_corr)) + result.update({ + "correlationStatus": "available", + "correlation": correlation, + "rSquared": max(0.0, min(1.0, correlation**2)), + }) + return result + + +def _calculate_capture_metrics(returns: pd.DataFrame) -> dict[str, Any]: + result: dict[str, Any] = { + "captureStatus": "insufficient_history", + "upCapture": None, + "downCapture": None, + "captureRatio": None, + "upMarketPeriods": 0, + "downMarketPeriods": 0, + } + if returns.empty: + return result + + up_returns = returns.loc[returns["benchmark"] > 0] + down_returns = returns.loc[returns["benchmark"] < 0] + result["upMarketPeriods"] = int(len(up_returns.index)) + result["downMarketPeriods"] = int(len(down_returns.index)) + up_capture = _capture_ratio(up_returns) + down_capture = _capture_ratio(down_returns) + capture_ratio = None + if up_capture is not None and down_capture is not None and abs(down_capture) > 1e-15: + capture_ratio = up_capture / down_capture + + if up_capture is not None and down_capture is not None: + status = "available" + elif up_capture is not None or down_capture is not None: + status = "partial" + else: + status = "insufficient_regimes" + result.update({ + "captureStatus": status, + "upCapture": up_capture, + "downCapture": down_capture, + "captureRatio": capture_ratio, + }) + return result + + +def _capture_ratio(regime_returns: pd.DataFrame) -> float | None: + if regime_returns.empty: + return None + observations = len(regime_returns.index) + benchmark_return = float((1.0 + regime_returns["benchmark"]).prod()) ** (1.0 / observations) - 1.0 + if abs(benchmark_return) <= 1e-15: + return None + portfolio_return = float((1.0 + regime_returns["portfolio"]).prod()) ** (1.0 / observations) - 1.0 + return portfolio_return / benchmark_return + + def benchmark_level_curve(frame: pd.DataFrame | None) -> list[dict[str, Any]]: """Return valid native-frequency benchmark levels without forward filling.""" if frame is None or frame.empty or "close" not in frame.columns: diff --git a/backend_api_python/tests/test_backtest_metrics.py b/backend_api_python/tests/test_backtest_metrics.py index 81e033b56..c1b5c8038 100644 --- a/backend_api_python/tests/test_backtest_metrics.py +++ b/backend_api_python/tests/test_backtest_metrics.py @@ -305,8 +305,10 @@ def test_calculate_beta_and_alpha_known_values(): assert result["beta"] == pytest.approx(2.0) assert result["alphaAnnualized"] == pytest.approx(0.0) assert result["treynorRatio"] == pytest.approx(2.52) + assert result["treynorStatus"] == "available" assert result["correlation"] == pytest.approx(1.0) assert result["rSquared"] == pytest.approx(1.0) + assert result["correlationStatus"] == "available" def test_calculate_beta_and_alpha_with_positive_risk_free_rate(): @@ -342,15 +344,10 @@ def test_calculate_capture_ratios_known_values(): assert result["observations"] == 4 assert result["upMarketPeriods"] == 2 assert result["downMarketPeriods"] == 2 - # Up market: benchmark = [0.05, 0.03], port = [0.06, 0.04] - # bench_up = 1.05 * 1.03 - 1 = 0.0815 - # port_up = 1.06 * 1.04 - 1 = 0.1024 - expected_up_capture = 0.1024 / 0.0815 + # Capture uses geometric-mean returns within each market regime. + expected_up_capture = ((1.06 * 1.04) ** 0.5 - 1.0) / ((1.05 * 1.03) ** 0.5 - 1.0) assert result["upCapture"] == pytest.approx(expected_up_capture) - # Down market: benchmark = [-0.04, -0.02], port = [-0.02, -0.01] - # bench_down = 0.96 * 0.98 - 1 = -0.0592 - # port_down = 0.98 * 0.99 - 1 = -0.0298 - expected_down_capture = -0.0298 / -0.0592 + expected_down_capture = ((0.98 * 0.99) ** 0.5 - 1.0) / ((0.96 * 0.98) ** 0.5 - 1.0) assert result["downCapture"] == pytest.approx(expected_down_capture) assert result["captureRatio"] == pytest.approx(expected_up_capture / expected_down_capture) @@ -363,7 +360,7 @@ def test_capture_ratios_only_up_market(): market="USStock", ) - assert result["status"] == "available" + assert result["status"] == "partial" assert result["upMarketPeriods"] == 3 assert result["downMarketPeriods"] == 0 assert result["upCapture"] is not None @@ -379,7 +376,7 @@ def test_capture_ratios_only_down_market(): market="USStock", ) - assert result["status"] == "available" + assert result["status"] == "partial" assert result["upMarketPeriods"] == 0 assert result["downMarketPeriods"] == 3 assert result["upCapture"] is None @@ -397,6 +394,8 @@ def test_zero_benchmark_variance(): ) assert result["status"] == "zero_benchmark_variance" + assert result["correlationStatus"] == "undefined_constant_benchmark" + assert result["treynorStatus"] == "zero_benchmark_variance" assert result["beta"] is None assert result["alphaAnnualized"] is None assert result["treynorRatio"] is None @@ -414,6 +413,52 @@ def test_zero_beta_treynor_is_none(): assert result["status"] == "available" assert result["beta"] == pytest.approx(0.0, abs=1e-12) assert result["treynorRatio"] is None + assert result["treynorStatus"] == "zero_beta" + + +def test_constant_portfolio_leaves_correlation_undefined(): + result = calculate_beta_and_alpha( + _curve([0.0, 0.0, 0.0]), + _curve([0.01, -0.02, 0.03]), + frequency="1d", + annualization_factor=252.0, + market="Crypto", + ) + + assert result["status"] == "available" + assert result["beta"] == pytest.approx(0.0) + assert result["correlationStatus"] == "undefined_constant_portfolio" + assert result["correlation"] is None + assert result["rSquared"] is None + + +def test_flat_benchmark_does_not_suppress_information_ratio(): + portfolio = _curve([0.01, -0.02, 0.03]) + benchmark = _curve([0.0, 0.0, 0.0]) + + information = calculate_information_ratio( + portfolio, + benchmark, + benchmark="flat", + frequency="1d", + annualization_factor=365.25, + market="Crypto", + ) + combined = calculate_benchmark_relative_metrics( + portfolio, + benchmark, + benchmark="flat", + frequency="1d", + annualization_factor=365.25, + market="Crypto", + ) + + assert combined["status"] == information["status"] == "available" + assert combined["informationRatio"] == pytest.approx(information["informationRatio"]) + assert combined["classification"] == information["classification"] + assert combined["betaStatus"] == "zero_benchmark_variance" + assert combined["correlationStatus"] == "undefined_constant_benchmark" + assert combined["beta"] is None def test_benchmark_metrics_insufficient_history(): @@ -429,6 +474,7 @@ def test_benchmark_metrics_insufficient_history(): assert result_zero["observations"] == 0 assert result_zero["portfolioReturnAnnualized"] is None assert result_zero["beta"] is None + assert result_zero["riskFreeRateSource"] == "default_zero" result_one = calculate_benchmark_relative_metrics( [{"time": "2026-01-01T00:00:00Z", "value": 100.0}, {"time": "2026-01-02T00:00:00Z", "value": 102.0}], @@ -488,6 +534,11 @@ def test_calculate_benchmark_relative_metrics_full_integration(): assert result["informationRatio"] == pytest.approx(8.270196225621152) assert result["classification"] == "exceptional" assert result["riskFreeRate"] == pytest.approx(0.02) + assert result["riskFreeRateSource"] == "configured" + assert result["betaStatus"] == "available" + assert result["correlationStatus"] == "available" + assert result["treynorStatus"] == "available" + assert result["captureStatus"] == "available" assert result["beta"] == pytest.approx(1.991525423728826) assert result["alphaAnnualized"] == pytest.approx(0.08923728813558407) assert result["treynorRatio"] == pytest.approx(0.7808085106382866) @@ -495,9 +546,14 @@ def test_calculate_benchmark_relative_metrics_full_integration(): assert result["rSquared"] == pytest.approx(0.9984180790960459) assert result["upMarketPeriods"] == 3 assert result["downMarketPeriods"] == 1 - assert result["upCapture"] == pytest.approx(2.0697185613667264) + expected_up_capture = ( + (1.01 * 1.02 * 1.005) ** (1.0 / 3.0) - 1.0 + ) / ( + (1.005 * 1.01 * 1.002) ** (1.0 / 3.0) - 1.0 + ) + assert result["upCapture"] == pytest.approx(expected_up_capture) assert result["downCapture"] == pytest.approx(2.0) - assert result["captureRatio"] == pytest.approx(1.0348592806833516) + assert result["captureRatio"] == pytest.approx(expected_up_capture / 2.0) def test_calculate_benchmark_relative_metrics_zero_tracking_error(): diff --git a/docs/product/BACKTEST_CENTER.md b/docs/product/BACKTEST_CENTER.md index 959ba470f..f2461f1e1 100644 --- a/docs/product/BACKTEST_CENTER.md +++ b/docs/product/BACKTEST_CENTER.md @@ -51,14 +51,22 @@ The calculation accepts alternative bands for mandates that require them. In addition to the Information Ratio, `benchmarkRelativeMetrics` provides: - **Beta ($\beta$):** Systematic risk relative to the benchmark ($\text{Cov}(R_p, R_b) / \text{Var}(R_b)$). Zero benchmark variance yields `zero_benchmark_variance` status. -- **Jensen's Alpha ($\alpha_{\text{ann}}$):** Annualized risk-adjusted excess return over CAPM expected return: $(R_{p,\text{ann}} - R_f) - \beta (R_{b,\text{ann}} - R_f)$. +- **Jensen's Alpha ($\alpha_{\text{ann}}$):** Annualized risk-adjusted excess return over CAPM expected return: $(R_{p,\text{ann}} - R_f) - \beta (R_{b,\text{ann}} - R_f)$. Strategy API V2 currently uses an explicit zero annual risk-free rate and returns that assumption in `riskFreeRate`. - **Treynor Ratio:** Annualized excess return per unit of systematic risk: $(R_{p,\text{ann}} - R_f) / \beta$. - **Correlation & $R^2$:** Pearson correlation coefficient ($r \in [-1, 1]$) and coefficient of determination ($R^2 \in [0, 1]$). - **Capture Ratios:** - - `upCapture`: Compounded portfolio return divided by compounded benchmark return across periods where $R_b > 0$. - - `downCapture`: Compounded portfolio return divided by compounded benchmark return across periods where $R_b < 0$. + - `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. +The top-level `status` remains the Information Ratio status for compatibility. +`betaStatus`, `correlationStatus`, `treynorStatus`, and `captureStatus` +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 +correlation is undefined in that case. Capture values use ratio units: `1.0` +means 100% capture. + Benchmark choice remains part of the research hypothesis. CDI can be suitable for Brazilian cash-like and low-duration fixed-income strategies, but it is not an automatic default for inflation-linked, longer-duration, or credit-risk diff --git a/docs/product/BACKTEST_CENTER_CN.md b/docs/product/BACKTEST_CENTER_CN.md index 9e65a9f68..7e5515bd9 100644 --- a/docs/product/BACKTEST_CENTER_CN.md +++ b/docs/product/BACKTEST_CENTER_CN.md @@ -23,6 +23,29 @@ 5. **基准比较**：同时看绝对收益和相对基准表现。 6. **稳健性**：更换时间区间、参数和市场状态，避免只保留最好的一次结果。 +### 基准相对指标 + +有可用基准数据时，Strategy API V2 会返回 `benchmarkRelativeMetrics`。 +计算会先按基准的原生时间点对齐策略权益曲线，再计算周期收益。这样在长区间 +日内回测使用较低频基准时，策略与基准使用相同的收益区间和年化因子，也不会 +因为向前填充价格而制造虚假的跟踪误差。连续市场的数据缺口和股票市场不连续 +交易日会被排除。 + +结果包括： + +- **信息比率**：年化主动收益除以年化跟踪误差。 +- **Beta**：策略相对基准的系统性风险；基准收益方差为零时不可计算。 +- **Jensen Alpha**：相对 CAPM 预期收益的年化风险调整超额收益。Strategy API V2 当前明确使用年化无风险利率 0，并通过 `riskFreeRate` 返回该假设。 +- **Treynor Ratio**：每单位系统性风险对应的年化超额收益。 +- **相关系数和 $R^2$**：策略与基准收益的 Pearson 相关性及其解释比例。任一收益序列恒定时，这两个值保持为空，因为此时相关系数没有数学定义。 +- **上涨/下跌捕获率**：分别在基准上涨和下跌周期内，用策略几何平均收益除以基准几何平均收益；`1.0` 表示 100% 捕获。 +- **捕获比率**：上涨捕获率除以下跌捕获率。 + +为保持兼容，顶层 `status` 继续表示信息比率状态；`betaStatus`、 +`correlationStatus`、`treynorStatus` 和 `captureStatus` 分别说明对应指标是否 +可用。基准收益恒定只会使 Beta 不可用，不会清空仍然可以计算的信息比率。 +读取指标前应先检查相应状态和有效观察数。 + ## 常见误判 - 零成交可能来自数据缺失、预热不足或条件无法满足。