☰
生产级金融数据采集系统:轻量闭环架构与边缘部署实践
2026/10/24 12:47:21 网站建设 项目流程

简介:本资源是一份面向Python初学者与爬虫入门者的系统性实践指南,聚焦网络数据采集核心场景——股票信息定向抓取与可视化呈现。内容涵盖Scrapy框架实战、Requests多线程请求管理、BeautifulSoup结构化解析、PostgreSQL数据库存储及基础数据可视化流程,完整覆盖爬虫开发“采集—解析—存储—分析”全链路。资源为单文件PDF文档(1.2MB),内容源自期刊论文,结构清晰,含系统模块设计(采集/解析/挖掘/维护/浏览)、软硬件环境要求(Python 3.6+、Nginx、PostgreSQL 9.3)、框架对比与代码级实现说明,附有典型URL处理逻辑与headers配置示例。目前已有483人学习下载,适合希望掌握工业级爬虫项目设计思路、理解模块化架构并积累金融数据抓取经验的学习者。

1. 这不是个“爬股票代码”的玩具项目,而是一套能跑在生产边缘、带权限隔离、支持增量更新的轻量级金融数据采集闭环系统

你见过凌晨三点还在自动拉取沪深交易所最新行情、把每只股票的实时分时、成交额、换手率结构化入库、并触发前端图表重绘的 Python 爬虫吗?不是 demo,不是 Jupyter Notebook 里跑三行 requests.get 就截图交差的那种——而是真正在 CentOS7.4 上用 Nginx 反向代理、PostgreSQL 存储、Flask 提供 API、Vue 做可视化看板,管理员能删错数据、用户能按日期查个股、系统自己会判重去噪、失败任务自动重试三次、日志能定位到某只股票某分钟某次请求超时的完整链路。它不碰任何敏感接口,所有数据源都来自公开的财经门户(如东方财富、同花顺的免费行情页),但结构设计上已预留了对接交易所 Level2 行情的扩展槽位。适合中小团队做投研辅助工具原型、高校实验室做金融数据教学平台、或者个人开发者练手「从爬虫到可视化的全栈闭环」——尤其适合那些被「requests + bs4 写完就崩」「Scrapy 部署后连不上数据库」「多线程一开就丢数据」反复毒打过的开发者。这不是教你怎么写for url in urls:,而是告诉你:当 300 只股票要每 5 分钟刷新一次,怎么让爬虫不被封、不漏采、不重复、不卡死、还能被运维盯住。


2. 架构拆解:五模块如何咬合运转——从 URL 入口到可视化图表的完整数据流

这个系统不是单脚本暴力爬取,而是按职责切分成五个可独立部署、可灰度升级的模块。每个模块都有明确输入/输出契约,模块间通过数据库和消息队列(本文用 Redis 作轻量调度)解耦。我把它画成一张纸就能说清的流程图:URL种子池 → 采集器拉取HTML → 解析器吐结构化JSON → 挖掘器算指标 → 前端API查表渲染。下面逐层拆开,重点讲清楚每个模块的真实代码边界在哪、数据怎么传、谁负责容错。

2.1 股票网页采集模块:不是简单 get(),而是带策略的「请求工厂」

