回测

在量化交易策略的研发过程中，回测系统的性能与易用性，直接决定了策略验证的效率以及迭代开发的速度。为了更好地支持多资产、中低频及高频策略的开发需求，Swordfish 提供了回测引擎，涵盖框架设计、策略实现、回测执行以及结果分析等环节，面向具备 Python 编程能力及基础量化知识的开发者，帮助快速上手并高效开展策略开发与验证工作。

基本框架

Swordfish 回测框架支持多种资产类别，包括股票、期货、期权、债券及加密货币等。整体架构由以下几个核心模块组成：

模块	说明	举例
StrategyTemplate	策略模板基类，定义策略的回调函数，如 initialize、on_snapshot、on_bar 等	# 定义一个自定义交易策略类，继承自回测框架提供的 StrategyTemplateclass MyStrategy(backtest.StrategyTemplate): # 初始化方法，在回测开始时调用一次 def initialize(self, context): pass # 当前未定义初始化逻辑 # 每根K线（Bar）到来时被调用的方法 def on_bar(self, context, msg, indicator): pass # 当前未定义交易逻辑
AssetMixin	提供各类资产的专用交易接口，如 StockOrderMixin （股票）、 FuturesOrderMixin （期货）等	class MyStrategy(backtest.StrategyTemplate, backtest.StockOrderMixin): def initialize(self, context): pass def on_bar(self, context, msg, indicator): self.submit_stock_order(...) # 用于发出股票买卖订单（你需要传入具体参数）
Config	回测参数配置类，不同资产类型有独立配置类，如 StockConfig、OptionConfig	config = backtest.StockConfig() # 创建一个股票回测配置对象
Backtester	回测器类，加载策略配置与数据，执行回测过程	backtest.Backtester(MyStrategy, config) # 创建回测器

整个回测流程可概括为以下五个步骤：

1. 策略开发：基于 StrategyTemplate 实现具体策略逻辑；

2. 参数配置：设置资产类型、回测时间范围、初始资金等参数；

3. 回测器构建：创建 Backtester 对象；

4. 执行回测：导入数据并启动回测流程；

5. 结果分析：获取回测结果并进行评估与可视化。

这种模块化、解耦式的设计结构，显著提升了策略的可复用性与框架的可维护性，使得策略开发更加高效与灵活。

本节代码示例展示了如何使用 Swordfish 框架构建一个最小可运行的策略回测框架。

# 导入回测模块和工具库
import swordfish.plugins.backtest as backtest
import swordfish as sf
import swordfish.function as F
import pandas as pd
# 定义策略类，继承回测框架的策略模板
class MyBasicStrategy(backtest.StrategyTemplate):
   def initialize(self, context):
       """
       策略初始化函数，在回测开始前自动调用。
       context: 提供当前回测上下文信息，如时间、资金、品种等。
       """
       print("Initial Strategy Context:", context)
   def on_bar(self, context, msg, indicator):
       """
       每根K线（bar）数据到达时自动调用。此处编写交易逻辑。
       context: 当前上下文，如资金、持仓等信息。
       msg: 当前bar的行情数据（如时间、价格、成交量）。
       indicator: 指标数据，可用于辅助判断买卖。
       """
       pass  # 可添加买入/卖出逻辑，如判断均线金叉/死叉等
