FXの自動取引をするために調べたこと

FXが面白い。チャートを見ているだけであっという間に時間が過ぎていく。市場は土日を除く平日24時間開いており、大なり小なりレンジ（一定の変動幅の中で価格が上がったり、下がったりを何度も繰り返す状態）がある。本業もあり、１日中チャートに張り付いている訳にもいかないので、プログラムで売買すれば儲けられると考えた。

自動取引の障壁は、意外と高い

為替情報の取得

とりあえずデータを集めて分析するか、と考えていたがオープンな（無料）データが全くない。過去の資料で現在は取得できないものが多くあった。私の調べた範囲では、MetaTrader5 (MT5) を使うのが良いと思う。MT5は、FX取引ツールで、Pythonコードからの操作がが可能です。

証券会社はOANDA証券

MT5を扱える証券会社は限られていて日本国内では数社しかない。（2023/06現在）その中からOANDA証券を選ぶ。OANDAではデモ口座を作成してMT5のデモ体験が可能。当たり前だけど口座の開設には時間がかかる。1、２週間ぐらいだったはず。

MacだとMT5は動かない

MT5はWindows環境でないと動かない。ConohaのWindowsServerを契約してMT5を動かした。macOSからWineを使ってMT5を動かすこともできたが、Pythonから操作する際にライブラリが足りなくてエラーになってしまった。

Wineを使ってWindowsアプリケーション(.exe)をMac/Linuxで動かす | FRONTL1NE (フロントライン)

MT5取引プログラム

PythonからMT5を操作するプログラムコードを載せておきます。OANDAのデモ口座が30日間しかアクセスできないため、取引をするところまで行っていません。Pythonだとバックテストも自分で実装する必要がありますが、それもできません。

ログイン

MT5を起動、ログインします。LOGINパラメータは、ご自身の環境に合わせて変更する必要があります。

import MetaTrader5 as mt5

LOGIN_ID = 123456789
LOGIN_SERVER = "DemoAccount"
LOGIN_PASS = "DemoPassword"

def login():
    # MT5と接続
    if not mt5.initialize(login=LOGIN_ID, server=LOGIN_SERVER, password=LOGIN_PASS):
        print(f"initialize() failed, error code = {mt5.last_error()}")
        return False

    # 接続ができたらMT5のバージョンを表示する
    print(f"MetaTrader5 package version {mt5.__version__}")
    
    # 接続状態とパラメータをリクエストする
    print(f"terminal_info={mt5.terminal_info()}")
    
    return True

if login():
    print(f"Login OK")

データ取得

OANDA証券では、過去数年の１分足データは残っていない。１０分足とかなら取れたはず。取得したデータによってTIMEFRAMEを切り替える。
取得するデータのタイムゾーンはニューヨークが基準となっている。日本時間で取得したい場合は変換が必要。サマータイムの実装は適当なので注意。

from datetime import datetime, timezone
import pandas as pd
import pytz

# set time zone to UTC
utc = pytz.timezone("Etc/UTC")
tokyo = pytz.timezone("Asia/Tokyo")
est = pytz.timezone('US/Eastern') # ニューヨーク EST=東部標準時 

def convert_jst(time):
    # タイムゾーンの再設定のため、tzinfo属性を削除
    time = time.replace(tzinfo=None)
    return tokyo.localize(time)

def convert_est(time):
    # タイムゾーンの再設定のため、tzinfo属性を削除
    time = time.replace(tzinfo=None)
    return est.localize(time)

def convert_mt5_utc(time):
    convert_time = time
    # 夏時間と冬時間の切替りについて未検証、概算
    if convert_time.month > 3 and convert_time.month < 11:
        convert_time = convert_time - timedelta(hours=3)
    else:
        convert_time = convert_time - timedelta(hours=2)
    # タイムゾーンの再設定のため、tzinfo属性を削除
    convert_time = convert_time.replace(tzinfo=None)
    return utc.localize(convert_time)

def convert_utc_mt5(time):
    convert_time = time
    # 夏時間と冬時間の切替りについて未検証、概算
    if convert_time.month > 3 and convert_time.month < 11:
        convert_time = convert_time + timedelta(hours=3)
    else:
        convert_time = convert_time + timedelta(hours=2)
    # タイムゾーンの再設定のため、tzinfo属性を削除
    convert_time = convert_time.replace(tzinfo=None)
    return utc.localize(convert_time)

# ローカル時間帯オフセットの実装を避けるために「datetime」オブジェクトをUTC時間帯で作成する
dt_from = datetime.now(utc)
dt_from = convert_utc_mt5(dt_from)

symbol = "USDJPY"
tf = mt5.TIMEFRAME_M1 # M1, M2, M3, M4, M5, M6, M10, M12, M15, M20, M30, H1, H2, H3, H4, H6, H8, H12, D1, W1, MN
rates = mt5.copy_rates_from(symbol, tf, dt_from, 1000)

if rates is None:
    print(f"copy_rates_range() failed, error code = {mt5.last_error()}")
else:
    df = pd.DataFrame(rates)
    df['time'] = pd.to_datetime(df['time'], unit='s') # UNIX時間を変換
    df['utc_time'] = df['time'].apply(convert_mt5_utc)
    df['jst_time'] = df['utc_time'].apply(convert_jst)
    df['est_time'] = df['utc_time'].apply(convert_est)

df.tail()

簡単なチャート

折れ線グラフ表示です。

