日経225先物・TOPIX先物のアルゴリズム取引戦略: 裁定取引からナイトセッション活用まで

戦略サマリー

日経225先物・TOPIX先物は、個人投資家がアルゴリズム取引を実践する上で最適な商品です。高い流動性、24時間近い取引時間（ナイトセッション含む）、明確な理論価格との裁定機会が存在します。

2026年3月時点の市場データ:

• 日経225先物: 約57,260円
• TOPIX先物: 約3,816円
• 日経225先物取引量: 日平均10万枚超
• ナイトセッション比率: 全体の約40%

先物取引の基礎知識

日経225先物の仕様

先物価格の理論値

先物の理論価格は、現物価格に金利コストを加え、配当を差し引いて算出します。

理論価格 = 現物価格 × (1 + 金利 × 日数/365) - 配当の現在価値

1. 裁定取引（アービトラージ）

現物-先物アービトラージ

先物価格が理論価格から乖離した際に、現物と先物を同時に売買して利益を狙う戦略です。

import pandas as pd
import numpy as np
from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class ArbitrageSignal:
    """裁定取引シグナル"""
    timestamp: datetime
    spot_price: float
    futures_price: float
    theoretical_price: float
    premium_rate: float  # プレミアム率（%）
    signal: str  # "buy_spot_sell_futures", "sell_spot_buy_futures", "hold"

class SpotFuturesArbitrage:
    """現物-先物裁定取引"""

    def __init__(
        self,
        risk_free_rate: float = 0.005,  # 年率0.5%
        dividend_yield: float = 0.02,   # 配当利回り2%
        transaction_cost: float = 0.001  # 往復0.1%
    ):
        self.risk_free_rate = risk_free_rate
        self.dividend_yield = dividend_yield
        self.transaction_cost = transaction_cost

    def calculate_theoretical_price(
        self,
        spot_price: float,
        days_to_expiry: int
    ) -> float:
        """先物理論価格を計算"""
        t = days_to_expiry / 365
        # 理論価格 = S * e^((r - d) * t)
        theoretical = spot_price * np.exp(
            (self.risk_free_rate - self.dividend_yield) * t
        )
        return theoretical

    def generate_signal(
        self,
        spot_price: float,
        futures_price: float,
        days_to_expiry: int,
        threshold: float = 0.003  # 0.3%以上の乖離でシグナル
    ) -> ArbitrageSignal:
        """裁定シグナルを生成"""
        theoretical = self.calculate_theoretical_price(spot_price, days_to_expiry)
        premium_rate = (futures_price - theoretical) / theoretical

        # 取引コストを考慮した閾値判定
        net_threshold = threshold + self.transaction_cost

        if premium_rate > net_threshold:
            # 先物が割高 → 現物買い・先物売り
            signal = "buy_spot_sell_futures"
        elif premium_rate < -net_threshold:
            # 先物が割安 → 現物売り・先物買い
            signal = "sell_spot_buy_futures"
        else:
            signal = "hold"

        return ArbitrageSignal(
            timestamp=datetime.now(),
            spot_price=spot_price,
            futures_price=futures_price,
            theoretical_price=theoretical,
            premium_rate=premium_rate * 100,
            signal=signal
        )

# 使用例
arbitrage = SpotFuturesArbitrage()
signal = arbitrage.generate_signal(
    spot_price=57000,
    futures_price=57260,
    days_to_expiry=30
)
print(f"プレミアム率: {signal.premium_rate:.2f}%")
print(f"シグナル: {signal.signal}")

限月間スプレッド取引

異なる限月間の価格差（カレンダースプレッド）を利用した戦略です。

