跳转至

如何在 Python 中创建交易机器人

本文和视频提供了在 Python 中创建交易机器人的分步说明。

注意

cTrader 是唯一一个具有原生 Python 支持 的主要交易平台，允许您在 cTrader Windows 或 Mac 中直接编写 Python 代码 并构建机器人、技术指标和工具。 无需适配器或复杂的变通方法。

创建交易机器人

在 Algo 应用程序的 cBots 选项卡中，点击 新建 按钮以开始创建新的交易机器人。

为 cBot 命名并选择 Python 作为编程语言。

注意

cTrader 提供了多种 预制算法，这些 Python 算法涵盖了广泛的策略和自动化交易操作。 这些算法模板已经包含了交易逻辑和可自定义的参数，保存并构建后即可运行。

选择 使用模板 作为创建方法，选择 Grid Sample，然后点击 创建。

Grid cBot 实施了一种网格交易策略，其中以固定的价格间隔或步长放置多个买入或卖出订单，从而形成一系列头寸。

让我们 构建 我们的 cBot 以验证其代码。

按下 Ctrl+B 或点击 构建。

按下 Cmd+B 或点击 构建 图标。

添加指标

让我们将指标集成到我们的交易机器人中以提高其准确性。

相对强弱指数 (RSI) 是一个流行的指标，用于指示市场中的超买或超卖情况。 该指标可以帮助我们的 cBot 在市场向一个方向过度延伸时避免开立新头寸。

将 RSI 集成到机器人逻辑中。 使用默认周期 14 初始化指标，并将超买和超卖水平分别设置为 70 和 30。 使用标的资产的收盘价来创建 RSI。

1
2
3
4
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 值的消息。

1
2
3
4
5
6
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 过滤器阻止时打印一条消息。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
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)

调整当前市场价格与头寸建仓价之间距离（以点为单位）的计算逻辑。

1
2
3
4
5
6
7
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 代码：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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

启动实例

让我们回到我们的 cBot 并构建它。 现在，点击任何实例的 启动 cBot 图标，交易机器人将开始运行。

或者，您可以点击 添加实例，指定您偏好的实例参数，添加新实例并启动它。

通过 添加 和 自定义 新实例，您可以在不同的标的资产、周期或参数上同时运行 cBot。

在其他地方运行机器人

启用 云同步 后，cBot 会自动出现在您的 cTID 处于活动状态的所有 cTrader 应用程序中。 我们可以在任何 cTrader 应用程序（包括 cTrader Mobile 和 Web）中在云端启动相同的交易机器人。

在 cTrader Algo 中启动 cBot。

本文展示了如何创建 Python 交易机器人并启动它，使您能够自动化您的策略并在任何设备上运行算法。

Image title