各通貨のレート情報を自動取得する

この投稿文は次の言語で読めます: 日本語

概要

MT4/MT5の中でバックテストをする際にはその機能の中でヒストリカルデータを管理すれば良いのですが、
MetaTraderから離れた世界で(たとえば機械学習等を活用して)分析したいと思った場合、
ヒストリカルデータはどのように準備すればよいのでしょうか?

と思い、色々手段を調べた結果です。

スポンサーリンク

要件

  • できるだけツールに依存しないデータ形式で取得したい(csvとか)
  • 自動化ができること。日々レートが更新されていくものを人間系でダウンロードしていられない
  • 信頼できるデータ提供元から取得できることが望ましい
スポンサーリンク

調査結果

自作スクリプト

やっぱり、というか。 自作が一番自由が効きます。

DucasCopy Bank というスイスの銀行が提供するレート情報を、コマンドラインでダウンロードする方法で実装しました。
しかもそれが既にPythonパッケージとしてGithub上に公開されていましたので、活用させて頂きました。

usage: duka [options]

positional arguments:
SYMBOLS symbol list using format EURUSD EURGBP

optional arguments:
-h show help message and exit
-v show program's version number and exit
-d DAY specific day format YYYY-MM-DD (default today)
-s STARTDATE start date format YYYY-MM-DD (default today)
-e ENDDATE end date format YYYY-MM-DD (default today)
-c CANDLE use candles instead of ticks. Accepted values M1 M2 M5 M10 M15 M30 H1 H4 D1
-f FOLDER the dowloaded data will be saved in FOLDER (default '.')
-t THREAD number of threads (default 10)
--header include CSV header (default false)

引数の指定により、

  • 通貨ペア指定
  • 日付範囲指定
  • 期間(M1 M2 M5 M10 M15 M30 H1 H4 D1 とか)指定

が可能で、完全に要求を満たしています。

 

Pythonでの開発の際には、JetBrains Suiteの PyCharm が利用でき、
IntelliJと基本設定や使用感も同じ感覚で開発ができるのでオススメです。
(IntelliJの設定をエクスポートしてPycharmにインポートできる)

以下はボツ案です。

MetaTrader

手動でヒストリカルデータをダウンロードする方法はありましたが、自動化が難しそうです。

ヘルプを見る限り、一部コマンドラインから実行できる操作はあるものの、ヒストリカルデータのダウンロードは見つからず。

コマンドラインからの実行

プラットフォームは事前に定義されたパラメータを使って手動で実行することができます。これは、コマンドラインでの異なるオプション、及び異なる設定ファイルの使用によってなされます。

プラットフォームは、コマンドラインからオプションで実行することができます。そこにプラットフォーム実行ファイルへのパス( file\terminal.exe へのパス)を指定し、スペースの後に以下のキーの1つ以上追加します。
•/login: ログイン番号 — プラットフォームをアカウントで実行します。例は terminal.exe / login:100000です。
•/config: 設定ファイルへのパス — プラットフォームを指定された設定ファイルを使って実行します。例は terminal.exe /config:c:\myconfiguration.iniです。デフォルトの設定ファイルは common.iniです。
•/profile: プロファイル名 — プラットフォームを特定のプロファイルで実行します。プロファイルはプラットフォーム/profiles/charts/に位置しなければなりません。例は terminal.exe /profile:Euroです。
•/portable — プラットフォームをポータブルモードで実行します。このモードはプラットフォームがすでにメインモードで起動された場合に必要かもしれません。プラットフォームをポータブルモードで実行するには、オペレーティングシステムユーザーには適切な権限が必要です。

キー割り当てが誤って設定された場合(無効なログイン、プロファイル名または構成ファイル)は、デフォルト値が使用されます。

カスタム設定ファイルを使用しての実行

プラットフォームは、パラメータのカスタムセットを使用して実行することができます。デフォルトのcommon.iniに基づいて独自の構成ファイルが作成されます。プラットフォームを特定の構成ファイルで実行するにはコマンドラインで以下のコマンドを使います。

path_to_platform\terminal exe /config:c:\myconfiguration.ini

ここで c:\myconfiguration.iniは構成ファイルへのパスです。

カスタム構成ファイルは、プラットフォームの作業中には「読み取り専用」モードで使用されます。プラットフォームインターフェイスから行われた設定の変更は、使用されているカスタム構成ファイルには書き込まれません。

