基于离线数据的股票量化回测系统构建指南：从数据获取到模拟交易

简介：这是一套面向编程初学者与金融兴趣者的模拟炒股软件开发学习资源，聚焦股票交易逻辑实现与离线数据分析实践，帮助用户理解K线图展示、买卖模拟、技术指标计算等核心功能。压缩包共82个文件，以55个ASP服务端脚本为主干，支撑用户登录、账户管理、行情查看、交易操作及后台管理等完整业务流程；辅以13个TXT说明文档、5个HTML前端页面、4个HTM文件及CSS样式表，构成典型的B/S架构股票模拟系统雏形，整体大小仅711KB，轻量易部署。已有224人下载学习，适合在本地IIS环境运行调试，可直接观察用户注册、资金划转、股票持仓、K线加载（依赖外部数据导入）等模块交互逻辑，并通过ADMIN目录下的管理ASP文件深入理解权限控制与数据维护机制。

1. 项目缘起：从“离线数据”到“模拟交易”的完整闭环

最近在整理硬盘，翻出来一个尘封已久的压缩包，文件名是monichaogu.zip_K._模拟炒股代码_股票_股票离线数据_股票软件源码。看到这个名字，估计很多做量化或者对股票数据感兴趣的朋友都会会心一笑。这几乎是一个典型的个人开发者或学生早期项目的“考古现场”：一个包含了数据、代码、甚至可能是简陋界面的“大杂烩”。它背后反映的，是一个从数据获取、处理、分析到最终模拟交易执行的完整需求链。很多人入门量化或股票分析，第一步往往不是去研究高深的算法，而是先想办法把数据“搞到手”，然后写几行代码看看能不能跑起来，最后再尝试构建一个能模拟买卖的“玩具系统”。这个压缩包的名字，恰好精准地概括了这三个核心环节：股票离线数据、模拟炒股代码、股票软件源码。

我拆解并重构了这个项目，发现其核心价值不在于代码本身有多精妙，而在于它提供了一个非常接地气的、从零到一的实践范本。它避开了云端API的复杂性和网络依赖，直接从最原始的离线CSV或TXT文件开始，教你如何组织数据、计算指标、制定策略并模拟交易。这对于想理解量化交易底层逻辑，但又苦于没有稳定数据源或担心网络延迟的新手来说，是一条非常实用的学习路径。今天，我就基于这个“考古”项目，结合当前Python生态中最主流的工具链，重新梳理出一套清晰、可复现的本地化股票分析及模拟交易系统构建指南。我们将完全使用离线数据，一步步实现数据管理、策略回测和模拟交易，让你彻底掌握这个闭环中的每一个技术要点。

2. 基石构建：离线股票数据的获取、清洗与本地化管理

一切分析的基础是数据。所谓“股票离线数据”，通常指已经下载到本地的、包含股票历史行情（如日K线）的结构化文件，常见格式有CSV、TXT或数据库文件。我们的第一步，就是建立一套高效、规范的本地数据管理体系。

2.1 数据来源与获取策略

虽然项目强调“离线”，但数据总得有个最初的来源。对于学习和回测，我们通常需要几年的历史日线数据。有几种常见的获取方式：

1. 第三方数据平台导出：这是最直接的方式。你可以使用一些财经网站（如东方财富、新浪财经）或专业数据终端（需注意数据使用权限），通过其数据导出功能，将特定股票的历史行情数据以CSV格式下载到本地。通常包含日期、开盘价、最高价、最低价、收盘价、成交量等字段。
2. 使用开源库批量抓取（用于初始构建）：像akshare、yfinance(针对美股) 这样的Python库，可以方便地获取历史数据。关键在于，我们只需在项目初始化时运行一次抓取脚本，将数据持久化到本地，后续所有分析都基于这份本地副本，从而实现真正的“离线”。这避免了每次运行代码都去请求网络，也解决了某些API有调用频率限制的问题。

这里给出一个使用akshare初始化本地数据仓库的示例脚本：

