1. 项目缘起为什么我要自己动手写一个问财选股工具做量化交易或者策略研究的朋友应该都绕不开一个场景如何高效地获取符合特定条件的股票列表无论是回测还是实盘监控第一步都是“选股”。市面上有很多工具比如同花顺的问财它提供了一个非常强大的自然语言选股界面你可以输入“连续三天放量上涨”、“MACD金叉且股价站上20日均线”这样的描述它就能给你筛选出结果。这功能确实方便尤其是对于不熟悉编程的投资者。但问题也随之而来。当你需要定期、批量地执行这些选股条件或者想把选股结果无缝接入到自己的Python分析、回测框架里时手动在网页上操作就显得力不从心了。每次都要打开浏览器、登录、输入条件、点击查询、等待结果、再手动复制数据……这个过程不仅繁琐而且无法自动化更别提在策略迭代中快速测试成百上千个条件组合了。所以一个很自然的需求就产生了能不能用Python代码直接调用同花顺问财的选股逻辑把结果拿回来自己处理这就是“同花顺问财选股Python源码”这个标题背后绝大多数开发者或量化爱好者的核心诉求。我们想要的不是一个简单的网页爬虫而是一个能够稳定、可靠、模拟真实用户行为从问财获取结构化选股数据的程序化接口。我最初也是被这个需求驱动经历了从简单爬虫到模拟登录再到处理各种反爬和数据结构问题的完整过程。网上能找到的零星代码要么过于陈旧已经失效要么功能不全、稳定性差。今天我就把自己踩过坑、验证过的相对完整的实现思路和核心代码分享出来目标是构建一个健壮的、可复用的问财选股Python模块。请注意以下内容基于技术研究和学习目的实际使用中务必尊重数据源的服务条款控制请求频率避免对目标服务器造成不必要的压力。2. 核心思路拆解问财选股的数据链路与模拟策略在动手写代码之前我们必须先搞清楚同花顺问财页面是怎么工作的。只有理解了数据流转的链条才能找到合适的介入点。通过浏览器开发者工具按F12对问财页面进行网络抓包分析是必不可少的第一步。2.1 网络请求分析从自然语言到股票列表当你访问同花顺问财页面并输入一个选股条件时例如“市值大于100亿”页面会发起一系列网络请求。我们需要找到最关键的那个——负责将你的查询条件发送给服务器并返回股票列表的请求。打开问财页面访问同花顺问财的官方网站。输入查询条件在搜索框输入“市值大于100亿”。开启开发者工具按F12切换到Network网络选项卡。记得勾选Preserve log保留日志以防页面跳转时请求记录被清除。执行查询点击搜索按钮或按回车。筛选请求在纷繁复杂的请求中我们需要寻找特征。通常返回股票列表数据的请求是一个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实现策略各有优劣直接请求API推荐但需处理反爬原理直接模拟浏览器向上述分析得到的核心API地址发送HTTP请求携带必要的参数和请求头。优点效率最高速度最快代码最简洁。直接获取结构化的数据通常是JSON解析方便。挑战需要精准构造请求参数和请求头特别是动态生成的token、sign等加密参数。同花顺的反爬机制可能会升级需要持续维护。对于需要登录的查询需要先模拟登录获取有效的Cookie。适用场景技术能力强愿意持续维护代码追求最高效率和稳定性的用户。通过Selenium等浏览器自动化工具原理使用Selenium库控制一个真实的浏览器如Chrome打开问财页面模拟用户输入、点击、等待页面加载然后从页面HTML中解析出股票数据。优点几乎可以应对任何反爬措施因为行为与真人操作完全一致。不需要深入分析复杂的API参数。缺点速度慢资源消耗大需要运行浏览器实例。代码稳定性受网页UI变化影响。解析HTML结构相对复杂。适用场景API难以逆向、查询条件非常简单、对运行速度不敏感的一次性任务。寻找替代数据源或第三方封装库原理使用其他财经数据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, page1, perpage50, retry3): 根据同花顺问财条件筛选股票 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, */*; q0.01, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, br, Content-Type: application/x-www-form-urlencoded; charsetUTF-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, paramsparams, headersheaders, timeout10) 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(fAPI返回异常状态: {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第{i1}次请求失败: {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, perpage100) 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或自定义算法生成。处理思路定位生成代码在浏览器的开发者工具中搜索包含sign或相关参数名的JS文件。通常生成签名的逻辑在前端JavaScript代码中。分析算法找到相关的JS函数尝试理解其算法。它可能只是简单的字符串拼接后取MD5也可能是更复杂的HMAC。Python复现将JS算法翻译成Python代码。可以使用hashlib库进行MD5、SHA等计算。简化方案如果算法过于复杂 如果逆向JS算法难度太大一个取巧但不推荐长期使用的方法是先用Selenium打开页面让页面JS自然执行生成必要的参数然后通过Selenium的execute_script方法从浏览器上下文中提取出这些参数再交给requests去发送请求。但这失去了requests高效的意义。一个假设的签名生成示例 假设我们分析发现sign是由questionpageperpage一个固定密钥拼接后取MD5值的前8位。import hashlib def generate_sign(question, page, perpage, secret_keysome_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, delay2): 批量查询多个选股条件 Args: question_list (list): 选股条件字符串列表 delay (int): 每次查询间隔秒数避免请求过快 Returns: dict: 键为条件值为对应的DataFrame results {} for idx, q in enumerate(question_list): print(f正在查询 ({idx1}/{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 fSH{code} elif market 2: return fSZ{code} else: return code df_clean[full_code] df_clean.apply(format_stock_code, axis1) # 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(,, ), errorscoerce) # 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, filenamestock_pick_result.csv, formatcsv): 将选股结果导出到文件 Args: df (pd.DataFrame): 要导出的数据 filename (str): 文件名 format (str): 导出格式支持 csv, excel, json if df.empty: print(数据为空无需导出。) return if format csv: df.to_csv(filename, indexFalse, encodingutf-8-sig) # utf-8-sig支持Excel中文 print(f结果已导出到CSV文件: {filename}) elif format excel: # 需要安装 openpyxl: pip install openpyxl df.to_excel(filename, indexFalse) print(f结果已导出到Excel文件: {filename}) elif format json: df.to_json(filename, orientrecords, force_asciiFalse) print(f结果已导出到JSON文件: {filename}) else: print(f不支持的格式: {format}) # 使用示例 df get_stock_pick(换手率5% 且 量比1.5) export_results(df, high_turnover_stocks.xlsx, formatexcel)5. 常见问题排查与稳定性优化在实际使用中你会遇到各种各样的问题。下面是我踩过的一些坑和解决方案。5.1 请求返回空数据或错误状态码问题调用get_stock_pick后返回的df是空的或者status_code不是0。排查步骤检查网络首先确认你的Python环境可以正常访问https://www.iwencai.com。更新请求头用最新的浏览器如Chrome重新抓包对比你的headers特别是User-Agent是否过时。Referer的值是否与当前查询匹配。验证参数这是最常见的问题。仔细对比你代码中的params和浏览器实际发送的Params或Form Data确保每一个键值对都完全一致包括那些看似随机的数字或字符串。检查Cookie如果查询需要登录而你未提供Cookie或者Cookie已过期就会失败。尝试在headers中加入从浏览器复制的有效Cookie再试。打印调试信息在代码中打印出完整的请求URL和参数手动在浏览器或curl中测试看是否能复现。查看响应内容即使返回错误也要打印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字段。排查步骤确认响应格式首先打印response.text的前几百个字符看看它到底是JSON还是HTML。有时请求出错会返回一个错误页面HTML。检查JSON路径如果确实是JSON用工具如在线JSON格式化器或Python的json.loads()仔细查看其完整结构。数据存放的路径可能发生了变化。处理字段名变化API返回的字段名可能微调比如从stock_code变成了code。你需要更新数据清洗函数中的字段名列表。异常处理在解析代码周围使用try...except确保即使某个字段缺失或格式异常程序也不会崩溃而是记录错误并继续处理其他数据。5.4 应对接口变更与维护同花顺作为商业平台其接口随时可能变更。你的代码可能在某一天突然失效。策略封装与隔离将核心的请求和解析逻辑封装在少数几个函数里如get_stock_pick。这样当接口变化时你只需要修改这几个函数而不必改动整个项目。定期测试建立一个简单的测试脚本定期如每周运行一次基础查询验证功能是否正常。关注网络请求如果测试失败第一时间重新抓包对比新旧请求的差异。差异点通常就是需要修改的地方。备份思路如果直接API调用变得过于困难可以将方案降级为使用Selenium的浏览器自动化方案作为备用。虽然慢但通常更稳定。6. 进阶应用构建简易的本地选股系统有了稳定的数据获取能力我们就可以做一些更有意思的事情比如构建一个本地的、可自定义的选股系统雏形。6.1 策略组合与交集/并集运算我们经常需要找出同时满足多个条件交集或满足任一条件并集的股票。def combine_strategies(strategy_dict, combine_methodintersection): 对多个选股策略的结果进行集合运算 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_methodintersection) print(f同时满足高成长和低估值的股票有 {len(growth_and_cheap)} 只。) # 找出满足“高成长”、“低估值”、“高分红”中任意一个条件的股票并集 any_good combine_strategies(strategies, combine_methodunion) 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(ROE15%) df_roe clean_stock_data(df_roe) # 2. 组合策略取交集 final_stocks combine_strategies({成长: df_growth, ROE: df_roe}, combine_methodintersection) # 3. 处理结果 if not final_stocks.empty: # 保存到文件 filename fdaily_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(indexFalse) # 发送邮件 (需要实现send_email函数) # send_email(mail_subject, mail_body, attachmentfilename) 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_pathstock_picks.db, table_nameselected_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 fINSERT INTO {table_name} ({column_names}) VALUES ({placeholders}) # 将DataFrame转换为元组列表 data_to_insert [tuple(row) for row in df.itertuples(indexFalse, nameNone)] 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和网络知识来维护和调试代码。它可能随时会“失效”。投资风险最后也是最重要的任何基于量化筛选的股票列表都只是辅助工具。股市有风险投资需谨慎。代码选出的股票不代表必然上涨你需要结合更深度的基本面分析、技术面分析和风险管理来做出投资决策。构建这样一个工具的过程其价值远不止于得到一个股票列表。它强迫你去理解数据背后的逻辑去处理网络请求中的各种异常去设计稳健的程序结构。这份经历本身对于任何从事数据分析、量化研究或后端开发的人来说都是非常宝贵的。希望这篇长文能为你提供一个坚实的起点而不仅仅是几行可以“复制粘贴”的代码。在实际操作中耐心调试和持续学习才是成功的关键。