Python量化选股实战：三天构建自动化股票分析工具

如果你是一名程序员，同时又对投资理财感兴趣，那么你很可能有过这样的念头：能不能用我的技术能力，写个程序来帮我分析股票，甚至自动选股？这个想法听起来很酷，但往往止步于“想法”，因为数据获取、指标计算、策略回测……每一步都像一座大山。

最近，我就花了三天时间，用 Python 把这个想法变成了现实。结果跑出来的数据，确实让我有点惊讶——不是因为它选出了什么“十倍牛股”，而是通过这个亲手搭建的工具，我清晰地看到了技术分析、数据驱动与主观臆断之间的巨大鸿沟。更重要的是，我验证了一个核心判断：对于开发者而言，构建一个基础的量化分析工具，技术门槛远没有想象中高，真正的挑战在于策略逻辑的严谨性和对市场规律的敬畏。

这篇文章，我将分享这个 Python 选股工具从零到一的完整构建过程，并附上全套源码。无论你是想学习 Python 数据分析、对量化投资感兴趣，还是单纯想了解如何将编程技能应用于实际场景，这篇文章都将为你提供一条清晰的路径。我们将避开复杂的金融工程理论，聚焦于可落地的代码实现，让你能亲手复现并理解其背后的每一个环节。

1. 这个工具到底解决了什么问题？

在开始看代码之前，我们必须明确这个工具的目标和边界。它不是一个“圣杯”，不能保证盈利。它的核心价值在于：

1. 自动化数据收集：替代手动从财经网站复制粘贴数据，实现高效、准确的数据获取。
2. 策略规则化执行：将模糊的选股思路（如“市盈率低、近期上涨”）转化为明确的、可重复执行的代码逻辑，消除情绪干扰。
3. 快速回溯验证：可以对历史数据进行回测，初步检验一个想法在过去的市场中是否有效，尽管过去不代表未来，但这是理性决策的第一步。
4. 学习与实践平台：它是学习pandas、numpy、数据可视化以及面向对象编程的绝佳实战项目。

简单说，这个工具是一个“决策辅助系统”。它帮你从4000多只股票中，根据你设定的条件快速筛选出一个小范围的股票池，为你更深度的基本面研究或技术分析节省大量前期时间。它的输出是“候选列表”，而不是“买入指令”。

2. 核心设计思路与技术栈

整个工具的设计遵循“数据输入 -> 规则处理 -> 结果输出”的流水线模式。

技术栈选择：

• Python 3.8+: 量化分析领域的事实标准，生态丰富。
• pandas & numpy: 数据处理与计算的基石，高效且易用。
• akshare: 一个免费、开源的财经数据接口库，数据源可靠，是替代付费数据API的绝佳选择。
• matplotlib / seaborn: 用于可视化展示选股结果或指标分布。
• SQLite (可选): 用于本地存储历史数据，避免频繁网络请求。

核心模块划分：

1. 数据获取模块 (DataFetcher): 负责从网络获取股票列表、实时或历史行情数据、基本面数据（如市盈率PE、市净率PB）。
2. 数据处理模块 (DataProcessor): 清洗原始数据，计算所需的技术指标（如移动平均线MA、相对强弱指数RSI）。
3. 策略引擎模块 (StrategyEngine): 核心所在。定义并执行选股策略。例如，一个简单的双均线策略：收盘价上穿20日均线，且市盈率小于30。
4. 结果输出与可视化模块 (ResultExporter): 将筛选出的股票列表导出为CSV或Excel文件，并生成简单的图表进行分析。

3. 环境准备与依赖安装

首先，确保你的Python环境已经就绪。推荐使用conda或venv创建独立的虚拟环境。

# 创建并激活虚拟环境 (以conda为例) conda create -n stock_selector python=3.8 conda activate stock_selector # 安装核心依赖库 pip install pandas numpy akshare matplotlib seaborn

如果你的网络环境访问akshare的数据源较慢，可以考虑使用国内镜像源安装，但akshare本身会从各大财经网站获取数据，数据获取速度取决于源站。

4. 分步拆解：从获取数据到执行策略

4.1 第一步：获取全市场股票列表

