MT4ヒストリカルデータをpython上で扱えるようにしたりcsvに保存する

何をするためのスクリプト？

ヒストリカルデータをpythonを使用してpandas DataFrameとして読み出したり、csvやpickleに書き込みを行います。

インストール

github - u1and0/stockplotからcloneしてください。

git clone https://github.com/u1and0/stockplot.git

binディレクトリ下のread_hst.pyを使用してください。
その他のファイルは次のページで説明しています。

• pythonでローソク足(candle chart)の描画
• plotlyでキャンドルチャートプロット
• Plotlyでぐりぐり動かせる為替チャートを作る(1)
• Plotlyでぐりぐり動かせる為替チャートを作る(2)

なんかリポジトリの容量が重たいのでいつか整理します。(cloneすることはできる。たぶんjupyter notebookのファイルとかgifファイルが重い)
read_hst.pyだけ必要なので、最後に書いたコードをコピペしたほうが早いです。

データのダウンロード

FXDD Trading などからヒストリカルデータ(一分足のティックデータ)の圧縮ファイルをダウンロードしてください。

wget, aria2などのコマンドが使える環境にあれば

$ wget http://tools.fxdd.com/tools/M1Data/USDJPY.zip

などとしてヒストリカルデータの圧縮ファイルをダウンロードできます。 容量は50MB程度です。

使用方法

jupyter notebook や ipython上で使うとき

1. read_hstモジュールをインポートします。
2. read_hst()関数にダウンロードしたzipファイルのパス、または解凍したhstファイルのパスを入れます。
3. 結果はpandas DataFrameとして返されます。

import read_hst as h
df = h.read_hst('data/USDJPY.zip')  # zipファイルの相対/絶対パス
# hstファイル以外の拡張子が与えられると、展開したhstファイルは削除されます。

df = h.read_hst('data/USDJPY.hst')  # hstファイルの相対/絶対パス
# zipを解凍してhstファイルを引数に与えたらファイルを削除しません。

df.tail

                        open     high      low    close  volume
time
2017-11-17 08:32:00  112.573  112.584  112.573  112.581    50.0
2017-11-17 08:33:00  112.581  112.583  112.578  112.580    38.0
2017-11-17 08:34:00  112.580  112.583  112.578  112.580    51.0
2017-11-17 08:35:00  112.580  112.580  112.572  112.572    44.0
2017-11-17 08:36:00  112.572  112.574  112.572  112.572    24.0

bashなどのshell上で使うとき

以下のコマンドは~/Data/USDJPY.zipを~/Data/USDJPY.csvとして保存します。
-pとすればpickleファイル(拡張子はpkl)としても保存できます。

$ cd ~/python/stockplot
$ bin/read_hst.py -c ~/Data/USDJPY.zip  # Convert .hst to .csv
$ bin/read_hst.py -h

usage: bin/read_hst.py [-h] [-c] [-p] filenames [filenames ...]

Convering historical file (.hst) to csv or pickle file.

positional arguments:
  filenames

optional arguments:
  -h, --help    show this help message and exit
  -c, --csv     Convert to csv file
  -p, --pickle  Convert to pickle file

参考

• numpyを使用して高速にバイナリ→テキスト変換 >> (´・ω・｀；)ﾋｨｨｯ　すいません - pythonでMT4のヒストリファイルを読み込む
• 引数読み込み >> Converting MT4 binary history files: hst to csv using a python script

コード

read_hst.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import argparse
import zipfile
import pandas as pd
import os
import numpy as np


def zip2hst(fullpath):
    """Extract zip file.

    Usage:
        zip2hst('~/Data/USDJPY.zip')
        > ~/Data/USDJPY.hst
        zip2hst('USDJPY.zip')
        > USDJPY.hst

    args:
        fullpath: Zip filename or path
    return:
        Extract filename or path
    """
    if zipfile.is_zipfile(fullpath):
        with zipfile.ZipFile(fullpath, 'r') as zf:
            zf.extractall()  # zip展開
            ziplist = zf.namelist()
            if not len(ziplist) == 1:
                print('There are {} files in zipfile. Try again.'.format(len(ziplist)))
                raise IOError
        hstfile = ziplist[0]
        return hstfile  # フルパスかファイルネームだけを返す
    else:  # zipファイルでなければそのまま返す
        return fullpath


