AI 股票分析系统搭建:Python 爬虫数据采集实战
本文将介绍如何搭建一个属于自己的 AI 股票分析系统。该系统名为 daily_stock_analysis,是一个开源项目。其核心特点是利用云端资源和 AI 大模型,将繁琐的复盘工作自动化。
介绍基于 Python 爬虫的 AI 股票分析系统搭建方案。涵盖云端环境配置、使用 requests 和 BeautifulSoup 抓取新浪财经实时行情、批量处理自选股数据、设置定时任务、数据清洗与 JSON 格式化,以及将自定义数据源接入 AI 分析流程。最终实现自动化数据采集与智能分析报告推送,辅助投资决策。
本文将介绍如何搭建一个属于自己的 AI 股票分析系统。该系统名为 daily_stock_analysis,是一个开源项目。其核心特点是利用云端资源和 AI 大模型,将繁琐的复盘工作自动化。
在动手之前,我们先了解一下这个系统的功能。
daily_stock_analysis 的核心思路是:让机器代替人去做那些重复、枯燥的信息收集和初步分析工作。
想象一下,一个专业的股票分析师每天要做什么?他需要:
这个 AI 系统干的就是前面三步的'苦力活',然后把整理好的数据和初步分析交给一个 AI 大模型(比如 Gemini 或者 DeepSeek),让它生成一份像模像样的'决策仪表盘'。最后,这份报告会通过微信、飞书或者邮件自动发给你。
它不会替你下决定,但它能把你从海量信息里解放出来,让你只关注最有价值的结论。
我们选择在云端 Linux 环境中部署,主要是图它方便。你不用自己折腾服务器环境,也不用担心家里的电脑关机了程序就停了。云端提供了现成的、带 GPU 的容器环境,跑 AI 应用正合适。
首先,你得有一个云端平台的账号。登录之后,找到'创建实例'或者'新建工作空间'的按钮。
在镜像选择这里,是关键一步。我们不需要从零开始安装 Python 环境,那样太慢了。你可以直接搜索'Python'或者'PyTorch'这类基础镜像。我推荐选择预装了 Python 3.10 或 Python 3.11,并且带有 CUDA 支持的镜像,这样后续调用一些需要 GPU 加速的库会更顺畅。
实例配置上,对于这个分析系统,CPU 和内存不用特别高,起步配置通常就够用。重点是把存储空间给足,建议至少 50GB,因为我们要安装不少 Python 库,后续运行也会产生一些缓存文件。
点击创建,稍等几分钟,一个属于你的云端开发环境就准备好了。
环境有了,接下来要把 daily_stock_analysis 的代码放进去。这里我们用 git 命令,这就像是一个代码搬运工。
打开云端平台提供的终端(通常是 Web Terminal 或者 Jupyter Lab 里的 Terminal),输入以下命令:
# 克隆项目代码到你的工作空间
git clone https://github.com/ZhuLinsen/daily_stock_analysis.git
# 进入项目文件夹
cd daily_stock_analysis
两条命令执行完,项目的所有文件就都下载到你的云端环境里了。你可以用 ls 命令看看,是不是多了一个 daily_stock_analysis 的文件夹。
代码有了,但它还跑不起来,因为它依赖很多其他的 Python 库,比如网络请求的 requests、数据分析的 pandas 等等。我们需要一次性把这些'零件'都装上。
项目很贴心地准备了一个 requirements.txt 文件,里面列出了所有需要的库。我们只需要一条命令:
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
这里我加了一个 -i 参数,指定使用清华大学的镜像源来下载。国内网络访问这个源会快很多,能避免安装过程卡住。
这个安装过程可能需要几分钟,取决于网速。泡杯茶,耐心等一下。如果中间有某个库安装报错,别慌,通常是网络波动,重新运行一次上面的命令就行。
好了,基础环境搭好了。现在来到最核心、也是最有意思的部分——数据采集。daily_stock_analysis 项目本身支持多种数据源,比如 AkShare、Tushare。但我想教你更底层、更通用的方法:自己写爬虫去抓。
为什么?因为自己写的爬虫最灵活。万一某个数据源接口变了,或者你想抓一些特定网站的数据,自己动手就能搞定。
我们从一个最简单的例子开始:从新浪财经抓取某只股票的实时价格。别被'爬虫'这个词吓到,其实就是让程序去访问一个网页,然后把我们需要的信息'抠'出来。
我们先安装一个专门用来解析网页的库:
pip install beautifulsoup4 lxml
然后,创建一个新的 Python 文件,比如叫 stock_spider.py,把下面的代码复制进去:
import requests
from bs4 import BeautifulSoup
import time
def get_stock_price(stock_code):
"""
从新浪财经获取 A 股股票实时价格
:param stock_code: 股票代码,例如 'sh600519' (上海) 或 'sz000001' (深圳)
:return: 股票名称和当前价格
"""
# 构建新浪财经股票页面的 URL
url = f'https://hq.sinajs.cn/list={stock_code}'
# 设置请求头,模拟浏览器访问
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Referer': 'https://finance.sina.com.cn'
}
try:
# 发送网络请求
response = requests.get(url, headers=headers)
response.encoding = 'gbk' # 新浪财经使用 gbk 编码
if response.status_code == 200: # 解析返回的数据(格式类似:var hq_str_sh600519="茅台,1850.50,...";)
data_str = response.text
# 提取引号内的数据部分
data_content = data_str.split('"')[1]
data_list = data_content.split(',')
if len(data_list) > 1:
stock_name = data_list[0]
current_price = data_list[3] # 当前价格通常在第四个位置
print(f"股票 {stock_name}({stock_code}) 当前价格:{current_price} 元")
return stock_name, current_price
else:
print("解析数据失败")
return None, None
else:
print(f"请求失败,状态码:{response.status_code}")
return None, None
except Exception as e:
print(f"抓取过程中出现错误:{e}")
return None, None
# 试试抓取贵州茅台的股价
if __name__ == "__main__":
# 贵州茅台在上海证券交易所的代码是 sh600519
name, price = get_stock_price('sh600519')
if price:
print(f"成功获取到 {name} 的价格:{price}")
运行这个脚本,你就能在终端看到贵州茅台(600519)的实时股价了。这段代码做了几件事:
requests.get 去访问新浪财经的一个特定数据接口。BeautifulSoup 虽然这里没直接用上,但它是解析复杂 HTML 的利器,我们后面会用到。这就是爬虫最基本的工作原理:访问 -> 获取 -> 解析。
只抓一只股票不过瘾,我们通常关注一个股票池。这就需要用到循环和列表了。
假设你有一个自选股列表,存放在一个文本文件 my_stocks.txt 里,每行一个代码:
sh600519 sz000858 sz300750
我们可以写一个函数来批量抓取:
def batch_get_prices(stock_file='my_stocks.txt'):
"""批量从文件读取股票代码并获取价格"""
stock_data = []
try:
with open(stock_file, 'r', encoding='utf-8') as f:
stock_codes = [line.strip() for line in f if line.strip()]
print(f"开始批量获取 {len(stock_codes)} 只股票数据...")
for code in stock_codes:
name, price = get_stock_price(code)
if name and price:
stock_data.append({
'code': code,
'name': name,
'price': price
})
# 礼貌一点,每次请求间隔 1 秒,避免把别人服务器搞崩
time.sleep(1)
print("\n=== 自选股行情汇总 ===")
for item in stock_data:
print(f"{item['name']}({item['code']}): {item['price']}")
return stock_data
except FileNotFoundError:
print(f"文件 {stock_file} 不存在")
return []
# 运行批量抓取
if __name__ == "__main__":
my_portfolio = batch_get_prices()
这样,你只需要维护好那个 my_stocks.txt 文件,运行一次脚本,整个股票池的行情就一目了然了。
手动运行脚本还是麻烦。既然是 AI 分析师,就得让它自动化。在 Linux 环境下,最常用的定时任务工具是 crontab。
在终端里输入 crontab -e,会打开一个编辑器。在里面添加一行:
# 每天上午 9 点 30 分(开盘后)和下午 3 点(收盘后)各运行一次爬虫脚本
30 9 * * 1-5 /usr/bin/python3 /你的路径/stock_spider.py >> /你的路径/stock.log 2>&1
0 15 * * 1-5 /usr/bin/python3 /你的路径/stock_spider.py >> /你的路径/stock.log 2>&1
这行命令的意思是:每周一到周五(1-5)的 9 点 30 分和 15 点整,自动执行我们的 Python 脚本,并且把运行日志追加到 stock.log 文件里。
这样一来,你的爬虫就成了一个不知疲倦的小助手,每天准时帮你采集数据。
爬虫抓回来的数据往往是原始的、杂乱的。直接丢给 AI,它可能'吃坏肚子'。所以我们需要做一步'数据清洗',把数据整理成 AI 容易理解的格式。
爬虫过程中可能会遇到网络错误、页面改版等问题,导致抓回来的数据是空的或者格式不对。我们需要在代码里增加一些'容错'处理。
改进一下之前的 get_stock_price 函数:
def get_stock_price_enhanced(stock_code):
url = f'https://hq.sinajs.cn/list={stock_code}'
headers = {'User-Agent': 'Mozilla/5.0 ...'}
try:
response = requests.get(url, headers=headers, timeout=10) # 设置 10 秒超时
response.encoding = 'gbk'
if response.status_code != 200:
print(f"警告:请求{stock_code}失败,状态码{response.status_code}")
# 这里可以尝试重试,或者记录到错误日志
return None, None, None # 多返回一个状态
data_str = response.text
if 'hq_str' not in data_str:
print(f"警告:{stock_code}返回的数据格式异常")
return None, None, '格式错误'
# ... 原有的解析逻辑 ...
# 假设我们成功解析出了价格
price_float = float(current_price)
# 数据合理性校验:股价通常不会低于 0.01 元或高于 10 万元
if price_float < 0.01 or price_float > 100000:
print(f"警告:{stock_code}价格{price_float}异常,可能数据有误")
return stock_name, current_price, '价格异常'
return stock_name, current_price, '成功'
except requests.exceptions.Timeout:
print(f"错误:获取{stock_code}超时")
return None, None, '超时'
except ValueError as e:
print(f"错误:解析{stock_code}价格时出错:{e}")
return None, None, '解析错误'
except Exception as e:
print(f"未知错误:{e}")
return None, None, '未知错误'
你看,我们增加了超时控制、数据格式校验、数值合理性判断,并且对不同类型的错误进行了分类。这样,即使某个股票数据抓取失败,也不会影响整个程序,我们还能知道失败的原因是什么。
daily_stock_analysis 项目内部有它自己约定的数据格式。我们的爬虫抓取到数据后,最好能转换成项目能直接使用的格式,比如一个 JSON 文件或者直接写入数据库。
假设项目需要一个 stock_data.json 文件,格式如下:
[
{
"symbol": "600519.SH",
"name": "贵州茅台",
"current_price": 1850.50,
"change": "0.85%",
"update_time": "2024-01-01 15:00:00"
}
]
那我们可以在批量抓取函数最后,加上数据保存的逻辑:
import json
from datetime import datetime
def save_to_json(data_list, filename='stock_data.json'):
"""将股票数据保存为 JSON 格式"""
output_data = []
for item in data_list:
output_data.append({
"symbol": item['code'].replace('sh', '').replace('sz', '') + '.SH' if 'sh' in item['code'] else '.SZ',
"name": item['name'],
"current_price": float(item['price']),
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
})
with open(filename, 'w', encoding='utf-8') as f:
json.dump(output_data, f, ensure_ascii=False, indent=2)
print(f"数据已保存至 {filename}")
保存成 JSON 的好处是,结构清晰,很多编程语言都能方便地读取。daily_stock_analysis 的主程序就可以直接加载这个文件,获取最新的股票数据。
数据有了,环境也有了,最后一步就是把它们串联起来,让 daily_stock_analysis 系统能用上我们爬取的数据。
回到 daily_stock_analysis 项目目录,我们需要进行一些配置。最简单的方式是复制环境变量模板文件:
cp .env.example .env
然后用文本编辑器打开 .env 文件,你会看到很多配置项。最关键的是这几行:
# 1. 设置你的自选股列表 (用逗号分隔)
STOCK_LIST=600519,000858,300750
# 2. 设置 AI 模型 (二选一)
# 如果你有 Google Gemini 的 API Key(免费申请)
GEMINI_API_KEY=your_gemini_api_key_here
# 或者,如果你用 DeepSeek、通义千问等兼容 OpenAI 的模型
OPENAI_API_KEY=your_openai_api_key
OPENAI_BASE_URL=https://api.deepseek.com/v1
OPENAI_MODEL=deepseek-chat
# 3. 设置通知方式 (比如企业微信)
WECHAT_WEBHOOK_URL=your_wechat_webhook_url
把 STOCK_LIST 改成你自己的股票代码。然后去申请一个 AI 模型的 API Key(Gemini 免费,DeepSeek 也便宜)。最后,如果你希望每天收到推送,就去配置一个企业微信或飞书的机器人,把 Webhook 地址填进来。
默认情况下,daily_stock_analysis 使用 AkShare 等库获取数据。为了让它使用我们爬虫抓取的数据,我们需要做一点小小的'嫁接'。
找到项目里获取股票行情的代码文件(通常在 data_provider 或 sources 目录下)。我们不需要改动核心逻辑,只需在合适的位置,添加一个从我们本地 JSON 文件读取数据的函数,并让它优先使用我们的数据。
例如,创建一个新文件 custom_data_loader.py:
import json
import os
from datetime import datetime, timedelta
def load_custom_stock_data(json_file='stock_data.json'):
"""从本地 JSON 文件加载爬虫获取的股票数据"""
if not os.path.exists(json_file):
print(f"自定义数据文件 {json_file} 不存在,将使用默认数据源。")
return None
try:
with open(json_file, 'r', encoding='utf-8') as f:
data = json.load(f)
# 检查数据是否新鲜(比如是今天的数据)
if data:
latest_time = data[0].get('update_time')
# 这里可以添加时间有效性校验逻辑
# ...
print(f"成功从自定义文件加载 {len(data)} 条股票数据。")
return data
else:
return None
except Exception as e:
print(f"加载自定义数据失败:{e}")
return None
# 然后,在项目主逻辑中,可以这样调用:
# custom_data = load_custom_stock_data()
# if custom_data:
# # 使用自定义数据
# process_data(custom_data)
# else:
# # 回退到默认数据源
# default_data = get_default_data()
这样,当我们的爬虫定时运行并更新了 stock_data.json 文件后,AI 分析系统在运行时就会优先采用这份更新鲜、更定制化的数据。
所有配置都搞定后,在项目根目录下运行:
python main.py
如果一切顺利,你会看到程序开始运行,获取数据、调用 AI 分析、生成报告。稍等片刻,如果你配置了通知,你的手机就会'叮'的一声,收到一份像下面这样的 AI 分析报告:
[AI 决策仪表盘] 🟢 买入 | 贵州茅台 (600519) 缩量回踩 MA5 支撑,乖离率 1.2% 处于最佳买点
💰 建议:买入 1800 | 止损 1750 | 目标 1900
多头排列 乖离安全 量能配合
--- 生成时间:18:00
恭喜你!到这里,一个集成了自动数据采集(Python 爬虫)和智能分析(AI 大模型)的股票分析系统,就完全在你的掌控之中了。
整个过程从无到有,我们经历了环境部署、爬虫编写、数据清洗、系统集成。最棒的是,它完全在你的私有环境里运行,数据经过你的手,分析逻辑由你定制。
回过头看,技术并没有想象中那么高深。它就是一系列自动化步骤的组合:定时抓取、清洗整理、交给 AI 总结、推送结果。你每天需要做的,就是花一分钟看看手机上的推送报告,把省下来的几个小时,用在更深入的思考和决策上。
这个系统就像一个不知疲倦的初级研究员,帮你完成了信息收集和整理的工作。但它终究是个工具,最后的投资决策,还需要你结合自己的经验和判断。用它来提高效率、辅助决策,而不是完全依赖它,这才是技术的正确打开方式。

微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
使用加密算法(如AES、TripleDES、Rabbit或RC4)加密和解密文本明文。 在线工具,加密/解密文本在线工具,online
生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online
基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online