钛媒体 Python 爬虫实战：构建科技股概念与板块分析数据引擎

钛媒体 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 结构如下：html<article class="post-content"> <h1 class="post-title">AI概念股全线爆发，算力板块领涨</h1> <div class="post-meta"> <span class="date">2026-08-10</span> <span class="author">钛媒体</span> <span class="tags"> <a href="/tag/ai">AI</a> <a href="/tag/算力">算力</a> </span> </div> <div class="post-body"> <!– 文章正文内容 –> </div> <div class="stock-list"> <!– 提及的相关股票 –> <span class="stock-item">中科曙光(603019.SH)</span> <span class="stock-item">浪潮信息(000977.SZ)</span> </div></article>### 3.3 反爬策略清单通过分析，钛媒体采用了以下反爬措施：1. User-Agent 检测：拒绝非浏览器请求2. 请求频率限制：同一 IP 每秒超过 5 次请求会被封禁3. Cookie 有效性验证：部分页面需要会话 Cookie4. 动态渲染：部分内容通过 JavaScript 加载5. 验证码：高频请求触发滑动验证码## 四、完整爬虫代码实现（分模块详解）### 4.1 配置模块（config.py）pythonimport 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.py）pythonimport randomimport timefrom functools import wrapsfrom fake_useragent import UserAgentfrom loguru import loggerfrom config import CRAWLER_CONFIG, LOG_CONFIG# 配置日志logger.add( LOG_CONFIG["log_file"], rotation=LOG_CONFIG["rotation"], retention=LOG_CONFIG["retention"], level=LOG_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/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive", "Referer": "https://www.tmtpost.com/", }def retry_on_failure(max_retries=None): """重试装饰器""" 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.py）pythonimport sqlite3import jsonfrom datetime import datetimefrom contextlib import contextmanagerfrom config import DB_CONFIGfrom loguru import loggerclass DatabaseManager: """数据库管理器""" def __init__(self, db_path=None): self.db_path = db_path or DB_CONFIG["db_path"] self.db_path.parent.mkdir(parents=True, exist_ok=True) 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_ascii=False) article_data["stock_mentions"] = json.dumps( [s["code"] for s in stock_mentions], ensure_ascii=False ) 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, category=None, date_from=None, date_to=None, stock_code=None, limit=100): """查询文章""" 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 "1=1" 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.py）pythonimport 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, headers=headers, timeout=self.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(strip=True), 'category': category, 'summary': '', 'publish_date': '', } # 提取摘要 summary_elem = item.select_one('.post-summary, .article-summary') if summary_elem: article['summary'] = summary_elem.get_text(strip=True) # 提取日期 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(strip=True) 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(strip=True) 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(strip=True)) # 提取作者 author_elem = soup.select_one('.author-name, .post-author') author = author_elem.get_text(strip=True) 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, pages=5): """爬取一个分类下的文章""" 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, categories=None, pages=5): """运行爬虫""" 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.py）pythonimport 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_ok=True) Path("logs").mkdir(exist_ok=True) Path("exports").mkdir(exist_ok=True)def export_to_csv(articles, filename=None): """导出为 CSV 格式""" import pandas as pd if not filename: filename = f"exports/stock_news_{datetime.now().strftime('%Y%m%d')}.csv" df = pd.DataFrame(articles) df.to_csv(filename, index=False, encoding='utf-8-sig') logger.info(f"数据已导出到 {filename}") return filenamedef main(): parser = argparse.ArgumentParser( description="钛媒体科技股爬虫 – 构建概念与板块分析数据引擎" ) parser.add_argument( "–categories", nargs="+", default=CRAWLER_CONFIG["categories"], help="要爬取的分类列表" ) parser.add_argument( "–pages", type=int, default=5, help="每个分类爬取的页数" ) parser.add_argument( "–export", action="store_true", help="导出为 CSV 文件" ) parser.add_argument( "–no-db", action="store_true", help="不存入数据库" ) args = parser.parse_args() setup_directories() # 初始化爬虫 crawler = TMTPostCrawler() articles = crawler.run( categories=args.categories, pages=args.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(headless=True) page = browser.new_page() page.goto(url, wait_until="networkidle") content = page.content() browser.close() return content### 6.3 未来扩展建议1. 分布式爬虫：使用 Scrapy + Redis 实现分布式爬取2. NLP 情感分析：对提取的股票提及进行情感分析（正面/负面/中性）3. 定时任务：配置定时任务，每天自动增量更新4. 数据可视化：构建 Web 仪表盘，展示概念热度和板块轮动趋势## 七、总结本文从零开始构建了一个完整的财经爬虫系统，覆盖了从网站分析、代码实现到数据存储的完整流程。通过这个项目，你可以：1. 掌握 Python 爬虫的完整开发流程2. 理解反爬策略和应对方案3. 学会使用数据库存储结构化数据4. 为量化分析提供数据基础完整代码已开源，你可以直接下载运行，也可以根据需求扩展更多数据源和功能。