diff --git a/tests/test_backtest.py b/tests/test_backtest.py new file mode 100644 index 0000000..f642d4f --- /dev/null +++ b/tests/test_backtest.py @@ -0,0 +1,124 @@ +"""Backtest contract tests for weights, NAV, rebalancing, and benchmarks.""" + +from __future__ import annotations + +import pandas as pd +import pytest + +from quant_engine.backtest import ( + compare_to_benchmark, + compute_nav_from_weights, + compute_returns_from_nav, + rebalance_periodic, + 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) diff --git a/tests/test_factor_library.py b/tests/test_factor_library.py new file mode 100644 index 0000000..25591fe --- /dev/null +++ b/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) diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..2f29cf2 --- /dev/null +++ b/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="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 diff --git a/tests/test_risk.py b/tests/test_risk.py new file mode 100644 index 0000000..f1899be --- /dev/null +++ b/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)))