策略回测

策略回测（Backtesting）

“我该如何在真实历史数据中验证我的策略是否真的有效？”

这一步，叫做策略回测（Backtesting）。在 Testnet 里你只能看到程序能不能跑通，但你完全不知道——你的策略在过去一年、过去一个牛熊周期中是否真的能赚钱。

原理

获取真实历史行情数据（如 Binance 的 BTC/USDT）
模拟你的策略逻辑（例如均线交叉、RSI、突破）
在每一个时间点计算信号（买/卖）
模拟资金变化（买入/卖出、手续费、滑点）
计算绩效指标：
收益率、最大回撤、夏普比率、胜率等
可视化交易过程（价格 + 买卖信号）

获取历史数据

你可以直接用 ccxt 或 python-binance 从 Binance 获取历史K线。

首先安装 cctx:

pip install ccxt

然后

import ccxt
import pandas as pd

exchange = ccxt.binance()

# 获取最近1000个1小时K线
bars = exchange.fetch_ohlcv('BTC/USDT', timeframe='1h', limit=1000)

df = pd.DataFrame(bars, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)

# 保存为 CSV 文件
df.to_csv('btc_usdt_1h.csv')
# 读取
# df = pd.read_csv('btc_usdt_1h.csv', parse_dates=['timestamp'], index_col='timestamp')

接下来是双均线策略的回测代码

import numpy as np
import pandas as pd

def ma_crossover_backtest(
    df: pd.DataFrame,
    short=12, long=26,
    ema=True,
    long_only=True,
    fee=0.001,          # 单边费率（示例：0.1%）
    slippage=0.0002,    # 单边滑点（示例：0.02%）
    execute_on='next_open'  # 'next_open' 或 'next_close'
):
    """
    df: index 为时间，至少包含列 ['open','close']（若 next_open 执行需要 'open'）
    short, long: 短长均线窗口
    ema: True 用 EMA，False 用 SMA
    long_only: True=多/空仓, False=可做多做空（合约）
    fee, slippage: 单边成本
    execute_on: 信号在下一根的 'open' 或 'close' 执行
    """

    df = df.copy()

    # 1) 计算均线
    if ema:
        df['ma_s'] = df['close'].ewm(span=short, adjust=False).mean()
        df['ma_l'] = df['close'].ewm(span=long, adjust=False).mean()
    else:
        df['ma_s'] = df['close'].rolling(short).mean()
        df['ma_l'] = df['close'].rolling(long).mean()

    df.dropna(inplace=True)

    # 2) 生成基础信号：1 or -1（或 0）
    if long_only:
        # 多/空仓（long/flat）
        raw = np.where(df['ma_s'] > df['ma_l'], 1, 0)
    else:
        # 多空双向（long/short）
        raw = np.where(df['ma_s'] > df['ma_l'], 1, -1)

    df['signal'] = pd.Series(raw, index=df.index)

    # 3) 只在翻转时交易：计算换手（position 变化）
    df['position'] = df['signal']
    # 执行时点：避免未来函数
    if execute_on == 'next_open':
        price_exec = df['open'].shift(-1)  # 用下一根开盘成交价
        ret_base  = df['close'].pct_change()  # 基准用收盘到收盘
    else:
        price_exec = df['close'].shift(-1) # 下一根收盘成交价
        ret_base  = df['close'].pct_change()

    df['position_exec'] = df['position'].shift(1).fillna(0)  # 按下一根执行
    df['turnover'] = (df['position_exec'].diff().abs()).fillna(0)  # 均为 0/1/2

    # 4) 成本：每次换仓扣双边成本（开/平各一次）
    # 对 long_only 来说，0->1 或 1->0 都是一次换仓
    # 对 long/short 来说，-1->1 视作平空+开多，可按两次处理（这里 turnover=2）
    cost_per_side = fee + slippage
    df['trade_cost'] = df['turnover'] * cost_per_side

    # 5) 策略收益（按仓位 * 基础涨跌），再减去交易成本
    # 用 position_exec 对应区间暴露；收益用收盘到收盘（简化）
    df['strategy_ret_gross'] = df['position_exec'] * ret_base
    df['strategy_ret_net']   = df['strategy_ret_gross'] - df['trade_cost']

    # 6) 绩效
    def _cumprod(s):
        return (1 + s.fillna(0)).cumprod()

    equity = _cumprod(df['strategy_ret_net'])
    mkt    = _cumprod(ret_base)

    # 年化与风险指标（以 1h 频率为例；若是日线，把 24*365 改成 365）
    N = len(df)
    if N < 2:
        raise ValueError("样本太短")

    if df.index.inferred_type in ('datetime64', 'datetime'):
        # 估一个年化频率（粗略）：按相邻 bar 的平均间隔推
        dt = (df.index[-1] - df.index[0]) / (N - 1)
        bars_per_year = pd.Timedelta('365D') / dt
    else:
        bars_per_year = 24*365  # 兜底

    total_return = equity.iloc[-1] - 1
    ann_return   = equity.iloc[-1] ** (bars_per_year / N) - 1
    vol          = df['strategy_ret_net'].std() * np.sqrt(bars_per_year)
    sharpe       = (df['strategy_ret_net'].mean() * bars_per_year) / (df['strategy_ret_net'].std() + 1e-12)
    dd_series    = 1 - equity / equity.cummax()
    max_dd       = dd_series.max()

    summary = {
        'total_return': float(total_return),
        'annual_return': float(ann_return),
        'volatility': float(vol),
        'sharpe': float(sharpe),
        'max_drawdown': float(max_dd),
        'trades': int(df['turnover'].sum()),   # 粗略：换手次数
        'bars': int(N),
        'bars_per_year_est': float(bars_per_year)
    }

    return df, equity, mkt, summary

bt_df, equity, market, stats = ma_crossover_backtest(
    df, short=12, long=26, ema=True,
    long_only=True, fee=0.001, slippage=0.0002,
    execute_on='next_open'
)
print(stats)