Pandas 与 NumPy 策略回测

Tick 数据（每笔交易） → 分钟 K 线 → 小时 K 线 → 日 K 线 → 周 K 线

import pandas as pd
import numpy as np
from os import path


def assert_msg(condition, msg):
    """自定义断言，抛出带信息的异常"""
    if not condition:
        raise Exception(msg)


def read_file(filename):
    """读取 CSV 格式的 OHLCV 数据并返回 DataFrame"""
    filepath = path.join(path.dirname(__file__), filename)

    # 判定文件是否存在
    assert_msg(path.exists(filepath), f"文件不存在: {filepath}")

    # 读取 CSV 文件
    df = pd.read_csv(
        filepath,
        index_col=0,              # 第一列作为索引（通常是日期）
        parse_dates=True,         # 自动解析日期
        infer_datetime_format=True
    )

    # 验证必要列存在
    required_cols = {'Open', 'High', 'Low', 'Close'}
    missing = required_cols - set(df.columns)
    assert_msg(len(missing) == 0, f"缺少必要列: {missing}")

    # 验证无缺失值
    assert_msg(
        not df[['Open', 'High', 'Low', 'Close']].max().isnull().any(),
        'OHLC 数据包含缺失值，请清洗后再使用'
    )

    # 确保按时间升序排列
    if not df.index.is_monotonic_increasing:
        df = df.sort_index()

    return df


# 使用示例
# BTCUSD = read_file('BTCUSD_GEMINI.csv')
# print(BTCUSD.head())

from numbers import Number


class Backtest:
    """
    向量化回测引擎。

    参数:
        data:           OHLCV 历史数据
        strategy_type:  策略类（Strategy 的子类）
        broker_type:    交易所模拟类（ExchangeAPI 的子类）
        cash:           初始资金
        commission:     手续费率（如 0.002 表示千分之二）
    """

    def __init__(self,
                 data: pd.DataFrame,
                 strategy_type: type,
                 broker_type: type,
                 cash: float = 10000,
                 commission: float = .0):
        # 类型校验
        assert_msg(issubclass(strategy_type, Strategy),
                   'strategy_type 必须是 Strategy 的子类')
        assert_msg(issubclass(broker_type, ExchangeAPI),
                   'broker_type 必须是 ExchangeAPI 的子类')
        assert_msg(isinstance(commission, Number),
                   'commission 必须是数值类型')

        data = data.copy(False)

        # 如果没有 Volume 列，填充 NaN
        if 'Volume' not in data:
            data['Volume'] = np.nan

        # 验证 OHLC 数据格式
        required = {'Open', 'High', 'Low', 'Close', 'Volume'}
        assert_msg(
            len(data.columns & required) == 5,
            f"数据格式不正确，至少需要包含: {required}"
        )

        # 初始化交易所和策略
        self._data = data
        self._broker = broker_type(data, cash, commission)
        self._strategy = strategy_type(self._broker, self._data)
        self._results = None

    def run(self):
        """运行回测，迭代历史数据并返回结果"""
        strategy = self._strategy
        broker = self._broker

        # 策略初始化（计算指标）
        strategy.init()

        # 回测主循环
        start = 100  # 跳过前 100 条数据（留给指标计算足够的 warm-up）
        end = len(self._data)

        for i in range(start, end):
            broker.next(i)       # 更新市场价格到第 i 时刻
            strategy.next(i)     # 策略根据第 i 时刻的市场做决策

        # 计算并返回结果
        self._results = self._compute_result(broker)
        return self._results

    def _compute_result(self, broker):
        """汇总回测结果"""
        result = pd.Series(dtype=float)
        result['初始市值'] = broker.initial_cash
        result['结束市值'] = broker.market_value
        result['收益'] = broker.market_value - broker.initial_cash
        result['收益率'] = (broker.market_value / broker.initial_cash - 1)
        return result

import abc
from typing import Callable


