算法交易员

交易策略构建器、回测工具和实时交易系统。

pip install algorithmic-trader

简介

algo-trader 是一个算法交易策略执行器和回测工具的实现。 它能够在本地进行策略回测，并通过您的经纪商 API 实现实时交易。

请注意，这目前仍处于 开发中，该交易系统缺少外部市场数据和交易提供商支持。 如果您希望使用该交易系统进行真实交易，您需要自行实现与经纪商 API 的对接。 尽管实时交易功能尚未完成，但回测功能已经完全可用，因此您可以先对已实现的策略进行回测，并在实时交易功能完善后将其应用于实际交易中。

algo-trader 使用 Python 编写，当前的技术栈包括：

1. MongoDB 作为回测策略的后端数据存储；
2. tulipy —— Tulip Indicators 的 Python 绑定库， 用于计算技术指标；
3. ib_client（可选）—— 用于与 Interactive Brokers 网关通信的 Python 库。仅当您计划通过 IB 进行实盘交易时才需要此库。

架构

流水线

Pipeline 是流式处理的基本协调者。它负责从 Source 读取数据， 并将其传递给流水线中的各个处理器。除了协调各组件外，它本身并不包含任何业务逻辑。 流水线及其所有子组件都可以被序列化为 JSON 格式，这样系统就可以在文件中定义、加载和保存整个流水线及其配置。 这一特性非常重要，因为它可以作为基于 UI/CLI 的运行程序的前端界面。 示例的可序列化（且可运行）流水线可以在 examples/pipeline-templates 目录中找到。 如何将这些流水线加载到 Pipeline 中，并使用 PipelineRunner 来运行它们的示例，可以在 main.py 中找到。

PipelineRunner

PipelineRunner 接受一个初始的 Pipeline 列表或单个流水线， 以及一个可选的起始 SharedContext。如果在构造时未提供 SharedContext，则会初始化一个新的上下文。 每个 Pipeline 都会按照其在列表中的顺序，使用前一个上下文调用 run() 方法。 上下文会在每个 Pipeline 中传递，从而允许在数据收集和输出之前进行一些操作， 如加载、缓存和验证。

源

Source 是蜡烛数据迭代器的实现。它是流水线的起点， 也是流入流水线并被处理的蜡烛数据的来源。

处理器

Processor 是流水线中的主要处理单元。处理器可以按照任意顺序构建， 而蜡烛数据则会从源开始，依次流经所有处理器。 每个处理器负责将其处理后的蜡烛数据发送到下一个处理器（除非有特殊原因不这样做）。

process() 函数在处理每根蜡烛时，还会接收一个名为 SharedContext 的对象。 该上下文是一个内存中的键值存储实例，用于在处理器之间共享上下文和信息。

另一种在处理器之间共享数据的方式是利用蜡烛本身的 attachments 字段。 该字段用于存储蜡烛的元数据，例如与该时间点相关的计算结果和状态。 蜡烛附件在整个流水线中都是持久化的。

重处理

重处理是一种功能，允许处理器将已经处理过的蜡烛重新发送给下一个处理器。 重处理对于那些在流水线主流程之外执行某些逻辑的处理器非常有用，例如在发生事件、外部触发、市场或交易提供商出现问题等情况时。 重处理的示例可以在 AssetCorrelationProcessor 中找到。

事件

Event 如其名称所示，定义了系统中发生的某个事件。 它遵循与蜡烛相同的流动路径，在处理器之间传递。每个处理器都有责任将事件传播到下一个处理器（如果需要的话）。

由于流水线是数据驱动的，而不是时间驱动的，因此事件可以用作检查点，以指示数据流中发生的事情。 例如，从数据库源和实时市场数据源运行同一套流水线，可能会因为处理器依赖于时间而产生不同的效果。

为了使回测具有参考价值，从数据库快速运行和实时运行必须保持一致的行为。

策略

策略由 StrategyProcessor 按照每根蜡烛逐个执行。 策略负责执行交易逻辑，并发出信号 (StrategySignal)。 当策略发出交易信号时，StrategyProcessor 会捕获该信号，并将其传递给 SignalsExecutor 进行执行。

终止器

Terminator 是流水线中的可选最后一环。它会在流水线的最后阶段执行， 即当 Source 迭代器中的数据被完全消费完毕时。 终止器对于单元测试、回测和环境清理非常有用。

本地运行

algo-trader 使用 MongoDB 作为数据存储。要在本地运行 MongoDB，请使用 docker-compose。

docker-compose -f docker-compose.yml up -d

命令行界面

命令行界面是一个简单的系统运行接口。它可以用来运行流水线、回测策略或实盘运行策略。 目前，命令行界面尚未完全实现，大部分功能仅限于描述和列出可用的处理器和策略。 如果不带任何参数运行命令行界面，将会显示帮助菜单。

通过 PIP 安装该软件包后，即可使用 algo-trader 命令。 如果您选择克隆此仓库，则应使用 python main.py 作为入口点。

algo-trader

列出所有可用的处理器：

algo-trader processor list

描述特定的处理器：

algo-trader processor describe <processor_name>

