私のauカブコム証券のkabuステーションで自動売買プログラムの説明(3) 現在の株価を監視 (board.py)

スポンサーリンク

board.pyのソースコード

import urllib.request
import json
import pprint
import time
import pandas as pd
import numpy as np
import settings
import sqlite3
from statistics import mean 
import datetime
import sendorder_entry

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

  curPriceList = []

  #標準偏差初期化
  std_val = 0

 #無限ループ
  while True:
    try:
      print('###board')
      with urllib.request.urlopen(req) as res:
        content = json.loads(res.read())

        #現在価格を取得
        curPrice = content["CurrentPrice"]
        if curPrice is None:
          #現在価格が取れない場合はまだ相場が開いてないのでスキップ
          time.sleep(60)
          continue

        #現在価格を配列に追加
        curPriceList.append(curPrice)
        
        #期間が20なので、20個 値が溜まるまで処理は行わない
        #逆に20個より前の値は捨てる
        if len(curPriceList) > 20:
          del curPriceList[0:len(curPriceList)-20]

          #1つ前の標準偏差値を退避
          pre_std_val = std_val
          #標準偏差を求める
          std_val = np.std(curPriceList)

          #RSIを求める
          delta_list = np.diff(curPriceList)
          up_list, down_list = delta_list.copy(), delta_list.copy()
          up_list[up_list < 0] = 0
          down_list[down_list > 0] = 0
          up_list_mean = up_list.mean()
          down_list_mean = down_list.mean()
          rsi =  up_list_mean / (up_list_mean + abs(down_list_mean)) * 100

          #RSIと標準偏差が閾値を超えているかつ、
          #標準偏差が1つ前より減少していたらエントリ
          if rsi > settings.rsi_threshold and \
             std_val > settings.sd_threshold and \
             pre_std_val >std_val:
            # 指定時間を超えていたらエントリしない
            nowtime = datetime.datetime.now()
            if (nowtime > settings.morningStartTime \
                and nowtime < settings.morningStopTime):
              # エントリ
              sendorder_entry.sendorder_entry()

    except urllib.error.HTTPError as e:
      print(e)
      content = json.loads(e.read())
      pprint.pprint(content)
    except Exception as e:
      print(e)
    
    #1分足なので60秒スリープ
    time.sleep(60)

if __name__ == "__main__":
  import sys
  board()

基本は、kabuステーションAPIのサンプルをベースに株価情報を取得して、エントリ条件を判定し、エントリ基準に達したら、エントリ処理を呼び出します。

詳細説明

1分間隔で株価を取得

 #無限ループ
  while True:
    try:
      print('###board')
      with urllib.request.urlopen(req) as res:
        content = json.loads(res.read())

        #現在価格を取得
        curPrice = content["CurrentPrice"]
    #1分足なので60秒スリープ
    time.sleep(60)

全体を「while True:」の中に入れることによって、無限に同じ処理を繰り返します。
そして、その最後で「time.sleep(60)」で60秒の待つことにより、60秒ごとに処理を繰り返します。

このプログラムでは、1分足のデータを使っているので、このwhile中で株価情報を取得しデータを貯めていくことで1分足のデータを作成します。

1分足のデータがほしい場合はsleep時間が60秒ですが、5分足の場合は300秒、15分足の場合は900秒にすることにより、同じように時間足のデータが取得できます。

市場がオープンしていない場合は株価情報が取れないので何もしない

        if curPrice is None:
          #現在価格が取れない場合はまだ相場が開いてないのでスキップ
          time.sleep(60)
          continue

東京証券取引所の開場は9:00なので、その前に株価を取得すると値が取得できません。その状態はPythonでは「None」という状態なので、その場合は何もせずに処理を抜けます。

ここでも忘れずにsleep処理を入れないと、休みなく呼び出しをしてしまいます。kabuステーションAPIは呼び出しの流量制限が1秒間あたり10回と、割と余裕があるのですが、流量制限にひっかからないことに越したことはありません。

取得した1分毎の株価を20分配列に格納する

        #現在価格を配列に追加
        curPriceList.append(curPrice)
        
        #期間が20なので、20個 値が溜まるまで処理は行わない
        #逆に20個より前の値は捨てる
        if len(curPriceList) > 20:
          del curPriceList[0:len(curPriceList)-20]

