docs: document unified backtest result workflow

.gitea/workflows/ci.yml

@@ -0,0 +1,32 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+concurrency:
+  group: ci-${{ github.ref }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  lite:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@524e936cd9e579adf00e308bfdf971aebc7de09e
+        with:
+          persist-credentials: false
+      - name: Gitea 合规校验
+        run: |
+          if git ls-tree -r HEAD | awk '$1==160000' | grep -q .; then echo "HEAD 含 gitlink"; exit 1; fi
+          if git ls-files .DS_Store | grep -q .; then echo "跟踪 .DS_Store"; exit 1; fi
+          if git grep -n -I -E 'sk-[A-Za-z0-9]{20,}|AKIA[0-9A-Z]{16}|ghp_[A-Za-z0-9]{36}|xox[baprs]-[A-Za-z0-9-]{10,}' HEAD | grep -q .; then echo "检出疑似凭证"; exit 1; fi
+          echo "Gitea 合规校验通过"
+      - name: 架构模块契约测试
+        run: python3 tests/governance/test_module_spec.py
+      - name: Syntax check
+        run: git ls-files -z '*.py' | xargs -0 python3 -m py_compile

MODULE_SPEC.yaml

@@ -0,0 +1,41 @@
+{
+  "schema_version": 1,
+  "module_id": "quant_engine",
+  "authority": {"scope": "module_metadata", "subject": "quant_engine", "owner": "quant-engine-owner", "source": "MODULE_SPEC.yaml", "revision": 1, "effective_from": "2026-08-20T00:00:00+08:00"},
+  "repository": {"name": "quant_engine", "workspace_id": "researchhub", "type": "research_engine", "maturity": "operational"},
+  "bounded_context": {
+    "domain": "quantitative-research-engine",
+    "responsibility": "Own reusable quantitative research calculations for factors, indicators, execution simulation, backtesting, portfolio decomposition, risk, and performance metrics.",
+    "prohibited_responsibilities": [
+      "Presenting research calculations, simulations, backtests, or metrics as investment advice or guaranteed performance",
+      "Submitting live orders, routing trades, managing brokerage accounts, or claiming transaction execution",
+      "Owning market-data source facts, research-result publication, or platform presentation state",
+      "Loading provider credentials, brokerage credentials, or production secrets",
+      "Changing financial model semantics through module metadata"
+    ]
+  },
+  "capabilities": [
+    {"id": "factor-and-indicator-calculation", "summary": "Calculate reusable alpha factors and technical indicators from caller-supplied data.", "status": "operational"},
+    {"id": "execution-simulation", "summary": "Simulate costs, slippage, market constraints, fills, NAV, and PnL without live order routing.", "status": "operational"},
+    {"id": "portfolio-backtesting", "summary": "Run weight-based backtests and benchmark comparisons.", "status": "operational"},
+    {"id": "risk-and-performance-analysis", "summary": "Calculate portfolio decomposition, risk contribution, and performance statistics.", "status": "operational"}
+  ],
+  "data": {"owns": [
+    {"asset_id": "quantitative-model-implementations", "kind": "model", "classification": "internal"},
+    {"asset_id": "simulation-and-metric-results", "kind": "artifact", "classification": "confidential"}
+  ]},
+  "contracts": {"provides": [], "consumes": []},
+  "dependencies": [],
+  "agent_context": {
+    "default_entrypoints": [
+      {"path": "README.md", "purpose": "Current engine scope, modules, dependencies, and non-live usage examples."},
+      {"path": "pyproject.toml", "purpose": "Supported Python version, package version, dependencies, and quality configuration."}
+    ],
+    "excluded_paths": ["graphify-out", ".venv", "build", "dist"],
+    "max_default_tokens": 8000
+  },
+  "verification": {"commands": [
+    {"id": "module-metadata", "argv": ["python3", "tests/governance/test_module_spec.py"], "cwd": ".", "network": false, "required": true},
+    {"id": "unit-tests", "argv": ["python3", "-m", "pytest"], "cwd": ".", "network": false, "required": true}
+  ]}
+}

README.md

@@ -60,6 +60,7 @@ from quant_engine.alpha_factors import alpha_001, alpha_005, ALPHA158_REGISTRY
 from quant_engine.execution import (
     ExecutionConfig, simulate_with_daily_data, compute_realized_pnl,
 )
+from quant_engine.backtest import run_weight_backtest
 from quant_engine.indicators import macd, bollinger, kdj
 from quant_engine.data_adapter import (
     long_to_wide, wide_to_long, rename_tushare_columns,
@@ -72,6 +73,16 @@ from quant_engine.data_adapter import (
 df = load_qtdb_daily(["000001.SZ"], "2024-01-01", with_adj=True)
 prices, volumes = prepare_execution_inputs(df)
 result = simulate_with_daily_data(prices, initial_cash=1_000_000.0)
+
+# 权重回测 → 稳定结果对象 → 绩效/基准分析
+backtest = run_weight_backtest(
+    weights=rebalance_weights,
+    stock_returns=daily_returns,
+    initial_capital=1_000_000.0,
+    benchmark_nav=benchmark_nav,
+)
+print(backtest.stats())
+print(backtest.benchmark_report())
 ```
 
 ## 与 research_results 的关系

src/quant_engine/backtest.py

@@ -13,29 +13,28 @@
 
 ```python
 from quant_engine.backtest import (
-    compute_nav_from_weights,    # 调仓表 → 净值
-    rebalance_table,              # 周期性再平衡
-    compare_to_benchmark,         # 策略 vs 基准
+    rebalance_periodic,           # 周期性再平衡
+    run_weight_backtest,          # 权重 → 统一结果对象
     weights_to_long_short,        # 多空组合
 )
 
-# 1. 调仓表 → 净值
-nav = compute_nav_from_weights(
+# 调仓表 → 净值、收益、绩效与基准报告
+rebalance_table = rebalance_periodic(target_weights, rebalance_dates, returns.index)
+result = run_weight_backtest(
     weights=rebalance_table,    # 每周/每月调仓
     stock_returns=returns,      # 个股日收益
     initial_capital=1.0,
+    benchmark_nav=benchmark_nav,
 )
-
-# 2. 跟基准比
-result = compare_to_benchmark(nav, benchmark_nav)
-print(result.summary())
+print(result.stats())
+print(result.benchmark_report())
 ```
 """
 
 from __future__ import annotations
 
-from pathlib import Path
 from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
 
 import numpy as np
 import pandas as pd
@@ -46,6 +45,26 @@ from quant_engine.metrics import summary as metrics_summary
 logger = get_logger(__name__)
 
 
+@dataclass(frozen=True, slots=True, eq=False)
+class BacktestResult:
+    """一次权重回测的稳定结果快照。"""
+
+    nav: pd.Series
+    returns: pd.Series
+    weights: pd.DataFrame
+    benchmark_nav: pd.Series | None = None
+
+    def stats(self, rf: float = 0.0) -> Mapping[str, float]:
+        """返回标准绩效指标。"""
+        return metrics_summary(self.returns, rf)
+
+    def benchmark_report(self, rf: float = 0.0) -> pd.DataFrame:
+        """返回策略与基准的对比报告。"""
+        if self.benchmark_nav is None:
+            raise ValueError("benchmark_nav is required for benchmark comparison")
+        return compare_to_benchmark(self.nav, self.benchmark_nav, rf)
+
+
 # ── 调仓表 → 净值 ──────────────────────────────────────
 
 
@@ -113,6 +132,29 @@ def compute_returns_from_nav(nav: pd.Series) -> pd.Series:
     return nav.pct_change().fillna(0.0)
 
 
+def run_weight_backtest(
+    weights: pd.DataFrame,
+    stock_returns: pd.DataFrame,
+    initial_capital: float = 1.0,
+    tc_rate: float = 0.0,
+    benchmark_nav: pd.Series | None = None,
+) -> BacktestResult:
+    """执行权重回测并返回隔离于调用方输入的结果快照。"""
+    weights_snapshot = weights.copy(deep=True)
+    nav = compute_nav_from_weights(
+        weights=weights_snapshot,
+        stock_returns=stock_returns,
+        initial_capital=initial_capital,
+        tc_rate=tc_rate,
+    )
+    return BacktestResult(
+        nav=nav,
+        returns=compute_returns_from_nav(nav),
+        weights=weights_snapshot,
+        benchmark_nav=None if benchmark_nav is None else benchmark_nav.copy(deep=True),
+    )
+
+
 # ── 调仓工具 ──────────────────────────────────────
 
 
@@ -131,6 +173,9 @@ def rebalance_periodic(
     Returns:
         调仓表 DataFrame（all_dates × 股票代码）
     """
+    if all_dates.empty:
+        return pd.DataFrame(index=all_dates, columns=target_weights.index, dtype=float)
+
     table = pd.DataFrame(0.0, index=all_dates, columns=target_weights.index)
     for date in rebalance_dates:
         if date not in all_dates:
@@ -201,6 +246,8 @@ def compare_to_benchmark(
     """
     # 对齐 index
     common = strategy_nav.index.intersection(benchmark_nav.index)
+    if common.empty:
+        raise ValueError("strategy and benchmark must have overlapping dates")
     s = strategy_nav.loc[common]
     b = benchmark_nav.loc[common]
 

src/quant_engine/factor_library.py

@@ -312,8 +312,8 @@ def ols_regress(
     ss_tot = float(((y_arr - y_arr.mean()) ** 2).sum())
     r_sq = 1.0 - ss_res / ss_tot if ss_tot > 0 else np.nan
     sigma2 = ss_res / max(n - k, 1)
-    # 协方差矩阵 = sigma2 * (X'X)^-1
-    xtx_inv = np.linalg.inv(x_arr.T @ x_arr) if sigma2 > 0 else np.full((k, k), np.nan)
+    # 广义协方差矩阵 = sigma2 * (X'X)^+，伪逆兼容共线因子。
+    xtx_inv = np.linalg.pinv(x_arr.T @ x_arr) if sigma2 > 0 else np.full((k, k), np.nan)
     se = np.sqrt(np.diag(xtx_inv) * sigma2)
     t_vals = coef / se if sigma2 > 0 else np.full_like(coef, np.nan)
     if add_constant:
@@ -513,6 +513,8 @@ def apply_factor_direction(
     Returns:
         方向调整后的因子（同向 = 越大越好）
     """
+    if direction not in {"auto", "forward", "reverse"}:
+        raise ValueError(f"direction={direction!r} not supported (auto / forward / reverse)")
     if factor.empty:
         return factor.copy()
     if direction == "auto":
@@ -541,6 +543,8 @@ def cross_sectional_rank_with_direction(
     Returns:
         pd.Series（百分位排名 [0, 1]，越大越优）
     """
+    if direction not in {"auto", "forward", "reverse"}:
+        raise ValueError(f"direction={direction!r} not supported (auto / forward / reverse)")
     if df.empty or factor_col not in df.columns:
         return pd.Series(dtype=float)
     factor = df[factor_col]

src/quant_engine/metrics.py

@@ -71,7 +71,8 @@ def max_drawdown(r: pd.Series) -> float:
     if len(r) < 2:
         return 0.0
     nav = (1 + r).cumprod()
-    peak = nav.cummax()
+    # 初始资金净值为 1；否则首个观测日的亏损会被误当成新的历史高点。
+    peak = nav.cummax().clip(lower=1.0)
     drawdown = (nav - peak) / peak
     return float(drawdown.min())
 

src/quant_engine/risk.py

@@ -5,9 +5,22 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 import numpy as np
 from numpy.typing import NDArray
-from typing import Any
+
+
+def _validate_inputs(weights: NDArray[Any], cov: NDArray[Any]) -> tuple[NDArray[Any], NDArray[Any]]:
+    """Normalize a portfolio vector and its covariance matrix."""
+    w = np.asarray(weights, dtype=float).ravel()
+    covariance = np.asarray(cov, dtype=float)
+    k = w.size
+    if k == 0:
+        raise ValueError("weights must contain at least one asset")
+    if covariance.shape != (k, k):
+        raise ValueError(f"cov shape {covariance.shape} does not match weights length {k}")
+    return w, covariance
 
 
 def risk_contribution(weights: NDArray[Any], cov: NDArray[Any]) -> NDArray[Any]:
@@ -25,11 +38,8 @@ def risk_contribution(weights: NDArray[Any], cov: NDArray[Any]) -> NDArray[Any]:
     Returns:
         RC: 风险贡献向量 (k,), Σ=1
     """
-    w = np.asarray(weights, dtype=float).ravel()
-    cov = np.asarray(cov, dtype=float)
+    w, cov = _validate_inputs(weights, cov)
     k = w.size
-    if cov.shape != (k, k):
-        raise ValueError(f"cov 形状 {cov.shape} 与 weights 长度 {k} 不匹配")
 
     port_var = float(w @ cov @ w)
     if port_var <= 0:
@@ -41,13 +51,11 @@ def risk_contribution(weights: NDArray[Any], cov: NDArray[Any]) -> NDArray[Any]:
 
 def marginal_risk_contribution(weights: NDArray[Any], cov: NDArray[Any]) -> NDArray[Any]:
     """边际风险贡献 (MRC_i): (Σw)_i。"""
-    w = np.asarray(weights, dtype=float).ravel()
-    cov = np.asarray(cov, dtype=float)
+    w, cov = _validate_inputs(weights, cov)
     return cov @ w  # type: ignore[no-any-return]
 
 
 def component_var(weights: NDArray[Any], cov: NDArray[Any]) -> NDArray[Any]:
     """成分方差: w_i · (Σw)_i; 与 RC 的关系 RC_i = CV_i / w'Σw。"""
-    w = np.asarray(weights, dtype=float).ravel()
-    cov = np.asarray(cov, dtype=float)
+    w, cov = _validate_inputs(weights, cov)
     return w * (cov @ w)  # type: ignore[no-any-return]

tests/governance/test_ci_contract.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+import re
+import unittest
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+class CiContractTests(unittest.TestCase):
+    def test_ci_is_one_dependency_free_lite_gate(self) -> None:
+        workflow = (ROOT / ".gitea/workflows/ci.yml").read_text(encoding="utf-8")
+        jobs = workflow.split("jobs:", 1)[1]
+        self.assertEqual(re.findall(r"(?m)^  ([a-z][a-z0-9_-]*):\s*$", jobs), ["lite"])
+        self.assertIn("actions/checkout@524e936cd9e579adf00e308bfdf971aebc7de09e", workflow)
+        self.assertIn("persist-credentials: false", workflow)
+        self.assertIn("python3 tests/governance/test_module_spec.py", workflow)
+        for forbidden in ("setup-python", "pip ", "curl ", "wget ", "docker pull"):
+            self.assertNotIn(forbidden, workflow)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/governance/test_module_spec.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+import json
+import unittest
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+class ModuleSpecTests(unittest.TestCase):
+    def test_module_spec_declares_pure_research_engine_boundary(self) -> None:
+        spec = json.loads((ROOT / "MODULE_SPEC.yaml").read_text(encoding="utf-8"))
+        self.assertEqual(spec["module_id"], "quant_engine")
+        self.assertEqual(spec["authority"]["subject"], spec["module_id"])
+        self.assertEqual(spec["repository"]["type"], "research_engine")
+        self.assertEqual(spec["bounded_context"]["domain"], "quantitative-research-engine")
+        prohibited = " ".join(spec["bounded_context"]["prohibited_responsibilities"]).lower()
+        for term in ("investment advice", "live order", "credentials", "source facts"):
+            self.assertIn(term, prohibited)
+        self.assertEqual(spec["contracts"], {"provides": [], "consumes": []})
+        self.assertEqual(spec["dependencies"], [])
+        self.assertTrue(
+            all(
+                command["required"] and not command["network"]
+                for command in spec["verification"]["commands"]
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_backtest.py

@@ -0,0 +1,209 @@
+"""Backtest contract tests for weights, NAV, rebalancing, and benchmarks."""
+
+from __future__ import annotations
+
+import pandas as pd
+import pytest
+
+from quant_engine.backtest import (
+    BacktestResult,
+    compare_to_benchmark,
+    compute_nav_from_weights,
+    compute_returns_from_nav,
+    rebalance_periodic,
+    run_weight_backtest,
+    weights_to_long_short,
+)
+
+
+def test_compute_nav_from_weights_forward_fills_rebalance_weights() -> None:
+    dates = pd.date_range("2026-01-05", periods=3, freq="B")
+    weights = pd.DataFrame({"A": [0.5], "B": [0.5]}, index=dates[:1])
+    returns = pd.DataFrame({"A": [0.10, 0.00, -0.10], "B": [0.00, 0.10, 0.00]}, index=dates)
+
+    nav = compute_nav_from_weights(weights, returns, initial_capital=100.0)
+
+    expected = pd.Series([105.0, 110.25, 104.7375], index=dates)
+    pd.testing.assert_series_equal(nav, expected)
+
+
+def test_compute_nav_stays_in_cash_before_first_rebalance() -> None:
+    dates = pd.date_range("2026-01-05", periods=3, freq="B")
+    weights = pd.DataFrame({"A": [1.0]}, index=dates[1:2])
+    returns = pd.DataFrame({"A": [0.50, 0.10, 0.10]}, index=dates)
+
+    nav = compute_nav_from_weights(weights, returns)
+
+    pd.testing.assert_series_equal(nav, pd.Series([1.0, 1.1, 1.21], index=dates))
+
+
+def test_compute_nav_ignores_weight_columns_without_returns() -> None:
+    dates = pd.date_range("2026-01-05", periods=2, freq="B")
+    weights = pd.DataFrame({"A": [0.5], "MISSING": [0.5]}, index=dates[:1])
+    returns = pd.DataFrame({"A": [0.10, 0.10]}, index=dates)
+
+    nav = compute_nav_from_weights(weights, returns)
+
+    pd.testing.assert_series_equal(nav, pd.Series([1.05, 1.1025], index=dates))
+
+
+def test_compute_nav_charges_configured_turnover_cost() -> None:
+    dates = pd.date_range("2026-01-05", periods=2, freq="B")
+    weights = pd.DataFrame({"A": [1.0]}, index=dates[:1])
+    returns = pd.DataFrame({"A": [0.0, 0.0]}, index=dates)
+
+    nav = compute_nav_from_weights(weights, returns, tc_rate=0.01)
+
+    pd.testing.assert_series_equal(nav, pd.Series([0.995, 0.995], index=dates))
+
+
+def test_compute_returns_from_nav_preserves_index_and_sets_initial_zero() -> None:
+    nav = pd.Series([100.0, 110.0, 99.0], index=pd.date_range("2026-01-05", periods=3))
+
+    result = compute_returns_from_nav(nav)
+
+    pd.testing.assert_series_equal(result, pd.Series([0.0, 0.1, -0.1], index=nav.index))
+
+
+def test_rebalance_periodic_maps_weekend_to_previous_trading_day() -> None:
+    dates = pd.date_range("2026-01-05", periods=5, freq="B")
+    target = pd.Series({"A": 0.6, "B": 0.4})
+
+    result = rebalance_periodic(target, [pd.Timestamp("2026-01-10")], dates)
+
+    assert result.loc[pd.Timestamp("2026-01-08")].sum() == 0.0
+    pd.testing.assert_series_equal(
+        result.loc[pd.Timestamp("2026-01-09")], target, check_names=False
+    )
+
+
+def test_rebalance_periodic_accepts_empty_trading_calendar() -> None:
+    target = pd.Series({"A": 1.0})
+
+    result = rebalance_periodic(
+        target,
+        [pd.Timestamp("2026-01-05")],
+        pd.DatetimeIndex([]),
+    )
+
+    assert result.empty
+    assert result.columns.tolist() == ["A"]
+
+
+def test_weights_to_long_short_allocates_each_leg() -> None:
+    result = weights_to_long_short(["A", "B"], ["C"], long_weight=0.6, short_weight=0.4)
+
+    assert result["A"] == pytest.approx(0.3)
+    assert result["B"] == pytest.approx(0.3)
+    assert result["C"] == pytest.approx(-0.4)
+    assert result.sum() == pytest.approx(0.2)
+
+
+def test_weights_to_long_short_keeps_explicit_universe() -> None:
+    result = weights_to_long_short(["A"], [], all_tickers=["A", "B"])
+
+    pd.testing.assert_series_equal(result, pd.Series({"A": 0.5, "B": 0.0}))
+
+
+def test_compare_to_benchmark_returns_report_table() -> None:
+    dates = pd.date_range("2026-01-05", periods=4, freq="B")
+    strategy = pd.Series([1.0, 1.1, 1.0, 1.2], index=dates)
+    benchmark = pd.Series([1.0, 1.0, 1.05, 1.1], index=dates)
+
+    result = compare_to_benchmark(strategy, benchmark)
+
+    assert result.columns.tolist() == ["策略", "基准"]
+    assert result.loc["n_days", "策略"] == 4
+    assert result.loc["累计收益", "策略"] == pytest.approx(0.2)
+    assert result.loc["累计收益", "基准"] == pytest.approx(0.1)
+
+
+def test_compare_to_benchmark_rejects_non_overlapping_dates() -> None:
+    strategy = pd.Series([1.0], index=[pd.Timestamp("2026-01-05")])
+    benchmark = pd.Series([1.0], index=[pd.Timestamp("2026-02-05")])
+
+    with pytest.raises(ValueError, match="overlapping dates"):
+        compare_to_benchmark(strategy, benchmark)
+
+
+# ── 统一回测结果门面 ──────────────────────────────────────
+
+
+def test_run_weight_backtest_returns_nav_returns_and_input_snapshot() -> None:
+    dates = pd.date_range("2026-01-05", periods=3, freq="B")
+    weights = pd.DataFrame({"A": [1.0]}, index=dates[:1])
+    stock_returns = pd.DataFrame({"A": [0.10, -0.10, 0.20]}, index=dates)
+
+    result = run_weight_backtest(weights, stock_returns, initial_capital=100.0)
+
+    assert isinstance(result, BacktestResult)
+    pd.testing.assert_series_equal(
+        result.nav,
+        pd.Series([110.0, 99.0, 118.8], index=dates),
+    )
+    pd.testing.assert_series_equal(
+        result.returns,
+        pd.Series([0.0, -0.1, 0.2], index=dates),
+    )
+    pd.testing.assert_frame_equal(result.weights, weights)
+
+
+def test_backtest_result_stats_reuses_standard_metrics_contract() -> None:
+    dates = pd.date_range("2026-01-05", periods=3, freq="B")
+    result = run_weight_backtest(
+        pd.DataFrame({"A": [1.0]}, index=dates[:1]),
+        pd.DataFrame({"A": [0.10, -0.10, 0.20]}, index=dates),
+    )
+
+    stats = result.stats(rf=0.02)
+
+    assert stats["n_days"] == 3
+    assert stats["ann_return"] == pytest.approx(
+        (1.0 * 0.9 * 1.2) ** (252 / 3) - 1.0
+    )
+    assert "sharpe" in stats
+    assert stats["drawback"] == stats["max_drawdown"]
+
+
+def test_backtest_result_builds_benchmark_report() -> None:
+    dates = pd.date_range("2026-01-05", periods=3, freq="B")
+    benchmark = pd.Series([1.0, 1.05, 1.10], index=dates, name="benchmark")
+    result = run_weight_backtest(
+        pd.DataFrame({"A": [1.0]}, index=dates[:1]),
+        pd.DataFrame({"A": [0.10, -0.10, 0.20]}, index=dates),
+        benchmark_nav=benchmark,
+    )
+
+    report = result.benchmark_report()
+
+    assert report.columns.tolist() == ["策略", "基准"]
+    assert report.loc["累计收益", "基准"] == pytest.approx(0.10)
+
+
+def test_backtest_result_requires_benchmark_for_comparison() -> None:
+    dates = pd.date_range("2026-01-05", periods=2, freq="B")
+    result = run_weight_backtest(
+        pd.DataFrame({"A": [1.0]}, index=dates[:1]),
+        pd.DataFrame({"A": [0.0, 0.0]}, index=dates),
+    )
+
+    with pytest.raises(ValueError, match="benchmark_nav"):
+        result.benchmark_report()
+
+
+def test_backtest_result_isolated_from_mutated_caller_inputs() -> None:
+    dates = pd.date_range("2026-01-05", periods=2, freq="B")
+    weights = pd.DataFrame({"A": [1.0]}, index=dates[:1])
+    benchmark = pd.Series([1.0, 1.1], index=dates)
+    result = run_weight_backtest(
+        weights,
+        pd.DataFrame({"A": [0.0, 0.0]}, index=dates),
+        benchmark_nav=benchmark,
+    )
+
+    weights.iloc[0, 0] = 0.0
+    benchmark.iloc[1] = 99.0
+
+    assert result.weights.iloc[0, 0] == 1.0
+    assert result.benchmark_nav is not None
+    assert result.benchmark_nav.iloc[1] == 1.1

tests/test_factor_library.py

@@ -0,0 +1,173 @@
+"""Contracts for reusable factor diagnostics and transformations."""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from quant_engine.factor_library import (
+    annualized_sharpe,
+    apply_factor_direction,
+    cross_sectional_momentum,
+    cross_sectional_pct_rank,
+    cross_sectional_rank_with_direction,
+    ic_summary,
+    jb_test,
+    kurtosis,
+    ols_regress,
+    rolling_annual_vol,
+    rolling_zscore,
+    skewness,
+    spearman_ic,
+    time_series_momentum,
+    turnover,
+    winsorize,
+)
+
+
+def test_turnover_supports_one_way_and_round_trip_conventions() -> None:
+    weights = pd.DataFrame({"A": [1.0, 0.0], "B": [0.0, 1.0]})
+
+    pd.testing.assert_series_equal(turnover(weights), pd.Series([1.0], index=[1]))
+    pd.testing.assert_series_equal(
+        turnover(weights, divide_by_two=False), pd.Series([2.0], index=[1])
+    )
+    assert turnover(weights.iloc[:1]).empty
+
+
+def test_ic_functions_measure_monotonic_relationship() -> None:
+    factor = pd.Series([1.0, 2.0, 3.0, 4.0])
+    forward = pd.Series([10.0, 20.0, 30.0, 40.0])
+
+    assert spearman_ic(factor, forward) == pytest.approx(1.0)
+    result = ic_summary(factor, forward, periods=(1,), method="pearson")
+    assert result.loc[1, "ic_mean"] == pytest.approx(1.0)
+    assert result.loc[1, "n"] == 4
+
+
+def test_ic_summary_rejects_unknown_method() -> None:
+    with pytest.raises(ValueError, match="not supported"):
+        ic_summary(pd.Series([1, 2, 3]), pd.Series([1, 2, 3]), method="kendall")
+
+
+def test_winsorize_clips_tails_and_preserves_nan() -> None:
+    values = pd.Series([0.0, 1.0, 2.0, 100.0, np.nan])
+
+    result = winsorize(values, lower=0.25, upper=0.75)
+
+    assert result.iloc[0] == pytest.approx(0.75)
+    assert result.iloc[3] == pytest.approx(26.5)
+    assert pd.isna(result.iloc[4])
+
+
+def test_distribution_diagnostics_handle_short_samples() -> None:
+    assert np.isnan(skewness(pd.Series([1.0, 2.0])))
+    assert np.isnan(kurtosis(pd.Series([1.0, 2.0, 3.0])))
+    jb, p_value = jb_test(pd.Series(range(7), dtype=float))
+    assert np.isnan(jb)
+    assert np.isnan(p_value)
+
+
+def test_distribution_diagnostics_return_finite_values() -> None:
+    values = pd.Series([-2.0, -1.0, -0.5, 0.0, 0.25, 0.75, 1.0, 3.0])
+
+    assert np.isfinite(skewness(values))
+    assert np.isfinite(kurtosis(values))
+    jb, p_value = jb_test(values)
+    assert jb >= 0
+    assert 0 <= p_value <= 1
+
+
+def test_ols_recovers_linear_coefficients_and_residual_index() -> None:
+    index = pd.date_range("2026-01-01", periods=8)
+    factor = pd.Series(np.arange(8, dtype=float), index=index, name="factor")
+    target = 1.5 + 2.0 * factor
+
+    result = ols_regress(target, factor)
+
+    assert result.alpha == pytest.approx(1.5)
+    assert result.beta["factor"] == pytest.approx(2.0)
+    assert result.r_squared == pytest.approx(1.0)
+    assert result.n == 8
+    assert result.resid.index.equals(index)
+
+
+def test_ols_handles_collinear_factors_without_crashing() -> None:
+    x = pd.DataFrame({"a": np.arange(8, dtype=float), "b": np.arange(8, dtype=float)})
+    y = pd.Series(1.0 + x["a"])
+
+    result = ols_regress(y, x)
+
+    assert result.n == 8
+    assert np.isfinite(result.beta).all()
+    np.testing.assert_allclose(result.resid, 0.0, atol=1e-12)
+
+
+def test_ols_short_sample_returns_empty_estimate() -> None:
+    result = ols_regress(pd.Series([1.0, 2.0]), pd.Series([1.0, 2.0], name="x"))
+
+    assert np.isnan(result.alpha)
+    assert result.beta.empty
+    assert result.n == 2
+
+
+def test_momentum_and_rolling_transforms_match_manual_values() -> None:
+    prices = pd.DataFrame({"A": [100.0, 110.0, 121.0, 133.1]})
+    momentum = cross_sectional_momentum(prices, lookback=2, skip=0)
+    assert momentum.iloc[2, 0] == pytest.approx(0.21)
+
+    returns = pd.Series([0.1, 0.1, -0.5, -0.5])
+    pd.testing.assert_series_equal(
+        time_series_momentum(returns, lookback=2),
+        pd.Series([0, 1, -1, -1]),
+    )
+
+    values = pd.Series([1.0, 2.0, 3.0])
+    zscore = rolling_zscore(values, window=3)
+    assert zscore.iloc[-1] == pytest.approx(1.0)
+    annual_vol = rolling_annual_vol(returns, window=2, min_periods=2, trading_days=4)
+    assert annual_vol.iloc[1] == pytest.approx(0.0)
+
+
+def test_rank_helpers_support_global_and_grouped_ranking() -> None:
+    frame = pd.DataFrame(
+        {"factor": [3.0, 1.0, 2.0, 4.0], "industry": ["x", "x", "y", "y"]}
+    )
+
+    global_rank = cross_sectional_pct_rank(frame, "factor", ascending=True)
+    grouped_rank = cross_sectional_pct_rank(
+        frame, "factor", group_col="industry", ascending=True
+    )
+
+    assert global_rank.tolist() == [0.75, 0.25, 0.5, 1.0]
+    assert grouped_rank.tolist() == [1.0, 0.5, 0.5, 1.0]
+    assert cross_sectional_pct_rank(frame, "missing").empty
+
+
+def test_factor_direction_and_directional_rank() -> None:
+    pe = pd.Series([10.0, 20.0], name="pe_ttm")
+    pd.testing.assert_series_equal(apply_factor_direction(pe), -pe)
+
+    frame = pd.DataFrame({"pe_ttm": [10.0, 20.0], "roe": [0.1, 0.2]})
+    assert cross_sectional_rank_with_direction(frame, "pe_ttm").tolist() == [1.0, 0.5]
+    assert cross_sectional_rank_with_direction(frame, "roe").tolist() == [0.5, 1.0]
+
+
+@pytest.mark.parametrize("direction", ["sideways", "", "REVERSE"])
+def test_factor_direction_rejects_unknown_values(direction: str) -> None:
+    factor = pd.Series([1.0, 2.0], name="roe")
+
+    with pytest.raises(ValueError, match="direction"):
+        apply_factor_direction(factor, direction=direction)
+    with pytest.raises(ValueError, match="direction"):
+        cross_sectional_rank_with_direction(
+            pd.DataFrame({"roe": factor}), "roe", direction=direction
+        )
+
+
+def test_annualized_sharpe_handles_empty_and_nonzero_returns() -> None:
+    assert annualized_sharpe(pd.Series(dtype=float)) == 0.0
+    returns = pd.Series([0.01, -0.01, 0.02, 0.0])
+    expected = returns.mean() * 252 / (returns.std() * np.sqrt(252))
+    assert annualized_sharpe(returns) == pytest.approx(expected)

tests/test_metrics.py

@@ -0,0 +1,93 @@
+"""Mathematical contracts for the standard performance metrics."""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from quant_engine.metrics import (
+    TRADING_DAYS_PER_YEAR,
+    annualized_return,
+    annualized_volatility,
+    calmar_ratio,
+    max_drawdown,
+    sharpe_ratio,
+    summary,
+    win_rate,
+)
+
+
+def test_annualized_return_uses_compounded_simple_returns() -> None:
+    returns = pd.Series([0.10, -0.10])
+    expected = 0.99 ** (TRADING_DAYS_PER_YEAR / 2) - 1.0
+
+    assert annualized_return(returns) == pytest.approx(expected)
+
+
+def test_annualized_volatility_uses_sample_standard_deviation() -> None:
+    returns = pd.Series([0.01, 0.03, 0.02])
+
+    assert annualized_volatility(returns) == pytest.approx(
+        returns.std() * np.sqrt(TRADING_DAYS_PER_YEAR)
+    )
+
+
+def test_sharpe_ratio_subtracts_annual_risk_free_rate() -> None:
+    returns = pd.Series([0.01, -0.005, 0.02, 0.0])
+
+    result = sharpe_ratio(returns, rf=0.02)
+
+    assert result == pytest.approx(
+        (annualized_return(returns) - 0.02) / annualized_volatility(returns)
+    )
+
+
+def test_zero_volatility_metrics_return_zero() -> None:
+    returns = pd.Series([0.0, 0.0, 0.0])
+
+    assert sharpe_ratio(returns) == 0.0
+    assert calmar_ratio(returns) == 0.0
+
+
+def test_max_drawdown_includes_loss_from_initial_capital() -> None:
+    returns = pd.Series([-0.20, 0.0])
+
+    assert max_drawdown(returns) == pytest.approx(-0.20)
+
+
+def test_max_drawdown_tracks_peak_to_trough_loss() -> None:
+    returns = pd.Series([0.10, -0.20, 0.05])
+
+    assert max_drawdown(returns) == pytest.approx(-0.20)
+
+
+def test_metrics_clean_nan_and_infinite_values() -> None:
+    returns = pd.Series([0.10, np.nan, np.inf, -0.05, -np.inf])
+
+    assert win_rate(returns) == 0.5
+    assert summary(returns)["n_days"] == 2
+
+
+def test_summary_aliases_match_canonical_fields() -> None:
+    result = summary(pd.Series([0.01, -0.02, 0.03]))
+
+    assert result["annual_yield"] == result["ann_return"]
+    assert result["annual_sd"] == result["ann_volatility"]
+    assert result["drawback"] == result["max_drawdown"]
+
+
+@pytest.mark.parametrize(
+    "metric",
+    [annualized_return, annualized_volatility, sharpe_ratio, max_drawdown, calmar_ratio, win_rate],
+)
+def test_metrics_reject_non_series_input(metric) -> None:
+    with pytest.raises(TypeError, match=r"expected pd\.Series"):
+        metric([0.01, 0.02])
+
+
+def test_short_and_empty_series_return_zero() -> None:
+    assert annualized_return(pd.Series(dtype=float)) == 0.0
+    assert annualized_volatility(pd.Series([0.01])) == 0.0
+    assert max_drawdown(pd.Series([0.01])) == 0.0
+    assert win_rate(pd.Series(dtype=float)) == 0.0

tests/test_risk.py

@@ -0,0 +1,54 @@
+"""Risk contribution contracts and validation tests."""
+
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+from quant_engine.risk import component_var, marginal_risk_contribution, risk_contribution
+
+
+def test_risk_contribution_sums_to_one_for_positive_portfolio_variance() -> None:
+    weights = np.array([0.5, 0.5])
+    covariance = np.diag([1.0, 4.0])
+
+    result = risk_contribution(weights, covariance)
+
+    np.testing.assert_allclose(result, [0.2, 0.8])
+    assert result.sum() == pytest.approx(1.0)
+
+
+def test_zero_variance_portfolio_falls_back_to_equal_contribution() -> None:
+    result = risk_contribution(np.array([0.2, 0.3, 0.5]), np.zeros((3, 3)))
+
+    np.testing.assert_allclose(result, np.full(3, 1 / 3))
+
+
+def test_marginal_and_component_risk_follow_matrix_identities() -> None:
+    weights = np.array([0.25, 0.75])
+    covariance = np.array([[0.04, 0.01], [0.01, 0.09]])
+
+    marginal = marginal_risk_contribution(weights, covariance)
+    component = component_var(weights, covariance)
+
+    np.testing.assert_allclose(marginal, covariance @ weights)
+    np.testing.assert_allclose(component, weights * marginal)
+    assert component.sum() == pytest.approx(weights @ covariance @ weights)
+
+
+@pytest.mark.parametrize(
+    "function",
+    [risk_contribution, marginal_risk_contribution, component_var],
+)
+def test_risk_functions_reject_covariance_shape_mismatch(function) -> None:
+    with pytest.raises(ValueError, match="does not match weights length"):
+        function(np.array([0.5, 0.5]), np.eye(3))
+
+
+@pytest.mark.parametrize(
+    "function",
+    [risk_contribution, marginal_risk_contribution, component_var],
+)
+def test_risk_functions_reject_empty_portfolio(function) -> None:
+    with pytest.raises(ValueError, match="at least one asset"):
+        function(np.array([]), np.empty((0, 0)))