import akshare as ak import pandas as pd import os from pathlib import Path def fetch_and_save_stock_data(stock_code, start_date='20180101', end_date='20231231'): """ 获取单只股票历史数据并保存到本地CSV :param stock_code: 股票代码，如 'sh600000'（akshare格式） :param start_date: 开始日期 :param end_date: 结束日期 """ # 创建数据存储目录 data_dir = Path('./stock_data/ashare/daily') data_dir.mkdir(parents=True, exist_ok=True) try: # 使用 ak.stock_zh_a_hist 获取A股日线数据 df = ak.stock_zh_a_hist(symbol=stock_code, period="daily", start_date=start_date, end_date=end_date, adjust="qfq") # 这里使用前复权，保证价格连续性 if df.empty: print(f"未获取到数据: {stock_code}") return # 标准化列名（不同数据源列名可能不同） df.rename(columns={ '日期': 'date', '开盘': 'open', '最高': 'high', '最低': 'low', '收盘': 'close', '成交量': 'volume', '成交额': 'amount', '振幅': 'amplitude', '涨跌幅': 'pct_chg', '涨跌额': 'change', '换手率': 'turnover' }, inplace=True) # 确保日期列为datetime类型，并设为索引（便于后续resample等操作） df['date'] = pd.to_datetime(df['date']) df.set_index('date', inplace=True) # 保存到CSV，文件名如 `sh600000.csv` file_path = data_dir / f"{stock_code}.csv" df.to_csv(file_path) print(f"数据已保存至: {file_path}") except Exception as e: print(f"获取 {stock_code} 数据时出错: {e}") # 示例：获取几只股票的数据 stock_list = ['sh600000', 'sz000001', 'sz300750'] for code in stock_list: fetch_and_save_stock_data(code)

注意：数据获取涉及网络请求和第三方数据源，务必遵守相关网站的使用条款，不要进行高频、大量的请求，以免对对方服务器造成压力或被封禁IP。本示例仅用于演示数据本地化的流程。

2.2 本地数据仓库的架构设计

一个良好的本地数据仓库能极大提升后续分析和回测的效率。我建议采用如下目录结构：

stock_data/ ├── ashar/ # A股数据 │ ├── daily/ # 日线数据 (CSV格式，每只股票一个文件) │ ├── 5min/ # 5分钟线数据（可选） │ └── basics/ # 基本面数据（如财务报表，可选） ├── index/ # 指数数据，如上证指数 │ └── daily/ ├── futures/ # 期货数据（可选） │ └── daily/ └── metadata.pkl # 元数据文件，记录所有已下载股票的代码、名称、更新日期等

为什么这样设计？

• 按市场和品种分离：清晰明了，避免混淆。
• 按周期分离：不同分析策略需要不同时间粒度的数据。
• 元数据文件：这是关键。当你有成百上千只股票的数据时，一个集中的元数据文件可以快速告诉你本地有哪些数据、数据更新到哪一天，方便做增量更新和批量操作。

2.3 数据清洗与一致性检查

从网络获取的数据难免有脏数据或格式问题，加载本地CSV后，必须进行清洗：

import pandas as pd def load_and_clean_data(file_path): """加载并清洗单个股票数据文件""" df = pd.read_csv(file_path, index_col='date', parse_dates=True) # 1. 处理缺失值：对于价格数据，通常用前向填充，成交量可以填0 # 但更常见的做法是直接删除缺失行，因为交易日数据本应连续 df.dropna(subset=['open', 'high', 'low', 'close'], inplace=True) # 2. 检查并处理价格异常值（如收盘价为0或负值） # 这可能是数据源错误，需要根据情况处理（删除或标记） abnormal = df[(df['close'] <= 0) | (df['volume'] < 0)] if not abnormal.empty: print(f"发现异常数据行，将删除: {len(abnormal)} 行") df.drop(abnormal.index, inplace=True) # 3. 确保数据按日期升序排列 df.sort_index(inplace=True) # 4. 去重（同一日期可能出现多次） df = df[~df.index.duplicated(keep='first')] # 5. 计算并添加常用衍生列（也可在策略中实时计算，但预处理能提升回测速度） df['returns'] = df['close'].pct_change() # 日收益率 df['ma5'] = df['close'].rolling(window=5).mean() # 5日均线 df['ma20'] = df['close'].rolling(window=20).mean() # 20日均线 # ... 可以计算更多指标，如ATR, RSI, MACD等 return df