# 创建回测配置对象
config = backtest.StockConfig()
config.start_date = sf.scalar("2021.01.01", type="DATE")   # 回测起始日期
config.end_date = sf.scalar("2021.12.31", type="DATE")     # 回测结束日期
config.asset_type = backtest.AssetType.STOCK               # 回测标的类型为股票
config.cash = 100_000_000                                  # 初始资金：1亿元
config.commission = 0.00015                                # 手续费率：万分之1.5
config.data_type = backtest.MarketType.MINUTE              # 行情数据为分钟线
# 创建回测引擎，传入策略和配置
backtester = backtest.Backtester(MyBasicStrategy, config)
# 当执行此行时，策略会初始化，并打印如下信息（来自 initialize 函数）：
# Starting Strategy Context: tradeDate->2021.01.01
# tradeTime->1970.01.01T00:00:00.000
# barTime->1970.01.01T00:00:00.000
# engine->53135104
# 准备行情数据（此处为示例，需用户自行准备真实的分钟级行情数据）
stocks = pd.DataFrame({
   # 示例字段（需根据框架要求准备）：symbol、datetime、open、high、low、close、volume 等
   # 当前只模拟一条数据
   'symbol': ['000001.XSHE'],
   'tradeTime': [sf.scalar("2021.01.01T09:30:00", type="TIMESTAMP")],
   'open': [10.0],
   'high': [10.5],
   'low': [9.5],
   'close': [10.2],
   'volume': [1000],
   'upLimitPrice': [11.0],
   'downLimitPrice': [9.0],
   'prevClosePrice': [10.0]
})
# 将 pandas DataFrame 转为 Swordfish 表结构
stocks = F.table(stocks)
# 注入行情数据供回测器使用
backtester.append_data(stocks)
# 当执行此行时，策略会再次初始化，并打印相同的数据：
# Starting Strategy Context: tradeDate->2021.01.01
# tradeTime->1970.01.01T00:00:00.000
# barTime->1970.01.01T00:00:00.000
# engine->53135104

上述代码构成了一个基础的回测流程，主要包括以下步骤：

1. 定义策略类：继承 StrategyTemplate，并实现关键回调函数，如 initialize() 和 on_bar()；

2. 配置回测参数：设定资产类型、起止时间、初始资金、手续费率、数据频率等；

3. 构建回测器对象：使用策略类与配置实例创建 Backtester；

4. 准备行情数据：将行情数据转换为 swordfish.function.table 格式；

5. 注入数据并启动回测：回测器接收到数据后会自动按时间驱动策略逻辑（如调用 on_bar()）。

备注

initialize() 函数在回测器创建时自动调用，适合执行加载参数、设定状态、注册指标等准备工作。

策略编辑

Swordfish 回测引擎基于事件驱动机制设计，支持多种类型的事件回调函数，涵盖策略初始化、每日盘前/盘后处理、逐笔行情、快照行情、K线数据、委托与成交回报等环节。用户可以根据自身策略的粒度和需求，在相应事件函数中定义逻辑、计算指标或下单操作。

回测引擎支持的事件函数

事件函数	说明
def initialize(self, context)	策略初始化函数，仅触发一次。用于定义初始状态、加载数据或初始化指标等。
def before_trading(self, context)	每日盘前回调，适合执行日内准备操作，如订阅行情、清空变量等。
def on_tick(self, context, msg, indicator)	逐笔行情回调，处理每笔委托或成交。适用于高频策略。
def on_snapshot(self, context, msg, indicator)	快照行情回调，获取市场某一时刻的完整状态。
def on_bar(self, context, msg, indicator)	K 线（Bar）行情回调，适用于中低频策略，如分钟级或日级策略。
def on_transaction(self, context, msg, indicator)	成交明细回调，仅支持上交所债券品种。
def on_order(self, context, orders)	委托状态变更时触发的回调，用于追踪订单生命周期。
def on_trade(self, context, trades)	成交回报事件，发生实际成交时触发。
def after_trading(self, context)	每日盘后回调函数，可用于计算当日收益、统计持仓等。
def finalize(self, context)	回测结束前触发一次。可用于结果输出、资源清理等操作。
def on_timer(self, context, msg, indicator)	定时器事件，在指定时间或频率下触发。适合定期执行策略逻辑。

回调函数参数说明

• context：策略上下文对象，类型为字典。用于保存策略运行过程中的自定义变量。引擎还内置以下四个全局属性：

  • context.tradeTime：当前行情的最新时间戳；

  • context.tradeDate：当前交易日期；

  • context.BarTime：当前 Bar（ K 线）的时间戳（用于低频快 照合成）；

  • context.engine：回测引擎实例，提供策略运行期间的控制接口。

• msg：行情数据对象，类型为字典，结构依行情类型而异：

  • 高频行情（如 on_tick, on_snapshot）：每条数据为一个独立字典，可通过 msg.lastPrice 或 msg.price 获取价格；

  • K线行情（如 on_bar）：为嵌套字典结构，第一层为标的代码（如 msg['600000']），第二层为该标的的行情字段（如 msg['600000'].close）。

