企业官网建设流程全解析

简介：这是一套基于Python的股票分析系统完整源码包，面向金融数据分析爱好者、Python开发者及量化入门者，提供从行情数据获取、指标计算到可视化展示的一站式分析框架。资源共176个文件，压缩包约31.34MB，涵盖116个Markdown说明文档、34张PNG图表、3个核心Python源文件、3个Excel工作簿、3个Word文档、2个CSV数据文件及2个PDF手册等。其中Markdown文档讲解设计理念与使用说明，Python源码覆盖数据获取、处理和结果展示等关键模块，PNG图表直观呈现价格、成交量等分析结果，Excel与CSV便于数据记录和二次计算，Word与PDF提供详细参考。整套资源既是一份可运行的学习工具，也是理解Python金融分析完整流程的典型案例，适合对照文档逐步阅读源码，尤其适合在真实数据上练习分析操作。目前已有642人学习下载。

1. 股票分析系统源码：先看清它替你把哪些环节自动化了

大多数想把技术分析落到代码里的人，第一反应是“找个现成的看盘软件”。但这套基于 Python 语言的股票分析系统设计源码，思路和我惯常推荐的一致：它不是一个黑匣子，而是把行情获取、数据清洗、指标计算、信号生成、回测验证和可视化这条链路拆成了可独立修改的模块。你先用自己的股票池跑一遍历史日线，再逐步替换觉得不对的规则，比打开软件点指标要踏实得多。适合具备基础 Python 语法、对 pandas 有初步了解、想自己掌控数据和分析逻辑的从业者或学生；纯粹想“一键出买卖点”的用户，这套源码对你反而是负担，因为所有参数都得自己调。下载后先跑通数据模块，再动指标和回测，顺序反了容易越改越乱。

2. 数据管道设计：行情获取、清洗与本地缓存的实现

判断一套股票分析系统源码靠不靠谱，我第一眼看数据层而不是指标层。原因很简单：指标再花哨，喂进去的数据如果是脏的，结论全是白做。常见的设计是把它拆成三层：采集、清洗、缓存。采集负责从数据源拿原始行情；清洗负责统一字段、去掉停牌日和空值；缓存负责把已经拉过的数据落到本地文件，防止跑一次策略就要全量重拉一次。

2.1 数据源选型：免费接口的稳定性与限流差异

A股日线行情可选的开源数据源里，我实际维护过 baostock、akshare、tushare pro 三条线路，选型时主要看三个维度：是否需要 token、是否有单次行数限制、字段是否齐全。这里先给一张对比表，后面再写代码。

我一般建议把 baostock 作为回测默认源，因为它不需要注册 token，返回的日线字段里直接带 tradestatus（交易状态）和 isST（是否 ST 股），这两列在构造股票池和剔除停牌样本时非常实用。akshare 适合补当日实时数据，但它的接口本质是对网页接口的封装，每隔几个月就可能改字段名，所以尽量封装在自己的 fetch 函数后面，不要让上层代码直接依赖它。

下面是一个基于 baostock 的日线获取函数，股票代码用完整前缀，例如 sh.600000 或 sz.000001：

import baostock as bs import pandas as pd def fetch_daily(stock_code, start_date, end_date, adjustflag="2"): lg = bs.login() rs = bs.query_history_k_data_plus( stock_code, "date,code,open,high,low,close,preclose,volume,amount,turn,tradestatus,pctChg,isST", start_date=start_date, end_date=end_date, frequency="d", adjustflag=adjustflag, ) rows = [] while rs.error_code == "0" and rs.next(): rows.append(rs.get_row_data()) df = pd.DataFrame(rows, columns=rs.fields) bs.logout() return df

这段代码有几个容易被忽略的参数：adjustflag 取 "2" 表示前复权，"1" 是不复权，"3" 是后复权，后面讲复权坑时还要提到；turn 是换手率，pctChg 是涨跌幅，但这两个字段在停牌日经常是空字符串而不是 NaN；isST 用来标记风险警示股票。如果你打算用这个系统做全市场筛选，建议把 tradestatus == "0" 的行在清洗阶段直接过滤掉，否则停牌日的 close 字段经常是空值，拿到指标层会连锁产生一堆 NaN。

2.2 数据清洗与本地缓存：避免每天重复拉全量数据

