简介这是一份面向计算机、通信、自动化等专业学生与从业者的股票自动交易系统毕业设计源码包适合用作毕业设计、课程设计或期末大作业的参考底稿。项目围绕行情数据获取、自动交易策略生成与订单执行的完整链路展开内含Java服务端业务逻辑、前端操作页面、SQLite数据库与辅助脚本整体结构贴近真实工程项目。压缩包共686个文件以Java源码、HTML/CSS/JS前端资源、JPG/PNG界面截图及配置数据为主Java源码承担核心交易逻辑前端文件负责交互界面呈现图片素材便于比对运行效果另含图标字体、Dockerfile、数据文件等部署辅助内容解压后约14.77MB。资源由个人毕设项目整理而来答辩评审分达到95分代码经过调试可运行已吸引336人学习下载。对于想要快速搭建同类交易系统或深入理解毕设模块拆分、前后端联调与数据库设计的读者这份包既可作为入门学习的基础也能支撑二次开发改造。1. 用 Python 写股票自动交易系统最难的其实不是下单代码拿 Python 做股票自动交易系统是毕业设计里最热门的方向之一也是很多想入行量化的人第一个动手项目。但真正做完的人往往发现最难的不是把买卖逻辑写成代码而是把「数据、策略、下单、风控」四块串成一个闭环时每一步都有让人头大的工程细节。这个项目标题里的源码包本质上就是解决这四块怎么落地的问题——数据从哪来、信号怎么算、订单怎么发、亏了怎么止损以及回测和实盘之间为什么总是差一截。这篇笔记按照我平时做这类系统的习惯把整个链路拆开讲给你一份可以直接照着写、照着改的落地方案。适合正在做毕设的学生也适合想用 Python 搭第一版自动化交易框架的开发者。2. 从数据到下单自动交易系统的四大模块与整体流程2.1 系统架构为什么要把「策略」和「执行」拆开一个能跑的股票自动交易系统最少要包含四个模块数据模块、策略模块、执行模块、风控模块。很多初学者喜欢在一个脚本里既拉数据又算信号又下单跑通的时候很爽但后面每改一个参数都要动整个文件回测和实盘也纠缠在一起出了错根本不知道是哪一环的问题。我一般会按这样的分层来组织代码# 项目结构建议 stock_trader/ ├── data/ # 数据获取与缓存 ├── strategy/ # 策略信号计算 ├── executor/ # 订单执行与状态管理 ├── risk/ # 风控检查 ├── backtest/ # 回测引擎 ├── config.py # 全局参数配置 └── main.py # 入口回测 / 模拟盘 / 实盘这个结构的好处是策略永远是纯函数输入历史行情输出买卖信号不关心数据怎么来的、订单怎么发的。这样回测和实盘用的是同一套策略代码不会出现「回测一套逻辑、实盘另一套逻辑」的经典翻车。执行模块只负责把策略信号翻译成订单再处理订单的状态流转。风控独立成一个模块专门做「策略说买但今天能不能真买」的判断。参数配置单独放一个config.py这是容易被忽略但很重要的事。策略参数、交易标的、资金上限、单笔仓位、滑点假设、手续费率全部集中在这里。毕设答辩时老师问「你改了哪些参数」你能直接说出改了什么、为什么改这比代码本身更加分。2.2 用 akshare 拉取行情数据不要自己写爬虫数据是自动交易的地基。很多人想到拉行情第一反应是自己写爬虫去抓网页这是最浪费时间也最脆弱的路子。现在 Python 生态里已经有很成熟的数据接口免费、无需 Key适合回测和模拟盘使用。akshare 是常用的选择tushare 也很流行但 tushare 的积分门槛对新手不友好所以我这里用 akshare 举例。# data/fetcher.py import akshare as ak import pandas as pd from datetime import datetime, timedelta def fetch_daily(symbol: str, days: int 500) - pd.DataFrame: 获取日线行情并做基础清洗 symbol: 股票代码如 600519不带市场前缀 days: 拉取最近多少天 end datetime.now().strftime(%Y%m%d) start (datetime.now() - timedelta(daysdays)).strftime(%Y%m%d) # 这里以东财接口为例akshare 的接口名随版本会有调整 # 版本升级后建议先打印 ak.__version__ 再对照文档确认函数名。 df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart, end_dateend, adjustqfq # 前复权回测必须用复权价 ) # 统一列名为英文方便后续计算 df.columns [date, open, close, high, low, volume, amount, amplitude, pct_change, change, turnover] df[date] pd.to_datetime(df[date]) df df[[date, open, close, high, low, volume]].copy() df df.sort_values(date).reset_index(dropTrue) return df if __name__ __main__: data fetch_daily(600519, days250) print(data.tail())这段代码有几个参数需要注意。adjustqfq是回测的生命线如果不用前复权分红除权日股价会跳空均线会被假信号干扰策略可能莫名出现「一夜暴跌然后反弹」的假买卖点这是新手最容易踩的坑。days500大约覆盖两年交易日做双均线这类慢策略够用如果做短周期策略建议直接拉全量历史不传days参数或传较大值本地缓存后切片使用。akshare 接口偶尔会因为上游数据源变动而调整函数名或字段名这点用的时候要有心理准备。碰到报错先看异常信息的最后一行通常是列名对不上或接口失效更新库版本后再试就好。2.3 信号计算把「金叉买入」变成可执行的条件策略模块是整个系统的核心也是毕设论文里最好写的部分。这里用双均线策略作为最小可用示例因为它逻辑清晰、参数少、容易讲明白而且你能很容易地把均线换成 MACD、KDJ 或你自己设计的因子。# strategy/ma_cross.py import pandas as pd def ma_cross_signal(data: pd.DataFrame, fast: int 5, slow: int 20) - pd.DataFrame: 双均线交叉策略 规则: 快线上穿慢线 - 买入(1)快线下穿慢线 - 卖出(-1)否则持有(0) 返回带 signal 列的 DataFrame df data.copy() df[ma_fast] df[close].rolling(windowfast).mean() df[ma_slow] df[close].rolling(windowslow).mean() # 前一天的金叉/死叉状态用 shift(1) 避免用到当天收盘后的信息 df[pre_fast] df[ma_fast].shift(1) df[pre_slow] df[ma_slow].shift(1) df[signal] 0 # 金叉今天快线在慢线上方且昨天快线在慢线下方 df.loc[(df[ma_fast] df[ma_slow]) (df[pre_fast] df[pre_slow]), signal] 1 # 死叉今天快线在慢线下方且昨天快线在慢线上方 df.loc[(df[ma_fast] df[ma_slow]) (df[pre_fast] df[pre_slow]), signal] -1 return df这里有一个关键细节用shift(1)做信号判断。你在盘中收盘后才能确认当天的均线关系所以信号要滞后一天才可执行这避免了一个很隐蔽的问题——未来函数。如果不做 shift回测里当天收盘出信号、当天就以收盘价成交实盘根本做不到回测收益率会虚高得离谱这是量化里最常见的「自己骗自己」。参数fast5, slow20是双均线最常用的默认组合代表短周期 5 天、长周期 20 天。改成fast10, slow30或fast20, slow60信号频率会变低、每次持有的周期变长适合不同交易风格。做毕设时建议做一组参数敏感性分析同一个时间段跑 3~4 组快慢参数看收益率、胜率、最大回撤的变化这能让论文内容厚实很多。2.4 把信号变成订单状态机与模拟撮合策略算出了 signal下一步是把它变成真实的买入卖出动作。这里最容易翻车的地方是信号是连续序列但你要保证「买过一次之后在卖出信号出现前不再重复买」不能每个金叉都下一单。所以执行模块需要一个状态机——记录当前是否有持仓只有在「空仓时收到买入信号」或「持仓时收到卖出信号」才真正下单。# executor/simulator.py class SimulatedBroker: def __init__(self, init_cash: float 100000.0, commission: float 0.0003): self.cash init_cash self.position 0 # 持仓股数 self.total_asset init_cash self.commission commission self.trades [] # 成交记录 def execute(self, signal: int, price: float, date: str, lots: int 100): 模拟撮合只按整手100股成交 signal: 1买入 / -1卖出 / 0不动 if signal 1 and self.position 0: # 计算可买数量留出手续费空间 max_shares int(self.cash * 0.95 / (price * 100)) * 100 if max_shares 100: return cost max_shares * price fee cost * self.commission if cost fee self.cash: return self.cash - (cost fee) self.position max_shares self.trades.append({date: date, type: BUY, price: price, shares: max_shares, fee: fee}) elif signal -1 and self.position 0: income self.position * price fee income * self.commission self.cash (income - fee) self.trades.append({date: date, type: SELL, price: price, shares: self.position, fee: fee}) self.position 0 # 更新总资产 self.total_asset self.cash self.position * price模拟撮合里有几个参数决定了回测的可信度。commission0.0003是常见的佣金费率万三实际券商可能更低但在回测中留一点余地总比按零费率乐观好。lots100是 A 股的整手制度买入股数必须是 100 的整数倍卖出时不足 100 股可以一次性卖出但买入必须按手来。max_shares的计算里留了 5% 的现金缓冲防止手续费吃掉全部本金导致余额不足。如果这是真实盘环境执行模块还要对接券商 API处理委托回报、撤单、查持仓这些更复杂的交互订单状态也不再是简单的一次买入就完事而是 pending - filled - 确认回报的异步流转。以目前的经验看国内实盘 API 的门槛和合规要求都比较高建议把毕设落点放在「本地的回测 模拟撮合」这个层面既能讲清楚原理又不会踩到合规红线。3. 回测引擎怎么写别让你的收益曲线骗了你3.1 回测数据按时间推进一条一条喂给策略回测的核心逻辑很简单把历史数据按时间从前往后逐日推进每天让策略看一眼「截至今天的数据」输出信号然后让模拟撮合去成交记录资金变化。但这里有一个新手容易写错的点策略在计算今日信号时只能用今天之前的数据不能用今天之后的数据做任何计算。# backtest/engine.py import pandas as pd from data.fetcher import fetch_daily from strategy.ma_cross import ma_cross_signal from executor.simulator import SimulatedBroker def run_backtest(symbol: str, fast: int 5, slow: int 20, init_cash: float 100000.0): df fetch_daily(symbol) df ma_cross_signal(df, fast, slow) broker SimulatedBroker(init_cashinit_cash) daily_records [] # 只从 slow 之后开始因为前面 slow 天没有慢均线数据 for i in range(slow, len(df)): row df.iloc[i] date row[date] signal int(row[signal]) price float(row[close]) broker.execute(signal, price, date) daily_records.append({ date: date, total_asset: broker.total_asset, cash: broker.cash, position: broker.position }) result pd.DataFrame(daily_records) return result, broker.trades这个回测引擎的关键点有两个。第一循环从range(slow, len(df))开始因为前slow天的慢均线是 NaN信号没有意义强行计算会产生大量误判。第二signal在策略里已经用shift(1)处理过所以当天的信号在当天开盘或盘中理论上是可以执行的这里用收盘价成交是一种近似。严格一点的回测应该用「次日开盘价」成交也能减少偏差。broker.total_asset每天的记录是后续画净值曲线、算最大回撤的基础。把这段数据导成 CSV用 pandas 或 matplotlib 画出来就是论文里最直观的「资金曲线图」。3.2 三个必看的回测指标收益率、最大回撤、胜率只跑一个总收益率就下结论是典型的自嗨行為。你需要三个指标来综合判断策略质量年化收益率、最大回撤从峰值跌下来的最大幅度、胜率赚钱交易笔数 ÷ 总交易笔数。# backtest/metrics.py import numpy as np def calculate_metrics(daily_df: pd.DataFrame, trades: list, init_cash: float 100000.0): total_asset daily_df[total_asset] final_value total_asset.iloc[-1] # 总收益率 total_return (final_value - init_cash) / init_cash # 最大回撤: 每个时间点相对历史最高点的回撤幅度 peak total_asset.cummax() drawdown (total_asset - peak) / peak max_drawdown drawdown.min() # 胜率 if len(trades) 2: return {total_return: total_return, max_drawdown: max_drawdown, win_rate: 0.0} buy_prices {} wins 0 total_closed 0 for t in trades: if t[type] BUY: buy_prices[t[date]] t[price] elif t[type] SELL: total_closed 1 # 找到最近一次买入价格 buy_dates [d for d in buy_prices.keys() if d t[date]] if not buy_dates: continue buy_date max(buy_dates) buy_price buy_prices[buy_date] if t[price] buy_price: wins 1 win_rate wins / total_closed if total_closed 0 else 0.0 return { total_return: round(total_return, 4), max_drawdown: round(max_drawdown, 4), win_rate: round(win_rate, 4) }看输出的时候重点不是某个指标高不高而是三个指标一起看。一个年化 50% 但最大回撤 40% 的策略和一个年化 15% 但最大回撤 5% 的策略后者在实际执行中往往更可能拿得住、活下来。做毕设时建议在论文里放一张表不同参数组合下的三指标对比这比任何文字描述都有说服力。3.3 为什么回测曲线很漂亮、实盘就翻车回测和实盘的差距是量化领域最经典的话题。原因通常出在几个地方用了未来函数我们前面避免了成交价格用了理想价格没有考虑滑点手续费设置得太低忽视了冲击成本以及策略恰好过拟合了某一段历史行情。# 给回测加滑点模拟价格偏差 def simulate_slippage(price: float, slippage_rate: float 0.001) - float: 简单滑点模型: 买入时价格上浮卖出时价格下浮 return price * (1 slippage_rate) # 这里以买入为例滑点设置为你交易频率的参考每天交易的策略滑点影响很大几天才交易一次的策略影响相对小。在实盘之前建议把滑点调到 0.002、手续费调到 0.0005再跑一遍回测如果收益大幅缩水说明策略对交易成本太敏感实盘大概率也跑不出回测的漂亮结果。4. 风控模块自动交易系统的「刹车」在哪4.1 单笔止损与账户止损两条铁律策略解决「怎么赚钱」风控解决「怎么不亏光」。一个没有风控的自动交易系统本质上是给代码发了张无限额度的信用卡。风控模块在信号到达执行模块之前先做校验不通过就拦截订单。# risk/controller.py class RiskController: def __init__(self, max_single_loss: float 0.02, max_drawdown: float 0.10): self.max_single_loss max_single_loss # 单笔最大亏损比例 self.max_drawdown max_drawdown # 账户最大回撤阈值 self.peak_asset 0.0 def check(self, signal: int, current_asset: float, position_value: float, entry_price: float, current_price: float) - int: # 更新历史峰值 self.peak_asset max(self.peak_asset, current_asset) # 规则1: 账户回撤超过阈值禁止开新仓清仓或停止交易 if (self.peak_asset - current_asset) / self.peak_asset self.max_drawdown: return -1 # 强制卖出或停止 # 规则2: 持仓亏损超过单笔止损线强制止损 if position_value 0 and entry_price 0: loss_ratio (current_price - entry_price) / entry_price if loss_ratio -self.max_single_loss: return -1 # 止损卖出 # 规则3: 正常情况放行策略信号 return signalmax_single_loss0.02表示单笔交易亏损超过 2% 就被强制止损这是短线交易者常用的纪律线。max_drawdown0.10是账户级别的熔断总资产从最高点回撤 10%系统停止交易避免连续亏损导致心态崩坏虽然自动交易没有心态但资金曲线崩了就是真的崩了。风控模块必须在策略信号之后、订单发出之前执行。顺序是策略算信号 - 风控检查 - 通过则下单 / 不通过则拦截同时写一条日志记录原因。日志很重要之后复盘时你能看到每一次拦截是被哪条规则触发的。4.2 黑匣子保护方案日志与状态快照自动交易系统跑起来之后你不可能 24 小时盯着它所以日志是保护你的黑匣子。每笔订单的发出时间、价格、数量、信号来源、风控是否拦截全都要记录下来。一旦某天资金曲线突然异常你可以根据日志回溯到具体时点发生了什么而不是对着 K 线图猜。# 日志记录示例: 用 logging 输出结构化信息 import logging logging.basicConfig( levellogging.INFO, format%(asctime)s | %(levelname)s | %(message)s, filenametrader.log ) def log_trade(signal: int, price: float, shares: int, after_risk: int): logging.info(fsignal{signal} - after_risk{after_risk} | price{price} | shares{shares})记录频率不用太高——每秒都记不仅费磁盘还会让你在排查问题时淹没在无关信息里。按订单级别和风控级别记录就够每笔成交、每次拦截、每次策略信号变化各记一条。启动和停止系统时也要记一条方便确认系统是什么时候开始跑的、中间有没有重启过。4.3 模拟盘先行用真实数据试运行两周实盘之前用一个叫「模拟盘」的过渡阶段来验证系统稳定性是代价最低的演练方式。模拟盘的意思是说用真实的行情数据通过你的策略和风控自动发出买卖信号但下单动作对接的是模拟账户比如券商的仿真环境或者自己写一个模拟撮合不产生真实资金流动。这段时间里你需要观察的不是收益而是三件事信号的触发频率是否符合预期太频繁说明参数太敏感太稀疏说明策略太钝风控规则在行情波动剧烈时是否正常工作大跌时有没有触发熔断有没有漏单日志是否完整记录了所有动作有无断档、时间戳是否准确。两周运行正常之后再决定是否接真金白银。很多人跳过模拟盘直接实盘结果第一周就把策略改得面目全非——因为回测没暴露的 bug 全都冒出来了。模拟盘的意义就是给你一个后悔药发现问题推倒重来不花钱。5. 落地过程避坑毕设答辩前最容易翻车的五个「意外」5.1 数据接口突然失效akshare 版本更新导致函数名变更现象昨天还能正常跑通的ak.stock_zh_a_hist今天突然报错AttributeError或返回空 DataFrame。原因akshare 迭代非常频繁上游数据源接口调整后函数签名或返回字段名会跟着变。你锁定的版本可能在别人电脑上正常但换一台机器重新pip install akshare --upgrade后老代码就不兼容了。解决一是把akshare版本固定在requirements.txt里如akshare1.14.0具体版本以你实测为准不要用这种宽松约束二是代码里加一层「表头重命名」兼容逻辑先打印df.columns再统一转换避免硬编码列名顺序。答辩现场最怕的就是演示时数据拉不出来提前把数据缓存成 CSV 文件作为本地备份演示时断网也能跑。5.2 复权价和原始价混用均线在除权日出现假信号现象回测收益不错但拉出交易记录发现某些买入点恰好在除权除息日附近价格有明显跳空。原因数据源默认返回未复权价除权日开盘价会大幅跳水均线计算被带偏策略误判为「超跌反弹」或「死叉卖出」。解决统一在数据获取阶段使用adjustqfq并且全程只在数据层做一次复权处理后续所有模块策略、回测、执行都只消费这个已经处理好的数据不要在其中一层叠加二次复权。回测结果对比时可以单独出一张「不复权 vs 前复权」的收益对比表作为论文里的一个分析点反而加分。5.3 信号重复触发导致连续买入忘了持仓状态判断现象回测成交记录里出现连续几天都买入仓位不断叠加但策略本意是「每次只有一个持仓」。原因策略只输出了「金叉买入」信号执行模块没有判断当前是否已有持仓。信号是连续的状态没有记忆。解决在execute()方法里加入if signal 1 and self.position 0的前置判断「空仓才买、持仓才卖」。这对应了前面模拟撮合代码里的状态机逻辑。更稳健的做法是把这个状态判断提升为独立函数供回测和实盘共用避免两套代码逻辑不一致。5.4 回测用了收盘价成交信号与成交价之间有时差现象回测中每次买入都在信号出现当天以收盘价成交实际模拟盘里按这个价格根本成交不了。原因真实交易中收盘价是全天交易结束后的结果盘中不可能预先知道。策略信号在收盘后才能确定理论上最早应在次日开盘执行。解决回测里把成交价改为「次日开盘价」——循环里多保留一行的开盘价数据信号出现后在下一次循环以row[open]成交。这会让你回测的收益略微下降但更接近真实可成交的价格。如果做的是日内高频策略还需要引入分钟级别的数据来做更精细的回测日线级别的假设就不够用了。5.5 局部最优参数换一段时间区间就跑不动了现象某组参数在 2022 年表现很好换到 2023 年就亏损把时间段改短收益从 30% 变成 -8%。原因策略参数在那段历史行情里「过拟合」了捕捉到的不是市场规律而是那段行情的噪声。解决用滚动窗口做样本外测试比如用 2021-2022 年做训练区间选参数用 2023 年做验证区间看效果验证区间不能参与参数选择。论文里可以明确写出「参数在训练区间确定验证区间只跑一次」的实验设计这比一句「参数经过反复调试」更能体现专业度。6. 让系统更可信从回测到「看得见的可解释性」做到这一步系统能跑、能回测、能有收益曲线了。但要让别人尤其答辩老师相信它值得投入还需要一层「可解释性」——不是只看最终数字而是能说清楚每一笔交易为什么发生、策略在什么市场环境下有效。我习惯给系统加一个报告模块把每次交易的买入理由、卖出理由、持仓周期、盈亏明细自动导出成表格。# 简单交易报告生成 def generate_report(trades: list): report_rows [] for t in trades: report_rows.append({ date: t[date], type: t[type], price: round(t[price], 2), shares: t[shares], fee: round(t[fee], 2) }) report_df pd.DataFrame(report_rows) report_df.to_csv(trade_report.csv, indexFalse, encodingutf-8-sig) return report_df导出 CSV 用utf-8-sig编码是给 Windows 上的 Excel 直接打开不乱码的细节。这个报告的价值在回测之后立刻显现你可以挑出一笔亏损交易回看那几天的 K 线判断是策略逻辑的问题还是市场环境的问题。这种「打开黑匣子」的复盘习惯才是自动交易系统真正值钱的地方——它把每次交易变成可审计的记录而不是一笔糊涂账。另外一个值得做的进阶验证是「多股票泛化测试」把同一套策略放在 5 只不同行业的股票上跑如果只有 1 只赚钱、另外 4 只都亏说明策略只是恰好适合某只股票的走势不具备泛化能力如果 3 只以上表现稳定策略才更可能是发现了某种共性。这个测试不需要额外代码改一下主程序里的股票列表循环就行但结论很有说服力。我做这类系统时最后一步永远不会省的是「让系统自己给自己写小结」每跑完一次回测自动输出一份包含总收益、最大回撤、交易次数、胜率、平均持仓天数的文本摘要直接作为运行留档。这样一段时间后回头看能清楚看到每个版本改了什么、结果好了还是坏了而不是靠记忆里模糊的印象。这个习惯帮我避免了很多次「好像改过但不知道改成啥」的尴尬。希望这次写下的细节能让你的系统少踩几次坑早一点跑在正确的路上。本文还有配套的精品资源点击获取