• indicator：策略中订阅的指标数据，其结构与对应的 msg 保持一致。常用于在 on_bar、on_tick 中结合技术指标进行判断。

• orders：订单信息，类型为字典，在 on_order 中传入，字段依资产类型而异。

• trades：成交信息，类型为字典，在 on_trade 中传入，代表实际成交订单。

备注

context 是一个贯穿整个策略的对象，可以看作一个共享的运行时字典（dict 类型），用于存储和传递策略状态、参数、缓存数据等。它允许在策略初始化时设置变量，并在后续函数中共享和修改这些变量，比如可以包含：

• 策略参数（如最大持仓数量、止盈止损线等）

• 状态标志（如是否已开仓、是否触发某个信号）

• 数据缓存（如历史价格、指标中间值）

• 自定义统计量（如累计收益、信号触发次数）

回测参数设定

在启动回测之前，首先需要构建并配置对应的回测参数对象。Swordfish 回测引擎为不同资产类别提供了专用的配置类，使得策略开发更加灵活并具备良好的扩展性，例如：

• StockConfig：股票回测配置

• FutureConfig：期货回测配置

• BacktestBasicConfig：通用参数配置（适用于多资产组合）

查看默认配置项

若用户不确定某个配置类中包含哪些参数项，或想查看其默认值，可以直接实例化配置类并打印：

import swordfish.plugins.backtest as backtest
# 创建一个默认的股票策略配置对象，用于设置回测参数
config = backtest.StockConfig()
# 打印当前配置对象的内容，查看默认参数或调试使用
print(config)

输出结果会展示所有支持的配置项及其默认值，例如：

{
  'add_time_column_in_indicator': False,
  # 是否在指标数据中添加时间列（通常用于调试或结果对齐）
  'benchmark': None,
  # 业绩基准（例如沪深300等），用于比较策略收益
  'callback_for_snapshot': 0,
  # 快照回调模式
  'cash': None,
  # 初始资金，单位为元，例如 100_000_000 表示 1 亿元
  'commission': None,
  # 交易手续费率，如 0.00015 表示千分之0.15（万分之1.5）
  'context': None,
  # 运行上下文（一般由框架自动填充，不需手动设置）
  'data_retention_window': None,
  # 数据保留窗口，用于指标回看等（例如保留最近30个Bar）
  'enable_indicator_optimize': False,
  # 是否启用指标计算优化（提升速度，但可能影响灵活性）
  'enable_subscription_to_tick_quotes': False,
  # 是否订阅逐笔成交/报价数据（用于高频或盘口策略）
  'frequency': 0,
  # 数据频率，例如分钟、日线等，对应于 MarketType 枚举
  'is_backtest_mode': True,
  # 是否处于回测模式（True 表示回测，False 表示实盘）
  'latency': None,
  # 延迟模拟（例如网络延迟、撮合延迟），一般用于仿真环境
  'matching_mode': None,
  # 撮合模式，例如价格优先/时间优先等，具体依框架定义
  'matching_ratio': 1.0,
  # 普通订单撮合比例（1.0 表示全量成交，<1 会部分成交）
  'msg_as_table': False,
  # 是否将行情消息结构化成表（Table），便于操作
  'orderbook_matching_ratio': 1.0,
  # 基于订单簿的撮合比例（高级场景）
  'output_order_info': False,
  # 是否输出每笔订单详细信息（用于调试）
  'output_queue_position': 0,
  # 是否输出订单在队列中的位置（高级订单流仿真）
  'prev_close_price': None,
  # 上一交易日收盘价（如需用作开盘参考）
  'set_last_day_position': None,
  # 是否自动带入前一日持仓（常用于策略连续性测试）
  'stock_dividend': None,
  # 股票分红配置（是否考虑分红送股）
  'tax': None,
  # 印花税（例如卖出时征收 0.001）
  'universe': None
  # 股票池设置，例如指定参与交易的股票列表
}

自定义配置参数

在获取到配置对象后，可根据实际需求进行修改，例如设置回测时间区间、资产类型、初始资金、手续费等：