纯拉数据不做清洗，这个系统活不过第一轮。baostock 返回的原始 DataFrame 里，所有字段都是字符串，数字列也是 str 类型，必须统一转成 float。常见做法是写一个 clean_daily 函数，把所有空字符串先替换成 NaN，再逐列做 pd.to_numeric：

import numpy as np def clean_daily(df): df = df.copy() for col in ["open", "high", "low", "close", "preclose", "volume", "amount", "turn", "pctChg"]: df.loc[df[col] == "", col] = np.nan df[col] = pd.to_numeric(df[col], errors="coerce") df["date"] = pd.to_datetime(df["date"]) df = df[df["tradestatus"] == "1"] # 只保留正常交易日 df = df.dropna(subset=["open", "high", "low", "close"]) df = df.sort_values("date").reset_index(drop=True) return df

为什么先做 replace("", np.nan) 而不是直接 to_numeric？因为 baostock 返回的空字符串经 to_numeric 转出来也是 NaN，但会伴随 CoercionWarning，大量数据时内存会被反复拷贝，慢得明显。另一个细节是停牌日处理，我把 tradestatus == "1" 之外的行全部删掉，这样后续 rolling、ewm 计算时窗口里不会混入无成交的假价格。

数据清洗完，下一步是做好本地缓存。一个完整系统的日常使用流程是：今天跑一次策略，明天增量更新一次，而不是每天重新下载十年历史。我会把每只股票落成一个单独 CSV，文件名用带市场前缀的代码命名，更新时只从本地最后日期开始拉：

import os def update_cache(symbol, start_date, end_date, cache_dir="data"): os.makedirs(cache_dir, exist_ok=True) cache_file = os.path.join(cache_dir, f"{symbol}.csv") if os.path.exists(cache_file): old = pd.read_csv(cache_file, parse_dates=["date"]) last_date = old["date"].max() if last_date >= pd.Timestamp(end_date): return old new = fetch_daily(symbol, last_date.strftime("%Y-%m-%d"), end_date) new = clean_daily(new) merged = pd.concat([old, new]) merged = merged.drop_duplicates(subset=["date"]).sort_values("date") merged.to_csv(cache_file, index=False) return merged full = clean_daily(fetch_daily(symbol, start_date, end_date)) full.to_csv(cache_file, index=False) return full

这里有一个容易翻车的细节：增量拉取的起始日期直接取 old 的 max(date) 是可以的，但如果上次运行后当天尚未收盘，本地最后一行其实是“半成品”，第二天用同一天的日线去拼接就会出现同日期数据被后一次覆盖或出现重复。我一般会在 update_cache 外层再加一个“是否已收盘”的判断，或干脆约定当天行情要到下午收盘后一段时间才允许入库，避免盘中行情污染历史数据集。

2.3 多标的批量拉取与字段统一

当你从单只股票扩展到几十只甚至几百只，最忌讳写一个没有停顿的 for 循环直接怼接口。即使 baostock 不要求 token，也会因为频繁登录登出触发连接限制。我常用的办法有两个：一是复用 baostock 的登录连接而不是每次 fetch 都登录登出，二是给每个拉取函数套一层重试机制。对二手项目，重点先建立统一字段规范，所有行情 DataFrame 拉到本地前都先映射成一套列名：

FIELD_MAP = { "date": "date", "code": "symbol", "open": "open", "high": "high", "low": "low", "close": "close", "volume": "volume", "amount": "amount", "turn": "turnover", "tradestatus": "trade_status", } def normalize_columns(df): return df.rename(columns=FIELD_MAP)

这个映射的好处是，上层指标模块只认 date、open、high、low、close、volume 这六列，未来即使换数据源，也只改 data_fetcher 和 FIELD_MAP。很多股票分析系统源码改不动，本质原因是数据源接口散落在各个业务函数里，换数据源等于重写整个项目。

3. 指标计算引擎：把K线价格变成可执行信号的向量化实现

指标层是股票分析系统里最接近“业务逻辑”的部分。我拿到一套源码会先看它的指标函数是循环实现还是向量化实现——向量化用 pandas 原生方法一跑就是几千条数据，循环实现则会让整个回测慢到没法用。下面以均线、MACD、RSI、KDJ 为例，把常见实现和参数陷阱一次讲清。

3.1 均线与MACD：EMA的平滑方式决定指标是否对齐

