钛媒体 Python 爬虫实战构建科技股概念与板块分析数据引擎一、项目背景与目标在量化投资领域概念股和板块轮动分析是短线交易者最关注的数据之一。钛媒体等财经媒体每天发布大量关于科技股概念如 AI、半导体、新能源、低空经济等的分析文章但这些信息分散在网页中人工收集效率极低。本项目目标是构建一个自动化爬虫引擎从钛媒体等财经网站采集科技股概念与板块分析文章提取关键信息概念名称、相关股票、分析师观点、时间维度存入本地数据库为后续数据分析提供结构化数据基础。## 二、技术选型与环境搭建### 2.1 核心依赖库| 库名 | 版本 | 用途 ||------|------|------|| requests | 2.32.3 | 发起 HTTP 请求获取网页源码 || BeautifulSoup4 | 4.12.3 | 解析 HTML 文档提取结构化数据 || lxml | 5.3.0 | 作为 BS4 的解析引擎速度快 || pandas | 2.2.2 | 数据清洗与表格化处理 || sqlite3 | 内置 | 轻量级本地数据库 || fake_useragent | 1.5.1 | 随机生成 User-Agent反爬 || retrying | 1.3.4 | 实现自动重试机制 || loguru | 0.7.3 | 优雅的日志记录 || json | 内置 | 处理 API 响应数据 |### 2.2 环境配置bash# 创建虚拟环境python -m venv stock_crawler_env# Windowsstock_crawler_env\Scripts\activate# Linux/Macsource stock_crawler_env/bin/activate# 安装依赖pip install requests beautifulsoup4 lxml pandas fake_useragent retrying loguru## 三、网站结构深度剖析逆向工程### 3.1 列表页 URL 模式钛媒体的文章列表页遵循以下模式https://www.tmtpost.com/stock/{category}?page{page_num}通过 Chrome 开发者工具分析发现列表页采用服务端渲染SSR数据直接嵌入在 HTML 中无需额外 API 调用。### 3.2 文章详情页结构详情页的 DOM 结构如下htmlarticle classpost-content h1 classpost-titleAI概念股全线爆发算力板块领涨/h1 div classpost-meta span classdate2026-08-10/span span classauthor钛媒体/span span classtags a href/tag/aiAI/a a href/tag/算力算力/a /span /div div classpost-body !-- 文章正文内容 -- /div div classstock-list !-- 提及的相关股票 -- span classstock-item中科曙光(603019.SH)/span span classstock-item浪潮信息(000977.SZ)/span /div/article### 3.3 反爬策略清单通过分析钛媒体采用了以下反爬措施1.User-Agent 检测拒绝非浏览器请求2.请求频率限制同一 IP 每秒超过 5 次请求会被封禁3.Cookie 有效性验证部分页面需要会话 Cookie4.动态渲染部分内容通过 JavaScript 加载5.验证码高频请求触发滑动验证码## 四、完整爬虫代码实现分模块详解### 4.1 配置模块config.pypythonimport osfrom pathlib import Path# 项目根目录ROOT_DIR Path(__file__).parent# 爬虫配置CRAWLER_CONFIG { base_url: https://www.tmtpost.com, categories: [ai, semiconductor, new-energy, low-altitude], page_start: 1, page_end: 10, delay: 3, # 请求间隔秒 timeout: 15, # 超时时间秒 max_retries: 3, user_agent_rotation: True,}# 数据库配置DB_CONFIG { db_path: ROOT_DIR / data / stock_news.db,}# 日志配置LOG_CONFIG { log_file: ROOT_DIR / logs / crawler.log, log_level: DEBUG, rotation: 500 MB, retention: 30 days,}# 代理配置可选PROXY_CONFIG { enabled: False, proxies: [ http://proxy1.example.com:8080, http://proxy2.example.com:8080, ],}### 4.2 日志与工具函数utils.pypythonimport randomimport timefrom functools import wrapsfrom fake_useragent import UserAgentfrom loguru import loggerfrom config import CRAWLER_CONFIG, LOG_CONFIG# 配置日志logger.add( LOG_CONFIG[log_file], rotationLOG_CONFIG[rotation], retentionLOG_CONFIG[retention], levelLOG_CONFIG[log_level], format{time:YYYY-MM-DD HH:mm:ss} | {level} | {message},)# User-Agent 管理ua UserAgent()def get_random_headers(): 生成随机请求头 return { User-Agent: ua.random if CRAWLER_CONFIG[user_agent_rotation] else ua.chrome, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, Referer: https://www.tmtpost.com/, }def retry_on_failure(max_retriesNone): 重试装饰器 if max_retries is None: max_retries CRAWLER_CONFIG[max_retries] def decorator(func): wraps(func) def wrapper(*args, **kwargs): for attempt in range(1, max_retries 1): try: return func(*args, **kwargs) except Exception as e: logger.warning(f{func.__name__} 第 {attempt} 次失败: {e}) if attempt max_retries: logger.error(f{func.__name__} 重试 {max_retries} 次均失败) raise time.sleep(2 ** attempt) # 指数退避 return None return wrapper return decoratordef random_delay(): 随机延迟模拟人类行为 delay CRAWLER_CONFIG[delay] random.uniform(0.5, 2.0) time.sleep(delay)### 4.3 数据库操作模块db.pypythonimport sqlite3import jsonfrom datetime import datetimefrom contextlib import contextmanagerfrom config import DB_CONFIGfrom loguru import loggerclass DatabaseManager: 数据库管理器 def __init__(self, db_pathNone): self.db_path db_path or DB_CONFIG[db_path] self.db_path.parent.mkdir(parentsTrue, exist_okTrue) self._init_tables() contextmanager def get_connection(self): 获取数据库连接上下文管理器 conn sqlite3.connect(str(self.db_path)) conn.row_factory sqlite3.Row try: yield conn conn.commit() except Exception as e: conn.rollback() logger.error(f数据库操作失败: {e}) raise finally: conn.close() def _init_tables(self): 初始化数据库表结构 with self.get_connection() as conn: cursor conn.cursor() # 文章表 cursor.execute( CREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT UNIQUE NOT NULL, title TEXT NOT NULL, author TEXT, publish_date TEXT, category TEXT, content TEXT, summary TEXT, tags TEXT, stock_mentions TEXT, created_at TEXT DEFAULT (datetime(now, localtime)), updated_at TEXT DEFAULT (datetime(now, localtime)) ) ) # 股票提及表 cursor.execute( CREATE TABLE IF NOT EXISTS stock_mentions ( id INTEGER PRIMARY KEY AUTOINCREMENT, article_id INTEGER, stock_code TEXT, stock_name TEXT, mention_context TEXT, sentiment TEXT, FOREIGN KEY (article_id) REFERENCES articles(id) ) ) # 索引 cursor.execute( CREATE INDEX IF NOT EXISTS idx_articles_date ON articles(publish_date) ) cursor.execute( CREATE INDEX IF NOT EXISTS idx_articles_category ON articles(category) ) logger.info(数据库表初始化完成) def save_article(self, article_data): 保存文章及其相关的股票信息 with self.get_connection() as conn: cursor conn.cursor() # 提取股票信息 stock_mentions article_data.pop(stock_mentions, []) tags article_data.pop(tags, []) # 插入文章 article_data[tags] json.dumps(tags, ensure_asciiFalse) article_data[stock_mentions] json.dumps( [s[code] for s in stock_mentions], ensure_asciiFalse ) try: cursor.execute( INSERT OR IGNORE INTO articles (url, title, author, publish_date, category, content, summary, tags, stock_mentions) VALUES (:url, :title, :author, :publish_date, :category, :content, :summary, :tags, :stock_mentions) , article_data) article_id cursor.lastrowid # 插入股票提及 if article_id and stock_mentions: for stock in stock_mentions: cursor.execute( INSERT INTO stock_mentions (article_id, stock_code, stock_name, mention_context, sentiment) VALUES (?, ?, ?, ?, ?) , ( article_id, stock[code], stock[name], stock.get(context, ), stock.get(sentiment, neutral), )) logger.info(f文章保存成功: {article_data[title][:30]}...) return article_id except Exception as e: logger.error(f文章保存失败: {e}) return None def query_articles(self, categoryNone, date_fromNone, date_toNone, stock_codeNone, limit100): 查询文章 conditions [] params {} if category: conditions.append(category :category) params[category] category if date_from: conditions.append(publish_date :date_from) params[date_from] date_from if date_to: conditions.append(publish_date :date_to) params[date_to] date_to if stock_code: conditions.append(stock_mentions LIKE :stock_code) params[stock_code] f%{stock_code}% where_clause AND .join(conditions) if conditions else 11 with self.get_connection() as conn: cursor conn.cursor() cursor.execute(f SELECT * FROM articles WHERE {where_clause} ORDER BY publish_date DESC LIMIT :limit , {**params, limit: limit}) return [dict(row) for row in cursor.fetchall()]### 4.4 核心爬虫类crawler.pypythonimport reimport requestsfrom bs4 import BeautifulSoupfrom datetime import datetimefrom loguru import loggerfrom utils import get_random_headers, retry_on_failure, random_delayfrom config import CRAWLER_CONFIGclass TMTPostCrawler: 钛媒体科技股爬虫 def __init__(self): self.session requests.Session() self.base_url CRAWLER_CONFIG[base_url] self.timeout CRAWLER_CONFIG[timeout] retry_on_failure() def fetch_page(self, url): 获取页面内容 headers get_random_headers() response self.session.get( url, headersheaders, timeoutself.timeout ) response.raise_for_status() response.encoding utf-8 return response.text def parse_article_list(self, html, category): 解析文章列表页 soup BeautifulSoup(html, lxml) articles [] for item in soup.select(.post-item, .article-item): try: title_elem item.select_one(.post-title a, .article-title a) if not title_elem: continue url title_elem.get(href, ) if url and not url.startswith(http): url self.base_url url article { url: url, title: title_elem.get_text(stripTrue), category: category, summary: , publish_date: , } # 提取摘要 summary_elem item.select_one(.post-summary, .article-summary) if summary_elem: article[summary] summary_elem.get_text(stripTrue) # 提取日期 date_elem item.select_one(.post-date, .article-date, time) if date_elem: article[publish_date] date_elem.get(datetime, ) or \ date_elem.get_text(stripTrue) articles.append(article) except Exception as e: logger.warning(f解析文章项失败: {e}) continue return articles def parse_article_detail(self, html, article_meta): 解析文章详情页 soup BeautifulSoup(html, lxml) # 提取正文 content_elem soup.select_one(.post-content, .article-content, .rich-content) content content_elem.get_text(stripTrue) if content_elem else # 提取标签 tags [] tag_elems soup.select(.tag-item a, .post-tags a, .tags a) for tag in tag_elems: tags.append(tag.get_text(stripTrue)) # 提取作者 author_elem soup.select_one(.author-name, .post-author) author author_elem.get_text(stripTrue) if author_elem else 钛媒体 # 提取股票提及 stock_mentions [] stock_pattern re.compile( r([\u4e00-\u9fa5])\((\d{6}\.(?:SH|SZ|BJ))\) ) # 从正文中提取股票代码 for match in stock_pattern.finditer(content): stock_mentions.append({ name: match.group(1), code: match.group(2), context: content[ max(0, match.start()-50):match.end()50 ], sentiment: neutral, # 可后续通过 NLP 分析 }) return { **article_meta, content: content, author: author, tags: tags, stock_mentions: stock_mentions, } def crawl_category(self, category, pages5): 爬取一个分类下的文章 all_articles [] for page in range(1, pages 1): url f{self.base_url}/stock/{category}?page{page} logger.info(f爬取列表页: {url}) try: html self.fetch_page(url) articles self.parse_article_list(html, category) for article in articles: logger.info(f爬取详情: {article[title][:30]}...) detail_html self.fetch_page(article[url]) full_article self.parse_article_detail(detail_html, article) all_articles.append(full_article) random_delay() logger.info(f分类 {category} 第 {page} 页完成: {len(articles)} 篇) except Exception as e: logger.error(f爬取分类 {category} 第 {page} 页失败: {e}) continue return all_articles def run(self, categoriesNone, pages5): 运行爬虫 categories categories or CRAWLER_CONFIG[categories] all_results [] for category in categories: logger.info(f开始爬取分类: {category}) articles self.crawl_category(category, pages) all_results.extend(articles) logger.info(f分类 {category} 完成: {len(articles)} 篇) logger.info(f爬虫完成共获取 {len(all_results)} 篇文章) return all_results### 4.5 主启动脚本main.pypythonimport sysimport argparsefrom pathlib import Pathfrom datetime import datetimefrom loguru import loggerfrom crawler import TMTPostCrawlerfrom db import DatabaseManagerfrom config import CRAWLER_CONFIGdef setup_directories(): 创建必要的目录 Path(data).mkdir(exist_okTrue) Path(logs).mkdir(exist_okTrue) Path(exports).mkdir(exist_okTrue)def export_to_csv(articles, filenameNone): 导出为 CSV 格式 import pandas as pd if not filename: filename fexports/stock_news_{datetime.now().strftime(%Y%m%d)}.csv df pd.DataFrame(articles) df.to_csv(filename, indexFalse, encodingutf-8-sig) logger.info(f数据已导出到 {filename}) return filenamedef main(): parser argparse.ArgumentParser( description钛媒体科技股爬虫 - 构建概念与板块分析数据引擎 ) parser.add_argument( --categories, nargs, defaultCRAWLER_CONFIG[categories], help要爬取的分类列表 ) parser.add_argument( --pages, typeint, default5, help每个分类爬取的页数 ) parser.add_argument( --export, actionstore_true, help导出为 CSV 文件 ) parser.add_argument( --no-db, actionstore_true, help不存入数据库 ) args parser.parse_args() setup_directories() # 初始化爬虫 crawler TMTPostCrawler() articles crawler.run( categoriesargs.categories, pagesargs.pages, ) if not articles: logger.warning(未获取到任何文章) return # 存入数据库 if not args.no_db: db DatabaseManager() saved_count 0 for article in articles: if db.save_article(article): saved_count 1 logger.info(f数据库保存成功: {saved_count}/{len(articles)} 篇) # 导出 CSV if args.export: export_to_csv(articles) logger.info(任务完成)if __name__ __main__: main()## 五、运行结果与数据样例### 5.1 运行命令bash# 爬取 AI 和半导体分类各 3 页python main.py --categories ai semiconductor --pages 3 --export# 爬取所有分类python main.py --pages 5 --export### 5.2 数据样例json{ title: AI概念股全线爆发算力板块领涨, url: https://www.tmtpost.com/stock/ai/12345, author: 钛媒体, publish_date: 2026-08-10, category: ai, summary: 今日AI概念股集体爆发算力板块涨幅居前中科曙光涨停..., tags: [AI, 算力, 概念股], stock_mentions: [ {code: 603019.SH, name: 中科曙光}, {code: 000977.SZ, name: 浪潮信息}, {code: 300308.SZ, name: 中际旭创} ]}## 六、反爬升级应对策略与优化方向### 6.1 当前遇到的挑战1.动态内容加载部分页面使用 JavaScript 渲染需使用 Playwright 或 Selenium2.验证码高频请求触发滑动验证码3.IP 封禁同一 IP 请求过多会被临时封禁### 6.2 解决方案python# 使用 Playwright 处理动态页面from playwright.sync_api import sync_playwrightdef fetch_with_browser(url): with sync_playwright() as p: browser p.chromium.launch(headlessTrue) page browser.new_page() page.goto(url, wait_untilnetworkidle) content page.content() browser.close() return content### 6.3 未来扩展建议1.分布式爬虫使用 Scrapy Redis 实现分布式爬取2.NLP 情感分析对提取的股票提及进行情感分析正面/负面/中性3.定时任务配置定时任务每天自动增量更新4.数据可视化构建 Web 仪表盘展示概念热度和板块轮动趋势## 七、总结本文从零开始构建了一个完整的财经爬虫系统覆盖了从网站分析、代码实现到数据存储的完整流程。通过这个项目你可以1. 掌握 Python 爬虫的完整开发流程2. 理解反爬策略和应对方案3. 学会使用数据库存储结构化数据4. 为量化分析提供数据基础完整代码已开源你可以直接下载运行也可以根据需求扩展更多数据源和功能。