我们需要知道分析的对象有哪些。akshare提供了A股基本列表。

# 文件：data_fetcher.py import akshare as ak def get_stock_list(): """ 获取沪深京A股股票代码和名称列表 :return: DataFrame，包含代码和名称 """ # 获取沪深京A股列表 stock_info_a_code_name_df = ak.stock_info_a_code_name() # 通常我们只关注主板、创业板、科创板，这里可以根据需要过滤，例如排除ST股票 # 简单示例，先不做复杂过滤 return stock_info_a_code_name_df if __name__ == '__main__': df = get_stock_list() print(f"共获取 {len(df)} 只股票") print(df.head())

运行这段代码，你会得到一个包含所有股票代码（如000001）和名称（如平安银行）的表格。这是我们的分析基础池。

4.2 第二步：获取单只股票的日线行情数据

有了股票列表，我们需要获取每只股票的历史价格和成交量数据，用于计算指标。

# 继续在 data_fetcher.py 中 def get_daily_data(stock_code, start_date='20230101', end_date='20231231'): """ 获取单只股票的日线行情数据 :param stock_code: 股票代码，带市场前缀，如 'sz000001' 或 '000001' :param start_date: 开始日期，格式 'YYYYMMDD' :param end_date: 结束日期，格式 'YYYYMMDD' :return: DataFrame，包含日期、开盘、收盘、最高、最低、成交量等 """ # akshare 需要标准格式代码，如 ‘000001’ # 注意：ak.stock_zh_a_hist 可能需要调整参数，请以akshare最新文档为准 try: # 方法1：使用 stock_zh_a_hist (常见) df = ak.stock_zh_a_hist(symbol=stock_code, period="daily", start_date=start_date, end_date=end_date, adjust="qfq") # 调整列名以便统一处理 if not df.empty: df.rename(columns={ '日期': 'date', '开盘': 'open', '收盘': 'close', '最高': 'high', '最低': 'low', '成交量': 'volume' }, inplace=True) df['date'] = pd.to_datetime(df['date']) df.set_index('date', inplace=True) return df except Exception as e: print(f"获取股票 {stock_code} 数据失败: {e}") return pd.DataFrame() # 返回空DataFrame if __name__ == '__main__': # 测试获取平安银行2023年数据 test_data = get_daily_data('000001', '20230101', '20230331') print(test_data.head()) print(test_data.tail())

关键点：这里使用了adjust="qfq"参数获取前复权数据，这非常重要，能保证价格序列在除权除息后依然连续，是进行技术分析的基础。

4.3 第三步：计算关键技术指标

数据获取后，我们需要计算策略依赖的指标。这里以最常用的简单移动平均线（SMA）为例。

# 文件：data_processor.py import pandas as pd import numpy as np def calculate_technical_indicators(df): """ 计算技术指标并添加到DataFrame中 :param df: 包含OHLCV数据的DataFrame :return: 添加了指标列的DataFrame """ if df.empty: return df df = df.copy() # 计算5日、20日、60日简单移动平均线 df['ma5'] = df['close'].rolling(window=5, min_periods=1).mean() df['ma20'] = df['close'].rolling(window=20, min_periods=1).mean() df['ma60'] = df['close'].rolling(window=60, min_periods=1).mean() # 计算相对强弱指数RSI (以14日为例) delta = df['close'].diff() gain = (delta.where(delta > 0, 0)).rolling(window=14).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean() rs = gain / loss df['rsi'] = 100 - (100 / (1 + rs)) # 计算布林带 (20日均线，2倍标准差) df['bb_middle'] = df['close'].rolling(window=20).mean() bb_std = df['close'].rolling(window=20).std() df['bb_upper'] = df['bb_middle'] + 2 * bb_std df['bb_lower'] = df['bb_middle'] - 2 * bb_std return df if __name__ == '__main__': # 假设从data_fetcher模块获取了数据 from data_fetcher import get_daily_data raw_data = get_daily_data('000001', '20230101', '20230630') processed_data = calculate_technical_indicators(raw_data) print(processed_data[['close', 'ma5', 'ma20', 'rsi']].tail(10))