MA 就是 close 的滚动均值，没什么玄学，但要注意窗口长度和 min_periods 的关系。如果 rolling 没有指定 min_periods，前 w-1 行是 NaN，对齐到 date 索引后，最前面一段信号天然缺失，回测开始时要主动截断。MACD 的计算则有一个非常隐蔽的坑：EMA 的递归权重。用 pandas 的 ewm 默认参数算出来，和国内行情软件上的 MACD 数值经常对不上，原因在于 adjust 参数。软件里常用的 EMA 是递推式，等价于 adjust=False：

def add_ma(df, windows=(5, 10, 20, 60)): for w in windows: df[f"ma{w}"] = df["close"].rolling(w, min_periods=w).mean() return df def add_macd(df, fast=12, slow=26, signal=9): ema_fast = df["close"].ewm(span=fast, adjust=False).mean() ema_slow = df["close"].ewm(span=slow, adjust=False).mean() df["dif"] = ema_fast - ema_slow df["dea"] = df["dif"].ewm(span=signal, adjust=False).mean() df["macd_bar"] = (df["dif"] - df["dea"]) * 2 return df

这里 (dif - dea) * 2 不是随手乘的：国内主流行情软件把 MACD 柱状图放大一倍，因子设为 2。如果你的策略要和行情软件对照信号，建议保留这个系数。另一个参数细节是 span 与 alpha 的换算，pandas 的 span 等于用 alpha = 2 / (span + 1)，所以 fast=12、slow=26、signal=9 对应的一般参数组合，直接传 span 即可，不要额外再传 alpha，否则双重换算会把周期变成原来的一半。

3.2 RSI与KDJ：平滑算法与除零保护

RSI 的公式看起来简单，但开源社区实现五花八门。国内软件和中国式技术分析文档里，RSI 的均值常用 SMA(X, N, 1) 这种递推均值；国外 TA-Lib 用的是 Wilder 平滑。两者在某些行情阶段计算出的 RSI 会差好几个点，你拿它做超买超卖阈值判断时，30 和 27 可能就不是同一种信号。pandas 里可以用 adjust=False 的 ewm 来模拟递推均值，alpha 取 1/N：

def add_rsi(df, n=14): delta = df["close"].diff() gain = delta.clip(lower=0) loss = -delta.clip(upper=0) avg_gain = gain.ewm(alpha=1 / n, adjust=False).mean() avg_loss = loss.ewm(alpha=1 / n, adjust=False).mean() rs = avg_gain / avg_loss.replace(0, np.nan) df[f"rsi{n}"] = 100 - 100 / (1 + rs) df[f"rsi{n}"] = df[f"rsi{n}"].fillna(50) return df

avg_loss.replace(0, np.nan) 这段是做除零保护：如果连续上涨没有下跌，avg_loss 是 0，RSI 应该取 100，而不是产生 inf。填 50 是保守策略，避免最前面一段 NaN 把后续排序搞乱。KDJ 则要处理另一个问题，即在高低价完全相同的高位横盘区，RSV 的分母为 0：

def add_kdj(df, n=9, k=3, d=3): low_n = df["low"].rolling(n, min_periods=1).min() high_n = df["high"].rolling(n, min_periods=1).max() denom = (high_n - low_n).replace(0, np.nan) rsv = (df["close"] - low_n) / denom * 100 rsv = rsv.fillna(50) df["K"] = rsv.ewm(alpha=1 / k, adjust=False).mean() df["D"] = df["K"].ewm(alpha=1 / d, adjust=False).mean() df["J"] = 3 * df["K"] - 2 * df["D"] return df

老代码里经常看到 KDJ 用 rolling(k).mean() 来平滑 K 值，那是错的。KDJ 的 K、D 两线在典型实现里使用的是递推均值，而不是简单时间窗平均；用 rolling 均值算出来的 K 线会落后软件显示好几天，在横盘震荡时尤其明显，信号对不上就查这里。

3.3 指标输出的边界处理

指标全部算完后，最容易被忽略的是前一段 NaN。MA(60) 的前 59 行必然是 NaN，MACD 起始段也有约几十行不稳定区。我一般会另写一个对齐函数：把 DataFrame 截断到所有指标都不是空值的区间开始。截断时机应在信号生成之后、进入回测之前，否则回测的涨跌幅和持仓序列会对不上。

