简介基于Python的股票分析系统设计源码包定位为股票分析与Python开发进阶者的综合学习项目完整呈现从行情数据获取、指标计算到可视化展示的系统实现思路。资源共176个文件压缩包31.34MB其中包含116个Markdown文档设计说明、使用指南、34个PNG图表走势图、指标图、3个Python源码核心算法模块、3个Excel工作簿、2个CSV数据文件、3个Word文档与2个PDF报告另有徐翔方法学习、准备事项、红宝书公式系统初/中/高级等特色资料以及测试数据与许可证文件构成了从理论到实践的完整链条。这份源码包不仅便于读者对照代码理解股票分析系统如何搭建还能借助自带的数据与图表复现分析流程学习经典操盘方法与公式系统设计提升金融数据处理与可视化能力。目前已有642人学习/下载适配有一定Python基础、想通过实战项目快速上手股票分析系统开发的学习者。1. 一个基于 Python 语言的股票分析系统听起来像是策略问题真正做起来你会发现它是数据工程问题很多新手拿到「股票分析系统」这个需求第一反应是写策略金叉买、死叉卖均线突破RSI 超卖。真正动手后才会发现开发时间的大头根本不在策略而在数据清洗、复权处理、停牌对齐、回测防作弊这些一点都不性感的环节上。一个基于 Python 语言的股票分析系统本质上是把「行情获取 → 指标计算 → 策略回测 → 结果可视化」这条链路用可复现的代码固化下来让你从「肉眼翻行情软件」变成「用 DataFrame 批量验证想法」。它适合三类人想验证某条均线规则是不是真的有效的个人投资者正在做课程设计需要完整落地的开发者以及要给团队快速搭行情原型的工程师。如果你以为装上库就能直接跑出漂亮收益曲线大概率会在第一步就卡住。2. 数据层用 akshare 拉行情建本地库先解决「数据从哪来」行情数据是整个系统的地基。常见做法是用akshare这类公开接口拉取日线再用 SQLite 做本地存储。之所以不直接落 CSV是因为 CSV 在多标的、增量更新、并发读写场景下非常难受之所以不一开始就上 PostgreSQL是因为个人级分析系统用不到那么重的服务单文件 SQLite 拷贝即备份足够省心。2.1 为什么选 akshare SQLite 这套组合akshare 把多个公开数据源封装成了统一的 DataFrame 接口省去了你自己去处理请求头、反爬、字段映射的重复劳动。SQLite 是 Python 标准库自带支持的嵌入式数据库不需要单独起服务建库、查询、备份都极其简单。这套组合对单机、百来个标的、日线级别频率的场景完全够用。另一个常见选择是 tushare但 pro 接口需要 token 且部分接口有积分门槛。如果你只是想快速把系统跑通akshare 零门槛的优势非常明显。要注意的是这类接口本质是公开数据源的二次封装不是商业行情终端不要指望毫秒级实时推送做日线级别分析没问题做 tick 级交易就超出它的定位了。表结构设计上我的建议是建一张daily_bar明细表字段为交易日期、证券代码、开高低收、成交量、成交金额。主键用(symbol, date)这样天然支持增量更新去重。日期字段用 TEXT 存储而不是 DATETIME因为 SQLite 没有原生日期类型TEXT 在比较和排序时依然符合字典序Python 侧再通过 pandas 转时间戳即可。2.2 建表与首次写入从拉取到落库的完整代码下面是系统里最核心的一段数据层代码完成建表、拉取、清洗、入库四个动作。注意我用了INSERT OR REPLACE而不是to_sql这样可以避免重复运行脚本时产生重复数据。import sqlite3 import akshare as ak import pandas as pd DB_PATH stock.db def init_db(db_pathDB_PATH): 建表主键 (symbol, date)保证同一标的一天只有一条记录。 conn sqlite3.connect(db_path) conn.execute( CREATE TABLE IF NOT EXISTS daily_bar ( date TEXT NOT NULL, symbol TEXT NOT NULL, open REAL, high REAL, low REAL, close REAL, volume INTEGER, amount REAL, PRIMARY KEY (symbol, date) ) ) conn.commit() conn.close() def download_daily(symbol: str, start_date: str, end_date: str, db_pathDB_PATH): 拉取前复权日线并入库。 df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustqfq, # 前复权保证历史价格连续均线/MACD计算不受除权跳空影响 ) if df is None or df.empty: print(f{symbol} 无数据跳过) return df.rename( columns{ 日期: date, 开盘: open, 最高: high, 最低: low, 收盘: close, 成交量: volume, 成交额: amount, }, inplaceTrue, ) df[symbol] symbol df[date] pd.to_datetime(df[date]).dt.strftime(%Y-%m-%d) conn sqlite3.connect(db_path) rows df[[date, symbol, open, high, low, close, volume, amount]].values.tolist() conn.executemany( INSERT OR REPLACE INTO daily_bar (date, symbol, open, high, low, close, volume, amount) VALUES (?, ?, ?, ?, ?, ?, ?, ?) , rows, ) conn.commit() conn.close() print(f{symbol} 入库 {len(rows)} 条{start_date} ~ {end_date})代码的逻辑分四步走第一步建表主键约束保证了重复写入时不会产生脏数据第二步拉数据核心参数是adjustqfq第三步清洗把中文列名改成英文日期统一成YYYY-MM-DD字符串第四步入库用executemany批量写入速度远快于逐条 insert。参数说明里最值得关注的是adjust。qfq表示前复权适合均线、MACD 这类技术指标计算如果你要做除权除息事件分析可以用hfq后复权分析当日真实成交盈亏场景时才需要考虑用不复权数据。我的习惯是本地库里存一份前复权数据作为指标计算主数据源不复权数据单独存另一张表备用。2.3 数据完整性校验缺口、停牌、缺失复权因子三件套数据入库之后绝不能直接拿去算指标。行情接口偶尔会返回缺失日、重复日甚至某一天高低价为 0 的脏数据。我一般会做一个校验函数重点看三件事记录数是否合理、日期有没有大缺口、关键字段有没有异常值。def check_daily(db_path, symbol): 检查某个标的的数据完整性和明显脏数据。 conn sqlite3.connect(db_path) cur conn.cursor() cur.execute(SELECT COUNT(*), MIN(date), MAX(date) FROM daily_bar WHERE symbol?, (symbol,)) cnt, min_date, max_date cur.fetchone() print(f{symbol}: {cnt} 条, {min_date} ~ {max_date}) cur.execute(SELECT date FROM daily_bar WHERE symbol? ORDER BY date, (symbol,)) dates [r[0] for r in cur.fetchall()] gaps [] for i in range(1, len(dates)): d0 pd.Timestamp(dates[i - 1]) d1 pd.Timestamp(dates[i]) if (d1 - d0).days 4: # 周末间隔2天普通节假日3-4天超过4天需要人工确认 gaps.append((dates[i - 1], dates[i])) cur.execute( SELECT date, open, high, low, close, volume FROM daily_bar WHERE symbol? AND (high low OR volume 0 OR close 0), (symbol,), ) bad_rows cur.fetchall() conn.close() return cnt, gaps, bad_rowsgap 判断里的阈值 4 天是个经验值。周末是 2 天间隔国庆、春节这类长假会产生 7 到 9 天的间隔所以超过 4 天不一定就是数据缺失需要人工确认是不是法定长假。停牌和缺失不一样停牌日本来就不产生 K 线回测时直接跳过就好不需要补数据但如果是接口漏数据就需要重新拉取或换数据源补齐。千万不要用fillna给缺失日硬填前收盘价那样回测里会出现「停牌日也能成交」的假象。3. 指标层用 pandas 向量化实现 MA / MACD / RSI 三个核心指标数据就绪之后进入指标计算层。这一层的目的不是实现教科书公式而是把市场上最常用的技术指标变成系统里可复用的函数。很多第一次写指标的人会用 for 循环逐行算数据量几百行时感觉还行一旦标的数量增加到几百个性能问题马上暴露。pandas 的向量化计算才是正确做法。3.1 为什么指标计算要向量化而不能写 for 循环pandas 的rolling、ewm、shift这些方法底层是 C 语言实现计算一次窗口聚合非常快。而 Python 的 for 循环每迭代一次都要做类型检查和对象构造数据量到几十万行时差距是数量级的。更重要的是向量化代码的意图更清晰一行rolling(20).mean()表达的是「20 日均值」而 for 循环需要一屏代码才能看明白。另一个理由是复现性。向量化计算天然和 DataFrame 的索引对齐不容易出现循环里索引错位的问题。指标计算是策略的信号源这里的 bug 会直接污染后续全部回测结果所以必须用最不容易出错、最容易被 review 的写法。3.2 MA、MACD、RSI 的落地代码与参数说明下面是一个统一的指标计算函数输入是清洗好的日线 DataFrame输出是带指标列的同一个 DataFrame。三个指标分别对应趋势、动能和超买超卖三类逻辑。def calc_indicators( df: pd.DataFrame, ma_short: int 5, ma_long: int 20, macd_fast: int 12, macd_slow: int 26, macd_signal: int 9, rsi_period: int 14, ) - pd.DataFrame: 在日线数据上追加 MA / MACD / RSI 指标列。 df df.sort_values(date).copy() # 均线rolling 窗口均值最小窗口内会得到 NaN df[ma_short] df[close].rolling(ma_short).mean() df[ma_long] df[close].rolling(ma_long).mean() # MACD快慢指数均线差 信号线 ema_fast df[close].ewm(spanmacd_fast, adjustFalse).mean() ema_slow df[close].ewm(spanmacd_slow, adjustFalse).mean() df[macd_dif] ema_fast - ema_slow df[macd_dea] df[macd_dif].ewm(spanmacd_signal, adjustFalse).mean() df[macd_hist] (df[macd_dif] - df[macd_dea]) * 2 # 国内软件习惯柱状值乘2 # RSI用 Wilder 平滑法等效 alpha 1 / period diff df[close].diff() gain diff.clip(lower0).ewm(alpha1 / rsi_period, adjustFalse).mean() loss (-diff.clip(upper0)).ewm(alpha1 / rsi_period, adjustFalse).mean() df[rsi] 100 - 100 / (1 gain / loss) return df参数说明是理解这段代码的关键。ma_short5和ma_long20是短中期均线最常用的组合MACD 的三个参数 12、26、9 是整个市场默认的标准组不要轻易改改完之后你得到的 MACD 和别人说的 MACD 就不是一回事了。ewm(adjustFalse)表示使用递归式指数加权而不是修正起始值的方式这样计算结果才和行情软件对上。RSI 这里用的是 Wilder 平滑而不是简单移动平均。两者的区别在于 Wilder 的衰减系数是1/period对近期价格变化的响应更平滑这是通达信、同花顺等软件里 RSI 指标的标准算法。如果你用ewm(spanrsi_period)结果会有差异单看数值可能差别不大但和别人讨论时会出现对不上的尴尬。3.3 指标计算里的时序陷阱未来函数和 NaN 对齐指标计算最容易翻车的有两处。第一处是未来函数比如你在回测里用df[close] df[ma_short]生成买入信号然后当天收盘价成交这在实盘里做不到因为信号产生时收盘价还没确定。正确做法是信号列整体shift(1)交易日 T 收盘后生成信号T1 开盘再执行。回测引擎里会重点讲这个但指标层就要为它做好准备。第二处是 NaN 对齐。rolling(20).mean()在数据前 19 行会产生 NaN如果你不管它直接拿去生成信号那么前 19 天会被当作「不满足条件」实际上是缺失而非信号为 False。我习惯在指标计算完成后显式丢弃指标为 NaN 的行df calc_indicators(df) df df.dropna(subset[ma_short, ma_long, macd_dif, macd_dea, rsi])这一步能避免很多隐蔽 bug尤其是你后面做多标的循环回测时每个标的的起始日期可能不一样NaN 的分布也不一样统一 dropna 能让结果更干净。4. 回测层一套最小系统把策略从黑匣子变成可复现结果回测是整个系统的核心也是新手最容易自我欺骗的地方。一个回测结果不可复现、成交价不切实际、手续费被忽略的系统跑出来的收益曲线再漂亮也只是一张图。我见过太多人卡在这一层同样的策略别人跑年化 30%自己一跑就是亏损原因往往不在策略本身而在回测引擎的细节处理上。4.1 模块划分数据、指标、策略、回测、可视化各管什么整个系统我习惯分成五个模块边界清楚了代码才不至于变成一团麻数据模块负责拉取、清洗、存储行情数据对外提供load_data(symbol)接口。指标模块负责计算 MA、MACD、RSI 等指标输入 DataFrame输出带指标列的 DataFrame。策略模块负责根据指标列生成信号输入指标数据输出 signal 列0 或 1。回测模块负责根据信号模拟交易处理成交价、手续费、持仓输出交易记录和净值序列。可视化模块负责画净值曲线、回撤曲线、买卖点标记。这五个模块之间的依赖是单向的数据 - 指标 - 策略 - 回测 - 可视化。任何时候想加一个新策略只需要替换策略模块回测引擎不用改动。4.2 最小回测引擎下一日开盘价成交手续费和滑点一个不少我一般会先写一个最小可用的回测循环不追求性能先保证逻辑正确。核心规则只有一条信号只决定次日的动作绝不允许当天信号当天成交。def backtest(df: pd.DataFrame, signal_col: str signal, init_cash: float 100000.0, fee_rate: float 0.0003, slippage_rate: float 0.001): 极简回测引擎T日产生信号T1日用开盘价成交。 返回持仓记录、交易记录和逐日权益。 data df.sort_values(date).copy() data[signal] data[signal_col].shift(1).fillna(0) # 关键信号后移一天 cash init_cash position 0 # 持股数量按整手100股取整 trade_log [] equity_curve [] for idx, row in data.iterrows(): price row[open] sig row[signal] # 买入信号为1且空仓 if sig 1 and position 0: if price 0: buy_volume int(cash / (price * (1 fee_rate slippage_rate)) / 100) * 100 if buy_volume 0: cost buy_volume * price fee cost * fee_rate slip cost * slippage_rate cash - (cost fee slip) position buy_volume trade_log.append((row[date], BUY, price, buy_volume, fee slip)) # 卖出信号为0且持仓 elif sig 0 and position 0: income position * price fee income * fee_rate slip income * slippage_rate cash (income - fee - slip) trade_log.append((row[date], SELL, price, position, fee slip)) position 0 # 每日权益 现金 持仓市值 equity cash position * row[close] equity_curve.append((row[date], equity)) return trade_log, equity_curve这段代码里最关键的参数是slippage_rate0.001。很多人回测翻车就是忽略了滑点真实交易里你的买入价不会正好是开盘价总有几分的偏差。0.001 是相对温和的千一滑点如果你是做小票或高频一点的策略我建议直接设到 0.002 甚至 0.003。fee_rate0.0003是单边万三佣金卖出时还会加印花税这里为了可读性没有单独拆实际使用时把卖出费率调整到千一以上更稳妥。买入按整手取整的逻辑也会影响结果int(... / 100) * 100保证不买零股更贴近 A 股真实交易规则。4.3 绩效指标年化收益、最大回撤、夏普有了权益曲线下一步就是算绩效。这一层有一堆现成库可以用但手写一个也不复杂还能避免黑匣子效应。import numpy as np def performance(equity_curve: list, periods: int 252): 输入 (date, equity) 列表输出常用绩效指标。 df pd.DataFrame(equity_curve, columns[date, equity]) df[ret] df[equity].pct_change() df df.dropna() total_return df[equity].iloc[-1] / df[equity].iloc[0] - 1 ann_return (1 total_return) ** (periods / len(df)) - 1 # 最大回撤净值从峰值回落的幅度计算累计净值对历史峰值的最大跌幅 roll_max df[equity].cummax() drawdown df[equity] / roll_max - 1 max_drawdown drawdown.min() # 夏普比率无风险利率近似取 0按年化标准差折算 sharpe df[ret].mean() / (df[ret].std() 1e-12) * np.sqrt(periods) return { total_return: total_return, annual_return: ann_return, max_drawdown: max_drawdown, sharpe: sharpe, trade_count: len(df), }最大回撤的计算我特别提醒一句一定要用净值对历史峰值的回撤而不是对初始净值的回撤。很多人算出来的「回撤」其实是累计亏损这会在策略经历先涨后跌时严重低估风险。夏普比率里的1e-12是防止某段时间收益率为 0 导致标准差为 0 时除零报错。5. 高频避坑数据、指标、回测里 5 个让人翻车的细节这一章写的是我自己在实际搭建系统过程中真实踩过的坑每一个都对应着「现象 → 原因 → 解决」三步。有些坑藏得很深不等到策略实盘你根本意识不到。5.1 前复权数据在均线金叉上反复横跳现象同一只股票换了数据源或调整了复权区间后MA20 的数值变了金叉死叉信号完全对不上。原因前复权价格是以「最新价格」为基准回算历史价格每一次除权除息后历史价格都会被重新计算。这意味着你拉取的数据区间不同复权基准就不同指标自然对不上。解决全系统统一使用前复权数据并且把复权因子也落库保存保证任何一次分析都基于同一份数据快照。不要在同一个项目里混用前复权和后复权这是最容易犯的隐性错误。5.2 停牌日不产生 K 线rolling 窗口「跳格」算现象某股票停牌 10 天复牌后第一个交易日MA20 和 RSI 的值看起来怪怪的和行情软件对不上。原因停牌期间没有 K 线数据pandas 的 rolling 窗口按实际存在的行数滑动而不是按日历日滑动。停牌前的最后一天到复牌日之间缺失了 10 天但 rolling 认为它们就是相邻的两天直接算进窗口里了。解决如果只是少量停牌可以不用处理回测时跳过缺失日即可如果策略对停牌敏感就需要在指标计算前把缺失日补成前收盘价补完之后再算指标最后回测时把这些填充日标记为「不可交易」。5.3 回测收益虚高信号信号你到底在哪一天生效现象回测年化 60%实盘一个月亏掉回测半年的收益怎么看怎么不对劲。原因最典型的偷价行为。代码里用了当日收盘价生成信号又在同一日的收盘价上成交这等于开了天眼——你还没收盘就知道收盘价是多少。每一笔交易都偷跑一天复利放大后收益严重虚高。解决信号列shift(1)强制信号在下一个交易日生效。这是回测引擎里的铁律我用四年的血泪经验告诉你所有「看起来太美好」的回测曲线先查信号和成交日的关系。5.4 数据源限频循环拉全市场时被服务端拒绝现象一次性拉 500 个标的的日线跑到第 120 个开始大量报错返回空 DataFrame。原因公开数据接口普遍有频率限制短时间大量请求会被临时封禁或限流。这不是代码 bug是接口的自我保护机制。解决在循环拉取的代码里加time.sleep(1)控制节奏同时把拉下来的数据缓存到本地 SQLite下次优先读本地。即使以后接口挂了历史数据还在这就是本地仓库最大的价值。5.5 MACD 信号在窄幅震荡里反复打脸现象MACD 金叉买入死叉卖出一个月里来回交易七八次手续费吃掉了大部分利润净值曲线锯齿状下滑。原因震荡行情里 MACD 的 DIF 和 DEA 线频繁穿越零轴产生大量噪音信号。这是趋势指标在无趋势市场里的通病不是代码算错了是策略本身的选择问题。解决给信号加趋势过滤器比如要求收盘价在 MA60 之上才允许做多或者在打分制里降低单一指标的权重。这也是下一章要展开的进阶技巧。6. 进阶技巧用打分制组合信号摆脱单指标反复打脸单指标信号最大的问题是脆弱金叉买、死叉卖在震荡市里会连续亏损止损一次就是真金白银。我后来把思路从「单指标触发」改成了「多指标打分」效果提升非常明显。核心思想很简单不再要求某一个信号精确命中而是让多个弱信号相互验证累计达到一定阈值才动手。打分规则我一般这样设计条件分值设计理由收盘价 MA202中期趋势向上这是最强过滤器MACD 的 DIF DEA2动能处于多头区间RSI 在 50-70 之间1强势但不超买避免追高风险收盘价高于 20 日前1中期动量仍然有效总分 6 分我一般设定 4 分作为入场阈值。低于 4 分不动作等于把所有单指标信号都过滤了一遍。代码实现非常直接def score_signal(df: pd.DataFrame) - pd.DataFrame: 把多指标打分合并为一个最终信号。 df df.copy() df[score] 0 df[score] (df[close] df[ma_short]).astype(int) * 2 df[score] (df[macd_dif] df[macd_dea]).astype(int) * 2 df[score] ((df[rsi] 50) (df[rsi] 70)).astype(int) df[score] (df[close] df[close].shift(20)).astype(int) df[signal] (df[score] 4).astype(int) return df这里的阈值设置是有讲究的。4 分意味着至少要拿到「趋势 动能」两个核心条件或者「趋势 两个辅助条件」单一指标不可能触发入场。这样在震荡市里RSI 和动量方向不一致分数自然上不去交易频率显著下降。我自己的经验是打分制策略的交易次数通常只有单指标策略的三分之一手续费成本大幅降低净值曲线平滑很多。这个打分框架是完全可以扩展的你想加入成交量确认就加一列volume volume_ma5条件你想加入风险控制就把总分数低于 3 分时强制清仓写成一个规则。打分制的本质是让多个维度的证据互相印证而不是把宝押在某一个指标上。最后说一个我一直以来的习惯每跑完一轮回测先看最大回撤再看年化收益先画净值曲线再算夏普比率。回测结果如果最大回撤超过 20%无论收益多高我都不会投入真金白银。我曾经用收盘价成交的回测引擎跑出年化 80% 的漂亮曲线后来改成次日开盘成交后直接腰斩那次翻车让我养成了「先怀疑数据、再怀疑策略」的排查顺序。这套系统的价值不在收益预测而在让你每一次决策都有据可查、每一笔交易都能复盘希望帮到你。本文还有配套的精品资源点击获取