kabuステーション®を自動で起動して、ログインする方法【株の自動売買プログラムをPythonで自作しよう⑦】

記事内に広告を含みます

この記事には広告を含む場合があります。

記事内で紹介する商品を購入することで、当サイトに売り上げの一部が還元されることがあります。

「kabuステーション®APIの使い方を解説」の記事でも触れましたが、kabuステーション®APIを使用するにはkabuステーション®を起動しておく必要があります。また、kabuステーション®の利用可能時間は、6:30から翌早朝6:15までとなっています。kabuステーション®を起動していても朝の6:15に強制的にログアウトされるので、6:30以降に再ログインしなければいけません。

通常であれば手動でkabuステーション®を閉じて、再起動させる必要がありますが、Pythonのプログラムで解決しましょう。

Pythonでは、外部ファイルを実行できたり、マウス操作やキーボード操作などを自動化できる便利なライブラリがあります。

本記事では、kabuステーション®を自動で起動して、ログインする方法について解説します。

自動化に必要なライブラリ
1. subprocess
2. PyAutoGUI
kabuステーション®を起動してから停止させるサンプルコード
自動売買プログラムのサンプルコード
ポイントの解説
おわりに

目次へ

自動化に必要なライブラリ

まずは自動化に必要なライブラリのsubprocessとPyAutoGUIを紹介します。

subprocess

subprocessは、Pythonの標準ライブラリで、Pythonから他のプログラムやコマンドを実行するためのモジュールです。

今回は、Popenクラスを使用してkabuステーション®の起動と停止を行います。

PyAutoGUI

PyAutoGUIは、マウス操作やキーボード操作などを制御できる自動化ライブラリです。

今回は、パスワードを入力してログインするために使用します。

初めて使用する場合はインストールする必要があります。コマンドプロンプトから以下のコマンドを実行します。

TeX

   pip install pyautogui

PyCharmを使用されている場合は、画面下部のPythonパッケージからでもインストールできます。

kabuステーション®を起動してから停止させるサンプルコード

まずは単純にkabuステーション®を起動して、停止させるプログラムを作成します。動作の流れは次の通りです。

kabuステーション®を起動する。
ログインパスワードを入力する。
ログインする。
10秒待つ。
kabuステーション®を停止する。

それではサンプルコードの紹介です。

Python

import subprocess
import pyautogui
import time

kabu_station_path = r'C:\Users\XXXXXXXXXX\KabuS.exe'
login_password = 'YOUR_PASSWORD'


# kabuステーションを起動
def start_kabu_station():
    starting_state = subprocess.Popen(kabu_station_path)
    time.sleep(5)
    pyautogui.typewrite(login_password)
    time.sleep(1)
    pyautogui.keyDown('Enter')

    return starting_state


# kabuステーションを停止
def stop_kabu_station(starting_state):
    starting_state.kill()


kabu_station = start_kabu_station()
time.sleep(10)
stop_kabu_station(kabu_station)

kabu_station_pathは、kabuステーション®の実行ファイルがインストールされているパスを記載してください。

パスの調べ方ですが、kabuステーション®を起動した状態でタスクマネージャーを開き、kabuステーション®を右クリックして「ファイルの場所を開く」を選択します。

開かれたフォルダのKabuS.exeを右クリックして「パスのコピー」でkabuステーション®のパスが取得できます。

kabuステーション®にログインで使用するパスワードをlogin_passwordに記載して下さい。PyAutoGUIでパスワードを入力して、ログインすることができます。

このプログラムを実行すると、kabuステーション®が起動して、停止する一連の動作ができます。

自動売買プログラムのサンプルコード

前回の記事で紹介した株の自動売買プログラムに、kabuステーション®の起動・停止機能を追加したサンプルコードを紹介します。

Python

from datetime import datetime
import time
import urllib.request
import urllib.error
import pprint
import json
import yfinance as yf
import talib
import subprocess
import pyautogui


