mirror of https://ghfast.top/https://github.com/aeroxw/easy-tdx.git synced 2026-09-12 20:24:16 +08:00

Files

T

GitHubandClaude Opus 4.8 5f14c44791 docs: add backtest engine implementation plan

10 tasks, TDD approach, covering:
- Core data types (Signal/Trade/Position/BacktestResult)
- Strategy base class with DataProxy and crossover
- OrderSimulator (5 execution modes + reject policy)
- PortfolioTracker (equity curve + drawdown)
- PerformanceAnalyzer (18 metrics)
- BacktestEngine (vectorized execution pipeline)
- DSL skeleton (P1)
- CLI integration with auto data fetch

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

2026-06-09 16:25:11 +08:00

76 KiB

Raw Blame History

Backtest Engine Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 为 easy-tdx 新增 easy_tdx.backtest 纯计算回测模块，支持 Python 类策略定义、向量化执行、5 种撮合规则、18 项绩效指标、CLI 一键回测。

Architecture: 自底向上构建——数据类型 → 策略基类 → 撮合器 → 持仓追踪 → 绩效分析 → 引擎编排 → CLI 集成。每层独立可测，无网络依赖。

Tech Stack: Python 3.10+, pandas, numpy, click（项目已有依赖）

Spec: docs/superpowers/specs/2026-06-09-backtest-engine-design.md (rev 2.1)

File Structure

# 新建文件
src/easy_tdx/backtest/__init__.py          # 公开 API 导出
src/easy_tdx/backtest/types.py             # Signal / Trade / Position / BacktestResult
src/easy_tdx/backtest/strategy.py          # Strategy 基类 + StrategyDataProxy
src/easy_tdx/backtest/orders.py            # OrderSimulator（撮合规则 + 拒绝策略）
src/easy_tdx/backtest/portfolio.py         # PortfolioTracker（持仓/资金曲线）
src/easy_tdx/backtest/performance.py       # PerformanceAnalyzer（18 项绩效指标）
src/easy_tdx/backtest/engine.py            # BacktestEngine（向量化执行管道）
src/easy_tdx/backtest/dsl.py              # @dsl_strategy 装饰器 + DSL 函数桥接 (P1，骨架)
src/easy_tdx/backtest/cli.py              # CLI 命令

# 修改文件
src/easy_tdx/cli/__init__.py              # 注册 backtest 命令
src/easy_tdx/__init__.py                  # 导出 BacktestEngine, Strategy 等（可选）

# 测试文件
tests/unit/test_backtest_types.py          # 数据类型 round-trip
tests/unit/test_backtest_strategy.py       # Strategy 基类 + DataProxy
tests/unit/test_backtest_orders.py         # 撮合规则 + 拒绝策略
tests/unit/test_backtest_portfolio.py      # 持仓追踪 + 资金曲线
tests/unit/test_backtest_performance.py    # 绩效指标（手工验证）
tests/unit/test_backtest_engine.py         # 引擎端到端
tests/unit/test_backtest_cli.py           # CLI 命令

Task 1: 数据类型 — types.py

Files:

Create: src/easy_tdx/backtest/__init__.py
Create: src/easy_tdx/backtest/types.py
Create: tests/unit/test_backtest_types.py
Step 1: 创建包结构和空 __init__.py

# src/easy_tdx/backtest/__init__.py
"""easy_tdx.backtest — 向量化策略回测引擎（纯计算，零网络依赖）。"""

Step 2: 写 types.py 的失败测试

# tests/unit/test_backtest_types.py
"""回测模块数据类型测试。"""

from __future__ import annotations

import json

import pandas as pd
import pytest

from easy_tdx.backtest.types import BacktestResult, Position, Signal, Trade


class TestSignal:
    def test_buy_signal_defaults(self):
        s = Signal(datetime=20240101, direction="BUY", size=0)
        assert s.direction == "BUY"
        assert s.price is None
        assert s.stop_loss is None
        assert s.take_profit is None
        assert s.size == 0  # 0 = 全仓

    def test_sell_signal_with_price(self):
        s = Signal(datetime=20240102, direction="SELL", size=100, price=50.0, stop_loss=48.0)
        assert s.price == 50.0
        assert s.stop_loss == 48.0


class TestTrade:
    def test_trade_defaults(self):
        t = Trade(datetime=20240101, direction="BUY", size=100, price=50.0, commission=1.5, slippage=0.0, pnl=0.0)
        assert t.rejected is False

    def test_rejected_trade(self):
        t = Trade(datetime=20240101, direction="BUY", size=100, price=50.0, commission=0.0, slippage=0.0, pnl=0.0, rejected=True)
        assert t.rejected is True


class TestPosition:
    def test_position_fields(self):
        p = Position(datetime=20240101, size=100, avg_price=50.0, market_value=5000.0, unrealized_pnl=100.0)
        assert p.size == 100
        assert p.avg_price == 50.0


class TestBacktestResult:
    def test_to_dict(self):
        result = BacktestResult(
            performance={"total_return": 0.1},
            equity_curve=pd.DataFrame({"total": [100000, 110000]}),
            trades=pd.DataFrame(),
            positions=pd.DataFrame(),
            config={"cash": 100000},
        )
        d = result.to_dict()
        assert d["performance"]["total_return"] == 0.1
        assert d["config"]["cash"] == 100000

    def test_to_json(self):
        result = BacktestResult(
            performance={"total_return": 0.1},
            equity_curve=pd.DataFrame({"total": [100000, 110000]}),
            trades=pd.DataFrame(),
            positions=pd.DataFrame(),
            config={"cash": 100000},
        )
        j = result.to_json()
        parsed = json.loads(j)
        assert parsed["performance"]["total_return"] == 0.1

    def test_empty_trades(self):
        result = BacktestResult(
            performance={},
            equity_curve=pd.DataFrame(),
            trades=pd.DataFrame(),
            positions=pd.DataFrame(),
            config={},
        )
        assert result.trades.empty

Step 3: 运行测试确认失败

Run: python -m pytest tests/unit/test_backtest_types.py -v Expected: FAIL — ModuleNotFoundError: No module named 'easy_tdx.backtest.types'

Step 4: 实现 types.py

# src/easy_tdx/backtest/types.py
"""回测引擎核心数据类型。"""

from __future__ import annotations

import json
from dataclasses import asdict, dataclass
from typing import Any, Literal

import pandas as pd


@dataclass
class Signal:
    """策略产生的交易信号。"""

    datetime: int
    direction: Literal["BUY", "SELL"]
    size: float  # 0 = 全仓/清仓
    price: float | None = None  # None = 市价
    stop_loss: float | None = None
    take_profit: float | None = None


@dataclass
class Trade:
    """已成交记录。"""

    datetime: int
    direction: Literal["BUY", "SELL"]
    size: float
    price: float
    commission: float
    slippage: float
    pnl: float  # 仅平仓时计算
    rejected: bool = False


@dataclass
class Position:
    """持仓快照。"""

    datetime: int
    size: float  # 正=多头，负=空头，0=空仓
    avg_price: float
    market_value: float
    unrealized_pnl: float


@dataclass
class BacktestResult:
    """回测完整结果。"""

    performance: dict[str, float]
    equity_curve: pd.DataFrame  # datetime, cash, position_value, total, drawdown, drawdown_pct
    trades: pd.DataFrame  # datetime, direction, size, price, commission, pnl, rejected
    positions: pd.DataFrame  # datetime, size, avg_price, market_value, unrealized_pnl
    config: dict[str, Any]

    def to_dict(self) -> dict[str, Any]:
        """转为可 JSON 序列化的字典。"""
        return {
            "performance": self.performance,
            "config": self.config,
            "equity_curve": self.equity_curve.to_dict(orient="records") if not self.equity_curve.empty else [],
            "trades": self.trades.to_dict(orient="records") if not self.trades.empty else [],
            "positions": self.positions.to_dict(orient="records") if not self.positions.empty else [],
        }

    def to_json(self) -> str:
        """转为 JSON 字符串。"""
        return json.dumps(self.to_dict(), ensure_ascii=False, default=str)

    def summary(self) -> None:
        """打印回测概要到 stdout。"""
        p = self.performance
        click_echo = print  # 避免 import click
        click_echo(f"总收益率: {p.get('total_return', 0):.2%}")
        click_echo(f"年化收益率: {p.get('annual_return', 0):.2%}")
        click_echo(f"最大回撤: {p.get('max_drawdown', 0):.2%}")
        click_echo(f"夏普比率: {p.get('sharpe', 0):.2f}")
        click_echo(f"胜率: {p.get('win_rate', 0):.2%}")
        click_echo(f"总交易: {p.get('total_trades', 0)} 次")

Step 5: 运行测试确认通过

Run: python -m pytest tests/unit/test_backtest_types.py -v Expected: 5 passed

Step 6: Commit

git add src/easy_tdx/backtest/__init__.py src/easy_tdx/backtest/types.py tests/unit/test_backtest_types.py
git commit -m "feat(backtest): add core data types (Signal/Trade/Position/BacktestResult)"

Task 2: 策略基类 + 数据代理 — strategy.py

Files:

Create: src/easy_tdx/backtest/strategy.py
Create: tests/unit/test_backtest_strategy.py
Step 1: 写 strategy.py 的失败测试

# tests/unit/test_backtest_strategy.py
"""Strategy 基类 + StrategyDataProxy 测试。"""

from __future__ import annotations

import numpy as np
import pandas as pd
import pytest

from easy_tdx.backtest.strategy import Strategy, StrategyDataProxy, crossover


