trade/engine/example/backtest_demo.py

"""
回测引擎使用示例 — 双策略演示

用法：
    source .venv/bin/activate && python example/backtest_demo.py
"""

import asyncio
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

_project_root = Path(__file__).resolve().parent.parent.parent
if str(_project_root) not in sys.path:
    sys.path.insert(0, str(_project_root))

from engine.common.base import BaseStrategy, Signal, StrategyConfig
from engine.common.models import Kline
from engine.common.config import config
from engine.backtest import BacktestEngine, BacktestConfig
from engine.indicators import sma, rsi


# ============================================================
# 策略 1：双均线交叉
# ============================================================


class MACrossConfig(StrategyConfig):
    fast_period: int = 7
    slow_period: int = 25


class MACrossStrategy(BaseStrategy):
    """双均线交叉策略 — 使用 engine.indicators.sma 计算均线"""

    strategy_type = "ma_cross"

    def __init__(self, config: MACrossConfig):
        super().__init__(config)
        self.config: MACrossConfig = config
        self._closes: list[float] = []
        self._last_signal: Optional[str] = None

    async def on_kline(self, kline: Kline) -> Optional[Signal]:
        self._closes.append(kline.close)

        # 使用指标库计算 SMA
        fast_ma = sma(self._closes, self.config.fast_period)
        slow_ma = sma(self._closes, self.config.slow_period)
        fast = fast_ma[-1]
        slow = slow_ma[-1]

        if fast == 0.0 or slow == 0.0:
            return None

        if fast > slow and self._last_signal != "BUY":
            self._last_signal = "BUY"
            return Signal(
                symbol=self.config.symbol,
                side="BUY",
                signal_type="MARKET",
                confidence=0.8,
                reason=f"金叉 MA{self.config.fast_period}>{self.config.slow_period}",
                timestamp=kline.open_time,
            )

        if fast < slow and self._last_signal != "SELL":
            self._last_signal = "SELL"
            return Signal(
                symbol=self.config.symbol,
                side="SELL",
                signal_type="MARKET",
                confidence=0.8,
                reason=f"死叉 MA{self.config.fast_period}<{self.config.slow_period}",
                timestamp=kline.open_time,
            )

        return None


# ============================================================
# 策略 2：RSI 超买超卖
# ============================================================


class RSIStrategyConfig(StrategyConfig):
    period: int = 14
    oversold: float = 30.0   # 超卖阈值
    overbought: float = 70.0  # 超买阈值


class RSIStrategy(BaseStrategy):
    """RSI 超买超卖策略 — 使用 engine.indicators.rsi 计算 RSI

    RSI 低于超卖线 → 买入；RSI 高于超买线 → 卖出。
    """

    strategy_type = "rsi"

    def __init__(self, config: RSIStrategyConfig):
        super().__init__(config)
        self.config: RSIStrategyConfig = config
        self._closes: list[float] = []
        self._has_position = False

    async def on_kline(self, kline: Kline) -> Optional[Signal]:
        self._closes.append(kline.close)

        # 使用指标库计算 RSI
        rsi_vals = rsi(self._closes, self.config.period)
        current_rsi = rsi_vals[-1]

        if current_rsi == 0.0:
            return None

        # 超卖 → 买入
        if current_rsi < self.config.oversold and not self._has_position:
            self._has_position = True
            return Signal(
                symbol=self.config.symbol,
                side="BUY",
                signal_type="MARKET",
                confidence=0.7,
                reason=f"RSI超卖 ({current_rsi:.1f} < {self.config.oversold})",
                timestamp=kline.open_time,
            )

        # 超买 → 卖出
        if current_rsi > self.config.overbought and self._has_position:
            self._has_position = False
            return Signal(
                symbol=self.config.symbol,
                side="SELL",
                signal_type="MARKET",
                confidence=0.7,
                reason=f"RSI超买 ({current_rsi:.1f} > {self.config.overbought})",
                timestamp=kline.open_time,
            )

        return None


# ============================================================
# 主函数
# ============================================================


async def run_backtest(
    engine: BacktestEngine,
    strategy_cls,
    strategy_config: StrategyConfig,
    label: str,
):
    """运行一次回测并打印结果"""
    print(f"\n{'━' * 60}")
    print(f"  {label}")
    print(f"{'━' * 60}")

    result = await engine.run(strategy_cls, strategy_config)
    print(result.summary())

    # 最近 5 笔交易
    if result.trades:
        print(f"\n  最近 5 笔交易:")
        print(f"  {'时间':<22} {'方向':<6} {'价格':>10} {'数量':>10} {'盈亏':>10}  原因")
        for t in result.trades[-5:]:
            dt = datetime.fromtimestamp(t.timestamp / 1000, tz=timezone.utc).strftime("%Y-%m-%d %H:%M")
            pnl_str = f"{t.pnl:+.4f}" if t.pnl is not None else "—"
            print(f"  {dt:<22} {t.side:<6} {t.price:>10.4f} {t.quantity:>10.6f} {pnl_str:>10}  {t.reason}")

    return result


async def main():
    # ── 回测配置 ──
    bt_config = BacktestConfig(
        symbol="ETHUSDT",
        interval="4h",
        start_time=datetime(2024, 1, 1),
        end_time=datetime(2026, 1, 1),
        initial_capital=10_000.0,
        commission_pct=0.001,
        slippage_pct=0.0005,
        warmup_bars=100,
    )

    print(f"\n回测: {bt_config.symbol} {bt_config.interval}")
    print(f"时间: {bt_config.start_time.date()} ~ {bt_config.end_time.date()}")
    print(f"初始资金: {bt_config.initial_capital:.2f} USDT")

    engine = BacktestEngine(bt_config, db_config=config.db)

    # ── 策略 1：双均线交叉 ──
    ma_config = MACrossConfig(
        name="ma_cross_eth",
        symbol="ETHUSDT",
        fast_period=7,
        slow_period=25,
    )
    await run_backtest(engine, MACrossStrategy, ma_config, "策略 1：双均线交叉 (MA7/MA25)")

    # ── 策略 2：RSI 超买超卖 ──
    rsi_config = RSIStrategyConfig(
        name="rsi_eth",
        symbol="ETHUSDT",
        period=14,
        oversold=30.0,
        overbought=70.0,
    )
    await run_backtest(engine, RSIStrategy, rsi_config, "策略 2：RSI 超买超卖 (30/70)")

    print("\n全部回测完成。")


if __name__ == "__main__":
    asyncio.run(main())