Claude Agent SDK で自分用株式分析 Agent を作ってみた

こんにちは、Ai Workforce 事業部でテクニカルプロジェクトマネージャーをしている Joe です。

この記事はLayerX AI エージェントブログリレー、26日目の記事です。昨日は同じAi Workforce事業部のFDEグループ藤田さんによる Agentic WorkflowをAgentと共に構築するために：ValidatorとLSPで支えるAI協働開発でした。

ChatGPT に株価の分析をしてもらったり生成AIで株式投資の分析が簡単になったと思いますが、今回自分用の Agent が欲しくなり、自分用の株式分析 AI Agent を作成してみました。

先に実際のターミナルでの操作を画像で添付します。

分析させてみた結果は最後に紹介をしますがターミナル上で起動した Agent に対して期間を指定すると株価を取得して分析してくれます。

※今回作成した AI Agent はあくまで分析のために利用したもので投資の売買などを推奨するものではありませんのでご注意ください。

今回の AI Agent 開発に使った技術スタックを紹介します。

技術スタック

Python 3.13
Claude Agent SDK
yfinance
Claude Code
Codex CLI

加えて、必要に応じて numpy などを利用しています。調査部分においては Codex が活躍してくれ、実装フェーズでは Claude Code が活躍してくれました。また、今回の開発では個人の ChatGPT Plus と Claude Pro で十分に開発することができ、どちらか片方でも可能です。

Claude Agent SDK と Claude Code を使うため利用量の制限にかかりやすい傾向はありますのでその点は注意してご利用ください。

Claude Agent SDK の解説についてはバクラク勤怠に所属している @upamune さんが詳しく解説してくれていますので是非こちらもご覧ください。リミットにかからない限りは定額で利用できるので安心して利用できます。

Claude Code SDK ではじめる定額 AI Agent 開発入門 - LayerX エンジニアブログ

Claude Code SDK からの Claude Agent SDK への移行でAI Agentのポータビリティを高める - LayerX エンジニアブログ

実装の中身

今回は以下のようなフローで分析を行っています。

Claude Agent SDKクライアントがリクエストを受け取りフェーズに応じて適切な MCP ツールを選択し、分析結果を返してくれるものです。

シーケンスの中でMCPサーバーとありますが、実際には３つのツールを作成したので次にそのツールの紹介をします。

３つの MCP Tool

今回の開発において３つのMCP Tool を作成しました。

1. fetch_latest_nikkei_data

yfinance を利用して直近最後の株価の始値と終値を返してくれるツールです。

あまり利用したりする想定はありませんが直近の15分足の株価を取得する場合に利用します。

本来は定期的にこのツールを呼び出すことで株価を取得して分析させることを構想しており、今後はこのツールを利用して定期的に株価を取得して分析させることを目指しています。

import asyncio
import yfinance as yf

async def fetch_latest_nikkei_data(args: FetchLatestArgs) -> dict[str, Any]:
  """最新日中足を取得する MCP ツール."""
    symbol = args.get("symbol", DEFAULT_SYMBOL) #NIY=F
    period = args.get("period", DEFAULT_PERIOD)
    interval = DEFAULT_INTERVAL  # 15分足に固定
    timeout = float(args.get("timeout", 30.0))

    try:
        result = await asyncio.to_thread(
            fetch_latest_nikkei_data_sync,
            symbol=symbol,
            period=period,
            interval=interval,
            timeout=timeout,
        )
    except Exception as exc:  # noqa: BLE001 - エラーメッセージをそのまま返す
        return format_tool_error(f"最新データ取得に失敗しました: {exc}")

    return format_tool_output(result)

