信用取引を活用したアルゴリズム投資戦略 — ロングショート・空売りシグナル自動化

戦略サマリー

信用取引は、レバレッジ（約3倍）と空売りという2つの武器をアルゴリズム取引に提供します。現物取引では不可能な「下落局面での利益獲得」と「市場中立型戦略」を実現でき、特にペアトレードや統計的アービトラージとの相性が抜群です。

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

• 信用取引買い残: 約4.0兆円（東証全体）
• 信用取引売り残: 約0.9兆円
• 信用倍率: 約4.5倍
• デイトレ信用利用率: 前年比20%増加傾向

信用取引の基礎

信用取引の3つの種類

信用取引のコスト構造

総コスト = 売買手数料 + 金利（買い方）or 貸株料（売り方） + 逆日歩リスク + 管理費

保証金の仕組み

from dataclasses import dataclass

@dataclass
class MarginAccount:
    """信用取引口座の状態"""
    cash_deposit: float        # 現金保証金
    stock_collateral: float    # 代用有価証券評価額
    collateral_ratio: float    # 代用掛目（通常80%）
    long_positions: float      # 信用買い建玉評価額
    short_positions: float     # 信用売り建玉評価額
    unrealized_pnl: float      # 含み損益

    @property
    def total_collateral(self) -> float:
        """保証金合計を計算"""
        return self.cash_deposit + self.stock_collateral * self.collateral_ratio

    @property
    def required_margin(self) -> float:
        """必要保証金（建玉の30%）"""
        total_positions = self.long_positions + self.short_positions
        return total_positions * 0.30

    @property
    def maintenance_margin_ratio(self) -> float:
        """保証金維持率を計算"""
        net_collateral = self.total_collateral + self.unrealized_pnl
        if self.long_positions + self.short_positions == 0:
            return float("inf")
        return net_collateral / (self.long_positions + self.short_positions) * 100

    @property
    def margin_status(self) -> str:
        """保証金状態を判定"""
        ratio = self.maintenance_margin_ratio
        if ratio >= 30:
            return "正常"
        elif ratio >= 25:
            return "警告（追証接近）"
        elif ratio >= 20:
            return "追証発生"
        else:
            return "強制決済水準"

# 使用例
account = MarginAccount(
    cash_deposit=3_000_000,
    stock_collateral=2_000_000,
    collateral_ratio=0.80,
    long_positions=5_000_000,
    short_positions=5_000_000,
    unrealized_pnl=-500_000
)
print(f"保証金合計: {account.total_collateral:,.0f}円")
print(f"保証金維持率: {account.maintenance_margin_ratio:.1f}%")
print(f"状態: {account.margin_status}")

1. ロングショート戦略（ペアトレード）

ペアトレードの基本原理

相関性の高い2銘柄の価格差（スプレッド）が平均から乖離した際に、割高な銘柄を空売り、割安な銘柄を買い建てる市場中立型戦略です。

ペア選定とスプレッド分析

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

@dataclass
class PairSignal:
    """ペアトレードシグナル"""
    timestamp: datetime
    stock_a: str
    stock_b: str
    spread: float
    z_score: float
    signal: str  # "long_a_short_b", "short_a_long_b", "close", "hold"
    hedge_ratio: float