实操心得：数据清洗是量化分析的“隐形工程”，做得不扎实，后面所有策略结果都可能失真。务必对每一只股票的数据进行上述基本检查，并形成清洗日志。对于A股，要特别注意除权除息日的数据，使用前复权（adjust=“qfq”）数据能保证价格序列的连续性，是回测的标准选择。

3. 核心引擎：模拟交易与回测系统的设计与实现

有了干净的数据，接下来就是构建系统的核心——模拟交易引擎。这个引擎需要模拟真实交易的几个关键要素：资金管理、订单执行、持仓跟踪、绩效统计。我们将完全基于Pandas和NumPy来实现一个轻量级但功能完整的回测框架。

3.1 定义核心数据结构：账户与订单

首先，我们需要定义交易发生时需要记录的信息。

from dataclasses import dataclass from datetime import datetime from typing import Optional @dataclass class Order: """订单类，记录一次买卖操作""" order_id: int symbol: str # 股票代码 direction: str # ‘buy’ 或 ‘sell’ order_type: str # ‘market’ (市价单，按下一个Bar的open价成交) quantity: int # 数量（股） filled_price: float # 成交价格 filled_time: datetime # 成交时间 commission: float = 0.0 # 佣金（手续费） @dataclass class Position: """持仓类，记录当前持有某只股票的状态""" symbol: str quantity: int = 0 # 持有数量，正数为多仓，负数为空仓（本例只做多） avg_cost: float = 0.0 # 平均成本 market_value: float = 0.0 # 当前市值（quantity * current_price） class BacktestAccount: """回测账户类，管理资金、持仓和订单历史""" def __init__(self, initial_cash=100000.0): self.initial_cash = initial_cash self.current_cash = initial_cash self.positions = {} # symbol -> Position self.orders = [] # 所有已成交订单列表 self.order_id_counter = 1 self.equity_curve = [] # 每日总权益记录 [(date, total_equity), ...] @property def total_equity(self, current_prices: dict = None): """计算当前总权益（现金 + 所有持仓市值）""" total = self.current_cash if current_prices: for sym, pos in self.positions.items(): if sym in current_prices: pos.market_value = pos.quantity * current_prices[sym] total += pos.market_value else: # 如果没有提供当前价，则用成本价近似（不准确，仅备用） total += pos.quantity * pos.avg_cost else: for pos in self.positions.values(): total += pos.quantity * pos.avg_cost return total

3.2 实现订单执行与资金结算逻辑

这是引擎最复杂的部分，需要处理买卖信号、检查资金是否充足、更新持仓和现金。

def place_market_order(self, symbol, direction, quantity, fill_price, fill_time, commission_rate=0.0003): """ 下达并执行一个市价订单。 :param commission_rate: 佣金率，默认万分之三（0.03%） """ if quantity <= 0: raise ValueError("订单数量必须大于0") # 计算佣金（买卖都收，最低5元是A股规则，此处简化按比例计算） commission = fill_price * quantity * commission_rate # 模拟最低佣金限制 commission = max(commission, 5.0) if commission_rate > 0 else 0 order_value = fill_price * quantity if direction == 'buy': # 检查资金是否足够（金额+佣金） required_cash = order_value + commission if self.current_cash < required_cash: print(f"[{fill_time}] 资金不足，无法买入 {symbol}。所需: {required_cash:.2f}, 可用: {self.current_cash:.2f}") return False # 执行买入 self.current_cash -= required_cash if symbol not in self.positions: self.positions[symbol] = Position(symbol) pos = self.positions[symbol] # 计算新的平均成本 total_cost = pos.avg_cost * pos.quantity + order_value pos.quantity += quantity pos.avg_cost = total_cost / pos.quantity if pos.quantity > 0 else 0 elif direction == 'sell': if symbol not in self.positions or self.positions[symbol].quantity < quantity: print(f"[{fill_time}] 持仓不足，无法卖出 {symbol}。") return False # 执行卖出 pos = self.positions[symbol] self.current_cash += order_value - commission pos.quantity -= quantity if pos.quantity == 0: pos.avg_cost = 0.0 # 可以选择删除零持仓的条目 # del self.positions[symbol] else: # 卖出不影响剩余持仓的平均成本 pass else: raise ValueError("direction 必须是 'buy' 或 'sell'") # 记录订单 order = Order(order_id=self.order_id_counter, symbol=symbol, direction=direction, order_type='market', quantity=quantity, filled_price=fill_price, filled_time=fill_time, commission=commission) self.orders.append(order) self.order_id_counter += 1 return True

