夜雨聆风🤖 AI + 量化交易实战：用自然语言从 0 搭建股票策略系统（完整代码）
📌 Prompt 模板 A：双均线趋势策略
📌 Prompt 模板 B：RSI 超卖反转策略（进阶）
📌 Prompt 模板 C：多因子选股策略（高阶）
# -*- coding: utf-8 -*-"""双均线趋势策略由 AI（Cursor/Claude Code）根据 Prompt 自动生成策略逻辑：MA5 上穿 MA20 买入，MA5 下穿 MA20 卖出回测框架：Backtrader | 数据：akshare | 时间：2022-01-01 ~ 2024-12-31"""# ============ 依赖安装（首次运行前执行）============# pip install backtrader akshare matplotlib pandas# ============ 1. 数据获取模块 ============import akshare as akimport pandas as pdimport datetime as dtimport backtrader as btimport matplotlibmatplotlib.use('Agg') # 非交互式后端，避免弹窗def get_stock_data(stock_code, start_date, end_date):"""通过 akshare 获取 A 股历史行情数据:param stock_code: 股票代码，如 '600519'（贵州茅台）:param start_date: 开始日期，格式 'YYYYMMDD':param end_date: 结束日期，格式 'YYYYMMDD':return: DataFrame，列名标准化为 Backtrader 所需格式"""print(ff"正在获取 {stock_code} 历史数据...")df = ak.stock_zh_a_hist(symbol=stock_code,period="daily",start_date=start_date,end_date=end_date,adjust="qfq" # 前复权)# akshare 返回列名重映射df = df[['日期','开盘','收盘','最高','最低','成交量']]df.columns = ['date','open','close','high','low','volume']df['date'] = pd.to_datetime(df['date'])df = df.set_index('date')print(ff"数据获取成功，共 {len(df)} 条日线数据")return df# ============ 2. 策略逻辑模块 ============class MACrossStrategy(bt.Strategy):"""双均线趋势策略- 买入：MA5 上穿 MA20（金叉）- 卖出：MA5 下穿 MA20（死叉）- 止损：亏损 5% 强制平仓- 止盈：盈利 15% 强制平仓"""# 策略参数：周期、回测数据从外部传入params = (('fast_ma', 5), # 快速均线周期('slow_ma', 20), # 慢速均线周期('stop_loss', 0.05), # 止损 5%('take_profit', 0.15), # 止盈 15%)def __init__(self):# 计算双均线指标self.ma_fast = bt.indicators.SimpleMovingAverage(self.data.close, period=self.params.fast_ma)self.ma_slow = bt.indicators.SimpleMovingAverage(self.data.close, period=self.params.slow_ma)self.crossover = bt.indicators.CrossOver(self.ma_fast, self.ma_slow)# 跟踪订单状态self.order = Noneself.buy_price = Nonedef next(self):# 如果有挂单未成交，跳过if self.order:return# ---- 买入逻辑：MA5 上穿 MA20 ----if self.crossover > 0 and not self.position:# 计算买入数量：30% 仓位size = int((self.broker.getvalue() * 0.30) / self.data.close[0])if size > 0:self.buy_price = self.data.close[0]self.order = self.buy(size=size)print(ff"[买入] 日期:{self.data.datetime.date(0)} | 价格:{self.buy_price:.2f} | 数量:{size}")# ---- 卖出逻辑：MA5 下穿 MA20 ----elif self.crossover < 0 and self.position:self.order = self.sell(size=self.position.size)print(ff"[卖出] 日期:{self.data.datetime.date(0)} | 价格:{self.data.close[0]:.2f} | 原因:死叉")# ---- 止损/止盈逻辑 ----if self.position and self.buy_price:pnl_ratio = (self.data.close[0] - self.buy_price) / self.buy_priceif pnl_ratio <= -self.params.stop_loss:self.order = self.sell(size=self.position.size)print(ff"[止损] 日期:{self.data.datetime.date(0)} | 亏损:{pnl_ratio*100:.1f}%")elif pnl_ratio >= self.params.take_profit:self.order = self.sell(size=self.position.size)print(ff"[止盈] 日期:{self.data.datetime.date(0)} | 盈利:{pnl_ratio*100:.1f}%")def notify_order(self, order):if order.status in [order.Completed]:self.order = None# ============ 3. 回测引擎模块 ============def run_backtest():"""初始化 Backtrader 引擎，运行回测，输出绩效报告"""cerebro = bt.Cerebro()# 添加策略（可同时测试多个策略）cerebro.addstrategy(MACrossStrategy)# 设置初始资金和佣金cerebro.broker.setcash(100000.0) # 10万元cerebro.broker.setcommission(commission=0.0003) # 双边万三佣金cerebro.broker.set_slippage_perc(0.001) # 千分之一滑点# 加载数据（贵州茅台 600519）df = get_stock_data('600519', '20220101', '20241231')data_feed = bt.feeds.PandasData(dataname=df)cerebro.adddata(data_feed)# 添加分析器：绩效指标cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe', riskfreerate=0.03)cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')# 设置图表样式cerebro.addwriter(bt.WriterFile, csv=True, outfile="backtest_result.csv")# 运行回测print("\n" + "="*50)print("回测开始...")strategies = cerebro.run()strategy = strategies[0]# 提取绩效指标final_value = cerebro.broker.getvalue()initial_cash = 100000.0total_return = (final_value - initial_cash) / initial_cash * 100sharpe_ratio = strategy.analyzers.sharpe.get_analysis().get('sharperatio', None)max_dd = strategy.analyzers.drawdown.get_analysis().get('max', {}).get('drawdown', 0)total_return_annual = strategy.analyzers.returns.get_analysis().get('rtot', 0) * 100# 交易统计trade_stats = strategy.analyzers.trades.get_analysis()total_trades = trade_stats.get('total', {}).get('total', 0)won_trades = trade_stats.get('won', {}).get('total', 0)lost_trades = trade_stats.get('lost', {}).get('total', 0)win_rate = (won_trades / total_trades * 100) if total_trades > 0 else 0# 打印绩效报告print("="*50)print("📊 回测绩效报告（贵州茅台 MA 双均线策略 2022-2024）")print("="*50)print(ff" 初始资金： ¥{initial_cash:,.2f}")print(ff" 最终净值： ¥{final_value:,.2f}")print(ff" 总收益率： {total_return:.2f}%")print(ff" 年化收益率： {total_return_annual:.2f}%")print(ff" 夏普比率： {sharpe_ratio:.3f}" if sharpe_ratio else " 夏普比率： N/A")print(ff" 最大回撤： {max_dd:.2f}%")print(ff" 总交易次数： {total_trades}")print(ff" 盈利次数： {won_trades}")print(ff" 亏损次数： {lost_trades}")print(ff" 胜率： {win_rate:.1f}%")print("="*50)print(ff"最终账户余额：¥{final_value:,.2f}（手续费和滑点已扣除）")# 绘制图表并保存cerebro.plot(style='candlestick', barup='#26a69a', bardown='#ef5350',volume=False, figsize=(16, 8))print("\n策略图表已保存为 backtest_chart.png")# ============ 4. 主程序入口 ============if __name__ == "__main__":run_backtest()
📌 安装依赖（Terminal / PowerShell）
# 创建独立 Python 环境（推荐）python -m venv venv.\venv\Scripts\activate # Windows# source venv/bin/activate # macOS/Linux# 一键安装所有依赖pip install backtrader akshare matplotlib pandas numpy# 验证安装python -c "import backtrader, akshare; print('✅ 依赖安装成功')"
📌 运行回测
# 运行双均线策略回测python ma_cross_strategy.py# -*- coding: utf-8 -*-"""风控增强模块 — 由 AI 根据风控需求自动生成功能：账户总止损 | 单日最大亏损 | 最大持仓限制 | 波动率过滤用法：将此类混入（mixin）到现有策略，或直接替换 MACrossStrategy"""class RiskManager:"""风控管理器 — 独立模块，可叠加到任意策略"""params = (('max_daily_loss_pct', 0.02), # 单日最大亏损 2% 时停止交易('max_portfolio_loss', 0.10), # 账户总回撤超 10% 时清仓('max_positions', 3), # 最多同时持仓 3 只('volatility_threshold', 0.03), # 日波动率超过 3% 时不新开仓('atr_stop_loss', 2.0), # ATR 的 2 倍作为动态止损)def __init__(self):self.daily_pnl = 0.0self.daily_start_value = Noneself.peak_value = None # 历史最高净值self.trading_stopped = Falseself.current_date = Noneself.atr = bt.indicators.ATR(self.data, period=14) # ATR 指标def next(self):current_date = self.data.datetime.date(0)# 每日重置单日盈亏if self.current_date != current_date:if self.daily_start_value:self.daily_pnl = (self.broker.getvalue() - self.daily_start_value) / self.daily_start_value# 检查单日亏损是否超限if self.daily_pnl <= -self.params.max_daily_loss_pct:self.trading_stopped = Trueself.close_all_positions()print(ff"[风控-日损] 日期:{current_date} | 单日亏损:{self.daily_pnl*100:.1f}% | 停止交易")self.daily_start_value = self.broker.getvalue()self.current_date = current_date# 更新历史最高净值，检查账户总回撤if self.peak_value is None or self.broker.getvalue() > self.peak_value:self.peak_value = self.broker.getvalue()else:drawdown = (self.broker.getvalue() - self.peak_value) / self.peak_valueif drawdown <= -self.params.max_portfolio_loss:self.trading_stopped = Trueself.close_all_positions()print(ff"[风控-总回撤] 日期:{current_date} | 回撤:{drawdown*100:.1f}% | 清仓止损")# 波动率过滤：不追高，不在极端波动时开仓daily_return = self.data.close[0] / self.data.close[-1] - 1if abs(daily_return) > self.params.volatility_threshold:# 波动过大时不产生新买入信号（由外部策略判断是否空仓）passdef can_open_position(self):"""检查是否可以开新仓位"""if self.trading_stopped:return Falseif len(self.broker.positions) >= self.params.max_positions:return Falsereturn Truedef get_atr_stop_price(self, entry_price):"""基于 ATR 计算动态止损价格"""return entry_price - self.params.atr_stop_loss * self.atr[0]def close_all_positions(self):"""紧急清仓"""for pos in self.broker.positions.values():if pos.size != 0:self.sell(size=abs(pos.size))
| 阶段 | 工具选择 | 关键动作 | 通过标准 |
|---|---|---|---|
| 阶段1 | |||
| 阶段2 | |||
| 阶段3 | |||
| 阶段4 |
❌ 错误路径
回测收益高 → 直接上全部资金实盘
跳过模拟盘验证阶段
实盘后频繁手动干预策略
出现大回撤后盲目修改参数
不知道策略何时失效就死扛
✅ 正确路径
虚拟回测 → 模拟盘 → 小资金实盘 → 全量
每个阶段设明确的通过/不通过标准
实盘严格遵循策略参数，不手动干预
定期做归因分析（策略收益来源）
设定硬性止损线和策略终止条件
| 指标 | 含义 | 优秀标准 | 危险信号 |
|---|---|---|---|
| 夏普比率 | > 1.0 优秀 | < 0.5 | |
| 最大回撤 | < 20% | > 40% | |
| 年化收益率 | > 15% | < 5% | |
| 胜率 | 40%~60% | > 70%（疑似过拟合） | |
| 盈亏比 | > 1.5 | < 1.0 | |
| 卡尔马比率 | > 0.5 | < 0.3 |
