Security Quant Backtest

AI-powered quantitative backtesting tool for China A-share strategies, supporting design, historical tests, performance attribution, walk-forward analysis, a...

Audits

Pass

Install

openclaw skills install security-quant-backtest

Quantitative Backtesting Laboratory / 量化回测实验室

English: AI-powered quantitative backtesting laboratory — covers strategy design, historical backtesting, performance attribution, walk-forward analysis, and Monte Carlo simulation. Built for quant analysts and algorithmic traders.

中文: 量化回测实验室——覆盖策略设计、历史回测、绩效归因、前向分析、蒙特卡洛模拟。适用:量化分析师、算法交易者、Python回测开发。


Industry Pain Points / 行业痛点

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

Trigger Keywords / 触发关键词

English Triggers: quantitative backtesting, algorithmic trading, strategy research, Python backtest, performance analysis, Monte Carlo, walk-forward analysis, A-share strategy

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


Core Capabilities / 核心能力

1. Backtesting Engine / 回测引擎

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
        
        volatility = equity_df["returns"].std() * np.sqrt(252) * 100
        
        sharpe_ratio = (annual_return - 2.75) / volatility  # 假设无风险利率2.75%
        
        # 最大回撤
        cummax = equity_df["portfolio_value"].cummax()
        drawdown = (equity_df["portfolio_value"] - cummax) / cummax
        max_drawdown = drawdown.min() * 100
        
        # 卡尔玛比率
        calmar_ratio = annual_return / abs(max_drawdown) if max_drawdown != 0 else 0
        
        return {
            "strategy": strategy_name,
            "period": f"{equity_df.index[0].date()} to {equity_df.index[-1].date()}",
            "total_return": round(total_return, 2),
            "annual_return": round(annual_return, 2),
            "volatility": round(volatility, 2),
            "sharpe_ratio": round(sharpe_ratio, 2),
            "max_drawdown": round(max_drawdown, 2),
            "calmar_ratio": round(calmar_ratio, 2),
            "total_trades": len([t for t in self.trades if t["action"] == "BUY"]),
            "win_rate": self._calculate_win_rate(),
            "equity_curve": equity_df
        }
    
    def _calculate_win_rate(self) -> float:
        """计算胜率"""
        if len(self.trades) < 2:
            return 0
        
        buy_trades = [t for t in self.trades if t["action"] == "BUY"]
        sell_trades = [t for t in self.trades if t["action"] == "SELL"]
        
        if len(sell_trades) == 0:
            return 0
        
        wins = sum(
            1 for i, sell in enumerate(sell_trades)
            if i < len(buy_trades) and 
            sell["price"] > buy_trades[i]["price"]
        )
        
        return wins / len(sell_trades) * 100

2. Strategy Examples / 策略示例

# 示例策略:双均线交叉策略
class DualMovingAverageStrategy:
    """双均线策略"""
    
    def __init__(self, short_window: int = 20, long_window: int = 60):
        self.short_window = short_window
        self.long_window = long_window
    
    def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame:
        """生成交易信号"""
        signals = pd.Series(index=data.index, dtype=int)
        
        # 计算均线
        ma_short = data["close"].rolling(self.short_window).mean()
        ma_long = data["close"].rolling(self.long_window).mean()
        
        # 金叉买入,死叉卖出
        position = 0
        for i in range(self.long_window, len(data)):
            if ma_short.iloc[i] > ma_long.iloc[i] and position == 0:
                signals.iloc[i] = 1  # 买入
                position = 1
            elif ma_short.iloc[i] < ma_long.iloc[i] and position == 1:
                signals.iloc[i] = -1  # 卖出
                position = 0
        
        return signals

# RSI均值回归策略
class RSIMeanReversionStrategy:
    """RSI均值回归策略"""
    
    def __init__(self, period: int = 14, 
                 oversold: float = 30, 
                 overbought: float = 70):
        self.period = period
        self.oversold = oversold
        self.overbought = overbought
    
    def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame:
        """RSI超卖买入,超买卖出"""
        delta = data["close"].diff()
        gain = (delta.where(delta > 0, 0)).rolling(self.period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(self.period).mean()
        
        rs = gain / loss
        rsi = 100 - (100 / (1 + rs))
        
        signals = pd.Series(index=data.index, dtype=int)
        position = 0
        
        for i in range(self.period, len(data)):
            if rsi.iloc[i] < self.oversold and position == 0:
                signals.iloc[i] = 1  # 买入
                position = 1
            elif rsi.iloc[i] > self.overbought and position == 1:
                signals.iloc[i] = -1  # 卖出
                position = 0
        
        return signals

3. Monte Carlo Simulation / 蒙特卡洛模拟

class MonteCarloSimulation:
    """蒙特卡洛模拟"""
    
    def run_simulation(self, historical_returns: pd.Series,
                      n_simulations: int = 1000,
                      n_periods: int = 252,
                      initial_value: float = 1000000) -> dict:
        """
        运行蒙特卡洛模拟
        """
        mu = historical_returns.mean()
        sigma = historical_returns.std()
        
        simulations = np.zeros((n_simulations, n_periods))
        simulations[:, 0] = initial_value
        
        for t in range(1, n_periods):
            random_returns = np.random.normal(mu, sigma, n_simulations)
            simulations[:, t] = simulations[:, t-1] * (1 + random_returns)
        
        # 统计结果
        final_values = simulations[:, -1]
        
        percentiles = {
            "5th": np.percentile(final_values, 5),
            "25th": np.percentile(final_values, 25),
            "50th": np.percentile(final_values, 50),
            "75th": np.percentile(final_values, 75),
            "95th": np.percentile(final_values, 95)
        }
        
        # 概率分析
        prob_loss = (final_values < initial_value).mean() * 100
        
        return {
            "percentiles": {k: round(v, 2) for k, v in percentiles.items()},
            "probability_of_loss": round(prob_loss, 2),
            "expected_return": round(final_values.mean() - initial_value, 2),
            "var_95": round(initial_value - percentiles["5th"], 2),
            "simulations": simulations
        }

Quick Command Templates / 快速指令模板

回测双均线策略:

回测双均线策略(MA20/MA60):
- 初始资金:100万
- 回测期:2020-01-01至2025-12-31
- 关注指标:收益率、夏普比率、最大回撤

蒙特卡洛模拟:

对当前持仓做蒙特卡洛模拟:
- 模拟次数:10000次
- 模拟期限:1年
- 置信区间:95%

Disclaimer

This skill provides backtesting tools for educational and research purposes. Backtesting results do not guarantee future performance. Past performance is not indicative of future results. Algorithmic trading involves substantial risk of loss.