def tickdata(filepath):
    """binary to pandas DataFrame using numpy.

    参考: (´・ω・｀；)ﾋｨｨｯ　すいません - pythonでMT4のヒストリファイルを読み込む
    http://fatbald.seesaa.net/article/447016624.html
    """
    with open(filepath, 'rb') as f:
        ver = np.frombuffer(f.read(148)[:4], 'i4')
        if ver == 400:
            dtype = [('time', 'u4'), ('open', 'f8'), ('low', 'f8'),
                     ('high', 'f8'), ('close', 'f8'), ('volume', 'f8')]
            df = pd.DataFrame(np.frombuffer(f.read(), dtype=dtype))
            df = df['time open high low close volume'.split()]
        elif ver == 401:
            dtype = [('time', 'u8'), ('open', 'f8'), ('high', 'f8'), ('low', 'f8'),
                     ('close', 'f8'), ('volume', 'i8'), ('s', 'i4'), ('r', 'i8')]
            df = pd.DataFrame(np.frombuffer(f.read(), dtype=dtype).astype(dtype[:-2]))
        df = df.set_index(pd.to_datetime(df['time'], unit='s')).drop('time', axis=1)
        return df


def read_hst(fullpath):
    """Extracting hst file from zip file.

    Usage:
        import hst_extract as h
        df = h.read_hst('~/Data/USDJPY.zip')

    args:
        fullpath: zip / hst file path
    return:
        pandas DataFrame
    """
    hstfile = zip2hst(fullpath)  # Extract zip in current directory.
    print('Extracting {}...'.format(hstfile))
    df = tickdata(hstfile)  # Convert binary to pandas DataFrame.
    if not os.path.splitext(fullpath)[1] == '.hst':  # fullpathにhstファイル以外が与えられた場合、ファイルを消す
        os.remove(hstfile)
    return df


def main():
    """Arg parser

    usage: bin/read_hst.py [-h] [-c] [-p] filenames [filenames ...]

    Convering historical file (.hst) to csv or pickle file.

    positional arguments:
      filenames

    optional arguments:
      -h, --help    show this help message and exit
      -c, --csv     Convert to csv file
      -p, --pickle  Convert to pickle file


    `stockplot/bin/read_hst.py -cp ~/Data/USDJPY.zip ~/Data/EURUSD.zip`
    Extracting '~/Data/USDJPY.zip' and '~/Data/EURUSD.zip' then save to

    * '~/Data/USDJPY.csv' and '~/Data/EURUSD.csv' as csv file.
    * '~/Data/USDJPY.pkl' and '~/Data/EURUSD.pkl' as pickle file.
    """
    description = 'Convering historical file (.hst) to csv or pickle file.'
    parser = argparse.ArgumentParser(prog=__file__, description=description)
    parser.add_argument('filenames', nargs='+')  # 1個以上のファイルネーム
    parser.add_argument('-c', '--csv', action='store_true', help='Convert to csv file')
    parser.add_argument('-p', '--pickle', action='store_true', help='Convert to pickle file')

    args = parser.parse_args()
    filenames = args.filenames
    csv = args.csv
    pickle = args.pickle

    if not filenames:
        raise KeyError("Enter a valid filenames")
    elif not (csv or pickle):
        raise KeyError("Enter a valid output - filetype '-c'(--csv) or '-p'(--pickle).")
    else:
        for filename in filenames:
            df = read_hst(filename)  # convert historical to pandas Dataframe
            basename = os.path.splitext(filename)[0]
            if csv:
                outfile = basename + '.csv'
                df.to_csv(outfile)
                yield outfile
            if pickle:
                outfile = basename + '.pkl'
                df.to_pickle(outfile)
                yield outfile


if __name__ == '__main__':
    for convert_filename in main():
        print(convert_filename)