炒股利器！用Python大模型筛选低估值股票 - 文章 - 开发者社区

picture.image

关注并将**「AIGC开发者」** 设为星标，

每天早上 08 : 35 按时送达

在股票投资中，寻找被低估的股票是实现高回报的重要途径。传统的分析方法需要大量手工处理数据，而Python数据科学和LLM大模型技术可以帮助我们更高效地完成这项工作。这个项目旨在利用Python编程语言和大模型来自动化筛选纳斯达克市场中的低估值股票。 本文所有代码已打包，并配有详细文档，下载方法详见文末 ，代码文件结构如下：

picture.image

最后代码执行后筛选结果如下：

picture.image

大模型给我们的财富密码是：PDD（拼多多）、CSCO（思科）、AEP（美国电力公司）、MU（美光科技）、CDW。大模型对筛选出的这几只股也作了解释，还是值得我们参考的。

好了，现在从头开始讲解如何进行开发，我们的主要目标是开发一个自动化系统，通过结合关键财务指标和大模型的智能分析，筛选出被低估且具有潜在高回报的股票。

我们选择了以下五个关键财务指标来评估股票的价值：市盈率 (PE Ratio)、市净率 (PB Ratio)、负债股本比 (Debt/Equity Ratio)、股息收益率 (Dividend Yield)、每股收益 (EPS) 增长率。

实现过程如下：

1、数据收集：

使用Python脚本从Yahoo Finance获取纳斯达克100指数成分股的历史数据。从维基百科页面获取纳斯达克100指数的成分股列表。

2、数据处理：

使用Pandas库对收集到的数据进行处理和清洗。提取每只股票的关键财务指标，包括市盈率、市净率、负债股本比、股息收益率和每股收益增长率。

3、智能分析：

使用Cohere API调用大模型对处理后的数据进行智能分析。基于大模型的分析结果，筛选出被低估且具有潜在高回报的股票。

4、结果输出：

将筛选结果保存到CSV文件中，方便进一步的分析和决策支持。打印筛选结果，提供对每只被选中股票的理由解释。

Cohere API 是一套应用程序接口，允许开发者访问和使用其强大的大规模语言模型（LLMs）来执行各种自然语言处理任务。Cohere 的 API 旨在简化机器学习模型的集成和使用，使得开发者无需深入了解复杂的机器学习算法和模型训练过程，就可以利用大模型技术进行文本分析、情感分析等任务。我们可以直接注册并获取API key，如下所示。

picture.image

其中最关键的过程是数据处理和智能分析的部分，以下是关于这部分代码的详细技术分析，解释其实现原理和流程。

导入库和模块


        
          
import os  
import pandas as pd  
import yfinance as yf  
from datetime import datetime, timedelta  
from cohere_interface import Cohere  # 确保导入Cohere类  
from concurrent.futures import ThreadPoolExecutor

这些库和模块提供了基本的数据处理和并行处理功能：

os：用于操作文件和目录。
pandas：强大的数据处理库。
yfinance：用于从Yahoo Finance获取金融数据。
datetime 和 timedelta：用于日期和时间操作。
Cohere：自定义模块，用于与Cohere API交互。
ThreadPoolExecutor：用于并行处理，以提高数据处理速度。

Screener 类定义


        
          
class Screener:  
    def \_\_init\_\_(self, folder):  
        self.index = folder  # 设置索引为文件夹名称  
        self.folder = f'/data/historical/analysis/{folder}/'  # 设置数据文件夹路径  
        self.files = os.listdir(self.folder)  # 获取文件夹中的所有文件  
        self.df_all = pd.DataFrame()  # 初始化空的数据框，用于存储最终数据  
  
        # 创建数据文件夹  
        if not os.path.exists(self.folder):  
            os.makedirs(self.folder)

初始化Screener类时，设置索引和数据文件夹路径。
获取文件夹中的所有文件名。
初始化一个空的DataFrame，用于存储最终的数据。
如果数据文件夹不存在，则创建该文件夹。

