基于Python与yfinance构建本地化股票量化筛选器：以PKScreener为例

1. 项目概述与核心价值最近在和一些做量化交易的朋友交流时发现大家普遍面临一个痛点虽然市面上有各种股票数据接口和量化平台但真正能快速、灵活地根据自定义条件进行股票筛选并且能本地化部署、深度定制的工具却不多。要么是封装得太死要么是学习成本太高要么就是数据源不稳定。直到我发现了 GitHub 上的一个开源项目pkjmesra/PKScreener它完美地解决了这个问题。简单来说PKScreener 是一个基于 Python 的、功能强大的股票筛选器它允许你通过编写简单的 Python 脚本利用丰富的数据源主要是 Yahoo Finance 和 NSE India来筛选符合特定技术或基本面条件的股票。这个项目的核心价值在于它的“可编程性”和“本地化”。它不是一个给你固定几个指标比如“市盈率低于20”、“MACD金叉”让你去点的黑盒软件而是一个工具箱。你可以像搭积木一样组合各种数据获取函数、技术指标计算函数和条件判断逻辑构建出完全属于你自己的、独一无二的筛选策略。比如你可以轻松实现“寻找过去5个交易日成交量持续放大且RSI从超卖区低于30刚刚上穿50同时股价站上20日均线的股票”这样的复杂组合条件。所有的计算和筛选都在你的本地机器上完成数据隐私有保障运行速度也取决于你的网络和算力避免了云服务的延迟和限制。对于量化爱好者、独立交易员甚至是金融专业的学生来说PKScreener 都是一个极佳的学习和实践平台。它让你从“使用工具”的层面深入到“创造工具”的层面真正理解量化筛选背后的数据流和逻辑链。接下来我将深入拆解这个项目的架构、核心用法并分享一套从零开始构建自定义筛选器的实战流程以及我踩过的一些坑和总结的经验。2. 项目架构与核心模块解析要玩转 PKScreener首先得理解它的“五脏六腑”。整个项目虽然代码量不小但模块化做得很好核心逻辑清晰。我们可以把它拆解为数据层、策略层和应用层来理解。2.1 数据获取层与市场的连接器数据是量化分析的血液。PKScreener 主要依赖yfinance库来获取广泛的全球市场数据尤其是美股同时对于印度国家证券交易所NSE的数据它有一套自己的补充机制。在PKScreener的源代码中你会找到一个核心的数据处理类通常命名为StockData或类似它封装了以下关键功能历史行情数据获取通过yfinance.download()函数可以获取指定股票代码、指定时间段如“1y”代表一年的日级或更细粒度的行情数据包括开盘价、最高价、最低价、收盘价、成交量以及调整后的收盘价。这是计算所有技术指标的基础。基本面数据抓取除了行情yfinance.Ticker对象提供了丰富的公司基本面信息如市盈率PE、市净率PB、股息率、资产负债表摘要等。PKScreener 的相关模块会解析这些信息供策略使用。股票列表管理一个筛选器需要知道筛选的范围。项目通常会维护一个股票代码列表文件如symbols.csv或nifty500.csv里面包含了待扫描的所有股票代码。对于NSE项目可能内置了从官方网页或API获取最新列表的功能。数据缓存与更新为了避免频繁请求网络数据可能触发速率限制好的实现会加入缓存机制。例如将下载的数据以DataFrame的格式保存为本地的.pkl或.csv文件并设置一个过期时间如1小时。下次请求时先检查缓存是否有效有效则直接读取无效则重新下载。这能极大提升批量扫描数百只股票时的效率。注意yfinance是一个非官方库其数据源和稳定性依赖于 Yahoo Finance 的公开接口。虽然大部分情况下很可靠但在极端市场行情或雅虎服务器调整时偶尔会出现数据缺失或延迟。在编写策略时必须加入健壮的错误处理try-except对无法获取数据的股票进行跳过或记录日志避免整个扫描进程因单只股票而崩溃。2.2 策略逻辑层大脑与规则引擎这是 PKScreener 最核心、最灵活的部分。策略层定义了“什么样的股票是我想要的”。在项目中策略通常以 Python 函数或类的形式存在。一个典型的策略函数结构如下def my_custom_screener(stock_data): 自定义筛选策略 :param stock_data: 包含股票历史数据的 pandas DataFrame :return: (bool, dict) 是否符合条件以及用于展示的详细信息字典 # 1. 计算所需指标 df stock_data.copy() df[MA20] df[Close].rolling(window20).mean() df[RSI] compute_rsi(df[Close], period14) # 假设有compute_rsi函数 df[Volume_MA5] df[Volume].rolling(window5).mean() # 获取最近一天的数据 latest df.iloc[-1] prev df.iloc[-2] # 2. 定义筛选条件 condition1 latest[Close] latest[MA20] # 股价在20日均线上方 condition2 latest[RSI] 50 and prev[RSI] 50 # RSI上穿50 condition3 latest[Volume] 1.5 * latest[Volume_MA5] # 成交量放量50% # 3. 组合条件 if condition1 and condition2 and condition3: details { Current Price: latest[Close], MA20: latest[MA20], RSI: latest[RSI], Volume_Ratio: latest[Volume] / latest[Volume_MA5] } return True, details else: return False, {}在这个例子中策略清晰地表达了三个技术条件的“与”关系。你可以无限扩展这个逻辑加入更多指标布林带、MACD、ATR等、基本面条件PE范围、营收增长率甚至是自定义的形态识别如“连续三根阳线”。策略层的设计精髓在于“模块化”。优秀的做法是将通用的指标计算如计算RSI、移动平均线封装成独立的工具函数然后在策略函数中调用。这样构建新策略就像拼装乐高积木只需要关注条件组合的逻辑而不必重复编写底层计算代码。PKScreener 的源码中通常就包含了这样一个丰富的“指标工具库”。2.3 应用执行与输出层调度器与报告生成有了数据获取和策略逻辑还需要一个“调度器”来把它们串联起来并处理输出。这一层通常由一个主循环脚本负责遍历股票列表读取symbols.csv逐个获取每只股票的数据。应用策略将获取到的股票数据传递给策略函数进行判断。收集结果将策略函数返回为True的股票及其详细信息details字典收集到一个列表或DataFrame中。输出报告将最终结果以易于阅读的格式输出。最常见的是生成一个DataFrame并用pandas的to_csv或to_excel方法保存为文件。更高级的输出可以包括控制台表格打印使用tabulate库美化输出在终端直接查看。HTML报告生成一个带有简单样式和排序功能的网页可视化效果更好。邮件通知将筛选结果通过电子邮件发送给自己实现自动化监控。集成到交易平台将筛选出的股票列表自动推送到你的模拟或实盘交易系统进行进一步分析或下单这需要额外的开发。这一层还需要考虑性能优化。同步地、一只接一只地获取几百只股票的数据会非常慢。常见的优化手段是使用并发例如 Python 的concurrent.futures.ThreadPoolExecutor来并发下载数据注意yfinance可能对并发请求数有限制或者使用异步IO库asyncio。PKScreener 的某些高级版本或分支可能已经实现了这些优化。3. 从零构建自定义筛选器实战七步法理解了架构我们就可以动手打造自己的筛选器了。下面我以一个实战案例为例目标是“筛选出在印度NSE上市当前股价高于200日均线且低于布林带上轨同时过去一周平均成交量较前一周放大超过20%的股票”。3.1 第一步环境搭建与依赖安装首先需要一个干净的 Python 环境建议使用 3.8 及以上版本。使用虚拟环境是一个好习惯。# 创建并激活虚拟环境 (以 conda 为例) conda create -n pkscreener python3.9 conda activate pkscreener # 安装核心依赖 pip install yfinance pandas numpy ta-lib这里重点说一下ta-lib。它是一个广泛使用的技术分析库计算指标速度快且准确。但它的安装有时比较麻烦特别是Windows系统。如果直接pip install ta-lib失败可以到 TA-Lib 的非官方预编译包网站 下载对应你Python版本和系统位数的.whl文件然后用pip install 文件名.whl安装。如果实在装不上也可以用pandas或numpy自己实现指标计算PKScreener项目源码里通常也有纯Python的实现可供参考。3.2 第二步获取并管理股票代码列表对于NSE市场我们需要一个最新的股票代码列表。我们可以从PKScreener项目源码中寻找相关函数或者自己写一个简单的爬虫/下载器。这里假设我们从一个可靠的静态文件开始。import pandas as pd # 方法1: 从本地文件读取 (假设你有一个nifty500.csv文件) # df_symbols pd.read_csv(nifty500.csv) # 方法2: 模拟一个简单的列表 (实际使用时请替换为真实、全面的列表) symbols [RELIANCE.NS, TCS.NS, INFY.NS, HDFCBANK.NS, ICICIBANK.NS] # .NS 是雅虎财经对NSE股票的后缀 # 在实际项目中这个列表可能包含数百个代码可以从NSE官网或其它数据提供商处获取。实操心得维护一个准确、全面的股票代码列表是持续运行筛选器的前提。对于NSE股票会增发、退市、更名。建议每周或每月更新一次列表。一个实用的技巧是将代码列表及其获取日期一起保存在运行筛选器前先检查列表是否“新鲜”如果太旧则触发更新流程。3.3 第三步封装数据获取函数我们将数据获取逻辑封装起来加入缓存和错误处理。import yfinance as yf import pandas as pd import os from datetime import datetime, timedelta import time CACHE_DIR ./data_cache os.makedirs(CACHE_DIR, exist_okTrue) def fetch_stock_data(symbol, period6mo, interval1d, use_cacheTrue): 获取股票数据支持缓存 cache_file os.path.join(CACHE_DIR, f{symbol.replace(., _)}.pkl) # 检查缓存 if use_cache and os.path.exists(cache_file): cache_age datetime.now() - datetime.fromtimestamp(os.path.getmtime(cache_file)) if cache_age timedelta(hours1): # 缓存1小时内有效 print(f[Cache Hit] Loading {symbol} from cache.) return pd.read_pickle(cache_file) # 从网络获取 print(f[Fetching] Downloading data for {symbol}...) try: ticker yf.Ticker(symbol) # 获取历史行情 df ticker.history(periodperiod, intervalinterval) if df.empty: print(f[Warning] No data for {symbol}) return None # 获取一些基本面信息示例 info ticker.info df[sector] info.get(sector, N/A) df[marketCap] info.get(marketCap, None) # 保存缓存 df.to_pickle(cache_file) # 礼貌性暂停避免请求过快 time.sleep(0.1) return df except Exception as e: print(f[Error] Failed to fetch {symbol}: {e}) return None这个函数实现了带时效性的缓存并尝试获取一些基本面信息附加到DataFrame中。period6mo获取最近6个月的数据对于计算200日均线约40周可能有点紧可以调整为1y或2y以获取更长的历史数据。3.4 第四步实现策略逻辑函数现在实现我们的核心筛选逻辑。我们将使用ta-lib来计算布林带。import talib import numpy as np def custom_screener_logic(df): 策略股价 200日均线股价 布林带上轨周成交量放大 20% if df is None or len(df) 210: # 确保有足够数据计算200日均线 return False, {} close_prices df[Close].values volumes df[Volume].values # 计算指标 # 1. 200日简单移动平均线 sma200 talib.SMA(close_prices, timeperiod200) # 2. 布林带 (默认20日均线2倍标准差) upper_band, middle_band, lower_band talib.BBANDS(close_prices, timeperiod20, nbdevup2, nbdevdn2) # 获取最新值 current_close close_prices[-1] current_sma200 sma200[-1] current_upper_band upper_band[-1] # 3. 计算成交量变化过去5个交易日一周均值 vs 再前5个交易日均值 if len(volumes) 10: recent_vol_avg np.mean(volumes[-5:]) prev_vol_avg np.mean(volumes[-10:-5]) volume_ratio recent_vol_avg / prev_vol_avg if prev_vol_avg 0 else 1 else: volume_ratio 1 # 定义条件 cond_price_above_sma current_close current_sma200 cond_price_below_bb current_close current_upper_band cond_volume_surge volume_ratio 1.2 # 放大20% if cond_price_above_sma and cond_price_below_bb and cond_volume_surge: details { Symbol: df.index[-1], # 实际代码需要从外部传入这里用日期示意 Close: round(current_close, 2), SMA200: round(current_sma200, 2) if not np.isnan(current_sma200) else None, BB_Upper: round(current_upper_band, 2) if not np.isnan(current_upper_band) else None, Volume_Ratio: round(volume_ratio, 2) } return True, details else: return False, {}这个函数严格遵循了策略层的设计输入数据计算指标判断条件返回结果和详情。注意我们使用了talib.SMA和talib.BBANDS它们直接对numpy数组进行计算效率很高。3.5 第五步构建主循环与调度器现在把数据获取和策略应用串联起来。def run_screener(symbol_list): 主筛选函数 results [] for symbol in symbol_list: # 1. 获取数据 stock_data fetch_stock_data(symbol, period1y, use_cacheTrue) # 2. 应用策略 passed, details custom_screener_logic(stock_data) # 3. 收集结果 if passed: details[Symbol] symbol # 将股票代码加入详情 results.append(details) print(f✅ Found: {symbol}) return results这是一个最简单的同步版本。对于大量股票这会很慢。我们可以引入线程池进行优化from concurrent.futures import ThreadPoolExecutor, as_completed def run_screener_concurrent(symbol_list, max_workers5): 使用线程池并发运行的筛选函数 results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交任务 future_to_symbol {executor.submit(process_single_stock, sym): sym for sym in symbol_list} for future in as_completed(future_to_symbol): symbol future_to_symbol[future] try: passed, details future.result() if passed: details[Symbol] symbol results.append(details) print(f✅ Found: {symbol}) except Exception as e: print(f❌ Error processing {symbol}: {e}) return results def process_single_stock(symbol): 包装单只股票的处理流程供线程池调用 stock_data fetch_stock_data(symbol, period1y, use_cacheTrue) return custom_screener_logic(stock_data)max_workers不宜设置过大通常5-10个线程为宜避免对数据源服务器造成过大压力或被封IP。3.6 第六步美化输出与报告生成得到结果列表后我们需要把它变成可读的报告。def generate_report(results, output_formatconsole): 生成筛选报告 if not results: print(No stocks passed the screening criteria.) return df_results pd.DataFrame(results) # 按股价或成交量比率排序 df_results df_results.sort_values(byVolume_Ratio, ascendingFalse) if output_format console: from tabulate import tabulate print(\n *60) print(SCREENING RESULTS) print(*60) print(tabulate(df_results, headerskeys, tablefmtpsql, showindexFalse)) elif output_format csv: filename fscreening_results_{datetime.now().strftime(%Y%m%d_%H%M%S)}.csv df_results.to_csv(filename, indexFalse) print(f\nResults saved to {filename}) elif output_format html: filename fscreening_results_{datetime.now().strftime(%Y%m%d_%H%M%S)}.html # 简单的HTML样式 html_content df_results.to_html(classestable table-striped, indexFalse) with open(filename, w) as f: f.write(fhtmlheadtitleScreening Results/title/headbody{html_content}/body/html) print(f\nHTML report saved to {filename})3.7 第七步整合与执行最后创建一个主入口文件将以上所有步骤整合。# main.py import sys def main(): # 1. 加载股票列表 (这里用示例列表实际应从文件读取) symbols [RELIANCE.NS, TCS.NS, INFY.NS, HDFCBANK.NS, ICICIBANK.NS, KOTAKBANK.NS, HINDUNILVR.NS] # 在实际项目中你应该从一个更全面的文件中加载例如 # symbols pd.read_csv(nse_symbols.csv)[Symbol].tolist() print(fStarting screening for {len(symbols)} symbols...) # 2. 运行筛选器 (选择同步或并发版本) # results run_screener(symbols) # 同步版 results run_screener_concurrent(symbols, max_workers5) # 并发版 # 3. 生成报告 generate_report(results, output_formatconsole) # 输出到控制台 generate_report(results, output_formatcsv) # 同时保存为CSV print(\nScreening completed.) if __name__ __main__: main()运行python main.py你就完成了一次完整的自定义股票筛选。控制台会打印出符合条件的股票同时会在当前目录下生成一个带有时间戳的CSV结果文件。4. 高级技巧、常见问题与避坑指南在实战中你会遇到各种各样的问题。下面是我在长期使用和定制 PKScreener 类工具过程中积累的一些经验和解决方案。4.1 数据质量与一致性处理问题1数据缺失或异常值雅虎财经的数据偶尔会有异常比如在非交易日有零值或NaN或者股价出现极端错误如“闪崩”导致的异常低点。直接使用这些数据计算指标如移动平均线会导致结果失真。解决方案清洗数据在计算前检查并处理NaN值。对于日线数据通常可以用前向填充ffill或直接删除包含NaN的行。df_cleaned df.dropna(subset[Close, Volume]) # 删除关键列为NaN的行 # 或者用前值填充 df_filled df.ffill()异常值过滤对于股价可以设定一个合理的范围例如当日涨跌幅超过20%则视为异常并采用相邻日期的平均值进行替换或直接排除该日数据。数据验证增加一个数据质量检查步骤如果数据长度远小于预期或缺失率过高则记录警告并跳过该股票。问题2复权价格问题yfinance默认提供的Adj Close调整后收盘价已经考虑了分红、拆股等因素是进行长期技术分析的首选。但有些策略可能需要使用实际收盘价Close。务必清楚你使用的价格序列是哪种并在策略说明中注明。对于短线策略使用Close可能更贴近实际交易。4.2 策略过拟合与回测的重要性陷阱你精心设计了一个包含七八个条件的复杂策略在历史数据上回测表现惊人。但一用到实盘或新的数据上效果就大打折扣。这很可能是过拟合——你的策略过于完美地“记忆”了历史噪声而非捕捉到了有效的市场规律。避坑方法简化策略奥卡姆剃刀原则。从一两个核心逻辑开始逐步增加条件并观察每个条件对结果数量和质量的边际贡献。如果一个条件加入后通过股票数锐减且未来表现没有显著提升则考虑剔除。样本外测试将历史数据分为两部分“训练集”用于开发策略“测试集”用于验证策略效果。确保测试集的数据在时间上晚于训练集且从未在开发过程中使用过。交叉验证对于时间序列数据可以使用“前向滚动”验证。例如用2000-2010年的数据优化参数在2011年测试然后用2001-2011年优化在2012年测试以此类推。理解策略的经济学逻辑问自己这个策略为什么应该有效是因为它捕捉了趋势动能、均值回归、还是市场情绪一个有逻辑支撑的策略比纯粹数据挖掘出来的策略更可能具有持续性。4.3 性能优化实战当股票列表达到上千只历史数据要求更长如5年日线同步循环可能会跑上小时。优化至关重要。并发获取数据如前所述使用ThreadPoolExecutor。但要注意yfinance的Ticker对象不是线程安全的。最佳实践是在每个线程内独立创建yf.Ticker(symbol)并获取数据而不是共享一个对象。向量化计算在策略函数中尽量使用pandas和numpy的向量化操作避免在数据行上进行Python级别的for循环。ta-lib本身已是向量化计算。缓存策略升级除了按股票代码缓存可以考虑按“股票代码数据周期”作为缓存键。使用更快的序列化格式如parquet或内存缓存如joblib.Memory。增量更新如果每天运行没必要每次都下载全部历史数据。可以只下载最新的几天数据然后与本地缓存的历史数据合并。yfinance的download函数支持start和end参数。4.4 错误处理与日志记录一个健壮的筛选器必须能处理各种异常并留下清晰的日志方便排查问题。import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[logging.FileHandler(screener.log), logging.StreamHandler()]) def robust_fetch(symbol): try: data fetch_stock_data(symbol) if data is None or data.empty: logging.warning(fSymbol {symbol}: returned empty data.) return None return data except yf.YFinanceError as e: logging.error(fYFinance error for {symbol}: {e}) return None except Exception as e: logging.exception(fUnexpected error for {symbol}: {e}) # 记录完整异常堆栈 return None在主循环中用robust_fetch代替直接的fetch_stock_data确保单只股票的失败不会导致整个程序崩溃。4.5 策略扩展与模块化设计当你积累的策略越来越多时一个好的代码结构能让你事半功倍。策略工厂模式创建一个策略注册表。每个策略都是一个独立的函数或类在一个中心化的地方注册。主程序根据配置加载指定的策略。STRATEGY_REGISTRY {} def register_strategy(name): def decorator(func): STRATEGY_REGISTRY[name] func return func return decorator register_strategy(breakout_above_ma) def breakout_strategy(df): # ... 策略实现 pass register_strategy(rsi_oversold) def rsi_strategy(df): # ... 策略实现 pass # 主程序中根据名称调用 strategy_name breakout_above_ma if strategy_name in STRATEGY_REGISTRY: strategy_func STRATEGY_REGISTRY[strategy_name] passed, details strategy_func(stock_data)参数化策略将策略中的阈值如RSI的30/70成交量放大倍数1.2提取为参数这样无需修改代码就能快速调整策略灵敏度。结果分析模块不仅输出通过列表还可以增加简单的统计分析如本次筛选的通过率、各行业分布、平均市值等帮助你从宏观上理解当前市场下该策略的选股偏好。5. 将筛选器融入自动化工作流一个本地运行的脚本只是开始。要让筛选器真正产生持续价值可以考虑将其自动化。定时任务使用系统的定时任务工具。在Linux上可以用cron在Windows上可以用“任务计划程序”。例如设置每天下午4点市场收盘后自动运行筛选脚本。# Linux crontab 示例每天16:05运行 5 16 * * 1-5 cd /path/to/your/project /usr/bin/python3 /path/to/your/project/main.py /path/to/logs/screener.log 21结果通知脚本运行结束后自动发送结果。可以用smtplib发邮件或者用requests调用钉钉、企业微信、Telegram等平台的Webhook发送消息。将最重要的几只股票信息推送到手机实现即时提醒。与交易平台对接进阶这是量化交易的核心环节。筛选出的股票列表可以通过券商提供的API如Interactive Brokers, Alpaca等自动下单或者导入到专业的量化回测/交易平台如Backtrader, Zipline, 或国内的JoinQuant, RiceQuant进行更复杂的回测和模拟交易。请注意实盘交易涉及巨大风险务必在充分理解策略、完成严格回测和模拟盘测试后再用极小资金尝试。PKScreener 项目本身可能已经包含了上述某些功能的雏形或完整实现。我的建议是先基于它的核心架构理解原理然后根据自己的需求像搭积木一样将这些自动化、通知、分析模块一个个添加进去最终形成一套完全为你量身定制的、自动化运行的量化筛选系统。这个过程本身就是学习和提升量化技能的最佳路径。