# 导入 swordfish 主模块，包含 scalar 类型等辅助功能
import swordfish as sf
# 设置回测开始日期（需要用 sf.scalar 创建 DATE 类型对象）
config.start_date = sf.scalar("2022.04.11", type="DATE")
# 设置回测结束日期（与开始日期相同表示只回测一天）
config.end_date = sf.scalar("2022.04.11", type="DATE")
# 设置资产类型为股票（也可以是期权、ETF 等，依框架支持）
config.asset_type = backtest.AssetType.STOCK
# 设置行情数据类型为快照级（SNAPSHOT）
# SNAPSHOT 一般为逐时快照数据，适用于中高频策略
# 可选值可能还有：MarketType.MINUTE（分钟）、MarketType.DAILY（日线）等
config.data_type = backtest.MarketType.SNAPSHOT
# 设置初始资金为 1 亿元（单位：人民币元）
config.cash = 100_000_000
# 设置交易印花税（通常仅在卖出时收取）
config.tax = 0.001

数据注入

Swordfish 的回测引擎中使用 Swordfish 的数据表作为行情输入。用户可以通过 sf.table() 构造表结构或者将 pandas DataFrame 进行转为 Swordfish 表结构，并用 backtester.append_data() 传入数据。

msg_table = sf.table({
    'symbol': sf.vector(["000001.XSHE", "000001.XSHE"], type="STRING"),
    'timestamp': sf.vector(["2022.04.11 10:10:00.000", "2022.04.11 10:10:03.000"], type="TIMESTAMP"),
    'lastPrice': sf.vector([7.0, 7.5], type="DOUBLE"),
    'prevClosePrice': sf.vector([6.5, 7.0], type="DOUBLE"),
    'offerPrice': sf.array_vector([[7.1], [7.6]], type="DOUBLE")
})

或者，可以用 F.loadText 导入一个csv文件。

无论使用何种方式准备数据，最终都需通过 backtester.append_data() 方法注入：

backtester = backtest.Backtester(MyBasicStrategy, config)
backtester.append_data(msg_table)

不同的资产类型和行情频率对输入数据字段的要求不同。以下以股票快照（Snapshot）策略为例，说明必需字段及其含义：

字段	类型	备注
symbol	SYMBOL	股票代码
symbolSource	STRING	"XSHG"（上交所）或者"XSHE"（深交所）
timestamp	TIMESTAMP	时间戳
lastPrice	DOUBLE	最新成交价
upLimitPrice	DOUBLE	涨停价
downLimitPrice	DOUBLE	跌停价
totalBidQty	LONG	区间买量
totalOfferQty	LONG	区间卖量
bidPrice	DOUBLE[]	委买价格列表
bidQty	LONG[]	委买量列表
offerPrice	DOUBLE[]	委卖价格列表
offerQty	LONG[]	委卖量列表

其他频率及资产类型所需的行情字段请参考 各个资产的回测引擎教程文档Stock — Swordfish documentation 。

示例：一个简单的趋势策略

本节将通过 Swordfish 回测框架，构建并回测一个基于分钟级 K 线的简单趋势策略。该策略逻辑如下：

• 当某个标的的当前收盘价低于上一根 K 线收盘价时，尝试买入；

• 若当前已持有该标的，且当前价格高于上一根 K 线收盘价，则尝试卖出；

• 每次交易固定买入/卖出1000 股；

• 策略运行于分钟级别 K 线行情上。

引入模块

import swordfish.plugins.backtest as backtest
import swordfish as sf
import swordfish.function as F

编写策略