class CalendarSpread:
    """限月間スプレッド取引"""

    def __init__(self, spread_history: list[float]):
        """
        Args:
            spread_history: 過去のスプレッド（近限月 - 遠限月）のリスト
        """
        self.spread_history = spread_history
        self.mean = np.mean(spread_history)
        self.std = np.std(spread_history)

    def generate_signal(
        self,
        near_price: float,
        far_price: float,
        z_threshold: float = 2.0
    ) -> dict:
        """スプレッドシグナルを生成"""
        current_spread = near_price - far_price
        z_score = (current_spread - self.mean) / self.std

        if z_score > z_threshold:
            # スプレッド拡大 → 近限月売り・遠限月買い
            signal = "sell_near_buy_far"
            expected_profit = (current_spread - self.mean) * 100  # ミニ換算
        elif z_score < -z_threshold:
            # スプレッド縮小 → 近限月買い・遠限月売り
            signal = "buy_near_sell_far"
            expected_profit = (self.mean - current_spread) * 100
        else:
            signal = "hold"
            expected_profit = 0

        return {
            "current_spread": current_spread,
            "mean_spread": self.mean,
            "z_score": z_score,
            "signal": signal,
            "expected_profit_mini": expected_profit
        }

# 使用例（過去30日のスプレッドデータ）
spread_history = [15, 12, 18, 14, 16, 13, 17, 15, 14, 16,
                  18, 20, 15, 12, 14, 16, 15, 17, 13, 14,
                  15, 16, 14, 18, 15, 13, 16, 17, 14, 15]
calendar = CalendarSpread(spread_history)

result = calendar.generate_signal(near_price=57260, far_price=57230)
print(f"現在のスプレッド: {result['current_spread']}円")
print(f"Zスコア: {result['z_score']:.2f}")
print(f"シグナル: {result['signal']}")

裁定取引のリスク要因

2. トレンドフォロー戦略

先物市場は24時間取引可能なため、グローバルなトレンドを捉えやすい特徴があります。

移動平均クロスオーバー

class FuturesTrendFollow:
    """先物トレンドフォロー戦略"""

    def __init__(
        self,
        fast_period: int = 20,
        slow_period: int = 60,
        atr_period: int = 14
    ):
        self.fast_period = fast_period
        self.slow_period = slow_period
        self.atr_period = atr_period

    def calculate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
        """トレンドフォローシグナルを計算"""
        # 移動平均
        df['SMA_Fast'] = df['Close'].rolling(self.fast_period).mean()
        df['SMA_Slow'] = df['Close'].rolling(self.slow_period).mean()

        # ATR（Average True Range）でボラティリティ計算
        df['TR'] = np.maximum(
            df['High'] - df['Low'],
            np.maximum(
                abs(df['High'] - df['Close'].shift(1)),
                abs(df['Low'] - df['Close'].shift(1))
            )
        )
        df['ATR'] = df['TR'].rolling(self.atr_period).mean()

        # シグナル生成
        df['Signal'] = 0
        df.loc[df['SMA_Fast'] > df['SMA_Slow'], 'Signal'] = 1   # ロング
        df.loc[df['SMA_Fast'] < df['SMA_Slow'], 'Signal'] = -1  # ショート

        # ポジションサイズ（ATRベース）
        # リスク1%として、ATRの2倍をストップロスとする
        df['Position_Size'] = 1 / (df['ATR'] * 2 / df['Close'])

        return df

    def calculate_breakout_signals(
        self,
        df: pd.DataFrame,
        lookback: int = 20
    ) -> pd.DataFrame:
        """ブレイクアウトシグナルを計算"""
        df['High_N'] = df['High'].rolling(lookback).max()
        df['Low_N'] = df['Low'].rolling(lookback).min()

        df['Breakout_Signal'] = 0
        # 直近N日高値更新でロング
        df.loc[df['Close'] > df['High_N'].shift(1), 'Breakout_Signal'] = 1
        # 直近N日安値更新でショート
        df.loc[df['Close'] < df['Low_N'].shift(1), 'Breakout_Signal'] = -1

        return df

# 使用例
trend = FuturesTrendFollow(fast_period=20, slow_period=60)
# df = 先物価格データ（OHLC）
# df = trend.calculate_signals(df)

トレンドフォローのバックテスト結果（2024-2026年）