为什么这样设计订单逻辑？

• 市价单简化：回测中普遍使用下一个Bar的开盘价（open）作为成交价，这避免了盘中复杂的价格撮合，是行业通用做法。
• 佣金模型：这里实现了按比例收取且有最低限额的佣金模型，更贴近A股实际。你也可以扩展印花税、过户费等。
• 平均成本法：这是股票持仓成本核算最常用的方法，简单且符合会计原则。

3.3 构建策略回测循环

现在，我们将数据、账户和策略串联起来。策略就是一个函数，它接收当前的市场数据、账户状态，并输出交易信号。

def simple_ma_crossover_strategy(data, account, current_date, lookback_days=20): """ 一个简单的双均线交叉策略示例。 当短期均线上穿长期均线时买入，下穿时卖出。 """ signals = [] # 返回信号列表，每个信号是 (symbol, direction, quantity) # 假设data是一个包含多只股票DataFrame的字典，索引是日期 for symbol, df in data.items(): # 确保有足够的历史数据 if current_date not in df.index or len(df.loc[:current_date]) < lookback_days + 1: continue # 获取到当前日期为止的数据切片 hist = df.loc[:current_date] current_bar = hist.loc[current_date] prev_bar = hist.iloc[-2] if len(hist) > 1 else None # 计算指标（这里应在数据清洗时预计算，此处为演示） short_ma = hist['close'].rolling(window=5).mean().iloc[-1] long_ma = hist['close'].rolling(window=20).mean().iloc[-1] prev_short_ma = hist['close'].rolling(window=5).mean().iloc[-2] if prev_bar is not None else None prev_long_ma = hist['close'].rolling(window=20).mean().iloc[-2] if prev_bar is not None else None # 生成信号 position = account.positions.get(symbol, Position(symbol)) if prev_short_ma and prev_long_ma: # 金叉：短期均线上穿长期均线，且当前无持仓 if prev_short_ma <= prev_long_ma and short_ma > long_ma and position.quantity == 0: # 计算买入数量（例如，使用固定资金比例） cash_to_use = account.current_cash * 0.95 # 使用95%的可用现金 price = current_bar['open'] # 假设以次日开盘价买入 quantity = int(cash_to_use / price / 100) * 100 # A股按手（100股）交易 if quantity > 0: signals.append((symbol, 'buy', quantity)) # 死叉：短期均线下穿长期均线，且当前有持仓 elif prev_short_ma >= prev_long_ma and short_ma < long_ma and position.quantity > 0: signals.append((symbol, 'sell', position.quantity)) # 全仓卖出 return signals def run_backtest(data_dict, initial_cash=100000.0, start_date=None, end_date=None): """ 运行回测的主函数。 :param data_dict: 字典，{股票代码: 对应的清洗后的DataFrame} """ # 确定回测日期范围（所有股票数据的交集） common_dates = None for df in data_dict.values(): if common_dates is None: common_dates = set(df.index) else: common_dates = common_dates.intersection(set(df.index)) common_dates = sorted(list(common_dates)) if start_date: common_dates = [d for d in common_dates if d >= pd.Timestamp(start_date)] if end_date: common_dates = [d for d in common_dates if d <= pd.Timestamp(end_date)] account = BacktestAccount(initial_cash=initial_cash) for current_date in common_dates: # 在交易日开始时，先记录前一日收盘后的权益（用于画曲线） # 这里简化处理，记录当日开盘前的权益（需用前一日收盘价计算持仓市值） if account.equity_curve: prev_date = account.equity_curve[-1][0] # 获取前一日收盘价计算市值 prev_prices = {} for sym, df in data_dict.items(): if prev_date in df.index: prev_prices[sym] = df.loc[prev_date, 'close'] account.equity_curve.append((current_date, account.total_equity(prev_prices))) else: # 第一天，记录初始权益 account.equity_curve.append((current_date, initial_cash)) # 运行策略，获取交易信号 signals = simple_ma_crossover_strategy(data_dict, account, current_date) # 执行信号（假设以当日开盘价成交） for symbol, direction, quantity in signals: if symbol in data_dict and current_date in data_dict[symbol].index: fill_price = data_dict[symbol].loc[current_date, 'open'] fill_time = current_date account.place_market_order(symbol, direction, quantity, fill_price, fill_time) # 回测结束，计算最终权益（使用最后一天的收盘价） final_prices = {} last_date = common_dates[-1] for sym, df in data_dict.items(): if last_date in df.index: final_prices[sym] = df.loc[last_date, 'close'] final_equity = account.total_equity(final_prices) account.equity_curve.append((last_date, final_equity)) # 记录最终权益点 return account