構成ファイルパラメータは、いくつかのブロックに分割し、プラットフォーム構成ウィンドウタブの設定に対応しています。以下は構成ファイルの中で最も重要な設定です。

TickStory

MT4用ヒストリカルデータの取得用として、有償版を購入して利用していましたが、
コマンドラインベースでのデータダウンロードには対応していないとのこと。

手順

ファイル形式でダウンロードしたものを、Amazon S3 を経由し、Redshift へインポートする流れです。
Redshift 上にレート情報とテスト結果、実口座での売買実績等あらゆるデータを集め、統合的に分析できるようにしています。

 

準備

  1. Python がインストールされていない場合は、インストールしておきます。
  2.  duka をインストールしておきます。
    pip install duka

サンプル

以下、長々書いていますが、ポイントは

        p = subprocess.Popen(['duka', symbol, '-c', period, '-d', load_date, '-f', self.rate_temp_directory])
        p.wait()

だけ。

コマンドラインから上記を実行しても結果は取得できますので、あえてPythonでコード書くのがしんどい方でも大丈夫です。

 

"""
レート情報をダウンロードする
"""
# -*- coding: utf-8 -*-

import sys
import os
import os.path
from datetime import datetime, timedelta
import subprocess
import codecs
from configparser import ConfigParser
import gzip
import pandas

# グローバル変数
app_home = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
sys.path.append(os.path.join(app_home, 'lib'))

from s3 import S3
from redshift import Redshift


