Python自动化选股：模拟同花顺问财API实现量化策略数据获取

1. 项目缘起：为什么我要自己动手写一个问财选股工具？

做量化交易或者策略研究的朋友，应该都绕不开一个场景：如何高效地获取符合特定条件的股票列表？无论是回测还是实盘监控，第一步都是“选股”。市面上有很多工具，比如同花顺的问财，它提供了一个非常强大的自然语言选股界面，你可以输入“连续三天放量上涨”、“MACD金叉且股价站上20日均线”这样的描述，它就能给你筛选出结果。这功能确实方便，尤其是对于不熟悉编程的投资者。

但问题也随之而来。当你需要定期、批量地执行这些选股条件，或者想把选股结果无缝接入到自己的Python分析、回测框架里时，手动在网页上操作就显得力不从心了。每次都要打开浏览器、登录、输入条件、点击查询、等待结果、再手动复制数据……这个过程不仅繁琐，而且无法自动化，更别提在策略迭代中快速测试成百上千个条件组合了。

所以，一个很自然的需求就产生了：能不能用Python代码直接调用同花顺问财的选股逻辑，把结果拿回来自己处理？这就是“同花顺问财选股Python源码”这个标题背后，绝大多数开发者或量化爱好者的核心诉求。我们想要的不是一个简单的网页爬虫，而是一个能够稳定、可靠、模拟真实用户行为，从问财获取结构化选股数据的程序化接口。

我最初也是被这个需求驱动，经历了从简单爬虫到模拟登录，再到处理各种反爬和数据结构问题的完整过程。网上能找到的零星代码要么过于陈旧已经失效，要么功能不全、稳定性差。今天，我就把自己踩过坑、验证过的相对完整的实现思路和核心代码分享出来，目标是构建一个健壮的、可复用的问财选股Python模块。请注意，以下内容基于技术研究和学习目的，实际使用中务必尊重数据源的服务条款，控制请求频率，避免对目标服务器造成不必要的压力。

2. 核心思路拆解：问财选股的数据链路与模拟策略

在动手写代码之前，我们必须先搞清楚同花顺问财页面是怎么工作的。只有理解了数据流转的链条，才能找到合适的介入点。通过浏览器开发者工具（按F12）对问财页面进行网络抓包分析，是必不可少的第一步。

2.1 网络请求分析：从自然语言到股票列表

当你访问同花顺问财页面并输入一个选股条件时，例如“市值大于100亿”，页面会发起一系列网络请求。我们需要找到最关键的那个——负责将你的查询条件发送给服务器并返回股票列表的请求。

1. 打开问财页面：访问同花顺问财的官方网站。
2. 输入查询条件：在搜索框输入“市值大于100亿”。
3. 开启开发者工具：按F12，切换到Network（网络）选项卡。记得勾选Preserve log（保留日志），以防页面跳转时请求记录被清除。
4. 执行查询：点击搜索按钮或按回车。
5. 筛选请求：在纷繁复杂的请求中，我们需要寻找特征。通常，返回股票列表数据的请求是一个XHR或Fetch请求，其Response（响应）是JSON格式或包含表格数据的文本。你可以通过Type列为xhr或fetch进行筛选，或者直接查看Preview（预览）选项卡，看哪个请求的返回数据是股票代码和名称的列表。

经过分析，你会发现核心请求的URL通常包含一个特定的API端点，例如https://search.10jqka.com.cn/unifiedwap/unified-wap/result/get-stock-pick。请求方法一般是POST或GET，并且会携带一系列参数，其中最重要的就是question参数，它的值就是你输入的选股条件“市值大于100亿”。

此外，请求头（Headers）中通常包含Cookie、User-Agent、Referer等字段，这些都是服务器用来验证请求来源、识别用户会话的关键信息。特别是Cookie，它可能包含了你的登录态或会话标识，对于需要登录后才能使用的复杂条件（比如涉及自选股、财务数据深度筛选）至关重要。

2.2 数据模拟策略：三种实现路径的权衡

理解了数据链路后，我们可以设计几种不同的Python实现策略，各有优劣：