import matplotlib.pyplot as plt
from pandas.plotting import register_matplotlib_converters
register_matplotlib_converters()

# チャートにティックを表示する
plt.plot(df['time'], df['open'], label='open')
plt.plot(df['time'], df['close'], label='close')
 
# 凡例を表示する
plt.legend(loc='upper left')
 
# ヘッダを追加する
plt.title('time rates')
 
# チャートを表示する
plt.show()

ローソク足チャート

ローソク足のグラフ表示です。
mplfinanceをインストールする必要があります。

import mplfinance as mpf
import matplotlib.pyplot as plt

plot_df = df.set_index(['jst_time'])
plot_df = plot_df.head(100)
mpf.plot(plot_df,type='candle', figratio=(2,1), style="binance")

移動平均

加重移動平均と指数移動平均を計算します。

import numpy as np

def wma(value):
    # 加重移動平均
    weight = np.arange(len(value)) + 1
    wma = np.sum(weight * value) / weight.sum()
    
    return wma

def ema(values, period):
    # 指数移動平均
    ema = np.zeros(len(values))
    ema[:] = np.nan # NaN で一旦初期化
    ema[period-1] = values[:period].mean() # 最初だけ単純移動平均で算出
    
    for day in range(period, len(values)):
        ema[day] = ema[day-1] + (values[day] - ema[day-1]) / (period + 1) * 2
    
    return ema

df["SMA3"]=df["close"].rolling(3).mean().round(3)
df["SMA5"]=df["close"].rolling(5).mean().round(3)
df["SMA7"]=df["close"].rolling(7).mean().round(3)
df["WMA3"]=df["close"].rolling(3).apply(wma, raw=True).round(3)
df["WMA5"]=df["close"].rolling(5).apply(wma, raw=True).round(3)
df["WMA7"]=df["close"].rolling(7).apply(wma, raw=True).round(3)
df["EMA3"] = df["close"].ewm(span=3, adjust=False).mean().round(3)
df["EMA5"] = df["close"].ewm(span=5, adjust=False).mean().round(3)
df["EMA7"] = df["close"].ewm(span=7, adjust=False).mean().round(3)
display(df.tail(5))

ジグザグ

チャートの上限、下限を計算します。

import numpy as np
import pandas as pd


PEAK, VALLEY = 1, -1

def _identify_initial_pivot(X, up_thresh, down_thresh):
    """Quickly identify the X[0] as a peak or valley."""
    x_0 = X[0]
    max_x = x_0
    max_t = 0
    min_x = x_0
    min_t = 0
    up_thresh += 1
    down_thresh += 1

    for t in range(1, len(X)):
        x_t = X[t]

        if x_t / min_x >= up_thresh:
            return VALLEY if min_t == 0 else PEAK

        if x_t / max_x <= down_thresh:
            return PEAK if max_t == 0 else VALLEY

        if x_t > max_x:
            max_x = x_t
            max_t = t

        if x_t < min_x:
            min_x = x_t
            min_t = t

    t_n = len(X)-1
    return VALLEY if x_0 < X[t_n] else PEAK

def peak_valley_pivots_candlestick(close, high, low, up_thresh, down_thresh):
    if down_thresh > 0:
        raise ValueError('The down_thresh must be negative.')

    initial_pivot = _identify_initial_pivot(close, up_thresh, down_thresh)

    t_n = len(close)
    pivots = np.zeros(t_n, dtype='i1')
    pivots[0] = initial_pivot

    # Adding one to the relative change thresholds saves operations. Instead
    # of computing relative change at each point as x_j / x_i - 1, it is
    # computed as x_j / x_1. Then, this value is compared to the threshold + 1.
    # This saves (t_n - 1) subtractions.
    up_thresh += 1
    down_thresh += 1

    trend = -initial_pivot
    last_pivot_t = 0
    last_pivot_x = close[0]
    for t in range(1, len(close)):

        if trend == -1:
            x = low[t]
            r = x / last_pivot_x
            if r >= up_thresh:
                pivots[last_pivot_t] = trend#
                trend = 1
                #last_pivot_x = x
                last_pivot_x = high[t]
                last_pivot_t = t
            elif x < last_pivot_x:
                last_pivot_x = x
                last_pivot_t = t
        else:
            x = high[t]
            r = x / last_pivot_x
            if r <= down_thresh:
                pivots[last_pivot_t] = trend
                trend = -1
                #last_pivot_x = x
                last_pivot_x = low[t]
                last_pivot_t = t
            elif x > last_pivot_x:
                last_pivot_x = x
                last_pivot_t = t


    if last_pivot_t == t_n-1:
        pivots[last_pivot_t] = trend
    elif pivots[t_n-1] == 0:
        pivots[t_n-1] = trend

    return pivots


pivots = peak_valley_pivots_candlestick(df.close, df.high, df.low , 0.0001, -0.0001)
df['Pivots'] = pivots
df['Pivot Price'] = np.nan  # This line clears old pivot prices
df.loc[df['Pivots'] == 1, 'Pivot Price'] = df.high
df.loc[df['Pivots'] == -1, 'Pivot Price'] = df.low

pivot_df = df[df['Pivots'] != 0]
display(pivot_df)

最後に

株や為替の自動取引はとても需要があると思うが、取引環境が整備されていない。今回は自動取引どころか為替データへアクセスするまでで終了。また気が向いたらチャレンジする。