diff --git a/backend_api_python/app/services/backtest/__init__.py b/backend_api_python/app/services/backtest/__init__.py new file mode 100644 index 000000000..e355aa54d --- /dev/null +++ b/backend_api_python/app/services/backtest/__init__.py @@ -0,0 +1,2 @@ +"""Reusable backtest calculations and report helpers.""" + diff --git a/backend_api_python/app/services/backtest/metrics.py b/backend_api_python/app/services/backtest/metrics.py new file mode 100644 index 000000000..e6198d702 --- /dev/null +++ b/backend_api_python/app/services/backtest/metrics.py @@ -0,0 +1,140 @@ +"""Benchmark-relative performance metrics for backtest result curves.""" + +from __future__ import annotations + +import math +from collections.abc import Iterable, Sequence +from typing import Any + +import pandas as pd + + +DEFAULT_INFORMATION_RATIO_BANDS: tuple[tuple[float, str], ...] = ( + (0.30, "weak"), + (0.50, "acceptable"), + (1.00, "good"), + (math.inf, "exceptional"), +) + + +def calculate_information_ratio( + portfolio_curve: Iterable[dict[str, Any]], + benchmark_curve: Iterable[dict[str, Any]], + *, + benchmark: str | None, + frequency: str, + annualization_factor: float, + classification_bands: Sequence[tuple[float, str]] = DEFAULT_INFORMATION_RATIO_BANDS, +) -> dict[str, Any]: + """Calculate an Information Ratio from timestamped portfolio and benchmark curves. + + Curves contain level observations with ``time`` and ``value`` keys. Returns + are paired only when both curves cover the exact same start and end + timestamps, preventing a missing observation from comparing intervals with + different lengths. Annualized returns use arithmetic periodic means so the + active-return numerator is consistent with the tracking-error denominator. + Tracking error uses sample standard deviation (``ddof=1``). + """ + 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) + portfolio_returns = _periodic_returns(portfolio_curve, "portfolioReturn") + benchmark_returns = _periodic_returns(benchmark_curve, "benchmarkReturn") + aligned = portfolio_returns.join(benchmark_returns, how="inner") + if not aligned.empty: + aligned = aligned.loc[aligned["portfolioStart"] == aligned["benchmarkStart"]] + + observations = int(len(aligned.index)) + result = { + "status": "insufficient_history", + "portfolioReturnAnnualized": None, + "benchmarkReturnAnnualized": None, + "activeReturnAnnualized": None, + "trackingErrorAnnualized": None, + "informationRatio": None, + "classification": None, + "benchmark": benchmark, + "observations": observations, + "frequency": str(frequency), + "annualizationFactor": factor, + } + if observations == 0: + return result + + active_returns = aligned["portfolioReturn"] - aligned["benchmarkReturn"] + portfolio_annualized = float(aligned["portfolioReturn"].mean()) * factor + benchmark_annualized = float(aligned["benchmarkReturn"].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 _periodic_returns(curve: Iterable[dict[str, Any]], prefix: str) -> pd.DataFrame: + rows = [] + for point in curve: + timestamp = pd.to_datetime(point.get("time"), errors="coerce", utc=True) + value = pd.to_numeric(point.get("value"), errors="coerce") + if pd.isna(timestamp) or pd.isna(value): + continue + numeric_value = float(value) + if not math.isfinite(numeric_value) or numeric_value <= 0: + continue + rows.append((timestamp, numeric_value)) + + if len(rows) < 2: + return pd.DataFrame(columns=[prefix, f"{prefix.removesuffix('Return')}Start"]) + + levels = pd.Series( + (value for _, value in rows), + index=pd.DatetimeIndex(timestamp for timestamp, _ in rows), + dtype="float64", + ) + levels = levels.loc[~levels.index.duplicated(keep="last")].sort_index() + starts = pd.Series(levels.index, index=levels.index).shift(1) + returns = levels.pct_change(fill_method=None) + start_column = f"{prefix.removesuffix('Return')}Start" + return pd.DataFrame({prefix: returns, start_column: starts}).dropna() + + +def _validate_classification_bands( + classification_bands: Sequence[tuple[float, str]], +) -> tuple[tuple[float, str], ...]: + bands = tuple((float(limit), str(label)) for limit, label in classification_bands) + if not bands or any(not label for _, label in bands): + raise ValueError("classification_bands must contain labelled upper bounds") + if any(current <= previous for (previous, _), (current, _) in zip(bands, bands[1:])): + raise ValueError("classification band upper bounds must be strictly increasing") + return bands + + +def _classify_information_ratio( + value: float, + bands: Sequence[tuple[float, str]], +) -> str: + for upper_bound, label in bands: + if value < upper_bound: + return label + return bands[-1][1] + diff --git a/backend_api_python/app/services/strategy_v2/service.py b/backend_api_python/app/services/strategy_v2/service.py index 0d24870ce..5fa77a0a3 100644 --- a/backend_api_python/app/services/strategy_v2/service.py +++ b/backend_api_python/app/services/strategy_v2/service.py @@ -13,6 +13,7 @@ import pandas as pd from app.data_sources.errors import MarketDataUnavailableError +from app.services.backtest.metrics import calculate_information_ratio from app.services.backtest_limits import ( BacktestRangeLimitError, backtest_warmup_calendar_days, @@ -276,6 +277,13 @@ def resolve_universe(reference: str, timestamp: pd.Timestamp) -> list[str]: ) result.update(benchmark) result["excessReturn"] = float(result.get("totalReturn") or 0.0) - float(result.get("benchmarkTotalReturn") or 0.0) + result["benchmarkRelativeMetrics"] = calculate_information_ratio( + result.get("equityCurve") or [], + result.get("benchmarkCurve") or [], + benchmark=benchmark_spec.key if benchmark_spec is not None else None, + frequency=manifest.driving_frequency, + annualization_factor=float(result.get("periodsPerYear") or 1.0), + ) timeframe_provenance = { item: [ _frame_provenance( diff --git a/backend_api_python/tests/test_backtest_metrics.py b/backend_api_python/tests/test_backtest_metrics.py new file mode 100644 index 000000000..b8260d944 --- /dev/null +++ b/backend_api_python/tests/test_backtest_metrics.py @@ -0,0 +1,113 @@ +import math + +import pytest + +from app.services.backtest.metrics import calculate_information_ratio + + +def _curve(returns, *, missing_index=None): + value = 100.0 + points = [{"time": "2026-01-01T00:00:00Z", "value": value}] + for index, periodic_return in enumerate(returns, start=1): + value *= 1.0 + periodic_return + if index != missing_index: + points.append({ + "time": f"2026-01-{index + 1:02d}T00:00:00Z", + "value": value, + }) + return points + + +def test_information_ratio_uses_aligned_periodic_returns_and_sample_tracking_error(): + result = calculate_information_ratio( + _curve([0.01, 0.02, -0.01, 0.005]), + _curve([0.005, 0.01, -0.005, 0.002]), + benchmark="USStock:SPY", + frequency="1d", + annualization_factor=252, + ) + + active_returns = [0.005, 0.01, -0.005, 0.003] + active_mean = sum(active_returns) / len(active_returns) + tracking_error = pytest.approx(0.006238322424070967 * math.sqrt(252)) + + assert result["status"] == "available" + assert result["portfolioReturnAnnualized"] == pytest.approx(1.575) + assert result["benchmarkReturnAnnualized"] == pytest.approx(0.756) + assert result["activeReturnAnnualized"] == pytest.approx(active_mean * 252) + assert result["trackingErrorAnnualized"] == tracking_error + assert result["informationRatio"] == pytest.approx(8.270196225621152) + assert result["classification"] == "exceptional" + assert result["benchmark"] == "USStock:SPY" + assert result["observations"] == 4 + assert result["frequency"] == "1d" + assert result["annualizationFactor"] == 252 + + +def test_information_ratio_drops_dates_missing_from_either_curve(): + result = calculate_information_ratio( + _curve([0.01, 0.02, -0.01, 0.005]), + _curve([0.005, 0.01, -0.005, 0.002], missing_index=2), + benchmark="USStock:SPY", + frequency="1d", + annualization_factor=252, + ) + + assert result["status"] == "available" + assert result["observations"] == 2 + + +def test_information_ratio_handles_zero_tracking_error_explicitly(): + result = calculate_information_ratio( + _curve([0.01, 0.02, 0.03]), + _curve([0.005, 0.015, 0.025]), + benchmark="USStock:SPY", + frequency="1d", + annualization_factor=252, + ) + + assert result["status"] == "zero_tracking_error" + assert result["trackingErrorAnnualized"] == pytest.approx(0.0, abs=1e-12) + assert result["informationRatio"] is None + assert result["classification"] is None + + +def test_information_ratio_reports_insufficient_aligned_history(): + result = calculate_information_ratio( + _curve([0.01]), + _curve([0.005]), + benchmark="USStock:SPY", + frequency="1d", + annualization_factor=252, + ) + + assert result["status"] == "insufficient_history" + assert result["observations"] == 1 + assert result["informationRatio"] is None + + +def test_information_ratio_preserves_negative_values(): + result = calculate_information_ratio( + _curve([-0.01, -0.02, 0.005, -0.01]), + _curve([0.005, -0.005, 0.01, 0.002]), + benchmark="USStock:SPY", + frequency="1d", + annualization_factor=252, + ) + + assert result["status"] == "available" + assert result["informationRatio"] < 0 + assert result["classification"] == "weak" + + +def test_information_ratio_accepts_configurable_interpretation_bands(): + result = calculate_information_ratio( + _curve([0.01, 0.02, -0.01, 0.005]), + _curve([0.005, 0.01, -0.005, 0.002]), + benchmark="custom", + frequency="1d", + annualization_factor=252, + classification_bands=((10.0, "ordinary"), (math.inf, "excellent")), + ) + + assert result["classification"] == "ordinary" diff --git a/backend_api_python/tests/test_strategy_v2_service.py b/backend_api_python/tests/test_strategy_v2_service.py index e15e9b402..265f0d1b0 100644 --- a/backend_api_python/tests/test_strategy_v2_service.py +++ b/backend_api_python/tests/test_strategy_v2_service.py @@ -277,6 +277,11 @@ def handle_data(context, data): assert result["diagnostics"]["sourceControlled"] is True assert result["benchmarkStatus"] == "available" assert len(result["benchmarkCurve"]) == len(result["equityCurve"]) + assert result["benchmarkRelativeMetrics"]["status"] == "available" + assert result["benchmarkRelativeMetrics"]["benchmark"] == "USStock:AAPL" + assert result["benchmarkRelativeMetrics"]["frequency"] == "1d" + assert result["benchmarkRelativeMetrics"]["annualizationFactor"] == result["periodsPerYear"] + assert result["benchmarkRelativeMetrics"]["observations"] == len(result["equityCurve"]) - 1 assert all(point["time"].endswith("Z") for point in result["benchmarkCurve"]) assert result["dataProvenance"]["kind"] == "market" assert result["audit"]["passed"] is True diff --git a/docs/product/BACKTEST_CENTER.md b/docs/product/BACKTEST_CENTER.md index 67ea89e4a..2b1c4e01c 100644 --- a/docs/product/BACKTEST_CENTER.md +++ b/docs/product/BACKTEST_CENTER.md @@ -23,6 +23,38 @@ Review in this order: 5. **Benchmark:** compare both absolute return and relative performance. 6. **Robustness:** vary ranges, parameters, and regimes instead of keeping only the best run. +### Benchmark-relative metrics + +When benchmark data is available, Strategy API V2 includes +`benchmarkRelativeMetrics`. The calculation first aligns portfolio and +benchmark observations to identical return intervals, then reports arithmetic +annualized portfolio, benchmark, and active returns. Annualized tracking error +uses the sample standard deviation of periodic active returns; the Information +Ratio is annualized active return divided by annualized tracking error. The +annualization factor is the same one used by the Strategy V2 backtest: 252 +trading days for non-crypto markets and 365.25 days for crypto, with intraday +frequencies expanded using the corresponding session length; weekly data uses +52 periods per year. + +Check `status` before reading the ratio. `insufficient_history` means fewer than +two aligned return observations were available. An interval is retained only +when both curves have valid positive levels at its exact start and end times, +so a missing endpoint discards that interval rather than stretching it across +multiple periods. `zero_tracking_error` leaves +the ratio and classification unset rather than emitting infinity. Missing or +non-finite observations are excluded, and negative ratios are preserved. + +The default interpretation bands (`weak`, `acceptable`, `good`, and +`exceptional`) are operational labels, not a universal market standard. The +calculation accepts alternative bands when a research mandate requires them. + +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 +mandates. Use an appropriate aligned benchmark series such as an IRF-M, IMA-B, +or credit index where the mandate calls for it; do not substitute a static +annual CDI rate for periodic benchmark observations. + ## Common misreadings - Zero executions may mean missing data, insufficient warmup, or unreachable conditions.