簡単にPythonプログラムでのトレード手法を試せるプログラムを用意しました。
サンプルとして、MACDのゴールデンクロスだけでトレードするプログラムで試します。
※2024/01.06 追記 この記事の仕組みで、勝てるかもしれないプログラムも公開しています。
この記事を読んで実行方法を理解してから見てください。
実行環境
必要なツール
- Python3.9 (3.7以上なら大抵動くと思います)
Pythonインストール - VSCode 1.85.1 (2023/12/24時点での最新)
VSCode
※拡張機能からPythonをインストールすること - MT5
FXトレードアプリケーション
MT５セットアップ - GIt
プログラムバージョン管理ツール。プログラムをGIt管理でGithubというサイトにアップしているので、Githubからのプログラム取得に利用します。
Gitのインストール
プログラム取得
VSCodeを起動しterminalを起動し
ターミナルでプログラムをダウンロードしたいフォルダに移動し、git cloneを実行
cd [ダウンロードしたいフォルダパス]
git clone https://github.com/sorapecopi/TradeSample.gitTradeSampleフォルダがダウンロードされるので、VSCodeのファイル->フォルダを開くから、TradeSampleフォルダを開いてください。
Python仮想環境の準備
ダウンロードしたプログラムの、.vscode/launch.jsonに仮想環境ベースでPythonライブラリパスを設定しています。この設定を利用してPythonを実行できるようにするため、Python仮想環境をセットアップします。
VSCodeのターミナルを起動し、ダウンロードしたTradeSampleフォルダ直下で以下を実行
python -m venv venvTradeSampleフォルダ直下にvenvフォルダができているのを確認してください。
この中に仮想環境のライブラリやアクティベートファイルが収まっています。
ターミナルから以下を実行しPython仮想環境をアクティブにします。
.\venv\Scripts\Activate.ps1(※ターミナルがコマンドプロンプトの場合は、.\venv\Scripts\Activate.batになります。)
ターミナルのコマンドラインのディレクトリ位置表示の先頭に(venv)がついていることを確認してください。
アクティベート前
アクティベート後
“(venv)”と表記されている状態が仮想環境で動作していることになり、この状態で以下を実行
pip install -r requirements.txtこれで必要なpythonライブラリがPython仮想環境にインストールされます。
試験プログラムの実行
MT5を起動してから
VScodeのエディタで”trading_test.py”ファイルを開いた状態にして、（↓このようにtrading_test.pyのプログラムが見れる状態）
左の虫のデバッグマークをクリック
←これ
次に[RUN AND DEBUG]の[Python:ファイルをデバッグ]の緑△をクリック
試験が始まりまり、ターミナルに実行状況が出現されます。
バックテストが完了すると最後に結果を表示します。
print("GET count:", len(df_result[df_result.result == 'GET']))
print("LOSS count:", len(df_result[df_result.result == 'LOSS']))ちなみに、2023/12/26にこのプログラムを動かすと11/27～12/26の試験で結果は少し負けです。
ですが、押し目つけてから入るようにしたり、上位足のトレンドを見たりするだけで、だいぶ勝率が上がると思います。
プログラムの説明
プログラム構成
trading_test.py トレード試験プログラムメイン
utils.py ユーティリティプログラム。夏時間、冬時間判定、ワンピップス計算等々便利機能
mt5/mt5_terminal.py MT5アクセス用プログラム
analyzer/indicator.py 移動平均、指数移動、RSI、MACDなどの計算処理
trading_test.py
import datetime
import pandas as pd
from mt5.mt5_terminal import Mt5Terminal
from analyzer.indicator import macd_dw, macd_up
from utils import get_one_pips
def main():
instrument = 'USDJPY'
onepips = get_one_pips(instrument)
mt5 = Mt5Terminal()
last_idx = None
mergin_count = 0
order = None
trade_history = []
for past_idx in range(28800, -1, -1):
if last_idx is None:
df_candle = mt5.get_candle_from(
instrument,
'M1',
count=2000,
past_idx=past_idx)
last_idx = df_candle.index[-1]
# print(df_candle.index[-1])
current_time = last_idx + datetime.timedelta(minutes=1)
else:
df_candle_tmp = mt5.get_candle_from(
instrument,
'M1',
count=1,
past_idx=past_idx+mergin_count)
# print(df_candle_tmp.index[-1], last_idx, past_idx+mergin_count)
if last_idx + datetime.timedelta(minutes=1) < df_candle_tmp.index[-1] and last_idx + datetime.timedelta(hours=1) > df_candle_tmp.index[-1]:
# データかけ
print(f"set pre data {last_idx} to {last_idx + datetime.timedelta(minutes=1)}")
mergin_count += 1
df_candle.loc[last_idx + datetime.timedelta(minutes=1)] = df_candle.loc[last_idx]
else:
df_candle = df_candle.combine_first(df_candle_tmp)
if len(df_candle) > 2000:
df_candle = df_candle.iloc[-2000:]
last_idx = df_candle.index[-1]
current_time = last_idx + datetime.timedelta(minutes=1)
if order is None:
if macd_dw(df_candle):
ticks = mt5.get_ticks_from(instrument, current_time, 10)
print(f"SELL:time:{current_time} price:{ticks.iloc[0].bid}")
order = {
'start_time': current_time,
'price': ticks.iloc[0].bid,
'action': 'SELL'
}
elif macd_up(df_candle):
ticks = mt5.get_ticks_from(instrument, current_time, 10)
print(f"BUY:time:{current_time} price:{ticks.iloc[0].ask}")
order = {
'start_time': current_time,
'price': ticks.iloc[0].ask,
'action': 'BUY'
}
else:
ticks = mt5.get_ticks_from_to(instrument, order['start_time'], current_time)
get_20p = False
loss_20p = False
if order['action'] == 'BUY':
if any(ticks.bid > order['price'] + onepips*20):
get_20p = True
if any(ticks.bid < order['price'] - onepips*20):
loss_20p = True
if get_20p and loss_20p:
if ticks[ticks.bid > order['price'] + onepips*20].index[0] < ticks[ticks.bid < order['price'] - onepips*20].index[0]:
loss_20p = False
else:
get_20p = False
if get_20p:
print(f"GET 20pips time:{current_time}")
order['result'] = 'GET'
trade_history.append(order)
order = None
elif loss_20p:
print(f"LOSS 20pips time:{current_time}")
order['result'] = 'LOSS'
trade_history.append(order)
order = None
elif order['action'] == 'SELL':
if any(ticks.ask < order['price'] - onepips*20):
get_20p = True
if any(ticks.ask > order['price'] + onepips*20):
loss_20p = True
if get_20p and loss_20p:
if ticks[ticks.ask < order['price'] - onepips*20].index[0] < ticks[ticks.ask > order['price'] + onepips*20].index[0]:
loss_20p = False
else:
get_20p = False
if get_20p:
print(f"GET 20pips time:{current_time}")
order['result'] = 'GET'
order['end_time']: current_time
trade_history.append(order)
order = None
elif loss_20p:
print(f"LOSS 20pips time:{current_time}")
order['result'] = 'LOSS'
order['end_time']: current_time
trade_history.append(order)
order = None
# print(trade_history)
df_result = pd.DataFrame(trade_history)
print("GET count:", len(df_result[df_result.result == 'GET']))
print("LOSS count:", len(df_result[df_result.result == 'LOSS']))”MT5のcandleデータをDataframeで取得する”で説明したMt5Terminalクラスを使って、１分のローソク足を取得し、MACDのゴールデンクロスを使用して、上昇か下降かを判定して、売り、買い設定をします。
買いの場合、20pips上昇すれば、勝ち(result:GET)。20pips下降すれば、負け(result:LOSS)と判定します。（売りはその逆）
試験範囲は、最新から28800個前のローソク足（60x24x20 大体20日前）から最新までです。
初めは、mt5.get_candle_from()で2000個ローソク足を取得し、変数名df_candleに保存します。その後は次のローソク足1本取得して、df_candleを更新します。
値動きがないときなどMT5からデータが取得できない時があるので、その場合は、前の時刻のデータで欠損データを埋めます。
if last_idx + datetime.timedelta(minutes=1) < df_candle_tmp.index[-1] and last_idx + datetime.timedelta(hours=1) > df_candle_tmp.index[-1]:
# データかけ
print(f"set pre data {last_idx} to {last_idx + datetime.timedelta(minutes=1)}")
mergin_count += 1
df_candle.loc[last_idx + datetime.timedelta(minutes=1)] = df_candle.loc[last_idx]ローソク足は例えば、1分足で12時13分のデータは12時13分0秒から12時13分59秒までの値動きを表したデータなので1分足が完了すると時刻は12時14分0秒ということになります。
ですので、現在時刻は1分足の時刻データの1分後にしています。
current_time = last_idx + datetime.timedelta(minutes=1)df_candleを用いて、MACDのゴールデンクロスを判定し、BUYかSELLを決定し、結果をorderに保存します。
orderは辞書型で、
start_time:取引開始時間
price:取引開始価格
action: BUY or SELL
を保存します。
if macd_dw(df_candle):
ticks = mt5.get_ticks_from(instrument, current_time, 10)
print(f"SELL:time:{current_time} price:{ticks.iloc[0].bid}")
order = {
'start_time': current_time,
'price': ticks.iloc[0].bid,
'action': 'SELL'
}
elif macd_up(df_candle):
ticks = mt5.get_ticks_from(instrument, current_time, 10)
print(f"BUY:time:{current_time} price:{ticks.iloc[0].ask}")
order = {
'start_time': current_time,
'price': ticks.iloc[0].ask,
'action': 'BUY'
}orderにデータがある場合は、上下に20pips移動したかどうかを判定し、BUYで20pips上昇すればresultをGET、20pips下降すればresultをLOSSとして、リスト型変数のtrade_historyに取引結果を保存します。
20pips上昇、下降の判定には、ティックデータをMT5から取得してその結果を使用します。
if any(ticks.ask < order['price'] - onepips*20):
get_20p = True
if any(ticks.ask > order['price'] + onepips*20):
loss_20p = True
if get_20p and loss_20p:
if ticks[ticks.ask < order['price'] - onepips*20].index[0] < ticks[ticks.ask > order['price'] + onepips*20].index[0]:
loss_20p = False
else:
get_20p = False
if get_20p:
print(f"GET 20pips time:{current_time}")
order['result'] = 'GET'
order['end_time']: current_time
trade_history.append(order)
order = None
elif loss_20p:
print(f"LOSS 20pips time:{current_time}")
order['result'] = 'LOSS'
order['end_time']: current_time
trade_history.append(order)
order = None最後にtrade_historyをDataFrame型に変換し、GETの数、LOSSの数をprint()し終了です。
df_result = pd.DataFrame(trade_history)
print("GET count:", len(df_result[df_result.result == 'GET']))
print("LOSS count:", len(df_result[df_result.result == 'LOSS']))以上。
簡単なプログラムで設定も簡単なのですぐに試せると思います。
実行結果をグラフで確認る方法はこちら
コメント