1. 直接请求API（推荐，但需处理反爬）：

   • 原理：直接模拟浏览器向上述分析得到的核心API地址发送HTTP请求，携带必要的参数和请求头。
   • 优点：效率最高，速度最快，代码最简洁。直接获取结构化的数据（通常是JSON），解析方便。
   • 挑战：需要精准构造请求参数和请求头，特别是动态生成的token、sign等加密参数。同花顺的反爬机制可能会升级，需要持续维护。对于需要登录的查询，需要先模拟登录获取有效的Cookie。
   • 适用场景：技术能力强，愿意持续维护代码，追求最高效率和稳定性的用户。

2. 通过Selenium等浏览器自动化工具：

   • 原理：使用Selenium库控制一个真实的浏览器（如Chrome）打开问财页面，模拟用户输入、点击、等待页面加载，然后从页面HTML中解析出股票数据。
   • 优点：几乎可以应对任何反爬措施，因为行为与真人操作完全一致。不需要深入分析复杂的API参数。
   • 缺点：速度慢，资源消耗大（需要运行浏览器实例）。代码稳定性受网页UI变化影响。解析HTML结构相对复杂。
   • 适用场景：API难以逆向、查询条件非常简单、对运行速度不敏感的一次性任务。

3. 寻找替代数据源或第三方封装库：

   • 原理：使用其他财经数据API（如Tushare、AkShare、Baostock等）来实现类似的选股逻辑。或者使用他人已经封装好的、维护中的同花顺问财Python库。
   • 优点：可能更稳定、更规范，有社区支持。
   • 缺点：功能可能无法与问财完全同步，可能有使用限制或费用。第三方库也可能失效。
   • 适用场景：对数据源要求灵活，愿意接受功能差异，希望快速上手。

考虑到性能、可控性和学习价值，本文将重点探讨第一种策略——直接请求API的实现。这是最具挑战性但也最核心的方法，掌握了它，你就能应对大多数类似的网络数据获取场景。

3. 实战构建：Python模拟请求的核心代码实现

接下来，我们进入实战环节。我将分步骤展示如何构建一个基础的同花顺问财选股函数。请注意，由于同花顺的接口和反爬策略可能随时变动，以下代码主要展示思路和核心结构，你需要根据实际情况调整参数和解析逻辑。

3.1 环境准备与依赖库安装

首先，确保你的Python环境（建议3.7以上）已经安装了必要的库。我们将主要使用requests来处理HTTP请求，用pandas来方便地处理返回的表格数据。

pip install requests pandas

如果需要对返回的复杂JSON或加密参数进行处理，可能还会用到json,hashlib,time,random等Python标准库。

3.2 核心函数构造：get_stock_pick

我们来构建一个名为get_stock_pick的核心函数。它的目标是输入一个问财查询语句，返回一个包含股票代码、名称等信息的pandas.DataFrame。

