現在動かなくなってしまっている自作Androidアプリ 「画像検索一括ダウンロード」(Windows版は「ImageSpider」)を Flutterを使って作り直してみる。
Flutterを使うことで今度はAndroidに加えてiOS、あわよくばWindowsやLinuxにも対応させられれば。
で、こんな感じ。
基本的な機能ができたので一旦Android版を公開…しようとしたら 久しぶりすぎて色々対応が必要そう…。
…おまちください。
電子ペーパー、クアデルノを買いました。
富士通 10.3型 電子ペーパー(A5サイズ)FUJITSU QUADERNO(クアデルノ) FMV-DPP04
クアデルノはタブレットのようなもので、かばんに入れて持ち運ぶ際にケースが欲しくなりました。
せっかくの軽い端末だし、荒く扱うつもりもないので 傷を防ぐ程度の、軽くて、端末をスッと出し入れできるケースはないものか…。
探してみても意外とありません。 つくってみるか…で割と簡単に使えるものができたので紹介します。
フェルトでできています。材料も道具もすべて100均で買えます。
はじめに以下の図のようなサイズでフェルトを切り出します。つまり以下のサイズですね。
定規を使ってチャコペンで線を引いておいて、線に沿ってハサミで切るといいです。 グルーガンのグルーをつける位置、フェルト生地を折った縁が来る位置にもチャコペンで線を引いておくと目印になります。
生地のサイズはタブレットの厚みや工作の精度で変わってきます。 一度作ってイマイチだったらサイズを変更してもう一度つくりましょう。 材料が安いので失敗も怖くありません。
次に上辺、下辺を折ってグルーガンでくっつけます。数cmグルーを乗せては圧着…を繰り返してくっつけていきます。
そして、上辺、下辺ができたら、真ん中で折って、左右をくっつけて…
…完成です!
電子書籍は便利なのですが
自分は電子ペーパーで読みたくて
持っている Kindle Paperwhite だと画面が小さい。
本によっては問題ないのですが
読みたい本の多くは大きな画面でないと読みづらく
それがために渋々紙の本をかうことが何度も。
比較的安価で画面の大きなQUADERNOとかで
Kindleの電子書籍が読めたらもっと世界が広がるのになぁ。
ということで
Kindleの電子書籍をPDFファイルにするツールをつくりました。
Reader Capture - 電子書籍をPDFファイル/ZIPファイルに変換!
Kindleの電子書籍を
Kindle for PC (Windows)のスクリーンショットを撮ることで
PDFファイルにします。
ちょっと調べると他にもやり方はあるようですが
簡単に使えること、使い続けられること、を重視してつくりました。
仕組みは汎用的なので他の電子書籍リーダーアプリでも使えるかもしれません。
保証はしませんけど…。
前回はシミュレーターでの自動取引を実際の取引所での自動取引に対応しやすくするためにプログラムの改修を行いました。
今回はそれを更に進めていよいよ実際の取引所での自動取引を実現します。
実現しますがアルゴリズムがしょぼくて勝てないことはシミュレーションで確認済みなので実行は動作確認にとどめたいと思います…。
前回準備を行ったのでやるべきことはそう多くはありません。
順番に進めていきます。
Apiクラスを継承し、GMOコインのAPIを呼んで、実際の取引所であるGMOコインとやり取りするGmoCoinApiクラスを作成します。
長くなるのでソースコードは割愛します。
GMOコインのAPIの呼び出しについては
Pythonでビットコインの価格を取得する(1) - てけとーぶろぐ。
を参考にしていただければと思います。
2点アドバイスがあります。
メインプログラムを作成します。概要としては以下の通りです。
from gmo_coin_api import GmoCoinApi from trader import Trader from apscheduler.schedulers.background import BackgroundScheduler import logging.config from app_const import AppConst import os from email.mime.text import MIMEText from email.utils import formatdate import smtplib from time import sleep GMAIL_USER = os.environ.get("GMAIL_USER") GMAIL_PASSWORD = os.environ.get("GMAIL_PASSWORD") TO_ADDRESS = os.environ.get("TO_ADDRESS") logging.config.fileConfig(AppConst.LOGGING_CONFIG_FILE_PATH) logger = logging.getLogger('fileLogger') gmo_coin_api = GmoCoinApi('https://api.coin.z.com') trader = Trader(gmo_coin_api) def send_mail(subject, body): msg = MIMEText(body) msg['Subject'] = subject msg['From'] = GMAIL_USER msg['To'] = TO_ADDRESS msg['Bcc'] = '' msg['Date'] = formatdate() smtpobj = smtplib.SMTP_SSL('smtp.gmail.com', 465, timeout=10) smtpobj.login(GMAIL_USER, GMAIL_PASSWORD) smtpobj.sendmail(GMAIL_USER, TO_ADDRESS, msg.as_string()) smtpobj.close() def update(): global is_exception_raised try: trader.act() except Exception as exception: logger.error('exception:\n{}'.format(exception)) send_mail('Trade Exception Alert', 'exception:\n{}'.format(exception)) is_exception_raised = True if __name__ == '__main__': # apscheduler のロギングを抑制 logging.getLogger('apscheduler.executors.default').setLevel(logging.WARNING) logging.getLogger('apscheduler.scheduler').setLevel(logging.WARNING) scheduler = BackgroundScheduler() scheduler.add_job(update, trigger='interval', seconds=3, max_instances=10) scheduler.start() is_exception_raised = False try: while True: # TODO 監視システムに生きていることを知らせる if is_exception_raised: break sleep(0.001) except KeyboardInterrupt: logger.info('KeyboardInterrupt') scheduler.shutdown()
3秒毎に特定の処理を実行するところは
Pythonでビットコインの価格を取得する(1) - てけとーぶろぐ。
のプログラムと同じです。
メール通知については
Pythonで株価アラートをつくる(1) - てけとーぶろぐ。
で扱っていますので参考にしていただければと思います。
前回は簡単な自動取引のアルゴリズムをつくって
シミュレーター上でそのアルゴリズムを動かしてみました。
取引の結果は大敗でしたがシミュレーション自体はできました。
ここからアルゴリズムを改良していくのもいいのですが
記事としては自動取引の実現を優先したいと思います。
今回は前回作成したプログラムに手を入れて
実際の取引所での自動取引に対応しやすくします。
実際の取引所での取引への対応とはどういうことでしょうか?
これはシミュレーションに使った自動取引のアルゴリズムをそのまま使って
実際の取引所で取引ができるようにすることと言えます。
前回作成したプログラムで言えば
自動取引のアルゴリズムはTraderクラスとして実装しました。
TraderクラスがSimulatorクラスとやり取りをして取引をしていました。
図にすると以下になります。
これをTraderクラスはそのままに、やり取りの相手を取引所にして取引ができればいいわけです。
図にすると以下になります。
Traderクラスをそのままにやり取りの相手をシミュレーターにしたり取引所にしたりできればいいと言ったのですが、そのためには前回のプログラムから少し改修が必要です。
なぜならSimulatorクラスと取引所ではやり取りの口、すなわちインターフェースが違うからです。
例えばSimulatorクラスは get_cur_ticker() メソッドを呼ぶことで現在のビットコインの価格を返すようになっていましたが、取引所は、この連載で使っているGMOコインであればWeb API の GET /public/v1/ticker を呼ぶことで現在のビットコインの価格を返します。
また今のTraderクラスはシミュレーターとのやり取りに特化した作りになってしまっています。
これらに対してApiクラスを追加して対応します。
TraderクラスがSimulatorクラスや取引所とやり取りする際にApiクラスを介するようにして、Traderクラスからはどちらとやり取りする場合も同じメソッド群でApiクラスとやり取りすることでその後ろにあるSimulatorクラスや取引所とやり取りできるようにするのです。
図にすると以下の通りです。
SimulatorApiクラス、GmoCoinApiクラス、どちらのクラスもTraderクラスから取引をするためのインターフェースが同じになるよう、共通の基底クラスとなるApiクラスを作ります。
Apiクラスには以下のようなメソッドを用意しています。
from datetime import datetime from dataclasses import dataclass from enum import Enum, auto from typing import List import json import requests import os import hmac import time from datetime_util import DatetimeUtil import hashlib from abc import ABCMeta, abstractmethod class OrderStatus(Enum): WAITING = auto() ORDERED = auto() MODIFYING = auto() CANCELLING = auto() CANCELED = auto() EXECUTED = auto() EXPIRED = auto() class Symbol(Enum): JPY = auto() BTC = auto() ETH = auto() BCH = auto() LTC = auto() XRP = auto() @dataclass class Ticker: ask_price: int bid_price: int high_price: int last_price: int low_price: int symbol: str timestamp: datetime volume: float @dataclass class Asset: amount: float symbol: Symbol class Api(metaclass=ABCMeta): @abstractmethod def get_available_amount(self) -> int: pass @abstractmethod def get_ticker(self, symbol: Symbol) -> Ticker: pass @abstractmethod def place_market_buying_order(self, symbol: Symbol, size: float) -> int: pass @abstractmethod def place_market_selling_order(self, symbol: Symbol, size: float) -> int: pass @abstractmethod def get_order_status(self, order_id: int) -> List[OrderStatus]: pass @abstractmethod def get_assets(self) -> List[Asset]: pass
Traderクラスを改修します。改修のポイントは以下です。
とりあえずTraderクラスで資産の量を持つ必要はなくなったためTraderクラスから資産の量を取り除いています。しかし例えば取引のアルゴリズムが今の資産の量を加味する場合は度々Apiクラスを通じて今の資産の量を取得するのも時間がかかるのでTraderクラスが資産の量をもつこともありえます。その場合であっても本当の資産の量はSimulatorクラスや取引所にあるものであり、Traderクラスが持っているものはその写しということになります。
from enum import Enum, auto from api import Api, Symbol, OrderStatus from price_data_manager import PriceDataManager import datetime class TradingStatus(Enum): # 買い発注待ち STANDING_BY_FOR_BUYING = auto() # 買い注文約定待ち ORDERED_BUYING_ORDER = auto() # 売り発注待ち STANDING_BY_FOR_SELLING = auto() # 売り注文約定待ち ORDERED_SELLING_ORDER = auto() class SellingType(Enum): DEADLINE = auto() PROFIT_TAKING = auto() STOP_LOSS = auto() class Trader: SHORT_MA_PERIOD_SEC = 1 * 60 LONG_MA_PERIOD_SEC = 10 * 60 MAX_HOLDING_SEC = 5 * 60 * 60 SELLING_TH_PRICE_RATE = 1.01 STOP_LOSS_SELLING_TH_PRICE_RATE = 0.995 def __init__(self, api: Api) : self.state = TradingStatus.STANDING_BY_FOR_BUYING self.api = api self.order_id = None # 取引判断用データ ---- self.price_data_manager = PriceDataManager() # 前回の移動平均 self.last_long_ma_price = None self.last_short_ma_price = None self.bought_price = None self.bought_datetime = None def act(self): ticker = self.api.get_ticker(Symbol.BTC) self.price_data_manager.add_ticker(ticker) short_ma_price = self.price_data_manager.get_moving_average_ask_price(self.SHORT_MA_PERIOD_SEC) long_ma_price = self.price_data_manager.get_moving_average_ask_price(self.LONG_MA_PERIOD_SEC) if self.state == TradingStatus.STANDING_BY_FOR_BUYING: # ゴールデンクロスのチェック if (self.last_short_ma_price is not None and self.last_short_ma_price < self.last_long_ma_price and short_ma_price > long_ma_price): # 買い self.order_id = self.api.place_market_buying_order(Symbol.BTC, 1) self.state = TradingStatus.ORDERED_BUYING_ORDER self.bought_price = ticker.ask_price self.bought_datetime = ticker.timestamp print(f'{ticker.timestamp} 買い {ticker.ask_price}') elif self.state == TradingStatus.ORDERED_BUYING_ORDER: order_status_list = self.api.get_order_status(self.order_id) is_executed = False for status in order_status_list: if status == OrderStatus.EXECUTED: is_executed = True if is_executed: self.state = TradingStatus.STANDING_BY_FOR_SELLING elif self.state == TradingStatus.STANDING_BY_FOR_SELLING: # 時間経過、指定の割合上昇、指定の割合下降で売る selling_type = None if ticker.timestamp >= self.bought_datetime + datetime.timedelta(seconds=self.MAX_HOLDING_SEC): selling_type = SellingType.DEADLINE if ticker.bid_price >= self.bought_price * self.SELLING_TH_PRICE_RATE: selling_type = SellingType.PROFIT_TAKING if ticker.bid_price <= self.bought_price * self.STOP_LOSS_SELLING_TH_PRICE_RATE: selling_type = SellingType.STOP_LOSS if selling_type: # 売り self.order_id = self.api.place_market_selling_order(Symbol.BTC, 1) self.state = TradingStatus.ORDERED_SELLING_ORDER if selling_type == SellingType.DEADLINE: action = '売り(期限)' elif selling_type == SellingType.PROFIT_TAKING: action = '売り(利確)' elif selling_type == SellingType.STOP_LOSS: action = '売り(ロスカット)' print(f'{ticker.timestamp} {action} {ticker.bid_price}') elif self.state == TradingStatus.ORDERED_SELLING_ORDER: order_status_list = self.api.get_order_status(self.order_id) is_executed = False for status in order_status_list: if status == OrderStatus.EXECUTED: is_executed = True if is_executed: self.state = TradingStatus.STANDING_BY_FOR_BUYING self.last_short_ma_price = short_ma_price self.last_long_ma_price = long_ma_price
Apiクラスの追加、Traderクラスの改修にあわせてSimulatorクラスも改修します。改修のポイントは以下です。
前回はTraderクラス上で資産の量を更新するだけで済ませていた注文の処理をSimulatorクラスが仮想の取引所としての役割を担って行うようにします。
place_market_buying_order()メソッドとplace_market_selling_order()メソッドで、それぞれ買い注文と売り注文を受けます。注文を受けたらそれらの注文は即座に約定したこととし、資産の量を更新し、order_status_list_dict として持つ各注文の状態も全量約定とします。
get_order_status()メソッドで注文の状態を問われたらorder_status_list_dictから返します。
from api import Ticker, OrderStatus import pandas as pd from typing import List class Simulator: def __init__(self): self.cur_ticker = None self.yen_amount = 2000000 self.btc_amount = 0 self.next_order_id = 0 self.order_status_list_dict = {} from simulator_api import SimulatorApi self.simulator_api = SimulatorApi(self) def simulate(self): from trader import Trader self.trader = Trader(self.simulator_api) print(f'開始 資産: {self.yen_amount}') # CSVファイルを読み込む df = pd.read_csv('ticker_20200514.csv') df['timestamp'] = pd.to_datetime(df['timestamp']) # 一行ずつ処理 for _, row in df.iterrows(): self.cur_ticker = Ticker(row['ask'], row['bid'], row['high'], row['last'], row['low'], row['symbol'], row['timestamp'], row['volume']) self.trader.act() # 全行処理が終わったら結果を表示する print(f'終了 資産: {self.yen_amount + self.btc_amount * self.cur_ticker.bid_price}') def get_cur_ticker(self) -> Ticker: return self.cur_ticker def place_market_buying_order(self, symbol: str, size: float) -> int: # 簡易的に即約定とする order_id = self.next_order_id self.yen_amount -= self.cur_ticker.ask_price self.btc_amount += 1 self.order_status_list_dict[order_id] = [OrderStatus.EXECUTED] self.next_order_id += 1 return order_id def place_market_selling_order(self, symbol: str, size: float) -> int: # 簡易的に即約定とする order_id = self.next_order_id self.yen_amount += self.cur_ticker.bid_price self.btc_amount -= 1 self.order_status_list_dict[order_id] = [OrderStatus.EXECUTED] self.next_order_id += 1 return order_id def get_order_status(self, order_id: int) -> List[OrderStatus]: if order_id not in self.order_status_list_dict: return [] return self.order_status_list_dict[order_id]
Apiクラスの各メソッドでSimulatorクラスに対して取引を行うSimulatorApiクラスを作成します。
from api import Api, Ticker, OrderStatus, Symbol, Asset from simulator import Simulator from typing import List class SimulatorApi(Api): def __init__(self, simulator: Simulator) : self.simulator = simulator def get_available_amount(self) -> int: return self.simulator.yen_amount def get_ticker(self, symbol) -> Ticker: return self.simulator.get_cur_ticker() def place_market_buying_order(self, symbol: str, size: float) -> int: return self.simulator.place_market_buying_order(symbol, size) def place_market_selling_order(self, symbol: str, size: float) -> int: return self.simulator.place_market_selling_order(symbol, size) def get_order_status(self, order_id: int) -> List[OrderStatus]: return self.simulator.get_order_status(order_id) def get_assets(self) -> List[Asset]: assets = [] assets.append(self.simulator.yen_amount, Symbol.JPY) assets.append(self.simulator.btc_amount, Symbol.BTC) return assets
シミュレーションの実行を行う main.py は変更なしです。シミュレーションを実行してみましょう。
Traderクラスがトレードの状態として「買い注文約定待ち」「売り注文約定待ち」という状態を持つようになり、3秒間約定の確認に取られるため若干結果に違いがでましたが特に大きな違いはありません。
相変わらずの大敗です…。
from simulator import Simulator if __name__ == '__main__': simulator = Simulator() simulator.simulate()
次回はTraderクラスはそのままに、GMOコインのAPIを介して取引を行うGmoCoinApiクラスを作成して、いよいよ自動取引を実現させましょう。
前回までで、実際のビットコインの価格データを収集したり、その価格データをグラフ化したりしました。
今回はゴールデンクロスを使った簡単な取引のアルゴリズムをつくり取引をシミュレーションしてみます。
シミュレーションはシミュレーターという仮想の取引所の中である期間仮想のトレーダーに取引させて行うこととします。
そして開始時と終了時でトレーダーがもつ資産がどう変わったか等を見てアルゴリズムがうまく働いているかを確認します。
今回行うシミュレーションでは実装を簡単にするために実際の取引からいくらか単純化します。
シミュレーションの概要を以下に示します。
以下のようなクラス構成を考えました。
Simulatorクラス
Traderクラス
PriceDataManagerクラス
クラス間の所有関係を図にすると以下のようになります。
Simulatorクラスのインスタンスのsimulate()を実行するだけです。
simulate()がシミュレーションを行って結果を表示してくれるはずです。
from simulator import Simulator if __name__ == '__main__': simulator = Simulator() simulator.simulate()
Pythonでビットコインの価格を取得する(2) - てけとーぶろぐ。で作成した価格データのCSVファイルを読み込み、先頭のデータから1行よんではその価格をシミュレーション上の今の価格としてトレーダーを行動させます。
トレーダーは行動する中でSimulatorクラスのget_cur_ticker()メソッドを呼んで価格を取得するのでシミュレーション上の今の価格を取得することになります。
from api import Ticker import pandas as pd class Simulator: def __init__(self): self.cur_ticker = None def simulate(self): from trader import Trader self.trader = Trader(self) self.trader.yen_amount = 2000000 self.trader.btc_amount = 0 print(f'開始 資産: {self.trader.yen_amount}') # CSVファイルを読み込む df = pd.read_csv('ticker_20200514.csv') df['timestamp'] = pd.to_datetime(df['timestamp']) # 一行ずつ処理 for _, row in df.iterrows(): self.cur_ticker = Ticker(row['ask'], row['bid'], row['high'], row['last'], row['low'], row['symbol'], row['timestamp'], row['volume']) self.trader.act() # 全行処理が終わったら結果を表示する print(f'終了 資産: {self.trader.yen_amount + self.trader.btc_amount * self.cur_ticker.bid_price}') def get_cur_ticker(self) -> Ticker: return self.cur_ticker
act()メソッドがシミュレーター内で一定仮想時間ごとに実行されます。
現在の状態(ビットコインを持っているか否か)に応じて売買の判断をします。
具体的には、ビットコインを持っていないときは1分の移動平均線と10分の移動平均線のゴールデンクロスで買いの判断をします。ビットコインを持っているときは買いから5時間経過するか、買いから1%価格が上がるか、0.5%価格が下がるかしたら売りの判断をします。
from enum import Enum, auto from simulator import Simulator from price_data_manager import PriceDataManager import datetime class TradingStatus(Enum): # 買い発注待ち STANDING_BY_FOR_BUYING = auto() # 売り発注待ち STANDING_BY_FOR_SELLING = auto() class SellingType(Enum): DEADLINE = auto() PROFIT_TAKING = auto() STOP_LOSS = auto() class Trader: SHORT_MA_PERIOD_SEC = 1 * 60 LONG_MA_PERIOD_SEC = 10 * 60 MAX_HOLDING_SEC = 5 * 60 * 60 SELLING_TH_PRICE_RATE = 1.01 STOP_LOSS_SELLING_TH_PRICE_RATE = 0.995 def __init__(self, simulator: Simulator) : # TODO Simulator ではなくAPIを渡してAPIの通信先が # Simulator でも GMOコインのAPI でも同じように扱えるようにする self.state = TradingStatus.STANDING_BY_FOR_BUYING self.simulator = simulator # 資産 ---- self.yen_amount = 0 self.btc_amount = 0 # 取引判断用データ ---- self.price_data_manager = PriceDataManager() # 前回の移動平均 self.last_long_ma_price = None self.last_short_ma_price = None self.bought_price = None self.bought_datetime = None def act(self): ticker = self.simulator.get_cur_ticker() self.price_data_manager.add_ticker(ticker) short_ma_price = self.price_data_manager.get_moving_average_ask_price(self.SHORT_MA_PERIOD_SEC) long_ma_price = self.price_data_manager.get_moving_average_ask_price(self.LONG_MA_PERIOD_SEC) if self.state == TradingStatus.STANDING_BY_FOR_BUYING: # ゴールデンクロスのチェック if (self.last_short_ma_price is not None and self.last_short_ma_price < self.last_long_ma_price and short_ma_price > long_ma_price): # 買い # 簡易的に今の価格で買えたこととする self.yen_amount -= ticker.ask_price self.btc_amount += 1 self.state = TradingStatus.STANDING_BY_FOR_SELLING self.bought_price = ticker.ask_price self.bought_datetime = ticker.timestamp print(f'{ticker.timestamp} 買い {ticker.ask_price}') elif self.state == TradingStatus.STANDING_BY_FOR_SELLING: # 時間経過、指定の割合上昇、指定の割合下降で売る selling_type = None if ticker.timestamp >= self.bought_datetime + datetime.timedelta(seconds=self.MAX_HOLDING_SEC): selling_type = SellingType.DEADLINE if ticker.bid_price >= self.bought_price * self.SELLING_TH_PRICE_RATE: selling_type = SellingType.PROFIT_TAKING if ticker.bid_price <= self.bought_price * self.STOP_LOSS_SELLING_TH_PRICE_RATE: selling_type = SellingType.STOP_LOSS if selling_type: # 売り # 簡易的に今の価格で売れたこととする self.yen_amount += ticker.bid_price self.btc_amount -= 1 self.state = TradingStatus.STANDING_BY_FOR_BUYING if selling_type == SellingType.DEADLINE: action = '売り(期限)' elif selling_type == SellingType.PROFIT_TAKING: action = '売り(利確)' elif selling_type == SellingType.STOP_LOSS: action = '売り(ロスカット)' print(f'{ticker.timestamp} {action} {ticker.bid_price}') self.last_short_ma_price = short_ma_price self.last_long_ma_price = long_ma_price
add_ticker()メソッドで価格データを蓄積しget_moving_average_ask_price()メソッドやget_moving_average_bid_price()メソッドで蓄積した価格データを使ってaskやbidの価格の移動平均値を求めます。
from api import Ticker from collections import deque import datetime class PriceDataManager: MAX_MOVING_AVERAGE_PERIOD_SEC = 10 * 60 _PRICE_TYPE_ASK = 'ask' _PRICE_TYPE_BID = 'bid' def __init__(self) : self.ticker_queue = deque() def add_ticker(self, ticker: Ticker) : self.ticker_queue.append(ticker) # Tickerが増えすぎないように指定期間より過去のTickerは削除する th_datetime = ticker.timestamp - datetime.timedelta(seconds=self.MAX_MOVING_AVERAGE_PERIOD_SEC) while self.ticker_queue[0].timestamp < th_datetime: self.ticker_queue.popleft() def get_moving_average_ask_price(self, period_sec) -> float: return self._get_moving_average_price(period_sec, self._PRICE_TYPE_ASK) def get_moving_average_bid_price(self, period_sec) -> float: return self._get_moving_average_price(period_sec, self._PRICE_TYPE_BID) def _get_moving_average_price(self, period_sec, price_type) -> float: th_datetime = self.ticker_queue[-1].timestamp - datetime.timedelta(seconds=period_sec) num = 0 sum = 0.0 for ticker in reversed(self.ticker_queue): if th_datetime < ticker.timestamp: num += 1 if price_type == self._PRICE_TYPE_ASK: sum += ticker.ask_price elif price_type == self._PRICE_TYPE_BID: sum += ticker.bid_price return sum / num
プログラムを実行してみます。
> python .\main.py 開始 資産: 2000000 2020-05-06 10:44:33+00:00 買い 981000 2020-05-06 10:49:15+00:00 売り(ロスカット) 978350 2020-05-06 10:53:48+00:00 買い 979990 2020-05-06 13:23:30+00:00 売り(ロスカット) 975000 2020-05-06 13:31:15+00:00 買い 977130 2020-05-06 18:31:15+00:00 売り(期限) 978670 2020-05-06 18:31:33+00:00 買い 980520 2020-05-06 18:41:56+00:00 売り(ロスカット) 978020 2020-05-06 18:48:15+00:00 買い 979990 2020-05-06 23:48:18+00:00 売り(期限) 985000 2020-05-07 00:07:57+00:00 買い 971000 2020-05-07 02:39:57+00:00 売り(利確) 990969 … 2020-05-13 05:44:57+00:00 買い 955187 2020-05-13 05:53:02+00:00 売り(ロスカット) 950501 終了 資産: 1616646
買いと売りを繰り返して…
買いのタイミングは前回のチャートと見比べるとゴールデンクロスのタイミングになっていそうです。
うーん、なんかロスカットがおおいかな…。
結果…200万円が161万円に!
…改良が必要ですね。シミュレーションしてよかった!と前向きにとらえましょう…。
前回収集した価格データをグラフ化して、自動売買のアルゴリズムを考えてみましょう。
前回収集した価格データのCSVファイルが以下です。
こちらをPythonプログラムでグラフ化してみます。
グラフの描画にはnumpyパッケージ, matplotlibパッケージを使います。
またCSVファイルの読み込み等にpandasパッケージを使います。
それぞれをインストールします。
> pip install matplotlib > pip install pandas
matplotlibを使ったグラフの描画については以下のサイトを参考にしました。
全体のコードは以下になります。
import numpy as np import pandas as pd import matplotlib.pyplot as plt if __name__ == '__main__': df = pd.read_csv('ticker_20200514.csv') df['timestamp'] = pd.to_datetime(df['timestamp']) START_TIMESTAMP = '2020-05-06 10:00' END_TIMESTAMP = '2020-05-06 18:00' SAMPLING_PERIOD_SEC = 3 SHORT_MA_PERIOD_SEC = 1 * 60 LONG_MA_PERIOD_SEC = 10 * 60 # データの期間を絞る df = df[(df['timestamp'] >= START_TIMESTAMP) & (df['timestamp'] < END_TIMESTAMP)] # 移動平均の列を作成 df['askShortMa'] = df['ask'].rolling(SHORT_MA_PERIOD_SEC // SAMPLING_PERIOD_SEC, min_periods=1).mean() df['askLongMa'] = df['ask'].rolling(LONG_MA_PERIOD_SEC // SAMPLING_PERIOD_SEC, min_periods=1).mean() # グラフ描画 fig, ax = plt.subplots() ax.plot(df['timestamp'], df['ask'], label='ask') ax.plot(df['timestamp'], df['askShortMa'], label='ask ma 1 min') ax.plot(df['timestamp'], df['askLongMa'], label='ask ma 10 min') ax.legend(fontsize=16) ax.set_xlabel('timestamp', fontsize=16) ax.set_ylabel('¥', fontsize=16) ax.set_title('Ticker', fontsize=16) plt.show()
上記サイトから幾つか変更点があります。
まず全データを描画すると時間がかかるためデータを絞りました。
以下のコードで指定の期間分に絞ったDataFrameを作成しています。
# データの期間を絞る df = df[(df['timestamp'] >= START_TIMESTAMP) & (df['timestamp'] < END_TIMESTAMP)]
また移動平均の求め方について
上記サイトではnumpyのconvolve()を使っていますが
pandasでできる範囲はpandasでということで以下のサイトを参考にpandasのrolling()で移動平均を求めています。
以下のコードになります。
「min_periods=1」として先頭の方でデータが足りなくても1つ以上データがあればある分だけで平均値を求めるようにしています。
# 移動平均の列を作成 df['askShortMa'] = df['ask'].rolling(SHORT_MA_PERIOD_SEC // SAMPLING_PERIOD_SEC, min_periods=1).mean() df['askLongMa'] = df['ask'].rolling(LONG_MA_PERIOD_SEC // SAMPLING_PERIOD_SEC, min_periods=1).mean()
実行すると以下のグラフが描画されます。
グラフを眺めてみるとどうでしょうか?
移動平均の期間を変えたり、表示する期間を変えたりしてもいいかと思います。
1分の移動平均線が10分の移動平均線を下から突き抜けるとある程度価格が上昇しているように見えます。
いわゆるゴールデンクロスですね。
参考: 「ゴールデンクロス」「デッドクロス」とは? | お金のキャンパス
このゴールデンクロスを買いシグナルとする自動取引のアルゴリズムでいくことにしましょう。
分析と言うには程遠い感じですしゆくゆくは機械学習の学習結果を使った自動取引などにも挑戦したいとは思うのですが
まずは実現しやすいものでどの程度の成績がでるか試してみます。