金融达人必备：教你用Python轻松简化数据分析流程！

作者：老余捞鱼

原创不易，转载请标明出处及原作者。

写在前面的话：今天我想和大家分享一篇关于如何利用FMP API和Python来简化金融数据分析流程的文章。本文将详细介绍如何运用Python，结合FMP API的强大功能，高效地处理和分析SP500指数公司的财务数据。从获取股票价格、财务报表，到分析关键财务指标，每一步都将通过实际案例为您一一展现。以期提高工作效率，并在投资决策中发挥更大的作用。

一、数据准备

我使用 FMP API 已经有一段时间了，它的易用性、交互友好以及独特的端点产品，使其成为任何想要访问高级金融数据的程序员、分析师或数据专家的绝佳选择。

地址：https://site.financialmodelingprep.com/

而本文中的编程主要包括面向对象编程（Object-Oriented Programming），简称 OOP。OOP是一种代码组织方式，它将数据和功能（即属性和方法）封装在一起，便于管理和使用。这种方式极大地便利了我使用FMP API的过程，因为我能够通过操作这个“对象”来轻松获取所需数据或进行数据分析。如果对象缺少某些功能，我还可以随时为其添加方法，从此告别了编写额外函数的烦恼！

1.1 创建数据类

我创建的对象名为 SP500data 类。该对象与 FMP API 相结合，可以导入 SP500 指数中最新公司的价格、财务报表、概况数据和关键财务指标。我还计划添加更多方法，以导入更多形式的数据并快速进行分析。让我们来看看这个对象的初始化函数以及它所需要的 Python 库。

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
import pandas as pd
from tqdm import tqdm
import time

class SP500data:
    def __init__(self, api_key):
        self.api_key = api_key
        
        ## Symbols and Profiles
        self.symbol_list = []
        self.company_profile_df = None
        
        ## Price
        self.price_data = {}

        ## Financials
        self.income_statement_data_annual = {}
        self.balance_sheet_data_annual = {}
        self.cash_flow_data_annual = {}
        self.key_metrics_data_annual = {}
        self.income_statement_data_quarter = {}
        self.balance_sheet_data_quarter = {}
        self.cash_flow_data_quarter = {}
        self.key_metrics_data_quarter = {}

如您所见，代码中需要提供的唯一变量就是您自己的 FMP API。

1.2 获取标志和公司简介

def get_symbols(self):
        url = f'https://financialmodelingprep.com/api/v3/sp500_constituent?apikey={self.api_key}'
        stocks = requests.get(url).json()
        self.symbol_list = [stock['symbol'] for stock in stocks]

get_symbols 方法将通过 sp500_constituent 端点为我们检索这些符号。

获取公司简介和我计划导入的其他数据可能会很耗时。此外，导入这些数据的顺序并不重要。考虑到这两点，我创建了一套利用并行编程导入数据的方法。并行编程指示计算机将多个工作负载分散到可用的处理器上同时运行。