import requests import pandas as pd import time import json def get_stock_pick(question, page=1, perpage=50, retry=3): """ 根据同花顺问财条件筛选股票 Args: question (str): 问财选股条件，例如：“市值大于100亿” page (int): 页码，默认第1页 perpage (int): 每页返回数量，默认50 retry (int): 请求失败重试次数，默认3次 Returns: pandas.DataFrame: 包含股票代码、名称等信息的DataFrame，如果失败返回空DataFrame """ # 1. 基础URL (此URL为示例，实际需要从网络分析中获取) base_url = "https://search.10jqka.com.cn/unifiedwap/unified-wap/result/get-stock-pick" # 2. 构造请求头 (关键！需要模拟浏览器) 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', 'Referer': 'https://www.iwencai.com/unifiedwap/result?w=市值大于100亿', # Referer需要动态构造，包含查询词 'Accept': 'application/json, text/javascript, */*; q=0.01', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8', 'Origin': 'https://www.iwencai.com', 'Connection': 'keep-alive', # 'Cookie': '你的Cookie字符串', # 对于复杂查询可能需要 } # 3. 构造请求参数 (关键！这些参数名和值需要从网络请求中分析) # 注意：参数可能包含动态生成的 `v`、`sign` 等，这里展示静态版本 params = { 'question': question, 'page': page, 'perpage': perpage, 'r': int(time.time() * 1000), # 一个时间戳参数，常用于防缓存 # 可能还有其他必要参数，如 `query_type`, `secondary_intent`, 等 } # 4. 发送请求 for i in range(retry): try: # 这里使用GET还是POST，参数放在params还是data里，需要根据实际抓包决定 response = requests.get(base_url, params=params, headers=headers, timeout=10) response.raise_for_status() # 检查HTTP状态码是否为200 # 5. 解析响应 data_json = response.json() # 重要：响应结构需要具体分析。这里假设数据在 `data` -> `list` 字段下 if data_json.get('status_code') == 0 and 'data' in data_json: stock_list = data_json['data'].get('list', []) if stock_list: # 将列表转换为DataFrame # 字段名也需要根据实际响应调整，如 `code`, `name`, `market` 等 df = pd.DataFrame(stock_list) # 可能需要对股票代码进行格式化，例如加上市场前缀 # df['code'] = df['market'].astype(str) + df['code'].astype(str) return df else: print(f"查询条件‘{question}’未找到符合条件的股票。") return pd.DataFrame() else: print(f"API返回异常状态: {data_json.get('status_code')}, 消息: {data_json.get('message')}") # 可以在这里加入对特定错误码的处理，比如需要登录 if data_json.get('status_code') == 10001: # 假设10001代表需要登录 print("此查询可能需要登录，请尝试添加有效的Cookie。") break # 非网络错误，重试可能无效 except requests.exceptions.RequestException as e: print(f"第{i+1}次请求失败: {e}") if i < retry - 1: time.sleep(2) # 重试前等待 else: print("重试次数用尽，请求失败。") except json.JSONDecodeError as e: print(f"响应内容JSON解析失败: {e}") print(f"原始响应文本: {response.text[:500]}") # 打印前500字符辅助调试 break return pd.DataFrame() # 所有尝试都失败，返回空DataFrame # 示例调用 if __name__ == '__main__': query = "沪深300成分股 且 市盈率小于20" df_result = get_stock_pick(query, perpage=100) if not df_result.empty: print(f"找到 {len(df_result)} 只股票:") print(df_result[['code', 'name']].head()) # 假设返回的字段包含code和name else: print("未获取到数据。")

代码关键点解析：

• URL与参数：base_url和params字典中的字段名和值必须通过抓包分析获得，不能臆想。示例中的question,page,perpage是常见的，但很可能还有query_type,secondary_intent,v,sign等关键参数。sign参数往往是加密的，是最大的难点。
• 请求头：User-Agent和Referer是最基本的反爬检查点。Referer最好动态构造，使其看起来是从正确的搜索页面跳转而来。
• Cookie：对于简单的公开数据查询，可能不需要Cookie。但如果查询涉及用户隐私数据（如自选股组合）或频率较高时，服务器可能会要求登录。此时你需要一个有效的Cookie。获取方式可以手动登录后从浏览器复制，或者用Selenium模拟登录后获取。请注意，分享或盗用他人Cookie是严重违规行为。
• 响应解析：response.json()将响应体解析为Python字典。你需要仔细查看这个字典的结构，找到股票数据实际存放的位置。路径可能是[‘data’][‘list’]，也可能是其他形式。字段名如code（代码）、name（名称）、market（市场）也需要确认。
• 错误处理与重试：网络请求不稳定，加入重试机制是提高鲁棒性的好习惯。同时，对不同的HTTP状态码和业务状态码（如status_code）进行区分处理，能让你更快定位问题。
• 防反爬与频率控制：在循环中调用此函数时，务必在请求间添加随机延时（例如time.sleep(random.uniform(1, 3))），避免请求过于频繁被服务器封禁IP。这是网络爬虫的基本道德和生存法则。

3.3 应对动态参数与签名（Sign）的挑战

这是本项目最核心的难点。现代Web应用普遍使用动态参数和签名来防止接口被轻易调用。你可能会在抓包时发现一个名为sign、token或v的参数，每次请求值都不同，并且由其他参数通过某种加密算法（如MD5、SHA1、AES或自定义算法）生成。

处理思路：

1. 定位生成代码：在浏览器的开发者工具中，搜索包含sign或相关参数名的JS文件。通常，生成签名的逻辑在前端JavaScript代码中。
2. 分析算法：找到相关的JS函数，尝试理解其算法。它可能只是简单的字符串拼接后取MD5，也可能是更复杂的HMAC。
3. Python复现：将JS算法翻译成Python代码。可以使用hashlib库进行MD5、SHA等计算。

简化方案（如果算法过于复杂）： 如果逆向JS算法难度太大，一个取巧但不推荐长期使用的方法是：先用Selenium打开页面，让页面JS自然执行生成必要的参数，然后通过Selenium的execute_script方法从浏览器上下文中提取出这些参数，再交给requests去发送请求。但这失去了requests高效的意义。