踩坑提醒：回测中一个经典的陷阱是“未来函数”。在上述策略中，我们严格使用hist = df.loc[:current_date]来获取截至当前日期的数据，计算指标。这意味着在current_date这一天，我们只能看到这一天及之前的数据，绝对不能用current_date之后的数据。这是回测系统可信度的生命线。

4. 绩效评估与可视化：从数字到洞见

回测跑完了，账户里有一堆订单和一条权益曲线，但这远远不够。我们需要一套科学的评估体系来衡量策略的好坏，并直观地展示出来。

4.1 关键绩效指标计算

仅仅看最终收益率是片面的。我们需要多维度评估策略的风险和收益特征。

import numpy as np import pandas as pd def calculate_performance_metrics(account, initial_cash, risk_free_rate=0.03): """ 计算一系列回测绩效指标。 :param risk_free_rate: 年化无风险利率，用于计算夏普比率等，默认3% """ # 将权益曲线转为DataFrame equity_df = pd.DataFrame(account.equity_curve, columns=['date', 'equity']) equity_df.set_index('date', inplace=True) equity_df['returns'] = equity_df['equity'].pct_change().fillna(0) # 1. 基础收益指标 total_return = (equity_df['equity'].iloc[-1] - initial_cash) / initial_cash annualized_return = (1 + total_return) ** (252 / len(equity_df)) - 1 # 假设252个交易日 # 2. 风险指标 volatility = equity_df['returns'].std() * np.sqrt(252) # 年化波动率 # 3. 风险调整后收益指标 excess_returns = equity_df['returns'] - risk_free_rate / 252 sharpe_ratio = excess_returns.mean() / excess_returns.std() * np.sqrt(252) if excess_returns.std() != 0 else 0 # 最大回撤（Max Drawdown） - 非常重要！ equity_series = equity_df['equity'] rolling_max = equity_series.expanding().max() drawdown = (equity_series - rolling_max) / rolling_max max_drawdown = drawdown.min() # 计算回撤持续期（可选） underwater = drawdown < 0 underwater_start = underwater.ne(underwater.shift()) & underwater underwater_end = underwater.ne(underwater.shift(-1)) & underwater # 这里可以进一步分析每次回撤的起止时间和深度 # 4. 交易相关指标 orders_df = pd.DataFrame([(o.symbol, o.direction, o.filled_price, o.quantity, o.commission, o.filled_time) for o in account.orders], columns=['symbol', 'direction', 'price', 'quantity', 'commission', 'time']) if not orders_df.empty: orders_df.set_index('time', inplace=True) win_trades = [] loss_trades = [] # 简化计算：遍历订单，匹配买卖对计算单笔盈亏（实际应基于持仓标记） # 更严谨的做法是在订单类或持仓类中记录每笔交易的盈亏 total_commission = orders_df['commission'].sum() # 这里省略复杂的逐笔交易盈亏匹配计算... total_trades = len(orders_df) # 胜率、平均盈亏比等需要更精细的交易记录，此处略过 metrics = { '初始资金': initial_cash, '最终权益': equity_df['equity'].iloc[-1], '总收益率': f"{total_return:.2%}", '年化收益率': f"{annualized_return:.2%}", '年化波动率': f"{volatility:.2%}", '夏普比率': f"{sharpe_ratio:.2f}", '最大回撤': f"{max_drawdown:.2%}", '总交易次数': total_trades if 'total_trades' in locals() else 0, '总手续费': total_commission if 'total_commission' in locals() else 0.0, } return metrics, equity_df, drawdown