class FxRate:
    def __init__(self,  symbol, period):

        self.__symbol = ''
        self.__period = ''
        self.__date_from = None
        self.__date_to = None
        self.__date_from_str = ''
        self.__date_to_str = ''
        self.__mt4_config = None
        self.__rate_directory = None
        self.__rate_temp_directory = None

        self.symbol = symbol
        self.period = period

        if self.symbol == '' or self.period == '':
            print('symbol, period のいずれかが未入力です。処理を中断します。')
            sys.exit(-1)

        # pythonスクリプト用config
        self.config = ConfigParser()
        conf_path = os.path.join(app_home, 'conf', 'autotest.conf')
        self.config.read(conf_path)

        # MT4用configファイル格納ディレクトリ 無ければ作成
        if not os.path.isdir(self.config.get('autotest', 'rate_directory')):
            os.mkdir(self.config.get('autotest', 'rate_directory'))

        self.rate_directory = self.config.get('autotest', 'rate_directory')

        if not os.path.isdir(self.config.get('autotest', 'rate_temp_directory')):
            os.mkdir(self.config.get('autotest', 'rate_temp_directory'))

        self.rate_temp_directory = self.config.get('autotest', 'rate_temp_directory')

    @property
    def symbol(self):
        """通貨ペア名称"""
        return self.__symbol

    @symbol.setter
    def symbol(self, value):
        """通貨ペア名称"""
        self.__symbol = value

    @property
    def period(self):
        """時間軸名称"""
        return self.__period

    @period.setter
    def period(self, value):
        """時間軸名称"""
        self.__period = value

    @property
    def date_from(self):
        """日付(From)(yyyy-MM-dd)"""
        return self.__date_from

    @date_from.setter
    def date_from(self, value):
        """日付(From)(yyyy-MM-dd)"""
        self.__date_from_str = value.replace('/', '-').replace('.', '-')
        self.__date_from = datetime.strptime(self.__date_from_str, '%Y-%m-%d')

    @property
    def date_to(self):
        """日付(To)(yyyy-MM-dd)"""
        return self.__date_to

    @date_to.setter
    def date_to(self, value):
        """日付(To)(yyyy-MM-dd)"""
        self.__date_to_str = value.replace('/', '-').replace('.', '-')
        self.__date_to = datetime.strptime(self.__date_to_str, '%Y-%m-%d')

    @property
    def date_from_str(self):
        return self.__date_from_str

    @property
    def date_to_str(self):
        return self.__date_to_str

    @property
    def rate_directory(self):
        """レートファイル格納ディレクトリ"""
        return self.__rate_directory
    
    @rate_directory.setter
    def rate_directory(self, value):
        """レートファイル格納ディレクトリ"""
        self.__rate_directory = value
        
    @property
    def rate_temp_directory(self):
        """レートファイル一時格納ディレクトリ"""
        return self.__rate_temp_directory
    
    @rate_temp_directory.setter
    def rate_temp_directory(self, value):
        """レートファイル一時格納ディレクトリ"""
        self.__rate_temp_directory = value

    def get(self, date_from, date_to):
        """DucasCopy社から指定範囲のレートをダウンロードする"""

        self.date_from = date_from
        self.date_to = date_to

        for d in range((self.date_to - self.date_from).days):
            load_date = self.date_from + timedelta(d)

            file_name = '-'.join((self.symbol, self.period, load_date.strftime('%Y%m%d'))) + '.csv'
            gzip_file_name = '-'.join((self.symbol, self.period, load_date.strftime('%Y%m%d'))) + '.gz'
            self.download_rate(self.symbol, self.period, load_date.strftime('%Y-%m-%d'), file_name)

            if os.path.isfile(os.path.join(self.rate_directory, file_name)):
                # 休日日はファイルがダウンロードされない.
                # ダウンロードされたファイルが存在する場合のみ処理する

                # ファイル内容調整
                # pandas で読取可能な状態にするため、改行コードを置換する
                text_temp = open(os.path.join(self.rate_directory, file_name)).read()
                text_temp = text_temp.replace('\n\n', '\r\n')
                f = codecs.open(os.path.join(self.rate_directory, file_name), 'w', 'utf-8')
                f.write(text_temp)
                f.close()

                # レート桁数、期間
                df = pandas.read_csv(os.path.join(self.rate_directory, file_name),
                                     sep=',',
                                     lineterminator='\n',
                                     names=('rate_time', 'open', 'close', 'high', 'low'))

                df_symbol = pandas.DataFrame([[self.symbol]] * len(df))
                df_period = pandas.DataFrame([[self.period]] * len(df))
                # open, close, high, low を100倍する
                df['open'] *= 100
                df['close'] *= 100
                df['high'] *= 100
                df['low'] *= 100

                df_modify = pandas.concat([df_symbol, df_period,
                                           df['rate_time'], df['open'], df['close'], df['high'], df['low']], axis=1)

                text = ''
                for index, row in df_modify.iterrows():
                    text += row.values[0] + ',' + \
                            row.values[1] + ',' + \
                            row.values[2] + ',' + \
                            str(round(row.values[3], 3)) + ',' + \
                            str(round(row.values[4], 3)) + ',' + \
                            str(round(row.values[5], 3)) + ',' + \
                            str(round(row.values[6], 3)) + ',' + '\n'

                f = gzip.open(os.path.join(self.rate_directory, gzip_file_name), 'wt', 9, encoding='utf-8')
                f.write(text)
                f.close()

                # gzip生成完了後、csvファイル削除
                os.remove(os.path.join(self.rate_directory, file_name))

    def load_to_redshift(self):
        """Redshift取込"""

        # S3転送
        s3 = S3('mt4bucket')
        files = os.listdir(self.rate_directory)
        for file in files:
            f = os.path.join(self.rate_directory, file)

            if os.path.isfile(f):
                # ディレクトリは処理対象外
                s3.file_upload(f, 'rate/' + file)
                os.remove(f)

        # Redshift取込
        redshift = Redshift('mt4')
        redshift.import_rate(s3.bucket_name)

    def download_rate(self, symbol, period, load_date, file_name):
        """レートファイルダウンロード"""

        # 一時フォルダを空にする
        files = os.listdir(self.rate_temp_directory)
        for file in files:
            os.remove(os.path.join(self.rate_temp_directory, file))

        p = subprocess.Popen(['duka', symbol, '-c', period, '-d', load_date, '-f', self.rate_temp_directory])
        p.wait()

        files = os.listdir(self.rate_temp_directory)
        for file in files:
            if os.path.isfile(os.path.join(self.rate_directory, file_name)):
                os.remove(os.path.join(self.rate_directory, file_name))

            os.rename(os.path.join(self.rate_temp_directory, file), os.path.join(self.rate_directory, file_name))


def main(args):
    assert args
    args = sys.argv
    try:
        # 引数設定
        symbol = args[1]
        period = args[2]
        date_from = args[3]
        date_to = args[4]

        rate = FxRate(symbol, period)
        rate.get(date_from, date_to)
        rate.load_to_redshift()

    except IndexError:
        print('Error: Invalid Arguments. ')
        print('arg1 : symbol, arg2 : period, arg3: date_from, arg4:date_to')

    except Exception as e:
        print(e)
        sys.exit(-1)


if __name__ == '__main__':
    main(sys.argv[1:])
スポンサーリンク

Twitterでフォローしよう

おすすめの記事