1. 项目概述从手动到自动的选股进化作为一名在量化交易和金融数据抓取领域摸爬滚打了十多年的老码农我深知手动在同花顺问财里一遍遍输入条件、筛选股票的痛苦。效率低、容易出错、无法回溯历史更别提想做一些复杂的多条件组合或者高频测试了。所以自己动手写一个基于Python的“同花顺问财选股”自动化工具几乎是每个有想法的个人投资者或量化研究员的必经之路。这个项目本质上就是利用Python的网络请求和数据处理能力模拟我们在同花顺问财网页上的操作将选股逻辑代码化、自动化从而解放双手实现策略的快速验证和批量执行。它能做什么简单说就是你不再需要打开浏览器手动输入“市盈率小于20、净利润增长率大于30%、换手率大于3%”这样的语句。你可以把这些条件写成Python脚本一键运行程序会自动去问财网站获取符合条件的所有股票列表并以结构化的数据比如CSV、Excel返回给你。更进一步你可以设定定时任务让程序每天收盘后自动运行把选股结果发到你的邮箱或者钉钉实现无人值守的智能投顾雏形。适合谁来参考无论是刚入门Python、对金融数据感兴趣的程序员还是有一定投资经验、想用技术提升效率的散户甚至是小型私募的研究员都能从这个项目中获得直接的收益——将想法快速转化为可执行、可回溯的代码。2. 核心思路与技术选型解析2.1 逆向工程理解问财的数据接口同花顺问财网站本身并没有对外提供官方的API。我们要实现自动化核心思路就是“模拟浏览器行为”也就是常说的“爬虫”或“网络抓取”。但这并不是简单的页面抓取BeautifulSoup解析静态HTML因为问财的搜索结果是通过Ajax动态加载的。我们的首要任务是找到它背后真正的数据接口。通过浏览器的开发者工具F12切换到Network网络选项卡然后在问财页面输入一个选股条件并搜索观察网络请求。你会发现一个关键的XHR异步请求。这个请求的URL、参数Payload和返回的数据格式通常是JSON就是我们程序需要模拟的对象。这是整个项目的基石一切自动化都建立在对这个接口的正确调用之上。注意逆向工程获取的接口地址和参数结构并非一成不变同花顺可能会不定期更新其前端架构或接口格式。因此一个健壮的程序必须包含对接口变化的检测和一定的容错机制不能把代码写死。2.2 技术栈选择轻量、高效、易维护基于上述核心思路我选择了以下技术栈它们共同的特点是轻量、高效且拥有庞大的社区支持非常适合个人开发者和小型项目。Requests库这是Python中进行HTTP请求的“瑞士军刀”。我们需要用它来构造和发送模拟搜索的POST请求并接收服务器返回的JSON数据。它比Python内置的urllib更简洁、更强大。Pandas库金融数据分析的“标配”。问财返回的股票数据代码、名称、指标数值是结构化的Pandas的DataFrame是处理这类表格数据的完美容器。我们可以方便地进行数据清洗、筛选、排序和保存为文件。Schedule / APScheduler库可选如果你需要定时自动运行选股任务如每日收盘后那么需要一个任务调度器。Schedule库非常简单轻量适合基础定时APScheduler功能更强大支持持久化和更复杂的触发规则。Logging模块一个容易被忽视但至关重要的部分。自动化脚本在无人值守运行时完善的日志记录是排查问题的唯一依据。必须记录每次请求的参数、返回状态、数据条数以及任何异常信息。为什么不选用更“重型”的框架如Scrapy因为我们的目标单一且明确——调用一个特定的数据接口。Scrapy更适合大规模、复杂的网站爬取。用RequestsPandas的组合代码更简洁依赖更少部署和分享都更方便。3. 核心代码实现与分步详解3.1 接口分析与请求构造首先我们需要定位到问财搜索的真实接口。以一次实际的搜索为例我在浏览器中搜索“沪深A股市值大于100亿市盈率小于30”。在开发者工具的Network里过滤XHR请求我找到了一个形如https://www.iwencai.com/unifiedwap/unified-wap/v2/result/get-robot-data的请求。查看其请求头Headers和负载Payload是关键。请求头中需要重点关注User-Agent模拟浏览器身份和Content-Type通常为application/json。负载是一个JSON对象核心字段通常包括question我们的选股条件字符串如“沪深A股市值大于100亿市盈率小于30”。perpage每页返回的数据条数。page当前页码。以及其他一些用于标识来源、计算类型的固定参数。我们的Python代码需要完美复现这个请求。下面是一个高度还原的请求函数示例import requests import json import pandas as pd import time def fetch_iwencai_data(question, per_page50): 从同花顺问财获取选股数据 :param question: 选股条件字符串 :param per_page: 每页数据量 :return: 包含股票数据的pandas DataFrame url https://www.iwencai.com/unifiedwap/unified-wap/v2/result/get-robot-data # 关键构造请求头模拟浏览器 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36, Content-Type: application/json, Accept: application/json, text/plain, */*, Origin: https://www.iwencai.com, Referer: https://www.iwencai.com/unifiedwap/result?w沪深A股, } # 关键构造请求体Payload这是核心 payload { question: question, perpage: per_page, page: 1, secondary_intent: stock, log_info: {input_type:typewrite}, source: Ths_iwencai_Xuangu, version: 2.0, query_area: , block_list: , add_info: {urp:{scene:1,company:1,business:1},contentType:json}, rsh: Ths_iwencai_Xuangu_7u4tmb6g # 这个参数可能会变需要定期检查 } try: response requests.post(url, headersheaders, datajson.dumps(payload), timeout10) response.raise_for_status() # 检查HTTP请求是否成功 data_json response.json() # 解析返回的JSON数据提取股票列表 if data_json.get(status_code) 0: stock_list data_json[data][answer][0][txt][0][content][components][0][data] # 将数据转换为DataFrame df pd.DataFrame(stock_list) return df else: print(f请求失败返回信息{data_json.get(message)}) return pd.DataFrame() except requests.exceptions.RequestException as e: print(f网络请求异常{e}) return pd.DataFrame() except (KeyError, json.JSONDecodeError) as e: print(f解析响应数据异常{e}) return pd.DataFrame() # 示例调用 if __name__ __main__: # 构建你的选股问题 my_question 沪深A股市值大于100亿市盈率小于30换手率大于3% stock_df fetch_iwencai_data(my_question, per_page100) if not stock_df.empty: print(f共找到 {len(stock_df)} 只股票) # 选择需要的列例如股票代码、名称、市值、市盈率 # 注意返回的列名是中文的如最新价市值 selected_columns [股票代码, 股票简称, 最新价, 市值, 市盈率(动), 换手率] # 确保列存在 available_cols [col for col in selected_columns if col in stock_df.columns] print(stock_df[available_cols].head()) else: print(未获取到数据)代码要点解析User-Agent必须设置否则服务器可能拒绝请求或返回错误页面。Payload参数question是最重要的。rsh这类参数看起来像是一个加密或会话标识有时不变有时会变。如果某天发现代码不工作了首先检查这个参数是否已更新。比较新旧请求的差异是解决问题的关键。错误处理网络请求充满不确定性必须用try...except包裹并对HTTP状态码和返回的JSON结构进行判断确保程序健壮性。数据解析返回的JSON结构较深需要通过data_json[data][answer][0][txt][0][content][components][0][data]这样的路径才能到达真正的数据列表。这个路径也可能随网站改版而变化。3.2 数据清洗与格式化问财返回的数据并非直接可用通常需要清洗和格式化。列名处理返回的DataFrame列名是中文的如“股票代码”、“市盈率(动)”。为了后续处理方便我通常会将其转换为英文或拼音缩写例如code,pe_ttm。数据类型转换数字类字段如市值、市盈率在JSON中可能是字符串类型且可能包含“亿”、“万”、“%”等字符。需要写函数将其转换为纯数字浮点型。例如“123.45亿” - 123.45 * 1e8 12345000000.0“15.6%” - 0.156缺失值处理某些股票可能缺少部分指标数据显示为“-”或空。需要用pd.to_numeric(..., errorscoerce)将其转换为NaN然后根据策略决定是填充如用行业均值还是删除该行。去重理论上问财返回的数据不应重复但加上去重操作更保险。def clean_iwencai_data(df): 清洗问财返回的原始DataFrame if df.empty: return df df_clean df.copy() # 1. 重命名列示例 column_mapping { 股票代码: code, 股票简称: name, 最新价: price, 市值: market_cap, 市盈率(动): pe_ttm, 换手率: turnover_rate } # 只重命名存在的列 existing_cols {k: v for k, v in column_mapping.items() if k in df_clean.columns} df_clean.rename(columnsexisting_cols, inplaceTrue) # 2. 转换市值例如“500.00亿” - 500.0e8 if market_cap in df_clean.columns: def convert_market_cap(val): if isinstance(val, str): if 亿 in val: return float(val.replace(亿, ).replace(,, )) * 1e8 elif 万 in val: return float(val.replace(万, ).replace(,, )) * 1e4 try: return float(val) except: return None df_clean[market_cap] df_clean[market_cap].apply(convert_market_cap) # 3. 转换百分比例如“5.25%” - 0.0525 percent_cols [turnover_rate] # 以及其他百分比列 for col in percent_cols: if col in df_clean.columns: df_clean[col] df_clean[col].astype(str).str.replace(%, ).astype(float) / 100.0 # 4. 转换数值列将错误转为NaN numeric_cols [pe_ttm, price] for col in numeric_cols: if col in df_clean.columns: df_clean[col] pd.to_numeric(df_clean[col], errorscoerce) # 5. 去重基于股票代码 if code in df_clean.columns: df_clean.drop_duplicates(subset[code], inplaceTrue) df_clean.reset_index(dropTrue, inplaceTrue) return df_clean3.3 策略封装与多条件组合将单一的请求函数升级为一个策略执行器。我们可以把不同的选股条件基本面、技术面封装成独立的函数或类方法然后灵活组合。class StockSelector: def __init__(self): self.base_condition 沪深A股上市时间超过1年非ST def add_condition(self, new_condition): 添加选股条件用分号分隔 self.base_condition f{new_condition} def select_by_fundamental(self, pe_max30, profit_growth_min20, market_cap_min50): 添加基本面条件 self.add_condition(f市盈率小于{pe_max}) self.add_condition(f净利润同比增长率大于{profit_growth_min}%) self.add_condition(f市值大于{market_cap_min}亿) def select_by_technical(self, ma_short5, ma_long20, turnover_min3): 添加技术面条件示例短期均线上穿长期均线 # 注意问财的语法支持一些技术指标但可能不如专业软件丰富 # 这里是一个示例实际语法需要根据问财的规则调整 condition f收盘价大于{ma_short}日均线{ma_short}日均线上穿{ma_long}日均线 self.add_condition(condition) self.add_condition(f换手率大于{turnover_min}%) def run_selection(self, per_page100): 执行选股并返回清洗后的数据 print(f执行选股条件{self.base_condition}) raw_df fetch_iwencai_data(self.base_condition, per_page) cleaned_df clean_iwencai_data(raw_df) return cleaned_df # 使用示例 selector StockSelector() selector.select_by_fundamental(pe_max25, profit_growth_min25, market_cap_min100) selector.select_by_technical(ma_short10, ma_long30, turnover_min5) result_df selector.run_selection(per_page200) if not result_df.empty: # 可以进一步筛选例如按市盈率排序 result_df result_df.sort_values(bype_ttm).head(20) print(result_df[[code, name, pe_ttm, market_cap]]) # 保存结果 result_df.to_csv(selected_stocks.csv, indexFalse, encodingutf-8-sig)这种方式将选股逻辑模块化你可以轻松地创建“高成长低估值”、“趋势突破”、“资金流入”等不同的策略类方便管理和回测。3.4 定时任务与自动化部署为了实现每日自动选股我们需要引入定时任务。这里使用轻量级的schedule库。import schedule import time from datetime import datetime import logging # 配置日志 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(stock_selection.log), logging.StreamHandler() ]) def daily_selection_job(): 每日执行的选股任务 logging.info(开始执行每日自动选股任务...) try: selector StockSelector() selector.base_condition 沪深A股上市时间超过1年非ST # 添加你每日的策略条件 selector.select_by_fundamental(pe_max35, profit_growth_min15) selector.add_condition(今日涨幅大于0) # 示例条件 df selector.run_selection() if not df.empty: # 生成带有日期的文件名 filename fselected_stocks_{datetime.now().strftime(%Y%m%d)}.csv df.to_csv(filename, indexFalse, encodingutf-8-sig) logging.info(f选股完成共{len(df)}只股票已保存至 {filename}) # 这里可以添加发送邮件或钉钉消息的代码 # send_email_with_attachment(filename) else: logging.warning(今日未筛选到符合条件的股票。) except Exception as e: logging.error(f选股任务执行失败{e}, exc_infoTrue) # 设定每天下午15:30收盘后执行 schedule.every().day.at(15:30).do(daily_selection_job) logging.info(定时选股任务已启动等待执行...) # 保持程序运行 while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次将上述脚本部署在一台始终开机的电脑或服务器上如家中的NAS、树莓派或云服务器即可实现完全自动化。对于云服务器推荐使用systemd或supervisor来管理进程确保程序在后台稳定运行。4. 常见问题与实战避坑指南在实际开发和长期运行中你会遇到各种各样的问题。下面是我踩过坑后总结出的经验。4.1 接口失效或返回空数据这是最常见的问题表现为程序突然无法获取数据或返回空列表。原因1请求头或Payload参数过期。同花顺前端更新可能导致rsh、log_info等参数失效。排查与解决重新抓包用浏览器手动执行一次搜索在开发者工具中捕获最新的请求。将新请求的Headers和Payload与代码中的进行逐字段对比。重点检查User-Agent可以更新到新版本、Cookie有时需要携带会话Cookie、以及Payload中那些看似随机的字符串参数。模拟登录进阶如果接口开始要求登录态事情会变得复杂。你可能需要模拟登录流程处理验证码、维护会话或者寻找是否有无需登录的公开数据源作为备选。原因2请求频率过高被反爬。短时间内发送大量请求IP可能被暂时限制。排查与解决增加延迟在循环请求分页数据时务必在请求间加入随机延时例如time.sleep(random.uniform(1, 3))。使用代理IP池对于大规模、高频的数据抓取这是必备的。但对于个人每日几次的选股控制频率通常就够了。设置合理的超时和重试使用requests的timeout参数并实现简单的重试逻辑。def robust_fetch(url, headers, payload, max_retries3): for i in range(max_retries): try: resp requests.post(url, headersheaders, jsonpayload, timeout15) resp.raise_for_status() return resp except requests.exceptions.RequestException as e: logging.warning(f第{i1}次请求失败: {e}) if i max_retries - 1: wait_time (i 1) * 5 # 退避等待 logging.info(f等待{wait_time}秒后重试...) time.sleep(wait_time) else: logging.error(f请求失败已达最大重试次数) raise return None4.2 数据解析错误返回的JSON结构发生变化导致按原有路径提取数据失败。解决在解析数据时不要直接使用多层级的[data][answer][0]...这样的硬编码。应该逐层检查并使用.get()方法提供默认值避免KeyError。def safe_parse_response(data_json): try: # 使用.get()安全访问避免KeyError answer_list data_json.get(data, {}).get(answer, []) if not answer_list: return [] txt_list answer_list[0].get(txt, []) if not txt_list: return [] components txt_list[0].get(content, {}).get(components, []) if not components: return [] stock_data components[0].get(data, []) return stock_data except (IndexError, AttributeError, TypeError) as e: logging.error(f解析响应数据结构异常: {e}) logging.debug(f原始响应结构: {json.dumps(data_json, indent2, ensure_asciiFalse)[:500]}) # 打印部分结构用于调试 return []4.3 选股条件语法错误问财有自己的自然语言处理规则复杂的条件组合可能无法被正确解析导致返回结果不符合预期。实战心得从简到繁先在网页上手动测试你的选股语句确保能返回正确结果再将其放入代码。使用标准分隔符条件之间用中文分号“”分隔这是问财识别的标准格式。避免歧义词尽量使用问财界面提供的标准指标名称。例如用“市盈率(动)”而不是简单的“市盈率”用“归属母公司股东的净利润同比增长率”而不是“利润增长”。指标单位在条件中明确单位有时更可靠例如“市值大于500亿”比“市值大于500”更好。4.4 性能与数据完整性当一次查询可能返回成千上万只股票时例如“沪深A股”问财接口可能分页返回。解决方案实现自动翻页。在首次请求后检查返回数据中的总条数和当前页码然后循环请求直到获取所有数据。注意控制请求频率。def fetch_all_pages(question, max_pages10, delay1): all_data [] page 1 while page max_pages: payload[page] page logging.info(f正在获取第 {page} 页数据...) df fetch_iwencai_data_with_payload(payload) # 需要一个能接受完整payload的函数 if df is None or df.empty: break all_data.append(df) if len(df) payload[perpage]: break # 如果返回的数据少于每页数量说明是最后一页 page 1 time.sleep(delay) # 页间延迟避免触发反爬 if all_data: return pd.concat(all_data, ignore_indexTrue) else: return pd.DataFrame()4.5 策略的有效性与回测最重要的一点通过这个工具选出的股票绝不等于未来一定会涨。它只是一个高效的“初筛工具”。回测是必须的你需要将选股逻辑和历史数据结合进行回测验证策略在历史阶段的有效性胜率、盈亏比、最大回撤等。可以使用akshare、tushare等库获取历史行情和财务数据用backtrader、zipline等框架进行回测。避免过度拟合不要在历史数据上反复调整参数直到曲线完美这样得到的策略在未来大概率失效。坚持策略的逻辑性比追求高历史收益更重要。实盘前的模拟在投入真金白银前至少进行3-6个月的模拟盘跟踪观察策略在实际市场环境下的表现。5. 进阶扩展与安全合规考量5.1 功能扩展方向一个基础的选股器搭建完成后你可以沿着以下几个方向深化多策略并行与评分同时运行多个选股策略如价值策略、成长策略、动量策略对每只股票进行打分综合排名选出最终股票池。集成其他数据源问财的数据主要用于初筛。你可以将筛选出的股票代码再用其他接口如akshare获取更细的历史K线、资金流向tushare pro获取深度财务数据进行二次分析和验证。构建可视化监控面板使用Flask或Streamlit快速搭建一个Web界面展示每日选股结果、策略历史表现图表更加直观。对接交易接口极度谨慎这是最终的一步将选股信号自动转化为交易指令。这涉及到券商API如华泰、国金等提供的量化接口或第三方平台。此步骤风险极高必须经过充分模拟测试且初期建议只做预警人工确认后再下单。5.2 法律与合规风险提醒在享受技术便利的同时必须时刻绷紧合规这根弦。数据使用同花顺问财的数据仅供个人学习、研究使用。严禁用于任何商业用途、严禁大规模抓取对其服务器造成压力、严禁将抓取的数据公开传播或售卖。访问频率务必保持低频、友好的访问模式模拟正常用户行为。设置合理的延时避免在短时间内发起海量请求。尊重版权在代码注释或相关文档中应声明数据来源。投资风险本项目产生的所有选股结果仅供参考不构成任何投资建议。金融市场风险巨大任何自动化策略都有失效的可能请理性投资自负盈亏。我个人最深刻的体会是技术工具解决的是“如何高效地找”的问题但它永远无法回答“该不该买”和“该买多少”这两个更核心的问题。后者需要你对市场、对商业模式、对人性的深刻理解。这个Python选股源码是一个强大的杠杆它能放大你的研究能力但运用杠杆的方向和力度永远取决于你自身的认知。把它当作一个不知疲倦的研究助理而不是一个点石成金的“圣杯”你的投资之路才能走得更稳、更远。