一个假设的签名生成示例： 假设我们分析发现sign是由question+page+perpage+一个固定密钥拼接后取MD5值的前8位。

import hashlib def generate_sign(question, page, perpage, secret_key='some_secret'): raw_string = f"{question}{page}{perpage}{secret_key}" md5_hash = hashlib.md5(raw_string.encode('utf-8')).hexdigest() return md5_hash[:8] # 取前8位作为sign # 然后在构造params时加入 params['sign'] = generate_sign(question, page, perpage)

重要提醒：实际的签名算法绝不会这么简单，且secret_key通常是隐藏的。这里的示例仅说明逻辑。你需要投入时间进行逆向分析，或者寻找是否有开源社区已经破解了相关算法（但需注意法律和合规风险）。

4. 功能扩展与数据深度处理

基础选股功能实现后，我们可以围绕它构建更实用的工具链。

4.1 多条件批量查询与结果聚合

我们常常需要测试多个选股条件。可以写一个简单的批量查询函数。

def batch_stock_pick(question_list, delay=2): """ 批量查询多个选股条件 Args: question_list (list): 选股条件字符串列表 delay (int): 每次查询间隔秒数，避免请求过快 Returns: dict: 键为条件，值为对应的DataFrame """ results = {} for idx, q in enumerate(question_list): print(f"正在查询 ({idx+1}/{len(question_list)}): {q}") df = get_stock_pick(q) results[q] = df if not df.empty: print(f" 找到 {len(df)} 只股票。") time.sleep(delay) # 非常重要的延时，尊重服务器 return results # 使用示例 conditions = ["ROE > 15%", "营收同比增长率 > 30%", "市盈率 < 30 且 市净率 < 3"] all_results = batch_stock_pick(conditions) for cond, df in all_results.items(): if not df.empty: print(f"\n条件 ‘{cond}’ 的结果（前5只）:") print(df[['code', 'name']].head())

4.2 数据清洗与格式化

从API获取的原始数据可能包含不需要的字段、格式不统一的代码等，需要进行清洗。

def clean_stock_data(df): """ 清洗和格式化股票数据DataFrame """ if df.empty: return df df_clean = df.copy() # 1. 选择需要的列 (根据实际API返回字段调整) # 假设API返回了：code, name, market, latest_price, pe_ratio columns_to_keep = ['code', 'name', 'market', 'latest_price', 'pe_ratio'] # 只保留存在的列 existing_cols = [col for col in columns_to_keep if col in df_clean.columns] df_clean = df_clean[existing_cols] # 2. 格式化股票代码：通常同花顺返回的code是6位数字，需要加上市场前缀 if 'market' in df_clean.columns and 'code' in df_clean.columns: # 假设 market: 1-沪市, 2-深市, 3-创业板... def format_stock_code(row): code = str(row['code']).zfill(6) # 补零到6位 market = row['market'] if market == 1: return f"SH{code}" elif market == 2: return f"SZ{code}" else: return code df_clean['full_code'] = df_clean.apply(format_stock_code, axis=1) # 3. 处理数值字段：去除字符串中的逗号，转换为浮点数 numeric_cols = ['latest_price', 'pe_ratio'] for col in numeric_cols: if col in df_clean.columns: # 先转换为字符串，替换逗号，再转浮点 df_clean[col] = pd.to_numeric(df_clean[col].astype(str).str.replace(',', ''), errors='coerce') # 4. 去重（如果有的话） if 'full_code' in df_clean.columns: df_clean = df_clean.drop_duplicates(subset=['full_code']) return df_clean # 使用示例 df_raw = get_stock_pick("净利润同比增长率>50%") df_clean = clean_stock_data(df_raw) print(df_clean.head())

4.3 结果导出与持久化

将筛选出的股票列表保存到文件，方便后续使用或分享。

