feat(backtest): add benchmark-relative information ratio metrics

backend_api_python/app/services/backtest/__init__.py

@@ -0,0 +1,2 @@
+"""Reusable backtest calculations and report helpers."""
+

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]
+

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(

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"

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

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.