def _make_df(n: int = 20, seed: int = 42) -> pd.DataFrame:
    """构造测试用 K 线 DataFrame。"""
    rng = np.random.default_rng(seed)
    close = 100 + np.cumsum(rng.standard_normal(n))
    high = close + np.abs(rng.standard_normal(n))
    low = close - np.abs(rng.standard_normal(n))
    open_ = low + (high - low) * rng.random(n)
    vol = (rng.random(n) * 1e6).astype(float)
    return pd.DataFrame({
        "datetime": pd.date_range("2024-01-01", periods=n, freq="D"),
        "open": open_,
        "high": high,
        "low": low,
        "close": close,
        "vol": vol,
        "amount": vol * close,
    })


def _make_df_with_extras(n: int = 20) -> pd.DataFrame:
    """构造带额外指标列的 DataFrame。"""
    df = _make_df(n)
    df["MACD_DIF"] = np.random.randn(n)
    df["BOLL_UPPER"] = df["close"] + 5
    return df


class TestStrategyDataProxy:
    def test_basic_columns(self):
        df = _make_df()
        proxy = StrategyDataProxy(df)
        proxy._set_index(10)
        assert isinstance(proxy.close[0], float)
        assert isinstance(proxy.open[0], float)

    def test_previous_bar(self):
        df = _make_df()
        proxy = StrategyDataProxy(df)
        proxy._set_index(5)
        assert proxy.close[0] == df["close"].iloc[5]
        assert proxy.close[-1] == df["close"].iloc[4]

    def test_extra_columns_via_getattr(self):
        df = _make_df_with_extras()
        proxy = StrategyDataProxy(df)
        proxy._set_index(0)
        assert isinstance(proxy.MACD_DIF[0], float)
        assert isinstance(proxy.BOLL_UPPER[0], float)

    def test_missing_column_raises(self):
        df = _make_df()
        proxy = StrategyDataProxy(df)
        proxy._set_index(0)
        with pytest.raises(AttributeError):
            _ = proxy.NONEXISTENT[0]

    def test_len(self):
        df = _make_df()
        proxy = StrategyDataProxy(df)
        proxy._set_index(0)
        assert len(proxy.close) >= 10


class TestCrossover:
    def test_crossover_true(self):
        a = np.array([1, 2, 3, 4, 5], dtype=float)
        b = np.array([5, 4, 3, 2, 1], dtype=float)
        # a crosses above b between index 1 and 2
        mask = crossover(a, b)
        assert mask[2] is np.True_

    def test_crossover_false_no_cross(self):
        a = np.array([1, 2, 3, 4, 5], dtype=float)
        b = np.array([10, 10, 10, 10, 10], dtype=float)
        mask = crossover(a, b)
        assert not mask.any()

    def test_crossover_series(self):
        s1 = pd.Series([1, 2, 5])
        s2 = pd.Series([3, 3, 3])
        mask = crossover(s1, s2)
        assert mask[2] is np.True_


class TestStrategyBase:
    def test_subclass_init_and_next(self):
        class TestStrat(Strategy):
            inited = False
            nexted = False

            def init(self):
                self.inited = True

            def next(self):
                self.nexted = True

        df = _make_df()
        strat = TestStrat()
        strat._bind_data(df)
        strat._call_init()
        assert strat.inited

        strat._set_bar_index(5)
        strat._call_next()
        assert strat.nexted

    def test_buy_sell_recording(self):
        class TestStrat(Strategy):
            def next(self):
                if self._bar_index == 0:
                    self.buy(size=100)
                elif self._bar_index == 5:
                    self.sell(size=100)

        df = _make_df()
        strat = TestStrat()
        strat._bind_data(df)
        strat._call_init()

        strat._set_bar_index(0)
        strat._call_next()
        assert len(strat._signals) == 1
        assert strat._signals[0].direction == "BUY"
        assert strat._signals[0].size == 100

        strat._set_bar_index(5)
        strat._call_next()
        assert len(strat._signals) == 2
        assert strat._signals[1].direction == "SELL"

    def test_I_registers_indicator(self):
        from easy_tdx import MyTT

        class TestStrat(Strategy):
            def init(self):
                self.ma5 = self.I(MyTT.MA, self.data.close, 5)

            def next(self):
                pass

        df = _make_df(50)
        strat = TestStrat()
        strat._bind_data(df)
        strat._call_init()
        assert len(strat.ma5) == 50

    def test_full_position_buy(self):
        class TestStrat(Strategy):
            def next(self):
                self.buy(size=0)  # 全仓

        df = _make_df()
        strat = TestStrat()
        strat._bind_data(df)
        strat._call_init()
        strat._set_bar_index(0)
        strat._call_next()
        assert strat._signals[0].size == 0  # 0 = 全仓标记

Step 2: 运行测试确认失败

Run: python -m pytest tests/unit/test_backtest_strategy.py -v Expected: FAIL — ModuleNotFoundError

Step 3: 实现 strategy.py

# src/easy_tdx/backtest/strategy.py
"""策略基类和数据代理。"""

from __future__ import annotations

from abc import ABC, abstractmethod
from collections.abc import Callable
from typing import Any

import numpy as np
import pandas as pd

from .types import Signal


class _SeriesAccessor:
    """支持 [0] 当前值、[-1] 前一根、切片。"""

    def __init__(self, series: np.ndarray, bar_index: int) -> None:
        self._series = series
        self._bar = bar_index

    def __getitem__(self, key: int) -> float:
        idx = self._bar + key
        return float(self._series[idx])

    def __len__(self) -> int:
        return len(self._series)

    def __array__(self) -> np.ndarray:
        """允许传入 MyTT 函数。"""
        return self._series

    @property
    def raw(self) -> np.ndarray:
        """获取完整原始数组。"""
        return self._series


class StrategyDataProxy:
    """K 线数据代理。

    支持标准列：.close[0] 当前值, .close[-1] 前一根。
    通过 __getattr__ 自动暴露 DataFrame 中的额外列（预计算指标）。
    """

    _STANDARD_COLS = ("open", "close", "high", "low", "vol", "amount")

    def __init__(self, df: pd.DataFrame) -> None:
        self._df = df
        self._arrays: dict[str, np.ndarray] = {}
        for col in df.columns:
            if col == "datetime":
                continue
            self._arrays[col] = df[col].to_numpy(dtype=float)
        self._bar: int = 0

    def _set_index(self, idx: int) -> None:
        self._bar = idx

    def _make_accessor(self, col: str) -> _SeriesAccessor:
        return _SeriesAccessor(self._arrays[col], self._bar)

    @property
    def open(self) -> _SeriesAccessor:  # noqa: A003
        return self._make_accessor("open")

    @property
    def close(self) -> _SeriesAccessor:
        return self._make_accessor("close")

    @property
    def high(self) -> _SeriesAccessor:
        return self._make_accessor("high")

    @property
    def low(self) -> _SeriesAccessor:  # noqa: A003
        return self._make_accessor("low")

    @property
    def vol(self) -> _SeriesAccessor:
        return self._make_accessor("vol")

    @property
    def amount(self) -> _SeriesAccessor:
        return self._make_accessor("amount")

    def __getattr__(self, name: str) -> _SeriesAccessor:
        if name.startswith("_"):
            raise AttributeError(name)
        if name in self._arrays:
            return self._make_accessor(name)
        raise AttributeError(f"DataFrame 中不存在列: {name}")


def crossover(a: np.ndarray | pd.Series | _SeriesAccessor, b: np.ndarray | pd.Series | _SeriesAccessor) -> np.ndarray:
    """检测 a 从下方穿越 b（金叉）。返回 bool 数组。"""
    arr_a = np.asarray(a)
    arr_b = np.asarray(b)
    prev_above = arr_a[:-1] > arr_b[:-1]
    curr_below = arr_a[1:] <= arr_b[1:]
    result = np.zeros(len(arr_a), dtype=bool)
    result[1:] = ~prev_above & ~curr_below
    return result