# 定义策略类，继承回测框架的策略模板（包含框架生命周期）和下单辅助功能
class MyStrategy(backtest.StrategyTemplate, backtest.StockOrderMixin):
    def initialize(self, context):
        """
        策略初始化函数，在回测开始前调用一次。
        用于定义和订阅所需的指标。
        """
        # 使用 swordfish 的元代码模块构建指标逻辑
        with sf.meta_code() as m:
            lastp = F.prev(m.col("close"))  # 计算前一根K线的收盘价（prev(close)）
        # 向框架订阅指标：类型为K线，指标名为 'lastp'
        self.subscribe_indicator(backtest.MarketDataType.KLINE, {
            'lastp': lastp
        })
    def on_bar(self, context, msg, indicator):
        """
        每根K线触发一次（例如每分钟/每天，取决于 data_type）。
        执行买入/卖出逻辑。
        """
        for istock in msg.keys():
            prevp = indicator[istock]["lastp"]  # 获取前一根K线的收盘价
            lastPrice = msg[istock]["close"]    # 当前K线的收盘价
            # 获取当前默认账户的多头持仓数量
            position = self.accounts[backtest.AccountType.DEFAULT].get_position(symbol=istock)["longPosition"]
            # === 买入逻辑 ===
            if position == 0:
                if lastPrice < prevp:  # 当前价格低于前一K线：判断为超跌买入信号
                    print(context["tradeTime"], "buy", istock, lastPrice)
                    # 发起买入订单
                    # 参数分别为：股票代码、时间、盘口级别、价格、数量、方向(1为买入)、标签
                    self.submit_stock_order(
                        istock, context["tradeTime"], 5, lastPrice, 1000, 1, label="buy"
                    )
            # === 卖出逻辑 ===
            else:
                if lastPrice > prevp:  # 当前价格高于前一K线：判断为反弹卖出信号
                    print(context["tradeTime"], "sell", istock, lastPrice)
                    # 发起卖出订单
                    # 参数中方向设为2，表示卖出
                    self.submit_stock_order(
                        istock, context["tradeTime"], 5, lastPrice, 1000, 2, label="sell"
                    )

在 initialize 中：

• 使用 sf.meta_code() 构建一个指标：上一根K线的收盘价**（``prev(close)``）**；

• 使用 subscribe_indicator 订阅此指标，类型为 KLINE（即K线行情数据）；

• 订阅结果将自动在后续的 on_bar 回调中提供；

在 on_bar 中：

• 遍历当前时间点的所有标的；

• 使用 indicator[istock]["lastp"] 获取上一根K线收盘价；

• 获取当前标的的持仓信息；

• 判断买入或卖出条件，调用 submit_stock_order 进行下单；

配置回测参数

• 资产类型为股票，数据为分钟级别K线；

• 设置初始资金为 1 亿；

• 设置交易佣金为万分之1.5，免税；

• 使用 frequency=0 表示不做降频回测，而是按照原始数据的时间戳驱动。

# 创建回测配置
config = backtest.StockConfig()
# 设置回测时间范围
config.start_date = sf.scalar("2021.01.01", type="DATE")
config.end_date = sf.scalar("2021.12.31", type="DATE")
# 设置资产类型和数据频率
config.asset_type = backtest.AssetType.STOCK
config.frequency = 0  # 不做降频回测
# 初始资金和交易成本
config.cash = 100000000
config.commission = 0.00015  # 佣金
config.tax = 0.0             # 税费
# 使用分钟级市场数据
config.data_type = backtest.MarketType.MINUTE

数据加载

# 加载股票数据文件
stocks = F.loadText('PATH_TO_DATA.csv')
# SQL 查询：按股票（symbol）分组并按时间排序，提取关键信息
msg_table = sf.sql("""
    select symbol,
           timestamp(tradeTime) as tradeTime,       // 时间戳
           open, low, high,                         // 当期开盘、最低、最高价
           prev(open) as close,                     // 上一条记录的开盘价作为收盘价
           long(volume) as volume,                  // 成交量
           amount,                                  // 成交额
           double(upLimitPrice) as upLimitPrice,    // 涨停价
           double(downLimitPrice) as downLimitPrice,// 跌停价
           prevClose as prevClosePrice              // 昨日收盘价
    from stocks
    context by symbol                               // 按股票分组处理（时间序列上下文）
    order by tradeTime                              // 按时间排序
""", vars={'stocks': stocks})

• 使用 loadText 读取CSV格式行情；

• 通过 SQL 构造标准化的 K 线行情表；

• 字段包含 symbol, open, close, volume, prevClose 等回测需要的字段；

启动回测并导入数据

# 创建回测器，使用自定义策略和配置
backtester = backtest.Backtester(MyStrategy, config)
# 加载市场数据到回测器中
backtester.append_data(msg_table)

完成上述步骤后，回测器将自动驱动策略运行，输出买卖点信息。

分析结果

完成回测后，用户可以通过 account 对象获取策略的各类回测结果数据，包括每日持仓、账户净值、收益统计与交易明细等。