def fetch_latest_nikkei_data_sync(
    symbol: str = DEFAULT_SYMBOL,
    period: str = DEFAULT_PERIOD,
    interval: str = DEFAULT_INTERVAL,
    timeout: float = 30.0,
) -> dict:
    """
    日経平均先物の最新15分足データを取得する

    Args:
        symbol: 取得対象のシンボル (デフォルト: NIY=F)
        period: 取得期間 (デフォルト: 1d)
        interval: 取得インターバル (デフォルト: 15m)
        timeout: HTTPタイムアウト秒数

    Returns:
        dict: 最新の株価データ（timestamp, Open, Close）
    """
    ticker = yf.Ticker(symbol)
    history_kwargs: dict[str, object] = {
        "period": period,
        "interval": interval,
        "auto_adjust": False,
        "prepost": False,
        "actions": False,
        "raise_errors": True,
        "timeout": timeout,
    }
    data = ticker.history(**history_kwargs)

    normalized = normalize_frame(data, symbol)
    localized = localize_to_tokyo(normalized)
    daytime = filter_day_session(localized)
    selected = select_price_columns(daytime)

    # 最新の1行のみ返す
    if len(selected) > 0:
        latest = selected.tail(1)
        return {
            "timestamp": latest.index[0].isoformat(),
            "open": int(latest["Open"].iloc[0]),
            "close": int(latest["Close"].iloc[0]),
        }

    raise ValueError("データが取得できませんでした")

2. fetch_nikkei_history

指定した期間の株価を取得するツールになっています。何も指定しない場合は過去１日に絞ってデータを取得するようになっており株価は15分刻みの始値と終値を取得して取得したデータを後続のツールで分析できるように csv 形式にして出力します。

実は csv よりも json のほうがAIが理解しやすいという特性があるのですが今回はトークン数を少なくできる csv にしてなるべくトークンを節約できるような構成にしています。

from claude_agent_sdk import tool

@tool(
    name="fetch_nikkei_history",
    description="日経平均先物の履歴15分足データ（始値・終値）をCSV形式で返します。intervalは15分足に固定されています。",
    input_schema={
        "type": "object",
        "properties": {
            "symbol": {
                "type": "string",
                "description": "yfinance 互換のシンボル。既定値は NIY=F。",
                "default": DEFAULT_SYMBOL, #NIY=F
            },
            "period": {
                "type": "string",
                "description": "期間指定。start/end 指定時は無視。",
                "default": "1mo",
            },
            "timeout": {
                "type": "number",
                "description": "HTTP タイムアウト秒。既定値は 30 秒。",
                "default": 30.0,
            },
            "start": {
                "type": "string",
                "description": "取得開始日 (YYYY-MM-DD)。",
            },
            "end": {
                "type": "string",
                "description": "取得終了日 (YYYY-MM-DD)。",
            },
        },
        "additionalProperties": False,
    },
)
async def fetch_nikkei_history(args: FetchHistoryArgs) -> dict[str, Any]:

"""株価を取得する処理、処理自体は fetch_latest_nikkei_data_sync と同様"""

3. analyze_kagi_chart

fetch_nikkei_history で取得したデータを使ってチャート分析をします。

今回はカギ足と呼ばれるチャートを採用して分析を実施しています。nicjps230さんの「投資のためのデータサイエンス」というブログ記事の中でカギ足について紹介されているものがあり、厳密にルールが定義できて分析がしやすいと考え採用しました。

カギ足のルールについては以下のサイトのルールを参考に実装をしました。 https://www.sevendata.co.jp/shihyou/technical/kagiashi.html

カギ足は反転幅を設定する必要があるので50円をデフォルト値としています。

@tool(
    name="analyze_kagi_chart",
    description="CSV形式の価格データからカギ足チャートを生成し、売買シグナルとパフォーマンスを分析します。",
    input_schema={
        "type": "object",
        "properties": {
            "csv_data": {
                "type": "string",
                "description": "CSV形式の価格データ（timestamp, Open, Close列を含む）。fetch_nikkei_historyの出力をそのまま渡せます。",
            },
            "reversal": {
                "type": "number",
                "description": "カギ足の反転幅（円）。デフォルトは50円。",
                "default": 50,
            },
        },
        "required": ["csv_data"],
        "additionalProperties": False,
    },
)
async def analyze_kagi_chart(args: AnalyzeKagiArgs) -> dict[str, Any]:
"""カギ足チャートの生成とパフォーマンス評価"""
  reversal = int(args.get("reversal", 50))

  kagi_lines = build_kagi_chart(price_data, reversal) #カギ足チャートの生成
  performance = calculate_performance(kagi_lines) #カギ足チャート売買分析

  return {
          "reversal": reversal,
          "data_points": len(price_data),
          "kagi_lines": len(kagi_lines),
          "performance": performance,
          "latest_signal": next(
              (line for line in reversed(kagi_lines) if line["signal"] is not None),
              None,
          ),
      }