3. 平均回帰戦略

先物価格が理論価格や移動平均から乖離した際に、回帰を狙う戦略です。

class FuturesMeanReversion:
    """先物平均回帰戦略"""

    def __init__(
        self,
        bb_period: int = 20,
        bb_std: float = 2.0,
        rsi_period: int = 14
    ):
        self.bb_period = bb_period
        self.bb_std = bb_std
        self.rsi_period = rsi_period

    def calculate_bollinger_bands(self, df: pd.DataFrame) -> pd.DataFrame:
        """ボリンジャーバンドを計算"""
        df['BB_Middle'] = df['Close'].rolling(self.bb_period).mean()
        df['BB_Std'] = df['Close'].rolling(self.bb_period).std()
        df['BB_Upper'] = df['BB_Middle'] + self.bb_std * df['BB_Std']
        df['BB_Lower'] = df['BB_Middle'] - self.bb_std * df['BB_Std']

        # %B（ボリンジャーバンド内での位置）
        df['BB_Pct'] = (df['Close'] - df['BB_Lower']) / (df['BB_Upper'] - df['BB_Lower'])

        return df

    def calculate_rsi(self, df: pd.DataFrame) -> pd.DataFrame:
        """RSIを計算"""
        delta = df['Close'].diff()
        gain = delta.where(delta > 0, 0).rolling(self.rsi_period).mean()
        loss = (-delta).where(delta < 0, 0).rolling(self.rsi_period).mean()

        rs = gain / loss
        df['RSI'] = 100 - (100 / (1 + rs))

        return df

    def generate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
        """平均回帰シグナルを生成"""
        df = self.calculate_bollinger_bands(df)
        df = self.calculate_rsi(df)

        df['Signal'] = 0

        # ボリンジャーバンド下限 & RSI<30 → ロング
        df.loc[
            (df['Close'] < df['BB_Lower']) & (df['RSI'] < 30),
            'Signal'
        ] = 1

        # ボリンジャーバンド上限 & RSI>70 → ショート
        df.loc[
            (df['Close'] > df['BB_Upper']) & (df['RSI'] > 70),
            'Signal'
        ] = -1

        return df

平均回帰のパラメータ比較

4. SQ（特別清算指数）を利用した戦略

SQ日（毎月第2金曜日）は先物・オプションの清算日であり、価格変動が大きくなります。

SQカレンダー（2026年）

SQ戦略の実装

from datetime import date, timedelta

class SQStrategy:
    """SQ関連戦略"""

    def __init__(self, sq_dates: list[date]):
        self.sq_dates = sq_dates

    def days_to_sq(self, current_date: date) -> int:
        """次のSQまでの日数を計算"""
        future_sqs = [d for d in self.sq_dates if d > current_date]
        if not future_sqs:
            return -1
        return (future_sqs[0] - current_date).days

    def generate_pre_sq_signal(
        self,
        current_date: date,
        futures_price: float,
        open_interest_change: float,  # 建玉増減
        put_call_ratio: float
    ) -> dict:
        """SQ前のシグナルを生成"""
        days = self.days_to_sq(current_date)

        signal = "hold"
        confidence = 0.0
        reason = ""

        if days <= 5 and days > 0:
            # SQ週のシグナル
            if open_interest_change > 0 and put_call_ratio < 0.7:
                # 建玉増加 & プット少ない → 上昇期待
                signal = "long"
                confidence = 0.7
                reason = "建玉増加・コール優勢"
            elif open_interest_change < 0 and put_call_ratio > 1.3:
                # 建玉減少 & プット多い → 下落期待
                signal = "short"
                confidence = 0.6
                reason = "建玉減少・プット優勢"

        return {
            "days_to_sq": days,
            "signal": signal,
            "confidence": confidence,
            "reason": reason
        }

    def sq_day_volatility_trade(
        self,
        pre_market_futures: float,
        expected_sq_value: float,
        threshold_pct: float = 0.5
    ) -> dict:
        """SQ当日のボラティリティトレード"""
        gap_pct = (expected_sq_value - pre_market_futures) / pre_market_futures * 100

        if abs(gap_pct) > threshold_pct:
            # SQ値と乖離があれば、収束方向にポジション
            signal = "short" if gap_pct > 0 else "long"
            target = expected_sq_value
        else:
            signal = "hold"
            target = None

        return {
            "gap_pct": gap_pct,
            "signal": signal,
            "target_price": target
        }