# 获取默认账户
account = backtester.accounts[backtest.AccountType.DEFAULT]
# 打印常用回测结果
print("每日持仓：", account.get_daily_position())          # 每日股票持仓情况
print("每日权益指标：", account.daily_total_portfolios)     # 每日总资产（权益）数据
print("收益汇总：", account.return_summary)                 # 总体收益统计（如年化收益、最大回撤等）
print("成交记录：", account.trade_details)                 # 每笔交易的明细记录

常用回测结果接口说明

接口名称	数据结构	说明
cash	DOUBLE	当前账户剩余现金
trade_details	TABLE	全部交易记录明细（含时间、方向、价格、数量等）
get_daily_position()	TABLE	每日持仓情况（含标的代码、方向、持仓量、浮动盈亏等）
total_portfolios	TABLE	策略运行期内的总账户指标（如当前市值、单位净值、收益率等）
daily_total_portfolios	TABLE	每日账户指标，包括每日市值、单位净值、盈亏、收益率等
return_summary	TABLE	策略回测的收益概览指标，如累计收益率、年化收益、夏普比率等（具体字段由框架计算提供）

进阶：订阅指标

在本节中，我们将为策略引入一个简单的涨跌幅指标，并展示如何基于这个指标进行下单。Swordfish 回测框架提供了快捷的方式来注册、订阅、调用这些指标。

完整代码在附件“第七节回测脚本”中，由于 @F.swordfish_udf 无法在交互命令行中使用，因此将附件中完整代码保存为 .py 文件后，在命令行中输入 python 文件名.py 即可运行文件。

定义指标函数

# 定义一个有状态（is_state=True）的 UDF，用于计算涨跌幅
@F.swordfish_udf(mode="translate", is_state=True)
def zdf(last_price, prev_close_price):
    return last_price / prev_close_price - 1  # 涨跌幅 = (现价 / 昨收) - 1

该函数计算涨跌幅（涨幅 = (当前价格 ÷ 昨日收盘价) − 1）。

• @swordfish_udf(...)：用于注册该函数为可在回测中使用的指标；

• mode="translate"：表示该函数将转换为元数据表达式，可用于 subscribe_indicator()；

策略中订阅指标

在 initialize() 函数中订阅该指标，用于后续行情触发时调用：

def initialize(self, context):
    # 设置每个标的的最大持仓数量
    context["max_pos"] = 500
    # 新建一个字典存储标的的开仓信息
    context["open"] = sf.dictionary(key_type="STRING", val_type="BOOL")
    # 定义指标计算逻辑（这里使用之前定义的涨跌幅 zdf 函数）
    with sf.meta_code() as m:
        m_zdf = zdf(m.col("lastPrice"), m.col("prevClosePrice"))
    # 订阅快照数据中的自定义指标（zdf）
    self.subscribe_indicator(backtest.MarketDataType.SNAPSHOT, {
        'zdf': m_zdf
    })

• 在上下文字典 context 中存储每个标的的最大持仓数量以及开仓信息。

• sf.meta_code() 是 Swordfish 中将指标转换成元代码对象。

• m.col("...") 获取市场数据中的字段引用。

• 最后通过 subscribe_indicator() 订阅自定义指标。可以看到第一个参数为backtest.MarketDataType.SNAPSHOT，代表订阅后，系统会自动在每个 snapshot 时间点计算该指标，并传入 on_snapshot() 中供调用。第二个参数为一个字典，key 值为指标名称，value 为对应指标的元代码。

使用指标触发交易逻辑

在 on_snapshot() 中，我们使用指标值进行策略判断，并下达交易指令：

def on_snapshot(self, context, msg, indicator):
    symbol = msg["symbol"]                          # 当前股票代码
    best_ask = msg["offerPrice"][0]                 # 最优卖一价
    long_pos = self.accounts[backtest.AccountType.DEFAULT]\
        .get_position(symbol)["longPosition"]       # 当前持仓数量
    opens = context["open"]                         # 已开仓标记字典
    # 满足以下条件则下单买入：
    # 1. 涨幅超过1%
    # 2. 持仓未超最大限制
    # 3. 尚未对该股票下过单
    if indicator["zdf"] > 0.01 and long_pos <= context["max_pos"] and not opens.get(symbol, False):
        self.submit_stock_order(
            symbol, context["tradeTime"],           # 股票代码和时间
            5, best_ask, 100, 1,                    # 委托参数：限价买、价格、数量等
            label="buy"
        )
        opens[symbol] = True  # 标记该股票已下单，防止重复买入
    context["open"] = opens  # 更新上下文状态

