1. 项目概述从数据洪流中淘金做量化分析或者基本面研究的朋友手里没点资金流向数据心里总是不踏实。股价的涨跌背后是资金的进进出出在推动。你能看到K线图能看到成交量但更细粒度的——主力资金、散户资金、超大单、大单、中单、小单——这些分档数据才是洞察市场情绪和主力意图的关键。东方财富网的资金流向页面几乎是国内股民和数据分析师的“数据水源地”之一它提供了个股和板块非常详尽的历史资金流向明细。但这个页面是给人看的不是给程序直接吃的。手动一天天去翻、去抄效率低到令人发指而且容易出错。所以用Python写个爬虫把这些结构化的资金流向数据自动抓下来并规整地存入数据库就成了一个非常实际且高频的需求。这不仅仅是“爬数据”更是构建个人量化分析数据底座的起点。有了这个自动化流程你才能腾出手来专注于更重要的策略研究和模型构建。接下来我会带你完整走一遍这个流程。从分析页面结构、设计数据表到编写稳健的爬虫、处理反爬策略最后将数据持久化到数据库。我会分享我在这个过程中踩过的坑和总结的技巧目标是让你拿到一个可以直接运行、易于扩展的解决方案。2. 核心思路与工具选型2.1 目标数据源分析与策略制定东方财富的资金流向数据页面通常URL模式类似http://data.eastmoney.com/zjlx/股票代码.html。例如贵州茅台就是http://data.eastmoney.com/zjlx/600519.html。我们的核心目标是抓取页面中的“历史资金流向”表格数据。策略选择动态渲染 vs. 静态接口首先需要判断页面数据是直接加载在HTML中还是通过JavaScript动态渲染的。这是爬虫设计的第一步方向错了后面全白费。初步探查用浏览器打开页面右键“查看网页源代码”。在源代码里搜索表格里的典型数据比如“净流入”、“主力资金”。如果搜不到基本可以断定是动态加载的。确认动态加载在浏览器开发者工具的“网络”Network选项卡中刷新页面筛选XHR或Fetch请求。你会很快发现一个关键的接口请求其URL通常包含API、GetData等字样返回的是JSON格式的数据。这才是数据的“真身”。接口分析找到这个接口后重点分析它的请求参数。通常包括code股票代码可能带市场前缀如SH600519。type数据类型可能区分日、周、月。page和size分页参数。一些时间戳或加密参数用于反爬。我们的选择直接请求这个后端JSON接口而不是用Selenium或Playwright去模拟浏览器渲染整个页面。理由很充分效率极高省去了加载CSS、JS、图片的时间消耗资源少速度更快也更稳定。难点在于如何找到并模拟这个接口的请求。工具选型背后的逻辑Requests json用于发起HTTP请求和解析JSON数据这是核心。Pandas并非必须但极其推荐。因为接口返回的JSON数据很容易被转换为DataFrame而Pandas在数据清洗、转换如日期格式处理方面是“神器”一行代码能顶十行原生Python操作。SQLAlchemy数据库ORM工具。为什么不用简单的sqlite3模块因为SQLAlchemy提供了数据库无关的抽象层。今天你用SQLite做测试明天想迁移到MySQL或PostgreSQL几乎不需要修改代码。它还能很好地处理连接池、数据类型映射让代码更健壮、更专业。Schedule / APScheduler如果你需要定时任务比如每天收盘后自动爬取这些库能帮你实现可靠的定时调度。2.2 数据库表结构设计在写爬虫之前先想好数据怎么存。一个好的表结构是后续所有分析的基础。资金流向表的核心字段通常包括字段名数据类型说明设计理由idINTEGER PRIMARY KEY自增主键每条记录的唯一标识便于管理和关联。symbolVARCHAR(10)股票代码如600519 必须建立索引因为这是最常用的查询条件。trade_dateDATE交易日期如2023-10-27 与symbol组成唯一约束防止重复插入。close_priceDECIMAL(10,2)收盘价记录当日的价格用于计算和分析。change_pctDECIMAL(6,2)涨跌幅(%)股价变动比例。main_net_inflowDECIMAL(16,2)主力净流入(元)核心指标通常指超大单大单的净额。main_net_inflow_rateDECIMAL(6,2)主力净流入占比(%)净流入占成交额的比例比绝对值更有意义。retail_net_inflowDECIMAL(16,2)散户净流入(元)通常指小单的净额与主力资金形成对照。超大单流入超大单流出DECIMAL(16,2)分档资金流更细粒度的数据有助于深度分析。建议分开存储。turnover_rateDECIMAL(6,2)换手率(%)市场活跃度指标。created_atTIMESTAMP记录创建时间默认CURRENT_TIMESTAMP用于追踪数据获取时间。设计要点与避坑指南唯一约束一定要在(symbol, trade_date)上建立唯一约束UNIQUE CONSTRAINT。这是防止数据重复的“铁闸”。爬虫运行时无论是网络重试还是手动执行都能确保数据库里同一只股票同一天只有一条记录。索引策略除了唯一约束自带的索引如果经常按日期范围查询如“查询某股票2023年所有数据”可以考虑在trade_date上单独建立索引。但索引不是越多越好会影响写入速度。金额字段精度资金数据可能很大使用DECIMAL(16,2)可以安全存储千亿级别的金额并保留两位小数。DECIMAL类型能保证精确计算避免浮点数误差。日期类型务必使用数据库的DATE类型存储trade_date而不是VARCHAR。这能让你直接使用数据库的日期函数进行高效的区间查询和聚合。3. 爬虫核心实现与反爬对抗3.1 逆向分析接口与参数构造这是最具技术挑战性的一步。以东方财富为例我们打开开发者工具定位到那个返回历史资金流JSON数据的请求。假设我们找到的接口是http://datacenter.eastmoney.com/api/data/get?typeRPTA_WEB_RZRQ_GGMXstyALLsourceWEB...这只是一个示例实际接口可能会变。关键步骤复制为cURL在Network中找到该请求右键选择“Copy” - “Copy as cURL (bash)”。这将给你一个完整的命令行请求。转换与分析将cURL命令粘贴到在线工具如 https://curlconverter.com/python/中可以将其转换为Python的Requests代码。这能快速得到完整的请求头headers和参数params。参数解构仔细研究这些参数。你会发现一些固定参数如type、sty、source。以及一些动态参数code可能是SH600519或SZ000001这种格式。page页码。size每页条数可能是20、50、100。filter可能是一个包含查询条件的字符串如(TRADE_DATE‘2022-01-01’)。最关键的t或_后面跟的一串数字这通常是一个当前时间戳用于防止缓存。这是最常见的反爬手段之一。请求头重点关注User-Agent、Referer、Accept。Referer通常需要设置为资金流向页面的URL这是服务器验证请求来源的常用方法。实操心得东方财富的接口参数命名可能不那么直观filter参数里的表达式语法需要仔细从接口示例中揣摩。一个技巧是先通过浏览器正常访问抓取第一页的请求然后尝试只修改page和size参数来获取更多数据。如果失败再检查是否有其他动态生成的令牌token。3.2 稳健的爬虫代码编写下面是一个融合了关键技巧的爬虫核心函数示例import requests import pandas as pd import time from datetime import datetime, timedelta import json def fetch_stock_capital_flow(symbol, start_dateNone, end_dateNone): 获取单只股票的历史资金流向数据 Args: symbol (str): 股票代码如 600519 start_date (str): 开始日期格式 2023-01-01默认为一年前 end_date (str): 结束日期格式 2023-10-27默认为今天 Returns: pandas.DataFrame: 资金流向数据列名已清洗 # 1. 构造请求头Headers - 模拟浏览器 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36, Referer: fhttp://data.eastmoney.com/zjlx/{symbol}.html, # 关键Referer必须正确 Accept: application/json, text/plain, */*, Accept-Language: zh-CN,zh;q0.9,en;q0.8, } # 2. 构造请求参数Params # 这里需要根据实际找到的接口调整参数名和值 params { type: RPTA_WEB_RZRQ_GGMX, # 示例类型需替换 sty: ALL, source: WEB, code: fSH{symbol} if symbol.startswith((6, 5, 9)) else fSZ{symbol}, # 判断市场 page: 1, size: 100, # 每页条数可调大以减少请求次数 filter: f(TRADE_DATE{start_date}) if start_date else None, t: int(time.time() * 1000), # 关键时间戳防缓存 } # 清理空参数 params {k: v for k, v in params.items() if v is not None} all_data [] max_pages 10 # 安全限制防止意外死循环 for page in range(1, max_pages 1): params[page] page print(f正在抓取 {symbol} 第 {page} 页...) try: # 3. 发送请求 # 接口URL需替换为实际地址 response requests.get( http://datacenter.eastmoney.com/api/data/get, paramsparams, headersheaders, timeout15 # 设置超时避免长时间等待 ) response.raise_for_status() # 如果状态码不是200抛出异常 # 4. 解析JSON json_data response.json() # 东方财富的接口数据通常嵌套在 result - data 路径下 data_list json_data.get(result, {}).get(data, []) if not data_list: print(f第 {page} 页无数据停止爬取。) break all_data.extend(data_list) # 5. 礼貌性延迟降低请求频率 time.sleep(1.5) # 重要避免请求过快被封IP except requests.exceptions.RequestException as e: print(f请求第 {page} 页时发生错误: {e}) break except json.JSONDecodeError as e: print(f解析第 {page} 页JSON时发生错误: {e}) print(f响应文本: {response.text[:500]}) # 打印前500字符辅助调试 break # 6. 转换为DataFrame并清洗 if not all_data: return pd.DataFrame() df pd.DataFrame(all_data) # 重命名列使其更易懂 column_mapping { SECURITY_CODE: symbol, TRADE_DATE: trade_date, CLOSE_PRICE: close_price, CHANGE_PCT: change_pct, MAIN_NET_INFLOW: main_net_inflow, MAIN_NET_INFLOW_RATE: main_net_inflow_rate, # ... 其他字段映射 } df.rename(columnscolumn_mapping, inplaceTrue) # 转换日期格式 if trade_date in df.columns: df[trade_date] pd.to_datetime(df[trade_date]).dt.date # 转换数值类型 numeric_columns [close_price, change_pct, main_net_inflow, main_net_inflow_rate] for col in numeric_columns: if col in df.columns: df[col] pd.to_numeric(df[col], errorscoerce) # 错误值转为NaN return df注意事项错误处理网络请求必须包含try...except并处理超时、状态码异常、JSON解析失败等情况。爬虫要能“优雅地失败”而不是整个崩溃。延迟Sleeptime.sleep(random.uniform(1, 3))比固定延迟更好模拟人类操作的不确定性。这是对目标网站最基本的尊重也是保护自己IP不被封禁的有效手段。User-Agent轮换如果需要大规模爬取准备一个User-Agent列表进行轮换可以进一步降低被识别风险。4. 数据持久化写入数据库数据抓取到Pandas DataFrame后下一步就是高效、安全地存入数据库。这里我们使用SQLAlchemy配合SQLite本地开发测试首选或MySQL/PostgreSQL。4.1 使用SQLAlchemy建立连接与映射from sqlalchemy import create_engine, Column, Integer, String, Date, DECIMAL, TIMESTAMP, UniqueConstraint from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime # 1. 定义基类 Base declarative_base() # 2. 定义数据表模型ORM类 class StockCapitalFlow(Base): __tablename__ stock_capital_flow id Column(Integer, primary_keyTrue, autoincrementTrue) symbol Column(String(10), nullableFalse, indexTrue) # 建立索引 trade_date Column(Date, nullableFalse) close_price Column(DECIMAL(10, 2)) change_pct Column(DECIMAL(6, 2)) main_net_inflow Column(DECIMAL(16, 2)) main_net_inflow_rate Column(DECIMAL(6, 2)) retail_net_inflow Column(DECIMAL(16, 2)) # ... 其他字段 created_at Column(TIMESTAMP, defaultdatetime.now) # 3. 定义唯一约束防止重复数据 __table_args__ (UniqueConstraint(symbol, trade_date, nameuix_symbol_trade_date),) # 4. 创建数据库连接引擎 # 使用SQLite文件数据库适合本地测试 # engine create_engine(sqlite:///./stock_data.db, echoFalse) # echoTrue 可查看SQL日志 # 使用MySQL生产环境推荐 # 格式mysqlpymysql://用户名:密码主机:端口/数据库名 engine create_engine(mysqlpymysql://user:passwordlocalhost:3306/stock_db?charsetutf8mb4) # 5. 创建表如果不存在 Base.metadata.create_all(engine) # 6. 创建会话工厂 SessionLocal sessionmaker(bindengine)4.2 高效批量写入与去重策略直接将DataFrame一行行插入效率极低。Pandas的to_sql方法虽然方便但默认的插入方式if_existsappend无法处理唯一约束冲突会导致整个插入失败。我们需要更精细的控制。推荐方案使用SQLAlchemy Core进行批量“upsert”“Upsert”是Update和Insert的合成词即存在则更新不存在则插入。MySQL 5.7和PostgreSQL都支持ON DUPLICATE KEY UPDATE语法SQLite也有类似的INSERT OR REPLACE/IGNORE。def upsert_capital_flow_to_db(df, engine): 将DataFrame数据高效写入数据库并处理重复数据。 Args: df (pd.DataFrame): 清洗后的资金流向数据 engine: SQLAlchemy引擎 if df.empty: print(DataFrame为空跳过写入。) return # 确保列名与数据库模型一致 # 假设df的列名已经是我们定义好的英文名 # 方法使用Pandas SQLAlchemy Core执行原生SQL进行批量upsert # 这里以MySQL为例 table_name stock_capital_flow temp_table_name ftemp_{table_name} # 创建一个临时表用于暂存本次抓取的数据 with engine.begin() as conn: # 使用事务 # 1. 将DataFrame写入临时表 df.to_sql(temp_table_name, conn, if_existsreplace, indexFalse) # 2. 执行Upsert操作 # 核心SQL语句将临时表的数据合并到主表冲突时更新某些字段 upsert_sql f INSERT INTO {table_name} (symbol, trade_date, close_price, change_pct, main_net_inflow, main_net_inflow_rate, ...) SELECT symbol, trade_date, close_price, change_pct, main_net_inflow, main_net_inflow_rate, ... FROM {temp_table_name} ON DUPLICATE KEY UPDATE close_price VALUES(close_price), change_pct VALUES(change_pct), main_net_inflow VALUES(main_net_inflow), main_net_inflow_rate VALUES(main_net_inflow_rate), -- ... 更新其他需要覆盖的字段created_at 通常不更新 updated_at CURRENT_TIMESTAMP; -- 可以加一个更新时间的字段 conn.execute(text(upsert_sql)) # 3. 删除临时表 conn.execute(text(fDROP TABLE IF EXISTS {temp_table_name})) print(f成功写入/更新 {len(df)} 条记录到数据库。)实操心得批量Upsert是生产级爬虫数据入库的关键。直接循环插入每秒可能只能处理几十条而批量Upsert每秒可以处理上万条。临时表的思路避免了在Python内存中拼接庞大的SQL语句性能和安全性都更好。记住对于像收盘价、资金流这种客观历史数据一旦写入即使重复抓取也应该用新数据覆盖旧数据如果接口数据有修正的话。created_at记录首次抓取时间updated_at记录最后一次更新时间这个设计很实用。5. 项目整合与调度实战5.1 构建可配置的主程序将上述模块组装起来并增加一些工程化配置。# config.py import os from dataclasses import dataclass dataclass class Config: # 数据库连接字符串 DATABASE_URL os.getenv(DATABASE_URL, sqlite:///./stock_data.db) # 请求基础延迟秒 BASE_DELAY 1.5 # 每页数据大小 PAGE_SIZE 100 # 要爬取的股票代码列表 STOCK_LIST [600519, 000858, 300750, 002594] # 示例 # 数据抓取时间范围 START_DATE 2023-01-01 END_DATE 2023-10-27 # main.py import logging from config import Config from capital_flow_spider import fetch_stock_capital_flow from database import upsert_capital_flow_to_db, engine import time logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) def main(): config Config() failed_symbols [] for symbol in config.STOCK_LIST: logger.info(f开始处理股票: {symbol}) try: # 抓取数据 df fetch_stock_capital_flow( symbolsymbol, start_dateconfig.START_DATE, end_dateconfig.END_DATE ) if df.empty: logger.warning(f股票 {symbol} 未获取到数据。) continue logger.info(f股票 {symbol} 抓取到 {len(df)} 条记录。) # 写入数据库 upsert_capital_flow_to_db(df, engine) # 随机延迟避免对服务器造成压力 time.sleep(config.BASE_DELAY) except Exception as e: logger.error(f处理股票 {symbol} 时发生错误: {e}, exc_infoTrue) failed_symbols.append(symbol) # 可以在每个股票处理完后加一个稍长的延迟 time.sleep(2) if failed_symbols: logger.error(f以下股票处理失败: {failed_symbols}) else: logger.info(所有股票处理完成) if __name__ __main__: main()5.2 定时任务与日志管理对于需要每日更新的场景定时任务必不可少。# scheduler.py import schedule import time from main import main import logging def job(): logging.info(定时爬虫任务开始执行...) try: main() logging.info(定时爬虫任务执行完毕。) except Exception as e: logging.error(f定时任务执行失败: {e}, exc_infoTrue) if __name__ __main__: # 设置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(capital_flow_spider.log), # 输出到文件 logging.StreamHandler() # 输出到控制台 ] ) # 每天下午16:30收盘后执行一次 schedule.every().day.at(16:30).do(job) logging.info(爬虫定时调度器已启动等待执行...) while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次日志管理将日志同时输出到文件和控制台便于后期排查问题。日志文件要设置轮转Rotating避免单个文件过大。6. 常见问题与排查技巧实录在实际运行中你肯定会遇到各种问题。这里记录了几个最典型的“坑”和解决方法。6.1 接口变更或数据为空问题昨天还能跑通的爬虫今天突然抓不到数据了或者返回空列表。排查手动访问页面首先用浏览器打开目标页面确认数据是否正常显示。如果页面都变了那接口大概率也变了。检查网络请求再次打开开发者工具查看之前那个关键的XHR请求是否还在参数是否有变化。重点看type、filter格式、是否有新增的必选参数如token、v版本号。模拟请求用Postman或直接在Python脚本里打印出完整的请求URL和响应内容response.text看看服务器返回了什么错误信息如{“code“: -1, “message“: “参数错误“}。解决根据新的接口文档或抓包结果调整爬虫的请求参数和URL。这是爬虫维护的常态。6.2 IP被封禁或访问频率限制问题请求返回403 Forbidden、429 Too Many Requests或者返回一些提示“操作频繁”的HTML页面。表现最初几次请求成功连续快速请求后开始失败。解决首要措施大幅降低请求频率。这是最有效的方法。在循环中增加time.sleep(random.uniform(3, 7))让请求间隔更随机、更长。使用代理IP池如果需要爬取大量股票或历史数据这是必选项。可以购买付费代理服务或者自建代理池。在Requests中设置代理很简单proxies {“http“: “http://your-proxy:port“, “https“: “https://your-proxy:port“}然后在请求中传入。完善请求头确保User-Agent、Referer、Accept-Language、Cookie如果需要等头部信息与真实浏览器一致。有些网站会校验这些信息。6.3 数据字段映射错误或类型转换异常问题数据能抓到但存入数据库时出错比如字符串无法转为数字日期格式解析失败。排查打印原始数据在清洗数据前打印几行原始的JSON数据查看字段名和值到底是什么样子。东方财富的接口字段名可能是中文也可能是缩写英文。逐步清洗不要试图一步到位。先确保能正确解析JSON为列表再转换为DataFrame然后一步一步进行重命名、类型转换。在每个步骤后打印df.dtypes和df.head()来检查。处理缺失值接口返回的某些字段可能为空字符串““或null。使用pd.to_numeric(..., errors‘coerce‘)可以将无效解析转为NaN避免程序崩溃。6.4 数据库连接与并发写入问题问题在多进程或异步爬虫中同时写入数据库导致锁表、连接超时或重复数据。解决连接池SQLAlchemy的create_engine默认启用了连接池。确保你的数据库连接字符串是全局唯一的并在所有爬虫线程/进程中共享这个引擎。会话管理每个线程或进程使用独立的Session并在任务结束后正确关闭 (session.close())。不要跨线程共享Session。更细粒度的锁如果使用SQLite写入密集型操作建议用WAL(Write-Ahead Logging) 模式可以提高并发性。对于MySQL/PostgreSQL确保使用InnoDB等支持行级锁的引擎。终极方案消息队列对于超大规模爬虫可以将爬取任务和数据写入解耦。爬虫进程只负责抓取将数据扔到Redis或RabbitMQ这样的消息队列中再由独立的消费者进程从队列中取出数据批量写入数据库。这能极大提升系统的稳定性和扩展性。6.5 数据更新逻辑的权衡问题每天跑爬虫是全部重新抓取覆盖还是只抓取新增日期方案全量覆盖逻辑简单代码好写。每天定时任务跑一遍所有历史数据。缺点是浪费带宽和计算资源且对目标网站不友好。增量更新更优雅。先从数据库查询某只股票最新的trade_date然后只请求这个日期之后的数据。这需要爬虫接口支持按日期过滤。优点是高效、友好。推荐此方案。# 增量更新示例片段 def get_latest_date_from_db(symbol, session): result session.query(StockCapitalFlow.trade_date).filter_by(symbolsymbol).order_by(StockCapitalFlow.trade_date.desc()).first() return result[0] if result else None # 在抓取前调用 latest_date get_latest_date_from_db(symbol, session) start_date (latest_date timedelta(days1)).strftime(‘%Y-%m-%d‘) if latest_date else config.START_DATE # 然后用这个 start_date 去构造请求参数爬虫项目从来不是“一劳永逸”的它是一个需要持续观察、维护和调整的系统。核心在于构建一个健壮的框架当数据源发生变化时你能快速定位问题并修复。把数据抓下来只是第一步确保数据管道长期稳定、准确、高效地运行才是真正的价值所在。 
长河君 · 2026/8/5 22:22:53 · 阅读 1.2k Python爬虫实战：抓取东方财富资金流向数据并构建量化分析数据库
1. 项目概述从数据洪流中淘金做量化分析或者基本面研究的朋友手里没点资金流向数据心里总是不踏实。股价的涨跌背后是资金的进进出出在推动。你能看到K线图能看到成交量但更细粒度的——主力资金、散户资金、超大单、大单、中单、小单——这些分档数据才是洞察市场情绪和主力意图的关键。东方财富网的资金流向页面几乎是国内股民和数据分析师的“数据水源地”之一它提供了个股和板块非常详尽的历史资金流向明细。但这个页面是给人看的不是给程序直接吃的。手动一天天去翻、去抄效率低到令人发指而且容易出错。所以用Python写个爬虫把这些结构化的资金流向数据自动抓下来并规整地存入数据库就成了一个非常实际且高频的需求。这不仅仅是“爬数据”更是构建个人量化分析数据底座的起点。有了这个自动化流程你才能腾出手来专注于更重要的策略研究和模型构建。接下来我会带你完整走一遍这个流程。从分析页面结构、设计数据表到编写稳健的爬虫、处理反爬策略最后将数据持久化到数据库。我会分享我在这个过程中踩过的坑和总结的技巧目标是让你拿到一个可以直接运行、易于扩展的解决方案。2. 核心思路与工具选型2.1 目标数据源分析与策略制定东方财富的资金流向数据页面通常URL模式类似http://data.eastmoney.com/zjlx/股票代码.html。例如贵州茅台就是http://data.eastmoney.com/zjlx/600519.html。我们的核心目标是抓取页面中的“历史资金流向”表格数据。策略选择动态渲染 vs. 静态接口首先需要判断页面数据是直接加载在HTML中还是通过JavaScript动态渲染的。这是爬虫设计的第一步方向错了后面全白费。初步探查用浏览器打开页面右键“查看网页源代码”。在源代码里搜索表格里的典型数据比如“净流入”、“主力资金”。如果搜不到基本可以断定是动态加载的。确认动态加载在浏览器开发者工具的“网络”Network选项卡中刷新页面筛选XHR或Fetch请求。你会很快发现一个关键的接口请求其URL通常包含API、GetData等字样返回的是JSON格式的数据。这才是数据的“真身”。接口分析找到这个接口后重点分析它的请求参数。通常包括code股票代码可能带市场前缀如SH600519。type数据类型可能区分日、周、月。page和size分页参数。一些时间戳或加密参数用于反爬。我们的选择直接请求这个后端JSON接口而不是用Selenium或Playwright去模拟浏览器渲染整个页面。理由很充分效率极高省去了加载CSS、JS、图片的时间消耗资源少速度更快也更稳定。难点在于如何找到并模拟这个接口的请求。工具选型背后的逻辑Requests json用于发起HTTP请求和解析JSON数据这是核心。Pandas并非必须但极其推荐。因为接口返回的JSON数据很容易被转换为DataFrame而Pandas在数据清洗、转换如日期格式处理方面是“神器”一行代码能顶十行原生Python操作。SQLAlchemy数据库ORM工具。为什么不用简单的sqlite3模块因为SQLAlchemy提供了数据库无关的抽象层。今天你用SQLite做测试明天想迁移到MySQL或PostgreSQL几乎不需要修改代码。它还能很好地处理连接池、数据类型映射让代码更健壮、更专业。Schedule / APScheduler如果你需要定时任务比如每天收盘后自动爬取这些库能帮你实现可靠的定时调度。2.2 数据库表结构设计在写爬虫之前先想好数据怎么存。一个好的表结构是后续所有分析的基础。资金流向表的核心字段通常包括字段名数据类型说明设计理由idINTEGER PRIMARY KEY自增主键每条记录的唯一标识便于管理和关联。symbolVARCHAR(10)股票代码如600519 必须建立索引因为这是最常用的查询条件。trade_dateDATE交易日期如2023-10-27 与symbol组成唯一约束防止重复插入。close_priceDECIMAL(10,2)收盘价记录当日的价格用于计算和分析。change_pctDECIMAL(6,2)涨跌幅(%)股价变动比例。main_net_inflowDECIMAL(16,2)主力净流入(元)核心指标通常指超大单大单的净额。main_net_inflow_rateDECIMAL(6,2)主力净流入占比(%)净流入占成交额的比例比绝对值更有意义。retail_net_inflowDECIMAL(16,2)散户净流入(元)通常指小单的净额与主力资金形成对照。超大单流入超大单流出DECIMAL(16,2)分档资金流更细粒度的数据有助于深度分析。建议分开存储。turnover_rateDECIMAL(6,2)换手率(%)市场活跃度指标。created_atTIMESTAMP记录创建时间默认CURRENT_TIMESTAMP用于追踪数据获取时间。设计要点与避坑指南唯一约束一定要在(symbol, trade_date)上建立唯一约束UNIQUE CONSTRAINT。这是防止数据重复的“铁闸”。爬虫运行时无论是网络重试还是手动执行都能确保数据库里同一只股票同一天只有一条记录。索引策略除了唯一约束自带的索引如果经常按日期范围查询如“查询某股票2023年所有数据”可以考虑在trade_date上单独建立索引。但索引不是越多越好会影响写入速度。金额字段精度资金数据可能很大使用DECIMAL(16,2)可以安全存储千亿级别的金额并保留两位小数。DECIMAL类型能保证精确计算避免浮点数误差。日期类型务必使用数据库的DATE类型存储trade_date而不是VARCHAR。这能让你直接使用数据库的日期函数进行高效的区间查询和聚合。3. 爬虫核心实现与反爬对抗3.1 逆向分析接口与参数构造这是最具技术挑战性的一步。以东方财富为例我们打开开发者工具定位到那个返回历史资金流JSON数据的请求。假设我们找到的接口是http://datacenter.eastmoney.com/api/data/get?typeRPTA_WEB_RZRQ_GGMXstyALLsourceWEB...这只是一个示例实际接口可能会变。关键步骤复制为cURL在Network中找到该请求右键选择“Copy” - “Copy as cURL (bash)”。这将给你一个完整的命令行请求。转换与分析将cURL命令粘贴到在线工具如 https://curlconverter.com/python/中可以将其转换为Python的Requests代码。这能快速得到完整的请求头headers和参数params。参数解构仔细研究这些参数。你会发现一些固定参数如type、sty、source。以及一些动态参数code可能是SH600519或SZ000001这种格式。page页码。size每页条数可能是20、50、100。filter可能是一个包含查询条件的字符串如(TRADE_DATE‘2022-01-01’)。最关键的t或_后面跟的一串数字这通常是一个当前时间戳用于防止缓存。这是最常见的反爬手段之一。请求头重点关注User-Agent、Referer、Accept。Referer通常需要设置为资金流向页面的URL这是服务器验证请求来源的常用方法。实操心得东方财富的接口参数命名可能不那么直观filter参数里的表达式语法需要仔细从接口示例中揣摩。一个技巧是先通过浏览器正常访问抓取第一页的请求然后尝试只修改page和size参数来获取更多数据。如果失败再检查是否有其他动态生成的令牌token。3.2 稳健的爬虫代码编写下面是一个融合了关键技巧的爬虫核心函数示例import requests import pandas as pd import time from datetime import datetime, timedelta import json def fetch_stock_capital_flow(symbol, start_dateNone, end_dateNone): 获取单只股票的历史资金流向数据 Args: symbol (str): 股票代码如 600519 start_date (str): 开始日期格式 2023-01-01默认为一年前 end_date (str): 结束日期格式 2023-10-27默认为今天 Returns: pandas.DataFrame: 资金流向数据列名已清洗 # 1. 构造请求头Headers - 模拟浏览器 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36, Referer: fhttp://data.eastmoney.com/zjlx/{symbol}.html, # 关键Referer必须正确 Accept: application/json, text/plain, */*, Accept-Language: zh-CN,zh;q0.9,en;q0.8, } # 2. 构造请求参数Params # 这里需要根据实际找到的接口调整参数名和值 params { type: RPTA_WEB_RZRQ_GGMX, # 示例类型需替换 sty: ALL, source: WEB, code: fSH{symbol} if symbol.startswith((6, 5, 9)) else fSZ{symbol}, # 判断市场 page: 1, size: 100, # 每页条数可调大以减少请求次数 filter: f(TRADE_DATE{start_date}) if start_date else None, t: int(time.time() * 1000), # 关键时间戳防缓存 } # 清理空参数 params {k: v for k, v in params.items() if v is not None} all_data [] max_pages 10 # 安全限制防止意外死循环 for page in range(1, max_pages 1): params[page] page print(f正在抓取 {symbol} 第 {page} 页...) try: # 3. 发送请求 # 接口URL需替换为实际地址 response requests.get( http://datacenter.eastmoney.com/api/data/get, paramsparams, headersheaders, timeout15 # 设置超时避免长时间等待 ) response.raise_for_status() # 如果状态码不是200抛出异常 # 4. 解析JSON json_data response.json() # 东方财富的接口数据通常嵌套在 result - data 路径下 data_list json_data.get(result, {}).get(data, []) if not data_list: print(f第 {page} 页无数据停止爬取。) break all_data.extend(data_list) # 5. 礼貌性延迟降低请求频率 time.sleep(1.5) # 重要避免请求过快被封IP except requests.exceptions.RequestException as e: print(f请求第 {page} 页时发生错误: {e}) break except json.JSONDecodeError as e: print(f解析第 {page} 页JSON时发生错误: {e}) print(f响应文本: {response.text[:500]}) # 打印前500字符辅助调试 break # 6. 转换为DataFrame并清洗 if not all_data: return pd.DataFrame() df pd.DataFrame(all_data) # 重命名列使其更易懂 column_mapping { SECURITY_CODE: symbol, TRADE_DATE: trade_date, CLOSE_PRICE: close_price, CHANGE_PCT: change_pct, MAIN_NET_INFLOW: main_net_inflow, MAIN_NET_INFLOW_RATE: main_net_inflow_rate, # ... 其他字段映射 } df.rename(columnscolumn_mapping, inplaceTrue) # 转换日期格式 if trade_date in df.columns: df[trade_date] pd.to_datetime(df[trade_date]).dt.date # 转换数值类型 numeric_columns [close_price, change_pct, main_net_inflow, main_net_inflow_rate] for col in numeric_columns: if col in df.columns: df[col] pd.to_numeric(df[col], errorscoerce) # 错误值转为NaN return df注意事项错误处理网络请求必须包含try...except并处理超时、状态码异常、JSON解析失败等情况。爬虫要能“优雅地失败”而不是整个崩溃。延迟Sleeptime.sleep(random.uniform(1, 3))比固定延迟更好模拟人类操作的不确定性。这是对目标网站最基本的尊重也是保护自己IP不被封禁的有效手段。User-Agent轮换如果需要大规模爬取准备一个User-Agent列表进行轮换可以进一步降低被识别风险。4. 数据持久化写入数据库数据抓取到Pandas DataFrame后下一步就是高效、安全地存入数据库。这里我们使用SQLAlchemy配合SQLite本地开发测试首选或MySQL/PostgreSQL。4.1 使用SQLAlchemy建立连接与映射from sqlalchemy import create_engine, Column, Integer, String, Date, DECIMAL, TIMESTAMP, UniqueConstraint from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime # 1. 定义基类 Base declarative_base() # 2. 定义数据表模型ORM类 class StockCapitalFlow(Base): __tablename__ stock_capital_flow id Column(Integer, primary_keyTrue, autoincrementTrue) symbol Column(String(10), nullableFalse, indexTrue) # 建立索引 trade_date Column(Date, nullableFalse) close_price Column(DECIMAL(10, 2)) change_pct Column(DECIMAL(6, 2)) main_net_inflow Column(DECIMAL(16, 2)) main_net_inflow_rate Column(DECIMAL(6, 2)) retail_net_inflow Column(DECIMAL(16, 2)) # ... 其他字段 created_at Column(TIMESTAMP, defaultdatetime.now) # 3. 定义唯一约束防止重复数据 __table_args__ (UniqueConstraint(symbol, trade_date, nameuix_symbol_trade_date),) # 4. 创建数据库连接引擎 # 使用SQLite文件数据库适合本地测试 # engine create_engine(sqlite:///./stock_data.db, echoFalse) # echoTrue 可查看SQL日志 # 使用MySQL生产环境推荐 # 格式mysqlpymysql://用户名:密码主机:端口/数据库名 engine create_engine(mysqlpymysql://user:passwordlocalhost:3306/stock_db?charsetutf8mb4) # 5. 创建表如果不存在 Base.metadata.create_all(engine) # 6. 创建会话工厂 SessionLocal sessionmaker(bindengine)4.2 高效批量写入与去重策略直接将DataFrame一行行插入效率极低。Pandas的to_sql方法虽然方便但默认的插入方式if_existsappend无法处理唯一约束冲突会导致整个插入失败。我们需要更精细的控制。推荐方案使用SQLAlchemy Core进行批量“upsert”“Upsert”是Update和Insert的合成词即存在则更新不存在则插入。MySQL 5.7和PostgreSQL都支持ON DUPLICATE KEY UPDATE语法SQLite也有类似的INSERT OR REPLACE/IGNORE。def upsert_capital_flow_to_db(df, engine): 将DataFrame数据高效写入数据库并处理重复数据。 Args: df (pd.DataFrame): 清洗后的资金流向数据 engine: SQLAlchemy引擎 if df.empty: print(DataFrame为空跳过写入。) return # 确保列名与数据库模型一致 # 假设df的列名已经是我们定义好的英文名 # 方法使用Pandas SQLAlchemy Core执行原生SQL进行批量upsert # 这里以MySQL为例 table_name stock_capital_flow temp_table_name ftemp_{table_name} # 创建一个临时表用于暂存本次抓取的数据 with engine.begin() as conn: # 使用事务 # 1. 将DataFrame写入临时表 df.to_sql(temp_table_name, conn, if_existsreplace, indexFalse) # 2. 执行Upsert操作 # 核心SQL语句将临时表的数据合并到主表冲突时更新某些字段 upsert_sql f INSERT INTO {table_name} (symbol, trade_date, close_price, change_pct, main_net_inflow, main_net_inflow_rate, ...) SELECT symbol, trade_date, close_price, change_pct, main_net_inflow, main_net_inflow_rate, ... FROM {temp_table_name} ON DUPLICATE KEY UPDATE close_price VALUES(close_price), change_pct VALUES(change_pct), main_net_inflow VALUES(main_net_inflow), main_net_inflow_rate VALUES(main_net_inflow_rate), -- ... 更新其他需要覆盖的字段created_at 通常不更新 updated_at CURRENT_TIMESTAMP; -- 可以加一个更新时间的字段 conn.execute(text(upsert_sql)) # 3. 删除临时表 conn.execute(text(fDROP TABLE IF EXISTS {temp_table_name})) print(f成功写入/更新 {len(df)} 条记录到数据库。)实操心得批量Upsert是生产级爬虫数据入库的关键。直接循环插入每秒可能只能处理几十条而批量Upsert每秒可以处理上万条。临时表的思路避免了在Python内存中拼接庞大的SQL语句性能和安全性都更好。记住对于像收盘价、资金流这种客观历史数据一旦写入即使重复抓取也应该用新数据覆盖旧数据如果接口数据有修正的话。created_at记录首次抓取时间updated_at记录最后一次更新时间这个设计很实用。5. 项目整合与调度实战5.1 构建可配置的主程序将上述模块组装起来并增加一些工程化配置。# config.py import os from dataclasses import dataclass dataclass class Config: # 数据库连接字符串 DATABASE_URL os.getenv(DATABASE_URL, sqlite:///./stock_data.db) # 请求基础延迟秒 BASE_DELAY 1.5 # 每页数据大小 PAGE_SIZE 100 # 要爬取的股票代码列表 STOCK_LIST [600519, 000858, 300750, 002594] # 示例 # 数据抓取时间范围 START_DATE 2023-01-01 END_DATE 2023-10-27 # main.py import logging from config import Config from capital_flow_spider import fetch_stock_capital_flow from database import upsert_capital_flow_to_db, engine import time logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) def main(): config Config() failed_symbols [] for symbol in config.STOCK_LIST: logger.info(f开始处理股票: {symbol}) try: # 抓取数据 df fetch_stock_capital_flow( symbolsymbol, start_dateconfig.START_DATE, end_dateconfig.END_DATE ) if df.empty: logger.warning(f股票 {symbol} 未获取到数据。) continue logger.info(f股票 {symbol} 抓取到 {len(df)} 条记录。) # 写入数据库 upsert_capital_flow_to_db(df, engine) # 随机延迟避免对服务器造成压力 time.sleep(config.BASE_DELAY) except Exception as e: logger.error(f处理股票 {symbol} 时发生错误: {e}, exc_infoTrue) failed_symbols.append(symbol) # 可以在每个股票处理完后加一个稍长的延迟 time.sleep(2) if failed_symbols: logger.error(f以下股票处理失败: {failed_symbols}) else: logger.info(所有股票处理完成) if __name__ __main__: main()5.2 定时任务与日志管理对于需要每日更新的场景定时任务必不可少。# scheduler.py import schedule import time from main import main import logging def job(): logging.info(定时爬虫任务开始执行...) try: main() logging.info(定时爬虫任务执行完毕。) except Exception as e: logging.error(f定时任务执行失败: {e}, exc_infoTrue) if __name__ __main__: # 设置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(capital_flow_spider.log), # 输出到文件 logging.StreamHandler() # 输出到控制台 ] ) # 每天下午16:30收盘后执行一次 schedule.every().day.at(16:30).do(job) logging.info(爬虫定时调度器已启动等待执行...) while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次日志管理将日志同时输出到文件和控制台便于后期排查问题。日志文件要设置轮转Rotating避免单个文件过大。6. 常见问题与排查技巧实录在实际运行中你肯定会遇到各种问题。这里记录了几个最典型的“坑”和解决方法。6.1 接口变更或数据为空问题昨天还能跑通的爬虫今天突然抓不到数据了或者返回空列表。排查手动访问页面首先用浏览器打开目标页面确认数据是否正常显示。如果页面都变了那接口大概率也变了。检查网络请求再次打开开发者工具查看之前那个关键的XHR请求是否还在参数是否有变化。重点看type、filter格式、是否有新增的必选参数如token、v版本号。模拟请求用Postman或直接在Python脚本里打印出完整的请求URL和响应内容response.text看看服务器返回了什么错误信息如{“code“: -1, “message“: “参数错误“}。解决根据新的接口文档或抓包结果调整爬虫的请求参数和URL。这是爬虫维护的常态。6.2 IP被封禁或访问频率限制问题请求返回403 Forbidden、429 Too Many Requests或者返回一些提示“操作频繁”的HTML页面。表现最初几次请求成功连续快速请求后开始失败。解决首要措施大幅降低请求频率。这是最有效的方法。在循环中增加time.sleep(random.uniform(3, 7))让请求间隔更随机、更长。使用代理IP池如果需要爬取大量股票或历史数据这是必选项。可以购买付费代理服务或者自建代理池。在Requests中设置代理很简单proxies {“http“: “http://your-proxy:port“, “https“: “https://your-proxy:port“}然后在请求中传入。完善请求头确保User-Agent、Referer、Accept-Language、Cookie如果需要等头部信息与真实浏览器一致。有些网站会校验这些信息。6.3 数据字段映射错误或类型转换异常问题数据能抓到但存入数据库时出错比如字符串无法转为数字日期格式解析失败。排查打印原始数据在清洗数据前打印几行原始的JSON数据查看字段名和值到底是什么样子。东方财富的接口字段名可能是中文也可能是缩写英文。逐步清洗不要试图一步到位。先确保能正确解析JSON为列表再转换为DataFrame然后一步一步进行重命名、类型转换。在每个步骤后打印df.dtypes和df.head()来检查。处理缺失值接口返回的某些字段可能为空字符串““或null。使用pd.to_numeric(..., errors‘coerce‘)可以将无效解析转为NaN避免程序崩溃。6.4 数据库连接与并发写入问题问题在多进程或异步爬虫中同时写入数据库导致锁表、连接超时或重复数据。解决连接池SQLAlchemy的create_engine默认启用了连接池。确保你的数据库连接字符串是全局唯一的并在所有爬虫线程/进程中共享这个引擎。会话管理每个线程或进程使用独立的Session并在任务结束后正确关闭 (session.close())。不要跨线程共享Session。更细粒度的锁如果使用SQLite写入密集型操作建议用WAL(Write-Ahead Logging) 模式可以提高并发性。对于MySQL/PostgreSQL确保使用InnoDB等支持行级锁的引擎。终极方案消息队列对于超大规模爬虫可以将爬取任务和数据写入解耦。爬虫进程只负责抓取将数据扔到Redis或RabbitMQ这样的消息队列中再由独立的消费者进程从队列中取出数据批量写入数据库。这能极大提升系统的稳定性和扩展性。6.5 数据更新逻辑的权衡问题每天跑爬虫是全部重新抓取覆盖还是只抓取新增日期方案全量覆盖逻辑简单代码好写。每天定时任务跑一遍所有历史数据。缺点是浪费带宽和计算资源且对目标网站不友好。增量更新更优雅。先从数据库查询某只股票最新的trade_date然后只请求这个日期之后的数据。这需要爬虫接口支持按日期过滤。优点是高效、友好。推荐此方案。# 增量更新示例片段 def get_latest_date_from_db(symbol, session): result session.query(StockCapitalFlow.trade_date).filter_by(symbolsymbol).order_by(StockCapitalFlow.trade_date.desc()).first() return result[0] if result else None # 在抓取前调用 latest_date get_latest_date_from_db(symbol, session) start_date (latest_date timedelta(days1)).strftime(‘%Y-%m-%d‘) if latest_date else config.START_DATE # 然后用这个 start_date 去构造请求参数爬虫项目从来不是“一劳永逸”的它是一个需要持续观察、维护和调整的系统。核心在于构建一个健壮的框架当数据源发生变化时你能快速定位问题并修复。把数据抓下来只是第一步确保数据管道长期稳定、准确、高效地运行才是真正的价值所在。