Security Quant Backtest — 安全量化回测

量化回测实验室 ================

AI驱动的量化回测实验室，覆盖策略设计、历史回测、绩效归因、前向分析和蒙特卡洛模拟。适用于量化分析师、算法交易者和Python回测开发。

行业痛点 ---------

回测虚高，实盘亏损：信号对齐检查+严格回测规范 过拟合：参数过度优化，实盘失效；样本外测试+统计显著性检验 滑点假设：低估交易成本，实盘收益缩水；多场景滑点模拟 幸存者偏差：只用现存股票，忽视退市股；使用完整历史数据 执行缺口：回测vs实盘收益差异大；分层回测+执行模拟

触发关键词 ------------

量化回测 算法交易 策略研究 Python回测 绩效归因 蒙特卡洛 前向分析 趋势跟踪 均值回归 配对交易 双均线 海龟策略 RSI策略 布林带策略 策略优化 参数寻优 机器学习选股 Alpha因子 多因子策略

核心能力 --------

回测引擎

import pandas as pd
import numpy as np
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')
class BacktestEngine:
    """量化回测引擎"""
    def __init__(self, initial_capital: float = 1000000, commission_rate: float = 0.0003, stamp_tax: float = 0.001, slippage: float = 0.001):
        """
        Args:
            initial_capital: 初始资金
            commission_rate: 佣金费率（含规费）
            stamp_tax: 印花税率（仅卖出）
            slippage: 滑点（百分比）
        """
        self.initial_capital = initial_capital
        self.commission_rate = commission_rate
        self.stamp_tax = stamp_tax
        self.slippage = slippage
        # 持仓状态
        self.cash = initial_capital
        self.position = {}  # {stock_code: shares}
        self.equity_curve = []
        self.trades = []
    def run(self, data: pd.DataFrame, signals: pd.DataFrame, strategy_name: str = "Strategy") -> dict:
        """
        执行回测
        Args:
            data: 价格数据（含收盘价、开盘价、最高、最低价）
            signals: 交易信号（1=买入, -1=卖出, 0=持有）
            strategy_name: 策略名称
        """
        results = []
        for date in data.index:
            price = data.loc[date, "close"]
            # 获取当日信号
            if date in signals.index:
                signal = signals.loc[date]
                if signal == 1:
                    # 买入信号
                    self._buy(date, price, self.cash  0.95)  # 保留5%现金
                elif signal == -1:
                    # 卖出信号
                    self._sell(date, price)
            # 更新权益
            portfolio_value = self._calculate_portfolio_value(price)
            self.equity_curve.append({
                "date": date,
                "portfolio_value": portfolio_value,
                "cash": self.cash
            })
        return self._generate_report(strategy_name)
    def _buy(self, date, price, target_amount):
        """买入执行（含滑点+佣金）"""
        buy_price = price  (1 + self.slippage)
        shares = int(target_amount / buy_price / 100)  100  # 100股整数
        if shares > 0:
            cost = shares  buy_price
            commission = cost  self.commission_rate
            if cost + commission <= self.cash:
                self.cash -= (cost + commission)
                self.trades.append({
                    "date": date,
                    "action": "BUY",
                    "price": buy_price,
                    "shares": shares,
                    "commission": commission
                })
    def _sell(self, date, price):
        """卖出执行（含滑点+佣金+印花税）"""
        sell_price = price  (1 - self.slippage)
        for stock, shares in list(self.position.items()):
            if shares > 0:
                proceeds = shares  sell_price
                commission = proceeds  self.commission_rate
                tax = proceeds  self.stamp_tax
                self.cash -= (commission + tax)
                self.cash += proceeds
                self.trades.append({
                    "date": date,
                    "action": "SELL",
                    "price": sell_price,
                    "shares": shares,
                    "commission": commission,
                    "tax": tax
                })
    def _calculate_portfolio_value(self, current_price):
        """计算组合市值"""
        position_value = sum(
            shares  current_price for stock, shares in self.position.items()
        )
        return self.cash + position_value
    def _generate_report(self, strategy_name: str) -> dict:
        """生成回测报告"""
        equity_df = pd.DataFrame(self.equity_curve)
        equity_df.set_index("date", inplace=True)
        equity_df["returns"] = equity_df["portfolio_value"].pct_change()
        # 核心指标计算
        total_return = (equity_df["portfolio_value"].iloc[-1] / self.initial_capital - 1)  100
        annual_return = ((1 + total_return/100)  (252/len(equity_df)) - 1)  100