class Strategy(ABC):
    """用户策略基类。

    继承此类，实现 init() 注册指标，next() 生成信号。
    """

    def __init__(self) -> None:
        self._data_proxy: StrategyDataProxy | None = None
        self._bar_index: int = 0
        self._signals: list[Signal] = []
        self._indicators: dict[str, np.ndarray] = {}
        self._chanlun_result: Any = None
        self._position_size: float = 0.0
        self._cash: float = 0.0

    # --- 生命周期（用户实现）---

    def init(self) -> None:
        """策略初始化。注册指标、设置参数。"""

    def next(self) -> None:
        """每根 K 线调用。在此生成买卖信号。"""

    # --- 指标注册 ---

    def I(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> np.ndarray:
        """注册指标函数。init() 后一次性计算，返回完整数组。"""
        # 将 _SeriesAccessor 参数解包为 numpy 数组
        resolved_args = []
        for arg in args:
            if isinstance(arg, _SeriesAccessor):
                resolved_args.append(arg.raw)
            else:
                resolved_args.append(arg)
        result = func(*resolved_args, **kwargs)
        if isinstance(result, np.ndarray):
            return result
        # 某些指标返回 tuple（如 MACD），取第一个
        return np.asarray(result)

    # --- 交易指令 ---

    def buy(self, size: float = 0, price: float | None = None,
            stop_loss: float | None = None, take_profit: float | None = None) -> None:
        """生成买入信号。size=0 表示全仓。"""
        dt = self._get_datetime()
        self._signals.append(Signal(
            datetime=dt, direction="BUY", size=size,
            price=price, stop_loss=stop_loss, take_profit=take_profit,
        ))

    def sell(self, size: float = 0, price: float | None = None,
             stop_loss: float | None = None, take_profit: float | None = None) -> None:
        """生成卖出信号。size=0 表示清仓。"""
        dt = self._get_datetime()
        self._signals.append(Signal(
            datetime=dt, direction="SELL", size=size,
            price=price, stop_loss=stop_loss, take_profit=take_profit,
        ))

    # --- 属性 ---

    @property
    def data(self) -> StrategyDataProxy:
        assert self._data_proxy is not None, "Strategy 未绑定数据"
        return self._data_proxy

    @property
    def position(self) -> dict[str, float]:
        """当前持仓信息（简化 dict，非 Position dataclass）。"""
        return {"size": self._position_size}

    @property
    def chanlun(self) -> Any:
        """缠论分析结果（如果引擎注入了）。"""
        return self._chanlun_result

    # --- 内部方法（引擎调用）---

    def _bind_data(self, df: pd.DataFrame) -> None:
        """绑定 K 线数据。"""
        self._data_proxy = StrategyDataProxy(df)

    def _call_init(self) -> None:
        """调用用户 init()，在数据绑定后。"""
        self.init()

    def _set_bar_index(self, idx: int) -> None:
        """设置当前 Bar 索引。"""
        self._bar_index = idx
        if self._data_proxy is not None:
            self._data_proxy._set_index(idx)

    def _call_next(self) -> None:
        """调用用户 next()。"""
        self.next()

    def _get_datetime(self) -> int:
        """获取当前 Bar 的 datetime 整数表示。"""
        if self._data_proxy is None:
            return 0
        dt_val = self._data_proxy._df["datetime"].iloc[self._bar_index]
        if hasattr(dt_val, "strftime"):
            return int(dt_val.strftime("%Y%m%d"))
        return int(dt_val)

    def _clear_signals(self) -> list[Signal]:
        """取出并清空信号队列。"""
        signals = self._signals[:]
        self._signals.clear()
        return signals

Step 4: 运行测试确认通过

Run: python -m pytest tests/unit/test_backtest_strategy.py -v Expected: 全部通过

Step 5: Commit

git add src/easy_tdx/backtest/strategy.py tests/unit/test_backtest_strategy.py
git commit -m "feat(backtest): add Strategy base class with DataProxy and crossover"

Task 3: 撮合规则 — orders.py

Files:

Create: src/easy_tdx/backtest/orders.py
Create: tests/unit/test_backtest_orders.py
Step 1: 写 orders.py 的失败测试

# tests/unit/test_backtest_orders.py
"""OrderSimulator 撮合规则测试。"""

from __future__ import annotations

import numpy as np
import pandas as pd
import pytest

from easy_tdx.backtest.orders import OrderSimulator
from easy_tdx.backtest.types import Signal


def _make_df(n: int = 10) -> pd.DataFrame:
    """构造测试用 K 线 DataFrame，价格固定递增。"""
    return pd.DataFrame({
        "datetime": range(20240101, 20240101 + n),
        "open": np.arange(100, 100 + n, dtype=float),
        "close": np.arange(101, 101 + n, dtype=float),
        "high": np.arange(102, 102 + n, dtype=float),
        "low": np.arange(99, 99 + n, dtype=float),
        "vol": np.full(n, 1e6),
        "amount": np.full(n, 1e8),
    })


def _buy_signal(bar_idx: int, size: float = 0) -> Signal:
    return Signal(datetime=20240101 + bar_idx, direction="BUY", size=size)


def _sell_signal(bar_idx: int, size: float = 0) -> Signal:
    return Signal(datetime=20240101 + bar_idx, direction="SELL", size=size)


class TestExecutionModes:
    def test_next_open(self):
        """买入信号在 bar 0，成交价为 bar 1 的 open。"""
        df = _make_df(5)
        sim = OrderSimulator(df, execution="next_open")
        trades = sim.simulate([_buy_signal(0)], cash=100000, position=0.0, position_mode="full")
        assert len(trades) == 1
        assert trades[0].price == 101.0  # df["open"].iloc[1]

    def test_next_close(self):
        df = _make_df(5)
        sim = OrderSimulator(df, execution="next_close")
        trades = sim.simulate([_buy_signal(0)], cash=100000, position=0.0, position_mode="full")
        assert trades[0].price == 102.0  # df["close"].iloc[1]

    def test_this_close(self):
        df = _make_df(5)
        sim = OrderSimulator(df, execution="this_close")
        trades = sim.simulate([_buy_signal(0)], cash=100000, position=0.0, position_mode="full")
        assert trades[0].price == 101.0  # df["close"].iloc[0]

    def test_this_close_future_leak_warning(self):
        """this_close 模式应标记 future_leak_warning。"""
        df = _make_df(5)
        sim = OrderSimulator(df, execution="this_close")
        assert sim.future_leak_warning is True

    def test_worst_price_buy(self):
        """worst 模式买入取 high。"""
        df = _make_df(5)
        sim = OrderSimulator(df, execution="worst")
        trades = sim.simulate([_buy_signal(0)], cash=100000, position=0.0, position_mode="full")
        # next bar worst buy price = high of bar 1 = 103.0
        assert trades[0].price == 103.0

    def test_best_price_buy(self):
        """best 模式买入取 low。"""
        df = _make_df(5)
        sim = OrderSimulator(df, execution="best")
        trades = sim.simulate([_buy_signal(0)], cash=100000, position=0.0, position_mode="full")
        # next bar best buy price = low of bar 1 = 100.0
        assert trades[0].price == 100.0


class TestPositionModes:
    def test_full_position(self):
        """full 模式买入全部现金。"""
        df = _make_df(5)
        sim = OrderSimulator(df, execution="this_close")
        trades = sim.simulate([_buy_signal(0, size=0)], cash=10000, position=0.0, position_mode="full")
        # price = 101.0 (this_close), cash=10000 → 10000/101 ≈ 99 shares (100股整手)
        assert trades[0].size > 0

    def test_fixed_position(self):
        df = _make_df(5)
        sim = OrderSimulator(df, execution="this_close")
        trades = sim.simulate([_buy_signal(0, size=100)], cash=100000, position=0.0, position_mode="fixed")
        assert trades[0].size == 100


class TestRejectPolicy:
    def test_reduce_on_insufficient_cash(self):
        """资金不足时 reduce 模式自动减仓。"""
        df = _make_df(5)
        sim = OrderSimulator(df, execution="this_close", reject_policy="reduce")
        # 现金只有 50，价格 101，买不了 100 股
        trades = sim.simulate([_buy_signal(0, size=100)], cash=50, position=0.0, position_mode="fixed")
        assert len(trades) == 1
        assert trades[0].size < 100

    def test_skip_on_insufficient_cash(self):
        """资金不足时 skip 模式跳过。"""
        df = _make_df(5)
        sim = OrderSimulator(df, execution="this_close", reject_policy="skip")
        trades = sim.simulate([_buy_signal(0, size=100)], cash=50, position=0.0, position_mode="fixed")
        assert len(trades) == 1
        assert trades[0].rejected is True
        assert trades[0].size == 0

    def test_sell_with_no_position_skip(self):
        """无持仓时卖出被跳过。"""
        df = _make_df(5)
        sim = OrderSimulator(df, execution="this_close", reject_policy="skip")
        trades = sim.simulate([_sell_signal(0, size=100)], cash=100000, position=0.0, position_mode="full")
        assert len(trades) == 1
        assert trades[0].rejected is True


class TestFees:
    def test_commission_on_buy(self):
        df = _make_df(5)
        sim = OrderSimulator(df, execution="this_close", commission=0.0003, min_commission=5.0)
        trades = sim.simulate([_buy_signal(0, size=100)], cash=100000, position=0.0, position_mode="fixed")
        assert trades[0].commission >= 5.0  # 最低佣金

    def test_stamp_tax_on_sell(self):
        df = _make_df(5)
        sim = OrderSimulator(df, execution="this_close", stamp_tax=0.001)
        trades = sim.simulate(
            [_buy_signal(0, size=100), _sell_signal(2, size=100)],
            cash=100000, position=0.0, position_mode="fixed",
        )
        sell_trade = [t for t in trades if t.direction == "SELL"][0]
        # 印花税 = size * price * 0.001
        assert sell_trade.commission > 0

Step 2: 运行测试确认失败

Run: python -m pytest tests/unit/test_backtest_orders.py -v Expected: FAIL

Step 3: 实现 orders.py

# src/easy_tdx/backtest/orders.py
"""订单撮合模拟器。"""

from __future__ import annotations

import warnings

import numpy as np
import pandas as pd

from .types import Signal, Trade


class OrderSimulator:
    """将信号转化为成交记录。可配置撮合规则和拒绝策略。"""

    def __init__(
        self,
        df: pd.DataFrame,
        execution: str = "next_open",
        position_mode: str = "full",
        reject_policy: str = "reduce",
        commission: float = 0.0003,
        min_commission: float = 5.0,
        stamp_tax: float = 0.001,
        slippage: float = 0.0,
    ) -> None:
        self._df = df
        self._execution = execution
        self._position_mode = position_mode
        self._reject_policy = reject_policy
        self._commission = commission
        self._min_commission = min_commission
        self._stamp_tax = stamp_tax
        self._slippage = slippage
        self.future_leak_warning = execution == "this_close"

        if self.future_leak_warning:
            warnings.warn(
                "⚠️  执行模式 'this_close' 存在未来函数风险（look-ahead bias），"
                "回测结果可能过度乐观。",
                UserWarning,
                stacklevel=2,
            )

    def simulate(
        self,
        signals: list[Signal],
        cash: float,
        position: float,
        position_mode: str | None = None,
    ) -> list[Trade]:
        """执行信号列表，返回成交记录。

        Args:
            signals: 按时间顺序排列的信号列表
            cash: 当前现金
            position: 当前持仓量
            position_mode: 覆盖构造时的 position_mode
        """
        pos_mode = position_mode or self._position_mode
        trades: list[Trade] = []
        cur_cash = cash
        cur_pos = position

        for signal in signals:
            bar_idx = self._find_bar_index(signal.datetime)
            exec_idx = self._resolve_exec_index(bar_idx)

            if exec_idx is None or exec_idx >= len(self._df):
                continue

            price = self._get_price(exec_idx, signal.direction)
            price += self._slippage * (1 if signal.direction == "BUY" else -1)

            if signal.direction == "BUY":
                trade = self._execute_buy(signal, price, cur_cash, pos_mode)
                if trade is not None:
                    trades.append(trade)
                    if not trade.rejected:
                        cur_cash -= trade.size * trade.price + trade.commission
                        cur_pos += trade.size
            else:
                trade = self._execute_sell(signal, price, cur_pos, cur_cash, pos_mode)
                if trade is not None:
                    trades.append(trade)
                    if not trade.rejected:
                        revenue = trade.size * trade.price - trade.commission
                        cur_cash += revenue
                        cur_pos -= trade.size

        return trades

    def _find_bar_index(self, datetime_val: int) -> int | None:
        """根据 datetime 找到对应的 bar 索引。"""
        dt_col = self._df["datetime"]
        if dt_col.dtype == object or hasattr(dt_col.iloc[0], "strftime"):
            # datetime 类型
            for i, val in enumerate(dt_col):
                if hasattr(val, "strftime"):
                    if int(val.strftime("%Y%m%d")) == datetime_val:
                        return i
                elif int(val) == datetime_val:
                    return i
        else:
            matches = dt_col[dt_col == datetime_val]
            if not matches.empty:
                return int(matches.index[0])
        return None

    def _resolve_exec_index(self, bar_idx: int | None) -> int | None:
        """根据执行模式确定成交的 bar 索引。"""
        if bar_idx is None:
            return None
        if self._execution == "this_close":
            return bar_idx
        # next_open / next_close / worst / best → 下一根
        return bar_idx + 1

    def _get_price(self, idx: int, direction: str) -> float:
        """根据执行模式获取成交价。"""
        if self._execution in ("next_open", "this_close"):
            if self._execution == "next_open":
                return float(self._df["open"].iloc[idx])
            return float(self._df["close"].iloc[idx])
        if self._execution == "next_close":
            return float(self._df["close"].iloc[idx])
        if self._execution == "worst":
            if direction == "BUY":
                return float(self._df["high"].iloc[idx])
            return float(self._df["low"].iloc[idx])
        if self._execution == "best":
            if direction == "BUY":
                return float(self._df["low"].iloc[idx])
            return float(self._df["high"].iloc[idx])
        return float(self._df["open"].iloc[idx])

    def _execute_buy(
        self, signal: Signal, price: float, cash: float, pos_mode: str,
    ) -> Trade | None:
        """执行买入。"""
        if pos_mode == "full" or signal.size == 0:
            # 全仓：用所有现金买，按 100 股整手
            max_shares = int(cash / (price * (1 + self._commission)))
            size = max_shares // 100 * 100  # 整手
        elif pos_mode == "fixed":
            size = int(signal.size)
        elif pos_mode == "percent":
            size = int(cash * signal.size / price)
        else:
            size = int(signal.size)

        if size <= 0 or size * price > cash:
            if self._reject_policy == "skip":
                return Trade(
                    datetime=signal.datetime, direction="BUY", size=0, price=price,
                    commission=0, slippage=0, pnl=0, rejected=True,
                )
            # reduce: 买能买的
            size = int(cash / (price * (1 + self._commission)))
            size = size // 100 * 100
            if size <= 0:
                return Trade(
                    datetime=signal.datetime, direction="BUY", size=0, price=price,
                    commission=0, slippage=0, pnl=0, rejected=True,
                )

        comm = max(size * price * self._commission, self._min_commission)
        return Trade(
            datetime=signal.datetime, direction="BUY", size=size, price=price,
            commission=comm, slippage=self._slippage * size, pnl=0,
        )

    def _execute_sell(
        self, signal: Signal, price: float, position: float, cash: float, pos_mode: str,
    ) -> Trade | None:
        """执行卖出。"""
        if pos_mode == "full" or signal.size == 0:
            size = position
        elif pos_mode == "fixed":
            size = min(signal.size, position)
        elif pos_mode == "percent":
            size = position * signal.size
        else:
            size = min(signal.size, position)

        if size <= 0 or size > position:
            return Trade(
                datetime=signal.datetime, direction="SELL", size=0, price=price,
                commission=0, slippage=0, pnl=0, rejected=True,
            )

        comm_buy = 0  # 买入佣金不计入卖出 Trade
        comm_sell = max(size * price * self._commission, self._min_commission)
        stamp = size * price * self._stamp_tax
        total_comm = comm_sell + stamp

        return Trade(
            datetime=signal.datetime, direction="SELL", size=size, price=price,
            commission=total_comm, slippage=self._slippage * size, pnl=0,
        )

Step 4: 运行测试确认通过

Run: python -m pytest tests/unit/test_backtest_orders.py -v Expected: 全部通过

Step 5: Commit

git add src/easy_tdx/backtest/orders.py tests/unit/test_backtest_orders.py
git commit -m "feat(backtest): add OrderSimulator with 5 execution modes and reject policy"

Task 4: 持仓追踪 — portfolio.py

Files:

Create: src/easy_tdx/backtest/portfolio.py
Create: tests/unit/test_backtest_portfolio.py
Step 1: 写 portfolio.py 的失败测试

# tests/unit/test_backtest_portfolio.py
"""PortfolioTracker 持仓追踪测试。"""

from __future__ import annotations

import numpy as np
import pandas as pd
import pytest

from easy_tdx.backtest.portfolio import PortfolioTracker
from easy_tdx.backtest.types import Trade


def _make_df(n: int = 10) -> pd.DataFrame:
    close = np.full(n, 100.0)
    close[5:] = 110.0  # bar 5 开始涨到 110
    return pd.DataFrame({
        "datetime": range(20240101, 20240101 + n),
        "open": close,
        "close": close,
        "high": close + 1,
        "low": close - 1,
        "vol": np.full(n, 1e6),
        "amount": np.full(n, 1e8),
    })


class TestPortfolioTracker:
    def test_initial_state(self):
        df = _make_df()
        tracker = PortfolioTracker(df, initial_cash=100000)
        assert tracker.initial_cash == 100000

    def test_buy_then_sell(self):
        df = _make_df()
        tracker = PortfolioTracker(df, initial_cash=100000)

        # bar 0 买入 100 股，价格 100
        buy_trade = Trade(datetime=20240101, direction="BUY", size=100, price=100.0, commission=5.0, slippage=0, pnl=0)
        # bar 5 卖出 100 股，价格 110
        sell_trade = Trade(datetime=20240106, direction="SELL", size=100, price=110.0, commission=11.0, slippage=0, pnl=0)

        tracker.apply_trades([buy_trade, sell_trade])

        curve = tracker.equity_curve
        assert len(curve) == 10

        # 最终现金 = 100000 - 100*100 - 5 + 100*110 - 11 = 10984
        final_cash = curve["cash"].iloc[-1]
        assert abs(final_cash - 10984.0) < 1.0

    def test_drawdown_calculation(self):
        df = _make_df()
        tracker = PortfolioTracker(df, initial_cash=100000)
        curve = tracker.equity_curve
        assert "drawdown" in curve.columns
        assert "drawdown_pct" in curve.columns
        # 无交易时 drawdown 为 0
        assert (curve["drawdown"] == 0).all()

    def test_equity_curve_columns(self):
        df = _make_df()
        tracker = PortfolioTracker(df, initial_cash=100000)
        curve = tracker.equity_curve
        for col in ["datetime", "cash", "position_value", "total", "drawdown", "drawdown_pct"]:
            assert col in curve.columns

    def test_position_tracking(self):
        df = _make_df()
        tracker = PortfolioTracker(df, initial_cash=100000)
        buy_trade = Trade(datetime=20240101, direction="BUY", size=100, price=100.0, commission=5.0, slippage=0, pnl=0)
        tracker.apply_trades([buy_trade])

        positions = tracker.positions
        assert len(positions) == 10
        # bar 0 后持仓应为 100
        assert positions["size"].iloc[0] == 100
        assert positions["size"].iloc[-1] == 100

Step 2: 运行测试确认失败

Run: python -m pytest tests/unit/test_backtest_portfolio.py -v Expected: FAIL

Step 3: 实现 portfolio.py

# src/easy_tdx/backtest/portfolio.py
"""持仓追踪器：逐 Bar 计算资金曲线。"""

from __future__ import annotations

import numpy as np
import pandas as pd

from .types import Trade


class PortfolioTracker:
    """逐 Bar 追踪资金曲线。"""

    def __init__(self, df: pd.DataFrame, initial_cash: float = 100000) -> None:
        self._df = df
        self._initial_cash = initial_cash
        self._n = len(df)
        self._close = df["close"].to_numpy(dtype=float)
        self._datetime = df["datetime"].to_numpy()

        # 预分配数组
        self._cash = np.full(self._n, initial_cash)
        self._position = np.zeros(self._n)
        self._avg_price = np.zeros(self._n)

    @property
    def initial_cash(self) -> float:
        return self._initial_cash

    def apply_trades(self, trades: list[Trade]) -> None:
        """将成交记录应用到资金曲线。"""
        # 构建 datetime → trade 的映射
        buy_map: dict[int, Trade] = {}
        sell_map: dict[int, Trade] = {}
        for t in trades:
            if t.rejected:
                continue
            if t.direction == "BUY":
                buy_map[t.datetime] = t
            else:
                sell_map[t.datetime] = t

        cur_cash = self._initial_cash
        cur_pos = 0.0
        cur_avg = 0.0

        for i in range(self._n):
            dt = self._datetime[i]
            dt_int = int(dt.strftime("%Y%m%d")) if hasattr(dt, "strftime") else int(dt)

            # 处理买入
            if dt_int in buy_map:
                t = buy_map[dt_int]
                cost = t.size * t.price + t.commission + t.slippage
                cur_cash -= cost
                # 更新均价
                if cur_pos + t.size > 0:
                    cur_avg = (cur_avg * cur_pos + t.price * t.size) / (cur_pos + t.size)
                cur_pos += t.size

            # 处理卖出
            if dt_int in sell_map:
                t = sell_map[dt_int]
                revenue = t.size * t.price - t.commission - t.slippage
                cur_cash += revenue
                cur_pos -= t.size
                if cur_pos <= 0:
                    cur_pos = 0
                    cur_avg = 0

            self._cash[i] = cur_cash
            self._position[i] = cur_pos
            self._avg_price[i] = cur_avg

    @property
    def equity_curve(self) -> pd.DataFrame:
        """返回资金曲线 DataFrame。"""
        pos_value = self._position * self._close
        total = self._cash + pos_value
        peak = np.maximum.accumulate(total)
        drawdown = peak - total
        drawdown_pct = np.where(peak > 0, drawdown / peak, 0)

        return pd.DataFrame({
            "datetime": self._datetime,
            "cash": self._cash,
            "position_value": pos_value,
            "total": total,
            "drawdown": drawdown,
            "drawdown_pct": drawdown_pct,
        })

    @property
    def positions(self) -> pd.DataFrame:
        """返回持仓历史 DataFrame。"""
        pos_value = self._position * self._close
        unrealized = np.where(
            self._position > 0,
            (self._close - self._avg_price) * self._position,
            0,
        )
        return pd.DataFrame({
            "datetime": self._datetime,
            "size": self._position,
            "avg_price": self._avg_price,
            "market_value": pos_value,
            "unrealized_pnl": unrealized,
        })

Step 4: 运行测试确认通过

Run: python -m pytest tests/unit/test_backtest_portfolio.py -v Expected: 全部通过

Step 5: Commit

git add src/easy_tdx/backtest/portfolio.py tests/unit/test_backtest_portfolio.py
git commit -m "feat(backtest): add PortfolioTracker with equity curve and drawdown"

Task 5: 绩效分析 — performance.py

Files:

Create: src/easy_tdx/backtest/performance.py
Create: tests/unit/test_backtest_performance.py
Step 1: 写 performance.py 的失败测试

# tests/unit/test_backtest_performance.py
"""PerformanceAnalyzer 绩效指标测试。"""

from __future__ import annotations

import numpy as np
import pandas as pd
import pytest

from easy_tdx.backtest.performance import PerformanceAnalyzer


def _make_equity_curve(n: int = 252, total_return: float = 0.1) -> pd.DataFrame:
    """构造资金曲线，总收益率为 total_return。"""
    daily_ret = (1 + total_return) ** (1 / n) - 1
    total = 100000 * np.cumprod(np.full(n, 1 + daily_ret))
    peak = np.maximum.accumulate(total)
    return pd.DataFrame({
        "datetime": pd.date_range("2024-01-01", periods=n, freq="D"),
        "cash": total,
        "position_value": np.zeros(n),
        "total": total,
        "drawdown": peak - total,
        "drawdown_pct": np.where(peak > 0, (peak - total) / peak, 0),
    })


def _make_trades() -> pd.DataFrame:
    """构造交易记录。"""
    return pd.DataFrame({
        "datetime": [20240101, 20240110, 20240120, 20240130],
        "direction": ["BUY", "SELL", "BUY", "SELL"],
        "size": [100, 100, 100, 100],
        "price": [100.0, 105.0, 95.0, 90.0],
        "commission": [5.0, 5.0, 5.0, 5.0],
        "pnl": [0.0, 500.0, 0.0, -500.0],
        "rejected": [False, False, False, False],
    })


class TestPerformanceAnalyzer:
    def test_total_return(self):
        curve = _make_equity_curve(252, 0.1)
        analyzer = PerformanceAnalyzer(curve, _make_trades())
        perf = analyzer.compute()
        assert abs(perf["total_return"] - 0.1) < 0.01

    def test_max_drawdown_zero_when_monotonic(self):
        """单调递增的资金曲线最大回撤为 0。"""
        curve = _make_equity_curve(252, 0.1)
        analyzer = PerformanceAnalyzer(curve, _make_trades())
        perf = analyzer.compute()
        assert perf["max_drawdown"] < 0.001  # 接近 0

    def test_sharpe_positive_for_profit(self):
        curve = _make_equity_curve(252, 0.2)
        analyzer = PerformanceAnalyzer(curve, _make_trades())
        perf = analyzer.compute()
        assert perf["sharpe"] > 0

    def test_win_rate(self):
        trades = pd.DataFrame({
            "direction": ["SELL", "SELL"],
            "pnl": [100.0, -50.0],
            "rejected": [False, False],
        })
        curve = _make_equity_curve(10, 0.0)
        analyzer = PerformanceAnalyzer(curve, trades)
        perf = analyzer.compute()
        assert abs(perf["win_rate"] - 0.5) < 0.01

    def test_total_trades(self):
        curve = _make_equity_curve(10, 0.0)
        analyzer = PerformanceAnalyzer(curve, _make_trades())
        perf = analyzer.compute()
        assert perf["total_trades"] == 4

    def test_empty_trades(self):
        curve = _make_equity_curve(10, 0.0)
        trades = pd.DataFrame(columns=["direction", "pnl", "rejected"])
        analyzer = PerformanceAnalyzer(curve, trades)
        perf = analyzer.compute()
        assert perf["total_trades"] == 0
        assert perf["win_rate"] == 0.0

    def test_all_keys_present(self):
        curve = _make_equity_curve(252, 0.1)
        analyzer = PerformanceAnalyzer(curve, _make_trades())
        perf = analyzer.compute()
        expected_keys = [
            "total_return", "annual_return", "max_drawdown", "max_dd_duration",
            "sharpe", "sortino", "calmar", "total_trades", "win_trades",
            "lose_trades", "rejected_trades", "win_rate", "profit_factor",
            "avg_win", "avg_loss", "max_win", "max_loss", "avg_holding_days",
            "volatility",
        ]
        for key in expected_keys:
            assert key in perf, f"缺少指标: {key}"

Step 2: 运行测试确认失败

Run: python -m pytest tests/unit/test_backtest_performance.py -v Expected: FAIL

Step 3: 实现 performance.py

# src/easy_tdx/backtest/performance.py
"""绩效分析器：从资金曲线和交易记录计算 18 项绩效指标。"""

from __future__ import annotations

import numpy as np
import pandas as pd


class PerformanceAnalyzer:
    """从资金曲线和交易记录计算绩效指标。"""

    ANNUAL_DAYS = 252
    RISK_FREE_RATE = 0.03

    def __init__(
        self,
        equity_curve: pd.DataFrame,
        trades: pd.DataFrame,
        risk_free_rate: float = 0.03,
    ) -> None:
        self._curve = equity_curve
        self._trades = trades
        self._rf = risk_free_rate

    def compute(self) -> dict[str, float]:
        """计算全部绩效指标。"""
        total = self._curve["total"].to_numpy() if not self._curve.empty else np.array([])
        if len(total) < 2:
            return self._empty_metrics()

        # 收益率序列
        daily_ret = np.diff(total) / total[:-1]
        daily_ret = daily_ret[~np.isnan(daily_ret)]

        # 总收益率
        total_return = total[-1] / total[0] - 1

        # 年化收益率
        n_days = len(total)
        annual_return = (1 + total_return) ** (self.ANNUAL_DAYS / n_days) - 1 if n_days > 0 else 0.0

        # 最大回撤
        peak = np.maximum.accumulate(total)
        drawdown = (peak - total) / np.where(peak > 0, peak, 1)
        max_dd = float(np.max(drawdown))

        # 最大回撤天数
        dd_peak_idx = np.argmax(drawdown)
        peak_before = np.argmax(total[:dd_peak_idx + 1]) if dd_peak_idx > 0 else 0
        max_dd_duration = dd_peak_idx - peak_before

        # 夏普比率
        rf_daily = self._rf / self.ANNUAL_DAYS
        if len(daily_ret) > 1 and np.std(daily_ret) > 0:
            sharpe = (np.mean(daily_ret) - rf_daily) / np.std(daily_ret) * np.sqrt(self.ANNUAL_DAYS)
        else:
            sharpe = 0.0

        # 索提诺比率
        neg_ret = daily_ret[daily_ret < 0]
        if len(neg_ret) > 1 and np.std(neg_ret) > 0:
            sortino = (np.mean(daily_ret) - rf_daily) / np.std(neg_ret) * np.sqrt(self.ANNUAL_DAYS)
        else:
            sortino = sharpe

        # 卡玛比率
        calmar = annual_return / max_dd if max_dd > 0 else 0.0

        # 波动率
        volatility = float(np.std(daily_ret) * np.sqrt(self.ANNUAL_DAYS)) if len(daily_ret) > 1 else 0.0

        # 交易统计
        trades = self._trades
        if not trades.empty:
            closed = trades[trades["direction"] == "SELL"]
            total_trades = len(trades)
            rejected_trades = int(trades["rejected"].sum()) if "rejected" in trades.columns else 0

            if not closed.empty and "pnl" in closed.columns:
                pnls = closed["pnl"].to_numpy()
                win_mask = pnls > 0
                lose_mask = pnls <= 0
                win_trades = int(win_mask.sum())
                lose_trades = int(lose_mask.sum())
                win_rate = win_trades / len(closed) if len(closed) > 0 else 0.0
                total_win = float(pnls[win_mask].sum()) if win_mask.any() else 0.0
                total_lose = float(np.abs(pnls[lose_mask].sum())) if lose_mask.any() else 1.0
                profit_factor = total_win / total_lose if total_lose > 0 else float("inf")
                avg_win = float(pnls[win_mask].mean()) if win_mask.any() else 0.0
                avg_loss = float(pnls[lose_mask].mean()) if lose_mask.any() else 0.0
                max_win = float(pnls.max()) if len(pnls) > 0 else 0.0
                max_loss = float(pnls.min()) if len(pnls) > 0 else 0.0
            else:
                win_trades, lose_trades, win_rate = 0, 0, 0.0
                profit_factor, avg_win, avg_loss, max_win, max_loss = 0.0, 0.0, 0.0, 0.0, 0.0

            avg_holding = 5.0  # 简化：后续可精确计算
        else:
            total_trades, rejected_trades = 0, 0
            win_trades, lose_trades, win_rate = 0, 0, 0.0
            profit_factor, avg_win, avg_loss, max_win, max_loss = 0.0, 0.0, 0.0, 0.0, 0.0
            avg_holding = 0.0

        return {
            "total_return": float(total_return),
            "annual_return": float(annual_return),
            "max_drawdown": float(max_dd),
            "max_dd_duration": float(max_dd_duration),
            "sharpe": float(sharpe),
            "sortino": float(sortino),
            "calmar": float(calmar),
            "total_trades": total_trades,
            "win_trades": win_trades,
            "lose_trades": lose_trades,
            "rejected_trades": rejected_trades,
            "win_rate": float(win_rate),
            "profit_factor": float(profit_factor) if profit_factor != float("inf") else 999.0,
            "avg_win": float(avg_win),
            "avg_loss": float(avg_loss),
            "max_win": float(max_win),
            "max_loss": float(max_loss),
            "avg_holding_days": float(avg_holding),
            "volatility": float(volatility),
        }

    def _empty_metrics(self) -> dict[str, float]:
        """空数据时返回零值指标。"""
        return {
            "total_return": 0.0, "annual_return": 0.0, "max_drawdown": 0.0,
            "max_dd_duration": 0.0, "sharpe": 0.0, "sortino": 0.0, "calmar": 0.0,
            "total_trades": 0, "win_trades": 0, "lose_trades": 0, "rejected_trades": 0,
            "win_rate": 0.0, "profit_factor": 0.0, "avg_win": 0.0, "avg_loss": 0.0,
            "max_win": 0.0, "max_loss": 0.0, "avg_holding_days": 0.0, "volatility": 0.0,
        }

Step 4: 运行测试确认通过

Run: python -m pytest tests/unit/test_backtest_performance.py -v Expected: 全部通过

Step 5: Commit

git add src/easy_tdx/backtest/performance.py tests/unit/test_backtest_performance.py
git commit -m "feat(backtest): add PerformanceAnalyzer with 18 metrics"

Task 6: 回测引擎编排 — engine.py

Files:

Create: src/easy_tdx/backtest/engine.py
Create: tests/unit/test_backtest_engine.py
Step 1: 写 engine.py 的失败测试

# tests/unit/test_backtest_engine.py
"""BacktestEngine 端到端测试。"""

from __future__ import annotations

import numpy as np
import pandas as pd
import pytest

from easy_tdx.backtest.engine import BacktestEngine
from easy_tdx.backtest.strategy import Strategy, crossover

from easy_tdx import MyTT


def _make_df(n: int = 100, seed: int = 42) -> pd.DataFrame:
    rng = np.random.default_rng(seed)
    close = 100 + np.cumsum(rng.standard_normal(n) * 0.5)
    high = close + np.abs(rng.standard_normal(n))
    low = close - np.abs(rng.standard_normal(n))
    open_ = low + (high - low) * rng.random(n)
    vol = (rng.random(n) * 1e6).astype(float)
    return pd.DataFrame({
        "datetime": pd.date_range("2024-01-01", periods=n, freq="D"),
        "open": open_,
        "high": high,
        "low": low,
        "close": close,
        "vol": vol,
        "amount": vol * close,
    })


class MACrossStrategy(Strategy):
    """双均线交叉策略。"""

    def init(self):
        self.ma5 = self.I(MyTT.MA, self.data.close, 5)
        self.ma20 = self.I(MyTT.MA, self.data.close, 20)

    def next(self):
        if crossover(self.ma5, self.ma20):
            self.buy(size=0)
        elif crossover(self.ma20, self.ma5):
            self.sell(size=0)


class FixedBuyStrategy(Strategy):
    """固定买入策略（第 5 根 K 线买入 100 股）。"""

    def next(self):
        if self._bar_index == 5:
            self.buy(size=100)
        if self._bar_index == 50:
            self.sell(size=100)


class TestBacktestEngine:
    def test_basic_run(self):
        df = _make_df(200)
        engine = BacktestEngine(strategy=MACrossStrategy, cash=100000)
        result = engine.run(df)

        assert result.performance["total_return"] is not None
        assert len(result.equity_curve) == 200
        assert not result.trades.empty or result.performance["total_trades"] == 0

    def test_fixed_strategy(self):
        df = _make_df(100)
        engine = BacktestEngine(strategy=FixedBuyStrategy, cash=100000)
        result = engine.run(df)

        assert result.performance["total_trades"] >= 2
        trades = result.trades
        buy_trades = trades[trades["direction"] == "BUY"]
        sell_trades = trades[trades["direction"] == "SELL"]
        assert len(buy_trades) >= 1
        assert len(sell_trades) >= 1

    def test_result_columns(self):
        df = _make_df(200)
        engine = BacktestEngine(strategy=MACrossStrategy, cash=100000)
        result = engine.run(df)

        # equity_curve 列
        for col in ["datetime", "cash", "position_value", "total", "drawdown", "drawdown_pct"]:
            assert col in result.equity_curve.columns

        # trades 列
        for col in ["datetime", "direction", "size", "price", "commission", "pnl", "rejected"]:
            assert col in result.trades.columns

    def test_to_dict(self):
        df = _make_df(100)
        engine = BacktestEngine(strategy=FixedBuyStrategy, cash=100000)
        result = engine.run(df)
        d = result.to_dict()
        assert "performance" in d
        assert "config" in d

    def test_chanlun_injection(self):
        """测试缠论结果注入（不验证缠论逻辑，只验证通道可用）。"""

        class ChanlunStrategy(Strategy):
            def init(self):
                pass

            def next(self):
                if self._bar_index == 5 and self.chanlun is not None:
                    self.buy(size=100)
                if self._bar_index == 50:
                    self.sell(size=100)

        df = _make_df(100)
        engine = BacktestEngine(strategy=ChanlunStrategy, cash=100000)
        result = engine.run(df, chanlun_result="fake_result")
        assert result.performance["total_trades"] >= 1

    def test_this_close_warning_in_config(self):
        df = _make_df(100)
        engine = BacktestEngine(strategy=FixedBuyStrategy, cash=100000, execution="this_close")
        result = engine.run(df)
        assert result.config.get("future_leak_warning") is True

    def test_config_snapshot(self):
        df = _make_df(50)
        engine = BacktestEngine(strategy=FixedBuyStrategy, cash=50000, commission=0.001)
        result = engine.run(df)
        assert result.config["cash"] == 50000
        assert result.config["commission"] == 0.001

    def test_precomputed_indicator_columns(self):
        """测试 DataFrame 中预计算指标列可直接在策略中引用。"""

        class BollStrategy(Strategy):
            def init(self):
                pass

            def next(self):
                # 直接访问预计算的 BOLL_UPPER 列
                if hasattr(self.data, "_arrays") and "BOLL_UPPER" in self.data._arrays:
                    upper = self.data._arrays["BOLL_UPPER"]
                    if self._bar_index < len(upper) and self.data.close[0] > upper[self._bar_index]:
                        self.sell(size=0)

        df = _make_df(100)
        df["BOLL_UPPER"] = df["close"] + 5  # 模拟预计算列
        engine = BacktestEngine(strategy=BollStrategy, cash=100000)
        result = engine.run(df)
        # 不崩溃就算通过
        assert result.performance is not None

Step 2: 运行测试确认失败

Run: python -m pytest tests/unit/test_backtest_engine.py -v Expected: FAIL

Step 3: 实现 engine.py

# src/easy_tdx/backtest/engine.py
"""回测引擎：向量化执行管道。"""

from __future__ import annotations

from typing import Any

import numpy as np
import pandas as pd

from .orders import OrderSimulator
from .performance import PerformanceAnalyzer
from .portfolio import PortfolioTracker
from .strategy import Strategy
from .types import BacktestResult, Signal


class BacktestEngine:
    """回测引擎主入口。纯计算，接收 DataFrame，输出 BacktestResult。"""

    def __init__(
        self,
        strategy: type[Strategy] | Strategy,
        cash: float = 100000.0,
        commission: float = 0.0003,
        min_commission: float = 5.0,
        stamp_tax: float = 0.001,
        slippage: float = 0.0,
        execution: str = "next_open",
        position_mode: str = "full",
        reject_policy: str = "reduce",
        benchmark: pd.DataFrame | None = None,
    ) -> None:
        if isinstance(strategy, type):
            self._strategy = strategy()
        else:
            self._strategy = strategy
        self._cash = cash
        self._commission = commission
        self._min_commission = min_commission
        self._stamp_tax = stamp_tax
        self._slippage = slippage
        self._execution = execution
        self._position_mode = position_mode
        self._reject_policy = reject_policy
        self._benchmark = benchmark

    def run(self, df: pd.DataFrame, chanlun_result: Any | None = None) -> BacktestResult:
        """执行回测。"""
        n = len(df)
        if n == 0:
            return self._empty_result()

        # Step 1: 信号生成
        signals = self._generate_signals(df, chanlun_result)

        # Step 2: 撮合
        simulator = OrderSimulator(
            df,
            execution=self._execution,
            position_mode=self._position_mode,
            reject_policy=self._reject_policy,
            commission=self._commission,
            min_commission=self._min_commission,
            stamp_tax=self._stamp_tax,
            slippage=self._slippage,
        )
        trades = simulator.simulate(signals, cash=self._cash, position=0.0)

        # Step 3: 持仓追踪
        tracker = PortfolioTracker(df, initial_cash=self._cash)
        # 计算每笔卖出的 pnl
        trades = self._compute_pnls(trades)
        tracker.apply_trades(trades)

        # Step 4: 绩效分析
        trades_df = self._trades_to_df(trades)
        analyzer = PerformanceAnalyzer(tracker.equity_curve, trades_df)
        performance = analyzer.compute()

        # 配置快照
        config: dict[str, Any] = {
            "cash": self._cash,
            "commission": self._commission,
            "min_commission": self._min_commission,
            "stamp_tax": self._stamp_tax,
            "slippage": self._slippage,
            "execution": self._execution,
            "position_mode": self._position_mode,
            "reject_policy": self._reject_policy,
            "future_leak_warning": simulator.future_leak_warning,
        }

        return BacktestResult(
            performance=performance,
            equity_curve=tracker.equity_curve,
            trades=trades_df,
            positions=tracker.positions,
            config=config,
        )

    def _generate_signals(self, df: pd.DataFrame, chanlun_result: Any | None) -> list[Signal]:
        """通过 Strategy 生成交易信号。"""
        strat = self._strategy
        strat._bind_data(df)
        if chanlun_result is not None:
            strat._chanlun_result = chanlun_result
        strat._call_init()

        all_signals: list[Signal] = []
        for i in range(len(df)):
            strat._set_bar_index(i)
            strat._call_next()
            bar_signals = strat._clear_signals()
            all_signals.extend(bar_signals)

        return all_signals

    def _compute_pnls(self, trades: list) -> list:
        """计算每笔卖出的 pnl。买入记录 pnl=0。"""
        position_cost = 0.0
        position_size = 0.0
        for t in trades:
            if t.rejected:
                continue
            if t.direction == "BUY":
                position_cost += t.size * t.price + t.commission
                position_size += t.size
                t.pnl = 0.0
            else:
                if position_size > 0:
                    avg_cost = position_cost / position_size
                    t.pnl = (t.price - avg_cost) * t.size - t.commission
                    position_cost -= avg_cost * t.size
                    position_size -= t.size
                else:
                    t.pnl = 0.0
        return trades

    def _trades_to_df(self, trades: list) -> pd.DataFrame:
        """将 Trade 列表转为 DataFrame。"""
        if not trades:
            return pd.DataFrame(columns=["datetime", "direction", "size", "price", "commission", "pnl", "rejected"])
        return pd.DataFrame([
            {
                "datetime": t.datetime,
                "direction": t.direction,
                "size": t.size,
                "price": t.price,
                "commission": t.commission,
                "pnl": t.pnl,
                "rejected": t.rejected,
            }
            for t in trades
        ])

    def _empty_result(self) -> BacktestResult:
        return BacktestResult(
            performance={},
            equity_curve=pd.DataFrame(),
            trades=pd.DataFrame(),
            positions=pd.DataFrame(),
            config={"cash": self._cash},
        )

Step 4: 运行测试确认通过

Run: python -m pytest tests/unit/test_backtest_engine.py -v Expected: 全部通过

Step 5: 运行全部回测测试

Run: python -m pytest tests/unit/test_backtest_*.py -v Expected: 全部通过

Step 6: Commit

git add src/easy_tdx/backtest/engine.py tests/unit/test_backtest_engine.py
git commit -m "feat(backtest): add BacktestEngine with vectorized execution pipeline"

Task 7: DSL 骨架 — dsl.py (P1 骨架)

Files:

Create: src/easy_tdx/backtest/dsl.py
Step 1: 创建 dsl.py 骨架文件

# src/easy_tdx/backtest/dsl.py
"""DSL 策略定义模块 (P1 — 骨架)。

v1 提供 @dsl_strategy 装饰器的基本实现。
字符串 DSL 解析器将在后续版本实现。
"""

from __future__ import annotations

from collections.abc import Callable
from typing import Any

import numpy as np
import pandas as pd

from .strategy import Strategy


def dsl_strategy(func: Callable[..., Any]) -> type[Strategy]:
    """将函数编译为 Strategy 子类。

    函数签名: (df: pd.DataFrame) -> tuple[np.ndarray[bool], np.ndarray[bool]]
    返回 (buy_mask, sell_mask)。

    用法::

        @dsl_strategy
        def dual_ma(df):
            buy = CROSS(MA(df.close, 5), MA(df.close, 20))
            sell = CROSS(MA(df.close, 20), MA(df.close, 5))
            return buy, sell
    """

    class DSLStrategy(Strategy):
        _signal_func = staticmethod(func)
        _buy_mask: np.ndarray | None = None
        _sell_mask: np.ndarray | None = None

        def init(self) -> None:
            pass

        def next(self) -> None:
            if self._buy_mask is None:
                return
            idx = self._bar_index
            if idx < len(self._buy_mask) and self._buy_mask[idx]:
                self.buy(size=0)
            elif idx < len(self._sell_mask) and self._sell_mask[idx]:
                self.sell(size=0)

    DSLStrategy.__name__ = func.__name__
    DSLStrategy.__qualname__ = func.__qualname__

    # 存储 signal 函数引用，引擎会在 run 时调用
    DSLStrategy._signal_func = func  # type: ignore[attr-defined]

    return DSLStrategy

Step 2: Commit

git add src/easy_tdx/backtest/dsl.py
git commit -m "feat(backtest): add DSL strategy skeleton (P1)"

Task 8: 更新 `init.py` 导出

Files:

Modify: src/easy_tdx/backtest/__init__.py
Step 1: 更新 __init__.py

# src/easy_tdx/backtest/__init__.py
"""easy_tdx.backtest — 向量化策略回测引擎（纯计算，零网络依赖）。

快速开始::

    from easy_tdx.backtest import BacktestEngine, Strategy

    class MyStrategy(Strategy):
        def init(self):
            self.ma5 = self.I(MA, self.data.close, 5)
            self.ma20 = self.I(MA, self.data.close, 20)

        def next(self):
            if crossover(self.ma5, self.ma20):
                self.buy()
            elif crossover(self.ma20, self.ma5):
                self.sell()

    engine = BacktestEngine(strategy=MyStrategy, cash=100000)
    result = engine.run(df)
    print(result.performance)
"""

from easy_tdx.backtest.engine import BacktestEngine  # noqa: F401
from easy_tdx.backtest.strategy import Strategy, StrategyDataProxy, crossover  # noqa: F401
from easy_tdx.backtest.types import BacktestResult, Position, Signal, Trade  # noqa: F401

__all__ = [
    "BacktestEngine",
    "BacktestResult",
    "Strategy",
    "StrategyDataProxy",
    "Signal",
    "Trade",
    "Position",
    "crossover",
]

Step 2: Commit

git add src/easy_tdx/backtest/__init__.py
git commit -m "feat(backtest): update __init__.py with public API exports"

Task 9: CLI 集成 — cli.py

Files:

Create: src/easy_tdx/backtest/cli.py
Create: tests/unit/test_backtest_cli.py
Modify: src/easy_tdx/cli/__init__.py
Step 1: 写 CLI 失败测试

# tests/unit/test_backtest_cli.py
"""backtest CLI 命令测试。"""

from __future__ import annotations

from click.testing import CliRunner

import pytest


class TestBacktestCLI:
    def test_help(self):
        from easy_tdx.cli import cli
        runner = CliRunner()
        result = runner.invoke(cli, ["backtest", "--help"])
        assert result.exit_code == 0
        assert "strategy" in result.output.lower() or "strategy-file" in result.output.lower()

    def test_missing_strategy_fails(self):
        from easy_tdx.cli import cli
        runner = CliRunner()
        result = runner.invoke(cli, ["backtest", "SH", "600519"])
        assert result.exit_code != 0

Step 2: 运行测试确认失败

Run: python -m pytest tests/unit/test_backtest_cli.py -v Expected: FAIL — No command 'backtest'

Step 3: 实现 cli.py

# src/easy_tdx/backtest/cli.py
"""回测 CLI 命令。"""

from __future__ import annotations

import importlib.util
import json
import sys

import click


@click.command()
@click.argument("market")
@click.argument("code")
@click.option("--strategy", "strategy_str", default=None, help="DSL 策略表达式 (P1)")
@click.option("--strategy-file", "strategy_file", default=None, help="Python 策略文件路径")
@click.option("--cash", default=100000.0, type=float, help="初始资金")
@click.option("--commission", default=0.0003, type=float, help="佣金率")
@click.option("--execution", default="next_open", type=click.Choice(["next_open", "next_close", "this_close", "worst", "best"]), help="成交价规则")
@click.option("--period", default="DAILY", help="K线周期: DAILY/5MIN/15MIN/30MIN/60MIN/1MIN/WEEKLY/MONTHLY")
@click.option("--adjust", default="NONE", help="复权: NONE/QFQ/HFQ")
@click.option("--count", default=500, type=int, help="K线数量")
@click.option("--indicators", default=None, help="预计算指标（逗号分隔，如 MACD,KDJ）")
@click.option("--table", "use_table", is_flag=True, help="表格输出")
@click.option("--output", "output_fmt", type=click.Choice(["json", "table", "csv"]), default="json")
def backtest(
    market: str,
    code: str,
    strategy_str: str | None,
    strategy_file: str | None,
    cash: float,
    commission: float,
    execution: str,
    period: str,
    adjust: str,
    count: int,
    indicators: str | None,
    use_table: bool,
    output_fmt: str,
) -> None:
    """策略回测：获取 K 线数据 + 执行回测 + 输出绩效报告。

    示例：

      easy-tdx backtest SH 600519 --strategy-file ma_cross.py --cash 100000 --table

      easy-tdx backtest SH 600519 --strategy-file ma_cross.py --indicators MACD,KDJ
    """
    from .engine import BacktestEngine
    from ..cli.conn import get_mac_client
    from ..cli.parsers import parse_adjust, parse_market, parse_period
    from ..indicator import compute_indicators

    # 确定策略
    strategy_cls = _load_strategy(strategy_str, strategy_file)

    # 获取数据
    mkt = parse_market(market)
    with get_mac_client() as client:
        df = client.get_stock_kline(
            mkt,
            code,
            period=parse_period(period),
            start=0,
            count=count,
            adjust=parse_adjust(adjust),
        )

        # 预计算指标
        if indicators:
            ind_list = [i.strip().upper() for i in indicators.split(",")]
            df = compute_indicators(df, ind_list, keep_ohlcv=True)

    # 执行回测
    engine = BacktestEngine(
        strategy=strategy_cls,
        cash=cash,
        commission=commission,
        execution=execution,
    )
    result = engine.run(df)

    # 输出
    fmt = "table" if use_table else output_fmt
    if fmt == "json":
        click.echo(json.dumps(result.to_dict(), ensure_ascii=False, indent=2, default=str))
    elif fmt == "csv":
        click.echo(result.trades.to_csv(index=False))
    else:
        _print_table(result)


def _load_strategy(strategy_str: str | None, strategy_file: str | None) -> type:
    """加载策略类。"""
    if strategy_file:
        return _load_strategy_from_file(strategy_file)

    if strategy_str:
        # P1: DSL 字符串解析暂未实现，提示用户使用 --strategy-file
        raise click.UsageError(
            "DSL 字符串策略暂未实现 (P1)。请使用 --strategy-file 指定 Python 策略文件。"
        )

    raise click.UsageError("必须指定 --strategy 或 --strategy-file")


def _load_strategy_from_file(path: str) -> type:
    """从 Python 文件加载 Strategy 子类。"""
    import os

    if not os.path.exists(path):
        raise click.UsageError(f"策略文件不存在: {path}")

    from .strategy import Strategy

    spec = importlib.util.spec_from_file_location("user_strategy", path)
    if spec is None or spec.loader is None:
        raise click.UsageError(f"无法加载策略文件: {path}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)

    # 查找 Strategy 子类
    strategy_classes = [
        obj for obj in vars(mod).values()
        if isinstance(obj, type) and issubclass(obj, Strategy) and obj is not Strategy
    ]
    if not strategy_classes:
        raise click.UsageError(f"策略文件中未找到 Strategy 子类: {path}")

    return strategy_classes[0]


def _print_table(result) -> None:
    """以表格形式输出回测结果。"""
    p = result.performance
    c = result.config

    click.echo(f"初始资金: ¥{c.get('cash', 0):,.0f}  执行模式: {c.get('execution', 'next_open')}")
    if c.get("future_leak_warning"):
        click.echo("⚠️  警告: 使用了 this_close 模式，存在未来函数风险！")
    click.echo()

    click.echo("═══ 绩效概要 ═══")
    click.echo(f"总收益率:     {p.get('total_return', 0):.2%}        年化收益率:   {p.get('annual_return', 0):.2%}")
    click.echo(f"最大回撤:     {p.get('max_drawdown', 0):.2%}        夏普比率:      {p.get('sharpe', 0):.2f}")
    click.echo(f"索提诺比率:   {p.get('sortino', 0):.2f}        卡玛比率:      {p.get('calmar', 0):.2f}")
    click.echo(f"收益波动率:   {p.get('volatility', 0):.2%}")
    click.echo()

    click.echo("═══ 交易统计 ═══")
    click.echo(f"总交易:       {p.get('total_trades', 0)} 次          胜率:          {p.get('win_rate', 0):.2%}")
    click.echo(f"盈利交易:     {p.get('win_trades', 0)} 次          亏损交易:      {p.get('lose_trades', 0)} 次")
    click.echo(f"盈亏比:       {p.get('profit_factor', 0):.2f}          平均盈利:      ¥{p.get('avg_win', 0):,.0f}")
    click.echo(f"平均亏损:     ¥{p.get('avg_loss', 0):,.0f}        最大单笔盈利:  ¥{p.get('max_win', 0):,.0f}")
    click.echo(f"最大单笔亏损: ¥{p.get('max_loss', 0):,.0f}        平均持仓:      {p.get('avg_holding_days', 0):.1f} 天")

    if not result.trades.empty:
        click.echo()
        click.echo("═══ 最近交易 ═══")
        recent = result.trades.tail(10)
        try:
            import tabulate
            click.echo(tabulate.tabulate(recent, headers="keys", tablefmt="grid", showindex=False))
        except ImportError:
            for _, row in recent.iterrows():
                click.echo(f"  {row['datetime']} {row['direction']} {row['size']}@{row['price']:.2f} pnl={row['pnl']:.2f}")

Step 4: 注册命令到 CLI

在 src/easy_tdx/cli/__init__.py 中添加：

# 在 import 区域添加
from .backtest_cmd import backtest  # 暂用别名避免冲突

# 在 cli.add_command 区域添加
cli.add_command(backtest)

实际上 backtest 命令定义在 src/easy_tdx/backtest/cli.py，需要在 src/easy_tdx/cli/__init__.py 中导入：

在 src/easy_tdx/cli/__init__.py 的 import 区域添加：

from easy_tdx.backtest.cli import backtest

在 cli.add_command 区域添加：

cli.add_command(backtest)

Step 5: 运行 CLI 测试

Run: python -m pytest tests/unit/test_backtest_cli.py -v Expected: 全部通过

Step 6: 手动验证 CLI help

Run: python -m easy_tdx.cli backtest --help Expected: 显示 backtest 命令帮助

Step 7: Commit

git add src/easy_tdx/backtest/cli.py src/easy_tdx/cli/__init__.py tests/unit/test_backtest_cli.py
git commit -m "feat(backtest): add CLI command with auto data fetch and table output"

Task 10: 最终验证 + 清理

Files:

Modify: src/easy_tdx/__init__.py (可选)
Step 1: 运行全部回测测试

Run: python -m pytest tests/unit/test_backtest_*.py -v Expected: 全部通过

Step 2: 运行项目全部单元测试（确认无回归）

Run: python -m pytest tests/unit/ -v Expected: 全部通过

Step 3: 运行 mypy 类型检查

Run: mypy src/easy_tdx/backtest/ Expected: 无错误（可能需要微调类型注解）

Step 4: 运行 ruff lint

Run: ruff check src/easy_tdx/backtest/ tests/unit/test_backtest_*.py Expected: 无错误（可能需要微调格式）

Step 5: 可选 — 更新顶层 __init__.py 导出

在 src/easy_tdx/__init__.py 的 __all__ 中添加（可选）：

# 在 import 区域
from .backtest import BacktestEngine, Strategy  # noqa: F401

Step 6: Final commit

git add -A
git commit -m "feat(backtest): final cleanup and type checking"

Self-Review Checklist

1. Spec Coverage

Spec 要求	对应 Task
Signal/Trade/Position/BacktestResult 数据类型	Task 1
Strategy 基类 + init()/next()	Task 2
StrategyDataProxy + 预计算列	Task 2
crossover() 辅助函数	Task 2
缠论预留 self.chanlun	Task 2 + Task 6
DSL @dsl_strategy 装饰器	Task 7 (P1 骨架)
5 种执行模式	Task 3
仓位管理 (full/fixed/percent)	Task 3
订单拒绝策略 (reduce/skip)	Task 3
费用模型 (佣金/印花税/滑点)	Task 3
this_close 未来函数警告	Task 3
PortfolioTracker 资金曲线	Task 4
回撤计算	Task 4
18 项绩效指标	Task 5
BacktestEngine 四步管道	Task 6
PnL 计算	Task 6
CLI 命令 + 自动获取 K 线	Task 9
--indicators 预计算	Task 9
CLI JSON/table/csv 输出	Task 9

2. Placeholder Scan

✅ 无 TBD/TODO
✅ 所有测试包含实际断言
✅ 所有实现包含完整代码

3. Type Consistency

Signal.direction: Literal["BUY", "SELL"] — types.py ↔ strategy.py ↔ orders.py 一致 ✅
Trade.rejected: bool — types.py ↔ orders.py ↔ engine.py 一致 ✅
BacktestResult.performance: dict[str, float] — types.py ↔ engine.py ↔ performance.py 一致 ✅
crossover() 签名: strategy.py 定义，test_backtest_strategy.py 和 test_backtest_engine.py 引用一致 ✅

76 KiB Raw Blame History Unescape Escape

Backtest Engine Implementation Plan

File Structure

Task 1: 数据类型 — types.py

Task 2: 策略基类 + 数据代理 — strategy.py

Task 3: 撮合规则 — orders.py

Task 4: 持仓追踪 — portfolio.py

Task 5: 绩效分析 — performance.py

Task 6: 回测引擎编排 — engine.py

Task 7: DSL 骨架 — dsl.py (P1 骨架)

Task 8: 更新 __init__.py 导出

Task 9: CLI 集成 — cli.py

Task 10: 最终验证 + 清理

Self-Review Checklist

1. Spec Coverage

2. Placeholder Scan

3. Type Consistency

76 KiB

Raw Blame History

Task 8: 更新 `init.py` 导出