def export_results(df, filename='stock_pick_result.csv', format='csv'): """ 将选股结果导出到文件 Args: df (pd.DataFrame): 要导出的数据 filename (str): 文件名 format (str): 导出格式，支持 'csv', 'excel', 'json' """ if df.empty: print("数据为空，无需导出。") return if format == 'csv': df.to_csv(filename, index=False, encoding='utf-8-sig') # utf-8-sig支持Excel中文 print(f"结果已导出到CSV文件: {filename}") elif format == 'excel': # 需要安装 openpyxl: pip install openpyxl df.to_excel(filename, index=False) print(f"结果已导出到Excel文件: {filename}") elif format == 'json': df.to_json(filename, orient='records', force_ascii=False) print(f"结果已导出到JSON文件: {filename}") else: print(f"不支持的格式: {format}") # 使用示例 df = get_stock_pick("换手率>5% 且 量比>1.5") export_results(df, 'high_turnover_stocks.xlsx', format='excel')

5. 常见问题排查与稳定性优化

在实际使用中，你会遇到各种各样的问题。下面是我踩过的一些坑和解决方案。

5.1 请求返回空数据或错误状态码

• 问题：调用get_stock_pick后，返回的df是空的，或者status_code不是0。
• 排查步骤：
  1. 检查网络：首先确认你的Python环境可以正常访问https://www.iwencai.com。
  2. 更新请求头：用最新的浏览器（如Chrome）重新抓包，对比你的headers（特别是User-Agent）是否过时。Referer的值是否与当前查询匹配。
  3. 验证参数：这是最常见的问题。仔细对比你代码中的params和浏览器实际发送的Params或Form Data，确保每一个键值对都完全一致，包括那些看似随机的数字或字符串。
  4. 检查Cookie：如果查询需要登录，而你未提供Cookie，或者Cookie已过期，就会失败。尝试在headers中加入从浏览器复制的有效Cookie再试。
  5. 打印调试信息：在代码中打印出完整的请求URL和参数，手动在浏览器或curl中测试，看是否能复现。
  6. 查看响应内容：即使返回错误，也要打印response.text，里面往往包含了服务器给出的具体错误信息，如“参数错误”、“签名验证失败”等，这是最直接的线索。

5.2 遇到IP封锁或访问频率限制

• 现象：最初几次请求成功，后续请求返回403 Forbidden、429 Too Many Requests或直接断开连接。
• 解决方案：
  • 降低请求频率：这是必须的。在每次请求之间加入随机延时time.sleep(random.uniform(3, 7))。批量查询时尤其要注意。
  • 使用代理IP池：如果单IP请求量很大，可以考虑使用代理IP。但普通个人使用，通过控制频率基本可以满足需求。
  • 伪装得更像浏览器：确保你的headers完整，包含Accept、Accept-Language、Connection等字段。有些反爬系统会检查这些。
  • 处理Cookie会话：如果服务器使用会话（Session）管理，考虑使用requests.Session()对象来保持会话，而不是每次新建请求。这能让你的行为更像一个连贯的用户。

5.3 数据解析失败或字段缺失

• 问题：response.json()解析失败，或者解析后找不到预期的data或list字段。
• 排查步骤：
  1. 确认响应格式：首先打印response.text的前几百个字符，看看它到底是JSON还是HTML。有时请求出错会返回一个错误页面（HTML）。
  2. 检查JSON路径：如果确实是JSON，用工具（如在线JSON格式化器）或Python的json.loads()仔细查看其完整结构。数据存放的路径可能发生了变化。
  3. 处理字段名变化：API返回的字段名可能微调，比如从stock_code变成了code。你需要更新数据清洗函数中的字段名列表。
  4. 异常处理：在解析代码周围使用try...except，确保即使某个字段缺失或格式异常，程序也不会崩溃，而是记录错误并继续处理其他数据。

5.4 应对接口变更与维护

同花顺作为商业平台，其接口随时可能变更。你的代码可能在某一天突然失效。

• 策略：
  • 封装与隔离：将核心的请求和解析逻辑封装在少数几个函数里（如get_stock_pick）。这样当接口变化时，你只需要修改这几个函数，而不必改动整个项目。
  • 定期测试：建立一个简单的测试脚本，定期（如每周）运行一次基础查询，验证功能是否正常。
  • 关注网络请求：如果测试失败，第一时间重新抓包，对比新旧请求的差异。差异点通常就是需要修改的地方。
  • 备份思路：如果直接API调用变得过于困难，可以将方案降级为使用Selenium的浏览器自动化方案作为备用。虽然慢，但通常更稳定。

6. 进阶应用：构建简易的本地选股系统

有了稳定的数据获取能力，我们就可以做一些更有意思的事情，比如构建一个本地的、可自定义的选股系统雏形。

