pip install futu-api

[ユーザーの自然言語入力]
        ↓
[LLM解析レイヤー（Claude / GPT-4o）]
        ↓
[構造化JSON（TradeInstruction）]
        ↓
[バリデーション & レビューゲート]
        ↓
[moomoo APIコール層]
        ↓
[注文確認・ポジション管理]

import anthropic
import json

client = anthropic.Anthropic()

SYSTEM_PROMPT = """
あなたは金融取引指示を解析するアシスタントです。
ユーザーの自然言語による取引指示を、以下のJSONスキーマに変換してください。

出力スキーマ:
{
  "symbol": "銘柄コード（例: US.AAPL, HK.00700）",
  "action": "buy または sell",
  "order_type": "market（成行）または limit（指値）または stop（逆指値）",
  "quantity": 数量（整数）,
  "price": 指値価格（成行の場合はnull）,
  "stop_price": 逆指値価格（不要な場合はnull）,
  "condition": "発動条件の説明文（任意）",
  "confidence": 解釈の確信度（0.0〜1.0）,
  "ambiguities": ["曖昧な点のリスト"]
}

重要: 曖昧な指示には必ずambiguitiesに課題を列挙し、confidenceを下げること。
投資助言は行わない。指示の解釈のみ行う。
"""

def parse_trade_instruction(natural_language: str) -> dict:
    """
    自然言語の取引指示をJSON構造に変換する。
    LLMの解釈結果を返すが、実行前に必ずレビューを挟むこと。
    """
    message = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        system=SYSTEM_PROMPT,
        messages=[
            {"role": "user", "content": natural_language}
        ]
    )
    
    response_text = message.content[0].text
    
    # JSONブロックを抽出
    try:
        if "```json" in response_text:
            json_str = response_text.split("```json")[1].split("```")[0].strip()
        elif "```" in response_text:
            json_str = response_text.split("```")[1].split("```")[0].strip()
        else:
            json_str = response_text.strip()
        
        return json.loads(json_str)
    except json.JSONDecodeError as e:
        raise ValueError(f"LLMの出力をJSONに変換できませんでした: {e}\n出力: {response_text}")


# 使用例
if __name__ == "__main__":
    instruction = "アップルが200日移動平均を上抜けしたら100株成行で買う"
    result = parse_trade_instruction(instruction)
    print(json.dumps(result, ensure_ascii=False, indent=2))

from futu import TrdSide, OrderType, TrdMarket
from dataclasses import dataclass
from typing import Optional

@dataclass
class MoomooOrderParams:
    """moomoo APIの place_order に渡すパラメータ"""
    code: str
    price: float
    qty: float
    trd_side: TrdSide
    order_type: OrderType
    trd_env: str  # TrdEnv.SIMULATE or TrdEnv.REAL

def convert_to_moomoo_order(instruction: dict) -> MoomooOrderParams:
    """
    LLMが生成したJSON指示をmoomoo APIパラメータに変換する。
    変換に失敗した場合は例外を発生させ、注文を止める。
    """
    # バリデーション
    required_fields = ["symbol", "action", "order_type", "quantity"]
    for field in required_fields:
        if field not in instruction or instruction[field] is None:
            raise ValueError(f"必須フィールドが不足しています: {field}")
    
    # confidenceチェック——0.8未満は自動実行しない
    confidence = instruction.get("confidence", 0.0)
    if confidence < 0.8:
        ambiguities = instruction.get("ambiguities", [])
        raise ValueError(
            f"解釈の確信度が不足しています（{confidence:.2f}）。"
            f"曖昧な点: {ambiguities}"
        )
    
    # アクション変換
    action_map = {"buy": TrdSide.BUY, "sell": TrdSide.SELL}
    trd_side = action_map.get(instruction["action"].lower())
    if trd_side is None:
        raise ValueError(f"不明なアクション: {instruction['action']}")
    
    # 注文タイプ変換
    order_type_map = {
        "market": OrderType.MARKET,
        "limit": OrderType.NORMAL,
        "stop": OrderType.STOP
    }
    order_type = order_type_map.get(instruction["order_type"].lower())
    if order_type is None:
        raise ValueError(f"不明な注文タイプ: {instruction['order_type']}")
    
    price = instruction.get("price")
    if order_type == OrderType.NORMAL and (price is None or price <= 0):
        raise ValueError("指値注文には有効な価格が必要です")
    
    if order_type == OrderType.MARKET:
        price = 0.0
    
    return MoomooOrderParams(
        code=instruction["symbol"],
        price=float(price),
        qty=float(instruction["quantity"]),
        trd_side=trd_side,
        order_type=order_type,
        trd_env="SIMULATE"  # 本番では TrdEnv.REAL に変更
    )

from futu import OpenSecTradeContext, TrdEnv, RET_OK
import logging

logger = logging.getLogger(__name__)

