Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 21 additions & 4 deletions src/hk_equity_strategies/backtest/combo_simulator.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,9 @@
from __future__ import annotations

import math
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, Literal, Mapping
from typing import Any, Literal

import pandas as pd

Expand Down Expand Up @@ -109,6 +110,16 @@ def _combo_target_weights(
regime,
)

leg_budget = math.fsum((etf_target_weight, div_target_weight))
if (
any(
not math.isfinite(float(weight)) or float(weight) < 0.0
for weight in (etf_target_weight, div_target_weight)
)
or leg_budget > 1.0
):
raise ValueError("combo target weights must be finite, non-negative and sum to at most one")

selected = {symbol: float(etf_weights.get(symbol, 0.0)) for symbol in close.columns}
if any(not math.isfinite(weight) or weight < 0.0 for weight in selected.values()) or math.fsum(
selected.values()
Expand All @@ -124,9 +135,15 @@ def _combo_target_weights(

row = {symbol: float(scaled.get(symbol, 0.0)) for symbol in asset_columns}
row[DIVIDEND_SYMBOL] = float(row.get(DIVIDEND_SYMBOL, 0.0)) + float(div_target_weight)
if any(not math.isfinite(weight) or weight < 0.0 for weight in row.values()) or math.fsum(
row.values()
) > 1.0:
row_total = math.fsum(row.values())
if 1.0 < row_total <= math.nextafter(1.0, math.inf) and leg_budget <= 1.0:
largest_symbol = max(
(symbol for symbol, weight in row.items() if weight > 0.0),
key=lambda symbol: (row[symbol], symbol),
)
row[largest_symbol] -= row_total - 1.0
row_total = math.fsum(row.values())
if any(not math.isfinite(weight) or weight < 0.0 for weight in row.values()) or row_total > 1.0:
raise ValueError("combo target weights must be finite, non-negative and sum to at most one")
rows.append({"date": as_of, **row})

Expand Down
89 changes: 86 additions & 3 deletions tests/test_orchestrator_runner.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,7 @@
from __future__ import annotations

import math

import hashlib
import math
import os
import subprocess
import sys
Expand All @@ -14,6 +13,7 @@
import pandas as pd
import pytest

from hk_equity_strategies.backtest.combo_simulator import HkComboBacktestConfig
from hk_equity_strategies.backtest.orchestrator_runner import (
SUPPORTED_PROFILES,
SYNTHETIC_MARKET_HISTORY_GENERATOR_VERSION,
Expand All @@ -31,8 +31,8 @@


def _run_observed_price_case(history, runner, weights=None):
from hk_equity_strategies.backtest.etf_rotation_simulator import HkRotationBacktestConfig, run_etf_rotation_backtest
from hk_equity_strategies.backtest.combo_simulator import HkComboBacktestConfig, run_combo_backtest
from hk_equity_strategies.backtest.etf_rotation_simulator import HkRotationBacktestConfig, run_etf_rotation_backtest

def signal(_):
return weights if weights is not None else {"A": 1.0}, {}
Expand All @@ -53,6 +53,89 @@ def _observed_price_history():
})


def test_combo_target_rounding_reclaims_one_ulp_without_relaxing_budget() -> None:
from hk_equity_strategies.backtest.combo_simulator import (
_combo_target_weights,
run_combo_backtest,
)
from hk_equity_strategies.backtest.etf_rotation_simulator import HkRotationBacktestConfig

dates = pd.to_datetime(["2024-01-31", "2024-02-01", "2024-02-02"] * 3)
history = pd.DataFrame(
{
"date": dates,
"symbol": ["A"] * 3 + ["B"] * 3 + ["03110"] * 3,
"close": [100.0] * 9,
}
)

combo_config = HkComboBacktestConfig(
combo_mode="static",
etf_weight=0.6,
dividend_weight=0.4,
min_history_days=1,
cost_bps=0.0,
)
rotation_config = HkRotationBacktestConfig(min_history_days=1, cost_bps=0.0)
close = history.pivot(index="date", columns="symbol", values="close")
targets = _combo_target_weights(
history,
close,
signal_fn=lambda _history: ({"A": 0.3, "B": 0.1, "03110": 0.3}, {}),
rotation_config=rotation_config,
combo_config=combo_config,
strategy_kwargs={},
asset_columns=close.columns,
)
target = targets.dropna(how="all").iloc[0]
assert math.fsum(target.to_dict().values()) <= 1.0

result = run_combo_backtest(
history,
lambda _history: ({"A": 0.3, "B": 0.1, "03110": 0.3}, {}),
combo_config=combo_config,
rotation_config=rotation_config,
universe_symbols=["A", "B", "03110"],
)

assert result.daily_returns.eq(0.0).all()


@pytest.mark.parametrize(
"config",
[
HkComboBacktestConfig(combo_mode="static", etf_weight=0.61, dividend_weight=0.40, min_history_days=1),
HkComboBacktestConfig(
combo_mode="static",
etf_weight=math.nextafter(math.nextafter(0.6, math.inf), math.inf),
dividend_weight=0.4,
min_history_days=1,
),
HkComboBacktestConfig(combo_mode="static", etf_weight=float("nan"), dividend_weight=0.4, min_history_days=1),
HkComboBacktestConfig(combo_mode="static", etf_weight=-0.1, dividend_weight=1.0, min_history_days=1),
],
)
def test_combo_target_rejects_invalid_leg_budgets(config) -> None:
from hk_equity_strategies.backtest.combo_simulator import run_combo_backtest
from hk_equity_strategies.backtest.etf_rotation_simulator import HkRotationBacktestConfig

history = pd.DataFrame(
{
"date": pd.to_datetime(["2024-01-31", "2024-02-01"] * 3),
"symbol": ["A"] * 2 + ["B"] * 2 + ["03110"] * 2,
"close": [100.0] * 6,
}
)
with pytest.raises(ValueError, match="combo target weights"):
run_combo_backtest(
history,
lambda _history: ({"A": 0.3, "B": 0.1, "03110": 0.3}, {}),
combo_config=config,
rotation_config=HkRotationBacktestConfig(min_history_days=1, cost_bps=0.0),
universe_symbols=["A", "B", "03110"],
)


@pytest.mark.parametrize("runner", ["rotation", "combo"])
@pytest.mark.parametrize("invalid", [float("nan"), 0.0, -1.0, float("inf"), "omitted"])
def test_held_asset_missing_or_invalid_marks_are_not_filled_or_dropped(runner, invalid):
Expand Down