【Python金融データ】AlphaVantageの使い方(第2回)

投資・ファイナンス

前回は基本的なAlphaVantageの使い方として、APIの取得方法や、pandas_datareaderを使ったデータ取得、APIを直接使ったデータ取得の方法を紹介しました。今回は、様々なデータをダウンロードしていくための汎用的な関数を作成し、すべてのデータを簡単に取得できるようにしていきます。
前回の記事を見ていない場合は、前回の記事も参考にしてください。

全体のコードはこちらのGoogleColabに載せています。

コードの整理

基本的にAlpha Vantageのデータ取得では、エンドポイントとなるURLをrequestで呼び、結果として、jsonを得ます。
実際の分析では、jsonよりpandas dataframeの方が使いやすいことも多いので、エンドポイントurlを入力するとデータフレームを返す関数を作成します。
また、ドキュメントには各APIごとに、どんな結果が返るか試すことのできるデモ用のエンドポイントが用意されていますので、どのようなデータが得られるか試したい場合にはそちらを利用すると、制限などに引っかからなくなります。

コード全体

import requests
import pandas as pd
from io import StringIO

def get_data(function, api_key, params=None, is_df=True, df_name=None, csv=False):
    # ベースURL
    base_url = f'https://www.alphavantage.co/query?function={function}'

    # パラメータを追加
    if params:
        param_list = []
        for key, value in params.items():
            if isinstance(value, list):
                for v in value:
                    param_list.append(f'{key}={v}')
            else:
                param_list.append(f'{key}={value}')

        param_str = '&'.join(param_list)
        url = f'{base_url}&{param_str}'
    else:
        url = base_url

    url += f'&apikey={api_key}'
    print(url)
    if csv:
        with requests.Session() as s:
            download = s.get(url)
            decoded_content = download.content.decode('utf-8')

            # StringIOを使ってPandasのデータフレームに変換
            df = pd.read_csv(StringIO(decoded_content))
        return df

    r = requests.get(url)
    data = r.json()

    if is_df:
        if df_name and df_name in data:
            df_data = data[df_name]
        else:
            # 最も長いリストを持つキーを検索
            max_len = 0
            df_data = None
            for key, value in data.items():
                if isinstance(value, list) and len(value) > max_len:
                    max_len = len(value)
                    df_data = value
        if df_data:
            df = pd.DataFrame(df_data)
        else:
            df = pd.DataFrame([data])  

        return df
    else:
        return data
# 使用例
function = 'SYMBOL_SEARCH'
params = {'keywords': 'tesco'}
df = get_data(function, api_key, params)
df

functionはAPIドキュメントで、functionと記載されている部分、paramsはそれ以外のパラメータの値を辞書で渡します。
引数

以下に詳細を解説していきます。

get_data 関数の定義

import requests
import pandas as pd
from io import StringIO

def get_data(function, api_key, params=None, is_df=True, df_name=None,csv=False):
  • function: Alpha Vantage API の機能（例えば、SYMBOL_SEARCH）。
  • api_key: API キー。
  • params: キーワードやその他のパラメータを含む辞書。
  • is_df: 返されたデータを DataFrame に変換するかどうかを指定するブール値。
  • df_name: DataFrame に変換する特定のキー（オプション）。
  • csv: csv形式で提供されているデータの場合はこちらをTrueに。
    df_nameはJSONの中でどの情報をデータフレームに変換するか指定する変数です。
    例えば、先ほどのTicker検索の場合は、
{'bestMatches': [{'1. symbol': 'SCT.LIS',
   '2. name': 'Toyota Caetano',
   '3. type': 'Equity',
   '4. region': 'Lisbon',
   '5. marketOpen': '08:00',
   '6. marketClose': '16:30',

なので、df_name='bestMatches'としたいです。
市場のOpen/Closeはmarketsとしたいです。
基本的には、一番データの多い部分をデータフレームに変換したいので、指定なしNoneで最大のデータを含む部分をデータフレームに変換します。
想定した部分がデータフレームにならない場合は、この引数を渡します。

URLの作成

    base_url = f'https://www.alphavantage.co/query?function={function}'

    # パラメータを追加
    if params:
        param_list = []
        for key, value in params.items():
            if isinstance(value, list):
                for v in value:
                    param_list.append(f'{key}={v}')
            else:
                param_list.append(f'{key}={value}')

        param_str = '&'.join(param_list)
        url = f'{base_url}&{param_str}'
    else:
        url = base_url

    url += f'&apikey={api_key}'
  • base_url は基本的なリクエストURLを構成します。
  • params が指定されている場合、それを URL に追加します。
  • 最後に、API キーを URL に追加します。
    paramsはfunctionごとに異なるパラメータの値をdictで渡して、URL
    URLに入れ込むための変数です。後半の使用例で使い方を確認してみてください。

データの取得

一部のデータはjsonではなく、csv形式で提供されています。それらに対応するためにcsvがTrueの場合は、ダウンロード方法をjsonの場合とは別に記載します。

if csv:
        with requests.Session() as s:
            download = s.get(url)
            decoded_content = download.content.decode('utf-8')

            # StringIOを使ってPandasのデータフレームに変換
            df = pd.read_csv(StringIO(decoded_content))
        return df

jsonの場合は以下で読み込みます。成形については次の項で説明します。

    r = requests.get(url)
    data = r.json()
  • 生成したURL を使ってリクエストを送信し、JSON 形式に変換します。

データの処理

    if is_df:
        if df_name and df_name in data:
            df_data = data[df_name]
        else:
            # 最も長いリストを持つキーを検索
            max_len = 0
            df_data = None
            for key, value in data.items():
                if isinstance(value, list) and len(value) > max_len:
                    max_len = len(value)
                    df_data = value
        if df_data:
            df = pd.DataFrame(df_data)
        else:
            df = pd.DataFrame()  

        return df
    else:
        return data
  • is_df が True の場合、データを DataFrame に変換する処理を行います。
  • df_name が指定され、そのキーがデータに存在する場合、そのキーのデータを使用します。
  • df_name が指定されていない場合、最も長いリストを持つキーのデータを検索します。

使用例

function = 'SYMBOL_SEARCH'
api_key = 'demo'
params = {'keywords': 'tesco'}
df = get_data(function, api_key, params)
df
  • function, api_key, params を指定して get_data 関数を呼び出します。
  • 取得したデータを DataFrame に変換し、それを df に格納して表示します。

さて、今回必要なコードを整理できたので、そのほかの取得可能データについても最後に次の記事で見ていきます。

タイトルとURLをコピーしました