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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
|
"""Regression tests: broker rules, benchmark normalization, drawdown zero."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
sys.path.insert(0, str(Path(__file__).resolve().parent))
from fixtures_synth import synthetic_daily # noqa: E402
from worker.backtest import run_backtest # noqa: E402
def build_manifest(tmp_path, specs):
d = tmp_path / "data"
d.mkdir(exist_ok=True)
objects = []
for sym, code, asset, base in specs:
df = synthetic_daily(sym, base=base, extra_fields=False) # code param unused
slug = sym.lower().replace("#", "_")
path = f"{slug}.csv"
df.to_csv(d / path, index=False)
objects.append({"instrument": {"symbol": sym, "market": "cn", "asset_type": asset},
"path": path, "row_count": len(df)})
manifest = {"hash": "fix-hash", "_synthetic": True, "objects": objects, "warnings": []}
return d, manifest
def make_request(tmp_path, code, specs, benchmark=None, params=None):
d, manifest = build_manifest(tmp_path, specs)
return {
"_synthetic": True,
"code": code,
"config": {"capital": 100000.0, "commission": 0.0003, "slippage": 0.001,
"benchmark_symbol": benchmark, "parameters": params or {}},
"dataset_manifest": manifest,
"data_root": str(d),
}
BUY_STOCK_ONLY = '''
import backtrader as bt
class Strategy(bt.Strategy):
def __init__(self): self.done = False
def next(self):
if not self.done and len(self) > 2:
self.buy(data=self.getdatabyname("SH#600000"), size=250)
self.done = True
'''
BUY_INDEX = '''
import backtrader as bt
class Strategy(bt.Strategy):
def __init__(self): self.done = False
def next(self):
if not self.done and len(self) > 2:
self.buy(data=self.getdatabyname("SH000300"), size=100)
self.done = True
'''
BUY_BM = '''
import backtrader as bt
class Strategy(bt.Strategy):
def __init__(self): self.done = False
def next(self):
if not self.done and len(self) > 2:
self.buy(data=self.getdatabyname("SH#600000"), size=100)
self.done = True
'''
SELL_OPEN = '''
import backtrader as bt
class Strategy(bt.Strategy):
def __init__(self): self.state = 0
def next(self):
if self.state == 0 and len(self) > 2:
self.buy(data=self.getdatabyname("SH#600000"), size=200)
self.state = 1
elif self.state == 1:
self.sell(data=self.getdatabyname("SH#600000"), size=150)
self.state = 2
'''
SHORT = '''
import backtrader as bt
class Strategy(bt.Strategy):
def __init__(self): self.state = 0
def next(self):
if self.state == 0 and len(self) > 2:
self.buy(data=self.getdatabyname("SH#600000"), size=100)
self.state = 1
elif self.state == 1:
self.sell(data=self.getdatabyname("SH#600000"), size=400)
self.state = 2
'''
BUY250_TEST = '''
import backtrader as bt
class Strategy(bt.Strategy):
def __init__(self): self.done = False
def next(self):
if not self.done and len(self) > 2:
self.buy(data=self.getdatabyname("SH#600000"), size=250)
self.done = True
'''
def test_zero_drawdown_is_zero_not_null(tmp_path):
req = make_request(tmp_path, SELL_OPEN, [("SH#600000", "sh_600000", "stock", 10.0)])
res = run_backtest(req)
assert res["status"] == "succeeded"
m = res["metrics"]
req2 = make_request(tmp_path / "flat" if False else tmp_path,
'import backtrader as bt\nclass Strategy(bt.Strategy):\n pass\n',
[("SH#600000", "sh_600000", "stock", 10.0)])
res2 = run_backtest(req2)
assert res2["metrics"]["max_drawdown"] == 0.0
assert res2["metrics"]["total_return"] == 0.0
assert res2["metrics"]["trade_count"] == 0
# initial capital and per-bar positions recorded
assert res2["initial_cash"] == 100000.0
assert all("positions" in e for e in res2["equity"])
def test_benchmark_normalized_to_initial_capital(tmp_path):
# stock 10 -> wobbles; benchmark BJ feed base differs so raw price != equity scale
req = make_request(
tmp_path, SELL_OPEN,
[("SH#600000", "sh_600000", "stock", 10.0),
("SH000300", "sh000300", "index", 3800.0)],
benchmark="SH000300")
res = run_backtest(req)
assert all(e["benchmark"] is not None for e in res["equity"])
# raw price never mixed into the equity scale
assert res["equity"][0]["benchmark"] == res["initial_cash"]
assert res["equity"][-1]["closes"]["SH000300"] > 3000 # raw close available separately
assert any("normalized to initial capital" in w for w in res["warnings"])
assert res["orders"]
def test_index_order_rejected_broker_level(tmp_path):
specs = [("SH#600000", "sh_600000", "stock", 10.0),
("SH000300", "sh000300", "index", 3800.0)]
req = make_request(tmp_path, BUY_INDEX, specs)
res = run_backtest(req)
assert res["status"] == "succeeded"
index_fills = [t for t in res["trades"] if t["symbol"] == "SH000300"]
assert not index_fills
assert not [t for t in res["trades"] if t["symbol"] == "SH#600000"] # strategy only bought index
user_orders = [o for o in res["orders"] if o.get("date")] # real order notifications
rejected = [o for o in user_orders if o["symbol"] == "SH000300"]
assert rejected and rejected[-1]["order_state"] == "Rejected"
rejects = [o for o in res["orders"] if o.get("order_state") == "rejected"]
assert rejects and "nontradable" in rejects[0]["reject_reason"]
assert any("nontradable" in w for w in res["warnings"])
def test_index_only_as_benchmark_accepted(tmp_path):
specs = [("SH#600000", "sh_600000", "stock", 10.0),
("SH000300", "sh000300", "index", 3800.0)]
req = make_request(tmp_path, BUY_BM, specs, benchmark="SH000300")
res = run_backtest(req)
assert res["status"] == "succeeded"
assert res["trades"][0]["symbol"] == "SH#600000"
assert res["equity"][-1]["benchmark"] is not None
def test_lot_size_rounded_down_to_100(tmp_path):
req = make_request(tmp_path, BUY250_TEST, [("SH#600000", "sh_600000", "stock", 10.0)])
res = run_backtest(req)
assert res["status"] == "succeeded"
assert res["trades"], "rejection would leave no fill; rounding should keep 200"
q = res["trades"][0]["quantity"]
assert q == 200, f"expected 250 -> 200 (2 lots), got {q}"
def test_no_naked_short_rejected(tmp_path):
req = make_request(tmp_path, SHORT, [("SH#600000", "sh_600000", "stock", 10.0)])
res = run_backtest(req)
short_fills = [t for t in res["trades"] if t["side"] == "sell" and t["quantity"] > 100]
# position was 100 (bought) but sell attempted 400 -> trimmed or rejected, not shorted
pos_last = res["equity"][-1]["positions"]["SH#600000"]
assert pos_last >= 0
assert any("no naked short" in w or "would exceed" in w for w in res["warnings"]) or \
not short_fills or short_fills[0]["quantity"] <= 100
def test_t1_samebar_buy_sell_guarded(tmp_path):
# sell submitted on the same bar as buy (position still 0 at submit) must be trimmed
req = make_request(tmp_path, SELL_OPEN, [("SH#600000", "sh_600000", "stock", 10.0)])
res = run_backtest(req)
assert res["status"] == "succeeded"
positions = [e["positions"]["SH#600000"] for e in res["equity"]]
assert min(positions) >= 0 # never negative (no short anywhere in series)
def test_cash_plus_position_reconciles_equity(tmp_path):
req = make_request(tmp_path, BUY_STOCK_ONLY, [("SH#600000", "sh_600000", "stock", 10.0)])
res = run_backtest(req)
last = res["equity"][-1]
held = {s: p for s, p in last["positions"].items() if p}
expected = last["cash"] + sum(p * last["closes"][s] for s, p in held.items())
assert abs(last["equity"] - expected) < 1e-6
|