4.2 使用Matplotlib进行可视化

图表能让绩效评估结果一目了然。

import matplotlib.pyplot as plt import matplotlib.dates as mdates def plot_backtest_results(equity_df, drawdown, metrics): """绘制回测结果图表""" fig, axes = plt.subplots(3, 1, figsize=(12, 10), sharex=True) # 1. 资产曲线 ax1 = axes[0] ax1.plot(equity_df.index, equity_df['equity'], label='策略净值', linewidth=2, color='blue') ax1.axhline(y=metrics['初始资金'], color='grey', linestyle='--', alpha=0.7, label='初始资金') ax1.set_ylabel('资产净值 (元)') ax1.set_title('策略净值曲线') ax1.legend() ax1.grid(True, alpha=0.3) # 2. 回撤曲线 ax2 = axes[1] ax2.fill_between(drawdown.index, 0, drawdown.values * 100, color='red', alpha=0.3) ax2.plot(drawdown.index, drawdown.values * 100, color='darkred', linewidth=1) ax2.set_ylabel('回撤 (%)') ax2.set_title('策略回撤曲线') ax2.grid(True, alpha=0.3) # 3. 每日收益率分布（直方图） ax3 = axes[2] returns = equity_df['returns'].dropna() * 100 # 转换为百分比 ax3.hist(returns, bins=50, edgecolor='black', alpha=0.7, color='steelblue') ax3.axvline(x=returns.mean(), color='red', linestyle='--', label=f'均值: {returns.mean():.2f}%') ax3.axvline(x=returns.std(), color='orange', linestyle='--', alpha=0.7, label=f'标准差: {returns.std():.2f}%') ax3.axvline(x=-returns.std(), color='orange', linestyle='--', alpha=0.7) ax3.set_xlabel('日收益率 (%)') ax3.set_ylabel('频次') ax3.set_title('日收益率分布') ax3.legend() ax3.grid(True, alpha=0.3) # 格式化x轴日期 for ax in axes: ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m')) ax.xaxis.set_major_locator(mdates.MonthLocator(interval=3)) plt.tight_layout() plt.show() # 打印关键指标 print("=== 回测绩效摘要 ===") for key, value in metrics.items(): print(f"{key}: {value}")

经验之谈：最大回撤是比收益率更重要的指标。它告诉你策略在最坏的情况下会亏多少钱，直接关系到你的心理承受能力和策略的生存能力。一个年化收益20%但最大回撤50%的策略，很可能在一次大的回调中被击穿止损线而被迫出局，实际根本拿不住。务必把回撤曲线和净值曲线放在一起看。

5. 从回测到“软件”：构建图形化前端与实战优化

原始的monichaogu.zip里可能包含一些简陋的GUI代码，比如用Tkinter或PyQt写的界面。对于个人项目，一个图形界面能极大提升交互体验。这里我们探讨如何用现代一点的方式（例如Plotly Dash或Streamlit）快速构建一个Web应用，将上述分析模块封装起来。

5.1 使用Streamlit快速搭建原型

Streamlit是目前构建数据应用最快捷的工具之一，几行代码就能把数据分析脚本变成交互式Web应用。