4. 信号生成与回测：从选股规则到收益验证的完整闭环

当指标层跑完，源码里每个信号函数就变成“给我一行行情，我给你一个布尔值”。但把布尔值直接当交易信号去回测，是我见过的最大翻车点——这里必须把未来函数问题处理清楚。

4.1 多条件选股器的结构设计

选股逻辑不该写死在主流程里。我见到的可维护设计，是把选股条件做成一个规则列表，每个规则是一个输入 DataFrame、输出 bool 序列的纯函数：

def rule_ma_bullish(df): return df["ma5"] > df["ma10"] def rule_close_above_ma20(df): return df["close"] > df["ma20"] def rule_macd_gold_cross(df): return (df["dif"] > df["dea"]) & (df["dif"].shift(1) <= df["dea"].shift(1)) def run_rules(df, rules): signal = pd.Series(True, index=df.index) for rule in rules: signal &= rule(df) return signal

run_rules 里用链式与运算把多个条件合并，多空条件组合时还可以在 rule 里直接取反。这种结构的价值在于，你不必为每个新条件重写循环，加一条规则就等于加一个函数。需要注意&与and的区别：Series 与 Series 之间只能用位运算符，写成 if ... and ... 会直接报 ValueError，很多初跑者是在这里被卡住的。

4.2 信号标记与次交易日执行：防未来函数

假设某天收盘后 MA5 上穿 MA10，这一天的“金叉信号”在收盘后才能确认，你的策略最早也只能在下一个交易日开盘执行。如果回测里当天信号当天成交，盘中价格还没走完就买入，等于用了未来数据。解决方法是把信号整体 shift 一格：

df["signal"] = run_rules(df, rules) df["signal_tomorrow"] = df["signal"].shift(1).fillna(False) df["trade_ret"] = df["signal_tomorrow"] * df["close"].pct_change()

shift(1) 之后，第 t 天的 signal 只影响第 t+1 天的持仓和收益，回测结果才接近实盘。另一个细节是 fillna(False)：第一行没有前一日信号，默认空仓处理，避免回测第一笔收益被错误计入。我在检查别人的源码时，会先搜索有没有在信号后面调用 shift；找不到，就直接判定这套回测结果不可信。

4.3 回测结果评估：年化收益、最大回撤与参数遍历

信号序列对齐后，回测的收益计算可以直接向量化完成，不需要 for 循环逐日模拟：

df["cum_ret"] = (1 + df["trade_ret"].fillna(0)).cumprod() df["peak"] = df["cum_ret"].cummax() df["drawdown"] = df["cum_ret"] / df["peak"] - 1 max_drawdown = df["drawdown"].min() total_days = len(df) annual_ret = df["cum_ret"].iloc[-1] ** (252 / total_days) - 1

这里 252 是 A 股一年大约的交易日数量，计算年化收益时按自然日算会虚高，按 252 算是业内默认基准。最大回撤算的是持仓路径的峰值到谷底的幅度，它比年化收益更能反映策略的承受能力。我会再用一组参数组合跑一个简单的循环，把不同均线窗口下的年化收益和回撤打印成表格，对比后再决定用哪组参数；但要注意，遍历参数找最优结果的过拟合风险也在这里，样本外验证一定要留一段数据不参与寻参。

5. 避坑指南：运行股票分析系统时最常踩的五个坑

这一章把我在维护类似股票分析系统源码时踩过、也替别人排查过的坑列一遍，按“现象–原因–解决”写，每一条都对应实际报错或回测失真。

5.1 数据源接口失败，还没到指标层就崩

现象：批量拉取时，跑到第 30 只股票程序直接抛异常退出，前面的数据没缓存、后面全部中断。 原因：baostock 或 akshare 对短时间内的连接次数有限制，或者目标接口临时维护，而原代码没有重试机制。 解决：给 fetch 函数包一层带延时重试的装饰器，失败后等 1 到 2 秒再试；同时把每只股票的成功结果立刻写本地缓存，下次断点续跑，不再重复拉已经落盘的标的。

5.2 回测曲线很漂亮，实盘完全走样

