开篇为什么我写了个爬虫来选股票先说结论这个项目不是让你躺着赚钱的而是把“每天翻几百只股票看财报和走势”这件事变成一条命令跑完的自动化流水线。名字叫“爬虫技术选股Python 自动化筛选潜力股”核心就三件事抓公开数据、清洗入库、按规则打分。我自己用Python写了这套东西跑了几个月最大的感受是选股的体力活少了但判断力的事还得自己来。这个项目适合谁如果你有Python基础懂一点pandas又不想每天手动刷新行情页面、复制粘贴财务指标那这套思路值得你花一个周末搭起来。它解决的核心痛点不是“预测涨跌”而是“把全市场几千只股票快速缩小到一个可以人工深度研究的候选池”。说白了爬虫解决的是信息获取效率筛选模型解决的是信息过滤标准。另外必须先说清楚这不是荐股工具所有输出都是基于公开数据的规则信号不构成投资建议。我搭这套东西的初衷是让自己从大量重复性浏览里解脱出来把时间留给真正需要动脑的基本面分析。下面我把整个项目的设计思路、技术选型、实操代码、踩坑记录一次讲完尽量让有基础的人能照着搭也让还没入门的人看懂关键逻辑。1. 选股项目的整体思路从需求到框架1.1 一个朴素问题选股难在哪里打开行情软件几百个板块、几千只股票光看一遍就得花半天。更麻烦的是看盘时直觉很容易被短期波动带跑今天涨得猛的票明天可能就回落。靠人工盯盘做初筛效率低标准还不稳定。所以我想要的不是“预测”而是“把数理化的条件固化成代码”。比如我想找“市值适中、ROE连续三年不错、近期成交量温和放大”的股票这个条件写成人话容易但人要每天手工跑一遍所有股票不现实。于是需求就清晰了让程序定期抓取全市场公开交易数据和财务指标然后我只需要在配置文件里改条件程序跑完自动输出一份按分数排序的表格。1.2 架构设计抓什么、存哪里、怎么筛整个流水线我分成了四层每层职责单一出了问题也好定位采集层负责从公开数据源抓取行情列表、每日交易数据、财务报表摘要。输出原始JSON或CSV。存储层用SQLite做本地库轻量、免安装、单文件适合个人项目。每个交易日的数据追加存储同时保留历史记录。处理层用pandas做数据清洗、字段标准化、计算衍生指标。筛选层按既定规则打分排序输出候选股票池并生成一个简单的HTML或Excel报告。这样的好处是分层解耦哪天想换数据源只需要改采集层想换选股策略只需要动筛选层其他模块不用大改。一个人维护这种小项目最怕改一处牵全身所以边界一定要清晰。1.3 合规与伦理边界先说这块再谈技术做爬虫选股最容易忽略的不是反爬而是合规。我的原则很简单只采集公开数据、不绕开登录墙、不暴力请求、尊重网站的robots协议和服务条款。股票行情、公开财报这些信息本身面向公众披露抓取下来做个人研究参考风险可控但绝不意味着可以无限频次、全速抓取。实操中我给自己定了三条规矩第一单次请求间隔不低于1秒第二每天每个数据源的请求总量限制在几百次以内第三只在交易日开盘后固定时段抓取避开盘中高峰。这套约束既保护对方服务器也降低自己被封的风险。数据合规的底线说穿了就一句话拿公开数据可以别做破坏性操作别拿去商用牟利。2. 数据源与工具链选型不盲目抓数据先选对水源2.1 数据源对比免费公开接口、财经页面、还是现成库项目初始我评估过三类数据来源。第一类是现成的Python开源库比如akshare、tushare封装好了大量接口调用方便省去自己解析页面的麻烦。第二类是各大财经网站公开的行情接口返回JSON格式字段齐全但需要自己维护接口地址和参数。第三类是直接请求网页再做HTML解析灵活但最脆弱页面结构一改就要修代码。我的选择是混合方案日常行情和财务摘要优先用开源库接口因为这些接口背后已经做了不少兼容工作如果某个数据缺失或接口异常再降级到备份的财经公开接口。不建议把所有鸡蛋放一个篮子里数据源的冗余本身就是一种可靠性设计。这里有一个必须理解的点免费接口往往有频率限制和字段更新延迟。比如有些接口单次最多返回100条记录全市场几千只股票就得循环拉取有些财务指标接口更新时间是晚上白天抓到的可能是前一天的数据。这些限制在项目设计阶段就要考虑进去否则后面跑起来才发现坑的是自己。2.2 抓取工具链requests、httpx、pandas怎么配合核心抓取库我用的是requests简单可靠。遇到需要异步并发或连接池复用的场景可以换成httpx但个人项目其实没太大必要上并发串行加延时反而更稳。解析JSON直接用内置的json模块如果被迫要解析HTML优先用lxml配合BeautifulSoup能有效处理不规范标签。数据处理是pandas的主场。抓回来的数据往往是“宽表”几百个字段堆在一起我需要按需抽取列、重命名、去重、填充缺失值。pandas的DataFrame处理这些场景很顺手一条链式操作就能完成多步清洗。再配合numpy做数值计算筛选条件的实现非常直观。所以这个项目的技术栈看起来平平无奇但胜在生态成熟、资料多。对于个人自动化工具稳定性和可维护性永远比炫技重要。别一上来就整分布式爬虫和消息队列那是给自己找麻烦。2.3 请求频率与错误处理礼貌爬虫才能活得久我踩过的最大一个坑是为了赶在开盘前把数据抓完把请求间隔设成了0.2秒结果跑了不到半小时就被限流接下来一整天数据都是断的。后来老老实实改成1秒间隔虽然总耗时变长但连续跑了几个月都没出问题。错误处理上我做了三层保障第一层单次请求失败自动重试3次退避时间按1、3、5秒递增第二层连续失败超过5次就跳过当前批次并记录日志不中断整个流程第三层每次抓取结束做一次数据量校验如果某张表记录数明显少于昨日触发告警提醒人工检查。这套机制让我从“半夜收到程序崩溃消息”的状态里解放出来大部分小故障程序自己就消化了。3. 核心功能实现从抓取到候选池3.1 抓取逻辑与目标定义到底要哪些字段开始写代码前我先把目标字段列成了清单这个动作特别值得做。清单包括两类字段一类是行情类比如代码、名称、最新价、涨跌幅、成交量、换手率、总市值、流通市值另一类是财务类比如市盈率TTM、市净率、ROE、营收同比增速、净利润同比增速、资产负债率、毛利率。这份清单直接决定了后面所有工作。没钱买商用数据源也没有专门团队维护数据质量所以字段尽量少而精只筛真正会用到的。事实证明前期多花半小时规划字段后面能少写一堆清洗代码。字段清单也是配置项的核心后续调整筛选条件时只需看这个清单里有没有对应字段即可。3.2 数据清洗与标准化抓回来不是重点洗能用才是爬虫抓回来的数据第一眼永远是脏的。常见问题包括字段名不统一一会儿是“pe_ttm”一会儿是“dynamic_pe”空值满天飞新股经常缺财务指标数值类型混乱“12.5万手”这种带单位的字符串重复记录同一只股票一天被抓了两遍。我的清洗流程固定五步列名标准化、类型转换、去重、缺值处理、异常值剔除。列名标准化是先维护一张映射表把不同数据源的字段名映射到统一命名。类型转换时把带“亿”“万”的字符串转成浮点数这一步用正则表达式配合pandas的replace方法非常高效。去重按“代码日期”双键做drop_duplicates。缺值处理不轻易填零因为财务指标缺失和数值为0含义完全不同通常我标注为NaN并在筛选时排除。3.3 定时任务与增量更新让流程自己跑起来桌面程序靠手动点击不够“自动”。环境不同方案也不同我本地用的计划任务服务器上则用crontab。核心诉求就一条每个交易日收盘后固定时间自动运行脚本抓当日数据、更新SQLite、重新计算评分并生成报告。增量更新逻辑是关键。全量重抓太浪费我的做法是每次抓取先查询本地库中每个代码的最新日期只抓取那个日期之后的数据。这样即使某天程序挂了第二天补跑也能自动补齐缺口。这个“断点续抓”机制用一句简单的SQL就能实现。另外每个月做一次全量校准防止个别代码因停牌、新股等原因长期缺失数据。4. 筛选模型搭建潜力股的量化画像4.1 指标选择从财务到技术拒绝单维度“潜力股”这个词听起来玄但落到代码里就是一组可量化的画像。我的筛选条件分成三组每组权重不同。第一组是估值类指标包括市盈率TTM、市净率核心逻辑是找相对便宜的标的避免追高第二组是质量类指标包括ROE、毛利率、资产负债率衡量赚钱能力和财务稳健度第三组是动量类指标包括近期涨幅、成交量变化捕捉资金关注度。这三组条件互有制衡。单独看低市盈率容易落入“价值陷阱”买一堆不涨的银行股单独看高动量又容易追在山顶。所以我用加权评分把几类维度压缩成一个综合分数排序输出。这样虽然不能保证选出大牛股但能稳定过滤掉明显不合规、不健康的标的节省大量人工筛选时间。4.2 评分权重的设计逻辑为什么是这几个数具体权重我设置如下估值30分质量40分动量30分。这个比例不是拍脑袋而是根据我自己的风险偏好调整出来的。质量维度占比最高是因为我认为长期来看赚钱质量好的公司会持续跑赢而估值便宜但质量差的票可能永远是便宜的。每个指标内部再细分成打分档位。比如ROE大于15%给满分介于10%到15%之间给80%的分数低于10%给0分。这里有个容易犯的错误用原始指标值直接线性映射分数。实际上不IPO或新上市的公司财务指标波动极大直接用线性会放大异常值。采用分段打分能天然抑制极端值的影响更贴近人工判断的逻辑。4.3 回测与验证别急着实盘先看看历史表现模型搭完一定要做的一件事是历史回测。我用SQLite里存的历史数据做了一次简单验证跑过去6个月的筛选结果看候选股票池后续3个月的平均涨幅是否优于全市场平均。实测下来我的模型在震荡市里超额收益不明显但在趋势市里能跑赢基准几个百分点。这个结果告诉我工具的有效性是有条件的13万不要迷信任何策略尤其是单一策略。回测时另一个重要心得是防止“幸存者偏差”。分析时不要把已经退市的股票剔除掉否则你会高估模型的盈利能力。我最初犯过这个错误剔除后的结果比剔除前好看太多明显失真。后来把历史快照保留下来每次回测都基于当时的候选池才得到相对可信的结果。5. 实操过程与代码骨架一条命令跑通全流程5.1 环境准备与依赖清单我用的是Python 3.10版本Windows和Linux都跑过没发现大的兼容性问题。核心依赖就四个requests、pandas、numpy、openpyxl用于生成Excel报告。换句话说用到的都是资历老、坑少、资料多的库新手照着装也不容易出错。安装方式建议直接用pip我习惯先为项目建一个虚拟环境避免依赖冲突。如果你在Windows上遇到过装pandas失败的问题多半是pip版本太老或者没装Microsoft C Build Tools这个问题在官方文档里有明确说明按文档装好再装pandas就没有障碍了。跑起来之前确保时间同步也是关键点因为签名校验和接口限频依赖准确客户端时间。5.2 采集模块核心代码解析先看最核心的抓取部分。这里以免费行情接口为例说明我的抓取循环import time import json import requests import pandas as pd HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } def fetch_stock_list(): url https://example-public-market-api.com/stock/list resp requests.get(url, headersHEADERS, timeout10) resp.raise_for_status() data resp.json() return pd.DataFrame(data[data]) def fetch_daily_quote(code): url fhttps://example-public-market-api.com/stock/daily params {code: code, date: latest} resp requests.get(url, paramsparams, headersHEADERS, timeout10) if resp.status_code ! 200: return None return resp.json() def batch_fetch_quotes(codes, interval1.0): rows [] for idx, code in enumerate(codes): try: item fetch_daily_quote(code) if item: rows.append(item[data]) except Exception as exc: print(f[WARN] {code} fetch failed: {exc}) time.sleep(interval) if (idx 1) % 100 0: print(f[INFO] fetched {idx 1}/{len(codes)}) return pd.DataFrame(rows)这段代码的关键点有两个。其一是循环里加入了time.sleep(interval)这是“礼貌爬虫”的最低要求其二是异常捕获粒度细化到单只股票而不是整个批次fail这样个别数据源波动不会拖垮整体任务。实际项目中我还在失败重试和连续失败熔断逻辑上做了加强但核心骨架就是上面这样。5.3 清洗与入库把DataFrame写进SQLite数据抓回来下一步是清洗入库。我的清洗函数长这样import sqlite3 from datetime import datetime def clean_quotes(df): if df.empty: return df # 标准列名 rename_map { symbol: code, name: name, price: close, change_pct: pct_chg, volume: volume, turnover: turnover_rate, mkt_cap: total_mv, } df df.rename(columnsrename_map) # 只保留需要的列 cols [code, name, close, pct_chg, volume, turnover_rate, total_mv] df df[[c for c in cols if c in df.columns]] # 类型转换 for col in [close, pct_chg, volume, turnover_rate, total_mv]: if col in df.columns: df[col] pd.to_numeric(df[col], errorscoerce) # 去重 df df.drop_duplicates(subset[code]) # 交易日期 df[trade_date] datetime.now().strftime(%Y-%m-%d) return df def save_to_sqlite(df, db_pathstock.db): conn sqlite3.connect(db_path) tbl daily_quote if_table_exists conn.execute( SELECT name FROM sqlite_master WHERE typetable AND name?, (tbl,) ).fetchone() if not if_table_exists: conn.execute(f CREATE TABLE {tbl} ( code TEXT, name TEXT, close REAL, pct_chg REAL, volume REAL, turnover_rate REAL, total_mv REAL, trade_date TEXT, PRIMARY KEY (code, trade_date) ) ) df.to_sql(tbl, conn, if_existsappend, indexFalse) conn.commit() conn.close()这里有几个值得注意的取舍。第一字段只挑要用的不要一股脑入库节省空间也减少后续维护成本。第二主键设为code trade_date自然去重重复执行同一天的抓取不会产生脏数据。第三pd.to_numeric(errorscoerce)把无法转成数字的值变成NaN后面筛选时统一处理避免字符串“12.5万”把排序逻辑搞坏。5.4 筛选评分模块把规则翻译成代码清洗入库之后的筛选评分模块是这个项目的灵魂。我的实现思路import pandas as pd import numpy as np def build_scores(df): df df.copy() # 估值分市盈率越低分越高 def pe_score(pe): if pd.isna(pe): return 0 if pe 15: return 100 elif pe 25: return 80 elif pe 40: return 50 else: return 20 # 质量分ROE越高越好 负债率适中 def roe_score(roe): if pd.isna(roe): return 0 if roe 15: return 100 elif roe 10: return 70 else: return 30 def debt_score(debt): if pd.isna(debt): return 0 if debt 40: return 100 elif debt 60: return 70 else: return 30 # 动量分近20日涨幅 成交量放大 def momentum_score(return_20d, volume_ratio): if pd.isna(return_20d): return 0 if return_20d 0 and volume_ratio 1.2: return 100 elif return_20d 0: return 70 elif return_20d -10: return 20 else: return 50 # 计算加权总分 df[score] ( 0.3 * df[pe].apply(pe_score) 0.25 * df[roe].apply(roe_score) 0.15 * df[debt_ratio].apply(debt_score) 0.2 * df[return_20d].apply( lambda x: momentum_score(x, 1.1) ) 0.1 * df[volume_ratio].apply( lambda x: 80 if x 1.5 else (50 if x 1 else 20) ) ) return df def rank_candidates(df, top_n50): df df[df[score] 60] df df.sort_values(score, ascendingFalse) return df.head(top_n)权重可以调整但调整时必须想清楚影响。比如加大估值权重候选池会偏向低位蓝筹加大动量权重候选池会偏向短线活跃股。这个项目的定位是“潜力股中长线观察池”所以质量分权重最高动量只是辅助。5.5 一键运行把爬取、清洗、计算、报告串起来最后的调度脚本通过一个main()把所有环节串起来def main(): df_list fetch_stock_list() quote_df batch_fetch_quotes(df_list[code].tolist()) clean_df clean_quotes(quote_df) save_to_sqlite(clean_df) # 读取历史库并按新数据打分 merged load_for_scoring() result rank_candidates(build_scores(merged)) result.to_excel(candidate_stocks.xlsx, indexFalse) print(f[DONE] saved {len(result)} candidates) if __name__ __main__: main()运行方式就是python main.py输出的Excel表格里我按分数从高到低列了所有候选股每条附带关键指标可以直接用Excel再中做透视、人工核验。整个流程从抓取到报告产出我实测下来耗时10多分钟基本不占用精力。6. 常见问题与避坑实录把这些坑提前埋上别让它坑你6.1 数据源字段对不上怎么办这是换数据源或数据源更新字段时最常见的问题。某天接口突然把pe_ttm改成了pe清洗代码直接报KeyError。我的解决办法是写一个字段映射函数兼容新旧字段名同时用try/except做缺列保护。更稳的做法是在清洗模块开头校验必选字段如果缺失马上记录日志并跳过当批数据避免脏数据入库。6.2 请求被限流怎么办限流表现有两种直接返回403或者返回一份错误提示JSON。如果你已经设置了合理的请求间隔但仍然被限大概率是总请求量超了阈值。这时候优先做两件事一是确认是否可以用批量接口替代逐个请求将请求量降低一个数量级二是增加随机延时让请求模式看起来更自然。实测下来固定1秒间隔比随机1到3秒更容易被识别原因可能是固定间隔具有机器特征。6.3 财务数据的复权与缺失问题行情类指标好抓财务类指标容易踩坑。比如“市盈率TTM”在数据源里可能是负值或极大值微利企业这时候直接套评分规则会把异常值当作高质量信号。我的处理方法是对财务指标做上下限截断。还有复权问题如果直接用不复权价格计算20日涨幅遇到除权当天价格会“断崖式下跌”导致动量分错误。我后来改用前复权价格来计算涨跌幅这个问题才彻底解决。6.4 定时任务跑挂后怎么自动恢复我遇到得最多的场景是某天网络抖动脚本在抓取半途中断SQLite里留了一批半成品数据。如果任务没有幂等设计第二次运行就会产生大量重复或错位数据。我的幂等方案是前面提过的主键去重外加每天写一个状态文件记录最新成功日期。任务启动时先读状态文件从最近成功日期开始续跑如果某段数据无法补齐状态文件不更新下次任务会再次尝试抓取。这套机制看起来简单但让我少熬了无数个夜。写在最后的一点个人体会这个项目做到现在我最大的收获不是“选出过多少涨得好的票”而是真正理解了“信息和规则必须分离”。人工选股时情绪会混入判断而程序筛选规则是死的结果是一致的我能随时复盘为什么某只股票被选中或被排除。这种透明度对想提升自己投资方法论的人来说非常有用。再分享一个小技巧所有配置我都写在一个config.yaml里包括请求间隔、重试次数、评分权重、候选数量。改策略时我不碰代码只改配置。这样既降低了自己误改逻辑的风险也让整个流程更像一个可以反复调用的“产品”而不只是一团脚本。最后提醒一句这套东西的价值上限取决于你对规则本身的思考深度。爬虫帮你把数据搬回家Python帮你把筛选变成流水线但“什么才是好的选股标准”这个问题永远需要你自己回答。