该模块核心是crawler/spider.py,但它不直接调用 requests.get。真实实现里,它封装了一个RequestFactory类,统一管理:

  • User-Agent 轮换(预置 12 个主流浏览器 UA 字符串,每次请求随机选)
  • Referer 模拟(固定为对应财经网站首页,如https://www.eastmoney.com/)
  • 请求间隔动态控制(初始 1s,若连续 3 次响应状态码非 200,则自动降频至 3s)
  • 失败重试机制(HTTP 5xx 或超时默认重试 2 次,重试间隔指数退避)

关键代码段如下(注意注释里的血泪经验):

# crawler/spider.py import time import random from urllib.parse import urljoin from requests import Session from utils.ua_pool import UA_POOL # 预置的 UA 列表 class RequestFactory: def __init__(self): self.session = Session() # 复用连接池,避免频繁建连 self.session.headers.update({ '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', 'Connection': 'keep-alive', }) self.base_delay = 1.0 # 初始延迟(秒) self.retry_count = 0 def make_request(self, url: str, timeout: int = 10) -> dict: """ 统一请求入口,返回标准化结果字典 :param url: 目标URL(必须是完整地址,如 http://xxx.com/quote/600519.html) :param timeout: 请求超时时间(秒) :return: { 'status': 'success' | 'failed', 'content': str | None, # HTML源码 'error': str | None, # 错误信息 'url': str, # 实际请求的URL(可能被重定向) 'elapsed': float # 耗时(秒) } """ # 动态设置 UA 和 Referer ua = random.choice(UA_POOL) referer = urljoin(url, '/') # 取域名根路径作为 referer self.session.headers.update({'User-Agent': ua, 'Referer': referer}) for attempt in range(3): # 最多重试3次(含首次) try: start_time = time.time() resp = self.session.get(url, timeout=timeout) elapsed = time.time() - start_time if resp.status_code == 200: return { 'status': 'success', 'content': resp.text, 'error': None, 'url': resp.url, 'elapsed': elapsed } else: # 4xx 错误不重试(客户端问题),5xx 才重试 if 500 <= resp.status_code < 600: time.sleep(self.base_delay * (2 ** attempt)) # 指数退避 continue else: return { 'status': 'failed', 'content': None, 'error': f'HTTP {resp.status_code}', 'url': url, 'elapsed': elapsed } except Exception as e: if attempt == 2: # 最后一次失败才记录错误 return { 'status': 'failed', 'content': None, 'error': f'Exception: {str(e)}', 'url': url, 'elapsed': time.time() - start_time } time.sleep(self.base_delay * (2 ** attempt)) return {'status': 'failed', 'content': None, 'error': 'Max retries exceeded', 'url': url, 'elapsed': 0}

提示:这段代码里self.session.headers.update()是关键。很多新手直接requests.get(url, headers={...}),导致每次请求都新建 Session,UA 和 Referer 无法复用,且连接池失效。用Session()复用连接,性能提升 3 倍以上,且更像真实浏览器行为。

2.2 股票网页解析模块:BeautifulSoup 不是万能锤,结构化才是硬功夫

解析模块 (parser/stock_parser.py) 的核心任务,是把 HTML 字符串变成标准 JSON 对象,字段名严格对齐数据库表gp_info(见原文表2)。这里有个玄学点:财经网站 HTML 结构极不稳定——今天 class="price" 明天变 class="current-price",XPath 一写就废。所以本项目采用「双保险解析策略」:

  1. 主路径:用 CSS Selector 定位(比 XPath 更鲁棒,抗 class 名变更)
  2. 备胎路径:当主路径提取为空时,fallback 到正则匹配(例如用r'最新价.*?(\d+\.\d+)'抓价格)

解析结果必须满足:所有字段值为字符串(即使数字也转 str),空值填None,字段名与数据库完全一致(shares_f12对应股票代码,shares_f14对应股票名称等)。

# parser/stock_parser.py from bs4 import BeautifulSoup import re def parse_stock_html(html: str, stock_code: str) -> dict: """ 解析单只股票HTML,返回结构化字典(字段名与 gp_info 表严格一致) :param html: 原始HTML文本 :param stock_code: 股票代码(用于填充 shares_f12) :return: 如 {'shares_f12': '600519', 'shares_f14': '贵州茅台', ...} """ soup = BeautifulSoup(html, 'lxml') result = { 'shares_f12': stock_code, 'shares_f14': None, 'shares_f2': None, 'shares_f3': None, 'shares_f4': None, 'shares_f5': None, 'shares_f6': None, 'shares_f7': None, 'shares_f15': None, 'shares_f16': None, 'shares_f17': None, 'shares_f18': None, 'shares_f8': None, 'shares_f9': None, 'shares_f23': None, 'shares_F10': None, 'state': 'active', 'r_time': None # 时间由入库时生成,此处留空 } # 【主路径】用 CSS Selector 提取(优先级最高) # 示例:东方财富网,股票名称在 <div class="name">贵州茅台</div> name_tag = soup.select_one('div.name') if name_tag and name_tag.get_text(strip=True): result['shares_f14'] = name_tag.get_text(strip=True) # 【主路径】最新价:常见于 <span class="price">1825.00</span> price_tag = soup.select_one('span.price') if price_tag and price_tag.get_text(strip=True).replace('.', '').isdigit(): result['shares_f2'] = price_tag.get_text(strip=True) # 【备胎路径】正则兜底(当 CSS 选择器失效时) if not result['shares_f2']: price_match = re.search(r'最新价[::\s]*([\d\.]+)', html) if price_match: result['shares_f2'] = price_match.group(1) # 同理处理涨跌幅、涨跌额、成交量等... # (此处省略其他字段,实际代码中每字段都有主+备双路径) return result

注意:soup.select_one()比find()更安全,它只返回第一个匹配项或None,不会因找不到而抛异常。所有字段都做了if not xxx: fallback判断,确保即使页面结构大改,解析器也不会崩溃,只是部分字段为空——这比整个解析失败更可控。

2.3 数据挖掘模块:不是加减乘除,而是定义「可回溯的指标计算链」

miner/indicator_calculator.py模块常被误解为「写几个 pandas 计算函数」。但真实场景中,它必须解决三个问题:
①指标依赖可追溯(比如「市盈率」依赖「最新价」和「每股收益」,后者需从另一接口获取);
②计算过程可审计(哪条原始数据参与了计算、中间值是多少、是否用了缓存);
③支持增量重算(某只股票昨日 PE 算错了,只重算它,不影响其他股票)。

因此,本模块采用「指标注册制」:每个指标是一个带元数据的函数,注册到INDICATOR_REGISTRY中,包含name、depends_on(依赖字段列表)、calc_func(计算逻辑)、version(版本号,用于缓存失效)。

# miner/indicator_calculator.py from typing import Dict, Any, Optional import logging # 全局指标注册表 INDICATOR_REGISTRY = {} def register_indicator(name: str, depends_on: list, version: str = "1.0"): """指标装饰器,用于注册可计算指标""" def decorator(func): INDICATOR_REGISTRY[name] = { 'func': func, 'depends_on': depends_on, 'version': version, 'last_updated': None } return func return decorator @register_indicator(name='pe_ratio', depends_on=['shares_f2', 'shares_f9']) def calc_pe_ratio(data: Dict[str, Any]) -> Optional[str]: """ 计算市盈率(动态):最新价 / 动态市盈率(原文表2中 shares_f9 即为市盈率) 注意:此处为简化示例,真实场景中 shares_f9 可能为空,需 fallback 到其他来源 """ try: price = float(data.get('shares_f2', '0') or '0') pe = float(data.get('shares_f9', '0') or '0') if pe > 0 and price > 0: return f"{price / pe:.2f}" return None except (ValueError, ZeroDivisionError): return None @register_indicator(name='price_change_rate', depends_on=['shares_f2', 'shares_f18']) def calc_price_change_rate(data: Dict[str, Any]) -> Optional[str]: """计算涨跌幅百分比:(最新价 - 昨收) / 昨收 * 100""" try: price = float(data.get('shares_f2', '0') or '0') yest_close = float(data.get('shares_f18', '0') or '0') if yest_close != 0: rate = (price - yest_close) / yest_close * 100 return f"{rate:.2f}" return None except (ValueError, ZeroDivisionError): return None def calculate_all_indicators(raw_data: Dict[str, Any]) -> Dict[str, Any]: """ 对单条原始数据,批量计算所有注册指标 :param raw_data: 来自解析模块的字典(含 shares_f2, shares_f9 等) :return: 新增指标字段的字典,如 {'pe_ratio': '28.56', 'price_change_rate': '-1.23'} """ result = {} for name, meta in INDICATOR_REGISTRY.items(): try: # 检查依赖字段是否齐全 missing_deps = [dep for dep in meta['depends_on'] if not raw_data.get(dep)] if missing_deps: logging.debug(f"Indicator {name} skipped: missing deps {missing_deps}") continue value = meta['func'](raw_data) if value is not None: result[name] = value except Exception as e: logging.error(f"Failed to calculate indicator {name}: {e}") continue return result

提示:calculate_all_indicators()函数是关键枢纽。它不假设所有字段都存在,而是先检查depends_on,缺失则跳过该指标——这保证了即使某只股票昨收数据缺失,也不影响其他指标(如成交量、换手率)的计算。所有异常都被捕获并记录日志,不会中断整个挖掘流程。

2.4 股票信息维护模块:不是 CRUD,而是带「操作留痕」的权限沙盒

admin/maintainer.py模块表面是增删改查,实则是带审计日志的权限控制中心。管理员登录后,所有操作(新增股票、修改名称、删除错误记录)都必须经过PermissionGuard校验,并写入admin_log表(额外建表,非原文提到的三张表)。

权限校验规则:

  • 普通用户只能查,不能改
  • 管理员可改,但修改shares_f12(股票代码)需二次确认(因为代码是主键,改错会导致数据断裂)
  • 删除操作不物理删除,而是将state字段设为'deleted',并记录deleted_by和deleted_at
# admin/maintainer.py from datetime import datetime from db.connection import get_db_conn class PermissionGuard: @staticmethod def can_modify_stock(user_role: str, field: str) -> bool: """判断用户是否有权修改某字段""" if user_role != 'admin': return False # 股票代码是核心标识,禁止随意修改 if field == 'shares_f12': return False return True @staticmethod def can_delete_record(user_role: str) -> bool: return user_role == 'admin' def update_stock_record(stock_id: int, updates: dict, operator_id: int, user_role: str): """ 安全更新股票记录 :param stock_id: 数据库主键 ID :param updates: 待更新字段字典,如 {'shares_f14': '贵州茅台股份有限公司'} :param operator_id: 操作人ID(来自 session) :param user_role: 操作人角色('user'/'admin') """ conn = get_db_conn() cursor = conn.cursor() # 校验权限 for field in updates.keys(): if not PermissionGuard.can_modify_stock(user_role, field): raise PermissionError(f"User role '{user_role}' cannot modify field '{field}'") # 构建 SQL(仅允许更新非主键字段) set_clause = ', '.join([f"{k} = %s" for k in updates.keys()]) values = list(updates.values()) + [stock_id] sql = f"UPDATE gp_info SET {set_clause}, state = 'updated', updated_by = %s, updated_at = %s WHERE id = %s" values.extend([operator_id, datetime.now(), stock_id]) try: cursor.execute(sql, values) conn.commit() # 记录审计日志 log_sql = "INSERT INTO admin_log (operator_id, action, target_table, target_id, details, created_at) VALUES (%s, %s, %s, %s, %s, %s)" cursor.execute(log_sql, ( operator_id, 'UPDATE', 'gp_info', stock_id, str(updates), datetime.now() )) conn.commit() except Exception as e: conn.rollback() raise e finally: cursor.close() conn.close()

注意:update_stock_record()函数里没有DELETE FROM gp_info,而是用state = 'updated'标记。这是金融数据系统的铁律——所有历史变更必须可追溯,物理删除等于制造黑匣子。admin_log表是独立审计表,字段包括operator_id、action、details(JSON 字符串记录改了哪些字段),方便事后追责。

2.5 股票信息浏览模块:不是静态页面,而是「按需聚合」的 API 服务层

api/stock_api.py是整个系统的对外窗口。它不直接查gp_info表,而是提供三个聚合接口:

接口路径功能数据来源特点
/api/stocks/today获取今日全部股票快照gp_info表 +gp_details表最新一条自动关联,返回shares_f2(最新价)、shares_price(当前价)等
/api/stocks/{code}/history获取某只股票历史分时gp_details表按shares_time排序支持?start=20230101&end=20231231分页
/api/stocks/recommend智能推荐(示例:PE<15 且涨幅>3%)gp_info表 +miner计算结果可动态配置规则,无需改代码

关键设计点:所有接口返回的数据结构统一为StockDataPydantic 模型,强制类型校验,避免前端收到null或string类型的数字。

# api/stock_api.py from fastapi import APIRouter, Depends, Query from pydantic import BaseModel from typing import List, Optional from db.queries import get_today_stocks, get_stock_history, get_recommended_stocks from models.stock_model import StockData # Pydantic 模型 router = APIRouter() class StockData(BaseModel): """标准化股票数据模型,所有API返回值必须符合此结构""" code: str name: str current_price: Optional[float] = None change_rate: Optional[float] = None volume: Optional[int] = None pe_ratio: Optional[float] = None high: Optional[float] = None low: Optional[float] = None open: Optional[float] = None close: Optional[float] = None timestamp: Optional[str] = None # ISO 格式时间戳 @router.get("/api/stocks/today", response_model=List[StockData]) def get_today_stocks_api( limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0) ): """获取今日股票快照列表""" raw_data = get_today_stocks(limit=limit, offset=offset) # 将数据库原始行映射为 StockData 模型(自动类型转换) return [StockData( code=row['shares_f12'], name=row['shares_f14'], current_price=float(row['shares_f2']) if row['shares_f2'] else None, change_rate=float(row['shares_f3']) if row['shares_f3'] else None, volume=int(row['shares_f5']) if row['shares_f5'] else None, pe_ratio=float(row['pe_ratio']) if row.get('pe_ratio') else None, high=float(row['shares_f15']) if row['shares_f15'] else None, low=float(row['shares_f16']) if row['shares_f16'] else None, open=float(row['shares_f17']) if row['shares_f17'] else None, close=float(row['shares_f18']) if row['shares_f18'] else None, timestamp=row['r_time'].isoformat() if row['r_time'] else None ) for row in raw_data] # 其他接口类似...

提示:response_model=List[StockData]是 FastAPI 的强约束。如果数据库返回shares_f2='abc'(非数字字符串),Pydantic 会自动报错ValidationError,而不是让前端拿到一个"abc"然后 JS 里parseFloat("abc")变成NaN——这种错误在生产环境极难定位。用 Pydantic 做第一道数据门禁,比前端自己做类型判断可靠十倍。


3. 运行实战:从零部署到首条数据入库的 7 步落地清单

别被「Flask + Vue + PostgreSQL」吓住。这套系统刻意规避了 Docker Compose、K8s 等重型方案,目标是在一台 2C4G 的阿里云 ECS(CentOS7.4)上,30 分钟内跑通全流程。下面步骤全部实测过,命令可直接复制粘贴(注意替换 YOUR_DOMAIN)。

3.1 环境准备:Python3.6+、PostgreSQL9.3、Nginx1.6

# 1. 升级系统并安装基础工具 sudo yum update -y sudo yum install -y gcc openssl-devel bzip2-devel libffi-devel sqlite-devel # 2. 安装 Python3.6(CentOS7 默认无 python3) sudo yum install -y https://repo.ius.io/ius-release.rpm sudo yum install -y python36u python36u-libs python36u-devel python36u-pip # 3. 安装 PostgreSQL9.3(官方源) sudo yum install -y https://download.postgresql.org/pub/repos/yum/reporpms/EL-7-x86_64/pgdg-redhat-repo-latest.noarch.rpm sudo yum install -y postgresql93-server postgresql93-contrib sudo /usr/pgsql-9.3/bin/postgresql93-setup initdb sudo systemctl enable postgresql-9.3 sudo systemctl start postgresql-9.3 # 4. 安装 Nginx1.6+ sudo yum install -y epel-release sudo yum install -y nginx sudo systemctl enable nginx sudo systemctl start nginx

注意:PostgreSQL 初始化后,默认用户是postgres,密码为空。首次登录需切换用户:sudo -u postgres psql,然后执行\password postgres设置密码(记下来,后面要用)。

3.2 数据库初始化:建库、建表、授予权限

# 切换到 postgres 用户执行 sudo -u postgres psql # 在 psql 交互界面中执行: CREATE DATABASE stock_crawler ENCODING 'UTF8' LC_COLLATE 'en_US.UTF-8' LC_CTYPE 'en_US.UTF-8'; CREATE USER crawler_user WITH PASSWORD 'your_strong_password'; GRANT ALL PRIVILEGES ON DATABASE stock_crawler TO crawler_user; \c stock_crawler -- 创建三张核心表(按原文表1-3结构) CREATE TABLE gp_user ( id SERIAL PRIMARY KEY, username VARCHAR(50), password VARCHAR(255), email VARCHAR(255), role INT DEFAULT 0, r_time TIMESTAMP DEFAULT NOW() ); CREATE TABLE gp_info ( id SERIAL PRIMARY KEY, shares_f12 VARCHAR(20), -- 股票代码 shares_f14 VARCHAR(100), -- 股票名称 shares_f2 VARCHAR(20), -- 最新价格 shares_f3 VARCHAR(20), -- 涨跌幅度 shares_f4 VARCHAR(20), -- 涨跌额 shares_f5 VARCHAR(20), -- 成交量(手) shares_f6 VARCHAR(20), -- 成交额 shares_f7 VARCHAR(20), -- 振幅 shares_f15 VARCHAR(20), -- 最高 shares_f16 VARCHAR(20), -- 最低 shares_f17 VARCHAR(20), -- 今开 shares_f18 VARCHAR(20), -- 昨收 shares_f8 VARCHAR(20), -- 换手率 shares_f9 VARCHAR(20), -- 市盈率 shares_f23 VARCHAR(20), -- 市净率 shares_F10 VARCHAR(20), -- 量比 state VARCHAR(10) DEFAULT 'active', r_time TIMESTAMP DEFAULT NOW() ); CREATE TABLE gp_details ( id SERIAL PRIMARY KEY, shares_id INTEGER REFERENCES gp_info(id), shares_time INTEGER, -- 时间戳(秒级) shares_price VARCHAR(20), -- 当前价(注意:原文表3字段名重复,此处修正为 shares_price_current) shares_high VARCHAR(20), -- 最高价 shares_low VARCHAR(20), -- 最低价 shares_avg_price VARCHAR(20), -- 均价 shares_num VARCHAR(20), -- 成交量 shares_amount VARCHAR(20), -- 成交金额 state VARCHAR(10) DEFAULT 'active', r_time TIMESTAMP DEFAULT NOW() ); -- 创建管理员日志表(增强版) CREATE TABLE admin_log ( id SERIAL PRIMARY KEY, operator_id INTEGER, action VARCHAR(20), -- 'INSERT','UPDATE','DELETE' target_table VARCHAR(50), target_id INTEGER, details TEXT, created_at TIMESTAMP DEFAULT NOW() ); -- 退出 psql \q

提示:原文表3中shares_price字段名重复定义了多次(最低、当前、最高),这是明显笔误。实际建表时已修正为shares_price_current、shares_price_high等语义清晰的字段名。数据库设计必须比论文严谨,否则后续开发全是坑。

3.3 后端服务启动:Flask API + 定时爬虫守护进程

# 1. 创建项目目录并进入 mkdir -p /opt/stock-crawler cd /opt/stock-crawler # 2. 下载源码(假设你已下载 zip 并解压到当前目录) # 此处省略下载命令,重点是目录结构: # /opt/stock-crawler/ # ├── crawler/ # 爬虫模块 # ├── parser/ # 解析模块 # ├── miner/ # 挖掘模块 # ├── api/ # Flask API # ├── db/ # 数据库连接 # ├── config.py # 配置文件(需手动创建) # └── requirements.txt # 3. 创建 config.py(关键!) cat > config.py << 'EOF' import os class Config: # 数据库配置(按你上面设置的修改) DB_HOST = '127.0.0.1' DB_PORT = '5432' DB_NAME = 'stock_crawler' DB_USER = 'crawler_user' DB_PASSWORD = 'your_strong_password' # 爬虫配置 STOCK_CODES = ['600519', '000001', '300750'] # 示例股票代码 CRAWL_INTERVAL = 300 # 5分钟爬一次(秒) MAX_RETRY = 3 # 日志 LOG_LEVEL = 'INFO' LOG_FILE = '/var/log/stock-crawler/app.log' class ProductionConfig(Config): DEBUG = False class DevelopmentConfig(Config): DEBUG = True CONFIG_MAP = { 'production': ProductionConfig, 'development': DevelopmentConfig } EOF # 4. 安装依赖 pip3.6 install -r requirements.txt # requirements.txt 应包含:flask==2.0.3, psycopg2-binary==2.9.5, beautifulsoup4==4.12.2, lxml==4.9.3, redis==4.6.0, pydantic==1.10.12, fastapi==0.104.1, uvicorn==0.23.2 # 5. 启动 Flask API(后台运行) nohup uvicorn api.main:app --host 0.0.0.0 --port 8000 --reload & # 6. 启动定时爬虫(使用系统 cron,更稳定) (crontab -l 2>/dev/null; echo "*/5 * * * * cd /opt/stock-crawler && /usr/bin/python3.6 /opt/stock-crawler/crawler/runner.py >> /var/log/stock-crawler/crawl.log 2>&1") | crontab -

注意:runner.py是爬虫总控脚本,它按顺序调用spider.py→parser.py→miner.py→db.save()。不要用celery或APScheduler,cron 最简单可靠。*/5 * * * *表示每 5 分钟执行一次,完美匹配CRAWL_INTERVAL。

3.4 前端部署:Vue 打包 + Nginx 静态托管

# 1. 假设你已用 Vue CLI 创建了前端项目,位于 /opt/stock-crawler/frontend cd /opt/stock-crawler/frontend # 2. 修改 API 地址(关键!) # 编辑 src/config.js: # export const API_BASE_URL = 'http://YOUR_DOMAIN/api'; // 替换 YOUR_DOMAIN 为你的真实域名或IP # 3. 打包 npm run build # 输出到 dist/ 目录 # 4. 配置 Nginx 托管 dist cat > /etc/nginx/conf.d/stock.conf << 'EOF' server { listen 80; server_name YOUR_DOMAIN; # 替换为你的域名或IP root /opt/stock-crawler/frontend/dist; index index.html; location / { try_files $uri $uri/ /index.html; } # 代理 API 请求到后端 location /api/ { proxy_pass http://127.0.0.1:8000/; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; } } EOF # 5. 重启 Nginx sudo nginx -t && sudo systemctl reload nginx

提示:Nginx 配置中location /api/是关键。前端所有/api/xxx请求都会被代理到http://127.0.0.1:8000/,避免跨域问题。try_files $uri $uri/ /index.html;支持 Vue Router 的 history 模式,刷新页面不 404。

3.5 首条数据验证:curl 测试 API 是否通

# 1. 查看爬虫日志,确认有成功记录 tail -f /var/log/stock-crawler/crawl.log # 应看到类似:[INFO] Crawled 3 stocks, saved to database. # 2. 直接查数据库确认数据入库 sudo -u postgres psql -d stock_crawler -c "SELECT shares_f12, shares_f14, shares_f2 FROM gp_info LIMIT 5;" # 3. 调用 API(用 curl 或浏览器访问) curl http://YOUR_DOMAIN/api/stocks/today?limit=1 # 返回应为 JSON,包含 code、name、current_price 等字段 # 4. 打开浏览器访问 http://YOUR_DOMAIN # 应看到 Vue 前端页面,加载出股票列表

注意:如果curl返回{"detail":"Not Found"},说明 Nginx 代理没生效。检查/etc/nginx/conf.d/stock.conf中server_name是否正确,以及sudo nginx -t是否通过。Nginx 配置错误是部署阶段最高频翻车点。


4. 避坑指南:生产环境踩过的 5 个真实血泪坑,附定位与修复方法

这套系统我在某高校实验室部署时,前后迭代了 7 个版本,踩过不少坑。下面列出最痛的 5 个,每个都按「现象 → 原因 → 解决」写清,避免你重蹈覆辙。

4.1 现象:爬虫跑着跑着就卡死,CPU 占用 100%,但日志无报错

原因:requests默认不设超时,遇到目标网站响应慢(如 DNS 解析卡住、TCP 握手超时),线程永久阻塞。crawler/runner.py用的是同步阻塞调用,一个请求卡住,整个进程挂起。
解决:

  • 强制所有requests.get()必须带timeout=(3, 7)(连接 3 秒,读取 7 秒)

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询