6.1 策略组合与交集/并集运算

我们经常需要找出同时满足多个条件（交集）或满足任一条件（并集）的股票。

def combine_strategies(strategy_dict, combine_method='intersection'): """ 对多个选股策略的结果进行集合运算 Args: strategy_dict (dict): 策略名为键，策略结果DataFrame为值的字典。 每个DataFrame应包含‘full_code’列。 combine_method (str): ‘intersection’ (交集) 或 ‘union’ (并集)。 Returns: pd.DataFrame: 合并后的股票列表 """ if not strategy_dict: return pd.DataFrame() # 确保每个DataFrame都有‘full_code’列 code_sets = [] for name, df in strategy_dict.items(): if 'full_code' in df.columns and not df.empty: code_sets.append(set(df['full_code'])) else: print(f"策略 ‘{name}’ 结果为空或缺少‘full_code’列，已忽略。") code_sets.append(set()) # 加入空集避免影响运算 if not code_sets: return pd.DataFrame() if combine_method == 'intersection': final_codes = set.intersection(*code_sets) elif combine_method == 'union': final_codes = set.union(*code_sets) else: raise ValueError("combine_method 必须是 ‘intersection’ 或 ‘union’") # 从第一个非空的DataFrame中获取股票的完整信息 sample_df = next((df for df in strategy_dict.values() if not df.empty), pd.DataFrame()) if not sample_df.empty: # 根据最终代码列表筛选出完整的股票信息 result_df = sample_df[sample_df['full_code'].isin(final_codes)].copy() # 可能需要去重，如果多个策略提供了同一只股票的重复信息 result_df = result_df.drop_duplicates(subset=['full_code']) return result_df else: return pd.DataFrame() # 使用示例 # 假设我们已经通过 batch_stock_pick 获取了多个策略的结果 strategies = { "高成长": get_stock_pick("营收增长率>30% 且 净利润增长率>30%"), "低估值": get_stock_pick("市盈率<15 且 市净率<1.5"), "高分红": get_stock_pick("股息率>3%"), } # 清洗数据，生成 full_code for key in strategies: strategies[key] = clean_stock_data(strategies[key]) # 找出同时满足“高成长”和“低估值”的股票（交集） growth_and_cheap = combine_strategies({"成长": strategies["高成长"], "估值": strategies["低估值"]}, combine_method='intersection') print(f"同时满足高成长和低估值的股票有 {len(growth_and_cheap)} 只。") # 找出满足“高成长”、“低估值”、“高分红”中任意一个条件的股票（并集） any_good = combine_strategies(strategies, combine_method='union') print(f"满足任一条件的股票有 {len(any_good)} 只。")

6.2 定时任务与数据监控

我们可以使用Python的schedule库或操作系统的定时任务（如cron, Windows任务计划程序）来定期执行选股策略，并将结果通过邮件或消息应用发送给自己。

import schedule import time from datetime import datetime # 假设有一个发送邮件的函数 send_email def daily_screening_job(): """每日执行的选股任务""" print(f"{datetime.now()}: 开始执行每日选股...") try: # 1. 执行选股策略 df_growth = get_stock_pick("近三年净利润复合增长率>20%") df_growth = clean_stock_data(df_growth) df_roe = get_stock_pick("ROE>15%") df_roe = clean_stock_data(df_roe) # 2. 组合策略（取交集） final_stocks = combine_strategies({"成长": df_growth, "ROE": df_roe}, combine_method='intersection') # 3. 处理结果 if not final_stocks.empty: # 保存到文件 filename = f"daily_picks_{datetime.now().strftime('%Y%m%d')}.csv" export_results(final_stocks, filename, 'csv') # 构造邮件内容 mail_subject = f"每日选股结果 {datetime.now().strftime('%Y-%m-%d')}" mail_body = f"今日符合‘高成长且高ROE’策略的股票共 {len(final_stocks)} 只。\n\n" mail_body += final_stocks[['full_code', 'name', 'latest_price', 'pe_ratio']].to_string(index=False) # 发送邮件 (需要实现send_email函数) # send_email(mail_subject, mail_body, attachment=filename) print(f"发现 {len(final_stocks)} 只股票，结果已保存。") else: print("今日未发现符合条件的股票。") except Exception as e: print(f"选股任务执行失败: {e}") print(f"{datetime.now()}: 每日选股任务完成。") # 设置定时任务（例如，每个工作日早上9点30分运行） schedule.every().monday.at("09:30").do(daily_screening_job) schedule.every().tuesday.at("09:30").do(daily_screening_job) schedule.every().wednesday.at("09:30").do(daily_screening_job) schedule.every().thursday.at("09:30").do(daily_screening_job) schedule.every().friday.at("09:30").do(daily_screening_job) print("定时选股任务已启动...") while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次

