chore: establish Strategy Lab source baseline (development, not release)

1 files changed, 357 insertions, 0 deletions

diff --git a/worker/backtest.py b/worker/backtest.py
new file mode 100644
index 0000000..489ee9a
--- /dev/null
+++ b/worker/backtest.py
@@ -0,0 +1,357 @@
+"""Backtrader runner. User `Strategy` code executes with named feeds and
+next-bar fills. Recording happens in two places:
+  - a sibling Recorder strategy (never trades) records per-bar cash/equity/closes;
+  - a generated subclass of the user's Strategy intercepts notify_order /
+    notify_trade (documented hooks) to record orders and fills, then calls the
+    user's own hook methods.
+"""
+from __future__ import annotations
+
+import ast
+import math
+import resource
+import statistics
+import time
+from pathlib import Path
+
+import backtrader as bt
+import pandas as pd
+
+ENGINE_VERSION = bt.__version__
+EXECUTION_ASSUMPTIONS = {
+    "fill_timing": "signal bar -> order; order fills at NEXT bar open (no same-bar lookahead)",
+    "slippage": "percentage of fill price (broker set_slippage_perc)",
+    "commission": "percentage of order value, charged at execution",
+    "indicator_warmup": "indicators are computed in strategy code; bars before warmup carry no signal",
+    "cn_stock_etf_rules": "broker enforces at submission: buys rounded down to lot=100 "
+                          "(reject if <100), sells rejected if they would exceed the "
+                          "current position (no naked short) or exceed position minus "
+                          "same-symbol pending buys (no T+0 sell of same-bar buys). "
+                          "With next-bar-open daily fills, a sell submitted on the fill "
+                          "day executes next trading day, which is exactly the A-share "
+                          "T+1 rule at daily granularity.",
+    "unmodeled": "suspension days, price-limit halts, intraday sequencing, "
+                 "liquidity/market impact are NOT simulated; no lot-size "
+                 "difference between odd lots of pre-held position on sell",
+    "etf_rules": "A-share listed ETFs follow the same lot=100 buy rule; ETF sell "
+                 "lots in the real market can differ slightly (approximation noted)",
+}
+
+
+class CnDailyRulesBroker(bt.brokers.BackBroker):
+    """Broker rules for cn stock/ETF daily simulation, enforced at submission.
+
+    - index feeds (asset_type=index) are nontradable research proxies: any
+      order on them is rejected at the BROKER level (they only serve as
+      benchmark feeds).
+    - buys: size rounded down to a whole lot of 100; rejected if < 100 or not
+      an integer number of lots after rounding.
+    - sells: rejected when they would exceed the tradable position
+      (no naked short; pending same-symbol buys are excluded to prevent T+0
+      sell of shares that are only settled by a same-cycle buy).
+    """
+
+    def __init__(self, index_feeds=(), rule_assets=(), lot: int = 100):
+        super().__init__()
+        self.index_feeds = set(index_feeds)
+        self.rule_assets = set(rule_assets)
+        self.lot = int(lot)
+        self.rejections = []  # {symbol, reason}
+
+    def _reject(self, order, reason: str):
+        self.rejections.append({"symbol": order.data._name, "reason": reason})
+        order.reject()
+        self.notify(order)
+        return order
+
+    def submit(self, order, check=True):
+        feed = order.data._name
+        if feed in self.index_feeds:
+            return self._reject(order, "index_order_rejected")
+        if feed in self.rule_assets:
+            size = order.size
+            if not float(size).is_integer():
+                order.size = int(size)  # floor for buys handled below
+            if order.isbuy():
+                pending_buy = 0
+                for o in (*self.submitted, *self.pending):
+                    if o.isbuy() and o.data._name == feed and o.status not in \
+                            (o.Rejected, o.Cancelled, o.Completed):
+                        pending_buy += abs(int(o.size or 0)) if not o.executed.size else 0
+                base = int(size)
+                lots = (base + pending_buy) // self.lot * self.lot
+                adjusted = max(0, lots - pending_buy)
+                if adjusted < self.lot:
+                    return self._reject(order, "below_min_lot_100")
+                if adjusted != size:
+                    order.prereject_size = size  # audit trail of requested size
+                    order.size = adjusted
+                    order.executed.remsize = adjusted  # _execute reads remsize
+            elif order.issell():
+                pos = self.getposition(order.data).size
+                pending_buy = 0
+                for o in (*self.submitted, *self.pending):
+                    if o.isbuy() and o.data._name == feed and o.status not in \
+                            (o.Rejected, o.Cancelled, o.Completed):
+                        pending_buy += abs(int(o.size or 0))
+                allowed = max(0, pos - pending_buy)
+                if allowed <= 0:
+                    return self._reject(order, "no_position_to_sell")
+                if abs(order.size) > allowed:
+                    order.prereject_size = abs(order.size)
+                    order.size = -allowed  # trim to owned shares (no short)
+                    order.executed.remsize = order.size
+        return super().submit(order, check=check)
+
+
+def fail(code: str, message: str, details=None, t0: float = 0.0) -> dict:
+    return {
+        "status": "failed",
+        "error": {"code": code, "message": message, "details": details},
+        "engine": {"name": "backtrader", "version": ENGINE_VERSION},
+        "elapsed_ms": int((time.monotonic() - t0) * 1000),
+        "warnings": [],
+    }
+
+
+def _safe_object_path(data_root: str, relpath: str) -> Path:
+    root = Path(data_root).resolve()
+    p = (root / relpath).resolve()
+    if not str(p).startswith(str(root) + "/"):
+        raise ValueError(f"path traversal rejected: {relpath}")
+    if p.is_symlink():
+        raise ValueError(f"symlink rejected in data root: {relpath}")
+    if not p.is_file():
+        raise FileNotFoundError(relpath)
+    return p
+
+
+def _load_frame(path: Path) -> pd.DataFrame:
+    df = pd.read_parquet(path) if path.suffix == ".parquet" else pd.read_csv(path)
+    for c in ("date", "open", "high", "low", "close", "volume"):
+        if c not in df.columns:
+            raise ValueError(f"dataset object {path.name} missing column '{c}'")
+    df["date"] = pd.to_datetime(df["date"])
+    if df["date"].duplicated().any():
+        df = df.drop_duplicates("date", keep="last")
+    if not df["date"].is_monotonic_increasing:
+        df = df.sort_values("date")
+    return df.set_index("date").sort_index()
+
+
+def _compile_user_strategy(code: str):
+    try:
+        ast.parse(code)
+    except SyntaxError as e:
+        raise SyntaxError(f"strategy syntax error at line {e.lineno}: {e.msg}")
+    # ONE namespace serves as both exec globals and locals. With separate
+    # dicts, module-level imports/definitions land in the locals dict while
+    # method bodies resolve names via the globals dict -> NameError in methods
+    # (e.g. `import os` at top level, then os.path in __init__).
+    # The sandbox is the Docker container itself; do not fake Python globals.
+    ns: dict = {"__builtins__": __builtins__}
+    exec(compile(code, "<strategy-source>", "exec"), ns)
+    cls = ns.get("Strategy")
+    if not (isinstance(cls, type) and issubclass(cls, bt.Strategy)):
+        raise ValueError("source must define a class named 'Strategy' subclassing backtrader.Strategy")
+    return cls
+
+
+def run_backtest(request: dict) -> dict:
+    t0 = time.monotonic()
+    cfg = request.get("config") or {}
+    manifest = request.get("dataset_manifest") or {}
+    warnings = list(manifest.get("warnings") or [])
+    if manifest.get("_synthetic"):
+        warnings.append("dataset is a synthetic fixture (_synthetic=true); result is for tests only")
+
+    try:
+        user_cls = _compile_user_strategy(request["code"])
+    except SyntaxError as e:
+        return fail("strategy_syntax", str(e), t0=t0)
+    except ValueError as e:
+        return fail("strategy_invalid", str(e), t0=t0)
+
+    frames = {}
+    try:
+        for obj in manifest.get("objects", []):
+            p = _safe_object_path(request["data_root"], obj["path"])
+            frames[p.resolve().name] = _load_frame(p)
+        if not frames:
+            return fail("data_missing", "dataset manifest has no objects", t0=t0)
+    except FileNotFoundError as e:
+        return fail("data_missing", f"missing dataset object: {e}", t0=t0)
+    except ValueError as e:
+        return fail("data_invalid", str(e), t0=t0)
+
+    feed_frames = [(obj, frames[Path(obj["path"]).resolve().name])
+                   for obj in manifest["objects"] if Path(obj["path"]).resolve().name in frames]
+    symbols = [obj["instrument"]["symbol"] for obj, _ in feed_frames]
+
+    rec = {"orders": [], "fills": [], "closed_trades": [], "rows": []}
+
+    def RecordingStrategy():
+        class R(user_cls):
+            plotinfo = dict(plot=False, subplot=False)
+
+            def notify_order(self, order):
+                rec["orders"].append({
+                    "date": self.datas[0].datetime.date(0).strftime("%Y-%m-%d"),
+                    "symbol": order.data._name,
+                    "side": "buy" if order.isbuy() else "sell",
+                    "order_state": order.getstatusname(),
+                })
+                if order.status == order.Completed:
+                    ex = order.executed
+                    # turnover: actual executed QUANTITY x executed PRICE.
+                    # (Backtrader's `ex.value` for sells reports cost basis, not
+                    # executed turnover — honest accounting must use size*price.)
+                    rec["fills"].append({
+                        "date": self.datas[0].datetime.date(0).strftime("%Y-%m-%d"),
+                        "symbol": order.data._name,
+                        "side": "buy" if order.isbuy() else "sell",
+                        "quantity": abs(float(ex.size)),
+                        "price": float(ex.price),
+                        "commission": float(ex.comm),
+                        "value": abs(float(ex.size) * float(ex.price)),
+                    })
+                user_cls.notify_order(self, order)
+
+            def notify_trade(self, trade):
+                if trade.isclosed:
+                    rec["closed_trades"].append({
+                        "symbol": trade.data._name,
+                        "pnl": float(trade.pnl),
+                        "close_date": self.datas[0].datetime.date(0).strftime("%Y-%m-%d"),
+                    })
+                user_cls.notify_trade(self, trade)
+        return R
+
+    class RowsRecorder(bt.Strategy):
+        plotinfo = dict(plot=False, subplot=False)
+
+        def next(self):
+            closes = {d._name: float(d.close[0]) for d in self.datas if len(d) > 0}
+            rec["rows"].append({
+                "date": self.datas[0].datetime.date(0).strftime("%Y-%m-%d"),
+                "cash": float(self.broker.getcash()),
+                "equity": float(self.broker.getvalue()),
+                "closes": closes,
+                "positions": {d._name: self.getposition(d).size for d in self.datas},
+            })
+
+    cerebro = bt.Cerebro(stdstats=False)
+    for obj, frame in feed_frames:
+        cerebro.adddata(bt.feeds.PandasData(dataname=frame, plot=False,
+                                            timeframe=bt.TimeFrame.Days),
+                        name=obj["instrument"]["symbol"])
+    asset_types = {obj["instrument"]["symbol"]:
+                   obj["instrument"].get("asset_type") for obj, _ in feed_frames}
+    index_feeds = [s for s, a in asset_types.items() if a == "index"]
+    rule_assets = [s for s, a in asset_types.items() if a in ("stock", "etf")]
+    broker = CnDailyRulesBroker(index_feeds=index_feeds, rule_assets=rule_assets)
+    cerebro.setbroker(broker)
+    initial_capital = float(cfg.get("capital", 100000.0))
+    cerebro.broker.setcash(initial_capital)
+    cerebro.broker.setcommission(commission=float(cfg.get("commission", 0.0003)))
+    cerebro.broker.set_slippage_perc(perc=float(cfg.get("slippage", 0.001)))
+    cerebro.addstrategy(RowsRecorder)
+    cerebro.addstrategy(RecordingStrategy(), **(cfg.get("parameters") or {}))
+
+    try:
+        strat_list = cerebro.run(runonce=False, preload=False)
+    except Exception as exc:  # user-code runtime errors surface honestly
+        return fail("runtime_error", f"{type(exc).__name__}: {exc}", t0=t0)
+
+    reasons = {
+        "index_order_rejected": "index feeds are nontradable proxies; broker rejected the order",
+        "below_min_lot_100": "buy below minimum lot 100; broker rejected the order",
+        "no_position_to_sell": "sell would exceed owned shares (no naked short); broker rejected the order",
+    }
+    by_reason: dict[tuple[str, str], int] = {}
+    for rej in broker.rejections:
+        by_reason[(rej["symbol"], rej["reason"])] = by_reason.get((rej["symbol"], rej["reason"]), 0) + 1
+    for (sym, reason), n in by_reason.items():
+        rec["orders"].append({"date": None, "symbol": sym, "side": None,
+                              "order_state": "rejected", "reject_reason": reasons.get(reason, reason),
+                              "count": n})
+        warnings.append(f"broker rejected {n} order(s) for {sym}: {reasons.get(reason, reason)}")
+
+    warnings.append("execution assumptions: "
+                    + "; ".join(f"{k}: {v}" for k, v in EXECUTION_ASSUMPTIONS.items()))
+    rows = rec["rows"]
+
+    benchmark_symbol = cfg.get("benchmark_symbol")
+    bmk_rows = [r["closes"].get(benchmark_symbol) for r in rows if benchmark_symbol]
+    bmk_valid = [v for v in bmk_rows if v is not None]
+    bmk_base = bmk_valid[0] if bmk_valid else None
+    eq, bench_missing = [], False
+    for r in rows:
+        e = {"date": r["date"], "equity": r["equity"], "cash": r["cash"],
+             "closes": r["closes"], "positions": r["positions"]}
+        if benchmark_symbol:
+            close = r["closes"].get(benchmark_symbol)
+            if close is None:
+                bench_missing = True
+                e["benchmark"] = None
+            elif bmk_base:
+                # normalized to initial capital so equity and benchmark share a
+                # common start (raw price is also available in e["closes"])
+                e["benchmark"] = (close / bmk_base) * initial_capital
+            else:
+                e["benchmark"] = None
+        eq.append(e)
+    if bmk_rows and benchmark_symbol and bmk_base:
+        warnings.append(f"benchmark '{benchmark_symbol}' series normalized to initial capital "
+                        f"{initial_capital:g} (raw closes available in equity[].closes)")
+    if bench_missing and benchmark_symbol:
+        warnings.append(f"benchmark_symbol '{benchmark_symbol}' not present in dataset; benchmark omitted")
+
+    equity_vals = [r["equity"] for r in rows]
+    total_return = annual_return = sharpe = max_dd = final = None
+    if len(equity_vals) >= 2 and equity_vals[0] > 0:
+        final = equity_vals[-1]
+        total_return = final / equity_vals[0] - 1.0
+        n = len(equity_vals)
+        annual_return = (1.0 + total_return) ** (252.0 / n) - 1.0
+        peak, max_dd = -float("inf"), 0.0
+        for v in equity_vals:
+            peak = max(peak, v)
+            max_dd = max(max_dd, (peak - v) / peak)
+        rets = [b / a - 1.0 for a, b in zip(equity_vals, equity_vals[1:])]
+        std = statistics.pstdev(rets)
+        sharpe = (statistics.fmean(rets) / std) * (252 ** 0.5) if std > 1e-12 else None
+    elif equity_vals:
+        final = equity_vals[-1]
+    clean = lambda v: v if v is None or math.isfinite(v) else None
+    metrics = {
+        "total_return": clean(total_return),
+        "annual_return": clean(annual_return),
+        "max_drawdown": (None if max_dd is None else -max_dd),  # negative loss fraction; 0.0 is valid
+        "sharpe": clean(sharpe),
+        # trade_count = number of closed round-trip trades (开了又平完成一次), not fills
+        "trade_count": len(rec["closed_trades"]),
+        "final_equity": clean(final),
+    }
+
+    return {
+        "status": "succeeded",
+        "engine": {"name": "backtrader", "version": ENGINE_VERSION},
+        "initial_cash": initial_capital,
+        "equity": eq,
+        "orders": rec["orders"],
+        "trades": rec["fills"],  # executions/fills: {date,symbol,side,quantity,price,commission,value}
+        "closed_trades": rec["closed_trades"],
+        "metrics": metrics,
+        "logs": [
+            f"named feeds: {', '.join(symbols)}",
+            f"bars in equity series: {len(rows)}",
+            f"final equity: {metrics['final_equity']}",
+        ],
+        "warnings": warnings,
+        "execution_assumptions": EXECUTION_ASSUMPTIONS,
+        "elapsed_ms": int((time.monotonic() - t0) * 1000),
+        "peak_rss_kb": int(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss),
+        "data_manifest_hash": manifest.get("hash"),
+        "_synthetic": bool(request.get("_synthetic")) or bool(manifest.get("_synthetic")),
+    }