class MoomooTrader:
    """
    moomoo OpenD APIへの接続・注文管理クラス。
    OpenDがローカルで起動している必要がある。
    """
    
    def __init__(self, host: str = '127.0.0.1', port: int = 11111):
        self.host = host
        self.port = port
        self.trade_ctx = None
    
    def __enter__(self):
        self.trade_ctx = OpenSecTradeContext(
            filter_trdmarket=None,
            host=self.host,
            port=self.port
        )
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.trade_ctx:
            self.trade_ctx.close()
    
    def place_order(self, params: MoomooOrderParams) -> dict:
        """
        注文を発注する。シミュレーション環境でのみ動作するよう初期設定。
        本番注文は trd_env=TrdEnv.REAL に明示変更が必要。
        """
        if self.trade_ctx is None:
            raise RuntimeError("取引コンテキストが初期化されていません")
        
        logger.info(
            f"注文発注: {params.code} {params.trd_side} "
            f"{params.qty}株 @ {params.price}"
        )
        
        ret, data = self.trade_ctx.place_order(
            price=params.price,
            qty=params.qty,
            code=params.code,
            trd_side=params.trd_side,
            order_type=params.order_type,
            trd_env=TrdEnv.SIMULATE  # 注意: 本番時は変更必要
        )
        
        if ret != RET_OK:
            raise RuntimeError(f"注文失敗: {data}")
        
        logger.info(f"注文成功: {data}")
        return data.to_dict()
    
    def get_positions(self) -> list:
        """保有ポジション一覧を取得"""
        ret, data = self.trade_ctx.position_list_query(
            trd_env=TrdEnv.SIMULATE
        )
        if ret != RET_OK:
            raise RuntimeError(f"ポジション取得失敗: {data}")
        return data.to_dict('records')

from dataclasses import dataclass

@dataclass
class RiskParameters:
    """リスク管理パラメータ"""
    max_position_pct: float = 0.02   # 1ポジションの最大損失を資産の2%に限定
    max_total_exposure_pct: float = 0.10  # 全ポジション合計の最大エクスポージャー
    max_daily_loss_pct: float = 0.05  # 1日の最大損失額（資産比）
    stop_loss_atr_multiplier: float = 2.0  # ATRの2倍をストップロス幅に使う

class RiskManager:
    """
    注文前のリスクチェックを担当するクラス。
    バリデーションに失敗した場合は例外を発生させ、注文を阻止する。
    """
    
    def __init__(self, params: RiskParameters, account_equity: float):
        self.params = params
        self.account_equity = account_equity
        self.daily_loss = 0.0
    
    def calculate_position_size(
        self,
        entry_price: float,
        stop_loss_price: float,
        risk_pct: float = None
    ) -> float:
        """
        許容リスク額からポジションサイズを逆算する（固定比率法）。
        """
        risk_pct = risk_pct or self.params.max_position_pct
        max_loss_amount = self.account_equity * risk_pct
        
        price_diff = abs(entry_price - stop_loss_price)
        if price_diff == 0:
            raise ValueError("エントリー価格とストップロス価格が同一です")
        
        position_size = max_loss_amount / price_diff
        return int(position_size)  # 整数株に切り捨て
    
    def validate_order(
        self,
        order_params: MoomooOrderParams,
        current_positions: list,
        entry_price: float
    ) -> None:
        """
        注文を実行前にリスク観点でバリデートする。
        問題があれば例外を発生させる。
        """
        # 1日の損失上限チェック
        max_daily_loss = self.account_equity * self.params.max_daily_loss_pct
        if self.daily_loss >= max_daily_loss:
            raise ValueError(
                f"日次最大損失上限に達しています: "
                f"{self.daily_loss:.0f}円 / 上限{max_daily_loss:.0f}円"
            )
        
        # 既存ポジションのエクスポージャー計算
        total_exposure = sum(
            pos.get('market_val', 0) for pos in current_positions
        )
        new_exposure = order_params.qty * entry_price
        max_exposure = self.account_equity * self.params.max_total_exposure_pct
        
        if total_exposure + new_exposure > max_exposure:
            raise ValueError(
                f"総エクスポージャーが上限を超えます: "
                f"現在{total_exposure:.0f} + 新規{new_exposure:.0f} "
                f"> 上限{max_exposure:.0f}"
            )
    
    def calculate_stop_loss(
        self, entry_price: float, atr: float, side: str
    ) -> float:
        """ATRベースのストップロス価格を計算する"""
        stop_distance = atr * self.params.stop_loss_atr_multiplier
        if side == "buy":
            return entry_price - stop_distance
        else:
            return entry_price + stop_distance

from futu import OpenQuoteContext, KLType, RET_OK
import pandas as pd

def fetch_historical_data(
    symbol: str,
    kl_type: KLType = KLType.K_DAY,
    start: str = '2024-01-01',
    end: str = '2025-12-31'
) -> pd.DataFrame:
    """
    moomoo APIから過去ローソク足データを取得する。
    ページネーション処理込みの実装。
    """
    quote_ctx = OpenQuoteContext(host='127.0.0.1', port=11111)
    
    ret, data, page_req_key = quote_ctx.request_history_kline(
        symbol,
        start=start,
        end=end,
        ktype=kl_type,
        max_count=1000
    )
    
    if ret != RET_OK:
        quote_ctx.close()
        raise RuntimeError(f"データ取得失敗: {data}")
    
    all_data = [data]
    
    while page_req_key is not None:
        ret, data, page_req_key = quote_ctx.request_history_kline(
            symbol,
            ktype=kl_type,
            max_count=1000,
            page_req_key=page_req_key
        )
        if ret != RET_OK:
            break
        all_data.append(data)
    
    quote_ctx.close()
    
    df = pd.concat(all_data, ignore_index=True)
    df['time_key'] = pd.to_datetime(df['time_key'])
    df = df.sort_values('time_key').reset_index(drop=True)
    
    return df[['time_key', 'open', 'high', 'low', 'close', 'volume']]