"""Backtrader runner tests — deterministic synthetic accounting (_synthetic). Hand-computable expectations: price series closes 10.20, 10.00, 10.40, 10.20... commission 0.0003, slippage 0.001 of price, next-bar-open fills. """ import json import sys from pathlib import Path import pytest 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, SYNTH # noqa: E402 from worker.backtest import run_backtest # noqa: E402 STRAT = ''' import backtrader as bt class Strategy(bt.Strategy): params = (("size", 100),) def __init__(self): self.bought = False def next(self): if not self.bought and len(self) > 2: self.buy(data=self.getdatabyname("SH#600000"), size=self.p.size) self.bought = True ''' STRAT_NAMED = ''' import backtrader as bt class Strategy(bt.Strategy): def __init__(self): self.a = self.getdatabyname("SH#600000") self.b = self.getdatabyname("BJ#TD001") self.done = False def next(self): if not self.done and len(self) > 2: self.buy(data=self.a, size=100) self.done = True ''' def write_dataset(tmp_path): d = tmp_path / "data" d.mkdir() dfa = synthetic_daily("SH#600000", base=10.0, extra_fields=False) dfb = synthetic_daily("BJ#TD001", base=5.0, extra_fields=False) for name, df in (("SH#600000", dfa), ("BJ#TD001", dfb)): p = d / f"{name.lower().replace('#','_')}.csv" df.to_csv(p, index=False) manifest = { "id": "00000000-0000-0000-0000-000000000000", "hash": "fix-hash", "_synthetic": True, "objects": [ {"instrument": {"symbol": "SH#600000", "market": "cn", "asset_type": "stock"}, "path": f"{name}" , "object_hash": "h", "row_count": 20, "columns": ["date"]}, ], "warnings": [], } manifest["objects"] = [ {"instrument": {"symbol": "SH#600000", "market": "cn", "asset_type": "stock"}, "path": "sh_600000.csv", "object_hash": "h-a", "row_count": 20}, {"instrument": {"symbol": "BJ#TD001", "market": "cn", "asset_type": "stock"}, "path": "bj_td001.csv", "object_hash": "h-b", "row_count": 20}, ] return d, manifest def make_request(tmp_path, code, benchmark=None, params=None): d, manifest = write_dataset(tmp_path) 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), }, d def test_synthetic_labeled_runs_next_bar_fill(tmp_path): req, _ = make_request(tmp_path, STRAT) res = run_backtest(req) assert res["engine"]["name"] == "backtrader" assert res["_synthetic"] is True assert res["trades"], "expected at least one trade" t = res["trades"][0] # signal bar index 3; fills at next-bar open (10.20 * 1.001 slippage), NOT signal-bar close assert t["symbol"] == "SH#600000" assert t["side"] == "buy" assert t["quantity"] == 100 assert abs(t["price"] - 10.20 * 1.001) < 1e-9 assert abs(t["commission"] - t["value"] * 0.0003) < 1e-6 def test_assumptions_out_of_warnings(tmp_path): """执行假设说明移出用户面 warnings，独立 assumptions 字段承载。""" from worker.backtest import EXECUTION_ASSUMPTIONS req, _ = make_request(tmp_path, STRAT) res = run_backtest(req) assert res["assumptions"] == EXECUTION_ASSUMPTIONS for w in res["warnings"]: assert "execution assumptions" not in w for k in EXECUTION_ASSUMPTIONS: assert str(k) not in w def test_equity_cash_accounting(tmp_path): req, _ = make_request(tmp_path, STRAT) res = run_backtest(req) eq = res["equity"] assert len(eq) == 20 first = eq[0] assert first["cash"] == 100000.0 assert first["equity"] == 100000.0 last = eq[-1] filled = 100 * 10.40 * 1.001 comm = filled * 0.0003 # after buy: cash reduced; equity = cash + 100 * final close (10.20? compute from fixture) assert last["cash"] < 100000.0 expected_equity = last["cash"] + 100 * last["closes"]["SH#600000"] assert abs(last["equity"] - expected_equity) < 1e-6 def test_metrics_no_nan_nulls_and_fields(tmp_path): req, _ = make_request(tmp_path, STRAT) res = run_backtest(req) m = res["metrics"] for key in ("total_return", "annual_return", "max_drawdown", "trade_count", "final_equity"): assert key in m assert m[key] is None or isinstance(m[key], (int, float)) if isinstance(m[key], float): assert not (m[key] != m[key] or m[key] in (float("inf"), float("-inf"))) # max drawdown is a non-positive fraction or null assert m["max_drawdown"] is None or m["max_drawdown"] <= 0 assert m["trade_count"] >= 0 # closed round-trips; buy-alone runs have 0 assert res["data_manifest_hash"] == "fix-hash" assert res["elapsed_ms"] >= 0 assert isinstance(res["logs"], list) and res["logs"] assert len(res["equity"]) == 20 def test_named_feeds_present_no_cross_lookahead(tmp_path): req, _ = make_request(tmp_path, STRAT_NAMED) res = run_backtest(req) assert res["trades"] assert res["trades"][0]["symbol"] == "SH#600000" # the second feed was never consumed for first-symbol pricing assert all(t["symbol"] != "BJ#TD001" for t in res["trades"]) def test_benchmark_series_included(tmp_path): req, _ = make_request(tmp_path, STRAT, benchmark="BJ#TD001") res = run_backtest(req) assert all("benchmark" in e and e["benchmark"] is not None for e in res["equity"]) def test_missing_strategy_class_fails_clean(tmp_path): req, _ = make_request(tmp_path, "import backtrader as bt\n\nclass Foo(bt.Strategy):\n pass\n") res = run_backtest(req) assert res["status"] == "failed" assert "Strategy" in res["error"]["message"] def test_syntax_error_fails_clean(tmp_path): req, _ = make_request(tmp_path, "def broken(:\n") res = run_backtest(req) assert res["status"] == "failed" assert res["error"]["code"] == "strategy_syntax" def test_missing_data_object_fails(tmp_path): req, _ = make_request(tmp_path, STRAT) req["dataset_manifest"]["objects"][0]["path"] = "nope.csv" res = run_backtest(req) assert res["status"] == "failed" assert res["error"]["code"] == "data_missing" def test_lookahead_signal_uses_prior_close_not_same_day(tmp_path): # Strategy trades only on the last bar; a legal next-bar fill must not exist yet. strat = ''' import backtrader as bt class Strategy(bt.Strategy): def next(self): if len(self) == 20: self.buy(data=self.getdatabyname("SH#600000"), size=100) ''' req, _ = make_request(tmp_path, strat) res = run_backtest(req) # order placed on final bar; notification/fill cannot occur after data end -> no trade assert not res["trades"] def test_strategy_module_imports_visible_in_methods(tmp_path): # Regression: module-level imports must remain visible inside __init__ and # next. Transport-level isolation is Docker, NOT restricted Python globals. strat = ''' import os import math import backtrader as bt class Strategy(bt.Strategy): def __init__(self): self.foo = os.sep # os imported at module level, used in a method def next(self): c = self.getdatabyname("SH#600000").close if len(self) > 2 and abs(math.copysign(1.0, c[0] - c[-1])) == 1.0: self.foo = math.sqrt(abs(c[0])) ''' req, _ = make_request(tmp_path, strat) res = run_backtest(req) assert res["status"] == "succeeded", res.get("error") def test_strategy_genuine_nameerror_still_fails(tmp_path): # a genuinely undefined name must still surface honestly as runtime_error strat = ''' import backtrader as bt class Strategy(bt.Strategy): def next(self): undefined_variable_xyz.bar() ''' req, _ = make_request(tmp_path, strat) res = run_backtest(req) assert res["status"] == "failed" assert res["error"]["code"] == "runtime_error" def test_fill_value_is_executed_turnover_not_cost_basis(tmp_path): """RED/GREEN accounting regression: fills must record actual turnover abs(ex.size * ex.price). Backtrader's ex.value for SELL orders reports the position cost basis, NOT sale proceeds (real QA showed identical 659.66 for a real buy and sell at different prices). Equity/cash are not affected: broker cash and equity already use executed price and commission.""" strat = ''' import backtrader as bt class Strategy(bt.Strategy): def __init__(self): self.done = False def next(self): if len(self) == 3: self.buy(data=self.getdatabyname("SH#600000"), size=100) elif len(self) == 10: self.sell(data=self.getdatabyname("SH#600000"), size=100) self.done = True ''' req, _ = make_request(tmp_path, strat) res = run_backtest(req) assert res["status"] == "succeeded", res.get("error") sides = [(t["side"], t) for t in res["trades"]] buys = [t for s, t in sides if s == "buy"] sells = [t for s, t in sides if s == "sell"] assert buys and sells, f"expected a buy AND a sell fill, got {sides}" # discount-adjusted commission is charged on the turnover, not on cost basis for t in res["trades"]: assert abs(t["value"] - abs(t["quantity"] * t["price"])) < 1e-9, \ f"fill value must be quantity*price turnover: {t}" assert abs(t["commission"] - t["value"] * 0.0003) < 1e-6, \ f"commission follows turnover: {t}" # honest check: buy and sell execute at different prices so their turnover # differs (cost-basis bug reported identical values for both sides) assert buys and sells assert abs(buys[0]["price"] - sells[0]["price"]) > 1e-9, \ f"buy/sell fill prices must differ: {buys[0]['price']} vs {sells[0]['price']}" assert abs(buys[0]["value"] - sells[0]["value"]) > 1e-9, \ "distinct prices must yield distinct fill values (cost-basis bug regression)" # trade_count remains closed round trips, not fills assert res["metrics"]["trade_count"] == 1