def fetch_profile(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/profile/{symbol}?apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

def fetch_profile_data(self):
        profile_dfs = []
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_profile, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    single_df = future.result()
                    profile_dfs.append(single_df)
                except Exception as exc:
                    print(f'Exception for {future_to_symbol[future]}: {exc}')
        self.company_profile_df = pd.concat(profile_dfs, ignore_index=True)

1.3 获取价格、财务报表和关键指标

使用方法导入以下内容：

• 每家公司最近五年的 OHLC 价格数据；
• 收入、资产负债表和现金流量表数据历史数据；
• 基于财务报表数据的关键财务指标。

当您运用这些方法后，您的对象将变成一个充满财务数据的宝库。比如，您想要展示一家公司关键利润率的历史数据，或者更上一层楼，您希望同时展示多家公司的数据以进行对比分析！只需简单调用这些方法，一行代码即可让您的实例获取所有必需的数据。另外，值得注意的是，API也能够检索季度和年度的财务报表，满足您不同的数据需求。

    def fetch_price(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/historical-price-full/{symbol}?apikey={self.api_key}'
        response = requests.get(url).json()
        prices_df = pd.DataFrame(response['historical'])
        prices_df = prices_df.sort_values(by='date').reset_index(drop=True)
        return prices_df

    def fetch_price_data(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_price, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    prices_df = future.result()
                    self.price_data[symbol] = prices_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_income_statement_annual(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/income-statement/{symbol}?period=annual&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_income_statement_quarter(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/income-statement/{symbol}?period=quarter&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_income_statement_data_annual_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_income_statement_annual, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    income_df = future.result()
                    self.income_statement_data_annual[symbol] = income_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_income_statement_data_quarter_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_income_statement_quarter, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    income_df = future.result()
                    self.income_statement_data_quarter[symbol] = income_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')


    def fetch_balance_sheet_annual(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/balance-sheet-statement/{symbol}?period=annual&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_balance_sheet_quarter(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/balance-sheet-statement/{symbol}?period=quarter&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_balance_sheet_data_annual_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_balance_sheet_annual, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    balance_df = future.result()
                    self.balance_sheet_data_annual[symbol] = balance_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_balance_sheet_data_quarter_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_balance_sheet_quarter, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    balance_df = future.result()
                    self.balance_sheet_data_quarter[symbol] = balance_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_cash_flow_annual(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/cash-flow-statement/{symbol}?period=annual&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_cash_flow_quarter(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/cash-flow-statement/{symbol}?period=quarter&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_cash_flow_data_annual_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_cash_flow_annual, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    cash_flow_df = future.result()
                    self.cash_flow_data_annual[symbol] = cash_flow_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_cash_flow_data_quarter_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_cash_flow_quarter, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    cash_flow_df = future.result()
                    self.cash_flow_data_quarter[symbol] = cash_flow_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_key_metrics_annual(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/key-metrics/{symbol}?period=annual&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_key_metrics_quarter(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/key-metrics/{symbol}?period=quarter&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_key_metrics_data_annual_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_key_metrics_annual, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    key_metrics_df = future.result()
                    self.key_metrics_data_annual[symbol] = key_metrics_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_key_metrics_data_quarter_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_key_metrics_quarter, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    key_metrics_df = future.result()
                    self.key_metrics_data_quarter[symbol] = key_metrics_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

1.4 用一行代码获取所有数据

至此，我已经编写了一系列方法，如果要进行详细分析，我可能需要这些方法获取的所有数据。为此，我创建了一个名为 fetch_all_data 的方法，它负责调用之前我所定义的所有函数。我还加入了一个延迟变量，允许最终用户根据他们的API调用限制进行调整。以FMP API的高级计划为例，它允许每分钟最多750次调用。尽管这个限制对我来说已经相当宽裕，但这种方法有可能在短时间内发起数千次API请求，因此我们需要适当设置延迟时间。值得庆幸的是，一旦数据提取完成，对象的实例就将包含我们所需的一切数据！

   def fetch_all_data(self,delay):
        print("Fetching symbols...")
        self.get_symbols()

        print("Fetching company profiles...")
        self.fetch_profile_data()

        print(f"Waiting for {delay} seconds to comply with API rate limits...")
        time.sleep(delay)

        print("Fetching price data...")
        self.fetch_price_data()

        print(f"Waiting for {delay} seconds to comply with API rate limits...")
        time.sleep(delay)

        print("Fetching annual income statements...")
        self.fetch_income_statement_data_annual_concurrent()

        print(f"Waiting for {delay} seconds to comply with API rate limits...")
        time.sleep(delay)

        print("Fetching quarterly income statements...")
        self.fetch_income_statement_data_quarter_concurrent()

        print(f"Waiting for {delay} seconds to comply with API rate limits...")
        time.sleep(delay)

        print("Fetching annual balance sheets...")
        self.fetch_balance_sheet_data_annual_concurrent()

        print(f"Waiting for {delay} seconds to comply with API rate limits...")
        time.sleep(delay)

        print("Fetching quarterly balance sheets...")
        self.fetch_balance_sheet_data_quarter_concurrent()

        print(f"Waiting for {delay} seconds to comply with API rate limits...")
        time.sleep(delay)

        print("Fetching annual cash flow statements...")
        self.fetch_cash_flow_data_annual_concurrent()

        print(f"Waiting for {delay} seconds to comply with API rate limits...")
        time.sleep(delay)

        print("Fetching quarterly cash flow statements...")
        self.fetch_cash_flow_data_quarter_concurrent()

        print(f"Waiting for {delay} seconds to comply with API rate limits...")
        time.sleep(delay)

        print("Fetching annual key metrics...")
        self.fetch_key_metrics_data_annual_concurrent()

        print(f"Waiting for {delay} seconds to comply with API rate limits...")
        time.sleep(delay)

        print("Fetching quarterly key metrics...")
        self.fetch_key_metrics_data_quarter_concurrent()

二、实用案例

我最近读完了第二遍本杰明-格雷厄姆（Benjamin Graham）的名著《聪明的投资者》（The Intelligent Investor）。从这本书中得到了灵感，我想根据格雷厄姆数（Graham Number），按行业和部门组合检索出一份折扣最大的股票列表。首先，让我们创建一个对象实例，然后调用 fetch_all_data 方法。

api_key = 'your_api_key_here'
data_object = SP500data(api_key = api_key)
data_object.fetch_all_data(delay = 45)

我们需要将公司概况数据帧中的行业和部门列合并成一列新列。这样，我们就可以按照各自的行业和产业组合对股票进行快速分组。创建一个名为 profile_df 的单独数据帧，它是我们当前数据帧的副本。从关键指标数据中检索格雷厄姆数字，并将其添加到 profile_df。

最后，我们将创建一个新指标：最新股价与格雷厄姆数值之间的差额。请注意，根据本杰明-格雷厄姆的逻辑，如果该数字为负数，则表示股票目前以折扣价交易。

profile_df = data_object.company_profile_df

profile_df['sector_industry'] = profile_df['sector'] + '_' + profile_df['industry']
profile_df['Graham_Number'] = 0

for symbol in data_object.symbol_list:
    gn = data_object.key_metrics_data_quarter[symbol].tail(1)['grahamNumber'].values[0]
    profile_df.loc[profile_df['symbol'] == symbol,'Graham_Number'] = gn

profile_df['price_gn_delta'] = profile_df['price'] / profile_df['Graham_Number'] - 1

创建一个名为 discount_df 的新数据框，其中将包含每个行业和产业组合中折价率最高的股票。

discount_df = profile_df.groupby('sector_industry').apply(
    lambda x: x.sort_values('price_gn_delta').head(1)).reset_index(drop=True)

discount_df.sort_values(by = 'price_gn_delta')

三、观点总结

• FMP API 是一个强大的工具，可以帮助开发者、分析师和数据专业人士轻松访问高级金融数据。
• 面向对象编程（OOP） 是组织和重用代码的有效方法，特别是在处理复杂数据导入和分析任务时。
• 并行编程 可以显著提高数据处理速度，尤其是在处理大量数据时。
• 创建一个全面的数据类（如 SP500data）可以极大地简化数据获取和分析流程。
• 遵守 API 速率限制 是使用任何 API 时的重要考虑因素，以避免被限制调用。
• 实用案例的提供 有助于展示如何将理论知识应用于实际问题解决中。
• 持续更新和改进代码，添加新方法和属性，以适应不断变化的数据需求和分析技术。

本文完整源代码如下：

Libraries

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
import pandas as pd
import numpy as np
from tqdm import tqdm
import time
import matplotlib.pyplot as plt
import seaborn as sns
import re
from docx import Document
from docx.shared import Inches
from docx.shared import RGBColor
     
Data Class

class SP500data:
    def __init__(self, api_key):
        self.api_key = api_key

        ## Symbols and Profiles
        self.symbol_list = []
        self.company_profile_df = None

        ## Price
        self.price_data = {}

        ## Dividend Data
        self.dividend_data = {}
        self.Gordon_Growth_Model_df = None

        ## Financials
        self.income_statement_data_annual = {}
        self.balance_sheet_data_annual = {}
        self.cash_flow_data_annual = {}
        self.key_metrics_data_annual = {}
        self.income_statement_data_quarter = {}
        self.balance_sheet_data_quarter = {}
        self.cash_flow_data_quarter = {}
        self.key_metrics_data_quarter = {}

        ## Transcripts
        self.transcript_dates = {}

    def get_symbols(self):
        url = f'https://financialmodelingprep.com/api/v3/sp500_constituent?apikey={self.api_key}'
        stocks = requests.get(url).json()
        self.symbol_list = [stock['symbol'] for stock in stocks]

    def fetch_profile(self, symbol):
            url = f'https://financialmodelingprep.com/api/v3/profile/{symbol}?apikey={self.api_key}'
            response = requests.get(url).json()
            return pd.DataFrame(response)

    def fetch_profile_data(self):
            profile_dfs = []
            with ThreadPoolExecutor(max_workers=10) as executor:
                future_to_symbol = {executor.submit(self.fetch_profile, symbol): symbol for symbol in self.symbol_list}
                for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                    try:
                        single_df = future.result()
                        profile_dfs.append(single_df)
                    except Exception as exc:
                        print(f'Exception for {future_to_symbol[future]}: {exc}')
            self.company_profile_df = pd.concat(profile_dfs, ignore_index=True)

    def fetch_price(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/historical-price-full/{symbol}?apikey={self.api_key}'
        response = requests.get(url).json()
        prices_df = pd.DataFrame(response['historical'])
        prices_df = prices_df.sort_values(by='date').reset_index(drop=True)
        return prices_df

    def fetch_price_data(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_price, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    prices_df = future.result()
                    self.price_data[symbol] = prices_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_income_statement_annual(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/income-statement/{symbol}?period=annual&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_income_statement_quarter(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/income-statement/{symbol}?period=quarter&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_income_statement_data_annual_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_income_statement_annual, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    income_df = future.result()
                    self.income_statement_data_annual[symbol] = income_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_income_statement_data_quarter_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_income_statement_quarter, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    income_df = future.result()
                    self.income_statement_data_quarter[symbol] = income_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_balance_sheet_annual(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/balance-sheet-statement/{symbol}?period=annual&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_balance_sheet_quarter(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/balance-sheet-statement/{symbol}?period=quarter&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_balance_sheet_data_annual_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_balance_sheet_annual, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    balance_df = future.result()
                    self.balance_sheet_data_annual[symbol] = balance_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_balance_sheet_data_quarter_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_balance_sheet_quarter, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    balance_df = future.result()
                    self.balance_sheet_data_quarter[symbol] = balance_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_cash_flow_annual(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/cash-flow-statement/{symbol}?period=annual&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_cash_flow_quarter(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/cash-flow-statement/{symbol}?period=quarter&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_cash_flow_data_annual_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_cash_flow_annual, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    cash_flow_df = future.result()
                    self.cash_flow_data_annual[symbol] = cash_flow_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_cash_flow_data_quarter_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_cash_flow_quarter, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    cash_flow_df = future.result()
                    self.cash_flow_data_quarter[symbol] = cash_flow_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_key_metrics_annual(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/key-metrics/{symbol}?period=annual&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_key_metrics_quarter(self, symbol):
        url = f'https://financialmodelingprep.com/api/v3/key-metrics/{symbol}?period=quarter&apikey={self.api_key}'
        response = requests.get(url).json()
        return pd.DataFrame(response)

    def fetch_key_metrics_data_annual_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_key_metrics_annual, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    key_metrics_df = future.result()
                    self.key_metrics_data_annual[symbol] = key_metrics_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_key_metrics_data_quarter_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_key_metrics_quarter, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    key_metrics_df = future.result()
                    self.key_metrics_data_quarter[symbol] = key_metrics_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_dividend_data(self, symbol):
        url = f"https://financialmodelingprep.com/api/v3/historical-price-full/stock_dividend/{symbol}?apikey={self.api_key}"
        response = requests.get(url).json()
        if 'historical' in response:
            div_df = pd.DataFrame(response['historical'])
            div_df = div_df.sort_values(by='date').reset_index(drop=True)
            div_df['year'] = pd.to_datetime(div_df['date']).dt.year
            return div_df
        else:
            return pd.DataFrame()

    def fetch_dividend_data_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_dividend_data, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    key_metrics_df = future.result()
                    self.dividend_data[symbol] = key_metrics_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_transcript_dates(self,symbol):
        url = f'https://financialmodelingprep.com/api/v4/earning_call_transcript?symbol={symbol}&apikey={self.api_key}'
        response = requests.get(url).json()
        transcript_df = pd.DataFrame(response)
        return transcript_df

    def fetch_transcript_dates_concurrent(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            future_to_symbol = {executor.submit(self.fetch_transcript_dates, symbol): symbol for symbol in self.symbol_list}
            for future in tqdm(as_completed(future_to_symbol), total=len(self.symbol_list)):
                try:
                    symbol = future_to_symbol[future]
                    transcript_df = future.result()
                    self.transcript_dates[symbol] = transcript_df
                except Exception as exc:
                    print(f'Exception for {symbol}: {exc}')

    def fetch_individual_stock_transcripts(self,symbol):
        transcript_dates_df = self.transcript_dates[symbol]
        transcripts = []

        for row in transcript_dates_df.index:
            q = transcript_dates_df.loc[row][0]
            y = transcript_dates_df.loc[row][1]
            url = f'https://financialmodelingprep.com/api/v3/earning_call_transcript/{symbol}?year={y}&quarter={q}&apikey={api_key}'
            response = requests.get(url).json()
            transcripts.append(response[0]['content'])

        transcript_dates_df['transcripts'] = transcripts
        return transcript_dates_df
     

感谢您阅读到最后，希望本文能给您带来新的收获。祝您投资顺利！如果对文中的内容有任何疑问，请给我留言，必复。

本文内容仅限技术探讨和学习，不构成任何投资建议。