米国株決算説明会の内容を自動的にまとめて投稿する(Seeking AlphaからEarnings Call Transcriptを自動取得しClaude APIでサマってQiitaに自動投稿)
背景
大した話ではないのですが、やってる人が居なそうなのと、メモがわりです。先に結論を言っておくと、いい感じにEarnings Callの内容をナラティブで情報欠落なく報告してくれるのはClaude 3 OpusをUIから使った時だけ、、、と言う印象です。APIで自動化すると結構品質が落ちます。
Claude APIでClaude3 Opusを使ってEarnings Callをまとめると、料金が1 Callで0.5USDぐらい行ってしまうのと、アウトプットのトークン数制限で引っかかってしまい、複数回に分けて内容をまとめる必要があるのですが、それが原因で出力が正直安定しないです(内容の粒度がぶれたり、全然違うフォーマットで返してきたり)。
一方で、SonnetやHaikuはAPIからだとかなり厳しくて、口酸っぱく箇条書き禁止だとか、日本語で回答しろと言っているのに全部無視してイタリア語でまとめてきたりとか、ほんとなめてんのかお前w
ソースコード全文載せておくので、適当にコピって使いたいところだけ使うと良いかと思います。
Seeking Alphaからのデータ取得API
まずはSeeking Alphaからのデータ取得です。Rapid APIにSeeking Alphaからデータを取得できるAPIがありますので、それを使います。詳しくはこちら：Freemiumである程度使えるので、10社ぐらいしか米国株を見ていない人ならSeeking Alphaを契約なしでこのAPIからTranscript取得するのでも十分じゃないですかね。
Rapid API？
Rapid APIとは、個人から法人まで様々なAPIを集めたプラットフォームで、使いたいAPIをSubscribeするとトークンが払い出され、Rapid APIのRest APIに当該トークンでアクセスしてそのAPIが使える、、、ざっくり言うとそんなAPIです。
Seeking Alpha APIは月間でそこそこな量までは無料で使えます。それを超えると、有料になります。
世界中の凄まじい数のAPIを集めて、それらAPIのゲートウェイみたいな役割をしてるサービスで、いいビジネスですね。大したアイディアじゃないのに・・・・めちゃくちゃ便利です。
アカウントを作りSubscribe
Googleアカウントでシームレスにアカウント作れるので、その上で上記のSeeking AlphaのAPIページでSubscribeすればトークンが払い出されますので、それを使います。
Claude APIを使う
Claude APIを使うにはAPIキーが必要です。適当にググってください。いくらでも情報は落ちてます。
LangChain-Anthropicを使う
今回、LangChain-Anthropicを使っているので、
pip install langchain-anthropicが必要です。
Qiita API？
なぜQiitaかというと、NoteはAPIでの自動投稿に対応してません。そして、マークダウンにもきちんと対応していません。改善する雰囲気もないしひどいもんです。
よって、自動投稿にはマークダウンにしっかり対応していて公式なAPIが提供されているプラットフォームが良いですので、Qiitaとしました。
Pythonコード
結論は冒頭に書いた通りなので、ぶっちゃけ自分でも今後は使わないと思いますが、一部でも参考になれば、自己責任でお願いします。ちなみに、1日にこれで投稿しすぎて制限かかってQiitaに投稿できなくなりましたので、1日に100個とか投稿しないように注意しましょうw
なお、以下のコードは、プロンプトで”箇条書きでまとめろ”と言う記載にしてます。決算説明会の内容を箇条書きさせると情報欠落が発生しやすくなるので基本的にはお勧めはしませんが、箇条書きにしないと出力トークンが4096を超えるため、後半の文章が空っぽになったり、途中でキレたりとかしますので、意図的にこう言う記載にしてます(summarize_transcript_via_gen_aiの関数の中でプロンプト書かれてますので、必要があればここを修正する感じです)。
無駄にコードが長いのはQiita APIのクラスも一緒に貼ってあるからです。Qiita投稿が不要ならこの辺は全部削除でいいでしょうね。
各種キーやトークンをご自分のものにセットすればGoogle Colabでもサクッと動きます(!pip install langchain-anthropic の実行を忘れないように)。
それなりにdocstringも書いてますので、読めばわかるかなと思います。自己責任でご参考ください。
# pip install langchain-anthropicが必要なので先に実行しておくこと
import requests
import re
from datetime import datetime, timedelta
import pandas as pd
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
# 定数の定義
TRANSCRIPT_LIST_URL = "https://seeking-alpha.p.rapidapi.com/transcripts/v2/list"
TRANSCRIPT_DETAIL_URL = "https://seeking-alpha.p.rapidapi.com/transcripts/v2/get-details"
RAPID_API_KEY = "YOUR_KEY"
RAPID_API_HOST = "seeking-alpha.p.rapidapi.com"
CLAUDE_API_KEY = "YOUR_KEY"
QIITA_ACCESS_TOKEN = "YOUR_KEY"
CLAUDE_MODEL = "claude-3-haiku-20240307"
import logging
def setup_logger(log_level=logging.DEBUG):
# ロガーの作成
logger = logging.getLogger(__name__)
logger.setLevel(log_level)
# ロガーに既にハンドラが追加されているかチェックし、ある場合はクリアする
if logger.handlers:
logger.handlers = [] # 既存のハンドラを全てクリア
# ログのフォーマットを設定
formatter = logging.Formatter('%(asctime)s - %(levelname)s: *** %(message)s:')
# コンソールハンドラーを作成
console_handler = logging.StreamHandler()
console_handler.setLevel(log_level)
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
return logger
# ロガーのセットアップ。ログファイル名やフィイルパスは適宜修正してください
logger = setup_logger(log_level=logging.DEBUG)
def get_transcript_list(ticker_symbols, since=None, until=None, top_n=None):
"""
Seeking AlphaのAPIを使用して、指定されたティッカーシンボルのTranscript一覧を取得する関数。
:param ticker_symbols: 取得するティッカーシンボルのリスト。
:param since: 取得する期間の開始日時（Unixタイムスタンプ）。デフォルトは現在時刻の180日前。
:param until: 取得する期間の終了日時（Unixタイムスタンプ）。デフォルトは現在時刻。
:param top_n: 各ティッカーシンボルごとに取得するTranscriptの最大件数。デフォルトは全件。
:return: 取得したTranscriptのリスト。
"""
headers = {
"X-RapidAPI-Key": RAPID_API_KEY,
"X-RapidAPI-Host": RAPID_API_HOST
}
# sinceとuntilが未指定の場合、現在時刻から直近半年間を指定
if since is None:
since = int((datetime.now() - timedelta(days=180)).timestamp())
if until is None:
until = int(datetime.now().timestamp())
transcript_list = []
for symbol in ticker_symbols:
querystring = {"id": symbol, "size": "20", "number": "1", "since": str(since), "until": str(until)} # ここの詳細はRapid APIのSeeking Alphaのページ見てね。
response = requests.get(TRANSCRIPT_LIST_URL, headers=headers, params=querystring, verify=True) # Mac OS対策でSSL証明書の検証を一時オフにしてSSL通信している。問題なければverify=Trueは削除してもOK。
response.raise_for_status()
# 最新のN件だけを取得
data = response.json()["data"]
if top_n is not None:
data = data[:top_n]
# 各Transcriptに対応するティッカーシンボルを追加
for transcript in data:
transcript["ticker"] = symbol
transcript_list.extend(data)
return transcript_list
def get_transcript_detail(transcript_id):
"""
Seeking AlphaのAPIを使用して、指定されたTranscriptの詳細を取得する関数。
:param transcript_id: 取得するTranscriptのID。
:return: 取得したTranscriptの詳細（本文）。
"""
headers = {
"X-RapidAPI-Key": RAPID_API_KEY,
"X-RapidAPI-Host": RAPID_API_HOST
}
querystring = {"id":transcript_id}
response = requests.get(TRANSCRIPT_DETAIL_URL, headers=headers, params=querystring, verify=True)
response.raise_for_status()
return response.json()["data"]["attributes"]["content"]
def summarize_transcript_via_gen_ai(transcript,model="claude-3-sonnet-20240229"):
"""
Anthropic APIを使用して、指定されたTranscriptを要約する関数。
:param transcript: 要約するTranscriptの本文。
:param model: モデル名を指定可能。デフォルトでClaude 3 sonnetにしておく。
:return: 要約されたTranscript。
"""
chat = ChatAnthropic(
temperature=0.5
, model_name=model
, api_key=CLAUDE_API_KEY
,max_tokens=4096
)
system = (
"""
あなたはプロフェッショナル投資アドバイザーです。ユーザープロンプトの内容に従い、Earnings Callの内容を報告してください。投資判断には、内容を割愛・要約して情報が欠落するのは絶対に避けなければなりません。
コンパクトにまとめつつ、重要な要素が漏れないようにしてください。
"""
)
human = """
Earnings Callの内容を投資家向け報告してください。報告結果だけを出力してください。
以下の条件を守ってください。
# 報告内容の要件：
・全体構造： 会社紹介、コールの内容をセクションに分けた目次、各セクションの内容の箇条書き、の順番に記載する。
・フォーマット：マークダウンで記載すること。全体の構成は以下の通り：
- 全体のタイトル: 見出し1. 例="# <対象の四半期> Earings Call Transcriptまとめ"のように、
- 会社紹介: 見出しは常に "## 会社紹介"
- 各セクション: 見出しは常に"## セクション名"、セクション名はCallの内容に応じて変えること
- 質疑応答がある場合、最後のセクションは必ず質疑応答であること(見出し： "## 質疑応答")
- 見出し3(### タイトル)は使わないこと
・会社紹介：事業内容と事業PFとPFごとの市場シェアや売り上げ規模などの数字ベースの内容、市場でのポジショニングなどについて記載すること
・まとめ方について：実績ベースの業績とその根拠、業績予想とその根拠、についてはマーケットの状況や重要なトレンドを漏らさずにまとめること
・質疑応答について： 最後のセクションは質疑応答だが、ここは重要なので漏れがないようにすること。質疑応答は抜粋せず全部記載すること
・箇条書きOKとするが、内容が薄すぎないように可能な限り背景や詳細などを記載すること。投資家の判断材料にするためには、情報は削ぎ落としてまとめるのではなく、詳細を記載した方が良い
・出力は日本語
・質疑応答の各質問は見出しのマークダウンにしないこと
・Earnings Call Transcriptは以下の通りです:
{transcript}
"""
prompt = ChatPromptTemplate.from_messages([("system", system), ("human", human)])
logger.debug(f"Calling Claude API for summarization on former part ..... target transcript is:{transcript[:100]}")
chain = prompt | chat
result = chain.invoke(
{
"input_language": "Japanese",
"output_language": "Japanese",
"transcript": transcript,
}
)
logger.debug(f"Done: Calling Claude API for summarization on former part ..... target transcript is:{transcript[:100]}")
# 決算の前半部分
kessan_text = result.content
kessan_text = kessan_text+"""
"""
# prompt = ChatPromptTemplate.from_messages([("system", system), ("human", human),("assistant",result.content), ("human", "残り全てを出力してください。要件はこれまでと同等です。繰り返しになりますが、報告内容は基本的にナラティブとし、箇条書きは禁止です。")])
# chain = prompt | chat
# logger.debug(f"Calling Claude API for summarization on latter part ..... target transcript is:{transcript[:100]}")
# result = chain.invoke(
# {
# "input_language": "Japanese",
# "output_language": "Japanese",
# "transcript": transcript,
# }
# )
# logger.debug(f"Done: Calling Claude API for summarization on latter part ..... target transcript is:{transcript[:100]}")
# # 決算の後半部分
# latter_text = result.content
# # 決算報告マージ
# kessan_text = former_text+"¥n¥n"+latter_text
return kessan_text
def remove_html_tags(text):
"""
指定された文字列からHTMLタグを除去する関数。
:param text: HTMLタグを除去する文字列。
:return: HTMLタグが除去された文字列。
"""
return re.sub('<.*?>', '', text)
def process_transcripts(ticker_symbols, since=None, until=None, top_n=None, require_summary=False,model="claude-3-sonnet-20240229"):
"""
Transcriptを処理する関数。
指定されたティッカーシンボルのTranscriptを取得し、要約の要否に応じて処理してデータフレームに格納する。
:param ticker_symbols: 取得するティッカーシンボルのリスト。
:param since: 取得する期間の開始日時（Unixタイムスタンプ）。デフォルトは現在時刻の180日前。
:param until: 取得する期間の終了日時（Unixタイムスタンプ）。デフォルトは現在時刻。
:param top_n: 各ティッカーシンボルごとに取得するTranscriptの最大件数。デフォルトは全件。
:param require_summary: 要約の要否を指定するフラグ。Trueの場合は要約を行う。デフォルトはFalse。
:return: None
"""
# Transcript一覧を取得し、publishOnの降順でソート
logger.debug(f"Getting transcripts: ticker symbols={', '.join(ticker_symbols)}")
transcript_list = get_transcript_list(ticker_symbols, since, until, top_n)
transcript_list.sort(key=lambda x: datetime.strptime(x["attributes"]["publishOn"], "%Y-%m-%dT%H:%M:%S%z"), reverse=True)
logger.debug(f"Done:Getting transcripts: ticker symbols={', '.join(ticker_symbols)}, number of transcripts={len(transcript_list)}")
# データフレームを作成
columns = ["ticker", "publish_on", "transcript"]
if require_summary:
columns.append("summary")
df = pd.DataFrame(columns=columns)
for loop,transcript in enumerate(transcript_list):
# TranscriptのIDとURLを取得
transcript_id = transcript["id"]
transcript_url = transcript["links"]["self"]
ticker=transcript["ticker"]
# Transcriptの詳細を取得
transcript_detail = get_transcript_detail(transcript_id)
# HTMLタグを除去
transcript_detail = remove_html_tags(transcript_detail)
publish_on = datetime.strptime(transcript["attributes"]["publishOn"], "%Y-%m-%dT%H:%M:%S%z").strftime("%Y-%m-%d")
# 要約の要否に応じて処理
logger.debug(f"Summarizing transcript of ticker:{ticker} with loop num={loop+1}")
if require_summary:
summary = summarize_transcript_via_gen_ai(transcript_detail, model="claude-3-sonnet-20240229")
data = {"ticker": [ticker], "publish_on": [publish_on], "transcript": [transcript_detail], "summary": [summary]}
else:
data = {"ticker": [ticker], "publish_on": [publish_on], "transcript": [transcript_detail], "summary": None}
import time
seconds=30
logger.debug(f"sleeping {seconds}seconds to prevent rate limit exeed")
time.sleep(seconds) # seconds秒間スリープします
logger.debug(f"Done:summarizing transcript of ticker:{ticker} with loop num={loop+1}")
# データフレームに行を追加
logger.debug(f"Converting transcript of ticker:{ticker} with loop num={loop+1}")
df = pd.concat([df, pd.DataFrame(data)], ignore_index=True)
logger.debug(f"Done:Converting transcript of ticker:{ticker} with loop num={loop+1}")
# df.reset_index(inplace=True)
return df
def export_to_csv(df, file_path):
"""
DataFrameをCSV形式でファイルに出力する関数。
:param df: 出力するDataFrame。
:param file_path: 出力先のファイルパス。
"""
try:
# DataFrameをCSV形式で出力
df.to_csv(file_path, index=False, quoting=3, escapechar='\\')
logger.info(f"DataFrameを正常にCSV形式で出力しました: {file_path}")
except Exception as e:
logger.error(f"DataFrameのCSV出力に失敗しました: {str(e)}")
# Qiita投稿用クラス
class QiitaAPIClient:
"""
Qiita API v2を操作するためのクライアントクラスです。
Attributes:
BASE_URL (str): Qiita APIのベースURL。
access_token (str): Qiita APIにアクセスするためのアクセストークン。
headers (dict): APIリクエストに含めるHTTPヘッダー。
Methods:
get(endpoint, params=None): GETリクエストを送信します。
post(endpoint, data=None): POSTリクエストを送信します。
put(endpoint, data=None): PUTリクエストを送信します。
delete(endpoint): DELETEリクエストを送信します。
get_authenticated_user(): 認証中のユーザー情報を取得します。
list_items(params=None): 記事一覧を取得します。
create_item(data): 新しい記事を作成します。
get_item(item_id): 指定した記事を取得します。
update_item(item_id, data): 指定した記事を更新します。
delete_item(item_id): 指定した記事を削除します。
list_tags(params=None): タグ一覧を取得します。
get_tag(tag_id): 指定したタグを取得します。
list_users(params=None): ユーザー一覧を取得します。
get_user(user_id): 指定したユーザーを取得します。
"""
BASE_URL = "https://qiita.com/api/v2"
def __init__(self, access_token):
"""
QiitaAPIClientを初期化します。
Args:
access_token (str): Qiita APIにアクセスするためのアクセストークン。
"""
self.access_token = access_token
self.headers = {"Authorization": f"Bearer {self.access_token}"}
def get(self, endpoint, params=None):
"""
GETリクエストを送信します。
Args:
endpoint (str): APIエンドポイント。
params (dict, optional): リクエストパラメータ。
Returns:
dict: APIレスポンスのJSONデータ。
"""
url = f"{self.BASE_URL}/{endpoint}"
response = requests.get(url, headers=self.headers, params=params, verify=True)
response.raise_for_status()
return response.json()
def post(self, endpoint, data=None):
"""
POSTリクエストを送信します。
Args:
endpoint (str): APIエンドポイント。
data (dict, optional): リクエストボディのJSONデータ。
Returns:
dict: APIレスポンスのJSONデータ。
"""
url = f"{self.BASE_URL}/{endpoint}"
response = requests.post(url, headers=self.headers, json=data, verify=True)
response.raise_for_status()
return response.json()
def put(self, endpoint, data=None):
"""
PUTリクエストを送信します。
Args:
endpoint (str): APIエンドポイント。
data (dict, optional): リクエストボディのJSONデータ。
Returns:
dict: APIレスポンスのJSONデータ。
"""
url = f"{self.BASE_URL}/{endpoint}"
response = requests.put(url, headers=self.headers, json=data, verify=True)
response.raise_for_status()
return response.json()
def delete(self, endpoint):
"""
DELETEリクエストを送信します。
Args:
endpoint (str): APIエンドポイント。
Returns:
dict: APIレスポンスのJSONデータ。
"""
url = f"{self.BASE_URL}/{endpoint}"
response = requests.delete(url, headers=self.headers)
response.raise_for_status()
return response.json()
def get_authenticated_user(self):
"""
認証中のユーザー情報を取得します。
Returns:
dict: 認証中のユーザー情報。
"""
return self.get("authenticated_user")
def list_items(self, params=None):
"""
記事一覧を取得します。
Args:
params (dict, optional): リクエストパラメータ。
Returns:
list: 記事一覧。
"""
return self.get("items", params=params)
def create_item(self, title="test title", body="test body text"):
"""
新しい記事を作成します。
Args:
data (dict): 記事の作成に必要なデータ。
Returns:
dict: 作成された記事の情報。
"""
data = {
"title": title,
"body": body,
"tags": [{"name": "Python"}, {"name": "API"}],
"private": True
}
return self.post("items", data=data)
def get_item(self, item_id):
"""
指定した記事を取得します。
Args:
item_id (str): 取得する記事のID。
Returns:
dict: 記事の情報。
"""
return self.get(f"items/{item_id}")
def update_item(self, item_id, data):
"""
指定した記事を更新します。
Args:
item_id (str): 更新する記事のID。
data (dict): 記事の更新に必要なデータ。
Returns:
dict: 更新された記事の情報。
"""
return self.patch(f"items/{item_id}", data=data)
def delete_item(self, item_id):
"""
指定した記事を削除します。
Args:
item_id (str): 削除する記事のID。
Returns:
dict: 削除された記事の情報。
"""
return self.delete(f"items/{item_id}")
def list_tags(self, params=None):
"""
タグ一覧を取得します。
Args:
params (dict, optional): リクエストパラメータ。
Returns:
list: タグ一覧。
"""
return self.get("tags", params=params)
def get_tag(self, tag_id):
"""
指定したタグを取得します。
Args:
tag_id (str): 取得するタグのID。
Returns:
dict: タグの情報。
"""
return self.get(f"tags/{tag_id}")
def list_users(self, params=None):
"""
ユーザー一覧を取得します。
Args:
params (dict, optional): リクエストパラメータ。
Returns:
list: ユーザー一覧。
"""
return self.get("users", params=params)
def get_user(self, user_id):
"""
指定したユーザーを取得します。
Args:
user_id (str): 取得するユーザーのID。
Returns:
dict: ユーザーの情報。
"""
return self.get(f"users/{user_id}")
ticker_symbols = ["ARM","CRTO"] # 取得したいティッカーシンボルのリスト。適宜変更すること。見ての通り配列形式
transcript_df = process_transcripts(ticker_symbols, top_n=1, require_summary=True,model=CLAUDE_MODEL)
# 各ティッカーシンボルごとに記事を投稿
client = QiitaAPIClient(QIITA_ACCESS_TOKEN)
# ティッカーシンボルごとにトランスクリプトをマージする
grouped = transcript_df.groupby('ticker')['summary'].apply("""
""".join).reset_index()
# マージしたトランスクリプトに対してQiitaに記事を作成する。
for _, row in grouped.iterrows():
logger.debug(f"Creating Qiita items with ticker:{row['ticker']}")
# 記事を作成
client.create_item(title=row['ticker']+" Earnings Call Transcriptまとめ("+CLAUDE_MODEL+"を用いた自動投稿)", body=row['summary'])
logger.debug(f"Done: Creating Qiita items with ticker:{row['ticker']}")
logger.debug(f"Everything is OK.")この記事が気に入ったらサポートをしてみませんか？