算法交易的 Python 基础知识

交易者使用 Python 构建交易机器人、技术指标、插件和其他工具来辅助算法交易或手动操作，本文解释了 Python 编码基础知识。

Python 是一种流行的高级通用编程语言，用于网络开发、数据科学、自动化，最近还用于 cTrader 中的算法交易。 无论您是创建简单的基于信号的交易机器人还是复杂的多指标策略，Python 都能使这个过程更快、更高效、更易于管理。

变量和数据类型

在 Python 中，变量是动态类型的，这意味着您不必显式声明它们的数据类型。 解释器根据分配的值在运行时确定类型。

您只需为变量分配一个值即可创建变量：

1
2
3
4

price = 105.50         # float
enabled = True         # boolean
symbol = "EURUSD"      # string
orders = []            # empty list

常见数据类型

在 cTrader 算法中，变量存储与市场信息、交易决策和 cBot 配置相关的重要数据。 这些变量通常代表计算的交易量、指标值、布尔标志、交易品种名称、标签等。

以下是一个示例：

1
2
3
4

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

在这种情况下，self.volumeInUnits 是用于下单的 float，而 self.enoughMoney 是控制是否继续在网格策略中开仓的 bool。

类和对象

Python 是一种面向对象编程（OOP）语言，提供了通过类和对象来组织代码的工具，这些类和对象可以模拟现实世界的实体和行为。 类定义了创建对象的蓝图，对象是该类的实例，具有自己独特的数据和行为。 关键概念如下：

• 使用 class 关键字定义类。
• 通常使用 __init__ 方法（构造函数）来初始化对象状态。
• 实例方法定义行为。
• 属性保存特定于每个对象的数据。

例如，您可以创建一个 Car 类：

1
2
3
4
5
6
7

class Car:
    def __init__(self, color, speed):
        self.color = color
        self.speed = speed

    def accelerate(self):
        self.speed += 10

然后您可以创建 Car 类的对象并与之交互：

1
2
3

my_car = Car("red", 50)
my_car.accelerate()
print(my_car.speed)  # This returns 60

在 cTrader 环境中

在 cTrader 中，cBot、指标和插件都是作为封装其整个逻辑的类来实现的。 算法通过特殊的 api 对象与 cTrader 平台交互，该对象提供对交易品种、图表、订单、指标等的访问。

您可以在 cTrader 中这样创建一个 SimplecBot 类：

1
2
3

class SimplecBot():
    def on_start(self):
        api.Print("cBot started")

SimplecBot 类包含一个单一方法（on_start(self)），该方法在交易机器人启动时执行一次。 api.Print 函数将消息写入日志，演示了对象方法如何封装交易行为。

类的每个实例实际上代表一个自包含的交易算法，使用 OOP 原则管理自己的生命周期和交易决策。 在标准的 cTrader cBot 中，一个类可能存储内部状态数据，如交易量、指标结果等，并使用几种方法来管理交易逻辑。

考虑这个 MACD 交叉示例，它在启动时初始化 MACD 指标，并在发生 MACD 交叉信号时下买单：

1
2
3
4
5
6
7

class MACDCrossover():
    def on_start(self):
        self.macd = api.Indicators.MacdCrossOver(api.Source, api.LongCycle, api.ShortCycle, api.SignalPeriods)

    def on_bar_closed(self):
        if self.macd.MACD.Last(0) > self.macd.Signal.Last(0):
            api.ExecuteMarketOrder(TradeType.Buy, api.SymbolName, volume)

self 关键字

在类内部，self 指的是类的实例，并提供对其变量和方法的访问。 self 用于：

• 存储数据，如指标结果、持仓量和配置值。
• 在不同方法调用（如 on_start()、on_tick() 或 on_bar_closed()）之间跟踪内部状态。

这个例子演示了如何使用 self：

1
2
3
4

class MACDCrossover():
    def on_start(self):
        self.volume = api.Symbol.QuantityToVolumeInUnits(api.VolumeInLots)
        self.macd = api.Indicators.MacdCrossOver(api.Source, api.LongCycle, api.ShortCycle, api.SignalPeriods)

在这种情况下，self.volume 存储用于交易的计算交易量，而 self.macd 存储 MACD 指标对象，允许在 cBot 的整个生命周期中访问其值。 如果没有 self，这些变量将仅限于 on_start() 内部，而在可能做出交易决策的 on_bar_closed() 中将无法访问。