策略逻辑说明：

• 取出当前标的的涨跌幅指标值：indicator["zdf"]；

• 如果涨幅超过 1%，且未达到最大持仓，且尚未对该标的下单，则执行买入；

• 使用 submit_stock_order() 提交订单；

• 通过 context["open"] 标记该标的已开仓，避免重复交易。

完整的代码详见附件。

回测引擎最佳实践

本章将介绍在使用 Swordfish 回测框架中应关注的重要事项，包含回测框架、代码逻辑以及数据处理三个方面。

回测框架配置面

指标优化配置

Swordfish 支持实时订阅指标。在策略中涉及大量技术指标（如均线、MACD、涨跌幅等）时，建议开启 config.enable_indicator_optimize = True。启用后，在回测时会对指标计算逻辑进行批量优化，避免重复计算从而提升性能。

指标计算推荐方式优先级（从高到低）：

1. 加载外部已计算指标（性能最佳）

2. 在初始化中订阅指标，由框架实时计算

3. 在回调函数中手动计算指标（性能最差，不推荐）

合理配置撮合模式

复杂撮合（如考虑订单队列位置、部分成交拆分）会增加计算成本。若回测阶段对成交率无严格要求，可采用简化撮合： config.matching_mode = 3 。此模式直接按委托价成交，可有效减少撮合运算时间。

代码逻辑优化

合理使用 context 存储复用对象

回调函数（如 on_bar, on_tick 等）在高频策略里每秒可能触发数千次，若在回调中频繁创建对象（如列表、字典、DataFrame），会造成大量内存分配与 GC，降低性能，建议，在 initialize 函数中，通过 context 预先创建可能会用到的对象或者缓存容器，在回调函数中仅复用对象，避免重复创建。

减少回调中的 IO / 大操作

避免在回调函数中执行 IO 操作，如 print 、文件写入、数据库查询之类。可以考虑将需要的信息存入 context 中，最后再通过 finalize 函数将其导出。比如要记录每次开仓时的参数日志，可以这样操作：

def initialize(self, context):
    context["entry_log"] = []
def on_bar(self, context, msg, indicator):
    context["entry_log"].append(....)
    # ... 交易逻辑

使用短路逻辑减少无效计算

将最易判断、且最可能提前退出的条件放在前面，减少不必要计算。

# 反面示例（先算昂贵指标）：
if indicator["ma5"] < msg['close'] and long_pos == 0:  # 先判断指标，再判断持仓
    # ... 交易逻辑
# 正面示例（先判断简单条件）：
# 先判断持仓（简单逻辑），再计算指标
if long_pos == 0 and indicator["ma5"] < msg['close']:
    # ... 交易逻辑

数据处理

离线计算静态指标

对于无需实时更新的静态指标，避免在回测中订阅指标进行实时计算。可以通过 sf.sql 向量化计算，代替 Python 的循环计算，避免在回测中逐条行情触发时重复计算，充分发挥 Swordfish 的计算性能优势。

批量注入行情数据

在回测过程中，Swordfish 会以事件方式将数据批量注入回测引擎。建议在回测阶段批量写入行情。当回测引擎接收到同一时间的行情时，可以同时批量读取缓存数据，提高行情解析的速度。

总结

本文介绍了如何使用 Swordfish 回测框架构建和运行量化策略，包括策略模板编写、回测参数配置、行情数据加载、自定义指标订阅及交易逻辑实现等内容。

Swordfish 提供了对多资产、多频率、高性能计算的良好支持，具备强大的可扩展性，适合进行中高频、多标的或组合类策略的研究与验证。

附件

• 第一节回测框架脚本: simplest_framework.py

• 第五节模拟数据: mink_data.csv

• 第五节回测脚本: trend_following_strategy.py

• 第七节回测脚本: indicator_subscription.py