炒股利器！用Python大模型筛选低估值股票

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

每天早上 08 : 35 按时送达

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

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

导入库和模块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之类的一些金融行业的大模型，确保分析结果的前沿性和准确性。开发用户友好的界面，提供可视化报告和实时筛选结果展示。

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

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