方法和函数

函数是用于执行特定任务的可重用代码块。 当函数定义在类内部时，它们被称为方法，并通过 self 参数操作类的内部状态。 在 cTrader 算法交易的背景下，方法定义了您的交易算法在不同条件下的行为，如市场报价、K线收盘或初始化。

平台会在适当的时候自动调用特殊方法(称为事件处理程序)，例如:

on_start(self) – 机器人初始化时调用一次。

1
2

def on_start(self):
    api.Print("cBot started")

on_tick(self) – 每次收到市场报价时调用。

1
2

def on_tick(self):
    api.Print("Tick received")

on_bar_closed(self) – 新蜡烛图/K线收盘时调用(适用于基于时间的策略)。

1
2
3
4

def on_bar_closed(self):
    if Functions.HasCrossedAbove(...):
        self.close_positions(TradeType.Sell)
        api.ExecuteMarketOrder(...)

自定义或辅助方法

为避免重写和重复代码，您可以定义自己的方法。 这些自定义方法应封装交易机器人可能重复使用的常见逻辑。

下面的方法返回与当前 cBot 标签相关的所有开放头寸，便于一致地跟踪或关闭头寸。

1
2

def get_bot_positions(self):
    return api.Positions.FindAll(api.Label)

考虑另一个计算当前价格与头寸入场价之间点数距离的方法，这对网格策略的决策至关重要。

1
2

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

定义方法后，您可以使用 self 关键字在同一类中调用它。 这种方法让您可以在机器人的任何地方重用逻辑而无需重复代码。

例如，定义 get_bot_positions() 后，您可能在另一个方法中调用它来关闭所有头寸:

1
2
3
4

def close_positions(self, tradeType):
    for position in self.get_bot_positions():   # Helper method call
        if position.TradeType == tradeType:
            api.ClosePosition(position)

这里，self.get_bot_positions() 调用辅助方法，只返回与当前机器人标签相关的头寸。

同样，get_distance_in_pips() 方法可以重复用于交易决策。 例如，在网格策略中，您可能只想在市场从上一个头寸移动足够远时才开新订单:

1
2
3
4
5
6
7

def on_tick(self):
    grid_positions = self.get_bot_positions()
    if len(grid_positions) > 0:
        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)  # Helper method call
        if distance >= api.StepPips:
            self.open_position()

其他概念

命名空间和导入

Python 中的命名空间通过将相关标识符(变量、函数、类)分组来帮助组织代码并避免名称冲突。 导入允许您引入外部模块、库或包以扩展 Python 标准功能之外的功能。

在 Python 中，您可以导入标准模块或自定义包:

1
2

import math
print(math.sqrt(16))  # 4.0

您还可以导入特定的函数或类:

1
2

from math import sqrt
print(sqrt(25))  # 5.0

在 Python 中为 cTrader 编写算法时，导入的工作方式略有不同，因为 cTrader 与基于 .NET 的 cTrader Algo API 集成。 这种设置涉及一个允许 Python 代码与 .NET 程序集交互的桥梁，这解释了为什么典型的 cBot 以以下内容开始:

1
2
3
4
5
6
7
8
9

import clr

clr.AddReference("cAlgo.API")

# Import cAlgo API types
from cAlgo.API import *

# Import trading wrapper functions
from robot_wrapper import *

其中:

• import clr 加载公共语言运行时(CLR)，使 Python 能够与 .NET 库互操作。
• clr.AddReference("cAlgo.API") 引用 cAlgo API 程序集，其中包含所有与交易相关的类、方法和对象。
• from cAlgo.API import * 引入 API 中的所有类型(例如 TradeType、Color、Position)。
• from robot_wrapper import * 加载简化常见交易操作的辅助函数和包装器。

命名空间系统确保:

• Python 的内置名称(如 print)避免与 cTrader 的名称(如 api.Print)冲突。
• 所有特定于交易的对象(api.Symbol、api.Indicators、api.ExecuteMarketOrder)来自 cTrader Algo API 命名空间，使它们保持有组织和独特。
• 您可以通过导入自己的自定义包来扩展功能，与 cTrader 的内置类型一起工作。

循环

循环允许您在一系列值上重复操作或直到满足条件为止。 在算法交易中，循环通常用于处理头寸(例如，关闭、过滤或更新它们)、检查多个资产或指标的条件、实现需要重复下单的网格或马丁格尔策略等。