class Strategy(metaclass=abc.ABCMeta):
    """
    策略抽象基类。所有具体策略必须继承此类并实现 init() 和 next()。

    使用方式:
        class MyStrategy(Strategy):
            def init(self):
                # 计算指标
                self.sma = self.I(SMA, self.data.Close, 20)

            def next(self, tick):
                # 根据指标决定买卖
                if 买入条件:
                    self.buy()
                elif 卖出条件:
                    self.sell()
    """

    def __init__(self, broker, data):
        self._indicators = []
        self._broker = broker
        self._data = data
        self._tick = 0

    def I(self, func: Callable, *args) -> np.ndarray:
        """
        计算指标向量。指标是一个与历史数据等长的数组。

        示例:
            self.sma = self.I(utils.SMA, self.data.Close, 20)
        """
        value = func(*args)
        value = np.asarray(value)
        assert_msg(
            value.shape[-1] == len(self._data.Close),
            f'指标长度 ({value.shape[-1]}) 必须与数据长度 ({len(self._data.Close)}) 相同'
        )
        self._indicators.append(value)
        return value

    @property
    def tick(self):
        return self._tick

    @abc.abstractmethod
    def init(self):
        """初始化策略。在回测开始时调用一次，用于计算指标。"""
        pass

    @abc.abstractmethod
    def next(self, tick):
        """
        步进函数。在每个时间点调用，根据当前市场状态决定买卖。

        参数:
            tick: 当前时间索引，data[tick] 可访问当前市场数据
        """
        pass

    def buy(self):
        """买入信号：用全部现金按市价买入"""
        self._broker.buy()

    def sell(self):
        """卖出信号：卖出全部持仓"""
        self._broker.sell()

    @property
    def data(self):
        return self._data

def SMA(values, n):
    """计算简单移动平均"""
    return pd.Series(values).rolling(n).mean()


def crossover(series1, series2) -> bool:
    """
    检测 series1 是否在最后一个位置从下方向上穿越 series2。

    判断逻辑:
        series1[-2] < series2[-2]  # 前一刻: series1 在 series2 下方
        series1[-1] > series2[-1]  # 当前刻: series1 在 series2 上方
        → 发生了上穿
    """
    return series1[-2] < series2[-2] and series1[-1] > series2[-1]

class SmaCross(Strategy):
    """移动均线交叉策略"""

    # 小窗口 SMA 的窗口大小（快线）
    fast = 10

    # 大窗口 SMA 的窗口大小（慢线）
    slow = 20

    def init(self):
        # 预计算历史上每个时刻的快线和慢线
        self.sma1 = self.I(SMA, self.data.Close, self.fast)
        self.sma2 = self.I(SMA, self.data.Close, self.slow)

    def next(self, tick):
        # 快线上穿慢线 → 金叉 → 买入
        if crossover(self.sma1[:tick], self.sma2[:tick]):
            self.buy()

        # 慢线上穿快线 → 死叉 → 卖出
        elif crossover(self.sma2[:tick], self.sma1[:tick]):
            self.sell()

        # 否则持仓不动
        else:
            pass

class ExchangeAPI:
    """
    模拟交易所，负责维护账户状态和执行模拟交易。

    简化假设:
        - 使用 Close 价格作为当前市价
        - 买卖使用全部资金/仓位
        - 市场容量足够大，订单立即完全成交
    """

    def __init__(self, data, cash, commission):
        assert_msg(cash > 0, f"初始现金必须大于 0，当前: {cash}")
        assert_msg(
            0 <= commission <= 0.05,
            f"手续费率应在 0~5% 之间，当前: {commission}"
        )
        self._initial_cash = cash
        self._data = data
        self._commission = commission
        self._position = 0.0   # 当前持仓数量
        self._cash = cash       # 当前现金
        self._i = 0             # 当前时间索引

    # ---- 属性 ----
    @property
    def cash(self):
        return self._cash

    @property
    def position(self):
        return self._position

    @property
    def initial_cash(self):
        return self._initial_cash

    @property
    def current_price(self):
        """当前市场价格（使用 Close 价格）"""
        return self._data.Close[self._i]

    @property
    def market_value(self):
        """当前总市值 = 现金 + 持仓市值"""
        return self._cash + self._position * self.current_price

    # ---- 交易操作 ----
    def buy(self):
        """
        用当前账户全部现金，按市价买入。

        买入数量 = 现金 / (价格 × (1 + 手续费率))
        """
        self._position = float(
            self._cash / (self.current_price * (1 + self._commission))
        )
        self._cash = 0.0

    def sell(self):
        """
        卖出当前账户全部持仓。

        卖出收益 = 持仓数量 × 价格 × (1 - 手续费率)
        """
        self._cash += float(
            self._position * self.current_price * (1 - self._commission)
        )
        self._position = 0.0

    def next(self, tick):
        """推进到下一个时间点"""
        self._i = tick

