Pythonで取引ロボットを作成する方法

この記事と動画では、Pythonで取引ロボットを作成するためのステップバイステップの手順を提供します。

注意

cTraderは、ネイティブのPythonサポートを備えた唯一の主要な取引プラットフォームであり、cTrader WindowsまたはMacで直接Pythonコードを記述してボット、テクニカルインジケーター、ツールを構築できます。アダプターや複雑な回避策は必要ありません。

取引ロボットを作成する

AlgoアプリのcBotsタブで、新規ボタンをクリックして新しい取引ロボットの作成を開始します。

WindowsMac

cBotに名前を付け、プログラミング言語としてPythonを選択します。

WindowsMac

注意

cTraderは、幅広い戦略と自動取引アクションをカバーするPythonのプリメイドアルゴリズムを提供しています。これらのアルゴリズムテンプレートには既に取引ロジックとカスタマイズ可能なパラメーターが含まれており、保存してビルドするだけで実行可能です。

作成方法としてテンプレートを使用を選択し、Grid Sampleを選択してから作成をクリックします。

WindowsMac

Grid cBotは、シンボルの複数の買いまたは売り注文を定期的な価格間隔またはステップで配置し、ポジションのグリッドを形成するグリッド取引戦略を実装します。

cBotのコードを検証するためにビルドしましょう。

WindowsMac

Ctrl+Bを押すか、ビルドをクリックします。

Cmd+Bを押すか、ビルドアイコンをクリックします。

インジケーターを追加する

取引ロボットの精度を向上させるためにインジケーターを組み込みましょう。

Relative Strength Index (RSI)は、市場の過買いまたは過売り状態を示す人気のあるインジケーターです。このインジケーターは、市場が一方向に伸びているときに新しいポジションを開くのを防ぐのに役立ちます。

RSIをロボットロジックに統合します。デフォルトの期間14でインジケーターを初期化し、過買いと過売りのレベルをそれぞれ70と30に設定します。シンボルの終値を使用してRSIを作成します。

self.rsi_period = getattr(api, "RsiPeriod", 14)
self.rsi_overbought = getattr(api, "RsiOverbought", 70)
self.rsi_oversold = getattr(api, "RsiOversold", 30)
self.rsi = api.Indicators.RelativeStrengthIndex(api.Bars.ClosePrices, self.rsi_period)

ポジションを開く前にRSI条件が取引を許可するかどうかを確認するロジックを追加します。フィルターが取引をブロックした場合、現在のRSI値を示すメッセージをログに記録します。

if len(self.get_grid_positions()) == 0:
    # self.open_position()
    if self.passes_rsi_filter():
        self.open_position()
    else:
        api.Print("RSI filter blocked initial entry (RSI={:.2f})".format(self.get_rsi_value()))

価格が十分に動き、RSI値が中立範囲内にある場合にのみグリッドポジションが開かれるようにするコードを記述します。安定性のために最後に閉じたバーのRSI値を使用し、RSIフィルターによってブロックされたときにメッセージを出力します。

if distance >= api.StepPips:
    # self.open_position()
    if self.passes_rsi_filter():
        self.open_position()
    else:
        api.Print("RSI filter: no new entry (RSI={:.2f}, range {}-{})"
                    .format(self.get_rsi_value(), self.rsi_oversold, self.rsi_overbought))

def get_rsi_value(self):
    # Use the last CLOSED bar for stability
    return self.rsi.Result.Last(1)

def passes_rsi_filter(self):
    r = self.get_rsi_value()
    return (r > self.rsi_oversold) and (r < self.rsi_overbought)

現在の市場価格とポジションのエントリー価格の間のpips単位の距離の計算ロジックを調整します。

def get_distance_in_pips(self, position):
    # return (position.EntryPrice - api.Symbol.Ask if position.TradeType == TradeType.Buy else api.Symbol.Bid - position.EntryPrice) / api.Symbol.PipSize
    if position.TradeType == TradeType.Buy:
        diff = position.EntryPrice - api.Symbol.Ask
    else:
        diff = api.Symbol.Bid - position.EntryPrice
    return diff / api.Symbol.PipSize

以下に、RSIインジケーターを統合した完全なPythonコードをコピーできます：

import clr

clr.AddReference("cAlgo.API")

# Import cAlgo API types
from cAlgo.API import *

# Import trading wrapper functions
from robot_wrapper import *

class SupercBot():
    def on_start(self):
        self.volumeInUnits = api.Symbol.QuantityToVolumeInUnits(api.VolumeInLots)
        self.enoughMoney = True

        self.rsi_period = getattr(api, "RsiPeriod", 14)
        self.rsi_overbought = getattr(api, "RsiOverbought", 70)
        self.rsi_oversold = getattr(api, "RsiOversold", 30)
        self.rsi = api.Indicators.RelativeStrengthIndex(api.Bars.ClosePrices, self.rsi_period)

        if len(self.get_grid_positions()) == 0:
            if self.passes_rsi_filter():
                self.open_position()
            else:
                api.Print("RSI filter blocked initial entry (RSI={:.2f})".format(self.get_rsi_value()))

    def on_tick(self):
        grid_positions = self.get_grid_positions()
        net_profit_sum = sum([p.NetProfit for p in grid_positions])
        if net_profit_sum >= api.TargetProfit:
            api.Print("Target profit is reached. Closing all grid positions")
            self.close_grid_positions()
            api.Print("All grid positions are closed. Stopping cBot")
            api.Stop()

        if len(grid_positions) > 0 and self.enoughMoney == True:
            position_with_highest_pips = sorted(grid_positions, key=lambda pos: pos.Pips, reverse=True)[0]
            distance = self.get_distance_in_pips(position_with_highest_pips)          
            if distance >= api.StepPips:
                if self.passes_rsi_filter():
                    self.open_position()
                else:
                    api.Print("RSI filter: no new entry (RSI={:.2f}, range {}-{})"
                              .format(self.get_rsi_value(), self.rsi_oversold, self.rsi_overbought))

    def get_rsi_value(self):
        # Use the last CLOSED bar for stability
        return self.rsi.Result.Last(1)

    def passes_rsi_filter(self):
        r = self.get_rsi_value()
        return (r > self.rsi_oversold) and (r < self.rsi_overbought)

    def get_grid_positions(self):
        return [pos for pos in api.Positions if pos.SymbolName == api.SymbolName and pos.TradeType == api.TradeType]

    def open_position(self):
        result = api.ExecuteMarketOrder(api.TradeType, api.SymbolName, self.volumeInUnits, "Grid")
        if result.Error == ErrorCode.NoMoney:
            self.enoughMoney = False
            api.Print("Not enough money to open additional positions")

    def close_grid_positions(self):
        for position in self.get_grid_positions():
            position.Close()

        if len(self.get_grid_positions()) > 0:
            self.close_grid_positions()

    def get_distance_in_pips(self, position):
        if position.TradeType == TradeType.Buy:
            diff = position.EntryPrice - api.Symbol.Ask
        else:
            diff = api.Symbol.Bid - position.EntryPrice
        return diff / api.Symbol.PipSize