class PairTradingStrategy:
    """ペアトレード戦略"""

    def __init__(
        self,
        lookback: int = 60,
        entry_z: float = 2.0,
        exit_z: float = 0.5,
        stop_z: float = 3.5
    ):
        self.lookback = lookback
        self.entry_z = entry_z
        self.exit_z = exit_z
        self.stop_z = stop_z

    def calculate_hedge_ratio(
        self,
        prices_a: pd.Series,
        prices_b: pd.Series
    ) -> float:
        """OLS回帰でヘッジ比率を計算"""
        # log価格でのヘッジ比率
        log_a = np.log(prices_a)
        log_b = np.log(prices_b)

        # 単純OLS: log(A) = beta * log(B) + alpha
        beta = np.cov(log_a, log_b)[0, 1] / np.var(log_b)
        return beta

    def calculate_spread(
        self,
        prices_a: pd.Series,
        prices_b: pd.Series,
        hedge_ratio: float
    ) -> pd.Series:
        """スプレッドを計算"""
        return np.log(prices_a) - hedge_ratio * np.log(prices_b)

    def generate_signal(
        self,
        prices_a: pd.Series,
        prices_b: pd.Series,
        stock_a: str,
        stock_b: str
    ) -> PairSignal:
        """ペアトレードシグナルを生成"""
        hedge_ratio = self.calculate_hedge_ratio(
            prices_a[-self.lookback:],
            prices_b[-self.lookback:]
        )

        spread = self.calculate_spread(prices_a, prices_b, hedge_ratio)
        spread_recent = spread[-self.lookback:]

        mean = spread_recent.mean()
        std = spread_recent.std()
        current_z = (spread.iloc[-1] - mean) / std

        # シグナル判定
        if current_z > self.entry_z:
            signal = "short_a_long_b"  # Aが割高 → A空売り・B買い
        elif current_z < -self.entry_z:
            signal = "long_a_short_b"  # Bが割高 → A買い・B空売り
        elif abs(current_z) < self.exit_z:
            signal = "close"  # スプレッド収束 → ポジション解消
        elif abs(current_z) > self.stop_z:
            signal = "stop_loss"  # ストップロス
        else:
            signal = "hold"

        return PairSignal(
            timestamp=datetime.now(),
            stock_a=stock_a,
            stock_b=stock_b,
            spread=spread.iloc[-1],
            z_score=current_z,
            signal=signal,
            hedge_ratio=hedge_ratio
        )

# 使用例: トヨタ(7203) vs ホンダ(7267)
pair = PairTradingStrategy(lookback=60, entry_z=2.0)
# prices_toyota = pd.Series([...])  # トヨタの終値データ
# prices_honda = pd.Series([...])   # ホンダの終値データ
# signal = pair.generate_signal(prices_toyota, prices_honda, "7203", "7267")
# print(f"Zスコア: {signal.z_score:.2f}, シグナル: {signal.signal}")

有力なペア候補（2026年3月時点）

実際の株価（2026年3月16日終値）:

• トヨタ(7203): 3,338円
• ホンダ(7267): 1,344.5円
• 三菱UFJ(8306): 2,634円
• ソニー(6758): 3,384円
• 日立(6501): 4,800円

セクターニュートラル戦略

セクター全体のリスクを中立化し、銘柄固有のアルファのみを抽出する手法です。

