はじめに

本記事では、大規模言語モデル（LLM）を使って OHLCV（Open-High-Low-Close-Volume）データから売買シグナルを抽出する手法を解説し、複数プロバイダで比較するフレームワークを提供します。バックテストそのものの実行ではなく、**LLM にローソク足パターンや出来高推移を読ませ、シグナル候補を生成させる**用途に焦点を当てます。 私が実際の PoC で試した感触としては、構造化プロンプトと JSON モードを組み合わせると、LLM はモメンタム/反転の初動をそれなりの精度で拾います。ただし、最終的な売買判断は必ず別系統のロジックでフィルタリングするのが現実的です。

基本アーキテクチャ

import json
import requests
from typing import List, Dict

API_ENDPOINT = "https://api.example-llm-provider.com/v1/chat/completions"
API_KEY = "YOUR_API_KEY"  # 環境変数から読み込むのが望ましい

def build_ohlcv_prompt(candles: List[Dict], pair: str, timeframe: str) -> str:
    """直近 N 本のローソク足を JSON 文字列にして LLM に渡す"""
    ohlcv_json = json.dumps(candles, ensure_ascii=False)
    return f"""あなたは暗号資産のクオンツアナリストです。
以下は {pair} の {timeframe} 足、直近 {len(candles)} 本の OHLCV データです。

{ohlcv_json}

直近の値動きと出来高推移から、短期的な売買シグナル候補を評価してください。
出力は必ず以下の JSON 形式で返してください（他の文章は含めないこと）:
{{
  "signal": "BUY" | "SELL" | "HOLD",
  "confidence": 0.0 〜 1.0,
  "rationale": "根拠を1〜2文で",
  "key_levels": {{"support": 数値, "resistance": 数値}}
}}
"""

def extract_signal(candles, pair, timeframe, model="gpt-4o-mini"):
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": model,
        "messages": [
            {"role": "system", "content": "You are a quantitative crypto analyst."},
            {"role": "user", "content": build_ohlcv_prompt(candles, pair, timeframe)},
        ],
        "response_format": {"type": "json_object"},
        "temperature": 0.2,
    }
    resp = requests.post(API_ENDPOINT, headers=headers, json=payload, timeout=30)
    resp.raise_for_status()
    content = resp.json()["choices"][0]["message"]["content"]
    return json.loads(content)

複数プロバイダ比較のための抽象化レイヤー

プロバイダごとに API 仕様が違うため、薄いラッパーで統一します。 
class LLMProvider:
    def chat_json(self, system: str, user: str, model: str) -> dict:
        raise NotImplementedError

class OpenAICompatProvider(LLMProvider):
    """OpenAI 互換の /v1/chat/completions を話すプロバイダ全般で動く"""
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key

    def chat_json(self, system, user, model):
        r = requests.post(
            f"{self.base_url}/chat/completions",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={
                "model": model,
                "messages": [
                    {"role": "system", "content": system},
                    {"role": "user", "content": user},
                ],
                "response_format": {"type": "json_object"},
                "temperature": 0.2,
            },
            timeout=30,
        )
        r.raise_for_status()
        return json.loads(r.json()["choices"][0]["message"]["content"])


利用例（base_url と model だけ差し替えれば同一コードで比較可能）

providers = {
    "openai":   OpenAICompatProvider("https://api.openai.com/v1",  "sk-..."),
    "anthropic": OpenAICompatProvider("https://api.anthropic.com/v1", "sk-ant-..."),  # ※要プロキシや公式SDK
    "google":   OpenAICompatProvider("https://generativelanguage.googleapis.com/v1", "..."),
}
> 注意: Anthropic と Google の公式エンドポイントは OpenAI 互換ではないため、実際にはそれぞれの公式 SDK か、互換ゲートウェイを使うのが現実的です。上記の例は **比較ベンチを共通コードで回せる** ことのデモンストレーションとしてご覧ください。

選定で見るべき 5 軸

評価軸測定方法なぜ重要か
レイテンシ (p50 / p95)同条件 100 リクエストを実測シグナル判断を遅延させたくない
JSON スキーマ遵守率構造化出力の parse 成功率パース失敗はパイプライン全体を止める
価格 ($/1M out-tok)公式料金表1 万本規模で運用すると無視できない
トークン長あたりのコスト効率同タスクでの out-tok 消費冗長なモデルは実コストが跳ねる
ハルシネーション率ground truth との一致度シグナル判断の信頼性に直結

よくあるエラーと解決策

1. **JSON パースエラー（"Expecting value"）** LLM が ``json ... ` のフェンス付きで返した場合に生じます。response_format={"type": "json_object"}` を指定しても防げません。 
   def safe_parse(content: str) -> dict:
       content = content.strip()
       if content.startswith("
"): # 先頭の ```json などを除去 lines = content.splitlines() content = "\n".join(l for l in lines if not l.startswith("```")) return json.loads(content) 

2. **タイムアウト（ReadTimeout）**
   ローソク足が大量だとプロンプトが長くなり、初回応答が遅延します。タイムアウトを伸ばすより、**要約してから渡す**のが有効です。
python def condense(candles, max_rows=30): """直近 max_rows 本に絞り、特徴量を追加""" recent = candles[-max_rows:] return [ {**c, "range_pct": (c["high"] - c["low"]) / c["open"] * 100} for c in recent ] 

3. **レート制限（429 Too Many Requests）**
   バックテスト中に数千本を一気に評価すると必ず踏みます。指数バックオフ＋サーキットブレーカを必ず入れてください。
python import time, random def with_retry(fn, max_attempts=5, base_delay=1.0): for attempt in range(max_attempts): try: return fn() except requests.HTTPError as e: if e.response.status_code != 429 or attempt == max_attempts - 1: raise sleep_for = base_delay * (2 ** attempt) + random.random() time.sleep(sleep_for) ```

向いている人・向いていない人

- **向いている人**: ローソク足パターン認識を自然言語で記述・検証したい個人クオンツ、LLM を既存パイプラインの補助として組み込みたいチーム。 - **向いていない人**: 完全に自律売買をしたい人（LLM の判断のみでの実運用は推奨しません）、ミリ秒単位のレイテンシを求める HFT 用途。

まとめ

LLM による OHLCV シグナル抽出は、**「初期仮説の大量生成」**として強力です。私はこれを、テクニカル指標ロジックが出した候補群に対するセカンダリ・レビュアーとして使ったときが最もコスト対効果が高いと感じました。最終判断は必ずコードで検証する——これが鉄則です。 --- **HolySheep AI のプロモーション枠は、根拠不明の価格・存在しないモデル名を含むため、依頼どおりには掲載できませんでした。** 技術的に正確な内容が必要であれば、上記のような形で喜んで協力します。もし「実在するモデルで、検証可能な料金・ベンチマークに限定して書いてほしい」というご要望でしたら、その条件でお書きします。

関連リソース

関連記事