现象：信号代码在历史数据上收益很高，最大回撤也小，拿到当下市场却连续亏损。 原因：八成是在信号生成或成交价上用了未来数据，或者把“当天收盘信号”直接用于“当天开盘成交”。 解决：把信号整体 shift(1)，并强制约定回测成交价只能取次日开盘价或次日收盘价。建议在回测里加一个断言，检查所有买入信号日期对应的成交日期都比信号日期晚，防止未来函数悄悄混进去。

5.3 除权日前后指标跳变，复权方式混用

现象：某只股票分红送股后，均线在除权日当天出现异常拐点，MACD 出现假金叉。 原因：拉取历史行情时用了不复权，而指标体系里又按前复权价格计算；或不同股票用了不同复权参数。 解决：全项目固定一种复权方式。做历史回测我推荐前复权，因为当前视角看历史价格是连续调整过的；但在算真实收益率时不要直接用复权后的 close 去做 pct_change，而是用原始价格结合复权因子单独算收益，否则每股真实涨跌幅会被复权因子扭曲。

5.4 pandas链式赋值与内存爆掉

现象：控制台不断弹 SettingWithCopyWarning，跑几百只股票时内存占用涨到几个 G。 原因：代码里用 df[df["xx"] > 0]["close"] = ... 这种链式赋值，pandas 会创建临时副本，写不进原表；多标的循环又每个副本都留在内存里。 解决：第一步改成 df.loc 条件索引赋值；第二步按股票循环后主动 del 中间变量，或在函数作用域内完成合并后再返回，避免大 DataFrame 在内存里到处留引用。

5.5 数值列全是字符串，指标算出来全是NaN

现象：指标函数算出来的 ma5、macd 整列都是 NaN，回测结果又全部为空。 原因：原始行情来自 CSV 或 baostock 返回结果，price 列是 object 类型，rolling 均值不会对字符串做数值运算。 解决：在 clean 阶段统一 pd.to_numeric，并用 errors="coerce" 把无法转换的脏数据变成 NaN；转换后立刻检查 df.dtypes，顺手加一条断言，凡是价格列为 object 就直接报错，别让脏数据流到指标层。

6. 可视化与导出：把信号标到K线上验证策略

信号验证不能只听数字，把买卖点画到K线图上，旁边搭两条均线，肉眼能看出大部分信号逻辑是否合理。我默认用 matplotlib 做一套轻量可视化函数，不引重型 GUI 框架，方便直接嵌入现有分析代码：

import matplotlib.pyplot as plt def plot_strategy(df, symbol, buy_dates, sell_dates): fig, ax = plt.subplots(figsize=(12, 7)) ax.plot(df.index, df["close"], color="#333333", linewidth=1.2, label="close") ax.plot(df.index, df["ma20"], color="#e67e00", linewidth=1, label="ma20") if "rsi14" in df.columns: axr = ax.twinx() axr.plot(df.index, df["rsi14"], color="#888888", linewidth=0.8, alpha=0.6) axr.axhline(70, linestyle="--", color="#999999", linewidth=0.8) axr.axhline(30, linestyle="--", color="#999999", linewidth=0.8) axr.set_ylim(0, 100) ax.scatter(buy_dates, df.loc[buy_dates, "close"], marker="^", color="#d62728", s=70, label="buy") ax.scatter(sell_dates, df.loc[sell_dates, "close"], marker="v", color="#2ca02c", s=70, label="sell") ax.set_title(f"{symbol} strategy signal", loc="left") ax.legend() plt.tight_layout() return fig

这个函数里的 buy_dates、sell_dates 建议直接取信号确认日，而真实成交日是次日，画图时把两条线对比看：如果买入点总在下跌途中出现，说明筛选规则偏向抄底；如果买入点密集出现在上涨末端，说明均线参数太灵敏。导出结果的最后一步可以把选股结果和关键指标落成 Excel：

df.loc[df["signal_tomorrow"], ["date", "close", "ma5", "ma10", "ma20", "rsi14", "macd_bar"]].to_excel("signals_export.xlsx", index=False)

从那以后，我每完成一次指标修改，都会强制走一遍：先看收益率数字，再把买卖点画到图上，最后导出一份 Excel 明细核对日期。光有数字看不出信号错位，光看曲线又记不住准确日期，三个动作连起来，坑就少了一半。这套源码下载后，建议保留 data 和 backtest 两个模块的目录结构，改的时候只动 indicators.py 和 rules.py，希望帮到你。

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