日経225マイクロ先物 × Monte Carlo 自動売買判定 — Claude + 1万通りシミュレーションで勝率55%超のときだけ発注する実装

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

"""
日経225 マイクロ先物 × Monte Carlo 売買判定
- 取引単位: 指数 × 10 円
- 証拠金: 約 20,000 円/枚（SPAN により変動、ここでは固定値で簡略化）
"""
import numpy as np
import pandas as pd
import yfinance as yf
from dataclasses import dataclass
from datetime import datetime, timedelta

MICRO_MULTIPLIER = 10        # マイクロ先物 1 枚 = 指数 × 10 円
MICRO_MARGIN = 20_000        # 証拠金概算（SPAN により変動）
TRADING_DAYS = 252

# ------------------------------------------------------------
# 1. 価格データ取得（yfinance を MVP として、本番は J-Quants API 推奨）
# ------------------------------------------------------------
def fetch_nikkei225(lookback_days: int = 252) -> pd.Series:
    end = datetime.now()
    start = end - timedelta(days=lookback_days * 2)
    df = yf.Ticker("^N225").history(start=start, end=end)
    return df["Close"].dropna().tail(lookback_days)

# ------------------------------------------------------------
# 2. GBM パラメータ推定
# ------------------------------------------------------------
@dataclass
class GBMParams:
    mu: float       # 年率ドリフト
    sigma: float    # 年率ボラ
    s0: float       # 現在価格

def estimate_gbm(prices: pd.Series) -> GBMParams:
    log_ret = np.log(prices / prices.shift(1)).dropna()
    return GBMParams(
        mu=float(log_ret.mean() * TRADING_DAYS),
        sigma=float(log_ret.std() * np.sqrt(TRADING_DAYS)),
        s0=float(prices.iloc[-1]),
    )

# ------------------------------------------------------------
# 3. Monte Carlo シミュレーション
# ------------------------------------------------------------
def simulate_paths(p: GBMParams, n_paths: int = 10_000,
                   horizon_days: int = 1) -> np.ndarray:
    """horizon_days 後の価格分布を GBM で生成"""
    dt = horizon_days / TRADING_DAYS
    z = np.random.standard_normal(n_paths)
    drift = (p.mu - 0.5 * p.sigma ** 2) * dt
    diffusion = p.sigma * np.sqrt(dt) * z
    return p.s0 * np.exp(drift + diffusion)

# ------------------------------------------------------------
# 4. 売買判定（勝率 + 期待値 + ケリー基準で枚数決定）
# ------------------------------------------------------------
@dataclass
class TradeSignal:
    action: str          # "LONG" / "SHORT" / "FLAT"
    win_rate: float
    expected_pt: float   # 期待損益（指数ポイント）
    var95_pt: float      # 95% VaR
    contracts: int

def _kelly_fraction(win_rate: float, ev: float, var95: float) -> float:
    """簡易ケリー: f* = (b*p - q) / b、b = ev/var95"""
    if ev <= 0 or var95 <= 0:
        return 0.0
    b = ev / var95
    f = (b * win_rate - (1 - win_rate)) / b
    return float(np.clip(f, 0.0, 0.25))  # 25% でキャップ

def decide_trade(p: GBMParams, paths: np.ndarray, capital: float,
                 win_threshold: float = 0.55,
                 ev_threshold_pt: float = 50.0) -> TradeSignal:
    pnl_long = paths - p.s0
    pnl_short = p.s0 - paths
    max_contracts = int(capital // MICRO_MARGIN)

    candidates = [
        ("LONG",  pnl_long),
        ("SHORT", pnl_short),
    ]
    best = TradeSignal("FLAT", 0.0, 0.0, 0.0, 0)
    for side, pnl in candidates:
        win = float((pnl > 0).mean())
        ev = float(pnl.mean())
        var95 = float(-np.percentile(pnl, 5))
        if win >= win_threshold and ev >= ev_threshold_pt:
            f = _kelly_fraction(win, ev, var95)
            contracts = max(1, int(max_contracts * f))
            if ev > best.expected_pt:
                best = TradeSignal(side, win, ev, var95, contracts)
    return best

# ------------------------------------------------------------
# 5. 実行
# ------------------------------------------------------------
def main(capital: float = 500_000):
    prices = fetch_nikkei225()
    params = estimate_gbm(prices)
    paths = simulate_paths(params, n_paths=10_000, horizon_days=1)

    print(f"現在値        : {params.s0:>10,.0f}")
    print(f"年率ドリフト   : {params.mu:>10.2%}")
    print(f"年率ボラ       : {params.sigma:>10.2%}")
    print(f"シナリオ平均   : {paths.mean():>10,.0f}")
    print(f" 5%ile / 95%ile: {np.percentile(paths,5):,.0f} / {np.percentile(paths,95):,.0f}")

    sig = decide_trade(params, paths, capital=capital)
    print("\n=== シグナル ===")
    print(f"アクション : {sig.action}")
    print(f"勝率       : {sig.win_rate:.1%}")
    print(f"期待損益   : {sig.expected_pt:+.1f} pt "
          f"= {sig.expected_pt * MICRO_MULTIPLIER:+,.0f} 円/枚")
    print(f"95% VaR    : {sig.var95_pt:.1f} pt "
          f"= {sig.var95_pt * MICRO_MULTIPLIER:,.0f} 円/枚")
    print(f"発注枚数   : {sig.contracts} 枚 "
          f"(必要証拠金 {sig.contracts * MICRO_MARGIN:,} 円)")

if __name__ == "__main__":
    main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

from anthropic import Anthropic

client = Anthropic()

def adjust_mu_with_claude(base_mu: float, news_summary: str) -> float:
    msg = client.messages.create(
        model="claude-opus-4-7",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": (
                "以下は直近24時間の日経225関連マクロニュースです。\n"
                "翌営業日のドリフト方向に対する影響度を -1.0〜+1.0 で答えてください。\n"
                "数値のみ JSON で返してください: {\"impact\": <float>}\n\n"
                f"{news_summary}"
            )
        }]
    )
    import json, re
    m = re.search(r'\{[^}]+\}', msg.content[0].text)
    if m is None:
        return base_mu  # JSON 抽出失敗時は補正しない
    try:
        impact = float(json.loads(m.group()).get("impact", 0.0))
    except (json.JSONDecodeError, ValueError):
        return base_mu
    impact = max(-1.0, min(1.0, impact))  # クリッピング
    # 年率換算で最大 ±5% 補正
    return base_mu + impact * 0.05