SQ戦略のバックテスト結果

5. ナイトセッション活用

ナイトセッション（16:30〜翌5:30）は、米国市場や欧州市場の動きを先取りできます。

ナイトセッションの特徴

海外市場連動戦略

class NightSessionStrategy:
    """ナイトセッション戦略"""

    def __init__(self):
        self.correlation_weights = {
            "SP500": 0.4,
            "NASDAQ": 0.3,
            "VIX": -0.2,
            "USDJPY": 0.1
        }

    def calculate_expected_move(
        self,
        us_futures_change: float,  # S&P500先物変動率（%）
        nasdaq_change: float,
        vix_change: float,
        usdjpy_change: float
    ) -> dict:
        """日経先物の期待変動を計算"""
        weighted_change = (
            us_futures_change * self.correlation_weights["SP500"] +
            nasdaq_change * self.correlation_weights["NASDAQ"] +
            vix_change * self.correlation_weights["VIX"] +
            usdjpy_change * self.correlation_weights["USDJPY"]
        )

        # 過去のベータ係数（日経/S&P）を考慮
        beta = 1.2  # 日経はS&Pより変動が大きい傾向
        expected_nikkei_move = weighted_change * beta

        return {
            "expected_move_pct": expected_nikkei_move,
            "signal": "long" if expected_nikkei_move > 0.3 else
                      "short" if expected_nikkei_move < -0.3 else "hold",
            "confidence": min(abs(expected_nikkei_move) / 1.0, 1.0)
        }

    def overnight_gap_prediction(
        self,
        night_session_close: float,
        day_session_close: float,
        us_market_close_change: float
    ) -> dict:
        """翌朝ギャップ予測"""
        night_change = (night_session_close - day_session_close) / day_session_close * 100

        # 米国市場終値とナイトセッション終値の差分
        residual_move = us_market_close_change - night_change

        # 残りの動きが翌朝ギャップに反映される傾向
        expected_gap = residual_move * 0.5  # 約50%が翌朝に反映

        return {
            "night_change_pct": night_change,
            "expected_morning_gap_pct": expected_gap,
            "signal": "gap_up" if expected_gap > 0.2 else
                      "gap_down" if expected_gap < -0.2 else "flat"
        }

# 使用例
night = NightSessionStrategy()
result = night.calculate_expected_move(
    us_futures_change=1.5,   # S&P500先物+1.5%
    nasdaq_change=2.0,       # NASDAQ+2.0%
    vix_change=-5.0,         # VIX-5%（リスクオン）
    usdjpy_change=0.3        # ドル円+0.3%
)
print(f"期待変動: {result['expected_move_pct']:.2f}%")
print(f"シグナル: {result['signal']}")

ナイトセッション戦略のパフォーマンス

6. 証券APIでの先物発注

kabuステーションAPIの実装

import requests
from dataclasses import dataclass
from enum import Enum

class OrderSide(Enum):
    BUY = "2"   # 買い
    SELL = "1"  # 売り

class OrderType(Enum):
    MARKET = "1"  # 成行
    LIMIT = "2"   # 指値
    STOP = "3"    # 逆指値

@dataclass
class FuturesOrder:
    """先物注文"""
    symbol: str  # "167060019" (日経225先物ミニ 2026年6月限)
    side: OrderSide
    qty: int
    order_type: OrderType
    price: float | None = None  # 成行の場合はNone
    stop_price: float | None = None  # 逆指値の場合