在 Python 中，迭代很简单，支持 for 循环(用于遍历集合)和 while 循环(用于重复直到条件改变)。

您可以编写一个循环来关闭头寸:

1
2
3
4

def close_positions(self, tradeType):
    for position in self.get_bot_positions():
        if position.TradeType == tradeType:
            api.ClosePosition(position)

在这种情况下:

• self.get_bot_positions() 返回属于当前 cBot 的所有头寸。
• for 循环遍历每个头寸。
• 匹配的头寸通过 api.ClosePosition 关闭。

考虑另一个循环，它遍历所有开放的网格头寸，逐个关闭它们，并递归检查是否还有剩余头寸(以防部分平仓)。

1
2
3
4
5
6

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()

条件语句

条件语句通过允许您的算法基于逻辑表达式做出决策来控制 Python 中的执行流程。 条件是策略逻辑的支柱，决定何时进入、退出或管理头寸。

Python 支持三种主要的条件结构:

• if – 当条件为真时执行代码。
• elif (else if) – 如果第一个条件为假，则检查另一个条件。
• else – 当没有条件满足时提供一个后备选项。

考虑在这段代码中表达的 MACD 交叉条件:

1
2
3
4

if self.macd.MACD.Last(0) > self.macd.Signal.Last(0):
    api.ExecuteMarketOrder(TradeType.Buy, api.SymbolName, self.volumeInUnits, api.Label, api.StopLossInPips, api.TakeProfitInPips)
elif self.macd.MACD.Last(0) < self.macd.Signal.Last(0):
    api.ExecuteMarketOrder(TradeType.Sell, api.SymbolName, self.volumeInUnits, api.Label, api.StopLossInPips, api.TakeProfitInPips)

这个结构:

• 当 MACD 线在信号线上方时下买单。
• 当 MACD 线在信号线下方时下卖单。
• 如果没有明确信号则跳过交易。

列表

Python 提供了强大而简洁的工具来处理数据集合，如交易、价格或指标。 两个最常见的是列表推导式和内置聚合函数，如 sum()。

例如，您可以这样编写代码来遍历 numbers 并只保留偶数:

1
2

numbers = [1, 2, 3, 4, 5]
squares = [n**2 for n in numbers if n % 2 == 0]

您可以使用列表推导式来只选择当前交易品种的头寸，生成一个在多交易品种环境中有用的过滤列表。

1

positions = [p for p in api.Positions if p.SymbolName == api.SymbolName]

使用 sum() 聚合利润或交易量是网格交易策略中确定总利润是否达到目标的标准方法。

1

net_profit_sum = sum([p.NetProfit for p in grid_positions])

您可以这样组合过滤和聚合:

1
2

buy_positions = [p for p in api.Positions if p.TradeType == TradeType.Buy]
avg_buy_price = sum([p.EntryPrice * p.VolumeInUnits for p in buy_positions]) / sum([p.VolumeInUnits for p in buy_positions])

同步和异步操作

Python 默认遵循同步执行模型。 您的算法的生命周期方法，如 on_start、on_tick 和 on_bar_closed，由平台的事件循环触发，一次一个。

以下是同步交易流程的基本表示:

graph TD
    A(遇到信号) ==> B(执行订单)
    B ==> C(头寸达到止盈/止损)
    C ==> D(关闭头寸)
    D ==> A

在同步流程中，任何长时间运行的操作都可能阻塞线程，可能延迟下一个事件。 然而，在现实世界的策略中，算法通常启动非阻塞操作，如提交订单或设置止损和止盈，然后等待未来事件(如新的报价或K线)来处理结果。 这种设置导致类似异步的编排，即使没有使用 Python 的 async/await 结构。

cTrader 通过序列化您的事件回调来管理并发。 每个方法都是独立调用的，从不并行执行，从而消除了多线程保护的需求。 这种设计使策略能够在逻辑上同时响应多个交易条件，而不会阻塞。

在下面的示例中，策略独立评估买入和卖出信号并做出相应响应，执行上下文没有任何重叠。

graph TD
    A([收到信号]) ==> B([开立买单]) & C([开立卖单用于对冲]) ==> D([订单触及<br>止盈或止损])
    D ==> E([平仓])
    E ==> A

尽管决策看似并行，但它们实际上是由 cTrader 的内部队列系统按顺序处理的。 平台架构旨在防止并发问题，同时保持交易机器人对市场变化的响应能力。