SBI証券で自動売買はPython自作できるのか
公開APIがない現実と、口座を活かす4つの実装ルート

# データ取得・処理
pandas>=2.2
numpy>=1.26
yfinance>=0.2.40          # 検証用。実運用はJ-Quants等の公式APIを推奨
requests>=2.32
python-dotenv>=1.0

# 保存
pyarrow>=17.0             # Parquet保存用

# Discord Webhook（通知先）
DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/xxxxx/yyyyy

# J-Quants を使う場合
JQUANTS_MAIL=your_mail@example.com
JQUANTS_PASS=your_password

# 本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
# SBI証券では発注APIが使えないため、「判断まで」を自動化し発注は手動で行う設計。
import os
import time
from dataclasses import dataclass
from typing import List, Optional

import pandas as pd
import requests
import yfinance as yf
from dotenv import load_dotenv

load_dotenv()
WEBHOOK = os.getenv("DISCORD_WEBHOOK_URL")

# 監視する銘柄（東証は「コード.T」）
WATCH_LIST = ["7203.T", "6758.T", "9984.T", "8306.T", "6501.T"]

FAST, SLOW = 25, 75          # 移動平均の期間
VOL_MULT   = 1.5             # 出来高が平均の何倍以上か
RISK_RATE  = 0.02            # 1銘柄あたりのリスク率（資金比）
CAPITAL    = 1_000_000       # 運用資金（円）


@dataclass
class Signal:
    code: str
    side: str
    price: float
    atr: float
    shares: int
    reason: str


def fetch(code: str, period: str = "1y") -> Optional[pd.DataFrame]:
    """日足を取得する。取得できなければ None。"""
    try:
        df = yf.download(code, period=period, interval="1d",
                         auto_adjust=True, progress=False)
    except Exception as e:
        print(f"[WARN] {code}: 取得失敗 {e}")
        return None
    if df is None or df.empty or len(df) < SLOW + 20:
        return None
    # yfinanceの列がMultiIndexで返る場合があるため平坦化する
    if isinstance(df.columns, pd.MultiIndex):
        df.columns = df.columns.get_level_values(0)
    return df


def atr(df: pd.DataFrame, n: int = 14) -> float:
    """ATR（平均真の値幅）。損切り幅とロット計算に使う。"""
    high, low, close = df["High"], df["Low"], df["Close"]
    prev = close.shift(1)
    tr = pd.concat([high - low, (high - prev).abs(), (low - prev).abs()],
                   axis=1).max(axis=1)
    return float(tr.rolling(n).mean().iloc[-1])


