path: root/tests/worker/test_backtest.py

"""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_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