同样的模式也适用于策略和源。 要运行流水线，您需要创建一个流水线模板文件，并使用命令行界面来运行它。

algo-trader pipeline run <pipeline_template_file>

易于运行的示例：

• 来自 Yahoo Finance 的数据加载器可在 src/algotrader/examples/pipeline-templates/build_daily_yahoo_loader.py 中找到。
  运行此示例管道会将历史数据从 Yahoo Finance 加载到 MongoDB：

  algo-trader pipeline run examples/pipeline-templates/build_daily_yahoo_loader.json
  

• 来自 Binance 的数据加载器可在 src/algotrader/examples/pipeline-templates/build_daily_binance_loader.json 中找到。
  运行此示例管道会将历史数据从 Binance 加载到 MongoDB：

  algo-trader pipeline run examples/pipeline-templates/build_daily_binance_loader.json
  

• 来自 Binance 的实时加密货币管道可在 src/algotrader/examples/pipeline-templates/build_realtime_binance.json 中找到。
  运行此示例管道将处理来自 Binance 的实时秒级蜡烛数据：

  algo-trader pipeline run examples/pipeline-templates/build_realtime_binance.json
  

代码示例

带有注释的简短代码示例。以下是通常如何将 algo-trader 作为库来使用，以从 Yahoo Finance 将股票数据加载并计算技术指标，然后存储到 MongoDB 中的方法：

from datetime import datetime, timedelta

from algotrader.calc.calculations import TechnicalCalculation
from algotrader.entities.timespan import TimeSpan
from algotrader.pipeline.configs.indicator_config import IndicatorConfig
from algotrader.pipeline.configs.technical_processor_config import TechnicalsProcessorConfig
from algotrader.pipeline.pipeline import Pipeline
from algotrader.pipeline.processors.candle_cache import CandleCache
from algotrader.pipeline.processors.storage_provider_sink import StorageSinkProcessor
from algotrader.pipeline.processors.technicals import TechnicalsProcessor
from algotrader.pipeline.runner import PipelineRunner
from algotrader.pipeline.sources.yahoo_finance_history import YahooFinanceHistorySource
from algotrader.storage.mongodb_storage import MongoDBStorage


def main():
    # 创建一个 MongoDB 存储连接器
    mongodb_storage = MongoDBStorage()

    # 创建一个 Yahoo Finance 数据源连接器
    source = YahooFinanceHistorySource(
        ['AAPL', 'MSFT', 'GOOG'],  # 传入要获取的股票代码列表
        TimeSpan.Day,  # 选择蜡烛时间周期
        datetime.now() - timedelta(days=100),  # 开始获取数据的日期
        datetime.now()  # 结束获取数据的日期
    )

    # 创建一个 MongoDB 汇流处理器，它会使用存储连接器将所有处理后的蜡烛数据保存到 MongoDB。
    sink = StorageSinkProcessor(mongodb_storage)

    # 缓存处理器充当内存中的缓存，使各个处理器之间可以共享数据。
    cache_processor = CandleCache(sink)
    # 创建一个技术指标处理器，该处理器会为每根蜡烛添加指标数据。
    # 然后我们将蜡烛传递给缓存处理器，以便稍后可以使用这些数据，并在需要时与其他处理器共享。
    technicals_config = TechnicalsProcessorConfig([
        IndicatorConfig('sma5', TechnicalCalculation.SMA, [5]),
        IndicatorConfig('sma20', TechnicalCalculation.SMA, [20]),
        IndicatorConfig('cci7', TechnicalCalculation.CCI, [7]),
        IndicatorConfig('cci14', TechnicalCalculation.CCI, [14]),
        IndicatorConfig('rsi7', TechnicalCalculation.CCI, [7]),
        IndicatorConfig('rsi14', TechnicalCalculation.CCI, [14]),
        IndicatorConfig('stddev5', TechnicalCalculation.STDDEV, [5]),
    ])

    processor = TechnicalsProcessor(technicals_config, cache_processor)

    # 构建管道对象。该对象可以序列化为 JSON，保存并在需要时重新加载。
    pipeline = Pipeline(source, processor)

    # 将管道对象提供给执行器，由执行器来执行它。
    # 执行器可以依次执行多个管道。
    # 这使得构建一个管道流程成为可能，其中每个管道都依赖于另一个管道。
    PipelineRunner(pipeline).run()

在本地使用此仓库

虚拟环境

最好使用虚拟环境来运行 algo-trader。

python3 -m venv run
source run/bin/activate
pip3 install -r requirements.txt

运行测试

• 单元测试：./scripts/test-unit.sh
• 集成测试（需要运行 IB Gateway）：./scripts/test-integration.sh
• 全部测试：./scripts/test-all.sh

贡献

欢迎并非常需要大家的贡献。 请参阅贡献指南。

algo-trader 快速上手指南

algo-trader 是一个用 Python 编写的算法交易策略构建器、回测引擎和实时交易执行器。它支持本地策略回测，并可通过 Broker API 进行实时交易（需自行实现接口）。

环境准备

系统要求