现在，我们的数据已经包含了进行策略判断所需的多种指标。

4.4 第四步：实现策略引擎——定义选股规则

这是整个工具的大脑。我们用一个类来封装策略，方便扩展和维护。

# 文件：strategy_engine.py class SimpleStrategy: """ 一个简单的示例策略：双均线策略 + RSI过滤 规则： 1. 收盘价上穿20日均线 (金叉) 2. 当前价格高于60日均线 (处于中长期上升趋势) 3. RSI(14) 处于30到70之间 (避免超买超卖) 4. 当日成交量大于20日平均成交量 (有资金关注) """ def __init__(self): self.name = "简单双均线趋势策略" def check(self, df): """ 对单只股票的历史数据DataFrame执行策略检查 :param df: 包含计算好指标的DataFrame :return: (bool, dict) 是否通过筛选，以及详细信息 """ if df.empty or len(df) < 60: # 数据量不足 return False, {} latest = df.iloc[-1] # 最新一个交易日的数据 prev = df.iloc[-2] # 前一个交易日的数据 # 规则1: 最新收盘价上穿ma20 (金叉) condition1 = (prev['close'] < prev['ma20']) and (latest['close'] > latest['ma20']) # 规则2: 最新收盘价在ma60之上 condition2 = latest['close'] > latest['ma60'] # 规则3: RSI在30-70区间 condition3 = 30 < latest['rsi'] < 70 # 规则4: 成交量放大 condition4 = latest['volume'] > df['volume'].rolling(window=20).mean().iloc[-1] # 综合判断 if condition1 and condition2 and condition3 and condition4: detail = { 'close': latest['close'], 'ma20': latest['ma20'], 'ma60': latest['ma60'], 'rsi': latest['rsi'], 'signal_reason': f"金叉于{latest['ma20']:.2f}, RSI={latest['rsi']:.1f}" } return True, detail else: return False, {}

这个策略类定义了一套清晰的规则。check方法对一只股票的数据进行判断，返回是否通过以及关键信息。你可以轻松地修改__init__中的规则或创建新的策略类（如价值投资策略、动量策略）。

4.5 第五步：串联全流程，执行批量选股

现在，我们将所有模块组合起来，遍历股票列表，应用策略。

# 文件：main.py import pandas as pd from data_fetcher import get_stock_list, get_daily_data from data_processor import calculate_technical_indicators from strategy_engine import SimpleStrategy import time def run_screening(start_date='20240101', end_date='20240601'): """ 主运行函数：获取股票列表，逐只获取数据、计算指标、应用策略 """ print("开始获取股票列表...") stock_list_df = get_stock_list() # 为了演示，只取前50只，实际运行时可以注释掉这行跑全市场（但会很慢） stock_list_df = stock_list_df.head(50) strategy = SimpleStrategy() selected_stocks = [] total = len(stock_list_df) for idx, row in stock_list_df.iterrows(): code = row['code'] name = row['name'] print(f"正在处理 [{idx+1}/{total}]: {code} {name}") # 1. 获取数据 daily_df = get_daily_data(code, start_date, end_date) if daily_df.empty: continue # 2. 计算指标 processed_df = calculate_technical_indicators(daily_df) # 3. 策略判断 passed, detail = strategy.check(processed_df) if passed: selected_stocks.append({ '代码': code, '名称': name, '当前价格': detail.get('close'), 'MA20': detail.get('ma20'), 'RSI': detail.get('rsi'), '信号原因': detail.get('signal_reason') }) print(f" -> 通过筛选！") # 礼貌性延时，避免请求过快 time.sleep(0.1) # 4. 输出结果 if selected_stocks: result_df = pd.DataFrame(selected_stocks) output_file = f'selected_stocks_{end_date}.csv' result_df.to_csv(output_file, index=False, encoding='utf-8-sig') print(f"\n筛选完成！共选出 {len(selected_stocks)} 只股票。") print(f"结果已保存至: {output_file}") print(result_df) else: print("\n本次筛选未选出任何股票。") return selected_stocks if __name__ == '__main__': # 运行筛选，分析2024年上半年的数据 run_screening('20240101', '20240601')