class KabuStationAPI:
    """kabuステーションAPI クライアント"""

    def __init__(self, api_password: str, port: int = 18080):
        self.base_url = f"http://localhost:{port}/kabusapi"
        self.api_password = api_password
        self.token: str | None = None

    def authenticate(self) -> bool:
        """認証してトークンを取得"""
        response = requests.post(
            f"{self.base_url}/token",
            json={"APIPassword": self.api_password}
        )
        if response.status_code == 200:
            self.token = response.json()["Token"]
            return True
        return False

    def place_futures_order(self, order: FuturesOrder) -> dict:
        """先物注文を発注"""
        if not self.token:
            raise RuntimeError("Not authenticated")

        payload = {
            "Password": self.api_password,
            "Symbol": order.symbol,
            "Exchange": 2,  # 日通し
            "TradeType": 1,  # 新規
            "TimeInForce": 2,  # FAK
            "Side": order.side.value,
            "Qty": order.qty,
            "FrontOrderType": int(order.order_type.value),
        }

        if order.order_type == OrderType.LIMIT and order.price:
            payload["Price"] = order.price
        if order.order_type == OrderType.STOP and order.stop_price:
            payload["Price"] = order.stop_price

        response = requests.post(
            f"{self.base_url}/sendorder/future",
            headers={"X-API-KEY": self.token},
            json=payload
        )

        return response.json()

    def get_positions(self) -> list[dict]:
        """保有ポジションを取得"""
        response = requests.get(
            f"{self.base_url}/positions",
            headers={"X-API-KEY": self.token},
            params={"product": 2}  # 先物
        )
        return response.json()

    def get_margin_info(self) -> dict:
        """証拠金情報を取得"""
        response = requests.get(
            f"{self.base_url}/wallet/future",
            headers={"X-API-KEY": self.token}
        )
        return response.json()

# 使用例
# api = KabuStationAPI(api_password="your_password")
# api.authenticate()
# order = FuturesOrder(
#     symbol="167060019",  # 日経225ミニ
#     side=OrderSide.BUY,
#     qty=1,
#     order_type=OrderType.LIMIT,
#     price=57000.0
# )
# result = api.place_futures_order(order)

主要証券会社のAPI比較

7. 証拠金管理とリスク管理

証拠金の計算

@dataclass
class MarginRequirement:
    """証拠金要件"""
    initial_margin: float  # 新規建て証拠金
    maintenance_margin: float  # 維持証拠金
    margin_call_level: float  # 追証発生水準

class MarginManager:
    """証拠金管理"""

    def __init__(
        self,
        account_balance: float,
        max_leverage: float = 10.0,  # 最大レバレッジ
        risk_per_trade: float = 0.02  # 1トレードあたりリスク2%
    ):
        self.account_balance = account_balance
        self.max_leverage = max_leverage
        self.risk_per_trade = risk_per_trade

    def calculate_position_size(
        self,
        futures_price: float,
        stop_loss_points: float,
        contract_multiplier: float = 100  # ミニの場合
    ) -> dict:
        """ポジションサイズを計算"""
        # 1枚あたりの損失額
        loss_per_contract = stop_loss_points * contract_multiplier

        # リスク許容額
        risk_amount = self.account_balance * self.risk_per_trade

        # 最大枚数（リスクベース）
        max_contracts_risk = int(risk_amount / loss_per_contract)

        # 最大枚数（レバレッジベース）
        contract_value = futures_price * contract_multiplier
        max_contracts_leverage = int(
            self.account_balance * self.max_leverage / contract_value
        )

        # より厳しい方を採用
        recommended_size = min(max_contracts_risk, max_contracts_leverage)

        return {
            "recommended_contracts": max(1, recommended_size),
            "risk_amount": risk_amount,
            "loss_per_contract": loss_per_contract,
            "effective_leverage": recommended_size * contract_value / self.account_balance
        }

    def check_margin_status(
        self,
        current_margin: float,
        required_margin: float,
        unrealized_pnl: float
    ) -> dict:
        """証拠金状況をチェック"""
        net_equity = current_margin + unrealized_pnl
        margin_ratio = net_equity / required_margin * 100

        status = "normal"
        if margin_ratio < 100:
            status = "margin_call"
        elif margin_ratio < 150:
            status = "warning"

        return {
            "net_equity": net_equity,
            "margin_ratio": margin_ratio,
            "status": status,
            "action_required": status != "normal"
        }