处理单个文件的方法


        
          
    def process\_file(self, filename):  
        if (filename.endswith('xlsx') or filename.endswith('csv')) and not filename.startswith('~$'):  
            file_path = os.path.join(self.folder, filename)  
            print(f"Processing file: {file\_path}")  
  
            try:  
                df = pd.read_csv(file_path)  # 读取CSV文件  
                df['Date'] = pd.to_datetime(df['Date'], utc=True)  # 转换日期列为日期时间格式  
                df['Date'] = df['Date'].dt.date  # 提取日期部分  
                current_date = df.iloc[-1]['Date']  # 获取最新日期  
                symbol = filename.split()[0]  # 获取股票符号  
                info = yf.Ticker(symbol).info  # 获取股票信息  
  
                # 提取关键财务数据  
                pe = info.get('trailingPE', None)  
                pb = info.get('priceToBook', None)  
                debtToEquity = round(info.get('debtToEquity', 0) / 100.0, 2)  
                dividend_yield = info.get('dividendYield', None)  
                eps_growth = info.get('earningsQuarterlyGrowth', None)  
                close = df.iloc[-1]['Close']  
  
                # 生成数据字典  
                data = {  
                    'date': [current_date],  
                    'symbol': [symbol],  
                    'pe': [pe],  
                    'pb': [pb],  
                    'debtToEquity': [debtToEquity],  
                    'dividend\_yield': [dividend_yield],  
                    'eps\_growth': [eps_growth],  
                    'close': [close]  
                }  
  
                df_change = pd.DataFrame(data)  # 将数据字典转换为数据框  
                self.df_all = pd.concat([self.df_all], axis=0)  # 将新的数据框添加到总数据框中  
            except Exception as e:  
                print(f"Error processing file {file\_path}: {e}")

检查文件是否是Excel或CSV格式且不是临时文件。
读取CSV文件并转换日期格式。
获取最新的日期和股票符号。
使用 yfinance 获取股票的财务信息。
提取关键财务数据：市盈率、价格与账面比、负债股本比、股息收益率、每股收益增长率和收盘价。
生成包含这些数据的字典，并将其转换为DataFrame。
将新的DataFrame添加到总的DataFrame中。
如果处理文件时出错，捕获异常并打印错误信息。

并行处理文件的方法


        
          
    def run(self):  
        with ThreadPoolExecutor(max_workers=4) as executor:  # 使用4个线程并行处理  
            executor.map(self.process_file, self.files)

使用 ThreadPoolExecutor 并行处理文件。
max_workers=4 指定使用4个线程并行处理文件。
executor.map(self.process_file, self.files) 依次将文件传递给 process_file 方法进行处理。

将结果存储到CSV的方法


        
          
    def store\_in\_csv(self):  
        # 将最终数据存储到CSV文件中  
        folder = f'./data/historical/screener/'  
        if not os.path.exists(folder):  
            os.makedirs(folder)  
        filename = f'{self.index}\_{str(datetime.today()}.csv'  
        path = os.path.join(folder, filename)  
        self.df_all.to_csv(path, index=False)

将最终的数据存储到CSV文件中。
检查存储文件夹是否存在，如果不存在则创建。
生成包含当前日期的文件名。
将DataFrame保存到CSV文件中，不包含索引。

使用Cohere进行AI筛选的方法


        
          
    def ai\_screening(self):  
        # 使用Cohere进行AI筛选  
        co = Cohere()  
        message = f"""你将获得一些股票的数据。请分析这些数据并筛选出2只被低估且未来有可能产生更高回报的股票。以下是数据 {self.df\_all}"""  
        preamble = """你是一名优秀的股市分析师，能够很好地理解提供的数据并做出谨慎的决策。输出应严格采用以下JSON格式：  
        [  
            {  
                "stock": "股票名称",  
                "justification": "筛选或选择该股票的理由"  
            },  
            {  
                "stock": "股票名称",  
                "justification": "筛选或选择该股票的理由"  
            }  
        ]"""  
        result = co.chat_cohere(message=message, preamble=preamble)  
        print(result)

使用Cohere API进行AI筛选。
创建Cohere对象。
定义要发送给Cohere的消息，包含待分析的股票数据。
定义消息的前言，设置任务和输出格式的上下文。
调用 chat_cohere 方法发送消息并接收结果。
打印筛选结果。

通过以上步骤，我们就可以自动化地完成纳斯达克100指数成分股的财务数据分析和智能筛选，帮助识别被低估且具有潜在高回报的股票。未来，可以进一步优化和扩展该项目，包括：

增加更多财务指标，动态调整和增加筛选财务指标。集成更多数据源，丰富数据维度，提高分析深度。模型优化与更新，比如选用BloomberGPT之类的一些金融行业的大模型，确保分析结果的前沿性和准确性。开发用户友好的界面，提供可视化报告和实时筛选结果展示。

扫码获取本文所有代码文件、详细文档：

picture.image

点击阅读原文，加入知识星球