• 操作系统: Linux, macOS 或 Windows (推荐 Linux/macOS)
• Python: Python 3.8+
• 数据库: MongoDB (用于存储回测数据和 K 线)
• 容器工具: Docker & Docker Compose (用于快速启动 MongoDB)

前置依赖

• 技术指标库: 项目依赖 tulipy (Tulip Indicators 的 Python 绑定)，安装时会自动处理。
• 可选依赖: 若需连接 Interactive Brokers 进行实盘交易，需配置 ib_client 并运行 IB Gateway。

安装步骤

1. 启动 MongoDB 数据库

algo-trader 使用 MongoDB 作为后端数据存储。推荐使用 Docker Compose 快速启动：

docker-compose -f docker-compose.yml up -d

2. 安装 Python 包

你可以选择通过 pip 直接安装，或克隆源码开发。

方式 A：通过 pip 安装（推荐）

pip install algorithmic-trader

注：国内用户若下载缓慢，可尝试指定清华源：

pip install algorithmic-trader -i https://pypi.tuna.tsinghua.edu.cn/simple

方式 B：源码安装（开发模式）

git clone <repo_url>
cd algo-trader
python3 -m venv run
source run/bin/activate  # Windows 用户使用: run\Scripts\activate
pip3 install -r requirements.txt

基本使用

1. 命令行工具 (CLI)

安装完成后，algo-trader 命令即可用（源码模式下使用 python main.py）。

查看帮助菜单：

algo-trader

查看可用的处理器列表：

algo-trader processor list

查看特定处理器详情：

algo-trader processor describe <processor_name>

2. 运行示例管道 (Pipeline)

algo-trader 的核心是“管道”概念。你可以直接运行内置的示例模板来加载历史数据或进行实时数据处理。

示例 A：从 Yahoo Finance 加载历史数据到 MongoDB

algo-trader pipeline run examples/pipeline-templates/build_daily_yahoo_loader.json

示例 B：从 Binance 加载历史数据到 MongoDB

algo-trader pipeline run examples/pipeline-templates/build_daily_binance_loader.json

示例 C：运行 Binance 实时加密货币数据管道

algo-trader pipeline run examples/pipeline-templates/build_realtime_binance.json

3. 代码集成示例

你也可以将 algo-trader 作为库在 Python 脚本中使用，以下示例展示了如何从 Yahoo Finance 获取数据、计算技术指标并存入 MongoDB：

from datetime import datetime, timedelta

from algotrader.calc.calculations import TechnicalCalculation
from algotrader.entities.timespan import TimeSpan
from algotrader.pipeline.configs.indicator_config import IndicatorConfig
from algotrader.pipeline.configs.technical_processor_config import TechnicalsProcessorConfig
from algotrader.pipeline.pipeline import Pipeline
from algotrader.pipeline.processors.candle_cache import CandleCache
from algotrader.pipeline.processors.storage_provider_sink import StorageSinkProcessor
from algotrader.pipeline.processors.technicals import TechnicalsProcessor
from algotrader.pipeline.runner import PipelineRunner
from algotrader.pipeline.sources.yahoo_finance_history import YahooFinanceHistorySource
from algotrader.storage.mongodb_storage import MongoDBStorage


def main():
    # 创建 MongoDB 存储连接器
    mongodb_storage = MongoDBStorage()

    # 创建 Yahoo Finance 数据源
    source = YahooFinanceHistorySource(
        ['AAPL', 'MSFT', 'GOOG'],  # 股票代码列表
        TimeSpan.Day,              # K 线周期
        datetime.now() - timedelta(days=100),  # 开始日期
        datetime.now()             # 结束日期
    )

    # 创建存储处理器，将处理后的 K 线存入 Mongo
    sink = StorageSinkProcessor(mongodb_storage)

    # 缓存处理器：用于在处理器间共享内存数据
    cache_processor = CandleCache(sink)
    
    # 配置技术指标处理器
    technicals_config = TechnicalsProcessorConfig([
        IndicatorConfig('sma5', TechnicalCalculation.SMA, [5]),
        IndicatorConfig('sma20', TechnicalCalculation.SMA, [20]),
        IndicatorConfig('cci7', TechnicalCalculation.CCI, [7]),
        IndicatorConfig('cci14', TechnicalCalculation.CCI, [14]),
        IndicatorConfig('rsi7', TechnicalCalculation.RSI, [7]), # 原文此处为 CCI，逻辑上 RSI 应调用 RSI 计算
        IndicatorConfig('rsi14', TechnicalCalculation.RSI, [14]),
        IndicatorConfig('stddev5', TechnicalCalculation.STDDEV, [5]),
    ])

    # 初始化技术指标处理器
    processor = TechnicalsProcessor(technicals_config, cache_processor)

    # 构建管道对象 (可序列化为 JSON)
    pipeline = Pipeline(source, processor)

    # 运行管道
    PipelineRunner(pipeline).run()

if __name__ == "__main__":
    main()

注意：目前该工具仍处于开发阶段（Work in Progress）。回测功能已完全可用，但实时交易功能需要用户自行实现 Broker API 接口方可用于实盘。