# 設定項目
kabu_station_path = r'C:\Users\XXXXXXXXXX\KabuS.exe'
api_password = 'YOUR_PASSWORD'
login_password = 'YOUR_PASSWORD'
symbol = '9432'
qty = 100

signal = None


# トークンを発行する関数
def generate_token():
    obj = {'APIPassword': api_password}
    json_data = json.dumps(obj).encode('utf8')

    url = 'http://localhost:18080/kabusapi/token'
    req = urllib.request.Request(url, json_data, method='POST')
    req.add_header('Content-Type', 'application/json')

    try:
        with urllib.request.urlopen(req) as res:
            content = json.loads(res.read())
            token_value = content.get('Token')
    except urllib.error.HTTPError as e:
        print(e)

    return token_value


# 残高照会する関数
def check_positions(token):
    url = 'http://localhost:18080/kabusapi/positions'
    params = {'product': 1}  # product - 0:すべて、1:現物、2:信用、3:先物、4:OP
    params['symbol'] = symbol  # symbol='xxxx'
    params['side'] = '2'  # 1:売、2:買
    params['addinfo'] = 'false'  # true:追加情報を出力する、false:追加情報を出力しない　※追加情報は、「現在値」、「評価金額」、「評価損益額」、「評価損益率」を意味します
    req = urllib.request.Request('{}?{}'.format(url, urllib.parse.urlencode(params)), method='GET')
    req.add_header('Content-Type', 'application/json')
    req.add_header('X-API-KEY', token)

    try:
        with urllib.request.urlopen(req) as res:
            print(res.status, res.reason)
            for header in res.getheaders():
                print(header)
            print()
            content = json.loads(res.read())
            pprint.pprint(content)
    except urllib.error.HTTPError as e:
        print(e)
        content = json.loads(e.read())
        pprint.pprint(content)
    except Exception as e:
        print(e)

    if not content or content[0]['LeavesQty'] == 0.0:
        check_position = False
        check_positions_side = None

    else:
        check_position = True
        if content[0]['Side'] == '2':
            check_positions_side = 'BUY'

    return check_position, check_positions_side


# 板情報を取得する関数
def get_price_data():
    symbol_data = yf.download(symbol + '.T', period='5d', interval='30m')

    return symbol_data


def get_board_information(token):
    url = 'http://localhost:18080/kabusapi/board/' + symbol + '@1'
    req = urllib.request.Request(url, method='GET')
    req.add_header('Content-Type', 'application/json')
    req.add_header('X-API-KEY', token)

    try:
        with urllib.request.urlopen(req) as res:
            print(res.status, res.reason)
            for header in res.getheaders():
                print(header)
            print()
            content = json.loads(res.read())
            pprint.pprint(content)
    except urllib.error.HTTPError as e:
        print(e)
        content = json.loads(e.read())
        pprint.pprint(content)
    except Exception as e:
        print(e)

    return content


# シグナルを判定する関数
def entry_signal(df):
    df['RSI'] = talib.RSI(df['Close'], timeperiod=14)

    if df.iat[-2, 6] <= 30 < df.iat[-1, 6]:
        signal = 'BUY'
        print('買いのシグナルが発生しました')

    elif df.iat[-2, 6] >= 70 > df.iat[-1, 6]:
        signal = 'SELL'
        print('売りのシグナルが発生しました')

    else:
        signal = None

    return signal


# 買い注文を出す関数
def kabusapi_sendorder_cash(token):
    obj = {'Password': login_password,
           'Symbol': symbol,
           'Exchange': 1,
           'SecurityType': 1,
           'Side': '2',
           'CashMargin': 1,
           'DelivType': 2,
           'FundType': 'AA',
           'AccountType': 2,
           'Qty': qty,
           'FrontOrderType': 10,
           'Price': 0,
           'ExpireDay': 0
           }
    json_data = json.dumps(obj).encode('utf-8')

    url = 'http://localhost:18080/kabusapi/sendorder'
    req = urllib.request.Request(url, json_data, method='POST')
    req.add_header('Content-Type', 'application/json')
    req.add_header('X-API-KEY', token)

    try:
        with urllib.request.urlopen(req) as res:
            print(res.status, res.reason)
            for header in res.getheaders():
                print(header)
            print()
            content = json.loads(res.read())
            pprint.pprint(content)
    except urllib.error.HTTPError as e:
        print(e)
        content = json.loads(e.read())
        pprint.pprint(content)
    except Exception as e:
        print(e)


