diff --git a/backend_api_python/app/services/backtest/metrics.py b/backend_api_python/app/services/backtest/metrics.py index f4025175c..60be10d32 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,205 @@ 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: dict[str, Any] = { + "status": "insufficient_history", + "portfolioReturnAnnualized": None, + "benchmarkReturnAnnualized": None, + "activeReturnAnnualized": None, + "trackingErrorAnnualized": None, + "informationRatio": None, + "classification": None, + "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, + }) + if observations < 2: + return result + + periodic_tracking_error = float(active_returns.std(ddof=1)) + tracking_error_annualized = periodic_tracking_error * math.sqrt(factor) + result["trackingErrorAnnualized"] = tracking_error_annualized + if math.isclose(periodic_tracking_error, 0.0, rel_tol=0.0, abs_tol=1e-15): + result["status"] = "zero_tracking_error" + 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 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 = { + result: dict[str, Any] = { + "status": "insufficient_history", + "correlationStatus": "insufficient_history", + "treynorStatus": "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 + + 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 + + +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 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( + 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, + } + capture = _calculate_capture_metrics(returns) + result.update(capture) + result["status"] = capture["captureStatus"] + result.pop("captureStatus", None) + 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, @@ -82,6 +297,22 @@ def calculate_information_ratio( "trackingErrorAnnualized": None, "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, + "correlation": None, + "rSquared": None, + "upCapture": None, + "downCapture": None, + "captureRatio": None, + "upMarketPeriods": 0, + "downMarketPeriods": 0, "benchmark": benchmark, "observations": observations, "frequency": normalized_frequency, @@ -99,12 +330,23 @@ def calculate_information_ratio( "benchmarkReturnAnnualized": benchmark_annualized, "activeReturnAnnualized": active_annualized, }) + + result.update(_calculate_capture_metrics(returns)) + 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 + + 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 @@ -118,6 +360,118 @@ def calculate_information_ratio( 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/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..c1b5c8038 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,298 @@ 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["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(): + 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 + # 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) + 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) + + +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"] == "partial" + 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"] == "partial" + 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["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 + + +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 + 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(): + 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 + 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}], + [{"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["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) + assert result["correlation"] == pytest.approx(0.99920872649114) + assert result["rSquared"] == pytest.approx(0.9984180790960459) + assert result["upMarketPeriods"] == 3 + assert result["downMarketPeriods"] == 1 + 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(expected_up_capture / 2.0) + + +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..f2461f1e1 100644 --- a/docs/product/BACKTEST_CENTER.md +++ b/docs/product/BACKTEST_CENTER.md @@ -48,6 +48,25 @@ 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)$. 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`: 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 不可用，不会清空仍然可以计算的信息比率。 +读取指标前应先检查相应状态和有效观察数。 + ## 常见误判 - 零成交可能来自数据缺失、预热不足或条件无法满足。