5. 运行结果与初步分析

当你运行main.py后，控制台会输出处理进度，最终会在脚本同级目录下生成一个selected_stocks_20240601.csv文件。

一个可能的输出结果示例：

“惊了”的点在哪里？

1. 效率：手动浏览4000多只股票的K线图和指标，可能需要数周。而这个工具在优化后，可以在几十分钟内完成初步筛选。
2. 纪律性：程序严格无情地执行既定规则，避免了“我觉得这支股不错”的主观偏见。你会发现，很多你“感觉”会涨的股票，并不满足你设定的客观条件。
3. 回溯的启示：你可以轻松修改start_date和end_date，回测这个策略在过去任何一段时间的表现。例如，跑一下2022年的数据，你可能会发现这个简单的策略在单边下跌市中会频繁发出错误信号，从而让你理解策略的局限性。

6. 常见问题与排查思路

在复现过程中，你可能会遇到以下问题：

7. 进阶优化与工程实践建议

一个可用的原型只是起点。要让这个工具真正具备实用性和健壮性，需要考虑以下方面：

1. 数据层优化：

   • 本地缓存：将获取到的股票日线数据存入本地SQLite数据库。下次分析时，优先从本地读取，缺失部分再联网更新。这能极大提升回测速度。
   • 增量更新：每天只获取最新数据，而不是全量历史数据。
   • 错误重试与降级：网络请求必须包含重试机制和超时设置。当主数据源失败时，可以尝试备用源。

2. 策略层优化：

   • 策略回测框架：实现一个标准的回测引擎，可以计算策略的年化收益率、夏普比率、最大回撤等关键绩效指标。可以参考backtrader、zipline等开源框架的思路。
   • 多策略并行：可以同时运行多个策略类，比较它们的选股结果和绩效。
   • 参数优化：将策略中的参数（如均线周期、RSI阈值）提取出来，进行网格搜索，寻找历史最优参数组合（注意防范过拟合）。

3. 工程化与部署：

   • 配置文件：将股票池范围、策略参数、数据源地址等写入config.yaml或config.ini文件，使代码和配置分离。
   • 日志系统：使用Python内置的logging模块记录程序运行过程、错误信息，便于调试和监控。
   • 定时任务：结合系统的cron（Linux）或Task Scheduler（Windows），让工具在每天收盘后自动运行，将选股结果通过邮件或钉钉/微信机器人发送给你。

4. 风险与边界认知：

   • 历史不代表未来：这是量化分析的第一铁律。回测表现优异绝不等于未来能赚钱。
   • 过度拟合风险：在历史数据上反复优化参数，可能得到一个只完美匹配过去噪音的策略，在未来必然失效。
   • 市场有效性：A股市场并非完全有效，但简单的技术指标策略很容易被更复杂的算法或内幕信息击败。
   • 本工具仅为辅助：它生成的列表是研究的起点，而非终点。深度基本面分析、行业理解、宏观经济判断同样不可或缺。

8. 总结：从代码到认知的跨越

通过这三天的实践，我们完成了一个具备完整链路的Python选股工具。它的价值不仅仅在于那几行代码和最终生成的CSV文件，更在于这个过程本身带来的认知提升：

• 技术层面，你串联了数据获取、清洗、计算、策略逻辑和结果输出，这是一次标准的ETL（抽取、转换、加载）加业务规则应用的实战。
• 投资层面，你开始用程序员的思维看待市场：将模糊的感觉转化为清晰的规则，用历史数据验证想法的可行性，并深刻理解了策略的局限性和风险。

这个项目的全部源码，我已经整理在了文章中。你可以直接复制运行，但更建议你以此为蓝本进行修改：

• 尝试实现你自己的策略（例如，PB-ROE价值选股）。
• 加入基本面数据（akshare也提供财务数据）。
• 实现一个简单的可视化界面（用streamlit只需几十行代码）。
• 将其改造成一个定时运行的自动化脚本。

记住，工具是理性的延伸，但市场是复杂人性的集合。用代码构建你的分析框架，用智慧做出最终决策。希望这个项目能成为你探索量化投资与Python编程结合之路的一块有用的垫脚石。