# 売り注文を出す関数
def kabusapi_sendorder_cash_close(token):
    obj = {'Password': login_password,
           'Symbol': symbol,
           'Exchange': 1,
           'SecurityType': 1,
           'Side': '1',
           'CashMargin': 1,
           'DelivType': 0,
           'FundType': '  ',
           'AccountType': 2,
           'Qty': qty,
           'FrontOrderType': 10,
           'Price': 0,
           'ExpireDay': 0
           }
    json_data = json.dumps(obj).encode('utf-8')

    url = 'http://localhost:18080/kabusapi/sendorder'
    req = urllib.request.Request(url, json_data, method='POST')
    req.add_header('Content-Type', 'application/json')
    req.add_header('X-API-KEY', token)

    try:
        with urllib.request.urlopen(req) as res:
            print(res.status, res.reason)
            for header in res.getheaders():
                print(header)
            print()
            content = json.loads(res.read())
            pprint.pprint(content)
    except urllib.error.HTTPError as e:
        print(e)
        content = json.loads(e.read())
        pprint.pprint(content)
    except Exception as e:
        print(e)


# kabuステーションを起動
def start_kabu_station():
    starting_state = subprocess.Popen(kabu_station_path)
    time.sleep(5)
    pyautogui.typewrite(login_password)
    time.sleep(1)
    pyautogui.keyDown('Enter')

    return starting_state


# kabuステーションを停止
def stop_kabu_station(starting_state):
    starting_state.kill()


# メイン処理
kabu_station = start_kabu_station()
time.sleep(10)
token_value = generate_token()
position, side = check_positions(token_value)

while True:

    hour = datetime.now().hour

    if hour == 6:
        stop_kabu_station(kabu_station)
        time.sleep(3600)
        kabu_station = start_kabu_station()

    else:
        # ポジションがある場合
        if position is True and side == 'BUY':
            price = get_price_data()
            signal = entry_signal(price)

            # シグナルが発生した時
            if signal == 'SELL':
                kabusapi_sendorder_cash_close(token_value)
                signal = None
                time.sleep(10)
                position, side = check_positions(token_value)
                time.sleep(1800)

            else:
                time.sleep(60)

        # ポジションがない場合
        else:
            price = get_price_data()
            signal = entry_signal(price)

            # シグナルが発生した時
            if signal == 'BUY':
                kabusapi_sendorder_cash(token_value)
                signal = None
                time.sleep(10)
                position, side = check_positions(token_value)
                time.sleep(1800)

            else:
                time.sleep(60)

ポイントの解説

前回と比較すると、主にメイン処理を変更しています。

Python

# メイン処理
kabu_station = start_kabu_station()
time.sleep(10)
token_value = generate_token()
position, side = check_positions(token_value)

while True:

    hour = datetime.now().hour

    if hour == 6:
        stop_kabu_station(kabu_station)
        time.sleep(3600)
        kabu_station = start_kabu_station()

    else:
        # ポジションがある場合
        if position is True and side == 'BUY':
            price = get_price_data()
            signal = entry_signal(price)

            # シグナルが発生した時
            if signal == 'SELL':
                kabusapi_sendorder_cash_close(token_value)
                signal = None
                time.sleep(10)
                position, side = check_positions(token_value)
                time.sleep(1800)

            else:
                time.sleep(60)

        # ポジションがない場合
        else:
            price = get_price_data()
            signal = entry_signal(price)

            # シグナルが発生した時
            if signal == 'BUY':
                kabusapi_sendorder_cash(token_value)
                signal = None
                time.sleep(10)
                position, side = check_positions(token_value)
                time.sleep(1800)

            else:
                time.sleep(60)