Recevoir le websocket de l'API kabu station ® en Python

Aperçu

Dans la continuité de précédent, au Kabucom Securities utilisera l'API de la station kabu fournie aux individus de Python. Cette fois, la marque sera livrée par Websocket en Python. En même temps, nous présenterons également le code pour l'enregistrement, la désinscription et la désinscription de tous les problèmes.

environnement

• Windows 10 --Python 3.8.5 (installé à partir du Microsoft Store)

Forfaits supplémentaires

• websockets
• requests
• pyyaml

code

Enregistrement de la marque

import json
import requests
import yaml

# ---

def get_token():
    with open('auth.yaml', 'r') as yml:
        auth = yaml.safe_load(yml)

    url = 'http://localhost:18080/kabusapi/token'
    headers = {'content-type': 'application/json'}
    payload = json.dumps(
        {'APIPassword': auth['PASS'],}
        ).encode('utf8')

    response = requests.post(url, data=payload, headers=headers)

    return json.loads(response.text)['Token']

# ---

token = get_token()

EXCHANGES = {
    1: 'TSE',
    3: 'Certificat de nom',
    5: 'Fortune',
    6: 'Certificat de facture',
}

payload = json.dumps({
    'Symbols': [
        {'Symbol': 8306 ,'Exchange': 1},  # MUFG
        {'Symbol': 9433 ,'Exchange': 1},  # KDDI
        # ...Jusqu'à 50 peuvent être enregistrés
    ],}).encode('utf8')

url = 'http://localhost:18080/kabusapi/register'
headers = {'Content-Type': 'application/json', 'X-API-KEY': token,}
response = requests.put(url, payload, headers=headers)

regist_list = json.loads(response.text)

print('Marque d'enregistrement de la livraison')
for regist in regist_list['RegistList']:
    print("{} {}".format(
        regist['Symbol'],
        EXCHANGES[regist['Exchange']]))

Annulation de l'enregistrement de la marque

Changez uniquement l'URL. L'unité d'affichage peut être détournée.

url = 'http://localhost:18080/kabusapi/unregister'

Annuler tous les stocks enregistrés

payload n'est plus nécessaire.

url = 'http://localhost:18080/kabusapi/unregister/all'
headers = {'Content-Type': 'application/json', 'X-API-KEY': token,}
response = requests.put(url, headers=headers)

Affichage de livraison

Continuer à recevoir la distribution des marques déposées. Quittez avec Ctrl + C. Notez que «.text» n'est pas ajouté à «response» qui est «json.loads».

import asyncio
import json
import websockets

# ---

async def stream():
    uri = 'ws://localhost:18080/kabusapi/websocket'

    async with websockets.connect(uri, ping_timeout=None) as ws:
        while not ws.closed:
            response = await ws.recv()
            board = json.loads(response)
            print("{} {} {}".format(
                board['Symbol'],
                board['SymbolName'],
                board['CurrentPrice'],
            ))

loop = asyncio.get_event_loop()
loop.create_task(stream())
try:
    loop.run_forever()
except KeyboardInterrupt:
    exit()

Puisque le côté serveur n'implémente pas le battement de cœur, l'argument de ping_timeout = None est requis dans websockets.connect.

[Demande] Support ping / pong WebSocket 　　Issue#8 https://github.com/kabucom/kabusapi/issues/8