def detect(code: str, df: pd.DataFrame) -> Optional[Signal]:
    """ゴールデンクロス＋出来高増加でエントリーsignalを返す。"""
    df = df.copy()
    df["fast"] = df["Close"].rolling(FAST).mean()
    df["slow"] = df["Close"].rolling(SLOW).mean()
    df["vol_ma"] = df["Volume"].rolling(20).mean()

    prev, last = df.iloc[-2], df.iloc[-1]

    crossed_up = prev["fast"] <= prev["slow"] and last["fast"] > last["slow"]
    vol_ok     = last["Volume"] > last["vol_ma"] * VOL_MULT
    if not (crossed_up and vol_ok):
        return None

    price = float(last["Close"])
    a = atr(df)
    if a <= 0 or price <= 0:
        return None

    # 損切り幅を 2ATR とし、1銘柄のリスクを資金の RISK_RATE に収める
    stop_width = a * 2.0
    raw = (CAPITAL * RISK_RATE) / stop_width
    shares = int(raw // 100 * 100)          # 日本株は100株単位
    if shares < 100:
        return None

    return Signal(
        code=code, side="BUY", price=price, atr=a, shares=shares,
        reason=f"{FAST}日線が{SLOW}日線を上抜け / 出来高が20日平均の{VOL_MULT}倍超",
    )


def notify(signals: List[Signal]) -> None:
    if not WEBHOOK:
        print("[WARN] DISCORD_WEBHOOK_URL が未設定のため通知をスキップします")
        return
    if not signals:
        requests.post(WEBHOOK, json={"content": "本日のシグナルはありません。"}, timeout=15)
        return

    lines = ["**本日の発注候補（SBI証券で手動発注してください）**", ""]
    for s in signals:
        cost = s.price * s.shares
        lines += [
            f"**{s.code}** {s.side}",
            f"  参考価格 : {s.price:,.1f} 円",
            f"  数量     : {s.shares:,} 株（概算 {cost:,.0f} 円）",
            f"  損切目安 : {s.price - s.atr * 2:,.1f} 円（2ATR）",
            f"  根拠     : {s.reason}",
            "",
        ]
    lines.append("※ 発注前に板・出来高・決算発表予定を必ず自分で確認すること。")

    requests.post(WEBHOOK, json={"content": "\n".join(lines)[:1900]}, timeout=15)


def main() -> None:
    found: List[Signal] = []
    for code in WATCH_LIST:
        df = fetch(code)
        if df is None:
            continue
        sig = detect(code, df)
        if sig:
            found.append(sig)
            print(f"[SIGNAL] {sig.code} {sig.side} {sig.shares}株 @ {sig.price:,.1f}")
        time.sleep(1.0)          # データ提供元に負荷をかけない

    notify(found)
    print(f"完了：{len(found)} 件のシグナル")


if __name__ == "__main__":
    main()

@echo off
cd /d C:\stock-bot
call .venv\Scripts\activate.bat
python signal_bot.py >> logs\run.log 2>&1

# 平日 15:30（大引け後）に実行する
30 15 * * 1-5 cd /home/user/stock-bot && /home/user/stock-bot/.venv/bin/python signal_bot.py >> logs/run.log 2>&1

# 本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
# signal_bot.py と同じルールを過去データで検証する
import numpy as np
import pandas as pd
import yfinance as yf

FAST, SLOW = 25, 75
FEE_RATE   = 0.0005      # 売買手数料の想定（往復・料率は各社の体系に合わせる）
SLIPPAGE   = 0.001       # スリッページ 0.1%


def backtest(code: str, period: str = "10y") -> dict:
    df = yf.download(code, period=period, interval="1d",
                     auto_adjust=True, progress=False)
    if isinstance(df.columns, pd.MultiIndex):
        df.columns = df.columns.get_level_values(0)
    if df is None or df.empty:
        return {}

    df["fast"] = df["Close"].rolling(FAST).mean()
    df["slow"] = df["Close"].rolling(SLOW).mean()

    # シグナルは「翌日の寄付きで執行」とし、先読みバイアスを避ける
    df["pos"] = (df["fast"] > df["slow"]).astype(int).shift(1).fillna(0)

    df["ret"] = df["Close"].pct_change().fillna(0.0)
    df["trade"] = df["pos"].diff().abs().fillna(0.0)      # 売買が発生した日は1
    cost = df["trade"] * (FEE_RATE + SLIPPAGE)

    df["strategy"] = df["pos"] * df["ret"] - cost
    equity = (1.0 + df["strategy"]).cumprod()

    total = equity.iloc[-1] - 1.0
    years = len(df) / 252.0
    cagr  = equity.iloc[-1] ** (1 / years) - 1 if years > 0 else 0.0
    dd    = (equity / equity.cummax() - 1.0).min()
    sd    = df["strategy"].std()
    sharpe = (df["strategy"].mean() / sd * np.sqrt(252)) if sd > 0 else 0.0

    return {
        "code": code,
        "総リターン": f"{total:.1%}",
        "CAGR": f"{cagr:.1%}",
        "最大DD": f"{dd:.1%}",
        "シャープ": f"{sharpe:.2f}",
        "売買回数": int(df["trade"].sum()),
    }


if __name__ == "__main__":
    rows = [backtest(c) for c in ["7203.T", "6758.T", "9984.T", "8306.T", "6501.T"]]
    print(pd.DataFrame([r for r in rows if r]).to_string(index=False))