[Puppeteer] ＜初めてのPuppeteer＞ EMAを使ってGC,DC判定売買bot（雛形）
読者の方から「2本のEMAを使ったGC(ゴールデンクロス)とDC(デッドクロス)判定のbotを作りたい」と要望があったので、Puppeteerを使ってさくっと雛形を作ってみました。
GC(ゴールデンクロス)とDC(デッドクロス)
釈迦に説法かもしれませんが、GC,DCについては
あたりを参考にしてもらうとして、さっそくEMAのGC,DCで売買するbotの雛形を作ってみましょう。
EMA計算
EMA計算は以前の回で紹介しています。
その時のロジックをそっくり拝借して次のように実装します。
EMA期間を３つ取れるようにしておきましょう。今回使用するEMA期間は２種類だけですが、将来パーフェクトオーダーにも対応したいと思っています。
# ==========================================================
# 移動平均（EMA の算出）
# ==========================================================
def __calc_ema(self, df):
ema = pd.DataFrame(OrderedDict({
'EMA_S': df['close'].ewm(span=self._config['EMA_S']).mean(),
'EMA_M': df['close'].ewm(span=self._config['EMA_M']).mean(),
'EMA_L': df['close'].ewm(span=self._config['EMA_L']).mean()
}))
# print(ema)
return ema
これでEMAを３期間計算する関数ができました。
GC・DC判定
EMAのGCは
短期線「EMA_S」が中期線「EMA_M」を上抜いたら「買い(buy)」
として、DCは
中期線「EMA_M」が短期線「EMA_S」を上抜いたら「売り(sell)」
とします。
ソースコードは次のようにしました。
引数のmaにはEMAの短期、中期、長期の値のDataFrame(Pandas)が格納されているとします。
# ==========================================================
# EMA 短期、中期でのドテン計算
# ==========================================================
def __calc_doten(self, ma):
last = ma.iloc[-1] # 直近のEMA
pre = ma.iloc[-2] # その一つ前のEMA
ret = 'none'
if (last['EMA_S'] > last['EMA_M']) and (pre['EMA_S'] <= pre['EMA_M']) :
# 短期線が中期線を上抜いた ー＞ 買い
ret = 'buy'
elif (last['EMA_S'] < last['EMA_M']) and (pre['EMA_S'] >= pre['EMA_M']) :
# 短期線が中期線を下抜いた ー＞ 売り
ret = 'sell'
return ret
これで判定部分はできましたね。
今回、処理を簡単にするためにチャタリング(小さい交差が連続して現れてパタパタと売買シグナルを出してしまう現象)については考慮していません。
ポジションによる売買処理
保有ポジションに従って、どっちのシグナルを優先するかを決めます。
ポジションが無い(初期状態)で、買いシグナル(GC)を検知したら指定ロット数買い、売りシグナル(DC)を検知したら指定ロット数売り。
ポジションがプラスの時(つまり前回が買い)に売りシグナル(DC)を検知したら指定ロットの倍を売り。
ポジションがマイナスの時(つまり前回が売り)に買いシグナル(GC)を検知したら指定ロットの倍を買い。
チャネルブレイクアウトみたいにドタンバタンと売り買いを繰り返す仕様です。
ソースコードは次のようにしました。
# ------------------------------------------------------
# 売買
# ------------------------------------------------------
if pos_qty == 0:
# --------------------------------------------------
# position = 0
# --------------------------------------------------
if doten == 'buy':
# ----------------------------------------------
# 買い
# ----------------------------------------------
self.__market_order('buy', self._config['LOT_SIZE'])
elif doten == 'sell':
# ----------------------------------------------
# 売り
# ----------------------------------------------
self.__market_order('sell', self._config['LOT_SIZE'])
elif pos_qty > 0:
# --------------------------------------------------
# position > 0
# --------------------------------------------------
if doten == 'sell':
# ----------------------------------------------
# 売り
# ----------------------------------------------
self.__market_order('sell', self._config['LOT_SIZE'] * 2)
elif pos_qty < 0:
# --------------------------------------------------
# position < 0
# --------------------------------------------------
if doten == 'buy':
# ----------------------------------------------
# 買い
# ----------------------------------------------
self.__market_order('buy', self._config['LOT_SIZE'] * 2)
上記の「__market_order」は成行注文関数です。
後述しますので、今はこのまま読み流してください。
Puppetとして実装
PuppeteerのPuppetとして実装してみましょう。
PuppeteerのPuppetsフォルダの下に
- Puppets/
- ma/
- ma.py
- ma.jsonのようにフォルダとファイルを用意します。
それぞれのファイルの内容は以下のようにしました。
ma.py
# -*- coding: utf-8 -*-
# ==========================================
# サンプル・ストラテジ
# ==========================================
import time
from datetime import datetime as dt, timezone as tz, timedelta as delta
import pandas as pd
from collections import OrderedDict
from puppeteer import Puppeteer
# ==========================================
# Puppet(傀儡) クラス
# param:
# puppeteer: Puppeteerオブジェクト
# ==========================================
class Puppet(Puppeteer):
# ==========================================================
# 初期化
# param:
# puppeteer: Puppeteerオブジェクト
# ==========================================================
def __init__(self, Puppeteer):
self._exchange = Puppeteer._exchange # 取引所オブジェクト(ccxt.bitmex)
self._logger = Puppeteer._logger # logger
self._config = Puppeteer._config # 定義ファイル
# ==========================================================
# 売買実行
# param:
# ticker: Tick情報
# orderbook: 板情報
# position: ポジション情報
# balance: 資産情報
# candle: ローソク足
# ==========================================================
def run(self, ticker, orderbook, position, balance, candle):
# --------------------------
# ここに処理を記述します
# --------------------------
# ------------------------------------------------------
# ポジションサイズ
# ------------------------------------------------------
pos_qty = position[0]['currentQty'] if len(position) != 0 else 0
# for DEBUG
#self._logger.info('pos_qty:{}'.format(pos_qty))
# ------------------------------------------------------
# ローソク足
# ------------------------------------------------------
df = self.__get_candleDF(candle)
# for DEBUG
#self._logger.info('candle: {}'.format(df.tail(5)))
ema = self.__calc_ema(df)
# for DEBUG
#self._logger.info('ema: {}'.format(ema.tail(5)))
doten = self.__calc_doten(ema)
# for DEBUG
#self._logger.info('doten: {}'.format(doten))
# ------------------------------------------------------
# 売買
# ------------------------------------------------------
if pos_qty == 0:
# --------------------------------------------------
# position = 0
# --------------------------------------------------
if doten == 'buy':
# ----------------------------------------------
# 買い
# ----------------------------------------------
self.__market_order('buy', self._config['LOT_SIZE'])
elif doten == 'sell':
# ----------------------------------------------
# 売り
# ----------------------------------------------
self.__market_order('sell', self._config['LOT_SIZE'])
elif pos_qty > 0:
# --------------------------------------------------
# position > 0
# --------------------------------------------------
if doten == 'sell':
# ----------------------------------------------
# 売り
# ----------------------------------------------
self.__market_order('sell', self._config['LOT_SIZE'] * 2)
elif pos_qty < 0:
# --------------------------------------------------
# position < 0
# --------------------------------------------------
if doten == 'buy':
# ----------------------------------------------
# 買い
# ----------------------------------------------
self.__market_order('buy', self._config['LOT_SIZE'] * 2)
# ==========================================================
# ローソク足 DataFrame 取得
# ==========================================================
def __get_candleDF(self, candle):
# -----------------------------------------------
# Pandasのデータフレームに
# -----------------------------------------------
df = pd.DataFrame(candle,
columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
# -----------------------------------------------
# 日時データをDataFrameのインデックスにする
# -----------------------------------------------
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True, infer_datetime_format=True) # UNIX時間(ミリ秒)を変換, UTC=TrueでタイムゾーンがUTCに設定される, infer_datetime_format=Trueは高速化に寄与するとのこと。
df = df.set_index('timestamp')
return df
# ==========================================================
# 移動平均（EMA の算出）
# ==========================================================
def __calc_ema(self, df):
ema = pd.DataFrame(OrderedDict({
'EMA_S': df['close'].ewm(span=self._config['EMA_S']).mean(),
'EMA_M': df['close'].ewm(span=self._config['EMA_M']).mean(),
'EMA_L': df['close'].ewm(span=self._config['EMA_L']).mean()
}))
# print(ema)
return ema
# ==========================================================
# EMA 短期、中期でのドテン計算
# ==========================================================
def __calc_doten(self, ma):
last = ma.iloc[-1] # 直近のEMA
pre = ma.iloc[-2] # その一つ前のEMA
ret = 'none'
if (last['EMA_S'] > last['EMA_M']) and (pre['EMA_S'] <= pre['EMA_M']) :
# 短期線が中期線を上抜いた ー＞ 買い
ret = 'buy'
elif (last['EMA_S'] < last['EMA_M']) and (pre['EMA_S'] >= pre['EMA_M']) :
# 短期線が中期線を下抜いた ー＞ 売り
ret = 'sell'
return ret
# ==========================================================
# 成行注文
# param:
# side: buy or sell
# size: orderロット数
# return:
# order
# ==========================================================
def __market_order(self, side, size):
order = None
try:
order = self._exchange.create_order(
self._config['SYMBOL'],
type='market',
side=side,
amount=size
)
self._logger.debug('■ market order={}'.format(order))
except Exception as e:
self._logger.error('■ market order: exception={}'.format(e))
order = None
return order
ma.json
{
"//" : "===============================================",
"//" : " システムで利用",
"//" : "===============================================",
"//" : "取引所のapiKey, secretを設定します",
"APIKEY" : "YOUR_APIKEY",
"SECRET" : "YOUR_SECRET",
"//" : "bitmex取引所で対応する通貨ペア等を記述",
"SYMBOL" : "BTC/USD",
"INFO_SYMBOL" : "XBTUSD",
"COIN_BASE" : "BTC",
"COIN_QUOTE" : "USD",
"//" : "bitmex取引所の価格の最小幅(0.5ドル)",
"PRICE_UNIT" : 0.5,
"//" : "TestNetを使うか？(使う: true, 使わない: false)",
"USE_TESTNET" : true,
"//" : "ticker, orderbook, position, balance, candle のどれを利用するかを指定する。Falseを指定した場合はそのデータは取得しない",
"USE" : {
"TICKER" : false,
"ORDERBOOK" : false,
"POSITION" : true,
"BALANCE" : false,
"CANDLE" : true
},
"//" : "ローソク足の収集定義。",
"CANDLE" : {
"//" : "ローソク足の足幅を設定する。設定値= 1m, 5m, 1h, 1d",
"TIMEFRAME" : "5m",
"//" : "データ取得開始時刻(UNIXTIME：1ミリ秒)、使用しない場合 もしくは自動の場合は null(None) を指定",
"SINCE" : null,
"//" : "取得件数(未指定:100、MAX:500)",
"LIMIT" : null,
"//" : "True(New->Old)、False(Old->New) 未指定時はFlase",
"REVERSE" : false,
"//" : "True(最新の未確定足を含む)、False(含まない) 未指定はTrue",
"PARTIAL" : false
},
"//" : "板情報の収集定義。",
"ORDERBOOK" : {
"//" : "取得件数(未指定:25、MAX:取引所による？)",
"LIMIT" : null
},
"//" : "websocketを使用するかどうかを指定",
"USE_WEBSOCKET" : false,
"//" : "ログレベルを指定。（'CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG'）",
"LOG_LEVEL" : "INFO",
"//" : "インターバルを秒で設定",
"INTERVAL" :60,
"//" : "discord通知用URL",
"DISCORD_WEBHOOK_URL" : "",
"//" : "資産状況通知をするか",
"USE_SEND_BALANCE" : false,
"//" : "===============================================",
"//" : " ユーザで自由に定義",
"//" : "===============================================",
"//" : "売買するサイズ",
"LOT_SIZE" :30,
"//" : "EMAの短期、中期、長期の期間指定（単位：整数）",
"EMA_S" :3,
"EMA_M" :5,
"EMA_L" :15
}ma.jsonのAPIKEY,SECRETには皆さんの方で取得したtestnet.bitmex.comのAPIKEYとSECRETを設定してください。
実行
設定ができたら以下のコマンドで実行できます。
5分足を用いて、短期(3本)、中期(5本)のEMAで動作するはずです。
python3 puppeteer.py puppets/ma/ma.py puppets/ma/ma.jsonログはコメントアウトしていますので、動作中の値の変化を見たい場合は「#」を外して頂ければ動作状況を確認できます。
df = self.__get_candleDF(candle)
# for DEBUG
#self._logger.info('candle: {}'.format(df.tail(5)))
ema = self.__calc_ema(df)
# for DEBUG
#self._logger.info('ema: {}'.format(ema.tail(5)))
doten = self.__calc_doten(ema)
# for DEBUG
#self._logger.info('doten: {}'.format(doten))
#for DEBUGと書かれた行の下の行がログ出力部分です。
最後に
雛形としては以上です。
異常処理や弱い交差対応などなど、まだまだ実装すべき部分は多いと思いますが、ひとまず基本形ができました。
皆さんの方でも色々拡張してみてください。
Puppeteerはこちらから
楽しいbotライフを！
いいなと思ったら応援しよう！
