如果你是一名程序员同时又对投资理财感兴趣那么你很可能有过这样的念头能不能用我的技术能力写个程序来帮我分析股票甚至自动选股这个想法听起来很酷但往往止步于“想法”因为数据获取、指标计算、策略回测……每一步都像一座大山。最近我就花了三天时间用 Python 把这个想法变成了现实。结果跑出来的数据确实让我有点惊讶——不是因为它选出了什么“十倍牛股”而是通过这个亲手搭建的工具我清晰地看到了技术分析、数据驱动与主观臆断之间的巨大鸿沟。更重要的是我验证了一个核心判断对于开发者而言构建一个基础的量化分析工具技术门槛远没有想象中高真正的挑战在于策略逻辑的严谨性和对市场规律的敬畏。这篇文章我将分享这个 Python 选股工具从零到一的完整构建过程并附上全套源码。无论你是想学习 Python 数据分析、对量化投资感兴趣还是单纯想了解如何将编程技能应用于实际场景这篇文章都将为你提供一条清晰的路径。我们将避开复杂的金融工程理论聚焦于可落地的代码实现让你能亲手复现并理解其背后的每一个环节。1. 这个工具到底解决了什么问题在开始看代码之前我们必须明确这个工具的目标和边界。它不是一个“圣杯”不能保证盈利。它的核心价值在于自动化数据收集替代手动从财经网站复制粘贴数据实现高效、准确的数据获取。策略规则化执行将模糊的选股思路如“市盈率低、近期上涨”转化为明确的、可重复执行的代码逻辑消除情绪干扰。快速回溯验证可以对历史数据进行回测初步检验一个想法在过去的市场中是否有效尽管过去不代表未来但这是理性决策的第一步。学习与实践平台它是学习pandas、numpy、数据可视化以及面向对象编程的绝佳实战项目。简单说这个工具是一个“决策辅助系统”。它帮你从4000多只股票中根据你设定的条件快速筛选出一个小范围的股票池为你更深度的基本面研究或技术分析节省大量前期时间。它的输出是“候选列表”而不是“买入指令”。2. 核心设计思路与技术栈整个工具的设计遵循“数据输入 - 规则处理 - 结果输出”的流水线模式。技术栈选择Python 3.8: 量化分析领域的事实标准生态丰富。pandas numpy: 数据处理与计算的基石高效且易用。akshare: 一个免费、开源的财经数据接口库数据源可靠是替代付费数据API的绝佳选择。matplotlib / seaborn: 用于可视化展示选股结果或指标分布。SQLite (可选): 用于本地存储历史数据避免频繁网络请求。核心模块划分数据获取模块 (DataFetcher): 负责从网络获取股票列表、实时或历史行情数据、基本面数据如市盈率PE、市净率PB。数据处理模块 (DataProcessor): 清洗原始数据计算所需的技术指标如移动平均线MA、相对强弱指数RSI。策略引擎模块 (StrategyEngine): 核心所在。定义并执行选股策略。例如一个简单的双均线策略收盘价上穿20日均线且市盈率小于30。结果输出与可视化模块 (ResultExporter): 将筛选出的股票列表导出为CSV或Excel文件并生成简单的图表进行分析。3. 环境准备与依赖安装首先确保你的Python环境已经就绪。推荐使用conda或venv创建独立的虚拟环境。# 创建并激活虚拟环境 (以conda为例) conda create -n stock_selector python3.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_date20230101, end_date20231231): 获取单只股票的日线行情数据 :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(symbolstock_code, perioddaily, start_datestart_date, end_dateend_date, adjustqfq) # 调整列名以便统一处理 if not df.empty: df.rename(columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume }, inplaceTrue) df[date] pd.to_datetime(df[date]) df.set_index(date, inplaceTrue) 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())关键点这里使用了adjustqfq参数获取前复权数据这非常重要能保证价格序列在除权除息后依然连续是进行技术分析的基础。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(window5, min_periods1).mean() df[ma20] df[close].rolling(window20, min_periods1).mean() df[ma60] df[close].rolling(window60, min_periods1).mean() # 计算相对强弱指数RSI (以14日为例) delta df[close].diff() gain (delta.where(delta 0, 0)).rolling(window14).mean() loss (-delta.where(delta 0, 0)).rolling(window14).mean() rs gain / loss df[rsi] 100 - (100 / (1 rs)) # 计算布林带 (20日均线2倍标准差) df[bb_middle] df[close].rolling(window20).mean() bb_std df[close].rolling(window20).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(window20).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_date20240101, end_date20240601): 主运行函数获取股票列表逐只获取数据、计算指标、应用策略 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正在处理 [{idx1}/{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 fselected_stocks_{end_date}.csv result_df.to_csv(output_file, indexFalse, encodingutf-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文件。一个可能的输出结果示例代码名称当前价格MA20RSI信号原因000001平安银行12.3412.1055.3金叉于12.10, RSI55.3000858五粮液165.80162.5048.7金叉于162.50, RSI48.7..................“惊了”的点在哪里效率手动浏览4000多只股票的K线图和指标可能需要数周。而这个工具在优化后可以在几十分钟内完成初步筛选。纪律性程序严格无情地执行既定规则避免了“我觉得这支股不错”的主观偏见。你会发现很多你“感觉”会涨的股票并不满足你设定的客观条件。回溯的启示你可以轻松修改start_date和end_date回测这个策略在过去任何一段时间的表现。例如跑一下2022年的数据你可能会发现这个简单的策略在单边下跌市中会频繁发出错误信号从而让你理解策略的局限性。6. 常见问题与排查思路在复现过程中你可能会遇到以下问题问题现象可能原因排查方式解决方案akshare报错KeyError或返回空数据1. 股票代码格式不对。2.akshare接口更新函数名或参数已变化。3. 网络问题或数据源暂时不可用。1. 打印出正在请求的代码。2. 查阅akshare官方文档或GitHub仓库的最新示例。3. 尝试直接访问相关财经网站。1. 确保代码为6位数字如000001。2. 根据akshare文档调整函数调用。3. 添加异常捕获和重试机制或更换数据源。程序运行速度极慢1. 循环请求网络未做任何优化。2. 股票列表数量太多。1. 使用time模块计算单次请求耗时。2. 观察CPU和内存使用情况。1. 增加time.sleep()避免被封但可适当减少等待时间如0.05秒。2.引入并发请求如concurrent.futures.ThreadPoolExecutor这是性能提升的关键。3. 先在小范围股票池测试。计算出的指标如RSI为NaN滚动窗口如14期初期数据不足。检查df.tail(20)看前14行的RSI是否为NaN。这是正常现象。在策略判断时应确保用于计算信号的数据是有效的例如len(df) 60。pandas的rolling计算会自动处理。选出的股票数量为01. 策略条件过于严格。2. 所选时间段市场整体不符合策略逻辑。3. 数据获取失败导致所有股票被跳过。1. 逐一打印每只股票的规则判断结果。2. 检查策略逻辑代码特别是条件判断的边界还是。3. 确认是否有股票数据成功获取。1. 放宽策略条件或测试更长时间段。2. 编写单元测试用模拟数据验证策略逻辑是否正确。3. 确保数据获取函数稳定并记录失败日志。导入模块错误ModuleNotFoundError1. 依赖未安装。2. 文件不在同一目录或Python路径问题。1. 在命令行执行pip list检查。2. 检查文件命名和导入语句。1. 使用虚拟环境并正确安装依赖。2. 使用相对导入或确保所有.py文件在同一项目根目录下。7. 进阶优化与工程实践建议一个可用的原型只是起点。要让这个工具真正具备实用性和健壮性需要考虑以下方面数据层优化本地缓存将获取到的股票日线数据存入本地SQLite数据库。下次分析时优先从本地读取缺失部分再联网更新。这能极大提升回测速度。增量更新每天只获取最新数据而不是全量历史数据。错误重试与降级网络请求必须包含重试机制和超时设置。当主数据源失败时可以尝试备用源。策略层优化策略回测框架实现一个标准的回测引擎可以计算策略的年化收益率、夏普比率、最大回撤等关键绩效指标。可以参考backtrader、zipline等开源框架的思路。多策略并行可以同时运行多个策略类比较它们的选股结果和绩效。参数优化将策略中的参数如均线周期、RSI阈值提取出来进行网格搜索寻找历史最优参数组合注意防范过拟合。工程化与部署配置文件将股票池范围、策略参数、数据源地址等写入config.yaml或config.ini文件使代码和配置分离。日志系统使用Python内置的logging模块记录程序运行过程、错误信息便于调试和监控。定时任务结合系统的cronLinux或Task SchedulerWindows让工具在每天收盘后自动运行将选股结果通过邮件或钉钉/微信机器人发送给你。风险与边界认知历史不代表未来这是量化分析的第一铁律。回测表现优异绝不等于未来能赚钱。过度拟合风险在历史数据上反复优化参数可能得到一个只完美匹配过去噪音的策略在未来必然失效。市场有效性A股市场并非完全有效但简单的技术指标策略很容易被更复杂的算法或内幕信息击败。本工具仅为辅助它生成的列表是研究的起点而非终点。深度基本面分析、行业理解、宏观经济判断同样不可或缺。8. 总结从代码到认知的跨越通过这三天的实践我们完成了一个具备完整链路的Python选股工具。它的价值不仅仅在于那几行代码和最终生成的CSV文件更在于这个过程本身带来的认知提升技术层面你串联了数据获取、清洗、计算、策略逻辑和结果输出这是一次标准的ETL抽取、转换、加载加业务规则应用的实战。投资层面你开始用程序员的思维看待市场将模糊的感觉转化为清晰的规则用历史数据验证想法的可行性并深刻理解了策略的局限性和风险。这个项目的全部源码我已经整理在了文章中。你可以直接复制运行但更建议你以此为蓝本进行修改尝试实现你自己的策略例如PB-ROE价值选股。加入基本面数据akshare也提供财务数据。实现一个简单的可视化界面用streamlit只需几十行代码。将其改造成一个定时运行的自动化脚本。记住工具是理性的延伸但市场是复杂人性的集合。用代码构建你的分析框架用智慧做出最终决策。希望这个项目能成为你探索量化投资与Python编程结合之路的一块有用的垫脚石。