# 使用例
manager = MarginManager(account_balance=5_000_000)  # 500万円
sizing = manager.calculate_position_size(
    futures_price=57000,
    stop_loss_points=200,  # 200円のストップロス
    contract_multiplier=100
)
print(f"推奨枚数: {sizing['recommended_contracts']}枚")
print(f"実効レバレッジ: {sizing['effective_leverage']:.1f}倍")

リスク管理のベストプラクティス

8. 先物プレミアム/ディスカウントの活用

プレミアム監視システム

class PremiumMonitor:
    """先物プレミアム監視"""

    def __init__(self, history_days: int = 30):
        self.history_days = history_days
        self.premium_history: list[float] = []

    def calculate_premium(
        self,
        futures_price: float,
        spot_price: float
    ) -> dict:
        """プレミアム/ディスカウントを計算"""
        premium = futures_price - spot_price
        premium_rate = premium / spot_price * 100

        self.premium_history.append(premium_rate)
        if len(self.premium_history) > self.history_days:
            self.premium_history.pop(0)

        mean = np.mean(self.premium_history)
        std = np.std(self.premium_history)
        z_score = (premium_rate - mean) / std if std > 0 else 0

        return {
            "premium_points": premium,
            "premium_rate": premium_rate,
            "mean_rate": mean,
            "z_score": z_score,
            "status": "premium" if premium > 0 else "discount"
        }

    def generate_signal(
        self,
        premium_data: dict,
        z_threshold: float = 2.0
    ) -> str:
        """プレミアム異常に基づくシグナル"""
        z = premium_data["z_score"]

        if z > z_threshold:
            # 異常なプレミアム → 縮小期待
            return "sell_futures_buy_spot"
        elif z < -z_threshold:
            # 異常なディスカウント → 拡大期待
            return "buy_futures_sell_spot"
        return "hold"

プレミアムの季節性

9. 建玉残高（OI）分析

OI分析の実装

class OpenInterestAnalyzer:
    """建玉残高分析"""

    def __init__(self):
        self.oi_history: list[dict] = []

    def analyze_oi(
        self,
        current_oi: int,
        previous_oi: int,
        price_change: float
    ) -> dict:
        """OIと価格変動の関係を分析"""
        oi_change = current_oi - previous_oi
        oi_change_pct = oi_change / previous_oi * 100

        # OIと価格の組み合わせでトレンド判断
        if oi_change > 0 and price_change > 0:
            interpretation = "新規買い増加 - 上昇トレンド継続"
            signal = "bullish"
        elif oi_change > 0 and price_change < 0:
            interpretation = "新規売り増加 - 下落トレンド継続"
            signal = "bearish"
        elif oi_change < 0 and price_change > 0:
            interpretation = "ショートカバー - 上昇は一時的"
            signal = "neutral_up"
        elif oi_change < 0 and price_change < 0:
            interpretation = "ロング決済 - 下落は一時的"
            signal = "neutral_down"
        else:
            interpretation = "方向感なし"
            signal = "neutral"

        return {
            "oi_change": oi_change,
            "oi_change_pct": oi_change_pct,
            "interpretation": interpretation,
            "signal": signal
        }

    def analyze_player_positions(
        self,
        foreign_investors: int,  # 外国人建玉（枚）
        retail_investors: int,   # 個人建玉（枚）
        institutions: int        # 機関投資家建玉（枚）
    ) -> dict:
        """投資主体別建玉を分析"""
        total = abs(foreign_investors) + abs(retail_investors) + abs(institutions)

        foreign_ratio = foreign_investors / total * 100 if total > 0 else 0
        retail_ratio = retail_investors / total * 100 if total > 0 else 0

        # 外国人と個人の逆張り傾向を活用
        if foreign_investors > 0 and retail_investors < 0:
            signal = "follow_foreign"  # 外国人についていく
            confidence = min(abs(foreign_ratio) / 50, 1.0)
        elif foreign_investors < 0 and retail_investors > 0:
            signal = "fade_retail"  # 個人の逆を行く
            confidence = min(abs(retail_ratio) / 30, 1.0)
        else:
            signal = "mixed"
            confidence = 0.3

        return {
            "foreign_position": foreign_investors,
            "retail_position": retail_investors,
            "signal": signal,
            "confidence": confidence
        }