# 文件: app.py import streamlit as st import pandas as pd import plotly.graph_objects as go from plotly.subplots import make_subplots import sys sys.path.append('.') from backtest_engine import run_backtest, calculate_performance_metrics # 假设我们的回测代码在 backtest_engine.py st.set_page_config(page_title="股票模拟交易回测系统", layout="wide") st.title("📈 本地股票模拟交易与回测平台") # 侧边栏：参数配置 st.sidebar.header("回测参数设置") initial_cash = st.sidebar.number_input("初始资金（元）", min_value=10000, value=100000, step=10000) start_date = st.sidebar.date_input("开始日期", pd.to_datetime('2020-01-01')) end_date = st.sidebar.date_input("结束日期", pd.to_datetime('2023-12-31')) # 股票选择（假设本地数据目录已存在一些CSV文件） import os data_files = [f for f in os.listdir('./stock_data/ashare/daily') if f.endswith('.csv')] symbol_options = [f.replace('.csv', '') for f in data_files] selected_symbols = st.sidebar.multiselect("选择股票代码", symbol_options, default=symbol_options[:3]) # 策略选择 strategy_name = st.sidebar.selectbox("选择策略", ["双均线交叉", "动量突破", "均值回归"]) # 运行回测按钮 if st.sidebar.button("开始回测", type="primary"): if not selected_symbols: st.warning("请至少选择一只股票。") else: with st.spinner('正在加载数据并运行回测...'): # 1. 加载数据 data_dict = {} for sym in selected_symbols: file_path = f'./stock_data/ashare/daily/{sym}.csv' try: df = pd.read_csv(file_path, index_col='date', parse_dates=True) df = df.loc[start_date:end_date] # 切片 if not df.empty: data_dict[sym] = df except FileNotFoundError: st.error(f"未找到股票 {sym} 的数据文件。") if not data_dict: st.error("未能加载任何有效数据，请检查数据文件。") else: # 2. 运行回测（这里需要根据strategy_name动态选择策略函数） # 为简化，我们固定使用之前写的双均线策略 account = run_backtest(data_dict, initial_cash=initial_cash, start_date=start_date, end_date=end_date) # 3. 计算绩效 metrics, equity_df, drawdown_series = calculate_performance_metrics(account, initial_cash) # 4. 展示结果 col1, col2, col3 = st.columns(3) with col1: st.metric("总收益率", metrics['总收益率']) st.metric("年化收益率", metrics['年化收益率']) with col2: st.metric("夏普比率", metrics['夏普比率']) st.metric("最大回撤", metrics['最大回撤']) with col3: st.metric("总交易次数", metrics['总交易次数']) st.metric("总手续费", f"{metrics['总手续费']:.2f} 元") # 5. 使用Plotly绘制交互式图表 fig = make_subplots(rows=2, cols=1, shared_xaxes=True, vertical_spacing=0.05, subplot_titles=('策略净值曲线', '回撤曲线'), row_heights=[0.7, 0.3]) # 净值曲线 fig.add_trace(go.Scatter(x=equity_df.index, y=equity_df['equity'], mode='lines', name='净值', line=dict(color='royalblue', width=2)), row=1, col=1) fig.add_hline(y=initial_cash, line_dash="dash", line_color="grey", annotation_text="初始资金", row=1, col=1) # 回撤曲线 fig.add_trace(go.Scatter(x=drawdown_series.index, y=drawdown_series.values * 100, # 转百分比 fill='tozeroy', mode='lines', name='回撤', line=dict(color='red', width=1), fillcolor='rgba(255,0,0,0.3)'), row=2, col=1) fig.update_layout(height=600, showlegend=True, hovermode='x unified') fig.update_xaxes(title_text="日期", row=2, col=1) fig.update_yaxes(title_text="净值 (元)", row=1, col=1) fig.update_yaxes(title_text="回撤 (%)", row=2, col=1) st.plotly_chart(fig, use_container_width=True) # 6. 展示交易记录 st.subheader("交易记录") if account.orders: orders_df = pd.DataFrame([(o.filled_time, o.symbol, o.direction, o.quantity, o.filled_price, o.commission) for o in account.orders], columns=['时间', '代码', '方向', '数量', '成交价', '手续费']) st.dataframe(orders_df.sort_values('时间', ascending=False)) else: st.info("本次回测期间无交易。") # 如果没有运行回测，显示一些介绍信息 else: st.info("请在左侧配置参数，然后点击【开始回测】。") st.markdown(""" ### 系统功能说明 1. **数据完全离线**：所有分析基于本地CSV文件，无需网络连接。 2. **灵活策略回测**：支持自定义策略（需修改代码）。 3. **专业绩效分析**：提供夏普比率、最大回撤等关键指标。 4. **交互式可视化**：图表可缩放、平移，方便细节分析。 """)