class SectorNeutralStrategy:
    """セクターニュートラル戦略"""

    def __init__(self, sector_stocks: dict[str, list[str]]):
        """
        Args:
            sector_stocks: セクター名 → 銘柄コードリストのマッピング
        """
        self.sector_stocks = sector_stocks

    def rank_stocks_in_sector(
        self,
        sector: str,
        scores: dict[str, float]
    ) -> list[tuple[str, float, str]]:
        """セクター内で銘柄をランキング"""
        stocks = self.sector_stocks.get(sector, [])
        ranked = sorted(
            [(s, scores.get(s, 0)) for s in stocks],
            key=lambda x: x[1],
            reverse=True
        )

        n = len(ranked)
        top_n = max(1, n // 4)  # 上位25%をロング
        bottom_n = max(1, n // 4)  # 下位25%をショート

        result = []
        for i, (stock, score) in enumerate(ranked):
            if i < top_n:
                result.append((stock, score, "long"))
            elif i >= n - bottom_n:
                result.append((stock, score, "short"))
            else:
                result.append((stock, score, "neutral"))

        return result

    def calculate_sector_exposure(
        self,
        positions: dict[str, float]
    ) -> dict[str, float]:
        """セクターエクスポージャーを計算"""
        exposure = {}
        for sector, stocks in self.sector_stocks.items():
            sector_exposure = sum(
                positions.get(s, 0) for s in stocks
            )
            exposure[sector] = sector_exposure
        return exposure

# セクター定義
sectors = {
    "自動車": ["7203", "7267", "7201", "7270"],
    "メガバンク": ["8306", "8316", "8411"],
    "半導体": ["8035", "6857", "6723", "6526"],
}

strategy = SectorNeutralStrategy(sectors)
# 各銘柄のスコア（モメンタムなどで算出）
scores = {"7203": 0.8, "7267": -0.3, "7201": 0.2, "7270": -0.5}
result = strategy.rank_stocks_in_sector("自動車", scores)
for stock, score, position in result:
    print(f"銘柄{stock}: スコア{score:.1f} → {position}")

2. 空売りシグナルの検出手法

RSI逆転シグナル

RSIが買われすぎ水準から急落するパターンを検出します。

class ShortSignalDetector:
    """空売りシグナル検出器"""

    def __init__(
        self,
        rsi_period: int = 14,
        rsi_overbought: float = 70,
        rsi_reversal_threshold: float = 5,
        volume_spike_ratio: float = 2.0,
        ma_short: int = 5,
        ma_long: int = 25
    ):
        self.rsi_period = rsi_period
        self.rsi_overbought = rsi_overbought
        self.rsi_reversal_threshold = rsi_reversal_threshold
        self.volume_spike_ratio = volume_spike_ratio
        self.ma_short = ma_short
        self.ma_long = ma_long

    def calculate_rsi(self, closes: pd.Series) -> pd.Series:
        """RSIを計算"""
        delta = closes.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
        return 100 - (100 / (1 + rs))

    def detect_rsi_reversal(self, df: pd.DataFrame) -> pd.Series:
        """RSI逆転シグナル: 買われすぎから急落"""
        rsi = self.calculate_rsi(df["Close"])
        rsi_prev = rsi.shift(1)

        # 前日RSI>70 かつ 当日RSIが5ポイント以上下落
        signal = (
            (rsi_prev > self.rsi_overbought) &
            (rsi_prev - rsi > self.rsi_reversal_threshold)
        )
        return signal.astype(int)

    def detect_death_cross(self, df: pd.DataFrame) -> pd.Series:
        """移動平均デッドクロス: 短期MAが長期MAを下抜け"""
        ma_short = df["Close"].rolling(self.ma_short).mean()
        ma_long = df["Close"].rolling(self.ma_long).mean()
        ma_short_prev = ma_short.shift(1)
        ma_long_prev = ma_long.shift(1)

        # 前日: 短期MA >= 長期MA → 当日: 短期MA < 長期MA
        signal = (
            (ma_short_prev >= ma_long_prev) &
            (ma_short < ma_long)
        )
        return signal.astype(int)

    def detect_volume_spike_decline(self, df: pd.DataFrame) -> pd.Series:
        """出来高急増+下落シグナル"""
        avg_volume = df["Volume"].rolling(20).mean()
        volume_ratio = df["Volume"] / avg_volume
        price_change = df["Close"].pct_change()

        # 出来高が平均の2倍以上 かつ 株価下落
        signal = (
            (volume_ratio > self.volume_spike_ratio) &
            (price_change < -0.01)  # 1%以上の下落
        )
        return signal.astype(int)

    def composite_short_signal(self, df: pd.DataFrame) -> pd.DataFrame:
        """複合空売りシグナル"""
        df = df.copy()
        df["RSI_Reversal"] = self.detect_rsi_reversal(df)
        df["Death_Cross"] = self.detect_death_cross(df)
        df["Volume_Spike"] = self.detect_volume_spike_decline(df)

        # 複合スコア（2つ以上のシグナルで空売り判定）
        df["Short_Score"] = (
            df["RSI_Reversal"] + df["Death_Cross"] + df["Volume_Spike"]
        )
        df["Short_Signal"] = (df["Short_Score"] >= 2).astype(int)

        return df

# 使用例
detector = ShortSignalDetector()
# df = 株価データ（OHLCV）
# result = detector.composite_short_signal(df)
# short_days = result[result["Short_Signal"] == 1]
# print(f"空売りシグナル日数: {len(short_days)}")

空売りシグナルの種類と精度

3. デイトレ信用の手数料無料活用

デイトレ信用の仕組み

主要ネット証券では、当日中に返済する信用取引の手数料を無料としています。これをアルゴリズム取引と組み合わせることで、取引コストをほぼゼロにできます。

デイトレ信用スキャルピング戦略

from datetime import time
from enum import Enum

class DaytradeSession(Enum):
    MORNING_OPEN = "morning_open"      # 9:00-9:30
    MORNING_MID = "morning_mid"        # 9:30-11:00
    LUNCH_BREAK = "lunch"              # 11:30-12:30
    AFTERNOON_OPEN = "afternoon_open"  # 12:30-13:00
    AFTERNOON_CLOSE = "afternoon_close"  # 14:30-15:00

class DaytradeMarginScalper:
    """デイトレ信用スキャルピング"""

    def __init__(
        self,
        target_profit_pct: float = 0.3,   # 0.3%で利確
        stop_loss_pct: float = 0.2,       # 0.2%で損切り
        max_positions: int = 5,
        force_close_time: time = time(14, 50)  # 14:50に全決済
    ):
        self.target_profit = target_profit_pct
        self.stop_loss = stop_loss_pct
        self.max_positions = max_positions
        self.force_close_time = force_close_time

    def identify_session(self, current_time: time) -> DaytradeSession:
        """現在の取引セッションを判定"""
        if current_time < time(9, 30):
            return DaytradeSession.MORNING_OPEN
        elif current_time < time(11, 0):
            return DaytradeSession.MORNING_MID
        elif current_time < time(12, 30):
            return DaytradeSession.LUNCH_BREAK
        elif current_time < time(13, 0):
            return DaytradeSession.AFTERNOON_OPEN
        else:
            return DaytradeSession.AFTERNOON_CLOSE

    def generate_entry_signal(
        self,
        price: float,
        bid: float,
        ask: float,
        volume: int,
        vwap: float,
        tick_direction: str,  # "uptick", "downtick"
        session: DaytradeSession
    ) -> dict:
        """エントリーシグナルを生成"""
        spread_pct = (ask - bid) / price * 100
        price_vs_vwap = (price - vwap) / vwap * 100

        signal = "hold"
        side = None
        reason = ""

        # 朝の寄り付き直後: モメンタム追随
        if session == DaytradeSession.MORNING_OPEN:
            if price_vs_vwap > 0.3 and tick_direction == "uptick":
                signal = "entry"
                side = "long"
                reason = "寄り付きモメンタム（VWAP上方）"
            elif price_vs_vwap < -0.3 and tick_direction == "downtick":
                signal = "entry"
                side = "short"
                reason = "寄り付き売り圧力（VWAP下方）"

        # 日中: VWAP回帰
        elif session == DaytradeSession.MORNING_MID:
            if price_vs_vwap < -0.5 and tick_direction == "uptick":
                signal = "entry"
                side = "long"
                reason = "VWAP回帰（下方乖離反転）"
            elif price_vs_vwap > 0.5 and tick_direction == "downtick":
                signal = "entry"
                side = "short"
                reason = "VWAP回帰（上方乖離反転）"

        # 14:50以降は新規エントリー禁止
        if session == DaytradeSession.AFTERNOON_CLOSE:
            signal = "close_all"
            reason = "大引け前の全決済"

        return {
            "signal": signal,
            "side": side,
            "reason": reason,
            "spread_pct": spread_pct,
            "price_vs_vwap": price_vs_vwap
        }

    def calculate_daily_pnl(self, trades: list[dict]) -> dict:
        """デイトレの日次損益を計算（手数料ゼロ前提）"""
        gross_pnl = sum(t.get("pnl", 0) for t in trades)
        # デイトレ信用は手数料無料、金利も日割りでほぼゼロ
        net_pnl = gross_pnl
        win_trades = sum(1 for t in trades if t.get("pnl", 0) > 0)
        total_trades = len(trades)

        return {
            "gross_pnl": gross_pnl,
            "net_pnl": net_pnl,
            "trade_count": total_trades,
            "win_rate": win_trades / total_trades if total_trades > 0 else 0,
            "commission_saved": gross_pnl * 0.001  # 通常手数料比で節約額
        }

デイトレ信用の時間帯別パフォーマンス

4. 追証リスク管理と保証金維持率モニタリング

追証の発生メカニズム

追証リスク管理システム

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class RiskAlert:
    """リスクアラート"""
    level: str  # "info", "warning", "critical"
    message: str
    timestamp: datetime
    margin_ratio: float
    action: str

class MarginRiskManager:
    """追証リスク管理"""

    def __init__(
        self,
        warning_ratio: float = 35.0,    # 警告水準
        danger_ratio: float = 28.0,     # 危険水準
        margin_call_ratio: float = 25.0,  # 追証水準
        max_position_ratio: float = 2.5,  # 保証金対比最大ポジション倍率
        auto_reduce_enabled: bool = True
    ):
        self.warning_ratio = warning_ratio
        self.danger_ratio = danger_ratio
        self.margin_call_ratio = margin_call_ratio
        self.max_position_ratio = max_position_ratio
        self.auto_reduce = auto_reduce_enabled
        self.alerts: list[RiskAlert] = []

    def monitor_margin(
        self,
        collateral: float,
        total_positions: float,
        unrealized_pnl: float
    ) -> RiskAlert:
        """保証金維持率を監視"""
        net_collateral = collateral + unrealized_pnl
        margin_ratio = net_collateral / total_positions * 100 if total_positions > 0 else 100

        if margin_ratio < self.margin_call_ratio:
            alert = RiskAlert(
                level="critical",
                message=f"追証発生: 維持率{margin_ratio:.1f}%",
                timestamp=datetime.now(),
                margin_ratio=margin_ratio,
                action="immediate_reduce"
            )
        elif margin_ratio < self.danger_ratio:
            alert = RiskAlert(
                level="warning",
                message=f"追証接近: 維持率{margin_ratio:.1f}%",
                timestamp=datetime.now(),
                margin_ratio=margin_ratio,
                action="reduce_position_30pct"
            )
        elif margin_ratio < self.warning_ratio:
            alert = RiskAlert(
                level="info",
                message=f"注意: 維持率{margin_ratio:.1f}%",
                timestamp=datetime.now(),
                margin_ratio=margin_ratio,
                action="monitor_closely"
            )
        else:
            alert = RiskAlert(
                level="info",
                message=f"正常: 維持率{margin_ratio:.1f}%",
                timestamp=datetime.now(),
                margin_ratio=margin_ratio,
                action="none"
            )

        self.alerts.append(alert)
        return alert

    def calculate_stress_scenario(
        self,
        collateral: float,
        positions: list[dict],
        price_shock_pct: float = -10.0
    ) -> dict:
        """ストレステスト: 急落時の追証シミュレーション"""
        total_long = sum(p["value"] for p in positions if p["side"] == "long")
        total_short = sum(p["value"] for p in positions if p["side"] == "short")

        # 急落時の損益
        long_loss = total_long * (price_shock_pct / 100)
        short_gain = total_short * (-price_shock_pct / 100)  # 空売りは逆
        net_pnl = long_loss + short_gain

        stressed_collateral = collateral + net_pnl
        total_positions = total_long + total_short
        stressed_ratio = stressed_collateral / total_positions * 100 if total_positions > 0 else 100

        return {
            "scenario": f"全銘柄{price_shock_pct}%変動",
            "net_pnl": net_pnl,
            "stressed_collateral": stressed_collateral,
            "stressed_margin_ratio": stressed_ratio,
            "margin_call": stressed_ratio < self.margin_call_ratio,
            "shortfall": max(0, total_positions * 0.30 - stressed_collateral)
        }

    def auto_reduce_positions(
        self,
        positions: list[dict],
        target_ratio: float = 40.0,
        collateral: float = 0
    ) -> list[dict]:
        """ポジション自動縮小の提案"""
        total_value = sum(p["value"] for p in positions)
        current_ratio = collateral / total_value * 100 if total_value > 0 else 100

        if current_ratio >= target_ratio:
            return []

        # 目標維持率に必要なポジション削減額
        target_positions = collateral / (target_ratio / 100)
        reduce_amount = total_value - target_positions

        # 含み損が大きいポジションから優先的に決済
        sorted_positions = sorted(positions, key=lambda p: p.get("pnl", 0))
        orders = []
        reduced = 0

        for pos in sorted_positions:
            if reduced >= reduce_amount:
                break
            orders.append({
                "stock": pos["stock"],
                "action": "close",
                "value": pos["value"],
                "pnl": pos.get("pnl", 0)
            })
            reduced += pos["value"]

        return orders

# 使用例
risk_mgr = MarginRiskManager()

# 通常監視
alert = risk_mgr.monitor_margin(
    collateral=4_600_000,
    total_positions=10_000_000,
    unrealized_pnl=-1_800_000
)
print(f"[{alert.level}] {alert.message}")
print(f"推奨アクション: {alert.action}")

# ストレステスト
stress = risk_mgr.calculate_stress_scenario(
    collateral=4_600_000,
    positions=[
        {"stock": "7203", "side": "long", "value": 5_000_000, "pnl": -200_000},
        {"stock": "7267", "side": "short", "value": 5_000_000, "pnl": 100_000},
    ],
    price_shock_pct=-10.0
)
print(f"ストレス後維持率: {stress['stressed_margin_ratio']:.1f}%")
print(f"追証発生: {stress['margin_call']}")

追証リスク管理のルール

5. kabuステーションAPIでの信用発注

信用取引の自動発注実装

import requests
from dataclasses import dataclass
from enum import Enum
from typing import Optional

class MarginTradeType(Enum):
    """信用取引区分"""
    NEW_MARGIN = 2       # 新規（制度信用）
    NEW_GENERAL = 3      # 新規（一般信用）
    NEW_DAYTRADE = 4     # 新規（デイトレ信用）
    REPAY = 5            # 返済

class CreditType(Enum):
    """信用区分"""
    SYSTEM = 1    # 制度信用
    GENERAL = 2   # 一般信用
    DAYTRADE = 3  # デイトレ信用

@dataclass
class MarginOrder:
    """信用注文"""
    symbol: str
    exchange: int          # 1: 東証
    side: str              # "buy" or "sell"
    qty: int
    credit_type: CreditType
    order_type: str        # "market", "limit"
    price: Optional[float] = None
    stop_price: Optional[float] = None

class KabuStationMarginTrader:
    """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: Optional[str] = None

    def authenticate(self) -> bool:
        """認証"""
        resp = requests.post(
            f"{self.base_url}/token",
            json={"APIPassword": self.api_password}
        )
        if resp.status_code == 200:
            self.token = resp.json()["Token"]
            return True
        return False

    def _headers(self) -> dict:
        return {"X-API-KEY": self.token or ""}

    def place_margin_order(self, order: MarginOrder) -> dict:
        """信用注文を発注"""
        if not self.token:
            raise RuntimeError("Not authenticated")

        # 信用取引区分の決定
        if order.credit_type == CreditType.SYSTEM:
            trade_type = MarginTradeType.NEW_MARGIN.value
        elif order.credit_type == CreditType.GENERAL:
            trade_type = MarginTradeType.NEW_GENERAL.value
        else:
            trade_type = MarginTradeType.NEW_DAYTRADE.value

        # 売買区分
        side = "1" if order.side == "sell" else "2"

        # 注文タイプ
        front_order_type = 10 if order.order_type == "market" else 20

        payload = {
            "Password": self.api_password,
            "Symbol": f"{order.symbol}@{order.exchange}",
            "Exchange": order.exchange,
            "SecurityType": 1,  # 株式
            "Side": side,
            "CashMargin": trade_type,  # 信用取引区分
            "MarginTradeType": order.credit_type.value,
            "DelivType": 0,  # 指定なし
            "AccountType": 4,  # 特定口座
            "Qty": order.qty,
            "FrontOrderType": front_order_type,
            "Price": order.price or 0,
            "ExpireDay": 0,  # 当日限り
        }

        resp = requests.post(
            f"{self.base_url}/sendorder",
            headers=self._headers(),
            json=payload
        )
        return resp.json()

    def repay_margin_position(
        self,
        position_id: str,
        qty: int,
        order_type: str = "market",
        price: Optional[float] = None
    ) -> dict:
        """信用建玉の返済"""
        front_order_type = 10 if order_type == "market" else 20

        payload = {
            "Password": self.api_password,
            "HoldID": position_id,
            "Qty": qty,
            "FrontOrderType": front_order_type,
            "Price": price or 0,
        }

        resp = requests.post(
            f"{self.base_url}/sendorder",
            headers=self._headers(),
            json=payload
        )
        return resp.json()

    def get_margin_positions(self) -> list[dict]:
        """信用建玉一覧を取得"""
        resp = requests.get(
            f"{self.base_url}/positions",
            headers=self._headers(),
            params={"product": 1}  # 株式
        )
        positions = resp.json()
        # 信用建玉のみフィルタ
        return [p for p in positions if p.get("MarginTradeType", 0) > 0]

    def get_margin_info(self) -> dict:
        """保証金情報を取得"""
        resp = requests.get(
            f"{self.base_url}/wallet/margin",
            headers=self._headers()
        )
        return resp.json()

# ペアトレード自動実行の例
class AutoPairTrader:
    """ペアトレード自動執行"""

    def __init__(self, api: KabuStationMarginTrader):
        self.api = api

    def execute_pair_trade(
        self,
        long_stock: str,
        short_stock: str,
        long_qty: int,
        short_qty: int,
        credit_type: CreditType = CreditType.DAYTRADE
    ) -> dict:
        """ペアトレードを執行"""
        # 買い注文
        long_order = MarginOrder(
            symbol=long_stock,
            exchange=1,
            side="buy",
            qty=long_qty,
            credit_type=credit_type,
            order_type="market"
        )

        # 空売り注文
        short_order = MarginOrder(
            symbol=short_stock,
            exchange=1,
            side="sell",
            qty=short_qty,
            credit_type=credit_type,
            order_type="market"
        )

        # 同時発注
        long_result = self.api.place_margin_order(long_order)
        short_result = self.api.place_margin_order(short_order)

        return {
            "long": {"stock": long_stock, "qty": long_qty, "result": long_result},
            "short": {"stock": short_stock, "qty": short_qty, "result": short_result}
        }

# 使用例
# api = KabuStationMarginTrader(api_password="your_password")
# api.authenticate()
# trader = AutoPairTrader(api)
# result = trader.execute_pair_trade(
#     long_stock="7203",   # トヨタ買い
#     short_stock="7267",  # ホンダ空売り
#     long_qty=100,
#     short_qty=200,
#     credit_type=CreditType.DAYTRADE
# )

バックテスト結果

ペアトレード（トヨタ vs ホンダ）バックテスト: 2024-2026年

戦略別パフォーマンス比較

月別リターン分布（ペアトレード）

リスク要因

まとめ

推奨ステップ:

1. デイトレ信用で小ロット（100株）からペアトレードを練習
2. 保証金維持率モニタリングシステムを構築
3. 空売りシグナルの精度をバックテストで検証
4. 制度信用で中期ペアトレードに移行
5. セクターニュートラルポートフォリオへ段階的にスケール

出典

• 東京証券取引所 - 信用取引残高・逆日歩データ
• auカブコム証券 - kabuステーションAPI仕様書
• 日本証券金融 - 貸株・逆日歩情報
• Yahoo Finance - 株価データ（2026年3月16日取得）
• Grok調査（2026年3月17日）- 最新市場動向

---

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