1. 搭建最小回测引擎

任务目标

实现一个最小可用的单资产日频回测引擎：输入历史价格序列与一个简单的均线交叉策略，引擎逐日模拟交易，输出每日账户权益曲线，并计算总收益率、年化收益率、最大回撤、夏普比率与胜率五项基础绩效指标。

环境依赖

仅使用 Python 标准库（math、statistics），无需安装第三方包。数据以 CSV 格式提供，包含 date 与 close 两列。

核心实现

import csv
import math
from statistics import mean

def load_prices(path):
    """读取CSV，返回(date, close)列表，按日期升序"""
    with open(path, newline='') as f:
        reader = csv.DictReader(f)
        data = [(row['date'], float(row['close'])) for row in reader]
    return data

def sma(values, window):
    """简单移动平均，返回与输入等长列表，前window-1个为None"""
    out = [None] * len(values)
    if len(values) < window:
        return out
    s = sum(values[:window])
    out[window-1] = s / window
    for i in range(window, len(values)):
        s += values[i] - values[i-window]
        out[i] = s / window
    return out

def run_backtest(prices, short_win=5, long_win=20, initial_cash=10000.0):
    """
    单资产日频回测主循环。
    策略：短期均线上穿长期均线时全仓买入，下穿时清仓。
    返回 (equity_curve, metrics_dict)
    """
    dates = [p[0] for p in prices]
    closes = [p[1] for p in prices]
    n = len(closes)

    short_sma = sma(closes, short_win)
    long_sma = sma(closes, long_win)

    cash = initial_cash
    shares = 0
    equity_curve = []
    trades = []  # 记录每笔已平仓交易的收益率

    for i in range(n):
        price = closes[i]
        # 信号判断：需要两条均线都有效
        if short_sma[i] is not None and long_sma[i] is not None:
            prev_short = short_sma[i-1] if i > 0 else None
            prev_long = long_sma[i-1] if i > 0 else None
            if prev_short is not None and prev_long is not None:
                # 金叉买入（空仓时）
                if prev_short <= prev_long and short_sma[i] > long_sma[i] and shares == 0:
                    shares = cash / price
                    cash = 0.0
                # 死叉卖出（持仓时）
                elif prev_short >= prev_long and short_sma[i] < long_sma[i] and shares > 0:
                    cash = shares * price
                    trade_return = (cash - initial_cash) / initial_cash
                    trades.append(trade_return)
                    shares = 0.0

        # 记录当日总权益（现金 + 持仓市值）
        equity = cash + shares * price
        equity_curve.append((dates[i], equity))

    # 若期末仍持仓，按最后价格平仓计入交易
    if shares > 0:
        final_value = shares * closes[-1]
        trade_return = (final_value - initial_cash) / initial_cash
        trades.append(trade_return)

    # ---- 绩效指标计算 ----
    final_equity = equity_curve[-1][1]
    total_return = (final_equity - initial_cash) / initial_cash

    # 年化（假设252个交易日）
    years = n / 252.0
    annual_return = (1 + total_return) ** (1 / years) - 1 if years > 0 else 0.0

    # 最大回撤
    peak = equity_curve[0][1]
    max_dd = 0.0
    for _, eq in equity_curve:
        if eq > peak:
            peak = eq
        dd = (peak - eq) / peak if peak > 0 else 0.0
        if dd > max_dd:
            max_dd = dd

    # 夏普（基于每日收益率，无风险利率=0）
    daily_rets = []
    for j in range(1, len(equity_curve)):
        prev_eq = equity_curve[j-1][1]
        if prev_eq > 0:
            daily_rets.append(equity_curve[j][1] / prev_eq - 1.0)
    sharpe = 0.0
    if len(daily_rets) > 1:
        sd = statistics_stdev(daily_rets)
        if sd > 0:
            sharpe = mean(daily_rets) / sd * math.sqrt(252)

    # 胜率（已平仓交易中盈利比例）
    win_rate = 0.0
    if trades:
        win_rate = sum(1 for t in trades if t > 0) / len(trades)

    metrics = {
        'total_return': total_return,
        'annual_return': annual_return,
        'max_drawdown': max_dd,
        'sharpe': sharpe,
        'win_rate': win_rate,
        'num_trades': len(trades)
    }
    return equity_curve, metrics

def statistics_stdev(values):
    """手动实现样本标准差（statistics.stdev 的简化替代）"""
    m = mean(values)
    var = sum((x - m) ** 2 for x in values) / (len(values) - 1)
    return math.sqrt(var)

# ---- 最小调用入口 ----
if __name__ == '__main__':
    # 生成示例数据（20个交易日，确定性序列）
    sample = [('2024-01-0%d' % (i+1) if i < 9 else '2024-01-%d' % (i+1), 100 + i * 0.5) for i in range(20)]
    with open('sample_prices.csv', 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(['date', 'close'])
        writer.writerows(sample)

    curve, metrics = run_backtest(sample)
    print('最后5日权益:', curve[-5:])
    print('绩效指标:', metrics)

运行方式

1. 将上述代码保存为 backtest.py。
2. 准备数据文件 prices.csv，格式为两列：date（格式不限，仅作标签）、close（浮点数），按日期升序排列。
3. 在代码末尾的 if __name__ == '__main__' 块中，将 sample 替换为从 load_prices('prices.csv') 读取的数据，或直接调用 run_backtest(load_prices('prices.csv'))。
4. 执行 python backtest.py。

验证方法

运行后观察输出：

• 权益曲线：打印的最后5日权益应逐日变化，且始终大于0。若策略从未触发交易，权益曲线为水平直线（恒等于初始资金）。
• 指标合理性：total_return 与 final_equity/initial_cash - 1 一致；max_drawdown 在0~1之间；num_trades 为实际发生的买卖次数（每次完整买卖计1笔）。
• 手工核对：取前10日数据，手动计算5日与20日均线（前4日无信号），验证第5日是否产生金叉/死叉信号，并与代码输出对比。

通过条件：程序无异常运行完毕，输出的权益曲线长度等于输入数据行数，五项指标均为有限数值且与手工抽样计算一致。

常见问题

• 均线前段为None：前 window-1 个数据点无法计算均线，引擎在这些日期不产生交易信号，属正常行为。
• 数据不足：若价格数据少于长期均线窗口（如少于20条），策略将始终空仓，权益曲线为水平线，指标中 num_trades=0。
• 除零风险：当初始资金为0或权益为0时，部分指标计算会出错。本实现假设 initial_cash > 0 且价格恒为正。