运行这个Streamlit应用只需要在终端执行streamlit run app.py。它提供了一个直观的界面，让用户可以点选股票、调整参数、运行回测并即时查看图表和结果，这比直接运行脚本友好得多。

5.2 策略优化与过拟合陷阱

当你有了一个能跑通的回测系统，很容易陷入不断修改参数（如均线周期、仓位比例）以使历史曲线更漂亮的陷阱。这就是过拟合。为了避免生产一个“纸上谈兵”的策略，你必须：

1. 样本外测试：将历史数据分为训练集（用于开发策略）和测试集（用于验证策略）。绝对不能用测试集的数据来优化参数。
2. 交叉验证：对于时间序列数据，可以使用“前向滚动”交叉验证（Walk-Forward Analysis）。例如，用2000-2010年的数据优化参数，在2011年测试；然后用2001-2011年优化，在2012年测试，以此类推。
3. 保持策略逻辑简单：策略的核心逻辑应基于坚实的经济或行为金融学原理（如趋势跟踪、均值回归），而不是通过数据挖掘找到的复杂、脆弱的数学关系。
4. 考虑交易成本与滑点：我们的回测已经包含了佣金，但真实的交易还有滑点（订单成交价与预期价的偏差），在快速波动的市场中尤其明显。可以在回测中引入一个固定的滑点（如0.1%）来让模型更稳健。

一个简单的样本外测试框架思路：

def walk_forward_analysis(full_data, train_years=3, test_years=1): """ 前向滚动分析示例框架。 :param full_data: 完整时间序列的股票数据字典。 :param train_years: 训练集年数。 :param test_years: 测试集年数。 """ # 获取所有数据的共同日期范围 # ... (代码略) start_date = common_dates[0] end_date = common_dates[-1] all_results = [] current_train_start = start_date while True: current_train_end = current_train_start + pd.DateOffset(years=train_years) current_test_start = current_train_end current_test_end = current_test_start + pd.DateOffset(years=test_years) if current_test_end > end_date: break # 1. 在训练集上优化策略参数（例如，网格搜索最佳均线周期） # train_data = 切片数据[current_train_start: current_train_end] # best_params = optimize_strategy(train_data) # 2. 使用最优参数在测试集上运行回测 # test_data = 切片数据[current_test_start: current_test_end] # account = run_backtest(test_data, strategy_with_best_params) # metrics = calculate_performance_metrics(account) # all_results.append(metrics) print(f"训练期: {current_train_start.date()} 至 {current_train_end.date()}, " f"测试期: {current_test_start.date()} 至 {current_test_end.date()}") # 滚动窗口 current_train_start = current_train_start + pd.DateOffset(years=1) # 每年滚动一次 # 分析 all_results 中各项指标（如年化收益、夏普比率）的均值和稳定性 return all_results

重构并扩展这个古老的monichaogu.zip项目，其意义远不止于让一段旧代码重新运行。它是一次完整的、从数据到决策的工程实践。你不仅学会了如何处理离线数据、构建回测引擎、评估策略绩效，还接触到了如何将其产品化为一个可交互的工具，并深刻理解了回测中最重要的“过拟合”陷阱。这套方法论和代码骨架，完全可以作为你探索更复杂量化策略的起点。记住，所有复杂的系统，都是从这样一个简单的闭环开始迭代的。

本文还有配套的精品资源，点击获取