炒股利器！用Python大模型篩選低估值股票 原創(chuàng)
在股票投資中，尋找被低估的股票是實現(xiàn)高回報的重要途徑。傳統(tǒng)的分析方法需要大量手工處理數(shù)據(jù)，而Python數(shù)據(jù)科學(xué)和LLM大模型技術(shù)可以幫助我們更高效地完成這項工作。這個項目旨在利用Python編程語言和大模型來自動化篩選納斯達(dá)克市場中的低估值股票。本文所有代碼已打包，并配有詳細(xì)文檔，下載方法詳見文末，代碼文件結(jié)構(gòu)如下：
最后代碼執(zhí)行后篩選結(jié)果如下：
大模型給我們的財富密碼是：PDD（拼多多）、CSCO（思科）、AEP（美國電力公司）、MU（美光科技）、CDW。大模型對篩選出的這幾只股也作了解釋，還是值得我們參考的。
好了，現(xiàn)在從頭開始講解如何進(jìn)行開發(fā)，我們的主要目標(biāo)是開發(fā)一個自動化系統(tǒng)，通過結(jié)合關(guān)鍵財務(wù)指標(biāo)和大模型的智能分析，篩選出被低估且具有潛在高回報的股票。
我們選擇了以下五個關(guān)鍵財務(wù)指標(biāo)來評估股票的價值：市盈率 (PE Ratio)、市凈率 (PB Ratio)、負(fù)債股本比 (Debt/Equity Ratio)、股息收益率 (Dividend Yield)、每股收益 (EPS) 增長率。?
實現(xiàn)過程如下：
1、數(shù)據(jù)收集：
使用Python腳本從Yahoo Finance獲取納斯達(dá)克100指數(shù)成分股的歷史數(shù)據(jù)。從維基百科頁面獲取納斯達(dá)克100指數(shù)的成分股列表。
2、數(shù)據(jù)處理：
使用Pandas庫對收集到的數(shù)據(jù)進(jìn)行處理和清洗。提取每只股票的關(guān)鍵財務(wù)指標(biāo)，包括市盈率、市凈率、負(fù)債股本比、股息收益率和每股收益增長率。
3、智能分析：
使用Cohere API調(diào)用大模型對處理后的數(shù)據(jù)進(jìn)行智能分析?；诖竽Ｐ偷姆治鼋Y(jié)果，篩選出被低估且具有潛在高回報的股票。
4、結(jié)果輸出：
將篩選結(jié)果保存到CSV文件中，方便進(jìn)一步的分析和決策支持。打印篩選結(jié)果，提供對每只被選中股票的理由解釋。
Cohere API 是一套應(yīng)用程序接口，允許開發(fā)者訪問和使用其強(qiáng)大的大規(guī)模語言模型（LLMs）來執(zhí)行各種自然語言處理任務(wù)。Cohere 的 API 旨在簡化機(jī)器學(xué)習(xí)模型的集成和使用，使得開發(fā)者無需深入了解復(fù)雜的機(jī)器學(xué)習(xí)算法和模型訓(xùn)練過程，就可以利用大模型技術(shù)進(jìn)行文本分析、情感分析等任務(wù)。我們可以直接注冊并獲取API key，如下所示。
其中最關(guān)鍵的過程是數(shù)據(jù)處理和智能分析的部分，以下是關(guān)于這部分代碼的詳細(xì)技術(shù)分析，解釋其實現(xiàn)原理和流程。
導(dǎo)入庫和模塊
import os
import pandas as pd
import yfinance as yf
from datetime import datetime, timedelta
from cohere_interface import Cohere # 確保導(dǎo)入Cohere類
from concurrent.futures import ThreadPoolExecutor這些庫和模塊提供了基本的數(shù)據(jù)處理和并行處理功能：
- ?
?os??：用于操作文件和目錄。 - ?
?pandas??：強(qiáng)大的數(shù)據(jù)處理庫。 - ?
?yfinance??：用于從Yahoo Finance獲取金融數(shù)據(jù)。 - ?
?datetime??? 和??timedelta??：用于日期和時間操作。 - ?
?Cohere??：自定義模塊，用于與Cohere API交互。 - ?
?ThreadPoolExecutor??：用于并行處理，以提高數(shù)據(jù)處理速度。
Screener 類定義
class Screener:
def __init__(self, folder):
self.index = folder # 設(shè)置索引為文件夾名稱
self.folder = f'/data/historical/analysis/{folder}/' # 設(shè)置數(shù)據(jù)文件夾路徑
self.files = os.listdir(self.folder) # 獲取文件夾中的所有文件
self.df_all = pd.DataFrame() # 初始化空的數(shù)據(jù)框，用于存儲最終數(shù)據(jù)
# 創(chuàng)建數(shù)據(jù)文件夾
if not os.path.exists(self.folder):
os.makedirs(self.folder)- 初始化Screener類時，設(shè)置索引和數(shù)據(jù)文件夾路徑。
- 獲取文件夾中的所有文件名。
- 初始化一個空的DataFrame，用于存儲最終的數(shù)據(jù)。
- 如果數(shù)據(jù)文件夾不存在，則創(chuàng)建該文件夾。
處理單個文件的方法
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) # 轉(zhuǎn)換日期列為日期時間格式
df['Date'] = df['Date'].dt.date # 提取日期部分
current_date = df.iloc[-1]['Date'] # 獲取最新日期
symbol = filename.split()[0] # 獲取股票符號
info = yf.Ticker(symbol).info # 獲取股票信息
# 提取關(guān)鍵財務(wù)數(shù)據(jù)
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']
# 生成數(shù)據(jù)字典
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) # 將數(shù)據(jù)字典轉(zhuǎn)換為數(shù)據(jù)框
self.df_all = pd.concat([self.df_all], axis=0) # 將新的數(shù)據(jù)框添加到總數(shù)據(jù)框中
except Exception as e:
print(f"Error processing file {file_path}: {e}")- 檢查文件是否是Excel或CSV格式且不是臨時文件。
- 讀取CSV文件并轉(zhuǎn)換日期格式。
- 獲取最新的日期和股票符號。
- 使用?
?yfinance?? 獲取股票的財務(wù)信息。 - 提取關(guān)鍵財務(wù)數(shù)據(jù)：市盈率、價格與賬面比、負(fù)債股本比、股息收益率、每股收益增長率和收盤價。
- 生成包含這些數(shù)據(jù)的字典，并將其轉(zhuǎn)換為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?? 方法進(jìn)行處理。
將結(jié)果存儲到CSV的方法
def store_in_csv(self):
# 將最終數(shù)據(jù)存儲到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)- 將最終的數(shù)據(jù)存儲到CSV文件中。
- 檢查存儲文件夾是否存在，如果不存在則創(chuàng)建。
- 生成包含當(dāng)前日期的文件名。
- 將DataFrame保存到CSV文件中，不包含索引。
使用Cohere進(jìn)行AI篩選的方法
def ai_screening(self):
# 使用Cohere進(jìn)行AI篩選
co = Cohere()
message = f"""你將獲得一些股票的數(shù)據(jù)。請分析這些數(shù)據(jù)并篩選出2只被低估且未來有可能產(chǎn)生更高回報的股票。以下是數(shù)據(jù) {self.df_all}"""
preamble = """你是一名優(yōu)秀的股市分析師，能夠很好地理解提供的數(shù)據(jù)并做出謹(jǐn)慎的決策。輸出應(yīng)嚴(yán)格采用以下JSON格式：
[
{
"stock": "股票名稱",
"justification": "篩選或選擇該股票的理由"
},
{
"stock": "股票名稱",
"justification": "篩選或選擇該股票的理由"
}
]"""
result = co.chat_cohere(message=message, preamble=preamble)
print(result)- 使用Cohere API進(jìn)行AI篩選。
- 創(chuàng)建Cohere對象。
- 定義要發(fā)送給Cohere的消息，包含待分析的股票數(shù)據(jù)。
- 定義消息的前言，設(shè)置任務(wù)和輸出格式的上下文。
- 調(diào)用?
?chat_cohere?? 方法發(fā)送消息并接收結(jié)果。 - 打印篩選結(jié)果。
通過以上步驟，我們就可以自動化地完成納斯達(dá)克100指數(shù)成分股的財務(wù)數(shù)據(jù)分析和智能篩選，幫助識別被低估且具有潛在高回報的股票。未來，可以進(jìn)一步優(yōu)化和擴(kuò)展該項目，包括：
增加更多財務(wù)指標(biāo)，動態(tài)調(diào)整和增加篩選財務(wù)指標(biāo)。集成更多數(shù)據(jù)源，豐富數(shù)據(jù)維度，提高分析深度。模型優(yōu)化與更新，比如選用BloomberGPT之類的一些金融行業(yè)的大模型，確保分析結(jié)果的前沿性和準(zhǔn)確性。開發(fā)用戶友好的界面，提供可視化報告和實時篩選結(jié)果展示。
本文轉(zhuǎn)載自公眾號AIGC開發(fā)者，作者：阿橙AIGC
