wbt 0.7.0

Weight-based backtesting engine for quantitative trading
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

from __future__ import annotations

import json

import msgpack
import numpy as np
import pytest

import wbt
from wbt import WeightBacktest
from wbt.result import BacktestResult, _json_safe
from wbt.serialization import FORMAT, FORMAT_VERSION, load_msgpack, to_msgpack


@pytest.fixture
def result(wb: WeightBacktest) -> BacktestResult:
    return wb.to_result()


# ---------------------------------------------------------------------------
# 非有限浮点：to_dict 必须是严格 JSON（allow_nan=False 不报错）
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("full", [False, True])
def test_to_dict_is_strict_json(result: BacktestResult, full: bool) -> None:
    payload = result.to_dict(full=full)
    # allow_nan=False：一旦出现 NaN / Infinity 会抛 ValueError
    json.dumps(payload, allow_nan=False)


def test_json_safe_coerces_non_finite() -> None:
    assert _json_safe(float("nan")) is None
    assert _json_safe(float("inf")) is None
    assert _json_safe(float("-inf")) is None
    assert _json_safe(np.float64("nan")) is None
    assert _json_safe(np.float64("inf")) is None
    # 有限值原样保留
    assert _json_safe(1.5) == 1.5
    assert _json_safe(np.float64(2.0)) == 2.0
    # 数组内的 NaN 也被清洗
    assert _json_safe(np.array([1.0, np.nan, np.inf])) == [1.0, None, None]
    # bool 不被误当作 float
    assert _json_safe(True) is True


# ---------------------------------------------------------------------------
# JSON / MessagePack 共用 wire envelope：严格类型一致性
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("full", [False, True])
def test_json_and_msgpack_roundtrip_preserve_normalized_payload_types(
    result: BacktestResult, tmp_path, full: bool
) -> None:
    result.stats["wire_contract"] = {
        "unicode": "回测结果",
        "empty_list": [],
        "empty_object": {},
        "int": 1,
        "float": 1.5,
        "bool": True,
        "nested": [np.float64("nan"), {"positive": np.float64("inf"), "negative": float("-inf")}],
    }
    json_path = tmp_path / "backtest_results.json"
    msgpack_path = tmp_path / "backtest_results.msgpack"

    result.dump_json(json_path, full=full)
    result.dump_msgpack(msgpack_path, full=full)

    json_payload = wbt.load_json(json_path)
    msgpack_payload = wbt.load_msgpack(msgpack_path)
    wbt.assert_payload_equal(json_payload, msgpack_payload)
    assert json_payload["stats"]["wire_contract"]["nested"] == [None, {"positive": None, "negative": None}]


def test_payload_comparison_rejects_equal_but_different_types() -> None:
    with pytest.raises(AssertionError, match="type mismatch"):
        wbt.assert_payload_equal({"value": True}, {"value": 1})


# ---------------------------------------------------------------------------
# MessagePack 读写
# ---------------------------------------------------------------------------
def test_roundtrip_via_methods(result: BacktestResult, tmp_path) -> None:
    path = tmp_path / "backtest_results.msgpack"
    result.dump_msgpack(path, full=True)
    payload = load_msgpack(path)
    assert payload["symbol_count"] == result.symbol_count
    assert len(payload["dates"]) == len(result.dates)
    assert set(payload["curves"].keys()) == {"多空", "多头", "空头", "基准", "超额"}


def test_envelope_has_format_and_version(result: BacktestResult) -> None:
    data = result.to_msgpack(full=False)
    envelope = msgpack.unpackb(data, raw=False)
    assert envelope["format"] == FORMAT == "wbt.backtest_result"
    assert envelope["format_version"] == FORMAT_VERSION == 1
    assert isinstance(envelope["payload"], dict)


def test_full_flag_controls_payload(result: BacktestResult) -> None:
    compact = msgpack.unpackb(result.to_msgpack(full=False), raw=False)["payload"]
    full = msgpack.unpackb(result.to_msgpack(full=True), raw=False)["payload"]
    assert "drawdowns" not in compact
    assert "drawdowns" in full


def test_top_level_helpers_exposed(result: BacktestResult) -> None:
    assert wbt.to_msgpack is to_msgpack
    data = wbt.to_msgpack(result)
    assert isinstance(data, bytes)


# ---------------------------------------------------------------------------
# 错误封装被拒绝
# ---------------------------------------------------------------------------
def test_wrong_format_rejected(tmp_path) -> None:
    path = tmp_path / "bad_format.msgpack"
    path.write_bytes(msgpack.packb({"format": "nope", "format_version": 1, "payload": {}}, use_bin_type=True))
    with pytest.raises(ValueError, match="unexpected format"):
        load_msgpack(path)


def test_unknown_version_rejected(tmp_path) -> None:
    path = tmp_path / "bad_version.msgpack"
    path.write_bytes(msgpack.packb({"format": FORMAT, "format_version": 999, "payload": {}}, use_bin_type=True))
    with pytest.raises(ValueError, match="unsupported format_version"):
        load_msgpack(path)


def test_non_mapping_envelope_rejected(tmp_path) -> None:
    path = tmp_path / "bad_envelope.msgpack"
    path.write_bytes(msgpack.packb([1, 2, 3], use_bin_type=True))
    with pytest.raises(ValueError):
        load_msgpack(path)