# 模拟手续费对策略收益的影响
# 假设每次交易手续费率为千分之三 (0.3%)

# 如果每天交易 1 次，一年 365 次
daily_trades = 1
annual_trades = 365
commission_rate = 0.003

# 每次买卖都要交手续费（买入 + 卖出 = 2 次）
annual_commission_cost = (1 + commission_rate) ** (annual_trades * 2) - 1
print(f"年化手续费损耗: {annual_commission_cost:.1%}")
# 典型输出: 年化手续费损耗: ~800%

# 结论：高频策略必须严格控制交易频率和手续费

def main():
    # 读取数据
    BTCUSD = read_file('BTCUSD_GEMINI.csv')

    # 手续费率为 0 的回测（理想情况）
    print("=" * 40)
    print("手续费率: 0%")
    print("=" * 40)
    ret_no_fee = Backtest(BTCUSD, SmaCross, ExchangeAPI,
                           cash=10000.0, commission=0.00).run()
    print(ret_no_fee)

    # 手续费率为千分之三的回测（接近真实情况）
    print("\n" + "=" * 40)
    print("手续费率: 0.3%")
    print("=" * 40)
    ret_with_fee = Backtest(BTCUSD, SmaCross, ExchangeAPI,
                            cash=10000.0, commission=0.003).run()
    print(ret_with_fee)


if __name__ == '__main__':
    main()

========================================
手续费率: 0%
========================================
初始市值     10000.000000
结束市值    576361.772884
收益        566361.772884
收益率           56.636177

========================================
手续费率: 0.3%
========================================
初始市值     10000.000000
结束市值      2036.562001
收益        -7963.437999
收益率           -0.796344

def compute_metrics(portfolio_values, trades, risk_free_rate=0.03):
    """
    计算策略的核心性能指标。

    参数:
        portfolio_values: 每个时间点的总市值序列
        trades: 交易记录列表 [{'type': 'buy', 'price': ..., 'date': ...}, ...]
        risk_free_rate: 无风险利率（默认 3%）
    """
    values = pd.Series(portfolio_values)
    returns = values.pct_change().dropna()

    # 总收益率
    total_return = values.iloc[-1] / values.iloc[0] - 1

    # 年化收益率
    days = len(returns)
    annual_return = (1 + total_return) ** (365 / days) - 1

    # 夏普比率
    excess_returns = returns - risk_free_rate / 365
    sharpe = np.sqrt(365) * excess_returns.mean() / returns.std() \
        if returns.std() > 0 else 0

    # 最大回撤
    cumulative = (1 + returns).cumprod()
    running_max = cumulative.expanding().max()
    drawdown = (cumulative - running_max) / running_max
    max_drawdown = drawdown.min()

    # 胜率
    if trades:
        # 配对买卖来计算每笔交易的盈亏
        winning_trades = 0
        for i in range(0, len(trades) - 1, 2):
            if i + 1 < len(trades):
                buy = trades[i]
                sell = trades[i + 1]
                if sell['price'] > buy['price']:
                    winning_trades += 1
        win_rate = winning_trades / (len(trades) // 2) if len(trades) >= 2 else 0
    else:
        win_rate = 0

    return {
        '总收益率': f'{total_return:.2%}',
        '年化收益率': f'{annual_return:.2%}',
        '夏普比率': f'{sharpe:.2f}',
        '最大回撤': f'{max_drawdown:.2%}',
        '胜率': f'{win_rate:.2%}',
        '交易次数': len(trades)
    }

# 网格搜索最优 SMA 窗口组合
results = []
for fast in range(5, 31, 5):
    for slow in range(fast + 10, 61, 10):
        # 动态修改策略参数
        SmaCross.fast = fast
        SmaCross.slow = slow

        ret = Backtest(BTCUSD, SmaCross, ExchangeAPI,
                       cash=10000.0, commission=0.002).run()

        results.append({
            'fast': fast,
            'slow': slow,
            'return': ret['收益率']
        })

# 找出最优参数组合
best = max(results, key=lambda x: x['return'])
print(f"最优参数: fast={best['fast']}, slow={best['slow']}")
print(f"最优收益率: {best['return']:.2%}")