注意：上述定时任务代码是一个无限循环，适合在服务器或常年开机的电脑上运行。对于个人使用，更简单的方式是写一个脚本，然后用系统的定时任务工具来调用它。

6.3 与本地数据库或回测框架集成

对于更严肃的量化研究，你可能希望将选股结果存入数据库（如SQLite, MySQL）或直接喂给回测框架（如Zipline, Backtrader, Qlib）。

集成SQLite示例：

import sqlite3 def save_to_database(df, db_path='stock_picks.db', table_name='selected_stocks'): """将选股结果保存到SQLite数据库""" if df.empty: return # 确保有必要的列，比如‘full_code’作为唯一标识 if 'full_code' not in df.columns: print("DataFrame缺少‘full_code’列，无法保存。") return conn = sqlite3.connect(db_path) cursor = conn.cursor() # 创建表（如果不存在） # 这里假设df的列就是我们要存的列，实际中可能需要更精细的表结构设计 columns_def = ', '.join([f'"{col}" TEXT' for col in df.columns]) create_table_sql = f''' CREATE TABLE IF NOT EXISTS {table_name} ( id INTEGER PRIMARY KEY AUTOINCREMENT, query_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, {columns_def} ) ''' cursor.execute(create_table_sql) # 插入数据 # 使用占位符防止SQL注入 placeholders = ', '.join(['?' for _ in df.columns]) column_names = ', '.join([f'"{col}"' for col in df.columns]) insert_sql = f'INSERT INTO {table_name} ({column_names}) VALUES ({placeholders})' # 将DataFrame转换为元组列表 data_to_insert = [tuple(row) for row in df.itertuples(index=False, name=None)] cursor.executemany(insert_sql, data_to_insert) conn.commit() conn.close() print(f"成功将 {len(df)} 条记录插入数据库 {db_path}.{table_name}") # 使用示例 df_picks = get_stock_pick("技术形态: 突破20日高点") df_picks = clean_stock_data(df_picks) save_to_database(df_picks)

通过这种方式，你可以积累历史选股记录，用于后续的策略分析和绩效评估。

7. 伦理、合规与风险提示

在结束之前，我必须强调一些重要的非技术事项。技术是一把双刃剑，用对了地方能提升效率，用错了则可能带来麻烦。

• 尊重数据源：同花顺等平台的数据是其资产。你的自动化脚本应仅限于个人学习、研究和低频率使用。大规模、高频的抓取行为会占用服务器资源，可能违反其服务条款，导致你的IP甚至账户被封禁。
• 控制请求频率：务必在代码中设置合理的延时（time.sleep），模拟人类操作的速度。切勿尝试并发大量请求。
• 遵守robots.txt：虽然robots.txt不是法律文件，但它是网站所有者表达爬虫抓取意愿的途径。检查目标网站的相关规定是良好的网络公民行为。
• 数据用途：获取的数据应用于个人分析决策参考。严禁用于任何商业倒卖、数据贩售或侵害他人权益的行为。
• 技术风险：本文提供的代码示例基于特定时间点的网络分析，接口和反爬策略会更新。你需要具备一定的Python和网络知识来维护和调试代码。它可能随时会“失效”。
• 投资风险：最后，也是最重要的，任何基于量化筛选的股票列表都只是辅助工具。股市有风险，投资需谨慎。代码选出的股票不代表必然上涨，你需要结合更深度的基本面分析、技术面分析和风险管理来做出投资决策。

构建这样一个工具的过程，其价值远不止于得到一个股票列表。它强迫你去理解数据背后的逻辑，去处理网络请求中的各种异常，去设计稳健的程序结构。这份经历本身，对于任何从事数据分析、量化研究或后端开发的人来说，都是非常宝贵的。希望这篇长文能为你提供一个坚实的起点，而不仅仅是几行可以“复制粘贴”的代码。在实际操作中，耐心调试和持续学习才是成功的关键。