悠々自作

ぐうたら主婦のプログラミング日記

株の自動売買をしてみる：検証用の環境で売り注文

auカブコム証券のkabuステーション®APIを利用して、株の自動売買にチャレンジしています。 買い注文を出すことができたので、次は売り注文を出して見ます。

kabuステーション API ｜株のことならネット証券会社【auカブコム証券】 を見ると、買い注文とアクセスポイントが一緒なので、 プロパティの売買の箇所を修正すればよさそうです。

SellOrderSenderクラスを実装する

# sell_order_sender.py

import os
import urllib.request
import json
import pprint
from dotenv import load_dotenv

load_dotenv()


class SellOrderSender:

    def __init__(self, config, token):
        self.config = config
        self.token = token

    @property
    def order_details(self):
        return {
            'Password': self.config['APIPassword'],
            'Symbol': os.getenv('NT'),
            'Exchange': 1,
            'SecurityType': 1,
            'Side': '1',
            'CashMargin': 1,
            'DelivType': 0,
            'FundType': '  ',
            'AccountType': 4,
            'Qty': 100,
            'FrontOrderType': 20,
            'Price': 1380,
            'ExpireDay': 0
        }

    def send_order(self):

        json_data = json.dumps(self.order_details).encode('utf-8')

        url = f"{self.config['URL']}sendorder"
        req = urllib.request.Request(url, json_data, method='POST')
        self.set_request_headers(req)
        try:
            with urllib.request.urlopen(req) as res:
                print(res.status, res.reason)
                for header in res.getheaders():
                    print(header)
                content = json.loads(res.read())
                pprint.pprint(content)
        except urllib.error.HTTPError as e:
            print(f"HTTPError: {e}")
            content = json.loads(e.read())
            pprint.pprint(content)
        except Exception as e:
            print(f"ExceptionError: {e}")

    def set_request_headers(self, req):
        req.add_header('Content-Type', 'application/json')
        req.add_header('X-API-KEY', self.token)

買い注文同様、売り注文もできました。 しかし、現物取引なので、持っていない株は売れません。 自分が対象銘柄の株を持っているかどうか確認します

対象銘柄の株を持っているかどうか確認する

残高照会 が使えそうです。

import urllib.request
import json
import pprint
import os
from dotenv import load_dotenv

load_dotenv()



url = f"{self.config['URL']}positions"
params = { 'product': 1 }     
params['symbol'] = os.getenv('NT')        
params['side'] = '2'              
params['addinfo'] = 'true'    
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', self.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)

検証用の環境で実行すると、

200 OKと共に空配列が返ってきているようです。 うーん。
検証用の環境で株を持っているということがあり得ないからでしょうか。 本番用の環境で実行してみると、 上手く行きました。

# positions.py
（略）
       try:
            with urllib.request.urlopen(req) as res:
                content = json.loads(res.read())
                return content[0]['LeavesQty'] if self.config == 'ENV_PROD' else 100

（略）

検証用の環境で空配列を返されたら、テストできないので、 本番環境ではcontent[0]['LeavesQty']の値、検証用環境では100を返すようにします。

株を持っていたら売る。株を持っていなかったら買う

# main.py
（略）
if __name__ == '__main__':
    config = return_environment_from_args()
    token_manager = TokenManager(config)
    token = token_manager.fetch_token()

    positions = Positions(config, token)
    position_quantity = positions.get_positions()

    if position_quantity == 0:
        buy_order = BuyOrderSender(config, token)
        buy_order.send_order()
    else:
        sell_order = SellOrderSender(config, token)
        sell_order.send_order()

（略）

非常に単純なので、実際にこの条件で売買するわけにはいきませんが、 条件によって株を売ったり買ったりすることが出来ました。

ソースコード

# main.py
from src.token_manager import TokenManager
from src.buy_order_sender import BuyOrderSender
from src.sell_order_sender import SellOrderSender
from src.positions import Positions
from config import Config
import argparse


def return_environment_from_args():
    parser = argparse.ArgumentParser(description='Fetch API token')
    parser.add_argument('--env', choices=['test', 'prod'], default='test', help='Specify environment (test or prod)')
    args = parser.parse_args()
    return Config.ENV_TEST if args.env == 'test' else Config.ENV_PROD


if __name__ == '__main__':
    config = return_environment_from_args()
    token_manager = TokenManager(config)
    token = token_manager.fetch_token()

    positions = Positions(config, token)
    position_quantity = positions.get_positions()

    if position_quantity == 0:
        buy_order = BuyOrderSender(config, token)
        buy_order.send_order()
    else:
        sell_order = SellOrderSender(config, token)
        sell_order.send_order()



# positions.py
import urllib.request
import json
import pprint
import os
from dotenv import load_dotenv

load_dotenv()


class Positions:
    def __init__(self, config, token):
        self.config = config
        self.token = token

    def get_positions(self):
        url = f"{self.config['URL']}positions"
        params = {
            'product': '1',
            'symbol': os.getenv('NT'),
            'side': '2',
            'addinfo': 'true'
        }
        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', self.token)

        try:
            with urllib.request.urlopen(req) as res:
                content = json.loads(res.read())
                return content[0]['LeavesQty'] if self.config == 'ENV_PROD' else 0
        except urllib.error.HTTPError as e:
            print(e)
            content = json.loads(e.read())
            pprint.pprint(content)
        except Exception as e:
            print(e)