OIデータの解釈

バックテスト: 限月ロールとコンタンゴ考慮

先物バックテストでは、限月ロールとコンタンゴ/バックワーデーションの影響を考慮する必要があります。

class FuturesBacktester:
    """先物バックテスト（限月ロール考慮）"""

    def __init__(
        self,
        roll_days_before_expiry: int = 3,
        transaction_cost: float = 0.0002  # 0.02%
    ):
        self.roll_days = roll_days_before_expiry
        self.transaction_cost = transaction_cost

    def adjust_for_roll(
        self,
        continuous_data: pd.DataFrame,
        roll_dates: list[date]
    ) -> pd.DataFrame:
        """限月ロールを調整"""
        adjusted = continuous_data.copy()

        for roll_date in roll_dates:
            roll_idx = adjusted.index.get_loc(roll_date)

            # ロール前後の価格差を計算
            if roll_idx > 0:
                gap = adjusted.iloc[roll_idx]['Close'] - adjusted.iloc[roll_idx - 1]['Close']
                # ロール後のデータを調整
                adjusted.iloc[:roll_idx, adjusted.columns.get_loc('Close')] -= gap

        return adjusted

    def run_backtest(
        self,
        data: pd.DataFrame,
        signals: pd.Series,
        initial_capital: float = 10_000_000
    ) -> dict:
        """バックテストを実行"""
        data = data.copy()
        data['Signal'] = signals
        data['Returns'] = data['Close'].pct_change()
        data['Strategy_Returns'] = data['Signal'].shift(1) * data['Returns']

        # 取引コスト控除
        trades = data['Signal'].diff().abs()
        data['Strategy_Returns'] -= trades * self.transaction_cost

        # 累積リターン
        data['Cumulative'] = (1 + data['Strategy_Returns']).cumprod()
        data['Drawdown'] = data['Cumulative'] / data['Cumulative'].cummax() - 1

        total_return = data['Cumulative'].iloc[-1] - 1
        max_drawdown = data['Drawdown'].min()
        sharpe = data['Strategy_Returns'].mean() / data['Strategy_Returns'].std() * np.sqrt(252)
        win_rate = (data['Strategy_Returns'] > 0).mean()

        return {
            "total_return": f"{total_return:.1%}",
            "max_drawdown": f"{max_drawdown:.1%}",
            "sharpe_ratio": f"{sharpe:.2f}",
            "win_rate": f"{win_rate:.1%}",
            "trade_count": int(trades.sum() / 2)
        }

リスク要因

まとめ

推奨ステップ:

1. デモ環境（kabuステーション等）で各戦略を検証
2. ミニ1枚から小ロットで実運用開始
3. 証拠金管理とリスク管理を徹底
4. ナイトセッションを活用してリスクヘッジ
5. バックテストと実運用の乖離を定期的に検証

出典

• JPX（日本取引所グループ）- 先物・オプション取引データ
• auカブコム証券 - kabuステーションAPI仕様
• Investing.com - 日経225先物リアルタイム相場
• Grok調査（2026年3月14日）- 最新市場動向

---

免責事項: 本記事は情報提供を目的としており、特定の金融商品の売買を推奨するものではありません。先物取引は元本を上回る損失が発生する可能性があります。投資判断は自己責任で行ってください。