Install
openclaw skills install security-quant-backtestAI-powered quantitative backtesting tool for China A-share strategies, supporting design, historical tests, performance attribution, walk-forward analysis, a...
openclaw skills install security-quant-backtestEnglish: 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回测开发。
| Pain Point / 痛点 | Impact / 影响 | Solution / 本Skill解决方案 |
|---|---|---|
| 未来函数 | 回测虚高,实盘亏损 | 信号对齐检查+严格回测规范 |
| 过拟合 | 参数过度优化,实盘失效 | 样本外测试+统计显著性检验 |
| 滑点假设 | 低估交易成本,实盘收益缩水 | 多场景滑点模拟 |
| 幸存者偏差 | 只用现存股票,忽视退市股 | 使用完整历史数据 |
| 执行缺口 | 回测vs实盘收益差异大 | 分层回测+执行模拟 |
English Triggers: quantitative backtesting, algorithmic trading, strategy research, Python backtest, performance analysis, Monte Carlo, walk-forward analysis, A-share strategy
中文触发词(优先): 量化回测 / 算法交易 / 策略研究 / 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
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
# 示例策略:双均线交叉策略
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
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
}
回测双均线策略:
回测双均线策略(MA20/MA60):
- 初始资金:100万
- 回测期:2020-01-01至2025-12-31
- 关注指标:收益率、夏普比率、最大回撤
蒙特卡洛模拟:
对当前持仓做蒙特卡洛模拟:
- 模拟次数:10000次
- 模拟期限:1年
- 置信区间:95%
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.