取得した株価をどんどん「curPriceList」という名前の配列に入れていきます。

このプログラムでは、エントリ判断に1分足で20分ぶんのデータしか使わないので、配列の個数が20個を超えた場合、古いデータを削除します。

逆に、43行目のif文により、データが20分ぶん溜まるまではそれ以降の処理をしないようにしています。

厳密に言えば、「 > 20」なので21分ぶん溜まるまでは処理を行いません。しかし、そこにこだわるとプログラムが複雑になるし、最初の1分エントリが遅れたからどうだということもないのでそのままにしています。

標準偏差を求める

  #標準偏差初期化
  std_val = 0
          #1つ前の標準偏差値を退避
          pre_std_val = std_val
          #標準偏差を求める
          std_val = np.std(curPriceList)

pythonのNumPyというライブラリにある、「std」という関数を使うと配列から簡単に標準偏差が取得できます。

このプログラムでは、標準偏差が前回から下がっているかどうかをエントリ条件の1つとしているので、後で比較するために、標準偏差を新たに取得する前に、前回の値を別の「pre_std_val」という変数に退避しておきます。

1番最初は「std_val」が0で初期化された状態なので、「pre_std_val」も0になります。前回の標準偏差の値というpre_std_valの変数の位置づけとしては正確ではありません。
しかし、pre_std_valが0の場合は、その後でstd_valがどんな値になったとしても、pre_std_valより小さくなることはないので、エントリ条件が誤判定されることはなく、問題ありません。

RSI(カトラー式)を求める

普通に配列をループしながら値と値の間の差をサマリしていってもよいのですが、ちょうどよいサンプルコードが見つかったので、流用しました。

Relative Strength Index in python pandas
I am new to pandas. What is the best way to calculate the relative strength part in the RSI indicator in pandas? So far I got the following: from pylab import ...
          #RSIを求める
          delta_list = np.diff(curPriceList)

NumPyの「diff」という関数で、配列から、それぞれの要素とその1つ前の要素との差分の配列を作成します。

          up_list, down_list = delta_list.copy(), delta_list.copy()
          up_list[up_list < 0] = 0
          down_list[down_list > 0] = 0

それを、プラスの差分、マイナスの差分用の配列にそれぞれコピーし、Pythonの「スライス」というフィルタ機能を使用して、プラスの差分用の配列のマイナスの要素を0に、マイナスの差分用のプラスの要素を0に変換します。

          up_list_mean = up_list.mean()
          down_list_mean = down_list.mean()
          rsi =  up_list_mean / (up_list_mean + abs(down_list_mean)) * 100

そして、statisticsというpythonのライブラリの「mean」という配列の平均値を取得する関数を使って、それぞれの配列の平均値を取得して、カトラーのRSIの公式に当てはめまてRSIを算出します。

RS={\frac  {{\text{SMA}}(U,n)}{{\text{SMA}}(D,n)}}
Relative strength index

エントリ基準判定

          #RSIと標準偏差が閾値を超えているかつ、
          #標準偏差が1つ前より減少していたらエントリ
          if rsi > settings.rsi_threshold and \
             std_val > settings.sd_threshold and \
             pre_std_val >std_val:

コメントのとおりですね。RSIが70以上、標準偏差が100以上、標準偏差が1分前の値よりも下がっていたらエントリ基準クリアです。

具体的な各値は、グローバル変数格納用モジュール(setting.py)に定数として設定してあります。

エントリ可能時間判定

            # 指定時間を超えていたらエントリしない
            nowtime = datetime.datetime.now()
            if (nowtime > settings.morningStartTime \
                and nowtime < settings.morningStopTime) :

午前中9:20~10:30をエントリ可能時間としているので、現在時刻がその範囲内にあるかのチェックをしています。

エントリ時間に制限を設けているのは、このプログラムでは前場の中で逆張り反転を狙っているため、前場の中で反転が可能と思われるエントリ時間を逆算して設定しています。

エントリ処理呼び出し

              # エントリ
              sendorder_entry.sendorder_entry()

数珠繋ぎ方式により、次のエントリ処理を呼び出しています。

コメント